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 wfcqueue.h for linking
  10  * 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
 316         if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 317                 /*
 318                  * @node is probably the only node in the queue.
 319                  * Try to move the tail to &q->head.
 320                  * q->head.next is set to NULL here, and stays
 321                  * NULL if the cmpxchg succeeds. Should the
 322                  * cmpxchg fail due to a concurrent enqueue, the
 323                  * q->head.next will be set to the next node.
 324                  * The implicit memory barrier before
 325                  * uatomic_cmpxchg() orders load node->next
 326                  * before loading q->tail.
 327                  * The implicit memory barrier before uatomic_cmpxchg
 328                  * orders load q->head.next before loading node's
 329                  * content.
 330                  */
 331                 _cds_wfcq_node_init(&head->node);
 332                 if (uatomic_cmpxchg(&tail->p, node, &head->node) == node)
 333                         return node;
 334                 next = ___cds_wfcq_node_sync_next(node, blocking);
 335         }
 336
 337         /*
 338          * Move queue head forward.
 339          */
 340         head->node.next = next;
 341
 342         /* Load q->head.next before loading node's content */
 343         cmm_smp_read_barrier_depends();
 344         return node;
 345 }
 346
 347 /*
 348  * __cds_wfcq_dequeue_blocking: dequeue a node from the queue.
 349  *
 350  * Content written into the node before enqueue is guaranteed to be
 351  * consistent, but no other memory ordering is ensured.
 352  * It is valid to reuse and free a dequeued node immediately.
 353  * Dequeue/splice/iteration mutual exclusion should be ensured by the
 354  * caller.
 355  */
 356 static inline struct cds_wfcq_node *
 357 ___cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
 358                 struct cds_wfcq_tail *tail)
 359 {
 360         return ___cds_wfcq_dequeue(head, tail, 1);
 361 }
 362
 363 /*
 364  * __cds_wfcq_dequeue_nonblocking: dequeue a node from a wait-free queue.
 365  *
 366  * Same as __cds_wfcq_dequeue_blocking, but returns CDS_WFCQ_WOULDBLOCK
 367  * if it needs to block.
 368  */
 369 static inline struct cds_wfcq_node *
 370 ___cds_wfcq_dequeue_nonblocking(struct cds_wfcq_head *head,
 371                 struct cds_wfcq_tail *tail)
 372 {
 373         return ___cds_wfcq_dequeue(head, tail, 0);
 374 }
 375
 376 static inline int
 377 ___cds_wfcq_splice(
 378                 struct cds_wfcq_head *dest_q_head,
 379                 struct cds_wfcq_tail *dest_q_tail,
 380                 struct cds_wfcq_head *src_q_head,
 381                 struct cds_wfcq_tail *src_q_tail,
 382                 int blocking)
 383 {
 384         struct cds_wfcq_node *head, *tail;
 385
 386         if (_cds_wfcq_empty(src_q_head, src_q_tail))
 387                 return 0;
 388
 389         head = ___cds_wfcq_node_sync_next(&src_q_head->node, blocking);
 390         if (head == CDS_WFCQ_WOULDBLOCK)
 391                 return -1;
 392         _cds_wfcq_node_init(&src_q_head->node);
 393
 394         /*
 395          * Memory barrier implied before uatomic_xchg() orders store to
 396          * src_q->head before store to src_q->tail. This is required by
 397          * concurrent enqueue on src_q, which exchanges the tail before
 398          * updating the previous tail's next pointer.
 399          */
 400         tail = uatomic_xchg(&src_q_tail->p, &src_q_head->node);
 401
 402         /*
 403          * Append the spliced content of src_q into dest_q. Does not
 404          * require mutual exclusion on dest_q (wait-free).
 405          */
 406         ___cds_wfcq_append(dest_q_head, dest_q_tail, head, tail);
 407         return 0;
 408 }
 409
 410
 411 /*
 412  * __cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
 413  *
 414  * Dequeue all nodes from src_q.
 415  * dest_q must be already initialized.
 416  * Dequeue/splice/iteration mutual exclusion for src_q should be ensured
 417  * by the caller.
 418  */
 419 static inline void
 420 ___cds_wfcq_splice_blocking(
 421                 struct cds_wfcq_head *dest_q_head,
 422                 struct cds_wfcq_tail *dest_q_tail,
 423                 struct cds_wfcq_head *src_q_head,
 424                 struct cds_wfcq_tail *src_q_tail)
 425 {
 426         (void) ___cds_wfcq_splice(dest_q_head, dest_q_tail,
 427                 src_q_head, src_q_tail, 1);
 428 }
 429
 430 /*
 431  * __cds_wfcq_splice_nonblocking: enqueue all src_q nodes at the end of dest_q.
 432  *
 433  * Same as __cds_wfcq_splice_blocking, but returns nonzero if it needs to
 434  * block.
 435  */
 436 static inline int
 437 ___cds_wfcq_splice_nonblocking(
 438                 struct cds_wfcq_head *dest_q_head,
 439                 struct cds_wfcq_tail *dest_q_tail,
 440                 struct cds_wfcq_head *src_q_head,
 441                 struct cds_wfcq_tail *src_q_tail)
 442 {
 443         return ___cds_wfcq_splice(dest_q_head, dest_q_tail,
 444                 src_q_head, src_q_tail, 0);
 445 }
 446
 447 /*
 448  * cds_wfcq_dequeue_blocking: dequeue a node from a wait-free queue.
 449  *
 450  * Content written into the node before enqueue is guaranteed to be
 451  * consistent, but no other memory ordering is ensured.
 452  * Mutual exlusion with cds_wfcq_splice_blocking and dequeue lock is
 453  * ensured.
 454  * It is valid to reuse and free a dequeued node immediately.
 455  */
 456 static inline struct cds_wfcq_node *
 457 _cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
 458                 struct cds_wfcq_tail *tail)
 459 {
 460         struct cds_wfcq_node *retval;
 461
 462         _cds_wfcq_dequeue_lock(head, tail);
 463         retval = ___cds_wfcq_dequeue_blocking(head, tail);
 464         _cds_wfcq_dequeue_unlock(head, tail);
 465         return retval;
 466 }
 467
 468 /*
 469  * cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
 470  *
 471  * Dequeue all nodes from src_q.
 472  * dest_q must be already initialized.
 473  * Content written into the node before enqueue is guaranteed to be
 474  * consistent, but no other memory ordering is ensured.
 475  * Mutual exlusion with cds_wfcq_dequeue_blocking and dequeue lock is
 476  * ensured.
 477  */
 478 static inline void
 479 _cds_wfcq_splice_blocking(
 480                 struct cds_wfcq_head *dest_q_head,
 481                 struct cds_wfcq_tail *dest_q_tail,
 482                 struct cds_wfcq_head *src_q_head,
 483                 struct cds_wfcq_tail *src_q_tail)
 484 {
 485         _cds_wfcq_dequeue_lock(src_q_head, src_q_tail);
 486         ___cds_wfcq_splice_blocking(dest_q_head, dest_q_tail,
 487                         src_q_head, src_q_tail);
 488         _cds_wfcq_dequeue_unlock(src_q_head, src_q_tail);
 489 }
 490
 491 #ifdef __cplusplus
 492 }
 493 #endif
 494
 495 #endif /* _URCU_WFCQUEUE_STATIC_H */