diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index cda9064d..49b0611a 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -12,6 +12,7 @@ #include +#include #include #include #include @@ -374,6 +375,12 @@ class EventLoop //! Hook called on the event loop thread when a client has disconnected. std::function testing_hook_disconnected; + + //! Miscellaneous testing hook. Called from various places with an + //! argument identifying the call site (typically a string literal), so + //! tests can control timing or inject behavior at specific points without + //! requiring a dedicated hook for each one. + std::function testing_hook_misc; }; //! Single element task queue used to handle recursive capnp calls. (If the diff --git a/include/mp/type-context.h b/include/mp/type-context.h index 54007207..7cfc7e79 100644 --- a/include/mp/type-context.h +++ b/include/mp/type-context.h @@ -131,6 +131,18 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& std::tie(request_thread, inserted) = SetThread( GuardedRef{thread_context.waiter->m_mutex, request_threads}, server.m_context.connection, [&] { 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 + // return the response capnp caches on first use instead of + // reading Cap'n Proto connection state, which is unsafe off the + // event loop thread because RpcConnectionState::disconnect() + // overwrites it there without synchronization on an abrupt remote + // disconnect (bitcoin-core/libmultiprocess#348). After this call, + // the only call_context state accessed by the execution thread is + // the params reader and the results struct allocated here; any + // future change accessing other call_context state needs to move + // that access to the event loop thread the same way. + call_context.getResults(); }); // If an entry was inserted into the request_threads map, diff --git a/test/mp/test/foo-types.h b/test/mp/test/foo-types.h index ee12f490..85c36349 100644 --- a/test/mp/test/foo-types.h +++ b/test/mp/test/foo-types.h @@ -9,6 +9,7 @@ #include // IWYU pragma: begin_exports +#include #include #include #include @@ -75,6 +76,8 @@ inline void CustomBuildMessage(InvokeContext& invoke_context, const test::FooMessage& src, test::messages::FooMessage::Builder&& builder) { + const auto& hook{invoke_context.connection.m_loop->testing_hook_misc}; + if (hook) hook("build FooMessage"); builder.setMessage(src.message + " build"); } diff --git a/test/mp/test/foo.capnp b/test/mp/test/foo.capnp index 11bfd9c5..392719ec 100644 --- a/test/mp/test/foo.capnp +++ b/test/mp/test/foo.capnp @@ -38,6 +38,7 @@ interface FooInterface $Proxy.wrap("mp::test::FooImplementation") { callIntFnAsync @21 (context :Proxy.Context, arg :Int32) -> (result :Int32); passDataPointers @22 (arg :List(Data)) -> (result :List(Data)); listBars @25 (context :Proxy.Context, n :Int32) -> (result :List(BarInterface)); + callMessageAsync @26 (context :Proxy.Context) -> (result :FooMessage); } interface FooInit $Proxy.wrap("mp::test::FooInit") { diff --git a/test/mp/test/foo.h b/test/mp/test/foo.h index 8fc34eee..e8782708 100644 --- a/test/mp/test/foo.h +++ b/test/mp/test/foo.h @@ -124,6 +124,7 @@ class FooImplementation void callFn() { assert(m_fn); m_fn(); } void callFnAsync() { assert(m_fn); m_fn(); } int callIntFnAsync(int arg) { assert(m_int_fn); return m_int_fn(arg); } + FooMessage callMessageAsync() { assert(m_fn); m_fn(); return {}; } std::function m_fn; std::function m_int_fn; }; diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index 400aff73..7f465a33 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -5,6 +5,7 @@ #include #include +#include #include #include #include @@ -79,6 +80,7 @@ class TestSetup public: std::function server_disconnect; std::function server_disconnect_later; + std::function server_on_disconnect; std::function client_disconnect; std::promise>> client_promise; std::unique_ptr> client; @@ -109,8 +111,14 @@ class TestSetup loop.m_task_set->add(kj::evalLater([&] { server_connection.reset(); })); }; // Set handler to destroy the server when the client disconnects. This - // is ignored if server_disconnect() is called instead. - server_connection->onDisconnect([&] { server_connection.reset(); }); + // is ignored if server_disconnect() is called instead. Tests can + // assign server_on_disconnect to override the default behavior of + // destroying the server connection as soon as the disconnect is + // detected (in which case they need to destroy it themselves, + // e.g. by calling server_disconnect(), so the event loop can + // exit). + server_on_disconnect = [&] { server_connection.reset(); }; + server_connection->onDisconnect([&] { server_on_disconnect(); }); auto client_connection = std::make_unique(loop, kj::mv(pipe.ends[1])); auto client_proxy = std::make_unique>( @@ -368,6 +376,98 @@ KJ_TEST("Calling IPC method, disconnecting and blocking during the call") signal.set_value(); } +KJ_TEST("Calling async IPC method with a remote disconnect while results are built") +{ + // Regression test for bitcoin-core/libmultiprocess#348, a data race + // reported by ThreadSanitizer where a server worker thread called + // call_context.getResults() while the event loop thread was tearing down + // Cap'n Proto connection state (capnp::_::RpcConnectionState::disconnect() + // overwriting the RpcConnectionState::connection field) after an abrupt + // remote disconnect. + // + // The test makes an async IPC call (callMessageAsync) whose method body + // just signals the main thread. When the worker thread returns from the + // method body, it calls call_context.getResults(), reading the connection + // state, and starts serializing the FooMessage result, where the + // testing_hook_misc hook set below makes it sleep. Meanwhile the main + // thread destroys the client connection, and the event loop thread + // processes the resulting EOF, running RpcConnectionState::disconnect() + // and overwriting the connection state the worker thread just read, with + // no synchronization between the two accesses. + // + // Two details are essential for ThreadSanitizer to detect the race: + // + // - There must be no synchronization between the worker thread's + // getResults() call and the event loop thread's disconnect processing. + // The worker signals the main thread *before* getResults() and sleeps + // (sleeping creates no happens-before edge) across the disconnect, so + // the racing accesses are not ordered by any of the test's own + // synchronization. + // + // - The worker thread's read should come *before* the event loop thread's + // write, close in time. In the write-then-read order (e.g. method + // returning long after the disconnect), the race exists too, but + // ThreadSanitizer usually misses it: right after disconnect() writes the + // connection field, the loop thread's teardown destructors re-read it + // many times (~ImportClient etc. check connection.is() to + // decide whether to send messages), evicting the write from the + // per-granule shadow history before a late reader comes along. + // + // The server Connection object is deliberately kept alive during all this + // by overriding server_on_disconnect: destroying it would cancel the + // in-flight request (Connection::~Connection calls m_canceler.cancel(), + // setting request_canceled) and the worker would throw InterruptException + // instead of proceeding into getResults(). Keeping it alive matches the + // window in the original report, where the worker races with capnp's own + // internal teardown, which runs before any onDisconnect notification. + + TestSetup setup{/*client_owns_connection=*/false}; + ProxyClient* foo = setup.client.get(); + KJ_EXPECT(foo->add(1, 2) == 3); + foo->initThreadMap(); + + // Keep the server Connection object alive when the disconnect is detected + // so the in-flight request is not canceled (see comment above). The + // connection is destroyed at the end of the test instead. + setup.server_on_disconnect = [] {}; + + // Signaled by the worker thread when the method body runs, just before it + // returns and the worker calls getResults() and serializes the results. + std::promise fn_called; + setup.server->m_impl->m_fn = [&] { fn_called.set_value(); }; + + // Keep the worker thread inside the results-building step, without + // synchronizing, while the event loop thread processes the disconnect. + EventLoop& loop = *setup.server->m_context.connection->m_loop; + loop.testing_hook_misc = [](std::any arg) { + if (const char* const* tag{std::any_cast(&arg)}; + tag && std::string_view{*tag} == "build FooMessage") { + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + } + }; + + // Signaled by the worker thread when it is completely done with the + // request, including serializing the results. + std::promise request_done; + loop.testing_hook_async_request_done = [&] { request_done.set_value(); }; + + // Make the IPC call from a separate thread so this thread can trigger the + // client disconnect while the call is executing. + std::thread caller{[&] { + EXPECT_EXCEPTION(foo->callMessageAsync(), "IPC client method call interrupted by disconnect."); + }}; + + fn_called.get_future().get(); + setup.client_disconnect(); + caller.join(); + + // Wait for the worker thread to finish the request, then tear down the + // server connection that was deliberately kept alive above, so the event + // loop is able to exit. + request_done.get_future().get(); + setup.server_disconnect(); +} + KJ_TEST("Worker thread destroyed before it is initialized") { // Regression test for bitcoin/bitcoin#34711, bitcoin/bitcoin#34756 where a