1 // SPDX-FileCopyrightText: 2010-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
2 // SPDX-FileCopyrightText: 2011-2012 Lai Jiangshan <laijs@cn.fujitsu.com>
4 // SPDX-License-Identifier: LGPL-2.1-or-later
6 #ifndef _URCU_WFCQUEUE_STATIC_H
7 #define _URCU_WFCQUEUE_STATIC_H
10 * Userspace RCU library - Concurrent Queue with Wait-Free Enqueue/Blocking Dequeue
12 * TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See urcu/wfcqueue.h for
13 * linking dynamically with the userspace rcu library.
19 #include <urcu/assert.h>
20 #include <urcu/compiler.h>
21 #include <urcu/uatomic.h>
28 * Concurrent queue with wait-free enqueue/blocking dequeue.
30 * This queue has been designed and implemented collaboratively by
31 * Mathieu Desnoyers and Lai Jiangshan. Inspired from
32 * half-wait-free/half-blocking queue implementation done by Paul E.
35 * Mutual exclusion of cds_wfcq_* / __cds_wfcq_* API
37 * Synchronization table:
39 * External synchronization techniques described in the API below is
40 * required between pairs marked with "X". No external synchronization
41 * required between pairs marked with "-".
44 * [1] cds_wfcq_enqueue
45 * [2] __cds_wfcq_splice (destination queue)
46 * [3] __cds_wfcq_dequeue
47 * [4] __cds_wfcq_splice (source queue)
48 * [5] __cds_wfcq_first
51 * [1] [2] [3] [4] [5] [6]
59 * Mutual exclusion can be ensured by holding cds_wfcq_dequeue_lock().
61 * For convenience, cds_wfcq_dequeue_blocking() and
62 * cds_wfcq_splice_blocking() hold the dequeue lock.
64 * Besides locking, mutual exclusion of dequeue, splice and iteration
65 * can be ensured by performing all of those operations from a single
66 * thread, without requiring any lock.
69 #define WFCQ_ADAPT_ATTEMPTS 10 /* Retry if being set */
70 #define WFCQ_WAIT 10 /* Wait 10 ms if being set */
73 * cds_wfcq_node_init: initialize wait-free queue node.
75 static inline void _cds_wfcq_node_init(struct cds_wfcq_node
*node
)
80 static inline void _cds_wfcq_node_init_atomic(struct cds_wfcq_node
*node
)
82 uatomic_store(&node
->next
, NULL
, CMM_RELAXED
);
86 * cds_wfcq_init: initialize wait-free queue (with lock). Pair with
89 static inline void _cds_wfcq_init(struct cds_wfcq_head
*head
,
90 struct cds_wfcq_tail
*tail
)
94 /* Set queue head and tail */
95 _cds_wfcq_node_init(&head
->node
);
96 tail
->p
= &head
->node
;
97 ret
= pthread_mutex_init(&head
->lock
, NULL
);
98 urcu_posix_assert(!ret
);
102 * cds_wfcq_destroy: destroy wait-free queue (with lock). Pair with
105 static inline void _cds_wfcq_destroy(struct cds_wfcq_head
*head
,
106 struct cds_wfcq_tail
*tail
__attribute__((unused
)))
108 int ret
= pthread_mutex_destroy(&head
->lock
);
109 urcu_posix_assert(!ret
);
113 * __cds_wfcq_init: initialize wait-free queue (without lock). Don't
114 * pair with any destroy function.
116 static inline void ___cds_wfcq_init(struct __cds_wfcq_head
*head
,
117 struct cds_wfcq_tail
*tail
)
119 /* Set queue head and tail */
120 _cds_wfcq_node_init(&head
->node
);
121 tail
->p
= &head
->node
;
125 * cds_wfcq_empty: return whether wait-free queue is empty.
127 * No memory barrier is issued. No mutual exclusion is required.
129 * We perform the test on head->node.next to check if the queue is
130 * possibly empty, but we confirm this by checking if the tail pointer
131 * points to the head node because the tail pointer is the linearisation
132 * point of the enqueuers. Just checking the head next pointer could
133 * make a queue appear empty if an enqueuer is preempted for a long time
134 * between xchg() and setting the previous node's next pointer.
136 static inline bool _cds_wfcq_empty(cds_wfcq_head_ptr_t u_head
,
137 struct cds_wfcq_tail
*tail
)
139 struct __cds_wfcq_head
*head
= u_head
._h
;
141 * Queue is empty if no node is pointed by head->node.next nor
142 * tail->p. Even though the tail->p check is sufficient to find
143 * out of the queue is empty, we first check head->node.next as a
144 * common case to ensure that dequeuers do not frequently access
145 * enqueuer's tail->p cache line.
147 return uatomic_load(&head
->node
.next
, CMM_CONSUME
) == NULL
148 && uatomic_load(&tail
->p
, CMM_CONSUME
) == &head
->node
;
151 static inline void _cds_wfcq_dequeue_lock(struct cds_wfcq_head
*head
,
152 struct cds_wfcq_tail
*tail
__attribute__((unused
)))
156 ret
= pthread_mutex_lock(&head
->lock
);
157 urcu_posix_assert(!ret
);
160 static inline void _cds_wfcq_dequeue_unlock(struct cds_wfcq_head
*head
,
161 struct cds_wfcq_tail
*tail
__attribute__((unused
)))
165 ret
= pthread_mutex_unlock(&head
->lock
);
166 urcu_posix_assert(!ret
);
169 static inline bool ___cds_wfcq_append(cds_wfcq_head_ptr_t u_head
,
170 struct cds_wfcq_tail
*tail
,
171 struct cds_wfcq_node
*new_head
,
172 struct cds_wfcq_node
*new_tail
)
174 struct __cds_wfcq_head
*head
= u_head
._h
;
175 struct cds_wfcq_node
*old_tail
;
178 * Implicit memory barrier before uatomic_xchg() orders earlier
179 * stores to data structure containing node and setting
180 * node->next to NULL before publication.
182 old_tail
= uatomic_xchg_mo(&tail
->p
, new_tail
, CMM_SEQ_CST
);
185 * Implicit memory barrier after uatomic_xchg() orders store to
186 * q->tail before store to old_tail->next.
188 * At this point, dequeuers see a NULL tail->p->next, which
189 * indicates that the queue is being appended to. The following
190 * store will append "node" to the queue from a dequeuer
193 uatomic_store(&old_tail
->next
, new_head
, CMM_RELEASE
);
196 * Return false if queue was empty prior to adding the node,
199 return old_tail
!= &head
->node
;
203 * cds_wfcq_enqueue: enqueue a node into a wait-free queue.
205 * Operations prior to enqueue are consistant with respect to dequeuing or
206 * splicing and iterating.
208 * Returns false if the queue was empty prior to adding the node.
209 * Returns true otherwise.
211 static inline bool _cds_wfcq_enqueue(cds_wfcq_head_ptr_t head
,
212 struct cds_wfcq_tail
*tail
,
213 struct cds_wfcq_node
*new_tail
)
215 cmm_emit_legacy_smp_mb();
217 return ___cds_wfcq_append(head
, tail
, new_tail
, new_tail
);
221 * CDS_WFCQ_WAIT_SLEEP:
223 * By default, this sleeps for the given @msec milliseconds.
224 * This is a macro which LGPL users may #define themselves before
225 * including wfcqueue.h to override the default behavior (e.g.
226 * to log a warning or perform other background work).
228 #ifndef CDS_WFCQ_WAIT_SLEEP
229 #define CDS_WFCQ_WAIT_SLEEP(msec) ___cds_wfcq_wait_sleep(msec)
232 static inline void ___cds_wfcq_wait_sleep(int msec
)
234 (void) poll(NULL
, 0, msec
);
238 * ___cds_wfcq_busy_wait: adaptative busy-wait.
240 * Returns 1 if nonblocking and needs to block, 0 otherwise.
243 ___cds_wfcq_busy_wait(int *attempt
, int blocking
)
247 if (++(*attempt
) >= WFCQ_ADAPT_ATTEMPTS
) {
248 CDS_WFCQ_WAIT_SLEEP(WFCQ_WAIT
); /* Wait for 10ms */
257 * Waiting for enqueuer to complete enqueue and return the next node.
259 static inline struct cds_wfcq_node
*
260 ___cds_wfcq_node_sync_next(struct cds_wfcq_node
*node
, int blocking
)
262 struct cds_wfcq_node
*next
;
266 * Adaptative busy-looping waiting for enqueuer to complete enqueue.
268 * Load node.next before loading node's content
270 while ((next
= uatomic_load(&node
->next
, CMM_CONSUME
)) == NULL
) {
271 if (___cds_wfcq_busy_wait(&attempt
, blocking
))
272 return CDS_WFCQ_WOULDBLOCK
;
278 static inline struct cds_wfcq_node
*
279 ___cds_wfcq_first(cds_wfcq_head_ptr_t u_head
,
280 struct cds_wfcq_tail
*tail
,
283 struct __cds_wfcq_head
*head
= u_head
._h
;
284 struct cds_wfcq_node
*node
;
286 if (_cds_wfcq_empty(__cds_wfcq_head_cast(head
), tail
))
288 node
= ___cds_wfcq_node_sync_next(&head
->node
, blocking
);
294 * __cds_wfcq_first_blocking: get first node of a queue, without dequeuing.
296 * Content written into the node before enqueue is guaranteed to be
297 * consistent, but no other memory ordering is ensured.
298 * Dequeue/splice/iteration mutual exclusion should be ensured by the
301 * Used by for-like iteration macros in urcu/wfqueue.h:
302 * __cds_wfcq_for_each_blocking()
303 * __cds_wfcq_for_each_blocking_safe()
305 * Returns NULL if queue is empty, first node otherwise.
307 static inline struct cds_wfcq_node
*
308 ___cds_wfcq_first_blocking(cds_wfcq_head_ptr_t head
,
309 struct cds_wfcq_tail
*tail
)
311 return ___cds_wfcq_first(head
, tail
, 1);
316 * __cds_wfcq_first_nonblocking: get first node of a queue, without dequeuing.
318 * Same as __cds_wfcq_first_blocking, but returns CDS_WFCQ_WOULDBLOCK if
321 static inline struct cds_wfcq_node
*
322 ___cds_wfcq_first_nonblocking(cds_wfcq_head_ptr_t head
,
323 struct cds_wfcq_tail
*tail
)
325 return ___cds_wfcq_first(head
, tail
, 0);
328 static inline struct cds_wfcq_node
*
329 ___cds_wfcq_next(cds_wfcq_head_ptr_t head
__attribute__((unused
)),
330 struct cds_wfcq_tail
*tail
,
331 struct cds_wfcq_node
*node
,
334 struct cds_wfcq_node
*next
;
337 * Even though the following tail->p check is sufficient to find
338 * out if we reached the end of the queue, we first check
339 * node->next as a common case to ensure that iteration on nodes
340 * do not frequently access enqueuer's tail->p cache line.
342 * Load node->next before loading next's content
344 if ((next
= uatomic_load(&node
->next
, CMM_CONSUME
)) == NULL
) {
345 if (uatomic_load(&tail
->p
, CMM_RELAXED
) == node
)
347 next
= ___cds_wfcq_node_sync_next(node
, blocking
);
354 * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
356 * Content written into the node before enqueue is guaranteed to be
357 * consistent, but no other memory ordering is ensured.
358 * Dequeue/splice/iteration mutual exclusion should be ensured by the
361 * Used by for-like iteration macros in urcu/wfqueue.h:
362 * __cds_wfcq_for_each_blocking()
363 * __cds_wfcq_for_each_blocking_safe()
365 * Returns NULL if reached end of queue, non-NULL next queue node
368 static inline struct cds_wfcq_node
*
369 ___cds_wfcq_next_blocking(cds_wfcq_head_ptr_t head
,
370 struct cds_wfcq_tail
*tail
,
371 struct cds_wfcq_node
*node
)
373 return ___cds_wfcq_next(head
, tail
, node
, 1);
377 * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
379 * Same as __cds_wfcq_next_blocking, but returns CDS_WFCQ_WOULDBLOCK if
382 static inline struct cds_wfcq_node
*
383 ___cds_wfcq_next_nonblocking(cds_wfcq_head_ptr_t head
,
384 struct cds_wfcq_tail
*tail
,
385 struct cds_wfcq_node
*node
)
387 return ___cds_wfcq_next(head
, tail
, node
, 0);
390 static inline struct cds_wfcq_node
*
391 ___cds_wfcq_dequeue_with_state(cds_wfcq_head_ptr_t u_head
,
392 struct cds_wfcq_tail
*tail
,
396 struct __cds_wfcq_head
*head
= u_head
._h
;
397 struct cds_wfcq_node
*node
, *next
;
402 if (_cds_wfcq_empty(__cds_wfcq_head_cast(head
), tail
)) {
406 node
= ___cds_wfcq_node_sync_next(&head
->node
, blocking
);
407 if (!blocking
&& node
== CDS_WFCQ_WOULDBLOCK
) {
408 return CDS_WFCQ_WOULDBLOCK
;
411 if ((next
= uatomic_load(&node
->next
, CMM_CONSUME
)) == NULL
) {
413 * @node is probably the only node in the queue.
414 * Try to move the tail to &q->head.
415 * q->head.next is set to NULL here, and stays
416 * NULL if the cmpxchg succeeds. Should the
417 * cmpxchg fail due to a concurrent enqueue, the
418 * q->head.next will be set to the next node.
420 _cds_wfcq_node_init_atomic(&head
->node
);
421 if (uatomic_cmpxchg_mo(&tail
->p
, node
, &head
->node
,
422 CMM_SEQ_CST
, CMM_SEQ_CST
) == node
) {
424 *state
|= CDS_WFCQ_STATE_LAST
;
425 cmm_emit_legacy_smp_mb();
428 next
= ___cds_wfcq_node_sync_next(node
, blocking
);
430 * In nonblocking mode, if we would need to block to
431 * get node's next, set the head next node pointer
432 * (currently NULL) back to its original value.
434 if (!blocking
&& next
== CDS_WFCQ_WOULDBLOCK
) {
435 uatomic_store(&head
->node
.next
, node
, CMM_RELAXED
);
436 return CDS_WFCQ_WOULDBLOCK
;
441 * Move queue head forward.
443 uatomic_store(&head
->node
.next
, next
, CMM_RELAXED
);
444 cmm_emit_legacy_smp_mb();
450 * __cds_wfcq_dequeue_with_state_blocking: dequeue node from queue, with state.
452 * Content written into the node before enqueue is guaranteed to be
453 * consistent, but no other memory ordering is ensured.
454 * It is valid to reuse and free a dequeued node immediately.
455 * Dequeue/splice/iteration mutual exclusion should be ensured by the
458 static inline struct cds_wfcq_node
*
459 ___cds_wfcq_dequeue_with_state_blocking(cds_wfcq_head_ptr_t head
,
460 struct cds_wfcq_tail
*tail
, int *state
)
462 return ___cds_wfcq_dequeue_with_state(head
, tail
, state
, 1);
466 * ___cds_wfcq_dequeue_blocking: dequeue node from queue.
468 * Same as __cds_wfcq_dequeue_with_state_blocking, but without saving
471 static inline struct cds_wfcq_node
*
472 ___cds_wfcq_dequeue_blocking(cds_wfcq_head_ptr_t head
,
473 struct cds_wfcq_tail
*tail
)
475 return ___cds_wfcq_dequeue_with_state_blocking(head
, tail
, NULL
);
479 * __cds_wfcq_dequeue_with_state_nonblocking: dequeue node, with state.
481 * Same as __cds_wfcq_dequeue_blocking, but returns CDS_WFCQ_WOULDBLOCK
482 * if it needs to block.
484 static inline struct cds_wfcq_node
*
485 ___cds_wfcq_dequeue_with_state_nonblocking(cds_wfcq_head_ptr_t head
,
486 struct cds_wfcq_tail
*tail
, int *state
)
488 return ___cds_wfcq_dequeue_with_state(head
, tail
, state
, 0);
492 * ___cds_wfcq_dequeue_nonblocking: dequeue node from queue.
494 * Same as __cds_wfcq_dequeue_with_state_nonblocking, but without saving
497 static inline struct cds_wfcq_node
*
498 ___cds_wfcq_dequeue_nonblocking(cds_wfcq_head_ptr_t head
,
499 struct cds_wfcq_tail
*tail
)
501 return ___cds_wfcq_dequeue_with_state_nonblocking(head
, tail
, NULL
);
505 * __cds_wfcq_splice: enqueue all src_q nodes at the end of dest_q.
507 * Operations after splice are consistant with respect to enqueue.
509 * Dequeue all nodes from src_q.
510 * dest_q must be already initialized.
511 * Mutual exclusion for src_q should be ensured by the caller as
512 * specified in the "Synchronisation table".
513 * Returns enum cds_wfcq_ret which indicates the state of the src or
516 static inline enum cds_wfcq_ret
518 cds_wfcq_head_ptr_t u_dest_q_head
,
519 struct cds_wfcq_tail
*dest_q_tail
,
520 cds_wfcq_head_ptr_t u_src_q_head
,
521 struct cds_wfcq_tail
*src_q_tail
,
524 struct __cds_wfcq_head
*dest_q_head
= u_dest_q_head
._h
;
525 struct __cds_wfcq_head
*src_q_head
= u_src_q_head
._h
;
526 struct cds_wfcq_node
*head
, *tail
;
530 * Initial emptiness check to speed up cases where queue is
531 * empty: only require loads to check if queue is empty.
533 if (_cds_wfcq_empty(__cds_wfcq_head_cast(src_q_head
), src_q_tail
))
534 return CDS_WFCQ_RET_SRC_EMPTY
;
538 * Open-coded _cds_wfcq_empty() by testing result of
539 * uatomic_xchg, as well as tail pointer vs head node
542 head
= uatomic_xchg_mo(&src_q_head
->node
.next
, NULL
, CMM_SEQ_CST
);
544 break; /* non-empty */
545 if (uatomic_load(&src_q_tail
->p
, CMM_CONSUME
) == &src_q_head
->node
)
546 return CDS_WFCQ_RET_SRC_EMPTY
;
547 if (___cds_wfcq_busy_wait(&attempt
, blocking
))
548 return CDS_WFCQ_RET_WOULDBLOCK
;
552 * Memory barrier implied before uatomic_xchg() orders store to
553 * src_q->head before store to src_q->tail. This is required by
554 * concurrent enqueue on src_q, which exchanges the tail before
555 * updating the previous tail's next pointer.
557 cmm_emit_legacy_smp_mb();
558 tail
= uatomic_xchg_mo(&src_q_tail
->p
, &src_q_head
->node
, CMM_SEQ_CST
);
561 * Append the spliced content of src_q into dest_q. Does not
562 * require mutual exclusion on dest_q (wait-free).
564 if (___cds_wfcq_append(__cds_wfcq_head_cast(dest_q_head
), dest_q_tail
,
566 return CDS_WFCQ_RET_DEST_NON_EMPTY
;
568 return CDS_WFCQ_RET_DEST_EMPTY
;
572 * __cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
574 * Dequeue all nodes from src_q.
575 * dest_q must be already initialized.
576 * Mutual exclusion for src_q should be ensured by the caller as
577 * specified in the "Synchronisation table".
578 * Returns enum cds_wfcq_ret which indicates the state of the src or
579 * dest queue. Never returns CDS_WFCQ_RET_WOULDBLOCK.
581 static inline enum cds_wfcq_ret
582 ___cds_wfcq_splice_blocking(
583 cds_wfcq_head_ptr_t dest_q_head
,
584 struct cds_wfcq_tail
*dest_q_tail
,
585 cds_wfcq_head_ptr_t src_q_head
,
586 struct cds_wfcq_tail
*src_q_tail
)
588 return ___cds_wfcq_splice(dest_q_head
, dest_q_tail
,
589 src_q_head
, src_q_tail
, 1);
593 * __cds_wfcq_splice_nonblocking: enqueue all src_q nodes at the end of dest_q.
595 * Same as __cds_wfcq_splice_blocking, but returns
596 * CDS_WFCQ_RET_WOULDBLOCK if it needs to block.
598 static inline enum cds_wfcq_ret
599 ___cds_wfcq_splice_nonblocking(
600 cds_wfcq_head_ptr_t dest_q_head
,
601 struct cds_wfcq_tail
*dest_q_tail
,
602 cds_wfcq_head_ptr_t src_q_head
,
603 struct cds_wfcq_tail
*src_q_tail
)
605 return ___cds_wfcq_splice(dest_q_head
, dest_q_tail
,
606 src_q_head
, src_q_tail
, 0);
610 * cds_wfcq_dequeue_with_state_blocking: dequeue a node from a wait-free queue.
612 * Content written into the node before enqueue is guaranteed to be
613 * consistent, but no other memory ordering is ensured.
614 * Mutual exclusion with cds_wfcq_splice_blocking and dequeue lock is
616 * It is valid to reuse and free a dequeued node immediately.
618 static inline struct cds_wfcq_node
*
619 _cds_wfcq_dequeue_with_state_blocking(struct cds_wfcq_head
*head
,
620 struct cds_wfcq_tail
*tail
, int *state
)
622 struct cds_wfcq_node
*retval
;
624 _cds_wfcq_dequeue_lock(head
, tail
);
625 retval
= ___cds_wfcq_dequeue_with_state_blocking(cds_wfcq_head_cast(head
),
627 _cds_wfcq_dequeue_unlock(head
, tail
);
632 * cds_wfcq_dequeue_blocking: dequeue node from queue.
634 * Same as cds_wfcq_dequeue_blocking, but without saving state.
636 static inline struct cds_wfcq_node
*
637 _cds_wfcq_dequeue_blocking(struct cds_wfcq_head
*head
,
638 struct cds_wfcq_tail
*tail
)
640 return _cds_wfcq_dequeue_with_state_blocking(head
, tail
, NULL
);
644 * cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
646 * Dequeue all nodes from src_q.
647 * dest_q must be already initialized.
648 * Content written into the node before enqueue is guaranteed to be
649 * consistent, but no other memory ordering is ensured.
650 * Mutual exclusion with cds_wfcq_dequeue_blocking and dequeue lock is
652 * Returns enum cds_wfcq_ret which indicates the state of the src or
653 * dest queue. Never returns CDS_WFCQ_RET_WOULDBLOCK.
655 static inline enum cds_wfcq_ret
656 _cds_wfcq_splice_blocking(
657 struct cds_wfcq_head
*dest_q_head
,
658 struct cds_wfcq_tail
*dest_q_tail
,
659 struct cds_wfcq_head
*src_q_head
,
660 struct cds_wfcq_tail
*src_q_tail
)
662 enum cds_wfcq_ret ret
;
664 _cds_wfcq_dequeue_lock(src_q_head
, src_q_tail
);
665 ret
= ___cds_wfcq_splice_blocking(cds_wfcq_head_cast(dest_q_head
), dest_q_tail
,
666 cds_wfcq_head_cast(src_q_head
), src_q_tail
);
667 _cds_wfcq_dequeue_unlock(src_q_head
, src_q_tail
);
675 #endif /* _URCU_WFCQUEUE_STATIC_H */