folly/fibers/FiberManagerInternal.h

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