bug: fix ProxyClient<Thread> deadlock if disconnected as IPC call is returning

This bug is currently causing mptest "disconnecting and blocking" test to
occasionally hang as reported by maflcko in
bitcoin/bitcoin#33244.

The bug was actually first reported by Sjors in
Sjors/bitcoin#90 (comment) and there are
more details about it in
#189.

The bug is caused by the "disconnecting and blocking" test triggering a
disconnect right before a server IPC call returns. This results in a race
between the IPC server thread and the onDisconnect handler in the event loop
thread both trying to destroy the server's request_threads ProxyClient<Thread>
object when the IPC call is done.

There was a lack of synchronization in this case, fixed here by adding
loop->sync() a few places. Specifically the fixes were to:

- Always call SetThread on the event loop thread using the loop->sync() method,
  to prevent a race between the ProxyClient<Thread> creation code and the
  connection shutdown code if there was an ill-timed disconnect.

- Similarly in ~ProxyClient<Thread> and thread-context.h PassField(), use
  loop->sync() when destroying the thread object, in case a disconnect happens
  at that time.

A few other changes were made in this commit to make the resulting code safer
and simpler, even though they are not technically necessary for the fix:

- In thread-context.h PassField(), Waiter::m_mutex is now unlocked while
  destroying ProxyClient<Thread> just to respect EventLoop::m_mutex and
  Waiter::m_mutex lock order and never lock the Waiter first. This is just for
  consistency. There is no actually possibility for a deadlock here due to the
  new sync() call.

- This adds asserts to make sure functions expected to run on the event
  loop thread are only called on that thread.

- This inlines the ProxyClient<Thread>::setDisconnectCallback function, just
  because it was a short function only called in a single place.

Author: ryanofsky

include/mp/proxy-io.h

@@ -66,8 +66,6 @@ struct ProxyClient<Thread> : public ProxyClientBase<Thread, ::capnp::Void>
     ProxyClient(const ProxyClient&) = delete;
     ~ProxyClient();
 
-    void setDisconnectCallback(const std::function<void()>& fn);
-
     //! Reference to callback function that is run if there is a sudden
     //! disconnect and the Connection object is destroyed before this
     //! ProxyClient<Thread> object. The callback will destroy this object and

include/mp/type-context.h

@@ -89,29 +89,39 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                     // need to update the map.
                     auto& thread_context = g_thread_context;
                     auto& request_threads = thread_context.request_threads;
-                    auto [request_thread, inserted]{SetThread(
-                        GuardedRef{thread_context.waiter->m_mutex, request_threads},
-                        server.m_context.connection,
-                        [&] { return context_arg.getCallbackThread(); })};
+                    ConnThread request_thread;
+                    bool inserted;
+                    server.m_context.loop->sync([&] {
+                        std::tie(request_thread, inserted) = SetThread(
+                            GuardedRef{thread_context.waiter->m_mutex, request_threads}, server.m_context.connection,
+                            [&] { return context_arg.getCallbackThread(); });
+                    });
 
