From 5f465cb28b9b199366b2178ca286b2c11109072a Mon Sep 17 00:00:00 2001 From: Novo Date: Tue, 30 Jun 2026 11:57:35 +0200 Subject: [PATCH 1/6] test: recursive async IPC call This test will be useful for verifying that the ThreadPool threads, can handle other requests while waiting for callback to return. Also use the opportunity to refactor some of the tests to remove the try-catch blocks. --- test/mp/test/test.cpp | 92 +++++++++++++++++++------------------------ 1 file changed, 40 insertions(+), 52 deletions(-) diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index eb4b5ec7..1f62599a 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -38,6 +38,15 @@ #include #include +//! Assert that a call throws std::runtime_error with the given message. +#define EXPECT_EXCEPTION(call, message) \ + try { \ + call; \ + KJ_EXPECT(false); \ + } catch (const std::runtime_error& e) { \ + KJ_EXPECT(std::string_view{e.what()} == message); \ + } + namespace mp { namespace test { @@ -251,14 +260,7 @@ KJ_TEST("Call IPC method after client connection is closed") KJ_EXPECT(foo->add(1, 2) == 3); setup.client_disconnect(); - bool disconnected{false}; - try { - foo->add(1, 2); - } catch (const std::runtime_error& e) { - KJ_EXPECT(std::string_view{e.what()} == "IPC client method called after disconnect."); - disconnected = true; - } - KJ_EXPECT(disconnected); + EXPECT_EXCEPTION(foo->add(1, 2), "IPC client method called after disconnect."); } KJ_TEST("Calling IPC method after server connection is closed") @@ -268,14 +270,7 @@ KJ_TEST("Calling IPC method after server connection is closed") KJ_EXPECT(foo->add(1, 2) == 3); setup.server_disconnect(); - bool disconnected{false}; - try { - foo->add(1, 2); - } catch (const std::runtime_error& e) { - KJ_EXPECT(std::string_view{e.what()} == "IPC client method call interrupted by disconnect."); - disconnected = true; - } - KJ_EXPECT(disconnected); + EXPECT_EXCEPTION(foo->add(1, 2), "IPC client method call interrupted by disconnect."); } KJ_TEST("Calling IPC method and disconnecting during the call") @@ -288,14 +283,7 @@ KJ_TEST("Calling IPC method and disconnecting during the call") // handling the callFn call to make sure this case is handled cleanly. setup.server->m_impl->m_fn = setup.client_disconnect; - bool disconnected{false}; - try { - foo->callFn(); - } catch (const std::runtime_error& e) { - KJ_EXPECT(std::string_view{e.what()} == "IPC client method call interrupted by disconnect."); - disconnected = true; - } - KJ_EXPECT(disconnected); + EXPECT_EXCEPTION(foo->callFn(), "IPC client method call interrupted by disconnect."); } KJ_TEST("Calling IPC method, disconnecting and blocking during the call") @@ -330,14 +318,7 @@ KJ_TEST("Calling IPC method, disconnecting and blocking during the call") signal.get_future().get(); }; - bool disconnected{false}; - try { - foo->callFnAsync(); - } catch (const std::runtime_error& e) { - KJ_EXPECT(std::string_view{e.what()} == "IPC client method call interrupted by disconnect."); - disconnected = true; - } - KJ_EXPECT(disconnected); + EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); // Now that the disconnect has been detected, set signal allowing the // callFnAsync() IPC call to return. Since signalling may not wake up the @@ -374,14 +355,7 @@ KJ_TEST("Worker thread destroyed before it is initialized") std::this_thread::sleep_for(std::chrono::milliseconds(10)); }; - bool disconnected{false}; - try { - foo->callFnAsync(); - } catch (const std::runtime_error& e) { - KJ_EXPECT(std::string_view{e.what()} == "IPC client method call interrupted by disconnect."); - disconnected = true; - } - KJ_EXPECT(disconnected); + EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); } KJ_TEST("Calling async IPC method, with server disconnect racing the call") @@ -406,12 +380,7 @@ KJ_TEST("Calling async IPC method, with server disconnect racing the call") std::this_thread::sleep_for(std::chrono::milliseconds(10)); }; - try { - foo->callFnAsync(); - KJ_EXPECT(false); - } catch (const std::runtime_error& e) { - KJ_EXPECT(std::string_view{e.what()} == "IPC client method call interrupted by disconnect."); - } + EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); } KJ_TEST("Calling async IPC method, with server disconnect after cleanup") @@ -435,12 +404,7 @@ KJ_TEST("Calling async IPC method, with server disconnect after cleanup") setup.server_disconnect(); }; - try { - foo->callFnAsync(); - KJ_EXPECT(false); - } catch (const std::runtime_error& e) { - KJ_EXPECT(std::string_view{e.what()} == "IPC client method call interrupted by disconnect."); - } + EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); } KJ_TEST("Destroying ProxyClient<> with destroy method after peer disconnect") @@ -524,5 +488,29 @@ KJ_TEST("Make simultaneous IPC calls on single remote thread") KJ_EXPECT(expected == 400); } +KJ_TEST("Make IPC call to same remote thread while it is waiting for callback") +{ + TestSetup setup; + ProxyClient* foo = setup.client.get(); + + class Callback : public FooCallback + { + public: + Callback(ProxyClient* foo, int expect) : m_foo(foo), m_expect(expect) {} + int call(int arg) override + { + KJ_EXPECT(arg == m_expect); + return m_foo->passFn([this]{ return m_expect; }); + } + + ProxyClient* m_foo; + int m_expect; + }; + + foo->initThreadMap(); + Callback callback(foo, 1); + KJ_EXPECT(foo->callback(callback, 1) == 1); +} + } // namespace test } // namespace mp From 22389f7f9fdbe9171723663c19abec2b382dcd7a Mon Sep 17 00:00:00 2001 From: Novo Date: Tue, 30 Jun 2026 10:48:34 +0200 Subject: [PATCH 2/6] refactor: extract WorkerThread from ProxyServer ProxyServer previously bundled two concerns: the Cap'n Proto server capability (getName, reference-counted lifetime) and core thread management (OS thread, Waiter, m_thread_ready promise chain, post()). Extract the thread management concern into a new WorkerThread struct. ProxyServer becomes a thin wrapper that owns a WorkerThread and adds the capability layer on top. This deduplication prepares for the ThreadPool implementation where WorkerThread objects are owned directly without Cap'n Proto capability overhead. Key design choices: - WorkerThread uses EventLoop& not EventLoopRef: lifetime is bounded by owner, which always holds the appropriate reference to keep the loop alive. - WorkerThread::post(fn, on_done) uses a template OnDone parameter (defaulting to nullptr_t with if constexpr guard) to avoid requiring copy-constructibility, allowing callers to capture move-only values. - ProxyServer gains a second constructor wrapping an existing thread (ThreadContext&) for use as a callback-thread capability. - makeThread() simplified: thread creation now inside WorkerThread ctor. --- include/mp/proxy-io.h | 118 ++++++++++++++++++++++++++++---------- include/mp/type-context.h | 4 +- src/mp/proxy.cpp | 75 ++++++++++++++---------- 3 files changed, 134 insertions(+), 63 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 092ea42e..2985c19f 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -91,12 +91,58 @@ struct ProxyClient : public ProxyClientBase std::optional m_disconnect_cb; }; +//! Worker thread used to execute IPC requests. Owns an OS thread and its +//! associated ThreadContext (waiter, name, callback maps). Used directly by +//! ThreadPool for pool threads, and wrapped by ProxyServer for +//! per-client dedicated threads. +struct WorkerThread { + //! Create a new OS thread with the given name. + WorkerThread(EventLoop& loop, std::string name); + //! Wrap an existing thread (e.g. the current thread) as a WorkerThread. + //! No OS thread is created; m_thread is left non-joinable. + WorkerThread(EventLoop& loop, ThreadContext& existing); + ~WorkerThread(); + WorkerThread(const WorkerThread&) = delete; + WorkerThread& operator=(const WorkerThread&) = delete; + + //! Post fn to the worker thread. When the thread starts executing fn, + //! the returned promise is chained so subsequent post() calls can proceed + //! immediately without blocking on fn's completion. on_done (if set) is + //! called from the event-loop thread after fn returns and results are + //! dispatched; callers use this to perform cleanup that must run on the + //! event-loop thread (e.g. releasing a capability self-reference). + //! OnDone may be std::nullptr_t (no cleanup) or any callable with + //! signature void(). The callable is invoked on the event-loop thread + //! after results are dispatched. Using a template avoids requiring + //! copy-constructibility (e.g. when capturing move-only kj::Own values). + template + kj::Promise post(Fn&& fn, OnDone on_done = nullptr); + + //! Reference to the event loop. Safe as a plain reference because + //! WorkerThread's lifetime is always bounded by its owner: when owned by + //! ProxyServer the owner holds an EventLoopRef keeping the loop + //! alive; when owned by ThreadPool the loop owns the pool. + EventLoop& m_loop; + //! Pointer to the OS thread's thread-local ThreadContext. For new-thread + //! mode this is obtained synchronously after the thread starts; for + //! existing-thread mode this points to the caller-supplied context. + ThreadContext* m_thread_context; + std::thread m_thread; + //! Promise signaled when m_thread_context.waiter is ready and there is no + //! post() callback function waiting to execute. + kj::Promise m_thread_ready{kj::READY_NOW}; +}; + template <> struct ProxyServer final : public Thread::Server { public: - ProxyServer(Connection& connection, ThreadContext& thread_context, std::thread&& thread); - ~ProxyServer(); + //! Construct a ProxyServer wrapping a newly spawned WorkerThread named name. + ProxyServer(Connection& connection, std::string name); + //! Construct a ProxyServer wrapping an existing thread (e.g. the calling + //! thread) so it can be used as a Cap'n Proto callback target. + ProxyServer(Connection& connection, ThreadContext& existing); + ~ProxyServer() = default; kj::Promise getName(GetNameContext context) override; //! Run a callback function fn returning T on this thread. The function will @@ -107,11 +153,7 @@ struct ProxyServer final : public Thread::Server kj::Promise post(Fn&& fn); EventLoopRef m_loop; - ThreadContext& m_thread_context; - std::thread m_thread; - //! Promise signaled when m_thread_context.waiter is ready and there is no - //! post() callback function waiting to execute. - kj::Promise m_thread_ready{kj::READY_NOW}; + WorkerThread m_worker; }; //! Handler for kj::TaskSet failed task events. @@ -740,36 +782,30 @@ struct ThreadContext bool loop_thread = false; }; -template -kj::Promise ProxyServer::post(Fn&& fn) +template +kj::Promise WorkerThread::post(Fn&& fn, OnDone on_done) { auto ready = kj::newPromiseAndFulfiller(); // Signaled when waiter is ready to post again. auto cancel_monitor_ptr = kj::heap(); CancelMonitor& cancel_monitor = *cancel_monitor_ptr; - // Keep a reference to the ProxyServer instance by assigning it to - // the self variable. ProxyServer instances are reference-counted and if the - // client drops its reference, this variable keeps the instance alive until - // the thread finishes executing. The self variable needs to be destroyed on - // the event loop thread so it is freed in a sync() call below. - auto self = thisCap(); - auto ret = m_thread_ready.then([this, self = std::move(self), fn = std::forward(fn), ready_fulfiller = kj::mv(ready.fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr)]() mutable { + auto ret = m_thread_ready.then([this, fn = std::forward(fn), ready_fulfiller = kj::mv(ready.fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr), on_done = std::move(on_done)]() mutable { auto result = kj::newPromiseAndFulfiller(); // Signaled when fn() is called, with its return value. - bool posted = m_thread_context.waiter->post([this, self = std::move(self), fn = std::forward(fn), ready_fulfiller = kj::mv(ready_fulfiller), result_fulfiller = kj::mv(result.fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr)]() mutable { + bool posted = m_thread_context->waiter->post([this, fn = std::move(fn), ready_fulfiller = kj::mv(ready_fulfiller), result_fulfiller = kj::mv(result.fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr), on_done = std::move(on_done)]() mutable { // Fulfill ready.promise now, as soon as the Waiter starts executing - // this lambda, so the next ProxyServer::post() call can - // immediately call waiter->post(). It is important to do this - // before calling fn() because fn() can make an IPC call back to the - // client, which can make another IPC call to this server thread. - // (This typically happens when IPC methods take std::function - // parameters.) When this happens the second call to the server - // thread should not be blocked waiting for the first call. - m_loop->sync([ready_fulfiller = kj::mv(ready_fulfiller)]() mutable { + // this lambda, so the next post() call can immediately call + // waiter->post(). It is important to do this before calling fn() + // because fn() can make an IPC call back to the client, which can + // make another IPC call to this server thread. (This typically + // happens when IPC methods take std::function parameters.) When + // this happens the second call to the server thread should not be + // blocked waiting for the first call. + m_loop.sync([ready_fulfiller = kj::mv(ready_fulfiller)]() mutable { ready_fulfiller->fulfill(); ready_fulfiller = nullptr; }); std::optional result_value; kj::Maybe exception{kj::runCatchingExceptions([&]{ result_value.emplace(fn(*cancel_monitor_ptr)); })}; - m_loop->sync([this, &result_value, &exception, self = kj::mv(self), result_fulfiller = kj::mv(result_fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr)]() mutable { + m_loop.sync([&result_value, &exception, result_fulfiller = kj::mv(result_fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr), on_done = std::move(on_done)]() mutable { // Destroy CancelMonitor here before fulfilling or rejecting the // promise so it doesn't get triggered when the promise is // destroyed. @@ -786,11 +822,13 @@ kj::Promise ProxyServer::post(Fn&& fn) result_value.reset(); } result_fulfiller = nullptr; - // Use evalLater to destroy the ProxyServer self - // reference, if it is the last reference, because the - // ProxyServer destructor needs to join the thread, - // which can't happen until this sync() block has exited. - m_loop->m_task_set->add(kj::evalLater([self = kj::mv(self)] {})); + // Call optional cleanup hook (e.g. ProxyServer uses + // this to schedule deferred self-destruction via evalLater). + if constexpr (!std::is_same_v, std::nullptr_t>) { + on_done(); + } else { + (void)on_done; + } }); }); // Assert that calling Waiter::post did not fail. It could only return @@ -804,6 +842,24 @@ kj::Promise ProxyServer::post(Fn&& fn) return ret; } +template +kj::Promise ProxyServer::post(Fn&& fn) +{ + // Keep a reference to the ProxyServer instance by assigning it to + // the self variable. ProxyServer instances are reference-counted and if the + // client drops its reference, this variable keeps the instance alive until + // the thread finishes executing. The self variable needs to be destroyed on + // the event loop thread so it is freed in a sync() call below. + auto self = thisCap(); + return m_worker.post(std::forward(fn), [this, self = kj::mv(self)]() mutable { + // Use evalLater to destroy the ProxyServer self reference, if + // it is the last reference, because the ProxyServer destructor + // needs to join the thread, which can't happen until this sync() block + // has exited. + m_loop->m_task_set->add(kj::evalLater([self = kj::mv(self)] {})); + }); +} + //! 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. diff --git a/include/mp/type-context.h b/include/mp/type-context.h index 3ab3d4b0..6627c3e4 100644 --- a/include/mp/type-context.h +++ b/include/mp/type-context.h @@ -28,7 +28,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>(connection, thread_context, std::thread{})); })}; + [&] { return connection.m_threads.add(kj::heap>(connection, thread_context)); })}; // Call remote ThreadMap.makeThread function so server will create a // dedicated worker thread to run function calls from this thread. Store the @@ -209,7 +209,7 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& KJ_IF_MAYBE (thread_server, perhaps) { auto& thread = static_cast&>(*thread_server); MP_LOG(loop, Log::Debug) - << "IPC server post request #" << req << " {" << thread.m_thread_context.thread_name << "}"; + << "IPC server post request #" << req << " {" << thread.m_worker.m_thread_context->thread_name << "}"; return thread.template post(std::move(invoke)); } else { MP_LOG(loop, Log::Error) diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp index 963050c3..b906a908 100644 --- a/src/mp/proxy.cpp +++ b/src/mp/proxy.cpp @@ -372,13 +372,28 @@ ProxyClient::~ProxyClient() } } -ProxyServer::ProxyServer(Connection& connection, ThreadContext& thread_context, std::thread&& thread) - : m_loop{*connection.m_loop}, m_thread_context(thread_context), m_thread(std::move(thread)) +WorkerThread::WorkerThread(EventLoop& loop, std::string name) : m_loop(loop) { - assert(m_thread_context.waiter.get() != nullptr); + std::promise ctx_promise; + m_thread = std::thread([&loop, &ctx_promise, name = std::move(name)]() mutable { + g_thread_context.thread_name = std::move(name); + g_thread_context.waiter = std::make_unique(); + Lock lock(g_thread_context.waiter->m_mutex); + ctx_promise.set_value(&g_thread_context); + if (loop.testing_hook_makethread_created) loop.testing_hook_makethread_created(); + // Wait for shutdown signal: waiter being set to null in ~WorkerThread. + g_thread_context.waiter->wait(lock, [] { return !g_thread_context.waiter; }); + }); + m_thread_context = ctx_promise.get_future().get(); } -ProxyServer::~ProxyServer() +WorkerThread::WorkerThread(EventLoop& loop, ThreadContext& existing) + : m_loop(loop), m_thread_context(&existing) +{ + assert(m_thread_context->waiter.get() != nullptr); +} + +WorkerThread::~WorkerThread() { if (!m_thread.joinable()) return; // Stop async thread and wait for it to exit. Need to wait because the @@ -386,30 +401,40 @@ ProxyServer::~ProxyServer() // without an active exception" error. An alternative to waiting would be // detach the thread, but this would introduce nondeterminism which could // make code harder to debug or extend. - assert(m_thread_context.waiter.get()); + assert(m_thread_context->waiter.get()); std::unique_ptr waiter; { - const Lock lock(m_thread_context.waiter->m_mutex); - //! Reset thread context waiter pointer, as shutdown signal for done - //! lambda passed as waiter->wait() argument in makeThread code below. - waiter = std::move(m_thread_context.waiter); - //! Assert waiter is idle. This destructor shouldn't be getting called if it is busy. + const Lock lock(m_thread_context->waiter->m_mutex); + // Reset thread context waiter pointer, as shutdown signal for the done + // lambda passed as waiter->wait() argument in WorkerThread constructor. + waiter = std::move(m_thread_context->waiter); + // Assert waiter is idle. The destructor shouldn't be called if busy. assert(!waiter->m_fn); - // Clear client maps now to avoid deadlock in m_thread.join() call - // below. The maps contain Thread::Client objects that need to be - // destroyed from the event loop thread (this thread), which can't - // happen if this thread is busy calling join. - m_thread_context.request_threads.clear(); - m_thread_context.callback_threads.clear(); - //! Ping waiter. + // Clear client maps now to avoid deadlock in m_thread.join() below. + // The maps contain Thread::Client objects that need to be destroyed + // from the event loop thread (this thread), which can't happen if this + // thread is busy calling join. + m_thread_context->request_threads.clear(); + m_thread_context->callback_threads.clear(); + // Ping waiter so it wakes and checks the null waiter shutdown signal. waiter->m_cv.notify_all(); } m_thread.join(); } +ProxyServer::ProxyServer(Connection& connection, std::string name) + : m_loop{*connection.m_loop}, m_worker(*connection.m_loop, std::move(name)) +{ +} + +ProxyServer::ProxyServer(Connection& connection, ThreadContext& existing) + : m_loop{*connection.m_loop}, m_worker(*connection.m_loop, existing) +{ +} + kj::Promise ProxyServer::getName(GetNameContext context) { - context.getResults().setResult(m_thread_context.thread_name); + context.getResults().setResult(m_worker.m_thread_context->thread_name); return kj::READY_NOW; } @@ -420,18 +445,8 @@ kj::Promise ProxyServer::makeThread(MakeThreadContext context) EventLoop& loop{*m_connection.m_loop}; if (loop.testing_hook_makethread) loop.testing_hook_makethread(); const std::string from = context.getParams().getName(); - std::promise thread_context; - std::thread thread([&loop, &thread_context, from]() { - g_thread_context.thread_name = ThreadName(loop.m_exe_name) + " (from " + from + ")"; - g_thread_context.waiter = std::make_unique(); - Lock lock(g_thread_context.waiter->m_mutex); - thread_context.set_value(&g_thread_context); - if (loop.testing_hook_makethread_created) loop.testing_hook_makethread_created(); - // Wait for shutdown signal from ProxyServer destructor (signal - // is just waiter getting set to null.) - g_thread_context.waiter->wait(lock, [] { return !g_thread_context.waiter; }); - }); - auto thread_server = kj::heap>(m_connection, *thread_context.get_future().get(), std::move(thread)); + auto thread_server = kj::heap>( + m_connection, ThreadName(loop.m_exe_name) + " (from " + from + ")"); auto thread_client = m_connection.m_threads.add(kj::mv(thread_server)); context.getResults().setResult(kj::mv(thread_client)); return kj::READY_NOW; From 9e0f2a80cd9f35b9a0385c1b0c41861d2aeeea5b Mon Sep 17 00:00:00 2001 From: Novo Date: Tue, 30 Jun 2026 10:52:21 +0200 Subject: [PATCH 3/6] feat: add ThreadPool to EventLoop Add ThreadPool class holding a fixed number of WorkerThreads dispatched round-robin. Wire it into EventLoop via a new num_pool_threads constructor parameter (default 0, no pool). When num_pool_threads > 0, m_thread_pool is created after m_task_set so WorkerThread destructors (which join their threads) run before the task set is destroyed. Pool threads are named '-N (pool I)' and are not Cap'n Proto capabilities - clients never hold Thread::Client references to them. PassField dispatch integration follows in the next commit. --- include/mp/proxy-io.h | 41 ++++++++++++++++++++++++++++++++++++++--- src/mp/proxy.cpp | 15 ++++++++++++++- 2 files changed, 52 insertions(+), 4 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 2985c19f..9ef80933 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -156,6 +156,31 @@ struct ProxyServer final : public Thread::Server WorkerThread m_worker; }; +//! Fixed-size pool of WorkerThreads. Requests are dispatched round-robin +//! across the pool. Owned by EventLoop when num_pool_threads > 0. +class ThreadPool +{ +public: + ThreadPool(EventLoop& loop, int num_threads); + + //! Dispatch fn to a pool WorkerThread (round-robin). Returns a promise + //! fulfilled on the event-loop thread when fn returns. + template + kj::Promise post(Fn&& fn); + +private: + std::vector> m_threads; + size_t m_next_thread{0}; +}; + +template +kj::Promise ThreadPool::post(Fn&& fn) +{ + WorkerThread& worker = *m_threads[m_next_thread % m_threads.size()]; + ++m_next_thread; + return worker.post(std::forward(fn)); +} + //! Handler for kj::TaskSet failed task events. class LoggingErrorHandler : public kj::TaskSet::ErrorHandler { @@ -281,11 +306,16 @@ class EventLoop { public: //! Construct event loop object with default logging options. - EventLoop(const char* exe_name, LogFn log_fn, void* context = nullptr) - : EventLoop(exe_name, LogOptions{std::move(log_fn)}, context){} + EventLoop(const char* exe_name, LogFn log_fn, void* context = nullptr, + int num_pool_threads = 0) + : EventLoop(exe_name, LogOptions{std::move(log_fn)}, context, num_pool_threads){} //! Construct event loop object with specified logging options. - EventLoop(const char* exe_name, LogOptions log_opts, void* context = nullptr); + //! If num_pool_threads > 0, a ThreadPool with that many WorkerThreads is + //! created and requests arriving with no dedicated thread in the Context + //! will be dispatched to it (see PassField in type-context.h). + EventLoop(const char* exe_name, LogOptions log_opts, void* context = nullptr, + int num_pool_threads = 0); //! Backwards-compatible constructor for previous (deprecated) logging callback signature EventLoop(const char* exe_name, std::function old_callback, void* context = nullptr) @@ -374,6 +404,11 @@ class EventLoop //! Capnp list of pending promises. std::unique_ptr m_task_set; + //! Optional thread pool. Declared after m_task_set so it is destroyed + //! first (reverse declaration order): WorkerThread destructors join their + //! threads before m_task_set is gone. + std::unique_ptr m_thread_pool; + //! List of connections. std::list m_incoming_connections; diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp index b906a908..ec0c3b22 100644 --- a/src/mp/proxy.cpp +++ b/src/mp/proxy.cpp @@ -36,6 +36,7 @@ #include #include #include +#include namespace mp { @@ -197,7 +198,17 @@ void EventLoop::addAsyncCleanup(std::function fn) startAsyncThread(); } -EventLoop::EventLoop(const char* exe_name, LogOptions log_opts, void* context) +ThreadPool::ThreadPool(EventLoop& loop, int num_threads) +{ + m_threads.reserve(num_threads); + for (int i = 0; i < num_threads; ++i) { + m_threads.push_back(std::make_unique( + loop, ThreadName(loop.m_exe_name) + " (pool " + std::to_string(i) + ")")); + } +} + +EventLoop::EventLoop(const char* exe_name, LogOptions log_opts, void* context, + int num_pool_threads) : m_exe_name(exe_name), m_io_context(kj::setupAsyncIo()), m_task_set(new kj::TaskSet(m_error_handler)), @@ -208,6 +219,8 @@ EventLoop::EventLoop(const char* exe_name, LogOptions log_opts, void* context) KJ_SYSCALL(socketpair(AF_UNIX, SOCK_STREAM, 0, fds)); m_wait_fd = fds[0]; m_post_fd = fds[1]; + if (num_pool_threads > 0) + m_thread_pool = std::make_unique(*this, num_pool_threads); } EventLoop::~EventLoop() From 74f50d0c9f83a3cf45d87ac0efe44081ce411f99 Mon Sep 17 00:00:00 2001 From: Novo Date: Tue, 30 Jun 2026 11:28:23 +0200 Subject: [PATCH 4/6] feat: PassField falls back to pool when no thread in context When a Context argument has no thread set, dispatch the request to the EventLoop's thread pool (if configured) instead of failing. Also adds a num_pool_threads parameter to TestSetup and a test verifying pool lifecycle and direct work dispatch. --- include/mp/type-context.h | 15 +++++++++++++++ test/mp/test/test.cpp | 6 +++--- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/include/mp/type-context.h b/include/mp/type-context.h index 6627c3e4..7706eea3 100644 --- a/include/mp/type-context.h +++ b/include/mp/type-context.h @@ -200,6 +200,21 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& // asynchronously with getLocalServer(). const auto& params = server_context.call_context.getParams(); Context::Reader context_arg = Accessor::get(params); + + // If context has no thread set, fall back to the event loop's thread pool + // (if configured). Pool threads are not Cap'n Proto capabilities, so the + // client never holds a Thread::Client reference to them. + if (!context_arg.hasThread()) { + if (loop.m_thread_pool) { + MP_LOG(loop, Log::Debug) << "IPC server post request #" << req << " {pool}"; + return server.m_context.connection->m_canceler.wrap( + loop.m_thread_pool->post(std::move(invoke))); + } + MP_LOG(loop, Log::Error) + << "IPC server error request #" << req << ", no thread in context and no thread pool configured"; + throw std::runtime_error("no thread in context and no thread pool configured"); + } + auto thread_client = context_arg.getThread(); auto result = server.m_context.connection->m_threads.getLocalServer(thread_client) .then([&loop, invoke = kj::mv(invoke), req](const kj::Maybe& perhaps) mutable { diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index 1f62599a..8b44cb4e 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -83,13 +83,13 @@ class TestSetup //! not start until the other members are initialized. std::thread thread; - TestSetup(bool client_owns_connection = true) - : thread{[&] { + TestSetup(bool client_owns_connection = true, int num_pool_threads = 0) + : thread{[&, num_pool_threads] { EventLoop loop("mptest", [](mp::LogMessage log) { // Info logs are not printed by default, but will be shown with `mptest --verbose` KJ_LOG(INFO, log.level, log.message); if (log.level == mp::Log::Raise) throw std::runtime_error(log.message); - }); + }, nullptr, num_pool_threads); auto pipe = loop.m_io_context.provider->newTwoWayPipe(); auto server_connection = From a8e54134fa3a20437f6d14a3bbef0dec6c02c6cd Mon Sep 17 00:00:00 2001 From: Novo Date: Tue, 30 Jun 2026 19:45:30 +0200 Subject: [PATCH 5/6] use ThreadPool if threadmap is unset --- include/mp/proxy-io.h | 4 +- include/mp/type-context.h | 73 ++++++++++++------------------------- include/mp/type-threadmap.h | 2 +- 3 files changed, 28 insertions(+), 51 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 9ef80933..d9cd1aaa 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -548,7 +548,9 @@ class Connection // ThreadMap interface client, used to create a remote server thread when an // client IPC call is being made for the first time from a new thread. - ThreadMap::Client m_thread_map{nullptr}; + // Empty (unset) when the client has not called initThreadMap(), in which + // case the server is expected to dispatch requests via a thread pool. + std::optional m_thread_map; //! Collection of server-side IPC worker threads (ProxyServer objects previously returned by //! ThreadMap.makeThread) used to service requests to clients. diff --git a/include/mp/type-context.h b/include/mp/type-context.h index 7706eea3..5dbf04e7 100644 --- a/include/mp/type-context.h +++ b/include/mp/type-context.h @@ -30,27 +30,32 @@ void CustomBuildField(TypeList<>, GuardedRef{thread_context.waiter->m_mutex, thread_context.callback_threads}, &connection, [&] { return connection.m_threads.add(kj::heap>(connection, thread_context)); })}; + auto context = output.init(); + context.setCallbackThread(callback_thread->second->m_client); + // Call remote ThreadMap.makeThread function so server will create a // dedicated worker thread to run function calls from this thread. Store the // Thread::Client reference it returns in the request_threads map. - auto make_request_thread{[&]{ - // This code will only run if an IPC client call is being made for the - // first time on this thread. After the first call, subsequent calls - // will use the existing request thread. This code will also never run at - // all if the current thread is a request thread created for a different - // IPC client, because in that case PassField code (below) will have set - // request_thread to point to the calling thread. - auto request = connection.m_thread_map.makeThreadRequest(); - request.setName(thread_context.thread_name); - return request.send().getResult(); // Nonblocking due to capnp request pipelining. - }}; - auto [request_thread, _1]{SetThread( - GuardedRef{thread_context.waiter->m_mutex, thread_context.request_threads}, - &connection, make_request_thread)}; - - auto context = output.init(); - context.setThread(request_thread->second->m_client); - context.setCallbackThread(callback_thread->second->m_client); + // Skip this if m_thread_map is not initialized, in which case + // the context.thread field is left unset and the server is + // expected to dispatch the request via its thread pool. + if (connection.m_thread_map) { + auto make_request_thread{[&]{ + // This code will only run if an IPC client call is being made for the + // first time on this thread. After the first call, subsequent calls + // will use the existing request thread. This code will also never run at + // all if the current thread is a request thread created for a different + // IPC client, because in that case PassField code (below) will have set + // request_thread to point to the calling thread. + auto request = connection.m_thread_map->makeThreadRequest(); + request.setName(thread_context.thread_name); + return request.send().getResult(); // Nonblocking due to capnp request pipelining. + }}; + auto [request_thread, _1]{SetThread( + GuardedRef{thread_context.waiter->m_mutex, thread_context.request_threads}, + &connection, make_request_thread)}; + context.setThread(request_thread->second->m_client); + } } //! PassField override for mp.Context arguments. Return asynchronously and call @@ -232,38 +237,8 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& throw std::runtime_error("invalid thread handle"); } }, [&loop, req](::kj::Exception&& e) -> kj::Promise { - // If you see the error "(remote):0: failed: remote exception: - // Called null capability" here, it probably means your Init class - // is missing a declaration like: - // - // construct @0 (threadMap: Proxy.ThreadMap) -> (threadMap :Proxy.ThreadMap); - // - // which passes a ThreadMap reference from the client to the server, - // allowing the server to create threads to run IPC calls on the - // client, and also returns a ThreadMap reference from the server to - // the client, allowing the client to create threads on the server. - // (Typically the latter ThreadMap is used more often because there - // are more client-to-server calls.) - // - // If the other side of the connection did not previously get a - // ThreadMap reference from this side of the connection, when the - // other side calls `m_thread_map.makeThreadRequest()` in - // `BuildField` above, `m_thread_map` will be null, but that call - // will not fail immediately due to Cap'n Proto's request pipelining - // and delayed execution. Instead that call will return an invalid - // Thread reference, and when that reference is passed to this side - // of the connection as `thread_client` above, the - // `getLocalServer(thread_client)` call there will be the first - // thing to overtly fail, leading to an error here. - // - // Potentially there are also other things that could cause errors - // here, but this is the most likely cause. - // - // The log statement here is not strictly necessary since the same - // exception will also be logged in serverInvoke, but this logging - // may provide extra context that could be helpful for debugging. MP_LOG(loop, Log::Info) - << "IPC server error request #" << req << " CapabilityServerSet::getLocalServer call failed, did you forget to provide a ThreadMap to the client prior to this IPC call?"; + << "IPC server error request #" << req << " CapabilityServerSet::getLocalServer call failed"; return kj::mv(e); }); // Use connection m_canceler object to cancel the result promise if the diff --git a/include/mp/type-threadmap.h b/include/mp/type-threadmap.h index 3005d9de..c3ae33e9 100644 --- a/include/mp/type-threadmap.h +++ b/include/mp/type-threadmap.h @@ -34,7 +34,7 @@ decltype(auto) CustomReadField(TypeList<>, Input&& input, typename std::enable_if::value>::type* enable = nullptr) { - invoke_context.connection.m_thread_map = input.get(); + invoke_context.connection.m_thread_map.emplace(input.get()); } } // namespace mp From 633579fe9df5bac8ce467405599f8c015f207698 Mon Sep 17 00:00:00 2001 From: Novo Date: Tue, 30 Jun 2026 19:33:57 +0200 Subject: [PATCH 6/6] test: update async tests to also use threadpool --- test/mp/test/test.cpp | 162 +++++++++++++++++++++++++----------------- 1 file changed, 96 insertions(+), 66 deletions(-) diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index 8b44cb4e..f60426a4 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -137,6 +137,20 @@ class TestSetup } }; +//! Run a test body twice: once with on-demand threads (initThreadMap), once with a thread pool. +#define RUN_WITH_ASYNC_VARIANTS(client_owns_connection, ...) \ + { \ + TestSetup setup{client_owns_connection}; \ + ProxyClient* foo = setup.client.get(); \ + foo->initThreadMap(); \ + __VA_ARGS__ \ + } \ + { \ + TestSetup setup{client_owns_connection, 1}; \ + ProxyClient* foo = setup.client.get(); \ + __VA_ARGS__ \ + } + KJ_TEST("Call FooInterface methods") { TestSetup setup; @@ -305,27 +319,45 @@ KJ_TEST("Calling IPC method, disconnecting and blocking during the call") // the ~Connection destructor usually would immediately free all remaining // ProxyServer objects associated with the connection. Having an in-progress // RPC call requires keeping the ProxyServer longer. + { + std::promise signal; + TestSetup setup{/*client_owns_connection=*/false}; + ProxyClient* foo = setup.client.get(); + foo->initThreadMap(); + KJ_EXPECT(foo->add(1, 2) == 3); + + setup.server->m_impl->m_fn = [&] { + EventLoopRef loop{*setup.server->m_context.loop}; + setup.client_disconnect(); + signal.get_future().get(); + }; + + EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); + + // Now that the disconnect has been detected, set signal allowing the + // callFnAsync() IPC call to return. Since signalling may not wake up the + // thread right away, it is important for the signal variable to be declared + // *before* the TestSetup variable so is not destroyed while + // signal.get_future().get() is called. + signal.set_value(); + } + // Test with ThreadPool + { + std::promise signal; + TestSetup setup{false, 1}; + ProxyClient* foo = setup.client.get(); + KJ_EXPECT(foo->add(1, 2) == 3); - std::promise signal; - TestSetup setup{/*client_owns_connection=*/false}; - ProxyClient* foo = setup.client.get(); - KJ_EXPECT(foo->add(1, 2) == 3); - - foo->initThreadMap(); - setup.server->m_impl->m_fn = [&] { - EventLoopRef loop{*setup.server->m_context.loop}; - setup.client_disconnect(); - signal.get_future().get(); - }; + setup.server->m_impl->m_fn = [&] { + EventLoopRef loop{*setup.server->m_context.loop}; + setup.client_disconnect(); + signal.get_future().get(); + }; - EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); + EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); - // Now that the disconnect has been detected, set signal allowing the - // callFnAsync() IPC call to return. Since signalling may not wake up the - // thread right away, it is important for the signal variable to be declared - // *before* the TestSetup variable so is not destroyed while - // signal.get_future().get() is called. - signal.set_value(); + signal.set_value(); + } } KJ_TEST("Worker thread destroyed before it is initialized") @@ -367,20 +399,19 @@ KJ_TEST("Calling async IPC method, with server disconnect racing the call") // worker thread as soon as it begins to execute an async request. Without // the bugfix, the worker thread would trigger a SIGSEGV after this by // calling call_context.getParams(). - TestSetup setup; - ProxyClient* foo = setup.client.get(); - foo->initThreadMap(); - setup.server->m_impl->m_fn = [] {}; - - EventLoop& loop = *setup.server->m_context.connection->m_loop; - loop.testing_hook_async_request_start = [&] { - setup.server_disconnect(); - // Sleep is necessary to let the event loop fully clean up after the - // disconnect and trigger the SIGSEGV. - std::this_thread::sleep_for(std::chrono::milliseconds(10)); - }; - - EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); + RUN_WITH_ASYNC_VARIANTS(true, + setup.server->m_impl->m_fn = [] {}; + + EventLoop& loop = *setup.server->m_context.connection->m_loop; + loop.testing_hook_async_request_start = [&] { + setup.server_disconnect(); + // Sleep is necessary to let the event loop fully clean up after the + // disconnect and trigger the SIGSEGV. + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + }; + + EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); + ) } KJ_TEST("Calling async IPC method, with server disconnect after cleanup") @@ -394,17 +425,16 @@ KJ_TEST("Calling async IPC method, with server disconnect after cleanup") // Without the bugfix, the m_on_cancel callback would be called at this // point, accessing the cancel_mutex stack variable that had gone out of // scope. - TestSetup setup; - ProxyClient* foo = setup.client.get(); - foo->initThreadMap(); - setup.server->m_impl->m_fn = [] {}; + RUN_WITH_ASYNC_VARIANTS(true, + setup.server->m_impl->m_fn = [] {}; - EventLoop& loop = *setup.server->m_context.connection->m_loop; - loop.testing_hook_async_request_done = [&] { - setup.server_disconnect(); - }; + EventLoop& loop = *setup.server->m_context.connection->m_loop; + loop.testing_hook_async_request_done = [&] { + setup.server_disconnect(); + }; - EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); + EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); + ) } KJ_TEST("Destroying ProxyClient<> with destroy method after peer disconnect") @@ -419,18 +449,16 @@ KJ_TEST("Destroying ProxyClient<> with destroy method after peer disconnect") // the now dead connection; without the bugfix the exception escapes the // noexcept destructor and aborts the process. - TestSetup setup{/*client_owns_connection=*/false}; - ProxyClient* foo = setup.client.get(); - foo->initThreadMap(); - class Callback : public FooCallback { public: int call(int arg) override { return arg; } }; - foo->saveCallback(std::make_shared()); - setup.client_disconnect(); + RUN_WITH_ASYNC_VARIANTS(false, + foo->saveCallback(std::make_shared()); + setup.client_disconnect(); + ) } KJ_TEST("Make simultaneous IPC calls on single remote thread") @@ -490,26 +518,28 @@ KJ_TEST("Make simultaneous IPC calls on single remote thread") KJ_TEST("Make IPC call to same remote thread while it is waiting for callback") { - TestSetup setup; - ProxyClient* foo = setup.client.get(); - - class Callback : public FooCallback - { - public: - Callback(ProxyClient* foo, int expect) : m_foo(foo), m_expect(expect) {} - int call(int arg) override + // `RUN_WITH_ASYNC_VARIANTS` sets up a ThreadPool + // with one thread, so the async request from the + // callback has to be processed by the same remote + // thread + RUN_WITH_ASYNC_VARIANTS(true, + class Callback : public FooCallback { - KJ_EXPECT(arg == m_expect); - return m_foo->passFn([this]{ return m_expect; }); - } - - ProxyClient* m_foo; - int m_expect; - }; - - foo->initThreadMap(); - Callback callback(foo, 1); - KJ_EXPECT(foo->callback(callback, 1) == 1); + public: + Callback(ProxyClient* foo, int expect) : m_foo(foo), m_expect(expect) {} + int call(int arg) override + { + KJ_EXPECT(arg == m_expect); + return m_foo->passFn([this]{ return m_expect; }); + } + + ProxyClient* m_foo; + int m_expect; + }; + + Callback callback(foo, 1); + KJ_EXPECT(foo->callback(callback, 1) == 1); + ) } } // namespace test