src/common/waiter.cpp

   1 /*
   2  * Copyright (C) 2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   3  * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4  *
   5  * SPDX-License-Identifier: LGPL-2.1-only
   6  *
   7  */
   8
   9 #include "error.hpp"
  10 #include "macros.hpp"
  11 #include "waiter.hpp"
  12
  13 #include <poll.h>
  14 #include <urcu/futex.h>
  15 #include <urcu/uatomic.h>
  16
  17 namespace {
  18 /* Number of busy-loop attempts before waiting on futex. */
  19 constexpr auto wait_attempt_count = 1000;
  20
  21 enum waiter_state {
  22         /* WAITER_WAITING is compared directly (futex compares it). */
  23         WAITER_WAITING = 0,
  24         /* non-zero are used as masks. */
  25         WAITER_WOKEN_UP = (1 << 0),
  26         WAITER_RUNNING = (1 << 1),
  27         WAITER_TEARDOWN = (1 << 2),
  28 };
  29 } /* namespace */
  30
  31 lttng::synchro::waiter::waiter()
  32 {
  33         arm();
  34 }
  35
  36 void lttng::synchro::waiter::arm() noexcept
  37 {
  38         cds_wfs_node_init(&_wait_queue_node);
  39         uatomic_set(&_state, WAITER_WAITING);
  40         cmm_smp_mb();
  41 }
  42
  43 /*
  44  * User must arm "waiter" before passing its memory to waker thread.
  45  */
  46 void lttng::synchro::waiter::wait()
  47 {
  48         DBG("Beginning of waiter \"wait\" period");
  49
  50         /* Load and test condition before read state. */
  51         cmm_smp_rmb();
  52         for (unsigned int i = 0; i < wait_attempt_count; i++) {
  53                 if (uatomic_read(&_state) != WAITER_WAITING) {
  54                         goto skip_futex_wait;
  55                 }
  56
  57                 caa_cpu_relax();
  58         }
  59
  60         while (uatomic_read(&_state) == WAITER_WAITING) {
  61                 if (!futex_noasync(
  62                             &_state, FUTEX_WAIT, WAITER_WAITING, nullptr, nullptr, 0)) {
  63                         /*
  64                          * Prior queued wakeups queued by unrelated code
  65                          * using the same address can cause futex wait to
  66                          * return 0 even through the futex value is still
  67                          * WAITER_WAITING (spurious wakeups). Check
  68                          * the value again in user-space to validate
  69                          * whether it really differs from WAITER_WAITING.
  70                          */
  71                         continue;
  72                 }
  73
  74                 switch (errno) {
  75                 case EAGAIN:
  76                         /* Value already changed. */
  77                         goto skip_futex_wait;
  78                 case EINTR:
  79                         /* Retry if interrupted by signal. */
  80                         break; /* Get out of switch. Check again. */
  81                 default:
  82                         /* Unexpected error. */
  83                         PERROR("futex_noasync");
  84                         abort();
  85                 }
  86         }
  87 skip_futex_wait:
  88
  89         /* Tell waker thread than we are running. */
  90         uatomic_or(&_state, WAITER_RUNNING);
  91
  92         /*
  93          * Wait until waker thread lets us know it's ok to tear down
  94          * memory allocated for struct lttng_waiter.
  95          */
  96         for (unsigned int i = 0; i < wait_attempt_count; i++) {
  97                 if (uatomic_read(&_state) & WAITER_TEARDOWN) {
  98                         break;
  99                 }
 100
 101                 caa_cpu_relax();
 102         }
 103
 104         while (!(uatomic_read(&_state) & WAITER_TEARDOWN)) {
 105                 poll(nullptr, 0, 10);
 106         }
 107
 108         LTTNG_ASSERT(uatomic_read(&_state) & WAITER_TEARDOWN);
 109         DBG("End of waiter \"wait\" period");
 110 }
 111
 112 lttng::synchro::waker lttng::synchro::waiter::get_waker()
 113 {
 114         return lttng::synchro::waker(_state);
 115 }
 116
 117 /*
 118  * Note: lttng_waiter_wake needs waiter to stay allocated throughout its
 119  * execution. In this scheme, the waiter owns the node memory, and we only allow
 120  * it to free this memory when it sees the WAITER_TEARDOWN flag.
 121  */
 122 void lttng::synchro::waker::wake()
 123 {
 124         cmm_smp_mb();
 125
 126         LTTNG_ASSERT(uatomic_read(&_state) == WAITER_WAITING);
 127
 128         uatomic_set(&_state, WAITER_WOKEN_UP);
 129         if (!(uatomic_read(&_state) & WAITER_RUNNING)) {
 130                 if (futex_noasync(&_state, FUTEX_WAKE, 1, nullptr, nullptr, 0) < 0) {
 131                         PERROR("futex_noasync");
 132                         abort();
 133                 }
 134         }
 135
 136         /* Allow teardown of struct urcu_wait memory. */
 137         uatomic_or(&_state, WAITER_TEARDOWN);
 138 }
 139
 140 lttng::synchro::wait_queue::wait_queue()
 141 {
 142         cds_wfs_init(&_stack);
 143 }
 144
 145 void lttng::synchro::wait_queue::add(waiter &waiter) noexcept
 146 {
 147         (void) cds_wfs_push(&_stack, &waiter._wait_queue_node);
 148 }
 149
 150 void lttng::synchro::wait_queue::wake_all()
 151 {
 152         /* Move all waiters from the queue to our local stack. */
 153         auto *waiters = __cds_wfs_pop_all(&_stack);
 154
 155         /* Wake all waiters in our stack head. */
 156         cds_wfs_node *iter, *iter_n;
 157         cds_wfs_for_each_blocking_safe(waiters, iter, iter_n) {
 158                 auto& waiter = *lttng::utils::container_of(
 159                         iter, &lttng::synchro::waiter::_wait_queue_node);
 160
 161                 /* Don't wake already running threads. */
 162                 if (waiter._state & WAITER_RUNNING) {
 163                         continue;
 164                 }
 165
 166                 waiter.get_waker().wake();
 167         }
 168 }