EventLoopGuarded is like MutexGuarded but synchronizes by queuing operations to a single EventLoop.

c++/src/capnp/capability.c++

@@ -347,21 +347,18 @@ private:
 class LocalClient final: public ClientHook, public kj::Refcounted {
 public:
   LocalClient(const kj::EventLoop& eventLoop, kj::Own<Capability::Server>&& server)
-      : eventLoop(eventLoop), server(kj::mv(server)) {}
+      : server(eventLoop, kj::mv(server)) {}
 
   Request<ObjectPointer, ObjectPointer> newCall(
       uint64_t interfaceId, uint16_t methodId, uint firstSegmentWordSize) const override {
     auto hook = kj::heap<LocalRequest>(
-        eventLoop, interfaceId, methodId, firstSegmentWordSize, kj::addRef(*this));
+        server.getEventLoop(), interfaceId, methodId, firstSegmentWordSize, kj::addRef(*this));
     auto root = hook->message->getRoot();  // Do not inline `root` -- kj::mv may happen first.
     return Request<ObjectPointer, ObjectPointer>(root, kj::mv(hook));
   }
 
   VoidPromiseAndPipeline call(uint64_t interfaceId, uint16_t methodId,
                               kj::Own<CallContextHook>&& context) const override {
-    // We can const-cast the server because we're synchronizing on the event loop.
-    auto server = const_cast<Capability::Server*>(this->server.get());
-
     auto contextPtr = context.get();
 
     // We don't want to actually dispatch the call synchronously, because:
@@ -374,26 +371,26 @@ public:
     //
     // Note also that QueuedClient depends on this evalLater() to ensure that pipelined calls don't
     // complete before 'whenMoreResolved()' promises resolve.
-    auto promise = eventLoop.evalLater(
-        [=]() {
+    auto promise = server.applyLater(
+        [=](kj::Own<Capability::Server>& server) {
           return server->dispatchCall(interfaceId, methodId,
                                       CallContext<ObjectPointer, ObjectPointer>(*contextPtr));
         });
 
     // Make sure that this client cannot be destroyed until the promise completes.
-    promise = eventLoop.there(kj::mv(promise), kj::mvCapture(kj::addRef(*this),
-        [=](kj::Own<const LocalClient>&& ref) {}));
+    promise = promise.thenInAnyThread(kj::mvCapture(kj::addRef(*this),
+        [](kj::Own<const LocalClient>&& ref) {}));
 
     // We have to fork this promise for the pipeline to receive a copy of the answer.
-    auto forked = eventLoop.fork(kj::mv(promise));
+    auto forked = server.getEventLoop().fork(kj::mv(promise));
 
-    auto pipelinePromise = eventLoop.there(forked.addBranch(), kj::mvCapture(context->addRef(),
+    auto pipelinePromise = forked.addBranch().thenInAnyThread(kj::mvCapture(context->addRef(),
         [=](kj::Own<CallContextHook>&& context) -> kj::Own<const PipelineHook> {
           context->releaseParams();
           return kj::refcounted<LocalPipeline>(kj::mv(context));
         }));
 
-    auto completionPromise = eventLoop.there(forked.addBranch(), kj::mvCapture(context,
+    auto completionPromise = forked.addBranch().thenInAnyThread(kj::mvCapture(context,
         [=](kj::Own<CallContextHook>&& context) {
           // Nothing to do here.  We just wanted to make sure to hold on to a reference to the
           // context even if the pipeline was discarded.
@@ -403,7 +400,7 @@ public:
         }));
 
     return VoidPromiseAndPipeline { kj::mv(completionPromise),
-        kj::refcounted<QueuedPipeline>(eventLoop, kj::mv(pipelinePromise)) };
+        kj::refcounted<QueuedPipeline>(server.getEventLoop(), kj::mv(pipelinePromise)) };
   }
 
   kj::Maybe<kj::Promise<kj::Own<const ClientHook>>> whenMoreResolved() const override {
@@ -420,8 +417,7 @@ public:
   }
 
 private:
-  const kj::EventLoop& eventLoop;
-  kj::Own<Capability::Server> server;
+  kj::EventLoopGuarded<kj::Own<Capability::Server>> server;
 };
 
 }  // namespace

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

@@ -472,5 +472,69 @@ TEST(Async, TaskSet) {
   EXPECT_EQ(1u, errorHandler.exceptionCount);
 }
 
+TEST(Async, EventLoopGuarded) {
+  SimpleEventLoop loop;
+  EventLoopGuarded<int> guarded(loop, 123);
+
+  {
+    EXPECT_EQ(123, guarded.getValue());
+
+    // We're not in the event loop, so the function will be applied later.
+    auto promise = guarded.applyNow([](int& i) -> const char* {
+      EXPECT_EQ(123, i);
+      i = 234;
+      return "foo";
+    });
+
+    EXPECT_EQ(123, guarded.getValue());
+
+    EXPECT_STREQ("foo", loop.wait(kj::mv(promise)));
+
+    EXPECT_EQ(234, guarded.getValue());
+  }
+
+  {
+    auto promise = loop.evalLater([&]() {
+      EXPECT_EQ(234, guarded.getValue());
+
+      // Since we're in the event loop, applyNow() will apply synchronously.
+      auto promise = guarded.applyNow([](int& i) -> const char* {
+        EXPECT_EQ(234, i);
+        i = 345;
+        return "bar";
+      });
+
+      EXPECT_EQ(345, guarded.getValue());  // already applied
+
+      return kj::mv(promise);
+    });
+
+    EXPECT_STREQ("bar", loop.wait(kj::mv(promise)));
+
+    EXPECT_EQ(345, guarded.getValue());
+  }
+
+  {
+    auto promise = loop.evalLater([&]() {
+      EXPECT_EQ(345, guarded.getValue());
+
+      // applyLater() is never synchronous.
+      auto promise = guarded.applyLater([](int& i) -> const char* {
+        EXPECT_EQ(345, i);
+        i = 456;
+        return "baz";
+      });
+
+      EXPECT_EQ(345, guarded.getValue());
+
+      return kj::mv(promise);
+    });
+
+    EXPECT_STREQ("baz", loop.wait(kj::mv(promise)));
+
+    EXPECT_EQ(456, guarded.getValue());
+  }
+}
+
 }  // namespace
 }  // namespace kj

c++/src/kj/async.h

@@ -669,6 +669,59 @@ private:
   friend class EventLoop;
 };
 
