urcu/static/rculfqueue.h

   1 #ifndef _URCU_RCULFQUEUE_STATIC_H
   2 #define _URCU_RCULFQUEUE_STATIC_H
   3
   4 /*
   5  * rculfqueue-static.h
   6  *
   7  * Userspace RCU library - Lock-Free RCU Queue
   8  *
   9  * Copyright 2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  10  *
  11  * TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See rculfqueue.h for linking
  12  * dynamically with the userspace rcu library.
  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 <urcu-call-rcu.h>
  30 #include <urcu/uatomic.h>
  31 #include <assert.h>
  32 #include <errno.h>
  33
  34 #ifdef __cplusplus
  35 extern "C" {
  36 #endif
  37
  38 struct cds_lfq_node_rcu_dummy {
  39         struct cds_lfq_node_rcu parent;
  40         struct rcu_head head;
  41         struct cds_lfq_queue_rcu *q;
  42 };
  43
  44 /*
  45  * Lock-free RCU queue. Enqueue and dequeue operations hold a RCU read
  46  * lock to deal with cmpxchg ABA problem. This queue is *not* circular:
  47  * head points to the oldest node, tail points to the newest node.
  48  * A dummy node is kept to ensure enqueue and dequeue can always proceed
  49  * concurrently. Keeping a separate head and tail helps with large
  50  * queues: enqueue and dequeue can proceed concurrently without
  51  * wrestling for exclusive access to the same variables.
  52  *
  53  * Dequeue retry if it detects that it would be dequeueing the last node
  54  * (it means a dummy node dequeue-requeue is in progress). This ensures
  55  * that there is always at least one node in the queue.
  56  *
  57  * In the dequeue operation, we internally reallocate the dummy node
  58  * upon dequeue/requeue and use call_rcu to free the old one after a
  59  * grace period.
  60  */
  61
  62 static inline
  63 struct cds_lfq_node_rcu *make_dummy(struct cds_lfq_queue_rcu *q,
  64                                     struct cds_lfq_node_rcu *next)
  65 {
  66         struct cds_lfq_node_rcu_dummy *dummy;
  67
  68         dummy = malloc(sizeof(struct cds_lfq_node_rcu_dummy));
  69         assert(dummy);
  70         dummy->parent.next = next;
  71         dummy->parent.dummy = 1;
  72         dummy->q = q;
  73         return &dummy->parent;
  74 }
  75
  76 static inline
  77 void free_dummy_cb(struct rcu_head *head)
  78 {
  79         struct cds_lfq_node_rcu_dummy *dummy =
  80                 caa_container_of(head, struct cds_lfq_node_rcu_dummy, head);
  81         free(dummy);
  82 }
  83
  84 static inline
  85 void rcu_free_dummy(struct cds_lfq_node_rcu *node)
  86 {
  87         struct cds_lfq_node_rcu_dummy *dummy;
  88
  89         assert(node->dummy);
  90         dummy = caa_container_of(node, struct cds_lfq_node_rcu_dummy, parent);
  91         call_rcu(&dummy->head, free_dummy_cb);
  92 }
  93
  94 static inline
  95 void free_dummy(struct cds_lfq_node_rcu *node)
  96 {
  97         struct cds_lfq_node_rcu_dummy *dummy;
  98
  99         assert(node->dummy);
 100         dummy = caa_container_of(node, struct cds_lfq_node_rcu_dummy, parent);
 101         free(dummy);
 102 }
 103
 104 static inline
 105 void _cds_lfq_node_init_rcu(struct cds_lfq_node_rcu *node)
 106 {
 107         node->next = NULL;
 108         node->dummy = 0;
 109 }
 110
 111 static inline
 112 void _cds_lfq_init_rcu(struct cds_lfq_queue_rcu *q)
 113 {
 114         q->tail = make_dummy(q, NULL);
 115         q->head = q->tail;
 116 }
 117
 118 /*
 119  * The queue should be emptied before calling destroy.
 120  *
 121  * Return 0 on success, -EPERM if queue is not empty.
 122  */
 123 static inline
 124 int _cds_lfq_destroy_rcu(struct cds_lfq_queue_rcu *q)
 125 {
 126         struct cds_lfq_node_rcu *head;
 127
 128         head = rcu_dereference(q->head);
 129         if (!(head->dummy && head->next == NULL))
 130                 return -EPERM;  /* not empty */
 131         free_dummy(head);
 132         return 0;
 133 }
 134
 135 /*
 136  * Acts as a RCU reader.
 137  */
 138 static inline
 139 void _cds_lfq_enqueue_rcu(struct cds_lfq_queue_rcu *q,
 140                           struct cds_lfq_node_rcu *node)
 141 {
 142         /*
 143          * uatomic_cmpxchg() implicit memory barrier orders earlier stores to
 144          * node before publication.
 145          */
 146
 147         for (;;) {
 148                 struct cds_lfq_node_rcu *tail, *next;
 149
 150                 rcu_read_lock();
 151                 tail = rcu_dereference(q->tail);
 152                 next = uatomic_cmpxchg(&tail->next, NULL, node);
 153                 if (next == NULL) {
 154                         /*
 155                          * Tail was at the end of queue, we successfully
 156                          * appended to it. Now move tail (another
 157                          * enqueue might beat us to it, that's fine).
 158                          */
 159                         (void) uatomic_cmpxchg(&q->tail, tail, node);
 160                         rcu_read_unlock();
 161                         return;
 162                 } else {
 163                         /*
 164                          * Failure to append to current tail.
 165                          * Help moving tail further and retry.
 166                          */
 167                         (void) uatomic_cmpxchg(&q->tail, tail, next);
 168                         rcu_read_unlock();
 169                         continue;
 170                 }
 171         }
 172 }
 173
 174 static inline
 175 void enqueue_dummy(struct cds_lfq_queue_rcu *q)
 176 {
 177         struct cds_lfq_node_rcu *node;
 178
 179         /* We need to reallocate to protect from ABA. */
 180         node = make_dummy(q, NULL);
 181         _cds_lfq_enqueue_rcu(q, node);
 182 }
 183
 184 /*
 185  * Acts as a RCU reader.
 186  *
 187  * The caller must wait for a grace period to pass before freeing the returned
 188  * node or modifying the cds_lfq_node_rcu structure.
 189  * Returns NULL if queue is empty.
 190  */
 191 static inline
 192 struct cds_lfq_node_rcu *_cds_lfq_dequeue_rcu(struct cds_lfq_queue_rcu *q)
 193 {
 194         for (;;) {
 195                 struct cds_lfq_node_rcu *head, *next;
 196
 197                 rcu_read_lock();
 198                 head = rcu_dereference(q->head);
 199                 next = rcu_dereference(head->next);
 200                 if (head->dummy && next == NULL) {
 201                         rcu_read_unlock();
 202                         return NULL;    /* empty */
 203                 }
 204                 /*
 205                  * We never, ever allow dequeue to get to a state where
 206                  * the queue is empty (we need at least one node in the
 207                  * queue). This is ensured by checking if the head next
 208                  * is NULL, which means we need to enqueue a dummy node
 209                  * before we can hope dequeuing anything.
 210                  */
 211                 if (!next) {
 212                         enqueue_dummy(q);
 213                         next = rcu_dereference(head->next);
 214                 }
 215                 if (uatomic_cmpxchg(&q->head, head, next) != head) {
 216                         rcu_read_unlock();
 217                         continue;       /* Concurrently pushed. */
 218                 }
 219                 if (head->dummy) {
 220                         /* Free dummy after grace period. */
 221                         rcu_free_dummy(head);
 222                         rcu_read_unlock();
 223                         continue;       /* try again */
 224                 }
 225                 rcu_read_unlock();
 226                 return head;
 227         }
 228 }
 229
 230 #ifdef __cplusplus
 231 }
 232 #endif
 233
 234 #endif /* _URCU_RCULFQUEUE_STATIC_H */