urcu/static/wfcqueue.h

   1 #ifndef _URCU_WFCQUEUE_STATIC_H
   2 #define _URCU_WFCQUEUE_STATIC_H
   3
   4 /*
   5  * wfcqueue-static.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
  64 #define WFCQ_ADAPT_ATTEMPTS             10      /* Retry if being set */
  65 #define WFCQ_WAIT                       10      /* Wait 10 ms if being set */
  66
  67 /*
  68  * cds_wfcq_node_init: initialize wait-free queue node.
  69  */
  70 static inline void _cds_wfcq_node_init(struct cds_wfcq_node *node)
  71 {
  72         node->next = NULL;
  73 }
  74
  75 /*
  76  * cds_wfcq_init: initialize wait-free queue.
  77  */
  78 static inline void _cds_wfcq_init(struct cds_wfcq_head *head,
  79                 struct cds_wfcq_tail *tail)
  80 {
  81         int ret;
  82
  83         /* Set queue head and tail */
  84         _cds_wfcq_node_init(&head->node);
  85         tail->p = &head->node;
  86         ret = pthread_mutex_init(&head->lock, NULL);
  87         assert(!ret);
  88 }
  89
  90 /*
  91  * cds_wfcq_empty: return whether wait-free queue is empty.
  92  *
  93  * No memory barrier is issued. No mutual exclusion is required.
  94  */
  95 static inline bool _cds_wfcq_empty(struct cds_wfcq_head *head,
  96                 struct cds_wfcq_tail *tail)
  97 {
  98         /*
  99          * Queue is empty if no node is pointed by head->node.next nor
 100          * tail->p. Even though the tail->p check is sufficient to find
 101          * out of the queue is empty, we first check head->node.next as a
 102          * common case to ensure that dequeuers do not frequently access
 103          * enqueuer's tail->p cache line.
 104          */
 105         return CMM_LOAD_SHARED(head->node.next) == NULL
 106                 && CMM_LOAD_SHARED(tail->p) == &head->node;
 107 }
 108
 109 static inline void _cds_wfcq_dequeue_lock(struct cds_wfcq_head *head,
 110                 struct cds_wfcq_tail *tail)
 111 {
 112         int ret;
 113
 114         ret = pthread_mutex_lock(&head->lock);
 115         assert(!ret);
 116 }
 117
 118 static inline void _cds_wfcq_dequeue_unlock(struct cds_wfcq_head *head,
 119                 struct cds_wfcq_tail *tail)
 120 {
 121         int ret;
 122
 123         ret = pthread_mutex_unlock(&head->lock);
 124         assert(!ret);
 125 }
 126
 127 static inline void ___cds_wfcq_append(struct cds_wfcq_head *head,
 128                 struct cds_wfcq_tail *tail,
 129                 struct cds_wfcq_node *new_head,
 130                 struct cds_wfcq_node *new_tail)
 131 {
 132         struct cds_wfcq_node *old_tail;
 133
 134         /*
 135          * Implicit memory barrier before uatomic_xchg() orders earlier
 136          * stores to data structure containing node and setting
 137          * node->next to NULL before publication.
 138          */
 139         old_tail = uatomic_xchg(&tail->p, new_tail);
 140
 141         /*
 142          * Implicit memory barrier after uatomic_xchg() orders store to
 143          * q->tail before store to old_tail->next.
 144          *
 145          * At this point, dequeuers see a NULL tail->p->next, which
 146          * indicates that the queue is being appended to. The following
 147          * store will append "node" to the queue from a dequeuer
 148          * perspective.
 149          */
 150         CMM_STORE_SHARED(old_tail->next, new_head);
 151 }
 152
 153 /*
 154  * cds_wfcq_enqueue: enqueue a node into a wait-free queue.
 155  *
 156  * Issues a full memory barrier before enqueue. No mutual exclusion is
 157  * required.
 158  */
 159 static inline void _cds_wfcq_enqueue(struct cds_wfcq_head *head,
 160                 struct cds_wfcq_tail *tail,
 161                 struct cds_wfcq_node *new_tail)
 162 {
 163         ___cds_wfcq_append(head, tail, new_tail, new_tail);
 164 }
 165
 166 /*
 167  * Waiting for enqueuer to complete enqueue and return the next node.
 168  */
 169 static inline struct cds_wfcq_node *
 170 ___cds_wfcq_node_sync_next(struct cds_wfcq_node *node)
 171 {
 172         struct cds_wfcq_node *next;
 173         int attempt = 0;
 174
 175         /*
 176          * Adaptative busy-looping waiting for enqueuer to complete enqueue.
 177          */
 178         while ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 179                 if (++attempt >= WFCQ_ADAPT_ATTEMPTS) {
 180                         poll(NULL, 0, WFCQ_WAIT);       /* Wait for 10ms */
 181                         attempt = 0;
 182                 } else {
 183                         caa_cpu_relax();
 184                 }
 185         }
 186
 187         return next;
 188 }
 189
 190 /*
 191  * __cds_wfcq_first_blocking: get first node of a queue, without dequeuing.
 192  *
 193  * Content written into the node before enqueue is guaranteed to be
 194  * consistent, but no other memory ordering is ensured.
 195  * Should be called with cds_wfcq_dequeue_lock() held.
 196  *
 197  * Used by for-like iteration macros in urcu/wfqueue.h:
 198  * __cds_wfcq_for_each_blocking()
 199  * __cds_wfcq_for_each_blocking_safe()
 200  */
 201 static inline struct cds_wfcq_node *
 202 ___cds_wfcq_first_blocking(struct cds_wfcq_head *head,
 203                 struct cds_wfcq_tail *tail)
 204 {
 205         struct cds_wfcq_node *node;
 206
 207         if (_cds_wfcq_empty(head, tail))
 208                 return NULL;
 209         node = ___cds_wfcq_node_sync_next(&head->node);
 210         /* Load head->node.next before loading node's content */
 211         cmm_smp_read_barrier_depends();
 212         return node;
 213 }
 214
 215 /*
 216  * __cds_wfcq_next_blocking: get next node of a queue, without dequeuing.
 217  *
 218  * Content written into the node before enqueue is guaranteed to be
 219  * consistent, but no other memory ordering is ensured.
 220  * Should be called with cds_wfcq_dequeue_lock() held.
 221  *
 222  * Used by for-like iteration macros in urcu/wfqueue.h:
 223  * __cds_wfcq_for_each_blocking()
 224  * __cds_wfcq_for_each_blocking_safe()
 225  */
 226 static inline struct cds_wfcq_node *
 227 ___cds_wfcq_next_blocking(struct cds_wfcq_head *head,
 228                 struct cds_wfcq_tail *tail,
 229                 struct cds_wfcq_node *node)
 230 {
 231         struct cds_wfcq_node *next;
 232
 233         /*
 234          * Even though the following tail->p check is sufficient to find
 235          * out if we reached the end of the queue, we first check
 236          * node->next as a common case to ensure that iteration on nodes
 237          * do not frequently access enqueuer's tail->p cache line.
 238          */
 239         if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 240                 /* Load node->next before tail->p */
 241                 cmm_smp_rmb();
 242                 if (CMM_LOAD_SHARED(tail->p) == node)
 243                         return NULL;
 244                 next = ___cds_wfcq_node_sync_next(node);
 245         }
 246         /* Load node->next before loading next's content */
 247         cmm_smp_read_barrier_depends();
 248         return next;
 249 }
 250
 251 /*
 252  * __cds_wfcq_dequeue_blocking: dequeue a node from the queue.
 253  *
 254  * No need to go on a waitqueue here, as there is no possible state in which the
 255  * list could cause dequeue to busy-loop needlessly while waiting for another
 256  * thread to be scheduled. The queue appears empty until tail->next is set by
 257  * enqueue.
 258  *
 259  * Content written into the node before enqueue is guaranteed to be
 260  * consistent, but no other memory ordering is ensured.
 261  * It is valid to reuse and free a dequeued node immediately.
 262  * Should be called with cds_wfcq_dequeue_lock() held.
 263  */
 264 static inline struct cds_wfcq_node *
 265 ___cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
 266                 struct cds_wfcq_tail *tail)
 267 {
 268         struct cds_wfcq_node *node, *next;
 269
 270         if (_cds_wfcq_empty(head, tail))
 271                 return NULL;
 272
 273         node = ___cds_wfcq_node_sync_next(&head->node);
 274
 275         if ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 276                 /*
 277                  * @node is probably the only node in the queue.
 278                  * Try to move the tail to &q->head.
 279                  * q->head.next is set to NULL here, and stays
 280                  * NULL if the cmpxchg succeeds. Should the
 281                  * cmpxchg fail due to a concurrent enqueue, the
 282                  * q->head.next will be set to the next node.
 283                  * The implicit memory barrier before
 284                  * uatomic_cmpxchg() orders load node->next
 285                  * before loading q->tail.
 286                  * The implicit memory barrier before uatomic_cmpxchg
 287                  * orders load q->head.next before loading node's
 288                  * content.
 289                  */
 290                 _cds_wfcq_node_init(&head->node);
 291                 if (uatomic_cmpxchg(&tail->p, node, &head->node) == node)
 292                         return node;
 293                 next = ___cds_wfcq_node_sync_next(node);
 294         }
 295
 296         /*
 297          * Move queue head forward.
 298          */
 299         head->node.next = next;
 300
 301         /* Load q->head.next before loading node's content */
 302         cmm_smp_read_barrier_depends();
 303         return node;
 304 }
 305
 306 /*
 307  * __cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
 308  *
 309  * Dequeue all nodes from src_q.
 310  * dest_q must be already initialized.
 311  * Should be called with cds_wfcq_dequeue_lock() held on src_q.
 312  */
 313 static inline void
 314 ___cds_wfcq_splice_blocking(
 315                 struct cds_wfcq_head *dest_q_head,
 316                 struct cds_wfcq_tail *dest_q_tail,
 317                 struct cds_wfcq_head *src_q_head,
 318                 struct cds_wfcq_tail *src_q_tail)
 319 {
 320         struct cds_wfcq_node *head, *tail;
 321
 322         if (_cds_wfcq_empty(src_q_head, src_q_tail))
 323                 return;
 324
 325         head = ___cds_wfcq_node_sync_next(&src_q_head->node);
 326         _cds_wfcq_node_init(&src_q_head->node);
 327
 328         /*
 329          * Memory barrier implied before uatomic_xchg() orders store to
 330          * src_q->head before store to src_q->tail. This is required by
 331          * concurrent enqueue on src_q, which exchanges the tail before
 332          * updating the previous tail's next pointer.
 333          */
 334         tail = uatomic_xchg(&src_q_tail->p, &src_q_head->node);
 335
 336         /*
 337          * Append the spliced content of src_q into dest_q. Does not
 338          * require mutual exclusion on dest_q (wait-free).
 339          */
 340         ___cds_wfcq_append(dest_q_head, dest_q_tail, head, tail);
 341 }
 342
 343 /*
 344  * cds_wfcq_dequeue_blocking: dequeue a node from a wait-free queue.
 345  *
 346  * Content written into the node before enqueue is guaranteed to be
 347  * consistent, but no other memory ordering is ensured.
 348  * Mutual exlusion with (and only with) cds_wfcq_splice_blocking is
 349  * ensured.
 350  * It is valid to reuse and free a dequeued node immediately.
 351  */
 352 static inline struct cds_wfcq_node *
 353 _cds_wfcq_dequeue_blocking(struct cds_wfcq_head *head,
 354                 struct cds_wfcq_tail *tail)
 355 {
 356         struct cds_wfcq_node *retval;
 357
 358         _cds_wfcq_dequeue_lock(head, tail);
 359         retval = ___cds_wfcq_dequeue_blocking(head, tail);
 360         _cds_wfcq_dequeue_unlock(head, tail);
 361         return retval;
 362 }
 363
 364 /*
 365  * cds_wfcq_splice_blocking: enqueue all src_q nodes at the end of dest_q.
 366  *
 367  * Dequeue all nodes from src_q.
 368  * dest_q must be already initialized.
 369  * Content written into the node before enqueue is guaranteed to be
 370  * consistent, but no other memory ordering is ensured.
 371  * Mutual exlusion with (and only with) cds_wfcq_dequeue_blocking is
 372  * ensured.
 373  */
 374 static inline void
 375 _cds_wfcq_splice_blocking(
 376                 struct cds_wfcq_head *dest_q_head,
 377                 struct cds_wfcq_tail *dest_q_tail,
 378                 struct cds_wfcq_head *src_q_head,
 379                 struct cds_wfcq_tail *src_q_tail)
 380 {
 381         _cds_wfcq_dequeue_lock(src_q_head, src_q_tail);
 382         ___cds_wfcq_splice_blocking(dest_q_head, dest_q_tail,
 383                         src_q_head, src_q_tail);
 384         _cds_wfcq_dequeue_unlock(src_q_head, src_q_tail);
 385 }
 386
 387 #ifdef __cplusplus
 388 }
 389 #endif
 390
 391 #endif /* _URCU_WFCQUEUE_STATIC_H */