workqueue/waitqueue: use lock-free stack for wakeup
[userspace-rcu.git] / urcu / workqueue-fifo.h
CommitLineData
13652c4b
MD
1#ifndef _URCU_WORKQUEUE_FIFO_H
2#define _URCU_WORKQUEUE_FIFO_H
3
4/*
5 * urcu/workqueue-fifo.h
6 *
7 * Userspace RCU library - work queue scheme with FIFO semantic
8 *
9 * Copyright (c) 2014 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
10 *
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.
15 *
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.
20 *
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
24 */
25
26#include <urcu/uatomic.h>
7a618cf7 27#include <urcu/lfstack.h>
13652c4b
MD
28#include <urcu/waitqueue-lifo.h>
29#include <urcu/wfcqueue.h>
30#include <urcu/rculist.h>
31#include <pthread.h>
e10c65b3 32#include <assert.h>
13652c4b
MD
33
34/*
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.
39 */
40
41struct urcu_work {
42 struct cds_wfcq_node node;
43};
44
45struct urcu_workqueue {
46 /* FIFO work queue */
47 struct __cds_wfcq_head head;
48 struct cds_wfcq_tail tail;
49
50 /* Associated wait queue for LIFO wait/wakeup */
51 struct urcu_wait_queue waitqueue;
52
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 */
56};
57
58struct urcu_worker {
59 struct cds_wfcq_head head;
60 struct cds_wfcq_tail tail;
61
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 */
66};
67
68enum urcu_worker_flags {
69 URCU_WORKER_STEAL = (1 << 0),
70};
71
72static inline
73void urcu_workqueue_init(struct urcu_workqueue *queue)
74{
75 __cds_wfcq_init(&queue->head, &queue->tail);
76 urcu_wait_queue_init(&queue->waitqueue);
77 CDS_INIT_LIST_HEAD(&queue->sibling_head);
78}
79
80static inline
81void urcu_queue_work(struct urcu_workqueue *queue, struct urcu_work *work)
82{
83 bool was_empty;
84
85 cds_wfcq_node_init(&work->node);
86
87 /* Enqueue work. */
88 was_empty = !cds_wfcq_enqueue(&queue->head, &queue->tail,
89 &work->node);
90 /*
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.
102 */
d3afe039
MD
103 if (was_empty) {
104 rcu_read_lock(); /* Protect stack dequeue */
13652c4b 105 (void) urcu_dequeue_wake_single(&queue->waitqueue);
d3afe039
MD
106 rcu_read_unlock(); /* Protect stack dequeue */
107 }
13652c4b
MD
108}
109
110static inline
111void urcu_workqueue_wakeup_all(struct urcu_workqueue *queue)
112{
113 struct urcu_waiters waiters;
114
d3afe039 115 rcu_read_lock(); /* Protect stack dequeue */
13652c4b 116 urcu_move_waiters(&waiters, &queue->waitqueue);
d3afe039
MD
117 rcu_read_unlock(); /* Protect stack dequeue */
118
13652c4b
MD
119 (void) urcu_wake_all_waiters(&waiters);
120}
121
122static inline
123void urcu_worker_init(struct urcu_worker *worker, int flags)
124{
125 cds_wfcq_init(&worker->head, &worker->tail);
126 worker->flags = flags;
127 urcu_wait_node_init(&worker->wait_node, URCU_WAIT_RUNNING);
128}
129
130static inline
131void urcu_worker_register(struct urcu_workqueue *queue,
132 struct urcu_worker *worker)
133{
134 if (worker->flags & URCU_WORKER_STEAL) {
135 pthread_mutex_lock(&queue->sibling_lock);
136 cds_list_add_rcu(&worker->sibling_node, &queue->sibling_head);
137 pthread_mutex_unlock(&queue->sibling_lock);
138 }
139}
140
141static inline
142void urcu_worker_unregister(struct urcu_workqueue *queue,
143 struct urcu_worker *worker)
144{
145 enum cds_wfcq_ret wfcq_ret;
146
147 if (worker->flags & URCU_WORKER_STEAL) {
148 pthread_mutex_lock(&queue->sibling_lock);
149 cds_list_del_rcu(&worker->sibling_node);
150 pthread_mutex_unlock(&queue->sibling_lock);
13652c4b
MD
151 }
152
d3afe039
MD
153 /*
154 * Wait for grace period before freeing or reusing
155 * "worker" because used by RCU linked list.
156 * Also prevents ABA for waitqueue stack dequeue: matches RCU
157 * read-side critical sections around dequeue and move all
158 * operations on waitqueue).
159 */
160 synchronize_rcu();
161
13652c4b
MD
162 /*
163 * Put any local work we still have back into the workqueue.
164 */
165 wfcq_ret = __cds_wfcq_splice_blocking(&queue->head,
166 &queue->tail,
167 &worker->head,
168 &worker->tail);
169 if (wfcq_ret != CDS_WFCQ_RET_SRC_EMPTY
170 && wfcq_ret == CDS_WFCQ_RET_DEST_EMPTY) {
171 /*
172 * Wakeup worker thread if we have put work back into
173 * workqueue that was previously empty.
174 */
d3afe039 175 rcu_read_lock(); /* Protect stack dequeue */
13652c4b 176 (void) urcu_dequeue_wake_single(&queue->waitqueue);
d3afe039 177 rcu_read_unlock(); /* Protect stack dequeue */
13652c4b
MD
178 }
179}
180
181/*
182 * Try stealing work from siblings when we have nothing to do.
183 */
184static inline
e10c65b3 185bool ___urcu_steal_work(struct urcu_worker *worker,
13652c4b
MD
186 struct urcu_worker *sibling)
187{
e10c65b3
MD
188 enum cds_wfcq_ret splice_ret;
189
30926570
MD
190 /*
191 * Don't bother grabbing the sibling queue lock if it is empty.
192 */
193 if (cds_wfcq_empty(&sibling->head, &sibling->tail))
e10c65b3 194 return false;
13652c4b 195 cds_wfcq_dequeue_lock(&sibling->head, &sibling->tail);
e10c65b3 196 splice_ret = __cds_wfcq_splice_blocking(&worker->head,
13652c4b
MD
197 &worker->tail,
198 &sibling->head,
199 &sibling->tail);
200 cds_wfcq_dequeue_unlock(&sibling->head, &sibling->tail);
e10c65b3
MD
201 /* Ensure that we preserve FIFO work order. */
202 assert(splice_ret != CDS_WFCQ_RET_DEST_NON_EMPTY);
203 return splice_ret != CDS_WFCQ_RET_SRC_EMPTY;
13652c4b
MD
204}
205
206static inline
e10c65b3 207bool __urcu_steal_work(struct urcu_workqueue *queue,
13652c4b
MD
208 struct urcu_worker *worker)
209{
210 struct urcu_worker *sibling_prev, *sibling_next;
211 struct cds_list_head *sibling_node;
e10c65b3 212 bool steal_performed = 0;
13652c4b
MD
213
214 if (!(worker->flags & URCU_WORKER_STEAL))
e10c65b3 215 return false;
13652c4b
MD
216
217 rcu_read_lock();
218
219 sibling_node = rcu_dereference(worker->sibling_node.next);
220 if (sibling_node == &queue->sibling_head)
221 sibling_node = rcu_dereference(sibling_node->next);
222 sibling_next = caa_container_of(sibling_node, struct urcu_worker,
223 sibling_node);
224 if (sibling_next != worker)
e10c65b3
MD
225 steal_performed = ___urcu_steal_work(worker, sibling_next);
226 if (steal_performed)
227 goto end;
13652c4b
MD
228
229 sibling_node = rcu_dereference(worker->sibling_node.prev);
230 if (sibling_node == &queue->sibling_head)
231 sibling_node = rcu_dereference(sibling_node->prev);
232 sibling_prev = caa_container_of(sibling_node, struct urcu_worker,
233 sibling_node);
234 if (sibling_prev != worker && sibling_prev != sibling_next)
e10c65b3
MD
235 steal_performed = ___urcu_steal_work(worker, sibling_prev);
236end:
13652c4b
MD
237 rcu_read_unlock();
238
e10c65b3 239 return steal_performed;
13652c4b
MD
240}
241
242static inline
5d30bf32 243bool ___urcu_wakeup_sibling(struct urcu_worker *sibling)
13652c4b 244{
5d30bf32 245 return urcu_adaptative_wake_up(&sibling->wait_node);
13652c4b
MD
246}
247
248static inline
5d30bf32 249bool __urcu_wakeup_siblings(struct urcu_workqueue *queue,
13652c4b
MD
250 struct urcu_worker *worker)
251{
252 struct urcu_worker *sibling_prev, *sibling_next;
253 struct cds_list_head *sibling_node;
5d30bf32 254 bool wakeup_performed = 0;
13652c4b
MD
255
256 if (!(worker->flags & URCU_WORKER_STEAL))
257 return;
258
259 /* Only wakeup siblings if we have work in our own queue. */
260 if (cds_wfcq_empty(&worker->head, &worker->tail))
261 return;
262
263 rcu_read_lock();
264
265 sibling_node = rcu_dereference(worker->sibling_node.next);
266 if (sibling_node == &queue->sibling_head)
267 sibling_node = rcu_dereference(sibling_node->next);
268 sibling_next = caa_container_of(sibling_node, struct urcu_worker,
269 sibling_node);
270 if (sibling_next != worker)
5d30bf32
MD
271 wakeup_performed = ___urcu_wakeup_sibling(sibling_next);
272 if (wakeup_performed)
273 goto end;
13652c4b
MD
274
275 sibling_node = rcu_dereference(worker->sibling_node.prev);
276 if (sibling_node == &queue->sibling_head)
277 sibling_node = rcu_dereference(sibling_node->prev);
278 sibling_prev = caa_container_of(sibling_node, struct urcu_worker,
279 sibling_node);
280 if (sibling_prev != worker && sibling_prev != sibling_next)
5d30bf32
MD
281 wakeup_performed = ___urcu_wakeup_sibling(sibling_prev);
282end:
13652c4b 283 rcu_read_unlock();
5d30bf32
MD
284
285 return wakeup_performed;
13652c4b
MD
286}
287
288static inline
289void urcu_accept_work(struct urcu_workqueue *queue,
290 struct urcu_worker *worker,
291 int blocking)
292{
293 enum cds_wfcq_ret wfcq_ret;
294
295 wfcq_ret = __cds_wfcq_splice_blocking(&worker->head,
296 &worker->tail,
297 &queue->head,
298 &queue->tail);
299 /* Don't wait if we have work to do. */
300 if (wfcq_ret != CDS_WFCQ_RET_SRC_EMPTY
301 || !cds_wfcq_empty(&worker->head,
302 &worker->tail))
303 goto do_work;
304 /* Try to steal work from sibling instead of blocking */
305 if (__urcu_steal_work(queue, worker))
306 goto do_work;
307 if (!blocking)
308 return;
309 urcu_wait_set_state(&worker->wait_node,
310 URCU_WAIT_WAITING);
311 if (!CMM_LOAD_SHARED(worker->wait_node.node.next)) {
312 int was_empty;
313
314 /*
315 * NULL next pointer. We are therefore not in
316 * the queue.
317 */
7a618cf7 318 cds_lfs_node_init(&worker->wait_node.node);
d3afe039
MD
319 /* Protect stack dequeue against ABA */
320 synchronize_rcu();
13652c4b
MD
321 was_empty = !urcu_wait_add(&queue->waitqueue,
322 &worker->wait_node);
323 /*
324 * If the wait queue was empty, it means we are the
325 * first thread to be put back into an otherwise empty
326 * wait queue. Re-check if work queue is empty after
327 * adding ourself to wait queue, so we can wakeup the
328 * top of wait queue since new work have appeared, and
329 * work enqueuer may not have seen that it needed to do
330 * a wake up.
331 */
332 if (was_empty && !cds_wfcq_empty(&queue->head,
d3afe039
MD
333 &queue->tail)) {
334 rcu_read_lock(); /* Protect stack dequeue */
13652c4b 335 (void) urcu_dequeue_wake_single(&queue->waitqueue);
d3afe039
MD
336 rcu_read_unlock(); /* Protect stack dequeue */
337 }
13652c4b
MD
338 } else {
339 /*
340 * Non-NULL next pointer. We are therefore in
341 * the queue, or the dispatcher just removed us
342 * from it (after we read the next pointer), and
343 * is therefore awakening us. The state will
344 * therefore have been changed from WAITING to
345 * some other state, which will let the busy
346 * wait pass through.
347 */
348 }
349 urcu_adaptative_busy_wait(&worker->wait_node);
350 return;
351
352do_work:
353 /*
354 * We will be busy handling the work batch, awaken siblings so
355 * they can steal from us.
356 */
5d30bf32 357 (void) __urcu_wakeup_siblings(queue, worker);
13652c4b
MD
358}
359
360static inline
361struct urcu_work *urcu_dequeue_work(struct urcu_worker *worker)
362{
363 struct cds_wfcq_node *node;
364
365 /*
366 * If we are registered for work stealing, we need to dequeue
367 * safely against siblings.
368 */
30926570
MD
369 if (worker->flags & URCU_WORKER_STEAL) {
370 /*
371 * Don't bother grabbing the worker queue lock if it is
372 * empty.
373 */
374 if (cds_wfcq_empty(&worker->head, &worker->tail))
375 return NULL;
13652c4b
MD
376 node = cds_wfcq_dequeue_blocking(&worker->head,
377 &worker->tail);
30926570 378 } else {
13652c4b
MD
379 node = ___cds_wfcq_dequeue_with_state(&worker->head,
380 &worker->tail, NULL, 1, 0);
30926570 381 }
13652c4b
MD
382 if (!node)
383 return NULL;
384 return caa_container_of(node, struct urcu_work, node);
385}
386
387#endif /* _URCU_WORKQUEUE_FIFO_H */
This page took 0.038153 seconds and 4 git commands to generate.