folly/fibers/FiberManagerInternal-inl.h

   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 #pragma once
  17
  18 #include <cassert>
  19
  20 #include <folly/CPortability.h>
  21 #include <folly/Memory.h>
  22 #include <folly/Optional.h>
  23 #include <folly/Portability.h>
  24 #include <folly/ScopeGuard.h>
  25 #ifdef __APPLE__
  26 #include <folly/ThreadLocal.h>
  27 #endif
  28 #include <folly/fibers/Baton.h>
  29 #include <folly/fibers/Fiber.h>
  30 #include <folly/fibers/LoopController.h>
  31 #include <folly/fibers/Promise.h>
  32 #include <folly/Try.h>
  33
  34 namespace folly {
  35 namespace fibers {
  36
  37 namespace {
  38
  39 inline FiberManager::Options preprocessOptions(FiberManager::Options opts) {
  40 #ifdef FOLLY_SANITIZE_ADDRESS
  41   /* ASAN needs a lot of extra stack space.
  42      16x is a conservative estimate, 8x also worked with tests
  43      where it mattered.  Note that overallocating here does not necessarily
  44      increase RSS, since unused memory is pretty much free. */
  45   opts.stackSize *= 16;
  46 #endif
  47   return opts;
  48 }
  49
  50 } // anonymous
  51
  52 inline void FiberManager::ensureLoopScheduled() {
  53   if (isLoopScheduled_) {
  54     return;
  55   }
  56
  57   isLoopScheduled_ = true;
  58   loopController_->schedule();
  59 }
  60
  61 inline intptr_t FiberManager::activateFiber(Fiber* fiber) {
  62   DCHECK_EQ(activeFiber_, (Fiber*)nullptr);
  63
  64 #ifdef FOLLY_SANITIZE_ADDRESS
  65   DCHECK(!fiber->asanMainStackBase_);
  66   DCHECK(!fiber->asanMainStackSize_);
  67   auto stack = fiber->getStack();
  68   void* asanFakeStack;
  69   registerStartSwitchStackWithAsan(&asanFakeStack, stack.first, stack.second);
  70   SCOPE_EXIT {
  71     registerFinishSwitchStackWithAsan(asanFakeStack, nullptr, nullptr);
  72     fiber->asanMainStackBase_ = nullptr;
  73     fiber->asanMainStackSize_ = 0;
  74   };
  75 #endif
  76
  77   activeFiber_ = fiber;
  78   return jumpContext(&mainContext_, &fiber->fcontext_, fiber->data_);
  79 }
  80
  81 inline intptr_t FiberManager::deactivateFiber(Fiber* fiber) {
  82   DCHECK_EQ(activeFiber_, fiber);
  83
  84 #ifdef FOLLY_SANITIZE_ADDRESS
  85   DCHECK(fiber->asanMainStackBase_);
  86   DCHECK(fiber->asanMainStackSize_);
  87
  88   // Release fake stack if fiber is completed
  89   auto saveFakeStackPtr =
  90       fiber->state_ == Fiber::INVALID ? nullptr : &fiber->asanFakeStack_;
  91   registerStartSwitchStackWithAsan(
  92       saveFakeStackPtr, fiber->asanMainStackBase_, fiber->asanMainStackSize_);
  93   SCOPE_EXIT {
  94     registerFinishSwitchStackWithAsan(
  95         fiber->asanFakeStack_,
  96         &fiber->asanMainStackBase_,
  97         &fiber->asanMainStackSize_);
  98     fiber->asanFakeStack_ = nullptr;
  99   };
 100 #endif
 101
 102   activeFiber_ = nullptr;
 103   return jumpContext(&fiber->fcontext_, &mainContext_, 0);
 104 }
 105
 106 inline void FiberManager::runReadyFiber(Fiber* fiber) {
 107   SCOPE_EXIT {
 108     assert(currentFiber_ == nullptr);
 109     assert(activeFiber_ == nullptr);
 110   };
 111
 112   assert(
 113       fiber->state_ == Fiber::NOT_STARTED ||
 114       fiber->state_ == Fiber::READY_TO_RUN);
 115   currentFiber_ = fiber;
 116   fiber->rcontext_ = RequestContext::setContext(std::move(fiber->rcontext_));
 117   if (observer_) {
 118     observer_->starting(reinterpret_cast<uintptr_t>(fiber));
 119   }
 120
 121   while (fiber->state_ == Fiber::NOT_STARTED ||
 122          fiber->state_ == Fiber::READY_TO_RUN) {
 123     activateFiber(fiber);
 124     if (fiber->state_ == Fiber::AWAITING_IMMEDIATE) {
 125       try {
 126         immediateFunc_();
 127       } catch (...) {
 128         exceptionCallback_(std::current_exception(), "running immediateFunc_");
 129       }
 130       immediateFunc_ = nullptr;
 131       fiber->state_ = Fiber::READY_TO_RUN;
 132     }
 133   }
 134
 135   if (fiber->state_ == Fiber::AWAITING) {
 136     awaitFunc_(*fiber);
 137     awaitFunc_ = nullptr;
 138     if (observer_) {
 139       observer_->stopped(reinterpret_cast<uintptr_t>(fiber));
 140     }
 141     currentFiber_ = nullptr;
 142     fiber->rcontext_ = RequestContext::setContext(std::move(fiber->rcontext_));
 143   } else if (fiber->state_ == Fiber::INVALID) {
 144     assert(fibersActive_ > 0);
 145     --fibersActive_;
 146     // Making sure that task functor is deleted once task is complete.
 147     // NOTE: we must do it on main context, as the fiber is not
 148     // running at this point.
 149     fiber->func_ = nullptr;
 150     fiber->resultFunc_ = nullptr;
 151     if (fiber->finallyFunc_) {
 152       try {
 153         fiber->finallyFunc_();
 154       } catch (...) {
 155         exceptionCallback_(std::current_exception(), "running finallyFunc_");
 156       }
 157       fiber->finallyFunc_ = nullptr;
 158     }
 159     // Make sure LocalData is not accessible from its destructor
 160     if (observer_) {
 161       observer_->stopped(reinterpret_cast<uintptr_t>(fiber));
 162     }
 163     currentFiber_ = nullptr;
 164     fiber->rcontext_ = RequestContext::setContext(std::move(fiber->rcontext_));
 165     fiber->localData_.reset();
 166     fiber->rcontext_.reset();
 167
 168     if (fibersPoolSize_ < options_.maxFibersPoolSize ||
 169         options_.fibersPoolResizePeriodMs > 0) {
 170       fibersPool_.push_front(*fiber);
 171       ++fibersPoolSize_;
 172     } else {
 173       delete fiber;
 174       assert(fibersAllocated_ > 0);
 175       --fibersAllocated_;
 176     }
 177   } else if (fiber->state_ == Fiber::YIELDED) {
 178     if (observer_) {
 179       observer_->stopped(reinterpret_cast<uintptr_t>(fiber));
 180     }
 181     currentFiber_ = nullptr;
 182     fiber->rcontext_ = RequestContext::setContext(std::move(fiber->rcontext_));
 183     fiber->state_ = Fiber::READY_TO_RUN;
 184     yieldedFibers_.push_back(*fiber);
 185   }
 186 }
 187
 188 inline bool FiberManager::loopUntilNoReady() {
 189 #ifndef _WIN32
 190   if (UNLIKELY(!alternateSignalStackRegistered_)) {
 191     registerAlternateSignalStack();
 192   }
 193 #endif
 194
 195   // Support nested FiberManagers
 196   auto originalFiberManager = this;
 197   std::swap(currentFiberManager_, originalFiberManager);
 198
 199   SCOPE_EXIT {
 200     isLoopScheduled_ = false;
 201     if (!readyFibers_.empty()) {
 202       ensureLoopScheduled();
 203     }
 204     std::swap(currentFiberManager_, originalFiberManager);
 205     CHECK_EQ(this, originalFiberManager);
 206   };
 207
 208   bool hadRemoteFiber = true;
 209   while (hadRemoteFiber) {
 210     hadRemoteFiber = false;
 211
 212     while (!readyFibers_.empty()) {
 213       auto& fiber = readyFibers_.front();
 214       readyFibers_.pop_front();
 215       runReadyFiber(&fiber);
 216     }
 217
 218     remoteReadyQueue_.sweep([this, &hadRemoteFiber](Fiber* fiber) {
 219       runReadyFiber(fiber);
 220       hadRemoteFiber = true;
 221     });
 222
 223     remoteTaskQueue_.sweep([this, &hadRemoteFiber](RemoteTask* taskPtr) {
 224       std::unique_ptr<RemoteTask> task(taskPtr);
 225       auto fiber = getFiber();
 226       if (task->localData) {
 227         fiber->localData_ = *task->localData;
 228       }
 229       fiber->rcontext_ = std::move(task->rcontext);
 230
 231       fiber->setFunction(std::move(task->func));
 232       fiber->data_ = reinterpret_cast<intptr_t>(fiber);
 233       if (observer_) {
 234         observer_->runnable(reinterpret_cast<uintptr_t>(fiber));
 235       }
 236       runReadyFiber(fiber);
 237       hadRemoteFiber = true;
 238     });
 239   }
 240
 241   if (observer_) {
 242     for (auto& yielded : yieldedFibers_) {
 243       observer_->runnable(reinterpret_cast<uintptr_t>(&yielded));
 244     }
 245   }
 246   readyFibers_.splice(readyFibers_.end(), yieldedFibers_);
 247
 248   return fibersActive_ > 0;
 249 }
 250
 251 // We need this to be in a struct, not inlined in addTask, because clang crashes
 252 // otherwise.
 253 template <typename F>
 254 struct FiberManager::AddTaskHelper {
 255   class Func;
 256
 257   static constexpr bool allocateInBuffer =
 258       sizeof(Func) <= Fiber::kUserBufferSize;
 259
 260   class Func {
 261    public:
 262     Func(F&& func, FiberManager& fm) : func_(std::forward<F>(func)), fm_(fm) {}
 263
 264     void operator()() {
 265       try {
 266         func_();
 267       } catch (...) {
 268         fm_.exceptionCallback_(
 269             std::current_exception(), "running Func functor");
 270       }
 271       if (allocateInBuffer) {
 272         this->~Func();
 273       } else {
 274         delete this;
 275       }
 276     }
 277
 278    private:
 279     F func_;
 280     FiberManager& fm_;
 281   };
 282 };
 283
 284 template <typename F>
 285 void FiberManager::addTask(F&& func) {
 286   typedef AddTaskHelper<F> Helper;
 287
 288   auto fiber = getFiber();
 289   initLocalData(*fiber);
 290
 291   if (Helper::allocateInBuffer) {
 292     auto funcLoc = static_cast<typename Helper::Func*>(fiber->getUserBuffer());
 293     new (funcLoc) typename Helper::Func(std::forward<F>(func), *this);
 294
 295     fiber->setFunction(std::ref(*funcLoc));
 296   } else {
 297     auto funcLoc = new typename Helper::Func(std::forward<F>(func), *this);
 298
 299     fiber->setFunction(std::ref(*funcLoc));
 300   }
 301
 302   fiber->data_ = reinterpret_cast<intptr_t>(fiber);
 303   readyFibers_.push_back(*fiber);
 304   if (observer_) {
 305     observer_->runnable(reinterpret_cast<uintptr_t>(fiber));
 306   }
 307
 308   ensureLoopScheduled();
 309 }
 310
 311 template <typename F>
 312 void FiberManager::addTaskRemote(F&& func) {
 313   auto task = [&]() {
 314     auto currentFm = getFiberManagerUnsafe();
 315     if (currentFm && currentFm->currentFiber_ &&
 316         currentFm->localType_ == localType_) {
 317       return folly::make_unique<RemoteTask>(
 318           std::forward<F>(func), currentFm->currentFiber_->localData_);
 319     }
 320     return folly::make_unique<RemoteTask>(std::forward<F>(func));
 321   }();
 322   auto insertHead = [&]() {
 323     return remoteTaskQueue_.insertHead(task.release());
 324   };
 325   loopController_->scheduleThreadSafe(std::ref(insertHead));
 326 }
 327
 328 template <typename X>
 329 struct IsRvalueRefTry {
 330   static const bool value = false;
 331 };
 332 template <typename T>
 333 struct IsRvalueRefTry<folly::Try<T>&&> {
 334   static const bool value = true;
 335 };
 336
 337 // We need this to be in a struct, not inlined in addTaskFinally, because clang
 338 // crashes otherwise.
 339 template <typename F, typename G>
 340 struct FiberManager::AddTaskFinallyHelper {
 341   class Func;
 342
 343   typedef typename std::result_of<F()>::type Result;
 344
 345   class Finally {
 346    public:
 347     Finally(G finally, FiberManager& fm)
 348         : finally_(std::move(finally)), fm_(fm) {}
 349
 350     void operator()() {
 351       try {
 352         finally_(std::move(*result_));
 353       } catch (...) {
 354         fm_.exceptionCallback_(
 355             std::current_exception(), "running Finally functor");
 356       }
 357
 358       if (allocateInBuffer) {
 359         this->~Finally();
 360       } else {
 361         delete this;
 362       }
 363     }
 364
 365    private:
 366     friend class Func;
 367
 368     G finally_;
 369     folly::Optional<folly::Try<Result>> result_;
 370     FiberManager& fm_;
 371   };
 372
 373   class Func {
 374    public:
 375     Func(F func, Finally& finally)
 376         : func_(std::move(func)), result_(finally.result_) {}
 377
 378     void operator()() {
 379       result_ = folly::makeTryWith(std::move(func_));
 380
 381       if (allocateInBuffer) {
 382         this->~Func();
 383       } else {
 384         delete this;
 385       }
 386     }
 387
 388    private:
 389     F func_;
 390     folly::Optional<folly::Try<Result>>& result_;
 391   };
 392
 393   static constexpr bool allocateInBuffer =
 394       sizeof(Func) + sizeof(Finally) <= Fiber::kUserBufferSize;
 395 };
 396
 397 template <typename F, typename G>
 398 void FiberManager::addTaskFinally(F&& func, G&& finally) {
 399   typedef typename std::result_of<F()>::type Result;
 400
 401   static_assert(
 402       IsRvalueRefTry<typename FirstArgOf<G>::type>::value,
 403       "finally(arg): arg must be Try<T>&&");
 404   static_assert(
 405       std::is_convertible<
 406           Result,
 407           typename std::remove_reference<
 408               typename FirstArgOf<G>::type>::type::element_type>::value,
 409       "finally(Try<T>&&): T must be convertible from func()'s return type");
 410
 411   auto fiber = getFiber();
 412   initLocalData(*fiber);
 413
 414   typedef AddTaskFinallyHelper<
 415       typename std::decay<F>::type,
 416       typename std::decay<G>::type>
 417       Helper;
 418
 419   if (Helper::allocateInBuffer) {
 420     auto funcLoc = static_cast<typename Helper::Func*>(fiber->getUserBuffer());
 421     auto finallyLoc =
 422         static_cast<typename Helper::Finally*>(static_cast<void*>(funcLoc + 1));
 423
 424     new (finallyLoc) typename Helper::Finally(std::forward<G>(finally), *this);
 425     new (funcLoc) typename Helper::Func(std::forward<F>(func), *finallyLoc);
 426
 427     fiber->setFunctionFinally(std::ref(*funcLoc), std::ref(*finallyLoc));
 428   } else {
 429     auto finallyLoc =
 430         new typename Helper::Finally(std::forward<G>(finally), *this);
 431     auto funcLoc =
 432         new typename Helper::Func(std::forward<F>(func), *finallyLoc);
 433
 434     fiber->setFunctionFinally(std::ref(*funcLoc), std::ref(*finallyLoc));
 435   }
 436
 437   fiber->data_ = reinterpret_cast<intptr_t>(fiber);
 438   readyFibers_.push_back(*fiber);
 439   if (observer_) {
 440     observer_->runnable(reinterpret_cast<uintptr_t>(fiber));
 441   }
 442
 443   ensureLoopScheduled();
 444 }
 445
 446 template <typename F>
 447 typename std::result_of<F()>::type FiberManager::runInMainContext(F&& func) {
 448   if (UNLIKELY(activeFiber_ == nullptr)) {
 449     return func();
 450   }
 451
 452   typedef typename std::result_of<F()>::type Result;
 453
 454   folly::Try<Result> result;
 455   auto f = [&func, &result]() mutable {
 456     result = folly::makeTryWith(std::forward<F>(func));
 457   };
 458
 459   immediateFunc_ = std::ref(f);
 460   activeFiber_->preempt(Fiber::AWAITING_IMMEDIATE);
 461
 462   return std::move(result).value();
 463 }
 464
 465 inline FiberManager& FiberManager::getFiberManager() {
 466   assert(currentFiberManager_ != nullptr);
 467   return *currentFiberManager_;
 468 }
 469
 470 inline FiberManager* FiberManager::getFiberManagerUnsafe() {
 471   return currentFiberManager_;
 472 }
 473
 474 inline bool FiberManager::hasActiveFiber() const {
 475   return activeFiber_ != nullptr;
 476 }
 477
 478 inline void FiberManager::yield() {
 479   assert(currentFiberManager_ == this);
 480   assert(activeFiber_ != nullptr);
 481   assert(activeFiber_->state_ == Fiber::RUNNING);
 482   activeFiber_->preempt(Fiber::YIELDED);
 483 }
 484
 485 template <typename T>
 486 T& FiberManager::local() {
 487   if (std::type_index(typeid(T)) == localType_ && currentFiber_) {
 488     return currentFiber_->localData_.get<T>();
 489   }
 490   return localThread<T>();
 491 }
 492
 493 template <typename T>
 494 T& FiberManager::localThread() {
 495 #ifndef __APPLE__
 496   static thread_local T t;
 497   return t;
 498 #else // osx doesn't support thread_local
 499   static ThreadLocal<T> t;
 500   return *t;
 501 #endif
 502 }
 503
 504 inline void FiberManager::initLocalData(Fiber& fiber) {
 505   auto fm = getFiberManagerUnsafe();
 506   if (fm && fm->currentFiber_ && fm->localType_ == localType_) {
 507     fiber.localData_ = fm->currentFiber_->localData_;
 508   }
 509   fiber.rcontext_ = RequestContext::saveContext();
 510 }
 511
 512 template <typename LocalT>
 513 FiberManager::FiberManager(
 514     LocalType<LocalT>,
 515     std::unique_ptr<LoopController> loopController__,
 516     Options options)
 517     : loopController_(std::move(loopController__)),
 518       stackAllocator_(options.useGuardPages),
 519       options_(preprocessOptions(std::move(options))),
 520       exceptionCallback_([](std::exception_ptr eptr, std::string context) {
 521         try {
 522           std::rethrow_exception(eptr);
 523         } catch (const std::exception& e) {
 524           LOG(DFATAL) << "Exception " << typeid(e).name() << " with message '"
 525                       << e.what() << "' was thrown in "
 526                       << "FiberManager with context '" << context << "'";
 527         } catch (...) {
 528           LOG(DFATAL) << "Unknown exception was thrown in FiberManager with "
 529                       << "context '" << context << "'";
 530         }
 531       }),
 532       timeoutManager_(std::make_shared<TimeoutController>(*loopController_)),
 533       fibersPoolResizer_(*this),
 534       localType_(typeid(LocalT)) {
 535   loopController_->setFiberManager(this);
 536 }
 537
 538 template <typename F>
 539 typename FirstArgOf<F>::type::value_type inline await(F&& func) {
 540   typedef typename FirstArgOf<F>::type::value_type Result;
 541   typedef typename FirstArgOf<F>::type::baton_type BatonT;
 542
 543   return Promise<Result, BatonT>::await(std::forward<F>(func));
 544 }
 545 }
 546 }