diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 092ea42e..d9cd1aaa 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,13 +153,34 @@ 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; +}; + +//! 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 { @@ -239,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) @@ -332,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; @@ -471,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. @@ -740,36 +819,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 +859,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 +879,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..5dbf04e7 100644 --- a/include/mp/type-context.h +++ b/include/mp/type-context.h @@ -28,29 +28,34 @@ 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)); })}; + + 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 @@ -200,6 +205,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 { @@ -209,7 +229,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) @@ -217,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 diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp index 963050c3..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() @@ -372,13 +385,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) +{ + 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(); +} + +WorkerThread::WorkerThread(EventLoop& loop, ThreadContext& existing) + : m_loop(loop), m_thread_context(&existing) { - assert(m_thread_context.waiter.get() != nullptr); + assert(m_thread_context->waiter.get() != nullptr); } -ProxyServer::~ProxyServer() +WorkerThread::~WorkerThread() { if (!m_thread.joinable()) return; // Stop async thread and wait for it to exit. Need to wait because the @@ -386,30 +414,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 +458,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; diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index eb4b5ec7..f60426a4 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 { @@ -74,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 = @@ -128,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; @@ -251,14 +274,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 +284,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 +297,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") @@ -317,34 +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); + setup.server->m_impl->m_fn = [&] { + EventLoopRef loop{*setup.server->m_context.loop}; + setup.client_disconnect(); + signal.get_future().get(); + }; - foo->initThreadMap(); - 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."); - 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; + signal.set_value(); } - KJ_EXPECT(disconnected); - - // 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(); } KJ_TEST("Worker thread destroyed before it is initialized") @@ -374,14 +387,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") @@ -393,25 +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)); - }; - - 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."); - } + 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") @@ -425,22 +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(); + }; - 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") @@ -455,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") @@ -524,5 +516,31 @@ 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") +{ + // `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 + { + 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 } // namespace mp