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..1c5300668 100644 --- a/src/tests/rpp/test_timeout.cpp +++ b/src/tests/rpp/test_timeout.cpp @@ -18,9 +18,144 @@ #include #include #include +#include #include "disposable_observable.hpp" +#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. + class manual_scheduler final + { + public: + static inline rpp::schedulers::time_point s_now{}; + static inline std::atomic s_hold_now_on{}; + static inline gate s_now_held{}; + static inline gate s_release_now{}; + + 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.open(); + s_release_now.wait(); + } + 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 +304,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.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.open(); + 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{}));