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(); } };