folly/experimental/fibers/FiberManager.h

   1 /*
   2  * Copyright 2015 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 <typeindex>
  23 #include <unordered_set>
  24 #include <vector>
  25
  26 #include <folly/AtomicLinkedList.h>
  27 #include <folly/Likely.h>
  28 #include <folly/IntrusiveList.h>
  29 #include <folly/futures/Try.h>
  30
  31 #include <folly/experimental/fibers/BoostContextCompatibility.h>
  32 #include <folly/experimental/fibers/Fiber.h>
  33 #include <folly/experimental/fibers/traits.h>
  34
  35 #ifdef USE_GUARD_ALLOCATOR
  36 #include <folly/experimental/fibers/GuardPageAllocator.h>
  37 #endif
  38
  39 namespace folly { namespace fibers {
  40
  41 class Baton;
  42 class Fiber;
  43 class LoopController;
  44 class TimeoutController;
  45
  46 template <typename T>
  47 class LocalType {
  48 };
  49
  50 /**
  51  * @class FiberManager
  52  * @brief Single-threaded task execution engine.
  53  *
  54  * FiberManager allows semi-parallel task execution on the same thread. Each
  55  * task can notify FiberManager that it is blocked on something (via await())
  56  * call. This will pause execution of this task and it will be resumed only
  57  * when it is unblocked (via setData()).
  58  */
  59 class FiberManager {
  60  public:
  61   struct Options {
  62 #ifdef FOLLY_SANITIZE_ADDRESS
  63     /* ASAN needs a lot of extra stack space.
  64        16x is a conservative estimate, 8x also worked with tests
  65        where it mattered.  Note that overallocating here does not necessarily
  66        increase RSS, since unused memory is pretty much free. */
  67     static constexpr size_t kDefaultStackSize{16 * 16 * 1024};
  68 #else
  69     static constexpr size_t kDefaultStackSize{16 * 1024};
  70 #endif
  71     /**
  72      * Maximum stack size for fibers which will be used for executing all the
  73      * tasks.
  74      */
  75     size_t stackSize{kDefaultStackSize};
  76
  77     /**
  78      * Record exact amount of stack used.
  79      *
  80      * This is fairly expensive: we fill each newly allocated stack
  81      * with some known value and find the boundary of unused stack
  82      * with linear search every time we surrender the stack back to fibersPool.
  83      * 0 disables stack recording.
  84      */
  85     size_t recordStackEvery{0};
  86
  87     /**
  88      * Keep at most this many free fibers in the pool.
  89      * This way the total number of fibers in the system is always bounded
  90      * by the number of active fibers + maxFibersPoolSize.
  91      */
  92     size_t maxFibersPoolSize{1000};
  93
  94     constexpr Options() {}
  95   };
  96
  97   typedef std::function<void(std::exception_ptr, std::string)>
  98   ExceptionCallback;
  99
 100   FiberManager(const FiberManager&) = delete;
 101   FiberManager& operator=(const FiberManager&) = delete;
 102
 103   /**
 104    * Initializes, but doesn't start FiberManager loop
 105    *
 106    * @param loopController
 107    * @param options FiberManager options
 108    */
 109   explicit FiberManager(std::unique_ptr<LoopController> loopController,
 110                         Options options = Options());
 111
 112   /**
 113    * Initializes, but doesn't start FiberManager loop
 114    *
 115    * @param loopController
 116    * @param options FiberManager options
 117    * @tparam LocalT only local of this type may be stored on fibers.
 118    *                Locals of other types will be considered thread-locals.
 119    */
 120   template <typename LocalT>
 121   FiberManager(LocalType<LocalT>,
 122                std::unique_ptr<LoopController> loopController,
 123                Options options = Options());
 124
 125
 126   ~FiberManager();
 127
 128   /**
 129    * Controller access.
 130    */
 131   LoopController& loopController();
 132   const LoopController& loopController() const;
 133
 134   /**
 135    * Keeps running ready tasks until the list of ready tasks is empty.
 136    *
 137    * @return True if there are any waiting tasks remaining.
 138    */
 139   bool loopUntilNoReady();
 140
 141   /**
 142    * @return true if there are outstanding tasks.
 143    */
 144   bool hasTasks() const;
 145
 146   /**
 147    * Sets exception callback which will be called if any of the tasks throws an
 148    * exception.
 149    *
 150    * @param ec
 151    */
 152   void setExceptionCallback(ExceptionCallback ec);
 153
 154   /**
 155    * Add a new task to be executed. Must be called from FiberManager's thread.
 156    *
 157    * @param func Task functor; must have a signature of `void func()`.
 158    *             The object will be destroyed once task execution is complete.
 159    */
 160   template <typename F>
 161   void addTask(F&& func);
 162
 163   /**
 164    * Add a new task to be executed. Safe to call from other threads.
 165    *
 166    * @param func Task function; must have a signature of `void func()`.
 167    *             The object will be destroyed once task execution is complete.
 168    */
 169   template <typename F>
 170   void addTaskRemote(F&& func);
 171
 172   /**
 173    * Add a new task. When the task is complete, execute finally(Try<Result>&&)
 174    * on the main context.
 175    *
 176    * @param func Task functor; must have a signature of `T func()` for some T.
 177    * @param finally Finally functor; must have a signature of
 178    *                `void finally(Try<T>&&)` and will be passed
 179    *                the result of func() (including the exception if occurred).
 180    */
 181   template <typename F, typename G>
 182   void addTaskFinally(F&& func, G&& finally);
 183
 184   /**
 185    * If called from a fiber, immediately switches to the FiberManager's context
 186    * and runs func(), going back to the Fiber's context after completion.
 187    * Outside a fiber, just calls func() directly.
 188    *
 189    * @return value returned by func().
 190    */
 191   template <typename F>
 192   typename std::result_of<F()>::type
 193   runInMainContext(F&& func);
 194
 195   /**
 196    * Returns a refference to a fiber-local context for given Fiber. Should be
 197    * always called with the same T for each fiber. Fiber-local context is lazily
 198    * default-constructed on first request.
 199    * When new task is scheduled via addTask / addTaskRemote from a fiber its
 200    * fiber-local context is copied into the new fiber.
 201    */
 202   template <typename T>
 203   T& local();
 204
 205   template <typename T>
 206   static T& localThread();
 207
 208   /**
 209    * @return How many fiber objects (and stacks) has this manager allocated.
 210    */
 211   size_t fibersAllocated() const;
 212
 213   /**
 214    * @return How many of the allocated fiber objects are currently
 215    * in the free pool.
 216    */
 217   size_t fibersPoolSize() const;
 218
 219   /**
 220    * return     true if running activeFiber_ is not nullptr.
 221    */
 222   bool hasActiveFiber() const;
 223
 224   /**
 225    * @return What was the most observed fiber stack usage (in bytes).
 226    */
 227   size_t stackHighWatermark() const;
 228
 229   /**
 230    * Yield execution of the currently running fiber. Must only be called from a
 231    * fiber executing on this FiberManager. The calling fiber will be scheduled
 232    * when all other fibers have had a chance to run and the event loop is
 233    * serviced.
 234    */
 235   void yield();
 236
 237   static FiberManager& getFiberManager();
 238   static FiberManager* getFiberManagerUnsafe();
 239
 240  private:
 241   friend class Baton;
 242   friend class Fiber;
 243   template <typename F>
 244   struct AddTaskHelper;
 245   template <typename F, typename G>
 246   struct AddTaskFinallyHelper;
 247
 248   struct RemoteTask {
 249     template <typename F>
 250     explicit RemoteTask(F&& f) : func(std::forward<F>(f)) {}
 251     template <typename F>
 252     RemoteTask(F&& f, const Fiber::LocalData& localData_) :
 253         func(std::forward<F>(f)),
 254         localData(folly::make_unique<Fiber::LocalData>(localData_)) {}
 255     std::function<void()> func;
 256     std::unique_ptr<Fiber::LocalData> localData;
 257     AtomicLinkedListHook<RemoteTask> nextRemoteTask;
 258   };
 259
 260   typedef folly::IntrusiveList<Fiber, &Fiber::listHook_> FiberTailQueue;
 261
 262   Fiber* activeFiber_{nullptr}; /**< active fiber, nullptr on main context */
 263   /**
 264    * Same as active fiber, but also set for functions run from fiber on main
 265    * context.
 266    */
 267   Fiber* currentFiber_{nullptr};
 268
 269   FiberTailQueue readyFibers_;  /**< queue of fibers ready to be executed */
 270   FiberTailQueue yieldedFibers_;  /**< queue of fibers which have yielded
 271                                        execution */
 272   FiberTailQueue fibersPool_;   /**< pool of unitialized Fiber objects */
 273
 274   size_t fibersAllocated_{0};   /**< total number of fibers allocated */
 275   size_t fibersPoolSize_{0};    /**< total number of fibers in the free pool */
 276   size_t fibersActive_{0};      /**< number of running or blocked fibers */
 277   size_t fiberId_{0};           /**< id of last fiber used */
 278
 279   FContext::ContextStruct mainContext_;  /**< stores loop function context */
 280
 281   std::unique_ptr<LoopController> loopController_;
 282   bool isLoopScheduled_{false}; /**< was the ready loop scheduled to run? */
 283
 284   /**
 285    * When we are inside FiberManager loop this points to FiberManager. Otherwise
 286    * it's nullptr
 287    */
 288   static __thread FiberManager* currentFiberManager_;
 289
 290   /**
 291    * runInMainContext implementation for non-void functions.
 292    */
 293   template <typename F>
 294   typename std::enable_if<
 295     !std::is_same<typename std::result_of<F()>::type, void>::value,
 296     typename std::result_of<F()>::type>::type
 297   runInMainContextHelper(F&& func);
 298
 299   /**
 300    * runInMainContext implementation for void functions
 301    */
 302   template <typename F>
 303   typename std::enable_if<
 304     std::is_same<typename std::result_of<F()>::type, void>::value,
 305     void>::type
 306   runInMainContextHelper(F&& func);
 307
 308   /**
 309    * Allocator used to allocate stack for Fibers in the pool.
 310    * Allocates stack on the stack of the main context.
 311    */
 312 #ifdef USE_GUARD_ALLOCATOR
 313   /* This is too slow for production use; can be fixed
 314      if we allocated all stack storage once upfront */
 315   GuardPageAllocator stackAllocator_;
 316 #else
 317   std::allocator<unsigned char> stackAllocator_;
 318 #endif
 319
 320   const Options options_;       /**< FiberManager options */
 321
 322   /**
 323    * Largest observed individual Fiber stack usage in bytes.
 324    */
 325   size_t stackHighWatermark_{0};
 326
 327   /**
 328    * Schedules a loop with loopController (unless already scheduled before).
 329    */
 330   void ensureLoopScheduled();
 331
 332   /**
 333    * @return An initialized Fiber object from the pool
 334    */
 335   Fiber* getFiber();
 336
 337   /**
 338    * Sets local data for given fiber if all conditions are met.
 339    */
 340   void initLocalData(Fiber& fiber);
 341
 342   /**
 343    * Function passed to the await call.
 344    */
 345   std::function<void(Fiber&)> awaitFunc_;
 346
 347   /**
 348    * Function passed to the runInMainContext call.
 349    */
 350   std::function<void()> immediateFunc_;
 351
 352   ExceptionCallback exceptionCallback_; /**< task exception callback */
 353
 354   folly::AtomicLinkedList<Fiber, &Fiber::nextRemoteReady_> remoteReadyQueue_;
 355
 356   folly::AtomicLinkedList<RemoteTask, &RemoteTask::nextRemoteTask>
 357       remoteTaskQueue_;
 358
 359   std::shared_ptr<TimeoutController> timeoutManager_;
 360
 361   /**
 362    * Only local of this type will be available for fibers.
 363    */
 364   std::type_index localType_;
 365
 366   void runReadyFiber(Fiber* fiber);
 367   void remoteReadyInsert(Fiber* fiber);
 368 };
 369
 370 /**
 371  * @return      true iff we are running in a fiber's context
 372  */
 373 inline bool onFiber() {
 374   auto fm = FiberManager::getFiberManagerUnsafe();
 375   return fm ? fm->hasActiveFiber() : false;
 376 }
 377
 378 /**
 379  * Add a new task to be executed.
 380  *
 381  * @param func Task functor; must have a signature of `void func()`.
 382  *             The object will be destroyed once task execution is complete.
 383  */
 384 template <typename F>
 385 inline void addTask(F&& func) {
 386   return FiberManager::getFiberManager().addTask(std::forward<F>(func));
 387 }
 388
 389 /**
 390  * Add a new task. When the task is complete, execute finally(Try<Result>&&)
 391  * on the main context.
 392  * Task functor is run and destroyed on the fiber context.
 393  * Finally functor is run and destroyed on the main context.
 394  *
 395  * @param func Task functor; must have a signature of `T func()` for some T.
 396  * @param finally Finally functor; must have a signature of
 397  *                `void finally(Try<T>&&)` and will be passed
 398  *                the result of func() (including the exception if occurred).
 399  */
 400 template <typename F, typename G>
 401 inline void addTaskFinally(F&& func, G&& finally) {
 402   return FiberManager::getFiberManager().addTaskFinally(
 403     std::forward<F>(func), std::forward<G>(finally));
 404 }
 405
 406 /**
 407  * Blocks task execution until given promise is fulfilled.
 408  *
 409  * Calls function passing in a Promise<T>, which has to be fulfilled.
 410  *
 411  * @return data which was used to fulfill the promise.
 412  */
 413 template <typename F>
 414 typename FirstArgOf<F>::type::value_type
 415 inline await(F&& func);
 416
 417 /**
 418  * If called from a fiber, immediately switches to the FiberManager's context
 419  * and runs func(), going back to the Fiber's context after completion.
 420  * Outside a fiber, just calls func() directly.
 421  *
 422  * @return value returned by func().
 423  */
 424 template <typename F>
 425 typename std::result_of<F()>::type
 426 inline runInMainContext(F&& func) {
 427   auto fm = FiberManager::getFiberManagerUnsafe();
 428   if (UNLIKELY(fm == nullptr)) {
 429     return func();
 430   }
 431   return fm->runInMainContext(std::forward<F>(func));
 432 }
 433
 434 /**
 435  * Returns a refference to a fiber-local context for given Fiber. Should be
 436  * always called with the same T for each fiber. Fiber-local context is lazily
 437  * default-constructed on first request.
 438  * When new task is scheduled via addTask / addTaskRemote from a fiber its
 439  * fiber-local context is copied into the new fiber.
 440  */
 441 template <typename T>
 442 T& local() {
 443   auto fm = FiberManager::getFiberManagerUnsafe();
 444   if (fm) {
 445     return fm->local<T>();
 446   }
 447   return FiberManager::localThread<T>();
 448 }
 449
 450 inline void yield() {
 451   auto fm = FiberManager::getFiberManagerUnsafe();
 452   if (fm) {
 453     fm->yield();
 454   } else {
 455     std::this_thread::yield();
 456   }
 457 }
 458
 459 }}
 460
 461 #include "FiberManager-inl.h"