urcu/waitqueue-lifo.h

   1 #ifndef _URCU_WAITQUEUE_LIFO_H
   2 #define _URCU_WAITQUEUE_LIFO_H
   3
   4 /*
   5  * urcu/waitqueue-lifo.h
   6  *
   7  * Userspace RCU library - wait queue scheme with LIFO semantic
   8  *
   9  * Copyright (c) 2012-2014 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  10  *
  11  * This library is free software; you can redistribute it and/or
  12  * modify it under the terms of the GNU Lesser General Public
  13  * License as published by the Free Software Foundation; either
  14  * version 2.1 of the License, or (at your option) any later version.
  15  *
  16  * This library is distributed in the hope that it will be useful,
  17  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  19  * Lesser General Public License for more details.
  20  *
  21  * You should have received a copy of the GNU Lesser General Public
  22  * License along with this library; if not, write to the Free Software
  23  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  24  */
  25
  26 #include <urcu/uatomic.h>
  27 #include <urcu/lfstack.h>
  28 #include <urcu/futex.h>
  29
  30 /*
  31  * Number of busy-loop attempts before waiting on futex for grace period
  32  * batching.
  33  */
  34 #define URCU_WAIT_ATTEMPTS 1000
  35
  36 enum urcu_wait_state {
  37         /* URCU_WAIT_WAITING is compared directly (futex compares it). */
  38         URCU_WAIT_WAITING =     0,
  39         /* non-zero are used as masks. */
  40         URCU_WAIT_WAKEUP =      (1 << 0),
  41         URCU_WAIT_RUNNING =     (1 << 1),
  42         URCU_WAIT_TEARDOWN =    (1 << 2),
  43 };
  44
  45 struct urcu_wait_node {
  46         struct cds_lfs_node node;
  47         int32_t state;  /* enum urcu_wait_state */
  48 };
  49
  50 #define URCU_WAIT_NODE_INIT(name, _state)               \
  51         { .state = _state }
  52
  53 #define DEFINE_URCU_WAIT_NODE(name, state)              \
  54         struct urcu_wait_node name = URCU_WAIT_NODE_INIT(name, state)
  55
  56 #define DECLARE_URCU_WAIT_NODE(name)                    \
  57         struct urcu_wait_node name
  58
  59 struct urcu_wait_queue {
  60         struct __cds_lfs_stack stack;
  61 };
  62
  63 #define URCU_WAIT_QUEUE_HEAD_INIT(name)                 \
  64         { .stack.head = CDS_LFS_END, }
  65
  66 #define DECLARE_URCU_WAIT_QUEUE(name)                   \
  67         struct urcu_wait_queue name
  68
  69 #define DEFINE_URCU_WAIT_QUEUE(name)                    \
  70         struct urcu_wait_queue name = URCU_WAIT_QUEUE_HEAD_INIT(name)
  71
  72 static inline
  73 void urcu_wait_queue_init(struct urcu_wait_queue *queue)
  74 {
  75         __cds_lfs_init(&queue->stack);
  76 }
  77
  78 struct urcu_waiters {
  79         struct cds_lfs_head *head;
  80 };
  81
  82 /*
  83  * Add ourself atomically to a wait queue. Return 0 if queue was
  84  * previously empty, else return 1.
  85  * A full memory barrier is issued before being added to the wait queue.
  86  */
  87 static inline
  88 bool urcu_wait_add(struct urcu_wait_queue *queue,
  89                 struct urcu_wait_node *node)
  90 {
  91         return cds_lfs_push(&queue->stack, &node->node);
  92 }
  93
  94 /*
  95  * Atomically move all waiters from wait queue into our local struct
  96  * urcu_waiters.
  97  */
  98 static inline
  99 void urcu_move_waiters(struct urcu_waiters *waiters,
 100                 struct urcu_wait_queue *queue)
 101 {
 102         waiters->head = __cds_lfs_pop_all(&queue->stack);
 103 }
 104
 105 static inline
 106 void urcu_wait_set_state(struct urcu_wait_node *node,
 107                 enum urcu_wait_state state)
 108 {
 109         node->state = state;
 110 }
 111
 112 static inline
 113 void urcu_wait_or_state(struct urcu_wait_node *node,
 114                 enum urcu_wait_state state)
 115 {
 116         uatomic_or(&node->state, state);
 117 }
 118
 119 static inline
 120 void urcu_wait_node_init(struct urcu_wait_node *node,
 121                 enum urcu_wait_state state)
 122 {
 123         urcu_wait_set_state(node, state);
 124         cds_lfs_node_init(&node->node);
 125 }
 126
 127 /*
 128  * Note: urcu_adaptative_wake_up needs "value" to stay allocated
 129  * throughout its execution. In this scheme, the waiter owns the node
 130  * memory, and we only allow it to free this memory when it receives the
 131  * URCU_WAIT_TEARDOWN flag.
 132  * Return true if wakeup is performed, false if thread was already
 133  * running.
 134  */
 135 static inline
 136 bool urcu_adaptative_wake_up(struct urcu_wait_node *wait)
 137 {
 138         bool wakeup_performed = false;
 139
 140         cmm_smp_mb();
 141         /*
 142          * "or" of WAKEUP flag rather than "set" is useful for multiple
 143          * concurrent wakeup sources. Note that "WAIT_TEARDOWN" becomes
 144          * useless when we use multiple wakeup sources: lifetime of the
 145          * "value" should then be handled by the caller.
 146          */
 147         uatomic_or(&wait->state, URCU_WAIT_WAKEUP);
 148         if (!(uatomic_read(&wait->state) & URCU_WAIT_RUNNING)) {
 149                 futex_noasync(&wait->state, FUTEX_WAKE, 1, NULL, NULL, 0);
 150                 wakeup_performed = true;
 151         }
 152         /* Allow teardown of struct urcu_wait memory. */
 153         uatomic_or(&wait->state, URCU_WAIT_TEARDOWN);
 154         return wakeup_performed;
 155 }
 156
 157 /*
 158  * Caller must initialize "value" to URCU_WAIT_WAITING before passing its
 159  * memory to waker thread.
 160  */
 161 static inline
 162 void urcu_adaptative_busy_wait(struct urcu_wait_node *wait)
 163 {
 164         unsigned int i;
 165
 166         /* Load and test condition before read state */
 167         cmm_smp_rmb();
 168         for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
 169                 if (uatomic_read(&wait->state) != URCU_WAIT_WAITING)
 170                         goto skip_futex_wait;
 171                 caa_cpu_relax();
 172         }
 173         futex_noasync(&wait->state, FUTEX_WAIT,
 174                 URCU_WAIT_WAITING, NULL, NULL, 0);
 175 skip_futex_wait:
 176
 177         /* Tell waker thread than we are running. */
 178         uatomic_or(&wait->state, URCU_WAIT_RUNNING);
 179
 180         /*
 181          * Wait until waker thread lets us know it's ok to tear down
 182          * memory allocated for struct urcu_wait.
 183          */
 184         for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
 185                 if (uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN)
 186                         break;
 187                 caa_cpu_relax();
 188         }
 189         while (!(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN))
 190                 poll(NULL, 0, 10);
 191         assert(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN);
 192 }
 193
 194 /*
 195  * Need mutual exclusion against other wakeup and move waiters
 196  * operations. It is provided by the caller.
 197  */
 198 static inline
 199 int urcu_dequeue_wake_single(struct urcu_wait_queue *queue)
 200 {
 201         struct cds_lfs_node *node;
 202         struct urcu_wait_node *wait_node;
 203         int ret = 0;
 204
 205         node = __cds_lfs_pop(&queue->stack);
 206         if (!node)
 207                 return -ENOENT;
 208         wait_node = caa_container_of(node, struct urcu_wait_node, node);
 209         CMM_STORE_SHARED(wait_node->node.next, NULL);
 210         /* Don't wake already running threads */
 211         if (!(wait_node->state & URCU_WAIT_RUNNING))
 212                 ret = urcu_adaptative_wake_up(wait_node);
 213         return ret;
 214 }
 215
 216 /*
 217  * Need mutual exclusion against other wakeup and move waiters
 218  * operations. It is provided by the caller.
 219  */
 220 static inline
 221 int urcu_dequeue_wake_n(struct urcu_wait_queue *queue, int n)
 222 {
 223         int nr_wakeup = 0;
 224
 225         for (;;) {
 226                 int ret;
 227
 228                 ret = urcu_dequeue_wake_single(queue);
 229                 if (ret < 0)
 230                         return nr_wakeup;
 231                 else if (ret > 0)
 232                         nr_wakeup++;
 233                 else
 234                         break;
 235         }
 236         return nr_wakeup;
 237 }
 238
 239 static inline
 240 int urcu_wake_all_waiters(struct urcu_waiters *waiters)
 241 {
 242         struct cds_lfs_node *iter, *iter_n;
 243         int nr_wakeup = 0;
 244
 245         /* Wake all waiters in our stack head */
 246         cds_lfs_for_each_safe(waiters->head, iter, iter_n) {
 247                 struct urcu_wait_node *wait_node =
 248                         caa_container_of(iter, struct urcu_wait_node, node);
 249
 250                 CMM_STORE_SHARED(wait_node->node.next, NULL);
 251                 /* Don't wake already running threads */
 252                 if (wait_node->state & URCU_WAIT_RUNNING)
 253                         continue;
 254                 if (urcu_adaptative_wake_up(wait_node))
 255                         nr_wakeup++;
 256         }
 257         return nr_wakeup;
 258 }
 259
 260 #endif /* _URCU_WAITQUEUE_LIFO_H */