Allow simultaneous calls on same Context.thread

If multiple IPC requests happen at the same time specifying same Context.thread
to run the requests on, queue the requests to execute in the order they are
received instead of raising a "thread busy" exception.

This change has no effect on C++ clients using libmultiprocess as a client
library, since the libmultiprocess client only makes blocking calls and creates
a server thread for every client thread, so it's not possible for there to be
multiple calls on the same server thread.

But this change may be useful for rust and python clients as discussed
bitcoin/bitcoin#33923

Author: ryanofsky

include/mp/proxy-io.h

@@ -82,11 +82,20 @@ template <>
 struct ProxyServer<Thread> final : public Thread::Server
 {
 public:
-    ProxyServer(ThreadContext& thread_context, std::thread&& thread);
+    ProxyServer(Connection& connection, ThreadContext& thread_context, std::thread&& thread);
     ~ProxyServer();
     kj::Promise<void> getName(GetNameContext context) override;
+
+    //! Run a callback function returning T on this thread.
+    template<typename T, typename Fn>
+    kj::Promise<T> post(Fn&& fn);
+
+    EventLoopRef m_loop;
     ThreadContext& m_thread_context;
     std::thread m_thread;
+    //! Promise signaled when m_thread_context.waiter is idle and there is no
+    //! post() callback function waiting to execute.
+    kj::Promise<void> m_thread_ready{kj::READY_NOW};
 };
 
 //! Handler for kj::TaskSet failed task events.
@@ -322,7 +331,12 @@ class EventLoop
 //! thread is blocked waiting for server response, this is what allows the
 //! client to run the request in the same thread, the same way code would run in a
 //! single process, with the callback sharing the same thread stack as the original
-//! call.)
+//! call.) To support this, the clientInvoke function calls Waiter::wait() to
+//! block the client IPC thread while initial request is in progress. Then if
+//! there is a callback, it is executed with Waiter::post().
+//!
+//! The Waiter class is also used server-side by `ProxyServer<Thread>::post()`
+//! to execute IPC calls on worker threads.
 struct Waiter
 {
     Waiter() = default;
@@ -675,6 +689,37 @@ struct ThreadContext
     bool loop_thread = false;
 };
 
+template<typename T, typename Fn>
+kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
+{
+    auto ready = kj::newPromiseAndFulfiller<void>(); // Signaled when waiter is idle again.
+    auto ret = m_thread_ready.then([this, fn = std::move(fn), ready_fulfiller = kj::mv(ready.fulfiller)]() mutable {
+        auto result = kj::newPromiseAndFulfiller<T>(); // Signaled when fn() is called, with its return value.
+        bool posted = m_thread_context.waiter->post([this, fn = std::move(fn), ready_fulfiller = kj::mv(ready_fulfiller), result_fulfiller = kj::mv(result.fulfiller)]() mutable {
+            m_loop->sync([ready_fulfiller = kj::mv(ready_fulfiller)]() mutable { ready_fulfiller->fulfill(); });
+            std::optional<T> result;
+            kj::Maybe<kj::Exception> exception{kj::runCatchingExceptions([&]{ result.emplace(fn()); })};
+            m_loop->sync([&result, &exception, result_fulfiller = kj::mv(result_fulfiller)]() mutable {
+                KJ_IF_MAYBE(e, exception) {
+                    assert(!result);
+                    result_fulfiller->reject(kj::mv(*e));
+                } else {
+                    assert(result);
+                    result_fulfiller->fulfill(kj::mv(*result));
+                }
+            });
+        });
+        // Assert that calling Waiter::post did not fail. It could only return
+        // false if a new function was posted before the previous one finished
+        // executing, but new functions are only posted when m_thread_ready is
+        // is signaled, so this should never happen.
+        assert(posted);
+        return kj::mv(result.promise);
+    });
+    m_thread_ready = kj::mv(ready.promise);
+    return ret;
+}
+
 //! Given stream file descriptor, make a new ProxyClient object to send requests
 //! over the stream. Also create a new Connection object embedded in the
 //! client that is freed when the client is closed.

include/mp/type-context.h

@@ -26,7 +26,7 @@ void CustomBuildField(TypeList<>,
     // future calls over this connection can reuse it.
     auto [callback_thread, _]{SetThread(
         GuardedRef{thread_context.waiter->m_mutex, thread_context.callback_threads}, &connection,
-        [&] { return connection.m_threads.add(kj::heap<ProxyServer<Thread>>(thread_context, std::thread{})); })};
+        [&] { return connection.m_threads.add(kj::heap<ProxyServer<Thread>>(connection, thread_context, std::thread{})); })};
 
     // Call remote ThreadMap.makeThread function so server will create a
     // dedicated worker thread to run function calls from this thread. Store the
