folly/experimental/fibers/FiberManager-inl.h

   1 /*
   2  * Copyright 2015 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/Memory.h>
  21 #include <folly/Optional.h>
  22 #include <folly/Portability.h>
  23 #include <folly/ScopeGuard.h>
  24 #include <folly/experimental/fibers/Baton.h>
  25 #include <folly/experimental/fibers/Fiber.h>
  26 #include <folly/experimental/fibers/Promise.h>
  27 #include <folly/experimental/fibers/LoopController.h>
  28 #include <folly/futures/Try.h>
  29
  30 namespace folly { namespace fibers {
  31
  32 inline void FiberManager::ensureLoopScheduled() {
  33   if (isLoopScheduled_) {
  34     return;
  35   }
  36
  37   isLoopScheduled_ = true;
  38   loopController_->schedule();
  39 }
  40
  41 inline void FiberManager::runReadyFiber(Fiber* fiber) {
  42   assert(fiber->state_ == Fiber::NOT_STARTED ||
  43          fiber->state_ == Fiber::READY_TO_RUN);
  44   currentFiber_ = fiber;
  45
  46   while (fiber->state_ == Fiber::NOT_STARTED ||
  47          fiber->state_ == Fiber::READY_TO_RUN) {
  48     activeFiber_ = fiber;
  49     jumpContext(&mainContext_, &fiber->fcontext_, fiber->data_);
  50     if (fiber->state_ == Fiber::AWAITING_IMMEDIATE) {
  51       try {
  52         immediateFunc_();
  53       } catch (...) {
  54         exceptionCallback_(std::current_exception(), "running immediateFunc_");
  55       }
  56       immediateFunc_ = nullptr;
  57       fiber->state_ = Fiber::READY_TO_RUN;
  58     }
  59   }
  60
  61   if (fiber->state_ == Fiber::AWAITING) {
  62     awaitFunc_(*fiber);
  63     awaitFunc_ = nullptr;
  64   } else if (fiber->state_ == Fiber::INVALID) {
  65     assert(fibersActive_ > 0);
  66     --fibersActive_;
  67     // Making sure that task functor is deleted once task is complete.
  68     // NOTE: we must do it on main context, as the fiber is not
  69     // running at this point.
  70     fiber->func_ = nullptr;
  71     fiber->resultFunc_ = nullptr;
  72     if (fiber->finallyFunc_) {
  73       try {
  74         fiber->finallyFunc_();
  75       } catch (...) {
  76         exceptionCallback_(std::current_exception(), "running finallyFunc_");
  77       }
  78       fiber->finallyFunc_ = nullptr;
  79     }
  80     fiber->localData_.reset();
  81
  82     if (fibersPoolSize_ < options_.maxFibersPoolSize) {
  83       fibersPool_.push_front(*fiber);
  84       ++fibersPoolSize_;
  85     } else {
  86       delete fiber;
  87       assert(fibersAllocated_ > 0);
  88       --fibersAllocated_;
  89     }
  90   }
  91   currentFiber_ = nullptr;
  92 }
  93
  94 inline bool FiberManager::loopUntilNoReady() {
  95   SCOPE_EXIT {
  96     isLoopScheduled_ = false;
  97     currentFiberManager_ = nullptr;
  98   };
  99
 100   currentFiberManager_ = this;
 101
 102   bool hadRemoteFiber = true;
 103   while (hadRemoteFiber) {
 104     hadRemoteFiber = false;
 105
 106     while (!readyFibers_.empty()) {
 107       auto& fiber = readyFibers_.front();
 108       readyFibers_.pop_front();
 109       runReadyFiber(&fiber);
 110     }
 111
 112     remoteReadyQueue_.sweep(
 113       [this, &hadRemoteFiber] (Fiber* fiber) {
 114         runReadyFiber(fiber);
 115         hadRemoteFiber = true;
 116       }
 117     );
 118
 119     remoteTaskQueue_.sweep(
 120       [this, &hadRemoteFiber] (RemoteTask* taskPtr) {
 121         std::unique_ptr<RemoteTask> task(taskPtr);
 122         auto fiber = getFiber();
 123         if (task->localData) {
 124           fiber->localData_ = *task->localData;
 125         }
 126
 127         fiber->setFunction(std::move(task->func));
 128         fiber->data_ = reinterpret_cast<intptr_t>(fiber);
 129         runReadyFiber(fiber);
 130         hadRemoteFiber = true;
 131       }
 132     );
 133   }
 134
 135   return fibersActive_ > 0;
 136 }
 137
 138 // We need this to be in a struct, not inlined in addTask, because clang crashes
 139 // otherwise.
 140 template <typename F>
 141 struct FiberManager::AddTaskHelper {
 142   class Func;
 143
 144   static constexpr bool allocateInBuffer =
 145     sizeof(Func) <= Fiber::kUserBufferSize;
 146
 147   class Func {
 148    public:
 149     Func(F&& func, FiberManager& fm) :
 150         func_(std::forward<F>(func)), fm_(fm) {}
 151
 152     void operator()() {
 153       try {
 154         func_();
 155       } catch (...) {
 156         fm_.exceptionCallback_(std::current_exception(),
 157                                "running Func functor");
 158       }
 159       if (allocateInBuffer) {
 160         this->~Func();
 161       } else {
 162         delete this;
 163       }
 164     }
 165
 166    private:
 167     F func_;
 168     FiberManager& fm_;
 169   };
 170 };
 171
 172 template <typename F>
 173 void FiberManager::addTask(F&& func) {
 174   typedef AddTaskHelper<F> Helper;
 175
 176   auto fiber = getFiber();
 177   if (currentFiber_) {
 178     fiber->localData_ = currentFiber_->localData_;
 179   }
 180
 181   if (Helper::allocateInBuffer) {
 182     auto funcLoc = static_cast<typename Helper::Func*>(fiber->getUserBuffer());
 183     new (funcLoc) typename Helper::Func(std::forward<F>(func), *this);
 184
 185     fiber->setFunction(std::ref(*funcLoc));
 186   } else {
 187     auto funcLoc = new typename Helper::Func(std::forward<F>(func), *this);
 188
 189     fiber->setFunction(std::ref(*funcLoc));
 190   }
 191
 192   fiber->data_ = reinterpret_cast<intptr_t>(fiber);
 193   readyFibers_.push_back(*fiber);
 194
 195   ensureLoopScheduled();
 196 }
 197
 198 template <typename F>
 199 void FiberManager::addTaskRemote(F&& func) {
 200   auto task = [&]() {
 201     auto currentFm = getFiberManagerUnsafe();
 202     if (currentFm && currentFm->currentFiber_) {
 203       return folly::make_unique<RemoteTask>(
 204         std::forward<F>(func),
 205         currentFm->currentFiber_->localData_);
 206     }
 207     return folly::make_unique<RemoteTask>(std::forward<F>(func));
 208   }();
 209   if (remoteTaskQueue_.insertHead(task.release())) {
 210     loopController_->scheduleThreadSafe();
 211   }
 212 }
 213
 214 template <typename X>
 215 struct IsRvalueRefTry { static const bool value = false; };
 216 template <typename T>
 217 struct IsRvalueRefTry<folly::Try<T>&&> { static const bool value = true; };
 218
 219 // We need this to be in a struct, not inlined in addTaskFinally, because clang
 220 // crashes otherwise.
 221 template <typename F, typename G>
 222 struct FiberManager::AddTaskFinallyHelper {
 223   class Func;
 224   class Finally;
 225
 226   typedef typename std::result_of<F()>::type Result;
 227
 228   static constexpr bool allocateInBuffer =
 229     sizeof(Func) + sizeof(Finally) <= Fiber::kUserBufferSize;
 230
 231   class Finally {
 232    public:
 233     Finally(G&& finally,
 234             FiberManager& fm) :
 235         finally_(std::move(finally)),
 236         fm_(fm) {
 237     }
 238
 239     void operator()() {
 240       try {
 241         finally_(std::move(*result_));
 242       } catch (...) {
 243         fm_.exceptionCallback_(std::current_exception(),
 244                                "running Finally functor");
 245       }
 246
 247       if (allocateInBuffer) {
 248         this->~Finally();
 249       } else {
 250         delete this;
 251       }
 252     }
 253
 254    private:
 255     friend class Func;
 256
 257     G finally_;
 258     folly::Optional<folly::Try<Result>> result_;
 259     FiberManager& fm_;
 260   };
 261
 262   class Func {
 263    public:
 264     Func(F&& func, Finally& finally) :
 265         func_(std::move(func)), result_(finally.result_) {}
 266
 267     void operator()() {
 268       result_ = folly::makeTryFunction(std::move(func_));
 269
 270       if (allocateInBuffer) {
 271         this->~Func();
 272       } else {
 273         delete this;
 274       }
 275     }
 276
 277    private:
 278     F func_;
 279     folly::Optional<folly::Try<Result>>& result_;
 280   };
 281 };
 282
 283 template <typename F, typename G>
 284 void FiberManager::addTaskFinally(F&& func, G&& finally) {
 285   typedef typename std::result_of<F()>::type Result;
 286
 287   static_assert(
 288     IsRvalueRefTry<typename FirstArgOf<G>::type>::value,
 289     "finally(arg): arg must be Try<T>&&");
 290   static_assert(
 291     std::is_convertible<
 292       Result,
 293       typename std::remove_reference<
 294         typename FirstArgOf<G>::type
 295       >::type::element_type
 296     >::value,
 297     "finally(Try<T>&&): T must be convertible from func()'s return type");
 298
 299   auto fiber = getFiber();
 300   if (currentFiber_) {
 301     fiber->localData_ = currentFiber_->localData_;
 302   }
 303
 304   typedef AddTaskFinallyHelper<F,G> Helper;
 305
 306   if (Helper::allocateInBuffer) {
 307     auto funcLoc = static_cast<typename Helper::Func*>(
 308       fiber->getUserBuffer());
 309     auto finallyLoc = static_cast<typename Helper::Finally*>(
 310       static_cast<void*>(funcLoc + 1));
 311
 312     new (finallyLoc) typename Helper::Finally(std::move(finally), *this);
 313     new (funcLoc) typename Helper::Func(std::move(func), *finallyLoc);
 314
 315     fiber->setFunctionFinally(std::ref(*funcLoc), std::ref(*finallyLoc));
 316   } else {
 317     auto finallyLoc = new typename Helper::Finally(std::move(finally), *this);
 318     auto funcLoc = new typename Helper::Func(std::move(func), *finallyLoc);
 319
 320     fiber->setFunctionFinally(std::ref(*funcLoc), std::ref(*finallyLoc));
 321   }
 322
 323   fiber->data_ = reinterpret_cast<intptr_t>(fiber);
 324   readyFibers_.push_back(*fiber);
 325
 326   ensureLoopScheduled();
 327 }
 328
 329 template <typename F>
 330 typename std::result_of<F()>::type
 331 FiberManager::runInMainContext(F&& func) {
 332   return runInMainContextHelper(std::forward<F>(func));
 333 }
 334
 335 template <typename F>
 336 inline typename std::enable_if<
 337   !std::is_same<typename std::result_of<F()>::type, void>::value,
 338   typename std::result_of<F()>::type>::type
 339 FiberManager::runInMainContextHelper(F&& func) {
 340   if (UNLIKELY(activeFiber_ == nullptr)) {
 341     return func();
 342   }
 343
 344   typedef typename std::result_of<F()>::type Result;
 345
 346   folly::Try<Result> result;
 347   auto f = [&func, &result]() mutable {
 348     result = folly::makeTryFunction(std::forward<F>(func));
 349   };
 350
 351   immediateFunc_ = std::ref(f);
 352   activeFiber_->preempt(Fiber::AWAITING_IMMEDIATE);
 353
 354   return std::move(result.value());
 355 }
 356
 357 template <typename F>
 358 inline typename std::enable_if<
 359   std::is_same<typename std::result_of<F()>::type, void>::value,
 360   void>::type
 361 FiberManager::runInMainContextHelper(F&& func) {
 362   if (UNLIKELY(activeFiber_ == nullptr)) {
 363     func();
 364     return;
 365   }
 366
 367   immediateFunc_ = std::ref(func);
 368   activeFiber_->preempt(Fiber::AWAITING_IMMEDIATE);
 369 }
 370
 371 inline FiberManager& FiberManager::getFiberManager() {
 372   assert(currentFiberManager_ != nullptr);
 373   return *currentFiberManager_;
 374 }
 375
 376 inline FiberManager* FiberManager::getFiberManagerUnsafe() {
 377   return currentFiberManager_;
 378 }
 379
 380 inline bool FiberManager::hasActiveFiber() const {
 381   return activeFiber_ != nullptr;
 382 }
 383
 384 template <typename T>
 385 T& FiberManager::local() {
 386   if (currentFiber_) {
 387     return currentFiber_->localData_.get<T>();
 388   }
 389   return localThread<T>();
 390 }
 391
 392 template <typename T>
 393 T& FiberManager::localThread() {
 394   static thread_local T t;
 395   return t;
 396 }
 397
 398 template <typename F>
 399 typename FirstArgOf<F>::type::value_type
 400 inline await(F&& func) {
 401   typedef typename FirstArgOf<F>::type::value_type Result;
 402
 403   folly::Try<Result> result;
 404
 405   Baton baton;
 406   baton.wait([&func, &result, &baton]() mutable {
 407       func(Promise<Result>(result, baton));
 408     });
 409
 410   return folly::moveFromTry(std::move(result));
 411 }
 412
 413 }}