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