diff --git a/.github/workflows/bitcoin-core-ci.yml b/.github/workflows/bitcoin-core-ci.yml index f14d5682..8d8b7d05 100644 --- a/.github/workflows/bitcoin-core-ci.yml +++ b/.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/ 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..05c74d30 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -23,6 +23,7 @@ #include #include #include +#include #include #include #include @@ -85,14 +86,16 @@ struct ProxyClient : public ProxyClientBase ~ProxyClient(); //! Reference to callback function that is run if there is a sudden - //! disconnect and the Connection object is destroyed before this - //! ProxyClient object. The callback will destroy this object and - //! remove its entry from the thread's request_threads or callback_threads - //! map. It will also reset m_disconnect_cb so the destructor does not - //! access it. In the normal case where there is no sudden disconnect, the - //! destructor will unregister m_disconnect_cb so the callback is never run. - //! Since this variable is accessed from multiple threads, accesses should - //! be guarded with the associated Waiter::m_mutex. + //! disconnect and the Connection is disconnected before this + //! ProxyClient object is destroyed. The callback will destroy this + //! object and remove its entry from the thread's request_threads or + //! callback_threads map (see SetThread, which explains why removal must + //! happen eagerly at disconnect time). It will also reset m_disconnect_cb + //! so the destructor does not access it. In the normal case where there is + //! no sudden disconnect, the destructor will unregister m_disconnect_cb so + //! the callback is never run. Since this variable is accessed from + //! multiple threads, accesses should be guarded with the associated + //! Waiter::m_mutex. std::optional m_disconnect_cb; }; @@ -299,6 +302,15 @@ class EventLoop //! Check if loop should exit. bool done() const MP_REQUIRES(m_mutex); + //! Type of m_incoming_connections list. + using Connections = std::list>; + + //! View of incoming connections yielding Connection& for each entry. + auto incomingConnections() + { + return m_incoming_connections | std::views::transform([](auto& ptr) -> Connection& { return *ptr; }); + } + //! Process name included in thread names so combined debug output from //! multiple processes is easier to understand. const char* m_exe_name; @@ -345,8 +357,11 @@ class EventLoop //! Capnp list of pending promises. std::unique_ptr m_task_set; - //! List of connections. - std::list m_incoming_connections; + //! List of connections. Holds one shared_ptr reference per connection; + //! proxy objects created for a connection hold additional references (see + //! ProxyContext::connection), so erasing a connection from this list does + //! not necessarily destroy it. + Connections m_incoming_connections; //! Logging options LogOptions m_log_opts; @@ -376,6 +391,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 +451,248 @@ 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 can be a plain Connection member because proxy objects hold +//! shared ownership of their Connection (see ProxyContext::connection), so a +//! ProxyServer object kept alive by an in-flight call cannot outlive the +//! Connection, and its destructor can always safely reach the counter through +//! m_context.connection. +//! +//! 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. +//! connection, or an outgoing client connection. It must be created on the +//! event loop thread with the make() factory function, which returns a +//! shared_ptr owner whose custom deleter destroys the object on the event loop +//! thread. Proxy objects created for the connection share ownership of it (see +//! ProxyContext::connection), so the Connection is guaranteed to outlive them +//! and stays valid -- as a disconnected husk -- even after disconnect(). //! In addition to Cap'n Proto state, it also holds lists of callbacks to run //! when the connection is closed. -class Connection +//! +//! 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 ServerObjectTracker, public std::enable_shared_from_this { 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)) {} - 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). - ~Connection() noexcept(false); + //! Create a Connection with shared ownership. Connection objects must be + //! owned by shared_ptr because proxy objects created for the connection + //! take shared ownership of it in their ProxyContext (via + //! shared_from_this), which requires an existing shared_ptr owner. The + //! custom deleter runs the destructor on the event loop thread (running + //! it directly if the last reference is dropped on the event loop thread, + //! posting to the loop otherwise), which ~Connection requires. The + //! connection's own EventLoopRef keeps the loop running until the deleter + //! has run. + //! + //! Note: dropping shared_ptr references alone will not usually destroy a + //! connected Connection, because connected state holds capability + //! references to server objects which themselves hold references back to + //! the Connection (m_rpc_system exports -> ProxyServer objects -> + //! ProxyContext::connection). disconnect() breaks these cycles, so every + //! code path that tears down a connection must call it; after that, the + //! object is destroyed when the last reference is dropped. + template + static std::shared_ptr make(Args&&... args) + { + return {new Connection(std::forward(args)...), [](Connection* connection) { + EventLoop& loop{*connection->m_loop}; + loop.sync([&] { delete connection; }); + }}; + } - //! 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. - CleanupIt addSyncCleanup(std::function fn); - void removeSyncCleanup(CleanupIt it); + //! Start serving RPC requests on the connection, handling them with server + //! objects created by the make_client callback. Called (for server-side + //! connections) after make(), not during construction, because the server + //! objects created by the callback take shared ownership of this + //! connection via shared_from_this(), which requires the shared_ptr owner + //! returned by make() to already exist. + void serve(const std::function<::capnp::Capability::Client(Connection&)>& make_client) + { + assert(!m_rpc_system); + m_rpc_system.emplace(::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); - //! Add disconnect handler. + //! 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(); + + //! True once disconnect() has run (locally, or as part of destruction), + //! i.e. the connection has been severed on this side. False while the + //! connection is still connected, including after a remote disconnect that + //! this side has not yet reacted to. + bool disconnected() const { return !m_network; } + + //! 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 shared_from_this(); } + + //! 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). + //! Runs inside disconnect() as part of teardown. Returns a handle that can + //! be passed to cancelOnDisconnect() to unregister the function before it + //! runs. + //! + //! The only current use is by SetThread, which registers callbacks that + //! eagerly remove this connection's entries from per-thread connection maps + //! (ProxyClient objects owned by other threads, see ThreadContext). + //! Eager removal is required because the owning threads may never touch + //! their maps again, and a surviving entry would hold this Connection object + //! -- and through its EventLoopRef the event loop -- alive indefinitely. + //! Other proxy objects need no disconnect notification (see disconnect()). + CleanupIt onDisconnect(std::function fn); + void cancelOnDisconnect(CleanupIt it); + + //! Register a handler for when the capnp network reports the connection + //! disconnected -- which happens on both a remote disconnect (the peer + //! closes it) and a local disconnect() on this side. The handler runs + //! deferred, on a later event loop turn. + //! + //! It is passed a Connection* that is either the still-live connection (safe + //! to use for the duration of the call) or null if the Connection was + //! already destroyed before the handler ran, which is the case when a + //! connection is torn down and freed locally. Handlers must decide + //! explicitly what to do in each case. + //! + //! Unless the event loop exits early, this callback will always be called + //! after disconnecting. So it is possible to use this callback to free the + //! last reference to the EventLoop, and exit after the last disconnect. template - void onDisconnect(F&& f) + void afterDisconnect(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 on both remote and local disconnects + // (disconnecting resets m_rpc_system, which drops capnp's last + // reference to the network and fulfills the promise). Lock a weak_ptr to + // this Connection and pass the raw pointer to f: it is null if the + // object was already destroyed, and while non-null the temporary + // shared_ptr from lock() keeps it alive for the duration of the call. + m_loop->m_task_set->add(m_network->onDisconnect().then( + [f = std::forward(f), weak = weak_from_this()]() mutable { + f(weak.lock().get()); + })); } 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; + //! 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(). + ServerObjectTracker m_server_objects; + std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId>> m_rpc_system; // ThreadMap interface client, used to create a remote server thread when an @@ -514,10 +715,25 @@ class Connection //! still executing at time of disconnection. kj::Canceler m_canceler; - //! Cleanup functions to run if connection is broken unexpectedly. List - //! will be empty if all ProxyClient are destroyed cleanly before the - //! connection is destroyed. + //! Cleanup functions to run when the connection is disconnected, removing + //! this connection's entries from per-thread connection maps. See + //! addSyncCleanup. List will be empty if all ProxyClient objects + //! are destroyed cleanly before the connection is disconnected. CleanupList m_sync_cleanup_fns; + +private: + //! Construct a client-side connection. Private; use make(). + Connection(EventLoop& loop, kj::Own&& stream_) + : Connection(loop, kj::mv(stream_), ::capnp::rpc::twoparty::Side::CLIENT) + { + m_rpc_system.emplace(::capnp::makeRpcClient(*m_network)); + } + //! Construct a connection for the given side without starting the RPC + //! system. Server-side connections start it with serve() after make() + //! returns. Private; use make(). + Connection(EventLoop& loop, kj::Own&& stream_, ::capnp::rpc::twoparty::Side side) + : m_loop(loop), m_stream(kj::mv(stream_)), + m_network(std::in_place, *m_stream, side, ::capnp::ReaderOptions()) {} }; //! Vat id for server side of connection. Required argument to RpcSystem::bootStrap() @@ -544,60 +760,45 @@ ProxyClientBase::ProxyClientBase(typename Interface::Client cli { MP_LOG(*m_context.loop, Log::Debug) << "Creating " << CxxTypeName(*this) << " " << this; - // Handler for the connection getting destroyed before this client object. - auto disconnect_cb = m_context.connection->addSyncCleanup([this]() { - // Release client capability by move-assigning to temporary. - { - typename Interface::Client(std::move(m_client)); - } - Lock lock{m_context.loop->m_mutex}; - m_context.connection = nullptr; - }); - - // Two shutdown sequences are supported: - // - // - A normal sequence where client proxy objects are deleted by external - // code that no longer needs them - // - // - A garbage collection sequence where the connection or event loop shuts - // down while external code is still holding client references. - // - // The first case is handled here when m_context.connection is not null. The - // second case is handled by the disconnect_cb function, which sets - // m_context.connection to null so nothing happens here. - m_context.cleanup_fns.emplace_front([this, destroy_connection, disconnect_cb]{ - { + // Cleanup function that runs when this object is destroyed. Unlike server + // objects, client objects are owned by application code, which can keep + // them alive arbitrarily long after a disconnect, but this needs no + // special handling: m_context holds shared ownership of the connection, + // so the Connection object is guaranteed to still exist here, and the + // m_client capability handle is safe to keep across a disconnect (Cap'n + // Proto keeps handles valid after the connection state is torn down; + // using them just fails with DISCONNECTED errors). The handle only needs + // to be released on the event loop thread, done in the sync() call below, + // because capability reference counts are not thread safe. + m_context.cleanup_fns.emplace_front([this, destroy_connection]{ // If the capnp interface defines a destroy method, call it to destroy // the remote object, waiting for it to be deleted server side. If the // capnp interface does not define a destroy method, this will just call // an empty stub defined in the ProxyClientBase class and do nothing. // Exceptions are caught and logged rather than propagated because - // ~ProxyClientBase is noexcept and the peer may be gone by the time - // this runs. + // ~ProxyClientBase is noexcept and the connection may have been + // disconnected. (In that case clientInvoke fails with "IPC client + // method called after disconnect", caught and logged here.) if (kj::runCatchingExceptions([&]{ Sub::destroy(*this); }) != nullptr) { MP_LOG(*m_context.loop, Log::Warning) << "Remote destroy call failed during cleanup. Continuing."; } - // FIXME: Could just invoke removed addCleanup fn here instead of duplicating code m_context.loop->sync([&]() { - // 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 - // before the removeSyncCleanup call. - if (m_context.connection) m_context.connection->removeSyncCleanup(disconnect_cb); - // Release client capability by move-assigning to temporary. { typename Interface::Client(std::move(m_client)); } if (destroy_connection) { - delete m_context.connection; - m_context.connection = nullptr; + // This client owns the connection: disconnect it and drop this + // object's reference while on the event loop thread, destroying + // the Connection here unless other proxy objects still reference + // it (in which case it is destroyed when the last of them is). + // m_context.connection is still valid here -- this object shares + // ownership of it and nothing resets the pointer on disconnect. + m_context.connection->disconnect(); + m_context.connection.reset(); } }); - } }); // If construct() fails, run the cleanup functions before rethrowing, // because ~ProxyClientBase will not run for an object whose constructor @@ -610,14 +811,18 @@ ProxyClientBase::ProxyClientBase(typename Interface::Client cli throw; } - // If this client owns the connection, delete the connection on disconnect. + // If this client owns the connection, register a handler to process a + // remote disconnect and shut down the connection. if (destroy_connection) { m_context.loop->sync([&] { EventLoop& loop = *m_context.loop; - Connection* connection = m_context.connection; - connection->onDisconnect([&loop, connection] { + m_context.connection->afterDisconnect([&loop](Connection* conn) { + // If conn is null or disconnected() is true, a disconnect was + // already trigged locally, so there is nothing to do and an + // unexpected network disconnect message should not be logged. + if (!conn || conn->disconnected()) return; MP_LOG(loop, Log::Warning) << "IPC client: unexpected network disconnect."; - delete connection; + conn->disconnect(); }); }); } @@ -635,6 +840,12 @@ template ProxyServerBase::ProxyServerBase(std::shared_ptr impl, Connection& connection) : m_impl(std::move(impl)), m_context(&connection) { + // 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. + connection.addServerObject(); MP_LOG(*m_context.loop, Log::Debug) << "Creating " << CxxTypeName(*this) << " " << this; assert(m_impl); } @@ -642,15 +853,13 @@ ProxyServerBase::ProxyServerBase(std::shared_ptr impl, Co //! ProxyServer destructor, called from the EventLoop thread by Cap'n Proto //! garbage collection code after there are no more references to this object. //! This will typically happen when the corresponding ProxyClient object on the -//! other side of the connection is destroyed. It can also happen earlier if the -//! connection is broken or destroyed. In the latter case this destructor will -//! typically be called inside m_rpc_system.reset() call in the ~Connection -//! destructor while the Connection object still exists. However, because -//! ProxyServer objects are refcounted, and the Connection object could be -//! destroyed while asynchronous IPC calls are still in-flight, it's possible -//! for this destructor to be called after the Connection object no longer -//! exists, so it is NOT valid to dereference the m_context.connection pointer -//! from this function. +//! other side of the connection is destroyed. It can also happen earlier if +//! the connection is broken or disconnected, in which case this destructor is +//! typically called inside the m_rpc_system.reset() call in +//! Connection::disconnect(). If an asynchronous IPC call is still in-flight at +//! disconnect time, Cap'n Proto keeps this object alive and this destructor +//! runs later, when the call body finishes; m_context.connection remains valid +//! even then, because m_context holds shared ownership of the Connection. template ProxyServerBase::~ProxyServerBase() { @@ -682,6 +891,14 @@ 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 + // (m_context.connection is always valid 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_context.connection->removeServerObject(); } //! If the capnp interface defined a special "destroy" method, as described the @@ -771,9 +988,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 +1009,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 @@ -864,41 +1087,101 @@ kj::Promise ProxyServer::post(Fn&& fn) //! If the init interface declares a construct() method, creating the client //! calls it, so this function may block making an IPC call and may throw if //! the call fails. +//! +//! If destroy_connection is false, the returned client does not take +//! ownership of the connection, and the caller is responsible for calling +//! disconnect() on it (accessible as client->m_context.connection) whenever +//! it is done with it. template -std::unique_ptr> ConnectStream(EventLoop& loop, Stream stream) +std::unique_ptr> ConnectStream(EventLoop& loop, Stream stream, bool destroy_connection = true) { typename InitInterface::Client init_client(nullptr); - std::unique_ptr connection; + std::shared_ptr connection; loop.sync([&] { - connection = std::make_unique(loop, kj::mv(stream)); + connection = Connection::make(loop, kj::mv(stream)); init_client = connection->m_rpc_system->bootstrap(ServerVatId().vat_id).castAs(); }); - return std::make_unique>( - kj::mv(init_client), connection.release(), /* destroy_connection= */ true); + // The local `connection` reference is dropped when this function returns; + // the returned ProxyClient keeps the connection alive (see + // ProxyContext::connection) and disconnects it on destruction if + // destroy_connection is true. + return std::make_unique>(kj::mv(init_client), connection.get(), destroy_connection); } //! Given stream and init objects, construct a new ProxyServer object that //! handles requests from the stream by calling the init object. Embed the -//! ProxyServer in a Connection object that is stored and erased if -//! disconnected. This should be called from the event loop thread. +//! ProxyServer in a Connection object that is stored in +//! loop.m_incoming_connections. This should be called from the event loop +//! thread. Returns the new ProxyServer along with a shared_ptr to its +//! Connection. +//! +//! If destroy_connection is false, the connection is not disconnected and +//! removed automatically when the peer disconnects, and the caller is +//! responsible for calling disconnect() and removing it from +//! loop.m_incoming_connections (by value, not iterator -- see the comment +//! below) whenever it is done with the connection. template -void _Serve(EventLoop& loop, kj::Own&& stream, InitImpl& init, OnDisconnect&& on_disconnect) +std::pair*, std::shared_ptr> _Serve(EventLoop& loop, + kj::Own&& stream, + std::shared_ptr init, + OnDisconnect&& on_disconnect, + bool destroy_connection = true) { - loop.m_incoming_connections.emplace_front(loop, kj::mv(stream), [&](Connection& connection) { - // Disable deleter so proxy server object doesn't attempt to delete the - // init implementation when the proxy client is destroyed or - // disconnected. - return kj::heap>(std::shared_ptr(&init, [](InitImpl*){}), connection); + ProxyServer* server = nullptr; + auto connection{Connection::make(loop, kj::mv(stream), ::capnp::rpc::twoparty::Side::SERVER)}; + loop.m_incoming_connections.emplace_front(connection); + connection->serve([&](Connection& connection) { + auto proxy_server = kj::heap>(std::move(init), connection); + server = proxy_server; + return capnp::Capability::Client(kj::mv(proxy_server)); }); - 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) as a sync cleanup, so it runs synchronously on every + // disconnect. The afterDisconnect handler below is deferred, so putting the + // decrement there could leave the listener's slot count stuck on a local + // disconnect() and stop it accepting. + connection->onDisconnect(std::forward(on_disconnect)); + connection->afterDisconnect([&loop, destroy_connection](Connection* conn) { + // If conn is null or disconnected() is true, a disconnect has been + // triggered locally and the code which triggered it is responsible for + // tearing down the Connection object. Otherwise the remote peer + // disconnected, and if destroy_connection is true, the connection is + // still on the incoming-connection list and must be disconnected and + // removed. + if (!conn || conn->disconnected()) return; MP_LOG(loop, Log::Info) << "IPC server: socket disconnected."; - loop.m_incoming_connections.erase(it); - on_disconnect(); + if (destroy_connection) { + // Remove by value, not through a captured iterator: other code may + // have reordered the list between this handler being queued and + // running. + conn->disconnect(); + loop.m_incoming_connections.remove(conn->shared_from_this()); + } if (loop.testing_hook_disconnected) loop.testing_hook_disconnected(); }); + return {server, connection}; +} + +//! Overload of _Serve that takes a reference to the init object instead of a +//! shared_ptr. ProxyServer objects use shared_ptr's internally to optionally +//! take ownership of interfaces being served, and free them when clients are +//! no longer using them. Some libmultiprocess callers take advantage of this +//! and pass shared_ptr's directly. But other callers that do not give away +//! ownership are not required to use shared_ptr, so this overload converts +//! references they pass into shared_ptr's with empty deleters. This prevents +//! the ProxyServer object from deleting the init object when the client is +//! disconnected. +template +std::pair*, std::shared_ptr> _Serve(EventLoop& loop, + kj::Own&& stream, + InitImpl& init, + OnDisconnect&& on_disconnect, + bool destroy_connection = true) +{ + return _Serve(loop, kj::mv(stream), std::shared_ptr(&init, [](InitImpl*){}), + std::forward(on_disconnect), destroy_connection); } struct Listener @@ -936,11 +1219,13 @@ void _Listen(const std::shared_ptr& listener, EventLoop& loop, InitImp } //! Given a stream and an init object, handle requests on the stream by calling -//! methods on the Init object. +//! methods on the Init object. See _Serve for details on the return value and +//! the destroy_connection parameter. template -void ServeStream(EventLoop& loop, Stream stream, InitImpl& init) +std::pair*, std::shared_ptr> ServeStream( + EventLoop& loop, Stream stream, InitImpl&& init, bool destroy_connection = true) { - _Serve(loop, kj::mv(stream), init, [] {}); + return _Serve(loop, kj::mv(stream), std::forward(init), /*on_disconnect=*/ [] {}, destroy_connection); } //! Given listening socket identifier and an init object, handle incoming diff --git a/include/mp/proxy-types.h b/include/mp/proxy-types.h index 3a5dd3d8..3c5e0da4 100644 --- a/include/mp/proxy-types.h +++ b/include/mp/proxy-types.h @@ -345,7 +345,7 @@ auto PassField(Priority<1>, TypeList, ServerContext& server_context, const auto& params = server_context.call_context.getParams(); const auto& input = Make(params); using Interface = typename Decay::Calls; - auto param = std::make_unique>(input.get(), server_context.proxy_server.m_context.connection, false); + auto param = std::make_unique>(input.get(), server_context.proxy_server.m_context.connection.get(), false); fn.invoke(server_context, std::forward(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."; diff --git a/include/mp/proxy.h b/include/mp/proxy.h index 2144b571..cac7c189 100644 --- a/include/mp/proxy.h +++ b/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; EventLoopRef loop; CleanupList cleanup_fns; diff --git a/include/mp/type-context.h b/include/mp/type-context.h index 7cfc7e79..503a016a 100644 --- a/include/mp/type-context.h +++ b/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 - // 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); - 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) diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp index eb2aee0c..ce3e28cf 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 @@ -104,15 +105,29 @@ void EventLoopRef::reset(bool relock) MP_NO_TSA } } -ProxyContext::ProxyContext(Connection* connection) : connection(connection), loop{*connection->m_loop} {} +ProxyContext::ProxyContext(Connection* connection) : connection(connection->shared_from_this()), loop{*connection->m_loop} {} 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; // Try to cancel any calls that may be executing. m_canceler.cancel("Interrupted by disconnect"); @@ -132,13 +147,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 @@ -158,71 +173,69 @@ Connection::~Connection() noexcept(false) } } - // ProxyClient cleanup handlers are in sync list, and ProxyServer cleanup - // handlers are in the async list. - // - // The ProxyClient cleanup handlers are synchronous because they are fast - // and don't do anything besides release capnp resources and reset state so - // future calls to client methods immediately throw exceptions instead of - // trying to communicate across the socket. The synchronous callbacks set - // ProxyClient capability pointers to null, so new method calls on client - // objects fail without triggering i/o or relying on event loop which may go - // out of scope or trigger obscure capnp i/o errors. - // - // The ProxyServer cleanup handlers call user defined destructors on the server - // object, which can run arbitrary blocking bitcoin code so they have to run - // asynchronously in a different thread. The asynchronous cleanup functions - // intentionally aren't started until after the synchronous cleanup - // functions run, so client objects are fully disconnected before bitcoin - // code in the destructors are run. This way if the bitcoin code tries to - // make client requests the requests will just fail immediately instead of - // sending i/o or accessing the event loop. - // - // The context where Connection objects are destroyed and this destructor is invoked - // is different depending on whether this is an outgoing connection being used - // to make an Init.makeX call() (e.g. Init.makeNode or Init.makeWalletClient) or an incoming - // 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. - // - // 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 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. + // Run cleanup functions registered with addSyncCleanup(). These remove + // this connection's ProxyClient entries from per-thread connection + // maps (see SetThread). The removal must happen eagerly here rather than + // whenever the owning threads next touch their maps, because the owning + // threads might never touch them again, and surviving entries would hold + // this Connection object -- and through its EventLoopRef the event loop -- + // alive indefinitely. // - // 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()); + // Other proxy objects associated with this connection do not need to be + // notified about the disconnect. Interface ProxyClient objects hold shared + // ownership of this Connection object, and their capability handles remain + // safe to hold and release after the disconnect (calls made through them + // just fail with "IPC client method called after disconnect" errors, see + // clientInvoke), so they are simply destroyed whenever the application + // code owning them gets around to it. ProxyServer objects that are not + // kept alive by in-flight calls are destroyed by the m_rpc_system.reset() + // call above; destructors of the m_impl objects they wrap can run + // arbitrary blocking application code, so ~ProxyServerBase schedules those + // on the EventLoop::m_async_fns worker thread instead of running them + // here, where they could deadlock the event loop thread. + { + 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) +CleanupIt Connection::onDisconnect(std::function fn) { const Lock lock(m_loop->m_mutex); // Add cleanup callbacks to the front of list, so sync cleanup functions run // in LIFO order. This is a good approach because sync cleanup functions are - // added as client objects are created, and it is natural to clean up - // objects in the reverse order they were created. In practice, however, - // 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. + // added as objects are created, and it is natural to clean up objects in + // the reverse order they were created. In practice, however, order should + // not be significant because the cleanup callbacks run synchronously in a + // single batch when the connection is disconnected, and each one acts on + // independent state: removing a map entry (see SetThread) or decrementing + // a listener's active-connection counter (see _Serve). return m_sync_cleanup_fns.emplace(m_sync_cleanup_fns.begin(), std::move(fn)); } -void Connection::removeSyncCleanup(CleanupIt it) +void Connection::cancelOnDisconnect(CleanupIt it) { // Require cleanup functions to be removed on the event loop thread to avoid // needing to deal with them being removed in the middle of a disconnect. @@ -394,20 +407,32 @@ 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); + // Register a cleanup callback eagerly removing this entry from the map + // when the connection is disconnected. This cannot be left to the + // thread owning the map, which might never touch the map again; a + // surviving entry would hold the disconnected Connection object -- and + // through its EventLoopRef the event loop -- alive indefinitely. See + // Connection::disconnect. + thread->second->m_disconnect_cb = connection->onDisconnect([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 +440,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 +452,52 @@ 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) { - m_context.connection->removeSyncCleanup(*m_disconnect_cb); + // Check connection->disconnected() in addition to m_disconnect_cb: + // if the connection was disconnected while this thread was waiting + // for the event loop, Connection::disconnect() has already run and + // freed every m_sync_cleanup_fns node, including the one + // m_disconnect_cb points at, so the iterator is dangling and must + // not be passed to cancelOnDisconnect. disconnect() runs to + // completion on the event loop thread without interleaving with + // this posted lambda, so disconnected() is true here exactly when + // the node has already been freed. + // + // m_disconnect_cb can be set here even though disconnect() freed + // the node: the SetThread callback only resets m_disconnect_cb when + // it still finds this object in the thread map (see SetThread); if + // ~ThreadContext extracted the map entry first, the callback + // returns early and leaves m_disconnect_cb set. Before this + // object's connection ownership was made shared, m_context.connection + // was nulled on disconnect and this check keyed off that instead. + if (m_disconnect_cb && !m_context.connection->disconnected()) { + m_context.connection->cancelOnDisconnect(*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..4bf6c716 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,37 @@ 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 with disconnect() must still free the + // listener's slot so the listener resumes accepting. The counter decrement + // runs as an onDisconnect cleanup, so it runs synchronously on local + // disconnects too; if it were in the deferred afterDisconnect handler it + // could be handed a null connection or dropped 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. Disconnect + // it before dropping the list's reference: under shared ownership, erasing + // it from the list alone would not tear it down (see Connection::make). + EventLoop& loop{**server.m_loop_ref}; + loop.sync([&] { + KJ_REQUIRE(loop.m_incoming_connections.size() == 1); + loop.m_incoming_connections.front()->disconnect(); + 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..b15dd855 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -74,13 +75,18 @@ static_assert(std::is_integral_v, "MP_MINOR_VERSION * this to be true to simplify shutdown and avoid needing to call * client_disconnect manually, but false allows testing more ProxyClient * behavior and the "IPC client method called after disconnect" code path. + * + * Also accepts an analogous server_owns_connection option controlling + * whether the server Connection is erased automatically when the peer + * disconnects. False is needed by tests that need to keep the server + * Connection alive across a disconnect notification, tearing it down + * explicitly later via server_disconnect() instead. */ 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; @@ -89,7 +95,7 @@ class TestSetup //! not start until the other members are initialized. std::thread thread; - TestSetup(bool client_owns_connection = true) + TestSetup(bool client_owns_connection = true, bool server_owns_connection = true) : thread{[&] { EventLoop loop("mptest", [](mp::LogMessage log) { // Info logs are not printed by default, but will be shown with `mptest --verbose` @@ -98,36 +104,49 @@ class TestSetup }); auto pipe = loop.m_io_context.provider->newTwoWayPipe(); - auto server_connection = - std::make_unique(loop, kj::mv(pipe.ends[0]), [&](Connection& connection) { - auto server_proxy = kj::heap>( - std::make_shared(), connection); - server = server_proxy; - return capnp::Capability::Client(kj::mv(server_proxy)); - }); - server_disconnect = [&] { loop.sync([&] { server_connection.reset(); }); }; + // Weak, not owning: nothing here needs to keep the connection + // alive. When server_owns_connection is true, ServeStream's own + // afterDisconnect handler owns tearing it down on a remote + // disconnect (via loop.m_incoming_connections); when false (or + // when a test disconnects it directly, bypassing + // server_disconnect()), the connection stays alive on that list + // until something else drops the last reference. An owning + // reference here would race both cases: it would leak the + // EventLoop forever in the first case unless dropped in lockstep + // with ServeStream's own handler, and would let this code tear + // the connection down early in the second case, before callers + // that expect it to survive until they call server_disconnect() + // are done with it. + std::weak_ptr server_connection; + { + auto server_result = ServeStream( + loop, kj::mv(pipe.ends[0]), std::make_shared(), server_owns_connection); + server = server_result.first; + server_connection = server_result.second; + } + auto server_close = [&] { + if (auto connection = server_connection.lock()) { + connection->disconnect(); + loop.m_incoming_connections.remove(connection); + } + }; + server_disconnect = [&] { loop.sync(server_close); }; server_disconnect_later = [&] { assert(std::this_thread::get_id() == loop.m_thread_id); - loop.m_task_set->add(kj::evalLater([&] { server_connection.reset(); })); + loop.m_task_set->add(kj::evalLater([&] { server_close(); })); }; - // Set handler to destroy the server when the client disconnects. This - // 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>( - client_connection->m_rpc_system->bootstrap(ServerVatId().vat_id).castAs(), - client_connection.get(), /* destroy_connection= */ client_owns_connection); - if (client_owns_connection) { - (void)client_connection.release(); - } else { - client_disconnect = [&] { loop.sync([&] { client_connection.reset(); }); }; + + auto client_proxy = + ConnectStream(loop, kj::mv(pipe.ends[1]), client_owns_connection); + std::shared_ptr client_connection; + if (!client_owns_connection) { + client_connection = client_proxy->m_context.connection; + client_disconnect = [&] { loop.sync([&] { + if (client_connection) { + client_connection->disconnect(); + client_connection.reset(); + } + }); }; } client_promise.set_value(std::move(client_proxy)); @@ -414,23 +433,20 @@ KJ_TEST("Calling async IPC method with a remote disconnect while results are bui // 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}; + // by constructing TestSetup with server_owns_connection=false: 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 + // TwoPartyVatNetwork::onDisconnect notification. The connection is + // destroyed explicitly at the end of the test instead. + + TestSetup setup{/*client_owns_connection=*/false, /*server_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; @@ -498,6 +514,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 +611,89 @@ 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 a shared reference to the server Connection on the event loop thread + // before disconnecting, and hold it for the rest of the test. This mirrors + // Ipc::disconnectIncoming, which keeps its own reference alive across the + // disconnect() + waitDrained() sequence: under shared ownership the last + // server proxy (destroyed once the drained body finishes) would otherwise + // free the Connection while the test is still observing it. The reference is + // released on the event loop thread at the end so ~Connection runs there. + std::shared_ptr connection; + 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(); + + // Drop the test's reference on the event loop thread, mirroring how + // Ipc::disconnectIncoming releases the last reference after draining. This + // destroys the drained connection (~Connection notices disconnect() has + // already run and does not tear things down twice). server_disconnect() is + // then a no-op that just lets the loop exit. + foo->m_context.loop->sync([&] { connection.reset(); }); + setup.server_disconnect(); +} + KJ_TEST("Destroying ProxyClient<> with destroy method after peer disconnect") { // Regression test for bitcoin-core/libmultiprocess#219 where @@ -588,8 +735,8 @@ KJ_TEST("Make simultaneous IPC calls on single remote thread") Thread::Client *callback_thread, *request_thread; foo->m_context.loop->sync([&] { Lock lock(tc.waiter->m_mutex); - callback_thread = &tc.callback_threads.at(foo->m_context.connection)->m_client; - request_thread = &tc.request_threads.at(foo->m_context.connection)->m_client; + callback_thread = &tc.callback_threads.at(foo->m_context.connection.get())->m_client; + request_thread = &tc.request_threads.at(foo->m_context.connection.get())->m_client; }); // Call callIntFnAsync 3 times with n=100, 200, 300