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() various places. There were also lock order issues where
Waiter::m_mutex could be incorrectly locked before EventLoop::m_mutex resulting
in a deadlock.

Author: ryanofsky

include/mp/proxy-io.h

@@ -58,6 +58,15 @@ struct ServerInvokeContext : InvokeContext
 template <typename Interface, typename Params, typename Results>
 using ServerContext = ServerInvokeContext<ProxyServer<Interface>, ::capnp::CallContext<Params, Results>>;
 
+//! Map from Connection to local or remote thread handle which will be used over
+//! that connection. This map will typically only contain one entry, but can
+//! contain multiple if a single thread makes IPC calls over multiple
+//! connections. A std::optional value type is used to avoid the map needing to
+//! be locked while ProxyClient<Thread> objects are constructred, see
+//! ThreadContext "Synchronization note" below.
+using ConnThreads = std::map<Connection*, std::optional<ProxyClient<Thread>>>;
+using ConnThread = ConnThreads::iterator;
+
 template <>
 struct ProxyClient<Thread> : public ProxyClientBase<Thread, ::capnp::Void>
 {
@@ -66,8 +75,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
@@ -534,9 +541,6 @@ void ProxyServerBase<Interface, Impl>::invokeDestroy()
     CleanupRun(m_context.cleanup_fns);
 }
 
-using ConnThreads = std::map<Connection*, ProxyClient<Thread>>;
-using ConnThread = ConnThreads::iterator;
-
 // Retrieve ProxyClient<Thread> object associated with this connection from a
 // map, or create a new one and insert it into the map. Return map iterator and
 // inserted bool.

include/mp/type-context.h

@@ -47,8 +47,8 @@ void CustomBuildField(TypeList<>,
         &connection, make_request_thread)};
 
     auto context = output.init();
-    context.setThread(request_thread->second.m_client);
-    context.setCallbackThread(callback_thread->second.m_client);
+    context.setThread(request_thread->second->m_client);
+    context.setCallbackThread(callback_thread->second->m_client);
 }
 
 //! PassField override for mp.Context arguments. Return asynchronously and call
@@ -89,10 +89,13 @@ 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(
-                        request_threads, thread_context.waiter->m_mutex,
-                        server.m_context.connection,
-                        [&] { return context_arg.getCallbackThread(); })};
+                    ConnThread request_thread;
+                    bool inserted;
+                    server.m_context.loop->sync([&] {
+                        std::tie(request_thread, inserted) = SetThread(
+                            request_threads, thread_context.waiter->m_mutex, server.m_context.connection,
+                            [&] { return context_arg.getCallbackThread(); });
+                    });
 
                     // If an entry was inserted into the requests_threads map,
                     // remove it after calling fn.invoke. If an entry was not
@@ -101,17 +104,27 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                     // makes another IPC call), so avoid modifying the map.
                     const bool erase_thread{inserted};
                     KJ_DEFER(if (erase_thread) {
-                        std::unique_lock<std::mutex> 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 since it may no longer
+                            // be there after a disconnect.
+                            std::unique_lock<std::mutex> lock(thread_context.waiter->m_mutex);
+                            ConnThread thread = request_threads.find(server.m_context.connection);
+                            if (thread != request_threads.end()) {
+                                // Release Waiter::m_mutex while calling the
+                                // ProxyClient<Thread> destructor so it can use
+                                // EventLoop::m_mutex.
+                                lock.unlock();
+                                thread->second.reset();
+                                // Reacquire Waiter::m_mutex to update the request_threads map.
+                                lock.lock();
+                                request_threads.erase(thread);
+                            }
+                        });
                     });
                     fn.invoke(server_context, args...);
                 }

src/mp/proxy.cpp

@@ -156,6 +156,9 @@ CleanupIt Connection::addSyncCleanup(std::function<void()> fn)
 
 void Connection::removeSyncCleanup(CleanupIt it)
 {
+    // Require cleanup functions to be removed 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);
 }
@@ -307,46 +310,50 @@ bool EventLoop::done() const
 
 std::tuple<ConnThread, bool> SetThread(ConnThreads& threads, std::mutex& mutex, Connection* connection, const std::function<Thread::Client()>& make_thread)
 {
-    const std::unique_lock<std::mutex> lock(mutex);
-    auto thread = threads.find(connection);
-    if (thread != threads.end()) return {thread, false};
-    thread = threads.emplace(
-        std::piecewise_construct, std::forward_as_tuple(connection),
-        std::forward_as_tuple(make_thread(), connection, /* destroy_connection= */ false)).first;
-    thread->second.setDisconnectCallback([&threads, &mutex, 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>
-        // 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.
-        // Remove connection pointer about to be destroyed from the map
+    assert(std::this_thread::get_id() == connection->m_loop->m_thread_id);
+    ConnThread thread;
+    bool inserted;
+    {
         const std::unique_lock<std::mutex> lock(mutex);
-        thread->second.m_disconnect_cb.reset();
-        threads.erase(thread);
-    });
-    return {thread, true};
+        std::tie(thread, inserted) = threads.try_emplace(connection);
+    }
+    if (inserted) {
+        thread->second.emplace(make_thread(), connection, /* destroy_connection= */ false);
+        thread->second->m_disconnect_cb = connection->addSyncCleanup([&threads, &mutex, 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>
+            // 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 std::unique_lock<std::mutex> lock(mutex);
+            threads.erase(thread);
+        });
+    }
+    return {thread, inserted};
 }
 
 ProxyClient<Thread>::~ProxyClient()
 {
-    // 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.
     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);
+                m_disconnect_cb.reset();
+            }
+        });
     }
 }
 
-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))
 {