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