Make EventLoop current when constructed, so that you can use the Promise methods…

Make EventLoop current when constructed, so that you can use the Promise methods even when the loop hasn't actually started yet.

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

@@ -29,17 +29,8 @@
 namespace kj {
 namespace {
 
-class DummyErrorHandler: public TaskSet::ErrorHandler {
-public:
-  void taskFailed(kj::Exception&& exception) override {
-    kj::throwRecoverableException(kj::mv(exception));
-  }
-};
-
 TEST(AsyncIo, SimpleNetwork) {
   UnixEventLoop loop;
-  DummyErrorHandler dummyHandler;
-  TaskSet tasks(loop, dummyHandler);
   auto network = Network::newSystemNetwork();
 
   Own<ConnectionReceiver> listener;
@@ -50,21 +41,16 @@ TEST(AsyncIo, SimpleNetwork) {
 
   auto port = newPromiseAndFulfiller<uint>();
 
-  tasks.add(loop.evalLater([&]() {
-    return port.promise
-        .then([&](uint portnum) {
+  loop.daemonize(port.promise.then([&](uint portnum) {
     return network->parseRemoteAddress("127.0.0.1", portnum);
   }).then([&](Own<RemoteAddress>&& result) {
     return result->connect();
   }).then([&](Own<AsyncIoStream>&& result) {
     client = kj::mv(result);
     return client->write("foo", 3);
-        });
   }));
 
-  kj::String result = loop.wait(loop.evalLater([&]() {
-    return network->parseLocalAddress("*")
-        .then([&](Own<LocalAddress>&& result) {
+  kj::String result = network->parseLocalAddress("*").then([&](Own<LocalAddress>&& result) {
     listener = result->listen();
     port.fulfiller->fulfill(listener->getPort());
     return listener->accept();
@@ -74,22 +60,17 @@ TEST(AsyncIo, SimpleNetwork) {
   }).then([&](size_t n) {
     EXPECT_EQ(3u, n);
     return heapString(receiveBuffer, n);
-        });
-  }));
+  }).wait();
 
   EXPECT_EQ("foo", result);
 }
 
 String tryParseLocal(EventLoop& loop, Network& network, StringPtr text, uint portHint = 0) {
-  return loop.wait(loop.evalLater([&]() {
-    return network.parseLocalAddress(text, portHint);
-  }))->toString();
+  return network.parseLocalAddress(text, portHint).wait()->toString();
 }
 
 String tryParseRemote(EventLoop& loop, Network& network, StringPtr text, uint portHint = 0) {
-  return loop.wait(loop.evalLater([&]() {
-    return network.parseRemoteAddress(text, portHint);
-  }))->toString();
+  return network.parseRemoteAddress(text, portHint).wait()->toString();
 }
 
 TEST(AsyncIo, AddressParsing) {
@@ -110,56 +91,42 @@ TEST(AsyncIo, AddressParsing) {
 
 TEST(AsyncIo, OneWayPipe) {
   UnixEventLoop loop;
-  DummyErrorHandler dummyHandler;
-  TaskSet tasks(loop, dummyHandler);
 
   auto pipe = newOneWayPipe();
   char receiveBuffer[4];
 
-  tasks.add(loop.evalLater([&]() {
-    return pipe.out->write("foo", 3);
-  }));
+  loop.daemonize(pipe.out->write("foo", 3));
 
-  kj::String result = loop.wait(loop.evalLater([&]() {
-    return pipe.in->tryRead(receiveBuffer, 3, 4)
-        .then([&](size_t n) {
+  kj::String result = pipe.in->tryRead(receiveBuffer, 3, 4).then([&](size_t n) {
     EXPECT_EQ(3u, n);
     return heapString(receiveBuffer, n);
-        });
-  }));
+  }).wait();
 
   EXPECT_EQ("foo", result);
 }
 
 TEST(AsyncIo, TwoWayPipe) {
   UnixEventLoop loop;
-  DummyErrorHandler dummyHandler;
 
   auto pipe = newTwoWayPipe();
   char receiveBuffer1[4];
   char receiveBuffer2[4];
 
-  auto promise = loop.evalLater([&]() {
-    return pipe.ends[0]->write("foo", 3)
-        .then([&]() {
+  auto promise = pipe.ends[0]->write("foo", 3).then([&]() {
     return pipe.ends[0]->tryRead(receiveBuffer1, 3, 4);
   }).then([&](size_t n) {
     EXPECT_EQ(3u, n);
     return heapString(receiveBuffer1, n);
   });
-  });
 
-  kj::String result = loop.wait(loop.evalLater([&]() {
-    return pipe.ends[1]->write("bar", 3)
-        .then([&]() {
+  kj::String result = pipe.ends[1]->write("bar", 3).then([&]() {
     return pipe.ends[1]->tryRead(receiveBuffer2, 3, 4);
   }).then([&](size_t n) {
     EXPECT_EQ(3u, n);
     return heapString(receiveBuffer2, n);
-        });
-  }));
+  }).wait();
 
-  kj::String result2 = loop.wait(kj::mv(promise));
+  kj::String result2 = promise.wait();
 
   EXPECT_EQ("foo", result);
   EXPECT_EQ("bar", result2);

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

@@ -29,6 +29,7 @@
 #include <sys/types.h>
 #include <sys/stat.h>
 #include <gtest/gtest.h>
+#include <pthread.h>
 
 namespace kj {
 
@@ -41,13 +42,6 @@ inline void delay() { usleep(10000); }
 #define EXPECT_SI_CODE(a,b)
 #endif
 
-class DummyErrorHandler: public TaskSet::ErrorHandler {
-public:
-  void taskFailed(kj::Exception&& exception) override {
-    kj::throwRecoverableException(kj::mv(exception));
-  }
-};
-
 class AsyncUnixTest: public testing::Test {
 public:
   static void SetUpTestCase() {
@@ -61,7 +55,7 @@ TEST_F(AsyncUnixTest, Signals) {
 
   kill(getpid(), SIGUSR2);
 
-  siginfo_t info = loop.wait(loop.onSignal(SIGUSR2));
+  siginfo_t info = loop.onSignal(SIGUSR2).wait();
   EXPECT_EQ(SIGUSR2, info.si_signo);
   EXPECT_SI_CODE(SI_USER, info.si_code);
 }
@@ -79,7 +73,7 @@ TEST_F(AsyncUnixTest, SignalWithValue) {
   value.sival_int = 123;
   sigqueue(getpid(), SIGUSR2, value);
 
-  siginfo_t info = loop.wait(loop.onSignal(SIGUSR2));
+  siginfo_t info = loop.onSignal(SIGUSR2).wait();
   EXPECT_EQ(SIGUSR2, info.si_signo);
   EXPECT_SI_CODE(SI_QUEUE, info.si_code);
   EXPECT_EQ(123, info.si_value.sival_int);
@@ -88,66 +82,48 @@ TEST_F(AsyncUnixTest, SignalWithValue) {
 
 TEST_F(AsyncUnixTest, SignalsMultiListen) {
   UnixEventLoop loop;
-  DummyErrorHandler dummyHandler;
 
-  TaskSet tasks(loop, dummyHandler);
-  tasks.add(loop.onSignal(SIGIO).thenInAnyThread([](siginfo_t&&) {
+  loop.daemonize(loop.onSignal(SIGIO).then([](siginfo_t&&) {
     ADD_FAILURE() << "Received wrong signal.";
   }));
 
   kill(getpid(), SIGUSR2);
 
-  siginfo_t info = loop.wait(loop.onSignal(SIGUSR2));
+  siginfo_t info = loop.onSignal(SIGUSR2).wait();
   EXPECT_EQ(SIGUSR2, info.si_signo);
   EXPECT_SI_CODE(SI_USER, info.si_code);
 }
 
 TEST_F(AsyncUnixTest, SignalsMultiReceive) {
   UnixEventLoop loop;
-  DummyErrorHandler dummyHandler;
 
   kill(getpid(), SIGUSR2);
   kill(getpid(), SIGIO);
 
-  siginfo_t info = loop.wait(loop.onSignal(SIGUSR2));
+  siginfo_t info = loop.onSignal(SIGUSR2).wait();
   EXPECT_EQ(SIGUSR2, info.si_signo);
   EXPECT_SI_CODE(SI_USER, info.si_code);
 
-  info = loop.wait(loop.onSignal(SIGIO));
+  info = loop.onSignal(SIGIO).wait();
   EXPECT_EQ(SIGIO, info.si_signo);
   EXPECT_SI_CODE(SI_USER, info.si_code);
 }
 
 TEST_F(AsyncUnixTest, SignalsAsync) {
-  // Arrange for another thread to wait on a UnixEventLoop...
-  auto exitThread = newPromiseAndFulfiller<void>();
-  UnixEventLoop unixLoop;
+  UnixEventLoop loop;
+
+  // Arrange for a signal to be sent from another thread.
+  pthread_t mainThread = pthread_self();
   Thread thread([&]() {
-    unixLoop.wait(kj::mv(exitThread.promise));
+    delay();
+    pthread_kill(mainThread, SIGUSR2);
   });
-  KJ_DEFER(exitThread.fulfiller->fulfill());
 
-  // Arrange to catch a signal in the other thread.  But we haven't sent one yet.
-  bool received = false;
-  Promise<void> promise = unixLoop.there(unixLoop.onSignal(SIGUSR2),
-      [&](siginfo_t&& info) {
-    received = true;
+  siginfo_t info = loop.onSignal(SIGUSR2).wait();
   EXPECT_EQ(SIGUSR2, info.si_signo);
 #if __linux__
   EXPECT_SI_CODE(SI_TKILL, info.si_code);
 #endif
-  });
-
-  delay();
-
-  EXPECT_FALSE(received);
-
-  thread.sendSignal(SIGUSR2);
-
-  SimpleEventLoop mainLoop;
-  mainLoop.wait(kj::mv(promise));
-
-  EXPECT_TRUE(received);
 }
 
 TEST_F(AsyncUnixTest, Poll) {
@@ -158,19 +134,17 @@ TEST_F(AsyncUnixTest, Poll) {
   KJ_SYSCALL(pipe(pipefds));
   KJ_SYSCALL(write(pipefds[1], "foo", 3));
 
-  EXPECT_EQ(POLLIN, loop.wait(loop.onFdEvent(pipefds[0], POLLIN | POLLPRI)));
+  EXPECT_EQ(POLLIN, loop.onFdEvent(pipefds[0], POLLIN | POLLPRI).wait());
 }
 
 TEST_F(AsyncUnixTest, PollMultiListen) {
   UnixEventLoop loop;
-  DummyErrorHandler dummyHandler;
 
   int bogusPipefds[2];
   KJ_SYSCALL(pipe(bogusPipefds));
   KJ_DEFER({ close(bogusPipefds[1]); close(bogusPipefds[0]); });
 
-  TaskSet tasks(loop, dummyHandler);
-  tasks.add(loop.onFdEvent(bogusPipefds[0], POLLIN | POLLPRI).thenInAnyThread([](short s) {
+  loop.daemonize(loop.onFdEvent(bogusPipefds[0], POLLIN | POLLPRI).then([](short s) {
     KJ_DBG(s);
     ADD_FAILURE() << "Received wrong poll.";
   }));
@@ -180,12 +154,11 @@ TEST_F(AsyncUnixTest, PollMultiListen) {
   KJ_DEFER({ close(pipefds[1]); close(pipefds[0]); });
   KJ_SYSCALL(write(pipefds[1], "foo", 3));
 
-  EXPECT_EQ(POLLIN, loop.wait(loop.onFdEvent(pipefds[0], POLLIN | POLLPRI)));
+  EXPECT_EQ(POLLIN, loop.onFdEvent(pipefds[0], POLLIN | POLLPRI).wait());
 }
 
 TEST_F(AsyncUnixTest, PollMultiReceive) {
   UnixEventLoop loop;
-  DummyErrorHandler dummyHandler;
 
   int pipefds[2];
   KJ_SYSCALL(pipe(pipefds));
@@ -197,40 +170,24 @@ TEST_F(AsyncUnixTest, PollMultiReceive) {
   KJ_DEFER({ close(pipefds2[1]); close(pipefds2[0]); });
   KJ_SYSCALL(write(pipefds2[1], "bar", 3));
 
-  EXPECT_EQ(POLLIN, loop.wait(loop.onFdEvent(pipefds[0], POLLIN | POLLPRI)));
-  EXPECT_EQ(POLLIN, loop.wait(loop.onFdEvent(pipefds2[0], POLLIN | POLLPRI)));
+  EXPECT_EQ(POLLIN, loop.onFdEvent(pipefds[0], POLLIN | POLLPRI).wait());
+  EXPECT_EQ(POLLIN, loop.onFdEvent(pipefds2[0], POLLIN | POLLPRI).wait());
 }
 
 TEST_F(AsyncUnixTest, PollAsync) {
-  // Arrange for another thread to wait on a UnixEventLoop...
-  auto exitThread = newPromiseAndFulfiller<void>();
-  UnixEventLoop unixLoop;
-  Thread thread([&]() {
-    unixLoop.wait(kj::mv(exitThread.promise));
-  });
-  KJ_DEFER(exitThread.fulfiller->fulfill());
+  UnixEventLoop loop;
 
-  // Make a pipe and wait on its read end in another thread.  But don't write to it yet.
+  // Make a pipe and wait on its read end while another thread writes to it.
   int pipefds[2];
   KJ_DEFER({ close(pipefds[1]); close(pipefds[0]); });
   KJ_SYSCALL(pipe(pipefds));
-  bool received = false;
-  Promise<void> promise = unixLoop.there(unixLoop.onFdEvent(pipefds[0], POLLIN | POLLPRI),
-      [&](short events) {
-    received = true;
-    EXPECT_EQ(POLLIN, events);
-  });
-
+  Thread thread([&]() {
     delay();
-
-  EXPECT_FALSE(received);
-
     KJ_SYSCALL(write(pipefds[1], "foo", 3));
+  });
 
-  SimpleEventLoop mainLoop;
-  mainLoop.wait(kj::mv(promise));
-
-  EXPECT_TRUE(received);
+  // Wait for the event in this thread.
+  EXPECT_EQ(POLLIN, loop.onFdEvent(pipefds[0], POLLIN | POLLPRI).wait());
 }
 
 }  // namespace kj

c++/src/kj/async.c++

@@ -165,18 +165,27 @@ bool EventLoop::isCurrent() const {
 }
 
 EventLoop::EventLoop()
-    : daemons(kj::heap<_::TaskSetImpl>(*this, _::LoggingErrorHandler::instance)) {}
+    : daemons(kj::heap<_::TaskSetImpl>(*this, _::LoggingErrorHandler::instance)) {
+  KJ_REQUIRE(threadLocalEventLoop == nullptr, "This thread already has an EventLoop.");
+  threadLocalEventLoop = this;
+}
 
-EventLoop::~EventLoop() noexcept(false) {}
+EventLoop::~EventLoop() noexcept(false) {
+  KJ_REQUIRE(threadLocalEventLoop == this,
+             "EventLoop being destroyed in a different thread than it was created.");
+  threadLocalEventLoop = nullptr;
+  queue.clearCallback();
+}
 
 void EventLoop::waitImpl(Own<_::PromiseNode> node, _::ExceptionOrValue& result) {
-  EventLoop* oldEventLoop = threadLocalEventLoop;
-  threadLocalEventLoop = this;
-  KJ_DEFER(threadLocalEventLoop = oldEventLoop);
+  KJ_REQUIRE(!running, "wait() is not allowed from within event callbacks.");
 
   BoolEvent event(*this);
   event.fired = node->onReady(event);
 
+  running = true;
+  KJ_DEFER(running = false);
+
   while (!event.fired) {
     KJ_IF_MAYBE(event, queue.peek(nullptr)) {
       // Arrange for events armed during the event callback to be inserted at the beginning

c++/src/kj/async.h

@@ -207,8 +207,13 @@ class EventLoop: private _::NewJobCallback {
   // EventLoop directly, but instead use `Promise`s to interact with it indirectly.  See the
   // documentation for `Promise`.
   //
-  // You will need to construct an `EventLoop` at the top level of your program.  You can then
-  // use it to construct some promises and wait on the result.  Example:
+  // Each thread can have at most one EventLoop.  When an EventLoop is created, it becomes the
+  // default loop for the current thread.  Async APIs require that the thread has a current
+  // EventLoop, or they will throw exceptions.
+  //
+  // Generally, you will want to construct an `EventLoop` at the top level of your program, e.g.
+  // in the main() function, or in the start function of a thread.  You can then use it to
+  // construct some promises and wait on the result.  Example:
   //
   //     int main() {
   //       SimpleEventLoop loop;
@@ -243,30 +248,24 @@ public:
   // Run the event loop until the promise is fulfilled, then return its result.  If the promise
   // is rejected, throw an exception.
   //
-  // It is possible to call wait() multiple times on the same event loop simultaneously, but you
-  // must be very careful about this.  Here's the deal:
-  // - If wait() is called from thread A when it is already being executed in thread B, then
-  //   thread A will block at least until thread B's call to wait() completes, _even if_ the
-  //   promise is fulfilled before that.
-  // - If wait() is called recursively from a thread in which wait() is already running, then
-  //   the inner wait() will proceed, but the outer wait() obviously cannot return until the inner
-  //   wait() completes.
-  // - Keep in mind that while wait() is running the event loop, it may be firing events that have
-  //   nothing to do with the thing you're actually waiting for.  Avoid holding any mutex locks
-  //   when you call wait() as if some other event handler happens to try to take that lock, you
-  //   will deadlock.
+  // wait() cannot be called recursively -- that is, an event callback cannot call wait().
+  // Instead, callbacks that need to perform more async operations should return a promise and
+  // rely on promise chaining.
   //
-  // In general, it is only a good idea to use `wait()` in high-level code that has a simple
-  // goal, e.g. in the main() function of a program that does one or two specific things and then
-  // exits.  On the other hand, `wait()` should be avoided in library code, unless you spawn a
-  // private thread and event loop to use it on.  Put another way, `wait()` is useful for quick
-  // prototyping but generally bad for "real code".
+  // wait() is primarily useful at the top level of a program -- typically, within the function
+  // that allocated the EventLoop.  For example, a program that performs one or two RPCs and then
+  // exits would likely use wait() in its main() function to wait on each RPC.  On the other hand,
+  // server-side code generally cannot use wait(), because it has to be able to accept multiple
+  // requests at once.
   //
   // If the promise is rejected, `wait()` throws an exception.  This exception is usually fatal,
   // so if compiled with -fno-exceptions, the process will abort.  You may work around this by
   // using `there()` with an error handler to handle this case.  If your error handler throws a
   // non-fatal exception and then recovers by returning a dummy value, wait() will also throw a
   // non-fatal exception and return the same dummy value.
+  //
+  // TODO(someday):  Implement fibers, and let them call wait() even when they are handling an
+  //   event.
 
   template <typename Func>
   PromiseForResult<Func, void> evalLater(Func&& func) const KJ_WARN_UNUSED_RESULT;
@@ -394,6 +393,9 @@ private:
     Event& event;
   };
 
+  bool running = false;
+  // True while looping -- wait() is then not allowed.
+
   _::WorkQueue<EventJob> queue;
 
   Maybe<_::WorkQueue<EventJob>::JobWrapper&> insertionPoint;
@@ -418,6 +420,7 @@ private:
 
   template <typename>
   friend class Promise;
+  friend Promise<void> yield();
 };
 
 // -------------------------------------------------------------------
@@ -634,11 +637,7 @@ public:
   // up doing the same thing as then(); don't use this unless you really know you need it.
 
   T wait();
-  // Equivalent to `EventLoop::current().wait(kj::mv(*this))`.  WARNING:  Although `wait()`
-  // advances the event loop, calls to `wait()` obviously can only return in the reverse of the
-  // order in which they were made.  `wait()` should therefore be considered a hack that should be
-  // avoided.  Consider using it only in high-level and one-off code.  In deep library code, use
-  // `then()` instead.
+  // Equivalent to `EventLoop::current().wait(kj::mv(*this))`.
   //
   // Note that `wait()` consumes the promise on which it is called, in the sense of move semantics.
   // After returning, the promise is no longer valid, and cannot be `wait()`ed on or `then()`ed
@@ -784,6 +783,11 @@ constexpr _::Void READY_NOW = _::Void();
 // Use this when you need a Promise<void> that is already fulfilled -- this value can be implicitly
 // cast to `Promise<void>`.
 
+template <typename Func>
+inline PromiseForResult<Func, void> evalLater(Func&& func) {
+  return EventLoop::current().evalLater(func);
+}
+
 // -------------------------------------------------------------------
 // Hack for creating a lambda that holds an owned pointer.
 
@@ -1481,7 +1485,7 @@ T EventLoop::wait(Promise<T>&& promise) {
 }
 
 template <typename Func>
-auto EventLoop::evalLater(Func&& func) const -> PromiseForResult<Func, void> {
+PromiseForResult<Func, void> EventLoop::evalLater(Func&& func) const {
   // Invoke thereImpl() on yieldIfSameThread().  Always spark the result.
   return PromiseForResult<Func, void>(false,
       _::spark<_::FixVoid<_::JoinPromises<_::ReturnType<Func, void>>>>(

c++/src/kj/work-queue.h

@@ -166,6 +166,8 @@ public:
   // Either the job's complete() or cancel() method will be called exactly once and will have
   // returned before the Own<const Job>'s destructor finishes.
 
+  void clearCallback() { newJobCallback = nullptr; }
+
 private:
   JobWrapper* head = nullptr;
   // Pointer to the first job.