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 /* A urcu implementation header should be already included. */
  34
  35 #ifdef __cplusplus
  36 extern "C" {
  37 #endif
  38
  39 struct cds_lfq_node_rcu_dummy {
  40         struct cds_lfq_node_rcu parent;
  41         struct rcu_head head;
  42         struct cds_lfq_queue_rcu *q;
  43 };
  44
  45 /*
  46  * Lock-free RCU queue. Enqueue and dequeue operations hold a RCU read
  47  * lock to deal with cmpxchg ABA problem. This queue is *not* circular:
  48  * head points to the oldest node, tail points to the newest node.
  49  * Dummy nodes are kept to ensure enqueue and dequeue can always proceed
  50  * concurrently. Keeping a separate head and tail helps with large
  51  * queues: enqueue and dequeue can proceed concurrently without
  52  * wrestling for exclusive access to the same variables.
  53  *
  54  * We keep two dummy nodes in the queue to distinguish between empty queue
  55  * state and intermediate state while a dummy node dequeue/requeue is
  56  * being performed. Dequeue retry if it detects that it would be
  57  * dequeueing the last node (it means a dummy node dequeue-requeue is in
  58  * progress). This ensures that there is always at least one node in
  59  * the queue. In a situation where the two dummy nodes are being
  60  * requeued (they therefore don't appear in the queue at a given
  61  * moment), we are certain that there is at least one non-dummy node in
  62  * the queue (ensured by the test for NULL next node upon dequeue).
  63  *
  64  * In the dequeue operation, we internally reallocate the dummy nodes
  65  * upon dequeue/requeue and use call_rcu to free them after a grace
  66  * period.
  67  */
  68
  69 static inline
  70 int is_dummy(struct cds_lfq_node_rcu *node)
  71 {
  72         return ((unsigned long) node) & 0x1UL;
  73 }
  74
  75 static inline
  76 struct cds_lfq_node_rcu *make_dummy(struct cds_lfq_queue_rcu *q,
  77                                     struct cds_lfq_node_rcu *next)
  78 {
  79         struct cds_lfq_node_rcu_dummy *dummy;
  80
  81         dummy = malloc(sizeof(struct cds_lfq_node_rcu_dummy));
  82         dummy->parent.next = next;
  83         dummy->q = q;
  84         return (struct cds_lfq_node_rcu *) (((unsigned long) &dummy->parent) | 0x1UL);
  85 }
  86
  87 static inline
  88 struct cds_lfq_node_rcu *get_node(struct cds_lfq_node_rcu *node)
  89 {
  90         return (struct cds_lfq_node_rcu *) (((unsigned long )node) & ~0x1UL);
  91 }
  92
  93 static inline
  94 void free_dummy(struct rcu_head *head)
  95 {
  96         struct cds_lfq_node_rcu_dummy *dummy =
  97                 caa_container_of(head, struct cds_lfq_node_rcu_dummy, head);
  98         free(dummy);
  99 }
 100
 101 static inline
 102 void rcu_free_dummy(struct cds_lfq_node_rcu *node)
 103 {
 104         struct cds_lfq_node_rcu_dummy *dummy;
 105
 106         dummy = caa_container_of(get_node(node), struct cds_lfq_node_rcu_dummy,
 107                                  parent);
 108         dummy->q->queue_call_rcu(&dummy->head, free_dummy);
 109 }
 110
 111 static inline
 112 void _cds_lfq_node_init_rcu(struct cds_lfq_node_rcu *node)
 113 {
 114         node->next = NULL;
 115 }
 116
 117 static inline
 118 void _cds_lfq_init_rcu(struct cds_lfq_queue_rcu *q,
 119                        void queue_call_rcu(struct rcu_head *head,
 120                                 void (*func)(struct rcu_head *head)))
 121 {
 122         q->tail = make_dummy(q, NULL);
 123         q->head = make_dummy(q, q->tail);
 124         q->queue_call_rcu = queue_call_rcu;
 125 }
 126
 127 /*
 128  * The queue should be emptied before calling destroy.
 129  *
 130  * Return 0 on success, -EPERM if queue is not empty.
 131  */
 132 static inline
 133 int _cds_lfq_destroy_rcu(struct cds_lfq_queue_rcu *q)
 134 {
 135         struct cds_lfq_node_rcu *head, *next;
 136
 137         head = rcu_dereference(q->head);
 138         next = rcu_dereference(get_node(head)->next);
 139         if (!(is_dummy(head) && is_dummy(next) && get_node(next)->next == NULL))
 140                 return -EPERM;  /* not empty */
 141         rcu_free_dummy(head);
 142         rcu_free_dummy(next);
 143         return 0;
 144 }
 145
 146 /*
 147  * Should be called under rcu read lock critical section.
 148  */
 149 static inline
 150 void _cds_lfq_enqueue_rcu(struct cds_lfq_queue_rcu *q,
 151                           struct cds_lfq_node_rcu *node)
 152 {
 153         /*
 154          * uatomic_cmpxchg() implicit memory barrier orders earlier stores to
 155          * node before publication.
 156          */
 157
 158         for (;;) {
 159                 struct cds_lfq_node_rcu *tail, *next;
 160
 161                 tail = rcu_dereference(q->tail);
 162                 next = uatomic_cmpxchg(&get_node(tail)->next, NULL, node);
 163                 if (next == NULL) {
 164                         /*
 165                          * Tail was at the end of queue, we successfully
 166                          * appended to it. Now move tail (another
 167                          * enqueue might beat us to it, that's fine).
 168                          */
 169                         (void) uatomic_cmpxchg(&q->tail, tail, node);
 170                         return;
 171                 } else {
 172                         /*
 173                          * Failure to append to current tail.
 174                          * Help moving tail further and retry.
 175                          */
 176                         (void) uatomic_cmpxchg(&q->tail, tail, next);
 177                         continue;
 178                 }
 179         }
 180 }
 181
 182 /*
 183  * Should be called under rcu read lock critical section.
 184  *
 185  * The caller must wait for a grace period to pass before freeing the returned
 186  * node or modifying the cds_lfq_node_rcu structure.
 187  * Returns NULL if queue is empty.
 188  */
 189 static inline
 190 struct cds_lfq_node_rcu *_cds_lfq_dequeue_rcu(struct cds_lfq_queue_rcu *q)
 191 {
 192         for (;;) {
 193                 struct cds_lfq_node_rcu *head, *next;
 194
 195                 head = rcu_dereference(q->head);
 196                 next = rcu_dereference(get_node(head)->next);
 197                 if (is_dummy(head) && is_dummy(next) && get_node(next)->next == NULL)
 198                         return NULL;    /* empty */
 199                 /*
 200                  * We never, ever allow dequeue to get to a state where
 201                  * the queue is empty (we need at least one node in the
 202                  * queue). This is ensured by checking if the head next
 203                  * is NULL and retry in that case (this means a
 204                  * concurrent dummy node re-enqueue is in progress).
 205                  */
 206                 if (next) {
 207                         if (uatomic_cmpxchg(&q->head, head, next) == head) {
 208                                 if (is_dummy(head)) {
 209                                         struct cds_lfq_node_rcu *node;
 210                                         /*
 211                                          * Requeue dummy. We need to
 212                                          * reallocate to protect from
 213                                          * ABA.
 214                                          */
 215                                         rcu_free_dummy(head);
 216                                         node = make_dummy(q, NULL);
 217                                         _cds_lfq_enqueue_rcu(q, node);
 218                                         continue;       /* try again */
 219                                 }
 220                                 return head;
 221                         } else {
 222                                 /* Concurrently pushed, retry */
 223                                 continue;
 224                         }
 225                 } else {
 226                         /* Dummy node re-enqueue is in progress, retry. */
 227                         continue;
 228                 }
 229         }
 230 }
 231
 232 #ifdef __cplusplus
 233 }
 234 #endif
 235
 236 #endif /* _URCU_RCULFQUEUE_STATIC_H */