folly/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 <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 } // anonymous 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   fiberManager_.unpoisonFiberStack(this);
 107 #endif
 108   fiberManager_.stackAllocator_.deallocate(fiberStackLimit_, fiberStackSize_);
 109 }
 110
 111 void Fiber::recordStackPosition() {
 112   int stackDummy;
 113   auto currentPosition = static_cast<size_t>(
 114       fiberStackLimit_ + fiberStackSize_ -
 115       static_cast<unsigned char*>(static_cast<void*>(&stackDummy)));
 116   fiberManager_.stackHighWatermark_ =
 117       std::max(fiberManager_.stackHighWatermark_, currentPosition);
 118   VLOG(4) << "Stack usage: " << currentPosition;
 119 }
 120
 121 [[noreturn]] void Fiber::fiberFunc() {
 122 #ifdef FOLLY_SANITIZE_ADDRESS
 123   fiberManager_.registerFinishSwitchStackWithAsan(
 124       nullptr, &asanMainStackBase_, &asanMainStackSize_);
 125 #endif
 126
 127   while (true) {
 128     DCHECK_EQ(state_, NOT_STARTED);
 129
 130     threadId_ = localThreadId();
 131     state_ = RUNNING;
 132
 133     try {
 134       if (resultFunc_) {
 135         DCHECK(finallyFunc_);
 136         DCHECK(!func_);
 137
 138         resultFunc_();
 139       } else {
 140         DCHECK(func_);
 141         func_();
 142       }
 143     } catch (...) {
 144       fiberManager_.exceptionCallback_(
 145           std::current_exception(), "running Fiber func_/resultFunc_");
 146     }
 147
 148     if (UNLIKELY(recordStackUsed_)) {
 149       fiberManager_.stackHighWatermark_ = std::max(
 150           fiberManager_.stackHighWatermark_,
 151           nonMagicInBytes(fiberStackLimit_, fiberStackSize_));
 152       VLOG(3) << "Max stack usage: " << fiberManager_.stackHighWatermark_;
 153       CHECK(
 154           fiberManager_.stackHighWatermark_ <
 155           fiberManager_.options_.stackSize - 64)
 156           << "Fiber stack overflow";
 157     }
 158
 159     state_ = INVALID;
 160
 161     fiberManager_.deactivateFiber(this);
 162   }
 163 }
 164
 165 void Fiber::preempt(State state) {
 166   auto preemptImpl = [&]() mutable {
 167     DCHECK_EQ(fiberManager_.activeFiber_, this);
 168     DCHECK_EQ(state_, RUNNING);
 169     DCHECK_NE(state, RUNNING);
 170
 171     state_ = state;
 172
 173     recordStackPosition();
 174
 175     fiberManager_.deactivateFiber(this);
 176
 177     DCHECK_EQ(fiberManager_.activeFiber_, this);
 178     DCHECK_EQ(state_, READY_TO_RUN);
 179     state_ = RUNNING;
 180   };
 181
 182   if (fiberManager_.preemptRunner_) {
 183     fiberManager_.preemptRunner_->run(std::ref(preemptImpl));
 184   } else {
 185     preemptImpl();
 186   }
 187 }
 188
 189 Fiber::LocalData::LocalData(const LocalData& other) : data_(nullptr) {
 190   *this = other;
 191 }
 192
 193 Fiber::LocalData& Fiber::LocalData::operator=(const LocalData& other) {
 194   reset();
 195   if (!other.data_) {
 196     return *this;
 197   }
 198
 199   dataSize_ = other.dataSize_;
 200   dataType_ = other.dataType_;
 201   dataDestructor_ = other.dataDestructor_;
 202   dataCopyConstructor_ = other.dataCopyConstructor_;
 203
 204   if (dataSize_ <= kBufferSize) {
 205     data_ = &buffer_;
 206   } else {
 207     data_ = allocateHeapBuffer(dataSize_);
 208   }
 209
 210   dataCopyConstructor_(data_, other.data_);
 211
 212   return *this;
 213 }
 214
 215 void Fiber::LocalData::reset() {
 216   if (!data_) {
 217     return;
 218   }
 219
 220   dataDestructor_(data_);
 221   data_ = nullptr;
 222 }
 223
 224 void* Fiber::LocalData::allocateHeapBuffer(size_t size) {
 225   return new char[size];
 226 }
 227
 228 void Fiber::LocalData::freeHeapBuffer(void* buffer) {
 229   delete[] reinterpret_cast<char*>(buffer);
 230 }
 231 }
 232 }