urcu/static/wfcqueue.h

   1 #ifndef _URCU_WFCQUEUE_STATIC_H
   2 #define _URCU_WFCQUEUE_STATIC_H
   3
   4 /*
   5  * urcu/static/wfcqueue.h
   6  *
   7  * Userspace RCU library - Concurrent Queue with Wait-Free Enqueue/Blocking Dequeue
   8  *
   9  * TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See urcu/wfcqueue.h for
  10  * linking dynamically with the userspace rcu library.
  11  *
  12  * Copyright 2010-2012 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  13  * Copyright 2011-2012 - Lai Jiangshan <laijs@cn.fujitsu.com>
  14  *
  15  * This library is free software; you can redistribute it and/or
  16  * modify it under the terms of the GNU Lesser General Public
  17  * License as published by the Free Software Foundation; either
  18  * version 2.1 of the License, or (at your option) any later version.
  19  *
  20  * This library is distributed in the hope that it will be useful,
  21  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  22  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  23  * Lesser General Public License for more details.
  24  *
  25  * You should have received a copy of the GNU Lesser General Public
  26  * License along with this library; if not, write to the Free Software
  27  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  28  */
  29
  30 #include <pthread.h>
  31 #include <assert.h>
  32 #include <poll.h>
  33 #include <stdbool.h>
  34 #include <urcu/compiler.h>
  35 #include <urcu/uatomic.h>
  36
  37 #ifdef __cplusplus
  38 extern "C" {
  39 #endif
  40
  41 /*
  42  * Concurrent queue with wait-free enqueue/blocking dequeue.
  43  *
  44  * This queue has been designed and implemented collaboratively by
  45  * Mathieu Desnoyers and Lai Jiangshan. Inspired from
  46  * half-wait-free/half-blocking queue implementation done by Paul E.
  47  * McKenney.
  48  *
  49  * Mutual exclusion of cds_wfcq_* / __cds_wfcq_* API
  50  *
  51  * Synchronization table:
  52  *
  53  * External synchronization techniques described in the API below is
  54  * required between pairs marked with "X". No external synchronization
  55  * required between pairs marked with "-".
  56  *
  57  * Legend:
  58  * [1] cds_wfcq_enqueue
  59  * [2] __cds_wfcq_splice (destination queue)
  60  * [3] __cds_wfcq_dequeue
  61  * [4] __cds_wfcq_splice (source queue)
  62  * [5] __cds_wfcq_first
  63  * [6] __cds_wfcq_next
  64  *
  65  *     [1] [2] [3] [4] [5] [6]
  66  * [1]  -   -   -   -   -   -
  67  * [2]  -   -   -   -   -   -
  68  * [3]  -   -   X   X   X   X
  69  * [4]  -   -   X   -   X   X
  70  * [5]  -   -   X   X   -   -
  71  * [6]  -   -   X   X   -   -
  72  *
  73  * Mutual exclusion can be ensured by holding cds_wfcq_dequeue_lock().
  74  *
  75  * For convenience, cds_wfcq_dequeue_blocking() and
  76  * cds_wfcq_splice_blocking() hold the dequeue lock.
  77  *
  78  * Besides locking, mutual exclusion of dequeue, splice and iteration
  79  * can be ensured by performing all of those operations from a single
  80  * thread, without requiring any lock.
  81  */
  82
  83 #define WFCQ_ADAPT_ATTEMPTS             10      /* Retry if being set */
  84 #define WFCQ_WAIT                       10      /* Wait 10 ms if being set */
  85
  86 /*
  87  * cds_wfcq_node_init: initialize wait-free queue node.
  88  */
  89 static inline void _cds_wfcq_node_init(struct cds_wfcq_node *node)
  90 {
  91         node->next = NULL;
  92 }
  93
  94 /*
  95  * cds_wfcq_init: initialize wait-free queue.
  96  */
  97 static inline void _cds_wfcq_init(struct cds_wfcq_head *head,
  98                 struct cds_wfcq_tail *tail)
  99 {
 100         int ret;
 101
 102         /* Set queue head and tail */
 103         _cds_wfcq_node_init(&head->node);
 104         tail->p = &head->node;
 105         ret = pthread_mutex_init(&head->lock, NULL);
 106         assert(!ret);
 107 }
 108
 109 /*
 110  * cds_wfcq_empty: return whether wait-free queue is empty.
 111  *
 112  * No memory barrier is issued. No mutual exclusion is required.
 113  *
 114  * We perform the test on head->node.next to check if the queue is
 115  * possibly empty, but we confirm this by checking if the tail pointer
 116  * points to the head node because the tail pointer is the linearisation
 117  * point of the enqueuers. Just checking the head next pointer could
 118  * make a queue appear empty if an enqueuer is preempted for a long time
 119  * between xchg() and setting the previous node's next pointer.
 120  */
 121 static inline bool _cds_wfcq_empty(struct cds_wfcq_head *head,
 122                 struct cds_wfcq_tail *tail)
 123 {
 124         /*
 125          * Queue is empty if no node is pointed by head->node.next nor
 126          * tail->p. Even though the tail->p check is sufficient to find
 127          * out of the queue is empty, we first check head->node.next as a
 128          * common case to ensure that dequeuers do not frequently access
 129          * enqueuer's tail->p cache line.
 130          */
 131         return CMM_LOAD_SHARED(head->node.next) == NULL
 132                 && CMM_LOAD_SHARED(tail->p) == &head->node;
 133 }
 134
 135 static inline void _cds_wfcq_dequeue_lock(struct cds_wfcq_head *head,
 136                 struct cds_wfcq_tail *tail)
 137 {
 138         int ret;
 139
 140         ret = pthread_mutex_lock(&head->lock);
 141         assert(!ret);
 142 }
 143
 144 static inline void _cds_wfcq_dequeue_unlock(struct cds_wfcq_head *head,
 145                 struct cds_wfcq_tail *tail)
 146 {
 147         int ret;
 148
 149         ret = pthread_mutex_unlock(&head->lock);
 150         assert(!ret);
 151 }
 152
 153 static inline bool ___cds_wfcq_append(struct cds_wfcq_head *head,
 154                 struct cds_wfcq_tail *tail,
 155                 struct cds_wfcq_node *new_head,
 156                 struct cds_wfcq_node *new_tail)
 157 {
 158         struct cds_wfcq_node *old_tail;
 159
 160         /*
 161          * Implicit memory barrier before uatomic_xchg() orders earlier
 162          * stores to data structure containing node and setting
 163          * node->next to NULL before publication.
 164          */
 165         old_tail = uatomic_xchg(&tail->p, new_tail);
 166
 167         /*
 168          * Implicit memory barrier after uatomic_xchg() orders store to
 169          * q->tail before store to old_tail->next.
 170          *
 171          * At this point, dequeuers see a NULL tail->p->next, which
 172          * indicates that the queue is being appended to. The following
 173          * store will append "node" to the queue from a dequeuer
 174          * perspective.
 175          */
 176         CMM_STORE_SHARED(old_tail->next, new_head);
 177         /*
 178          * Return false if queue was empty prior to adding the node,
 179          * else return true.
 180          */
 181         return old_tail != &head->node;
 182 }
 183
 184 /*
 185  * cds_wfcq_enqueue: enqueue a node into a wait-free queue.
 186  *
 187  * Issues a full memory barrier before enqueue. No mutual exclusion is
 188  * required.
 189  *
 190  * Returns false if the queue was empty prior to adding the node.
 191  * Returns true otherwise.
 192  */
 193 static inline bool _cds_wfcq_enqueue(struct cds_wfcq_head *head,
 194                 struct cds_wfcq_tail *tail,
 195                 struct cds_wfcq_node *new_tail)
 196 {
 197         return ___cds_wfcq_append(head, tail, new_tail, new_tail);
 198 }
 199
 200 /*
 201  * ___cds_wfcq_busy_wait: adaptative busy-wait.
 202  *
 203  * Returns 1 if nonblocking and needs to block, 0 otherwise.
 204  */
 205 static inline bool
 206 ___cds_wfcq_busy_wait(int *attempt, int blocking)
 207 {
 208         if (!blocking)
 209                 return 1;
 210         if (++(*attempt) >= WFCQ_ADAPT_ATTEMPTS) {
 211                 poll(NULL, 0, WFCQ_WAIT);       /* Wait for 10ms */
 212                 *attempt = 0;
 213         } else {
 214                 caa_cpu_relax();
 215         }
 216         return 0;
 217 }
 218
 219 /*
 220  * Waiting for enqueuer to complete enqueue and return the next node.
 221  */
 222 static inline struct cds_wfcq_node *
 223 ___cds_wfcq_node_sync_next(struct cds_wfcq_node *node, int blocking)
 224 {
 225         struct cds_wfcq_node *next;
 226         int attempt = 0;
 227
 228         /*
 229          * Adaptative busy-looping waiting for enqueuer to complete enqueue.
 230          */
 231         while ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 232                 if (___cds_wfcq_busy_wait(&attempt, blocking))
 233                         return CDS_WFCQ_WOULDBLOCK;
 234         }
 235
 236         return next;
 237 }
 238
 239 static inline struct cds_wfcq_node *
 240 ___cds_wfcq_first(struct cds_wfcq_head *head,
 241                 struct cds_wfcq_tail *tail,
 242                 int blocking)
 243 {
 244         struct cds_wfcq_node *node;
 245
 246         if (_cds_wfcq_empty(head, tail))
 247                 return NULL;
 248         node = ___cds_wfcq_node_sync_next(&head->node, blocking);
 249         /* Load head->node.next before loading node's content */
 250         cmm_smp_read_barrier_depends();
 251         return node;
 252 }
 253
 254 /*
 255  * __cds_wfcq_first_blocking: get first node of a queue, without dequeuing.
 256  *
 257  * Content written into the node before enqueue is guaranteed to be
 258  * consistent, but no other memory ordering is ensured.
 259  * Dequeue/splice/iteration mutual exclusion should be ensured by the
 260  * caller.
 261  *
 262  * Used by for-like iteration macros in urcu/wfqueue.h:
 263  * __cds_wfcq_for_each_blocking()
 264  * __cds_wfcq_for_each_blocking_safe()
 265  *
 266  * Returns NULL if queue is empty, first node otherwise.
 267  */
 268 static inline struct cds_wfcq_node *
 269 ___cds_wfcq_first_blocking(struct cds_wfcq_head *head,
 270                 struct cds_wfcq_tail *tail)
 271 {
 272         return ___cds_wfcq_first(head, tail, 1);
 273 }
 274
 275
 276 /*
 277  * __cds_wfcq_first_nonblocking: get first node of a queue, without dequeuing.
 278  *
 279  * Same as __cds_wfcq_first_blocking, but returns CDS_WFCQ_WOULDBLOCK if
 280  * it needs to block.
 281  */
 282 static inline struct cds_wfcq_node *
 283 ___cds_wfcq_first_nonblocking(struct cds_wfcq_head *head,
 284                 struct cds_wfcq_tail *tail)
 285 {
 286         return ___cds_wfcq_first(head, tail, 0);
 287 }
 288
 289 static inline struct cds_wfcq_node *
 290 ___cds_wfcq_next(struct cds_wfcq_head *head,
 291                 struct cds_wfcq_tail *tail,
 292                 struct cds_wfcq_node *node,
 293                 int blocking)
 294 {
 295         struct cds_wfcq_node *next;
 296
 297         /*
 298          * Even though the following tail->p check is sufficient to find
 299          * out if we reached the end of the queue, we first check
 300          * node->next as a common case to ensure that iteration on nodes
 301          * do not frequently access enqueuer's tail->p cache line.
 302          */
 303         if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 304                 /* Load node->next before tail->p */
 305                 cmm_smp_rmb();
 306                 if (CMM_LOAD_SHARED(tail->p) == node)
 307                         return NULL;
 308                 next = ___cds_wfcq_node_sync_next(node, blocking);
 309         }
 310         /* Load node->next before loading next's content */
 311         cmm_smp_read_barrier_depends();
 312         return next;
 313 }
 314
 315 /*
 316  * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
 317  *
 318  * Content written into the node before enqueue is guaranteed to be
 319  * consistent, but no other memory ordering is ensured.
 320  * Dequeue/splice/iteration mutual exclusion should be ensured by the
 321  * caller.
 322  *
 323  * Used by for-like iteration macros in urcu/wfqueue.h:
 324  * __cds_wfcq_for_each_blocking()
 325  * __cds_wfcq_for_each_blocking_safe()
 326  *
 327  * Returns NULL if reached end of queue, non-NULL next queue node
 328  * otherwise.
 329  */
 330 static inline struct cds_wfcq_node *
 331 ___cds_wfcq_next_blocking(struct cds_wfcq_head *head,
 332                 struct cds_wfcq_tail *tail,
 333                 struct cds_wfcq_node *node)
 334 {
 335         return ___cds_wfcq_next(head, tail, node, 1);
 336 }
 337
 338 /*
 339  * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
 340  *
 341  * Same as __cds_wfcq_next_blocking, but returns CDS_WFCQ_WOULDBLOCK if
 342  * it needs to block.
 343  */
 344 static inline struct cds_wfcq_node *
 345 ___cds_wfcq_next_nonblocking(struct cds_wfcq_head *head,
 346                 struct cds_wfcq_tail *tail,
 347                 struct cds_wfcq_node *node)
 348 {
 349         return ___cds_wfcq_next(head, tail, node, 0);
 350 }
 351
 352 static inline struct cds_wfcq_node *
 353 ___cds_wfcq_dequeue_with_state(struct cds_wfcq_head *head,
 354                 struct cds_wfcq_tail *tail,
 355                 int *state,
 356                 int blocking)
 357 {
 358         struct cds_wfcq_node *node, *next;
 359
 360         if (state)
 361                 *state = 0;
 362
 363         if (_cds_wfcq_empty(head, tail)) {
 364                 return NULL;
 365         }
 366
 367         node = ___cds_wfcq_node_sync_next(&head->node, blocking);
 368         if (!blocking && node == CDS_WFCQ_WOULDBLOCK) {
 369                 return CDS_WFCQ_WOULDBLOCK;
 370         }
 371
 372         if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 373                 /*
 374                  * @node is probably the only node in the queue.
 375                  * Try to move the tail to &q->head.
 376                  * q->head.next is set to NULL here, and stays
 377                  * NULL if the cmpxchg succeeds. Should the
 378                  * cmpxchg fail due to a concurrent enqueue, the
 379                  * q->head.next will be set to the next node.
 380                  * The implicit memory barrier before
 381                  * uatomic_cmpxchg() orders load node->next
 382                  * before loading q->tail.
 383                  * The implicit memory barrier before uatomic_cmpxchg
 384                  * orders load q->head.next before loading node's
 385                  * content.
 386                  */
 387                 _cds_wfcq_node_init(&head->node);
 388                 if (uatomic_cmpxchg(&tail->p, node, &head->node) == node) {
 389                         if (state)
 390                                 *state |= CDS_WFCQ_STATE_LAST;
 391                         return node;
 392                 }
 393                 next = ___cds_wfcq_node_sync_next(node, blocking);
 394                 /*
 395                  * In nonblocking mode, if we would need to block to
 396                  * get node's next, set the head next node pointer
 397                  * (currently NULL) back to its original value.
 398                  */
 399                 if (!blocking && next == CDS_WFCQ_WOULDBLOCK) {
 400                         head->node.next = node;
 401                         return CDS_WFCQ_WOULDBLOCK;
 402                 }
 403         }
 404
 405         /*
 406          * Move queue head forward.
 407          */
 408         head->node.next = next;
 409
 410         /* Load q->head.next before loading node's content */
 411         cmm_smp_read_barrier_depends();
 412         return node;
 413 }
 414
 415 /*
 416  * __cds_wfcq_dequeue_with_state_blocking: dequeue node from queue, with state.
 417  *
 418  * Content written into the node before enqueue is guaranteed to be
 419  * consistent, but no other memory ordering is ensured.
 420  * It is valid to reuse and free a dequeued node immediately.
 421  * Dequeue/splice/iteration mutual exclusion should be ensured by the
 422  * caller.
 423  */
 424 static inline struct cds_wfcq_node *
 425 ___cds_wfcq_dequeue_with_state_blocking(struct cds_wfcq_head *head,
 426                 struct cds_wfcq_tail *tail, int *state)
 427 {
 428         return ___cds_wfcq_dequeue_with_state(head, tail, state, 1);
 429 }
 430
 431 /*
 432  * ___cds_wfcq_dequeue_blocking: dequeue node from queue.
 433  *
 434  * Same as __cds_wfcq_dequeue_with_state_blocking, but without saving
 435  * state.
 436  */
 437 static inline struct cds_wfcq_node *
 438 ___cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
 439                 struct cds_wfcq_tail *tail)
 440 {
 441         return ___cds_wfcq_dequeue_with_state_blocking(head, tail, NULL);
 442 }
 443
 444 /*
 445  * __cds_wfcq_dequeue_with_state_nonblocking: dequeue node, with state.
 446  *
 447  * Same as __cds_wfcq_dequeue_blocking, but returns CDS_WFCQ_WOULDBLOCK
 448  * if it needs to block.
 449  */
 450 static inline struct cds_wfcq_node *
 451 ___cds_wfcq_dequeue_with_state_nonblocking(struct cds_wfcq_head *head,
 452                 struct cds_wfcq_tail *tail, int *state)
 453 {
 454         return ___cds_wfcq_dequeue_with_state(head, tail, state, 0);
 455 }
 456
 457 /*
 458  * ___cds_wfcq_dequeue_nonblocking: dequeue node from queue.
 459  *
 460  * Same as __cds_wfcq_dequeue_with_state_nonblocking, but without saving
 461  * state.
 462  */
 463 static inline struct cds_wfcq_node *
 464 ___cds_wfcq_dequeue_nonblocking(struct cds_wfcq_head *head,
 465                 struct cds_wfcq_tail *tail)
 466 {
 467         return ___cds_wfcq_dequeue_with_state_nonblocking(head, tail, NULL);
 468 }
 469
 470 /*
 471  * __cds_wfcq_splice: enqueue all src_q nodes at the end of dest_q.
 472  *
 473  * Dequeue all nodes from src_q.
 474  * dest_q must be already initialized.
 475  * Mutual exclusion for src_q should be ensured by the caller as
 476  * specified in the "Synchronisation table".
 477  * Returns enum cds_wfcq_ret which indicates the state of the src or
 478  * dest queue.
 479  */
 480 static inline enum cds_wfcq_ret
 481 ___cds_wfcq_splice(
 482                 struct cds_wfcq_head *dest_q_head,
 483                 struct cds_wfcq_tail *dest_q_tail,
 484                 struct cds_wfcq_head *src_q_head,
 485                 struct cds_wfcq_tail *src_q_tail,
 486                 int blocking)
 487 {
 488         struct cds_wfcq_node *head, *tail;
 489         int attempt = 0;
 490
 491         /*
 492          * Initial emptiness check to speed up cases where queue is
 493          * empty: only require loads to check if queue is empty.
 494          */
 495         if (_cds_wfcq_empty(src_q_head, src_q_tail))
 496                 return CDS_WFCQ_RET_SRC_EMPTY;
 497
 498         for (;;) {
 499                 /*
 500                  * Open-coded _cds_wfcq_empty() by testing result of
 501                  * uatomic_xchg, as well as tail pointer vs head node
 502                  * address.
 503                  */
 504                 head = uatomic_xchg(&src_q_head->node.next, NULL);
 505                 if (head)
 506                         break;  /* non-empty */
 507                 if (CMM_LOAD_SHARED(src_q_tail->p) == &src_q_head->node)
 508                         return CDS_WFCQ_RET_SRC_EMPTY;
 509                 if (___cds_wfcq_busy_wait(&attempt, blocking))
 510                         return CDS_WFCQ_RET_WOULDBLOCK;
 511         }
 512
 513         /*
 514          * Memory barrier implied before uatomic_xchg() orders store to
 515          * src_q->head before store to src_q->tail. This is required by
 516          * concurrent enqueue on src_q, which exchanges the tail before
 517          * updating the previous tail's next pointer.
 518          */
 519         tail = uatomic_xchg(&src_q_tail->p, &src_q_head->node);
 520
 521         /*
 522          * Append the spliced content of src_q into dest_q. Does not
 523          * require mutual exclusion on dest_q (wait-free).
 524          */
 525         if (___cds_wfcq_append(dest_q_head, dest_q_tail, head, tail))
 526                 return CDS_WFCQ_RET_DEST_NON_EMPTY;
 527         else
 528                 return CDS_WFCQ_RET_DEST_EMPTY;
 529 }
 530
 531 /*
 532  * __cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
 533  *
 534  * Dequeue all nodes from src_q.
 535  * dest_q must be already initialized.
 536  * Mutual exclusion for src_q should be ensured by the caller as
 537  * specified in the "Synchronisation table".
 538  * Returns enum cds_wfcq_ret which indicates the state of the src or
 539  * dest queue. Never returns CDS_WFCQ_RET_WOULDBLOCK.
 540  */
 541 static inline enum cds_wfcq_ret
 542 ___cds_wfcq_splice_blocking(
 543                 struct cds_wfcq_head *dest_q_head,
 544                 struct cds_wfcq_tail *dest_q_tail,
 545                 struct cds_wfcq_head *src_q_head,
 546                 struct cds_wfcq_tail *src_q_tail)
 547 {
 548         return ___cds_wfcq_splice(dest_q_head, dest_q_tail,
 549                 src_q_head, src_q_tail, 1);
 550 }
 551
 552 /*
 553  * __cds_wfcq_splice_nonblocking: enqueue all src_q nodes at the end of dest_q.
 554  *
 555  * Same as __cds_wfcq_splice_blocking, but returns
 556  * CDS_WFCQ_RET_WOULDBLOCK if it needs to block.
 557  */
 558 static inline enum cds_wfcq_ret
 559 ___cds_wfcq_splice_nonblocking(
 560                 struct cds_wfcq_head *dest_q_head,
 561                 struct cds_wfcq_tail *dest_q_tail,
 562                 struct cds_wfcq_head *src_q_head,
 563                 struct cds_wfcq_tail *src_q_tail)
 564 {
 565         return ___cds_wfcq_splice(dest_q_head, dest_q_tail,
 566                 src_q_head, src_q_tail, 0);
 567 }
 568
 569 /*
 570  * cds_wfcq_dequeue_with_state_blocking: dequeue a node from a wait-free queue.
 571  *
 572  * Content written into the node before enqueue is guaranteed to be
 573  * consistent, but no other memory ordering is ensured.
 574  * Mutual exclusion with cds_wfcq_splice_blocking and dequeue lock is
 575  * ensured.
 576  * It is valid to reuse and free a dequeued node immediately.
 577  */
 578 static inline struct cds_wfcq_node *
 579 _cds_wfcq_dequeue_with_state_blocking(struct cds_wfcq_head *head,
 580                 struct cds_wfcq_tail *tail, int *state)
 581 {
 582         struct cds_wfcq_node *retval;
 583
 584         _cds_wfcq_dequeue_lock(head, tail);
 585         retval = ___cds_wfcq_dequeue_with_state_blocking(head, tail, state);
 586         _cds_wfcq_dequeue_unlock(head, tail);
 587         return retval;
 588 }
 589
 590 /*
 591  * cds_wfcq_dequeue_blocking: dequeue node from queue.
 592  *
 593  * Same as cds_wfcq_dequeue_blocking, but without saving state.
 594  */
 595 static inline struct cds_wfcq_node *
 596 _cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
 597                 struct cds_wfcq_tail *tail)
 598 {
 599         return _cds_wfcq_dequeue_with_state_blocking(head, tail, NULL);
 600 }
 601
 602 /*
 603  * cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
 604  *
 605  * Dequeue all nodes from src_q.
 606  * dest_q must be already initialized.
 607  * Content written into the node before enqueue is guaranteed to be
 608  * consistent, but no other memory ordering is ensured.
 609  * Mutual exclusion with cds_wfcq_dequeue_blocking and dequeue lock is
 610  * ensured.
 611  * Returns enum cds_wfcq_ret which indicates the state of the src or
 612  * dest queue. Never returns CDS_WFCQ_RET_WOULDBLOCK.
 613  */
 614 static inline enum cds_wfcq_ret
 615 _cds_wfcq_splice_blocking(
 616                 struct cds_wfcq_head *dest_q_head,
 617                 struct cds_wfcq_tail *dest_q_tail,
 618                 struct cds_wfcq_head *src_q_head,
 619                 struct cds_wfcq_tail *src_q_tail)
 620 {
 621         enum cds_wfcq_ret ret;
 622
 623         _cds_wfcq_dequeue_lock(src_q_head, src_q_tail);
 624         ret = ___cds_wfcq_splice_blocking(dest_q_head, dest_q_tail,
 625                         src_q_head, src_q_tail);
 626         _cds_wfcq_dequeue_unlock(src_q_head, src_q_tail);
 627         return ret;
 628 }
 629
 630 #ifdef __cplusplus
 631 }
 632 #endif
 633
 634 #endif /* _URCU_WFCQUEUE_STATIC_H */