[lttng-dev] [URCU PATCH 0/3] wait-free concurrent queues (wfcqueue)

[lttng-dev] [URCU PATCH 0/3] wait-free concurrent queues (wfcqueue)

[Mathieu Desnoyers]

Implement wait-free concurrent queues, with a new API different from
wfqueue.h, which is already provided by Userspace RCU. The advantage of
splitting the head and tail objects of the queue into different
arguments is to allow these to sit on different cache-lines, thus
eliminating false-sharing, leading to a 2.3x speed increase.

This API also introduces a "splice" operation, which moves all nodes
from one queue into another, and postpones the synchronization to either
dequeue or iteration on the list. The splice operation does not need to
touch every single node of the queue it moves them from. Moreover, the
splice operation only needs to ensure mutual exclusion with other
dequeuers, iterations, and splice operations from the list it splices
from, but acts as a simple enqueuer on the list it splices into (no
mutual exclusion needed for that list).

Feedback is welcome,

Thanks!

Mathieu

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[lttng-dev] [URCU PATCH 1/3] wfcqueue: implement concurrency-efficient queue

[Mathieu Desnoyers]

This new API simplify the wfqueue implementation, and brings a 2.3x to
2.6x performance boost due to the ability to eliminate false-sharing
between enqueue and dequeue.

This work is derived from the patch from Lai Jiangshan submitted as
"urcu: new wfqueue implementation"
(http://lists.lttng.org/pipermail/lttng-dev/2012-August/018379.html)

Its changelog:

> Some guys would be surprised by this fact:
> There are already TWO implementations of wfqueue in urcu.
>
> The first one is in urcu/static/wfqueue.h:
> 1) enqueue: exchange the tail and then update previous->next
> 2) dequeue: wait for first node's next pointer and them shift, a dummy node
> 	is introduced to avoid the queue->tail become NULL when shift.
>
> The second one shares some code with the first one, and the left code
> are spreading in urcu-call-rcu-impl.h:
> 1) enqueue: share with the first one
> 2) no dequeue operation: and no shift, so it don't need dummy node,
> 	Although the dummy node is queued when initialization, but it is removed
> 	after the first dequeue_all operation in call_rcu_thread().
> 	call_rcu_data_free() forgets to handle the dummy node if it is not removed.
> 3)dequeue_all: record the old head and tail, and queue->head become the special
> 	tail node.(atomic record the tail and change the tail).
>
> The second implementation's code are spreading, bad for review, and it is not
> tested by tests/test_urcu_wfq.
>
> So we need a better implementation avoid the dummy node dancing and can service
> both generic wfqueue APIs and dequeue_all API for call rcu.
>
> The new implementation:
> 1) enqueue: share with the first one/original implementation.
> 2) dequeue: shift when node count >= 2, cmpxchg when node count = 1.
> 	no dummy node, save memory.
> 3) dequeue_all: simply set queue->head.next to NULL, xchg the tail
> 	and return the old head.next.
>
> More implementation details are in the code.
> tests/test_urcu_wfq will be update in future for testing new APIs.

The patch proposed by Lai brings a very interesting simplification to
the single-node handling (which is kept here), and moves all queue
handling code away from call_rcu implementation, back into the wfqueue
code. This has the benefit to allow testing enhancements.

I modified it so the API does not expose implementation details to the
user (e.g. ___cds_wfq_node_sync_next). I added a "splice" operation and
a for loop iterator which should allow wfqueue users to use the list
very efficiently both from LGPL/GPL code and from non-LGPL-compatible
code.

I also changed the API so the queue head and tail are now two separate
structures: it allows the queue user to place these as they like, either
on different cache lines (to eliminate false-sharing), or close one to
another (on same cache-line) in case a queue is spliced onto the stack
and not concurrently accessed.

Benchmarks performed on Intel(R) Core(TM) i7-3520M CPU @ 2.90GHz
(dual-core, with hyperthreading)

Benchmark invoked:
for a in $(seq 1 10); do ./test_urcu_wfq 1 1 10 -a 0 -a 2; done

(using cpu number 0 and 2, which should correspond to two cores of my
Intel 2-core/hyperthread processor)

Before patch:

testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     97274297 nr_dequeues     80745742 successful enqueues     97274297 successful dequeues     80745321 end_dequeues 16528976 nr_ops    178020039
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     92300568 nr_dequeues     75019529 successful enqueues     92300568 successful dequeues     74973237 end_dequeues 17327331 nr_ops    167320097
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     93516443 nr_dequeues     75846726 successful enqueues     93516443 successful dequeues     75826578 end_dequeues 17689865 nr_ops    169363169
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     94160362 nr_dequeues     77967638 successful enqueues     94160362 successful dequeues     77967638 end_dequeues 16192724 nr_ops    172128000
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     97491956 nr_dequeues     81001191 successful enqueues     97491956 successful dequeues     81000247 end_dequeues 16491709 nr_ops    178493147
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     94101298 nr_dequeues     75650510 successful enqueues     94101298 successful dequeues     75649318 end_dequeues 18451980 nr_ops    169751808
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     94742803 nr_dequeues     75402105 successful enqueues     94742803 successful dequeues     75341859 end_dequeues 19400944 nr_ops    170144908
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     92198835 nr_dequeues     75037877 successful enqueues     92198835 successful dequeues     75027605 end_dequeues 17171230 nr_ops    167236712
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     94159560 nr_dequeues     77895972 successful enqueues     94159560 successful dequeues     77858442 end_dequeues 16301118 nr_ops    172055532
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     96059399 nr_dequeues     80115442 successful enqueues     96059399 successful dequeues     80066843 end_dequeues 15992556 nr_ops    176174841

After patch:

testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    221229322 nr_dequeues    210645491 successful enqueues    221229322 successful dequeues    210645088 end_dequeues 10584234 nr_ops    431874813
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    219803943 nr_dequeues    210377337 successful enqueues    219803943 successful dequeues    210368680 end_dequeues 9435263 nr_ops    430181280
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    237006358 nr_dequeues    237035340 successful enqueues    237006358 successful dequeues    236997050 end_dequeues 9308 nr_ops    474041698
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    235822443 nr_dequeues    235815942 successful enqueues    235822443 successful dequeues    235814020 end_dequeues 8423 nr_ops    471638385
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    235825567 nr_dequeues    235811803 successful enqueues    235825567 successful dequeues    235810526 end_dequeues 15041 nr_ops    471637370
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    221974953 nr_dequeues    210938190 successful enqueues    221974953 successful dequeues    210938190 end_dequeues 11036763 nr_ops    432913143
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    237994492 nr_dequeues    237938119 successful enqueues    237994492 successful dequeues    237930648 end_dequeues 63844 nr_ops    475932611
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    220634365 nr_dequeues    210491382 successful enqueues    220634365 successful dequeues    210490995 end_dequeues 10143370 nr_ops    431125747
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    237388065 nr_dequeues    237401251 successful enqueues    237388065 successful dequeues    237380295 end_dequeues 7770 nr_ops    474789316
testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    221201436 nr_dequeues    210831162 successful enqueues    221201436 successful dequeues    210831162 end_dequeues 10370274 nr_ops    432032598

Summary: Both enqueue and dequeue speed increase: around 2.3x speedup
for enqueue, and around 2.6x for dequeue.

We can verify that:
   successful enqueues - successful dequeues = end_dequeues

For all runs (ensures correctness: no lost node).

CC: Lai Jiangshan <laijs at cn.fujitsu.com>
CC: Paul McKenney <paulmck at linux.vnet.ibm.com>
Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
---
diff --git a/Makefile.am b/Makefile.am
index 2396fcf..ffdca9a 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -16,7 +16,7 @@ nobase_dist_include_HEADERS = urcu/compiler.h urcu/hlist.h urcu/list.h \
 		urcu/uatomic/generic.h urcu/arch/generic.h urcu/wfstack.h \
 		urcu/wfqueue.h urcu/rculfstack.h urcu/rculfqueue.h \
 		urcu/ref.h urcu/cds.h urcu/urcu_ref.h urcu/urcu-futex.h \
