wfqueue: expand API, simplify implementation, small performance boost
A message that is not in the archive (sent off-list or not yet received): 20120812145000.GA32747@Krystal
Quoted 65 lines · Mathieu Desnoyers
This work is derived from the patch from Lai Jiangshan submitted as "urcu: new wfqueue implementation"
(http://lists.lttng.org/pipermail/lttng-dev/2012-August/018379.html)Its changelog:
Quoted 31 lines
Some guys would be surprised by this fact:
There are already TWO implementations of wfqueue in urcu.The first one is in
urcu/static/wfqueue.h:
1) enqueue: exchange the tail and then updateprevious->next
2) dequeue: wait for first node's next pointer and them shift, a dummy nodeis introduced to avoid the queue->tail become NULL when shift.The second one shares some code with the first one, and the left code are spreading in
urcu-call-rcu-impl.h:
1) enqueue: share with the first one
2) no dequeue operation: and no shift, so it don't need dummy node, Although the dummy node is queued when initialization, but it is removed after the firstdequeue_alloperation incall_rcu_thread().call_rcu_data_free()forgets to handle the dummy node if it is not removed.
3)dequeue_all: record the old head and tail, andqueue->headbecome the special
tail node.(atomic record the tail and change the tail).The second implementation's code are spreading, bad for review, and it is not tested by tests/test_urcu_wfq.
So we need a better implementation avoid the dummy node dancing and can service both generic wfqueue APIs and
dequeue_allAPI for call rcu.The new implementation:
1) enqueue: share with the first one/original implementation.
2) dequeue: shift when node count >= 2, cmpxchg when node count = 1. no dummy node, save memory.
3)dequeue_all: simply setqueue->head.nextto NULL, xchg the tail and return the old head.next.Quoted 2 lines
More implementation details are in the code. tests/test_urcu_wfq will be update in future for testing new APIs.
The patch proposed by Lai brings a very interesting simplification to the single-node handling (which is kept here), and moves all queue handling code away from
call_rcuimplementation, back into the wfqueue code. This has the benefit to allow testing enhancements.I modified it so the API does not expose implementation details to the user (e.g.
___cds_wfq_node_sync_next). I added a "splice" operation and a for loop iterator which should allow wfqueue users to use the list very efficiently both from LGPL/GPL code and from non-LGPL-compatible code.Benchmarks performed on Intel(R) Core(TM) i7-3520M CPU @ 2.90GHz (dual-core, with hyperthreading)
Benchmark invoked:
test_urcu_wfq2 2 10Only did 2 runs, but a small improvement seems to be clear for the dequeue speed:
Before patch:
testdur 10 nr_enqueuers 2 wdelay 0 nr_dequeuers 2 rdur 0 nr_enqueues 136251248 nr_dequeues 54694027 successful enqueues 136251248 successful dequeues 54693904 end_dequeues 81557344 nr_ops 190945275 testdur 10 nr_enqueuers 2 wdelay 0 nr_dequeuers 2 rdur 0 nr_enqueues 137258881 nr_dequeues 54463340 successful enqueues 137258881 successful dequeues 54463238 end_dequeues 82795643 nr_ops 191722221
Quoted 6 lines
After patch:
testdur 10 nr_enqueuers 2 wdelay 0 nr_dequeuers 2 rdur 0 nr_enqueues 138589301 nr_dequeues 56911253 successful enqueues 138589301 successful dequeues 56910916 end_dequeues 81678385 nr_ops 195500554 testdur 10 nr_enqueuers 2 wdelay 0 nr_dequeuers 2 rdur 0 nr_enqueues 139007622 nr_dequeues 57281502 successful enqueues 139007622 successful dequeues 57281348 end_dequeues 81726274 nr_ops 196289124Summary: Number of enqueues is slightly lower,
?!
I see the nr_enqueues and successful enqueues are both increased after after patch.
Quoted 63 lines
probably due to higher
dequeue rate. Number of dequeue increased. Respective rate change is within 1% (slowdown) for enqueue, 2% (performance improvement) for dequeue. Overall number of operations (dequeue+enqueue) increased with the patch.We can verify that:
successful enqueues - successful dequeues =end_dequeuesFor all runs (ensures correctness: no lost node).
CC Lai Jiangshan <[email protected]>CC Paul McKenney <[email protected]>Signed-off-by Mathieu Desnoyers <[email protected]>---
urcu-call-rcu-impl.h +8
-16@@ -21,6 +21,7 @@*/ #define _GNU_SOURCE +#define _LGPL_SOURCE#include <stdio.h> #include <pthread.h> #include <signal.h> @@ -220,10 +221,7 @@ static void call_rcu_wake_up(struct call_rcu_data *crdp)static void *call_rcu_thread (void *arg) {unsigned long cbcount; -struct cds_wfq_node *cbs; -struct cds_wfq_node **cbs_tail; -struct call_rcu_data *crdp = (struct call_rcu_data *)arg; -struct rcu_head *rhp; +struct call_rcu_data *crdp = (struct call_rcu_data *) arg;int rt = !!(uatomic_read (&crdp->flags) &URCU_CALL_RCU_RT );int ret;@@ -243,35 +241,29 @@ static void *call_rcu_thread(void *arg)cmm_smp_mb (); }for (;;) { -if (&crdp->cbs.head !=_CMM_LOAD_SHARED (crdp->cbs.tail)) { -while ((cbs =_CMM_LOAD_SHARED (crdp->cbs.head)) ==NULL ) -poll (NULL ,0 ,1 ); -_CMM_STORE_SHARED (crdp->cbs.head,NULL ); - cbs_tail = (struct cds_wfq_node **) -uatomic_xchg (&crdp->cbs.tail, &crdp->cbs.head); +struct cds_wfq_queue cbs_tmp; +struct cds_wfq_node *cbs; + +cds_wfq_init (&cbs_tmp); +__cds_wfq_splice_blocking (&cbs_tmp, &crdp->cbs); +if (!cds_wfq_empty (&cbs_tmp)) {synchronize_rcu (); cbcount =0 ; - do { - while (cbs->next == NULL && - &cbs->next != cbs_tail) - poll(NULL, 0, 1); - if (cbs == &crdp->cbs.dummy) { - cbs = cbs->next;
urcu-call-rcu-impl.h static void *call_rcu_thread(void *arg) · line 257
Quoted 10 lines
urcu-call-rcu-impl.h +5
-4-continue ; - } - rhp = (struct rcu_head *)cbs; - cbs = cbs->next; +__cds_wfq_for_each_blocking (&cbs_tmp, cbs) { +struct rcu_head *rhp; + + rhp =caa_container_of (cbs, + struct rcu_head, next); rhp->func (rhp);
cbs is freed hear, but it will be used in __cds_wfq_next_blocking(). Introduce __cds_wfq_for_each_blocking_safe() ?
Quoted 45 lines
urcu-call-rcu-impl.h +9
-13cbcount++; - }while (cbs !=NULL ); + }uatomic_sub (&crdp->qlen, cbcount); }if (uatomic_read (&crdp->flags) &URCU_CALL_RCU_STOP )break ;rcu_thread_offline ();if (!rt) { -if (&crdp->cbs.head - ==_CMM_LOAD_SHARED (crdp->cbs.tail)) { +if (cds_wfq_empty (&crdp->cbs)) {call_rcu_wait (crdp);poll (NULL ,0 ,10 );uatomic_dec (&crdp->futex);@@ -625,32 +617,32 @@ void call_rcu(struct rcu_head *head,*/ void call_rcu_data_free (struct call_rcu_data *crdp) { -struct cds_wfq_node *cbs; -struct cds_wfq_node **cbs_tail; -struct cds_wfq_node **cbs_endprev; -if (crdp ==NULL || crdp == default_call_rcu_data) {return ; } +if ((uatomic_read (&crdp->flags) &URCU_CALL_RCU_STOPPED ) ==0 ) {uatomic_or (&crdp->flags,URCU_CALL_RCU_STOP );wake_call_rcu_thread (crdp);while ((uatomic_read (&crdp->flags) &URCU_CALL_RCU_STOPPED ) ==0 )poll (NULL ,0 ,1 ); } -if (&crdp->cbs.head !=_CMM_LOAD_SHARED (crdp->cbs.tail)) { -while ((cbs =_CMM_LOAD_SHARED (crdp->cbs.head)) ==NULL ) -poll (NULL ,0 ,1 ); -_CMM_STORE_SHARED (crdp->cbs.head,NULL ); - cbs_tail = (struct cds_wfq_node **) -uatomic_xchg (&crdp->cbs.tail, &crdp->cbs.head); + +if (!cds_wfq_empty (&crdp->cbs)) { +struct cds_wfq_queue cbs_tmp; + +cds_wfq_init (&cbs_tmp); +__cds_wfq_splice_blocking (&cbs_tmp, &crdp->cbs);
urcu-call-rcu-impl.h void call_rcu(struct rcu_head *head, · line 642
Quoted 10 lines
urcu-call-rcu-impl.h +5
-3+/* Create default call rcu data if need be */ (void )get_default_call_rcu_data (); - cbs_endprev = (struct cds_wfq_node **) -uatomic_xchg (&default_call_rcu_data, cbs_tail); - *cbs_endprev = cbs; + +__cds_wfq_splice_blocking (&default_call_rcu_data->cbs, + &cbs_tmp); +
Too much code to me, cbs_tmp is not required here.
/* Create default call rcu data if need be */
(void) get_default_call_rcu_data();+ __cds_wfq_splice_blocking(&default_call_rcu_data->cbs,&crdp->cbs);
Quoted 143 lines
urcu-call-rcu-impl.h +1
-0uatomic_add (&default_call_rcu_data->qlen,uatomic_read (&crdp->qlen)); +wake_call_rcu_thread (default_call_rcu_data); }urcu/static/wfqueue.h +62
-18@@ -9,7 +9,8 @@* TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See wfqueue.h for linking * dynamically with the userspace rcu library. * -* Copyright 2010 - Mathieu Desnoyers <[email protected]> +* Copyright 2010-2012 - Mathieu Desnoyers <[email protected]> +* Copyright 2011-2012 - Lai Jiangshan <[email protected]> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public @@ -29,6 +30,7 @@#include <pthread.h> #include <assert.h> #include <poll.h> +#include <stdbool.h> #include <urcu/compiler.h> #include <urcu/uatomic.h> @@ -38,11 +40,16 @@ extern "C" {/* * Queue with wait-free enqueue/blocking dequeue. -* This implementation adds a dummy head node when the queue is empty to ensure -* we can always update the queue locklessly. * * Inspired from half-wait-free/half-blocking queue implementation done by * Paul E. McKenney. +* +* Caller must ensure mutual exclusion of queue update operations +* "dequeue" and "splice" source queue. Queue read operations "first" +* and "next" need to be protected against concurrent "dequeue" and +* "splice" (for source queue) by the caller. "enqueue", "splice" +* (destination queue), and "empty" are the only operations that can be +* used without any mutual exclusion. */ #define WFQ_ADAPT_ATTEMPTS 10/* Retry if being set */ @@ -57,31 +64,51 @@ static inline void _cds_wfq_init(struct cds_wfq_queue *q) {int ret; -_cds_wfq_node_init (&q->dummy);/* Set queue head and tail */ - q->head = &q->dummy; - q->tail = &q->dummy.next; +_cds_wfq_node_init (&q->head); + q->tail = &q->head; ret =pthread_mutex_init (&q->lock,NULL );assert (!ret); } -static inline void _cds_wfq_enqueue (struct cds_wfq_queue *q, -struct cds_wfq_node *node) +static inline bool _cds_wfq_empty (struct cds_wfq_queue *q) +{ +/* +* Queue is empty if no node is pointed by q->head.next nor q->tail. +*/ +return CMM_LOAD_SHARED (q->head.next) ==NULL + &&CMM_LOAD_SHARED (q->tail) == &q->head; +} + +static inline void ___cds_wfq_append (struct cds_wfq_queue *q, +struct cds_wfq_node *new_head, +struct cds_wfq_node *new_tail) { -struct cds_wfq_node **old_tail; +struct cds_wfq_node *old_tail;/* -* uatomic_xchg() implicit memory barrier orders earlier stores to data -* structure containing node and setting node->next to NULL before -* publication. +* Implicit memory barrier before uatomic_xchg() orders earlier +* stores to data structure containing node and setting +* node->next to NULL before publication. */ - old_tail =uatomic_xchg (&q->tail, &node->next); + old_tail =uatomic_xchg (&q->tail, new_tail); +/* -* At this point, dequeuers see a NULL old_tail->next, which indicates -* that the queue is being appended to. The following store will append -* "node" to the queue from a dequeuer perspective. +* Implicit memory barrier after uatomic_xchg() orders store to +* q->tail before store to old_tail->next. +* +* At this point, dequeuers see a NULL q->tail->next, which +* indicates that the queue is being appended to. The following +* store will append "node" to the queue from a dequeuer +* perspective. */ -CMM_STORE_SHARED (*old_tail, node); +CMM_STORE_SHARED (old_tail->next, new_head); +} + +static inline void _cds_wfq_enqueue (struct cds_wfq_queue *q, +struct cds_wfq_node *new_tail) +{ +___cds_wfq_append (q, new_tail, new_tail); }/* @@ -100,14 +127,45 @@ ___cds_wfq_node_sync_next(struct cds_wfq_node *node)if (++attempt >=WFQ_ADAPT_ATTEMPTS ) {poll (NULL ,0 ,WFQ_WAIT );/* Wait for 10ms */ attempt =0 ; - }else + }else {caa_cpu_relax (); + } }return next; }/* +* ___cds_wfq_first_blocking: get first node of a queue, without dequeuing. +* +* Mutual exclusion with "dequeue" and "splice" operations must be ensured +* by the caller. +*/ +static inline struct cds_wfq_node * +___cds_wfq_first_blocking (struct cds_wfq_queue *q) +{ +if (_cds_wfq_empty (q)) +return NULL ; +return ___cds_wfq_node_sync_next (&q->head); +} + +/* +* ___cds_wfq_next_blocking: get next node of a queue, without dequeuing. +*
urcu/static/wfqueue.h ___cds_wfq_node_sync_next(struct cds_wfq_node *node) · line 164
Quoted 10 lines
urcu/static/wfqueue.h +10
-0+* Mutual exclusion with "dequeue" and "splice" operations must be ensured +* by the caller. +*/ +static inline struct cds_wfq_node * +___cds_wfq_next_blocking (struct cds_wfq_queue *q,struct cds_wfq_node *node) +{ +if (CMM_LOAD_SHARED (q->tail) == node) +return NULL ; +return ___cds_wfq_node_sync_next (node); +}
The same BUG as you told me.
If q has only one node just enqueued by other thread.
but if q->head.next is seen, ___cds_wfq_first_blocking() returns a node, And the update of q->tail is not seen, it is still &q->head, ___cds_wfq_node_sync_next(node) will be loop for every if there is no other enqueue.
static inline struct cds_wfq_node * ___cds_wfq_first_blocking(struct cds_wfq_queue *q) {
+ struct cds_wfq_node *ret.
if (_cds_wfq_empty(q))
return NULL;
ret = ___cds_wfq_node_sync_next(&q->head);+ cmm_smp_rmb();
+ return ret;
}
Quoted 16 lines
urcu/static/wfqueue.h +5
-4+ +/* +* ___cds_wfq_dequeue_blocking: dequeue a node from the queue. +* * It is valid to reuse and free a dequeued node immediately. * * No need to go on a waitqueue here, as there is no possible state in which the @@ -120,42 +178,123 @@ ___cds_wfq_dequeue_blocking(struct cds_wfq_queue *q) {struct cds_wfq_node *node, *next; -/* -* Queue is empty if it only contains the dummy node. -*/ -if (q->head == &q->dummy &&CMM_LOAD_SHARED (q->tail) == &q->dummy.next) +if (_cds_wfq_empty (q))
urcu/static/wfqueue.h ___cds_wfq_dequeue_blocking(struct cds_wfq_queue *q) · line 189
Quoted 10 lines
urcu/static/wfqueue.h +6
-2return NULL ; - node = q->head; - next =___cds_wfq_node_sync_next (node); + node =___cds_wfq_node_sync_next (&q->head); + +if ((next =CMM_LOAD_SHARED (node->next)) ==NULL ) { +/* Load node->next before q->tail */ +cmm_smp_rmb (); +if (CMM_LOAD_SHARED (q->tail) == node) {
I don't know why I added this "if" since it is likely true. Could you remove the above 3 lines?
(I remember there is a mb() before uatomic_cmpxchg() which means this mb() is before the test in uatomic_cmpxchg())
Quoted 80 lines
urcu/static/wfqueue.h +66
-8+/* +* @node is the only node in the queue. +* Try to move the tail to &q->head +*/ +_cds_wfq_node_init (&q->head); +if (uatomic_cmpxchg (&q->tail, node, &q->head) == node) +return node; + } + next =___cds_wfq_node_sync_next (node); + }/* * Move queue head forward. */ - q->head = next; + q->head.next = next; + +return node; +} + +/* +* ___cds_wfq_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. +* caller ensures mutual exclusion of dequeue and splice operations on +* src_q. +*/ +static inline void +___cds_wfq_splice_blocking (struct cds_wfq_queue *dest_q, +struct cds_wfq_queue *src_q) +{ +struct cds_wfq_node *head, *tail; + +if (_cds_wfq_empty (src_q)) +return ; + + head =___cds_wfq_node_sync_next (&src_q->head); +_cds_wfq_node_init (&src_q->head); + +/* +* Memory barrier implied before uatomic_xchg() orders store to +* src_q->head before store to src_q->tail. This is required by +* concurrent enqueue on src_q, which exchanges the tail before +* updating the previous tail's next pointer. +*/ + tail =uatomic_xchg (&src_q->tail, &src_q->head); +/* -* Requeue dummy node if we just dequeued it. +* Append the spliced content of src_q into dest_q. Does not +* require mutual exclusion on dest_q (wait-free). */ -if (node == &q->dummy) { -_cds_wfq_node_init (node); -_cds_wfq_enqueue (q, node); -return ___cds_wfq_dequeue_blocking (q); - } -return node; +___cds_wfq_append (dest_q, head, tail); +} + +/* Locking performed within cds_wfq calls. */ +static inline struct cds_wfq_node * +_cds_wfq_first_blocking (struct cds_wfq_queue *q) +{ +struct cds_wfq_node *retval; +int ret; + + ret =pthread_mutex_lock (&q->lock); +assert (!ret); + retval =___cds_wfq_first_blocking (q); + ret =pthread_mutex_unlock (&q->lock); +assert (!ret); +return retval; +} + +static inline struct cds_wfq_node * +_cds_wfq_next_blocking (struct cds_wfq_queue *q,struct cds_wfq_node *node) +{
urcu/static/wfqueue.h ___cds_wfq_dequeue_blocking(struct cds_wfq_queue *q) · line 271
Quoted 10 lines
urcu/static/wfqueue.h +9
-0+struct cds_wfq_node *retval; +int ret; + + ret =pthread_mutex_lock (&q->lock); +assert (!ret); + retval =___cds_wfq_next_blocking (q, node); + ret =pthread_mutex_unlock (&q->lock); +assert (!ret); +return retval; }
I reject these _cds_wfq_first_blocking(), _cds_wfq_next_blocking() and cds_wfq_for_each_blocking(), because the claimed "Locking" makes no sense:
1. It protects nothing in _cds_wfq_next_blocking().
2. There is no "Locking" in the loop body, @node is not dequeued, it will be invalid if some other dequeue it,
and _cds_wfq_next_blocking() results BUG.
Quoted 64 lines
urcu/static/wfqueue.h +16
-3static inline struct cds_wfq_node *_cds_wfq_dequeue_blocking (struct cds_wfq_queue *q) { -struct cds_wfq_node *retnode; +struct cds_wfq_node *retval;int ret; ret =pthread_mutex_lock (&q->lock);assert (!ret); - retnode =___cds_wfq_dequeue_blocking (q); + retval =___cds_wfq_dequeue_blocking (q); ret =pthread_mutex_unlock (&q->lock);assert (!ret); -return retnode; +return retval; +} + +static inline void +_cds_wfq_splice_blocking (struct cds_wfq_queue *dest_q, +struct cds_wfq_queue *src_q) +{ +int ret; + + ret =pthread_mutex_lock (&src_q->lock); +assert (!ret); +___cds_wfq_splice_blocking (dest_q, src_q); + ret =pthread_mutex_unlock (&src_q->lock); +assert (!ret); }#ifdef __cplusplusurcu/wfqueue.h +3
-3@@ -6,7 +6,8 @@* * Userspace RCU library - Queue with Wait-Free Enqueue/Blocking Dequeue * -* Copyright 2010 - Mathieu Desnoyers <[email protected]> +* Copyright 2010-2012 - Mathieu Desnoyers <[email protected]> +* Copyright 2011-2012 - Lai Jiangshan <[email protected]> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public @@ -25,6 +26,7 @@#include <pthread.h> #include <assert.h> +#include <stdbool.h> #include <urcu/compiler.h> #ifdef __cplusplus@@ -33,8 +35,6 @@ extern "C" {/* * Queue with wait-free enqueue/blocking dequeue. -* This implementation adds a dummy head node when the queue is empty to ensure -* we can always update the queue locklessly. * * Inspired from half-wait-free/half-blocking queue implementation done by * Paul E. McKenney.
urcu/wfqueue.h struct cds_wfq_node { · line 51
Quoted 9 lines
urcu/wfqueue.h +2
-2@@ -45,8 +45,8 @@ struct cds_wfq_node { };struct cds_wfq_queue { -struct cds_wfq_node *head, **tail; -struct cds_wfq_node dummy;/* Dummy node */ +struct cds_wfq_node head, *tail; +struct cds_wfq_node padding;/* unused */ pthread_mutex_t lock; };
Why keep the padding?
Quoted 172 lines
urcu/wfqueue.h +73
-5@@ -55,22 +55,90 @@ struct cds_wfq_queue {#include <urcu/static/wfqueue.h> #define cds_wfq_node_init _cds_wfq_node_init -#define cds_wfq_init _cds_wfq_init -#define cds_wfq_enqueue _cds_wfq_enqueue -#define __cds_wfq_dequeue_blocking ___cds_wfq_dequeue_blocking +#define cds_wfq_init _cds_wfq_init +#define cds_wfq_empty _cds_wfq_empty +#define cds_wfq_enqueue _cds_wfq_enqueue + +/* Locking performed within cds_wfq calls. */ #define cds_wfq_dequeue_blocking _cds_wfq_dequeue_blocking +#define cds_wfq_splice_blocking _cds_wfq_splice_blocking +#define cds_wfq_first_blocking _cds_wfq_first_blocking +#define cds_wfq_next_blocking _cds_wfq_next_blocking + +/* Locking ensured by caller */ +#define __cds_wfq_dequeue_blocking ___cds_wfq_dequeue_blocking +#define __cds_wfq_splice_blocking ___cds_wfq_splice_blocking +#define __cds_wfq_first_blocking ___cds_wfq_first_blocking +#define __cds_wfq_next_blocking ___cds_wfq_next_blocking#else /* !_LGPL_SOURCE */ extern void cds_wfq_node_init (struct cds_wfq_node *node);extern void cds_wfq_init (struct cds_wfq_queue *q); +extern bool cds_wfq_empty (struct cds_wfq_queue *q);extern void cds_wfq_enqueue (struct cds_wfq_queue *q,struct cds_wfq_node *node); -/* __cds_wfq_dequeue_blocking: caller ensures mutual exclusion between dequeues */ -extern struct cds_wfq_node *__cds_wfq_dequeue_blocking (struct cds_wfq_queue *q); + +/* Locking performed within cds_wfq calls. */ extern struct cds_wfq_node *cds_wfq_dequeue_blocking (struct cds_wfq_queue *q); +extern void cds_wfq_splice_blocking (struct cds_wfq_queue *dest_q, +struct cds_wfq_queue *src_q); +extern struct cds_wfq_node *cds_wfq_first_blocking (struct cds_wfq_queue *q); +extern struct cds_wfq_node *cds_wfq_next_blocking (struct cds_wfq_queue *q, +struct cds_wfq_node *node); + +/* +* __cds_wfq_dequeue_blocking: caller ensures mutual exclusion of dequeue +* and splice operations. +*/ +extern struct cds_wfq_node *__cds_wfq_dequeue_blocking (struct cds_wfq_queue *q); + +/* +* __cds_wfq_splice_blocking: caller ensures mutual exclusion of dequeue and +* splice operations on src_q. dest_q must be already initialized. +*/ +extern void __cds_wfq_splice_blocking (struct cds_wfq_queue *dest_q, +struct cds_wfq_queue *src_q); + +/* +* __cds_wfq_first_blocking: mutual exclusion with "dequeue" and +* "splice" operations must be ensured by the caller. +*/ +extern struct cds_wfq_node *__cds_wfq_first_blocking (struct cds_wfq_queue *q); + +/* +* __cds_wfq_next_blocking: mutual exclusion with "dequeue" and "splice" +* operations must be ensured by the caller. +*/ +extern struct cds_wfq_node *__cds_wfq_next_blocking (struct cds_wfq_queue *q, +struct cds_wfq_node *node);#endif /* !_LGPL_SOURCE */ +/* +* cds_wfq_for_each_blocking: Iterate over all nodes in a queue, without +* dequeuing them. +* +* cds_wfq_for_each_blocking: mutual exclusion is performed within the +* cds_wfq calls. +*/ +#define cds_wfq_for_each_blocking (q, node) \ + for (node = cds_wfq_first_blocking(q); \ + node != NULL; \ + node = cds_wfq_next_blocking(q, node)) + +/* +* __cds_wfq_for_each_blocking: Iterate over all nodes in a queue, +* without dequeuing them. +* +* Mutual exclusion with "dequeue" and "splice" operations must be +* ensured by the caller. +*/ + +#define __cds_wfq_for_each_blocking (q, node) \ + for (node = __cds_wfq_first_blocking(q); \ + node != NULL; \ + node = __cds_wfq_next_blocking(q, node)) + #ifdef __cplusplus }#endif wfqueue.c +43
-4@@ -3,7 +3,8 @@* * Userspace RCU library - Queue with Wait-Free Enqueue/Blocking Dequeue * -* Copyright 2010 - Mathieu Desnoyers <[email protected]> +* Copyright 2010-2012 - Mathieu Desnoyers <[email protected]> +* Copyright 2011-2012 - Lai Jiangshan <[email protected]> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public @@ -38,17 +39,56 @@ void cds_wfq_init(struct cds_wfq_queue *q)_cds_wfq_init (q); } +bool cds_wfq_empty (struct cds_wfq_queue *q) +{ +return _cds_wfq_empty (q); +} +void cds_wfq_enqueue (struct cds_wfq_queue *q,struct cds_wfq_node *node) {_cds_wfq_enqueue (q, node); } +struct cds_wfq_node *cds_wfq_dequeue_blocking (struct cds_wfq_queue *q) +{ +return _cds_wfq_dequeue_blocking (q); +} + +void cds_wfq_splice_blocking (struct cds_wfq_queue *dest_q, +struct cds_wfq_queue *src_q) +{ +_cds_wfq_splice_blocking (dest_q, src_q); +} + +struct cds_wfq_node *cds_wfq_first_blocking (struct cds_wfq_queue *q) +{ +return _cds_wfq_first_blocking (q); +} + +struct cds_wfq_node *cds_wfq_next_blocking (struct cds_wfq_queue *q, +struct cds_wfq_node *node) +{ +return _cds_wfq_next_blocking (q, node); +} +struct cds_wfq_node *__cds_wfq_dequeue_blocking (struct cds_wfq_queue *q) {return ___cds_wfq_dequeue_blocking (q); } -struct cds_wfq_node *cds_wfq_dequeue_blocking (struct cds_wfq_queue *q) +void __cds_wfq_splice_blocking (struct cds_wfq_queue *dest_q, +struct cds_wfq_queue *src_q) { -return _cds_wfq_dequeue_blocking (q); +___cds_wfq_splice_blocking (dest_q, src_q); +} + +struct cds_wfq_node *__cds_wfq_first_blocking (struct cds_wfq_queue *q) +{ +return ___cds_wfq_first_blocking (q); +} + +struct cds_wfq_node *__cds_wfq_next_blocking (struct cds_wfq_queue *q, +struct cds_wfq_node *node) +{ +return ___cds_wfq_next_blocking (q, node); } --
Quoted 4 lines
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com
Thanks,
Lai
A message that is not in the archive (sent off-list or not yet received): CACvQF52366ewREnRrDgi+xUBVQ6_ifWjoDv==nHik8FbCnMugA@mail.gmail.com
Hi Lai,
Quoted 69 lines · * Lai Jiangshan ([email protected]) wrote:
Quoted 67 lines · Mathieu Desnoyers
This work is derived from the patch from Lai Jiangshan submitted as "urcu: new wfqueue implementation"
(http://lists.lttng.org/pipermail/lttng-dev/2012-August/018379.html)Its changelog:
Quoted 31 lines
Some guys would be surprised by this fact:
There are already TWO implementations of wfqueue in urcu.The first one is in
urcu/static/wfqueue.h:
1) enqueue: exchange the tail and then updateprevious->next
2) dequeue: wait for first node's next pointer and them shift, a dummy nodeis introduced to avoid the queue->tail become NULL when shift.The second one shares some code with the first one, and the left code are spreading in
urcu-call-rcu-impl.h:
1) enqueue: share with the first one
2) no dequeue operation: and no shift, so it don't need dummy node, Although the dummy node is queued when initialization, but it is removed after the firstdequeue_alloperation incall_rcu_thread().call_rcu_data_free()forgets to handle the dummy node if it is not removed.
3)dequeue_all: record the old head and tail, andqueue->headbecome the special
tail node.(atomic record the tail and change the tail).The second implementation's code are spreading, bad for review, and it is not tested by tests/test_urcu_wfq.
So we need a better implementation avoid the dummy node dancing and can service both generic wfqueue APIs and
dequeue_allAPI for call rcu.The new implementation:
1) enqueue: share with the first one/original implementation.
2) dequeue: shift when node count >= 2, cmpxchg when node count = 1. no dummy node, save memory.
3)dequeue_all: simply setqueue->head.nextto NULL, xchg the tail and return the old head.next.Quoted 2 lines
More implementation details are in the code. tests/test_urcu_wfq will be update in future for testing new APIs.
The patch proposed by Lai brings a very interesting simplification to the single-node handling (which is kept here), and moves all queue handling code away from
call_rcuimplementation, back into the wfqueue code. This has the benefit to allow testing enhancements.I modified it so the API does not expose implementation details to the user (e.g.
___cds_wfq_node_sync_next). I added a "splice" operation and a for loop iterator which should allow wfqueue users to use the list very efficiently both from LGPL/GPL code and from non-LGPL-compatible code.Benchmarks performed on Intel(R) Core(TM) i7-3520M CPU @ 2.90GHz (dual-core, with hyperthreading)
Benchmark invoked:
test_urcu_wfq2 2 10Only did 2 runs, but a small improvement seems to be clear for the dequeue speed:
Before patch:
testdur 10 nr_enqueuers 2 wdelay 0 nr_dequeuers 2 rdur 0 nr_enqueues 136251248 nr_dequeues 54694027 successful enqueues 136251248 successful dequeues 54693904 end_dequeues 81557344 nr_ops 190945275 testdur 10 nr_enqueuers 2 wdelay 0 nr_dequeuers 2 rdur 0 nr_enqueues 137258881 nr_dequeues 54463340 successful enqueues 137258881 successful dequeues 54463238 end_dequeues 82795643 nr_ops 191722221After patch:
Quoted 8 lines
Quoted 4 lines
testdur 10 nr_enqueuers 2 wdelay 0 nr_dequeuers 2 rdur 0 nr_enqueues 138589301 nr_dequeues 56911253 successful enqueues 138589301 successful dequeues 56910916 end_dequeues 81678385 nr_ops 195500554 testdur 10 nr_enqueuers 2 wdelay 0 nr_dequeuers 2 rdur 0 nr_enqueues 139007622 nr_dequeues 57281502 successful enqueues 139007622 successful dequeues 57281348 end_dequeues 81726274 nr_ops 196289124Summary: Number of enqueues is slightly lower,
?!
I see thenr_enqueuesand successful enqueues are both increased after after patch.
Oh, you're right. I wrote that summary in a hurry. Sorry about that, will fix.
Quoted 70 lines
Quoted 70 lines
probably due to higher
dequeue rate. Number of dequeue increased. Respective rate change is within 1% (slowdown) for enqueue, 2% (performance improvement) for dequeue. Overall number of operations (dequeue+enqueue) increased with the patch.We can verify that:
successful enqueues - successful dequeues =end_dequeuesFor all runs (ensures correctness: no lost node).
CC Lai Jiangshan <[email protected]>CC Paul McKenney <[email protected]>Signed-off-by Mathieu Desnoyers <[email protected]>---
urcu-call-rcu-impl.h +10
-20@@ -21,6 +21,7 @@*/ #define _GNU_SOURCE +#define _LGPL_SOURCE#include <stdio.h> #include <pthread.h> #include <signal.h> @@ -220,10 +221,7 @@ static void call_rcu_wake_up(struct call_rcu_data *crdp)static void *call_rcu_thread (void *arg) {unsigned long cbcount; -struct cds_wfq_node *cbs; -struct cds_wfq_node **cbs_tail; -struct call_rcu_data *crdp = (struct call_rcu_data *)arg; -struct rcu_head *rhp; +struct call_rcu_data *crdp = (struct call_rcu_data *) arg;int rt = !!(uatomic_read (&crdp->flags) &URCU_CALL_RCU_RT );int ret;@@ -243,35 +241,29 @@ static void *call_rcu_thread(void *arg)cmm_smp_mb (); }for (;;) { -if (&crdp->cbs.head !=_CMM_LOAD_SHARED (crdp->cbs.tail)) { -while ((cbs =_CMM_LOAD_SHARED (crdp->cbs.head)) ==NULL ) -poll (NULL ,0 ,1 ); -_CMM_STORE_SHARED (crdp->cbs.head,NULL ); - cbs_tail = (struct cds_wfq_node **) -uatomic_xchg (&crdp->cbs.tail, &crdp->cbs.head); +struct cds_wfq_queue cbs_tmp; +struct cds_wfq_node *cbs; + +cds_wfq_init (&cbs_tmp); +__cds_wfq_splice_blocking (&cbs_tmp, &crdp->cbs); +if (!cds_wfq_empty (&cbs_tmp)) {synchronize_rcu (); cbcount =0 ; - do { - while (cbs->next == NULL && - &cbs->next != cbs_tail) - poll(NULL, 0, 1); - if (cbs == &crdp->cbs.dummy) { - cbs = cbs->next; - continue; - } - rhp = (struct rcu_head *)cbs; - cbs = cbs->next; +__cds_wfq_for_each_blocking (&cbs_tmp, cbs) { +struct rcu_head *rhp;
Quoted 8 lines
urcu-call-rcu-impl.h static void *call_rcu_thread(void *arg) · line 257
Quoted 4 lines
urcu-call-rcu-impl.h +3
-0+ + rhp =caa_container_of (cbs, + struct rcu_head, next); rhp->func (rhp);cbs is freed hear, but it will be used in
__cds_wfq_next_blocking(). Introduce__cds_wfq_for_each_blocking_safe()?
Good point! Will do.
Quoted 60 lines
Quoted 46 lines
urcu-call-rcu-impl.h +9
-13cbcount++; - }while (cbs !=NULL ); + }uatomic_sub (&crdp->qlen, cbcount); }if (uatomic_read (&crdp->flags) &URCU_CALL_RCU_STOP )break ;rcu_thread_offline ();if (!rt) { -if (&crdp->cbs.head - ==_CMM_LOAD_SHARED (crdp->cbs.tail)) { +if (cds_wfq_empty (&crdp->cbs)) {call_rcu_wait (crdp);poll (NULL ,0 ,10 );uatomic_dec (&crdp->futex);@@ -625,32 +617,32 @@ void call_rcu(struct rcu_head *head,*/ void call_rcu_data_free (struct call_rcu_data *crdp) { -struct cds_wfq_node *cbs; -struct cds_wfq_node **cbs_tail; -struct cds_wfq_node **cbs_endprev; -if (crdp ==NULL || crdp == default_call_rcu_data) {return ; } +if ((uatomic_read (&crdp->flags) &URCU_CALL_RCU_STOPPED ) ==0 ) {uatomic_or (&crdp->flags,URCU_CALL_RCU_STOP );wake_call_rcu_thread (crdp);while ((uatomic_read (&crdp->flags) &URCU_CALL_RCU_STOPPED ) ==0 )poll (NULL ,0 ,1 ); } -if (&crdp->cbs.head !=_CMM_LOAD_SHARED (crdp->cbs.tail)) { -while ((cbs =_CMM_LOAD_SHARED (crdp->cbs.head)) ==NULL ) -poll (NULL ,0 ,1 ); -_CMM_STORE_SHARED (crdp->cbs.head,NULL ); - cbs_tail = (struct cds_wfq_node **) -uatomic_xchg (&crdp->cbs.tail, &crdp->cbs.head); + +if (!cds_wfq_empty (&crdp->cbs)) { +struct cds_wfq_queue cbs_tmp; + +cds_wfq_init (&cbs_tmp); +__cds_wfq_splice_blocking (&cbs_tmp, &crdp->cbs);urcu-call-rcu-impl.h void call_rcu(struct rcu_head *head, · line 642
Quoted 10 lines
urcu-call-rcu-impl.h +5
-3+/* Create default call rcu data if need be */ (void )get_default_call_rcu_data (); - cbs_endprev = (struct cds_wfq_node **) -uatomic_xchg (&default_call_rcu_data, cbs_tail); - *cbs_endprev = cbs; + +__cds_wfq_splice_blocking (&default_call_rcu_data->cbs, + &cbs_tmp); +Too much code to me,
cbs_tmpis not required here.
Quoted 3 lines
/* Create default call rcu data if need be */ (void) get_default_call_rcu_data();+
__cds_wfq_splice_blocking(&default_call_rcu_data->cbs,&crdp->cbs);
You're right. I initially thought that the order with the call to get_default_call_rcu_data() was important, but it rather looks like we call get_default_call_rcu_data() just to ensure that we have somewhere to send the already enqueued callbacks. Indeed, we can merge those as you propose. Will do!
Quoted 158 lines
Quoted 146 lines
urcu-call-rcu-impl.h +1
-0uatomic_add (&default_call_rcu_data->qlen,uatomic_read (&crdp->qlen)); +wake_call_rcu_thread (default_call_rcu_data); }urcu/static/wfqueue.h +62
-18@@ -9,7 +9,8 @@* TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See wfqueue.h for linking * dynamically with the userspace rcu library. * -* Copyright 2010 - Mathieu Desnoyers <[email protected]> +* Copyright 2010-2012 - Mathieu Desnoyers <[email protected]> +* Copyright 2011-2012 - Lai Jiangshan <[email protected]> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public @@ -29,6 +30,7 @@#include <pthread.h> #include <assert.h> #include <poll.h> +#include <stdbool.h> #include <urcu/compiler.h> #include <urcu/uatomic.h> @@ -38,11 +40,16 @@ extern "C" {/* * Queue with wait-free enqueue/blocking dequeue. -* This implementation adds a dummy head node when the queue is empty to ensure -* we can always update the queue locklessly. * * Inspired from half-wait-free/half-blocking queue implementation done by * Paul E. McKenney. +* +* Caller must ensure mutual exclusion of queue update operations +* "dequeue" and "splice" source queue. Queue read operations "first" +* and "next" need to be protected against concurrent "dequeue" and +* "splice" (for source queue) by the caller. "enqueue", "splice" +* (destination queue), and "empty" are the only operations that can be +* used without any mutual exclusion. */ #define WFQ_ADAPT_ATTEMPTS 10/* Retry if being set */ @@ -57,31 +64,51 @@ static inline void _cds_wfq_init(struct cds_wfq_queue *q) {int ret; -_cds_wfq_node_init (&q->dummy);/* Set queue head and tail */ - q->head = &q->dummy; - q->tail = &q->dummy.next; +_cds_wfq_node_init (&q->head); + q->tail = &q->head; ret =pthread_mutex_init (&q->lock,NULL );assert (!ret); } -static inline void _cds_wfq_enqueue (struct cds_wfq_queue *q, -struct cds_wfq_node *node) +static inline bool _cds_wfq_empty (struct cds_wfq_queue *q) +{ +/* +* Queue is empty if no node is pointed by q->head.next nor q->tail. +*/ +return CMM_LOAD_SHARED (q->head.next) ==NULL + &&CMM_LOAD_SHARED (q->tail) == &q->head; +} + +static inline void ___cds_wfq_append (struct cds_wfq_queue *q, +struct cds_wfq_node *new_head, +struct cds_wfq_node *new_tail) { -struct cds_wfq_node **old_tail; +struct cds_wfq_node *old_tail;/* -* uatomic_xchg() implicit memory barrier orders earlier stores to data -* structure containing node and setting node->next to NULL before -* publication. +* Implicit memory barrier before uatomic_xchg() orders earlier +* stores to data structure containing node and setting +* node->next to NULL before publication. */ - old_tail =uatomic_xchg (&q->tail, &node->next); + old_tail =uatomic_xchg (&q->tail, new_tail); +/* -* At this point, dequeuers see a NULL old_tail->next, which indicates -* that the queue is being appended to. The following store will append -* "node" to the queue from a dequeuer perspective. +* Implicit memory barrier after uatomic_xchg() orders store to +* q->tail before store to old_tail->next. +* +* At this point, dequeuers see a NULL q->tail->next, which +* indicates that the queue is being appended to. The following +* store will append "node" to the queue from a dequeuer +* perspective. */ -CMM_STORE_SHARED (*old_tail, node); +CMM_STORE_SHARED (old_tail->next, new_head); +} + +static inline void _cds_wfq_enqueue (struct cds_wfq_queue *q, +struct cds_wfq_node *new_tail) +{ +___cds_wfq_append (q, new_tail, new_tail); }/* @@ -100,14 +127,45 @@ ___cds_wfq_node_sync_next(struct cds_wfq_node *node)if (++attempt >=WFQ_ADAPT_ATTEMPTS ) {poll (NULL ,0 ,WFQ_WAIT );/* Wait for 10ms */ attempt =0 ; - }else + }else {caa_cpu_relax (); + } }return next; }/* +* ___cds_wfq_first_blocking: get first node of a queue, without dequeuing. +* +* Mutual exclusion with "dequeue" and "splice" operations must be ensured +* by the caller. +*/ +static inline struct cds_wfq_node * +___cds_wfq_first_blocking (struct cds_wfq_queue *q) +{ +if (_cds_wfq_empty (q)) +return NULL ; +return ___cds_wfq_node_sync_next (&q->head); +} + +/* +* ___cds_wfq_next_blocking: get next node of a queue, without dequeuing. +* urcu/static/wfqueue.h ___cds_wfq_node_sync_next(struct cds_wfq_node *node) · line 164
Quoted 10 lines
urcu/static/wfqueue.h +10
-0+* Mutual exclusion with "dequeue" and "splice" operations must be ensured +* by the caller. +*/ +static inline struct cds_wfq_node * +___cds_wfq_next_blocking (struct cds_wfq_queue *q,struct cds_wfq_node *node) +{ +if (CMM_LOAD_SHARED (q->tail) == node) +return NULL ; +return ___cds_wfq_node_sync_next (node); +}
Quoted 6 lines
The same BUG as you told me.
If q has only one node just enqueued by other thread.
but ifq->head.nextis seen,___cds_wfq_first_blocking()returns a node, And the update ofq->tailis not seen, it is still&q->head,___cds_wfq_node_sync_next(node)will be loop for every if there is no other enqueue.
Good catch ! :-)
Quoted 12 lines
static inline struct
cds_wfq_node*___cds_wfq_first_blocking(struct cds_wfq_queue *q){(quoted diff) +3
-0+ struct cds_wfq_node *ret. if (_cds_wfq_empty(q)) return NULL; ret = ___cds_wfq_node_sync_next(&q->head); + cmm_smp_rmb(); + return ret;
Quoted 1 line
}
However, I think we should add the rmb at the beginning of ___cds_wfq_next_blocking(), so it applies at each "next" call. Otherwise, I think we could end up in a situation where we wait for a NULL next forever in the second of two consecutive ___cds_wfq_next_blocking() calls. I therefore propose:
static inline struct cds_wfq_node * ___cds_wfq_next_blocking(struct cds_wfq_queue *q, struct cds_wfq_node *node)
{
/* Load previous node->next before q->tail */
cmm_smp_rmb();
if (CMM_LOAD_SHARED(q->tail) == node)
return NULL;
return ___cds_wfq_node_sync_next(node);
}Quoted 21 lines
Quoted 21 lines
urcu/static/wfqueue.h +5
-5+ +/* +* ___cds_wfq_dequeue_blocking: dequeue a node from the queue. +* * It is valid to reuse and free a dequeued node immediately. * * No need to go on a waitqueue here, as there is no possible state in which the @@ -120,42 +178,123 @@ ___cds_wfq_dequeue_blocking(struct cds_wfq_queue *q) {struct cds_wfq_node *node, *next; -/* -* Queue is empty if it only contains the dummy node. -*/ -if (q->head == &q->dummy &&CMM_LOAD_SHARED (q->tail) == &q->dummy.next) +if (_cds_wfq_empty (q))return NULL ; - node = q->head;
Quoted 12 lines
urcu/static/wfqueue.h ___cds_wfq_dequeue_blocking(struct cds_wfq_queue *q) · line 189
Quoted 7 lines
urcu/static/wfqueue.h +6
-1- next =___cds_wfq_node_sync_next (node); + node =___cds_wfq_node_sync_next (&q->head); + +if ((next =CMM_LOAD_SHARED (node->next)) ==NULL ) { +/* Load node->next before q->tail */ +cmm_smp_rmb (); +if (CMM_LOAD_SHARED (q->tail) == node) {I don't know why I added this "if" since it is likely true. Could you remove the above 3 lines?
(I remember there is amb()beforeuatomic_cmpxchg()which means thismb()is before the test inuatomic_cmpxchg())
Indeed. I replaced these by a large comment.
Quoted 88 lines
Quoted 88 lines
urcu/static/wfqueue.h +73
-8+/* +* @node is the only node in the queue. +* Try to move the tail to &q->head +*/ +_cds_wfq_node_init (&q->head); +if (uatomic_cmpxchg (&q->tail, node, &q->head) == node) +return node; + } + next =___cds_wfq_node_sync_next (node); + }/* * Move queue head forward. */ - q->head = next; + q->head.next = next; + +return node; +} + +/* +* ___cds_wfq_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. +* caller ensures mutual exclusion of dequeue and splice operations on +* src_q. +*/ +static inline void +___cds_wfq_splice_blocking (struct cds_wfq_queue *dest_q, +struct cds_wfq_queue *src_q) +{ +struct cds_wfq_node *head, *tail; + +if (_cds_wfq_empty (src_q)) +return ; + + head =___cds_wfq_node_sync_next (&src_q->head); +_cds_wfq_node_init (&src_q->head); + +/* +* Memory barrier implied before uatomic_xchg() orders store to +* src_q->head before store to src_q->tail. This is required by +* concurrent enqueue on src_q, which exchanges the tail before +* updating the previous tail's next pointer. +*/ + tail =uatomic_xchg (&src_q->tail, &src_q->head); +/* -* Requeue dummy node if we just dequeued it. +* Append the spliced content of src_q into dest_q. Does not +* require mutual exclusion on dest_q (wait-free). */ -if (node == &q->dummy) { -_cds_wfq_node_init (node); -_cds_wfq_enqueue (q, node); -return ___cds_wfq_dequeue_blocking (q); - } -return node; +___cds_wfq_append (dest_q, head, tail); +} + +/* Locking performed within cds_wfq calls. */ +static inline struct cds_wfq_node * +_cds_wfq_first_blocking (struct cds_wfq_queue *q) +{ +struct cds_wfq_node *retval; +int ret; + + ret =pthread_mutex_lock (&q->lock); +assert (!ret); + retval =___cds_wfq_first_blocking (q); + ret =pthread_mutex_unlock (&q->lock); +assert (!ret); +return retval; +} + +static inline struct cds_wfq_node * +_cds_wfq_next_blocking (struct cds_wfq_queue *q,struct cds_wfq_node *node) +{ +struct cds_wfq_node *retval; +int ret; + + ret =pthread_mutex_lock (&q->lock); +assert (!ret); + retval =___cds_wfq_next_blocking (q, node); + ret =pthread_mutex_unlock (&q->lock);
Quoted 8 lines
urcu/static/wfqueue.h ___cds_wfq_dequeue_blocking(struct cds_wfq_queue *q) · line 271
Quoted 3 lines
urcu/static/wfqueue.h +2
-0+assert (!ret); +return retval; }I reject these
_cds_wfq_first_blocking(),_cds_wfq_next_blocking()andcds_wfq_for_each_blocking(), because the claimed "Locking" makes no sense:
1. It protects nothing in_cds_wfq_next_blocking().
Good point!
Quoted 3 lines
2. There is no "Locking" in the loop body, @node is not dequeued, it will be invalid if some other dequeue it,
and_cds_wfq_next_blocking()results BUG.
Indeed.
So do you recommend we just leave locking to the callers then ? What locking rules would you recommend we document in the API ?
The only API members that would still have implicit locking are thus:
- cds_wfq_dequeue_blocking()
- cds_wfq_splice_blocking()
The only reason I leave them there is that we already expose a "cds_wfq_dequeue_blocking" to users which provides locking, and I don't want to break the API.
Quoted 67 lines
Quoted 67 lines
urcu/static/wfqueue.h +16
-3static inline struct cds_wfq_node *_cds_wfq_dequeue_blocking (struct cds_wfq_queue *q) { -struct cds_wfq_node *retnode; +struct cds_wfq_node *retval;int ret; ret =pthread_mutex_lock (&q->lock);assert (!ret); - retnode =___cds_wfq_dequeue_blocking (q); + retval =___cds_wfq_dequeue_blocking (q); ret =pthread_mutex_unlock (&q->lock);assert (!ret); -return retnode; +return retval; +} + +static inline void +_cds_wfq_splice_blocking (struct cds_wfq_queue *dest_q, +struct cds_wfq_queue *src_q) +{ +int ret; + + ret =pthread_mutex_lock (&src_q->lock); +assert (!ret); +___cds_wfq_splice_blocking (dest_q, src_q); + ret =pthread_mutex_unlock (&src_q->lock); +assert (!ret); }#ifdef __cplusplusurcu/wfqueue.h +3
-3@@ -6,7 +6,8 @@* * Userspace RCU library - Queue with Wait-Free Enqueue/Blocking Dequeue * -* Copyright 2010 - Mathieu Desnoyers <[email protected]> +* Copyright 2010-2012 - Mathieu Desnoyers <[email protected]> +* Copyright 2011-2012 - Lai Jiangshan <[email protected]> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public @@ -25,6 +26,7 @@#include <pthread.h> #include <assert.h> +#include <stdbool.h> #include <urcu/compiler.h> #ifdef __cplusplus@@ -33,8 +35,6 @@ extern "C" {/* * Queue with wait-free enqueue/blocking dequeue. -* This implementation adds a dummy head node when the queue is empty to ensure -* we can always update the queue locklessly. * * Inspired from half-wait-free/half-blocking queue implementation done by * Paul E. McKenney. @@ -45,8 +45,8 @@ struct cds_wfq_node { };
Quoted 9 lines
urcu/wfqueue.h struct cds_wfq_node { · line 51
Quoted 7 lines
urcu/wfqueue.h +2
-2struct cds_wfq_queue { -struct cds_wfq_node *head, **tail; -struct cds_wfq_node dummy;/* Dummy node */ +struct cds_wfq_node head, *tail; +struct cds_wfq_node padding;/* unused */ pthread_mutex_t lock; };Why keep the padding?
This is mainly to keep compatibility with applications already using wfqueue.h. We could indeed make the size of the cds_wfq_queue smaller, but we cannot enlarge it without breaking the API. Therefore, I think it is safe to keep some unused padding rather than shrink the size, if we ever need to put an extra flag or pointer in the structure.
What I can do, though, is to move the padding to the end of the structure, and give it a void * type. Is that OK ?
Thanks,
Mathieu
Signature · 4 lines
Mathieu Desnoyers Operating System Efficiency R&D Consultant EfficiOS Inc. http://www.efficios.com
A message that is not in the archive (sent off-list or not yet received): 20120813201219.GA27637@Krystal
Quoted 416 lines · Mathieu Desnoyers
Hi Lai,
Quoted 69 lines · * Lai Jiangshan ([email protected]) wrote:
Quoted 67 lines · Mathieu Desnoyers
This work is derived from the patch from Lai Jiangshan submitted as "urcu: new wfqueue implementation"
(http://lists.lttng.org/pipermail/lttng-dev/2012-August/018379.html)Its changelog:
Quoted 31 lines
Some guys would be surprised by this fact:
There are already TWO implementations of wfqueue in urcu.The first one is in
urcu/static/wfqueue.h:
1) enqueue: exchange the tail and then updateprevious->next
2) dequeue: wait for first node's next pointer and them shift, a dummy nodeis introduced to avoid the queue->tail become NULL when shift.The second one shares some code with the first one, and the left code are spreading in
urcu-call-rcu-impl.h:
1) enqueue: share with the first one
2) no dequeue operation: and no shift, so it don't need dummy node, Although the dummy node is queued when initialization, but it is removed after the firstdequeue_alloperation incall_rcu_thread().call_rcu_data_free()forgets to handle the dummy node if it is not removed.
3)dequeue_all: record the old head and tail, andqueue->headbecome the special
tail node.(atomic record the tail and change the tail).The second implementation's code are spreading, bad for review, and it is not tested by tests/test_urcu_wfq.
So we need a better implementation avoid the dummy node dancing and can service both generic wfqueue APIs and
dequeue_allAPI for call rcu.The new implementation:
1) enqueue: share with the first one/original implementation.
2) dequeue: shift when node count >= 2, cmpxchg when node count = 1. no dummy node, save memory.
3)dequeue_all: simply setqueue->head.nextto NULL, xchg the tail and return the old head.next.Quoted 2 lines
More implementation details are in the code. tests/test_urcu_wfq will be update in future for testing new APIs.
The patch proposed by Lai brings a very interesting simplification to the single-node handling (which is kept here), and moves all queue handling code away from
call_rcuimplementation, back into the wfqueue code. This has the benefit to allow testing enhancements.I modified it so the API does not expose implementation details to the user (e.g.
___cds_wfq_node_sync_next). I added a "splice" operation and a for loop iterator which should allow wfqueue users to use the list very efficiently both from LGPL/GPL code and from non-LGPL-compatible code.Benchmarks performed on Intel(R) Core(TM) i7-3520M CPU @ 2.90GHz (dual-core, with hyperthreading)
Benchmark invoked:
test_urcu_wfq2 2 10Only did 2 runs, but a small improvement seems to be clear for the dequeue speed:
Before patch:
testdur 10 nr_enqueuers 2 wdelay 0 nr_dequeuers 2 rdur 0 nr_enqueues 136251248 nr_dequeues 54694027 successful enqueues 136251248 successful dequeues 54693904 end_dequeues 81557344 nr_ops 190945275 testdur 10 nr_enqueuers 2 wdelay 0 nr_dequeuers 2 rdur 0 nr_enqueues 137258881 nr_dequeues 54463340 successful enqueues 137258881 successful dequeues 54463238 end_dequeues 82795643 nr_ops 191722221After patch:
Quoted 8 lines
Quoted 4 lines
testdur 10 nr_enqueuers 2 wdelay 0 nr_dequeuers 2 rdur 0 nr_enqueues 138589301 nr_dequeues 56911253 successful enqueues 138589301 successful dequeues 56910916 end_dequeues 81678385 nr_ops 195500554 testdur 10 nr_enqueuers 2 wdelay 0 nr_dequeuers 2 rdur 0 nr_enqueues 139007622 nr_dequeues 57281502 successful enqueues 139007622 successful dequeues 57281348 end_dequeues 81726274 nr_ops 196289124Summary: Number of enqueues is slightly lower,
?!
I see thenr_enqueuesand successful enqueues are both increased after after patch.Oh, you're right. I wrote that summary in a hurry. Sorry about that, will fix.
Quoted 70 lines
Quoted 70 lines
probably due to higher
dequeue rate. Number of dequeue increased. Respective rate change is within 1% (slowdown) for enqueue, 2% (performance improvement) for dequeue. Overall number of operations (dequeue+enqueue) increased with the patch.We can verify that:
successful enqueues - successful dequeues =end_dequeuesFor all runs (ensures correctness: no lost node).
CC Lai Jiangshan <[email protected]>CC Paul McKenney <[email protected]>Signed-off-by Mathieu Desnoyers <[email protected]>---
urcu-call-rcu-impl.h +10
-20@@ -21,6 +21,7 @@*/ #define _GNU_SOURCE +#define _LGPL_SOURCE#include <stdio.h> #include <pthread.h> #include <signal.h> @@ -220,10 +221,7 @@ static void call_rcu_wake_up(struct call_rcu_data *crdp)static void *call_rcu_thread (void *arg) {unsigned long cbcount; -struct cds_wfq_node *cbs; -struct cds_wfq_node **cbs_tail; -struct call_rcu_data *crdp = (struct call_rcu_data *)arg; -struct rcu_head *rhp; +struct call_rcu_data *crdp = (struct call_rcu_data *) arg;int rt = !!(uatomic_read (&crdp->flags) &URCU_CALL_RCU_RT );int ret;@@ -243,35 +241,29 @@ static void *call_rcu_thread(void *arg)cmm_smp_mb (); }for (;;) { -if (&crdp->cbs.head !=_CMM_LOAD_SHARED (crdp->cbs.tail)) { -while ((cbs =_CMM_LOAD_SHARED (crdp->cbs.head)) ==NULL ) -poll (NULL ,0 ,1 ); -_CMM_STORE_SHARED (crdp->cbs.head,NULL ); - cbs_tail = (struct cds_wfq_node **) -uatomic_xchg (&crdp->cbs.tail, &crdp->cbs.head); +struct cds_wfq_queue cbs_tmp; +struct cds_wfq_node *cbs; + +cds_wfq_init (&cbs_tmp); +__cds_wfq_splice_blocking (&cbs_tmp, &crdp->cbs); +if (!cds_wfq_empty (&cbs_tmp)) {synchronize_rcu (); cbcount =0 ; - do { - while (cbs->next == NULL && - &cbs->next != cbs_tail) - poll(NULL, 0, 1); - if (cbs == &crdp->cbs.dummy) { - cbs = cbs->next; - continue; - } - rhp = (struct rcu_head *)cbs; - cbs = cbs->next; +__cds_wfq_for_each_blocking (&cbs_tmp, cbs) { +struct rcu_head *rhp;Quoted 8 lines
urcu-call-rcu-impl.h static void *call_rcu_thread(void *arg) · line 257
Quoted 4 lines
urcu-call-rcu-impl.h +3
-0+ + rhp =caa_container_of (cbs, + struct rcu_head, next); rhp->func (rhp);cbs is freed hear, but it will be used in
__cds_wfq_next_blocking(). Introduce__cds_wfq_for_each_blocking_safe()?Good point! Will do.
Quoted 60 lines
Quoted 46 lines
urcu-call-rcu-impl.h +9
-13cbcount++; - }while (cbs !=NULL ); + }uatomic_sub (&crdp->qlen, cbcount); }if (uatomic_read (&crdp->flags) &URCU_CALL_RCU_STOP )break ;rcu_thread_offline ();if (!rt) { -if (&crdp->cbs.head - ==_CMM_LOAD_SHARED (crdp->cbs.tail)) { +if (cds_wfq_empty (&crdp->cbs)) {call_rcu_wait (crdp);poll (NULL ,0 ,10 );uatomic_dec (&crdp->futex);@@ -625,32 +617,32 @@ void call_rcu(struct rcu_head *head,*/ void call_rcu_data_free (struct call_rcu_data *crdp) { -struct cds_wfq_node *cbs; -struct cds_wfq_node **cbs_tail; -struct cds_wfq_node **cbs_endprev; -if (crdp ==NULL || crdp == default_call_rcu_data) {return ; } +if ((uatomic_read (&crdp->flags) &URCU_CALL_RCU_STOPPED ) ==0 ) {uatomic_or (&crdp->flags,URCU_CALL_RCU_STOP );wake_call_rcu_thread (crdp);while ((uatomic_read (&crdp->flags) &URCU_CALL_RCU_STOPPED ) ==0 )poll (NULL ,0 ,1 ); } -if (&crdp->cbs.head !=_CMM_LOAD_SHARED (crdp->cbs.tail)) { -while ((cbs =_CMM_LOAD_SHARED (crdp->cbs.head)) ==NULL ) -poll (NULL ,0 ,1 ); -_CMM_STORE_SHARED (crdp->cbs.head,NULL ); - cbs_tail = (struct cds_wfq_node **) -uatomic_xchg (&crdp->cbs.tail, &crdp->cbs.head); + +if (!cds_wfq_empty (&crdp->cbs)) { +struct cds_wfq_queue cbs_tmp; + +cds_wfq_init (&cbs_tmp); +__cds_wfq_splice_blocking (&cbs_tmp, &crdp->cbs);urcu-call-rcu-impl.h void call_rcu(struct rcu_head *head, · line 642
Quoted 10 lines
urcu-call-rcu-impl.h +5
-3+/* Create default call rcu data if need be */ (void )get_default_call_rcu_data (); - cbs_endprev = (struct cds_wfq_node **) -uatomic_xchg (&default_call_rcu_data, cbs_tail); - *cbs_endprev = cbs; + +__cds_wfq_splice_blocking (&default_call_rcu_data->cbs, + &cbs_tmp); +Too much code to me,
cbs_tmpis not required here.Quoted 3 lines
/* Create default call rcu data if need be */ (void) get_default_call_rcu_data();+
__cds_wfq_splice_blocking(&default_call_rcu_data->cbs,&crdp->cbs);You're right. I initially thought that the order with the call to
get_default_call_rcu_data()was important, but it rather looks like we callget_default_call_rcu_data()just to ensure that we have somewhere to send the already enqueued callbacks. Indeed, we can merge those as you propose. Will do!Quoted 158 lines
Quoted 146 lines
urcu-call-rcu-impl.h +1
-0uatomic_add (&default_call_rcu_data->qlen,uatomic_read (&crdp->qlen)); +wake_call_rcu_thread (default_call_rcu_data); }urcu/static/wfqueue.h +62
-18@@ -9,7 +9,8 @@* TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See wfqueue.h for linking * dynamically with the userspace rcu library. * -* Copyright 2010 - Mathieu Desnoyers <[email protected]> +* Copyright 2010-2012 - Mathieu Desnoyers <[email protected]> +* Copyright 2011-2012 - Lai Jiangshan <[email protected]> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public @@ -29,6 +30,7 @@#include <pthread.h> #include <assert.h> #include <poll.h> +#include <stdbool.h> #include <urcu/compiler.h> #include <urcu/uatomic.h> @@ -38,11 +40,16 @@ extern "C" {/* * Queue with wait-free enqueue/blocking dequeue. -* This implementation adds a dummy head node when the queue is empty to ensure -* we can always update the queue locklessly. * * Inspired from half-wait-free/half-blocking queue implementation done by * Paul E. McKenney. +* +* Caller must ensure mutual exclusion of queue update operations +* "dequeue" and "splice" source queue. Queue read operations "first" +* and "next" need to be protected against concurrent "dequeue" and +* "splice" (for source queue) by the caller. "enqueue", "splice" +* (destination queue), and "empty" are the only operations that can be +* used without any mutual exclusion. */ #define WFQ_ADAPT_ATTEMPTS 10/* Retry if being set */ @@ -57,31 +64,51 @@ static inline void _cds_wfq_init(struct cds_wfq_queue *q) {int ret; -_cds_wfq_node_init (&q->dummy);/* Set queue head and tail */ - q->head = &q->dummy; - q->tail = &q->dummy.next; +_cds_wfq_node_init (&q->head); + q->tail = &q->head; ret =pthread_mutex_init (&q->lock,NULL );assert (!ret); } -static inline void _cds_wfq_enqueue (struct cds_wfq_queue *q, -struct cds_wfq_node *node) +static inline bool _cds_wfq_empty (struct cds_wfq_queue *q) +{ +/* +* Queue is empty if no node is pointed by q->head.next nor q->tail. +*/ +return CMM_LOAD_SHARED (q->head.next) ==NULL + &&CMM_LOAD_SHARED (q->tail) == &q->head; +} + +static inline void ___cds_wfq_append (struct cds_wfq_queue *q, +struct cds_wfq_node *new_head, +struct cds_wfq_node *new_tail) { -struct cds_wfq_node **old_tail; +struct cds_wfq_node *old_tail;/* -* uatomic_xchg() implicit memory barrier orders earlier stores to data -* structure containing node and setting node->next to NULL before -* publication. +* Implicit memory barrier before uatomic_xchg() orders earlier +* stores to data structure containing node and setting +* node->next to NULL before publication. */ - old_tail =uatomic_xchg (&q->tail, &node->next); + old_tail =uatomic_xchg (&q->tail, new_tail); +/* -* At this point, dequeuers see a NULL old_tail->next, which indicates -* that the queue is being appended to. The following store will append -* "node" to the queue from a dequeuer perspective. +* Implicit memory barrier after uatomic_xchg() orders store to +* q->tail before store to old_tail->next. +* +* At this point, dequeuers see a NULL q->tail->next, which +* indicates that the queue is being appended to. The following +* store will append "node" to the queue from a dequeuer +* perspective. */ -CMM_STORE_SHARED (*old_tail, node); +CMM_STORE_SHARED (old_tail->next, new_head); +} + +static inline void _cds_wfq_enqueue (struct cds_wfq_queue *q, +struct cds_wfq_node *new_tail) +{ +___cds_wfq_append (q, new_tail, new_tail); }/* @@ -100,14 +127,45 @@ ___cds_wfq_node_sync_next(struct cds_wfq_node *node)if (++attempt >=WFQ_ADAPT_ATTEMPTS ) {poll (NULL ,0 ,WFQ_WAIT );/* Wait for 10ms */ attempt =0 ; - }else + }else {caa_cpu_relax (); + } }return next; }/* +* ___cds_wfq_first_blocking: get first node of a queue, without dequeuing. +* +* Mutual exclusion with "dequeue" and "splice" operations must be ensured +* by the caller. +*/ +static inline struct cds_wfq_node * +___cds_wfq_first_blocking (struct cds_wfq_queue *q) +{ +if (_cds_wfq_empty (q)) +return NULL ; +return ___cds_wfq_node_sync_next (&q->head); +} + +/* +* ___cds_wfq_next_blocking: get next node of a queue, without dequeuing. +* urcu/static/wfqueue.h ___cds_wfq_node_sync_next(struct cds_wfq_node *node) · line 164
Quoted 10 lines
urcu/static/wfqueue.h +10
-0+* Mutual exclusion with "dequeue" and "splice" operations must be ensured +* by the caller. +*/ +static inline struct cds_wfq_node * +___cds_wfq_next_blocking (struct cds_wfq_queue *q,struct cds_wfq_node *node) +{ +if (CMM_LOAD_SHARED (q->tail) == node) +return NULL ; +return ___cds_wfq_node_sync_next (node); +}Quoted 6 lines
The same BUG as you told me.
If q has only one node just enqueued by other thread.
but ifq->head.nextis seen,___cds_wfq_first_blocking()returns a node, And the update ofq->tailis not seen, it is still&q->head,___cds_wfq_node_sync_next(node)will be loop for every if there is no other enqueue.Good catch ! :-)
Quoted 12 lines
static inline struct
cds_wfq_node*___cds_wfq_first_blocking(struct cds_wfq_queue *q){(quoted diff) +3
-0+ struct cds_wfq_node *ret. if (_cds_wfq_empty(q)) return NULL; ret = ___cds_wfq_node_sync_next(&q->head); + cmm_smp_rmb(); + return ret;Quoted 1 line
}
Quoted 5 lines
However, I think we should add the rmb at the beginning of
___cds_wfq_next_blocking(), so it applies at each "next" call. Otherwise, I think we could end up in a situation where we wait for a NULL next forever in the second of two consecutive___cds_wfq_next_blocking()calls. I therefore propose:
How this can happen(wait for a NULL next forever in the second ...)?
A rmb is enough to leave the state(simgle node && q->tail == &q->head). "CMM_LOAD_SHARED(q->tail) == node" will be true in the loop except this state.
Quoted 259 lines
static inline struct
cds_wfq_node*___cds_wfq_next_blocking(struct cds_wfq_queue *q, struct cds_wfq_node *node){ /* Load previous node->next before q->tail */ cmm_smp_rmb(); if (CMM_LOAD_SHARED(q->tail) == node) return NULL; return ___cds_wfq_node_sync_next(node); }Quoted 21 lines
Quoted 21 lines
urcu/static/wfqueue.h +5
-5+ +/* +* ___cds_wfq_dequeue_blocking: dequeue a node from the queue. +* * It is valid to reuse and free a dequeued node immediately. * * No need to go on a waitqueue here, as there is no possible state in which the @@ -120,42 +178,123 @@ ___cds_wfq_dequeue_blocking(struct cds_wfq_queue *q) {struct cds_wfq_node *node, *next; -/* -* Queue is empty if it only contains the dummy node. -*/ -if (q->head == &q->dummy &&CMM_LOAD_SHARED (q->tail) == &q->dummy.next) +if (_cds_wfq_empty (q))return NULL ; - node = q->head;Quoted 12 lines
urcu/static/wfqueue.h ___cds_wfq_dequeue_blocking(struct cds_wfq_queue *q) · line 189
Quoted 7 lines
urcu/static/wfqueue.h +6
-1- next =___cds_wfq_node_sync_next (node); + node =___cds_wfq_node_sync_next (&q->head); + +if ((next =CMM_LOAD_SHARED (node->next)) ==NULL ) { +/* Load node->next before q->tail */ +cmm_smp_rmb (); +if (CMM_LOAD_SHARED (q->tail) == node) {I don't know why I added this "if" since it is likely true. Could you remove the above 3 lines?
(I remember there is amb()beforeuatomic_cmpxchg()which means thismb()is before the test inuatomic_cmpxchg())Indeed. I replaced these by a large comment.
Quoted 88 lines
Quoted 88 lines
urcu/static/wfqueue.h +73
-8+/* +* @node is the only node in the queue. +* Try to move the tail to &q->head +*/ +_cds_wfq_node_init (&q->head); +if (uatomic_cmpxchg (&q->tail, node, &q->head) == node) +return node; + } + next =___cds_wfq_node_sync_next (node); + }/* * Move queue head forward. */ - q->head = next; + q->head.next = next; + +return node; +} + +/* +* ___cds_wfq_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. +* caller ensures mutual exclusion of dequeue and splice operations on +* src_q. +*/ +static inline void +___cds_wfq_splice_blocking (struct cds_wfq_queue *dest_q, +struct cds_wfq_queue *src_q) +{ +struct cds_wfq_node *head, *tail; + +if (_cds_wfq_empty (src_q)) +return ; + + head =___cds_wfq_node_sync_next (&src_q->head); +_cds_wfq_node_init (&src_q->head); + +/* +* Memory barrier implied before uatomic_xchg() orders store to +* src_q->head before store to src_q->tail. This is required by +* concurrent enqueue on src_q, which exchanges the tail before +* updating the previous tail's next pointer. +*/ + tail =uatomic_xchg (&src_q->tail, &src_q->head); +/* -* Requeue dummy node if we just dequeued it. +* Append the spliced content of src_q into dest_q. Does not +* require mutual exclusion on dest_q (wait-free). */ -if (node == &q->dummy) { -_cds_wfq_node_init (node); -_cds_wfq_enqueue (q, node); -return ___cds_wfq_dequeue_blocking (q); - } -return node; +___cds_wfq_append (dest_q, head, tail); +} + +/* Locking performed within cds_wfq calls. */ +static inline struct cds_wfq_node * +_cds_wfq_first_blocking (struct cds_wfq_queue *q) +{ +struct cds_wfq_node *retval; +int ret; + + ret =pthread_mutex_lock (&q->lock); +assert (!ret); + retval =___cds_wfq_first_blocking (q); + ret =pthread_mutex_unlock (&q->lock); +assert (!ret); +return retval; +} + +static inline struct cds_wfq_node * +_cds_wfq_next_blocking (struct cds_wfq_queue *q,struct cds_wfq_node *node) +{ +struct cds_wfq_node *retval; +int ret; + + ret =pthread_mutex_lock (&q->lock); +assert (!ret); + retval =___cds_wfq_next_blocking (q, node); + ret =pthread_mutex_unlock (&q->lock);Quoted 8 lines
urcu/static/wfqueue.h ___cds_wfq_dequeue_blocking(struct cds_wfq_queue *q) · line 271
Quoted 3 lines
urcu/static/wfqueue.h +2
-0+assert (!ret); +return retval; }I reject these
_cds_wfq_first_blocking(),_cds_wfq_next_blocking()andcds_wfq_for_each_blocking(), because the claimed "Locking" makes no sense:
1. It protects nothing in_cds_wfq_next_blocking().Good point!
Quoted 3 lines
2. There is no "Locking" in the loop body, @node is not dequeued, it will be invalid if some other dequeue it,
and_cds_wfq_next_blocking()results BUG.Indeed.
So do you recommend we just leave locking to the callers then ? What locking rules would you recommend we document in the API ?
The only API members that would still have implicit locking are thus:
-
cds_wfq_dequeue_blocking()
-cds_wfq_splice_blocking()The only reason I leave them there is that we already expose a "
cds_wfq_dequeue_blocking" to users which provides locking, and I don't want to break the API.Quoted 67 lines
Quoted 67 lines
urcu/static/wfqueue.h +16
-3static inline struct cds_wfq_node *_cds_wfq_dequeue_blocking (struct cds_wfq_queue *q) { -struct cds_wfq_node *retnode; +struct cds_wfq_node *retval;int ret; ret =pthread_mutex_lock (&q->lock);assert (!ret); - retnode =___cds_wfq_dequeue_blocking (q); + retval =___cds_wfq_dequeue_blocking (q); ret =pthread_mutex_unlock (&q->lock);assert (!ret); -return retnode; +return retval; +} + +static inline void +_cds_wfq_splice_blocking (struct cds_wfq_queue *dest_q, +struct cds_wfq_queue *src_q) +{ +int ret; + + ret =pthread_mutex_lock (&src_q->lock); +assert (!ret); +___cds_wfq_splice_blocking (dest_q, src_q); + ret =pthread_mutex_unlock (&src_q->lock); +assert (!ret); }#ifdef __cplusplusurcu/wfqueue.h +3
-3@@ -6,7 +6,8 @@* * Userspace RCU library - Queue with Wait-Free Enqueue/Blocking Dequeue * -* Copyright 2010 - Mathieu Desnoyers <[email protected]> +* Copyright 2010-2012 - Mathieu Desnoyers <[email protected]> +* Copyright 2011-2012 - Lai Jiangshan <[email protected]> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public @@ -25,6 +26,7 @@#include <pthread.h> #include <assert.h> +#include <stdbool.h> #include <urcu/compiler.h> #ifdef __cplusplus@@ -33,8 +35,6 @@ extern "C" {/* * Queue with wait-free enqueue/blocking dequeue. -* This implementation adds a dummy head node when the queue is empty to ensure -* we can always update the queue locklessly. * * Inspired from half-wait-free/half-blocking queue implementation done by * Paul E. McKenney. @@ -45,8 +45,8 @@ struct cds_wfq_node { };Quoted 9 lines
urcu/wfqueue.h struct cds_wfq_node { · line 51
Quoted 7 lines
urcu/wfqueue.h +2
-2struct cds_wfq_queue { -struct cds_wfq_node *head, **tail; -struct cds_wfq_node dummy;/* Dummy node */ +struct cds_wfq_node head, *tail; +struct cds_wfq_node padding;/* unused */ pthread_mutex_t lock; };Why keep the padding?
This is mainly to keep compatibility with applications already using
wfqueue.h. We could indeed make the size of thecds_wfq_queuesmaller, but we cannot enlarge it without breaking the API. Therefore, I think it is safe to keep some unused padding rather than shrink the size, if we ever need to put an extra flag or pointer in the structure.What I can do, though, is to move the padding to the end of the structure, and give it a void * type. Is that OK ?
Thanks,
Mathieu
--
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com
A message that is not in the archive (sent off-list or not yet received): CACvQF51tWXaTdm9jCC=0S2XA+BdzJmR0W2sFVZAuVVZiWX8jAA@mail.gmail.com
Quoted 2 lines · * Lai Jiangshan ([email protected]) wrote:
On Tue, Aug 14, 2012 at 4:12 AM, Mathieu Desnoyers <[email protected]> wrote:
[...]
Quoted 58 lines
Quoted 52 lines
Quoted 29 lines
Quoted 17 lines
/*
+ *___cds_wfq_first_blocking: get first node of a queue, without dequeuing.
+ *
+ * Mutual exclusion with "dequeue" and "splice" operations must be ensured
+ * by the caller.
+ */(quoted diff) +11
-0+static inline struct cds_wfq_node * +___cds_wfq_first_blocking(struct cds_wfq_queue *q) +{ + if (_cds_wfq_empty(q)) + return NULL; + return ___cds_wfq_node_sync_next(&q->head); +} + +/* + * ___cds_wfq_next_blocking: get next node of a queue, without dequeuing. + *Review
Quoted 10 lines
(quoted diff) +10
-0+ * Mutual exclusion with "dequeue" and "splice" operations must be ensured + * by the caller. + */ +static inline struct cds_wfq_node * +___cds_wfq_next_blocking(struct cds_wfq_queue *q, struct cds_wfq_node *node) +{ + if (CMM_LOAD_SHARED(q->tail) == node) + return NULL; + return ___cds_wfq_node_sync_next(node); +}Quoted 6 lines
The same BUG as you told me.
If q has only one node just enqueued by other thread.
but ifq->head.nextis seen,___cds_wfq_first_blocking()returns a node, And the update ofq->tailis not seen, it is still&q->head,___cds_wfq_node_sync_next(node)will be loop for every if there is no other enqueue.Good catch ! :-)
Quoted 12 lines
static inline struct
cds_wfq_node*___cds_wfq_first_blocking(struct cds_wfq_queue *q){(quoted diff) +3
-0+ struct cds_wfq_node *ret. if (_cds_wfq_empty(q)) return NULL; ret = ___cds_wfq_node_sync_next(&q->head); + cmm_smp_rmb(); + return ret;Quoted 1 line
}
Quoted 5 lines
However, I think we should add the rmb at the beginning of
___cds_wfq_next_blocking(), so it applies at each "next" call. Otherwise, I think we could end up in a situation where we wait for a NULL next forever in the second of two consecutive___cds_wfq_next_blocking()calls. I therefore propose:
Quoted 4 lines
How this can happen(wait for a NULL next forever in the second ...)?
A rmb is enough to leave the
state(simgle node && q->tail == &q->head). "CMM_LOAD_SHARED(q->tail)== node" will be true in the loop except this state.
You are indeed right, but I now realise there is a small thing we need to change. I tried a couple of scenarios to convince myself of it, and here are my conclusions:
- In all cases (except for the first node), comparing q->tail with node ensures that we never go beyond the tail. Except for the first node (q->head), we always perform this comparison, so at no point can we reach the nodes through "next" pointers that would be beyond the q->tail node.
- In the case of the first node, we do the comparison with
CMM_LOAD_SHARED(q->head.next) == NULL
&& CMM_LOAD_SHARED(q->tail) == &q->head; in no particular order (checking if empty) before we move to the first node with ___cds_wfq_node_sync_next(&q->head).
The reason why the first node is different from all the others is because we don't use "CMM_LOAD_SHARED(q->tail) == &q->head" alone to distinguish between an empty queue and a queue that contains
nodes: we first check for NULL q->head.next, AFAIU for performance
reasons: whenever possible, we don't want to read the q->tail pointer because it is being heavily updated by the enqueuer.
So assuming the goal beyond the CMM_LOAD_SHARED(q->head.next) == NULL check in the empty queue check is for performance of dequeue vs enqueue, I think it would be appropriate to do a similar optimisation for the ___cds_wfq_next_blocking() case: first check the next pointer of the current node to see if it is NULL. If not, then we know we can fetch the next pointer without comparing our position to the tail (performance optimisation). However, if we do that, we'll need a cmm_smp_rmb() at each ___cds_wfq_next_blocking() invocation.
Here is the resulting code:
/*
* ___cds_wfq_first_blocking: get first node of a queue, without dequeuing. *
* Mutual exclusion with "dequeue" and "splice" operations must be ensured
* by the caller.
*/
static inline struct cds_wfq_node * ___cds_wfq_first_blocking(struct cds_wfq_queue *q)
{
if (_cds_wfq_empty(q))
return NULL;
return ___cds_wfq_node_sync_next(&q->head);
}/*
* ___cds_wfq_next_blocking: get next node of a queue, without dequeuing. *
* Mutual exclusion with "dequeue" and "splice" operations must be ensured
* by the caller.
*/
static inline struct cds_wfq_node * ___cds_wfq_next_blocking(struct cds_wfq_queue *q, struct cds_wfq_node *node)
{
struct cds_wfq_node *next;
if ((next = CMM_LOAD_SHARED(node->next)) != NULL)
return next;
/* Load node->next before q->tail */
cmm_smp_rmb();
if (CMM_LOAD_SHARED(q->tail) == node)
return NULL;
return ___cds_wfq_node_sync_next(node);
}Is my understanding correct ?
Thanks,
Mathieu
Signature · 4 lines
Mathieu Desnoyers Operating System Efficiency R&D Consultant EfficiOS Inc. http://www.efficios.com
A message that is not in the archive (sent off-list or not yet received): 20120814222724.GA9751@Krystal
Quoted 136 lines · * Mathieu Desnoyers ([email protected]) wrote:
Quoted 2 lines · * Lai Jiangshan ([email protected]) wrote:
On Tue, Aug 14, 2012 at 4:12 AM, Mathieu Desnoyers <[email protected]> wrote:
[...]
Quoted 58 lines
Quoted 52 lines
Quoted 29 lines
Quoted 17 lines
/*
+ *___cds_wfq_first_blocking: get first node of a queue, without dequeuing.
+ *
+ * Mutual exclusion with "dequeue" and "splice" operations must be ensured
+ * by the caller.
+ */(quoted diff) +11
-0+static inline struct cds_wfq_node * +___cds_wfq_first_blocking(struct cds_wfq_queue *q) +{ + if (_cds_wfq_empty(q)) + return NULL; + return ___cds_wfq_node_sync_next(&q->head); +} + +/* + * ___cds_wfq_next_blocking: get next node of a queue, without dequeuing. + *Review
Quoted 10 lines
(quoted diff) +10
-0+ * Mutual exclusion with "dequeue" and "splice" operations must be ensured + * by the caller. + */ +static inline struct cds_wfq_node * +___cds_wfq_next_blocking(struct cds_wfq_queue *q, struct cds_wfq_node *node) +{ + if (CMM_LOAD_SHARED(q->tail) == node) + return NULL; + return ___cds_wfq_node_sync_next(node); +}Quoted 6 lines
The same BUG as you told me.
If q has only one node just enqueued by other thread.
but ifq->head.nextis seen,___cds_wfq_first_blocking()returns a node, And the update ofq->tailis not seen, it is still&q->head,___cds_wfq_node_sync_next(node)will be loop for every if there is no other enqueue.Good catch ! :-)
Quoted 12 lines
static inline struct
cds_wfq_node*___cds_wfq_first_blocking(struct cds_wfq_queue *q){(quoted diff) +3
-0+ struct cds_wfq_node *ret. if (_cds_wfq_empty(q)) return NULL; ret = ___cds_wfq_node_sync_next(&q->head); + cmm_smp_rmb(); + return ret;Quoted 1 line
}
Quoted 5 lines
However, I think we should add the rmb at the beginning of
___cds_wfq_next_blocking(), so it applies at each "next" call. Otherwise, I think we could end up in a situation where we wait for a NULL next forever in the second of two consecutive___cds_wfq_next_blocking()calls. I therefore propose:Quoted 4 lines
How this can happen(wait for a NULL next forever in the second ...)?
A rmb is enough to leave the
state(simgle node && q->tail == &q->head). "CMM_LOAD_SHARED(q->tail)== node" will be true in the loop except this state.You are indeed right, but I now realise there is a small thing we need to change. I tried a couple of scenarios to convince myself of it, and here are my conclusions:
- In all cases (except for the first node), comparing
q->tailwith node ensures that we never go beyond the tail. Except for the first node (q->head), we always perform this comparison, so at no point can we reach the nodes through "next" pointers that would be beyond theq->tailnode.- In the case of the first node, we do the comparison with
CMM_LOAD_SHARED(q->head.next) == NULL && CMM_LOAD_SHARED(q->tail) == &q->head;in no particular order (checking if empty) before we move to the first node with
___cds_wfq_node_sync_next(&q->head).
The reason why the first node is different from all the others is because we don't use "CMM_LOAD_SHARED(q->tail)==&q->head" alone to distinguish between an empty queue and a queue that contains
nodes: we first check for NULLq->head.next, AFAIU for performance
reasons: whenever possible, we don't want to read theq->tailpointer because it is being heavily updated by the enqueuer.So assuming the goal beyond the
CMM_LOAD_SHARED(q->head.next)== NULL check in the empty queue check is for performance of dequeue vs enqueue, I think it would be appropriate to do a similar optimisation for the___cds_wfq_next_blocking()case: first check the next pointer of the current node to see if it is NULL. If not, then we know we can fetch the next pointer without comparing our position to the tail (performance optimisation). However, if we do that, we'll need acmm_smp_rmb()at each___cds_wfq_next_blocking()invocation.Here is the resulting code:
/*
*___cds_wfq_first_blocking: get first node of a queue, without dequeuing. *
* Mutual exclusion with "dequeue" and "splice" operations must be ensured
* by the caller.
*/
static inline structcds_wfq_node*___cds_wfq_first_blocking(struct cds_wfq_queue *q){ if (_cds_wfq_empty(q)) return NULL; return ___cds_wfq_node_sync_next(&q->head); }/*
*___cds_wfq_next_blocking: get next node of a queue, without dequeuing. *
* Mutual exclusion with "dequeue" and "splice" operations must be ensured
* by the caller.
*/
static inline structcds_wfq_node*___cds_wfq_next_blocking(struct cds_wfq_queue *q, struct cds_wfq_node *node){ struct cds_wfq_node *next; if ((next = CMM_LOAD_SHARED(node->next)) != NULL) return next; /* Load node->next before q->tail */ cmm_smp_rmb(); if (CMM_LOAD_SHARED(q->tail) == node) return NULL; return ___cds_wfq_node_sync_next(node); }
Quoted 1 line
Is my understanding correct ?
One more thing: I think we need to add cmm_smp_read_barrier_depends() before returning the node pointers in dequeue, first, and next operations. This memory barrier would match the implicit barrier in the enqueue ordering write to the prior tail's next pointer with writes to the node content that would have been performed beforehand by the caller.
Thoughts ?
Thanks,
Mathieu
Quoted 10 lines
Thanks,
Mathieu
Signature · 4 lines
Mathieu Desnoyers Operating System Efficiency R&D Consultant EfficiOS Inc. http://www.efficios.com
Signature · 4 lines
Mathieu Desnoyers Operating System Efficiency R&D Consultant EfficiOS Inc. http://www.efficios.com
A message that is not in the archive (sent off-list or not yet received): 20120814230634.GA11255@Krystal
* Mathieu Desnoyers ([email protected]) wrote:
[...]
Quoted 6 lines
One more thing: I think we need to add
cmm_smp_read_barrier_depends()before returning the node pointers in dequeue, first, and next operations. This memory barrier would match the implicit barrier in the enqueue ordering write to the prior tail's next pointer with writes to the node content that would have been performed beforehand by the caller.
Please note that the only reason why I propose to use "cmm_smp_read_barrier_depends()" before returning nodes in dequeue, first, and next functions rather than a full cmm_smp_mb() is because I fail to see a case where an architecture would reorder a dependent store before the load of the pointer.
I very well imagine a use-case where we have node N within the structure X like the scenario below, where we write to X immediately after dequeue:
CPU 0 CPU 1
store X content
(implicit full memory barrier
implied before uatomic_xchg)
enqueue X->N
dequeue X->N (which barrier here ?) write to X immediately.
If we assume that no reordering of a store that depends on loading a pointer can arrive, then I think we should be good with only a read_barrier_depends() on CPU 1. Otherwise, we might want to put a full barrier there to ensure that we don't race with CPU 0 storing content to X.
Thoughts ?
Thanks,
Mathieu
Signature · 4 lines
Mathieu Desnoyers Operating System Efficiency R&D Consultant EfficiOS Inc. http://www.efficios.com
Quoted 82 lines · Mathieu Desnoyers
Quoted 2 lines · * Lai Jiangshan ([email protected]) wrote:
On Tue, Aug 14, 2012 at 4:12 AM, Mathieu Desnoyers <[email protected]> wrote:
[...]
Quoted 58 lines
Quoted 52 lines
Quoted 29 lines
Quoted 17 lines
/*
+ *___cds_wfq_first_blocking: get first node of a queue, without dequeuing.
+ *
+ * Mutual exclusion with "dequeue" and "splice" operations must be ensured
+ * by the caller.
+ */(quoted diff) +11
-0+static inline struct cds_wfq_node * +___cds_wfq_first_blocking(struct cds_wfq_queue *q) +{ + if (_cds_wfq_empty(q)) + return NULL; + return ___cds_wfq_node_sync_next(&q->head); +} + +/* + * ___cds_wfq_next_blocking: get next node of a queue, without dequeuing. + *Review
Quoted 10 lines
(quoted diff) +10
-0+ * Mutual exclusion with "dequeue" and "splice" operations must be ensured + * by the caller. + */ +static inline struct cds_wfq_node * +___cds_wfq_next_blocking(struct cds_wfq_queue *q, struct cds_wfq_node *node) +{ + if (CMM_LOAD_SHARED(q->tail) == node) + return NULL; + return ___cds_wfq_node_sync_next(node); +}Quoted 6 lines
The same BUG as you told me.
If q has only one node just enqueued by other thread.
but ifq->head.nextis seen,___cds_wfq_first_blocking()returns a node, And the update ofq->tailis not seen, it is still&q->head,___cds_wfq_node_sync_next(node)will be loop for every if there is no other enqueue.Good catch ! :-)
Quoted 12 lines
static inline struct
cds_wfq_node*___cds_wfq_first_blocking(struct cds_wfq_queue *q){(quoted diff) +3
-0+ struct cds_wfq_node *ret. if (_cds_wfq_empty(q)) return NULL; ret = ___cds_wfq_node_sync_next(&q->head); + cmm_smp_rmb(); + return ret;Quoted 1 line
}
Quoted 5 lines
However, I think we should add the rmb at the beginning of
___cds_wfq_next_blocking(), so it applies at each "next" call. Otherwise, I think we could end up in a situation where we wait for a NULL next forever in the second of two consecutive___cds_wfq_next_blocking()calls. I therefore propose:Quoted 4 lines
How this can happen(wait for a NULL next forever in the second ...)?
A rmb is enough to leave the
state(simgle node && q->tail == &q->head). "CMM_LOAD_SHARED(q->tail)== node" will be true in the loop except this state.You are indeed right, but I now realise there is a small thing we need to change. I tried a couple of scenarios to convince myself of it, and here are my conclusions:
- In all cases (except for the first node), comparing
q->tailwith node ensures that we never go beyond the tail. Except for the first node (q->head), we always perform this comparison, so at no point can we reach the nodes through "next" pointers that would be beyond theq->tailnode.- In the case of the first node, we do the comparison with
CMM_LOAD_SHARED(q->head.next) == NULL && CMM_LOAD_SHARED(q->tail) == &q->head;
Quoted 8 lines
in no particular order (checking if empty) before we move to the first node with
___cds_wfq_node_sync_next(&q->head).
The reason why the first node is different from all the others is because we don't use "CMM_LOAD_SHARED(q->tail)==&q->head" alone to distinguish between an empty queue and a queue that contains
nodes: we first check for NULLq->head.next, AFAIU for performance
reasons: whenever possible, we don't want to read theq->tailpointer because it is being heavily updated by the enqueuer.
Is it false sharing?
Access to q->head.next and access to q->tail have the same performance because they are in the same cache line.
Quoted 57 lines
So assuming the goal beyond the
CMM_LOAD_SHARED(q->head.next)== NULL check in the empty queue check is for performance of dequeue vs enqueue, I think it would be appropriate to do a similar optimisation for the___cds_wfq_next_blocking()case: first check the next pointer of the current node to see if it is NULL. If not, then we know we can fetch the next pointer without comparing our position to the tail (performance optimisation). However, if we do that, we'll need acmm_smp_rmb()at each___cds_wfq_next_blocking()invocation.Here is the resulting code:
/*
*___cds_wfq_first_blocking: get first node of a queue, without dequeuing. *
* Mutual exclusion with "dequeue" and "splice" operations must be ensured
* by the caller.
*/
static inline structcds_wfq_node*___cds_wfq_first_blocking(struct cds_wfq_queue *q){ if (_cds_wfq_empty(q)) return NULL; return ___cds_wfq_node_sync_next(&q->head); }/*
*___cds_wfq_next_blocking: get next node of a queue, without dequeuing. *
* Mutual exclusion with "dequeue" and "splice" operations must be ensured
* by the caller.
*/
static inline structcds_wfq_node*___cds_wfq_next_blocking(struct cds_wfq_queue *q, struct cds_wfq_node *node){ struct cds_wfq_node *next; if ((next = CMM_LOAD_SHARED(node->next)) != NULL) return next; /* Load node->next before q->tail */ cmm_smp_rmb(); if (CMM_LOAD_SHARED(q->tail) == node) return NULL; return ___cds_wfq_node_sync_next(node); }Is my understanding correct ?
Thanks,
Mathieu
--
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com
A message that is not in the archive (sent off-list or not yet received): CACvQF51LxhB+H8X9MqmDTS4xmGUyttTMNk66=NLvqsBkw00qXg@mail.gmail.com
Quoted 93 lines · * Lai Jiangshan ([email protected]) wrote:
Quoted 82 lines · Mathieu Desnoyers
Quoted 2 lines · * Lai Jiangshan ([email protected]) wrote:
On Tue, Aug 14, 2012 at 4:12 AM, Mathieu Desnoyers <[email protected]> wrote:
[...]
Quoted 58 lines
Quoted 52 lines
Quoted 29 lines
Quoted 17 lines
/*
+ *___cds_wfq_first_blocking: get first node of a queue, without dequeuing.
+ *
+ * Mutual exclusion with "dequeue" and "splice" operations must be ensured
+ * by the caller.
+ */(quoted diff) +11
-0+static inline struct cds_wfq_node * +___cds_wfq_first_blocking(struct cds_wfq_queue *q) +{ + if (_cds_wfq_empty(q)) + return NULL; + return ___cds_wfq_node_sync_next(&q->head); +} + +/* + * ___cds_wfq_next_blocking: get next node of a queue, without dequeuing. + *Review
Quoted 10 lines
(quoted diff) +10
-0+ * Mutual exclusion with "dequeue" and "splice" operations must be ensured + * by the caller. + */ +static inline struct cds_wfq_node * +___cds_wfq_next_blocking(struct cds_wfq_queue *q, struct cds_wfq_node *node) +{ + if (CMM_LOAD_SHARED(q->tail) == node) + return NULL; + return ___cds_wfq_node_sync_next(node); +}Quoted 6 lines
The same BUG as you told me.
If q has only one node just enqueued by other thread.
but ifq->head.nextis seen,___cds_wfq_first_blocking()returns a node, And the update ofq->tailis not seen, it is still&q->head,___cds_wfq_node_sync_next(node)will be loop for every if there is no other enqueue.Good catch ! :-)
Quoted 12 lines
static inline struct
cds_wfq_node*___cds_wfq_first_blocking(struct cds_wfq_queue *q){(quoted diff) +3
-0+ struct cds_wfq_node *ret. if (_cds_wfq_empty(q)) return NULL; ret = ___cds_wfq_node_sync_next(&q->head); + cmm_smp_rmb(); + return ret;Quoted 1 line
}
Quoted 5 lines
However, I think we should add the rmb at the beginning of
___cds_wfq_next_blocking(), so it applies at each "next" call. Otherwise, I think we could end up in a situation where we wait for a NULL next forever in the second of two consecutive___cds_wfq_next_blocking()calls. I therefore propose:Quoted 4 lines
How this can happen(wait for a NULL next forever in the second ...)?
A rmb is enough to leave the
state(simgle node && q->tail == &q->head). "CMM_LOAD_SHARED(q->tail)== node" will be true in the loop except this state.You are indeed right, but I now realise there is a small thing we need to change. I tried a couple of scenarios to convince myself of it, and here are my conclusions:
- In all cases (except for the first node), comparing
q->tailwith node ensures that we never go beyond the tail. Except for the first node (q->head), we always perform this comparison, so at no point can we reach the nodes through "next" pointers that would be beyond theq->tailnode.- In the case of the first node, we do the comparison with
CMM_LOAD_SHARED(q->head.next) == NULL && CMM_LOAD_SHARED(q->tail) == &q->head;Quoted 8 lines
in no particular order (checking if empty) before we move to the first node with
___cds_wfq_node_sync_next(&q->head).
The reason why the first node is different from all the others is because we don't use "CMM_LOAD_SHARED(q->tail)==&q->head" alone to distinguish between an empty queue and a queue that contains
nodes: we first check for NULLq->head.next, AFAIU for performance
reasons: whenever possible, we don't want to read theq->tailpointer because it is being heavily updated by the enqueuer.
Quoted 3 lines
Is it false sharing?
Access toq->head.nextand access toq->tailhave the same performance because they are in the same cache line.
Yes! you are right! And a quick benchmark confirms it:
with head and tail on same cache line:
SUMMARY /home/compudj/doc/userspace-rcu/tests/.libs/lt-test_urcu_wfq testdur 10 nr_enqueuers 1 wdelay 0 nr_dequeuers 1 rdur 0 nr_enqueues 100833595 nr_dequeues 88647134 successful enqueues 100833595 successful dequeues 88646898 end_dequeues 12186697 nr_ops 189480729
with a 256 bytes padding between head and tail, keeping the mutex on the "head" cache line:
SUMMARY /home/compudj/doc/userspace-rcu/tests/.libs/lt-test_urcu_wfq testdur 10 nr_enqueuers 1 wdelay 0 nr_dequeuers 1 rdur 0 nr_enqueues 228992829 nr_dequeues 228921791 successful enqueues 228992829 successful dequeues 228921367 end_dequeues 71462 nr_ops 457914620
enqueue: 127% speedup
dequeue: 158% speedup
That is indeed a _really_ huge difference. However, to get this, we would have to increase the size of struct cds_wfq_queue beyond its current size, which would break API compatibility. Any idea on how to best do this without causing incompatibility would be welcome.
Thanks!
Mathieu
Quoted 58 lines
Quoted 58 lines
So assuming the goal beyond the
CMM_LOAD_SHARED(q->head.next)== NULL check in the empty queue check is for performance of dequeue vs enqueue, I think it would be appropriate to do a similar optimisation for the___cds_wfq_next_blocking()case: first check the next pointer of the current node to see if it is NULL. If not, then we know we can fetch the next pointer without comparing our position to the tail (performance optimisation). However, if we do that, we'll need acmm_smp_rmb()at each___cds_wfq_next_blocking()invocation.Here is the resulting code:
/*
*___cds_wfq_first_blocking: get first node of a queue, without dequeuing. *
* Mutual exclusion with "dequeue" and "splice" operations must be ensured
* by the caller.
*/
static inline structcds_wfq_node*___cds_wfq_first_blocking(struct cds_wfq_queue *q){ if (_cds_wfq_empty(q)) return NULL; return ___cds_wfq_node_sync_next(&q->head); }/*
*___cds_wfq_next_blocking: get next node of a queue, without dequeuing. *
* Mutual exclusion with "dequeue" and "splice" operations must be ensured
* by the caller.
*/
static inline structcds_wfq_node*___cds_wfq_next_blocking(struct cds_wfq_queue *q, struct cds_wfq_node *node){ struct cds_wfq_node *next; if ((next = CMM_LOAD_SHARED(node->next)) != NULL) return next; /* Load node->next before q->tail */ cmm_smp_rmb(); if (CMM_LOAD_SHARED(q->tail) == node) return NULL; return ___cds_wfq_node_sync_next(node); }Is my understanding correct ?
Thanks,
Mathieu
--
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com
Signature · 4 lines
Mathieu Desnoyers Operating System Efficiency R&D Consultant EfficiOS Inc. http://www.efficios.com
A message that is not in the archive (sent off-list or not yet received): 20120815125209.GA11737@Krystal
Quoted 16 lines
Quoted 4 lines
Is it false sharing?
Access toq->head.nextand access toq->tailhave the same performance because they are in the same cache line.Yes! you are right! And a quick benchmark confirms it:
with head and tail on same cache line:
SUMMARY
/home/compudj/doc/userspace-rcu/tests/.libs/lt-test_urcu_wfqtestdur 10nr_enqueuers1 wdelay 0nr_dequeuers1 rdur 0nr_enqueues100833595nr_dequeues88647134 successful enqueues 100833595 successful dequeues 88646898end_dequeues12186697nr_ops189480729with a 256 bytes padding between head and tail, keeping the mutex on the "head" cache line:
SUMMARY
/home/compudj/doc/userspace-rcu/tests/.libs/lt-test_urcu_wfqtestdur 10nr_enqueuers1 wdelay 0nr_dequeuers1 rdur 0nr_enqueues228992829nr_dequeues228921791 successful enqueues 228992829 successful dequeues 228921367end_dequeues71462nr_ops457914620
Quoted 8 lines
enqueue: 127% speedup
dequeue: 158% speedupThat is indeed a _really_ huge difference. However, to get this, we would have to increase the size of struct
cds_wfq_queuebeyond its current size, which would break API compatibility. Any idea on how to best do this without causing incompatibility would be welcome.
choice 1) two set of APIs?(cache-line-opt and none-cache-line-opt), many users don't need the cache-line-opt.
choice 2) Just break the compatibility for NONE-LGPL. I think NONE-LGPL-user of it is rare. And current version of urcu <1.0, I don't like too much burden when <1.0.
thanks,
Lai
A message that is not in the archive (sent off-list or not yet received): CACvQF52Z-wyz=LPOjZyJpfhHdF8cAtr8A5bWABM=3A95=q8Egg@mail.gmail.com
We need the smallest patch at first, all other things left in disscussion. no free_each(), simple changes.
thanks,
Lai
Quoted 34 lines · Lai Jiangshan
Quoted 16 lines
Quoted 4 lines
Is it false sharing?
Access toq->head.nextand access toq->tailhave the same performance because they are in the same cache line.Yes! you are right! And a quick benchmark confirms it:
with head and tail on same cache line:
SUMMARY
/home/compudj/doc/userspace-rcu/tests/.libs/lt-test_urcu_wfqtestdur 10nr_enqueuers1 wdelay 0nr_dequeuers1 rdur 0nr_enqueues100833595nr_dequeues88647134 successful enqueues 100833595 successful dequeues 88646898end_dequeues12186697nr_ops189480729with a 256 bytes padding between head and tail, keeping the mutex on the "head" cache line:
SUMMARY
/home/compudj/doc/userspace-rcu/tests/.libs/lt-test_urcu_wfqtestdur 10nr_enqueuers1 wdelay 0nr_dequeuers1 rdur 0nr_enqueues228992829nr_dequeues228921791 successful enqueues 228992829 successful dequeues 228921367end_dequeues71462nr_ops457914620Quoted 8 lines
enqueue: 127% speedup
dequeue: 158% speedupThat is indeed a _really_ huge difference. However, to get this, we would have to increase the size of struct
cds_wfq_queuebeyond its current size, which would break API compatibility. Any idea on how to best do this without causing incompatibility would be welcome.choice 1) two set of APIs?(cache-line-opt and none-cache-line-opt), many users don't need the cache-line-opt.
choice 2) Just break the compatibility for NONE-LGPL. I think NONE-LGPL-user of it is rare. And current version of urcu <1.0, I don't like too much burden when <1.0.thanks,
Lai
Quoted 25 lines · * Lai Jiangshan ([email protected]) wrote:
Quoted 16 lines
Quoted 4 lines
Is it false sharing?
Access toq->head.nextand access toq->tailhave the same performance because they are in the same cache line.Yes! you are right! And a quick benchmark confirms it:
with head and tail on same cache line:
SUMMARY
/home/compudj/doc/userspace-rcu/tests/.libs/lt-test_urcu_wfqtestdur 10nr_enqueuers1 wdelay 0nr_dequeuers1 rdur 0nr_enqueues100833595nr_dequeues88647134 successful enqueues 100833595 successful dequeues 88646898end_dequeues12186697nr_ops189480729with a 256 bytes padding between head and tail, keeping the mutex on the "head" cache line:
SUMMARY
/home/compudj/doc/userspace-rcu/tests/.libs/lt-test_urcu_wfqtestdur 10nr_enqueuers1 wdelay 0nr_dequeuers1 rdur 0nr_enqueues228992829nr_dequeues228921791 successful enqueues 228992829 successful dequeues 228921367end_dequeues71462nr_ops457914620Quoted 8 lines
enqueue: 127% speedup
dequeue: 158% speedupThat is indeed a _really_ huge difference. However, to get this, we would have to increase the size of struct
cds_wfq_queuebeyond its current size, which would break API compatibility. Any idea on how to best do this without causing incompatibility would be welcome.
Quoted 5 lines
choice 1) two set of APIs?(cache-line-opt and none-cache-line-opt), many users don't need the cache-line-opt.
choice 2) Just break the compatibility for NONE-LGPL. I think NONE-LGPL-user of it is rare. And current version of urcu <1.0, I don't like too much burden when <1.0.
Hi Lai,
Keeping backward compatibility for a while, and gradually deprecating APIs, is very important for userspace RCU in my opinion: failure to provide this is the first thing that will discourage users from using the library. It is also the first thing that will turn away package maintainers, and distributions, from this project.
I sent 2 patches that implement a compatibility ABI and the new ABI in a way that will let us continue development on wfqueue without having to have legacy code in our way: I plan to move all the legacy ABI to wfqueue0.c, and have a fresh start in terms of ABI for the new wfqueue. That way, we can eventually remove the wfqueue v0 ABI when we decide to bump the library version number (and break compatibility).
Feedback is welcome,
Thanks!
Mathieu
Signature · 4 lines
Mathieu Desnoyers Operating System Efficiency R&D Consultant EfficiOS Inc. http://www.efficios.com
A message that is not in the archive (sent off-list or not yet received): CACvQF53D1u4FYcY9ab5J6XRi0E4Z+5mM++DgpVR+Vumqvr15wQ@mail.gmail.com
Quoted 2 lines · * Lai Jiangshan ([email protected]) wrote:
We need the smallest patch at first, all other things left in disscussion. no
free_each(), simple changes.
I guess you mean for_each().
If we limit ourself to the versions where the user is doing the locking, I don't think there were any issues left.
In the version I provided (in the last series of 2 patches), I took care of all issues that were raised in our email discussion.
I prefer to introduce these new API members all in one go, mainly because I really don't want to add new API members (exposed through the public API) and then remove them afterward when we notice that they expose too many details. I think the current splice, dequeue, next, and first API members allow any user to do the kind of use-case that call_rcu is
doing: this lets us achieve your original goal of not duplicating the code everywhere.
If you still notice issues with __cds_wfq_for_each_blocking() and __cds_wfq_for_each_blocking_safe() in the last patch, please let me know,
Thanks !
Mathieu
Quoted 39 lines
thanks,
LaiQuoted 34 lines · Lai Jiangshan
Quoted 16 lines
Quoted 4 lines
Is it false sharing?
Access toq->head.nextand access toq->tailhave the same performance because they are in the same cache line.Yes! you are right! And a quick benchmark confirms it:
with head and tail on same cache line:
SUMMARY
/home/compudj/doc/userspace-rcu/tests/.libs/lt-test_urcu_wfqtestdur 10nr_enqueuers1 wdelay 0nr_dequeuers1 rdur 0nr_enqueues100833595nr_dequeues88647134 successful enqueues 100833595 successful dequeues 88646898end_dequeues12186697nr_ops189480729with a 256 bytes padding between head and tail, keeping the mutex on the "head" cache line:
SUMMARY
/home/compudj/doc/userspace-rcu/tests/.libs/lt-test_urcu_wfqtestdur 10nr_enqueuers1 wdelay 0nr_dequeuers1 rdur 0nr_enqueues228992829nr_dequeues228921791 successful enqueues 228992829 successful dequeues 228921367end_dequeues71462nr_ops457914620Quoted 8 lines
enqueue: 127% speedup
dequeue: 158% speedupThat is indeed a _really_ huge difference. However, to get this, we would have to increase the size of struct
cds_wfq_queuebeyond its current size, which would break API compatibility. Any idea on how to best do this without causing incompatibility would be welcome.choice 1) two set of APIs?(cache-line-opt and none-cache-line-opt), many users don't need the cache-line-opt.
choice 2) Just break the compatibility for NONE-LGPL. I think NONE-LGPL-user of it is rare. And current version of urcu <1.0, I don't like too much burden when <1.0.thanks,
Lai
Signature · 4 lines
Mathieu Desnoyers Operating System Efficiency R&D Consultant EfficiOS Inc. http://www.efficios.com
A message that is not in the archive (sent off-list or not yet received): 20120816211155.GA3487@Krystal
Patches are sent? I'm in vacation, I can't recieve them.
__cds_wfq_for_each_blocking_safe() sounds like we can delete any node in for_each(), it breaks the mantics of QUEUE. and the users and we don't know how to delete a node. for_each_safe() is just a special things for call_rcu_thread() which destroy all node in a tmp queue.
so I prefer to use sync_next() in the call_rcu_thread() as my original patch now, it is also very simple. the for_each_safe() sugar is not required in call_rcu_thread().
=====
I think we need dequeue_all() API.splice() = dequeue_all() + enqueue(list).dequeue_all(): 1 xchg()enqueue(): 1 xchg()splice(): 2 xchg()
And if no dequeue_all(), the users have to use splice()+tmp instead, not straightly.
So it is worth to have dequeue_all().
=====
the first smallest patch can be:
equeue(),sync_next(),dequeue(),dequeue_all(),splice().
Leave all the other things in future patches.
Thanks,
Lai.
"other things" list in my view:
What memroy barrier should be add to dequeue()? for_each() APIs and mantics.
cache-line-opt queue or current(none-cache-line-opt) or both and related compatibility thing.
Quoted 73 lines · Mathieu Desnoyers
Quoted 2 lines · * Lai Jiangshan ([email protected]) wrote:
We need the smallest patch at first, all other things left in disscussion. no
free_each(), simple changes.I guess you mean
for_each().If we limit ourself to the versions where the user is doing the locking, I don't think there were any issues left.
In the version I provided (in the last series of 2 patches), I took care of all issues that were raised in our email discussion.
I prefer to introduce these new API members all in one go, mainly because I really don't want to add new API members (exposed through the public API) and then remove them afterward when we notice that they expose too many details. I think the current splice, dequeue, next, and first API members allow any user to do the kind of use-case that
call_rcuis
doing: this lets us achieve your original goal of not duplicating the code everywhere.If you still notice issues with
__cds_wfq_for_each_blocking()and__cds_wfq_for_each_blocking_safe()in the last patch, please let me know,Thanks !
Mathieu
Quoted 39 lines
thanks,
LaiQuoted 34 lines · Lai Jiangshan
Quoted 16 lines
Quoted 4 lines
Is it false sharing?
Access toq->head.nextand access toq->tailhave the same performance because they are in the same cache line.Yes! you are right! And a quick benchmark confirms it:
with head and tail on same cache line:
SUMMARY
/home/compudj/doc/userspace-rcu/tests/.libs/lt-test_urcu_wfqtestdur 10nr_enqueuers1 wdelay 0nr_dequeuers1 rdur 0nr_enqueues100833595nr_dequeues88647134 successful enqueues 100833595 successful dequeues 88646898end_dequeues12186697nr_ops189480729with a 256 bytes padding between head and tail, keeping the mutex on the "head" cache line:
SUMMARY
/home/compudj/doc/userspace-rcu/tests/.libs/lt-test_urcu_wfqtestdur 10nr_enqueuers1 wdelay 0nr_dequeuers1 rdur 0nr_enqueues228992829nr_dequeues228921791 successful enqueues 228992829 successful dequeues 228921367end_dequeues71462nr_ops457914620Quoted 8 lines
enqueue: 127% speedup
dequeue: 158% speedupThat is indeed a _really_ huge difference. However, to get this, we would have to increase the size of struct
cds_wfq_queuebeyond its current size, which would break API compatibility. Any idea on how to best do this without causing incompatibility would be welcome.choice 1) two set of APIs?(cache-line-opt and none-cache-line-opt), many users don't need the cache-line-opt.
choice 2) Just break the compatibility for NONE-LGPL. I think NONE-LGPL-user of it is rare. And current version of urcu <1.0, I don't like too much burden when <1.0.thanks,
Lai--
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com