diff --git a/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp b/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp index 1f4bed9ca..e3d445012 100644 --- a/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp +++ b/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp @@ -1014,8 +1014,7 @@ reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx) bool more_handlers = !completed_ops_.empty(); if (!more_handlers && - (outstanding_work_.load(std::memory_order_acquire) == 0 || - timeout_us == 0)) + outstanding_work_.load(std::memory_order_acquire) == 0) { completed_ops_.push(&task_op_); return 0; @@ -1041,10 +1040,19 @@ reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx) } task_running_.store(false, std::memory_order_relaxed); + bool const reaped = !completed_ops_.empty(); completed_ops_.push(&task_op_); settle_retired(lock); - if (timeout_us > 0) + + // wait_one() returns after one pass; poll() and poll_one() + // return after a pass that reaps nothing. Either way, wake a + // parked peer to take the sentinel. + if (timeout_us > 0 || (timeout_us == 0 && !reaped)) + { + if (!one_thread_) + maybe_unlock_and_signal_one(lock); return 0; + } continue; } diff --git a/test/unit/bounded_run.cpp b/test/unit/bounded_run.cpp index f0402d7b0..012aa9e51 100644 --- a/test/unit/bounded_run.cpp +++ b/test/unit/bounded_run.cpp @@ -9,8 +9,8 @@ // Bounded run variants (run_for / run_one_for / poll) on the reactor // schedulers. Durations here bound how long a call may block; nothing -// asserts elapsed time, only that parked work stays parked and armed -// timers still fire within the bound. +// asserts elapsed time, only that parked work stays parked, ready work +// completes, and armed timers still fire within the bound. #include @@ -19,6 +19,8 @@ #include #include +#include +#include #include #include @@ -36,6 +38,8 @@ #include #include +#include + #include "context.hpp" #include "must.hpp" #include "test_suite.hpp" @@ -172,6 +176,65 @@ struct bounded_run_test BOOST_TEST(resumed); } + template + void checkPollCompletesReadyRead(Drive drive) + { + io_context ioc(Backend); + local_stream_socket a(ioc); + local_stream_socket b(ioc); + test::must_ok(connect_pair(a, b)); + + bool done = false; + auto reader = [&]() -> capy::task<> { + char buf[1]; + auto [ec, n] = co_await b.read_some(capy::mutable_buffer(buf, 1)); + BOOST_TEST(!ec); + BOOST_TEST_EQ(n, 1u); + done = true; + }; + capy::run_async(ioc.get_executor())(reader()); + std::ignore = ioc.poll(); // the read parks + BOOST_TEST(!done); + + char const c = 'x'; + BOOST_TEST_EQ(::send(a.native_handle(), &c, 1, 0), 1); + + for (int i = 0; i < 5 && !done; ++i) + { + ioc.restart(); + std::ignore = drive(ioc); + } + BOOST_TEST(done); + } + + void testPollCompletesReadyRead() + { + checkPollCompletesReadyRead([](io_context& c) { return c.poll(); }); + checkPollCompletesReadyRead( + [](io_context& c) { return c.poll_one(); }); + } + + void testPollFiresExpiredDelay() + { + io_context ioc(Backend); + bool fired = false; + auto task = [&]() -> capy::task<> { + std::ignore = + co_await corosio::delay(std::chrono::milliseconds(1)); + fired = true; + }; + capy::run_async(ioc.get_executor())(task()); + + for (int i = 0; i < 1000 && !fired; ++i) + { + ioc.restart(); + std::ignore = ioc.poll(); + if (!fired) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + BOOST_TEST(fired); + } + void testTwoThreadBoundedFollower() { // Two threads inside the scheduler: one holds reactor @@ -234,6 +297,8 @@ struct bounded_run_test testRunOneForLeavesParkedReadParked(); testRunForFiresArmedDelay(); testPollLeavesParkedReadParked(); + testPollCompletesReadyRead(); + testPollFiresExpiredDelay(); testTwoThreadBoundedFollower(); testOversizeBudgetThrows(); testRunOneOnIdleReturnsZero();