include/urcu/static/rculfqueue.h

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