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..cd612a690 100644 --- a/src/tests/rpp/test_scheduler.cpp +++ b/src/tests/rpp/test_scheduler.cpp @@ -1068,3 +1068,72 @@ 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(); +}