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{true};
  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   static FiberManager& getFiberManager();
 256   static FiberManager* getFiberManagerUnsafe();
 257
 258  private:
 259   friend class Baton;
 260   friend class Fiber;
 261   template <typename F>
 262   struct AddTaskHelper;
 263   template <typename F, typename G>
 264   struct AddTaskFinallyHelper;
 265
 266   struct RemoteTask {
 267     template <typename F>
 268     explicit RemoteTask(F&& f) :
 269         func(std::forward<F>(f)),
 270         rcontext(RequestContext::saveContext()) {}
 271     template <typename F>
 272     RemoteTask(F&& f, const Fiber::LocalData& localData_) :
 273         func(std::forward<F>(f)),
 274         localData(folly::make_unique<Fiber::LocalData>(localData_)),
 275         rcontext(RequestContext::saveContext()) {}
 276     std::function<void()> func;
 277     std::unique_ptr<Fiber::LocalData> localData;
 278     std::shared_ptr<RequestContext> rcontext;
 279     AtomicLinkedListHook<RemoteTask> nextRemoteTask;
 280   };
 281
 282   typedef folly::IntrusiveList<Fiber, &Fiber::listHook_> FiberTailQueue;
 283
 284   Fiber* activeFiber_{nullptr}; /**< active fiber, nullptr on main context */
 285   /**
 286    * Same as active fiber, but also set for functions run from fiber on main
 287    * context.
 288    */
 289   Fiber* currentFiber_{nullptr};
 290
 291   FiberTailQueue readyFibers_;  /**< queue of fibers ready to be executed */
 292   FiberTailQueue yieldedFibers_;  /**< queue of fibers which have yielded
 293                                        execution */
 294   FiberTailQueue fibersPool_;   /**< pool of unitialized Fiber objects */
 295
 296   size_t fibersAllocated_{0};   /**< total number of fibers allocated */
 297   size_t fibersPoolSize_{0};    /**< total number of fibers in the free pool */
 298   size_t fibersActive_{0};      /**< number of running or blocked fibers */
 299   size_t fiberId_{0};           /**< id of last fiber used */
 300
 301   FContext::ContextStruct mainContext_;  /**< stores loop function context */
 302
 303   std::unique_ptr<LoopController> loopController_;
 304   bool isLoopScheduled_{false}; /**< was the ready loop scheduled to run? */
 305
 306   /**
 307    * When we are inside FiberManager loop this points to FiberManager. Otherwise
 308    * it's nullptr
 309    */
 310   static __thread FiberManager* currentFiberManager_;
 311
 312   /**
 313    * runInMainContext implementation for non-void functions.
 314    */
 315   template <typename F>
 316   typename std::enable_if<
 317     !std::is_same<typename std::result_of<F()>::type, void>::value,
 318     typename std::result_of<F()>::type>::type
 319   runInMainContextHelper(F&& func);
 320
 321   /**
 322    * runInMainContext implementation for void functions
 323    */
 324   template <typename F>
 325   typename std::enable_if<
 326     std::is_same<typename std::result_of<F()>::type, void>::value,
 327     void>::type
 328   runInMainContextHelper(F&& func);
 329
 330   /**
 331    * Allocator used to allocate stack for Fibers in the pool.
 332    * Allocates stack on the stack of the main context.
 333    */
 334   GuardPageAllocator stackAllocator_;
 335
 336   const Options options_;       /**< FiberManager options */
 337
 338   /**
 339    * Largest observed individual Fiber stack usage in bytes.
 340    */
 341   size_t stackHighWatermark_{0};
 342
 343   /**
 344    * Schedules a loop with loopController (unless already scheduled before).
 345    */
 346   void ensureLoopScheduled();
 347
 348   /**
 349    * @return An initialized Fiber object from the pool
 350    */
 351   Fiber* getFiber();
 352
 353   /**
 354    * Sets local data for given fiber if all conditions are met.
 355    */
 356   void initLocalData(Fiber& fiber);
 357
 358   /**
 359    * Function passed to the await call.
 360    */
 361   std::function<void(Fiber&)> awaitFunc_;
 362
 363   /**
 364    * Function passed to the runInMainContext call.
 365    */
 366   std::function<void()> immediateFunc_;
 367
 368   /**
 369    * Fiber's execution observer.
 370    */
 371   ExecutionObserver* observer_{nullptr};
 372
 373   ExceptionCallback exceptionCallback_; /**< task exception callback */
 374
 375   folly::AtomicLinkedList<Fiber, &Fiber::nextRemoteReady_> remoteReadyQueue_;
 376
 377   folly::AtomicLinkedList<RemoteTask, &RemoteTask::nextRemoteTask>
 378       remoteTaskQueue_;
 379
 380   std::shared_ptr<TimeoutController> timeoutManager_;
 381
 382   /**
 383    * Only local of this type will be available for fibers.
 384    */
 385   std::type_index localType_;
 386
 387   void runReadyFiber(Fiber* fiber);
 388   void remoteReadyInsert(Fiber* fiber);
 389 };
 390
 391 /**
 392  * @return      true iff we are running in a fiber's context
 393  */
 394 inline bool onFiber() {
 395   auto fm = FiberManager::getFiberManagerUnsafe();
 396   return fm ? fm->hasActiveFiber() : false;
 397 }
 398
 399 /**
 400  * Add a new task to be executed.
 401  *
 402  * @param func Task functor; must have a signature of `void func()`.
 403  *             The object will be destroyed once task execution is complete.
 404  */
 405 template <typename F>
 406 inline void addTask(F&& func) {
 407   return FiberManager::getFiberManager().addTask(std::forward<F>(func));
 408 }
 409
 410 /**
 411  * Add a new task. When the task is complete, execute finally(Try<Result>&&)
 412  * on the main context.
 413  * Task functor is run and destroyed on the fiber context.
 414  * Finally functor is run and destroyed on the main context.
 415  *
 416  * @param func Task functor; must have a signature of `T func()` for some T.
 417  * @param finally Finally functor; must have a signature of
 418  *                `void finally(Try<T>&&)` and will be passed
 419  *                the result of func() (including the exception if occurred).
 420  */
 421 template <typename F, typename G>
 422 inline void addTaskFinally(F&& func, G&& finally) {
 423   return FiberManager::getFiberManager().addTaskFinally(
 424     std::forward<F>(func), std::forward<G>(finally));
 425 }
 426
 427 /**
 428  * Blocks task execution until given promise is fulfilled.
 429  *
 430  * Calls function passing in a Promise<T>, which has to be fulfilled.
 431  *
 432  * @return data which was used to fulfill the promise.
 433  */
 434 template <typename F>
 435 typename FirstArgOf<F>::type::value_type
 436 inline await(F&& func);
 437
 438 /**
 439  * If called from a fiber, immediately switches to the FiberManager's context
 440  * and runs func(), going back to the Fiber's context after completion.
 441  * Outside a fiber, just calls func() directly.
 442  *
 443  * @return value returned by func().
 444  */
 445 template <typename F>
 446 typename std::result_of<F()>::type
 447 inline runInMainContext(F&& func) {
 448   auto fm = FiberManager::getFiberManagerUnsafe();
 449   if (UNLIKELY(fm == nullptr)) {
 450     return func();
 451   }
 452   return fm->runInMainContext(std::forward<F>(func));
 453 }
 454
 455 /**
 456  * Returns a refference to a fiber-local context for given Fiber. Should be
 457  * always called with the same T for each fiber. Fiber-local context is lazily
 458  * default-constructed on first request.
 459  * When new task is scheduled via addTask / addTaskRemote from a fiber its
 460  * fiber-local context is copied into the new fiber.
 461  */
 462 template <typename T>
 463 T& local() {
 464   auto fm = FiberManager::getFiberManagerUnsafe();
 465   if (fm) {
 466     return fm->local<T>();
 467   }
 468   return FiberManager::localThread<T>();
 469 }
 470
 471 inline void yield() {
 472   auto fm = FiberManager::getFiberManagerUnsafe();
 473   if (fm) {
 474     fm->yield();
 475   } else {
 476     std::this_thread::yield();
 477   }
 478 }
 479
 480 }}
 481
 482 #include "FiberManager-inl.h"