src/common/waiter.c

   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 "waiter.h"
  10 #include <urcu/uatomic.h>
  11 #include <urcu/futex.h>
  12 #include <assert.h>
  13 #include "error.h"
  14 #include <poll.h>
  15
  16 /*
  17  * Number of busy-loop attempts before waiting on futex.
  18  */
  19 #define WAIT_ATTEMPTS 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
  30 LTTNG_HIDDEN
  31 void lttng_waiter_init(struct lttng_waiter *waiter)
  32 {
  33         cds_wfs_node_init(&waiter->wait_queue_node);
  34         uatomic_set(&waiter->state, WAITER_WAITING);
  35         cmm_smp_mb();
  36 }
  37
  38 /*
  39  * User must init "waiter" before passing its memory to waker thread.
  40  */
  41 LTTNG_HIDDEN
  42 void lttng_waiter_wait(struct lttng_waiter *waiter)
  43 {
  44         unsigned int i;
  45
  46         DBG("Beginning of waiter \"wait\" period");
  47
  48         /* Load and test condition before read state. */
  49         cmm_smp_rmb();
  50         for (i = 0; i < WAIT_ATTEMPTS; i++) {
  51                 if (uatomic_read(&waiter->state) != WAITER_WAITING) {
  52                         goto skip_futex_wait;
  53                 }
  54
  55                 caa_cpu_relax();
  56         }
  57
  58         while (uatomic_read(&waiter->state) == WAITER_WAITING) {
  59                 if (!futex_noasync(&waiter->state, FUTEX_WAIT, WAITER_WAITING, NULL, NULL, 0)) {
  60                         /*
  61                          * Prior queued wakeups queued by unrelated code
  62                          * using the same address can cause futex wait to
  63                          * return 0 even through the futex value is still
  64                          * WAITER_WAITING (spurious wakeups). Check
  65                          * the value again in user-space to validate
  66                          * whether it really differs from WAITER_WAITING.
  67                          */
  68                         continue;
  69                 }
  70
  71                 switch (errno) {
  72                 case EAGAIN:
  73                         /* Value already changed. */
  74                         goto skip_futex_wait;
  75                 case EINTR:
  76                         /* Retry if interrupted by signal. */
  77                         break;  /* Get out of switch. Check again. */
  78                 default:
  79                         /* Unexpected error. */
  80                         PERROR("futex_noasync");
  81                         abort();
  82                 }
  83         }
  84 skip_futex_wait:
  85
  86         /* Tell waker thread than we are running. */
  87         uatomic_or(&waiter->state, WAITER_RUNNING);
  88
  89         /*
  90          * Wait until waker thread lets us know it's ok to tear down
  91          * memory allocated for struct lttng_waiter.
  92          */
  93         for (i = 0; i < WAIT_ATTEMPTS; i++) {
  94                 if (uatomic_read(&waiter->state) & WAITER_TEARDOWN) {
  95                         break;
  96                 }
  97
  98                 caa_cpu_relax();
  99         }
 100
 101         while (!(uatomic_read(&waiter->state) & WAITER_TEARDOWN)) {
 102                 poll(NULL, 0, 10);
 103         }
 104
 105         assert(uatomic_read(&waiter->state) & WAITER_TEARDOWN);
 106         DBG("End of waiter \"wait\" period");
 107 }
 108
 109 /*
 110  * Note: lttng_waiter_wake needs waiter to stay allocated throughout its
 111  * execution. In this scheme, the waiter owns the node memory, and we only allow
 112  * it to free this memory when it sees the WAITER_TEARDOWN flag.
 113  */
 114 LTTNG_HIDDEN
 115 void lttng_waiter_wake(struct lttng_waiter *waiter)
 116 {
 117         cmm_smp_mb();
 118         assert(uatomic_read(&waiter->state) == WAITER_WAITING);
 119         uatomic_set(&waiter->state, WAITER_WOKEN_UP);
 120         if (!(uatomic_read(&waiter->state) & WAITER_RUNNING)) {
 121                 if (futex_noasync(&waiter->state, FUTEX_WAKE, 1,
 122                                 NULL, NULL, 0) < 0) {
 123                         PERROR("futex_noasync");
 124                         abort();
 125                 }
 126         }
 127
 128         /* Allow teardown of struct urcu_wait memory. */
 129         uatomic_or(&waiter->state, WAITER_TEARDOWN);
 130 }
 131
 132
 133 LTTNG_HIDDEN
 134 void lttng_wait_queue_init(struct lttng_wait_queue *queue)
 135 {
 136         cds_wfs_init(&queue->stack);
 137 }
 138
 139 LTTNG_HIDDEN
 140 void lttng_wait_queue_add(struct lttng_wait_queue *queue,
 141                 struct lttng_waiter *waiter)
 142 {
 143         (void) cds_wfs_push(&queue->stack, &waiter->wait_queue_node);
 144 }
 145
 146 LTTNG_HIDDEN
 147 void lttng_wait_queue_wake_all(struct lttng_wait_queue *queue)
 148 {
 149         struct cds_wfs_head *waiters;
 150         struct cds_wfs_node *iter, *iter_n;
 151
 152         /* Move all waiters from the queue to our local stack. */
 153         waiters = __cds_wfs_pop_all(&queue->stack);
 154
 155         /* Wake all waiters in our stack head. */
 156         cds_wfs_for_each_blocking_safe(waiters, iter, iter_n) {
 157                 struct lttng_waiter *waiter =
 158                         container_of(iter, struct lttng_waiter, wait_queue_node);
 159
 160                 /* Don't wake already running threads. */
 161                 if (waiter->state & WAITER_RUNNING) {
 162                         continue;
 163                 }
 164
 165                 lttng_waiter_wake(waiter);
 166         }
 167 }