folly/fibers/FiberManager.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/futures/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    * Setup fibers preempt runner.
 301    */
 302   void setPreemptRunner(InlineFunctionRunner* preemptRunner);
 303
 304   /**
 305    * Returns an estimate of the number of fibers which are waiting to run (does
 306    * not include fibers or tasks scheduled remotely).
 307    */
 308   size_t runQueueSize() const {
 309     return readyFibers_.size() + yieldedFibers_.size();
 310   }
 311
 312   static FiberManager& getFiberManager();
 313   static FiberManager* getFiberManagerUnsafe();
 314
 315  private:
 316   friend class Baton;
 317   friend class Fiber;
 318   template <typename F>
 319   struct AddTaskHelper;
 320   template <typename F, typename G>
 321   struct AddTaskFinallyHelper;
 322
 323   struct RemoteTask {
 324     template <typename F>
 325     explicit RemoteTask(F&& f)
 326         : func(std::forward<F>(f)), rcontext(RequestContext::saveContext()) {}
 327     template <typename F>
 328     RemoteTask(F&& f, const Fiber::LocalData& localData_)
 329         : func(std::forward<F>(f)),
 330           localData(folly::make_unique<Fiber::LocalData>(localData_)),
 331           rcontext(RequestContext::saveContext()) {}
 332     folly::Function<void()> func;
 333     std::unique_ptr<Fiber::LocalData> localData;
 334     std::shared_ptr<RequestContext> rcontext;
 335     AtomicIntrusiveLinkedListHook<RemoteTask> nextRemoteTask;
 336   };
 337
 338   intptr_t activateFiber(Fiber* fiber);
 339   intptr_t deactivateFiber(Fiber* fiber);
 340
 341   typedef folly::IntrusiveList<Fiber, &Fiber::listHook_> FiberTailQueue;
 342   typedef folly::IntrusiveList<Fiber, &Fiber::globalListHook_>
 343       GlobalFiberTailQueue;
 344
 345   Fiber* activeFiber_{nullptr}; /**< active fiber, nullptr on main context */
 346   /**
 347    * Same as active fiber, but also set for functions run from fiber on main
 348    * context.
 349    */
 350   Fiber* currentFiber_{nullptr};
 351
 352   FiberTailQueue readyFibers_; /**< queue of fibers ready to be executed */
 353   FiberTailQueue yieldedFibers_; /**< queue of fibers which have yielded
 354                                       execution */
 355   FiberTailQueue fibersPool_; /**< pool of unitialized Fiber objects */
 356
 357   GlobalFiberTailQueue allFibers_; /**< list of all Fiber objects owned */
 358
 359   size_t fibersAllocated_{0}; /**< total number of fibers allocated */
 360   size_t fibersPoolSize_{0}; /**< total number of fibers in the free pool */
 361   size_t fibersActive_{0}; /**< number of running or blocked fibers */
 362   size_t fiberId_{0}; /**< id of last fiber used */
 363
 364   /**
 365    * Maximum number of active fibers in the last period lasting
 366    * Options::fibersPoolResizePeriod milliseconds.
 367    */
 368   size_t maxFibersActiveLastPeriod_{0};
 369
 370   FContext::ContextStruct mainContext_; /**< stores loop function context */
 371
 372   std::unique_ptr<LoopController> loopController_;
 373   bool isLoopScheduled_{false}; /**< was the ready loop scheduled to run? */
 374
 375   /**
 376    * When we are inside FiberManager loop this points to FiberManager. Otherwise
 377    * it's nullptr
 378    */
 379   static FOLLY_TLS FiberManager* currentFiberManager_;
 380
 381   /**
 382    * Allocator used to allocate stack for Fibers in the pool.
 383    * Allocates stack on the stack of the main context.
 384    */
 385   GuardPageAllocator stackAllocator_;
 386
 387   const Options options_; /**< FiberManager options */
 388
 389   /**
 390    * Largest observed individual Fiber stack usage in bytes.
 391    */
 392   size_t stackHighWatermark_{0};
 393
 394   /**
 395    * Schedules a loop with loopController (unless already scheduled before).
 396    */
 397   void ensureLoopScheduled();
 398
 399   /**
 400    * @return An initialized Fiber object from the pool
 401    */
 402   Fiber* getFiber();
 403
 404   /**
 405    * Sets local data for given fiber if all conditions are met.
 406    */
 407   void initLocalData(Fiber& fiber);
 408
 409   /**
 410    * Function passed to the await call.
 411    */
 412   folly::Function<void(Fiber&)> awaitFunc_;
 413
 414   /**
 415    * Function passed to the runInMainContext call.
 416    */
 417   folly::Function<void()> immediateFunc_;
 418
 419   /**
 420    * Preempt runner.
 421    */
 422   InlineFunctionRunner* preemptRunner_{nullptr};
 423
 424   /**
 425    * Fiber's execution observer.
 426    */
 427   ExecutionObserver* observer_{nullptr};
 428
 429   ExceptionCallback exceptionCallback_; /**< task exception callback */
 430
 431   folly::AtomicIntrusiveLinkedList<Fiber, &Fiber::nextRemoteReady_>
 432       remoteReadyQueue_;
 433
 434   folly::AtomicIntrusiveLinkedList<RemoteTask, &RemoteTask::nextRemoteTask>
 435       remoteTaskQueue_;
 436
 437   std::shared_ptr<TimeoutController> timeoutManager_;
 438
 439   struct FibersPoolResizer {
 440     explicit FibersPoolResizer(FiberManager& fm) : fiberManager_(fm) {}
 441     void operator()();
 442
 443    private:
 444     FiberManager& fiberManager_;
 445   };
 446
 447   FibersPoolResizer fibersPoolResizer_;
 448   bool fibersPoolResizerScheduled_{false};
 449
 450   void doFibersPoolResizing();
 451
 452   /**
 453    * Only local of this type will be available for fibers.
 454    */
 455   std::type_index localType_;
 456
 457   void runReadyFiber(Fiber* fiber);
 458   void remoteReadyInsert(Fiber* fiber);
 459
 460 #ifdef FOLLY_SANITIZE_ADDRESS
 461
 462   // These methods notify ASAN when a fiber is entered/exited so that ASAN can
 463   // find the right stack extents when it needs to poison/unpoison the stack.
 464
 465   void registerFiberActivationWithAsan(Fiber* fiber);
 466   void registerFiberDeactivationWithAsan(Fiber* fiber);
 467   void unpoisonFiberStack(const Fiber* fiber);
 468
 469 #endif // FOLLY_SANITIZE_ADDRESS
 470
 471   bool alternateSignalStackRegistered_{false};
 472
 473   void registerAlternateSignalStack();
 474 };
 475
 476 /**
 477  * @return      true iff we are running in a fiber's context
 478  */
 479 inline bool onFiber() {
 480   auto fm = FiberManager::getFiberManagerUnsafe();
 481   return fm ? fm->hasActiveFiber() : false;
 482 }
 483
 484 /**
 485  * Add a new task to be executed.
 486  *
 487  * @param func Task functor; must have a signature of `void func()`.
 488  *             The object will be destroyed once task execution is complete.
 489  */
 490 template <typename F>
 491 inline void addTask(F&& func) {
 492   return FiberManager::getFiberManager().addTask(std::forward<F>(func));
 493 }
 494
 495 /**
 496  * Add a new task. When the task is complete, execute finally(Try<Result>&&)
 497  * on the main context.
 498  * Task functor is run and destroyed on the fiber context.
 499  * Finally functor is run and destroyed on the main context.
 500  *
 501  * @param func Task functor; must have a signature of `T func()` for some T.
 502  * @param finally Finally functor; must have a signature of
 503  *                `void finally(Try<T>&&)` and will be passed
 504  *                the result of func() (including the exception if occurred).
 505  */
 506 template <typename F, typename G>
 507 inline void addTaskFinally(F&& func, G&& finally) {
 508   return FiberManager::getFiberManager().addTaskFinally(
 509       std::forward<F>(func), std::forward<G>(finally));
 510 }
 511
 512 /**
 513  * Blocks task execution until given promise is fulfilled.
 514  *
 515  * Calls function passing in a Promise<T>, which has to be fulfilled.
 516  *
 517  * @return data which was used to fulfill the promise.
 518  */
 519 template <typename F>
 520 typename FirstArgOf<F>::type::value_type inline await(F&& func);
 521
 522 /**
 523  * If called from a fiber, immediately switches to the FiberManager's context
 524  * and runs func(), going back to the Fiber's context after completion.
 525  * Outside a fiber, just calls func() directly.
 526  *
 527  * @return value returned by func().
 528  */
 529 template <typename F>
 530 typename std::result_of<F()>::type inline runInMainContext(F&& func) {
 531   auto fm = FiberManager::getFiberManagerUnsafe();
 532   if (UNLIKELY(fm == nullptr)) {
 533     return func();
 534   }
 535   return fm->runInMainContext(std::forward<F>(func));
 536 }
 537
 538 /**
 539  * Returns a refference to a fiber-local context for given Fiber. Should be
 540  * always called with the same T for each fiber. Fiber-local context is lazily
 541  * default-constructed on first request.
 542  * When new task is scheduled via addTask / addTaskRemote from a fiber its
 543  * fiber-local context is copied into the new fiber.
 544  */
 545 template <typename T>
 546 T& local() {
 547   auto fm = FiberManager::getFiberManagerUnsafe();
 548   if (fm) {
 549     return fm->local<T>();
 550   }
 551   return FiberManager::localThread<T>();
 552 }
 553
 554 inline void yield() {
 555   auto fm = FiberManager::getFiberManagerUnsafe();
 556   if (fm) {
 557     fm->yield();
 558   } else {
 559     std::this_thread::yield();
 560   }
 561 }
 562 }
 563 }
 564
 565 #include "FiberManager-inl.h"