workqueue: keep one work item from stealing co-workers

Takes care of lack of progress issue that could theoretically occur due
to a steady work-stealing circular loop amongst co-workers. Keeping a
work item hidden from co-workers ensures forward progress.

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

diff --git a/urcu/workqueue-fifo.h b/urcu/workqueue-fifo.h
index 1292e04d6a173f6a276d6fab70dfaca004f7f26d..883f41f3feec4c0869f27c794f44a420ba80fdb5 100644 (file)
--- a/urcu/workqueue-fifo.h
+++ b/urcu/workqueue-fifo.h
@@ -63,9 +63,13 @@ struct urcu_workqueue {
 };
 
 struct urcu_worker {
+       /* Workqueue which can be either used by worker, or stolen. */
        struct cds_wfcq_head head;
        struct cds_wfcq_tail tail;
 
+       /* Work belonging to worker. Cannot be stolen. */
+       struct urcu_work *own;
+
        struct urcu_wait_node wait_node;
        /* RCU linked list node of siblings for work stealing. */
        struct cds_list_head sibling_node;
@@ -133,6 +137,7 @@ void urcu_worker_init(struct urcu_worker *worker, int flags)
        cds_wfcq_init(&worker->head, &worker->tail);
        worker->flags = flags;
        urcu_wait_node_init(&worker->wait_node, URCU_WAIT_RUNNING);
+       worker->own = NULL;
 }
 
 static inline
@@ -192,27 +197,80 @@ void urcu_worker_unregister(struct urcu_workqueue *queue,
        synchronize_rcu();
 }
 
-/*
- * Try stealing work from siblings when we have nothing to do.
- */
 static inline
-bool ___urcu_steal_work(struct urcu_worker *worker,
-               struct urcu_worker *sibling)
+bool ___urcu_grab_work(struct urcu_worker *worker,
+               cds_wfcq_head_ptr_t src_head,
+               struct cds_wfcq_tail *src_tail,
+               bool steal)
 {
        enum cds_wfcq_ret splice_ret;
+       struct cds_wfcq_head tmp_head;
+       struct cds_wfcq_tail tmp_tail;
+       struct cds_wfcq_node *node;
 
        /*
-        * Don't bother grabbing the sibling queue lock if it is empty.
+        * Don't bother grabbing the src queue lock if it is empty.
         */
-       if (cds_wfcq_empty(&sibling->head, &sibling->tail))
+       if (cds_wfcq_empty(src_head, src_tail))
                return false;
+       cds_wfcq_init(&tmp_head, &tmp_tail);
+
+       /* Ensure that we preserve FIFO work order. */
+       assert(!steal || worker->own == NULL);
+
+       /* Splice to temporary queue. */
+       if (steal)
+               cds_wfcq_dequeue_lock(src_head.h, src_tail);
+       splice_ret = __cds_wfcq_splice_blocking(&tmp_head,
+                       &tmp_tail,
+                       src_head,
+                       src_tail);
+       if (steal)
+               cds_wfcq_dequeue_unlock(src_head.h, src_tail);
+       if (splice_ret == CDS_WFCQ_RET_SRC_EMPTY)
+               return false;
+
+       /*
+        * Keep one work entry for ourself. This ensures forward
+        * progress amongst stealing co-workers. This also ensures that
+        * when a worker grab some work from the global workqueue, it
+        * will have at least one work item to deal with.
+        */
+       if (worker->own == NULL) {
+               if (!steal) {
+                       /*
+                        * Try to grab own work from worker workqueue to
+                        * preserve FIFO order.
+                        */
+                       node = cds_wfcq_dequeue_blocking(&worker->head,
+                               &worker->tail);
+                       if (node)
+                               goto got_node;
+               }
+               node = __cds_wfcq_dequeue_blocking(&tmp_head, &tmp_tail);
+               assert(node != NULL);
+got_node:
+               worker->own = caa_container_of(node, struct urcu_work, node);
+       }
+
+       /* Splice into worker workqueue. */
        splice_ret = cds_wfcq_splice_blocking(&worker->head,
                        &worker->tail,
-                       &sibling->head,
-                       &sibling->tail);
+                       &tmp_head,
+                       &tmp_tail);
        /* Ensure that we preserve FIFO work order. */
-       assert(splice_ret != CDS_WFCQ_RET_DEST_NON_EMPTY);
-       return splice_ret != CDS_WFCQ_RET_SRC_EMPTY;
+       assert(!steal || splice_ret != CDS_WFCQ_RET_DEST_NON_EMPTY);
+       return true;
+}
+
+/*
+ * Try stealing work from siblings when we have nothing to do.
+ */
+static inline
+bool ___urcu_steal_work(struct urcu_worker *worker,
+               struct urcu_worker *sibling)
+{
+       return ___urcu_grab_work(worker, &sibling->head, &sibling->tail, 1);
 }
 
 static inline
@@ -302,15 +360,11 @@ enum urcu_accept_ret urcu_accept_work(struct urcu_workqueue *queue,
                struct urcu_worker *worker)
 {
        enum cds_wfcq_ret wfcq_ret;
+       bool has_work;
 
-       wfcq_ret = __cds_wfcq_splice_blocking(&worker->head,
-                       &worker->tail,
-                       &queue->head,
-                       &queue->tail);
+       has_work = ___urcu_grab_work(worker, &queue->head, &queue->tail, 0);
        /* Don't wait if we have work to do. */
-       if (wfcq_ret != CDS_WFCQ_RET_SRC_EMPTY
-                       || !cds_wfcq_empty(&worker->head,
-                               &worker->tail))
+       if (has_work || !cds_wfcq_empty(&worker->head, &worker->tail))
                goto do_work;
        /* Try to steal work from sibling instead of blocking */
        if (__urcu_steal_work(queue, worker))
@@ -375,6 +429,14 @@ struct urcu_work *urcu_dequeue_work(struct urcu_worker *worker)
 {
        struct cds_wfcq_node *node;
 
+       if (worker->own) {
+               struct urcu_work *work;
+
+               /* Process our own work entry. */
+               work = worker->own;
+               worker->own = NULL;
+               return work;
+       }
        /*
         * If we are registered for work stealing, we need to dequeue
         * safely against siblings.