test: simultaneous IPC calls using same thread

Although the libmultiprocess client prevents this kind of situation from occuring, it easily occurs in non-libmultiprocess clients

Author: Eunovo

test/mp/test/test.cpp

@@ -5,8 +5,10 @@
 #include <mp/test/foo.capnp.h>
 #include <mp/test/foo.capnp.proxy.h>
 
+#include <atomic>
 #include <capnp/capability.h>
 #include <capnp/rpc.h>
+#include <condition_variable>
 #include <cstring>
 #include <functional>
 #include <future>
@@ -15,11 +17,15 @@
 #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>
+#include "mp/proxy.capnp.h"
 #include <mp/proxy-io.h>
+#include "mp/util.h"
 #include <optional>
 #include <set>
 #include <stdexcept>
@@ -297,5 +303,74 @@ KJ_TEST("Calling IPC method, disconnecting and blocking during the call")
     signal.set_value();
 }
 
+KJ_TEST("Make simultaneous IPC callbacks with same request_thread and callback_thread")
+{
+    TestSetup setup;
+    ProxyClient<messages::FooInterface>* foo = setup.client.get();
+
+    class Callback : public FooCallback
+    {
+    public:
+        Callback(int expect, int ret) : m_expect(expect), m_ret(ret) {}
+        int call(int arg) override
+        {
+            KJ_EXPECT(arg == m_expect);
+            return m_ret;
+        }
+        int m_expect, m_ret;
+    };
+
+    foo->initThreadMap();
+    auto callback{std::make_shared<Callback>(1, 2)};
+    // Note: Calling saveCallback initializes the callback_thread and request_thread
+    // for this connection. These threads will be reused for the simultaneous calls below.
+    foo->saveCallback(callback);
+
+    ThreadContext& tc{g_thread_context};
+    std::optional<Thread::Client> callback_thread, request_thread;
+    {
+        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;
+    }
+
+    auto client{foo->m_client};
+    bool caught_thread_busy = false;
+    // NOTE: '3' was choosen 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 callbackReq{client.callbackSavedRequest()};
+            auto context{callbackReq.initContext()};
+            context.setCallbackThread(*callback_thread);
+            context.setThread(*request_thread);
+            callbackReq.setArg(1);
+            foo->m_context.loop->m_task_set->add(callbackReq.send().then(
+                [&](auto&& results) {
+                    KJ_EXPECT(results.getResult() == 2);
+                    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 -= 1;
+                    tc.waiter->m_cv.notify_all();
+                }
+            ));
+        }
+    });
+    {
+        Lock lock(tc.waiter->m_mutex);
+        tc.waiter->wait(lock, [&running] { return running == 0; });
+    }
+    foo->saveCallback(nullptr); // Clear saved callback
+    KJ_EXPECT(caught_thread_busy);
+}
+
 } // namespace test
 } // namespace mp