folly/experimental/fibers/Fiber.cpp

   1 /*
   2  * Copyright 2016 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 "Fiber.h"
  17
  18 #include <sys/syscall.h>
  19 #include <unistd.h>
  20
  21 #include <glog/logging.h>
  22 #include <algorithm>
  23 #include <cstring>
  24 #include <stdexcept>
  25
  26 #include <folly/Likely.h>
  27 #include <folly/Portability.h>
  28 #include <folly/experimental/fibers/BoostContextCompatibility.h>
  29 #include <folly/experimental/fibers/FiberManager.h>
  30 #include <folly/portability/SysSyscall.h>
  31
  32 namespace folly {
  33 namespace fibers {
  34
  35 namespace {
  36 static const uint64_t kMagic8Bytes = 0xfaceb00cfaceb00c;
  37
  38 pid_t localThreadId() {
  39   // __thread doesn't allow non-const initialization.
  40   // OSX doesn't support thread_local.
  41   static FOLLY_TLS pid_t threadId = 0;
  42   if (UNLIKELY(threadId == 0)) {
  43     threadId = syscall(FOLLY_SYS_gettid);
  44   }
  45   return threadId;
  46 }
  47
  48 /* Size of the region from p + nBytes down to the last non-magic value */
  49 static size_t nonMagicInBytes(const FContext& context) {
  50   uint64_t* begin = static_cast<uint64_t*>(context.stackLimit());
  51   uint64_t* end = static_cast<uint64_t*>(context.stackBase());
  52
  53   auto firstNonMagic = std::find_if(
  54       begin, end, [](uint64_t val) { return val != kMagic8Bytes; });
  55
  56   return (end - firstNonMagic) * sizeof(uint64_t);
  57 }
  58
  59 } // anonymous namespace
  60
  61 void Fiber::setData(intptr_t data) {
  62   DCHECK_EQ(state_, AWAITING);
  63   data_ = data;
  64   state_ = READY_TO_RUN;
  65
  66   if (fiberManager_.observer_) {
  67     fiberManager_.observer_->runnable(reinterpret_cast<uintptr_t>(this));
  68   }
  69
  70   if (LIKELY(threadId_ == localThreadId())) {
  71     fiberManager_.readyFibers_.push_back(*this);
  72     fiberManager_.ensureLoopScheduled();
  73   } else {
  74     fiberManager_.remoteReadyInsert(this);
  75   }
  76 }
  77
  78 Fiber::Fiber(FiberManager& fiberManager) : fiberManager_(fiberManager) {
  79   auto size = fiberManager_.options_.stackSize;
  80   auto limit = fiberManager_.stackAllocator_.allocate(size);
  81
  82   fcontext_ = makeContext(limit, size, &Fiber::fiberFuncHelper);
  83
  84   fiberManager_.allFibers_.push_back(*this);
  85 }
  86
  87 void Fiber::init(bool recordStackUsed) {
  88 // It is necessary to disable the logic for ASAN because we change
  89 // the fiber's stack.
  90 #ifndef FOLLY_SANITIZE_ADDRESS
  91   recordStackUsed_ = recordStackUsed;
  92   if (UNLIKELY(recordStackUsed_ && !stackFilledWithMagic_)) {
  93     auto limit = fcontext_.stackLimit();
  94     auto base = fcontext_.stackBase();
  95
  96     std::fill(
  97         static_cast<uint64_t*>(limit),
  98         static_cast<uint64_t*>(base),
  99         kMagic8Bytes);
 100
 101     // newer versions of boost allocate context on fiber stack,
 102     // need to create a new one
 103     auto size = fiberManager_.options_.stackSize;
 104     fcontext_ = makeContext(limit, size, &Fiber::fiberFuncHelper);
 105
 106     stackFilledWithMagic_ = true;
 107   }
 108 #else
 109   (void)recordStackUsed;
 110 #endif
 111 }
 112
 113 Fiber::~Fiber() {
 114 #ifdef FOLLY_SANITIZE_ADDRESS
 115   fiberManager_.unpoisonFiberStack(this);
 116 #endif
 117   fiberManager_.stackAllocator_.deallocate(
 118       static_cast<unsigned char*>(fcontext_.stackLimit()),
 119       fiberManager_.options_.stackSize);
 120 }
 121
 122 void Fiber::recordStackPosition() {
 123   int stackDummy;
 124   auto currentPosition = static_cast<size_t>(
 125       static_cast<unsigned char*>(fcontext_.stackBase()) -
 126       static_cast<unsigned char*>(static_cast<void*>(&stackDummy)));
 127   fiberManager_.stackHighWatermark_ =
 128       std::max(fiberManager_.stackHighWatermark_, currentPosition);
 129   VLOG(4) << "Stack usage: " << currentPosition;
 130 }
 131
 132 void Fiber::fiberFuncHelper(intptr_t fiber) {
 133   reinterpret_cast<Fiber*>(fiber)->fiberFunc();
 134 }
 135
 136 void Fiber::fiberFunc() {
 137   while (true) {
 138     DCHECK_EQ(state_, NOT_STARTED);
 139
 140     threadId_ = localThreadId();
 141     state_ = RUNNING;
 142
 143     try {
 144       if (resultFunc_) {
 145         DCHECK(finallyFunc_);
 146         DCHECK(!func_);
 147
 148         resultFunc_();
 149       } else {
 150         DCHECK(func_);
 151         func_();
 152       }
 153     } catch (...) {
 154       fiberManager_.exceptionCallback_(
 155           std::current_exception(), "running Fiber func_/resultFunc_");
 156     }
 157
 158     if (UNLIKELY(recordStackUsed_)) {
 159       fiberManager_.stackHighWatermark_ = std::max(
 160           fiberManager_.stackHighWatermark_, nonMagicInBytes(fcontext_));
 161       VLOG(3) << "Max stack usage: " << fiberManager_.stackHighWatermark_;
 162       CHECK(
 163           fiberManager_.stackHighWatermark_ <
 164           fiberManager_.options_.stackSize - 64)
 165           << "Fiber stack overflow";
 166     }
 167
 168     state_ = INVALID;
 169
 170     auto context = fiberManager_.deactivateFiber(this);
 171
 172     DCHECK_EQ(reinterpret_cast<Fiber*>(context), this);
 173   }
 174 }
 175
 176 intptr_t Fiber::preempt(State state) {
 177   intptr_t ret;
 178
 179   auto preemptImpl = [&]() mutable {
 180     DCHECK_EQ(fiberManager_.activeFiber_, this);
 181     DCHECK_EQ(state_, RUNNING);
 182     DCHECK_NE(state, RUNNING);
 183
 184     state_ = state;
 185
 186     recordStackPosition();
 187
 188     ret = fiberManager_.deactivateFiber(this);
 189
 190     DCHECK_EQ(fiberManager_.activeFiber_, this);
 191     DCHECK_EQ(state_, READY_TO_RUN);
 192     state_ = RUNNING;
 193   };
 194
 195   if (fiberManager_.preemptRunner_) {
 196     fiberManager_.preemptRunner_->run(std::ref(preemptImpl));
 197   } else {
 198     preemptImpl();
 199   }
 200
 201   return ret;
 202 }
 203
 204 Fiber::LocalData::LocalData(const LocalData& other) : data_(nullptr) {
 205   *this = other;
 206 }
 207
 208 Fiber::LocalData& Fiber::LocalData::operator=(const LocalData& other) {
 209   reset();
 210   if (!other.data_) {
 211     return *this;
 212   }
 213
 214   dataSize_ = other.dataSize_;
 215   dataType_ = other.dataType_;
 216   dataDestructor_ = other.dataDestructor_;
 217   dataCopyConstructor_ = other.dataCopyConstructor_;
 218
 219   if (dataSize_ <= kBufferSize) {
 220     data_ = &buffer_;
 221   } else {
 222     data_ = allocateHeapBuffer(dataSize_);
 223   }
 224
 225   dataCopyConstructor_(data_, other.data_);
 226
 227   return *this;
 228 }
 229
 230 void Fiber::LocalData::reset() {
 231   if (!data_) {
 232     return;
 233   }
 234
 235   dataDestructor_(data_);
 236   data_ = nullptr;
 237 }
 238
 239 void* Fiber::LocalData::allocateHeapBuffer(size_t size) {
 240   return new char[size];
 241 }
 242
 243 void Fiber::LocalData::freeHeapBuffer(void* buffer) {
 244   delete[] reinterpret_cast<char*>(buffer);
 245 }
 246 }
 247 }