From ece735f41f48ab700099d2fec299033d13d51bd4 Mon Sep 17 00:00:00 2001 From: Markus Werle Date: Fri, 23 Oct 2026 16:44:12 +0200 Subject: [PATCH 1/2] schedulers: re-arm run_loop/new_thread timer wait when an earlier schedulable is queued Co-Authored-By: Claude Opus 5.5 --- src/rpp/rpp/schedulers/new_thread.hpp | 5 +- src/rpp/rpp/schedulers/run_loop.hpp | 5 +- src/tests/rpp/test_scheduler.cpp | 67 +++++++++++++++++++++++++++ 3 files changed, 75 insertions(+), 2 deletions(-) diff --git a/src/rpp/rpp/schedulers/new_thread.hpp b/src/rpp/rpp/schedulers/new_thread.hpp index e274675a7..30253a4bd 100644 --- a/src/rpp/rpp/schedulers/new_thread.hpp +++ b/src/rpp/rpp/schedulers/new_thread.hpp @@ -87,7 +87,10 @@ namespace rpp::schedulers { if (const auto now = worker_strategy::now(); now < state->queue.top()->get_timepoint()) { - state->cv.wait_for(lock, state->queue.top()->get_timepoint() - now, [&] { return state->queue.top()->is_disposed() || worker_strategy::now() >= state->queue.top()->get_timepoint(); }); + // Sleep until the current top is due, and wake as soon as the top changes + // (an earlier schedulable was queued), so the loop recomputes the wait. + const auto* const top = state->queue.top().get(); + state->cv.wait_until(lock, top->get_timepoint(), [&] { return state->queue.is_empty() || state->queue.top().get() != top || state->queue.top()->is_disposed() || worker_strategy::now() >= state->queue.top()->get_timepoint(); }); continue; } } diff --git a/src/rpp/rpp/schedulers/run_loop.hpp b/src/rpp/rpp/schedulers/run_loop.hpp index ce608ab74..602020ad5 100644 --- a/src/rpp/rpp/schedulers/run_loop.hpp +++ b/src/rpp/rpp/schedulers/run_loop.hpp @@ -66,7 +66,10 @@ namespace rpp::schedulers if (!wait) break; - m_cv.wait_for(lock, m_queue.top()->get_timepoint() - now, [&]() { return is_disposed() || (!m_queue.is_empty() && m_queue.top()->get_timepoint() <= worker_strategy::now()); }); + // Sleep until the current top is due, and wake as soon as the top changes + // (an earlier schedulable was queued), so the loop recomputes the wait. + const auto* const top = m_queue.top().get(); + m_cv.wait_until(lock, top->get_timepoint(), [&]() { return is_disposed() || m_queue.is_empty() || m_queue.top().get() != top || m_queue.top()->is_disposed() || m_queue.top()->get_timepoint() <= worker_strategy::now(); }); } return {}; } diff --git a/src/tests/rpp/test_scheduler.cpp b/src/tests/rpp/test_scheduler.cpp index b7777fd61..b970c5c22 100644 --- a/src/tests/rpp/test_scheduler.cpp +++ b/src/tests/rpp/test_scheduler.cpp @@ -1068,3 +1068,70 @@ TEST_CASE("thread_pool shares same thread") CHECK(f.get()); } + +TEST_CASE("timer queued while the worker waits for a later one runs at its own time") +{ + // The worker is already waiting for a schedulable due in `later`. A second one, due + // in `earlier` (much sooner), is queued from another thread. It has to run after + // about `earlier`, not when the wait for the first one ends. + constexpr auto later = std::chrono::seconds{3}; + constexpr auto earlier = std::chrono::milliseconds{50}; + constexpr auto bound = std::chrono::seconds{1}; + + auto d = rpp::composite_disposable_wrapper::make(); + auto obs = mock_observer_strategy{}.get_observer(d).as_dynamic(); + const auto noop = [](const auto&) { return rpp::schedulers::optional_delay_from_now{}; }; + + SUBCASE("run_loop") + { + auto scheduler = rpp::schedulers::run_loop{}; + auto worker = scheduler.create_worker(); + worker.schedule(later, noop, obs); + + std::atomic_bool earlier_ran{}; + std::chrono::steady_clock::time_point queued_at{}; + std::chrono::steady_clock::time_point ran_at{}; + + auto other = std::thread{[&] { + std::this_thread::sleep_for(std::chrono::milliseconds{100}); // dispatch() below waits for `later` by now + queued_at = std::chrono::steady_clock::now(); + worker.schedule( + earlier, + [&](const auto&) { + ran_at = std::chrono::steady_clock::now(); + earlier_ran.store(true); + return rpp::schedulers::optional_delay_from_now{}; + }, + obs); + }}; + + while (!earlier_ran.load()) + scheduler.dispatch(); + other.join(); + + CHECK(ran_at - queued_at < bound); + } + + SUBCASE("new_thread") + { + auto worker = rpp::schedulers::new_thread::create_worker(); + worker.schedule(later, noop, obs); + std::this_thread::sleep_for(std::chrono::milliseconds{100}); // the worker thread waits for `later` by now + + std::promise ran_at{}; + const auto queued_at = std::chrono::steady_clock::now(); + worker.schedule( + earlier, + [&ran_at](const auto&) { + ran_at.set_value(std::chrono::steady_clock::now()); + return rpp::schedulers::optional_delay_from_now{}; + }, + obs); + + auto f = ran_at.get_future(); + REQUIRE(f.wait_for(later * 2) == std::future_status::ready); + CHECK(f.get() - queued_at < bound); + } + + d.dispose(); +} From 3f23b64de40e990b10ce60de2c894a8b8ae53662 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Fri, 23 Oct 2026 17:39:22 +0000 Subject: [PATCH 2/2] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- src/tests/rpp/test_scheduler.cpp | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/tests/rpp/test_scheduler.cpp b/src/tests/rpp/test_scheduler.cpp index b970c5c22..cd612a690 100644 --- a/src/tests/rpp/test_scheduler.cpp +++ b/src/tests/rpp/test_scheduler.cpp @@ -1080,7 +1080,9 @@ TEST_CASE("timer queued while the worker waits for a later one runs at its own t auto d = rpp::composite_disposable_wrapper::make(); auto obs = mock_observer_strategy{}.get_observer(d).as_dynamic(); - const auto noop = [](const auto&) { return rpp::schedulers::optional_delay_from_now{}; }; + const auto noop = [](const auto&) { + return rpp::schedulers::optional_delay_from_now{}; + }; SUBCASE("run_loop") {