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-die.h"
  30
  31 /*
  32  * Number of busy-loop attempts before waiting on futex for grace period
  33  * batching.
  34  */
  35 #define URCU_WAIT_ATTEMPTS 1000
  36
  37 enum urcu_wait_state {
  38         /* URCU_WAIT_WAITING is compared directly (futex compares it). */
  39         URCU_WAIT_WAITING =     0,
  40         /* non-zero are used as masks. */
  41         URCU_WAIT_WAKEUP =      (1 << 0),
  42         URCU_WAIT_RUNNING =     (1 << 1),
  43         URCU_WAIT_TEARDOWN =    (1 << 2),
  44 };
  45
  46 struct urcu_wait_node {
  47         struct cds_wfs_node node;
  48         int32_t state;  /* enum urcu_wait_state */
  49 };
  50
  51 #define URCU_WAIT_NODE_INIT(name, _state)               \
  52         { .state = _state }
  53
  54 #define DEFINE_URCU_WAIT_NODE(name, state)              \
  55         struct urcu_wait_node name = URCU_WAIT_NODE_INIT(name, state)
  56
  57 #define DECLARE_URCU_WAIT_NODE(name)                    \
  58         struct urcu_wait_node name
  59
  60 struct urcu_wait_queue {
  61         struct cds_wfs_stack stack;
  62 };
  63
  64 #define URCU_WAIT_QUEUE_HEAD_INIT(name)                         \
  65         {                                                       \
  66                  .stack = {                                     \
  67                         .head = CDS_WFS_END,                    \
  68                         .lock = PTHREAD_MUTEX_INITIALIZER,      \
  69                 },                                              \
  70         }
  71
  72 #define DECLARE_URCU_WAIT_QUEUE(name)                   \
  73         struct urcu_wait_queue name
  74
  75 #define DEFINE_URCU_WAIT_QUEUE(name)                    \
  76         struct urcu_wait_queue name = URCU_WAIT_QUEUE_HEAD_INIT(name)
  77
  78 struct urcu_waiters {
  79         struct cds_wfs_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_wfs_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_wfs_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_node_init(struct urcu_wait_node *node,
 114                 enum urcu_wait_state state)
 115 {
 116         urcu_wait_set_state(node, state);
 117         cds_wfs_node_init(&node->node);
 118 }
 119
 120 /*
 121  * Note: urcu_adaptative_wake_up needs "value" to stay allocated
 122  * throughout its execution. In this scheme, the waiter owns the node
 123  * memory, and we only allow it to free this memory when it receives the
 124  * URCU_WAIT_TEARDOWN flag.
 125  */
 126 static inline
 127 void urcu_adaptative_wake_up(struct urcu_wait_node *wait)
 128 {
 129         cmm_smp_mb();
 130         urcu_posix_assert(uatomic_read(&wait->state) == URCU_WAIT_WAITING);
 131         uatomic_set(&wait->state, URCU_WAIT_WAKEUP);
 132         if (!(uatomic_read(&wait->state) & URCU_WAIT_RUNNING)) {
 133                 if (futex_noasync(&wait->state, FUTEX_WAKE, 1,
 134                                 NULL, NULL, 0) < 0)
 135                         urcu_die(errno);
 136         }
 137         /* Allow teardown of struct urcu_wait memory. */
 138         uatomic_or(&wait->state, URCU_WAIT_TEARDOWN);
 139 }
 140
 141 /*
 142  * Caller must initialize "value" to URCU_WAIT_WAITING before passing its
 143  * memory to waker thread.
 144  */
 145 static inline
 146 void urcu_adaptative_busy_wait(struct urcu_wait_node *wait)
 147 {
 148         unsigned int i;
 149
 150         /* Load and test condition before read state */
 151         cmm_smp_rmb();
 152         for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
 153                 if (uatomic_read(&wait->state) != URCU_WAIT_WAITING)
 154                         goto skip_futex_wait;
 155                 caa_cpu_relax();
 156         }
 157         while (futex_noasync(&wait->state, FUTEX_WAIT, URCU_WAIT_WAITING,
 158                         NULL, NULL, 0)) {
 159                 switch (errno) {
 160                 case EWOULDBLOCK:
 161                         /* Value already changed. */
 162                         goto skip_futex_wait;
 163                 case EINTR:
 164                         /* Retry if interrupted by signal. */
 165                         break;  /* Get out of switch. */
 166                 default:
 167                         /* Unexpected error. */
 168                         urcu_die(errno);
 169                 }
 170         }
 171 skip_futex_wait:
 172
 173         /* Tell waker thread than we are running. */
 174         uatomic_or(&wait->state, URCU_WAIT_RUNNING);
 175
 176         /*
 177          * Wait until waker thread lets us know it's ok to tear down
 178          * memory allocated for struct urcu_wait.
 179          */
 180         for (i = 0; i < URCU_WAIT_ATTEMPTS; i++) {
 181                 if (uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN)
 182                         break;
 183                 caa_cpu_relax();
 184         }
 185         while (!(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN))
 186                 poll(NULL, 0, 10);
 187         urcu_posix_assert(uatomic_read(&wait->state) & URCU_WAIT_TEARDOWN);
 188 }
 189
 190 static inline
 191 void urcu_wake_all_waiters(struct urcu_waiters *waiters)
 192 {
 193         struct cds_wfs_node *iter, *iter_n;
 194
 195         /* Wake all waiters in our stack head */
 196         cds_wfs_for_each_blocking_safe(waiters->head, iter, iter_n) {
 197                 struct urcu_wait_node *wait_node =
 198                         caa_container_of(iter, struct urcu_wait_node, node);
 199
 200                 /* Don't wake already running threads */
 201                 if (wait_node->state & URCU_WAIT_RUNNING)
 202                         continue;
 203                 urcu_adaptative_wake_up(wait_node);
 204         }
 205 }
 206
 207 #endif /* _URCU_WAIT_H */