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   void loopUntilNoReady();
 160
 161   /**
 162    * This should only be called by a LoopController.
 163    */
 164   void loopUntilNoReadyImpl();
 165
 166   /**
 167    * @return true if there are outstanding tasks.
 168    */
 169   bool hasTasks() const;
 170
 171   /**
 172    * Sets exception callback which will be called if any of the tasks throws an
 173    * exception.
 174    *
 175    * @param ec
 176    */
 177   void setExceptionCallback(ExceptionCallback ec);
 178
 179   /**
 180    * Add a new task to be executed. Must be called from FiberManager's thread.
 181    *
 182    * @param func Task functor; must have a signature of `void func()`.
 183    *             The object will be destroyed once task execution is complete.
 184    */
 185   template <typename F>
 186   void addTask(F&& func);
 187
 188   /**
 189    * Add a new task to be executed and return a future that will be set on
 190    * return from func. Must be called from FiberManager's thread.
 191    *
 192    * @param func Task functor; must have a signature of `void func()`.
 193    *             The object will be destroyed once task execution is complete.
 194    */
 195   template <typename F>
 196   auto addTaskFuture(F&& func) -> folly::Future<
 197       typename folly::Unit::Lift<typename std::result_of<F()>::type>::type>;
 198   /**
 199    * Add a new task to be executed. Safe to call from other threads.
 200    *
 201    * @param func Task function; must have a signature of `void func()`.
 202    *             The object will be destroyed once task execution is complete.
 203    */
 204   template <typename F>
 205   void addTaskRemote(F&& func);
 206
 207   /**
 208    * Add a new task to be executed and return a future that will be set on
 209    * return from func. Safe to call from other threads.
 210    *
 211    * @param func Task function; must have a signature of `void func()`.
 212    *             The object will be destroyed once task execution is complete.
 213    */
 214   template <typename F>
 215   auto addTaskRemoteFuture(F&& func) -> folly::Future<
 216       typename folly::Unit::Lift<typename std::result_of<F()>::type>::type>;
 217
 218   // Executor interface calls addTaskRemote
 219   void add(folly::Func f) override {
 220     addTaskRemote(std::move(f));
 221   }
 222
 223   /**
 224    * Add a new task. When the task is complete, execute finally(Try<Result>&&)
 225    * on the main context.
 226    *
 227    * @param func Task functor; must have a signature of `T func()` for some T.
 228    * @param finally Finally functor; must have a signature of
 229    *                `void finally(Try<T>&&)` and will be passed
 230    *                the result of func() (including the exception if occurred).
 231    */
 232   template <typename F, typename G>
 233   void addTaskFinally(F&& func, G&& finally);
 234
 235   /**
 236    * If called from a fiber, immediately switches to the FiberManager's context
 237    * and runs func(), going back to the Fiber's context after completion.
 238    * Outside a fiber, just calls func() directly.
 239    *
 240    * @return value returned by func().
 241    */
 242   template <typename F>
 243   typename std::result_of<F()>::type runInMainContext(F&& func);
 244
 245   /**
 246    * Returns a refference to a fiber-local context for given Fiber. Should be
 247    * always called with the same T for each fiber. Fiber-local context is lazily
 248    * default-constructed on first request.
 249    * When new task is scheduled via addTask / addTaskRemote from a fiber its
 250    * fiber-local context is copied into the new fiber.
 251    */
 252   template <typename T>
 253   T& local();
 254
 255   template <typename T>
 256   static T& localThread();
 257
 258   /**
 259    * @return How many fiber objects (and stacks) has this manager allocated.
 260    */
 261   size_t fibersAllocated() const;
 262
 263   /**
 264    * @return How many of the allocated fiber objects are currently
 265    * in the free pool.
 266    */
 267   size_t fibersPoolSize() const;
 268
 269   /**
 270    * return     true if running activeFiber_ is not nullptr.
 271    */
 272   bool hasActiveFiber() const;
 273
 274   /**
 275    * @return The currently running fiber or null if no fiber is executing.
 276    */
 277   Fiber* currentFiber() const {
 278     return currentFiber_;
 279   }
 280
 281   /**
 282    * @return What was the most observed fiber stack usage (in bytes).
 283    */
 284   size_t stackHighWatermark() const;
 285
 286   /**
 287    * Yield execution of the currently running fiber. Must only be called from a
 288    * fiber executing on this FiberManager. The calling fiber will be scheduled
 289    * when all other fibers have had a chance to run and the event loop is
 290    * serviced.
 291    */
 292   void yield();
 293
 294   /**
 295    * Setup fibers execution observation/instrumentation. Fiber locals are
 296    * available to observer.
 297    *
 298    * @param observer  Fiber's execution observer.
 299    */
 300   void setObserver(ExecutionObserver* observer);
 301
 302   /**
 303    * @return Current observer for this FiberManager. Returns nullptr
 304    * if no observer has been set.
 305    */
 306   ExecutionObserver* getObserver();
 307
 308   /**
 309    * Setup fibers preempt runner.
 310    */
 311   void setPreemptRunner(InlineFunctionRunner* preemptRunner);
 312
 313   /**
 314    * Returns an estimate of the number of fibers which are waiting to run (does
 315    * not include fibers or tasks scheduled remotely).
 316    */
 317   size_t runQueueSize() const {
 318     return readyFibers_.size() + yieldedFibers_.size();
 319   }
 320
 321   static FiberManager& getFiberManager();
 322   static FiberManager* getFiberManagerUnsafe();
 323
 324  private:
 325   friend class Baton;
 326   friend class Fiber;
 327   template <typename F>
 328   struct AddTaskHelper;
 329   template <typename F, typename G>
 330   struct AddTaskFinallyHelper;
 331
 332   struct RemoteTask {
 333     template <typename F>
 334     explicit RemoteTask(F&& f)
 335         : func(std::forward<F>(f)), rcontext(RequestContext::saveContext()) {}
 336     template <typename F>
 337     RemoteTask(F&& f, const Fiber::LocalData& localData_)
 338         : func(std::forward<F>(f)),
 339           localData(folly::make_unique<Fiber::LocalData>(localData_)),
 340           rcontext(RequestContext::saveContext()) {}
 341     folly::Function<void()> func;
 342     std::unique_ptr<Fiber::LocalData> localData;
 343     std::shared_ptr<RequestContext> rcontext;
 344     AtomicIntrusiveLinkedListHook<RemoteTask> nextRemoteTask;
 345   };
 346
 347   void activateFiber(Fiber* fiber);
 348   void deactivateFiber(Fiber* fiber);
 349
 350   typedef folly::IntrusiveList<Fiber, &Fiber::listHook_> FiberTailQueue;
 351   typedef folly::IntrusiveList<Fiber, &Fiber::globalListHook_>
 352       GlobalFiberTailQueue;
 353
 354   Fiber* activeFiber_{nullptr}; /**< active fiber, nullptr on main context */
 355   /**
 356    * Same as active fiber, but also set for functions run from fiber on main
 357    * context.
 358    */
 359   Fiber* currentFiber_{nullptr};
 360
 361   FiberTailQueue readyFibers_; /**< queue of fibers ready to be executed */
 362   FiberTailQueue yieldedFibers_; /**< queue of fibers which have yielded
 363                                       execution */
 364   FiberTailQueue fibersPool_; /**< pool of unitialized Fiber objects */
 365
 366   GlobalFiberTailQueue allFibers_; /**< list of all Fiber objects owned */
 367
 368   size_t fibersAllocated_{0}; /**< total number of fibers allocated */
 369   size_t fibersPoolSize_{0}; /**< total number of fibers in the free pool */
 370   size_t fibersActive_{0}; /**< number of running or blocked fibers */
 371   size_t fiberId_{0}; /**< id of last fiber used */
 372
 373   /**
 374    * Maximum number of active fibers in the last period lasting
 375    * Options::fibersPoolResizePeriod milliseconds.
 376    */
 377   size_t maxFibersActiveLastPeriod_{0};
 378
 379   std::unique_ptr<LoopController> loopController_;
 380   bool isLoopScheduled_{false}; /**< was the ready loop scheduled to run? */
 381
 382   /**
 383    * When we are inside FiberManager loop this points to FiberManager. Otherwise
 384    * it's nullptr
 385    */
 386   static FOLLY_TLS FiberManager* currentFiberManager_;
 387
 388   /**
 389    * Allocator used to allocate stack for Fibers in the pool.
 390    * Allocates stack on the stack of the main context.
 391    */
 392   GuardPageAllocator stackAllocator_;
 393
 394   const Options options_; /**< FiberManager options */
 395
 396   /**
 397    * Largest observed individual Fiber stack usage in bytes.
 398    */
 399   size_t stackHighWatermark_{0};
 400
 401   /**
 402    * Schedules a loop with loopController (unless already scheduled before).
 403    */
 404   void ensureLoopScheduled();
 405
 406   /**
 407    * @return An initialized Fiber object from the pool
 408    */
 409   Fiber* getFiber();
 410
 411   /**
 412    * Sets local data for given fiber if all conditions are met.
 413    */
 414   void initLocalData(Fiber& fiber);
 415
 416   /**
 417    * Function passed to the await call.
 418    */
 419   folly::Function<void(Fiber&)> awaitFunc_;
 420
 421   /**
 422    * Function passed to the runInMainContext call.
 423    */
 424   folly::Function<void()> immediateFunc_;
 425
 426   /**
 427    * Preempt runner.
 428    */
 429   InlineFunctionRunner* preemptRunner_{nullptr};
 430
 431   /**
 432    * Fiber's execution observer.
 433    */
 434   ExecutionObserver* observer_{nullptr};
 435
 436   ExceptionCallback exceptionCallback_; /**< task exception callback */
 437
 438   folly::AtomicIntrusiveLinkedList<Fiber, &Fiber::nextRemoteReady_>
 439       remoteReadyQueue_;
 440
 441   folly::AtomicIntrusiveLinkedList<RemoteTask, &RemoteTask::nextRemoteTask>
 442       remoteTaskQueue_;
 443
 444   std::shared_ptr<TimeoutController> timeoutManager_;
 445
 446   struct FibersPoolResizer {
 447     explicit FibersPoolResizer(FiberManager& fm) : fiberManager_(fm) {}
 448     void operator()();
 449
 450    private:
 451     FiberManager& fiberManager_;
 452   };
 453
 454   FibersPoolResizer fibersPoolResizer_;
 455   bool fibersPoolResizerScheduled_{false};
 456
 457   void doFibersPoolResizing();
 458
 459   /**
 460    * Only local of this type will be available for fibers.
 461    */
 462   std::type_index localType_;
 463
 464   void runReadyFiber(Fiber* fiber);
 465   void remoteReadyInsert(Fiber* fiber);
 466
 467 #ifdef FOLLY_SANITIZE_ADDRESS
 468
 469   // These methods notify ASAN when a fiber is entered/exited so that ASAN can
 470   // find the right stack extents when it needs to poison/unpoison the stack.
 471
 472   void registerStartSwitchStackWithAsan(
 473       void** saveFakeStack,
 474       const void* stackBase,
 475       size_t stackSize);
 476   void registerFinishSwitchStackWithAsan(
 477       void* fakeStack,
 478       const void** saveStackBase,
 479       size_t* saveStackSize);
 480   void unpoisonFiberStack(const Fiber* fiber);
 481
 482 #endif // FOLLY_SANITIZE_ADDRESS
 483
 484 #ifndef _WIN32
 485   bool alternateSignalStackRegistered_{false};
 486
 487   void registerAlternateSignalStack();
 488 #endif
 489 };
 490
 491 /**
 492  * @return      true iff we are running in a fiber's context
 493  */
 494 inline bool onFiber() {
 495   auto fm = FiberManager::getFiberManagerUnsafe();
 496   return fm ? fm->hasActiveFiber() : false;
 497 }
 498
 499 /**
 500  * Add a new task to be executed.
 501  *
 502  * @param func Task functor; must have a signature of `void func()`.
 503  *             The object will be destroyed once task execution is complete.
 504  */
 505 template <typename F>
 506 inline void addTask(F&& func) {
 507   return FiberManager::getFiberManager().addTask(std::forward<F>(func));
 508 }
 509
 510 /**
 511  * Add a new task. When the task is complete, execute finally(Try<Result>&&)
 512  * on the main context.
 513  * Task functor is run and destroyed on the fiber context.
 514  * Finally functor is run and destroyed on the main context.
 515  *
 516  * @param func Task functor; must have a signature of `T func()` for some T.
 517  * @param finally Finally functor; must have a signature of
 518  *                `void finally(Try<T>&&)` and will be passed
 519  *                the result of func() (including the exception if occurred).
 520  */
 521 template <typename F, typename G>
 522 inline void addTaskFinally(F&& func, G&& finally) {
 523   return FiberManager::getFiberManager().addTaskFinally(
 524       std::forward<F>(func), std::forward<G>(finally));
 525 }
 526
 527 /**
 528  * Blocks task execution until given promise is fulfilled.
 529  *
 530  * Calls function passing in a Promise<T>, which has to be fulfilled.
 531  *
 532  * @return data which was used to fulfill the promise.
 533  */
 534 template <typename F>
 535 typename FirstArgOf<F>::type::value_type inline await(F&& func);
 536
 537 /**
 538  * If called from a fiber, immediately switches to the FiberManager's context
 539  * and runs func(), going back to the Fiber's context after completion.
 540  * Outside a fiber, just calls func() directly.
 541  *
 542  * @return value returned by func().
 543  */
 544 template <typename F>
 545 typename std::result_of<F()>::type inline runInMainContext(F&& func) {
 546   auto fm = FiberManager::getFiberManagerUnsafe();
 547   if (UNLIKELY(fm == nullptr)) {
 548     return func();
 549   }
 550   return fm->runInMainContext(std::forward<F>(func));
 551 }
 552
 553 /**
 554  * Returns a refference to a fiber-local context for given Fiber. Should be
 555  * always called with the same T for each fiber. Fiber-local context is lazily
 556  * default-constructed on first request.
 557  * When new task is scheduled via addTask / addTaskRemote from a fiber its
 558  * fiber-local context is copied into the new fiber.
 559  */
 560 template <typename T>
 561 T& local() {
 562   auto fm = FiberManager::getFiberManagerUnsafe();
 563   if (fm) {
 564     return fm->local<T>();
 565   }
 566   return FiberManager::localThread<T>();
 567 }
 568
 569 inline void yield() {
 570   auto fm = FiberManager::getFiberManagerUnsafe();
 571   if (fm) {
 572     fm->yield();
 573   } else {
 574     std::this_thread::yield();
 575   }
 576 }
 577 }
 578 }
 579
 580 #include <folly/fibers/FiberManagerInternal-inl.h>