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

@@ -297,5 +297,68 @@ 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};
+    auto callback_thread{tc.callback_threads.at(foo->m_context.connection)->m_client};
+    auto 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
+    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