Fix: wfcqueue nonblocking dequeue
[urcu.git] / urcu / static / wfcqueue.h
1 #ifndef _URCU_WFCQUEUE_STATIC_H
2 #define _URCU_WFCQUEUE_STATIC_H
3
4 /*
5 * urcu/static/wfcqueue.h
6 *
7 * Userspace RCU library - Concurrent Queue with Wait-Free Enqueue/Blocking Dequeue
8 *
9 * TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See urcu/wfcqueue.h for
10 * linking dynamically with the userspace rcu library.
11 *
12 * Copyright 2010-2012 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
13 * Copyright 2011-2012 - Lai Jiangshan <laijs@cn.fujitsu.com>
14 *
15 * This library is free software; you can redistribute it and/or
16 * modify it under the terms of the GNU Lesser General Public
17 * License as published by the Free Software Foundation; either
18 * version 2.1 of the License, or (at your option) any later version.
19 *
20 * This library is distributed in the hope that it will be useful,
21 * but WITHOUT ANY WARRANTY; without even the implied warranty of
22 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
23 * Lesser General Public License for more details.
24 *
25 * You should have received a copy of the GNU Lesser General Public
26 * License along with this library; if not, write to the Free Software
27 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
28 */
29
30 #include <pthread.h>
31 #include <assert.h>
32 #include <poll.h>
33 #include <stdbool.h>
34 #include <urcu/compiler.h>
35 #include <urcu/uatomic.h>
36
37 #ifdef __cplusplus
38 extern "C" {
39 #endif
40
41 /*
42 * Concurrent queue with wait-free enqueue/blocking dequeue.
43 *
44 * This queue has been designed and implemented collaboratively by
45 * Mathieu Desnoyers and Lai Jiangshan. Inspired from
46 * half-wait-free/half-blocking queue implementation done by Paul E.
47 * McKenney.
48 *
49 * Mutual exclusion of __cds_wfcq_* API
50 *
51 * Unless otherwise stated, the caller must ensure mutual exclusion of
52 * queue update operations "dequeue" and "splice" (for source queue).
53 * Queue read operations "first" and "next", which are used by
54 * "for_each" iterations, need to be protected against concurrent
55 * "dequeue" and "splice" (for source queue) by the caller.
56 * "enqueue", "splice" (for destination queue), and "empty" are the only
57 * operations that can be used without any mutual exclusion.
58 * Mutual exclusion can be ensured by holding cds_wfcq_dequeue_lock().
59 *
60 * For convenience, cds_wfcq_dequeue_blocking() and
61 * cds_wfcq_splice_blocking() hold the dequeue lock.
62 *
63 * Besides locking, mutual exclusion of dequeue, splice and iteration
64 * can be ensured by performing all of those operations from a single
65 * thread, without requiring any lock.
66 */
67
68 #define WFCQ_ADAPT_ATTEMPTS 10 /* Retry if being set */
69 #define WFCQ_WAIT 10 /* Wait 10 ms if being set */
70
71 /*
72 * cds_wfcq_node_init: initialize wait-free queue node.
73 */
74 static inline void _cds_wfcq_node_init(struct cds_wfcq_node *node)
75 {
76 node->next = NULL;
77 }
78
79 /*
80 * cds_wfcq_init: initialize wait-free queue.
81 */
82 static inline void _cds_wfcq_init(struct cds_wfcq_head *head,
83 struct cds_wfcq_tail *tail)
84 {
85 int ret;
86
87 /* Set queue head and tail */
88 _cds_wfcq_node_init(&head->node);
89 tail->p = &head->node;
90 ret = pthread_mutex_init(&head->lock, NULL);
91 assert(!ret);
92 }
93
94 /*
95 * cds_wfcq_empty: return whether wait-free queue is empty.
96 *
97 * No memory barrier is issued. No mutual exclusion is required.
98 */
99 static inline bool _cds_wfcq_empty(struct cds_wfcq_head *head,
100 struct cds_wfcq_tail *tail)
101 {
102 /*
103 * Queue is empty if no node is pointed by head->node.next nor
104 * tail->p. Even though the tail->p check is sufficient to find
105 * out of the queue is empty, we first check head->node.next as a
106 * common case to ensure that dequeuers do not frequently access
107 * enqueuer's tail->p cache line.
108 */
109 return CMM_LOAD_SHARED(head->node.next) == NULL
110 && CMM_LOAD_SHARED(tail->p) == &head->node;
111 }
112
113 static inline void _cds_wfcq_dequeue_lock(struct cds_wfcq_head *head,
114 struct cds_wfcq_tail *tail)
115 {
116 int ret;
117
118 ret = pthread_mutex_lock(&head->lock);
119 assert(!ret);
120 }
121
122 static inline void _cds_wfcq_dequeue_unlock(struct cds_wfcq_head *head,
123 struct cds_wfcq_tail *tail)
124 {
125 int ret;
126
127 ret = pthread_mutex_unlock(&head->lock);
128 assert(!ret);
129 }
130
131 static inline void ___cds_wfcq_append(struct cds_wfcq_head *head,
132 struct cds_wfcq_tail *tail,
133 struct cds_wfcq_node *new_head,
134 struct cds_wfcq_node *new_tail)
135 {
136 struct cds_wfcq_node *old_tail;
137
138 /*
139 * Implicit memory barrier before uatomic_xchg() orders earlier
140 * stores to data structure containing node and setting
141 * node->next to NULL before publication.
142 */
143 old_tail = uatomic_xchg(&tail->p, new_tail);
144
145 /*
146 * Implicit memory barrier after uatomic_xchg() orders store to
147 * q->tail before store to old_tail->next.
148 *
149 * At this point, dequeuers see a NULL tail->p->next, which
150 * indicates that the queue is being appended to. The following
151 * store will append "node" to the queue from a dequeuer
152 * perspective.
153 */
154 CMM_STORE_SHARED(old_tail->next, new_head);
155 }
156
157 /*
158 * cds_wfcq_enqueue: enqueue a node into a wait-free queue.
159 *
160 * Issues a full memory barrier before enqueue. No mutual exclusion is
161 * required.
162 */
163 static inline void _cds_wfcq_enqueue(struct cds_wfcq_head *head,
164 struct cds_wfcq_tail *tail,
165 struct cds_wfcq_node *new_tail)
166 {
167 ___cds_wfcq_append(head, tail, new_tail, new_tail);
168 }
169
170 /*
171 * Waiting for enqueuer to complete enqueue and return the next node.
172 */
173 static inline struct cds_wfcq_node *
174 ___cds_wfcq_node_sync_next(struct cds_wfcq_node *node, int blocking)
175 {
176 struct cds_wfcq_node *next;
177 int attempt = 0;
178
179 /*
180 * Adaptative busy-looping waiting for enqueuer to complete enqueue.
181 */
182 while ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
183 if (!blocking)
184 return CDS_WFCQ_WOULDBLOCK;
185 if (++attempt >= WFCQ_ADAPT_ATTEMPTS) {
186 poll(NULL, 0, WFCQ_WAIT); /* Wait for 10ms */
187 attempt = 0;
188 } else {
189 caa_cpu_relax();
190 }
191 }
192
193 return next;
194 }
195
196 static inline struct cds_wfcq_node *
197 ___cds_wfcq_first(struct cds_wfcq_head *head,
198 struct cds_wfcq_tail *tail,
199 int blocking)
200 {
201 struct cds_wfcq_node *node;
202
203 if (_cds_wfcq_empty(head, tail))
204 return NULL;
205 node = ___cds_wfcq_node_sync_next(&head->node, blocking);
206 /* Load head->node.next before loading node's content */
207 cmm_smp_read_barrier_depends();
208 return node;
209 }
210
211 /*
212 * __cds_wfcq_first_blocking: get first node of a queue, without dequeuing.
213 *
214 * Content written into the node before enqueue is guaranteed to be
215 * consistent, but no other memory ordering is ensured.
216 * Dequeue/splice/iteration mutual exclusion should be ensured by the
217 * caller.
218 *
219 * Used by for-like iteration macros in urcu/wfqueue.h:
220 * __cds_wfcq_for_each_blocking()
221 * __cds_wfcq_for_each_blocking_safe()
222 */
223 static inline struct cds_wfcq_node *
224 ___cds_wfcq_first_blocking(struct cds_wfcq_head *head,
225 struct cds_wfcq_tail *tail)
226 {
227 return ___cds_wfcq_first(head, tail, 1);
228 }
229
230
231 /*
232 * __cds_wfcq_first_nonblocking: get first node of a queue, without dequeuing.
233 *
234 * Same as __cds_wfcq_first_blocking, but returns CDS_WFCQ_WOULDBLOCK if
235 * it needs to block.
236 */
237 static inline struct cds_wfcq_node *
238 ___cds_wfcq_first_nonblocking(struct cds_wfcq_head *head,
239 struct cds_wfcq_tail *tail)
240 {
241 return ___cds_wfcq_first(head, tail, 0);
242 }
243
244 static inline struct cds_wfcq_node *
245 ___cds_wfcq_next(struct cds_wfcq_head *head,
246 struct cds_wfcq_tail *tail,
247 struct cds_wfcq_node *node,
248 int blocking)
249 {
250 struct cds_wfcq_node *next;
251
252 /*
253 * Even though the following tail->p check is sufficient to find
254 * out if we reached the end of the queue, we first check
255 * node->next as a common case to ensure that iteration on nodes
256 * do not frequently access enqueuer's tail->p cache line.
257 */
258 if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
259 /* Load node->next before tail->p */
260 cmm_smp_rmb();
261 if (CMM_LOAD_SHARED(tail->p) == node)
262 return NULL;
263 next = ___cds_wfcq_node_sync_next(node, blocking);
264 }
265 /* Load node->next before loading next's content */
266 cmm_smp_read_barrier_depends();
267 return next;
268 }
269
270 /*
271 * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
272 *
273 * Content written into the node before enqueue is guaranteed to be
274 * consistent, but no other memory ordering is ensured.
275 * Dequeue/splice/iteration mutual exclusion should be ensured by the
276 * caller.
277 *
278 * Used by for-like iteration macros in urcu/wfqueue.h:
279 * __cds_wfcq_for_each_blocking()
280 * __cds_wfcq_for_each_blocking_safe()
281 */
282 static inline struct cds_wfcq_node *
283 ___cds_wfcq_next_blocking(struct cds_wfcq_head *head,
284 struct cds_wfcq_tail *tail,
285 struct cds_wfcq_node *node)
286 {
287 return ___cds_wfcq_next(head, tail, node, 1);
288 }
289
290 /*
291 * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
292 *
293 * Same as __cds_wfcq_next_blocking, but returns CDS_WFCQ_WOULDBLOCK if
294 * it needs to block.
295 */
296 static inline struct cds_wfcq_node *
297 ___cds_wfcq_next_nonblocking(struct cds_wfcq_head *head,
298 struct cds_wfcq_tail *tail,
299 struct cds_wfcq_node *node)
300 {
301 return ___cds_wfcq_next(head, tail, node, 0);
302 }
303
304 static inline struct cds_wfcq_node *
305 ___cds_wfcq_dequeue(struct cds_wfcq_head *head,
306 struct cds_wfcq_tail *tail,
307 int blocking)
308 {
309 struct cds_wfcq_node *node, *next;
310
311 if (_cds_wfcq_empty(head, tail))
312 return NULL;
313
314 node = ___cds_wfcq_node_sync_next(&head->node, blocking);
315 if (!blocking && node == CDS_WFCQ_WOULDBLOCK)
316 return CDS_WFCQ_WOULDBLOCK;
317
318 if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
319 /*
320 * @node is probably the only node in the queue.
321 * Try to move the tail to &q->head.
322 * q->head.next is set to NULL here, and stays
323 * NULL if the cmpxchg succeeds. Should the
324 * cmpxchg fail due to a concurrent enqueue, the
325 * q->head.next will be set to the next node.
326 * The implicit memory barrier before
327 * uatomic_cmpxchg() orders load node->next
328 * before loading q->tail.
329 * The implicit memory barrier before uatomic_cmpxchg
330 * orders load q->head.next before loading node's
331 * content.
332 */
333 _cds_wfcq_node_init(&head->node);
334 if (uatomic_cmpxchg(&tail->p, node, &head->node) == node)
335 return node;
336 next = ___cds_wfcq_node_sync_next(node, blocking);
337 /*
338 * In nonblocking mode, if we would need to block to
339 * get node's next, set the head next node pointer
340 * (currently NULL) back to its original value.
341 */
342 if (!blocking && next == CDS_WFCQ_WOULDBLOCK) {
343 head->node.next = node;
344 return CDS_WFCQ_WOULDBLOCK;
345 }
346 }
347
348 /*
349 * Move queue head forward.
350 */
351 head->node.next = next;
352
353 /* Load q->head.next before loading node's content */
354 cmm_smp_read_barrier_depends();
355 return node;
356 }
357
358 /*
359 * __cds_wfcq_dequeue_blocking: dequeue a node from the queue.
360 *
361 * Content written into the node before enqueue is guaranteed to be
362 * consistent, but no other memory ordering is ensured.
363 * It is valid to reuse and free a dequeued node immediately.
364 * Dequeue/splice/iteration mutual exclusion should be ensured by the
365 * caller.
366 */
367 static inline struct cds_wfcq_node *
368 ___cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
369 struct cds_wfcq_tail *tail)
370 {
371 return ___cds_wfcq_dequeue(head, tail, 1);
372 }
373
374 /*
375 * __cds_wfcq_dequeue_nonblocking: dequeue a node from a wait-free queue.
376 *
377 * Same as __cds_wfcq_dequeue_blocking, but returns CDS_WFCQ_WOULDBLOCK
378 * if it needs to block.
379 */
380 static inline struct cds_wfcq_node *
381 ___cds_wfcq_dequeue_nonblocking(struct cds_wfcq_head *head,
382 struct cds_wfcq_tail *tail)
383 {
384 return ___cds_wfcq_dequeue(head, tail, 0);
385 }
386
387 static inline int
388 ___cds_wfcq_splice(
389 struct cds_wfcq_head *dest_q_head,
390 struct cds_wfcq_tail *dest_q_tail,
391 struct cds_wfcq_head *src_q_head,
392 struct cds_wfcq_tail *src_q_tail,
393 int blocking)
394 {
395 struct cds_wfcq_node *head, *tail;
396
397 if (_cds_wfcq_empty(src_q_head, src_q_tail))
398 return 0;
399
400 head = ___cds_wfcq_node_sync_next(&src_q_head->node, blocking);
401 if (head == CDS_WFCQ_WOULDBLOCK)
402 return -1;
403 _cds_wfcq_node_init(&src_q_head->node);
404
405 /*
406 * Memory barrier implied before uatomic_xchg() orders store to
407 * src_q->head before store to src_q->tail. This is required by
408 * concurrent enqueue on src_q, which exchanges the tail before
409 * updating the previous tail's next pointer.
410 */
411 tail = uatomic_xchg(&src_q_tail->p, &src_q_head->node);
412
413 /*
414 * Append the spliced content of src_q into dest_q. Does not
415 * require mutual exclusion on dest_q (wait-free).
416 */
417 ___cds_wfcq_append(dest_q_head, dest_q_tail, head, tail);
418 return 0;
419 }
420
421
422 /*
423 * __cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
424 *
425 * Dequeue all nodes from src_q.
426 * dest_q must be already initialized.
427 * Dequeue/splice/iteration mutual exclusion for src_q should be ensured
428 * by the caller.
429 */
430 static inline void
431 ___cds_wfcq_splice_blocking(
432 struct cds_wfcq_head *dest_q_head,
433 struct cds_wfcq_tail *dest_q_tail,
434 struct cds_wfcq_head *src_q_head,
435 struct cds_wfcq_tail *src_q_tail)
436 {
437 (void) ___cds_wfcq_splice(dest_q_head, dest_q_tail,
438 src_q_head, src_q_tail, 1);
439 }
440
441 /*
442 * __cds_wfcq_splice_nonblocking: enqueue all src_q nodes at the end of dest_q.
443 *
444 * Same as __cds_wfcq_splice_blocking, but returns nonzero if it needs to
445 * block.
446 */
447 static inline int
448 ___cds_wfcq_splice_nonblocking(
449 struct cds_wfcq_head *dest_q_head,
450 struct cds_wfcq_tail *dest_q_tail,
451 struct cds_wfcq_head *src_q_head,
452 struct cds_wfcq_tail *src_q_tail)
453 {
454 return ___cds_wfcq_splice(dest_q_head, dest_q_tail,
455 src_q_head, src_q_tail, 0);
456 }
457
458 /*
459 * cds_wfcq_dequeue_blocking: dequeue a node from a wait-free queue.
460 *
461 * Content written into the node before enqueue is guaranteed to be
462 * consistent, but no other memory ordering is ensured.
463 * Mutual exlusion with cds_wfcq_splice_blocking and dequeue lock is
464 * ensured.
465 * It is valid to reuse and free a dequeued node immediately.
466 */
467 static inline struct cds_wfcq_node *
468 _cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
469 struct cds_wfcq_tail *tail)
470 {
471 struct cds_wfcq_node *retval;
472
473 _cds_wfcq_dequeue_lock(head, tail);
474 retval = ___cds_wfcq_dequeue_blocking(head, tail);
475 _cds_wfcq_dequeue_unlock(head, tail);
476 return retval;
477 }
478
479 /*
480 * cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
481 *
482 * Dequeue all nodes from src_q.
483 * dest_q must be already initialized.
484 * Content written into the node before enqueue is guaranteed to be
485 * consistent, but no other memory ordering is ensured.
486 * Mutual exlusion with cds_wfcq_dequeue_blocking and dequeue lock is
487 * ensured.
488 */
489 static inline void
490 _cds_wfcq_splice_blocking(
491 struct cds_wfcq_head *dest_q_head,
492 struct cds_wfcq_tail *dest_q_tail,
493 struct cds_wfcq_head *src_q_head,
494 struct cds_wfcq_tail *src_q_tail)
495 {
496 _cds_wfcq_dequeue_lock(src_q_head, src_q_tail);
497 ___cds_wfcq_splice_blocking(dest_q_head, dest_q_tail,
498 src_q_head, src_q_tail);
499 _cds_wfcq_dequeue_unlock(src_q_head, src_q_tail);
500 }
501
502 #ifdef __cplusplus
503 }
504 #endif
505
506 #endif /* _URCU_WFCQUEUE_STATIC_H */
This page took 0.03931 seconds and 5 git commands to generate.