refactor: Replace EventLoop::post() with sync() taking kj::FunctionParam

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<void()> 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 <typename Callable>
-    void sync(Callable&& callable)
-    {
-        post(std::forward<Callable>(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<void()> 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<void()>* m_post_fn MP_GUARDED_BY(m_mutex) = nullptr;
+    //! Callback function to run on event loop thread during sync() call.
+    kj::FunctionParam<void()>* m_sync_fn MP_GUARDED_BY(m_mutex) = nullptr;
 
     //! Callback functions to run on async thread.
     std::optional<CleanupList> m_async_fns MP_GUARDED_BY(m_mutex);

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,18 +304,18 @@ 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
-            // of EventLoop::post() will return without any indication of failure,
+        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()
-            // would return true, to ensure that the post() m_post_writer->write()
+            // 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.
             break;
@@ -333,21 +333,21 @@ void EventLoop::loop()
     m_cv.notify_all();
 }
 
-void EventLoop::post(kj::Function<void()> fn)
+void EventLoop::sync(kj::FunctionParam<void()> fn)
 {
     if (std::this_thread::get_id() == m_thread_id) {
         fn();
         return;
     }
     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()

test/mp/test/listen_tests.cpp

@@ -16,6 +16,7 @@
 #include <kj/async.h>
 #include <kj/common.h>
 #include <kj/debug.h>
+#include <kj/function.h>
 #include <kj/memory.h>
 #include <kj/test.h>
 #include <memory>

test/mp/test/test.cpp

@@ -20,6 +20,7 @@
 #include <kj/common.h>
 #include <kj/exception.h>
 #include <kj/debug.h>
+#include <kj/function.h>
 #include <kj/memory.h>
 #include <kj/string.h>
 #include <kj/test.h>