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 (with lock). Pair with
  96  * cds_wfcq_destroy().
  97  */
  98 static inline void _cds_wfcq_init(struct cds_wfcq_head *head,
  99                 struct cds_wfcq_tail *tail)
 100 {
 101         int ret;
 102
 103         /* Set queue head and tail */
 104         _cds_wfcq_node_init(&head->node);
 105         tail->p = &head->node;
 106         ret = pthread_mutex_init(&head->lock, NULL);
 107         assert(!ret);
 108 }
 109
 110 /*
 111  * cds_wfcq_destroy: destroy wait-free queue (with lock). Pair with
 112  * cds_wfcq_init().
 113  */
 114 static inline void _cds_wfcq_destroy(struct cds_wfcq_head *head,
 115                 struct cds_wfcq_tail *tail)
 116 {
 117         int ret = pthread_mutex_destroy(&head->lock);
 118         assert(!ret);
 119 }
 120
 121 /*
 122  * cds_wfcq_empty: return whether wait-free queue is empty.
 123  *
 124  * No memory barrier is issued. No mutual exclusion is required.
 125  *
 126  * We perform the test on head->node.next to check if the queue is
 127  * possibly empty, but we confirm this by checking if the tail pointer
 128  * points to the head node because the tail pointer is the linearisation
 129  * point of the enqueuers. Just checking the head next pointer could
 130  * make a queue appear empty if an enqueuer is preempted for a long time
 131  * between xchg() and setting the previous node's next pointer.
 132  */
 133 static inline bool _cds_wfcq_empty(struct cds_wfcq_head *head,
 134                 struct cds_wfcq_tail *tail)
 135 {
 136         /*
 137          * Queue is empty if no node is pointed by head->node.next nor
 138          * tail->p. Even though the tail->p check is sufficient to find
 139          * out of the queue is empty, we first check head->node.next as a
 140          * common case to ensure that dequeuers do not frequently access
 141          * enqueuer's tail->p cache line.
 142          */
 143         return CMM_LOAD_SHARED(head->node.next) == NULL
 144                 && CMM_LOAD_SHARED(tail->p) == &head->node;
 145 }
 146
 147 static inline void _cds_wfcq_dequeue_lock(struct cds_wfcq_head *head,
 148                 struct cds_wfcq_tail *tail)
 149 {
 150         int ret;
 151
 152         ret = pthread_mutex_lock(&head->lock);
 153         assert(!ret);
 154 }
 155
 156 static inline void _cds_wfcq_dequeue_unlock(struct cds_wfcq_head *head,
 157                 struct cds_wfcq_tail *tail)
 158 {
 159         int ret;
 160
 161         ret = pthread_mutex_unlock(&head->lock);
 162         assert(!ret);
 163 }
 164
 165 static inline bool ___cds_wfcq_append(struct cds_wfcq_head *head,
 166                 struct cds_wfcq_tail *tail,
 167                 struct cds_wfcq_node *new_head,
 168                 struct cds_wfcq_node *new_tail)
 169 {
 170         struct cds_wfcq_node *old_tail;
 171
 172         /*
 173          * Implicit memory barrier before uatomic_xchg() orders earlier
 174          * stores to data structure containing node and setting
 175          * node->next to NULL before publication.
 176          */
 177         old_tail = uatomic_xchg(&tail->p, new_tail);
 178
 179         /*
 180          * Implicit memory barrier after uatomic_xchg() orders store to
 181          * q->tail before store to old_tail->next.
 182          *
 183          * At this point, dequeuers see a NULL tail->p->next, which
 184          * indicates that the queue is being appended to. The following
 185          * store will append "node" to the queue from a dequeuer
 186          * perspective.
 187          */
 188         CMM_STORE_SHARED(old_tail->next, new_head);
 189         /*
 190          * Return false if queue was empty prior to adding the node,
 191          * else return true.
 192          */
 193         return old_tail != &head->node;
 194 }
 195
 196 /*
 197  * cds_wfcq_enqueue: enqueue a node into a wait-free queue.
 198  *
 199  * Issues a full memory barrier before enqueue. No mutual exclusion is
 200  * required.
 201  *
 202  * Returns false if the queue was empty prior to adding the node.
 203  * Returns true otherwise.
 204  */
 205 static inline bool _cds_wfcq_enqueue(struct cds_wfcq_head *head,
 206                 struct cds_wfcq_tail *tail,
 207                 struct cds_wfcq_node *new_tail)
 208 {
 209         return ___cds_wfcq_append(head, tail, new_tail, new_tail);
 210 }
 211
 212 /*
 213  * ___cds_wfcq_busy_wait: adaptative busy-wait.
 214  *
 215  * Returns 1 if nonblocking and needs to block, 0 otherwise.
 216  */
 217 static inline bool
 218 ___cds_wfcq_busy_wait(int *attempt, int blocking)
 219 {
 220         if (!blocking)
 221                 return 1;
 222         if (++(*attempt) >= WFCQ_ADAPT_ATTEMPTS) {
 223                 poll(NULL, 0, WFCQ_WAIT);       /* Wait for 10ms */
 224                 *attempt = 0;
 225         } else {
 226                 caa_cpu_relax();
 227         }
 228         return 0;
 229 }
 230
 231 /*
 232  * Waiting for enqueuer to complete enqueue and return the next node.
 233  */
 234 static inline struct cds_wfcq_node *
 235 ___cds_wfcq_node_sync_next(struct cds_wfcq_node *node, int blocking)
 236 {
 237         struct cds_wfcq_node *next;
 238         int attempt = 0;
 239
 240         /*
 241          * Adaptative busy-looping waiting for enqueuer to complete enqueue.
 242          */
 243         while ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 244                 if (___cds_wfcq_busy_wait(&attempt, blocking))
 245                         return CDS_WFCQ_WOULDBLOCK;
 246         }
 247
 248         return next;
 249 }
 250
 251 static inline struct cds_wfcq_node *
 252 ___cds_wfcq_first(struct cds_wfcq_head *head,
 253                 struct cds_wfcq_tail *tail,
 254                 int blocking)
 255 {
 256         struct cds_wfcq_node *node;
 257
 258         if (_cds_wfcq_empty(head, tail))
 259                 return NULL;
 260         node = ___cds_wfcq_node_sync_next(&head->node, blocking);
 261         /* Load head->node.next before loading node's content */
 262         cmm_smp_read_barrier_depends();
 263         return node;
 264 }
 265
 266 /*
 267  * __cds_wfcq_first_blocking: get first node of a queue, without dequeuing.
 268  *
 269  * Content written into the node before enqueue is guaranteed to be
 270  * consistent, but no other memory ordering is ensured.
 271  * Dequeue/splice/iteration mutual exclusion should be ensured by the
 272  * caller.
 273  *
 274  * Used by for-like iteration macros in urcu/wfqueue.h:
 275  * __cds_wfcq_for_each_blocking()
 276  * __cds_wfcq_for_each_blocking_safe()
 277  *
 278  * Returns NULL if queue is empty, first node otherwise.
 279  */
 280 static inline struct cds_wfcq_node *
 281 ___cds_wfcq_first_blocking(struct cds_wfcq_head *head,
 282                 struct cds_wfcq_tail *tail)
 283 {
 284         return ___cds_wfcq_first(head, tail, 1);
 285 }
 286
 287
 288 /*
 289  * __cds_wfcq_first_nonblocking: get first node of a queue, without dequeuing.
 290  *
 291  * Same as __cds_wfcq_first_blocking, but returns CDS_WFCQ_WOULDBLOCK if
 292  * it needs to block.
 293  */
 294 static inline struct cds_wfcq_node *
 295 ___cds_wfcq_first_nonblocking(struct cds_wfcq_head *head,
 296                 struct cds_wfcq_tail *tail)
 297 {
 298         return ___cds_wfcq_first(head, tail, 0);
 299 }
 300
 301 static inline struct cds_wfcq_node *
 302 ___cds_wfcq_next(struct cds_wfcq_head *head,
 303                 struct cds_wfcq_tail *tail,
 304                 struct cds_wfcq_node *node,
 305                 int blocking)
 306 {
 307         struct cds_wfcq_node *next;
 308
 309         /*
 310          * Even though the following tail->p check is sufficient to find
 311          * out if we reached the end of the queue, we first check
 312          * node->next as a common case to ensure that iteration on nodes
 313          * do not frequently access enqueuer's tail->p cache line.
 314          */
 315         if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 316                 /* Load node->next before tail->p */
 317                 cmm_smp_rmb();
 318                 if (CMM_LOAD_SHARED(tail->p) == node)
 319                         return NULL;
 320                 next = ___cds_wfcq_node_sync_next(node, blocking);
 321         }
 322         /* Load node->next before loading next's content */
 323         cmm_smp_read_barrier_depends();
 324         return next;
 325 }
 326
 327 /*
 328  * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
 329  *
 330  * Content written into the node before enqueue is guaranteed to be
 331  * consistent, but no other memory ordering is ensured.
 332  * Dequeue/splice/iteration mutual exclusion should be ensured by the
 333  * caller.
 334  *
 335  * Used by for-like iteration macros in urcu/wfqueue.h:
 336  * __cds_wfcq_for_each_blocking()
 337  * __cds_wfcq_for_each_blocking_safe()
 338  *
 339  * Returns NULL if reached end of queue, non-NULL next queue node
 340  * otherwise.
 341  */
 342 static inline struct cds_wfcq_node *
 343 ___cds_wfcq_next_blocking(struct cds_wfcq_head *head,
 344                 struct cds_wfcq_tail *tail,
 345                 struct cds_wfcq_node *node)
 346 {
 347         return ___cds_wfcq_next(head, tail, node, 1);
 348 }
 349
 350 /*
 351  * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
 352  *
 353  * Same as __cds_wfcq_next_blocking, but returns CDS_WFCQ_WOULDBLOCK if
 354  * it needs to block.
 355  */
 356 static inline struct cds_wfcq_node *
 357 ___cds_wfcq_next_nonblocking(struct cds_wfcq_head *head,
 358                 struct cds_wfcq_tail *tail,
 359                 struct cds_wfcq_node *node)
 360 {
 361         return ___cds_wfcq_next(head, tail, node, 0);
 362 }
 363
 364 static inline struct cds_wfcq_node *
 365 ___cds_wfcq_dequeue_with_state(struct cds_wfcq_head *head,
 366                 struct cds_wfcq_tail *tail,
 367                 int *state,
 368                 int blocking)
 369 {
 370         struct cds_wfcq_node *node, *next;
 371
 372         if (state)
 373                 *state = 0;
 374
 375         if (_cds_wfcq_empty(head, tail)) {
 376                 return NULL;
 377         }
 378
 379         node = ___cds_wfcq_node_sync_next(&head->node, blocking);
 380         if (!blocking && node == CDS_WFCQ_WOULDBLOCK) {
 381                 return CDS_WFCQ_WOULDBLOCK;
 382         }
 383
 384         if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 385                 /*
 386                  * @node is probably the only node in the queue.
 387                  * Try to move the tail to &q->head.
 388                  * q->head.next is set to NULL here, and stays
 389                  * NULL if the cmpxchg succeeds. Should the
 390                  * cmpxchg fail due to a concurrent enqueue, the
 391                  * q->head.next will be set to the next node.
 392                  * The implicit memory barrier before
 393                  * uatomic_cmpxchg() orders load node->next
 394                  * before loading q->tail.
 395                  * The implicit memory barrier before uatomic_cmpxchg
 396                  * orders load q->head.next before loading node's
 397                  * content.
 398                  */
 399                 _cds_wfcq_node_init(&head->node);
 400                 if (uatomic_cmpxchg(&tail->p, node, &head->node) == node) {
 401                         if (state)
 402                                 *state |= CDS_WFCQ_STATE_LAST;
 403                         return node;
 404                 }
 405                 next = ___cds_wfcq_node_sync_next(node, blocking);
 406                 /*
 407                  * In nonblocking mode, if we would need to block to
 408                  * get node's next, set the head next node pointer
 409                  * (currently NULL) back to its original value.
 410                  */
 411                 if (!blocking && next == CDS_WFCQ_WOULDBLOCK) {
 412                         head->node.next = node;
 413                         return CDS_WFCQ_WOULDBLOCK;
 414                 }
 415         }
 416
 417         /*
 418          * Move queue head forward.
 419          */
 420         head->node.next = next;
 421
 422         /* Load q->head.next before loading node's content */
 423         cmm_smp_read_barrier_depends();
 424         return node;
 425 }
 426
 427 /*
 428  * __cds_wfcq_dequeue_with_state_blocking: dequeue node from queue, with state.
 429  *
 430  * Content written into the node before enqueue is guaranteed to be
 431  * consistent, but no other memory ordering is ensured.
 432  * It is valid to reuse and free a dequeued node immediately.
 433  * Dequeue/splice/iteration mutual exclusion should be ensured by the
 434  * caller.
 435  */
 436 static inline struct cds_wfcq_node *
 437 ___cds_wfcq_dequeue_with_state_blocking(struct cds_wfcq_head *head,
 438                 struct cds_wfcq_tail *tail, int *state)
 439 {
 440         return ___cds_wfcq_dequeue_with_state(head, tail, state, 1);
 441 }
 442
 443 /*
 444  * ___cds_wfcq_dequeue_blocking: dequeue node from queue.
 445  *
 446  * Same as __cds_wfcq_dequeue_with_state_blocking, but without saving
 447  * state.
 448  */
 449 static inline struct cds_wfcq_node *
 450 ___cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
 451                 struct cds_wfcq_tail *tail)
 452 {
 453         return ___cds_wfcq_dequeue_with_state_blocking(head, tail, NULL);
 454 }
 455
 456 /*
 457  * __cds_wfcq_dequeue_with_state_nonblocking: dequeue node, with state.
 458  *
 459  * Same as __cds_wfcq_dequeue_blocking, but returns CDS_WFCQ_WOULDBLOCK
 460  * if it needs to block.
 461  */
 462 static inline struct cds_wfcq_node *
 463 ___cds_wfcq_dequeue_with_state_nonblocking(struct cds_wfcq_head *head,
 464                 struct cds_wfcq_tail *tail, int *state)
 465 {
 466         return ___cds_wfcq_dequeue_with_state(head, tail, state, 0);
 467 }
 468
 469 /*
 470  * ___cds_wfcq_dequeue_nonblocking: dequeue node from queue.
 471  *
 472  * Same as __cds_wfcq_dequeue_with_state_nonblocking, but without saving
 473  * state.
 474  */
 475 static inline struct cds_wfcq_node *
 476 ___cds_wfcq_dequeue_nonblocking(struct cds_wfcq_head *head,
 477                 struct cds_wfcq_tail *tail)
 478 {
 479         return ___cds_wfcq_dequeue_with_state_nonblocking(head, tail, NULL);
 480 }
 481
 482 /*
 483  * __cds_wfcq_splice: enqueue all src_q nodes at the end of dest_q.
 484  *
 485  * Dequeue all nodes from src_q.
 486  * dest_q must be already initialized.
 487  * Mutual exclusion for src_q should be ensured by the caller as
 488  * specified in the "Synchronisation table".
 489  * Returns enum cds_wfcq_ret which indicates the state of the src or
 490  * dest queue.
 491  */
 492 static inline enum cds_wfcq_ret
 493 ___cds_wfcq_splice(
 494                 struct cds_wfcq_head *dest_q_head,
 495                 struct cds_wfcq_tail *dest_q_tail,
 496                 struct cds_wfcq_head *src_q_head,
 497                 struct cds_wfcq_tail *src_q_tail,
 498                 int blocking)
 499 {
 500         struct cds_wfcq_node *head, *tail;
 501         int attempt = 0;
 502
 503         /*
 504          * Initial emptiness check to speed up cases where queue is
 505          * empty: only require loads to check if queue is empty.
 506          */
 507         if (_cds_wfcq_empty(src_q_head, src_q_tail))
 508                 return CDS_WFCQ_RET_SRC_EMPTY;
 509
 510         for (;;) {
 511                 /*
 512                  * Open-coded _cds_wfcq_empty() by testing result of
 513                  * uatomic_xchg, as well as tail pointer vs head node
 514                  * address.
 515                  */
 516                 head = uatomic_xchg(&src_q_head->node.next, NULL);
 517                 if (head)
 518                         break;  /* non-empty */
 519                 if (CMM_LOAD_SHARED(src_q_tail->p) == &src_q_head->node)
 520                         return CDS_WFCQ_RET_SRC_EMPTY;
 521                 if (___cds_wfcq_busy_wait(&attempt, blocking))
 522                         return CDS_WFCQ_RET_WOULDBLOCK;
 523         }
 524
 525         /*
 526          * Memory barrier implied before uatomic_xchg() orders store to
 527          * src_q->head before store to src_q->tail. This is required by
 528          * concurrent enqueue on src_q, which exchanges the tail before
 529          * updating the previous tail's next pointer.
 530          */
 531         tail = uatomic_xchg(&src_q_tail->p, &src_q_head->node);
 532
 533         /*
 534          * Append the spliced content of src_q into dest_q. Does not
 535          * require mutual exclusion on dest_q (wait-free).
 536          */
 537         if (___cds_wfcq_append(dest_q_head, dest_q_tail, head, tail))
 538                 return CDS_WFCQ_RET_DEST_NON_EMPTY;
 539         else
 540                 return CDS_WFCQ_RET_DEST_EMPTY;
 541 }
 542
 543 /*
 544  * __cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
 545  *
 546  * Dequeue all nodes from src_q.
 547  * dest_q must be already initialized.
 548  * Mutual exclusion for src_q should be ensured by the caller as
 549  * specified in the "Synchronisation table".
 550  * Returns enum cds_wfcq_ret which indicates the state of the src or
 551  * dest queue. Never returns CDS_WFCQ_RET_WOULDBLOCK.
 552  */
 553 static inline enum cds_wfcq_ret
 554 ___cds_wfcq_splice_blocking(
 555                 struct cds_wfcq_head *dest_q_head,
 556                 struct cds_wfcq_tail *dest_q_tail,
 557                 struct cds_wfcq_head *src_q_head,
 558                 struct cds_wfcq_tail *src_q_tail)
 559 {
 560         return ___cds_wfcq_splice(dest_q_head, dest_q_tail,
 561                 src_q_head, src_q_tail, 1);
 562 }
 563
 564 /*
 565  * __cds_wfcq_splice_nonblocking: enqueue all src_q nodes at the end of dest_q.
 566  *
 567  * Same as __cds_wfcq_splice_blocking, but returns
 568  * CDS_WFCQ_RET_WOULDBLOCK if it needs to block.
 569  */
 570 static inline enum cds_wfcq_ret
 571 ___cds_wfcq_splice_nonblocking(
 572                 struct cds_wfcq_head *dest_q_head,
 573                 struct cds_wfcq_tail *dest_q_tail,
 574                 struct cds_wfcq_head *src_q_head,
 575                 struct cds_wfcq_tail *src_q_tail)
 576 {
 577         return ___cds_wfcq_splice(dest_q_head, dest_q_tail,
 578                 src_q_head, src_q_tail, 0);
 579 }
 580
 581 /*
 582  * cds_wfcq_dequeue_with_state_blocking: dequeue a node from a wait-free queue.
 583  *
 584  * Content written into the node before enqueue is guaranteed to be
 585  * consistent, but no other memory ordering is ensured.
 586  * Mutual exclusion with cds_wfcq_splice_blocking and dequeue lock is
 587  * ensured.
 588  * It is valid to reuse and free a dequeued node immediately.
 589  */
 590 static inline struct cds_wfcq_node *
 591 _cds_wfcq_dequeue_with_state_blocking(struct cds_wfcq_head *head,
 592                 struct cds_wfcq_tail *tail, int *state)
 593 {
 594         struct cds_wfcq_node *retval;
 595
 596         _cds_wfcq_dequeue_lock(head, tail);
 597         retval = ___cds_wfcq_dequeue_with_state_blocking(head, tail, state);
 598         _cds_wfcq_dequeue_unlock(head, tail);
 599         return retval;
 600 }
 601
 602 /*
 603  * cds_wfcq_dequeue_blocking: dequeue node from queue.
 604  *
 605  * Same as cds_wfcq_dequeue_blocking, but without saving state.
 606  */
 607 static inline struct cds_wfcq_node *
 608 _cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
 609                 struct cds_wfcq_tail *tail)
 610 {
 611         return _cds_wfcq_dequeue_with_state_blocking(head, tail, NULL);
 612 }
 613
 614 /*
 615  * cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
 616  *
 617  * Dequeue all nodes from src_q.
 618  * dest_q must be already initialized.
 619  * Content written into the node before enqueue is guaranteed to be
 620  * consistent, but no other memory ordering is ensured.
 621  * Mutual exclusion with cds_wfcq_dequeue_blocking and dequeue lock is
 622  * ensured.
 623  * Returns enum cds_wfcq_ret which indicates the state of the src or
 624  * dest queue. Never returns CDS_WFCQ_RET_WOULDBLOCK.
 625  */
 626 static inline enum cds_wfcq_ret
 627 _cds_wfcq_splice_blocking(
 628                 struct cds_wfcq_head *dest_q_head,
 629                 struct cds_wfcq_tail *dest_q_tail,
 630                 struct cds_wfcq_head *src_q_head,
 631                 struct cds_wfcq_tail *src_q_tail)
 632 {
 633         enum cds_wfcq_ret ret;
 634
 635         _cds_wfcq_dequeue_lock(src_q_head, src_q_tail);
 636         ret = ___cds_wfcq_splice_blocking(dest_q_head, dest_q_tail,
 637                         src_q_head, src_q_tail);
 638         _cds_wfcq_dequeue_unlock(src_q_head, src_q_tail);
 639         return ret;
 640 }
 641
 642 #ifdef __cplusplus
 643 }
 644 #endif
 645
 646 #endif /* _URCU_WFCQUEUE_STATIC_H */