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(&_state, FUTEX_WAIT, WAITER_WAITING, nullptr, nullptr, 0)) {
  62                         /*
  63                          * Prior queued wakeups queued by unrelated code
  64                          * using the same address can cause futex wait to
  65                          * return 0 even through the futex value is still
  66                          * WAITER_WAITING (spurious wakeups). Check
  67                          * the value again in user-space to validate
  68                          * whether it really differs from WAITER_WAITING.
  69                          */
  70                         continue;
  71                 }
  72
  73                 switch (errno) {
  74                 case EAGAIN:
  75                         /* Value already changed. */
  76                         goto skip_futex_wait;
  77                 case EINTR:
  78                         /* Retry if interrupted by signal. */
  79                         break; /* Get out of switch. Check again. */
  80                 default:
  81                         /* Unexpected error. */
  82                         PERROR("futex_noasync");
  83                         abort();
  84                 }
  85         }
  86 skip_futex_wait:
  87
  88         /* Tell waker thread than we are running. */
  89         uatomic_or(&_state, WAITER_RUNNING);
  90
  91         /*
  92          * Wait until waker thread lets us know it's ok to tear down
  93          * memory allocated for struct lttng_waiter.
  94          */
  95         for (unsigned int i = 0; i < wait_attempt_count; i++) {
  96                 if (uatomic_read(&_state) & WAITER_TEARDOWN) {
  97                         break;
  98                 }
  99
 100                 caa_cpu_relax();
 101         }
 102
 103         while (!(uatomic_read(&_state) & WAITER_TEARDOWN)) {
 104                 poll(nullptr, 0, 10);
 105         }
 106
 107         LTTNG_ASSERT(uatomic_read(&_state) & WAITER_TEARDOWN);
 108         DBG("End of waiter \"wait\" period");
 109 }
 110
 111 lttng::synchro::waker lttng::synchro::waiter::get_waker()
 112 {
 113         return lttng::synchro::waker(_state);
 114 }
 115
 116 /*
 117  * Note: lttng_waiter_wake needs waiter to stay allocated throughout its
 118  * execution. In this scheme, the waiter owns the node memory, and we only allow
 119  * it to free this memory when it sees the WAITER_TEARDOWN flag.
 120  */
 121 void lttng::synchro::waker::wake()
 122 {
 123         cmm_smp_mb();
 124
 125         LTTNG_ASSERT(uatomic_read(&_state.get()) == WAITER_WAITING);
 126
 127         uatomic_set(&_state.get(), WAITER_WOKEN_UP);
 128         if (!(uatomic_read(&_state.get()) & WAITER_RUNNING)) {
 129                 if (futex_noasync(&_state.get(), FUTEX_WAKE, 1, nullptr, nullptr, 0) < 0) {
 130                         PERROR("futex_noasync");
 131                         abort();
 132                 }
 133         }
 134
 135         /* Allow teardown of struct urcu_wait memory. */
 136         uatomic_or(&_state.get(), WAITER_TEARDOWN);
 137 }
 138
 139 lttng::synchro::wait_queue::wait_queue()
 140 {
 141         cds_wfs_init(&_stack);
 142 }
 143
 144 void lttng::synchro::wait_queue::add(waiter& waiter) noexcept
 145 {
 146         (void) cds_wfs_push(&_stack, &waiter._wait_queue_node);
 147 }
 148
 149 void lttng::synchro::wait_queue::wake_all()
 150 {
 151         /* Move all waiters from the queue to our local stack. */
 152         auto *waiters = __cds_wfs_pop_all(&_stack);
 153
 154         /* Wake all waiters in our stack head. */
 155         cds_wfs_node *iter, *iter_n;
 156         cds_wfs_for_each_blocking_safe (waiters, iter, iter_n) {
 157                 auto& waiter = *lttng::utils::container_of(
 158                         iter, &lttng::synchro::waiter::_wait_queue_node);
 159
 160                 /* Don't wake already running threads. */
 161                 if (waiter._state & WAITER_RUNNING) {
 162                         continue;
 163                 }
 164
 165                 waiter.get_waker().wake();
 166         }
 167 }