From 97fbd80fe67070d97a353c84e98bfb4db68fd119 Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Wed, 11 Nov 2026 23:15:38 +0100 Subject: [PATCH] perf: run handlers queued behind the reactor before polling again When the reactor sentinel reached the front with handlers still queued, the scheduler polled the reactor without blocking. Those handlers were mostly posted by the pass that just ran, such as a coroutine an I/O completion woke, so the poll usually found nothing. The sentinel now moves behind the queued handlers once and polls on its next turn regardless, so handlers that keep reposting still cannot starve I/O. restart() forgets a deferral still pending, so a run stopped with the sentinel moved back doesn't cost the next run its own. poll() and poll_one() are unchanged. --- .../detail/reactor/reactor_scheduler.hpp | 20 +++++ test/unit/fault/kqueue_faults.cpp | 19 ++--- test/unit/fault/reactor_faults.hpp | 73 +++++++++++++++++++ 3 files changed, 101 insertions(+), 11 deletions(-) diff --git a/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp b/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp index 1f4bed9ca..b2b3cf056 100644 --- a/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp +++ b/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp @@ -333,6 +333,9 @@ class reactor_scheduler : public scheduler std::atomic stopped_{false}; mutable std::atomic 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)) diff --git a/test/unit/fault/kqueue_faults.cpp b/test/unit/fault/kqueue_faults.cpp index 19da85ba5..77c36aa0e 100644 --- a/test/unit/fault/kqueue_faults.cpp +++ b/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(); diff --git a/test/unit/fault/reactor_faults.hpp b/test/unit/fault/reactor_faults.hpp index d4fab93f7..16c019be7 100644 --- a/test/unit/fault/reactor_faults.hpp +++ b/test/unit/fault/reactor_faults.hpp @@ -27,6 +27,7 @@ #include #include #include +#include #include #include @@ -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(); } };