refactor: Add ProxyContext EventLoop* member

This commit makes mechanical changes needed to simplify an upcoming commit
which replaces EventLoop* with an EventLoopRef.

This change also happens to be also useful on its own so clientInvoke can
detect disconnections in a non-racy way
(#123 (comment))
by seeing if the client Connection pointer is null while holding the event loop
mutex.

Author: ryanofsky

include/mp/proxy-io.h

@@ -446,7 +446,7 @@ ProxyClientBase<Interface, Impl>::ProxyClientBase(typename Interface::Client cli
         Sub::destroy(*this);
 
         // FIXME: Could just invoke removed addCleanup fn here instead of duplicating code
-        m_context.connection->m_loop.sync([&]() {
+        m_context.loop->sync([&]() {
             // Release client capability by move-assigning to temporary.
             {
                 typename Interface::Client(std::move(m_client));

include/mp/proxy-types.h

@@ -568,7 +568,7 @@ template <typename Client>
 void clientDestroy(Client& client)
 {
     if (client.m_context.connection) {
-        client.m_context.connection->m_loop.log() << "IPC client destroy " << typeid(client).name();
+        client.m_context.loop->log() << "IPC client destroy " << typeid(client).name();
     } else {
         KJ_LOG(INFO, "IPC interrupted client destroy", typeid(client).name());
     }
@@ -577,7 +577,7 @@ void clientDestroy(Client& client)
 template <typename Server>
 void serverDestroy(Server& server)
 {
-    server.m_context.connection->m_loop.log() << "IPC server destroy " << typeid(server).name();
+    server.m_context.loop->log() << "IPC server destroy " << typeid(server).name();
 }
 
 //! Entry point called by generated client code that looks like:
@@ -597,7 +597,7 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel
     }
     if (!g_thread_context.waiter) {
         assert(g_thread_context.thread_name.empty());
-        g_thread_context.thread_name = ThreadName(proxy_client.m_context.connection->m_loop.m_exe_name);
+        g_thread_context.thread_name = ThreadName(proxy_client.m_context.loop->m_exe_name);
         // If next assert triggers, it means clientInvoke is being called from
         // the capnp event loop thread. This can happen when a ProxyServer
         // method implementation that runs synchronously on the event loop
@@ -608,26 +608,26 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel
         // declaration so the server method runs in a dedicated thread.
         assert(!g_thread_context.loop_thread);
         g_thread_context.waiter = std::make_unique<Waiter>();
-        proxy_client.m_context.connection->m_loop.logPlain()
+        proxy_client.m_context.loop->logPlain()
             << "{" << g_thread_context.thread_name
             << "} IPC client first request from current thread, constructing waiter";
     }
     ClientInvokeContext invoke_context{*proxy_client.m_context.connection, g_thread_context};
     std::exception_ptr exception;
     std::string kj_exception;
     bool done = false;
-    proxy_client.m_context.connection->m_loop.sync([&]() {
+    proxy_client.m_context.loop->sync([&]() {
         auto request = (proxy_client.m_client.*get_request)(nullptr);
         using Request = CapRequestTraits<decltype(request)>;
         using FieldList = typename ProxyClientMethodTraits<typename Request::Params>::Fields;
         IterateFields().handleChain(invoke_context, request, FieldList(), typename FieldObjs::BuildParams{&fields}...);
-        proxy_client.m_context.connection->m_loop.logPlain()
+        proxy_client.m_context.loop->logPlain()
             << "{" << invoke_context.thread_context.thread_name << "} IPC client send "
             << TypeName<typename Request::Params>() << " " << LogEscape(request.toString());
 
-        proxy_client.m_context.connection->m_loop.m_task_set->add(request.send().then(
+        proxy_client.m_context.loop->m_task_set->add(request.send().then(
             [&](::capnp::Response<typename Request::Results>&& response) {
-                proxy_client.m_context.connection->m_loop.logPlain()
+                proxy_client.m_context.loop->logPlain()
                     << "{" << invoke_context.thread_context.thread_name << "} IPC client recv "
                     << TypeName<typename Request::Results>() << " " << LogEscape(response.toString());
                 try {
@@ -642,7 +642,7 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel
             },
             [&](const ::kj::Exception& e) {
                 kj_exception = kj::str("kj::Exception: ", e).cStr();
-                proxy_client.m_context.connection->m_loop.logPlain()
+                proxy_client.m_context.loop->logPlain()
                     << "{" << invoke_context.thread_context.thread_name << "} IPC client exception " << kj_exception;
                 const std::unique_lock<std::mutex> lock(invoke_context.thread_context.waiter->m_mutex);
                 done = true;
@@ -653,7 +653,7 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel
     std::unique_lock<std::mutex> lock(invoke_context.thread_context.waiter->m_mutex);
     invoke_context.thread_context.waiter->wait(lock, [&done]() { return done; });
     if (exception) std::rethrow_exception(exception);
-    if (!kj_exception.empty()) proxy_client.m_context.connection->m_loop.raise() << kj_exception;
+    if (!kj_exception.empty()) proxy_client.m_context.loop->raise() << kj_exception;
 }
 
 //! Invoke callable `fn()` that may return void. If it does return void, replace
@@ -687,7 +687,7 @@ kj::Promise<void> serverInvoke(Server& server, CallContext& call_context, Fn fn)
     using Results = typename decltype(call_context.getResults())::Builds;
 
     int req = ++server_reqs;
-    server.m_context.connection->m_loop.log() << "IPC server recv request  #" << req << " "
+    server.m_context.loop->log() << "IPC server recv request  #" << req << " "
                                      << TypeName<typename Params::Reads>() << " " << LogEscape(params.toString());
 
     try {
@@ -704,14 +704,14 @@ kj::Promise<void> serverInvoke(Server& server, CallContext& call_context, Fn fn)
         return ReplaceVoid([&]() { return fn.invoke(server_context, ArgList()); },
             [&]() { return kj::Promise<CallContext>(kj::mv(call_context)); })
             .then([&server, req](CallContext call_context) {
-                server.m_context.connection->m_loop.log() << "IPC server send response #" << req << " " << TypeName<Results>()
+                server.m_context.loop->log() << "IPC server send response #" << req << " " << TypeName<Results>()
                                                  << " " << LogEscape(call_context.getResults().toString());
             });
     } catch (const std::exception& e) {
-        server.m_context.connection->m_loop.log() << "IPC server unhandled exception: " << e.what();
+        server.m_context.loop->log() << "IPC server unhandled exception: " << e.what();
         throw;
     } catch (...) {
-        server.m_context.connection->m_loop.log() << "IPC server unhandled exception";
+        server.m_context.loop->log() << "IPC server unhandled exception";
         throw;
     }
 }

include/mp/proxy.h

@@ -72,9 +72,10 @@ class EventLoopRef
 struct ProxyContext
 {
     Connection* connection;
+    EventLoop* loop;
     CleanupList cleanup_fns;
 
-    ProxyContext(Connection* connection) : connection(connection) {}
+    ProxyContext(Connection* connection);
 };
 
 //! Base class for generated ProxyClient classes that implement a C++ interface

include/mp/type-context.h

@@ -132,13 +132,13 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                     return;
                 }
                 KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() {
-                    server.m_context.connection->m_loop.sync([&] {
+                    server.m_context.loop->sync([&] {
                         auto fulfiller_dispose = kj::mv(fulfiller);
                         fulfiller_dispose->fulfill(kj::mv(call_context));
                     });
                 }))
                 {
-                    server.m_context.connection->m_loop.sync([&]() {
+                    server.m_context.loop->sync([&]() {
                         auto fulfiller_dispose = kj::mv(fulfiller);
                         fulfiller_dispose->reject(kj::mv(*exception));
                     });
@@ -156,11 +156,11 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
             // thread.
             KJ_IF_MAYBE (thread_server, perhaps) {
                 const auto& thread = static_cast<ProxyServer<Thread>&>(*thread_server);
-                server.m_context.connection->m_loop.log()
+                server.m_context.loop->log()
                     << "IPC server post request  #" << req << " {" << thread.m_thread_context.thread_name << "}";
                 thread.m_thread_context.waiter->post(std::move(invoke));
             } else {
-                server.m_context.connection->m_loop.log()
+                server.m_context.loop->log()
                     << "IPC server error request #" << req << ", missing thread to execute request";
                 throw std::runtime_error("invalid thread handle");
             }

src/mp/proxy.cpp

@@ -65,6 +65,8 @@ bool EventLoopRef::reset()
     return done;
 }
 
+ProxyContext::ProxyContext(Connection* connection) : connection(connection), loop{&connection->m_loop} {}
+
 Connection::~Connection()
 {
     // Shut down RPC system first, since this will garbage collect Server