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/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   intptr_t activateFiber(Fiber* fiber);
 345   intptr_t 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   FContext::ContextStruct mainContext_; /**< stores loop function context */
 377
 378   std::unique_ptr<LoopController> loopController_;
 379   bool isLoopScheduled_{false}; /**< was the ready loop scheduled to run? */
 380
 381   /**
 382    * When we are inside FiberManager loop this points to FiberManager. Otherwise
 383    * it's nullptr
 384    */
 385   static FOLLY_TLS FiberManager* currentFiberManager_;
 386
 387   /**
 388    * Allocator used to allocate stack for Fibers in the pool.
 389    * Allocates stack on the stack of the main context.
 390    */
 391   GuardPageAllocator stackAllocator_;
 392
 393   const Options options_; /**< FiberManager options */
 394
 395   /**
 396    * Largest observed individual Fiber stack usage in bytes.
 397    */
 398   size_t stackHighWatermark_{0};
 399
 400   /**
 401    * Schedules a loop with loopController (unless already scheduled before).
 402    */
 403   void ensureLoopScheduled();
 404
 405   /**
 406    * @return An initialized Fiber object from the pool
 407    */
 408   Fiber* getFiber();
 409
 410   /**
 411    * Sets local data for given fiber if all conditions are met.
 412    */
 413   void initLocalData(Fiber& fiber);
 414
 415   /**
 416    * Function passed to the await call.
 417    */
 418   folly::Function<void(Fiber&)> awaitFunc_;
 419
 420   /**
 421    * Function passed to the runInMainContext call.
 422    */
 423   folly::Function<void()> immediateFunc_;
 424
 425   /**
 426    * Preempt runner.
 427    */
 428   InlineFunctionRunner* preemptRunner_{nullptr};
 429
 430   /**
 431    * Fiber's execution observer.
 432    */
 433   ExecutionObserver* observer_{nullptr};
 434
 435   ExceptionCallback exceptionCallback_; /**< task exception callback */
 436
 437   folly::AtomicIntrusiveLinkedList<Fiber, &Fiber::nextRemoteReady_>
 438       remoteReadyQueue_;
 439
 440   folly::AtomicIntrusiveLinkedList<RemoteTask, &RemoteTask::nextRemoteTask>
 441       remoteTaskQueue_;
 442
 443   std::shared_ptr<TimeoutController> timeoutManager_;
 444
 445   struct FibersPoolResizer {
 446     explicit FibersPoolResizer(FiberManager& fm) : fiberManager_(fm) {}
 447     void operator()();
 448
 449    private:
 450     FiberManager& fiberManager_;
 451   };
 452
 453   FibersPoolResizer fibersPoolResizer_;
 454   bool fibersPoolResizerScheduled_{false};
 455
 456   void doFibersPoolResizing();
 457
 458   /**
 459    * Only local of this type will be available for fibers.
 460    */
 461   std::type_index localType_;
 462
 463   void runReadyFiber(Fiber* fiber);
 464   void remoteReadyInsert(Fiber* fiber);
 465
 466 #ifdef FOLLY_SANITIZE_ADDRESS
 467
 468   // These methods notify ASAN when a fiber is entered/exited so that ASAN can
 469   // find the right stack extents when it needs to poison/unpoison the stack.
 470
 471   void registerFiberActivationWithAsan(Fiber* fiber);
 472   void registerFiberDeactivationWithAsan(Fiber* fiber);
 473   void unpoisonFiberStack(const Fiber* fiber);
 474
 475 #endif // FOLLY_SANITIZE_ADDRESS
 476
 477 #ifndef _WIN32
 478   bool alternateSignalStackRegistered_{false};
 479
 480   void registerAlternateSignalStack();
 481 #endif
 482 };
 483
 484 /**
 485  * @return      true iff we are running in a fiber's context
 486  */
 487 inline bool onFiber() {
 488   auto fm = FiberManager::getFiberManagerUnsafe();
 489   return fm ? fm->hasActiveFiber() : false;
 490 }
 491
 492 /**
 493  * Add a new task to be executed.
 494  *
 495  * @param func Task functor; must have a signature of `void func()`.
 496  *             The object will be destroyed once task execution is complete.
 497  */
 498 template <typename F>
 499 inline void addTask(F&& func) {
 500   return FiberManager::getFiberManager().addTask(std::forward<F>(func));
 501 }
 502
 503 /**
 504  * Add a new task. When the task is complete, execute finally(Try<Result>&&)
 505  * on the main context.
 506  * Task functor is run and destroyed on the fiber context.
 507  * Finally functor is run and destroyed on the main context.
 508  *
 509  * @param func Task functor; must have a signature of `T func()` for some T.
 510  * @param finally Finally functor; must have a signature of
 511  *                `void finally(Try<T>&&)` and will be passed
 512  *                the result of func() (including the exception if occurred).
 513  */
 514 template <typename F, typename G>
 515 inline void addTaskFinally(F&& func, G&& finally) {
 516   return FiberManager::getFiberManager().addTaskFinally(
 517       std::forward<F>(func), std::forward<G>(finally));
 518 }
 519
 520 /**
 521  * Blocks task execution until given promise is fulfilled.
 522  *
 523  * Calls function passing in a Promise<T>, which has to be fulfilled.
 524  *
 525  * @return data which was used to fulfill the promise.
 526  */
 527 template <typename F>
 528 typename FirstArgOf<F>::type::value_type inline await(F&& func);
 529
 530 /**
 531  * If called from a fiber, immediately switches to the FiberManager's context
 532  * and runs func(), going back to the Fiber's context after completion.
 533  * Outside a fiber, just calls func() directly.
 534  *
 535  * @return value returned by func().
 536  */
 537 template <typename F>
 538 typename std::result_of<F()>::type inline runInMainContext(F&& func) {
 539   auto fm = FiberManager::getFiberManagerUnsafe();
 540   if (UNLIKELY(fm == nullptr)) {
 541     return func();
 542   }
 543   return fm->runInMainContext(std::forward<F>(func));
 544 }
 545
 546 /**
 547  * Returns a refference to a fiber-local context for given Fiber. Should be
 548  * always called with the same T for each fiber. Fiber-local context is lazily
 549  * default-constructed on first request.
 550  * When new task is scheduled via addTask / addTaskRemote from a fiber its
 551  * fiber-local context is copied into the new fiber.
 552  */
 553 template <typename T>
 554 T& local() {
 555   auto fm = FiberManager::getFiberManagerUnsafe();
 556   if (fm) {
 557     return fm->local<T>();
 558   }
 559   return FiberManager::localThread<T>();
 560 }
 561
 562 inline void yield() {
 563   auto fm = FiberManager::getFiberManagerUnsafe();
 564   if (fm) {
 565     fm->yield();
 566   } else {
 567     std::this_thread::yield();
 568   }
 569 }
 570 }
 571 }
 572
 573 #include <folly/fibers/FiberManager-inl.h>