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         urcu_posix_assert(uatomic_read(&wait->state) == URCU_WAIT_WAITING);
 117         uatomic_store(&wait->state, URCU_WAIT_WAKEUP, CMM_RELEASE);
 118         if (!(uatomic_read(&wait->state) & URCU_WAIT_RUNNING)) {
 119                 if (futex_noasync(&wait->state, FUTEX_WAKE, 1,
 120                                 NULL, NULL, 0) < 0)
 121                         urcu_die(errno);
 122         }
 123         /* Allow teardown of struct urcu_wait memory. */
 124         uatomic_or_mo(&wait->state, URCU_WAIT_TEARDOWN, CMM_RELEASE);
 125 }
 126
 127 /*
 128  * Caller must initialize "value" to URCU_WAIT_WAITING before passing its
 129  * memory to waker thread.
 130  */
 131 static inline
 132 void urcu_adaptative_busy_wait(struct urcu_wait_node *wait)
 133 {
 134         unsigned int i;
 135
 136         /* Load and test condition before read state */
 137         cmm_smp_rmb();
 138         for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
 139                 if (uatomic_load(&wait->state, CMM_ACQUIRE) != URCU_WAIT_WAITING)
 140                         goto skip_futex_wait;
 141                 caa_cpu_relax();
 142         }
 143         while (uatomic_load(&wait->state, CMM_ACQUIRE) == URCU_WAIT_WAITING) {
 144                 if (!futex_noasync(&wait->state, FUTEX_WAIT, URCU_WAIT_WAITING, NULL, NULL, 0)) {
 145                         /*
 146                          * Prior queued wakeups queued by unrelated code
 147                          * using the same address can cause futex wait to
 148                          * return 0 even through the futex value is still
 149                          * URCU_WAIT_WAITING (spurious wakeups). Check
 150                          * the value again in user-space to validate
 151                          * whether it really differs from
 152                          * URCU_WAIT_WAITING.
 153                          */
 154                         continue;
 155                 }
 156                 switch (errno) {
 157                 case EAGAIN:
 158                         /* Value already changed. */
 159                         goto skip_futex_wait;
 160                 case EINTR:
 161                         /* Retry if interrupted by signal. */
 162                         break;  /* Get out of switch. Check again. */
 163                 default:
 164                         /* Unexpected error. */
 165                         urcu_die(errno);
 166                 }
 167         }
 168 skip_futex_wait:
 169
 170         /* Tell waker thread than we are running. */
 171         uatomic_or(&wait->state, URCU_WAIT_RUNNING);
 172
 173         /*
 174          * Wait until waker thread lets us know it's ok to tear down
 175          * memory allocated for struct urcu_wait.
 176          */
 177         for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
 178                 if (uatomic_load(&wait->state, CMM_RELAXED) & URCU_WAIT_TEARDOWN)
 179                         break;
 180                 caa_cpu_relax();
 181         }
 182         while (!(uatomic_load(&wait->state, CMM_ACQUIRE) & URCU_WAIT_TEARDOWN))
 183                 poll(NULL, 0, 10);
 184         urcu_posix_assert(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN);
 185 }
 186
 187 static inline
 188 void urcu_wake_all_waiters(struct urcu_waiters *waiters)
 189 {
 190         struct cds_wfs_node *iter, *iter_n;
 191
 192         /* Wake all waiters in our stack head */
 193         cds_wfs_for_each_blocking_safe(waiters->head, iter, iter_n) {
 194                 struct urcu_wait_node *wait_node =
 195                         caa_container_of(iter, struct urcu_wait_node, node);
 196
 197                 /* Don't wake already running threads */
 198                 if (uatomic_load(&wait_node->state, CMM_RELAXED) & URCU_WAIT_RUNNING)
 199                         continue;
 200                 urcu_adaptative_wake_up(wait_node);
 201         }
 202 }
 203
 204 #endif /* _URCU_WAIT_H */