Implementation better suited for supporting call_rcu().
Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
test_urcu_bp_dynamic_link_CFLAGS = -DDYNAMIC_LINK_TEST $(AM_CFLAGS)
test_urcu_lfq_SOURCES = test_urcu_lfq.c $(URCU_DEFER)
-test_urcu_wfq_SOURCES = test_urcu_wfq.c $(URCU_DEFER)
+test_urcu_wfq_SOURCES = test_urcu_wfq.c
test_urcu_lfs_SOURCES = test_urcu_lfs.c $(URCU_DEFER)
test_urcu_wfs_SOURCES = test_urcu_wfs.c $(URCU_DEFER)
#define _LGPL_SOURCE
#endif
#include <urcu.h>
-#include <urcu/rcuwfqueue.h>
-#include <urcu-defer.h>
+#include <urcu/wfqueue.h>
static volatile int test_go, test_stop;
static unsigned int nr_enqueuers;
static unsigned int nr_dequeuers;
-static struct rcu_wfq_queue q;
+static struct wfq_queue q;
void *thr_enqueuer(void *_count)
{
set_affinity();
- rcu_register_thread();
-
while (!test_go)
{
}
smp_mb();
for (;;) {
- struct rcu_wfq_node *node = malloc(sizeof(*node));
+ struct wfq_node *node = malloc(sizeof(*node));
if (!node)
goto fail;
- rcu_wfq_node_init(node);
- rcu_wfq_enqueue(&q, node);
+ wfq_node_init(node);
+ wfq_enqueue(&q, node);
nr_successful_enqueues++;
if (unlikely(wdelay))
break;
}
- rcu_unregister_thread();
-
count[0] = nr_enqueues;
count[1] = nr_successful_enqueues;
printf_verbose("enqueuer thread_end, thread id : %lx, tid %lu, "
}
-static void rcu_release_node(struct urcu_ref *ref)
-{
- struct rcu_wfq_node *node = container_of(ref, struct rcu_wfq_node, ref);
- defer_rcu(free, node);
- //synchronize_rcu();
- //free(node);
-}
-
void *thr_dequeuer(void *_count)
{
unsigned long long *count = _count;
set_affinity();
- rcu_defer_register_thread();
- rcu_register_thread();
-
while (!test_go)
{
}
smp_mb();
for (;;) {
- struct rcu_wfq_node *node =
- rcu_wfq_dequeue_blocking(&q, rcu_release_node);
+ struct wfq_node *node = wfq_dequeue_blocking(&q);
if (node) {
- urcu_ref_put(&node->ref, rcu_release_node);
+ free(node);
nr_successful_dequeues++;
}
loop_sleep(rduration);
}
- rcu_unregister_thread();
- rcu_defer_unregister_thread();
-
printf_verbose("dequeuer thread_end, thread id : %lx, tid %lu, "
"dequeues %llu, successful_dequeues %llu\n",
pthread_self(), (unsigned long)gettid(), nr_dequeues,
return ((void*)2);
}
-static void release_node(struct urcu_ref *ref)
-{
- struct rcu_wfq_node *node = container_of(ref, struct rcu_wfq_node, ref);
- free(node);
-}
-
-void test_end(struct rcu_wfq_queue *q, unsigned long long *nr_dequeues)
+void test_end(struct wfq_queue *q, unsigned long long *nr_dequeues)
{
- struct rcu_wfq_node *node;
+ struct wfq_node *node;
do {
- node = rcu_wfq_dequeue_blocking(q, release_node);
+ node = wfq_dequeue_blocking(q);
if (node) {
- urcu_ref_put(&node->ref, release_node);
+ free(node);
(*nr_dequeues)++;
}
} while (node);
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);
- rcu_wfq_init(&q);
+ wfq_init(&q);
next_aff = 0;
+++ /dev/null
-#ifndef _URCU_RCUWFQUEUE_H
-#define _URCU_RCUWFQUEUE_H
-
-/*
- * rcuwfqueue.h
- *
- * Userspace RCU library - RCU Queue with Wait-Free Enqueue/Blocking Dequeue
- *
- * Copyright 2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.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 <urcu/urcu_ref.h>
-#include <assert.h>
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#if (!defined(_GNU_SOURCE) && !defined(_LGPL_SOURCE))
-#error "Dynamic loader LGPL wrappers not implemented yet"
-#endif
-
-/*
- * RCU queue with wait-free enqueue/blocking dequeue using reference counting.
- * Enqueue and dequeue operations hold a RCU read lock to deal with cmpxchg ABA
- * problem. This implementation keeps a dummy head node to ensure we can always
- * update the queue locklessly. Given that this is a queue, the dummy head node
- * must always advance as we dequeue entries. Therefore, we keep a reference
- * count on each entry we are dequeueing, so they can be kept as dummy head node
- * until the next dequeue, at which point their reference count will be
- * decremented.
- */
-
-#define URCU_WFQ_PERMANENT_REF 128
-
-struct rcu_wfq_node {
- struct rcu_wfq_node *next;
- struct urcu_ref ref;
-};
-
-struct rcu_wfq_queue {
- struct rcu_wfq_node *head, *tail;
- struct rcu_wfq_node init; /* Dummy initialization node */
-};
-
-void rcu_wfq_node_init(struct rcu_wfq_node *node)
-{
- node->next = NULL;
- urcu_ref_init(&node->ref);
-}
-
-void rcu_wfq_init(struct rcu_wfq_queue *q)
-{
- rcu_wfq_node_init(&q->init);
- /* Make sure the initial node is never freed. */
- urcu_ref_set(&q->init.ref, URCU_WFQ_PERMANENT_REF);
- /* Set queue end */
- q->head = q->tail = &q->init;
-}
-
-void rcu_wfq_enqueue(struct rcu_wfq_queue *q, struct rcu_wfq_node *node)
-{
- struct rcu_wfq_node *old_tail;
-
- urcu_ref_get(&node->ref);
- /*
- * uatomic_xchg() implicit memory barrier orders earlier stores to node
- * (setting it to NULL and incrementing the refcount) before
- * publication.
- */
- old_tail = uatomic_xchg(&q->tail, node);
- /*
- * At this point, dequeuers see a NULL old_tail->next, which indicates
- * end of queue. The following store will append "node" to the queue
- * from a dequeuer perspective.
- */
- STORE_SHARED(old_tail->next, node);
-}
-
-/*
- * The entry returned by dequeue must be taken care of by doing a urcu_ref_put,
- * which calls the release primitive when the reference count drops to zero. A
- * grace period must be waited before performing the actual memory reclamation
- * in the release primitive. The wfq node returned by dequeue must not be
- * modified/re-used/freed until the reference count reaches zero and a grace
- * period has elapsed (after the refcount reached 0).
- *
- * 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.
- */
-struct rcu_wfq_node *
-rcu_wfq_dequeue_blocking(struct rcu_wfq_queue *q,
- void (*release)(struct urcu_ref *))
-{
- for (;;) {
- struct rcu_wfq_node *head, *next;
-
- rcu_read_lock();
- head = rcu_dereference(q->head);
- next = rcu_dereference(head->next);
- if (next) {
- if (uatomic_cmpxchg(&q->head, head, next) == head) {
- rcu_read_unlock();
- urcu_ref_put(&head->ref, release);
- return next;
- } else {
- /* Concurrently pushed, retry */
- rcu_read_unlock();
- continue;
- }
- } else {
- /* Empty */
- rcu_read_unlock();
- return NULL;
- }
- }
-}
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /* _URCU_RCUWFQUEUE_H */
--- /dev/null
+#ifndef _URCU_WFQUEUE_H
+#define _URCU_WFQUEUE_H
+
+/*
+ * wfqueue.h
+ *
+ * Userspace RCU library - Queue with Wait-Free Enqueue/Blocking Dequeue
+ *
+ * Copyright 2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.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 <urcu/compiler.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#if (!defined(_GNU_SOURCE) && !defined(_LGPL_SOURCE))
+#error "Dynamic loader LGPL wrappers not implemented yet"
+#endif
+
+/*
+ * Queue with wait-free enqueue/blocking dequeue.
+ * This implementation adds a dummy head node when the queue is empty to ensure
+ * we can always update the queue locklessly.
+ *
+ * Inspired from half-wait-free/half-blocking queue implementation done by
+ * Paul E. McKenney.
+ */
+
+struct wfq_node {
+ struct wfq_node *next;
+};
+
+struct wfq_queue {
+ struct wfq_node *head, **tail;
+ struct wfq_node dummy; /* Dummy node */
+ pthread_mutex_t lock;
+};
+
+void wfq_node_init(struct wfq_node *node)
+{
+ node->next = NULL;
+}
+
+void wfq_init(struct wfq_queue *q)
+{
+ int ret;
+
+ wfq_node_init(&q->dummy);
+ /* Set queue head and tail */
+ q->head = &q->dummy;
+ q->tail = &q->dummy.next;
+ ret = pthread_mutex_init(&q->lock, NULL);
+ assert(!ret);
+}
+
+void wfq_enqueue(struct wfq_queue *q, struct wfq_node *node)
+{
+ struct wfq_node **old_tail;
+
+ /*
+ * uatomic_xchg() implicit memory barrier orders earlier stores to data
+ * structure containing node and setting node->next to NULL before
+ * publication.
+ */
+ old_tail = uatomic_xchg(&q->tail, node);
+ /*
+ * At this point, dequeuers see a NULL old_tail->next, which indicates
+ * that the queue is being appended to. The following store will append
+ * "node" to the queue from a dequeuer perspective.
+ */
+ STORE_SHARED(*old_tail, node);
+}
+
+/*
+ * It is valid to reuse and free a dequeued node immediately.
+ *
+ * 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.
+ */
+struct wfq_node *
+__wfq_dequeue_blocking(struct wfq_queue *q)
+{
+ struct wfq_node *node, *next;
+ int busy_wait = 16;
+
+ /*
+ * Queue is empty if it only contains the dummy node.
+ */
+ if (q->head == &q->dummy && LOAD_SHARED(q->tail) == &q->dummy.next)
+ return NULL;
+ node = q->head;
+
+ /*
+ * Adaptative busy-looping waiting for enqueuer to complete enqueue.
+ */
+ while ((next = LOAD_SHARED(node->next)) == NULL) {
+ if (busy_wait > 0) {
+ cpu_relax();
+ busy_wait--;
+ } else
+ poll(NULL, 0, 1); /* Wait for 1ms */
+ }
+ /*
+ * Move queue head forward.
+ */
+ q->head = next;
+ /*
+ * Requeue dummy node if we just dequeued it.
+ */
+ if (node == &q->dummy) {
+ wfq_node_init(node);
+ wfq_enqueue(q, node);
+ return __wfq_dequeue_blocking(q);
+ }
+ return node;
+}
+
+struct wfq_node *
+wfq_dequeue_blocking(struct wfq_queue *q)
+{
+ struct wfq_node *retnode;
+ int ret;
+
+ ret = pthread_mutex_lock(&q->lock);
+ assert(!ret);
+ retnode = __wfq_dequeue_blocking(q);
+ ret = pthread_mutex_unlock(&q->lock);
+ assert(!ret);
+ return retnode;
+}
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _URCU_WFQUEUE_H */
projects / userspace-rcu.git / commitdiff
