src/urcu-wait.h

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