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