urcu/static/wfqueue.h

   1 #ifndef _URCU_WFQUEUE_STATIC_H
   2 #define _URCU_WFQUEUE_STATIC_H
   3
   4 /*
   5  * wfqueue-static.h
   6  *
   7  * Userspace RCU library - Queue with Wait-Free Enqueue/Blocking Dequeue
   8  *
   9  * TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See wfqueue.h for linking
  10  * dynamically with the userspace rcu library.
  11  *
  12  * Copyright 2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  13  *
  14  * This library is free software; you can redistribute it and/or
  15  * modify it under the terms of the GNU Lesser General Public
  16  * License as published by the Free Software Foundation; either
  17  * version 2.1 of the License, or (at your option) any later version.
  18  *
  19  * This library is distributed in the hope that it will be useful,
  20  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  21  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  22  * Lesser General Public License for more details.
  23  *
  24  * You should have received a copy of the GNU Lesser General Public
  25  * License along with this library; if not, write to the Free Software
  26  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  27  */
  28
  29 #include <pthread.h>
  30 #include <assert.h>
  31 #include <poll.h>
  32 #include <urcu/compiler.h>
  33 #include <urcu/uatomic.h>
  34
  35 #ifdef __cplusplus
  36 extern "C" {
  37 #endif
  38
  39 /*
  40  * Queue with wait-free enqueue/blocking dequeue.
  41  * This implementation adds a dummy head node when the queue is empty to ensure
  42  * we can always update the queue locklessly.
  43  *
  44  * Inspired from half-wait-free/half-blocking queue implementation done by
  45  * Paul E. McKenney.
  46  */
  47
  48 #define WFQ_ADAPT_ATTEMPTS              10      /* Retry if being set */
  49 #define WFQ_WAIT                        10      /* Wait 10 ms if being set */
  50
  51 static inline void _cds_wfq_node_init(struct cds_wfq_node *node)
  52 {
  53         node->next = NULL;
  54 }
  55
  56 static inline void _cds_wfq_init(struct cds_wfq_queue *q)
  57 {
  58         int ret;
  59
  60         _cds_wfq_node_init(&q->dummy);
  61         /* Set queue head and tail */
  62         q->head = &q->dummy;
  63         q->tail = &q->dummy.next;
  64         ret = pthread_mutex_init(&q->lock, NULL);
  65         assert(!ret);
  66 }
  67
  68 static inline void _cds_wfq_enqueue(struct cds_wfq_queue *q,
  69                                     struct cds_wfq_node *node)
  70 {
  71         struct cds_wfq_node **old_tail;
  72
  73         /*
  74          * uatomic_xchg() implicit memory barrier orders earlier stores to data
  75          * structure containing node and setting node->next to NULL before
  76          * publication.
  77          */
  78         old_tail = uatomic_xchg(&q->tail, &node->next);
  79         /*
  80          * At this point, dequeuers see a NULL old_tail->next, which indicates
  81          * that the queue is being appended to. The following store will append
  82          * "node" to the queue from a dequeuer perspective.
  83          */
  84         CMM_STORE_SHARED(*old_tail, node);
  85 }
  86
  87 /*
  88  * It is valid to reuse and free a dequeued node immediately.
  89  *
  90  * No need to go on a waitqueue here, as there is no possible state in which the
  91  * list could cause dequeue to busy-loop needlessly while waiting for another
  92  * thread to be scheduled. The queue appears empty until tail->next is set by
  93  * enqueue.
  94  */
  95 static inline struct cds_wfq_node *
  96 ___cds_wfq_dequeue_blocking(struct cds_wfq_queue *q)
  97 {
  98         struct cds_wfq_node *node, *next;
  99         int attempt = 0;
 100
 101         /*
 102          * Queue is empty if it only contains the dummy node.
 103          */
 104         if (q->head == &q->dummy && CMM_LOAD_SHARED(q->tail) == &q->dummy.next)
 105                 return NULL;
 106         node = q->head;
 107
 108         /*
 109          * Adaptative busy-looping waiting for enqueuer to complete enqueue.
 110          */
 111         while ((next = CMM_LOAD_SHARED(node->next)) == NULL) {
 112                 if (++attempt >= WFQ_ADAPT_ATTEMPTS) {
 113                         poll(NULL, 0, WFQ_WAIT);        /* Wait for 10ms */
 114                         attempt = 0;
 115                 } else
 116                         caa_cpu_relax();
 117         }
 118         /*
 119          * Move queue head forward.
 120          */
 121         q->head = next;
 122         /*
 123          * Requeue dummy node if we just dequeued it.
 124          */
 125         if (node == &q->dummy) {
 126                 _cds_wfq_node_init(node);
 127                 _cds_wfq_enqueue(q, node);
 128                 return ___cds_wfq_dequeue_blocking(q);
 129         }
 130         return node;
 131 }
 132
 133 static inline struct cds_wfq_node *
 134 _cds_wfq_dequeue_blocking(struct cds_wfq_queue *q)
 135 {
 136         struct cds_wfq_node *retnode;
 137         int ret;
 138
 139         ret = pthread_mutex_lock(&q->lock);
 140         assert(!ret);
 141         retnode = ___cds_wfq_dequeue_blocking(q);
 142         ret = pthread_mutex_unlock(&q->lock);
 143         assert(!ret);
 144         return retnode;
 145 }
 146
 147 #ifdef __cplusplus
 148 }
 149 #endif
 150
 151 #endif /* _URCU_WFQUEUE_STATIC_H */