diff --git a/src/ipc/capnp/protocol.cpp b/src/ipc/capnp/protocol.cpp index 3326f70931..7bc653d25c 100644 --- a/src/ipc/capnp/protocol.cpp +++ b/src/ipc/capnp/protocol.cpp @@ -41,10 +41,7 @@ class CapnpProtocol : public Protocol public: ~CapnpProtocol() noexcept(true) { - if (m_loop) { - std::unique_lock lock(m_loop->m_mutex); - m_loop->removeClient(lock); - } + m_loop_ref.reset(); if (m_loop_thread.joinable()) m_loop_thread.join(); assert(!m_loop); }; @@ -83,10 +80,7 @@ public: m_loop_thread = std::thread([&] { util::ThreadRename("capnp-loop"); m_loop.emplace(exe_name, &IpcLogFn, &m_context); - { - std::unique_lock lock(m_loop->m_mutex); - m_loop->addClient(lock); - } + m_loop_ref.emplace(*m_loop); promise.set_value(); m_loop->loop(); m_loop.reset(); @@ -95,7 +89,12 @@ public: } Context m_context; std::thread m_loop_thread; + //! EventLoop object which manages I/O events for all connections. std::optional m_loop; + //! Reference to the same EventLoop. Increments the loop’s refcount on + //! creation, decrements on destruction. The loop thread exits when the + //! refcount reaches 0. Other IPC objects also hold their own EventLoopRef. + std::optional m_loop_ref; }; } // namespace diff --git a/src/test/ipc_test.cpp b/src/test/ipc_test.cpp index b4d7ad354c..eeaf348812 100644 --- a/src/test/ipc_test.cpp +++ b/src/test/ipc_test.cpp @@ -55,7 +55,6 @@ void IpcPipeTest() { // Setup: create FooImplementation object and listen for FooInterface requests std::promise>> foo_promise; - std::function disconnect_client; std::thread thread([&]() { mp::EventLoop loop("IpcPipeTest", [](bool raise, const std::string& log) { LogPrintf("LOG%i: %s\n", raise, log); }); auto pipe = loop.m_io_context.provider->newTwoWayPipe(); @@ -63,9 +62,9 @@ void IpcPipeTest() auto connection_client = std::make_unique(loop, kj::mv(pipe.ends[0])); auto foo_client = std::make_unique>( connection_client->m_rpc_system->bootstrap(mp::ServerVatId().vat_id).castAs(), - connection_client.get(), /* destroy_connection= */ false); + connection_client.get(), /* destroy_connection= */ true); + connection_client.release(); foo_promise.set_value(std::move(foo_client)); - disconnect_client = [&] { loop.sync([&] { connection_client.reset(); }); }; auto connection_server = std::make_unique(loop, kj::mv(pipe.ends[1]), [&](mp::Connection& connection) { auto foo_server = kj::heap>(std::make_shared(), connection); @@ -106,8 +105,8 @@ void IpcPipeTest() auto script2{foo->passScript(script1)}; BOOST_CHECK_EQUAL(HexStr(script1), HexStr(script2)); - // Test cleanup: disconnect pipe and join thread - disconnect_client(); + // Test cleanup: disconnect and join thread + foo.reset(); thread.join(); }