rculfqueue: Keep a reference to the current dummy node rather than using low bit

[urcu.git] / urcu / static / rculfqueue.h

diff --git a/urcu/static/rculfqueue.h b/urcu/static/rculfqueue.h
index 0b83bc37ffb2c5b0e3b19218299286e571d70580..0970c0f3f0cf2fdfd1a1f31c936bac059d712804 100644 (file)
--- a/urcu/static/rculfqueue.h
+++ b/urcu/static/rculfqueue.h
@@ -30,7 +30,6 @@
 #include <urcu/uatomic.h>
 #include <assert.h>
 #include <errno.h>
-/* A urcu implementation header should be already included. */
 
 #ifdef __cplusplus
 extern "C" {
@@ -46,30 +45,24 @@ struct cds_lfq_node_rcu_dummy {
  * Lock-free RCU queue. Enqueue and dequeue operations hold a RCU read
  * lock to deal with cmpxchg ABA problem. This queue is *not* circular:
  * head points to the oldest node, tail points to the newest node.
- * Dummy nodes are kept to ensure enqueue and dequeue can always proceed
+ * A dummy node is kept to ensure enqueue and dequeue can always proceed
  * concurrently. Keeping a separate head and tail helps with large
  * queues: enqueue and dequeue can proceed concurrently without
  * wrestling for exclusive access to the same variables.
  *
- * We keep two dummy nodes in the queue to distinguish between empty queue
- * state and intermediate state while a dummy node dequeue/requeue is
- * being performed. Dequeue retry if it detects that it would be
- * dequeueing the last node (it means a dummy node dequeue-requeue is in
- * progress). This ensures that there is always at least one node in
- * the queue. In a situation where the two dummy nodes are being
- * requeued (they therefore don't appear in the queue at a given
- * moment), we are certain that there is at least one non-dummy node in
- * the queue (ensured by the test for NULL next node upon dequeue).
+ * Dequeue retry if it detects that it would be dequeueing the last node
+ * (it means a dummy node dequeue-requeue is in progress). This ensures
+ * that there is always at least one node in the queue.
  *
- * In the dequeue operation, we internally reallocate the dummy nodes
- * upon dequeue/requeue and use call_rcu to free them after a grace
- * period.
+ * In the dequeue operation, we internally reallocate the dummy node
+ * upon dequeue/requeue and use call_rcu to free the old one after a
+ * grace period.
  */
 
 static inline
-int is_dummy(struct cds_lfq_node_rcu *node)
+int is_dummy(struct cds_lfq_queue_rcu *q, struct cds_lfq_node_rcu *node)
 {
-       return ((unsigned long) node) & 0x1UL;
+       return node == q->dummy;
 }
 
 static inline
@@ -82,13 +75,7 @@ struct cds_lfq_node_rcu *make_dummy(struct cds_lfq_queue_rcu *q,
        assert(dummy);
        dummy->parent.next = next;
        dummy->q = q;
-       return (struct cds_lfq_node_rcu *) (((unsigned long) &dummy->parent) | 0x1UL);
-}
-
-static inline
-struct cds_lfq_node_rcu *get_node(struct cds_lfq_node_rcu *node)
-{
-       return (struct cds_lfq_node_rcu *) (((unsigned long )node) & ~0x1UL);
+       return &dummy->parent;
 }
 
 static inline
@@ -104,8 +91,7 @@ void rcu_free_dummy(struct cds_lfq_node_rcu *node)
 {
        struct cds_lfq_node_rcu_dummy *dummy;
 
-       dummy = caa_container_of(get_node(node), struct cds_lfq_node_rcu_dummy,
-                                parent);
+       dummy = caa_container_of(node, struct cds_lfq_node_rcu_dummy, parent);
        dummy->q->queue_call_rcu(&dummy->head, free_dummy);
 }
 
@@ -121,7 +107,8 @@ void _cds_lfq_init_rcu(struct cds_lfq_queue_rcu *q,
                                void (*func)(struct rcu_head *head)))
 {
        q->tail = make_dummy(q, NULL);
-       q->head = make_dummy(q, q->tail);
+       q->dummy = q->tail;
+       q->head = q->tail;
        q->queue_call_rcu = queue_call_rcu;
 }
 
@@ -133,14 +120,12 @@ void _cds_lfq_init_rcu(struct cds_lfq_queue_rcu *q,
 static inline
 int _cds_lfq_destroy_rcu(struct cds_lfq_queue_rcu *q)
 {
-       struct cds_lfq_node_rcu *head, *next;
+       struct cds_lfq_node_rcu *head;
 
        head = rcu_dereference(q->head);
-       next = rcu_dereference(get_node(head)->next);
-       if (!(is_dummy(head) && is_dummy(next) && get_node(next)->next == NULL))
+       if (!(is_dummy(q, head) && head->next == NULL))
                return -EPERM;  /* not empty */
        rcu_free_dummy(head);
-       rcu_free_dummy(next);
        return 0;
 }
 
@@ -160,7 +145,7 @@ void _cds_lfq_enqueue_rcu(struct cds_lfq_queue_rcu *q,
                struct cds_lfq_node_rcu *tail, *next;
 
                tail = rcu_dereference(q->tail);
-               next = uatomic_cmpxchg(&get_node(tail)->next, NULL, node);
+               next = uatomic_cmpxchg(&tail->next, NULL, node);
                if (next == NULL) {
                        /*
                         * Tail was at the end of queue, we successfully
@@ -194,8 +179,8 @@ struct cds_lfq_node_rcu *_cds_lfq_dequeue_rcu(struct cds_lfq_queue_rcu *q)
                struct cds_lfq_node_rcu *head, *next;
 
                head = rcu_dereference(q->head);
-               next = rcu_dereference(get_node(head)->next);
-               if (is_dummy(head) && is_dummy(next) && get_node(next)->next == NULL)
+               next = rcu_dereference(head->next);
+               if (is_dummy(q, head) && next == NULL)
                        return NULL;    /* empty */
                /*
                 * We never, ever allow dequeue to get to a state where
@@ -206,7 +191,7 @@ struct cds_lfq_node_rcu *_cds_lfq_dequeue_rcu(struct cds_lfq_queue_rcu *q)
                 */
                if (next) {
                        if (uatomic_cmpxchg(&q->head, head, next) == head) {
-                               if (is_dummy(head)) {
+                               if (is_dummy(q, head)) {
                                        struct cds_lfq_node_rcu *node;
                                        /*
                                         * Requeue dummy. We need to
@@ -215,6 +200,12 @@ struct cds_lfq_node_rcu *_cds_lfq_dequeue_rcu(struct cds_lfq_queue_rcu *q)
                                         */
                                        rcu_free_dummy(head);
                                        node = make_dummy(q, NULL);
+                                       /*
+                                        * We are the only thread
+                                        * allowed to update dummy (we
+                                        * own the old dummy).
+                                        */
+                                       q->dummy = node;
                                        _cds_lfq_enqueue_rcu(q, node);
                                        continue;       /* try again */
                                }