X-Git-Url: https://git.liburcu.org/?p=urcu.git;a=blobdiff_plain;f=urcu%2Fstatic%2Fwfcqueue.h;h=4ababbf0ee8e99974f585c1f3b236381a8f60b3c;hp=1200227f4d3f6ade1f118b47ec77d592b3a508b5;hb=4d0d7cbccec1c35f544f6fee24b762e398a2f0b9;hpb=23773356a9fd12bf12627df437d0c7bd20e8ef01 diff --git a/urcu/static/wfcqueue.h b/urcu/static/wfcqueue.h index 1200227..4ababbf 100644 --- a/urcu/static/wfcqueue.h +++ b/urcu/static/wfcqueue.h @@ -46,15 +46,30 @@ extern "C" { * half-wait-free/half-blocking queue implementation done by Paul E. * McKenney. * - * Mutual exclusion of __cds_wfcq_* API - * - * Unless otherwise stated, the caller must ensure mutual exclusion of - * queue update operations "dequeue" and "splice" (for source queue). - * Queue read operations "first" and "next", which are used by - * "for_each" iterations, need to be protected against concurrent - * "dequeue" and "splice" (for source queue) by the caller. - * "enqueue", "splice" (for destination queue), and "empty" are the only - * operations that can be used without any mutual exclusion. + * Mutual exclusion of cds_wfcq_* / __cds_wfcq_* API + * + * Synchronization table: + * + * External synchronization techniques described in the API below is + * required between pairs marked with "X". No external synchronization + * required between pairs marked with "-". + * + * Legend: + * [1] cds_wfcq_enqueue + * [2] __cds_wfcq_splice (destination queue) + * [3] __cds_wfcq_dequeue + * [4] __cds_wfcq_splice (source queue) + * [5] __cds_wfcq_first + * [6] __cds_wfcq_next + * + * [1] [2] [3] [4] [5] [6] + * [1] - - - - - - + * [2] - - - - - - + * [3] - - X X X X + * [4] - - X - X X + * [5] - - X X - - + * [6] - - X X - - + * * Mutual exclusion can be ensured by holding cds_wfcq_dequeue_lock(). * * For convenience, cds_wfcq_dequeue_blocking() and @@ -91,14 +106,33 @@ static inline void _cds_wfcq_init(struct cds_wfcq_head *head, assert(!ret); } +/* + * __cds_wfcq_init: initialize wait-free queue. + */ +static inline void ___cds_wfcq_init(struct __cds_wfcq_head *head, + struct cds_wfcq_tail *tail) +{ + /* Set queue head and tail */ + _cds_wfcq_node_init(&head->node); + tail->p = &head->node; +} + /* * cds_wfcq_empty: return whether wait-free queue is empty. * * No memory barrier is issued. No mutual exclusion is required. + * + * We perform the test on head->node.next to check if the queue is + * possibly empty, but we confirm this by checking if the tail pointer + * points to the head node because the tail pointer is the linearisation + * point of the enqueuers. Just checking the head next pointer could + * make a queue appear empty if an enqueuer is preempted for a long time + * between xchg() and setting the previous node's next pointer. */ -static inline bool _cds_wfcq_empty(struct cds_wfcq_head *head, +static inline bool _cds_wfcq_empty(cds_wfcq_head_ptr_t u_head, struct cds_wfcq_tail *tail) { + struct __cds_wfcq_head *head = u_head._h; /* * Queue is empty if no node is pointed by head->node.next nor * tail->p. Even though the tail->p check is sufficient to find @@ -128,11 +162,12 @@ static inline void _cds_wfcq_dequeue_unlock(struct cds_wfcq_head *head, assert(!ret); } -static inline bool ___cds_wfcq_append(struct cds_wfcq_head *head, +static inline bool ___cds_wfcq_append(cds_wfcq_head_ptr_t u_head, struct cds_wfcq_tail *tail, struct cds_wfcq_node *new_head, struct cds_wfcq_node *new_tail) { + struct __cds_wfcq_head *head = u_head._h; struct cds_wfcq_node *old_tail; /* @@ -168,13 +203,32 @@ static inline bool ___cds_wfcq_append(struct cds_wfcq_head *head, * Returns false if the queue was empty prior to adding the node. * Returns true otherwise. */ -static inline bool _cds_wfcq_enqueue(struct cds_wfcq_head *head, +static inline bool _cds_wfcq_enqueue(cds_wfcq_head_ptr_t head, struct cds_wfcq_tail *tail, struct cds_wfcq_node *new_tail) { return ___cds_wfcq_append(head, tail, new_tail, new_tail); } +/* + * ___cds_wfcq_busy_wait: adaptative busy-wait. + * + * Returns 1 if nonblocking and needs to block, 0 otherwise. + */ +static inline bool +___cds_wfcq_busy_wait(int *attempt, int blocking) +{ + if (!blocking) + return 1; + if (++(*attempt) >= WFCQ_ADAPT_ATTEMPTS) { + (void) poll(NULL, 0, WFCQ_WAIT); /* Wait for 10ms */ + *attempt = 0; + } else { + caa_cpu_relax(); + } + return 0; +} + /* * Waiting for enqueuer to complete enqueue and return the next node. */ @@ -188,27 +242,22 @@ ___cds_wfcq_node_sync_next(struct cds_wfcq_node *node, int blocking) * Adaptative busy-looping waiting for enqueuer to complete enqueue. */ while ((next = CMM_LOAD_SHARED(node->next)) == NULL) { - if (!blocking) + if (___cds_wfcq_busy_wait(&attempt, blocking)) return CDS_WFCQ_WOULDBLOCK; - if (++attempt >= WFCQ_ADAPT_ATTEMPTS) { - poll(NULL, 0, WFCQ_WAIT); /* Wait for 10ms */ - attempt = 0; - } else { - caa_cpu_relax(); - } } return next; } static inline struct cds_wfcq_node * -___cds_wfcq_first(struct cds_wfcq_head *head, +___cds_wfcq_first(cds_wfcq_head_ptr_t u_head, struct cds_wfcq_tail *tail, int blocking) { + struct __cds_wfcq_head *head = u_head._h; struct cds_wfcq_node *node; - if (_cds_wfcq_empty(head, tail)) + if (_cds_wfcq_empty(__cds_wfcq_head_cast(head), tail)) return NULL; node = ___cds_wfcq_node_sync_next(&head->node, blocking); /* Load head->node.next before loading node's content */ @@ -227,9 +276,11 @@ ___cds_wfcq_first(struct cds_wfcq_head *head, * Used by for-like iteration macros in urcu/wfqueue.h: * __cds_wfcq_for_each_blocking() * __cds_wfcq_for_each_blocking_safe() + * + * Returns NULL if queue is empty, first node otherwise. */ static inline struct cds_wfcq_node * -___cds_wfcq_first_blocking(struct cds_wfcq_head *head, +___cds_wfcq_first_blocking(cds_wfcq_head_ptr_t head, struct cds_wfcq_tail *tail) { return ___cds_wfcq_first(head, tail, 1); @@ -243,14 +294,14 @@ ___cds_wfcq_first_blocking(struct cds_wfcq_head *head, * it needs to block. */ static inline struct cds_wfcq_node * -___cds_wfcq_first_nonblocking(struct cds_wfcq_head *head, +___cds_wfcq_first_nonblocking(cds_wfcq_head_ptr_t head, struct cds_wfcq_tail *tail) { return ___cds_wfcq_first(head, tail, 0); } static inline struct cds_wfcq_node * -___cds_wfcq_next(struct cds_wfcq_head *head, +___cds_wfcq_next(cds_wfcq_head_ptr_t head, struct cds_wfcq_tail *tail, struct cds_wfcq_node *node, int blocking) @@ -286,9 +337,12 @@ ___cds_wfcq_next(struct cds_wfcq_head *head, * Used by for-like iteration macros in urcu/wfqueue.h: * __cds_wfcq_for_each_blocking() * __cds_wfcq_for_each_blocking_safe() + * + * Returns NULL if reached end of queue, non-NULL next queue node + * otherwise. */ static inline struct cds_wfcq_node * -___cds_wfcq_next_blocking(struct cds_wfcq_head *head, +___cds_wfcq_next_blocking(cds_wfcq_head_ptr_t head, struct cds_wfcq_tail *tail, struct cds_wfcq_node *node) { @@ -302,7 +356,7 @@ ___cds_wfcq_next_blocking(struct cds_wfcq_head *head, * it needs to block. */ static inline struct cds_wfcq_node * -___cds_wfcq_next_nonblocking(struct cds_wfcq_head *head, +___cds_wfcq_next_nonblocking(cds_wfcq_head_ptr_t head, struct cds_wfcq_tail *tail, struct cds_wfcq_node *node) { @@ -310,18 +364,25 @@ ___cds_wfcq_next_nonblocking(struct cds_wfcq_head *head, } static inline struct cds_wfcq_node * -___cds_wfcq_dequeue(struct cds_wfcq_head *head, +___cds_wfcq_dequeue_with_state(cds_wfcq_head_ptr_t u_head, struct cds_wfcq_tail *tail, + int *state, int blocking) { + struct __cds_wfcq_head *head = u_head._h; struct cds_wfcq_node *node, *next; - if (_cds_wfcq_empty(head, tail)) + if (state) + *state = 0; + + if (_cds_wfcq_empty(__cds_wfcq_head_cast(head), tail)) { return NULL; + } node = ___cds_wfcq_node_sync_next(&head->node, blocking); - if (!blocking && node == CDS_WFCQ_WOULDBLOCK) + if (!blocking && node == CDS_WFCQ_WOULDBLOCK) { return CDS_WFCQ_WOULDBLOCK; + } if ((next = CMM_LOAD_SHARED(node->next)) == NULL) { /* @@ -339,8 +400,11 @@ ___cds_wfcq_dequeue(struct cds_wfcq_head *head, * content. */ _cds_wfcq_node_init(&head->node); - if (uatomic_cmpxchg(&tail->p, node, &head->node) == node) + if (uatomic_cmpxchg(&tail->p, node, &head->node) == node) { + if (state) + *state |= CDS_WFCQ_STATE_LAST; return node; + } next = ___cds_wfcq_node_sync_next(node, blocking); /* * In nonblocking mode, if we would need to block to @@ -364,7 +428,7 @@ ___cds_wfcq_dequeue(struct cds_wfcq_head *head, } /* - * __cds_wfcq_dequeue_blocking: dequeue a node from the queue. + * __cds_wfcq_dequeue_with_state_blocking: dequeue node from queue, with state. * * Content written into the node before enqueue is guaranteed to be * consistent, but no other memory ordering is ensured. @@ -373,42 +437,95 @@ ___cds_wfcq_dequeue(struct cds_wfcq_head *head, * caller. */ static inline struct cds_wfcq_node * -___cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head, +___cds_wfcq_dequeue_with_state_blocking(cds_wfcq_head_ptr_t head, + struct cds_wfcq_tail *tail, int *state) +{ + return ___cds_wfcq_dequeue_with_state(head, tail, state, 1); +} + +/* + * ___cds_wfcq_dequeue_blocking: dequeue node from queue. + * + * Same as __cds_wfcq_dequeue_with_state_blocking, but without saving + * state. + */ +static inline struct cds_wfcq_node * +___cds_wfcq_dequeue_blocking(cds_wfcq_head_ptr_t head, struct cds_wfcq_tail *tail) { - return ___cds_wfcq_dequeue(head, tail, 1); + return ___cds_wfcq_dequeue_with_state_blocking(head, tail, NULL); } /* - * __cds_wfcq_dequeue_nonblocking: dequeue a node from a wait-free queue. + * __cds_wfcq_dequeue_with_state_nonblocking: dequeue node, with state. * * Same as __cds_wfcq_dequeue_blocking, but returns CDS_WFCQ_WOULDBLOCK * if it needs to block. */ static inline struct cds_wfcq_node * -___cds_wfcq_dequeue_nonblocking(struct cds_wfcq_head *head, +___cds_wfcq_dequeue_with_state_nonblocking(cds_wfcq_head_ptr_t head, + struct cds_wfcq_tail *tail, int *state) +{ + return ___cds_wfcq_dequeue_with_state(head, tail, state, 0); +} + +/* + * ___cds_wfcq_dequeue_nonblocking: dequeue node from queue. + * + * Same as __cds_wfcq_dequeue_with_state_nonblocking, but without saving + * state. + */ +static inline struct cds_wfcq_node * +___cds_wfcq_dequeue_nonblocking(cds_wfcq_head_ptr_t head, struct cds_wfcq_tail *tail) { - return ___cds_wfcq_dequeue(head, tail, 0); + return ___cds_wfcq_dequeue_with_state_nonblocking(head, tail, NULL); } +/* + * __cds_wfcq_splice: enqueue all src_q nodes at the end of dest_q. + * + * Dequeue all nodes from src_q. + * dest_q must be already initialized. + * Mutual exclusion for src_q should be ensured by the caller as + * specified in the "Synchronisation table". + * Returns enum cds_wfcq_ret which indicates the state of the src or + * dest queue. + */ static inline enum cds_wfcq_ret ___cds_wfcq_splice( - struct cds_wfcq_head *dest_q_head, + cds_wfcq_head_ptr_t u_dest_q_head, struct cds_wfcq_tail *dest_q_tail, - struct cds_wfcq_head *src_q_head, + cds_wfcq_head_ptr_t u_src_q_head, struct cds_wfcq_tail *src_q_tail, int blocking) { + struct __cds_wfcq_head *dest_q_head = u_dest_q_head._h; + struct __cds_wfcq_head *src_q_head = u_src_q_head._h; struct cds_wfcq_node *head, *tail; + int attempt = 0; - if (_cds_wfcq_empty(src_q_head, src_q_tail)) + /* + * Initial emptiness check to speed up cases where queue is + * empty: only require loads to check if queue is empty. + */ + if (_cds_wfcq_empty(__cds_wfcq_head_cast(src_q_head), src_q_tail)) return CDS_WFCQ_RET_SRC_EMPTY; - head = ___cds_wfcq_node_sync_next(&src_q_head->node, blocking); - if (head == CDS_WFCQ_WOULDBLOCK) - return CDS_WFCQ_RET_WOULDBLOCK; - _cds_wfcq_node_init(&src_q_head->node); + for (;;) { + /* + * Open-coded _cds_wfcq_empty() by testing result of + * uatomic_xchg, as well as tail pointer vs head node + * address. + */ + head = uatomic_xchg(&src_q_head->node.next, NULL); + if (head) + break; /* non-empty */ + if (CMM_LOAD_SHARED(src_q_tail->p) == &src_q_head->node) + return CDS_WFCQ_RET_SRC_EMPTY; + if (___cds_wfcq_busy_wait(&attempt, blocking)) + return CDS_WFCQ_RET_WOULDBLOCK; + } /* * Memory barrier implied before uatomic_xchg() orders store to @@ -422,28 +539,28 @@ ___cds_wfcq_splice( * Append the spliced content of src_q into dest_q. Does not * require mutual exclusion on dest_q (wait-free). */ - if (___cds_wfcq_append(dest_q_head, dest_q_tail, head, tail)) + if (___cds_wfcq_append(__cds_wfcq_head_cast(dest_q_head), dest_q_tail, + head, tail)) return CDS_WFCQ_RET_DEST_NON_EMPTY; else return CDS_WFCQ_RET_DEST_EMPTY; } - /* * __cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q. * * Dequeue all nodes from src_q. * dest_q must be already initialized. - * Dequeue/splice/iteration mutual exclusion for src_q should be ensured - * by the caller. + * Mutual exclusion for src_q should be ensured by the caller as + * specified in the "Synchronisation table". * Returns enum cds_wfcq_ret which indicates the state of the src or * dest queue. Never returns CDS_WFCQ_RET_WOULDBLOCK. */ static inline enum cds_wfcq_ret ___cds_wfcq_splice_blocking( - struct cds_wfcq_head *dest_q_head, + cds_wfcq_head_ptr_t dest_q_head, struct cds_wfcq_tail *dest_q_tail, - struct cds_wfcq_head *src_q_head, + cds_wfcq_head_ptr_t src_q_head, struct cds_wfcq_tail *src_q_tail) { return ___cds_wfcq_splice(dest_q_head, dest_q_tail, @@ -458,9 +575,9 @@ ___cds_wfcq_splice_blocking( */ static inline enum cds_wfcq_ret ___cds_wfcq_splice_nonblocking( - struct cds_wfcq_head *dest_q_head, + cds_wfcq_head_ptr_t dest_q_head, struct cds_wfcq_tail *dest_q_tail, - struct cds_wfcq_head *src_q_head, + cds_wfcq_head_ptr_t src_q_head, struct cds_wfcq_tail *src_q_tail) { return ___cds_wfcq_splice(dest_q_head, dest_q_tail, @@ -468,26 +585,39 @@ ___cds_wfcq_splice_nonblocking( } /* - * cds_wfcq_dequeue_blocking: dequeue a node from a wait-free queue. + * cds_wfcq_dequeue_with_state_blocking: dequeue a node from a wait-free queue. * * Content written into the node before enqueue is guaranteed to be * consistent, but no other memory ordering is ensured. - * Mutual exlusion with cds_wfcq_splice_blocking and dequeue lock is + * Mutual exclusion with cds_wfcq_splice_blocking and dequeue lock is * ensured. * It is valid to reuse and free a dequeued node immediately. */ static inline struct cds_wfcq_node * -_cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head, - struct cds_wfcq_tail *tail) +_cds_wfcq_dequeue_with_state_blocking(struct cds_wfcq_head *head, + struct cds_wfcq_tail *tail, int *state) { struct cds_wfcq_node *retval; _cds_wfcq_dequeue_lock(head, tail); - retval = ___cds_wfcq_dequeue_blocking(head, tail); + retval = ___cds_wfcq_dequeue_with_state_blocking(cds_wfcq_head_cast(head), + tail, state); _cds_wfcq_dequeue_unlock(head, tail); return retval; } +/* + * cds_wfcq_dequeue_blocking: dequeue node from queue. + * + * Same as cds_wfcq_dequeue_blocking, but without saving state. + */ +static inline struct cds_wfcq_node * +_cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head, + struct cds_wfcq_tail *tail) +{ + return _cds_wfcq_dequeue_with_state_blocking(head, tail, NULL); +} + /* * cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q. * @@ -495,7 +625,7 @@ _cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head, * dest_q must be already initialized. * Content written into the node before enqueue is guaranteed to be * consistent, but no other memory ordering is ensured. - * Mutual exlusion with cds_wfcq_dequeue_blocking and dequeue lock is + * Mutual exclusion with cds_wfcq_dequeue_blocking and dequeue lock is * ensured. * Returns enum cds_wfcq_ret which indicates the state of the src or * dest queue. Never returns CDS_WFCQ_RET_WOULDBLOCK. @@ -510,8 +640,8 @@ _cds_wfcq_splice_blocking( enum cds_wfcq_ret ret; _cds_wfcq_dequeue_lock(src_q_head, src_q_tail); - ret = ___cds_wfcq_splice_blocking(dest_q_head, dest_q_tail, - src_q_head, src_q_tail); + ret = ___cds_wfcq_splice_blocking(cds_wfcq_head_cast(dest_q_head), dest_q_tail, + cds_wfcq_head_cast(src_q_head), src_q_tail); _cds_wfcq_dequeue_unlock(src_q_head, src_q_tail); return ret; }