waitqueue: add in_waitqueue field

Using the waitqueue node next pointer to figure out if the node is
within the queue worked when wfstack was used to implement the queue
stack, but unfortunately it does not keep the same semantic with
lfstack: indeed, the bottom of stack is indicated by a NULL pointer too.

Keep this state in a separate field instead to eliminate coupling
between workqueue, waitqueue and lfstack.

Note that it is OK for a worker to see "in_waitqueue" as temporarily
false when the wait node has just been pop'd from the waitqueue. It's
the opposite that we never want to happen: a worker should never see
"in_waitqueue" as false when the node is actually in the waitqueue,
because it would cause queue corruption.

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>

diff --git a/urcu/waitqueue-lifo.h b/urcu/waitqueue-lifo.h
index fccba17844001cdd6f678ff9ad7bfb406b72465c..96b211c68355bbd9611c4895ebaab5a295e961fb 100644 (file)
--- a/urcu/waitqueue-lifo.h
+++ b/urcu/waitqueue-lifo.h
@@ -45,6 +45,7 @@ enum urcu_wait_state {
 struct urcu_wait_node {
        struct cds_lfs_node node;
        int32_t state;  /* enum urcu_wait_state */
+       int in_waitqueue;
 };
 
 #define URCU_WAIT_NODE_INIT(name, _state)              \
@@ -88,6 +89,8 @@ static inline
 bool urcu_wait_add(struct urcu_wait_queue *queue,
                struct urcu_wait_node *node)
 {
+       cds_lfs_node_init(&node->node);
+       CMM_STORE_SHARED(node->in_waitqueue, true);
        return cds_lfs_push(&queue->stack, &node->node);
 }
 
@@ -122,6 +125,13 @@ void urcu_wait_node_init(struct urcu_wait_node *node,
 {
        urcu_wait_set_state(node, state);
        cds_lfs_node_init(&node->node);
+       node->in_waitqueue = false;
+}
+
+static inline
+bool urcu_in_waitqueue(struct urcu_wait_node *node)
+{
+       return CMM_LOAD_SHARED(node->in_waitqueue);
 }
 
 /*
@@ -206,7 +216,7 @@ int urcu_dequeue_wake_single(struct urcu_wait_queue *queue)
        if (!node)
                return -ENOENT;
        wait_node = caa_container_of(node, struct urcu_wait_node, node);
-       CMM_STORE_SHARED(wait_node->node.next, NULL);
+       CMM_STORE_SHARED(wait_node->in_waitqueue, false);
        /* Don't wake already running threads */
        if (!(wait_node->state & URCU_WAIT_RUNNING))
                ret = urcu_adaptative_wake_up(wait_node);
@@ -247,7 +257,7 @@ int urcu_wake_all_waiters(struct urcu_waiters *waiters)
                struct urcu_wait_node *wait_node =
                        caa_container_of(iter, struct urcu_wait_node, node);
 
-               CMM_STORE_SHARED(wait_node->node.next, NULL);
+               CMM_STORE_SHARED(wait_node->in_waitqueue, false);
                /* Don't wake already running threads */
                if (wait_node->state & URCU_WAIT_RUNNING)
                        continue;

diff --git a/urcu/workqueue-fifo.h b/urcu/workqueue-fifo.h
index e2dd72f0c1dc66c47069c8b1f7292ff74938ee5c..6918a8217bdb3e289404ca3983540073d9234fd8 100644 (file)
--- a/urcu/workqueue-fifo.h
+++ b/urcu/workqueue-fifo.h
@@ -195,7 +195,7 @@ void urcu_worker_unregister(struct urcu_workqueue *queue,
        /*
         * Make sure we are removed from waitqueue.
         */
-       if (CMM_LOAD_SHARED(worker->wait_node.node.next))
+       if (urcu_in_waitqueue(&worker->wait_node))
                __urcu_workqueue_wakeup_all(queue);
 
        /*
@@ -404,14 +404,9 @@ enum urcu_accept_ret urcu_accept_work(struct urcu_worker *worker)
                return URCU_ACCEPT_SHUTDOWN;
        urcu_wait_set_state(&worker->wait_node,
                        URCU_WAIT_WAITING);
-       if (!CMM_LOAD_SHARED(worker->wait_node.node.next)) {
+       if (!urcu_in_waitqueue(&worker->wait_node)) {
                int was_empty;
 
-               /*
-                * NULL next pointer. We are therefore not in
-                * the queue.
-                */
-               cds_lfs_node_init(&worker->wait_node.node);
                /* Protect stack dequeue against ABA */
                synchronize_rcu();
                was_empty = !urcu_wait_add(&queue->waitqueue,
@@ -433,13 +428,11 @@ enum urcu_accept_ret urcu_accept_work(struct urcu_worker *worker)
                }
        } else {
                /*
-                * Non-NULL next pointer. We are therefore in
-                * the queue, or the dispatcher just removed us
-                * from it (after we read the next pointer), and
-                * is therefore awakening us. The state will
-                * therefore have been changed from WAITING to
-                * some other state, which will let the busy
-                * wait pass through.
+                * We are in the queue, or the dispatcher just removed
+                * us from it (after we read the next pointer), and is
+                * therefore awakening us. The state will therefore have
+                * been changed from WAITING to some other state, which
+                * will let the busy wait pass through.
                 */
        }
        urcu_adaptative_busy_wait(&worker->wait_node);