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/wfstack.h>
28 #include <urcu/waitqueue-lifo.h>
29 #include <urcu/wfcqueue.h>
30 #include <urcu/rculist.h>
35 * We use RCU to steal work from siblings. Therefore, one of RCU flavors
36 * need to be included before this header. All worker that participate
37 * in stealing (initialized with the URCU_WORKER_STEAL flag) need to be
38 * registered RCU readers threads.
42 struct cds_wfcq_node node
;
45 struct urcu_workqueue
{
47 struct __cds_wfcq_head head
;
48 struct cds_wfcq_tail tail
;
50 /* Associated wait queue for LIFO wait/wakeup */
51 struct urcu_wait_queue waitqueue
;
53 /* RCU linked list head of siblings for work stealing. */
54 struct cds_list_head sibling_head
;
55 pthread_mutex_t sibling_lock
; /* Protect sibling list updates */
59 struct cds_wfcq_head head
;
60 struct cds_wfcq_tail tail
;
62 struct urcu_wait_node wait_node
;
63 /* RCU linked list node of siblings for work stealing. */
64 struct cds_list_head sibling_node
;
65 int flags
; /* enum urcu_worker_flags */
68 enum urcu_worker_flags
{
69 URCU_WORKER_STEAL
= (1 << 0),
73 void urcu_workqueue_init(struct urcu_workqueue
*queue
)
75 __cds_wfcq_init(&queue
->head
, &queue
->tail
);
76 urcu_wait_queue_init(&queue
->waitqueue
);
77 CDS_INIT_LIST_HEAD(&queue
->sibling_head
);
81 void urcu_queue_work(struct urcu_workqueue
*queue
, struct urcu_work
*work
)
85 cds_wfcq_node_init(&work
->node
);
88 was_empty
= !cds_wfcq_enqueue(&queue
->head
, &queue
->tail
,
91 * If workqueue was previously empty, wakeup one worker thread.
92 * It will eventually grab the entire content of the work-queue
93 * (therefore grabbing a "work batch"). After having grabbed the
94 * work batch, while that thread is running and taking care of
95 * that work batch, when we enqueue more work, we will wake
96 * another thread (if there is one waiting), which will
97 * eventually grab the new batch, and so on. This scheme ensures
98 * that contiguous batch of work are handled by the same thread
99 * (for locality), and also ensures that we scale work to many
100 * worker threads when threads are busy enough to still be
101 * running when work is enqueued.
104 (void) urcu_dequeue_wake_single(&queue
->waitqueue
);
108 void urcu_workqueue_wakeup_all(struct urcu_workqueue
*queue
)
110 struct urcu_waiters waiters
;
112 urcu_move_waiters(&waiters
, &queue
->waitqueue
);
113 (void) urcu_wake_all_waiters(&waiters
);
117 void urcu_worker_init(struct urcu_worker
*worker
, int flags
)
119 cds_wfcq_init(&worker
->head
, &worker
->tail
);
120 worker
->flags
= flags
;
121 urcu_wait_node_init(&worker
->wait_node
, URCU_WAIT_RUNNING
);
125 void urcu_worker_register(struct urcu_workqueue
*queue
,
126 struct urcu_worker
*worker
)
128 if (worker
->flags
& URCU_WORKER_STEAL
) {
129 pthread_mutex_lock(&queue
->sibling_lock
);
130 cds_list_add_rcu(&worker
->sibling_node
, &queue
->sibling_head
);
131 pthread_mutex_unlock(&queue
->sibling_lock
);
136 void urcu_worker_unregister(struct urcu_workqueue
*queue
,
137 struct urcu_worker
*worker
)
139 enum cds_wfcq_ret wfcq_ret
;
141 if (worker
->flags
& URCU_WORKER_STEAL
) {
142 pthread_mutex_lock(&queue
->sibling_lock
);
143 cds_list_del_rcu(&worker
->sibling_node
);
144 pthread_mutex_unlock(&queue
->sibling_lock
);
147 * Wait for grace period before freeing or reusing
148 * "worker" because used by RCU linked list.
154 * Put any local work we still have back into the workqueue.
156 wfcq_ret
= __cds_wfcq_splice_blocking(&queue
->head
,
160 if (wfcq_ret
!= CDS_WFCQ_RET_SRC_EMPTY
161 && wfcq_ret
== CDS_WFCQ_RET_DEST_EMPTY
) {
163 * Wakeup worker thread if we have put work back into
164 * workqueue that was previously empty.
166 (void) urcu_dequeue_wake_single(&queue
->waitqueue
);
171 * Try stealing work from siblings when we have nothing to do.
174 bool ___urcu_steal_work(struct urcu_worker
*worker
,
175 struct urcu_worker
*sibling
)
177 enum cds_wfcq_ret splice_ret
;
180 * Don't bother grabbing the sibling queue lock if it is empty.
182 if (cds_wfcq_empty(&sibling
->head
, &sibling
->tail
))
184 cds_wfcq_dequeue_lock(&sibling
->head
, &sibling
->tail
);
185 splice_ret
= __cds_wfcq_splice_blocking(&worker
->head
,
189 cds_wfcq_dequeue_unlock(&sibling
->head
, &sibling
->tail
);
190 /* Ensure that we preserve FIFO work order. */
191 assert(splice_ret
!= CDS_WFCQ_RET_DEST_NON_EMPTY
);
192 return splice_ret
!= CDS_WFCQ_RET_SRC_EMPTY
;
196 bool __urcu_steal_work(struct urcu_workqueue
*queue
,
197 struct urcu_worker
*worker
)
199 struct urcu_worker
*sibling_prev
, *sibling_next
;
200 struct cds_list_head
*sibling_node
;
201 bool steal_performed
= 0;
203 if (!(worker
->flags
& URCU_WORKER_STEAL
))
208 sibling_node
= rcu_dereference(worker
->sibling_node
.next
);
209 if (sibling_node
== &queue
->sibling_head
)
210 sibling_node
= rcu_dereference(sibling_node
->next
);
211 sibling_next
= caa_container_of(sibling_node
, struct urcu_worker
,
213 if (sibling_next
!= worker
)
214 steal_performed
= ___urcu_steal_work(worker
, sibling_next
);
218 sibling_node
= rcu_dereference(worker
->sibling_node
.prev
);
219 if (sibling_node
== &queue
->sibling_head
)
220 sibling_node
= rcu_dereference(sibling_node
->prev
);
221 sibling_prev
= caa_container_of(sibling_node
, struct urcu_worker
,
223 if (sibling_prev
!= worker
&& sibling_prev
!= sibling_next
)
224 steal_performed
= ___urcu_steal_work(worker
, sibling_prev
);
228 return steal_performed
;
232 bool ___urcu_wakeup_sibling(struct urcu_worker
*sibling
)
234 return urcu_adaptative_wake_up(&sibling
->wait_node
);
238 bool __urcu_wakeup_siblings(struct urcu_workqueue
*queue
,
239 struct urcu_worker
*worker
)
241 struct urcu_worker
*sibling_prev
, *sibling_next
;
242 struct cds_list_head
*sibling_node
;
243 bool wakeup_performed
= 0;
245 if (!(worker
->flags
& URCU_WORKER_STEAL
))
248 /* Only wakeup siblings if we have work in our own queue. */
249 if (cds_wfcq_empty(&worker
->head
, &worker
->tail
))
254 sibling_node
= rcu_dereference(worker
->sibling_node
.next
);
255 if (sibling_node
== &queue
->sibling_head
)
256 sibling_node
= rcu_dereference(sibling_node
->next
);
257 sibling_next
= caa_container_of(sibling_node
, struct urcu_worker
,
259 if (sibling_next
!= worker
)
260 wakeup_performed
= ___urcu_wakeup_sibling(sibling_next
);
261 if (wakeup_performed
)
264 sibling_node
= rcu_dereference(worker
->sibling_node
.prev
);
265 if (sibling_node
== &queue
->sibling_head
)
266 sibling_node
= rcu_dereference(sibling_node
->prev
);
267 sibling_prev
= caa_container_of(sibling_node
, struct urcu_worker
,
269 if (sibling_prev
!= worker
&& sibling_prev
!= sibling_next
)
270 wakeup_performed
= ___urcu_wakeup_sibling(sibling_prev
);
274 return wakeup_performed
;
278 void urcu_accept_work(struct urcu_workqueue
*queue
,
279 struct urcu_worker
*worker
,
282 enum cds_wfcq_ret wfcq_ret
;
284 wfcq_ret
= __cds_wfcq_splice_blocking(&worker
->head
,
288 /* Don't wait if we have work to do. */
289 if (wfcq_ret
!= CDS_WFCQ_RET_SRC_EMPTY
290 || !cds_wfcq_empty(&worker
->head
,
293 /* Try to steal work from sibling instead of blocking */
294 if (__urcu_steal_work(queue
, worker
))
298 urcu_wait_set_state(&worker
->wait_node
,
300 if (!CMM_LOAD_SHARED(worker
->wait_node
.node
.next
)) {
304 * NULL next pointer. We are therefore not in
307 cds_wfs_node_init(&worker
->wait_node
.node
);
308 was_empty
= !urcu_wait_add(&queue
->waitqueue
,
311 * If the wait queue was empty, it means we are the
312 * first thread to be put back into an otherwise empty
313 * wait queue. Re-check if work queue is empty after
314 * adding ourself to wait queue, so we can wakeup the
315 * top of wait queue since new work have appeared, and
316 * work enqueuer may not have seen that it needed to do
319 if (was_empty
&& !cds_wfcq_empty(&queue
->head
,
321 (void) urcu_dequeue_wake_single(&queue
->waitqueue
);
324 * Non-NULL next pointer. We are therefore in
325 * the queue, or the dispatcher just removed us
326 * from it (after we read the next pointer), and
327 * is therefore awakening us. The state will
328 * therefore have been changed from WAITING to
329 * some other state, which will let the busy
333 urcu_adaptative_busy_wait(&worker
->wait_node
);
338 * We will be busy handling the work batch, awaken siblings so
339 * they can steal from us.
341 (void) __urcu_wakeup_siblings(queue
, worker
);
345 struct urcu_work
*urcu_dequeue_work(struct urcu_worker
*worker
)
347 struct cds_wfcq_node
*node
;
350 * If we are registered for work stealing, we need to dequeue
351 * safely against siblings.
353 if (worker
->flags
& URCU_WORKER_STEAL
) {
355 * Don't bother grabbing the worker queue lock if it is
358 if (cds_wfcq_empty(&worker
->head
, &worker
->tail
))
360 node
= cds_wfcq_dequeue_blocking(&worker
->head
,
363 node
= ___cds_wfcq_dequeue_with_state(&worker
->head
,
364 &worker
->tail
, NULL
, 1, 0);
368 return caa_container_of(node
, struct urcu_work
, node
);
371 #endif /* _URCU_WORKQUEUE_FIFO_H */