folly/fibers/Baton.cpp

   1 /*
   2  * Copyright 2014-present Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 #include "Baton.h"
  17
  18 #include <chrono>
  19
  20 #include <folly/detail/MemoryIdler.h>
  21 #include <folly/fibers/FiberManagerInternal.h>
  22 #include <folly/portability/Asm.h>
  23
  24 namespace folly {
  25 namespace fibers {
  26
  27 void Baton::wait() {
  28   wait([]() {});
  29 }
  30
  31 void Baton::wait(TimeoutHandler& timeoutHandler) {
  32   auto timeoutFunc = [this, &timeoutHandler] {
  33     if (!try_wait()) {
  34       postHelper(TIMEOUT);
  35     }
  36     timeoutHandler.timeoutPtr_ = 0;
  37   };
  38   timeoutHandler.timeoutFunc_ = std::ref(timeoutFunc);
  39   timeoutHandler.fiberManager_ = FiberManager::getFiberManagerUnsafe();
  40   wait();
  41   timeoutHandler.cancelTimeout();
  42 }
  43
  44 void Baton::waitThread() {
  45   if (spinWaitForEarlyPost()) {
  46     assert(waitingFiber_.load(std::memory_order_acquire) == POSTED);
  47     return;
  48   }
  49
  50   auto fiber = waitingFiber_.load();
  51
  52   if (LIKELY(
  53           fiber == NO_WAITER &&
  54           waitingFiber_.compare_exchange_strong(fiber, THREAD_WAITING))) {
  55     do {
  56       folly::detail::MemoryIdler::futexWait(
  57           futex_.futex, uint32_t(THREAD_WAITING));
  58       fiber = waitingFiber_.load(std::memory_order_acquire);
  59     } while (fiber == THREAD_WAITING);
  60   }
  61
  62   if (LIKELY(fiber == POSTED)) {
  63     return;
  64   }
  65
  66   // Handle errors
  67   if (fiber == TIMEOUT) {
  68     throw std::logic_error("Thread baton can't have timeout status");
  69   }
  70   if (fiber == THREAD_WAITING) {
  71     throw std::logic_error("Other thread is already waiting on this baton");
  72   }
  73   throw std::logic_error("Other fiber is already waiting on this baton");
  74 }
  75
  76 bool Baton::spinWaitForEarlyPost() {
  77   static_assert(
  78       PreBlockAttempts > 0,
  79       "isn't this assert clearer than an uninitialized variable warning?");
  80   for (int i = 0; i < PreBlockAttempts; ++i) {
  81     if (try_wait()) {
  82       // hooray!
  83       return true;
  84     }
  85     // The pause instruction is the polite way to spin, but it doesn't
  86     // actually affect correctness to omit it if we don't have it.
  87     // Pausing donates the full capabilities of the current core to
  88     // its other hyperthreads for a dozen cycles or so
  89     asm_volatile_pause();
  90   }
  91
  92   return false;
  93 }
  94
  95 bool Baton::timedWaitThread(TimeoutController::Duration timeout) {
  96   if (spinWaitForEarlyPost()) {
  97     assert(waitingFiber_.load(std::memory_order_acquire) == POSTED);
  98     return true;
  99   }
 100
 101   auto fiber = waitingFiber_.load();
 102
 103   if (LIKELY(
 104           fiber == NO_WAITER &&
 105           waitingFiber_.compare_exchange_strong(fiber, THREAD_WAITING))) {
 106     auto deadline = TimeoutController::Clock::now() + timeout;
 107     do {
 108       const auto wait_rv =
 109           futex_.futex.futexWaitUntil(uint32_t(THREAD_WAITING), deadline);
 110       if (wait_rv == folly::detail::FutexResult::TIMEDOUT) {
 111         return false;
 112       }
 113       fiber = waitingFiber_.load(std::memory_order_relaxed);
 114     } while (fiber == THREAD_WAITING);
 115   }
 116
 117   if (LIKELY(fiber == POSTED)) {
 118     return true;
 119   }
 120
 121   // Handle errors
 122   if (fiber == TIMEOUT) {
 123     throw std::logic_error("Thread baton can't have timeout status");
 124   }
 125   if (fiber == THREAD_WAITING) {
 126     throw std::logic_error("Other thread is already waiting on this baton");
 127   }
 128   throw std::logic_error("Other fiber is already waiting on this baton");
 129 }
 130
 131 void Baton::post() {
 132   postHelper(POSTED);
 133 }
 134
 135 void Baton::postHelper(intptr_t new_value) {
 136   auto fiber = waitingFiber_.load();
 137
 138   do {
 139     if (fiber == THREAD_WAITING) {
 140       assert(new_value == POSTED);
 141
 142       return postThread();
 143     }
 144
 145     if (fiber == POSTED || fiber == TIMEOUT) {
 146       return;
 147     }
 148   } while (!waitingFiber_.compare_exchange_weak(fiber, new_value));
 149
 150   if (fiber != NO_WAITER) {
 151     reinterpret_cast<Fiber*>(fiber)->resume();
 152   }
 153 }
 154
 155 bool Baton::try_wait() {
 156   return ready();
 157 }
 158
 159 void Baton::postThread() {
 160   auto expected = THREAD_WAITING;
 161
 162   if (!waitingFiber_.compare_exchange_strong(expected, POSTED)) {
 163     return;
 164   }
 165
 166   futex_.futex.futexWake(1);
 167 }
 168
 169 void Baton::reset() {
 170   waitingFiber_.store(NO_WAITER, std::memory_order_relaxed);
 171   ;
 172 }
 173
 174 void Baton::TimeoutHandler::scheduleTimeout(
 175     TimeoutController::Duration timeout) {
 176   assert(fiberManager_ != nullptr);
 177   assert(timeoutFunc_ != nullptr);
 178   assert(timeoutPtr_ == 0);
 179
 180   if (timeout.count() > 0) {
 181     timeoutPtr_ =
 182         fiberManager_->timeoutManager_->registerTimeout(timeoutFunc_, timeout);
 183   }
 184 }
 185
 186 void Baton::TimeoutHandler::cancelTimeout() {
 187   if (timeoutPtr_) {
 188     fiberManager_->timeoutManager_->cancel(timeoutPtr_);
 189   }
 190 }
 191 } // namespace fibers
 192 } // namespace folly