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_* API
  50  *
  51  * Unless otherwise stated, the caller must ensure mutual exclusion of
  52  * queue update operations "dequeue" and "splice" (for source queue).
  53  * Queue read operations "first" and "next", which are used by
  54  * "for_each" iterations, need to be protected against concurrent
  55  * "dequeue" and "splice" (for source queue) by the caller.
  56  * "enqueue", "splice" (for destination queue), and "empty" are the only
  57  * operations that can be used without any mutual exclusion.
  58  * Mutual exclusion can be ensured by holding cds_wfcq_dequeue_lock().
  59  *
  60  * For convenience, cds_wfcq_dequeue_blocking() and
  61  * cds_wfcq_splice_blocking() hold the dequeue lock.
  62  *
  63  * Besides locking, mutual exclusion of dequeue, splice and iteration
  64  * can be ensured by performing all of those operations from a single
  65  * thread, without requiring any lock.
  66  */
  67
  68 #define WFCQ_ADAPT_ATTEMPTS             10      /* Retry if being set */
  69 #define WFCQ_WAIT                       10      /* Wait 10 ms if being set */
  70
  71 /*
  72  * cds_wfcq_node_init: initialize wait-free queue node.
  73  */
  74 static inline void _cds_wfcq_node_init(struct cds_wfcq_node *node)
  75 {
  76         node->next = NULL;
  77 }
  78
  79 /*
  80  * cds_wfcq_init: initialize wait-free queue.
  81  */
  82 static inline void _cds_wfcq_init(struct cds_wfcq_head *head,
  83                 struct cds_wfcq_tail *tail)
  84 {
  85         int ret;
  86
  87         /* Set queue head and tail */
  88         _cds_wfcq_node_init(&head->node);
  89         tail->p = &head->node;
  90         ret = pthread_mutex_init(&head->lock, NULL);
  91         assert(!ret);
  92 }
  93
  94 /*
  95  * cds_wfcq_empty: return whether wait-free queue is empty.
  96  *
  97  * No memory barrier is issued. No mutual exclusion is required.
  98  *
  99  * We perform the test on head->node.next to check if the queue is
 100  * possibly empty, but we confirm this by checking if the tail pointer
 101  * points to the head node because the tail pointer is the linearisation
 102  * point of the enqueuers. Just checking the head next pointer could
 103  * make a queue appear empty if an enqueuer is preempted for a long time
 104  * between xchg() and setting the previous node's next pointer.
 105  */
 106 static inline bool _cds_wfcq_empty(struct cds_wfcq_head *head,
 107                 struct cds_wfcq_tail *tail)
 108 {
 109         /*
 110          * Queue is empty if no node is pointed by head->node.next nor
 111          * tail->p. Even though the tail->p check is sufficient to find
 112          * out of the queue is empty, we first check head->node.next as a
 113          * common case to ensure that dequeuers do not frequently access
 114          * enqueuer's tail->p cache line.
 115          */
 116         return CMM_LOAD_SHARED(head->node.next) == NULL
 117                 && CMM_LOAD_SHARED(tail->p) == &head->node;
 118 }
 119
 120 static inline void _cds_wfcq_dequeue_lock(struct cds_wfcq_head *head,
 121                 struct cds_wfcq_tail *tail)
 122 {
 123         int ret;
 124
 125         ret = pthread_mutex_lock(&head->lock);
 126         assert(!ret);
 127 }
 128
 129 static inline void _cds_wfcq_dequeue_unlock(struct cds_wfcq_head *head,
 130                 struct cds_wfcq_tail *tail)
 131 {
 132         int ret;
 133
 134         ret = pthread_mutex_unlock(&head->lock);
 135         assert(!ret);
 136 }
 137
 138 static inline bool ___cds_wfcq_append(struct cds_wfcq_head *head,
 139                 struct cds_wfcq_tail *tail,
 140                 struct cds_wfcq_node *new_head,
 141                 struct cds_wfcq_node *new_tail)
 142 {
 143         struct cds_wfcq_node *old_tail;
 144
 145         /*
 146          * Implicit memory barrier before uatomic_xchg() orders earlier
 147          * stores to data structure containing node and setting
 148          * node->next to NULL before publication.
 149          */
 150         old_tail = uatomic_xchg(&tail->p, new_tail);
 151
 152         /*
 153          * Implicit memory barrier after uatomic_xchg() orders store to
 154          * q->tail before store to old_tail->next.
 155          *
 156          * At this point, dequeuers see a NULL tail->p->next, which
 157          * indicates that the queue is being appended to. The following
 158          * store will append "node" to the queue from a dequeuer
 159          * perspective.
 160          */
 161         CMM_STORE_SHARED(old_tail->next, new_head);
 162         /*
 163          * Return false if queue was empty prior to adding the node,
 164          * else return true.
 165          */
 166         return old_tail != &head->node;
 167 }
 168
 169 /*
 170  * cds_wfcq_enqueue: enqueue a node into a wait-free queue.
 171  *
 172  * Issues a full memory barrier before enqueue. No mutual exclusion is
 173  * required.
 174  *
 175  * Returns false if the queue was empty prior to adding the node.
 176  * Returns true otherwise.
 177  */
 178 static inline bool _cds_wfcq_enqueue(struct cds_wfcq_head *head,
 179                 struct cds_wfcq_tail *tail,
 180                 struct cds_wfcq_node *new_tail)
 181 {
 182         return ___cds_wfcq_append(head, tail, new_tail, new_tail);
 183 }
 184
 185 /*
 186  * Waiting for enqueuer to complete enqueue and return the next node.
 187  */
 188 static inline struct cds_wfcq_node *
 189 ___cds_wfcq_node_sync_next(struct cds_wfcq_node *node, int blocking)
 190 {
 191         struct cds_wfcq_node *next;
 192         int attempt = 0;
 193
 194         /*
 195          * Adaptative busy-looping waiting for enqueuer to complete enqueue.
 196          */
 197         while ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 198                 if (!blocking)
 199                         return CDS_WFCQ_WOULDBLOCK;
 200                 if (++attempt >= WFCQ_ADAPT_ATTEMPTS) {
 201                         poll(NULL, 0, WFCQ_WAIT);       /* Wait for 10ms */
 202                         attempt = 0;
 203                 } else {
 204                         caa_cpu_relax();
 205                 }
 206         }
 207
 208         return next;
 209 }
 210
 211 static inline struct cds_wfcq_node *
 212 ___cds_wfcq_first(struct cds_wfcq_head *head,
 213                 struct cds_wfcq_tail *tail,
 214                 int blocking)
 215 {
 216         struct cds_wfcq_node *node;
 217
 218         if (_cds_wfcq_empty(head, tail))
 219                 return NULL;
 220         node = ___cds_wfcq_node_sync_next(&head->node, blocking);
 221         /* Load head->node.next before loading node's content */
 222         cmm_smp_read_barrier_depends();
 223         return node;
 224 }
 225
 226 /*
 227  * __cds_wfcq_first_blocking: get first node of a queue, without dequeuing.
 228  *
 229  * Content written into the node before enqueue is guaranteed to be
 230  * consistent, but no other memory ordering is ensured.
 231  * Dequeue/splice/iteration mutual exclusion should be ensured by the
 232  * caller.
 233  *
 234  * Used by for-like iteration macros in urcu/wfqueue.h:
 235  * __cds_wfcq_for_each_blocking()
 236  * __cds_wfcq_for_each_blocking_safe()
 237  */
 238 static inline struct cds_wfcq_node *
 239 ___cds_wfcq_first_blocking(struct cds_wfcq_head *head,
 240                 struct cds_wfcq_tail *tail)
 241 {
 242         return ___cds_wfcq_first(head, tail, 1);
 243 }
 244
 245
 246 /*
 247  * __cds_wfcq_first_nonblocking: get first node of a queue, without dequeuing.
 248  *
 249  * Same as __cds_wfcq_first_blocking, but returns CDS_WFCQ_WOULDBLOCK if
 250  * it needs to block.
 251  */
 252 static inline struct cds_wfcq_node *
 253 ___cds_wfcq_first_nonblocking(struct cds_wfcq_head *head,
 254                 struct cds_wfcq_tail *tail)
 255 {
 256         return ___cds_wfcq_first(head, tail, 0);
 257 }
 258
 259 static inline struct cds_wfcq_node *
 260 ___cds_wfcq_next(struct cds_wfcq_head *head,
 261                 struct cds_wfcq_tail *tail,
 262                 struct cds_wfcq_node *node,
 263                 int blocking)
 264 {
 265         struct cds_wfcq_node *next;
 266
 267         /*
 268          * Even though the following tail->p check is sufficient to find
 269          * out if we reached the end of the queue, we first check
 270          * node->next as a common case to ensure that iteration on nodes
 271          * do not frequently access enqueuer's tail->p cache line.
 272          */
 273         if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 274                 /* Load node->next before tail->p */
 275                 cmm_smp_rmb();
 276                 if (CMM_LOAD_SHARED(tail->p) == node)
 277                         return NULL;
 278                 next = ___cds_wfcq_node_sync_next(node, blocking);
 279         }
 280         /* Load node->next before loading next's content */
 281         cmm_smp_read_barrier_depends();
 282         return next;
 283 }
 284
 285 /*
 286  * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
 287  *
 288  * Content written into the node before enqueue is guaranteed to be
 289  * consistent, but no other memory ordering is ensured.
 290  * Dequeue/splice/iteration mutual exclusion should be ensured by the
 291  * caller.
 292  *
 293  * Used by for-like iteration macros in urcu/wfqueue.h:
 294  * __cds_wfcq_for_each_blocking()
 295  * __cds_wfcq_for_each_blocking_safe()
 296  */
 297 static inline struct cds_wfcq_node *
 298 ___cds_wfcq_next_blocking(struct cds_wfcq_head *head,
 299                 struct cds_wfcq_tail *tail,
 300                 struct cds_wfcq_node *node)
 301 {
 302         return ___cds_wfcq_next(head, tail, node, 1);
 303 }
 304
 305 /*
 306  * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
 307  *
 308  * Same as __cds_wfcq_next_blocking, but returns CDS_WFCQ_WOULDBLOCK if
 309  * it needs to block.
 310  */
 311 static inline struct cds_wfcq_node *
 312 ___cds_wfcq_next_nonblocking(struct cds_wfcq_head *head,
 313                 struct cds_wfcq_tail *tail,
 314                 struct cds_wfcq_node *node)
 315 {
 316         return ___cds_wfcq_next(head, tail, node, 0);
 317 }
 318
 319 static inline struct cds_wfcq_node *
 320 ___cds_wfcq_dequeue(struct cds_wfcq_head *head,
 321                 struct cds_wfcq_tail *tail,
 322                 int blocking)
 323 {
 324         struct cds_wfcq_node *node, *next;
 325
 326         if (_cds_wfcq_empty(head, tail))
 327                 return NULL;
 328
 329         node = ___cds_wfcq_node_sync_next(&head->node, blocking);
 330         if (!blocking && node == CDS_WFCQ_WOULDBLOCK)
 331                 return CDS_WFCQ_WOULDBLOCK;
 332
 333         if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 334                 /*
 335                  * @node is probably the only node in the queue.
 336                  * Try to move the tail to &q->head.
 337                  * q->head.next is set to NULL here, and stays
 338                  * NULL if the cmpxchg succeeds. Should the
 339                  * cmpxchg fail due to a concurrent enqueue, the
 340                  * q->head.next will be set to the next node.
 341                  * The implicit memory barrier before
 342                  * uatomic_cmpxchg() orders load node->next
 343                  * before loading q->tail.
 344                  * The implicit memory barrier before uatomic_cmpxchg
 345                  * orders load q->head.next before loading node's
 346                  * content.
 347                  */
 348                 _cds_wfcq_node_init(&head->node);
 349                 if (uatomic_cmpxchg(&tail->p, node, &head->node) == node)
 350                         return node;
 351                 next = ___cds_wfcq_node_sync_next(node, blocking);
 352                 /*
 353                  * In nonblocking mode, if we would need to block to
 354                  * get node's next, set the head next node pointer
 355                  * (currently NULL) back to its original value.
 356                  */
 357                 if (!blocking && next == CDS_WFCQ_WOULDBLOCK) {
 358                         head->node.next = node;
 359                         return CDS_WFCQ_WOULDBLOCK;
 360                 }
 361         }
 362
 363         /*
 364          * Move queue head forward.
 365          */
 366         head->node.next = next;
 367
 368         /* Load q->head.next before loading node's content */
 369         cmm_smp_read_barrier_depends();
 370         return node;
 371 }
 372
 373 /*
 374  * __cds_wfcq_dequeue_blocking: dequeue a node from the queue.
 375  *
 376  * Content written into the node before enqueue is guaranteed to be
 377  * consistent, but no other memory ordering is ensured.
 378  * It is valid to reuse and free a dequeued node immediately.
 379  * Dequeue/splice/iteration mutual exclusion should be ensured by the
 380  * caller.
 381  */
 382 static inline struct cds_wfcq_node *
 383 ___cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
 384                 struct cds_wfcq_tail *tail)
 385 {
 386         return ___cds_wfcq_dequeue(head, tail, 1);
 387 }
 388
 389 /*
 390  * __cds_wfcq_dequeue_nonblocking: dequeue a node from a wait-free queue.
 391  *
 392  * Same as __cds_wfcq_dequeue_blocking, but returns CDS_WFCQ_WOULDBLOCK
 393  * if it needs to block.
 394  */
 395 static inline struct cds_wfcq_node *
 396 ___cds_wfcq_dequeue_nonblocking(struct cds_wfcq_head *head,
 397                 struct cds_wfcq_tail *tail)
 398 {
 399         return ___cds_wfcq_dequeue(head, tail, 0);
 400 }
 401
 402 static inline enum cds_wfcq_ret
 403 ___cds_wfcq_splice(
 404                 struct cds_wfcq_head *dest_q_head,
 405                 struct cds_wfcq_tail *dest_q_tail,
 406                 struct cds_wfcq_head *src_q_head,
 407                 struct cds_wfcq_tail *src_q_tail,
 408                 int blocking)
 409 {
 410         struct cds_wfcq_node *head, *tail;
 411
 412         if (_cds_wfcq_empty(src_q_head, src_q_tail))
 413                 return CDS_WFCQ_RET_SRC_EMPTY;
 414
 415         head = ___cds_wfcq_node_sync_next(&src_q_head->node, blocking);
 416         if (head == CDS_WFCQ_WOULDBLOCK)
 417                 return CDS_WFCQ_RET_WOULDBLOCK;
 418         _cds_wfcq_node_init(&src_q_head->node);
 419
 420         /*
 421          * Memory barrier implied before uatomic_xchg() orders store to
 422          * src_q->head before store to src_q->tail. This is required by
 423          * concurrent enqueue on src_q, which exchanges the tail before
 424          * updating the previous tail's next pointer.
 425          */
 426         tail = uatomic_xchg(&src_q_tail->p, &src_q_head->node);
 427
 428         /*
 429          * Append the spliced content of src_q into dest_q. Does not
 430          * require mutual exclusion on dest_q (wait-free).
 431          */
 432         if (___cds_wfcq_append(dest_q_head, dest_q_tail, head, tail))
 433                 return CDS_WFCQ_RET_DEST_NON_EMPTY;
 434         else
 435                 return CDS_WFCQ_RET_DEST_EMPTY;
 436 }
 437
 438
 439 /*
 440  * __cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
 441  *
 442  * Dequeue all nodes from src_q.
 443  * dest_q must be already initialized.
 444  * Dequeue/splice/iteration mutual exclusion for src_q should be ensured
 445  * by the caller.
 446  * Returns enum cds_wfcq_ret which indicates the state of the src or
 447  * dest queue. Never returns CDS_WFCQ_RET_WOULDBLOCK.
 448  */
 449 static inline enum cds_wfcq_ret
 450 ___cds_wfcq_splice_blocking(
 451                 struct cds_wfcq_head *dest_q_head,
 452                 struct cds_wfcq_tail *dest_q_tail,
 453                 struct cds_wfcq_head *src_q_head,
 454                 struct cds_wfcq_tail *src_q_tail)
 455 {
 456         return ___cds_wfcq_splice(dest_q_head, dest_q_tail,
 457                 src_q_head, src_q_tail, 1);
 458 }
 459
 460 /*
 461  * __cds_wfcq_splice_nonblocking: enqueue all src_q nodes at the end of dest_q.
 462  *
 463  * Same as __cds_wfcq_splice_blocking, but returns
 464  * CDS_WFCQ_RET_WOULDBLOCK if it needs to block.
 465  */
 466 static inline enum cds_wfcq_ret
 467 ___cds_wfcq_splice_nonblocking(
 468                 struct cds_wfcq_head *dest_q_head,
 469                 struct cds_wfcq_tail *dest_q_tail,
 470                 struct cds_wfcq_head *src_q_head,
 471                 struct cds_wfcq_tail *src_q_tail)
 472 {
 473         return ___cds_wfcq_splice(dest_q_head, dest_q_tail,
 474                 src_q_head, src_q_tail, 0);
 475 }
 476
 477 /*
 478  * cds_wfcq_dequeue_blocking: dequeue a node from a wait-free queue.
 479  *
 480  * Content written into the node before enqueue is guaranteed to be
 481  * consistent, but no other memory ordering is ensured.
 482  * Mutual exlusion with cds_wfcq_splice_blocking and dequeue lock is
 483  * ensured.
 484  * It is valid to reuse and free a dequeued node immediately.
 485  */
 486 static inline struct cds_wfcq_node *
 487 _cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
 488                 struct cds_wfcq_tail *tail)
 489 {
 490         struct cds_wfcq_node *retval;
 491
 492         _cds_wfcq_dequeue_lock(head, tail);
 493         retval = ___cds_wfcq_dequeue_blocking(head, tail);
 494         _cds_wfcq_dequeue_unlock(head, tail);
 495         return retval;
 496 }
 497
 498 /*
 499  * cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
 500  *
 501  * Dequeue all nodes from src_q.
 502  * dest_q must be already initialized.
 503  * Content written into the node before enqueue is guaranteed to be
 504  * consistent, but no other memory ordering is ensured.
 505  * Mutual exlusion with cds_wfcq_dequeue_blocking and dequeue lock is
 506  * ensured.
 507  * Returns enum cds_wfcq_ret which indicates the state of the src or
 508  * dest queue. Never returns CDS_WFCQ_RET_WOULDBLOCK.
 509  */
 510 static inline enum cds_wfcq_ret
 511 _cds_wfcq_splice_blocking(
 512                 struct cds_wfcq_head *dest_q_head,
 513                 struct cds_wfcq_tail *dest_q_tail,
 514                 struct cds_wfcq_head *src_q_head,
 515                 struct cds_wfcq_tail *src_q_tail)
 516 {
 517         enum cds_wfcq_ret ret;
 518
 519         _cds_wfcq_dequeue_lock(src_q_head, src_q_tail);
 520         ret = ___cds_wfcq_splice_blocking(dest_q_head, dest_q_tail,
 521                         src_q_head, src_q_tail);
 522         _cds_wfcq_dequeue_unlock(src_q_head, src_q_tail);
 523         return ret;
 524 }
 525
 526 #ifdef __cplusplus
 527 }
 528 #endif
 529
 530 #endif /* _URCU_WFCQUEUE_STATIC_H */