src/urcu-wait.h

   1 // SPDX-FileCopyrightText: 2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   2 //
   3 // SPDX-License-Identifier: LGPL-2.1-or-later
   4
   5 #ifndef _URCU_WAIT_H
   6 #define _URCU_WAIT_H
   7
   8 /*
   9  * Userspace RCU library wait/wakeup management
  10  */
  11
  12 #include <urcu/assert.h>
  13 #include <urcu/uatomic.h>
  14 #include <urcu/wfstack.h>
  15 #include <urcu/futex.h>
  16 #include "urcu-die.h"
  17
  18 /*
  19  * Number of busy-loop attempts before waiting on futex for grace period
  20  * batching.
  21  */
  22 #define URCU_WAIT_ATTEMPTS 1000
  23
  24 enum urcu_wait_state {
  25         /* URCU_WAIT_WAITING is compared directly (futex compares it). */
  26         URCU_WAIT_WAITING =     0,
  27         /* non-zero are used as masks. */
  28         URCU_WAIT_WAKEUP =      (1 << 0),
  29         URCU_WAIT_RUNNING =     (1 << 1),
  30         URCU_WAIT_TEARDOWN =    (1 << 2),
  31 };
  32
  33 struct urcu_wait_node {
  34         struct cds_wfs_node node;
  35         int32_t state;  /* enum urcu_wait_state */
  36 };
  37
  38 #define URCU_WAIT_NODE_INIT(name, _state)               \
  39         { .node = { .next = NULL }, .state = _state }
  40
  41 #define DEFINE_URCU_WAIT_NODE(name, state)              \
  42         struct urcu_wait_node name = URCU_WAIT_NODE_INIT(name, state)
  43
  44 #define DECLARE_URCU_WAIT_NODE(name)                    \
  45         struct urcu_wait_node name
  46
  47 struct urcu_wait_queue {
  48         struct cds_wfs_stack stack;
  49 };
  50
  51 #define URCU_WAIT_QUEUE_HEAD_INIT(name)                         \
  52         {                                                       \
  53                  .stack = {                                     \
  54                         .head = CDS_WFS_END,                    \
  55                         .lock = PTHREAD_MUTEX_INITIALIZER,      \
  56                 },                                              \
  57         }
  58
  59 #define DECLARE_URCU_WAIT_QUEUE(name)                   \
  60         struct urcu_wait_queue name
  61
  62 #define DEFINE_URCU_WAIT_QUEUE(name)                    \
  63         struct urcu_wait_queue name = URCU_WAIT_QUEUE_HEAD_INIT(name)
  64
  65 struct urcu_waiters {
  66         struct cds_wfs_head *head;
  67 };
  68
  69 /*
  70  * Add ourself atomically to a wait queue. Return 0 if queue was
  71  * previously empty, else return 1.
  72  * A full memory barrier is issued before being added to the wait queue.
  73  */
  74 static inline
  75 bool urcu_wait_add(struct urcu_wait_queue *queue,
  76                 struct urcu_wait_node *node)
  77 {
  78         return cds_wfs_push(&queue->stack, &node->node);
  79 }
  80
  81 /*
  82  * Atomically move all waiters from wait queue into our local struct
  83  * urcu_waiters.
  84  */
  85 static inline
  86 void urcu_move_waiters(struct urcu_waiters *waiters,
  87                 struct urcu_wait_queue *queue)
  88 {
  89         waiters->head = __cds_wfs_pop_all(&queue->stack);
  90 }
  91
  92 static inline
  93 void urcu_wait_set_state(struct urcu_wait_node *node,
  94                 enum urcu_wait_state state)
  95 {
  96         node->state = state;
  97 }
  98
  99 static inline
 100 void urcu_wait_node_init(struct urcu_wait_node *node,
 101                 enum urcu_wait_state state)
 102 {
 103         urcu_wait_set_state(node, state);
 104         cds_wfs_node_init(&node->node);
 105 }
 106
 107 /*
 108  * Note: urcu_adaptative_wake_up needs "value" to stay allocated
 109  * throughout its execution. In this scheme, the waiter owns the node
 110  * memory, and we only allow it to free this memory when it receives the
 111  * URCU_WAIT_TEARDOWN flag.
 112  */
 113 static inline
 114 void urcu_adaptative_wake_up(struct urcu_wait_node *wait)
 115 {
 116         cmm_smp_mb();
 117         urcu_posix_assert(uatomic_read(&wait->state) == URCU_WAIT_WAITING);
 118         uatomic_set(&wait->state, URCU_WAIT_WAKEUP);
 119         if (!(uatomic_read(&wait->state) & URCU_WAIT_RUNNING)) {
 120                 if (futex_noasync(&wait->state, FUTEX_WAKE, 1,
 121                                 NULL, NULL, 0) < 0)
 122                         urcu_die(errno);
 123         }
 124         /* Allow teardown of struct urcu_wait memory. */
 125         uatomic_or(&wait->state, URCU_WAIT_TEARDOWN);
 126 }
 127
 128 /*
 129  * Caller must initialize "value" to URCU_WAIT_WAITING before passing its
 130  * memory to waker thread.
 131  */
 132 static inline
 133 void urcu_adaptative_busy_wait(struct urcu_wait_node *wait)
 134 {
 135         unsigned int i;
 136
 137         /* Load and test condition before read state */
 138         cmm_smp_rmb();
 139         for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
 140                 if (uatomic_read(&wait->state) != URCU_WAIT_WAITING)
 141                         goto skip_futex_wait;
 142                 caa_cpu_relax();
 143         }
 144         while (uatomic_read(&wait->state) == URCU_WAIT_WAITING) {
 145                 if (!futex_noasync(&wait->state, FUTEX_WAIT, URCU_WAIT_WAITING, NULL, NULL, 0)) {
 146                         /*
 147                          * Prior queued wakeups queued by unrelated code
 148                          * using the same address can cause futex wait to
 149                          * return 0 even through the futex value is still
 150                          * URCU_WAIT_WAITING (spurious wakeups). Check
 151                          * the value again in user-space to validate
 152                          * whether it really differs from
 153                          * URCU_WAIT_WAITING.
 154                          */
 155                         continue;
 156                 }
 157                 switch (errno) {
 158                 case EAGAIN:
 159                         /* Value already changed. */
 160                         goto skip_futex_wait;
 161                 case EINTR:
 162                         /* Retry if interrupted by signal. */
 163                         break;  /* Get out of switch. Check again. */
 164                 default:
 165                         /* Unexpected error. */
 166                         urcu_die(errno);
 167                 }
 168         }
 169 skip_futex_wait:
 170
 171         /* Tell waker thread than we are running. */
 172         uatomic_or(&wait->state, URCU_WAIT_RUNNING);
 173
 174         /*
 175          * Wait until waker thread lets us know it's ok to tear down
 176          * memory allocated for struct urcu_wait.
 177          */
 178         for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
 179                 if (uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN)
 180                         break;
 181                 caa_cpu_relax();
 182         }
 183         while (!(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN))
 184                 poll(NULL, 0, 10);
 185         urcu_posix_assert(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN);
 186 }
 187
 188 static inline
 189 void urcu_wake_all_waiters(struct urcu_waiters *waiters)
 190 {
 191         struct cds_wfs_node *iter, *iter_n;
 192
 193         /* Wake all waiters in our stack head */
 194         cds_wfs_for_each_blocking_safe(waiters->head, iter, iter_n) {
 195                 struct urcu_wait_node *wait_node =
 196                         caa_container_of(iter, struct urcu_wait_node, node);
 197
 198                 /* Don't wake already running threads */
 199                 if (wait_node->state & URCU_WAIT_RUNNING)
 200                         continue;
 201                 urcu_adaptative_wake_up(wait_node);
 202         }
 203 }
 204
 205 #endif /* _URCU_WAIT_H */