proxy-io: Reference-count Connection objects

.github/workflows/bitcoin-core-ci.yml

@@ -18,7 +18,8 @@ concurrency:
 
 env:
   BITCOIN_REPO: bitcoin/bitcoin
-  BITCOIN_CORE_REF: refs/heads/master
+  # Temporary: use PR #35932 until it merges; revert to refs/heads/master after
+  BITCOIN_CORE_REF: refs/pull/35932/merge
   LLVM_VERSION: 22
   LIBCXX_DIR: /tmp/libcxx-build/
 

example/calculator.cpp

@@ -10,7 +10,6 @@
 #include <fstream>
 #include <functional>
 #include <iostream>
-#include <kj/async.h>
 #include <kj/common.h>
 #include <kj/memory.h>
 #include <memory>

example/printer.cpp

@@ -10,7 +10,6 @@
 #include <cstring> // IWYU pragma: keep
 #include <fstream>
 #include <iostream>
-#include <kj/async.h>
 #include <kj/common.h>
 #include <kj/memory.h>
 #include <memory>

include/mp/proxy-types.h

@@ -345,7 +345,7 @@ auto PassField(Priority<1>, TypeList<LocalType&>, ServerContext& server_context,
     const auto& params = server_context.call_context.getParams();
     const auto& input = Make<StructField, Accessor>(params);
     using Interface = typename Decay<decltype(input.get())>::Calls;
-    auto param = std::make_unique<ProxyClient<Interface>>(input.get(), server_context.proxy_server.m_context.connection, false);
+    auto param = std::make_unique<ProxyClient<Interface>>(input.get(), server_context.proxy_server.m_context.connection.get(), false);
     fn.invoke(server_context, std::forward<Args>(args)..., *param);
 }
 
@@ -734,7 +734,12 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel
     bool done = false;
     const char* disconnected = nullptr;
     proxy_client.m_context.loop->sync([&]() {
-        if (!proxy_client.m_context.connection) {
+        // Fail immediately on a disconnected connection instead of trying to
+        // send a request through it. (The connection object itself is always
+        // valid, since m_context holds shared ownership of it.) disconnected()
+        // is only meaningful on the event loop thread, which this sync()
+        // callback runs on.
+        if (proxy_client.m_context.connection->disconnected()) {
             const Lock lock(thread_context.waiter->m_mutex);
             done = true;
             disconnected = "IPC client method called after disconnect.";

include/mp/proxy.h

@@ -69,7 +69,11 @@ class EventLoopRef
 //! Context data associated with proxy client and server classes.
 struct ProxyContext
 {
-    Connection* connection;
+    //! Connection this proxy object is associated with. Holding shared
+    //! ownership (Connection objects are always owned by shared_ptr, see
+    //! Connection::make) guarantees the Connection outlives this proxy
+    //! object, so the pointer stays valid even after a disconnect.
+    std::shared_ptr<Connection> connection;
     EventLoopRef loop;
     CleanupList cleanup_fns;
 

include/mp/type-context.h

@@ -129,7 +129,7 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
             // cancel_monitor.m_canceled was checked above and this
             // code is running on the event loop thread.
             std::tie(request_thread, inserted) = SetThread(
-                GuardedRef{thread_context.waiter->m_mutex, request_threads}, server.m_context.connection,
+                GuardedRef{thread_context.waiter->m_mutex, request_threads}, server.m_context.connection.get(),
                 [&] { return Accessor::get(call_context.getParams()).getCallbackThread(); });
             // Initialize the request's results struct here on the event loop
             // thread, so later getResults() calls on the execution thread
@@ -179,14 +179,14 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                 if (erase_thread) {
                     // Look up the thread again without using existing
                     // iterator since entry may no longer be there after
-                    // a disconnect. Destroy node after releasing
-                    // Waiter::m_mutex, so the ProxyClient<Thread>
-                    // destructor is able to use EventLoop::mutex
-                    // without violating lock order.
+                    // a disconnect. Destroy the node after releasing
+                    // Waiter::m_mutex, so ~ProxyClient<Thread> does not
+                    // run with the mutex held (see the SetThread
+                    // disconnect callback).
                     ConnThreads::node_type removed;
                     {
                         Lock lock(thread_context.waiter->m_mutex);
-                        removed = request_threads.extract(server.m_context.connection);
+                        removed = request_threads.extract(server.m_context.connection.get());
                     }
                 }
             });
@@ -215,7 +215,7 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
     if (!context_arg.hasThread()) {
         // No client thread specified — dispatch through the server thread
         // pool, picking the slot with the smallest in-flight depth.
-        auto* connection = server.m_context.connection;
+        auto* connection = server.m_context.connection.get();
         auto& pool = connection->m_thread_pool;
         if (pool.empty()) {
             MP_LOG(loop, Log::Error)