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