From 26249b6db7f5fc02969e90b5377fdc118ad89be5 Mon Sep 17 00:00:00 2001 From: Markus Werle Date: Thu, 12 Nov 2026 18:55:37 +0100 Subject: [PATCH 1/2] timeout: do not pass a value to the observer after the timer has handed it to the fallback on_next checked for disposal only before taking the lock. A value that passed that check while the timer held the lock reached the observer the timer had just moved into the fallback subscription. Check is_disposed() again under the lock, as on_error and on_completed do. Co-Authored-By: Claude Opus 5.5 --- src/rpp/rpp/operators/timeout.hpp | 3 + src/tests/rpp/test_timeout.cpp | 141 ++++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+) diff --git a/src/rpp/rpp/operators/timeout.hpp b/src/rpp/rpp/operators/timeout.hpp index c30ad0444..92ed9d3d1 100644 --- a/src/rpp/rpp/operators/timeout.hpp +++ b/src/rpp/rpp/operators/timeout.hpp @@ -82,6 +82,9 @@ namespace rpp::operators::details void on_next(T&& v) const { auto obs_with_timeout = disposable->get_observer_with_timeout_under_lock(); + // the timer may have disposed and handed the observer to the fallback while we waited for the lock + if (disposable->is_disposed()) + return; obs_with_timeout->observer.on_next(std::forward(v)); obs_with_timeout->timeout = rpp::schedulers::utils::get_worker_t::now() + disposable->get_period(); } diff --git a/src/tests/rpp/test_timeout.cpp b/src/tests/rpp/test_timeout.cpp index c529e635e..71ff26c4f 100644 --- a/src/tests/rpp/test_timeout.cpp +++ b/src/tests/rpp/test_timeout.cpp @@ -18,9 +18,117 @@ #include #include #include +#include #include "disposable_observable.hpp" +#include +#include +#include + +namespace +{ + // Runs its schedulables only when fire() is called, on the calling thread. On the thread set in + // s_hold_now_on, the next now() blocks until s_release_now is signalled: the timeout handler + // calls now() while it holds the operator's lock, so a test can pause it there. + class manual_scheduler final + { + public: + static inline rpp::schedulers::time_point s_now{}; + static inline std::atomic s_hold_now_on{}; + static inline std::binary_semaphore s_now_held{0}; + static inline std::binary_semaphore s_release_now{0}; + + class worker_strategy; + + struct state + { + rpp::schedulers::details::schedulables_queue queue{}; + }; + + class worker_strategy + { + public: + explicit worker_strategy(std::shared_ptr state) + : m_state{std::move(state)} + { + } + + template Fn> + void defer_to(rpp::schedulers::time_point time_point, Fn&& fn, Handler&& handler, Args&&... args) const + { + m_state->queue.emplace(time_point, std::forward(fn), std::forward(handler), std::forward(args)...); + } + + static rpp::schedulers::time_point now() + { + if (s_hold_now_on.load() == std::this_thread::get_id()) + { + s_hold_now_on.store(std::thread::id{}); + s_now_held.release(); + s_release_now.acquire(); + } + return s_now; + } + + private: + std::shared_ptr m_state; + }; + + rpp::schedulers::worker create_worker() const + { + return rpp::schedulers::worker{m_state}; + } + + void fire() const + { + const auto schedulable = m_state->queue.top(); + m_state->queue.pop(); + if (!schedulable->is_disposed()) + (void)(*schedulable)(); + } + + private: + std::shared_ptr m_state = std::make_shared(); + }; + + // Counts every call that reaches an instance which has been moved from. + class move_aware_strategy + { + public: + static constexpr auto preferred_disposables_mode = rpp::details::observers::disposables_mode::Auto; + + explicit move_aware_strategy(std::atomic& calls_on_moved_from) + : m_calls_on_moved_from{&calls_on_moved_from} + { + } + + move_aware_strategy(move_aware_strategy&& other) noexcept + : m_calls_on_moved_from{other.m_calls_on_moved_from} + { + other.m_moved_from = true; + } + + void on_next(const int&) const { count(); } + void on_next(int&&) const { count(); } + void on_error(const std::exception_ptr&) const { count(); } + void on_completed() const { count(); } + + static void set_upstream(const rpp::disposable_wrapper&) { } + static bool is_disposed() { return false; } + + private: + void count() const + { + if (m_moved_from) + m_calls_on_moved_from->fetch_add(1); + } + + std::atomic* m_calls_on_moved_from; + bool m_moved_from{}; + }; +} // namespace + TEST_CASE("timeout subscribes to passed observable in case of reaching timeout") { @@ -169,6 +277,39 @@ TEST_CASE("timeout handles exception from fallback") CHECK(mock.get_on_completed_count() == 0); } +TEST_CASE("timeout does not pass a value that races the timer to the observer handed to the fallback") +{ + // The timer runs on one thread and a value arrives on another. The value passes the disposed check + // before the timer disposes, then waits for the lock the timer holds while it hands the observer + // to the fallback. Once it gets the lock, the value must not reach that moved-from observer. + const auto scheduler = manual_scheduler{}; + auto source = rpp::subjects::publish_subject{}; + auto fallback = rpp::subjects::publish_subject{}; + + std::atomic calls_on_moved_from{}; + source.get_observable() + | rpp::ops::timeout(std::chrono::seconds{1}, fallback.get_observable(), scheduler) + | rpp::ops::subscribe(rpp::observer{calls_on_moved_from}); + + manual_scheduler::s_now += std::chrono::seconds{1}; + + std::thread timer_thread{[&] { + manual_scheduler::s_hold_now_on.store(std::this_thread::get_id()); + scheduler.fire(); + }}; + manual_scheduler::s_now_held.acquire(); + + std::thread value_thread{[&] { source.get_observer().on_next(1); }}; + // let the value pass the disposed check and block on the operator's lock + std::this_thread::sleep_for(std::chrono::milliseconds{200}); + + manual_scheduler::s_release_now.release(); + timer_thread.join(); + value_thread.join(); + + CHECK(calls_on_moved_from.load() == 0); +} + TEST_CASE("timeout satisfies disposable contracts") { test_operator_with_disposable(rpp::ops::timeout(std::chrono::seconds{10000000}, rpp::schedulers::test_scheduler{})); From f8ddd1d544b3c604961682a18801dc0d914664e4 Mon Sep 17 00:00:00 2001 From: Markus Werle Date: Thu, 12 Nov 2026 20:59:09 +0100 Subject: [PATCH 2/2] timeout test: use a mutex and condition variable instead of std::binary_semaphore GCC 10 has no , so the test did not compile in the ci-ubuntu-gcc jobs. Co-Authored-By: Claude Fable 5.2 --- src/tests/rpp/test_timeout.cpp | 41 ++++++++++++++++++++++++++++------ 1 file changed, 34 insertions(+), 7 deletions(-) diff --git a/src/tests/rpp/test_timeout.cpp b/src/tests/rpp/test_timeout.cpp index 71ff26c4f..1c5300668 100644 --- a/src/tests/rpp/test_timeout.cpp +++ b/src/tests/rpp/test_timeout.cpp @@ -23,11 +23,38 @@ #include "disposable_observable.hpp" #include -#include +#include +#include #include namespace { + // wait() blocks until open() has been called once, then closes again. + class gate + { + public: + void open() + { + { + const std::lock_guard lock{m_mutex}; + m_open = true; + } + m_cv.notify_one(); + } + + void wait() + { + std::unique_lock lock{m_mutex}; + m_cv.wait(lock, [this] { return m_open; }); + m_open = false; + } + + private: + std::mutex m_mutex{}; + std::condition_variable m_cv{}; + bool m_open{}; + }; + // Runs its schedulables only when fire() is called, on the calling thread. On the thread set in // s_hold_now_on, the next now() blocks until s_release_now is signalled: the timeout handler // calls now() while it holds the operator's lock, so a test can pause it there. @@ -36,8 +63,8 @@ namespace public: static inline rpp::schedulers::time_point s_now{}; static inline std::atomic s_hold_now_on{}; - static inline std::binary_semaphore s_now_held{0}; - static inline std::binary_semaphore s_release_now{0}; + static inline gate s_now_held{}; + static inline gate s_release_now{}; class worker_strategy; @@ -65,8 +92,8 @@ namespace if (s_hold_now_on.load() == std::this_thread::get_id()) { s_hold_now_on.store(std::thread::id{}); - s_now_held.release(); - s_release_now.acquire(); + s_now_held.open(); + s_release_now.wait(); } return s_now; } @@ -297,13 +324,13 @@ TEST_CASE("timeout does not pass a value that races the timer to the observer ha manual_scheduler::s_hold_now_on.store(std::this_thread::get_id()); scheduler.fire(); }}; - manual_scheduler::s_now_held.acquire(); + manual_scheduler::s_now_held.wait(); std::thread value_thread{[&] { source.get_observer().on_next(1); }}; // let the value pass the disposed check and block on the operator's lock std::this_thread::sleep_for(std::chrono::milliseconds{200}); - manual_scheduler::s_release_now.release(); + manual_scheduler::s_release_now.open(); timer_thread.join(); value_thread.join();