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