X-Git-Url: http://git.liburcu.org/?p=urcu.git;a=blobdiff_plain;f=urcu%2Fstatic%2Frculfqueue.h;h=af73c6f2ef7565dc69e16f62d539ded7ed66d583;hp=33e71f5c4ec9e1d9c78b848ba52537fa1fee3b6f;hb=4157e1ac43033c8b3ef5405b9c01e8311b2454ca;hpb=0cca1a2d1b9861591795df4e48df73ac9919d772

diff --git a/urcu/static/rculfqueue.h b/urcu/static/rculfqueue.h
index 33e71f5..af73c6f 100644
--- a/urcu/static/rculfqueue.h
+++ b/urcu/static/rculfqueue.h
@@ -28,6 +28,7 @@
 
 #include <urcu-call-rcu.h>
 #include <urcu/uatomic.h>
+#include <urcu-pointer.h>
 #include <assert.h>
 #include <errno.h>
 
@@ -59,12 +60,6 @@ struct cds_lfq_node_rcu_dummy {
  * grace period.
  */
 
-static inline
-int is_dummy(struct cds_lfq_node_rcu *node)
-{
-	return ((unsigned long) node) & 0x1UL;
-}
-
 static inline
 struct cds_lfq_node_rcu *make_dummy(struct cds_lfq_queue_rcu *q,
 				    struct cds_lfq_node_rcu *next)
@@ -74,18 +69,13 @@ struct cds_lfq_node_rcu *make_dummy(struct cds_lfq_queue_rcu *q,
 	dummy = malloc(sizeof(struct cds_lfq_node_rcu_dummy));
 	assert(dummy);
 	dummy->parent.next = next;
+	dummy->parent.dummy = 1;
 	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
-void free_dummy(struct rcu_head *head)
+void free_dummy_cb(struct rcu_head *head)
 {
 	struct cds_lfq_node_rcu_dummy *dummy =
 		caa_container_of(head, struct cds_lfq_node_rcu_dummy, head);
@@ -97,15 +87,26 @@ 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->q->queue_call_rcu(&dummy->head, free_dummy);
+	assert(node->dummy);
+	dummy = caa_container_of(node, struct cds_lfq_node_rcu_dummy, parent);
+	dummy->q->queue_call_rcu(&dummy->head, free_dummy_cb);
+}
+
+static inline
+void free_dummy(struct cds_lfq_node_rcu *node)
+{
+	struct cds_lfq_node_rcu_dummy *dummy;
+
+	assert(node->dummy);
+	dummy = caa_container_of(node, struct cds_lfq_node_rcu_dummy, parent);
+	free(dummy);
 }
 
 static inline
 void _cds_lfq_node_init_rcu(struct cds_lfq_node_rcu *node)
 {
 	node->next = NULL;
+	node->dummy = 0;
 }
 
 static inline
@@ -129,9 +130,9 @@ int _cds_lfq_destroy_rcu(struct cds_lfq_queue_rcu *q)
 	struct cds_lfq_node_rcu *head;
 
 	head = rcu_dereference(q->head);
-	if (!(is_dummy(head) && get_node(head)->next == NULL))
+	if (!(head->dummy && head->next == NULL))
 		return -EPERM;	/* not empty */
-	rcu_free_dummy(head);
+	free_dummy(head);
 	return 0;
 }
 
@@ -151,7 +152,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
@@ -171,6 +172,16 @@ void _cds_lfq_enqueue_rcu(struct cds_lfq_queue_rcu *q,
 	}
 }
 
+static inline
+void enqueue_dummy(struct cds_lfq_queue_rcu *q)
+{
+	struct cds_lfq_node_rcu *node;
+
+	/* We need to reallocate to protect from ABA. */
+	node = make_dummy(q, NULL);
+	_cds_lfq_enqueue_rcu(q, node);
+}
+
 /*
  * Should be called under rcu read lock critical section.
  *
@@ -185,39 +196,28 @@ 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) && next == NULL)
+		next = rcu_dereference(head->next);
+		if (head->dummy && next == NULL)
 			return NULL;	/* empty */
 		/*
 		 * We never, ever allow dequeue to get to a state where
 		 * the queue is empty (we need at least one node in the
 		 * queue). This is ensured by checking if the head next
-		 * is NULL and retry in that case (this means a
-		 * concurrent dummy node re-enqueue is in progress).
+		 * is NULL, which means we need to enqueue a dummy node
+		 * before we can hope dequeuing anything.
 		 */
-		if (next) {
-			if (uatomic_cmpxchg(&q->head, head, next) == head) {
-				if (is_dummy(head)) {
-					struct cds_lfq_node_rcu *node;
-					/*
-					 * Requeue dummy. We need to
-					 * reallocate to protect from
-					 * ABA.
-					 */
-					rcu_free_dummy(head);
-					node = make_dummy(q, NULL);
-					_cds_lfq_enqueue_rcu(q, node);
-					continue;	/* try again */
-				}
-				return head;
-			} else {
-				/* Concurrently pushed, retry */
-				continue;
-			}
-		} else {
-			/* Dummy node re-enqueue is in progress, retry. */
-			continue;
+		if (!next) {
+			enqueue_dummy(q);
+			next = rcu_dereference(head->next);
+		}
+		if (uatomic_cmpxchg(&q->head, head, next) != head)
+			continue;	/* Concurrently pushed. */
+		if (head->dummy) {
+			/* Free dummy after grace period. */
+			rcu_free_dummy(head);
+			continue;	/* try again */
 		}
+		return head;
 	}
 }