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