-                    // If an entry was inserted into the requests_threads map,
+                    // If an entry was inserted into the request_threads map,
                     // remove it after calling fn.invoke. If an entry was not
                     // inserted, one already existed, meaning this must be a
                     // recursive call (IPC call calling back to the caller which
                     // makes another IPC call), so avoid modifying the map.
                     const bool erase_thread{inserted};
                     KJ_DEFER(if (erase_thread) {
-                        Lock lock(thread_context.waiter->m_mutex);
-                        // Call erase here with a Connection* argument instead
-                        // of an iterator argument, because the `request_thread`
-                        // iterator may be invalid if the connection is closed
-                        // during this function call. More specifically, the
-                        // iterator may be invalid because SetThread adds a
-                        // cleanup callback to the Connection destructor that
-                        // erases the thread from the map, and also because the
-                        // ProxyServer<Thread> destructor calls
-                        // request_threads.clear().
-                        request_threads.erase(server.m_context.connection);
+                        // Erase the request_threads entry on the event loop
+                        // thread with loop->sync(), so if the connection is
+                        // broken there is not a race between this thread and
+                        // the disconnect handler trying to destroy the thread
+                        // client object.
+                        server.m_context.loop->sync([&] {
+                            // Look up the thread again without using existing
+                            // iterator since entry may no longer be there after
+                            // a disconnect. Destroy node after releasing
+                            // Waiter::m_mutex, so the ProxyClient<Thread>
+                            // destructor is able to use EventLoop::mutex
+                            // without violating lock order.
+                            ConnThreads::node_type removed;
+                            {
+                                Lock lock(thread_context.waiter->m_mutex);
+                                removed = request_threads.extract(server.m_context.connection);
+                            }
+                        });
                     });
                     fn.invoke(server_context, args...);
                 }

src/mp/proxy.cpp

@@ -12,6 +12,7 @@
 
 #include <atomic>
 #include <capnp/capability.h>
+#include <capnp/common.h> // IWYU pragma: keep
 #include <capnp/rpc.h>
 #include <condition_variable>
 #include <functional>
@@ -80,6 +81,11 @@ ProxyContext::ProxyContext(Connection* connection) : connection(connection), loo
 
 Connection::~Connection()
 {
+    // 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.
+    assert(std::this_thread::get_id() == m_loop->m_thread_id);
     // Shut down RPC system first, since this will garbage collect any
     // ProxyServer objects that were not freed before the connection was closed.
     // Typically all ProxyServer objects associated with this connection will be
@@ -155,6 +161,9 @@ CleanupIt Connection::addSyncCleanup(std::function<void()> fn)
 
 void Connection::removeSyncCleanup(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.
+    assert(std::this_thread::get_id() == m_loop->m_thread_id);
     const Lock lock(m_loop->m_mutex);
     m_sync_cleanup_fns.erase(it);
 }
@@ -306,6 +315,7 @@ bool EventLoop::done() const
 
 std::tuple<ConnThread, bool> SetThread(GuardedRef<ConnThreads> threads, Connection* connection, const std::function<Thread::Client()>& make_thread)
 {
+    assert(std::this_thread::get_id() == connection->m_loop->m_thread_id);
     ConnThread thread;
     bool inserted;
     {
@@ -314,7 +324,7 @@ std::tuple<ConnThread, bool> SetThread(GuardedRef<ConnThreads> threads, Connecti
     }
     if (inserted) {
         thread->second.emplace(make_thread(), connection, /* destroy_connection= */ false);
-        thread->second->setDisconnectCallback([threads, thread] {
+        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<Thread>
@@ -323,9 +333,10 @@ std::tuple<ConnThread, bool> SetThread(GuardedRef<ConnThreads> threads, Connecti
             // 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);
-            thread->second->m_disconnect_cb.reset();
             threads.ref.erase(thread);
         });
     }
@@ -338,17 +349,18 @@ ProxyClient<Thread>::~ProxyClient()
     // cleanup callback that was registered to handle the connection being
     // destroyed before the thread being destroyed.
     if (m_disconnect_cb) {
-        m_context.connection->removeSyncCleanup(*m_disconnect_cb);
+        // Remove disconnect callback on the event loop thread with
+        // loop->sync(), so if the connection is broken there is not a race
+        // 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);
+            }
+        });
     }
 }
 
-void ProxyClient<Thread>::setDisconnectCallback(const std::function<void()>& fn)
-{
-    assert(fn);
-    assert(!m_disconnect_cb);
-    m_disconnect_cb = m_context.connection->addSyncCleanup(fn);
-}
-
 ProxyServer<Thread>::ProxyServer(ThreadContext& thread_context, std::thread&& thread)
     : m_thread_context(thread_context), m_thread(std::move(thread))
 {