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