+template <typename T>
+class EventLoopGuarded {
+  // An instance of T that is bound to a particular EventLoop and may only be modified within that
+  // loop.
+
+public:
+  template <typename... Params>
+  inline EventLoopGuarded(const EventLoop& loop, Params&&... params)
+      : loop(loop), value(kj::fwd<Params>(params)...) {}
+
+  const T& getValue() const { return value; }
+  // Get a const (thread-safe) reference to the value.
+
+  const EventLoop& getEventLoop() const { return loop; }
+  // Get the EventLoop in which this value can be modified.
+
+  template <typename Func>
+  PromiseForResult<Func, T&> applyNow(Func&& func) const KJ_WARN_UNUSED_RESULT {
+    // Calls the given function, passing the guarded object to it as a mutable reference, and
+    // returning a pointer to the function's result.  When called from within the object's event
+    // loop, the function runs synchronously, but when called from any other thread, the function
+    // is queued to run on the object's loop later.
+
+    if (loop.isCurrent()) {
+      return func(const_cast<T&>(value));
+    } else {
+      return applyLater(kj::fwd<Func>(func));
+    }
+  }
+
+  template <typename Func>
+  PromiseForResult<Func, T&> applyLater(Func&& func) const KJ_WARN_UNUSED_RESULT {
+    // Like `applyNow` but always queues the function to run later regardless of which thread
+    // called it.
+
+    return loop.evalLater(Capture<Func> { const_cast<T&>(value), kj::fwd<Func>(func) });
+  }
+
+private:
+  const EventLoop& loop;
+  T value;
+
+  template <typename Func>
+  struct Capture {
+    T& value;
+    Func func;
+
+    decltype(func(value)) operator()() {
+      return func(value);
+    }
+  };
+};
+
 class TaskSet {
   // Holds a collection of Promise<void>s and ensures that each executes to completion.  Memory
   // associated with each promise is automatically freed when the promise completes.  Destroying

mega-test-quick.cfg

 linux-gcc-4.7      1731 ./super-test.sh tmpdir capnp-gcc-4.7 quick
 linux-gcc-4.8      1734 ./super-test.sh tmpdir capnp-gcc-4.8 quick gcc-4.8
 linux-clang        1754 ./super-test.sh tmpdir capnp-clang quick clang
-mac                 802 ./super-test.sh remote beat caffeinate quick
+mac                 807 ./super-test.sh remote beat caffeinate quick
 cygwin              810 ./super-test.sh remote Kenton@flashman quick