From d6f74433701985a3a934c19585cc5c7928b27cbd 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 --- include/mp/proxy-io.h | 26 ++++++++++++-------------- src/mp/proxy.cpp | 6 +++--- test/mp/test/listen_tests.cpp | 1 + test/mp/test/test.cpp | 1 + 4 files changed, 17 insertions(+), 17 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index cda9064d..ba6b1731 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,14 @@ 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)); - } + //! + //! The function is accepted as a kj::FunctionParam, which 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 this call 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. + void sync(kj::FunctionParam fn); //! Register cleanup function to run on asynchronous worker thread without //! blocking the event loop thread. @@ -309,8 +307,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 2b23e78b35fe0208a782db17e49b04c4f03aa088 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 ba6b1731..2ce82cf4 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -308,7 +308,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()