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 static inline struct cds_wfcq_node *
 267 ___cds_wfcq_first_blocking(struct cds_wfcq_head *head,
 268                 struct cds_wfcq_tail *tail)
 269 {
 270         return ___cds_wfcq_first(head, tail, 1);
 271 }
 272
 273
 274 /*
 275  * __cds_wfcq_first_nonblocking: get first node of a queue, without dequeuing.
 276  *
 277  * Same as __cds_wfcq_first_blocking, but returns CDS_WFCQ_WOULDBLOCK if
 278  * it needs to block.
 279  */
 280 static inline struct cds_wfcq_node *
 281 ___cds_wfcq_first_nonblocking(struct cds_wfcq_head *head,
 282                 struct cds_wfcq_tail *tail)
 283 {
 284         return ___cds_wfcq_first(head, tail, 0);
 285 }
 286
 287 static inline struct cds_wfcq_node *
 288 ___cds_wfcq_next(struct cds_wfcq_head *head,
 289                 struct cds_wfcq_tail *tail,
 290                 struct cds_wfcq_node *node,
 291                 int blocking)
 292 {
 293         struct cds_wfcq_node *next;
 294
 295         /*
 296          * Even though the following tail->p check is sufficient to find
 297          * out if we reached the end of the queue, we first check
 298          * node->next as a common case to ensure that iteration on nodes
 299          * do not frequently access enqueuer's tail->p cache line.
 300          */
 301         if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 302                 /* Load node->next before tail->p */
 303                 cmm_smp_rmb();
 304                 if (CMM_LOAD_SHARED(tail->p) == node)
 305                         return NULL;
 306                 next = ___cds_wfcq_node_sync_next(node, blocking);
 307         }
 308         /* Load node->next before loading next's content */
 309         cmm_smp_read_barrier_depends();
 310         return next;
 311 }
 312
 313 /*
 314  * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
 315  *
 316  * Content written into the node before enqueue is guaranteed to be
 317  * consistent, but no other memory ordering is ensured.
 318  * Dequeue/splice/iteration mutual exclusion should be ensured by the
 319  * caller.
 320  *
 321  * Used by for-like iteration macros in urcu/wfqueue.h:
 322  * __cds_wfcq_for_each_blocking()
 323  * __cds_wfcq_for_each_blocking_safe()
 324  */
 325 static inline struct cds_wfcq_node *
 326 ___cds_wfcq_next_blocking(struct cds_wfcq_head *head,
 327                 struct cds_wfcq_tail *tail,
 328                 struct cds_wfcq_node *node)
 329 {
 330         return ___cds_wfcq_next(head, tail, node, 1);
 331 }
 332
 333 /*
 334  * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
 335  *
 336  * Same as __cds_wfcq_next_blocking, but returns CDS_WFCQ_WOULDBLOCK if
 337  * it needs to block.
 338  */
 339 static inline struct cds_wfcq_node *
 340 ___cds_wfcq_next_nonblocking(struct cds_wfcq_head *head,
 341                 struct cds_wfcq_tail *tail,
 342                 struct cds_wfcq_node *node)
 343 {
 344         return ___cds_wfcq_next(head, tail, node, 0);
 345 }
 346
 347 static inline struct cds_wfcq_node *
 348 ___cds_wfcq_dequeue(struct cds_wfcq_head *head,
 349                 struct cds_wfcq_tail *tail,
 350                 int blocking)
 351 {
 352         struct cds_wfcq_node *node, *next;
 353
 354         if (_cds_wfcq_empty(head, tail))
 355                 return NULL;
 356
 357         node = ___cds_wfcq_node_sync_next(&head->node, blocking);
 358         if (!blocking && node == CDS_WFCQ_WOULDBLOCK)
 359                 return CDS_WFCQ_WOULDBLOCK;
 360
 361         if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 362                 /*
 363                  * @node is probably the only node in the queue.
 364                  * Try to move the tail to &q->head.
 365                  * q->head.next is set to NULL here, and stays
 366                  * NULL if the cmpxchg succeeds. Should the
 367                  * cmpxchg fail due to a concurrent enqueue, the
 368                  * q->head.next will be set to the next node.
 369                  * The implicit memory barrier before
 370                  * uatomic_cmpxchg() orders load node->next
 371                  * before loading q->tail.
 372                  * The implicit memory barrier before uatomic_cmpxchg
 373                  * orders load q->head.next before loading node's
 374                  * content.
 375                  */
 376                 _cds_wfcq_node_init(&head->node);
 377                 if (uatomic_cmpxchg(&tail->p, node, &head->node) == node)
 378                         return node;
 379                 next = ___cds_wfcq_node_sync_next(node, blocking);
 380                 /*
 381                  * In nonblocking mode, if we would need to block to
 382                  * get node's next, set the head next node pointer
 383                  * (currently NULL) back to its original value.
 384                  */
 385                 if (!blocking && next == CDS_WFCQ_WOULDBLOCK) {
 386                         head->node.next = node;
 387                         return CDS_WFCQ_WOULDBLOCK;
 388                 }
 389         }
 390
 391         /*
 392          * Move queue head forward.
 393          */
 394         head->node.next = next;
 395
 396         /* Load q->head.next before loading node's content */
 397         cmm_smp_read_barrier_depends();
 398         return node;
 399 }
 400
 401 /*
 402  * __cds_wfcq_dequeue_blocking: dequeue a node from the queue.
 403  *
 404  * Content written into the node before enqueue is guaranteed to be
 405  * consistent, but no other memory ordering is ensured.
 406  * It is valid to reuse and free a dequeued node immediately.
 407  * Dequeue/splice/iteration mutual exclusion should be ensured by the
 408  * caller.
 409  */
 410 static inline struct cds_wfcq_node *
 411 ___cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
 412                 struct cds_wfcq_tail *tail)
 413 {
 414         return ___cds_wfcq_dequeue(head, tail, 1);
 415 }
 416
 417 /*
 418  * __cds_wfcq_dequeue_nonblocking: dequeue a node from a wait-free queue.
 419  *
 420  * Same as __cds_wfcq_dequeue_blocking, but returns CDS_WFCQ_WOULDBLOCK
 421  * if it needs to block.
 422  */
 423 static inline struct cds_wfcq_node *
 424 ___cds_wfcq_dequeue_nonblocking(struct cds_wfcq_head *head,
 425                 struct cds_wfcq_tail *tail)
 426 {
 427         return ___cds_wfcq_dequeue(head, tail, 0);
 428 }
 429
 430 /*
 431  * __cds_wfcq_splice: enqueue all src_q nodes at the end of dest_q.
 432  *
 433  * Dequeue all nodes from src_q.
 434  * dest_q must be already initialized.
 435  * Mutual exclusion for src_q should be ensured by the caller as
 436  * specified in the "Synchronisation table".
 437  * Returns enum cds_wfcq_ret which indicates the state of the src or
 438  * dest queue.
 439  */
 440 static inline enum cds_wfcq_ret
 441 ___cds_wfcq_splice(
 442                 struct cds_wfcq_head *dest_q_head,
 443                 struct cds_wfcq_tail *dest_q_tail,
 444                 struct cds_wfcq_head *src_q_head,
 445                 struct cds_wfcq_tail *src_q_tail,
 446                 int blocking)
 447 {
 448         struct cds_wfcq_node *head, *tail;
 449         int attempt = 0;
 450
 451         /*
 452          * Initial emptiness check to speed up cases where queue is
 453          * empty: only require loads to check if queue is empty.
 454          */
 455         if (_cds_wfcq_empty(src_q_head, src_q_tail))
 456                 return CDS_WFCQ_RET_SRC_EMPTY;
 457
 458         for (;;) {
 459                 /*
 460                  * Open-coded _cds_wfcq_empty() by testing result of
 461                  * uatomic_xchg, as well as tail pointer vs head node
 462                  * address.
 463                  */
 464                 head = uatomic_xchg(&src_q_head->node.next, NULL);
 465                 if (head)
 466                         break;  /* non-empty */
 467                 if (CMM_LOAD_SHARED(src_q_tail->p) == &src_q_head->node)
 468                         return CDS_WFCQ_RET_SRC_EMPTY;
 469                 if (___cds_wfcq_busy_wait(&attempt, blocking))
 470                         return CDS_WFCQ_RET_WOULDBLOCK;
 471         }
 472
 473         /*
 474          * Memory barrier implied before uatomic_xchg() orders store to
 475          * src_q->head before store to src_q->tail. This is required by
 476          * concurrent enqueue on src_q, which exchanges the tail before
 477          * updating the previous tail's next pointer.
 478          */
 479         tail = uatomic_xchg(&src_q_tail->p, &src_q_head->node);
 480
 481         /*
 482          * Append the spliced content of src_q into dest_q. Does not
 483          * require mutual exclusion on dest_q (wait-free).
 484          */
 485         if (___cds_wfcq_append(dest_q_head, dest_q_tail, head, tail))
 486                 return CDS_WFCQ_RET_DEST_NON_EMPTY;
 487         else
 488                 return CDS_WFCQ_RET_DEST_EMPTY;
 489 }
 490
 491 /*
 492  * __cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
 493  *
 494  * Dequeue all nodes from src_q.
 495  * dest_q must be already initialized.
 496  * Mutual exclusion for src_q should be ensured by the caller as
 497  * specified in the "Synchronisation table".
 498  * Returns enum cds_wfcq_ret which indicates the state of the src or
 499  * dest queue. Never returns CDS_WFCQ_RET_WOULDBLOCK.
 500  */
 501 static inline enum cds_wfcq_ret
 502 ___cds_wfcq_splice_blocking(
 503                 struct cds_wfcq_head *dest_q_head,
 504                 struct cds_wfcq_tail *dest_q_tail,
 505                 struct cds_wfcq_head *src_q_head,
 506                 struct cds_wfcq_tail *src_q_tail)
 507 {
 508         return ___cds_wfcq_splice(dest_q_head, dest_q_tail,
 509                 src_q_head, src_q_tail, 1);
 510 }
 511
 512 /*
 513  * __cds_wfcq_splice_nonblocking: enqueue all src_q nodes at the end of dest_q.
 514  *
 515  * Same as __cds_wfcq_splice_blocking, but returns
 516  * CDS_WFCQ_RET_WOULDBLOCK if it needs to block.
 517  */
 518 static inline enum cds_wfcq_ret
 519 ___cds_wfcq_splice_nonblocking(
 520                 struct cds_wfcq_head *dest_q_head,
 521                 struct cds_wfcq_tail *dest_q_tail,
 522                 struct cds_wfcq_head *src_q_head,
 523                 struct cds_wfcq_tail *src_q_tail)
 524 {
 525         return ___cds_wfcq_splice(dest_q_head, dest_q_tail,
 526                 src_q_head, src_q_tail, 0);
 527 }
 528
 529 /*
 530  * cds_wfcq_dequeue_blocking: dequeue a node from a wait-free queue.
 531  *
 532  * Content written into the node before enqueue is guaranteed to be
 533  * consistent, but no other memory ordering is ensured.
 534  * Mutual exlusion with cds_wfcq_splice_blocking and dequeue lock is
 535  * ensured.
 536  * It is valid to reuse and free a dequeued node immediately.
 537  */
 538 static inline struct cds_wfcq_node *
 539 _cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
 540                 struct cds_wfcq_tail *tail)
 541 {
 542         struct cds_wfcq_node *retval;
 543
 544         _cds_wfcq_dequeue_lock(head, tail);
 545         retval = ___cds_wfcq_dequeue_blocking(head, tail);
 546         _cds_wfcq_dequeue_unlock(head, tail);
 547         return retval;
 548 }
 549
 550 /*
 551  * cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
 552  *
 553  * Dequeue all nodes from src_q.
 554  * dest_q must be already initialized.
 555  * Content written into the node before enqueue is guaranteed to be
 556  * consistent, but no other memory ordering is ensured.
 557  * Mutual exlusion with cds_wfcq_dequeue_blocking and dequeue lock is
 558  * ensured.
 559  * Returns enum cds_wfcq_ret which indicates the state of the src or
 560  * dest queue. Never returns CDS_WFCQ_RET_WOULDBLOCK.
 561  */
 562 static inline enum cds_wfcq_ret
 563 _cds_wfcq_splice_blocking(
 564                 struct cds_wfcq_head *dest_q_head,
 565                 struct cds_wfcq_tail *dest_q_tail,
 566                 struct cds_wfcq_head *src_q_head,
 567                 struct cds_wfcq_tail *src_q_tail)
 568 {
 569         enum cds_wfcq_ret ret;
 570
 571         _cds_wfcq_dequeue_lock(src_q_head, src_q_tail);
 572         ret = ___cds_wfcq_splice_blocking(dest_q_head, dest_q_tail,
 573                         src_q_head, src_q_tail);
 574         _cds_wfcq_dequeue_unlock(src_q_head, src_q_tail);
 575         return ret;
 576 }
 577
 578 #ifdef __cplusplus
 579 }
 580 #endif
 581
 582 #endif /* _URCU_WFCQUEUE_STATIC_H */