diff --git a/example/calculator.cpp b/example/calculator.cpp index 5b9990ea..51d5e587 100644 --- a/example/calculator.cpp +++ b/example/calculator.cpp @@ -10,7 +10,6 @@ #include #include #include -#include #include #include #include diff --git a/example/printer.cpp b/example/printer.cpp index 2ba3f600..e3d4ce63 100644 --- a/example/printer.cpp +++ b/example/printer.cpp @@ -10,7 +10,6 @@ #include // IWYU pragma: keep #include #include -#include #include #include #include diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 49b0611a..172a94c8 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -376,6 +376,11 @@ class EventLoop //! Hook called on the event loop thread when a client has disconnected. std::function testing_hook_disconnected; + //! Hook called at the start of ~ProxyClient, on whichever thread + //! is destroying the object, with the object being destroyed. Used by + //! tests to control timing during thread map teardown. + std::function*)> testing_hook_thread_client_destroy; + //! 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 @@ -431,67 +436,203 @@ struct Waiter //! to guard access to related state. Specifically, since the thread_local //! ThreadContext struct owns a Waiter, the Waiter::m_mutex is used to guard //! access to other parts of the struct to avoid needing to deal with more - //! mutexes than necessary. This mutex can be held at the same time as - //! EventLoop::m_mutex as long as Waiter::mutex is locked first and - //! EventLoop::m_mutex is locked second. + //! mutexes than necessary. + //! + //! Lock order: this mutex can be held at the same time as + //! EventLoop::m_mutex as long as Waiter::m_mutex is locked first and + //! EventLoop::m_mutex is locked second. ~ProxyServer locks them + //! in this order on the event loop thread. No code locks them in the + //! reverse order. + //! + //! Blocking rule: a thread other than the event loop thread must not hold + //! this mutex while calling EventLoop::sync() or EventLoop::post(). Those + //! calls block until the event loop thread runs the posted function, and + //! the event loop thread locks Waiter::m_mutex itself (in SetThread and + //! its disconnect callback, Waiter::post, and ~ProxyServer), so + //! holding the mutex across the call could deadlock even though the two + //! mutexes would be locked in the permitted order. This is why + //! ~ThreadContext releases the mutex before destroying ProxyClient + //! objects, whose destructor calls EventLoop::sync(). Mutex m_mutex; std::condition_variable m_cv MP_GUARDED_BY(m_mutex); std::optional> m_fn MP_GUARDED_BY(m_mutex); }; +//! Counter tracking the number of live ProxyServer objects associated with a +//! Connection, used to wait for a disconnected connection's server side to +//! become quiescent. +//! +//! This count is non-zero as long as the client is holding any object handles +//! or as long as any server methods are running, even after the client as +//! disconnected or freed all its handles. When this is zero it means no more +//! ProxyServer method calls can originate from the connection and it is safe to +//! free any resources accessed through the connection. +// +//! Why counting live server objects is a valid "no server call body running" +//! signal: a ProxyServer object is reference counted and is not destroyed +//! until its outstanding calls finish. Cap'n Proto keeps the target capability +//! alive for the duration of a call, and the mp.Context PassField overload and +//! ProxyServer::post() additionally pin it (self = thisCap()) until +//! the call body running on a worker thread completes and its result is +//! delivered. So "object destroyed" implies "its call bodies finished", and a +//! connection whose live-object count reached zero after a disconnect has no +//! server code running. This matters because disconnecting only cancels the +//! KJ promise of an in-flight call; it does not interrupt a call body that +//! was already dispatched to a worker thread (see Connection::disconnect). +//! +//! The counter is held via shared_ptr by the Connection and by every +//! ProxyServer object created for the connection, because a ProxyServer +//! object kept alive by an in-flight call can outlive the Connection (see +//! ~ProxyServerBase), and its destructor must decrement state that is still +//! valid. +//! +//! ProxyServer and ProxyServer are separate +//! specializations (not ProxyServerBase instances) and are intentionally not +//! counted: every application method body runs on an interface ProxyServer, +//! which is counted and stays alive for the duration of the body, so counting +//! those is sufficient. +struct ServerObjectTracker +{ + //! Called from the ProxyServerBase constructor (on the event loop thread). + void addServerObject() + { + const Lock lock(m_mutex); + m_count += 1; + } + + //! Called from the ProxyServerBase destructor (on the event loop thread). + void removeServerObject() + { + { + const Lock lock(m_mutex); + assert(m_count > 0); + m_count -= 1; + } + m_cv.notify_all(); + } + + //! Return the current count. May be called from any thread. + size_t pendingServerObjects() const + { + const Lock lock(m_mutex); + return m_count; + } + + //! Block until no server objects remain. Must NOT be called from the event + //! loop thread: in-flight call bodies need the event loop to deliver their + //! results before their server objects are destroyed, so blocking the loop + //! here would deadlock. + void waitDrained() + { + Lock lock(m_mutex); + if (m_count != 0 && testing_hook_wait) testing_hook_wait(); + m_cv.wait(lock.m_lock, [this]() MP_REQUIRES(m_mutex) { return m_count == 0; }); + } + + mutable Mutex m_mutex; + std::condition_variable m_cv; + size_t m_count MP_GUARDED_BY(m_mutex){0}; + std::function testing_hook_wait; +}; + //! Object holding network & rpc state associated with either an incoming server //! connection, or an outgoing client connection. It must be created and destroyed //! on the event loop thread. //! In addition to Cap'n Proto state, it also holds lists of callbacks to run //! when the connection is closed. +//! +//! A Connection may be severed with disconnect() and then kept alive (rather +//! than destroyed) so callers can wait for in-flight server calls to finish. +//! Once disconnect() has run the object holds no transport, RPC system, or +//! worker threads, so the only valid operations on it are destruction and +//! calling disconnect() again (a no-op). Methods that perform I/O or make +//! calls must not be used after disconnect(). class Connection { public: Connection(EventLoop& loop, kj::Own&& stream_) : m_loop(loop), m_stream(kj::mv(stream_)), - m_network(*m_stream, ::capnp::rpc::twoparty::Side::CLIENT, ::capnp::ReaderOptions()), - m_rpc_system(::capnp::makeRpcClient(m_network)) {} + m_network(std::in_place, *m_stream, ::capnp::rpc::twoparty::Side::CLIENT, ::capnp::ReaderOptions()), + m_rpc_system(::capnp::makeRpcClient(*m_network)) {} Connection(EventLoop& loop, kj::Own&& stream_, const std::function<::capnp::Capability::Client(Connection&)>& make_client) : m_loop(loop), m_stream(kj::mv(stream_)), - m_network(*m_stream, ::capnp::rpc::twoparty::Side::SERVER, ::capnp::ReaderOptions()), - m_rpc_system(::capnp::makeRpcServer(m_network, make_client(*this))) {} - - //! Run cleanup functions. Must be called from the event loop thread. First - //! calls synchronous cleanup functions while blocked (to free capnp - //! Capability::Client handles owned by ProxyClient objects), then schedules - //! asynchronous cleanup functions to run in a worker thread (to run - //! destructors of m_impl instances owned by ProxyServer objects). + m_network(std::in_place, *m_stream, ::capnp::rpc::twoparty::Side::SERVER, ::capnp::ReaderOptions()), + m_rpc_system(::capnp::makeRpcServer(*m_network, make_client(*this))) {} + + //! Destroy the connection. Calls disconnect() if it has not been called + //! already. Must be called from the event loop thread. ~Connection() noexcept(false); - //! Register synchronous cleanup function to run on event loop thread (with - //! access to capnp thread local variables) when disconnect() is called. - //! any new i/o. + //! Sever the connection without destroying this object: close the transport + //! so the peer observes the disconnect, and run the connection's cleanup + //! handlers. Idempotent, and called automatically by the destructor -- + //! calling it directly is only needed to disconnect while keeping the object + //! alive (e.g. to wait for in-flight calls to drain afterward). Must be + //! called from the event loop thread. + //! + //! Cancels the KJ promise of any in-flight call, but a server method body + //! already dispatched to a worker thread runs to completion. + void disconnect(); + + //! Return server object tracker allowing clients to block until no + //! ProxyServer objects associated with this connection remain, i.e. until + //! no server call body is still executing (see ServerObjectTracker). + using Tracker = std::shared_ptr; + Tracker tracker() { return m_server_objects; } + + //! Register a synchronous cleanup function to run on the event loop thread + //! (with access to capnp thread-local variables) when the connection is + //! disconnected -- for either a remote disconnect (the peer closes the + //! connection) or a local one (the connection is torn down on this side). + //! Contrast onDisconnect(), whose handler runs only on a remote + //! disconnect. Returns a handle that can be passed to removeSyncCleanup() + //! to unregister the function before it runs. CleanupIt addSyncCleanup(std::function fn); void removeSyncCleanup(CleanupIt it); - //! Add disconnect handler. + //! Register a handler to run on the event loop thread when the peer + //! disconnects. The handler runs at most once, and only while this + //! Connection is still alive: if the connection is torn down locally before + //! the handler runs, the handler is not called. template void onDisconnect(F&& f) { - // Add disconnect handler to local TaskSet to ensure it is canceled and - // will never run after connection object is destroyed. But when disconnect - // handler fires, do not call the function f right away, instead add it - // to the EventLoop TaskSet to avoid "Promise callback destroyed itself" - // error in the typical case where f deletes this Connection object. - m_on_disconnect.add(m_network.onDisconnect().then( - [f = std::forward(f), this]() mutable { m_loop->m_task_set->add(kj::evalLater(kj::mv(f))); })); + // m_network->onDisconnect() fires both on a remote disconnect and on a + // local one (disconnecting resets m_rpc_system, which drops capnp's + // last reference to the network and fulfills the promise). The m_alive + // weak_ptr tells the two apart -- disconnect() expires it, so f is + // skipped on local disconnects. This lets onDisconnect callbacks + // delete the Connection without a double deletion. + m_loop->m_task_set->add(m_network->onDisconnect().then( + [f = std::forward(f), alive = std::weak_ptr(m_alive)]() mutable { + if (!alive.expired()) f(); + })); } EventLoopRef m_loop; kj::Own m_stream; - LoggingErrorHandler m_error_handler{*m_loop}; - //! TaskSet used to cancel the m_network.onDisconnect() handler for remote - //! disconnections, if the connection is closed locally first by deleting - //! this Connection object. - kj::TaskSet m_on_disconnect{m_error_handler}; - ::capnp::TwoPartyVatNetwork m_network; + //! Liveness token checked by onDisconnect() callbacks (see there). + //! Could be dropped if Connection lifetime were reference-counted (#336). + std::shared_ptr m_alive{std::make_shared()}; + //! Wrapped in std::optional so disconnect() can tear it down (along with + //! the stream) to sever the transport while this object stays alive. + //! Closing the stream is what makes the peer observe the disconnect: it + //! reads EOF and fails its outstanding calls with DISCONNECTED errors. + std::optional<::capnp::TwoPartyVatNetwork> m_network; + + //! Tracker for live ProxyServer objects associated with this connection, + //! used by waitDrained(). Held via shared_ptr because ProxyServer objects + //! kept alive by in-flight calls can outlive the Connection (see + //! ServerObjectTracker and ~ProxyServerBase). + //! + //! Must be declared before m_rpc_system: constructing m_rpc_system runs + //! the make_client callback, which creates the bootstrap (Init) server + //! object, whose ProxyServerBase constructor registers itself with this + //! tracker. + std::shared_ptr m_server_objects{std::make_shared()}; + std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId>> m_rpc_system; // ThreadMap interface client, used to create a remote server thread when an @@ -583,8 +724,8 @@ ProxyClientBase::ProxyClientBase(typename Interface::Client cli // Remove disconnect callback on cleanup so it doesn't run and try // to access this object after it's destroyed. This call needs to // run inside loop->sync() on the event loop thread because - // otherwise, if there were an ill-timed disconnect, the - // onDisconnect handler could fire and delete the Connection object + // otherwise, if there were an ill-timed disconnect, the remote + // disconnect handler could fire and delete the Connection object // before the removeSyncCleanup call. if (m_context.connection) m_context.connection->removeSyncCleanup(disconnect_cb); @@ -633,8 +774,14 @@ ProxyClientBase::~ProxyClientBase() noexcept template ProxyServerBase::ProxyServerBase(std::shared_ptr impl, Connection& connection) - : m_impl(std::move(impl)), m_context(&connection) + : m_impl(std::move(impl)), m_context(&connection), m_server_objects(connection.m_server_objects) { + // Register this object with the connection's live-object tracker. This + // runs on the event loop thread, so it is ordered before any connection + // teardown (which also runs on the event loop thread): code that + // disconnects the connection and then calls Connection::waitDrained() is + // guaranteed to see this object. + m_server_objects->addServerObject(); MP_LOG(*m_context.loop, Log::Debug) << "Creating " << CxxTypeName(*this) << " " << this; assert(m_impl); } @@ -682,6 +829,15 @@ ProxyServerBase::~ProxyServerBase() } assert(m_context.cleanup_fns.empty()); MP_LOG(*m_context.loop, Log::Debug) << "Destroying " << CxxTypeName(*this) << " " << this; + // Deregister this object from the connection's live-object tracker, + // through the shared m_server_objects handle since m_context.connection + // may be dangling here (see comment above). Done at the end of the + // destructor so a zero count means destruction fully completed. Note that + // any m_impl destruction scheduled through addAsyncCleanup above is NOT + // covered by the tracker: it runs later on the async cleanup thread, so + // Connection::waitDrained() waits for server call bodies, not for + // m_impl destructors. + m_server_objects->removeServerObject(); } //! If the capnp interface defined a special "destroy" method, as described the @@ -771,9 +927,10 @@ struct ThreadContext //! However, individual ProxyClient objects in the maps will only be //! associated with one event loop and guarded by EventLoop::m_mutex. So //! Waiter::m_mutex does not need to be held while accessing individual - //! ProxyClient instances, and may even need to be released to - //! respect lock order and avoid locking Waiter::m_mutex before - //! EventLoop::m_mutex. + //! ProxyClient instances, and must be released before destroying + //! one from a thread other than the event loop thread, because + //! ~ProxyClient calls EventLoop::sync() (see the blocking rule in + //! the Waiter::m_mutex documentation). ConnThreads callback_threads MP_GUARDED_BY(waiter->m_mutex); //! When client is making a request to a server, this is the `thread` @@ -791,6 +948,11 @@ struct ThreadContext //! to assert false if there's an attempt to execute a blocking operation //! which could deadlock the thread. bool loop_thread = false; + + //! Destructor which destroys the thread maps, coordinating with event + //! loop threads that remove entries from them concurrently when + //! connections are broken (see the code comment). + ~ThreadContext(); }; template @@ -893,10 +1055,17 @@ void _Serve(EventLoop& loop, kj::Own&& stream, InitImpl& init auto it = loop.m_incoming_connections.begin(); MP_LOG(loop, Log::Info) << "IPC server: socket connected."; if (loop.testing_hook_connected) loop.testing_hook_connected(); - it->onDisconnect([&loop, it, on_disconnect = std::forward(on_disconnect)]() mutable { + // Run on_disconnect (e.g. the listener's active-connection counter + // decrement) on any disconnect. It is registered with addSyncCleanup rather + // than placed in the onDisconnect handler below because that handler + // only fires on a remote disconnect and is canceled when a connection is + // closed locally; if on_disconnect lived there, closing a connection + // locally would leave the listener's slot count stuck and stop it from + // accepting again. + it->addSyncCleanup(std::forward(on_disconnect)); + it->onDisconnect([&loop, it]() mutable { MP_LOG(loop, Log::Info) << "IPC server: socket disconnected."; loop.m_incoming_connections.erase(it); - on_disconnect(); if (loop.testing_hook_disconnected) loop.testing_hook_disconnected(); }); } diff --git a/include/mp/proxy.h b/include/mp/proxy.h index 2144b571..b02abfd8 100644 --- a/include/mp/proxy.h +++ b/include/mp/proxy.h @@ -20,6 +20,7 @@ namespace mp { class Connection; class EventLoop; +struct ServerObjectTracker; //! Mapping from capnp interface type to proxy client implementation (specializations are generated by //! proxy-codegen.cpp). template struct ProxyClient; // IWYU pragma: export @@ -172,6 +173,12 @@ struct ProxyServerBase : public virtual Interface_::Server * wrapped. */ std::shared_ptr m_impl; ProxyContext m_context; + //! Live-object tracker shared with this object's Connection, incremented + //! in the constructor and decremented in the destructor so shutdown code + //! can wait for a disconnected connection's server objects to drain. Held + //! via shared_ptr so it remains valid if this object (kept alive by an + //! in-flight call) outlives the Connection. See ServerObjectTracker. + std::shared_ptr m_server_objects; }; //! Customizable (through template specialization) base class which ProxyServer diff --git a/include/mp/type-context.h b/include/mp/type-context.h index 7cfc7e79..5f7e05b2 100644 --- a/include/mp/type-context.h +++ b/include/mp/type-context.h @@ -179,10 +179,10 @@ 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 - // destructor is able to use EventLoop::mutex - // without violating lock order. + // a disconnect. Destroy the node after releasing + // Waiter::m_mutex, so ~ProxyClient does not + // run with the mutex held (see the SetThread + // disconnect callback). ConnThreads::node_type removed; { Lock lock(thread_context.waiter->m_mutex); diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp index eb2aee0c..e59d7693 100644 --- a/src/mp/proxy.cpp +++ b/src/mp/proxy.cpp @@ -13,6 +13,7 @@ #include #include #include // IWYU pragma: keep +#include #include #include #include @@ -111,9 +112,32 @@ Connection::~Connection() noexcept(false) // Connection destructor is always called on the event loop thread. If this // is a local disconnect, it will trigger I/O, so this needs to run on the // event loop thread, and if there was a remote disconnect, this is called - // by an onDisconnect callback directly from the event loop thread. + // by a TwoPartyVatNetwork::onDisconnect callback directly from the event + // loop thread. + assert(std::this_thread::get_id() == m_loop->m_thread_id); + disconnect(); +} + +void Connection::disconnect() +{ + // Disconnecting triggers I/O and tears down capnp state, so it must run on + // the event loop thread, like the destructor. assert(std::this_thread::get_id() == m_loop->m_thread_id); + // m_network is reset at the end of teardown below, so treat it being null + // as the "already disconnected" state: a second call (including the one + // from the destructor) is a no-op. + if (!m_network) return; + + // Expire m_alive before severing the connection below so onDisconnect + // handlers will not trigger and delete this Connection object. The + // onDisconnect handlers trigger on remote disconnects and automatically + // delete Connection objects. But on local disconnects, they should not + // trigger, because local code that disconnects is responsible for freeing + // Connection objects, and it may want to wait for in-flight calls to finish + // before destroying them. + m_alive.reset(); + // Try to cancel any calls that may be executing. m_canceler.cancel("Interrupted by disconnect"); @@ -132,13 +156,13 @@ Connection::~Connection() noexcept(false) // // Sending pending data is important if the connection is a socketpair // because when one side of the socketpair is closed, the other side doesn't - // seem to receive any onDisconnect event. So it is important for the other - // side to instead receive Cap'n Proto "release" messages (see `struct - // Release` in capnp/rpc.capnp) from local Client objects being destroyed so - // the remote side can free resources and shut down cleanly. Without this, - // when one side of a socket pair is closed the other side may not receive - // these messages, preventing the remote side from freeing ProxyServer - // resources and shutting down cleanly. + // seem to receive any TwoPartyVatNetwork::onDisconnect event. So it is + // important for the other side to instead receive Cap'n Proto "release" + // messages (see `struct Release` in capnp/rpc.capnp) from local Client + // objects being destroyed so the remote side can free resources and shut + // down cleanly. Without this, when one side of a socket pair is closed the + // other side may not receive these messages, preventing the remote side + // from freeing ProxyServer resources and shutting down cleanly. // Use kj::runCatchingExceptions instead of try/catch because on macOS with // dynamic libraries, kj::Exception typeinfo differs between libcapnp and // the calling binary, so catch (const kj::Exception&) silently fails to @@ -184,28 +208,49 @@ Connection::~Connection() noexcept(false) // connection implementing the Init interface and handling the Init.makeX() calls. // // Either way when a connection is closed, capnp behavior is to call all - // ProxyServer object destructors first, and then trigger an onDisconnect - // callback. + // ProxyServer object destructors first, and then trigger a + // TwoPartyVatNetwork::onDisconnect callback. // - // On incoming side of the connection, the onDisconnect callback is written - // to delete the Connection object from the m_incoming_connections and call - // this destructor which calls Connection::disconnect. + // On incoming side of the connection, the TwoPartyVatNetwork::onDisconnect + // callback is written to delete the Connection object from the + // m_incoming_connections list and call this destructor. // - // On the outgoing side, the Connection object is owned by top level client - // object client, which onDisconnect handler doesn't have ready access to, - // so onDisconnect handler just calls Connection::disconnect directly - // instead. + // On the outgoing side, the Connection object is heap-allocated and owned + // by a top-level ProxyClient object. In this case, the + // TwoPartyVatNetwork::onDisconnect handler deletes the Connection object + // directly, calling this destructor, and loop below sets Connection + // pointers in all associated ProxyClient objects to null. // // Either way disconnect code runs in the event loop thread and called both // on clean and unclean shutdowns. In unclean shutdown case when the // connection is broken, sync and async cleanup lists will be filled with // callbacks. In the clean shutdown case both lists will be empty. - Lock lock{m_loop->m_mutex}; - while (!m_sync_cleanup_fns.empty()) { - CleanupList fn; - fn.splice(fn.begin(), m_sync_cleanup_fns, m_sync_cleanup_fns.begin()); - Unlock(lock, fn.front()); + { + Lock lock{m_loop->m_mutex}; + while (!m_sync_cleanup_fns.empty()) { + CleanupList fn; + fn.splice(fn.begin(), m_sync_cleanup_fns, m_sync_cleanup_fns.begin()); + Unlock(lock, fn.front()); + } } + + // Release Thread capabilities so idle worker threads are stopped and + // joined at disconnect time, whether or not this object is destroyed right + // away. (A worker thread currently executing a call body is unaffected: + // its ProxyServer object is pinned by the post() call and released + // when the body finishes.) + m_thread_pool.clear(); + m_thread_map = nullptr; + + // Destroy the network and close the stream so the peer observes the + // disconnect, reading EOF and failing its outstanding calls with + // DISCONNECTED errors. This has to be explicit because when disconnect() + // is called without destroying this object, nothing else severs the + // transport: m_rpc_system.reset() above stops reading from the stream but + // does not reliably close it. The network is destroyed first since it + // references the stream. + m_network.reset(); + m_stream = nullptr; } CleanupIt Connection::addSyncCleanup(std::function fn) @@ -218,7 +263,8 @@ CleanupIt Connection::addSyncCleanup(std::function fn) // order should not be significant because the cleanup callbacks run // synchronously in a single batch when the connection is broken, and they // only reset the connection pointers in the client objects without actually - // deleting the client objects. + // deleting the client objects, or update ListenConnections max_connections + // bookkeeping. return m_sync_cleanup_fns.emplace(m_sync_cleanup_fns.begin(), std::move(fn)); } @@ -394,20 +440,26 @@ std::tuple SetThread(GuardedRef threads, Connecti } if (inserted) { thread->second.emplace(make_thread(), connection, /* destroy_connection= */ false); - thread->second->m_disconnect_cb = connection->addSyncCleanup([threads, thread] { - // Note: it is safe to use the `thread` iterator in this cleanup - // function, because the iterator would only be invalid if the map entry - // was removed, and if the map entry is removed the ProxyClient - // destructor unregisters the cleanup. - - // Connection is being destroyed before thread client is, so reset - // thread client m_disconnect_cb member so thread client destructor does not - // try to unregister this callback after connection is destroyed. - thread->second->m_disconnect_cb.reset(); - - // Remove connection pointer about to be destroyed from the map - const Lock lock(threads.mutex); - threads.ref.erase(thread); + thread->second->m_disconnect_cb = connection->addSyncCleanup([threads, connection] { + // Remove and destroy this connection's map entry, unless the + // thread owning the map is exiting and ~ThreadContext already took + // the entry, in which case that thread destroys it. Look the entry + // up by key rather than capturing the iterator, which would be + // invalid in that case. + ConnThreads::node_type removed; + { + const Lock lock(threads.mutex); + auto it = threads.ref.find(connection); + if (it == threads.ref.end()) return; + // Reset so ~ProxyClient does not try to unregister + // this callback, which is already running. + it->second->m_disconnect_cb.reset(); + removed = threads.ref.extract(it); + } + // The entry is destroyed here with Waiter::m_mutex released, as in + // ~ThreadContext. (Holding it would be safe on the event loop + // thread, where EventLoop::sync() runs directly, but releasing it + // keeps the two consistent.) }); } return {thread, inserted}; @@ -415,6 +467,9 @@ std::tuple SetThread(GuardedRef threads, Connecti ProxyClient::~ProxyClient() { + EventLoop& loop{*m_context.loop}; + if (loop.testing_hook_thread_client_destroy) loop.testing_hook_thread_client_destroy(this); + // If thread is being destroyed before connection is destroyed, remove the // cleanup callback that was registered to handle the connection being // destroyed before the thread being destroyed. @@ -424,13 +479,43 @@ ProxyClient::~ProxyClient() // between this thread trying to remove the callback and the disconnect // handler attempting to call it. m_context.loop->sync([&]() { - if (m_disconnect_cb) { + // Skip if the connection was disconnected while this thread waited + // for the event loop: Connection::disconnect() has already run and + // freed the cleanup list m_disconnect_cb points into, and the + // ProxyClientBase disconnect callback has nulled + // m_context.connection. (The SetThread callback resets + // m_disconnect_cb only when it finds this object in the thread + // map, so if ~ThreadContext took the entry first, m_disconnect_cb + // is still set here.) + if (m_disconnect_cb && m_context.connection) { m_context.connection->removeSyncCleanup(*m_disconnect_cb); } }); } } +ThreadContext::~ThreadContext() +{ + // Server threads created by ProxyServer::makeThread have no + // waiter here: ~ProxyServer moves it away and clears the maps + // before the thread exits. + if (!waiter) return; + + // Take the thread client maps under Waiter::m_mutex, because the SetThread + // disconnect callback removes entries from them concurrently on the event + // loop thread when connections are broken, and whichever side removes an + // entry destroys it. Destroy the maps with the mutex released, because + // ~ProxyClient blocks in EventLoop::sync() and the event loop + // thread locks Waiter::m_mutex itself (see the blocking rule in the + // Waiter::m_mutex documentation). + ConnThreads request, callback; + { + const Lock lock(waiter->m_mutex); + request.swap(request_threads); + callback.swap(callback_threads); + } +} + ProxyServer::ProxyServer(Connection& connection, ThreadContext& thread_context, std::thread&& thread) : m_loop{*connection.m_loop}, m_thread_context(thread_context), m_thread(std::move(thread)) { diff --git a/test/mp/test/connect_tests.cpp b/test/mp/test/connect_tests.cpp index 4e4f051f..7f522ad6 100644 --- a/test/mp/test/connect_tests.cpp +++ b/test/mp/test/connect_tests.cpp @@ -118,11 +118,11 @@ KJ_TEST("ConnectStream defers disconnect failure to the first IPC request for in } catch (const std::runtime_error& e) { std::string_view reason = e.what(); - // There is a race between the event loop detecting the disconnect - // and foo->add() being called. If the onDisconnect callback fires - // first and nulls m_context.connection, the error is "called after - // disconnect"; if foo->add() submits before the callback fires, - // the error is "interrupted by disconnect". + // There is a race between the event loop detecting the disconnect and + // foo->add() being called. If the TwoPartyVatNetwork::onDisconnect + // callback fires first and nulls m_context.connection, the error is + // "called after disconnect"; if foo->add() submits before the callback + // fires, the error is "interrupted by disconnect". KJ_EXPECT(reason == "IPC client method called after disconnect." || reason == "IPC client method call interrupted by disconnect."); } diff --git a/test/mp/test/listen_tests.cpp b/test/mp/test/listen_tests.cpp index 8d06cb35..e7c98e6b 100644 --- a/test/mp/test/listen_tests.cpp +++ b/test/mp/test/listen_tests.cpp @@ -18,6 +18,7 @@ #include #include #include +#include #include #include #include @@ -192,6 +193,34 @@ KJ_TEST("ListenConnections enforces a local connection limit") KJ_EXPECT(client3->client->add(3, 4) == 7); } +KJ_TEST("ListenConnections resumes after a local disconnect") +{ + // A connection closed locally (here by erasing it from the incoming- + // connection list) must still free the listener's slot so the listener + // resumes accepting. The counter decrement runs as an addSyncCleanup cleanup; + // if it were in the onDisconnect handler it would be canceled by the + // local close and the slot would stay stuck. + ListenSetup server(/*max_connections=*/1); + + auto client1 = std::make_unique(server.listener.MakeConnectedSocket()); + server.WaitForConnectedCount(1); + KJ_EXPECT(client1->client->add(1, 2) == 3); + + auto client2 = std::make_unique(server.listener.MakeConnectedSocket()); + (**server.m_loop_ref).sync([] {}); + KJ_EXPECT(server.ConnectedCount() == 1); + + // Close the first connection locally on the event loop thread. + EventLoop& loop{**server.m_loop_ref}; + loop.sync([&] { + KJ_REQUIRE(loop.m_incoming_connections.size() == 1); + loop.m_incoming_connections.pop_front(); + }); + + server.WaitForConnectedCount(2); + KJ_EXPECT(client2->client->add(2, 3) == 5); +} + KJ_TEST("ListenConnections accepts multiple connections") { // With max-connections=2, two clients should be accepted and usable at the diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index 5ecb7cc4..de1f1604 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -419,7 +420,8 @@ KJ_TEST("Calling async IPC method with a remote disconnect while results are bui // 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. + // internal teardown, which runs before any TwoPartyVatNetwork::onDisconnect + // notification. TestSetup setup{/*client_owns_connection=*/false}; ProxyClient* foo = setup.client.get(); @@ -498,6 +500,54 @@ KJ_TEST("Worker thread destroyed before it is initialized") EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); } +KJ_TEST("Thread exiting while its connection is destroyed") +{ + // Regression test for a race between a thread exiting after making IPC + // calls and its connection being destroyed on the event loop thread. + // ~ThreadContext on the exiting thread and the SetThread disconnect + // callback run by ~Connection both remove the thread's map entries for + // the connection, and previously nothing synchronized them, so both could + // destroy the same ProxyClient object. + // + // The testing_hook_thread_client_destroy hook, called at the start of + // ~ProxyClient, blocks the exiting thread inside its first map + // entry destructor while the main thread destroys the connection. The + // disconnect callback must leave that entry alone: it resets + // m_disconnect_cb only when it finds the entry in the map and takes over + // destroying it, so the entry's m_disconnect_cb must still be set when + // the exiting thread resumes. + TestSetup setup{/*client_owns_connection=*/false}; + ProxyClient* foo = setup.client.get(); + foo->initThreadMap(); + setup.server->m_impl->m_fn = [] {}; + EventLoop& loop = *foo->m_context.loop; + + std::promise caller_exiting, release_caller; + bool caller_blocked{false}; // caller thread only + bool disconnect_cb_set{false}; // caller thread, read after join + loop.testing_hook_thread_client_destroy = [&](ProxyClient* client) { + // The hook also runs on the event loop thread for entries the + // disconnect callback destroys. Block only the exiting caller thread, + // in the first destructor it runs. + if (std::this_thread::get_id() == loop.m_thread_id || caller_blocked) return; + caller_blocked = true; + caller_exiting.set_value(); + release_caller.get_future().get(); + disconnect_cb_set = client->m_disconnect_cb.has_value(); + }; + + // Make a call taking an mp.Context argument, which adds callback and + // request thread entries for the connection to the caller's thread-local + // ThreadContext maps. They are destroyed when the thread exits. + std::thread caller{[&] { foo->callFnAsync(); }}; + caller_exiting.get_future().get(); + setup.client_disconnect(); + release_caller.set_value(); + caller.join(); + loop.testing_hook_thread_client_destroy = nullptr; + KJ_EXPECT(disconnect_cb_set); +} + KJ_TEST("Calling async IPC method, with server disconnect racing the call") { // Regression test for bitcoin/bitcoin#34777 heap-use-after-free where @@ -547,6 +597,81 @@ KJ_TEST("Calling async IPC method, with server disconnect after cleanup") EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); } +KJ_TEST("Waiting for in-flight server call to finish after disconnect") +{ + // Regression test for bitcoin/bitcoin#35845. Disconnecting a connection + // cancels the KJ promise of an in-flight call, but a C++ server method + // body already dispatched to a worker thread runs to completion. Verify + // that Connection::waitDrained() blocks until such a body finishes and its + // server object is destroyed, so shutdown code can wait for a disconnected + // connection to become quiescent before freeing state the body accesses. + + std::promise body_started, release_body; + TestSetup setup; + ProxyClient* foo = setup.client.get(); + foo->initThreadMap(); + + // A server call body that signals when it starts and then blocks until the + // test releases it, so the in-flight state can be observed + // deterministically. + setup.server->m_impl->m_fn = [&] { + body_started.set_value(); + release_body.get_future().get(); + }; + + // Grab the server Connection object on the event loop thread before + // disconnecting. It stays valid until server_disconnect() destroys it + // below. + Connection* connection{nullptr}; + foo->m_context.loop->sync([&] { connection = setup.server->m_context.connection; }); + + // Invoke the async method on a separate thread so its body blocks there + // while this thread makes assertions. callFnAsync() takes an mp.Context, + // so its body runs on a worker thread via ProxyServer::post(). + std::thread call_thread([&] { + EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); + }); + body_started.get_future().get(); + + // The FooInterface server object is the connection's only counted server + // object, and its call body is executing. + KJ_EXPECT(connection->tracker()->pendingServerObjects() == 1); + + // Disconnect. This cancels the call's promise (the client above sees the + // disconnect error), but the body is still blocked on the worker thread, + // so its server object must still be alive. + foo->m_context.loop->sync([&] { connection->disconnect(); }); + KJ_EXPECT(connection->tracker()->pendingServerObjects() == 1); + + // A drain must block while the body runs and return only once it + // finishes, which is what Ipc::disconnectIncoming relies on during + // shutdown. + std::promise drain_waiting; + connection->tracker()->testing_hook_wait = [&] { drain_waiting.set_value(); }; + std::atomic drained{false}; + std::thread drain_thread([&] { + connection->tracker()->waitDrained(); + drained = true; + }); + + // Wait until waitDrained() has observed the live server object and is + // about to block, then verify it does not return while the body is blocked. + drain_waiting.get_future().get(); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + KJ_EXPECT(!drained); + + // Let the body finish; the drain should now complete. + release_body.set_value(); + drain_thread.join(); + KJ_EXPECT(drained); + KJ_EXPECT(connection->tracker()->pendingServerObjects() == 0); + call_thread.join(); + + // Destroy the drained connection. (~Connection notices disconnect() has + // already run and does not tear things down twice.) + setup.server_disconnect(); +} + KJ_TEST("Destroying ProxyClient<> with destroy method after peer disconnect") { // Regression test for bitcoin-core/libmultiprocess#219 where @@ -734,6 +859,85 @@ KJ_TEST("Async cleanup thread has OS thread name") } #endif // HAVE_PTHREAD_GETNAME_NP +KJ_TEST("onDisconnect handler does not run after the connection is destroyed") +{ + // Regression test for a race condition where an onDisconnect handler + // could fire even after the connection had already been disconnected + // locally. Local disconnects are supposed to preempt onDisconnect + // handlers so they are able to free Connection objects without risking double + // deletions. + // + // To hit the previous race window deterministically without relying on + // socket timing, this test exploits two ordering facts: + // m_network.onDisconnect() hands out branches of a forked promise that fire + // in the order they were added, and kj::evalLater tasks queued during one + // event loop turn run on the next turn in FIFO order. A "destroyer" branch + // is registered before the handler under test, so when the peer disconnects + // the destroyer fires first and queues an evalLater task that destroys the + // Connection ahead of the handler. The handler records whether it ran after + // the Connection was destroyed, reading only test-owned flags (never the + // freed Connection), so a violation is a deterministic wrong result rather + // than a flaky crash. + // + // The Connection must be destroyed from that separate task, not from inside + // an onDisconnect continuation: destroying it there tears down m_network + // while a continuation of m_network's promise is still running, and crashes. + + std::atomic connection_destroyed{false}; + std::atomic handler_ran{false}; + std::atomic handler_ran_after_destroy{false}; + + std::thread thread{[&] { + EventLoop loop("mptest", [](mp::LogMessage log) { + KJ_LOG(INFO, log.level, log.message); + if (log.level == mp::Log::Raise) throw std::runtime_error(log.message); + }); + auto pipe = loop.m_io_context.provider->newTwoWayPipe(); + + // Server-side connection whose onDisconnect handler is under test. + auto server_conn = std::make_unique( + loop, kj::mv(pipe.ends[0]), [&](Connection& connection) { + return ::capnp::Capability::Client(kj::heap>( + std::make_shared(), connection)); + }); + + // Raw peer end of the pipe. Closing it makes the server connection's + // m_network.onDisconnect() resolve, i.e. a real remote disconnect. + kj::Own peer_end = kj::mv(pipe.ends[1]); + + // Destroyer branch, registered BEFORE the handler under test so it + // fires first and its destroy task E_d is queued before the buggy + // bounced handler task E_f. It schedules the destroy on a *separate* + // task, so the connection is not torn down from inside an onDisconnect + // continuation. + loop.m_task_set->add(server_conn->m_network->onDisconnect().then([&] { + loop.m_task_set->add(kj::evalLater([&] { + server_conn.reset(); + connection_destroyed = true; + })); + })); + + // Handler under test. It intentionally does not dereference the + // Connection; it only records whether it ran after the connection was + // destroyed, so the result is deterministic. + server_conn->onDisconnect([&] { + handler_ran = true; + handler_ran_after_destroy = connection_destroyed.load(); + }); + + // Trigger the remote disconnect once the loop is running. + loop.m_task_set->add(kj::evalLater([&] { peer_end = nullptr; })); + + loop.loop(); + }}; + thread.join(); + + // The handler must have run (both versions schedule it), but it must never + // run after the connection was destroyed. + KJ_EXPECT(handler_ran); + KJ_EXPECT(!handler_ran_after_destroy); +} + KJ_TEST("Call async IPC method without thread or pool errors correctly") { TestSetup setup;