-		urcu/uatomic_arch.h urcu/rculfhash.h \
+		urcu/uatomic_arch.h urcu/rculfhash.h urcu/wfcqueue.h \
 		$(top_srcdir)/urcu/map/*.h \
 		$(top_srcdir)/urcu/static/*.h \
 		urcu/tls-compat.h
@@ -53,7 +53,7 @@ lib_LTLIBRARIES = liburcu-common.la \
 # liburcu-common contains wait-free queues (needed by call_rcu) as well
 # as futex fallbacks.
 #
-liburcu_common_la_SOURCES = wfqueue.c wfstack.c $(COMPAT)
+liburcu_common_la_SOURCES = wfqueue.c wfcqueue.c wfstack.c $(COMPAT)
 
 liburcu_la_SOURCES = urcu.c urcu-pointer.c $(COMPAT)
 liburcu_la_LIBADD = liburcu-common.la
diff --git a/urcu/static/wfcqueue.h b/urcu/static/wfcqueue.h
new file mode 100644
index 0000000..a989984
--- /dev/null
+++ b/urcu/static/wfcqueue.h
@@ -0,0 +1,380 @@
+#ifndef _URCU_WFCQUEUE_STATIC_H
+#define _URCU_WFCQUEUE_STATIC_H
+
+/*
+ * wfcqueue-static.h
+ *
+ * Userspace RCU library - Concurrent Queue with Wait-Free Enqueue/Blocking Dequeue
+ *
+ * TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See wfcqueue.h for linking
+ * dynamically with the userspace rcu library.
+ *
+ * Copyright 2010-2012 - Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
+ * Copyright 2011-2012 - Lai Jiangshan <laijs at cn.fujitsu.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include <pthread.h>
+#include <assert.h>
+#include <poll.h>
+#include <stdbool.h>
+#include <urcu/compiler.h>
+#include <urcu/uatomic.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/*
+ * Concurrent queue with wait-free enqueue/blocking dequeue.
+ *
+ * Inspired from half-wait-free/half-blocking queue implementation done by
+ * Paul E. McKenney.
+ *
+ * Mutual exclusion of __cds_wfcq_* API
+ *
+ * Unless otherwise stated, the caller must ensure mutual exclusion of
+ * queue update operations "dequeue" and "splice" (for source queue).
+ * Queue read operations "first" and "next" need to be protected against
+ * concurrent "dequeue" and "splice" (for source queue) by the caller.
+ * "enqueue", "splice" (for destination queue), and "empty" are the only
+ * operations that can be used without any mutual exclusion.
+ * Mutual exclusion can be ensured by holding cds_wfcq_dequeue_lock().
+ *
+ * For convenience, cds_wfcq_dequeue_blocking() and
+ * cds_wfcq_splice_blocking() hold the dequeue lock.
+ */
+
+#define WFCQ_ADAPT_ATTEMPTS		10	/* Retry if being set */
+#define WFCQ_WAIT			10	/* Wait 10 ms if being set */
+
+/*
+ * cds_wfcq_node_init: initialize wait-free queue node.
+ */
+static inline void _cds_wfcq_node_init(struct cds_wfcq_node *node)
+{
+	node->next = NULL;
+}
+
+/*
+ * cds_wfcq_init: initialize wait-free queue.
+ */
+static inline void _cds_wfcq_init(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	int ret;
+
+	/* Set queue head and tail */
+	_cds_wfcq_node_init(&head->node);
+	tail->p = &head->node;
+	ret = pthread_mutex_init(&head->lock, NULL);
+	assert(!ret);
+}
+
+/*
+ * cds_wfcq_empty: return whether wait-free queue is empty.
+ *
+ * No memory barrier is issued. No mutual exclusion is required.
+ */
+static inline bool _cds_wfcq_empty(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	/*
+	 * Queue is empty if no node is pointed by head->node.next nor
+	 * tail->p. Even though the tail->p check is sufficient to find
+	 * out of the queue is empty, we first check head->node.next as a
+	 * common case to ensure that dequeuers do not frequently access
+	 * enqueuer's tail->p cache line.
+	 */
+	return CMM_LOAD_SHARED(head->node.next) == NULL
+		&& CMM_LOAD_SHARED(tail->p) == &head->node;
+}
+
+static inline void _cds_wfcq_dequeue_lock(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	int ret;
+
+	ret = pthread_mutex_lock(&head->lock);
+	assert(!ret);
+}
+
+static inline void _cds_wfcq_dequeue_unlock(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	int ret;
+
+	ret = pthread_mutex_unlock(&head->lock);
+	assert(!ret);
+}
+
+static inline void ___cds_wfcq_append(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail,
+		struct cds_wfcq_node *new_head,
+		struct cds_wfcq_node *new_tail)
+{
+	struct cds_wfcq_node *old_tail;
+
+	/*
+	 * Implicit memory barrier before uatomic_xchg() orders earlier
+	 * stores to data structure containing node and setting
+	 * node->next to NULL before publication.
+	 */
+	old_tail = uatomic_xchg(&tail->p, new_tail);
+
+	/*
+	 * Implicit memory barrier after uatomic_xchg() orders store to
+	 * q->tail before store to old_tail->next.
+	 *
+	 * At this point, dequeuers see a NULL tail->p->next, which
+	 * indicates that the queue is being appended to. The following
+	 * store will append "node" to the queue from a dequeuer
+	 * perspective.
+	 */
+	CMM_STORE_SHARED(old_tail->next, new_head);
+}
+
+/*
+ * cds_wfcq_enqueue: enqueue a node into a wait-free queue.
+ *
+ * Issues a full memory barrier before enqueue. No mutual exclusion is
+ * required.
+ */
+static inline void _cds_wfcq_enqueue(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail,
+		struct cds_wfcq_node *new_tail)
+{
+	___cds_wfcq_append(head, tail, new_tail, new_tail);
+}
+
+/*
+ * Waiting for enqueuer to complete enqueue and return the next node.
+ */
+static inline struct cds_wfcq_node *
+___cds_wfcq_node_sync_next(struct cds_wfcq_node *node)
+{
+	struct cds_wfcq_node *next;
+	int attempt = 0;
+
+	/*
+	 * Adaptative busy-looping waiting for enqueuer to complete enqueue.
+	 */
+	while ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
+		if (++attempt >= WFCQ_ADAPT_ATTEMPTS) {
+			poll(NULL, 0, WFCQ_WAIT);	/* Wait for 10ms */
+			attempt = 0;
+		} else {
+			caa_cpu_relax();
+		}
+	}
+
+	return next;
+}
+
+/*
+ * __cds_wfcq_first_blocking: get first node of a queue, without dequeuing.
+ *
+ * Content written into the node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * Should be called with cds_wfcq_dequeue_lock() held.
+ */
+static inline struct cds_wfcq_node *
+___cds_wfcq_first_blocking(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	struct cds_wfcq_node *node;
+
+	if (_cds_wfcq_empty(head, tail))
+		return NULL;
+	node = ___cds_wfcq_node_sync_next(&head->node);
+	/* Load head->node.next before loading node's content */
+	cmm_smp_read_barrier_depends();
+	return node;
+}
+
+/*
+ * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
+ *
+ * Content written into the node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * Should be called with cds_wfcq_dequeue_lock() held.
+ */
+static inline struct cds_wfcq_node *
+___cds_wfcq_next_blocking(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail,
+		struct cds_wfcq_node *node)
+{
+	struct cds_wfcq_node *next;
+
+	/*
+	 * Even though the following tail->p check is sufficient to find
+	 * out if we reached the end of the queue, we first check
+	 * node->next as a common case to ensure that iteration on nodes
+	 * do not frequently access enqueuer's tail->p cache line.
+	 */
+	if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
+		/* Load node->next before tail->p */
+		cmm_smp_rmb();
+		if (CMM_LOAD_SHARED(tail->p) == node)
+			return NULL;
+		next = ___cds_wfcq_node_sync_next(node);
+	}
+	/* Load node->next before loading next's content */
+	cmm_smp_read_barrier_depends();
+	return next;
+}
+
+/*
+ * __cds_wfcq_dequeue_blocking: dequeue a node from the queue.
+ *
+ * No need to go on a waitqueue here, as there is no possible state in which the
+ * list could cause dequeue to busy-loop needlessly while waiting for another
+ * thread to be scheduled. The queue appears empty until tail->next is set by
+ * enqueue.
+ *
+ * Content written into the node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * It is valid to reuse and free a dequeued node immediately.
+ * Should be called with cds_wfcq_dequeue_lock() held.
+ */
+static inline struct cds_wfcq_node *
+___cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	struct cds_wfcq_node *node, *next;
+
+	if (_cds_wfcq_empty(head, tail))
+		return NULL;
+
+	node = ___cds_wfcq_node_sync_next(&head->node);
+
+	if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
+		/*
+		 * @node is probably the only node in the queue.
+		 * Try to move the tail to &q->head.
+		 * q->head.next is set to NULL here, and stays
+		 * NULL if the cmpxchg succeeds. Should the
+		 * cmpxchg fail due to a concurrent enqueue, the
+		 * q->head.next will be set to the next node.
+		 * The implicit memory barrier before
+		 * uatomic_cmpxchg() orders load node->next
+		 * before loading q->tail.
+		 * The implicit memory barrier before uatomic_cmpxchg
+		 * orders load q->head.next before loading node's
+		 * content.
+		 */
+		_cds_wfcq_node_init(&head->node);
+		if (uatomic_cmpxchg(&tail->p, node, &head->node) == node)
+			return node;
+		next = ___cds_wfcq_node_sync_next(node);
+	}
+
+	/*
+	 * Move queue head forward.
+	 */
+	head->node.next = next;
+
+	/* Load q->head.next before loading node's content */
+	cmm_smp_read_barrier_depends();
+	return node;
+}
+
+/*
+ * __cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
+ *
+ * Dequeue all nodes from src_q.
+ * dest_q must be already initialized.
+ * Should be called with cds_wfcq_dequeue_lock() held on src_q.
+ */
+static inline void
+___cds_wfcq_splice_blocking(
+		struct cds_wfcq_head *dest_q_head,
+		struct cds_wfcq_tail *dest_q_tail,
+		struct cds_wfcq_head *src_q_head,
+		struct cds_wfcq_tail *src_q_tail)
+{
+	struct cds_wfcq_node *head, *tail;
+
+	if (_cds_wfcq_empty(src_q_head, src_q_tail))
+		return;
+
+	head = ___cds_wfcq_node_sync_next(&src_q_head->node);
+	_cds_wfcq_node_init(&src_q_head->node);
+
+	/*
+	 * Memory barrier implied before uatomic_xchg() orders store to
+	 * src_q->head before store to src_q->tail. This is required by
+	 * concurrent enqueue on src_q, which exchanges the tail before
+	 * updating the previous tail's next pointer.
+	 */
+	tail = uatomic_xchg(&src_q_tail->p, &src_q_head->node);
+
+	/*
+	 * Append the spliced content of src_q into dest_q. Does not
+	 * require mutual exclusion on dest_q (wait-free).
+	 */
+	___cds_wfcq_append(dest_q_head, dest_q_tail, head, tail);
+}
+
+/*
+ * cds_wfcq_dequeue_blocking: dequeue a node from a wait-free queue.
+ *
+ * Content written into the node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * Mutual exlusion with (and only with) cds_wfcq_splice_blocking is
+ * ensured.
+ * It is valid to reuse and free a dequeued node immediately.
+ */
+static inline struct cds_wfcq_node *
+_cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	struct cds_wfcq_node *retval;
+
+	_cds_wfcq_dequeue_lock(head, tail);
+	retval = ___cds_wfcq_dequeue_blocking(head, tail);
+	_cds_wfcq_dequeue_unlock(head, tail);
+	return retval;
+}
+
+/*
+ * cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
+ *
+ * Dequeue all nodes from src_q.
+ * dest_q must be already initialized.
+ * Content written into the node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * Mutual exlusion with (and only with) cds_wfcq_dequeue_blocking is
+ * ensured.
+ */
+static inline void
+_cds_wfcq_splice_blocking(
+		struct cds_wfcq_head *dest_q_head,
+		struct cds_wfcq_tail *dest_q_tail,
+		struct cds_wfcq_head *src_q_head,
+		struct cds_wfcq_tail *src_q_tail)
+{
+	_cds_wfcq_dequeue_lock(src_q_head, src_q_tail);
+	___cds_wfcq_splice_blocking(dest_q_head, dest_q_tail,
+			src_q_head, src_q_tail);
+	_cds_wfcq_dequeue_unlock(src_q_head, src_q_tail);
+}
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _URCU_WFCQUEUE_STATIC_H */
diff --git a/urcu/wfcqueue.h b/urcu/wfcqueue.h
new file mode 100644
index 0000000..5576cbf
--- /dev/null
+++ b/urcu/wfcqueue.h
@@ -0,0 +1,263 @@
+#ifndef _URCU_WFCQUEUE_H
+#define _URCU_WFCQUEUE_H
+
+/*
+ * wfcqueue.h
+ *
+ * Userspace RCU library - Concurrent Queue with Wait-Free Enqueue/Blocking Dequeue
+ *
+ * Copyright 2010-2012 - Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
+ * Copyright 2011-2012 - Lai Jiangshan <laijs at cn.fujitsu.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include <pthread.h>
+#include <assert.h>
+#include <stdbool.h>
+#include <urcu/compiler.h>
+#include <urcu/arch.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/*
+ * Concurrent queue with wait-free enqueue/blocking dequeue.
+ *
+ * Inspired from half-wait-free/half-blocking queue implementation done by
+ * Paul E. McKenney.
+ */
+
+struct cds_wfcq_node {
+	struct cds_wfcq_node *next;
+};
+
+/*
+ * Do not put head and tail on the same cache-line if concurrent
+ * enqueue/dequeue are expected from many CPUs. This eliminates
+ * false-sharing between enqueue and dequeue.
+ */
+struct cds_wfcq_head {
+	struct cds_wfcq_node node;
+	pthread_mutex_t lock;
+};
+
+struct cds_wfcq_tail {
+	struct cds_wfcq_node *p;
+};
+
+#ifdef _LGPL_SOURCE
+
+#include <urcu/static/wfcqueue.h>
+
+#define cds_wfcq_node_init		_cds_wfcq_node_init
+#define cds_wfcq_init			_cds_wfcq_init
+#define cds_wfcq_empty			_cds_wfcq_empty
+#define cds_wfcq_enqueue		_cds_wfcq_enqueue
+
+/* Dequeue locking */
+#define cds_wfcq_dequeue_lock		_cds_wfcq_dequeue_lock
+#define cds_wfcq_dequeue_unlock		_cds_wfcq_dequeue_unlock
+
+/* Locking performed within cds_wfcq calls. */
+#define cds_wfcq_dequeue_blocking	_cds_wfcq_dequeue_blocking
+#define cds_wfcq_splice_blocking	_cds_wfcq_splice_blocking
+#define cds_wfcq_first_blocking		_cds_wfcq_first_blocking
+#define cds_wfcq_next_blocking		_cds_wfcq_next_blocking
+
+/* Locking ensured by caller by holding cds_wfcq_dequeue_lock() */
+#define __cds_wfcq_dequeue_blocking	___cds_wfcq_dequeue_blocking
+#define __cds_wfcq_splice_blocking	___cds_wfcq_splice_blocking
+#define __cds_wfcq_first_blocking	___cds_wfcq_first_blocking
+#define __cds_wfcq_next_blocking	___cds_wfcq_next_blocking
+
+#else /* !_LGPL_SOURCE */
+
+/*
+ * Mutual exclusion of cds_wfcq_* / __cds_wfcq_* API
+ *
+ * Unless otherwise stated, the caller must ensure mutual exclusion of
+ * queue update operations "dequeue" and "splice" (for source queue).
+ * Queue read operations "first" and "next" need to be protected against
+ * concurrent "dequeue" and "splice" (for source queue) by the caller.
+ * "enqueue", "splice" (for destination queue), and "empty" are the only
+ * operations that can be used without any mutual exclusion.
+ * Mutual exclusion can be ensured by holding cds_wfcq_dequeue_lock().
+ *
+ * For convenience, cds_wfcq_dequeue_blocking() and
+ * cds_wfcq_splice_blocking() hold the dequeue lock.
+ */
+
+/*
+ * cds_wfcq_node_init: initialize wait-free queue node.
+ */
+extern void cds_wfcq_node_init(struct cds_wfcq_node *node);
+
+/*
+ * cds_wfcq_init: initialize wait-free queue.
+ */
+extern void cds_wfcq_init(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail);
+
+/*
+ * cds_wfcq_empty: return whether wait-free queue is empty.
+ *
+ * No memory barrier is issued. No mutual exclusion is required.
+ */
+extern bool cds_wfcq_empty(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail);
+
+/*
+ * cds_wfcq_dequeue_lock: take the dequeue mutual exclusion lock.
+ */
+extern void cds_wfcq_dequeue_lock(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail);
+
+/*
+ * cds_wfcq_dequeue_unlock: release the dequeue mutual exclusion lock.
+ */
+extern void cds_wfcq_dequeue_unlock(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail);
+
+/*
+ * cds_wfcq_enqueue: enqueue a node into a wait-free queue.
+ *
+ * Issues a full memory barrier before enqueue. No mutual exclusion is
+ * required.
+ */
+extern void cds_wfcq_enqueue(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail,
+		struct cds_wfcq_node *node);
+
+/*
+ * cds_wfcq_dequeue_blocking: dequeue a node from a wait-free queue.
+ *
+ * Content written into the node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * It is valid to reuse and free a dequeued node immediately.
+ * Mutual exlusion with dequeuers is ensured internally.
+ */
+extern struct cds_wfcq_node *cds_wfcq_dequeue_blocking(
+		struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail);
+
+/*
+ * cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
+ *
+ * Dequeue all nodes from src_q.
+ * dest_q must be already initialized.
+ * Content written into the node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * Mutual exlusion with dequeuers is ensured internally.
+ */
+extern void cds_wfcq_splice_blocking(
+		struct cds_wfcq_head *dest_q_head,
+		struct cds_wfcq_tail *dest_q_tail,
+		struct cds_wfcq_head *src_q_head,
+		struct cds_wfcq_tail *src_q_tail);
+
+/*
+ * __cds_wfcq_dequeue_blocking:
+ *
+ * Content written into the node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * It is valid to reuse and free a dequeued node immediately.
+ * Should be called with cds_wfcq_dequeue_lock() held.
+ */
+extern struct cds_wfcq_node *__cds_wfcq_dequeue_blocking(
+		struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail);
+
+/*
+ * __cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
+ *
+ * Dequeue all nodes from src_q.
+ * dest_q must be already initialized.
+ * Content written into the node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * Should be called with cds_wfcq_dequeue_lock() held.
+ */
+extern void __cds_wfcq_splice_blocking(
+		struct cds_wfcq_head *dest_q_head,
+		struct cds_wfcq_tail *dest_q_tail,
+		struct cds_wfcq_head *src_q_head,
+		struct cds_wfcq_tail *src_q_tail);
+
+/*
+ * __cds_wfcq_first_blocking: get first node of a queue, without dequeuing.
+ *
+ * Content written into the node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * Should be called with cds_wfcq_dequeue_lock() held.
+ */
+extern struct cds_wfcq_node *__cds_wfcq_first_blocking(
+		struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail);
+
+/*
+ * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
+ *
+ * Content written into the node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * Should be called with cds_wfcq_dequeue_lock() held.
+ */
+extern struct cds_wfcq_node *__cds_wfcq_next_blocking(
+		struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail,
+		struct cds_wfcq_node *node);
+
+#endif /* !_LGPL_SOURCE */
+
+/*
+ * __cds_wfcq_for_each_blocking: Iterate over all nodes in a queue,
+ * without dequeuing them.
+ * @head: head of the queue (struct cds_wfcq_head pointer).
+ * @tail: tail of the queue (struct cds_wfcq_tail pointer).
+ * @node: iterator on the queue (struct cds_wfcq_node pointer).
+ *
+ * Content written into each node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * Should be called with cds_wfcq_dequeue_lock() held.
+ */
+#define __cds_wfcq_for_each_blocking(head, tail, node)		\
+	for (node = __cds_wfcq_first_blocking(head, tail);	\
+		node != NULL;					\
+		node = __cds_wfcq_next_blocking(head, tail, node))
+
+/*
+ * __cds_wfcq_for_each_blocking_safe: Iterate over all nodes in a queue,
+ * without dequeuing them. Safe against deletion.
+ * @head: head of the queue (struct cds_wfcq_head pointer).
+ * @tail: tail of the queue (struct cds_wfcq_tail pointer).
+ * @node: iterator on the queue (struct cds_wfcq_node pointer).
+ * @n: struct cds_wfcq_node pointer holding the next pointer (used
+ *     internally).
+ *
+ * Content written into each node before enqueue is guaranteed to be
+ * consistent, but no other memory ordering is ensured.
+ * Should be called with cds_wfcq_dequeue_lock() held.
+ */
+#define __cds_wfcq_for_each_blocking_safe(head, tail, node, n)		       \
+	for (node = __cds_wfcq_first_blocking(head, tail),		       \
+			n = (node ? __cds_wfcq_next_blocking(head, tail, node) : NULL); \
+		node != NULL;						       \
+		node = n, n = (node ? __cds_wfcq_next_blocking(head, tail, node) : NULL))
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _URCU_WFCQUEUE_H */
diff --git a/wfcqueue.c b/wfcqueue.c
new file mode 100644
index 0000000..1fa27ac
--- /dev/null
+++ b/wfcqueue.c
@@ -0,0 +1,116 @@
+/*
+ * wfcqueue.c
+ *
+ * Userspace RCU library - Concurrent queue with Wait-Free Enqueue/Blocking Dequeue
+ *
+ * Copyright 2010-2012 - Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
+ * Copyright 2011-2012 - Lai Jiangshan <laijs at cn.fujitsu.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+/* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
+#include "urcu/wfcqueue.h"
+#include "urcu/static/wfcqueue.h"
+
+/*
+ * library wrappers to be used by non-LGPL compatible source code.
+ */
+
+void cds_wfcq_node_init(struct cds_wfcq_node *node)
+{
+	_cds_wfcq_node_init(node);
+}
+
+void cds_wfcq_init(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	_cds_wfcq_init(head, tail);
+}
+
+bool cds_wfcq_empty(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+
+{
+	return _cds_wfcq_empty(head, tail);
+}
+
+void cds_wfcq_enqueue(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail,
+		struct cds_wfcq_node *node)
+{
+	_cds_wfcq_enqueue(head, tail, node);
+}
+
+void cds_wfcq_dequeue_lock(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	cds_wfcq_dequeue_lock(head, tail);
+}
+
+void cds_wfcq_dequeue_unlock(struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	cds_wfcq_dequeue_unlock(head, tail);
+}
+
+struct cds_wfcq_node *cds_wfcq_dequeue_blocking(
+		struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	return _cds_wfcq_dequeue_blocking(head, tail);
+}
+
+void cds_wfcq_splice_blocking(
+		struct cds_wfcq_head *dest_q_head,
+		struct cds_wfcq_tail *dest_q_tail,
+		struct cds_wfcq_head *src_q_head,
+		struct cds_wfcq_tail *src_q_tail)
+{
+	_cds_wfcq_splice_blocking(dest_q_head, dest_q_tail,
+				src_q_head, src_q_tail);
+}
+
+struct cds_wfcq_node *__cds_wfcq_dequeue_blocking(
+		struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	return ___cds_wfcq_dequeue_blocking(head, tail);
+}
+
+void __cds_wfcq_splice_blocking(
+		struct cds_wfcq_head *dest_q_head,
+		struct cds_wfcq_tail *dest_q_tail,
+		struct cds_wfcq_head *src_q_head,
+		struct cds_wfcq_tail *src_q_tail)
+{
+	___cds_wfcq_splice_blocking(dest_q_head, dest_q_tail,
+				src_q_head, src_q_tail);
+}
+
+struct cds_wfcq_node *__cds_wfcq_first_blocking(
+		struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail)
+{
+	return ___cds_wfcq_first_blocking(head, tail);
+}
+
+struct cds_wfcq_node *__cds_wfcq_next_blocking(
+		struct cds_wfcq_head *head,
+		struct cds_wfcq_tail *tail,
+		struct cds_wfcq_node *node)
+{
+	return ___cds_wfcq_next_blocking(head, tail, node);
+}

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[lttng-dev] [URCU PATCH 2/3] wfcqueue test

[Mathieu Desnoyers]

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
---
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 7d5ea82..81718bb 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -14,7 +14,9 @@ noinst_PROGRAMS = test_urcu test_urcu_dynamic_link test_urcu_timing \
         test_uatomic test_urcu_assign test_urcu_assign_dynamic_link \
         test_urcu_bp test_urcu_bp_dynamic_link test_cycles_per_loop \
 	test_urcu_lfq test_urcu_wfq test_urcu_lfs test_urcu_wfs \
+	test_urcu_wfcq \
 	test_urcu_wfq_dynlink test_urcu_wfs_dynlink \
+	test_urcu_wfcq_dynlink \
 	test_urcu_lfq_dynlink test_urcu_lfs_dynlink test_urcu_hash
 noinst_HEADERS = rcutorture.h
 
@@ -169,6 +171,13 @@ test_urcu_wfq_dynlink_SOURCES = test_urcu_wfq.c
 test_urcu_wfq_dynlink_CFLAGS = -DDYNAMIC_LINK_TEST $(AM_CFLAGS)
 test_urcu_wfq_dynlink_LDADD = $(URCU_COMMON_LIB)
 
+test_urcu_wfcq_SOURCES = test_urcu_wfcq.c $(COMPAT)
+test_urcu_wfcq_LDADD = $(URCU_COMMON_LIB)
+
+test_urcu_wfcq_dynlink_SOURCES = test_urcu_wfcq.c
+test_urcu_wfcq_dynlink_CFLAGS = -DDYNAMIC_LINK_TEST $(AM_CFLAGS)
+test_urcu_wfcq_dynlink_LDADD = $(URCU_COMMON_LIB)
+
 test_urcu_lfs_SOURCES = test_urcu_lfs.c $(URCU)
 test_urcu_lfs_LDADD = $(URCU_CDS_LIB)
 
diff --git a/tests/test_urcu_wfcq.c b/tests/test_urcu_wfcq.c
new file mode 100644
index 0000000..3141268
--- /dev/null
+++ b/tests/test_urcu_wfcq.c
@@ -0,0 +1,418 @@
+/*
+ * test_urcu_wfcq.c
+ *
+ * Userspace RCU library - example RCU-based lock-free concurrent queue
+ *
+ * Copyright February 2010 - Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
+ * Copyright February 2010 - Paolo Bonzini <pbonzini at redhat.com>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#define _GNU_SOURCE
+#include "../config.h"
+#include <stdio.h>
+#include <pthread.h>
+#include <stdlib.h>
+#include <stdint.h>
+#include <stdbool.h>
+#include <string.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <unistd.h>
+#include <stdio.h>
+#include <assert.h>
+#include <sched.h>
+#include <errno.h>
+
+#include <urcu/arch.h>
+#include <urcu/tls-compat.h>
+
+#ifdef __linux__
+#include <syscall.h>
+#endif
+
+/* hardcoded number of CPUs */
+#define NR_CPUS 16384
+
+#if defined(_syscall0)
+_syscall0(pid_t, gettid)
+#elif defined(__NR_gettid)
+static inline pid_t gettid(void)
+{
+	return syscall(__NR_gettid);
+}
+#else
+#warning "use pid as tid"
+static inline pid_t gettid(void)
+{
+	return getpid();
+}
+#endif
+
+#ifndef DYNAMIC_LINK_TEST
+#define _LGPL_SOURCE
+#endif
+#include <urcu.h>
+#include <urcu/wfcqueue.h>
+
+static volatile int test_go, test_stop;
+
+static unsigned long rduration;
+
+static unsigned long duration;
+
+/* read-side C.S. duration, in loops */
+static unsigned long wdelay;
+
+static inline void loop_sleep(unsigned long l)
+{
+	while(l-- != 0)
+		caa_cpu_relax();
+}
+
+static int verbose_mode;
+
+#define printf_verbose(fmt, args...)		\
+	do {					\
+		if (verbose_mode)		\
+			printf(fmt, args);	\
+	} while (0)
+
+static unsigned int cpu_affinities[NR_CPUS];
+static unsigned int next_aff = 0;
+static int use_affinity = 0;
+
+pthread_mutex_t affinity_mutex = PTHREAD_MUTEX_INITIALIZER;
+
+#ifndef HAVE_CPU_SET_T
+typedef unsigned long cpu_set_t;
+# define CPU_ZERO(cpuset) do { *(cpuset) = 0; } while(0)
+# define CPU_SET(cpu, cpuset) do { *(cpuset) |= (1UL << (cpu)); } while(0)
+#endif
+
+static void set_affinity(void)
+{
+	cpu_set_t mask;
+	int cpu;
+	int ret;
+
+	if (!use_affinity)
+		return;
+
+#if HAVE_SCHED_SETAFFINITY
+	ret = pthread_mutex_lock(&affinity_mutex);
+	if (ret) {
+		perror("Error in pthread mutex lock");
+		exit(-1);
+	}
+	cpu = cpu_affinities[next_aff++];
+	ret = pthread_mutex_unlock(&affinity_mutex);
+	if (ret) {
+		perror("Error in pthread mutex unlock");
+		exit(-1);
+	}
+
+	CPU_ZERO(&mask);
+	CPU_SET(cpu, &mask);
+#if SCHED_SETAFFINITY_ARGS == 2
+	sched_setaffinity(0, &mask);
+#else
+	sched_setaffinity(0, sizeof(mask), &mask);
+#endif
+#endif /* HAVE_SCHED_SETAFFINITY */
+}
+
+/*
+ * returns 0 if test should end.
+ */
+static int test_duration_dequeue(void)
+{
+	return !test_stop;
+}
+
+static int test_duration_enqueue(void)
+{
+	return !test_stop;
+}
+
+static DEFINE_URCU_TLS(unsigned long long, nr_dequeues);
+static DEFINE_URCU_TLS(unsigned long long, nr_enqueues);
+
+static DEFINE_URCU_TLS(unsigned long long, nr_successful_dequeues);
+static DEFINE_URCU_TLS(unsigned long long, nr_successful_enqueues);
+
+static unsigned int nr_enqueuers;
+static unsigned int nr_dequeuers;
+
+static struct cds_wfcq_head __attribute__((aligned(CAA_CACHE_LINE_SIZE))) head;
+static struct cds_wfcq_tail __attribute__((aligned(CAA_CACHE_LINE_SIZE))) tail;
+
+void *thr_enqueuer(void *_count)
+{
+	unsigned long long *count = _count;
+
+	printf_verbose("thread_begin %s, thread id : %lx, tid %lu\n",
+			"enqueuer", pthread_self(), (unsigned long)gettid());
+
+	set_affinity();
+
+	while (!test_go)
+	{
+	}
+	cmm_smp_mb();
+
+	for (;;) {
+		struct cds_wfcq_node *node = malloc(sizeof(*node));
+		if (!node)
+			goto fail;
+		cds_wfcq_node_init(node);
+		cds_wfcq_enqueue(&head, &tail, node);
+		URCU_TLS(nr_successful_enqueues)++;
+
+		if (caa_unlikely(wdelay))
+			loop_sleep(wdelay);
+fail:
+		URCU_TLS(nr_enqueues)++;
+		if (caa_unlikely(!test_duration_enqueue()))
+			break;
+	}
+
+	count[0] = URCU_TLS(nr_enqueues);
+	count[1] = URCU_TLS(nr_successful_enqueues);
+	printf_verbose("enqueuer thread_end, thread id : %lx, tid %lu, "
+		       "enqueues %llu successful_enqueues %llu\n",
+		       pthread_self(), (unsigned long)gettid(),
+		       URCU_TLS(nr_enqueues), URCU_TLS(nr_successful_enqueues));
+	return ((void*)1);
+
+}
+
+void *thr_dequeuer(void *_count)
+{
+	unsigned long long *count = _count;
+
+	printf_verbose("thread_begin %s, thread id : %lx, tid %lu\n",
+			"dequeuer", pthread_self(), (unsigned long)gettid());
+
+	set_affinity();
+
+	while (!test_go)
+	{
+	}
+	cmm_smp_mb();
+
+	for (;;) {
+		struct cds_wfcq_node *node =
+			cds_wfcq_dequeue_blocking(&head, &tail);
+
+		if (node) {
+			free(node);
+			URCU_TLS(nr_successful_dequeues)++;
+		}
+
+		URCU_TLS(nr_dequeues)++;
+		if (caa_unlikely(!test_duration_dequeue()))
+			break;
+		if (caa_unlikely(rduration))
+			loop_sleep(rduration);
+	}
+
+	printf_verbose("dequeuer thread_end, thread id : %lx, tid %lu, "
+		       "dequeues %llu, successful_dequeues %llu\n",
+		       pthread_self(), (unsigned long)gettid(),
+		       URCU_TLS(nr_dequeues), URCU_TLS(nr_successful_dequeues));
+	count[0] = URCU_TLS(nr_dequeues);
+	count[1] = URCU_TLS(nr_successful_dequeues);
+	return ((void*)2);
+}
+
+void test_end(unsigned long long *nr_dequeues)
+{
+	struct cds_wfcq_node *node;
+
+	do {
+		node = cds_wfcq_dequeue_blocking(&head, &tail);
+		if (node) {
+			free(node);
+			(*nr_dequeues)++;
+		}
+	} while (node);
+}
+
+void show_usage(int argc, char **argv)
+{
+	printf("Usage : %s nr_dequeuers nr_enqueuers duration (s)", argv[0]);
+	printf(" [-d delay] (enqueuer period (in loops))");
+	printf(" [-c duration] (dequeuer period (in loops))");
+	printf(" [-v] (verbose output)");
+	printf(" [-a cpu#] [-a cpu#]... (affinity)");
+	printf("\n");
+}
+
+int main(int argc, char **argv)
+{
+	int err;
+	pthread_t *tid_enqueuer, *tid_dequeuer;
+	void *tret;
+	unsigned long long *count_enqueuer, *count_dequeuer;
+	unsigned long long tot_enqueues = 0, tot_dequeues = 0;
+	unsigned long long tot_successful_enqueues = 0,
+			   tot_successful_dequeues = 0;
+	unsigned long long end_dequeues = 0;
+	int i, a;
+
+	if (argc < 4) {
+		show_usage(argc, argv);
+		return -1;
+	}
+
+	err = sscanf(argv[1], "%u", &nr_dequeuers);
+	if (err != 1) {
+		show_usage(argc, argv);
+		return -1;
+	}
+
+	err = sscanf(argv[2], "%u", &nr_enqueuers);
+	if (err != 1) {
+		show_usage(argc, argv);
+		return -1;
+	}
+	
+	err = sscanf(argv[3], "%lu", &duration);
+	if (err != 1) {
+		show_usage(argc, argv);
+		return -1;
+	}
+
+	for (i = 4; i < argc; i++) {
+		if (argv[i][0] != '-')
+			continue;
+		switch (argv[i][1]) {
+		case 'a':
+			if (argc < i + 2) {
+				show_usage(argc, argv);
+				return -1;
+			}
+			a = atoi(argv[++i]);
+			cpu_affinities[next_aff++] = a;
+			use_affinity = 1;
+			printf_verbose("Adding CPU %d affinity\n", a);
+			break;
+		case 'c':
+			if (argc < i + 2) {
+				show_usage(argc, argv);
+				return -1;
+			}
+			rduration = atol(argv[++i]);
+			break;
+		case 'd':
+			if (argc < i + 2) {
+				show_usage(argc, argv);
+				return -1;
+			}
+			wdelay = atol(argv[++i]);
+			break;
+		case 'v':
+			verbose_mode = 1;
+			break;
+		}
+	}
+
+	printf_verbose("running test for %lu seconds, %u enqueuers, "
+		       "%u dequeuers.\n",
+		       duration, nr_enqueuers, nr_dequeuers);
+	printf_verbose("Writer delay : %lu loops.\n", rduration);
+	printf_verbose("Reader duration : %lu loops.\n", wdelay);
+	printf_verbose("thread %-6s, thread id : %lx, tid %lu\n",
+			"main", pthread_self(), (unsigned long)gettid());
+
+	tid_enqueuer = malloc(sizeof(*tid_enqueuer) * nr_enqueuers);
+	tid_dequeuer = malloc(sizeof(*tid_dequeuer) * nr_dequeuers);
+	count_enqueuer = malloc(2 * sizeof(*count_enqueuer) * nr_enqueuers);
+	count_dequeuer = malloc(2 * sizeof(*count_dequeuer) * nr_dequeuers);
+	cds_wfcq_init(&head, &tail);
+
+	next_aff = 0;
+
+	for (i = 0; i < nr_enqueuers; i++) {
+		err = pthread_create(&tid_enqueuer[i], NULL, thr_enqueuer,
+				     &count_enqueuer[2 * i]);
+		if (err != 0)
+			exit(1);
+	}
+	for (i = 0; i < nr_dequeuers; i++) {
+		err = pthread_create(&tid_dequeuer[i], NULL, thr_dequeuer,
+				     &count_dequeuer[2 * i]);
+		if (err != 0)
+			exit(1);
+	}
+
+	cmm_smp_mb();
+
+	test_go = 1;
+
+	for (i = 0; i < duration; i++) {
+		sleep(1);
+		if (verbose_mode)
+			write (1, ".", 1);
+	}
+
+	test_stop = 1;
+
+	for (i = 0; i < nr_enqueuers; i++) {
+		err = pthread_join(tid_enqueuer[i], &tret);
+		if (err != 0)
+			exit(1);
+		tot_enqueues += count_enqueuer[2 * i];
+		tot_successful_enqueues += count_enqueuer[2 * i + 1];
+	}
+	for (i = 0; i < nr_dequeuers; i++) {
+		err = pthread_join(tid_dequeuer[i], &tret);
+		if (err != 0)
+			exit(1);
+		tot_dequeues += count_dequeuer[2 * i];
+		tot_successful_dequeues += count_dequeuer[2 * i + 1];
+	}
+	
+	test_end(&end_dequeues);
+
+	printf_verbose("total number of enqueues : %llu, dequeues %llu\n",
+		       tot_enqueues, tot_dequeues);
+	printf_verbose("total number of successful enqueues : %llu, "
+		       "successful dequeues %llu\n",
+		       tot_successful_enqueues, tot_successful_dequeues);
+	printf("SUMMARY %-25s testdur %4lu nr_enqueuers %3u wdelay %6lu "
+		"nr_dequeuers %3u "
+		"rdur %6lu nr_enqueues %12llu nr_dequeues %12llu "
+		"successful enqueues %12llu successful dequeues %12llu "
+		"end_dequeues %llu nr_ops %12llu\n",
+		argv[0], duration, nr_enqueuers, wdelay,
+		nr_dequeuers, rduration, tot_enqueues, tot_dequeues,
+		tot_successful_enqueues,
+		tot_successful_dequeues, end_dequeues,
+		tot_enqueues + tot_dequeues);
+	if (tot_successful_enqueues != tot_successful_dequeues + end_dequeues)
+		printf("WARNING! Discrepancy between nr succ. enqueues %llu vs "
+		       "succ. dequeues + end dequeues %llu.\n",
+		       tot_successful_enqueues,
+		       tot_successful_dequeues + end_dequeues);
+
+	free(count_enqueuer);
+	free(count_dequeuer);
+	free(tid_enqueuer);
+	free(tid_dequeuer);
+	return 0;
+}
-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[lttng-dev] [URCU PATCH 3/3] call_rcu: use wfcqueue, eliminate false-sharing

[Mathieu Desnoyers]

Eliminate false-sharing between call_rcu (enqueuer) and worker threads
on the queue head and tail.

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
---
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 81718bb..c92bbe6 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -30,14 +30,14 @@ if COMPAT_FUTEX
 COMPAT+=$(top_srcdir)/compat_futex.c
 endif
 
-URCU=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
-URCU_QSBR=$(top_srcdir)/urcu-qsbr.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
+URCU=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
+URCU_QSBR=$(top_srcdir)/urcu-qsbr.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
 # URCU_MB uses urcu.c but -DRCU_MB must be defined
-URCU_MB=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
+URCU_MB=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
 # URCU_SIGNAL uses urcu.c but -DRCU_SIGNAL must be defined
-URCU_SIGNAL=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
-URCU_BP=$(top_srcdir)/urcu-bp.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
-URCU_DEFER=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
+URCU_SIGNAL=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
+URCU_BP=$(top_srcdir)/urcu-bp.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
+URCU_DEFER=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
 
 URCU_COMMON_LIB=$(top_builddir)/liburcu-common.la
 URCU_LIB=$(top_builddir)/liburcu.la
diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
index 13b24ff..cf65992 100644
--- a/urcu-call-rcu-impl.h
+++ b/urcu-call-rcu-impl.h
@@ -21,6 +21,7 @@
  */
 
 #define _GNU_SOURCE
+#define _LGPL_SOURCE
 #include <stdio.h>
 #include <pthread.h>
 #include <signal.h>
@@ -35,7 +36,7 @@
 #include <sched.h>
 
 #include "config.h"
-#include "urcu/wfqueue.h"
+#include "urcu/wfcqueue.h"
 #include "urcu-call-rcu.h"
 #include "urcu-pointer.h"
 #include "urcu/list.h"
@@ -46,7 +47,14 @@
 /* Data structure that identifies a call_rcu thread. */
 
 struct call_rcu_data {
-	struct cds_wfq_queue cbs;
+	/*
+	 * Align the tail on cache line size to eliminate false-sharing
+	 * with head.
+	 */
+	struct cds_wfcq_tail __attribute__((aligned(CAA_CACHE_LINE_SIZE))) cbs_tail;
+	/* Alignment on cache line size will add padding here */
+
+	struct cds_wfcq_head cbs_head;
 	unsigned long flags;
 	int32_t futex;
 	unsigned long qlen; /* maintained for debugging. */
@@ -220,10 +228,7 @@ static void call_rcu_wake_up(struct call_rcu_data *crdp)
 static void *call_rcu_thread(void *arg)
 {
 	unsigned long cbcount;
-	struct cds_wfq_node *cbs;
-	struct cds_wfq_node **cbs_tail;
-	struct call_rcu_data *crdp = (struct call_rcu_data *)arg;
-	struct rcu_head *rhp;
+	struct call_rcu_data *crdp = (struct call_rcu_data *) arg;
 	int rt = !!(uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT);
 	int ret;
 
@@ -243,35 +248,33 @@ static void *call_rcu_thread(void *arg)
 		cmm_smp_mb();
 	}
 	for (;;) {
-		if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
-			while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
-				poll(NULL, 0, 1);
-			_CMM_STORE_SHARED(crdp->cbs.head, NULL);
-			cbs_tail = (struct cds_wfq_node **)
-				uatomic_xchg(&crdp->cbs.tail, &crdp->cbs.head);
+		struct cds_wfcq_head cbs_tmp_head;
+		struct cds_wfcq_tail cbs_tmp_tail;
+		struct cds_wfcq_node *cbs, *cbs_tmp_n;
+
+		cds_wfcq_init(&cbs_tmp_head, &cbs_tmp_tail);
+		__cds_wfcq_splice_blocking(&cbs_tmp_head, &cbs_tmp_tail,
+			&crdp->cbs_head, &crdp->cbs_tail);
+		if (!cds_wfcq_empty(&cbs_tmp_head, &cbs_tmp_tail)) {
 			synchronize_rcu();
 			cbcount = 0;
-			do {
-				while (cbs->next == NULL &&
-				       &cbs->next != cbs_tail)
-				       	poll(NULL, 0, 1);
-				if (cbs == &crdp->cbs.dummy) {
-					cbs = cbs->next;
-					continue;
-				}
-				rhp = (struct rcu_head *)cbs;
-				cbs = cbs->next;
+			__cds_wfcq_for_each_blocking_safe(&cbs_tmp_head,
+					&cbs_tmp_tail, cbs, cbs_tmp_n) {
+				struct rcu_head *rhp;
+
+				rhp = caa_container_of(cbs,
+					struct rcu_head, next);
 				rhp->func(rhp);
 				cbcount++;
-			} while (cbs != NULL);
+			}
 			uatomic_sub(&crdp->qlen, cbcount);
 		}
 		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
 			break;
 		rcu_thread_offline();
 		if (!rt) {
-			if (&crdp->cbs.head
-			    == _CMM_LOAD_SHARED(crdp->cbs.tail)) {
+			if (cds_wfcq_empty(&crdp->cbs_head,
+					&crdp->cbs_tail)) {
 				call_rcu_wait(crdp);
 				poll(NULL, 0, 10);
 				uatomic_dec(&crdp->futex);
@@ -317,7 +320,7 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
 	if (crdp == NULL)
 		urcu_die(errno);
 	memset(crdp, '\0', sizeof(*crdp));
-	cds_wfq_init(&crdp->cbs);
+	cds_wfcq_init(&crdp->cbs_head, &crdp->cbs_tail);
 	crdp->qlen = 0;
 	crdp->futex = 0;
 	crdp->flags = flags;
@@ -590,12 +593,12 @@ void call_rcu(struct rcu_head *head,
 {
 	struct call_rcu_data *crdp;
 
-	cds_wfq_node_init(&head->next);
+	cds_wfcq_node_init(&head->next);
 	head->func = func;
 	/* Holding rcu read-side lock across use of per-cpu crdp */
 	rcu_read_lock();
 	crdp = get_call_rcu_data();
-	cds_wfq_enqueue(&crdp->cbs, &head->next);
+	cds_wfcq_enqueue(&crdp->cbs_head, &crdp->cbs_tail, &head->next);
 	uatomic_inc(&crdp->qlen);
 	wake_call_rcu_thread(crdp);
 	rcu_read_unlock();
@@ -625,10 +628,6 @@ void call_rcu(struct rcu_head *head,
  */
 void call_rcu_data_free(struct call_rcu_data *crdp)
 {
-	struct cds_wfq_node *cbs;
-	struct cds_wfq_node **cbs_tail;
-	struct cds_wfq_node **cbs_endprev;
-
 	if (crdp == NULL || crdp == default_call_rcu_data) {
 		return;
 	}
@@ -638,17 +637,12 @@ void call_rcu_data_free(struct call_rcu_data *crdp)
 		while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0)
 			poll(NULL, 0, 1);
 	}
-	if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
-		while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
-			poll(NULL, 0, 1);
-		_CMM_STORE_SHARED(crdp->cbs.head, NULL);
-		cbs_tail = (struct cds_wfq_node **)
-			uatomic_xchg(&crdp->cbs.tail, &crdp->cbs.head);
+	if (!cds_wfcq_empty(&crdp->cbs_head, &crdp->cbs_tail)) {
 		/* Create default call rcu data if need be */
 		(void) get_default_call_rcu_data();
-		cbs_endprev = (struct cds_wfq_node **)
-			uatomic_xchg(&default_call_rcu_data, cbs_tail);
-		*cbs_endprev = cbs;
+		__cds_wfcq_splice_blocking(&default_call_rcu_data->cbs_head,
+			&default_call_rcu_data->cbs_tail,
+			&crdp->cbs_head, &crdp->cbs_tail);
 		uatomic_add(&default_call_rcu_data->qlen,
 			    uatomic_read(&crdp->qlen));
 		wake_call_rcu_thread(default_call_rcu_data);
diff --git a/urcu-call-rcu.h b/urcu-call-rcu.h
index f7eac8d..1dad0e2 100644
--- a/urcu-call-rcu.h
+++ b/urcu-call-rcu.h
@@ -32,7 +32,7 @@
 #include <stdlib.h>
 #include <pthread.h>
 
-#include <urcu/wfqueue.h>
+#include <urcu/wfcqueue.h>
 
 #ifdef __cplusplus
 extern "C" {
@@ -55,7 +55,7 @@ struct call_rcu_data;
  */
 
 struct rcu_head {
-	struct cds_wfq_node next;
+	struct cds_wfcq_node next;
 	void (*func)(struct rcu_head *head);
 };
 
-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[lttng-dev] [rp] [URCU PATCH 0/3] wait-free concurrent queues (wfcqueue)

[Paul E. McKenney]

On Tue, Oct 02, 2012 at 10:13:07AM -0400, Mathieu Desnoyers wrote:
> Implement wait-free concurrent queues, with a new API different from
> wfqueue.h, which is already provided by Userspace RCU. The advantage of
> splitting the head and tail objects of the queue into different
> arguments is to allow these to sit on different cache-lines, thus
> eliminating false-sharing, leading to a 2.3x speed increase.
> 
> This API also introduces a "splice" operation, which moves all nodes
> from one queue into another, and postpones the synchronization to either
> dequeue or iteration on the list. The splice operation does not need to
> touch every single node of the queue it moves them from. Moreover, the
> splice operation only needs to ensure mutual exclusion with other
> dequeuers, iterations, and splice operations from the list it splices
> from, but acts as a simple enqueuer on the list it splices into (no
> mutual exclusion needed for that list).
> 
> Feedback is welcome,

These look sane to me, though I must confess that the tail pointer
referencing the node rather than the node's next pointer did throw
me for a bit.  ;-)

							Thanx, Paul

[lttng-dev] [rp] [URCU PATCH 0/3] wait-free concurrent queues (wfcqueue)

[Mathieu Desnoyers]

* Paul E. McKenney (paulmck at linux.vnet.ibm.com) wrote:
> On Tue, Oct 02, 2012 at 10:13:07AM -0400, Mathieu Desnoyers wrote:
> > Implement wait-free concurrent queues, with a new API different from
> > wfqueue.h, which is already provided by Userspace RCU. The advantage of
> > splitting the head and tail objects of the queue into different
> > arguments is to allow these to sit on different cache-lines, thus
> > eliminating false-sharing, leading to a 2.3x speed increase.
> > 
> > This API also introduces a "splice" operation, which moves all nodes
> > from one queue into another, and postpones the synchronization to either
> > dequeue or iteration on the list. The splice operation does not need to
> > touch every single node of the queue it moves them from. Moreover, the
> > splice operation only needs to ensure mutual exclusion with other
> > dequeuers, iterations, and splice operations from the list it splices
> > from, but acts as a simple enqueuer on the list it splices into (no
> > mutual exclusion needed for that list).
> > 
> > Feedback is welcome,
> 
> These look sane to me, though I must confess that the tail pointer
> referencing the node rather than the node's next pointer did throw
> me for a bit.  ;-)

Yes, this was originally introduced with Lai's original patch to
wfqueue, which I think is a nice simplification: it's pretty much the
same thing to use the last node address as tail rather than the address
of its first member (its next pointer address (_not_ value)). It ends up
being the same address in this case, but more interestingly, we don't
have to use a struct cds_wfcq_node ** type: a simple struct
cds_wfcq_node *  suffice.

Thanks Paul, I will therefore merge these 3 patches with your Acked-by.

Lai, you are welcome to provide improvements to this code against the
master branch. I will gladly consider any change you propose.

Thanks!

Mathieu

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[lttng-dev] [rp] [URCU PATCH 0/3] wait-free concurrent queues (wfcqueue)

[Paul E. McKenney]

On Wed, Oct 03, 2012 at 05:04:36PM -0400, Mathieu Desnoyers wrote:
> * Paul E. McKenney (paulmck at linux.vnet.ibm.com) wrote:
> > On Tue, Oct 02, 2012 at 10:13:07AM -0400, Mathieu Desnoyers wrote:
> > > Implement wait-free concurrent queues, with a new API different from
> > > wfqueue.h, which is already provided by Userspace RCU. The advantage of
> > > splitting the head and tail objects of the queue into different
> > > arguments is to allow these to sit on different cache-lines, thus
> > > eliminating false-sharing, leading to a 2.3x speed increase.
> > > 
> > > This API also introduces a "splice" operation, which moves all nodes
> > > from one queue into another, and postpones the synchronization to either
> > > dequeue or iteration on the list. The splice operation does not need to
> > > touch every single node of the queue it moves them from. Moreover, the
> > > splice operation only needs to ensure mutual exclusion with other
> > > dequeuers, iterations, and splice operations from the list it splices
> > > from, but acts as a simple enqueuer on the list it splices into (no
> > > mutual exclusion needed for that list).
> > > 
> > > Feedback is welcome,
> > 
> > These look sane to me, though I must confess that the tail pointer
> > referencing the node rather than the node's next pointer did throw
> > me for a bit.  ;-)
> 
> Yes, this was originally introduced with Lai's original patch to
> wfqueue, which I think is a nice simplification: it's pretty much the
> same thing to use the last node address as tail rather than the address
> of its first member (its next pointer address (_not_ value)). It ends up
> being the same address in this case, but more interestingly, we don't
> have to use a struct cds_wfcq_node ** type: a simple struct
> cds_wfcq_node *  suffice.
> 
> Thanks Paul, I will therefore merge these 3 patches with your Acked-by.

Good point -- just confirming:

Acked-by: Paul E. McKenney <paulmck at linux.vnet.ibm.com>

> Lai, you are welcome to provide improvements to this code against the
> master branch. I will gladly consider any change you propose.
> 
> Thanks!
> 
> Mathieu
> 
> -- 
> Mathieu Desnoyers
> Operating System Efficiency R&D Consultant
> EfficiOS Inc.
> http://www.efficios.com
> 
> _______________________________________________
> rp mailing list
> rp at svcs.cs.pdx.edu
> http://svcs.cs.pdx.edu/mailman/listinfo/rp
> 

[lttng-dev] [URCU PATCH 3/3] call_rcu: use wfcqueue, eliminate false-sharing

[Lai Jiangshan]

On 10/02/2012 10:16 PM, Mathieu Desnoyers wrote:
> Eliminate false-sharing between call_rcu (enqueuer) and worker threads
> on the queue head and tail.
> 
> Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
> ---
> diff --git a/tests/Makefile.am b/tests/Makefile.am
> index 81718bb..c92bbe6 100644
> --- a/tests/Makefile.am
> +++ b/tests/Makefile.am
> @@ -30,14 +30,14 @@ if COMPAT_FUTEX
>  COMPAT+=$(top_srcdir)/compat_futex.c
>  endif
>  
> -URCU=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> -URCU_QSBR=$(top_srcdir)/urcu-qsbr.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> +URCU=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> +URCU_QSBR=$(top_srcdir)/urcu-qsbr.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
>  # URCU_MB uses urcu.c but -DRCU_MB must be defined
> -URCU_MB=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> +URCU_MB=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
>  # URCU_SIGNAL uses urcu.c but -DRCU_SIGNAL must be defined
> -URCU_SIGNAL=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> -URCU_BP=$(top_srcdir)/urcu-bp.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> -URCU_DEFER=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> +URCU_SIGNAL=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> +URCU_BP=$(top_srcdir)/urcu-bp.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> +URCU_DEFER=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
>  
>  URCU_COMMON_LIB=$(top_builddir)/liburcu-common.la
>  URCU_LIB=$(top_builddir)/liburcu.la
> diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> index 13b24ff..cf65992 100644
> --- a/urcu-call-rcu-impl.h
> +++ b/urcu-call-rcu-impl.h
> @@ -21,6 +21,7 @@
>   */
>  
>  #define _GNU_SOURCE
> +#define _LGPL_SOURCE
>  #include <stdio.h>
>  #include <pthread.h>
>  #include <signal.h>
> @@ -35,7 +36,7 @@
>  #include <sched.h>
>  
>  #include "config.h"
> -#include "urcu/wfqueue.h"
> +#include "urcu/wfcqueue.h"
>  #include "urcu-call-rcu.h"
>  #include "urcu-pointer.h"
>  #include "urcu/list.h"
> @@ -46,7 +47,14 @@
>  /* Data structure that identifies a call_rcu thread. */
>  
>  struct call_rcu_data {
> -	struct cds_wfq_queue cbs;
> +	/*
> +	 * Align the tail on cache line size to eliminate false-sharing
> +	 * with head.
> +	 */
> +	struct cds_wfcq_tail __attribute__((aligned(CAA_CACHE_LINE_SIZE))) cbs_tail;
> +	/* Alignment on cache line size will add padding here */
> +
> +	struct cds_wfcq_head cbs_head;


wrong here. In this code, cbs_tail and cbs_head are in the same cache line.

---

struct call_rcu_data {
	struct cds_wfcq_tail cbs_tail;
	struct cds_wfcq_head __attribute__((aligned(CAA_CACHE_LINE_SIZE))) cbs_head;
	/* other fields, can move some fields up to use the room between tail and head */
};

# cat test.c

struct a {
	int __attribute__((aligned(64))) i;
	int j;
};
struct b {
	int i;
	int __attribute__((aligned(64))) j;
};

void main(void)
{
	printf("%d,%d\n", sizeof(struct a), sizeof(struct b));
}

# ./a.out
64,128


>  	unsigned long flags;
>  	int32_t futex;
>  	unsigned long qlen; /* maintained for debugging. */
> @@ -220,10 +228,7 @@ static void call_rcu_wake_up(struct call_rcu_data *crdp)
>  static void *call_rcu_thread(void *arg)
>  {
>  	unsigned long cbcount;
> -	struct cds_wfq_node *cbs;
> -	struct cds_wfq_node **cbs_tail;
> -	struct call_rcu_data *crdp = (struct call_rcu_data *)arg;
> -	struct rcu_head *rhp;
> +	struct call_rcu_data *crdp = (struct call_rcu_data *) arg;
>  	int rt = !!(uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT);
>  	int ret;
>  
> @@ -243,35 +248,33 @@ static void *call_rcu_thread(void *arg)
>  		cmm_smp_mb();
>  	}
>  	for (;;) {
> -		if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
> -			while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
> -				poll(NULL, 0, 1);
> -			_CMM_STORE_SHARED(crdp->cbs.head, NULL);
> -			cbs_tail = (struct cds_wfq_node **)
> -				uatomic_xchg(&crdp->cbs.tail, &crdp->cbs.head);
> +		struct cds_wfcq_head cbs_tmp_head;
> +		struct cds_wfcq_tail cbs_tmp_tail;
> +		struct cds_wfcq_node *cbs, *cbs_tmp_n;
> +
> +		cds_wfcq_init(&cbs_tmp_head, &cbs_tmp_tail);
> +		__cds_wfcq_splice_blocking(&cbs_tmp_head, &cbs_tmp_tail,
> +			&crdp->cbs_head, &crdp->cbs_tail);
> +		if (!cds_wfcq_empty(&cbs_tmp_head, &cbs_tmp_tail)) {
>  			synchronize_rcu();
>  			cbcount = 0;
> -			do {
> -				while (cbs->next == NULL &&
> -				       &cbs->next != cbs_tail)
> -				       	poll(NULL, 0, 1);
> -				if (cbs == &crdp->cbs.dummy) {
> -					cbs = cbs->next;
> -					continue;
> -				}
> -				rhp = (struct rcu_head *)cbs;
> -				cbs = cbs->next;
> +			__cds_wfcq_for_each_blocking_safe(&cbs_tmp_head,
> +					&cbs_tmp_tail, cbs, cbs_tmp_n) {
> +				struct rcu_head *rhp;
> +
> +				rhp = caa_container_of(cbs,
> +					struct rcu_head, next);
>  				rhp->func(rhp);
>  				cbcount++;
> -			} while (cbs != NULL);
> +			}
>  			uatomic_sub(&crdp->qlen, cbcount);
>  		}
>  		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
>  			break;
>  		rcu_thread_offline();
>  		if (!rt) {
> -			if (&crdp->cbs.head
> -			    == _CMM_LOAD_SHARED(crdp->cbs.tail)) {
> +			if (cds_wfcq_empty(&crdp->cbs_head,
> +					&crdp->cbs_tail)) {
>  				call_rcu_wait(crdp);
>  				poll(NULL, 0, 10);
>  				uatomic_dec(&crdp->futex);
> @@ -317,7 +320,7 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
>  	if (crdp == NULL)
>  		urcu_die(errno);
>  	memset(crdp, '\0', sizeof(*crdp));
> -	cds_wfq_init(&crdp->cbs);
> +	cds_wfcq_init(&crdp->cbs_head, &crdp->cbs_tail);
>  	crdp->qlen = 0;
>  	crdp->futex = 0;
>  	crdp->flags = flags;
> @@ -590,12 +593,12 @@ void call_rcu(struct rcu_head *head,
>  {
>  	struct call_rcu_data *crdp;
>  
> -	cds_wfq_node_init(&head->next);
> +	cds_wfcq_node_init(&head->next);
>  	head->func = func;
>  	/* Holding rcu read-side lock across use of per-cpu crdp */
>  	rcu_read_lock();
>  	crdp = get_call_rcu_data();
> -	cds_wfq_enqueue(&crdp->cbs, &head->next);
> +	cds_wfcq_enqueue(&crdp->cbs_head, &crdp->cbs_tail, &head->next);
>  	uatomic_inc(&crdp->qlen);
>  	wake_call_rcu_thread(crdp);
>  	rcu_read_unlock();
> @@ -625,10 +628,6 @@ void call_rcu(struct rcu_head *head,
>   */
>  void call_rcu_data_free(struct call_rcu_data *crdp)
>  {
> -	struct cds_wfq_node *cbs;
> -	struct cds_wfq_node **cbs_tail;
> -	struct cds_wfq_node **cbs_endprev;
> -
>  	if (crdp == NULL || crdp == default_call_rcu_data) {
>  		return;
>  	}
> @@ -638,17 +637,12 @@ void call_rcu_data_free(struct call_rcu_data *crdp)
>  		while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0)
>  			poll(NULL, 0, 1);
>  	}
> -	if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
> -		while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
> -			poll(NULL, 0, 1);
> -		_CMM_STORE_SHARED(crdp->cbs.head, NULL);
> -		cbs_tail = (struct cds_wfq_node **)
> -			uatomic_xchg(&crdp->cbs.tail, &crdp->cbs.head);
> +	if (!cds_wfcq_empty(&crdp->cbs_head, &crdp->cbs_tail)) {
>  		/* Create default call rcu data if need be */
>  		(void) get_default_call_rcu_data();
> -		cbs_endprev = (struct cds_wfq_node **)
> -			uatomic_xchg(&default_call_rcu_data, cbs_tail);
> -		*cbs_endprev = cbs;
> +		__cds_wfcq_splice_blocking(&default_call_rcu_data->cbs_head,
> +			&default_call_rcu_data->cbs_tail,
> +			&crdp->cbs_head, &crdp->cbs_tail);
>  		uatomic_add(&default_call_rcu_data->qlen,
>  			    uatomic_read(&crdp->qlen));
>  		wake_call_rcu_thread(default_call_rcu_data);
> diff --git a/urcu-call-rcu.h b/urcu-call-rcu.h
> index f7eac8d..1dad0e2 100644
> --- a/urcu-call-rcu.h
> +++ b/urcu-call-rcu.h
> @@ -32,7 +32,7 @@
>  #include <stdlib.h>
>  #include <pthread.h>
>  
> -#include <urcu/wfqueue.h>
> +#include <urcu/wfcqueue.h>
>  
>  #ifdef __cplusplus
>  extern "C" {
> @@ -55,7 +55,7 @@ struct call_rcu_data;
>   */
>  
>  struct rcu_head {
> -	struct cds_wfq_node next;
> +	struct cds_wfcq_node next;
>  	void (*func)(struct rcu_head *head);
>  };
>  

[lttng-dev] [rp] [URCU PATCH 0/3] wait-free concurrent queues (wfcqueue)

[Lai Jiangshan]

On 10/04/2012 05:04 AM, Mathieu Desnoyers wrote:
> * Paul E. McKenney (paulmck at linux.vnet.ibm.com) wrote:
>> On Tue, Oct 02, 2012 at 10:13:07AM -0400, Mathieu Desnoyers wrote:
>>> Implement wait-free concurrent queues, with a new API different from
>>> wfqueue.h, which is already provided by Userspace RCU. The advantage of
>>> splitting the head and tail objects of the queue into different
>>> arguments is to allow these to sit on different cache-lines, thus
>>> eliminating false-sharing, leading to a 2.3x speed increase.
>>>
>>> This API also introduces a "splice" operation, which moves all nodes
>>> from one queue into another, and postpones the synchronization to either
>>> dequeue or iteration on the list. The splice operation does not need to
>>> touch every single node of the queue it moves them from. Moreover, the
>>> splice operation only needs to ensure mutual exclusion with other
>>> dequeuers, iterations, and splice operations from the list it splices
>>> from, but acts as a simple enqueuer on the list it splices into (no
>>> mutual exclusion needed for that list).
>>>
>>> Feedback is welcome,
>>
>> These look sane to me, though I must confess that the tail pointer
>> referencing the node rather than the node's next pointer did throw
>> me for a bit.  ;-)
> 
> Yes, this was originally introduced with Lai's original patch to
> wfqueue, which I think is a nice simplification: it's pretty much the
> same thing to use the last node address as tail rather than the address
> of its first member (its next pointer address (_not_ value)). It ends up
> being the same address in this case, but more interestingly, we don't
> have to use a struct cds_wfcq_node ** type: a simple struct
> cds_wfcq_node *  suffice.
> 
> Thanks Paul, I will therefore merge these 3 patches with your Acked-by.
> 
> Lai, you are welcome to provide improvements to this code against the
> master branch. I will gladly consider any change you propose.
> 

I did not remember that there is any improvement idea not included.
The patchset is OK for me.

I think you can reimplement wfqueue via wfcqueue without cacheline opt.

Thanks,
Lai

[lttng-dev] [URCU PATCH 3/3] call_rcu: use wfcqueue, eliminate false-sharing

[Mathieu Desnoyers]

* Lai Jiangshan (laijs at cn.fujitsu.com) wrote:
> On 10/02/2012 10:16 PM, Mathieu Desnoyers wrote:
> > Eliminate false-sharing between call_rcu (enqueuer) and worker threads
> > on the queue head and tail.
> > 
> > Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
> > ---
> > diff --git a/tests/Makefile.am b/tests/Makefile.am
> > index 81718bb..c92bbe6 100644
> > --- a/tests/Makefile.am
> > +++ b/tests/Makefile.am
> > @@ -30,14 +30,14 @@ if COMPAT_FUTEX
> >  COMPAT+=$(top_srcdir)/compat_futex.c
> >  endif
> >  
> > -URCU=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> > -URCU_QSBR=$(top_srcdir)/urcu-qsbr.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> > +URCU=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> > +URCU_QSBR=$(top_srcdir)/urcu-qsbr.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> >  # URCU_MB uses urcu.c but -DRCU_MB must be defined
> > -URCU_MB=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> > +URCU_MB=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> >  # URCU_SIGNAL uses urcu.c but -DRCU_SIGNAL must be defined
> > -URCU_SIGNAL=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> > -URCU_BP=$(top_srcdir)/urcu-bp.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> > -URCU_DEFER=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> > +URCU_SIGNAL=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> > +URCU_BP=$(top_srcdir)/urcu-bp.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> > +URCU_DEFER=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> >  
> >  URCU_COMMON_LIB=$(top_builddir)/liburcu-common.la
> >  URCU_LIB=$(top_builddir)/liburcu.la
> > diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> > index 13b24ff..cf65992 100644
> > --- a/urcu-call-rcu-impl.h
> > +++ b/urcu-call-rcu-impl.h
> > @@ -21,6 +21,7 @@
> >   */
> >  
> >  #define _GNU_SOURCE
> > +#define _LGPL_SOURCE
> >  #include <stdio.h>
> >  #include <pthread.h>
> >  #include <signal.h>
> > @@ -35,7 +36,7 @@
> >  #include <sched.h>
> >  
> >  #include "config.h"
> > -#include "urcu/wfqueue.h"
> > +#include "urcu/wfcqueue.h"
> >  #include "urcu-call-rcu.h"
> >  #include "urcu-pointer.h"
> >  #include "urcu/list.h"
> > @@ -46,7 +47,14 @@
> >  /* Data structure that identifies a call_rcu thread. */
> >  
> >  struct call_rcu_data {
> > -	struct cds_wfq_queue cbs;
> > +	/*
> > +	 * Align the tail on cache line size to eliminate false-sharing
> > +	 * with head.
> > +	 */
> > +	struct cds_wfcq_tail __attribute__((aligned(CAA_CACHE_LINE_SIZE))) cbs_tail;
> > +	/* Alignment on cache line size will add padding here */
> > +
> > +	struct cds_wfcq_head cbs_head;
> 
> 
> wrong here. In this code, cbs_tail and cbs_head are in the same cache line.
> 
> ---
> 
> struct call_rcu_data {
> 	struct cds_wfcq_tail cbs_tail;
> 	struct cds_wfcq_head __attribute__((aligned(CAA_CACHE_LINE_SIZE))) cbs_head;
> 	/* other fields, can move some fields up to use the room between tail and head */
> };
> 
> # cat test.c
> 
> struct a {
> 	int __attribute__((aligned(64))) i;
> 	int j;
> };
> struct b {
> 	int i;
> 	int __attribute__((aligned(64))) j;
> };
> 
> void main(void)
> {
> 	printf("%d,%d\n", sizeof(struct a), sizeof(struct b));
> }
> 
> # ./a.out
> 64,128
> 

Good point! While we are there, I notice that the "qlen" count, kept for
debugging, is causing false-sharing too. I wonder if we should split
this counter in two counters: nr_enqueue and nr_dequeue, which would sit
in two different cache lines ? It's mainly Paul who cares about this
counter. Thoughts ?

Here is the fix to the problem you noticed above:

commit b9f893b69fbc31baea418794938f4eb74cc4923a
Author: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
Date:   Mon Oct 8 10:44:38 2012 -0400

    Fix urcu-call-rcu-impl.h: false-sharing
    
    > >  struct call_rcu_data {
    > > -   struct cds_wfq_queue cbs;
    > > +   /*
    > > +    * Align the tail on cache line size to eliminate false-sharing
    > > +    * with head.
    > > +    */
    > > +   struct cds_wfcq_tail __attribute__((aligned(CAA_CACHE_LINE_SIZE))) cbs_tail;
    > > +   /* Alignment on cache line size will add padding here */
    > > +
    > > +   struct cds_wfcq_head cbs_head;
    >
    >
    > wrong here. In this code, cbs_tail and cbs_head are in the same cache line.
    
    Reported-by: Lai Jiangshan <laijs at cn.fujitsu.com>
    Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>

Thanks!

Mathieu

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[lttng-dev] [rp] [URCU PATCH 0/3] wait-free concurrent queues (wfcqueue)

[Mathieu Desnoyers]

* Lai Jiangshan (laijs at cn.fujitsu.com) wrote:
> On 10/04/2012 05:04 AM, Mathieu Desnoyers wrote:
> > * Paul E. McKenney (paulmck at linux.vnet.ibm.com) wrote:
> >> On Tue, Oct 02, 2012 at 10:13:07AM -0400, Mathieu Desnoyers wrote:
> >>> Implement wait-free concurrent queues, with a new API different from
> >>> wfqueue.h, which is already provided by Userspace RCU. The advantage of
> >>> splitting the head and tail objects of the queue into different
> >>> arguments is to allow these to sit on different cache-lines, thus
> >>> eliminating false-sharing, leading to a 2.3x speed increase.
> >>>
> >>> This API also introduces a "splice" operation, which moves all nodes
> >>> from one queue into another, and postpones the synchronization to either
> >>> dequeue or iteration on the list. The splice operation does not need to
> >>> touch every single node of the queue it moves them from. Moreover, the
> >>> splice operation only needs to ensure mutual exclusion with other
> >>> dequeuers, iterations, and splice operations from the list it splices
> >>> from, but acts as a simple enqueuer on the list it splices into (no
> >>> mutual exclusion needed for that list).
> >>>
> >>> Feedback is welcome,
> >>
> >> These look sane to me, though I must confess that the tail pointer
> >> referencing the node rather than the node's next pointer did throw
> >> me for a bit.  ;-)
> > 
> > Yes, this was originally introduced with Lai's original patch to
> > wfqueue, which I think is a nice simplification: it's pretty much the
> > same thing to use the last node address as tail rather than the address
> > of its first member (its next pointer address (_not_ value)). It ends up
> > being the same address in this case, but more interestingly, we don't
> > have to use a struct cds_wfcq_node ** type: a simple struct
> > cds_wfcq_node *  suffice.
> > 
> > Thanks Paul, I will therefore merge these 3 patches with your Acked-by.
> > 
> > Lai, you are welcome to provide improvements to this code against the
> > master branch. I will gladly consider any change you propose.
> > 
> 
> I did not remember that there is any improvement idea not included.
> The patchset is OK for me.

Great! Would you be OK if I commit the following patch ? Let me know if
you want me to put your signed-off-by on this (I can even put your email
as From if you like):


wfcqueue: update credits in patch documentation

Give credits to those responsible for the design and implementation of
commit 8ad4ce587f001ae026d5560ac509c2e48986130b, "wfcqueue: implement
concurrency-efficient queue", which happened through rounds of email and
patch exchanges.

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
---
diff --git a/urcu/static/wfcqueue.h b/urcu/static/wfcqueue.h
index a989984..153143d 100644
--- a/urcu/static/wfcqueue.h
+++ b/urcu/static/wfcqueue.h
@@ -41,8 +41,10 @@ extern "C" {
 /*
  * Concurrent queue with wait-free enqueue/blocking dequeue.
  *
- * Inspired from half-wait-free/half-blocking queue implementation done by
- * Paul E. McKenney.
+ * This queue has been designed and implemented collaboratively by
+ * Mathieu Desnoyers and Lai Jiangshan. Inspired from
+ * half-wait-free/half-blocking queue implementation done by Paul E.
+ * McKenney.
  *
  * Mutual exclusion of __cds_wfcq_* API
  *
diff --git a/urcu/wfcqueue.h b/urcu/wfcqueue.h
index 5576cbf..940dc7d 100644
--- a/urcu/wfcqueue.h
+++ b/urcu/wfcqueue.h
@@ -37,8 +37,10 @@ extern "C" {
 /*
  * Concurrent queue with wait-free enqueue/blocking dequeue.
  *
- * Inspired from half-wait-free/half-blocking queue implementation done by
- * Paul E. McKenney.
+ * This queue has been designed and implemented collaboratively by
+ * Mathieu Desnoyers and Lai Jiangshan. Inspired from
+ * half-wait-free/half-blocking queue implementation done by Paul E.
+ * McKenney.
  */
 
 struct cds_wfcq_node {


> I think you can reimplement wfqueue via wfcqueue without cacheline opt.

Hrm, semantically this can indeed be done, but I fear that we might not
be strictly ABI-compatible with the old wfqueue. So I would be tempted
to leave the old wfqueue implementation as-is, and maybe deprecate it at
some point. Thoughts ?

Thanks!

Mathieu

> 
> Thanks,
> Lai

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[lttng-dev] [URCU PATCH 3/3] call_rcu: use wfcqueue, eliminate false-sharing

[Paul E. McKenney]

On Mon, Oct 08, 2012 at 10:49:16AM -0400, Mathieu Desnoyers wrote:
> * Lai Jiangshan (laijs at cn.fujitsu.com) wrote:
> > On 10/02/2012 10:16 PM, Mathieu Desnoyers wrote:
> > > Eliminate false-sharing between call_rcu (enqueuer) and worker threads
> > > on the queue head and tail.
> > > 
> > > Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
> > > ---
> > > diff --git a/tests/Makefile.am b/tests/Makefile.am
> > > index 81718bb..c92bbe6 100644
> > > --- a/tests/Makefile.am
> > > +++ b/tests/Makefile.am
> > > @@ -30,14 +30,14 @@ if COMPAT_FUTEX
> > >  COMPAT+=$(top_srcdir)/compat_futex.c
> > >  endif
> > >  
> > > -URCU=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> > > -URCU_QSBR=$(top_srcdir)/urcu-qsbr.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> > > +URCU=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> > > +URCU_QSBR=$(top_srcdir)/urcu-qsbr.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> > >  # URCU_MB uses urcu.c but -DRCU_MB must be defined
> > > -URCU_MB=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> > > +URCU_MB=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> > >  # URCU_SIGNAL uses urcu.c but -DRCU_SIGNAL must be defined
> > > -URCU_SIGNAL=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> > > -URCU_BP=$(top_srcdir)/urcu-bp.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> > > -URCU_DEFER=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfqueue.c $(COMPAT)
> > > +URCU_SIGNAL=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> > > +URCU_BP=$(top_srcdir)/urcu-bp.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> > > +URCU_DEFER=$(top_srcdir)/urcu.c $(top_srcdir)/urcu-pointer.c $(top_srcdir)/wfcqueue.c $(COMPAT)
> > >  
> > >  URCU_COMMON_LIB=$(top_builddir)/liburcu-common.la
> > >  URCU_LIB=$(top_builddir)/liburcu.la
> > > diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> > > index 13b24ff..cf65992 100644
> > > --- a/urcu-call-rcu-impl.h
> > > +++ b/urcu-call-rcu-impl.h
> > > @@ -21,6 +21,7 @@
> > >   */
> > >  
> > >  #define _GNU_SOURCE
> > > +#define _LGPL_SOURCE
> > >  #include <stdio.h>
> > >  #include <pthread.h>
> > >  #include <signal.h>
> > > @@ -35,7 +36,7 @@
> > >  #include <sched.h>
> > >  
> > >  #include "config.h"
> > > -#include "urcu/wfqueue.h"
> > > +#include "urcu/wfcqueue.h"
> > >  #include "urcu-call-rcu.h"
> > >  #include "urcu-pointer.h"
> > >  #include "urcu/list.h"
> > > @@ -46,7 +47,14 @@
> > >  /* Data structure that identifies a call_rcu thread. */
> > >  
> > >  struct call_rcu_data {
> > > -	struct cds_wfq_queue cbs;
> > > +	/*
> > > +	 * Align the tail on cache line size to eliminate false-sharing
> > > +	 * with head.
> > > +	 */
> > > +	struct cds_wfcq_tail __attribute__((aligned(CAA_CACHE_LINE_SIZE))) cbs_tail;
> > > +	/* Alignment on cache line size will add padding here */
> > > +
> > > +	struct cds_wfcq_head cbs_head;
> > 
> > 
> > wrong here. In this code, cbs_tail and cbs_head are in the same cache line.
> > 
> > ---
> > 
> > struct call_rcu_data {
> > 	struct cds_wfcq_tail cbs_tail;
> > 	struct cds_wfcq_head __attribute__((aligned(CAA_CACHE_LINE_SIZE))) cbs_head;
> > 	/* other fields, can move some fields up to use the room between tail and head */
> > };
> > 
> > # cat test.c
> > 
> > struct a {
> > 	int __attribute__((aligned(64))) i;
> > 	int j;
> > };
> > struct b {
> > 	int i;
> > 	int __attribute__((aligned(64))) j;
> > };
> > 
> > void main(void)
> > {
> > 	printf("%d,%d\n", sizeof(struct a), sizeof(struct b));
> > }
> > 
> > # ./a.out
> > 64,128
> > 
> 
> Good point! While we are there, I notice that the "qlen" count, kept for
> debugging, is causing false-sharing too. I wonder if we should split
> this counter in two counters: nr_enqueue and nr_dequeue, which would sit
> in two different cache lines ? It's mainly Paul who cares about this
> counter. Thoughts ?

Works for me, as long as nr_enqueue and nr_dequeue are both unsigned long
to avoid issues with overflow.

							Thanx, Paul

> Here is the fix to the problem you noticed above:
> 
> commit b9f893b69fbc31baea418794938f4eb74cc4923a
> Author: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
> Date:   Mon Oct 8 10:44:38 2012 -0400
> 
>     Fix urcu-call-rcu-impl.h: false-sharing
>     
>     > >  struct call_rcu_data {
>     > > -   struct cds_wfq_queue cbs;
>     > > +   /*
>     > > +    * Align the tail on cache line size to eliminate false-sharing
>     > > +    * with head.
>     > > +    */
>     > > +   struct cds_wfcq_tail __attribute__((aligned(CAA_CACHE_LINE_SIZE))) cbs_tail;
>     > > +   /* Alignment on cache line size will add padding here */
>     > > +
>     > > +   struct cds_wfcq_head cbs_head;
>     >
>     >
>     > wrong here. In this code, cbs_tail and cbs_head are in the same cache line.
>     
>     Reported-by: Lai Jiangshan <laijs at cn.fujitsu.com>
>     Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
> 
> Thanks!
> 
> Mathieu
> 
> -- 
> Mathieu Desnoyers
> Operating System Efficiency R&D Consultant
> EfficiOS Inc.
> http://www.efficios.com
> 

[lttng-dev] [URCU PATCH 1/3] wfcqueue: implement concurrency-efficient queue

[Paolo Bonzini]

Il 02/10/2012 16:14, Mathieu Desnoyers ha scritto:
> +/*
> + * Concurrent queue with wait-free enqueue/blocking dequeue.
> + *
> + * Inspired from half-wait-free/half-blocking queue implementation done by
> + * Paul E. McKenney.
> + *
> + * Mutual exclusion of __cds_wfcq_* API
> + *
> + * Unless otherwise stated, the caller must ensure mutual exclusion of
> + * queue update operations "dequeue" and "splice" (for source queue).
> + * Queue read operations "first" and "next" need to be protected against
> + * concurrent "dequeue" and "splice" (for source queue) by the caller.
> + * "enqueue", "splice" (for destination queue), and "empty" are the only
> + * operations that can be used without any mutual exclusion.
> + * Mutual exclusion can be ensured by holding cds_wfcq_dequeue_lock().
> + *
> + * For convenience, cds_wfcq_dequeue_blocking() and
> + * cds_wfcq_splice_blocking() hold the dequeue lock.
> + */

Hi,

can you add a for-like macro for iteration?  Iteration does not require
holding the lock and assumes that you are the sole user of the data
structure, which is useful together with splice.

Paolo

[lttng-dev] [URCU PATCH 3/3] call_rcu: use wfcqueue, eliminate false-sharing

[Mathieu Desnoyers]

* Paul E. McKenney (paulmck at linux.vnet.ibm.com) wrote:
> On Mon, Oct 08, 2012 at 10:49:16AM -0400, Mathieu Desnoyers wrote:
[...]
> > 
> > Good point! While we are there, I notice that the "qlen" count, kept for
> > debugging, is causing false-sharing too. I wonder if we should split
> > this counter in two counters: nr_enqueue and nr_dequeue, which would sit
> > in two different cache lines ? It's mainly Paul who cares about this
> > counter. Thoughts ?
> 
> Works for me, as long as nr_enqueue and nr_dequeue are both unsigned long
> to avoid issues with overflow.
> 

How about the following ? The 0.7% performance increase (which is this
small probably due to call rcu batching) makes me wonder if it's really
worth all the trouble through. If not, then we might want to consider
removing the head alignment altogether. Carefully placing head/tail into
different cache lines is really worth it if we have frequent dequeue,
but in this case we batch through "splice" operations on large queues.
Thoughts ?


Fix wfcqueue: false-sharing of qlen, flags, futex fields

The "qlen" count, kept for debugging, is causing false-sharing.  Split
this counter into two: nr_enqueue and nr_dequeue, which sit in two
different cache lines.

The flags field is read-only by enqueue, and read-mostly by dequeue. We
can eliminate all false-sharing between enqueue and dequeue by having
one copy of the flag for enqueue and one for dequeue. The STOP and
STOPPED flags are only used for dequeue (at "free" time), and the RT
flag is set at init time, and used by both enqueue and dequeue.

The "futex" field is very often read by enqueue, and less often
read and modified by dequeue (only when dequeue has emptied the queue).
Move it to the enqueuer cache-line too.

With these modifications, the test "test_urcu_hash 0 1 10" (one updater
for hash table, which uses call_rcu heavily for reclaim) gains 0.7%. The
small performance improvement in this case can be associated to the fact
that call_rcu dequeuer batches dequeuing of rcu head elements.

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
---
diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
index dca98e4..3f46be6 100644
--- a/urcu-call-rcu-impl.h
+++ b/urcu-call-rcu-impl.h
@@ -49,17 +49,18 @@
 struct call_rcu_data {
 	/*
 	 * Align the tail on cache line size to eliminate false-sharing
-	 * with head. Small note, however: the "qlen" field, kept for
-	 * debugging, will cause false-sharing between enqueue and
-	 * dequeue.
+	 * with head.
 	 */
 	struct cds_wfcq_tail cbs_tail;
+	unsigned long nr_enqueue;	/* maintained for debugging. */
+	unsigned long flags_enqueue;
+	int32_t futex;
+
 	/* Alignment on cache line size will add padding here */
 
 	struct cds_wfcq_head __attribute__((aligned(CAA_CACHE_LINE_SIZE))) cbs_head;
-	unsigned long flags;
-	int32_t futex;
-	unsigned long qlen; /* maintained for debugging. */
+	unsigned long nr_dequeue;	/* maintained for debugging. */
+	unsigned long flags_dequeue;
 	pthread_t tid;
 	int cpu_affinity;
 	struct cds_list_head list;
@@ -231,7 +232,7 @@ static void *call_rcu_thread(void *arg)
 {
 	unsigned long cbcount;
 	struct call_rcu_data *crdp = (struct call_rcu_data *) arg;
-	int rt = !!(uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT);
+	int rt = !!(uatomic_read(&crdp->flags_dequeue) & URCU_CALL_RCU_RT);
 	int ret;
 
 	ret = set_thread_cpu_affinity(crdp);
@@ -269,9 +270,9 @@ static void *call_rcu_thread(void *arg)
 				rhp->func(rhp);
 				cbcount++;
 			}
-			uatomic_sub(&crdp->qlen, cbcount);
+			uatomic_add(&crdp->nr_dequeue, cbcount);
 		}
-		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
+		if (uatomic_read(&crdp->flags_dequeue) & URCU_CALL_RCU_STOP)
 			break;
 		rcu_thread_offline();
 		if (!rt) {
@@ -300,7 +301,7 @@ static void *call_rcu_thread(void *arg)
 		cmm_smp_mb();
 		uatomic_set(&crdp->futex, 0);
 	}
-	uatomic_or(&crdp->flags, URCU_CALL_RCU_STOPPED);
+	uatomic_or(&crdp->flags_dequeue, URCU_CALL_RCU_STOPPED);
 	rcu_unregister_thread();
 	return NULL;
 }
@@ -323,9 +324,12 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
 		urcu_die(errno);
 	memset(crdp, '\0', sizeof(*crdp));
 	cds_wfcq_init(&crdp->cbs_head, &crdp->cbs_tail);
-	crdp->qlen = 0;
+	crdp->nr_enqueue = 0;
+	crdp->flags_enqueue = flags;
 	crdp->futex = 0;
-	crdp->flags = flags;
+
+	crdp->nr_dequeue = 0;
+	crdp->flags_dequeue = flags;
 	cds_list_add(&crdp->list, &call_rcu_data_list);
 	crdp->cpu_affinity = cpu_affinity;
 	cmm_smp_mb();  /* Structure initialized before pointer is planted. */
@@ -571,7 +575,7 @@ int create_all_cpu_call_rcu_data(unsigned long flags)
  */
 static void wake_call_rcu_thread(struct call_rcu_data *crdp)
 {
-	if (!(_CMM_LOAD_SHARED(crdp->flags) & URCU_CALL_RCU_RT))
+	if (!(_CMM_LOAD_SHARED(crdp->flags_enqueue) & URCU_CALL_RCU_RT))
 		call_rcu_wake_up(crdp);
 }
 
@@ -601,7 +605,7 @@ void call_rcu(struct rcu_head *head,
 	rcu_read_lock();
 	crdp = get_call_rcu_data();
 	cds_wfcq_enqueue(&crdp->cbs_head, &crdp->cbs_tail, &head->next);
-	uatomic_inc(&crdp->qlen);
+	uatomic_inc(&crdp->nr_enqueue);
 	wake_call_rcu_thread(crdp);
 	rcu_read_unlock();
 }
@@ -633,10 +637,10 @@ void call_rcu_data_free(struct call_rcu_data *crdp)
 	if (crdp == NULL || crdp == default_call_rcu_data) {
 		return;
 	}
-	if ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0) {
-		uatomic_or(&crdp->flags, URCU_CALL_RCU_STOP);
+	if ((uatomic_read(&crdp->flags_dequeue) & URCU_CALL_RCU_STOPPED) == 0) {
+		uatomic_or(&crdp->flags_dequeue, URCU_CALL_RCU_STOP);
 		wake_call_rcu_thread(crdp);
-		while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0)
+		while ((uatomic_read(&crdp->flags_dequeue) & URCU_CALL_RCU_STOPPED) == 0)
 			poll(NULL, 0, 1);
 	}
 	if (!cds_wfcq_empty(&crdp->cbs_head, &crdp->cbs_tail)) {
@@ -645,8 +649,10 @@ void call_rcu_data_free(struct call_rcu_data *crdp)
 		__cds_wfcq_splice_blocking(&default_call_rcu_data->cbs_head,
 			&default_call_rcu_data->cbs_tail,
 			&crdp->cbs_head, &crdp->cbs_tail);
-		uatomic_add(&default_call_rcu_data->qlen,
-			    uatomic_read(&crdp->qlen));
+		uatomic_add(&default_call_rcu_data->nr_enqueue,
+			    uatomic_read(&crdp->nr_enqueue));
+		uatomic_add(&default_call_rcu_data->nr_dequeue,
+			    uatomic_read(&crdp->nr_dequeue));
 		wake_call_rcu_thread(default_call_rcu_data);
 	}
 
@@ -750,7 +756,7 @@ void call_rcu_after_fork_child(void)
 	cds_list_for_each_entry_safe(crdp, next, &call_rcu_data_list, list) {
 		if (crdp == default_call_rcu_data)
 			continue;
-		uatomic_set(&crdp->flags, URCU_CALL_RCU_STOPPED);
+		uatomic_set(&crdp->flags_dequeue, URCU_CALL_RCU_STOPPED);
 		call_rcu_data_free(crdp);
 	}
 }

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[lttng-dev] [URCU PATCH 1/3] wfcqueue: implement concurrency-efficient queue

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> Il 02/10/2012 16:14, Mathieu Desnoyers ha scritto:
> > +/*
> > + * Concurrent queue with wait-free enqueue/blocking dequeue.
> > + *
> > + * Inspired from half-wait-free/half-blocking queue implementation done by
> > + * Paul E. McKenney.
> > + *
> > + * Mutual exclusion of __cds_wfcq_* API
> > + *
> > + * Unless otherwise stated, the caller must ensure mutual exclusion of
> > + * queue update operations "dequeue" and "splice" (for source queue).
> > + * Queue read operations "first" and "next" need to be protected against
> > + * concurrent "dequeue" and "splice" (for source queue) by the caller.
> > + * "enqueue", "splice" (for destination queue), and "empty" are the only
> > + * operations that can be used without any mutual exclusion.
> > + * Mutual exclusion can be ensured by holding cds_wfcq_dequeue_lock().
> > + *
> > + * For convenience, cds_wfcq_dequeue_blocking() and
> > + * cds_wfcq_splice_blocking() hold the dequeue lock.
> > + */
> 
> Hi,
> 
> can you add a for-like macro for iteration?  Iteration does not require
> holding the lock and assumes that you are the sole user of the data
> structure, which is useful together with splice.
> 
> Paolo

Hi Paolo,

We actually already have those, they are just not described in this
comment. I will fix this right away. By the way, you will notice the
wording:

+ * Queue read operations "first" and "next", which are used by
+ * "for_each" iterations, need to be protected against concurrent
+ * "dequeue" and "splice" (for source queue) by the caller.

Being the only one iterating on a queue with local head/tail after a
splice operation is one way to provide mutual exclusion. Holding a lock
is not the only way to achieve mutual exclusion.


commit f94061a3df4c9eab9ac869a19e4228de54771fcb
Author: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
Date:   Mon Oct 8 12:11:30 2012 -0400

    wfcqueue documentation: hint at for_each iterators
    
    Reported-by: Paolo Bonzini <pbonzini at redhat.com>
    Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>

diff --git a/urcu/static/wfcqueue.h b/urcu/static/wfcqueue.h
index a989984..944ee88 100644
--- a/urcu/static/wfcqueue.h
+++ b/urcu/static/wfcqueue.h
@@ -48,8 +48,9 @@ extern "C" {
  *
  * Unless otherwise stated, the caller must ensure mutual exclusion of
  * queue update operations "dequeue" and "splice" (for source queue).
- * Queue read operations "first" and "next" need to be protected against
- * concurrent "dequeue" and "splice" (for source queue) by the caller.
+ * Queue read operations "first" and "next", which are used by
+ * "for_each" iterations, need to be protected against concurrent
+ * "dequeue" and "splice" (for source queue) by the caller.
  * "enqueue", "splice" (for destination queue), and "empty" are the only
  * operations that can be used without any mutual exclusion.
  * Mutual exclusion can be ensured by holding cds_wfcq_dequeue_lock().
@@ -190,6 +191,10 @@ ___cds_wfcq_node_sync_next(struct cds_wfcq_node *node)
  * Content written into the node before enqueue is guaranteed to be
  * consistent, but no other memory ordering is ensured.
  * Should be called with cds_wfcq_dequeue_lock() held.
+ *
+ * Used by for-like iteration macros in urcu/wfqueue.h:
+ * __cds_wfcq_for_each_blocking()
+ * __cds_wfcq_for_each_blocking_safe()
  */
 static inline struct cds_wfcq_node *
 ___cds_wfcq_first_blocking(struct cds_wfcq_head *head,
@@ -211,6 +216,10 @@ ___cds_wfcq_first_blocking(struct cds_wfcq_head *head,
  * Content written into the node before enqueue is guaranteed to be
  * consistent, but no other memory ordering is ensured.
  * Should be called with cds_wfcq_dequeue_lock() held.
+ *
+ * Used by for-like iteration macros in urcu/wfqueue.h:
+ * __cds_wfcq_for_each_blocking()
+ * __cds_wfcq_for_each_blocking_safe()
  */
 static inline struct cds_wfcq_node *
 ___cds_wfcq_next_blocking(struct cds_wfcq_head *head,
diff --git a/urcu/wfcqueue.h b/urcu/wfcqueue.h
index 5576cbf..501120b 100644
--- a/urcu/wfcqueue.h
+++ b/urcu/wfcqueue.h
@@ -91,8 +91,9 @@ struct cds_wfcq_tail {
  *
  * Unless otherwise stated, the caller must ensure mutual exclusion of
  * queue update operations "dequeue" and "splice" (for source queue).
- * Queue read operations "first" and "next" need to be protected against
- * concurrent "dequeue" and "splice" (for source queue) by the caller.
+ * Queue read operations "first" and "next", which are used by
+ * "for_each" iterations, need to be protected against concurrent
+ * "dequeue" and "splice" (for source queue) by the caller.
  * "enqueue", "splice" (for destination queue), and "empty" are the only
  * operations that can be used without any mutual exclusion.
  * Mutual exclusion can be ensured by holding cds_wfcq_dequeue_lock().
@@ -202,6 +203,10 @@ extern void __cds_wfcq_splice_blocking(
  * Content written into the node before enqueue is guaranteed to be
  * consistent, but no other memory ordering is ensured.
  * Should be called with cds_wfcq_dequeue_lock() held.
+ *
+ * Used by for-like iteration macros:
+ * __cds_wfcq_for_each_blocking()
+ * __cds_wfcq_for_each_blocking_safe()
  */
 extern struct cds_wfcq_node *__cds_wfcq_first_blocking(
 		struct cds_wfcq_head *head,
@@ -213,6 +218,10 @@ extern struct cds_wfcq_node *__cds_wfcq_first_blocking(
  * Content written into the node before enqueue is guaranteed to be
  * consistent, but no other memory ordering is ensured.
  * Should be called with cds_wfcq_dequeue_lock() held.
+ *
+ * Used by for-like iteration macros:
+ * __cds_wfcq_for_each_blocking()
+ * __cds_wfcq_for_each_blocking_safe()
  */
 extern struct cds_wfcq_node *__cds_wfcq_next_blocking(
 		struct cds_wfcq_head *head,


-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[lttng-dev] [URCU PATCH 1/3] wfcqueue: implement concurrency-efficient queue

[Paolo Bonzini]

Il 08/10/2012 18:15, Mathieu Desnoyers ha scritto:
> Hi Paolo,
> 
> We actually already have those, they are just not described in this
> comment. I will fix this right away. By the way, you will notice the
> wording:
> 
> + * Queue read operations "first" and "next", which are used by
> + * "for_each" iterations, need to be protected against concurrent
> + * "dequeue" and "splice" (for source queue) by the caller.
> 
> Being the only one iterating on a queue with local head/tail after a
> splice operation is one way to provide mutual exclusion. Holding a lock
> is not the only way to achieve mutual exclusion.

Uh, I was confused by the _blocking suffix.  But when used together with
splice you know it is not blocking---only the splice will block.

Paolo

[lttng-dev] [URCU PATCH 1/3] wfcqueue: implement concurrency-efficient queue

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> Il 08/10/2012 18:15, Mathieu Desnoyers ha scritto:
> > Hi Paolo,
> > 
> > We actually already have those, they are just not described in this
> > comment. I will fix this right away. By the way, you will notice the
> > wording:
> > 
> > + * Queue read operations "first" and "next", which are used by
> > + * "for_each" iterations, need to be protected against concurrent
> > + * "dequeue" and "splice" (for source queue) by the caller.
> > 
> > Being the only one iterating on a queue with local head/tail after a
> > splice operation is one way to provide mutual exclusion. Holding a lock
> > is not the only way to achieve mutual exclusion.
> 
> Uh, I was confused by the _blocking suffix.  But when used together with
> splice you know it is not blocking---only the splice will block.

Well, in this case, the "blocking" can be understood as busy-waiting
(and actual blocking that invokes the OS happens if busy-waiting for long
periods only).

for_each iteration can indeed busy-wait: ___cds_wfcq_first_blocking()
and ___cds_wfcq_next_blocking() can both call
___cds_wfcq_node_sync_next(), which busy-waits if it encounters a NULL
next pointer that is not located in the tail node.

Seen through the use-case of splice to local queue + for_each, here is
what happens:

- splice moves the content of the queue into a "local" queue. However,
  it does _not_ issue a ___cds_wfcq_node_sync_next() on each node: no
  traversal is performed.

- then, within the for_each iteration, we perform the
  ___cds_wfcq_node_sync_next() synchronization as we iterate on the
  local queue.

This ensures that the synchronization performed for each node is only
performed lazily, only when those nodes are actually traversed. This
approach has the advantage of increasing the locality of reference, and
also to diminish the odds that this synchronization actually needs to be
executed uselessly by postponing it to the point where it is actually
needed.

Does it make more sense ?

Thanks,

Mathieu

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[lttng-dev] [rp] [URCU PATCH 0/3] wait-free concurrent queues (wfcqueue)

[Lai Jiangshan]

On 10/08/2012 11:07 PM, Mathieu Desnoyers wrote:
> * Lai Jiangshan (laijs at cn.fujitsu.com) wrote:
>> On 10/04/2012 05:04 AM, Mathieu Desnoyers wrote:
>>> * Paul E. McKenney (paulmck at linux.vnet.ibm.com) wrote:
>>>> On Tue, Oct 02, 2012 at 10:13:07AM -0400, Mathieu Desnoyers wrote:
>>>>> Implement wait-free concurrent queues, with a new API different from
>>>>> wfqueue.h, which is already provided by Userspace RCU. The advantage of
>>>>> splitting the head and tail objects of the queue into different
>>>>> arguments is to allow these to sit on different cache-lines, thus
>>>>> eliminating false-sharing, leading to a 2.3x speed increase.
>>>>>
>>>>> This API also introduces a "splice" operation, which moves all nodes
>>>>> from one queue into another, and postpones the synchronization to either
>>>>> dequeue or iteration on the list. The splice operation does not need to
>>>>> touch every single node of the queue it moves them from. Moreover, the
>>>>> splice operation only needs to ensure mutual exclusion with other
>>>>> dequeuers, iterations, and splice operations from the list it splices
>>>>> from, but acts as a simple enqueuer on the list it splices into (no
>>>>> mutual exclusion needed for that list).
>>>>>
>>>>> Feedback is welcome,
>>>>
>>>> These look sane to me, though I must confess that the tail pointer
>>>> referencing the node rather than the node's next pointer did throw
>>>> me for a bit.  ;-)
>>>
>>> Yes, this was originally introduced with Lai's original patch to
>>> wfqueue, which I think is a nice simplification: it's pretty much the
>>> same thing to use the last node address as tail rather than the address
>>> of its first member (its next pointer address (_not_ value)). It ends up
>>> being the same address in this case, but more interestingly, we don't
>>> have to use a struct cds_wfcq_node ** type: a simple struct
>>> cds_wfcq_node *  suffice.
>>>
>>> Thanks Paul, I will therefore merge these 3 patches with your Acked-by.
>>>
>>> Lai, you are welcome to provide improvements to this code against the
>>> master branch. I will gladly consider any change you propose.
>>>
>>
>> I did not remember that there is any improvement idea not included.
>> The patchset is OK for me.
> 
> Great! Would you be OK if I commit the following patch ? Let me know if
> you want me to put your signed-off-by on this (I can even put your email
> as From if you like):
> 
> 
> wfcqueue: update credits in patch documentation
> 
> Give credits to those responsible for the design and implementation of
> commit 8ad4ce587f001ae026d5560ac509c2e48986130b, "wfcqueue: implement
> concurrency-efficient queue", which happened through rounds of email and
> patch exchanges.
> 
> Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
> ---
> diff --git a/urcu/static/wfcqueue.h b/urcu/static/wfcqueue.h
> index a989984..153143d 100644
> --- a/urcu/static/wfcqueue.h
> +++ b/urcu/static/wfcqueue.h
> @@ -41,8 +41,10 @@ extern "C" {
>  /*
>   * Concurrent queue with wait-free enqueue/blocking dequeue.
>   *
> - * Inspired from half-wait-free/half-blocking queue implementation done by
> - * Paul E. McKenney.
> + * This queue has been designed and implemented collaboratively by
> + * Mathieu Desnoyers and Lai Jiangshan. Inspired from
> + * half-wait-free/half-blocking queue implementation done by Paul E.
> + * McKenney.
>   *
>   * Mutual exclusion of __cds_wfcq_* API
>   *
> diff --git a/urcu/wfcqueue.h b/urcu/wfcqueue.h
> index 5576cbf..940dc7d 100644
> --- a/urcu/wfcqueue.h
> +++ b/urcu/wfcqueue.h
> @@ -37,8 +37,10 @@ extern "C" {
>  /*
>   * Concurrent queue with wait-free enqueue/blocking dequeue.
>   *
> - * Inspired from half-wait-free/half-blocking queue implementation done by
> - * Paul E. McKenney.
> + * This queue has been designed and implemented collaboratively by
> + * Mathieu Desnoyers and Lai Jiangshan. Inspired from
> + * half-wait-free/half-blocking queue implementation done by Paul E.
> + * McKenney.
>   */
>  
>  struct cds_wfcq_node {
> 
> 
>> I think you can reimplement wfqueue via wfcqueue without cacheline opt.
> 
> Hrm, semantically this can indeed be done, but I fear that we might not
> be strictly ABI-compatible with the old wfqueue. So I would be tempted
> to leave the old wfqueue implementation as-is, and maybe deprecate it at
> some point. Thoughts ?
> 

All APIs is not changed, they are forwarded to wfcqueue APIs in implementation,
so ABI of APIs is compatible. The only thing is struct cds_wfq_queue:

struct cds_wfq_queue {
	struct cds_wfq_node *head, **tail;
	struct cds_wfq_node dummy;	/* Dummy node */
	pthread_mutex_t lock;
};


We can redefine it as:

#define cds_wfq_node cds_wfcq_node

struct cds_wfq_queue {
	union {
		struct cds_wfq_node *head; /* make bug-user who wrongly directly access to ->head happy */
		struct cds_wfcq_node *__pad; /* not used in new implement */
	};
	union {
		struct cds_wfq_node **tail;  /* make bug-user who wrongly directly access to ->tail happy */
		struct cds_wfcq_tail real_tail;
	};
	union {
		struct {
			struct cds_wfq_node dummy;	/* Dummy node */
			pthread_mutex_t lock;
		}
		struct cds_wfcq_head real_head;
	};
}

static inline void _cds_wfq_init(struct cds_wfq_queue *q)
{
	q->head = &q->dummy; /* make bug-user who wrongly directly access to ->head happy */
	_cds_wfcq_init(&q->real_head, &q->real_tail);
}

after this change, struct cds_wfq_queue is not changed.
Even bug-user wrongly directly access to struct cds_wfq_queue by old-view,
the queue is compatible:
	head->dummy node->real node->real node....
	tail->real tail node(or dummy node)

the only different is that: dummy node is always the first node by old-view.


And if we deprecate struct cds_wfq_queue, I think we should provide a new default wfqueue to users.

Thanks,
Lai

[lttng-dev] [URCU PATCH 1/3] wfcqueue: implement concurrency-efficient queue

[Lai Jiangshan]

On 10/02/2012 10:14 PM, Mathieu Desnoyers wrote:
> This new API simplify the wfqueue implementation, and brings a 2.3x to
> 2.6x performance boost due to the ability to eliminate false-sharing
> between enqueue and dequeue.
> 
> This work is derived from the patch from Lai Jiangshan submitted as
> "urcu: new wfqueue implementation"
> (http://lists.lttng.org/pipermail/lttng-dev/2012-August/018379.html)
> 
> Its changelog:
> 
>> Some guys would be surprised by this fact:
>> There are already TWO implementations of wfqueue in urcu.
>>
>> The first one is in urcu/static/wfqueue.h:
>> 1) enqueue: exchange the tail and then update previous->next
>> 2) dequeue: wait for first node's next pointer and them shift, a dummy node
>> 	is introduced to avoid the queue->tail become NULL when shift.
>>
>> The second one shares some code with the first one, and the left code
>> are spreading in urcu-call-rcu-impl.h:
>> 1) enqueue: share with the first one
>> 2) no dequeue operation: and no shift, so it don't need dummy node,
>> 	Although the dummy node is queued when initialization, but it is removed
>> 	after the first dequeue_all operation in call_rcu_thread().
>> 	call_rcu_data_free() forgets to handle the dummy node if it is not removed.
>> 3)dequeue_all: record the old head and tail, and queue->head become the special
>> 	tail node.(atomic record the tail and change the tail).
>>
>> The second implementation's code are spreading, bad for review, and it is not
>> tested by tests/test_urcu_wfq.
>>
>> So we need a better implementation avoid the dummy node dancing and can service
>> both generic wfqueue APIs and dequeue_all API for call rcu.
>>
>> The new implementation:
>> 1) enqueue: share with the first one/original implementation.
>> 2) dequeue: shift when node count >= 2, cmpxchg when node count = 1.
>> 	no dummy node, save memory.
>> 3) dequeue_all: simply set queue->head.next to NULL, xchg the tail
>> 	and return the old head.next.
>>
>> More implementation details are in the code.
>> tests/test_urcu_wfq will be update in future for testing new APIs.
> 
> The patch proposed by Lai brings a very interesting simplification to
> the single-node handling (which is kept here), and moves all queue
> handling code away from call_rcu implementation, back into the wfqueue
> code. This has the benefit to allow testing enhancements.
> 
> I modified it so the API does not expose implementation details to the
> user (e.g. ___cds_wfq_node_sync_next). I added a "splice" operation and
> a for loop iterator which should allow wfqueue users to use the list
> very efficiently both from LGPL/GPL code and from non-LGPL-compatible
> code.
> 
> I also changed the API so the queue head and tail are now two separate
> structures: it allows the queue user to place these as they like, either
> on different cache lines (to eliminate false-sharing), or close one to
> another (on same cache-line) in case a queue is spliced onto the stack
> and not concurrently accessed.
> 
> Benchmarks performed on Intel(R) Core(TM) i7-3520M CPU @ 2.90GHz
> (dual-core, with hyperthreading)
> 
> Benchmark invoked:
> for a in $(seq 1 10); do ./test_urcu_wfq 1 1 10 -a 0 -a 2; done
> 
> (using cpu number 0 and 2, which should correspond to two cores of my
> Intel 2-core/hyperthread processor)
> 
> Before patch:
> 
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     97274297 nr_dequeues     80745742 successful enqueues     97274297 successful dequeues     80745321 end_dequeues 16528976 nr_ops    178020039
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     92300568 nr_dequeues     75019529 successful enqueues     92300568 successful dequeues     74973237 end_dequeues 17327331 nr_ops    167320097
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     93516443 nr_dequeues     75846726 successful enqueues     93516443 successful dequeues     75826578 end_dequeues 17689865 nr_ops    169363169
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     94160362 nr_dequeues     77967638 successful enqueues     94160362 successful dequeues     77967638 end_dequeues 16192724 nr_ops    172128000
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     97491956 nr_dequeues     81001191 successful enqueues     97491956 successful dequeues     81000247 end_dequeues 16491709 nr_ops    178493147
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     94101298 nr_dequeues     75650510 successful enqueues     94101298 successful dequeues     75649318 end_dequeues 18451980 nr_ops    169751808
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     94742803 nr_dequeues     75402105 successful enqueues     94742803 successful dequeues     75341859 end_dequeues 19400944 nr_ops    170144908
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     92198835 nr_dequeues     75037877 successful enqueues     92198835 successful dequeues     75027605 end_dequeues 17171230 nr_ops    167236712
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     94159560 nr_dequeues     77895972 successful enqueues     94159560 successful dequeues     77858442 end_dequeues 16301118 nr_ops    172055532
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues     96059399 nr_dequeues     80115442 successful enqueues     96059399 successful dequeues     80066843 end_dequeues 15992556 nr_ops    176174841
> 
> After patch:
> 
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    221229322 nr_dequeues    210645491 successful enqueues    221229322 successful dequeues    210645088 end_dequeues 10584234 nr_ops    431874813
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    219803943 nr_dequeues    210377337 successful enqueues    219803943 successful dequeues    210368680 end_dequeues 9435263 nr_ops    430181280
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    237006358 nr_dequeues    237035340 successful enqueues    237006358 successful dequeues    236997050 end_dequeues 9308 nr_ops    474041698
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    235822443 nr_dequeues    235815942 successful enqueues    235822443 successful dequeues    235814020 end_dequeues 8423 nr_ops    471638385
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    235825567 nr_dequeues    235811803 successful enqueues    235825567 successful dequeues    235810526 end_dequeues 15041 nr_ops    471637370
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    221974953 nr_dequeues    210938190 successful enqueues    221974953 successful dequeues    210938190 end_dequeues 11036763 nr_ops    432913143
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    237994492 nr_dequeues    237938119 successful enqueues    237994492 successful dequeues    237930648 end_dequeues 63844 nr_ops    475932611
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    220634365 nr_dequeues    210491382 successful enqueues    220634365 successful dequeues    210490995 end_dequeues 10143370 nr_ops    431125747
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    237388065 nr_dequeues    237401251 successful enqueues    237388065 successful dequeues    237380295 end_dequeues 7770 nr_ops    474789316
> testdur   10 nr_enqueuers   1 wdelay      0 nr_dequeuers   1 rdur      0 nr_enqueues    221201436 nr_dequeues    210831162 successful enqueues    221201436 successful dequeues    210831162 end_dequeues 10370274 nr_ops    432032598
> 
> Summary: Both enqueue and dequeue speed increase: around 2.3x speedup
> for enqueue, and around 2.6x for dequeue.
> 
> We can verify that:
>    successful enqueues - successful dequeues = end_dequeues
> 
> For all runs (ensures correctness: no lost node).
> 
> CC: Lai Jiangshan <laijs at cn.fujitsu.com>
> CC: Paul McKenney <paulmck at linux.vnet.ibm.com>
> Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
> ---
> diff --git a/Makefile.am b/Makefile.am
> index 2396fcf..ffdca9a 100644
> --- a/Makefile.am
> +++ b/Makefile.am
> @@ -16,7 +16,7 @@ nobase_dist_include_HEADERS = urcu/compiler.h urcu/hlist.h urcu/list.h \
>  		urcu/uatomic/generic.h urcu/arch/generic.h urcu/wfstack.h \
>  		urcu/wfqueue.h urcu/rculfstack.h urcu/rculfqueue.h \
>  		urcu/ref.h urcu/cds.h urcu/urcu_ref.h urcu/urcu-futex.h \
> -		urcu/uatomic_arch.h urcu/rculfhash.h \
> +		urcu/uatomic_arch.h urcu/rculfhash.h urcu/wfcqueue.h \
>  		$(top_srcdir)/urcu/map/*.h \
>  		$(top_srcdir)/urcu/static/*.h \
>  		urcu/tls-compat.h
> @@ -53,7 +53,7 @@ lib_LTLIBRARIES = liburcu-common.la \
>  # liburcu-common contains wait-free queues (needed by call_rcu) as well
>  # as futex fallbacks.
>  #
> -liburcu_common_la_SOURCES = wfqueue.c wfstack.c $(COMPAT)
> +liburcu_common_la_SOURCES = wfqueue.c wfcqueue.c wfstack.c $(COMPAT)
>  
>  liburcu_la_SOURCES = urcu.c urcu-pointer.c $(COMPAT)
>  liburcu_la_LIBADD = liburcu-common.la
> diff --git a/urcu/static/wfcqueue.h b/urcu/static/wfcqueue.h
> new file mode 100644
> index 0000000..a989984
> --- /dev/null
> +++ b/urcu/static/wfcqueue.h
> @@ -0,0 +1,380 @@
> +#ifndef _URCU_WFCQUEUE_STATIC_H
> +#define _URCU_WFCQUEUE_STATIC_H
> +
> +/*
> + * wfcqueue-static.h
> + *
> + * Userspace RCU library - Concurrent Queue with Wait-Free Enqueue/Blocking Dequeue
> + *
> + * TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See wfcqueue.h for linking
> + * dynamically with the userspace rcu library.
> + *
> + * Copyright 2010-2012 - Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
> + * Copyright 2011-2012 - Lai Jiangshan <laijs at cn.fujitsu.com>
> + *
> + * This library is free software; you can redistribute it and/or
> + * modify it under the terms of the GNU Lesser General Public
> + * License as published by the Free Software Foundation; either
> + * version 2.1 of the License, or (at your option) any later version.
> + *
> + * This library is distributed in the hope that it will be useful,
> + * but WITHOUT ANY WARRANTY; without even the implied warranty of
> + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
> + * Lesser General Public License for more details.
> + *
> + * You should have received a copy of the GNU Lesser General Public
> + * License along with this library; if not, write to the Free Software
> + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
> + */
> +
> +#include <pthread.h>
> +#include <assert.h>
> +#include <poll.h>
> +#include <stdbool.h>
> +#include <urcu/compiler.h>
> +#include <urcu/uatomic.h>
> +
> +#ifdef __cplusplus
> +extern "C" {
> +#endif
> +
> +/*
> + * Concurrent queue with wait-free enqueue/blocking dequeue.
> + *
> + * Inspired from half-wait-free/half-blocking queue implementation done by
> + * Paul E. McKenney.
> + *
> + * Mutual exclusion of __cds_wfcq_* API
> + *
> + * Unless otherwise stated, the caller must ensure mutual exclusion of
> + * queue update operations "dequeue" and "splice" (for source queue).
> + * Queue read operations "first" and "next" need to be protected against
> + * concurrent "dequeue" and "splice" (for source queue) by the caller.
> + * "enqueue", "splice" (for destination queue), and "empty" are the only
> + * operations that can be used without any mutual exclusion.
> + * Mutual exclusion can be ensured by holding cds_wfcq_dequeue_lock().
> + *
> + * For convenience, cds_wfcq_dequeue_blocking() and
> + * cds_wfcq_splice_blocking() hold the dequeue lock.
> + */
> +
> +#define WFCQ_ADAPT_ATTEMPTS		10	/* Retry if being set */
> +#define WFCQ_WAIT			10	/* Wait 10 ms if being set */
> +
> +/*
> + * cds_wfcq_node_init: initialize wait-free queue node.
> + */
> +static inline void _cds_wfcq_node_init(struct cds_wfcq_node *node)
> +{
> +	node->next = NULL;
> +}
> +
> +/*
> + * cds_wfcq_init: initialize wait-free queue.
> + */
> +static inline void _cds_wfcq_init(struct cds_wfcq_head *head,
> +		struct cds_wfcq_tail *tail)
> +{
> +	int ret;
> +
> +	/* Set queue head and tail */
> +	_cds_wfcq_node_init(&head->node);
> +	tail->p = &head->node;
> +	ret = pthread_mutex_init(&head->lock, NULL);
> +	assert(!ret);
> +}
> +
> +/*
> + * cds_wfcq_empty: return whether wait-free queue is empty.
> + *
> + * No memory barrier is issued. No mutual exclusion is required.
> + */
> +static inline bool _cds_wfcq_empty(struct cds_wfcq_head *head,
> +		struct cds_wfcq_tail *tail)
> +{
> +	/*
> +	 * Queue is empty if no node is pointed by head->node.next nor
> +	 * tail->p. Even though the tail->p check is sufficient to find
> +	 * out of the queue is empty, we first check head->node.next as a
> +	 * common case to ensure that dequeuers do not frequently access
> +	 * enqueuer's tail->p cache line.
> +	 */
> +	return CMM_LOAD_SHARED(head->node.next) == NULL
> +		&& CMM_LOAD_SHARED(tail->p) == &head->node;
> +}
> +
> +static inline void _cds_wfcq_dequeue_lock(struct cds_wfcq_head *head,
> +		struct cds_wfcq_tail *tail)
> +{
> +	int ret;
> +
> +	ret = pthread_mutex_lock(&head->lock);
> +	assert(!ret);
> +}
> +
> +static inline void _cds_wfcq_dequeue_unlock(struct cds_wfcq_head *head,
> +		struct cds_wfcq_tail *tail)
> +{
> +	int ret;
> +
> +	ret = pthread_mutex_unlock(&head->lock);
> +	assert(!ret);
> +}
> +
> +static inline void ___cds_wfcq_append(struct cds_wfcq_head *head,
> +		struct cds_wfcq_tail *tail,
> +		struct cds_wfcq_node *new_head,
> +		struct cds_wfcq_node *new_tail)
> +{
> +	struct cds_wfcq_node *old_tail;
> +
> +	/*
> +	 * Implicit memory barrier before uatomic_xchg() orders earlier
> +	 * stores to data structure containing node and setting
> +	 * node->next to NULL before publication.
> +	 */
> +	old_tail = uatomic_xchg(&tail->p, new_tail);
> +
> +	/*
> +	 * Implicit memory barrier after uatomic_xchg() orders store to
> +	 * q->tail before store to old_tail->next.
> +	 *
> +	 * At this point, dequeuers see a NULL tail->p->next, which
> +	 * indicates that the queue is being appended to. The following
> +	 * store will append "node" to the queue from a dequeuer
> +	 * perspective.
> +	 */
> +	CMM_STORE_SHARED(old_tail->next, new_head);
> +}
> +
> +/*
> + * cds_wfcq_enqueue: enqueue a node into a wait-free queue.
> + *
> + * Issues a full memory barrier before enqueue. No mutual exclusion is
> + * required.
> + */
> +static inline void _cds_wfcq_enqueue(struct cds_wfcq_head *head,
> +		struct cds_wfcq_tail *tail,
> +		struct cds_wfcq_node *new_tail)
> +{
> +	___cds_wfcq_append(head, tail, new_tail, new_tail);
> +}
> +
> +/*
> + * Waiting for enqueuer to complete enqueue and return the next node.
> + */
> +static inline struct cds_wfcq_node *
> +___cds_wfcq_node_sync_next(struct cds_wfcq_node *node)
> +{
> +	struct cds_wfcq_node *next;
> +	int attempt = 0;
> +
> +	/*
> +	 * Adaptative busy-looping waiting for enqueuer to complete enqueue.
> +	 */
> +	while ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
> +		if (++attempt >= WFCQ_ADAPT_ATTEMPTS) {
> +			poll(NULL, 0, WFCQ_WAIT);	/* Wait for 10ms */
> +			attempt = 0;
> +		} else {
> +			caa_cpu_relax();
> +		}
> +	}
> +
> +	return next;
> +}
> +
> +/*
> + * __cds_wfcq_first_blocking: get first node of a queue, without dequeuing.
> + *
> + * Content written into the node before enqueue is guaranteed to be
> + * consistent, but no other memory ordering is ensured.
> + * Should be called with cds_wfcq_dequeue_lock() held.
> + */
> +static inline struct cds_wfcq_node *
> +___cds_wfcq_first_blocking(struct cds_wfcq_head *head,
> +		struct cds_wfcq_tail *tail)
> +{
> +	struct cds_wfcq_node *node;
> +
> +	if (_cds_wfcq_empty(head, tail))
> +		return NULL;
> +	node = ___cds_wfcq_node_sync_next(&head->node);
> +	/* Load head->node.next before loading node's content */
> +	cmm_smp_read_barrier_depends();
> +	return node;
> +}
> +
> +/*
> + * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
> + *
> + * Content written into the node before enqueue is guaranteed to be
> + * consistent, but no other memory ordering is ensured.
> + * Should be called with cds_wfcq_dequeue_lock() held.
> + */
> +static inline struct cds_wfcq_node *
> +___cds_wfcq_next_blocking(struct cds_wfcq_head *head,
> +		struct cds_wfcq_tail *tail,
> +		struct cds_wfcq_node *node)
> +{
> +	struct cds_wfcq_node *next;
> +
> +	/*
> +	 * Even though the following tail->p check is sufficient to find
> +	 * out if we reached the end of the queue, we first check
> +	 * node->next as a common case to ensure that iteration on nodes
> +	 * do not frequently access enqueuer's tail->p cache line.
> +	 */
> +	if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
> +		/* Load node->next before tail->p */
> +		cmm_smp_rmb();
> +		if (CMM_LOAD_SHARED(tail->p) == node)
> +			return NULL;
> +		next = ___cds_wfcq_node_sync_next(node);
> +	}
> +	/* Load node->next before loading next's content */
> +	cmm_smp_read_barrier_depends();
> +	return next;
> +}
> +
> +/*
> + * __cds_wfcq_dequeue_blocking: dequeue a node from the queue.
> + *
> + * No need to go on a waitqueue here, as there is no possible state in which the
> + * list could cause dequeue to busy-loop needlessly while waiting for another
> + * thread to be scheduled. The queue appears empty until tail->next is set by
> + * enqueue.
> + *
> + * Content written into the node before enqueue is guaranteed to be
> + * consistent, but no other memory ordering is ensured.
> + * It is valid to reuse and free a dequeued node immediately.
> + * Should be called with cds_wfcq_dequeue_lock() held.
> + */
> +static inline struct cds_wfcq_node *
> +___cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
> +		struct cds_wfcq_tail *tail)
> +{
> +	struct cds_wfcq_node *node, *next;
> +
> +	if (_cds_wfcq_empty(head, tail))
> +		return NULL;
> +
> +	node = ___cds_wfcq_node_sync_next(&head->node);
> +
> +	if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
> +		/*
> +		 * @node is probably the only node in the queue.
> +		 * Try to move the tail to &q->head.
> +		 * q->head.next is set to NULL here, and stays
> +		 * NULL if the cmpxchg succeeds. Should the
> +		 * cmpxchg fail due to a concurrent enqueue, the
> +		 * q->head.next will be set to the next node.
> +		 * The implicit memory barrier before
> +		 * uatomic_cmpxchg() orders load node->next
> +		 * before loading q->tail.
> +		 * The implicit memory barrier before uatomic_cmpxchg
> +		 * orders load q->head.next before loading node's
> +		 * content.
> +		 */
> +		_cds_wfcq_node_init(&head->node);
> +		if (uatomic_cmpxchg(&tail->p, node, &head->node) == node)
> +			return node;
> +		next = ___cds_wfcq_node_sync_next(node);
> +	}
> +
> +	/*
> +	 * Move queue head forward.
> +	 */
> +	head->node.next = next;
> +
> +	/* Load q->head.next before loading node's content */
> +	cmm_smp_read_barrier_depends();
> +	return node;
> +}
> +
> +/*
> + * __cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
> + *
> + * Dequeue all nodes from src_q.
> + * dest_q must be already initialized.
> + * Should be called with cds_wfcq_dequeue_lock() held on src_q.
> + */
> +static inline void
> +___cds_wfcq_splice_blocking(
> +		struct cds_wfcq_head *dest_q_head,
> +		struct cds_wfcq_tail *dest_q_tail,
> +		struct cds_wfcq_head *src_q_head,
> +		struct cds_wfcq_tail *src_q_tail)
> +{
> +	struct cds_wfcq_node *head, *tail;
> +
> +	if (_cds_wfcq_empty(src_q_head, src_q_tail))
> +		return;
> +
> +	head = ___cds_wfcq_node_sync_next(&src_q_head->node);
> +	_cds_wfcq_node_init(&src_q_head->node);
> +
> +	/*
> +	 * Memory barrier implied before uatomic_xchg() orders store to
> +	 * src_q->head before store to src_q->tail. This is required by
> +	 * concurrent enqueue on src_q, which exchanges the tail before
> +	 * updating the previous tail's next pointer.
> +	 */
> +	tail = uatomic_xchg(&src_q_tail->p, &src_q_head->node);
> +
> +	/*
> +	 * Append the spliced content of src_q into dest_q. Does not
> +	 * require mutual exclusion on dest_q (wait-free).
> +	 */
> +	___cds_wfcq_append(dest_q_head, dest_q_tail, head, tail);
> +}
> +
> +/*
> + * cds_wfcq_dequeue_blocking: dequeue a node from a wait-free queue.
> + *
> + * Content written into the node before enqueue is guaranteed to be
> + * consistent, but no other memory ordering is ensured.
> + * Mutual exlusion with (and only with) cds_wfcq_splice_blocking is
> + * ensured.
> + * It is valid to reuse and free a dequeued node immediately.
> + */
> +static inline struct cds_wfcq_node *
> +_cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
> +		struct cds_wfcq_tail *tail)
> +{
> +	struct cds_wfcq_node *retval;
> +
> +	_cds_wfcq_dequeue_lock(head, tail);
> +	retval = ___cds_wfcq_dequeue_blocking(head, tail);
> +	_cds_wfcq_dequeue_unlock(head, tail);
> +	return retval;
> +}
> +
> +/*
> + * cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
> + *
> + * Dequeue all nodes from src_q.
> + * dest_q must be already initialized.
> + * Content written into the node before enqueue is guaranteed to be
> + * consistent, but no other memory ordering is ensured.
> + * Mutual exlusion with (and only with) cds_wfcq_dequeue_blocking is
> + * ensured.
> + */
> +static inline void
> +_cds_wfcq_splice_blocking(
> +		struct cds_wfcq_head *dest_q_head,
> +		struct cds_wfcq_tail *dest_q_tail,
> +		struct cds_wfcq_head *src_q_head,
> +		struct cds_wfcq_tail *src_q_tail)
> +{
> +	_cds_wfcq_dequeue_lock(src_q_head, src_q_tail);
> +	___cds_wfcq_splice_blocking(dest_q_head, dest_q_tail,
> +			src_q_head, src_q_tail);
> +	_cds_wfcq_dequeue_unlock(src_q_head, src_q_tail);
> +}
> +
> +#ifdef __cplusplus
> +}
> +#endif
> +
> +#endif /* _URCU_WFCQUEUE_STATIC_H */
> diff --git a/urcu/wfcqueue.h b/urcu/wfcqueue.h
> new file mode 100644
> index 0000000..5576cbf
> --- /dev/null
> +++ b/urcu/wfcqueue.h
> @@ -0,0 +1,263 @@
> +#ifndef _URCU_WFCQUEUE_H
> +#define _URCU_WFCQUEUE_H
> +
> +/*
> + * wfcqueue.h
> + *
> + * Userspace RCU library - Concurrent Queue with Wait-Free Enqueue/Blocking Dequeue
> + *
> + * Copyright 2010-2012 - Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
> + * Copyright 2011-2012 - Lai Jiangshan <laijs at cn.fujitsu.com>
> + *
> + * This library is free software; you can redistribute it and/or
> + * modify it under the terms of the GNU Lesser General Public
> + * License as published by the Free Software Foundation; either
> + * version 2.1 of the License, or (at your option) any later version.
> + *
> + * This library is distributed in the hope that it will be useful,
> + * but WITHOUT ANY WARRANTY; without even the implied warranty of
> + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
> + * Lesser General Public License for more details.
> + *
> + * You should have received a copy of the GNU Lesser General Public
> + * License along with this library; if not, write to the Free Software
> + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
> + */
> +
> +#include <pthread.h>
> +#include <assert.h>
> +#include <stdbool.h>
> +#include <urcu/compiler.h>
> +#include <urcu/arch.h>
> +
> +#ifdef __cplusplus
> +extern "C" {
> +#endif
> +
> +/*
> + * Concurrent queue with wait-free enqueue/blocking dequeue.
> + *
> + * Inspired from half-wait-free/half-blocking queue implementation done by
> + * Paul E. McKenney.
> + */
> +
> +struct cds_wfcq_node {
> +	struct cds_wfcq_node *next;
> +};
> +
> +/*
> + * Do not put head and tail on the same cache-line if concurrent
> + * enqueue/dequeue are expected from many CPUs. This eliminates
> + * false-sharing between enqueue and dequeue.
> + */
> +struct cds_wfcq_head {
> +	struct cds_wfcq_node node;
> +	pthread_mutex_t lock;
> +};
> +
> +struct cds_wfcq_tail {
> +	struct cds_wfcq_node *p;
> +};

Why use "p" here?

I don't see the commits in the public git-tree. Or you forgot to update the
tree and make "http" address happy.

Thanks,
Lai

[lttng-dev] [URCU PATCH 1/3] wfcqueue: implement concurrency-efficient queue

[Mathieu Desnoyers]

* Lai Jiangshan (laijs at cn.fujitsu.com) wrote:
> On 10/02/2012 10:14 PM, Mathieu Desnoyers wrote:
[...]
> > --- /dev/null
> > +++ b/urcu/wfcqueue.h
> > @@ -0,0 +1,263 @@
> > +#ifndef _URCU_WFCQUEUE_H
> > +#define _URCU_WFCQUEUE_H
> > +
> > +/*
> > + * wfcqueue.h
> > + *
> > + * Userspace RCU library - Concurrent Queue with Wait-Free Enqueue/Blocking Dequeue
> > + *
> > + * Copyright 2010-2012 - Mathieu Desnoyers <mathieu.desnoyers at efficios.com>
> > + * Copyright 2011-2012 - Lai Jiangshan <laijs at cn.fujitsu.com>
> > + *
> > + * This library is free software; you can redistribute it and/or
> > + * modify it under the terms of the GNU Lesser General Public
> > + * License as published by the Free Software Foundation; either
> > + * version 2.1 of the License, or (at your option) any later version.
> > + *
> > + * This library is distributed in the hope that it will be useful,
> > + * but WITHOUT ANY WARRANTY; without even the implied warranty of
> > + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
> > + * Lesser General Public License for more details.
> > + *
> > + * You should have received a copy of the GNU Lesser General Public
> > + * License along with this library; if not, write to the Free Software
> > + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
> > + */
> > +
> > +#include <pthread.h>
> > +#include <assert.h>
> > +#include <stdbool.h>
> > +#include <urcu/compiler.h>
> > +#include <urcu/arch.h>
> > +
> > +#ifdef __cplusplus
> > +extern "C" {
> > +#endif
> > +
> > +/*
> > + * Concurrent queue with wait-free enqueue/blocking dequeue.
> > + *
> > + * Inspired from half-wait-free/half-blocking queue implementation done by
> > + * Paul E. McKenney.
> > + */
> > +
> > +struct cds_wfcq_node {
> > +	struct cds_wfcq_node *next;
> > +};
> > +
> > +/*
> > + * Do not put head and tail on the same cache-line if concurrent
> > + * enqueue/dequeue are expected from many CPUs. This eliminates
> > + * false-sharing between enqueue and dequeue.
> > + */
> > +struct cds_wfcq_head {
> > +	struct cds_wfcq_node node;
> > +	pthread_mutex_t lock;
> > +};
> > +
> > +struct cds_wfcq_tail {
> > +	struct cds_wfcq_node *p;
> > +};
> 
> Why use "p" here?

For lack of imagination. ;)  Do you have something better to propose ?

> 
> I don't see the commits in the public git-tree. Or you forgot to update the
> tree and make "http" address happy.

I have pushed those into the userspace RCU master branch a couple of
days after posting them to the mailing lists. I just synchronized my
urcu/wfcqueue volatile dev branch, but I expect to drop it soon, since
it's now merged back into master.

Thanks!

Mathieu

> 
> Thanks,
> Lai

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[lttng-dev] [rp] [URCU PATCH 0/3] wait-free concurrent queues (wfcqueue)

[Mathieu Desnoyers]

* Lai Jiangshan (laijs at cn.fujitsu.com) wrote:
> On 10/08/2012 11:07 PM, Mathieu Desnoyers wrote:
> > * Lai Jiangshan (laijs at cn.fujitsu.com) wrote:
> >> On 10/04/2012 05:04 AM, Mathieu Desnoyers wrote:
> >>> * Paul E. McKenney (paulmck at linux.vnet.ibm.com) wrote:
> >>>> On Tue, Oct 02, 2012 at 10:13:07AM -0400, Mathieu Desnoyers wrote:
> >>>>> Implement wait-free concurrent queues, with a new API different from
> >>>>> wfqueue.h, which is already provided by Userspace RCU. The advantage of
> >>>>> splitting the head and tail objects of the queue into different
> >>>>> arguments is to allow these to sit on different cache-lines, thus
> >>>>> eliminating false-sharing, leading to a 2.3x speed increase.
> >>>>>
> >>>>> This API also introduces a "splice" operation, which moves all nodes
> >>>>> from one queue into another, and postpones the synchronization to either
> >>>>> dequeue or iteration on the list. The splice operation does not need to
> >>>>> touch every single node of the queue it moves them from. Moreover, the
> >>>>> splice operation only needs to ensure mutual exclusion with other
> >>>>> dequeuers, iterations, and splice operations from the list it splices
> >>>>> from, but acts as a simple enqueuer on the list it splices into (no
> >>>>> mutual exclusion needed for that list).
> >>>>>
> >>>>> Feedback is welcome,
> >>>>
> >>>> These look sane to me, though I must confess that the tail pointer
> >>>> referencing the node rather than the node's next pointer did throw
> >>>> me for a bit.  ;-)
> >>>
> >>> Yes, this was originally introduced with Lai's original patch to
> >>> wfqueue, which I think is a nice simplification: it's pretty much the
> >>> same thing to use the last node address as tail rather than the address
> >>> of its first member (its next pointer address (_not_ value)). It ends up
> >>> being the same address in this case, but more interestingly, we don't
> >>> have to use a struct cds_wfcq_node ** type: a simple struct
> >>> cds_wfcq_node *  suffice.
> >>>
> >>> Thanks Paul, I will therefore merge these 3 patches with your Acked-by.
> >>>
> >>> Lai, you are welcome to provide improvements to this code against the
> >>> master branch. I will gladly consider any change you propose.
> >>>
> >>
> >> I did not remember that there is any improvement idea not included.
> >> The patchset is OK for me.
> > 
> > Great! Would you be OK if I commit the following patch ? Let me know if
> > you want me to put your signed-off-by on this (I can even put your email
> > as From if you like):
> > 
> > 
> > wfcqueue: update credits in patch documentation
> > 
> > Give credits to those responsible for the design and implementation of
> > commit 8ad4ce587f001ae026d5560ac509c2e48986130b, "wfcqueue: implement
> > concurrency-efficient queue", which happened through rounds of email and
> > patch exchanges.
> > 
> > Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers at efficios.com>

Hi Lai,

Are you OK with this credits patch ?

> > ---
> > diff --git a/urcu/static/wfcqueue.h b/urcu/static/wfcqueue.h
> > index a989984..153143d 100644
> > --- a/urcu/static/wfcqueue.h
> > +++ b/urcu/static/wfcqueue.h
> > @@ -41,8 +41,10 @@ extern "C" {
> >  /*
> >   * Concurrent queue with wait-free enqueue/blocking dequeue.
> >   *
> > - * Inspired from half-wait-free/half-blocking queue implementation done by
> > - * Paul E. McKenney.
> > + * This queue has been designed and implemented collaboratively by
> > + * Mathieu Desnoyers and Lai Jiangshan. Inspired from
> > + * half-wait-free/half-blocking queue implementation done by Paul E.
> > + * McKenney.
> >   *
> >   * Mutual exclusion of __cds_wfcq_* API
> >   *
> > diff --git a/urcu/wfcqueue.h b/urcu/wfcqueue.h
> > index 5576cbf..940dc7d 100644
> > --- a/urcu/wfcqueue.h
> > +++ b/urcu/wfcqueue.h
> > @@ -37,8 +37,10 @@ extern "C" {
> >  /*
> >   * Concurrent queue with wait-free enqueue/blocking dequeue.
> >   *
> > - * Inspired from half-wait-free/half-blocking queue implementation done by
> > - * Paul E. McKenney.
> > + * This queue has been designed and implemented collaboratively by
> > + * Mathieu Desnoyers and Lai Jiangshan. Inspired from
> > + * half-wait-free/half-blocking queue implementation done by Paul E.
> > + * McKenney.
> >   */
> >  
> >  struct cds_wfcq_node {
> > 
> > 
> >> I think you can reimplement wfqueue via wfcqueue without cacheline opt.
> > 
> > Hrm, semantically this can indeed be done, but I fear that we might not
> > be strictly ABI-compatible with the old wfqueue. So I would be tempted
> > to leave the old wfqueue implementation as-is, and maybe deprecate it at
> > some point. Thoughts ?
> > 
> 
> All APIs is not changed, they are forwarded to wfcqueue APIs in implementation,
> so ABI of APIs is compatible. The only thing is struct cds_wfq_queue:
> 
> struct cds_wfq_queue {
> 	struct cds_wfq_node *head, **tail;
> 	struct cds_wfq_node dummy;	/* Dummy node */
> 	pthread_mutex_t lock;
> };
> 
> 
> We can redefine it as:
> 
> #define cds_wfq_node cds_wfcq_node
> 
> struct cds_wfq_queue {
> 	union {
> 		struct cds_wfq_node *head; /* make bug-user who wrongly directly access to ->head happy */
> 		struct cds_wfcq_node *__pad; /* not used in new implement */
> 	};
> 	union {
> 		struct cds_wfq_node **tail;  /* make bug-user who wrongly directly access to ->tail happy */
> 		struct cds_wfcq_tail real_tail;
> 	};
> 	union {
> 		struct {
> 			struct cds_wfq_node dummy;	/* Dummy node */
> 			pthread_mutex_t lock;
> 		}
> 		struct cds_wfcq_head real_head;
> 	};
> }
> 
> static inline void _cds_wfq_init(struct cds_wfq_queue *q)
> {
> 	q->head = &q->dummy; /* make bug-user who wrongly directly access to ->head happy */
> 	_cds_wfcq_init(&q->real_head, &q->real_tail);
> }
> 
> after this change, struct cds_wfq_queue is not changed.
> Even bug-user wrongly directly access to struct cds_wfq_queue by old-view,
> the queue is compatible:
> 	head->dummy node->real node->real node....
> 	tail->real tail node(or dummy node)
> 
> the only different is that: dummy node is always the first node by old-view.
> 
> 
> And if we deprecate struct cds_wfq_queue, I think we should provide a
> new default wfqueue to users.

I guess the objective is nice, but I'm wondering if your mapping of
wfqueue onto wfcqueue covers scenarios where we have objects that
uses both the old and new view statically linked into the same program ?

My ABI concern is not just about queue users directly accessing the
fields of the queue, but also about binary-level compatibility of mixed
old-new queues.

Thoughts ?

Thanks,

Mathieu

> 
> Thanks,
> Lai

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com