urcu/rcuwfqueue.h

   1 /*
   2  * rcuwfqueue.h
   3  *
   4  * Userspace RCU library - RCU Queue with Wait-Free Enqueue/Blocking Dequeue
   5  *
   6  * Copyright 2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   7  *
   8  * This library is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License as published by the Free Software Foundation; either
  11  * version 2.1 of the License, or (at your option) any later version.
  12  *
  13  * This library is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  16  * Lesser General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU Lesser General Public
  19  * License along with this library; if not, write to the Free Software
  20  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  21  */
  22
  23 #include <urcu/urcu_ref.h>
  24 #include <assert.h>
  25
  26 #if (!defined(_GNU_SOURCE) && !defined(_LGPL_SOURCE))
  27 #error "Dynamic loader LGPL wrappers not implemented yet"
  28 #endif
  29
  30 /*
  31  * RCU queue with wait-free enqueue/blocking dequeue using reference counting.
  32  * Enqueue and dequeue operations hold a RCU read lock to deal with cmpxchg ABA
  33  * problem. This implementation keeps a dummy head node to ensure we can always
  34  * update the queue locklessly. Given that this is a queue, the dummy head node
  35  * must always advance as we dequeue entries. Therefore, we keep a reference
  36  * count on each entry we are dequeueing, so they can be kept as dummy head node
  37  * until the next dequeue, at which point their reference count will be
  38  * decremented.
  39  */
  40
  41 #define URCU_WFQ_PERMANENT_REF          128
  42
  43 struct rcu_wfq_node {
  44         struct rcu_wfq_node *next;
  45         struct urcu_ref ref;
  46 };
  47
  48 struct rcu_wfq_queue {
  49         struct rcu_wfq_node *head, *tail;
  50         struct rcu_wfq_node init;       /* Dummy initialization node */
  51 };
  52
  53 void rcu_wfq_node_init(struct rcu_wfq_node *node)
  54 {
  55         node->next = NULL;
  56         urcu_ref_init(&node->ref);
  57 }
  58
  59 void rcu_wfq_init(struct rcu_wfq_queue *q)
  60 {
  61         rcu_wfq_node_init(&q->init);
  62         /* Make sure the initial node is never freed. */
  63         urcu_ref_set(&q->init.ref, URCU_WFQ_PERMANENT_REF);
  64         /* Set queue end */
  65         q->head = q->tail = &q->init;
  66 }
  67
  68 void rcu_wfq_enqueue(struct rcu_wfq_queue *q, struct rcu_wfq_node *node)
  69 {
  70         struct rcu_wfq_node *old_tail;
  71
  72         urcu_ref_get(&node->ref);
  73         /*
  74          * uatomic_xchg() implicit memory barrier orders earlier stores to node
  75          * (setting it to NULL and incrementing the refcount) before
  76          * publication.
  77          */
  78         old_tail = uatomic_xchg(&q->tail, node);
  79         /*
  80          * At this point, dequeuers see a NULL old_tail->next, which indicates
  81          * end of queue. The following store will append "node" to the queue
  82          * from a dequeuer perspective.
  83          */
  84         STORE_SHARED(old_tail->next, node);
  85 }
  86
  87 /*
  88  * The entry returned by dequeue must be taken care of by doing a urcu_ref_put,
  89  * which calls the release primitive when the reference count drops to zero. A
  90  * grace period must be waited before performing the actual memory reclamation
  91  * in the release primitive. The wfq node returned by dequeue must not be
  92  * modified/re-used/freed until the reference count reaches zero and a grace
  93  * period has elapsed (after the refcount reached 0).
  94  *
  95  * TODO: implement adaptative busy-wait and wait/wakeup scheme rather than busy
  96  * loops. Better for UP.
  97  */
  98 struct rcu_wfq_node *
  99 rcu_wfq_dequeue_blocking(struct rcu_wfq_queue *q,
 100                          void (*release)(struct urcu_ref *))
 101 {
 102         for (;;) {
 103                 struct rcu_wfq_node *head, *next;
 104
 105                 rcu_read_lock();
 106                 head = rcu_dereference(q->head);
 107                 next = rcu_dereference(head->next);
 108                 if (next) {
 109                         if (uatomic_cmpxchg(&q->head, head, next) == head) {
 110                                 rcu_read_unlock();
 111                                 urcu_ref_put(&head->ref, release);
 112                                 return next;
 113                         } else {
 114                                 /* Concurrently pushed, retry */
 115                                 rcu_read_unlock();
 116                                 continue;
 117                         }
 118                 } else {
 119                         /* Empty */
 120                         rcu_read_unlock();
 121                         return NULL;
 122                 }
 123         }
 124 }