@@ -61,11 +61,10 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
 {
     const auto& params = server_context.call_context.getParams();
     Context::Reader context_arg = Accessor::get(params);
-    auto future = kj::newPromiseAndFulfiller<typename ServerContext::CallContext>();
     auto& server = server_context.proxy_server;
     int req = server_context.req;
-    auto invoke = [fulfiller = kj::mv(future.fulfiller),
-         call_context = kj::mv(server_context.call_context), &server, req, fn, args...]() mutable {
+    auto invoke = [call_context = kj::mv(server_context.call_context), &server, req, fn, args...]() mutable {
+                MP_LOG(*server.m_context.loop, Log::Debug) << "IPC server executing request #" << req;
                 const auto& params = call_context.getParams();
                 Context::Reader context_arg = Accessor::get(params);
                 ServerContext server_context{server, call_context, req};
@@ -125,18 +124,7 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                     });
                     fn.invoke(server_context, args...);
                 }
-                KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() {
-                    server.m_context.loop->sync([&] {
-                        auto fulfiller_dispose = kj::mv(fulfiller);
-                        fulfiller_dispose->fulfill(kj::mv(call_context));
-                    });
-                }))
-                {
-                    server.m_context.loop->sync([&]() {
-                        auto fulfiller_dispose = kj::mv(fulfiller);
-                        fulfiller_dispose->reject(kj::mv(*exception));
-                    });
-                }
+                return call_context;
             };
 
     // Lookup Thread object specified by the client. The specified thread should
@@ -149,23 +137,16 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
             // `invoke` lambda above which will invoke the function on that
             // thread.
             KJ_IF_MAYBE (thread_server, perhaps) {
-                const auto& thread = static_cast<ProxyServer<Thread>&>(*thread_server);
+                auto& thread = static_cast<ProxyServer<Thread>&>(*thread_server);
                 MP_LOG(*server.m_context.loop, Log::Debug)
                     << "IPC server post request  #" << req << " {" << thread.m_thread_context.thread_name << "}";
-                if (!thread.m_thread_context.waiter->post(std::move(invoke))) {
-                    MP_LOG(*server.m_context.loop, Log::Error)
-                        << "IPC server error request #" << req
-                        << " {" << thread.m_thread_context.thread_name << "}" << ", thread busy";
-                    throw std::runtime_error("thread busy");
-                }
+                return thread.template post<typename ServerContext::CallContext>(std::move(invoke));
             } else {
                 MP_LOG(*server.m_context.loop, Log::Error)
                     << "IPC server error request #" << req << ", missing thread to execute request";
                 throw std::runtime_error("invalid thread handle");
             }
-        })
-        // Wait for the invocation to finish before returning to the caller.
-        .then([invoke_wait = kj::mv(future.promise)]() mutable { return kj::mv(invoke_wait); });
+        });
 }
 } // namespace mp
 

src/mp/proxy.cpp

@@ -362,8 +362,8 @@ ProxyClient<Thread>::~ProxyClient()
     }
 }
 
-ProxyServer<Thread>::ProxyServer(ThreadContext& thread_context, std::thread&& thread)
-    : m_thread_context(thread_context), m_thread(std::move(thread))
+ProxyServer<Thread>::ProxyServer(Connection& connection, ThreadContext& thread_context, std::thread&& thread)
+    : m_loop{*connection.m_loop}, m_thread_context(thread_context), m_thread(std::move(thread))
 {
     assert(m_thread_context.waiter.get() != nullptr);
 }
@@ -418,7 +418,7 @@ kj::Promise<void> ProxyServer<ThreadMap>::makeThread(MakeThreadContext context)
         // is just waiter getting set to null.)
         g_thread_context.waiter->wait(lock, [] { return !g_thread_context.waiter; });
     });
-    auto thread_server = kj::heap<ProxyServer<Thread>>(*thread_context.get_future().get(), std::move(thread));
+    auto thread_server = kj::heap<ProxyServer<Thread>>(m_connection, *thread_context.get_future().get(), std::move(thread));
     auto thread_client = m_connection.m_threads.add(kj::mv(thread_server));
     context.getResults().setResult(kj::mv(thread_client));
     return kj::READY_NOW;

test/mp/test/foo.capnp

