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