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

@@ -93,6 +93,9 @@ struct ProxyServer<Thread> final : public Thread::Server
     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.
@@ -689,9 +692,14 @@ struct ThreadContext
 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::forward<Fn>(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::forward<Fn>(fn), result_fulfiller = kj::mv(result.fulfiller)]() mutable {
+        bool posted = m_thread_context.waiter->post([this, fn = std::forward<Fn>(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();
+                ready_fulfiller = nullptr;
+            });
             std::optional<T> result_value;
             kj::Maybe<kj::Exception> exception{kj::runCatchingExceptions([&]{ result_value.emplace(fn()); })};
             m_loop->sync([&result_value, &exception, result_fulfiller = kj::mv(result_fulfiller)]() mutable {
@@ -706,9 +714,15 @@ kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
                 result_fulfiller = nullptr;
             });
         });
-        if (!posted) throw std::runtime_error("thread busy");
+        // 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

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