Merge pull request #913 from capnproto/fibers

Implement fibers

c++/src/kj/async-inl.h

@@ -194,6 +194,22 @@ public:
   // If this node wraps some other PromiseNode, get the wrapped node.  Used for debug tracing.
   // Default implementation returns nullptr.
 
+  template <typename T>
+  static Own<PromiseNode> from(T&& promise) {
+    // Given a Promise, extract the PromiseNode.
+    return kj::mv(promise.node);
+  }
+  template <typename T>
+  static PromiseNode& from(T& promise) {
+    // Given a Promise, extract the PromiseNode.
+    return *promise.node;
+  }
+  template <typename T>
+  static T to(Own<PromiseNode>&& node) {
+    // Construct a Promise from a PromiseNode. (T should be a Promise type.)
+    return T(false, kj::mv(node));
+  }
+
 protected:
   class OnReadyEvent {
     // Helper class for implementing onReady().
@@ -213,6 +229,13 @@ protected:
 
 // -------------------------------------------------------------------
 
+template <typename T>
+inline NeverDone::operator Promise<T>() const {
+  return PromiseNode::to<Promise<T>>(neverDone());
+}
+
+// -------------------------------------------------------------------
+
 class ImmediatePromiseNodeBase: public PromiseNode {
 public:
   ImmediatePromiseNodeBase();
@@ -557,7 +580,7 @@ public:
   ForkHub(Own<PromiseNode>&& inner): ForkHubBase(kj::mv(inner), result) {}
 
   Promise<_::UnfixVoid<T>> addBranch() {
-    return Promise<_::UnfixVoid<T>>(false, kj::heap<ForkBranch<T>>(addRef(*this)));
+    return _::PromiseNode::to<Promise<_::UnfixVoid<T>>>(kj::heap<ForkBranch<T>>(addRef(*this)));
   }
 
   _::SplitTuplePromise<T> split() {
@@ -574,8 +597,8 @@ private:
 
   template <size_t index>
   ReducePromises<typename SplitBranch<T, index>::Element> addSplit() {
-    return ReducePromises<typename SplitBranch<T, index>::Element>(
-        false, maybeChain(kj::heap<SplitBranch<T, index>>(addRef(*this)),
+    return _::PromiseNode::to<ReducePromises<typename SplitBranch<T, index>::Element>>(
+        maybeChain(kj::heap<SplitBranch<T, index>>(addRef(*this)),
                    implicitCast<typename SplitBranch<T, index>::Element*>(nullptr)));
   }
 };
@@ -848,6 +871,81 @@ private:
   }
 };
 
+// -------------------------------------------------------------------
+
+class FiberBase: public PromiseNode, private Event {
+  // Base class for the outer PromiseNode representing a fiber.
+
+public:
+  FiberBase(size_t stackSize, _::ExceptionOrValue& result);
+  ~FiberBase() noexcept(false);
+
+  void start() { armDepthFirst(); }
+  // Call immediately after construction to begin executing the fiber.
+
+  class WaitDoneEvent;
+
+  void onReady(_::Event* event) noexcept override;
+  PromiseNode* getInnerForTrace() override;
+
+protected:
+  bool isFinished() { return state == FINISHED; }
+
+private:
+  enum { WAITING, RUNNING, CANCELED, FINISHED } state;
+
+  size_t stackSize;
+
+#if _WIN32 || __CYGWIN__
+  void* osFiber;
+#else
+  struct Impl;
+  Impl& impl;
+#endif
+
+  _::PromiseNode* currentInner = nullptr;
+  OnReadyEvent onReadyEvent;
+  _::ExceptionOrValue& result;
+
+  void run();
+  virtual void runImpl(WaitScope& waitScope) = 0;
+
+  struct StartRoutine;
+
+  void switchToFiber();
+  void switchToMain();
+
+  Maybe<Own<Event>> fire() override;
+  // Implements Event. Each time the event is fired, switchToFiber() is called.
+
+  friend class WaitScope;
+  friend void _::waitImpl(Own<_::PromiseNode>&& node, _::ExceptionOrValue& result,
+                          WaitScope& waitScope);
+  friend bool _::pollImpl(_::PromiseNode& node, WaitScope& waitScope);
+};
+
+template <typename Func>
+class Fiber final: public FiberBase {
+public:
+  Fiber(size_t stackSize, Func&& func): FiberBase(stackSize, result), func(kj::fwd<Func>(func)) {}
+
+  typedef FixVoid<decltype(kj::instance<Func&>()(kj::instance<WaitScope&>()))> ResultType;
+
+  void get(ExceptionOrValue& output) noexcept override {
+    KJ_IREQUIRE(isFinished());
+    output.as<ResultType>() = kj::mv(result);
+  }
+
+private:
+  Func func;
+  ExceptionOr<ResultType> result;
+
+  void runImpl(WaitScope& waitScope) override {
+    result.template as<ResultType>() =
+        MaybeVoidCaller<WaitScope&, ResultType>::apply(func, waitScope);
+  }
+};
+
 }  // namespace _ (private)
 
 // =======================================================================================
@@ -868,7 +966,7 @@ PromiseForResult<Func, T> Promise<T>::then(Func&& func, ErrorFunc&& errorHandler
   Own<_::PromiseNode> intermediate =
       heap<_::TransformPromiseNode<ResultT, _::FixVoid<T>, Func, ErrorFunc>>(
           kj::mv(node), kj::fwd<Func>(func), kj::fwd<ErrorFunc>(errorHandler));
-  auto result = _::ChainPromises<_::ReturnType<Func, T>>(false,
+  auto result = _::PromiseNode::to<_::ChainPromises<_::ReturnType<Func, T>>>(
       _::maybeChain(kj::mv(intermediate), implicitCast<ResultT*>(nullptr)));
   return _::maybeReduce(kj::mv(result), false);
 }
@@ -991,6 +1089,17 @@ inline PromiseForResult<Func, void> evalNow(Func&& func) {
   return result;
 }
 
+template <typename Func>
+inline PromiseForResult<Func, WaitScope&> startFiber(size_t stackSize, Func&& func) {
+  typedef _::FixVoid<_::ReturnType<Func, WaitScope&>> ResultT;
+
+  Own<_::FiberBase> intermediate = kj::heap<_::Fiber<Func>>(stackSize, kj::fwd<Func>(func));
+  intermediate->start();
+  auto result = _::PromiseNode::to<_::ChainPromises<_::ReturnType<Func, WaitScope&>>>(
+      _::maybeChain(kj::mv(intermediate), implicitCast<ResultT*>(nullptr)));
+  return _::maybeReduce(kj::mv(result), false);
+}
+
 template <typename T>
 template <typename ErrorFunc>
 void Promise<T>::detach(ErrorFunc&& errorHandler) {
@@ -1005,8 +1114,8 @@ void Promise<void>::detach(ErrorFunc&& errorHandler) {
 
 template <typename T>
 Promise<Array<T>> joinPromises(Array<Promise<T>>&& promises) {
-  return Promise<Array<T>>(false, kj::heap<_::ArrayJoinPromiseNode<T>>(
-      KJ_MAP(p, promises) { return kj::mv(p.node); },
+  return _::PromiseNode::to<Promise<Array<T>>>(kj::heap<_::ArrayJoinPromiseNode<T>>(
+      KJ_MAP(p, promises) { return _::PromiseNode::from(kj::mv(p)); },
       heapArray<_::ExceptionOr<T>>(promises.size())));
 }
 
@@ -1134,7 +1243,7 @@ _::ReducePromises<T> newAdaptedPromise(Params&&... adapterConstructorParams) {
   Own<_::PromiseNode> intermediate(
       heap<_::AdapterPromiseNode<_::FixVoid<T>, Adapter>>(
           kj::fwd<Params>(adapterConstructorParams)...));
-  return _::ReducePromises<T>(false,
+  return _::PromiseNode::to<_::ReducePromises<T>>(
       _::maybeChain(kj::mv(intermediate), implicitCast<T*>(nullptr)));
 }
 
@@ -1144,7 +1253,7 @@ PromiseFulfillerPair<T> newPromiseAndFulfiller() {
 
   Own<_::PromiseNode> intermediate(
       heap<_::AdapterPromiseNode<_::FixVoid<T>, _::PromiseAndFulfillerAdapter<T>>>(*wrapper));
-  _::ReducePromises<T> promise(false,
+  auto promise = _::PromiseNode::to<_::ReducePromises<T>>(
       _::maybeChain(kj::mv(intermediate), implicitCast<T*>(nullptr)));
 
   return PromiseFulfillerPair<T> { kj::mv(promise), kj::mv(wrapper) };
@@ -1171,9 +1280,6 @@ protected:
   // Run the function. If the function returns a promise, returns the inner PromiseNode, otherwise
   // returns null.
 
-  template <typename T>
-  Own<PromiseNode> extractNode(Promise<T> promise) { return kj::mv(promise.node); }
-
   // implements PromiseNode ----------------------------------------------------
   void onReady(Event* event) noexcept override;
 
@@ -1271,7 +1377,7 @@ public:
   typedef _::FixVoid<_::UnwrapPromise<PromiseForResult<Func, void>>> ResultT;
 
   kj::Maybe<Own<_::PromiseNode>> execute() override {
-    auto result = extractNode(func());
+    auto result = _::PromiseNode::from(func());
     KJ_IREQUIRE(result.get() != nullptr);
     return kj::mv(result);
   }
@@ -1300,7 +1406,7 @@ template <typename Func>
 PromiseForResult<Func, void> Executor::executeAsync(Func&& func) const {
   auto event = kj::heap<_::XThreadEventImpl<Func>>(kj::fwd<Func>(func), *this);
   send(*event, false);
-  return PromiseForResult<Func, void>(false, kj::mv(event));
+  return _::PromiseNode::to<PromiseForResult<Func, void>>(kj::mv(event));
 }
 
 }  // namespace kj

c++/src/kj/async-prelude.h

@@ -188,6 +188,7 @@ class PromiseNode;
 class ChainPromiseNode;
 template <typename T>
 class ForkHub;
+class FiberBase;
 
 class Event;
 class XThreadEvent;
@@ -203,15 +204,9 @@ private:
   PromiseBase() = default;
   PromiseBase(Own<PromiseNode>&& node): node(kj::mv(node)) {}
 
-  friend class kj::EventLoop;
-  friend class ChainPromiseNode;
   template <typename>
   friend class kj::Promise;
-  friend class kj::TaskSet;
-  template <typename U>
-  friend Promise<Array<U>> kj::joinPromises(Array<Promise<U>>&& promises);
-  friend Promise<void> kj::joinPromises(Array<Promise<void>>&& promises);
-  friend class XThreadEvent;
+  friend class PromiseNode;
 };
 
 void detach(kj::Promise<void>&& promise);
@@ -224,9 +219,7 @@ Own<PromiseNode> neverDone();
 class NeverDone {
 public:
   template <typename T>
-  operator Promise<T>() const {
-    return Promise<T>(false, neverDone());
-  }
+  operator Promise<T>() const;
 
   KJ_NORETURN(void wait(WaitScope& waitScope) const);
 };

c++/src/kj/async-test.c++

@@ -844,5 +844,93 @@ KJ_TEST("exclusiveJoin both events complete simultaneously") {
   KJ_EXPECT(!joined.poll(waitScope));
 }
 
+KJ_TEST("start a fiber") {
+  EventLoop loop;
+  WaitScope waitScope(loop);
+
+  auto paf = newPromiseAndFulfiller<int>();
+
+  Promise<StringPtr> fiber = startFiber(65536,
+      [promise = kj::mv(paf.promise)](WaitScope& fiberScope) mutable {
+    int i = promise.wait(fiberScope);
+    KJ_EXPECT(i == 123);
+    return "foo"_kj;
+  });
+
+  KJ_EXPECT(!fiber.poll(waitScope));
+
+  paf.fulfiller->fulfill(123);
+
+  KJ_ASSERT(fiber.poll(waitScope));
+  KJ_EXPECT(fiber.wait(waitScope) == "foo");
+}
+
+KJ_TEST("fiber promise chaining") {
+  EventLoop loop;
+  WaitScope waitScope(loop);
+
+  auto paf = newPromiseAndFulfiller<int>();
+  bool ran = false;
+
+  Promise<int> fiber = startFiber(65536,
+      [promise = kj::mv(paf.promise), &ran](WaitScope& fiberScope) mutable {
+    ran = true;
+    return kj::mv(promise);
+  });
+
+  KJ_EXPECT(!ran);
+  KJ_EXPECT(!fiber.poll(waitScope));
+  KJ_EXPECT(ran);
+
+  paf.fulfiller->fulfill(123);
+
+  KJ_ASSERT(fiber.poll(waitScope));
+  KJ_EXPECT(fiber.wait(waitScope) == 123);
+}
+
+KJ_TEST("throw from a fiber") {
+  EventLoop loop;
+  WaitScope waitScope(loop);
+
+  auto paf = newPromiseAndFulfiller<void>();
+
+  Promise<void> fiber = startFiber(65536,
+      [promise = kj::mv(paf.promise)](WaitScope& fiberScope) mutable {
+    promise.wait(fiberScope);
+    KJ_FAIL_EXPECT("wait() should have thrown");
+  });
+
+  KJ_EXPECT(!fiber.poll(waitScope));
+
+  paf.fulfiller->reject(KJ_EXCEPTION(FAILED, "test exception"));
+
+  KJ_ASSERT(fiber.poll(waitScope));
+  KJ_EXPECT_THROW_MESSAGE("test exception", fiber.wait(waitScope));
+}
+
+KJ_TEST("cancel a fiber") {
+  EventLoop loop;
+  WaitScope waitScope(loop);
+
+  auto paf = newPromiseAndFulfiller<int>();
+
+  bool exited = false;
+
+  {
+    Promise<StringPtr> fiber = startFiber(65536,
+        [promise = kj::mv(paf.promise), &exited](WaitScope& fiberScope) mutable {
+      KJ_DEFER(exited = true);
+      int i = promise.wait(fiberScope);
+      KJ_EXPECT(i == 123);
+      return "foo"_kj;
+    });
+
+    KJ_EXPECT(!fiber.poll(waitScope));
+    KJ_EXPECT(!exited);
+  }
+
+  KJ_EXPECT(exited);
+}
+
 }  // namespace
 }  // namespace kj

c++/src/kj/async.h

@@ -229,8 +229,16 @@ public:
   //   around them in arbitrary ways.  Therefore, callers really need to know if a function they
   //   are calling might wait(), and the `WaitScope&` parameter makes this clear.
   //
-  // TODO(someday):  Implement fibers, and let them call wait() even when they are handling an
-  //   event.
+  // Usually, there is only one `WaitScope` for each `EventLoop`, and it can only be used at the
+  // top level of the thread owning the loop. Calling `wait()` with this `WaitScope` is what
+  // actually causes the event loop to run at all. This top-level `WaitScope` cannot be used
+  // recursively, so cannot be used within an event callback.
+  //
+  // However, it is possible to obtain a `WaitScope` in lower-level code by using fibers. Use
+  // kj::startFiber() to start some code executing on an alternate call stack. That code will get
+  // its own `WaitScope` allowing it to operate in a synchronous style. In this case, `wait()`
+  // switches back to the main stack in order to run the event loop, returning to the fiber's stack
+  // once the awaited promise resolves.
 
   bool poll(WaitScope& waitScope);
   // Returns true if a call to wait() would complete without blocking, false if it would block.
@@ -244,6 +252,8 @@ public:
   // The first poll() verifies that the promise doesn't resolve early, which would otherwise be
   // hard to do deterministically. The second poll() allows you to check that the promise has
   // resolved and avoid a wait() that might deadlock in the case that it hasn't.
+  //
+  // poll() is not supported in fibers; it will throw an exception.
 
   ForkedPromise<T> fork() KJ_WARN_UNUSED_RESULT;
   // Forks the promise, so that multiple different clients can independently wait on the result.
@@ -305,24 +315,7 @@ private:
   Promise(bool, Own<_::PromiseNode>&& node): PromiseBase(kj::mv(node)) {}
   // Second parameter prevent ambiguity with immediate-value constructor.
 
-  template <typename>
-  friend class Promise;
-  friend class EventLoop;
-  template <typename U, typename Adapter, typename... Params>
-  friend _::ReducePromises<U> newAdaptedPromise(Params&&... adapterConstructorParams);
-  template <typename U>
-  friend PromiseFulfillerPair<U> newPromiseAndFulfiller();
-  template <typename>
-  friend class _::ForkHub;
-  friend class TaskSet;
-  friend Promise<void> _::yield();
-  friend Promise<void> _::yieldHarder();
-  friend class _::NeverDone;
-  template <typename U>
-  friend Promise<Array<U>> joinPromises(Array<Promise<U>>&& promises);
-  friend Promise<void> joinPromises(Array<Promise<void>>&& promises);
-  friend class _::XThreadEvent;
-  friend class Executor;
+  friend class _::PromiseNode;
 };
 
 template <typename T>
@@ -397,6 +390,29 @@ PromiseForResult<Func, void> evalLast(Func&& func) KJ_WARN_UNUSED_RESULT;
 // callback enqueues new events, then latter callbacks will not execute until those events are
 // drained.
 
+template <typename Func>
+PromiseForResult<Func, WaitScope&> startFiber(size_t stackSize, Func&& func) KJ_WARN_UNUSED_RESULT;
+// Executes `func()` in a fiber, returning a promise for the eventual reseult. `func()` will be
+// passed a `WaitScope&` as its parameter, allowing it to call `.wait()` on promises. Thus, `func()`
+// can be written in a synchronous, blocking style, instead of using `.then()`. This is often much
+// easier to write and read, and may even be significantly faster if it allows the use of stack
+// allocation rather than heap allocation.
+//
+// However, fibers have a major disadvantage: memory must be allocated for the fiber's call stack.
+// The entire stack must be allocated at once, making it necessary to choose a stack size upfront
+// that is big enough for whatever the fiber needs to do. Estimating this is often difficult. That
+// said, over-estimating is not too terrible since pages of the stack will actually be allocated
+// lazily when first accessed; actual memory usage will correspond to the "high watermark" of the
+// actual stack usage. That said, this lazy allocation forces page faults, which can be quite slow.
+// Worse, freeing a stack forces a TLB flush and shootdown -- all currently-executing threads will
+// have to be interrupted to flush their CPU cores' TLB caches.
+//
+// In short, when performance matters, you should try to avoid creating fibers very frequently.
+//
+// TODO(perf): We should add a mechanism for freelisting stacks. However, this improves CPU usage
+//   at the expense of memory usage: stacks on the freelist will consume however many pages they
+//   used at their high watermark, forever.
+
 template <typename T>
 Promise<Array<T>> joinPromises(Array<Promise<T>>&& promises);
 // Join an array of promises into a promise for an array.
@@ -891,6 +907,10 @@ private:
 
   Own<TaskSet> daemons;
 
+#if _WIN32 || __CYGWIN__
+  void* mainFiber = nullptr;
+#endif
+
   bool turn();
   void setRunnable(bool runnable);
   void enterScope();
@@ -907,6 +927,7 @@ private:
   friend class WaitScope;
   friend class Executor;
   friend class _::XThreadEvent;
+  friend class _::FiberBase;
 };
 
 class WaitScope {
@@ -920,20 +941,31 @@ class WaitScope {
 
 public:
   inline explicit WaitScope(EventLoop& loop): loop(loop) { loop.enterScope(); }
-  inline ~WaitScope() { loop.leaveScope(); }
+  inline ~WaitScope() { if (fiber == nullptr) loop.leaveScope(); }
   KJ_DISALLOW_COPY(WaitScope);
 
   void poll();
   // Pumps the event queue and polls for I/O until there's nothing left to do (without blocking).
+  //
+  // Not supported in fibers.
 
   void setBusyPollInterval(uint count) { busyPollInterval = count; }
   // Set the maximum number of events to run in a row before calling poll() on the EventPort to
   // check for new I/O.
+  //
+  // This has no effect when used in a fiber.
 
 private:
   EventLoop& loop;
   uint busyPollInterval = kj::maxValue;
+
+  kj::Maybe<_::FiberBase&> fiber;
+
+  explicit WaitScope(EventLoop& loop, _::FiberBase& fiber)
+      : loop(loop), fiber(fiber) {}
+
   friend class EventLoop;
+  friend class _::FiberBase;
   friend void _::waitImpl(Own<_::PromiseNode>&& node, _::ExceptionOrValue& result,
                           WaitScope& waitScope);
   friend bool _::pollImpl(_::PromiseNode& node, WaitScope& waitScope);