perf: run handlers queued behind the reactor before polling again

include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp

@@ -333,6 +333,9 @@ class reactor_scheduler : public scheduler
     std::atomic<bool> stopped_{false};
     mutable std::atomic<bool> task_running_{false};
     mutable bool task_interrupted_ = false;
+    // Set while the sentinel waits behind queued handlers (guarded by
+    // mutex_).
+    bool task_deferred_ = false;
 
     // Descriptors retired during the running reactor pass (guarded by
     // mutex_).
@@ -631,6 +634,10 @@ reactor_scheduler::stopped() const noexcept
 inline void
 reactor_scheduler::restart()
 {
+    // A run can end with the sentinel marked deferred; clearing the
+    // mark lets the next run defer it again.
+    lock_type lock(mutex_);
+    task_deferred_ = false;
     stopped_.store(false, std::memory_order_release);
 }
 
@@ -1013,6 +1020,19 @@ reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx)
         {
             bool more_handlers = !completed_ops_.empty();
 
+            // Handlers queued here were mostly posted by the pass that
+            // just ran, so polling ahead of them usually finds nothing.
+            // Defer at most once in a row so reposting work cannot
+            // starve I/O. poll() and poll_one() don't defer: the reactor
+            // turn they would skip may be their last before returning.
+            if (more_handlers && !task_deferred_ && timeout_us != 0)
+            {
+                task_deferred_ = true;
+                completed_ops_.push(&task_op_);
+                continue;
+            }
+            task_deferred_ = false;
+
             if (!more_handlers &&
                 (outstanding_work_.load(std::memory_order_acquire) == 0 ||
                  timeout_us == 0))

test/unit/fault/kqueue_faults.cpp

@@ -329,14 +329,12 @@ struct kqueue_faults
 
     void testRunLoopFaults()
     {
-        // Nothing calls kevent between the constructor and the run
-        // loop's first wait: posting the task does not interrupt a
-        // reactor that is not running yet
-        // (reactor_scheduler::wake_one_thread_and_unlock), so nth 1 is
-        // that wait.
+        // The run loop runs the posted task before its first wait, and
+        // arming the delay triggers the user event, so nth 1 is that
+        // trigger and nth 2 is the wait.
         {
             io_context ioc(kqueue);
-            fault_scope f(sys::kevent, EINTR);
+            fault_scope f(sys::kevent, EINTR, 2);
             bool done = false;
             auto body = [&]() -> capy::task<> {
                 std::ignore =
@@ -352,7 +350,7 @@ struct kqueue_faults
             // Any other errno leaves run() through an exception
             // (kqueue_scheduler::run_task).
             io_context ioc(kqueue);
-            fault_scope f(sys::kevent, EBADF);
+            fault_scope f(sys::kevent, EBADF, 2);
             auto body = [&]() -> capy::task<> {
                 std::ignore =
                     co_await corosio::delay(std::chrono::milliseconds(1));
@@ -377,10 +375,9 @@ struct kqueue_faults
     {
         io_context ioc(kqueue);
         {
-            // stop() is the one path that reaches NOTE_TRIGGER without
-            // a reactor thread (reactor_scheduler::stop), and it
-            // interrupts only on the transition, so each stop below is
-            // one kevent.
+            // stop() reaches NOTE_TRIGGER with no other kevent call in
+            // between, and it interrupts only on the transition, so each
+            // stop below is one kevent.
             fault_scope first(sys::kevent, EIO, 1);
             fault_scope second(sys::kevent, EIO, 2);
             ioc.stop();

test/unit/fault/reactor_faults.hpp

@@ -27,6 +27,7 @@
 #include <boost/corosio/test/socket_pair.hpp>
 #include <boost/corosio/udp_socket.hpp>
 #include <boost/corosio/wait_type.hpp>
+#include <boost/capy/ex/async_waker.hpp>
 #include <boost/capy/ex/run_async.hpp>
 #include <boost/capy/task.hpp>
 
@@ -84,6 +85,9 @@ struct reactor_common_faults
     static constexpr sys spec_write = sys::write;
 #endif
     static constexpr sys vec_write = is_kqueue ? sys::writev : sys::sendmsg;
+    static constexpr sys wait_sys  = is_select ? sys::select
+        : is_kqueue                            ? sys::kevent
+                                               : sys::epoll_wait;
 
     static endpoint loopback()
     {
@@ -1034,6 +1038,73 @@ struct reactor_common_faults
 #endif
     }
 
+    // A handler that an I/O completion posts runs before the reactor is
+    // polled again: right after a pass, a poll would find nothing new.
+    void testPostedWakeRunsBeforeNextPoll()
+    {
+        io_context ioc(Backend);
+        auto [a, b] = test::must(test::make_socket_pair(ioc));
+        capy::async_waker waker;
+        unsigned at_read = 0;
+        unsigned at_wake = 0;
+        // Counts this thread's calls to the wait symbol (every kevent on
+        // kqueue); the one it would fail is never reached.
+        fault_scope f(wait_sys, EINTR, 1u << 30);
+
+        auto waiter = [&]() -> capy::task<> {
+            auto [ec] = co_await waker.wait();
+            BOOST_TEST(!ec);
+            at_wake = f.count();
+        };
+        auto reader = [&]() -> capy::task<> {
+            char buf[1];
+            auto [ec, n] = co_await b.read_some(capy::mutable_buffer(buf, 1));
+            BOOST_TEST(!ec);
+            at_read = f.count();
+            waker.wake();
+        };
+        auto writer = [&]() -> capy::task<> {
+            char const c = 'x';
+            auto [ec, n] = co_await a.write_some(capy::const_buffer(&c, 1));
+            BOOST_TEST(!ec);
+        };
+        capy::run_async(ioc.get_executor())(waiter());
+        capy::run_async(ioc.get_executor())(reader());
+        capy::run_async(ioc.get_executor())(writer());
+        ioc.run();
+
+        // The read parked, so a poll completed it.
+        BOOST_TEST(at_read > 0);
+        BOOST_TEST_EQ(at_wake, at_read);
+    }
+
+    // A run stopped while the sentinel was deferred leaves the next run
+    // its own deferral, so a posted handler still runs before a poll.
+    void testRestartRenewsDeferral()
+    {
+        io_context ioc(Backend);
+        auto stopper = [&]() -> capy::task<> {
+            ioc.stop();
+            co_return;
+        };
+        capy::run_async(ioc.get_executor())(stopper());
+        ioc.run();
+        ioc.restart();
+
+        // Counts this thread's calls to the wait symbol (every kevent on
+        // kqueue); the one it would fail is never reached.
+        fault_scope f(wait_sys, EINTR, 1u << 30);
+        unsigned at_run = 1;
+        auto probe      = [&]() -> capy::task<> {
+            at_run = f.count();
+            co_return;
+        };
+        capy::run_async(ioc.get_executor())(probe());
+        ioc.run();
+
+        BOOST_TEST_EQ(at_run, 0u);
+    }
+
     // Accepted and assigned sockets learn their family from the
     // addresses the library already reads, so option calls never ask.
     void testAdoptedSocketsKnowFamily()
@@ -1352,6 +1423,8 @@ struct reactor_common_faults
         testSocketFamilyProbeFails();
         testOpenedSocketKnowsFamily();
         testAdoptedSocketsKnowFamily();
+        testPostedWakeRunsBeforeNextPoll();
+        testRestartRenewsDeferral();
         testDatagramFails();
     }
 };