@@ -33,6 +33,7 @@ interface FooInterface $Proxy.wrap("mp::test::FooImplementation") {
     passFn @16 (context :Proxy.Context, fn :FooFn) -> (result :Int32);
     callFn @17 () -> ();
     callFnAsync @18 (context :Proxy.Context) -> ();
+    callIntFnAsync @21 (context :Proxy.Context, arg :Int32) -> (result :Int32);
 }
 
 interface FooCallback $Proxy.wrap("mp::test::FooCallback") {

test/mp/test/foo.h

@@ -83,7 +83,9 @@ class FooImplementation
     std::shared_ptr<FooCallback> m_callback;
     void callFn() { assert(m_fn); m_fn(); }
     void callFnAsync() { assert(m_fn); m_fn(); }
+    int callIntFnAsync(int arg) { assert(m_int_fn); return m_int_fn(arg); }
     std::function<void()> m_fn;
+    std::function<int(int)> m_int_fn;
 };
 
 } // namespace test

test/mp/test/test.cpp

@@ -8,17 +8,17 @@
 #include <atomic>
 #include <capnp/capability.h>
 #include <capnp/rpc.h>
+#include <cassert>
 #include <condition_variable>
+#include <cstdint>
 #include <cstring>
 #include <functional>
 #include <future>
 #include <kj/async.h>
 #include <kj/async-io.h>
 #include <kj/common.h>
 #include <kj/debug.h>
-#include <kj/exception.h>
 #include <kj/memory.h>
-#include <kj/string.h>
 #include <kj/test.h>
 #include <memory>
 #include <mp/proxy.h>
@@ -31,7 +31,6 @@
 #include <stdexcept>
 #include <string>
 #include <string_view>
-#include <system_error>
 #include <thread>
 #include <type_traits>
 #include <utility>
@@ -317,7 +316,7 @@ KJ_TEST("Calling IPC method, disconnecting and blocking during the call")
     signal.set_value();
 }
 
-KJ_TEST("Make simultaneous IPC calls to trigger 'thread busy' error")
+KJ_TEST("Make simultaneous IPC calls on single remote thread")
 {
     TestSetup setup;
     ProxyClient<messages::FooInterface>* foo = setup.client.get();
@@ -336,51 +335,39 @@ KJ_TEST("Make simultaneous IPC calls to trigger 'thread busy' error")
         request_thread = &tc.request_threads.at(foo->m_context.connection)->m_client;
     });
 
-    setup.server->m_impl->m_fn = [&] {
-        try
-        {
-            signal.get_future().get();
-        }
-        catch (const std::future_error& e)
-        {
-            KJ_EXPECT(e.code() == std::make_error_code(std::future_errc::future_already_retrieved));
-        }
+    // Call callIntFnAsync 3 times with n=100, 200, 300
+    std::atomic<int> expected = 100;
+
+    setup.server->m_impl->m_int_fn = [&](int n) {
+        assert(n == expected);
+        expected += 100;
+        return n;
     };
 
     auto client{foo->m_client};
-    bool caught_thread_busy = false;
-    // NOTE: '3' was chosen because it was the lowest number
-    // of simultaneous calls required to reliably catch a "thread busy" error
     std::atomic<size_t> running{3};
     foo->m_context.loop->sync([&]
     {
         for (size_t i = 0; i < running; i++)
         {
-            auto request{client.callFnAsyncRequest()};
+            auto request{client.callIntFnAsyncRequest()};
             auto context{request.initContext()};
             context.setCallbackThread(*callback_thread);
             context.setThread(*request_thread);
+            request.setArg(100 * (i+1));
             foo->m_context.loop->m_task_set->add(request.send().then(
-                [&](auto&& results) {
-                    running -= 1;
-                    tc.waiter->m_cv.notify_all();
-                },
-                [&](kj::Exception&& e) {
-                    KJ_EXPECT(std::string_view{e.getDescription().cStr()} ==
-                        "remote exception: std::exception: thread busy");
-                    caught_thread_busy = true;
+                [&running, &tc, i](auto&& results) {
+                    assert(results.getResult() == static_cast<int32_t>(100 * (i+1)));
                     running -= 1;
-                    signal.set_value();
                     tc.waiter->m_cv.notify_all();
-                }
-            ));
+                }));
         }
     });
     {
         Lock lock(tc.waiter->m_mutex);
         tc.waiter->wait(lock, [&running] { return running == 0; });
     }
-    KJ_EXPECT(caught_thread_busy);
+    KJ_EXPECT(expected == 400);
 }
 
 } // namespace test