From 2330fbe81e255996710e3e258cb6a5ea94cfb250 Mon Sep 17 00:00:00 2001 From: ViniciusCestarii Date: Sun, 16 Aug 2026 11:13:46 -0300 Subject: [PATCH 1/2] refactor: replace EventLoop::post() with sync() taking kj::FunctionParam kj::FunctionParam only stores a pointer to the caller's callable instead of taking ownership of it. This avoids a heap allocation per call, and is safe because EventLoop:sync() is synchronous, so the callable is guaranteed not to go out of scope before the event loop thread is done running it. It also means the callable does not need to be copyable or movable. --- include/mp/proxy-io.h | 19 +++++-------------- src/mp/proxy.cpp | 6 +++--- test/mp/test/listen_tests.cpp | 1 + test/mp/test/test.cpp | 1 + 4 files changed, 10 insertions(+), 17 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index cda9064d..b56f7d08 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -233,8 +233,8 @@ Stream MakeStream(EventLoop&loop, SocketId socket); //! assumes it will only be accessed from one thread. So all this code needs to //! actually run on one thread, and the EventLoop::loop() method is the entry point for //! this thread. ProxyClient and ProxyServer objects that use other threads and -//! need to perform I/O operations post to this thread using EventLoop::post() -//! and EventLoop::sync() methods. +//! need to perform I/O operations post to this thread using the +//! EventLoop::sync() method. //! //! Specifically, because ProxyClient methods can be called from arbitrary //! threads, and ProxyServer methods can run on arbitrary threads, ProxyClient @@ -267,16 +267,7 @@ class EventLoop //! Run function on event loop thread. Does not return until function completes. //! Must be called while the loop() function is active. - void post(kj::Function fn); - - //! Wrapper around EventLoop::post that takes advantage of the - //! fact that callable will not go out of scope to avoid requirement that it - //! be copyable. - template - void sync(Callable&& callable) - { - post(std::forward(callable)); - } + void sync(kj::FunctionParam fn); //! Register cleanup function to run on asynchronous worker thread without //! blocking the event loop thread. @@ -309,8 +300,8 @@ class EventLoop //! method has not been called. std::thread m_async_thread; - //! Callback function to run on event loop thread during post() or sync() call. - kj::Function* m_post_fn MP_GUARDED_BY(m_mutex) = nullptr; + //! Callback function to run on event loop thread during sync() call. + kj::FunctionParam* m_post_fn MP_GUARDED_BY(m_mutex) = nullptr; //! Callback functions to run on async thread. std::optional m_async_fns MP_GUARDED_BY(m_mutex); diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp index eb2aee0c..ae05ca6d 100644 --- a/src/mp/proxy.cpp +++ b/src/mp/proxy.cpp @@ -306,7 +306,7 @@ void EventLoop::loop() Lock lock(m_mutex); if (m_post_fn) { // m_post_fn throwing is never expected. If it does happen, the caller - // of EventLoop::post() will return without any indication of failure, + // of EventLoop::sync() will return without any indication of failure, // which will likely cause other bugs. Log the error and continue. KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() MP_REQUIRES(m_mutex) { Unlock(lock, *m_post_fn); })) { MP_LOG(*this, Log::Error) << "EventLoop: m_post_fn threw: " << kj::str(*exception).cStr(); @@ -315,7 +315,7 @@ void EventLoop::loop() m_cv.notify_all(); } else if (done()) { // Intentionally do not break if m_post_fn was set, even if done() - // would return true, to ensure that the post() m_post_writer->write() + // would return true, to ensure that the sync() m_post_writer->write() // call always succeeds and the loop does not exit between the time // that the done condition is set and the write call is made. break; @@ -333,7 +333,7 @@ void EventLoop::loop() m_cv.notify_all(); } -void EventLoop::post(kj::Function fn) +void EventLoop::sync(kj::FunctionParam fn) { if (std::this_thread::get_id() == m_thread_id) { fn(); diff --git a/test/mp/test/listen_tests.cpp b/test/mp/test/listen_tests.cpp index 8d06cb35..41fb05aa 100644 --- a/test/mp/test/listen_tests.cpp +++ b/test/mp/test/listen_tests.cpp @@ -16,6 +16,7 @@ #include #include #include +#include #include #include #include diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index 400aff73..5e1b010e 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -20,6 +20,7 @@ #include #include #include +#include #include #include #include From 5c49666a1d8860028b6ed04b3aead64d06979d6d Mon Sep 17 00:00:00 2001 From: ViniciusCestarii Date: Sun, 16 Aug 2026 11:54:05 -0300 Subject: [PATCH 2/2] refactor: rename EventLoop::m_post_fn to m_sync_fn --- include/mp/proxy-io.h | 2 +- src/mp/proxy.cpp | 20 ++++++++++---------- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index b56f7d08..d716b8ec 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -301,7 +301,7 @@ class EventLoop std::thread m_async_thread; //! Callback function to run on event loop thread during sync() call. - kj::FunctionParam* m_post_fn MP_GUARDED_BY(m_mutex) = nullptr; + kj::FunctionParam* m_sync_fn MP_GUARDED_BY(m_mutex) = nullptr; //! Callback functions to run on async thread. std::optional m_async_fns MP_GUARDED_BY(m_mutex); diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp index ae05ca6d..552e3268 100644 --- a/src/mp/proxy.cpp +++ b/src/mp/proxy.cpp @@ -275,7 +275,7 @@ EventLoop::~EventLoop() { if (m_async_thread.joinable()) m_async_thread.join(); const Lock lock(m_mutex); - KJ_ASSERT(m_post_fn == nullptr); + KJ_ASSERT(m_sync_fn == nullptr); KJ_ASSERT(!m_async_fns); KJ_ASSERT(!m_wait_stream); KJ_ASSERT(!m_post_stream); @@ -304,17 +304,17 @@ void EventLoop::loop() const size_t read_bytes = wait_stream->read(&buffer, 0, 1).wait(m_io_context.waitScope); if (read_bytes != 1) throw std::logic_error("EventLoop wait_stream closed unexpectedly"); Lock lock(m_mutex); - if (m_post_fn) { - // m_post_fn throwing is never expected. If it does happen, the caller + if (m_sync_fn) { + // m_sync_fn throwing is never expected. If it does happen, the caller // of EventLoop::sync() will return without any indication of failure, // which will likely cause other bugs. Log the error and continue. - KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() MP_REQUIRES(m_mutex) { Unlock(lock, *m_post_fn); })) { - MP_LOG(*this, Log::Error) << "EventLoop: m_post_fn threw: " << kj::str(*exception).cStr(); + KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() MP_REQUIRES(m_mutex) { Unlock(lock, *m_sync_fn); })) { + MP_LOG(*this, Log::Error) << "EventLoop: m_sync_fn threw: " << kj::str(*exception).cStr(); } - m_post_fn = nullptr; + m_sync_fn = nullptr; m_cv.notify_all(); } else if (done()) { - // Intentionally do not break if m_post_fn was set, even if done() + // Intentionally do not break if m_sync_fn was set, even if done() // would return true, to ensure that the sync() m_post_writer->write() // call always succeeds and the loop does not exit between the time // that the done condition is set and the write call is made. @@ -341,13 +341,13 @@ void EventLoop::sync(kj::FunctionParam fn) } Lock lock(m_mutex); EventLoopRef ref(*this, &lock); - m_cv.wait(lock.m_lock, [this]() MP_REQUIRES(m_mutex) { return m_post_fn == nullptr; }); - m_post_fn = &fn; + m_cv.wait(lock.m_lock, [this]() MP_REQUIRES(m_mutex) { return m_sync_fn == nullptr; }); + m_sync_fn = &fn; Unlock(lock, [&] { char buffer = 0; m_post_writer->write(&buffer, 1); }); - m_cv.wait(lock.m_lock, [this, &fn]() MP_REQUIRES(m_mutex) { return m_post_fn != &fn; }); + m_cv.wait(lock.m_lock, [this, &fn]() MP_REQUIRES(m_mutex) { return m_sync_fn != &fn; }); } void EventLoop::startAsyncThread()