folly/fibers/FiberManagerInternal.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 <functional>
  19 #include <memory>
  20 #include <queue>
  21 #include <thread>
  22 #include <type_traits>
  23 #include <typeindex>
  24 #include <unordered_set>
  25 #include <vector>
  26
  27 #include <folly/AtomicIntrusiveLinkedList.h>
  28 #include <folly/Executor.h>
  29 #include <folly/IntrusiveList.h>
  30 #include <folly/Likely.h>
  31 #include <folly/Try.h>
  32 #include <folly/io/async/Request.h>
  33
  34 #include <folly/experimental/ExecutionObserver.h>
  35 #include <folly/fibers/BoostContextCompatibility.h>
  36 #include <folly/fibers/Fiber.h>
  37 #include <folly/fibers/GuardPageAllocator.h>
  38 #include <folly/fibers/TimeoutController.h>
  39 #include <folly/fibers/traits.h>
  40
  41 namespace folly {
  42
  43 template <class T>
  44 class Future;
  45
  46 namespace fibers {
  47
  48 class Baton;
  49 class Fiber;
  50 class LoopController;
  51 class TimeoutController;
  52
  53 template <typename T>
  54 class LocalType {};
  55
  56 class InlineFunctionRunner {
  57  public:
  58   virtual ~InlineFunctionRunner() {}
  59
  60   /**
  61    * func must be executed inline and only once.
  62    */
  63   virtual void run(folly::Function<void()> func) = 0;
  64 };
  65
  66 /**
  67  * @class FiberManager
  68  * @brief Single-threaded task execution engine.
  69  *
  70  * FiberManager allows semi-parallel task execution on the same thread. Each
  71  * task can notify FiberManager that it is blocked on something (via await())
  72  * call. This will pause execution of this task and it will be resumed only
  73  * when it is unblocked (via setData()).
  74  */
  75 class FiberManager : public ::folly::Executor {
  76  public:
  77   struct Options {
  78     static constexpr size_t kDefaultStackSize{16 * 1024};
  79
  80     /**
  81      * Maximum stack size for fibers which will be used for executing all the
  82      * tasks.
  83      */
  84     size_t stackSize{kDefaultStackSize};
  85
  86     /**
  87      * Record exact amount of stack used.
  88      *
  89      * This is fairly expensive: we fill each newly allocated stack
  90      * with some known value and find the boundary of unused stack
  91      * with linear search every time we surrender the stack back to fibersPool.
  92      * 0 disables stack recording.
  93      */
  94     size_t recordStackEvery{0};
  95
  96     /**
  97      * Keep at most this many free fibers in the pool.
  98      * This way the total number of fibers in the system is always bounded
  99      * by the number of active fibers + maxFibersPoolSize.
 100      */
 101     size_t maxFibersPoolSize{1000};
 102
 103     /**
 104      * Protect limited amount of fiber stacks with guard pages.
 105      */
 106     bool useGuardPages{true};
 107
 108     /**
 109      * Free unnecessary fibers in the fibers pool every fibersPoolResizePeriodMs
 110      * milliseconds. If value is 0, periodic resizing of the fibers pool is
 111      * disabled.
 112      */
 113     uint32_t fibersPoolResizePeriodMs{0};
 114
 115     constexpr Options() {}
 116   };
 117
 118   using ExceptionCallback =
 119       folly::Function<void(std::exception_ptr, std::string)>;
 120
 121   FiberManager(const FiberManager&) = delete;
 122   FiberManager& operator=(const FiberManager&) = delete;
 123
 124   /**
 125    * Initializes, but doesn't start FiberManager loop
 126    *
 127    * @param loopController
 128    * @param options FiberManager options
 129    */
 130   explicit FiberManager(
 131       std::unique_ptr<LoopController> loopController,
 132       Options options = Options());
 133
 134   /**
 135    * Initializes, but doesn't start FiberManager loop
 136    *
 137    * @param loopController
 138    * @param options FiberManager options
 139    * @tparam LocalT only local of this type may be stored on fibers.
 140    *                Locals of other types will be considered thread-locals.
 141    */
 142   template <typename LocalT>
 143   FiberManager(
 144       LocalType<LocalT>,
 145       std::unique_ptr<LoopController> loopController,
 146       Options options = Options());
 147
 148   ~FiberManager();
 149
 150   /**
 151    * Controller access.
 152    */
 153   LoopController& loopController();
 154   const LoopController& loopController() const;
 155
 156   /**
 157    * Keeps running ready tasks until the list of ready tasks is empty.
 158    *
 159    * @return True if there are any waiting tasks remaining.
 160    */
 161   bool loopUntilNoReady();
 162
 163   /**
 164    * @return true if there are outstanding tasks.
 165    */
 166   bool hasTasks() const;
 167
 168   /**
 169    * Sets exception callback which will be called if any of the tasks throws an
 170    * exception.
 171    *
 172    * @param ec
 173    */
 174   void setExceptionCallback(ExceptionCallback ec);
 175
 176   /**
 177    * Add a new task to be executed. Must be called from FiberManager's thread.
 178    *
 179    * @param func Task functor; must have a signature of `void func()`.
 180    *             The object will be destroyed once task execution is complete.
 181    */
 182   template <typename F>
 183   void addTask(F&& func);
 184
 185   /**
 186    * Add a new task to be executed and return a future that will be set on
 187    * return from func. Must be called from FiberManager's thread.
 188    *
 189    * @param func Task functor; must have a signature of `void func()`.
 190    *             The object will be destroyed once task execution is complete.
 191    */
 192   template <typename F>
 193   auto addTaskFuture(F&& func) -> folly::Future<
 194       typename folly::Unit::Lift<typename std::result_of<F()>::type>::type>;
 195   /**
 196    * Add a new task to be executed. Safe to call from other threads.
 197    *
 198    * @param func Task function; must have a signature of `void func()`.
 199    *             The object will be destroyed once task execution is complete.
 200    */
 201   template <typename F>
 202   void addTaskRemote(F&& func);
 203
 204   /**
 205    * Add a new task to be executed and return a future that will be set on
 206    * return from func. Safe to call from other threads.
 207    *
 208    * @param func Task function; must have a signature of `void func()`.
 209    *             The object will be destroyed once task execution is complete.
 210    */
 211   template <typename F>
 212   auto addTaskRemoteFuture(F&& func) -> folly::Future<
 213       typename folly::Unit::Lift<typename std::result_of<F()>::type>::type>;
 214
 215   // Executor interface calls addTaskRemote
 216   void add(folly::Func f) override {
 217     addTaskRemote(std::move(f));
 218   }
 219
 220   /**
 221    * Add a new task. When the task is complete, execute finally(Try<Result>&&)
 222    * on the main context.
 223    *
 224    * @param func Task functor; must have a signature of `T func()` for some T.
 225    * @param finally Finally functor; must have a signature of
 226    *                `void finally(Try<T>&&)` and will be passed
 227    *                the result of func() (including the exception if occurred).
 228    */
 229   template <typename F, typename G>
 230   void addTaskFinally(F&& func, G&& finally);
 231
 232   /**
 233    * If called from a fiber, immediately switches to the FiberManager's context
 234    * and runs func(), going back to the Fiber's context after completion.
 235    * Outside a fiber, just calls func() directly.
 236    *
 237    * @return value returned by func().
 238    */
 239   template <typename F>
 240   typename std::result_of<F()>::type runInMainContext(F&& func);
 241
 242   /**
 243    * Returns a refference to a fiber-local context for given Fiber. Should be
 244    * always called with the same T for each fiber. Fiber-local context is lazily
 245    * default-constructed on first request.
 246    * When new task is scheduled via addTask / addTaskRemote from a fiber its
 247    * fiber-local context is copied into the new fiber.
 248    */
 249   template <typename T>
 250   T& local();
 251
 252   template <typename T>
 253   static T& localThread();
 254
 255   /**
 256    * @return How many fiber objects (and stacks) has this manager allocated.
 257    */
 258   size_t fibersAllocated() const;
 259
 260   /**
 261    * @return How many of the allocated fiber objects are currently
 262    * in the free pool.
 263    */
 264   size_t fibersPoolSize() const;
 265
 266   /**
 267    * return     true if running activeFiber_ is not nullptr.
 268    */
 269   bool hasActiveFiber() const;
 270
 271   /**
 272    * @return The currently running fiber or null if no fiber is executing.
 273    */
 274   Fiber* currentFiber() const {
 275     return currentFiber_;
 276   }
 277
 278   /**
 279    * @return What was the most observed fiber stack usage (in bytes).
 280    */
 281   size_t stackHighWatermark() const;
 282
 283   /**
 284    * Yield execution of the currently running fiber. Must only be called from a
 285    * fiber executing on this FiberManager. The calling fiber will be scheduled
 286    * when all other fibers have had a chance to run and the event loop is
 287    * serviced.
 288    */
 289   void yield();
 290
 291   /**
 292    * Setup fibers execution observation/instrumentation. Fiber locals are
 293    * available to observer.
 294    *
 295    * @param observer  Fiber's execution observer.
 296    */
 297   void setObserver(ExecutionObserver* observer);
 298
 299   /**
 300    * @return Current observer for this FiberManager. Returns nullptr
 301    * if no observer has been set.
 302    */
 303   ExecutionObserver* getObserver();
 304
 305   /**
 306    * Setup fibers preempt runner.
 307    */
 308   void setPreemptRunner(InlineFunctionRunner* preemptRunner);
 309
 310   /**
 311    * Returns an estimate of the number of fibers which are waiting to run (does
 312    * not include fibers or tasks scheduled remotely).
 313    */
 314   size_t runQueueSize() const {
 315     return readyFibers_.size() + yieldedFibers_.size();
 316   }
 317
 318   static FiberManager& getFiberManager();
 319   static FiberManager* getFiberManagerUnsafe();
 320
 321  private:
 322   friend class Baton;
 323   friend class Fiber;
 324   template <typename F>
 325   struct AddTaskHelper;
 326   template <typename F, typename G>
 327   struct AddTaskFinallyHelper;
 328
 329   struct RemoteTask {
 330     template <typename F>
 331     explicit RemoteTask(F&& f)
 332         : func(std::forward<F>(f)), rcontext(RequestContext::saveContext()) {}
 333     template <typename F>
 334     RemoteTask(F&& f, const Fiber::LocalData& localData_)
 335         : func(std::forward<F>(f)),
 336           localData(folly::make_unique<Fiber::LocalData>(localData_)),
 337           rcontext(RequestContext::saveContext()) {}
 338     folly::Function<void()> func;
 339     std::unique_ptr<Fiber::LocalData> localData;
 340     std::shared_ptr<RequestContext> rcontext;
 341     AtomicIntrusiveLinkedListHook<RemoteTask> nextRemoteTask;
 342   };
 343
 344   void activateFiber(Fiber* fiber);
 345   void deactivateFiber(Fiber* fiber);
 346
 347   typedef folly::IntrusiveList<Fiber, &Fiber::listHook_> FiberTailQueue;
 348   typedef folly::IntrusiveList<Fiber, &Fiber::globalListHook_>
 349       GlobalFiberTailQueue;
 350
 351   Fiber* activeFiber_{nullptr}; /**< active fiber, nullptr on main context */
 352   /**
 353    * Same as active fiber, but also set for functions run from fiber on main
 354    * context.
 355    */
 356   Fiber* currentFiber_{nullptr};
 357
 358   FiberTailQueue readyFibers_; /**< queue of fibers ready to be executed */
 359   FiberTailQueue yieldedFibers_; /**< queue of fibers which have yielded
 360                                       execution */
 361   FiberTailQueue fibersPool_; /**< pool of unitialized Fiber objects */
 362
 363   GlobalFiberTailQueue allFibers_; /**< list of all Fiber objects owned */
 364
 365   size_t fibersAllocated_{0}; /**< total number of fibers allocated */
 366   size_t fibersPoolSize_{0}; /**< total number of fibers in the free pool */
 367   size_t fibersActive_{0}; /**< number of running or blocked fibers */
 368   size_t fiberId_{0}; /**< id of last fiber used */
 369
 370   /**
 371    * Maximum number of active fibers in the last period lasting
 372    * Options::fibersPoolResizePeriod milliseconds.
 373    */
 374   size_t maxFibersActiveLastPeriod_{0};
 375
 376   std::unique_ptr<LoopController> loopController_;
 377   bool isLoopScheduled_{false}; /**< was the ready loop scheduled to run? */
 378
 379   /**
 380    * When we are inside FiberManager loop this points to FiberManager. Otherwise
 381    * it's nullptr
 382    */
 383   static FOLLY_TLS FiberManager* currentFiberManager_;
 384
 385   /**
 386    * Allocator used to allocate stack for Fibers in the pool.
 387    * Allocates stack on the stack of the main context.
 388    */
 389   GuardPageAllocator stackAllocator_;
 390
 391   const Options options_; /**< FiberManager options */
 392
 393   /**
 394    * Largest observed individual Fiber stack usage in bytes.
 395    */
 396   size_t stackHighWatermark_{0};
 397
 398   /**
 399    * Schedules a loop with loopController (unless already scheduled before).
 400    */
 401   void ensureLoopScheduled();
 402
 403   /**
 404    * @return An initialized Fiber object from the pool
 405    */
 406   Fiber* getFiber();
 407
 408   /**
 409    * Sets local data for given fiber if all conditions are met.
 410    */
 411   void initLocalData(Fiber& fiber);
 412
 413   /**
 414    * Function passed to the await call.
 415    */
 416   folly::Function<void(Fiber&)> awaitFunc_;
 417
 418   /**
 419    * Function passed to the runInMainContext call.
 420    */
 421   folly::Function<void()> immediateFunc_;
 422
 423   /**
 424    * Preempt runner.
 425    */
 426   InlineFunctionRunner* preemptRunner_{nullptr};
 427
 428   /**
 429    * Fiber's execution observer.
 430    */
 431   ExecutionObserver* observer_{nullptr};
 432
 433   ExceptionCallback exceptionCallback_; /**< task exception callback */
 434
 435   folly::AtomicIntrusiveLinkedList<Fiber, &Fiber::nextRemoteReady_>
 436       remoteReadyQueue_;
 437
 438   folly::AtomicIntrusiveLinkedList<RemoteTask, &RemoteTask::nextRemoteTask>
 439       remoteTaskQueue_;
 440
 441   std::shared_ptr<TimeoutController> timeoutManager_;
 442
 443   struct FibersPoolResizer {
 444     explicit FibersPoolResizer(FiberManager& fm) : fiberManager_(fm) {}
 445     void operator()();
 446
 447    private:
 448     FiberManager& fiberManager_;
 449   };
 450
 451   FibersPoolResizer fibersPoolResizer_;
 452   bool fibersPoolResizerScheduled_{false};
 453
 454   void doFibersPoolResizing();
 455
 456   /**
 457    * Only local of this type will be available for fibers.
 458    */
 459   std::type_index localType_;
 460
 461   void runReadyFiber(Fiber* fiber);
 462   void remoteReadyInsert(Fiber* fiber);
 463
 464 #ifdef FOLLY_SANITIZE_ADDRESS
 465
 466   // These methods notify ASAN when a fiber is entered/exited so that ASAN can
 467   // find the right stack extents when it needs to poison/unpoison the stack.
 468
 469   void registerStartSwitchStackWithAsan(
 470       void** saveFakeStack,
 471       const void* stackBase,
 472       size_t stackSize);
 473   void registerFinishSwitchStackWithAsan(
 474       void* fakeStack,
 475       const void** saveStackBase,
 476       size_t* saveStackSize);
 477   void unpoisonFiberStack(const Fiber* fiber);
 478
 479 #endif // FOLLY_SANITIZE_ADDRESS
 480
 481 #ifndef _WIN32
 482   bool alternateSignalStackRegistered_{false};
 483
 484   void registerAlternateSignalStack();
 485 #endif
 486 };
 487
 488 /**
 489  * @return      true iff we are running in a fiber's context
 490  */
 491 inline bool onFiber() {
 492   auto fm = FiberManager::getFiberManagerUnsafe();
 493   return fm ? fm->hasActiveFiber() : false;
 494 }
 495
 496 /**
 497  * Add a new task to be executed.
 498  *
 499  * @param func Task functor; must have a signature of `void func()`.
 500  *             The object will be destroyed once task execution is complete.
 501  */
 502 template <typename F>
 503 inline void addTask(F&& func) {
 504   return FiberManager::getFiberManager().addTask(std::forward<F>(func));
 505 }
 506
 507 /**
 508  * Add a new task. When the task is complete, execute finally(Try<Result>&&)
 509  * on the main context.
 510  * Task functor is run and destroyed on the fiber context.
 511  * Finally functor is run and destroyed on the main context.
 512  *
 513  * @param func Task functor; must have a signature of `T func()` for some T.
 514  * @param finally Finally functor; must have a signature of
 515  *                `void finally(Try<T>&&)` and will be passed
 516  *                the result of func() (including the exception if occurred).
 517  */
 518 template <typename F, typename G>
 519 inline void addTaskFinally(F&& func, G&& finally) {
 520   return FiberManager::getFiberManager().addTaskFinally(
 521       std::forward<F>(func), std::forward<G>(finally));
 522 }
 523
 524 /**
 525  * Blocks task execution until given promise is fulfilled.
 526  *
 527  * Calls function passing in a Promise<T>, which has to be fulfilled.
 528  *
 529  * @return data which was used to fulfill the promise.
 530  */
 531 template <typename F>
 532 typename FirstArgOf<F>::type::value_type inline await(F&& func);
 533
 534 /**
 535  * If called from a fiber, immediately switches to the FiberManager's context
 536  * and runs func(), going back to the Fiber's context after completion.
 537  * Outside a fiber, just calls func() directly.
 538  *
 539  * @return value returned by func().
 540  */
 541 template <typename F>
 542 typename std::result_of<F()>::type inline runInMainContext(F&& func) {
 543   auto fm = FiberManager::getFiberManagerUnsafe();
 544   if (UNLIKELY(fm == nullptr)) {
 545     return func();
 546   }
 547   return fm->runInMainContext(std::forward<F>(func));
 548 }
 549
 550 /**
 551  * Returns a refference to a fiber-local context for given Fiber. Should be
 552  * always called with the same T for each fiber. Fiber-local context is lazily
 553  * default-constructed on first request.
 554  * When new task is scheduled via addTask / addTaskRemote from a fiber its
 555  * fiber-local context is copied into the new fiber.
 556  */
 557 template <typename T>
 558 T& local() {
 559   auto fm = FiberManager::getFiberManagerUnsafe();
 560   if (fm) {
 561     return fm->local<T>();
 562   }
 563   return FiberManager::localThread<T>();
 564 }
 565
 566 inline void yield() {
 567   auto fm = FiberManager::getFiberManagerUnsafe();
 568   if (fm) {
 569     fm->yield();
 570   } else {
 571     std::this_thread::yield();
 572   }
 573 }
 574 }
 575 }
 576
 577 #include <folly/fibers/FiberManagerInternal-inl.h>