1 #ifndef _URCU_WORKQUEUE_FIFO_H
2 #define _URCU_WORKQUEUE_FIFO_H
5 * urcu/workqueue-fifo.h
7 * Userspace RCU library - work queue scheme with FIFO semantic
9 * Copyright (c) 2014 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
11 * This library is free software; you can redistribute it and/or
12 * modify it under the terms of the GNU Lesser General Public
13 * License as published by the Free Software Foundation; either
14 * version 2.1 of the License, or (at your option) any later version.
16 * This library is distributed in the hope that it will be useful,
17 * but WITHOUT ANY WARRANTY; without even the implied warranty of
18 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
19 * Lesser General Public License for more details.
21 * You should have received a copy of the GNU Lesser General Public
22 * License along with this library; if not, write to the Free Software
23 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
26 #include <urcu/uatomic.h>
27 #include <urcu/lfstack.h>
28 #include <urcu/waitqueue-lifo.h>
29 #include <urcu/wfcqueue.h>
30 #include <urcu/rculist.h>
34 enum urcu_accept_ret
{
36 URCU_ACCEPT_SHUTDOWN
= 1,
39 enum urcu_enqueue_ret
{
41 URCU_ENQUEUE_FULL
= 1,
45 * We use RCU to steal work from siblings. Therefore, one of RCU flavors
46 * need to be included before this header. All worker that participate
47 * in stealing (initialized with the URCU_WORKER_STEAL flag) need to be
48 * registered RCU readers threads.
52 struct cds_wfcq_node node
;
55 struct urcu_workqueue
{
57 struct __cds_wfcq_head head
;
58 struct cds_wfcq_tail tail
;
60 /* Associated wait queue for LIFO wait/wakeup */
61 struct urcu_wait_queue waitqueue
;
63 /* RCU linked list head of siblings for work stealing. */
64 struct cds_list_head sibling_head
;
65 pthread_mutex_t sibling_lock
; /* Protect sibling list updates */
67 /* Maximum number of work entries (approximate). 0 means infinite. */
68 unsigned long nr_work_max
;
69 unsigned long nr_work
; /* Current number of work items */
71 int worker_flags
; /* Worker flags */
72 int shutdown
; /* Shutdown performed */
76 /* Workqueue which can be either used by worker, or stolen. */
77 struct cds_wfcq_head head
;
78 struct cds_wfcq_tail tail
;
80 /* Work belonging to worker. Cannot be stolen. */
81 struct urcu_work
*own
;
83 struct urcu_wait_node wait_node
;
84 /* RCU linked list node of siblings for work stealing. */
85 struct cds_list_head sibling_node
;
86 struct urcu_workqueue
*queue
;
87 int flags
; /* enum urcu_worker_flags */
90 enum urcu_worker_flags
{
91 URCU_WORKER_STEAL
= (1 << 0),
95 void urcu_workqueue_init(struct urcu_workqueue
*queue
,
96 unsigned long max_queue_len
,
99 __cds_wfcq_init(&queue
->head
, &queue
->tail
);
100 urcu_wait_queue_init(&queue
->waitqueue
);
101 CDS_INIT_LIST_HEAD(&queue
->sibling_head
);
102 pthread_mutex_init(&queue
->sibling_lock
, NULL
);
103 queue
->nr_work_max
= max_queue_len
;
105 queue
->shutdown
= false;
109 enum urcu_enqueue_ret
urcu_queue_work(struct urcu_workqueue
*queue
,
110 struct urcu_work
*work
)
113 unsigned long nr_work_max
;
115 nr_work_max
= queue
->nr_work_max
;
117 /* Approximate max queue size. */
118 if (uatomic_read(&queue
->nr_work
) >= nr_work_max
)
119 return URCU_ENQUEUE_FULL
;
120 uatomic_inc(&queue
->nr_work
);
122 cds_wfcq_node_init(&work
->node
);
125 was_empty
= !cds_wfcq_enqueue(&queue
->head
, &queue
->tail
,
128 * If workqueue was previously empty, wakeup one worker thread.
129 * It will eventually grab the entire content of the work-queue
130 * (therefore grabbing a "work batch"). After having grabbed the
131 * work batch, while that thread is running and taking care of
132 * that work batch, when we enqueue more work, we will wake
133 * another thread (if there is one waiting), which will
134 * eventually grab the new batch, and so on. This scheme ensures
135 * that contiguous batch of work are handled by the same thread
136 * (for locality), and also ensures that we scale work to many
137 * worker threads when threads are busy enough to still be
138 * running when work is enqueued.
141 rcu_read_lock(); /* Protect stack dequeue */
142 (void) urcu_dequeue_wake_single(&queue
->waitqueue
);
143 rcu_read_unlock(); /* Protect stack dequeue */
145 return URCU_ENQUEUE_OK
;
149 void __urcu_workqueue_wakeup_all(struct urcu_workqueue
*queue
)
151 struct urcu_waiters waiters
;
153 rcu_read_lock(); /* Protect stack dequeue */
154 urcu_move_waiters(&waiters
, &queue
->waitqueue
);
155 rcu_read_unlock(); /* Protect stack dequeue */
157 (void) urcu_wake_all_waiters(&waiters
);
161 void urcu_worker_init(struct urcu_workqueue
*queue
,
162 struct urcu_worker
*worker
)
164 cds_wfcq_init(&worker
->head
, &worker
->tail
);
165 urcu_wait_node_init(&worker
->wait_node
, URCU_WAIT_RUNNING
);
167 worker
->wait_node
.node
.next
= NULL
;
168 worker
->queue
= queue
;
169 worker
->flags
= queue
->worker_flags
;
173 void urcu_worker_register(struct urcu_workqueue
*queue
,
174 struct urcu_worker
*worker
)
176 if (worker
->flags
& URCU_WORKER_STEAL
) {
177 pthread_mutex_lock(&queue
->sibling_lock
);
178 cds_list_add_rcu(&worker
->sibling_node
, &queue
->sibling_head
);
179 pthread_mutex_unlock(&queue
->sibling_lock
);
184 void urcu_worker_unregister(struct urcu_workqueue
*queue
,
185 struct urcu_worker
*worker
)
187 enum cds_wfcq_ret wfcq_ret
;
189 if (worker
->flags
& URCU_WORKER_STEAL
) {
190 pthread_mutex_lock(&queue
->sibling_lock
);
191 cds_list_del_rcu(&worker
->sibling_node
);
192 pthread_mutex_unlock(&queue
->sibling_lock
);
196 * Make sure we are removed from waitqueue.
198 if (CMM_LOAD_SHARED(worker
->wait_node
.node
.next
))
199 __urcu_workqueue_wakeup_all(queue
);
202 * Put any local work we still have back into the workqueue.
204 wfcq_ret
= __cds_wfcq_splice_blocking(&queue
->head
,
208 if (wfcq_ret
!= CDS_WFCQ_RET_SRC_EMPTY
209 && wfcq_ret
== CDS_WFCQ_RET_DEST_EMPTY
) {
211 * Wakeup worker thread if we have put work back into
212 * workqueue that was previously empty.
214 rcu_read_lock(); /* Protect stack dequeue */
215 (void) urcu_dequeue_wake_single(&queue
->waitqueue
);
216 rcu_read_unlock(); /* Protect stack dequeue */
220 * Wait for grace period before freeing or reusing
221 * "worker" because used by RCU linked list.
222 * Also prevents ABA for waitqueue stack dequeue: matches RCU
223 * read-side critical sections around dequeue and move all
224 * operations on waitqueue).
230 bool ___urcu_grab_work(struct urcu_worker
*worker
,
231 cds_wfcq_head_ptr_t src_head
,
232 struct cds_wfcq_tail
*src_tail
,
235 enum cds_wfcq_ret splice_ret
;
236 struct __cds_wfcq_head tmp_head
;
237 struct cds_wfcq_tail tmp_tail
;
238 struct cds_wfcq_node
*node
;
241 * Don't bother grabbing the src queue lock if it is empty.
243 if (cds_wfcq_empty(src_head
, src_tail
))
245 __cds_wfcq_init(&tmp_head
, &tmp_tail
);
247 /* Ensure that we preserve FIFO work order. */
248 assert(!steal
|| worker
->own
== NULL
);
250 /* Splice to temporary queue. */
252 cds_wfcq_dequeue_lock(src_head
.h
, src_tail
);
253 splice_ret
= __cds_wfcq_splice_blocking(&tmp_head
,
258 cds_wfcq_dequeue_unlock(src_head
.h
, src_tail
);
259 if (splice_ret
== CDS_WFCQ_RET_SRC_EMPTY
)
263 * Keep one work entry for ourself. This ensures forward
264 * progress amongst stealing co-workers. This also ensures that
265 * when a worker grab some work from the global workqueue, it
266 * will have at least one work item to deal with.
268 if (worker
->own
== NULL
) {
271 * Try to grab own work from worker workqueue to
272 * preserve FIFO order.
274 node
= cds_wfcq_dequeue_blocking(&worker
->head
,
279 node
= __cds_wfcq_dequeue_blocking(&tmp_head
, &tmp_tail
);
280 assert(node
!= NULL
);
282 worker
->own
= caa_container_of(node
, struct urcu_work
, node
);
285 /* Splice into worker workqueue. */
286 splice_ret
= __cds_wfcq_splice_blocking(&worker
->head
,
290 /* Ensure that we preserve FIFO work order. */
291 assert(!steal
|| splice_ret
!= CDS_WFCQ_RET_DEST_NON_EMPTY
);
296 * Try stealing work from siblings when we have nothing to do.
299 bool ___urcu_steal_work(struct urcu_worker
*worker
,
300 struct urcu_worker
*sibling
)
302 return ___urcu_grab_work(worker
, &sibling
->head
, &sibling
->tail
, 1);
306 bool __urcu_steal_work(struct urcu_workqueue
*queue
,
307 struct urcu_worker
*worker
)
309 struct urcu_worker
*sibling_prev
, *sibling_next
;
310 struct cds_list_head
*sibling_node
;
311 bool steal_performed
= 0;
313 if (!(worker
->flags
& URCU_WORKER_STEAL
))
318 sibling_node
= rcu_dereference(worker
->sibling_node
.next
);
319 if (sibling_node
== &queue
->sibling_head
)
320 sibling_node
= rcu_dereference(sibling_node
->next
);
321 sibling_next
= caa_container_of(sibling_node
, struct urcu_worker
,
323 if (sibling_next
!= worker
)
324 steal_performed
= ___urcu_steal_work(worker
, sibling_next
);
328 sibling_node
= rcu_dereference(worker
->sibling_node
.prev
);
329 if (sibling_node
== &queue
->sibling_head
)
330 sibling_node
= rcu_dereference(sibling_node
->prev
);
331 sibling_prev
= caa_container_of(sibling_node
, struct urcu_worker
,
333 if (sibling_prev
!= worker
&& sibling_prev
!= sibling_next
)
334 steal_performed
= ___urcu_steal_work(worker
, sibling_prev
);
338 return steal_performed
;
342 bool ___urcu_wakeup_sibling(struct urcu_worker
*sibling
)
344 return urcu_adaptative_wake_up(&sibling
->wait_node
);
348 bool __urcu_wakeup_siblings(struct urcu_workqueue
*queue
,
349 struct urcu_worker
*worker
)
351 struct urcu_worker
*sibling_prev
, *sibling_next
;
352 struct cds_list_head
*sibling_node
;
353 bool wakeup_performed
= 0;
355 if (!(worker
->flags
& URCU_WORKER_STEAL
))
358 /* Only wakeup siblings if we have work in our own queue. */
359 if (cds_wfcq_empty(&worker
->head
, &worker
->tail
))
364 sibling_node
= rcu_dereference(worker
->sibling_node
.next
);
365 if (sibling_node
== &queue
->sibling_head
)
366 sibling_node
= rcu_dereference(sibling_node
->next
);
367 sibling_next
= caa_container_of(sibling_node
, struct urcu_worker
,
369 if (sibling_next
!= worker
)
370 wakeup_performed
= ___urcu_wakeup_sibling(sibling_next
);
371 if (wakeup_performed
)
374 sibling_node
= rcu_dereference(worker
->sibling_node
.prev
);
375 if (sibling_node
== &queue
->sibling_head
)
376 sibling_node
= rcu_dereference(sibling_node
->prev
);
377 sibling_prev
= caa_container_of(sibling_node
, struct urcu_worker
,
379 if (sibling_prev
!= worker
&& sibling_prev
!= sibling_next
)
380 wakeup_performed
= ___urcu_wakeup_sibling(sibling_prev
);
384 return wakeup_performed
;
388 enum urcu_accept_ret
urcu_accept_work(struct urcu_worker
*worker
)
390 struct urcu_workqueue
*queue
= worker
->queue
;
391 enum cds_wfcq_ret wfcq_ret
;
394 has_work
= ___urcu_grab_work(worker
, &queue
->head
, &queue
->tail
, 0);
395 /* Don't wait if we have work to do. */
396 if (has_work
|| worker
->own
397 || !cds_wfcq_empty(&worker
->head
, &worker
->tail
))
399 /* Try to steal work from sibling instead of blocking */
400 if (__urcu_steal_work(queue
, worker
))
402 /* No more work to do, check shutdown state */
403 if (CMM_LOAD_SHARED(queue
->shutdown
))
404 return URCU_ACCEPT_SHUTDOWN
;
405 urcu_wait_set_state(&worker
->wait_node
,
407 if (!CMM_LOAD_SHARED(worker
->wait_node
.node
.next
)) {
411 * NULL next pointer. We are therefore not in
414 cds_lfs_node_init(&worker
->wait_node
.node
);
415 /* Protect stack dequeue against ABA */
417 was_empty
= !urcu_wait_add(&queue
->waitqueue
,
420 * If the wait queue was empty, it means we are the
421 * first thread to be put back into an otherwise empty
422 * wait queue. Re-check if work queue is empty after
423 * adding ourself to wait queue, so we can wakeup the
424 * top of wait queue since new work have appeared, and
425 * work enqueuer may not have seen that it needed to do
428 if (was_empty
&& !cds_wfcq_empty(&queue
->head
,
430 rcu_read_lock(); /* Protect stack dequeue */
431 (void) urcu_dequeue_wake_single(&queue
->waitqueue
);
432 rcu_read_unlock(); /* Protect stack dequeue */
436 * Non-NULL next pointer. We are therefore in
437 * the queue, or the dispatcher just removed us
438 * from it (after we read the next pointer), and
439 * is therefore awakening us. The state will
440 * therefore have been changed from WAITING to
441 * some other state, which will let the busy
445 urcu_adaptative_busy_wait(&worker
->wait_node
);
450 * We will be busy handling the work batch, awaken siblings so
451 * they can steal from us.
453 (void) __urcu_wakeup_siblings(queue
, worker
);
454 return URCU_ACCEPT_WORK
;
458 struct urcu_work
*urcu_dequeue_work(struct urcu_worker
*worker
)
460 struct urcu_workqueue
*queue
= worker
->queue
;
461 struct cds_wfcq_node
*node
;
462 struct urcu_work
*work
;
465 /* Process our own work entry. */
471 * If we are registered for work stealing, we need to dequeue
472 * safely against siblings.
474 if (worker
->flags
& URCU_WORKER_STEAL
) {
476 * Don't bother grabbing the worker queue lock if it is
479 if (cds_wfcq_empty(&worker
->head
, &worker
->tail
))
481 node
= cds_wfcq_dequeue_blocking(&worker
->head
,
484 node
= ___cds_wfcq_dequeue_with_state(&worker
->head
,
485 &worker
->tail
, NULL
, 1, 0);
489 work
= caa_container_of(node
, struct urcu_work
, node
);
491 if (queue
->nr_work_max
)
492 uatomic_dec(&queue
->nr_work
);
497 void urcu_workqueue_shutdown(struct urcu_workqueue
*queue
)
500 CMM_STORE_SHARED(queue
->shutdown
, true);
501 /* Wakeup all workers */
502 __urcu_workqueue_wakeup_all(queue
);
506 * Use to let dispatcher steal work from the entire queue in case of
507 * stall. The "worker" parameter need to be intialized, but is usually
511 bool urcu_workqueue_steal_all(struct urcu_workqueue
*queue
,
512 struct urcu_worker
*worker
)
514 struct urcu_worker
*sibling
;
515 bool has_work
= false;
517 if (worker
->flags
& URCU_WORKER_STEAL
) {
519 /* Steal from each worker */
520 cds_list_for_each_entry_rcu(sibling
, &queue
->sibling_head
,
522 has_work
|= ___urcu_grab_work(worker
, &sibling
->head
,
527 /* Steal from global workqueue */
528 has_work
|= ___urcu_grab_work(worker
, &queue
->head
, &queue
->tail
, 0);
532 #endif /* _URCU_WORKQUEUE_FIFO_H */