diff --git a/include/exec/when_any.hpp b/include/exec/when_any.hpp index 2bde0e1d7..9005675bc 100644 --- a/include/exec/when_any.hpp +++ b/include/exec/when_any.hpp @@ -83,27 +83,8 @@ namespace experimental::execution STDEXEC_IMMOVABLE(__opstate_base); - struct __forward_stop_request - { - constexpr void operator()() const noexcept - { - // Temporarily increment the count to avoid concurrent/recursive - // arrivals to pull the rug under our feet. Relaxed memory order is - // fine here. - __op_->__count_.fetch_add(1, __std::memory_order_relaxed); - - __op_->__stop_source_.request_stop(); - - // Arrive in order to decrement the count again and complete if - // needed. - __op_->__arrive(); - } - - __opstate_base* __op_; - }; - using __on_stop = - stop_callback_for_t&>, __forward_stop_request>; + stop_callback_for_t&>, __forward_stop_request<>>; inplace_stop_source __stop_source_{}; std::optional<__on_stop> __on_stop_{}; @@ -227,8 +208,11 @@ namespace experimental::execution void start() & noexcept { + // Only forward the request to the children. The callback can still be + // running on another thread after the last child has arrived, so it + // must not arrive itself. this->__on_stop_.emplace(get_stop_token(get_env(this->__rcvr_)), - typename __op_base_t::__forward_stop_request{this}); + __forward_stop_request{this->__stop_source_}); if (this->__stop_source_.stop_requested()) { STDEXEC::set_stopped(static_cast<_Receiver&&>(this->__rcvr_)); diff --git a/include/stdexec/__detail/__when_all.hpp b/include/stdexec/__detail/__when_all.hpp index 7d7770e7a..c4b166438 100644 --- a/include/stdexec/__detail/__when_all.hpp +++ b/include/stdexec/__detail/__when_all.hpp @@ -561,35 +561,6 @@ namespace STDEXEC struct _INVALID_ARGUMENTS_TO_WHEN_ALL_ {}; - template - struct __forward_stop_request - { - constexpr void operator()() const noexcept - { - // Temporarily increment the count to avoid concurrent/recursive - // arrivals to pull the rug under our feet. Relaxed memory order is fine - // here. - __state_->__count_.fetch_add(1, __std::memory_order_relaxed); - - __state_t __expected = __started; - // Transition to the "stopped" state if and only if we're in the - // "started" state. (If this fails, it's because we're in an error - // state, which trumps cancellation.) - if (__state_->__state_.compare_exchange_strong(__expected, __stopped)) - { - if constexpr (_State::__uses_stop_source) - { - __state_->__stop_source_.request_stop(); - } - } - - // Arrive in order to decrement the count again and complete if needed. - __state_->__arrive(); - } - - _State* __state_; - }; - template >, __forward_stop_request<__state>>; + stop_callback_for_t>, __forward_stop_request<>>; using __stop_source_t = __if_c<_UsesStopSource, inplace_stop_source, __empty>; using __on_stop_t = __if_c<_UsesStopSource, __optional<__stop_callback_t>, __empty>; @@ -622,16 +593,6 @@ namespace STDEXEC // All child operations have completed and arrived at the barrier. switch (__state_.load(__std::memory_order_relaxed)) { - case __stopped: - if constexpr (_SendsStopped) - { - STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_)); - break; - } - // This is reachable because the stop callback sets __stopped whether - // or not any child can send set_stopped, therefore we handle this the - // same as __started - [[fallthrough]]; case __started: if constexpr (!__std::same_as<_ValuesTuple, __ignore>) { @@ -647,6 +608,17 @@ namespace STDEXEC static_cast<_ErrorsVariant&&>(__errors_)); } break; + case __stopped: + // Only a child completing with set_stopped puts us in this state. + if constexpr (_SendsStopped) + { + STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_)); + } + else + { + STDEXEC_UNREACHABLE(); + } + break; default:; } } @@ -797,8 +769,11 @@ namespace STDEXEC { if constexpr (_State::__uses_stop_source) { + // Only forward the request to the children. The callback can still + // be running on another thread after the last child has arrived, so + // it must not arrive or change __state_ itself. __state.__on_stop_.emplace(get_stop_token(STDEXEC::get_env(__state.__rcvr_)), - __forward_stop_request<_State>{&__state}); + __forward_stop_request{__state.__stop_source_}); } (STDEXEC::start(__child_ops), ...); }; @@ -864,24 +839,29 @@ namespace STDEXEC } else if constexpr (!__same_as<_ValuesTuple, __ignore>) { - auto& __opt_values = STDEXEC::__get<_Index::value>(__state.__values_); - using _Tuple = __decayed_tuple<_Args...>; - static_assert(__same_as, - "One of the senders in this when_all() is fibbing about what types it " - "sends"); - if constexpr ((__nothrow_decay_copyable<_Args> && ...)) + // We only need to bother recording the completion values if we're + // not already in the "error" or "stopped" state. + if (__state.__state_.load() == __started) { - __opt_values.emplace(_Tuple{static_cast<_Args&&>(__args)...}); - } - else - { - STDEXEC_TRY + auto& __opt_values = STDEXEC::__get<_Index::value>(__state.__values_); + using _Tuple = __decayed_tuple<_Args...>; + static_assert(__same_as, + "One of the senders in this when_all() is fibbing about what types it " + "sends"); + if constexpr ((__nothrow_decay_copyable<_Args> && ...)) { __opt_values.emplace(_Tuple{static_cast<_Args&&>(__args)...}); } - STDEXEC_CATCH_ALL + else { - __set_error(__state, std::current_exception()); + STDEXEC_TRY + { + __opt_values.emplace(_Tuple{static_cast<_Args&&>(__args)...}); + } + STDEXEC_CATCH_ALL + { + __set_error(__state, std::current_exception()); + } } } } diff --git a/test/exec/test_when_any.cpp b/test/exec/test_when_any.cpp index 7a5518f1b..c7f21b4c0 100644 --- a/test/exec/test_when_any.cpp +++ b/test/exec/test_when_any.cpp @@ -14,13 +14,17 @@ * limitations under the License. */ +#include +#include #include #include #include #include #include #include +#include #include +#include #include @@ -332,6 +336,40 @@ namespace REQUIRE(result == 42); } + TEST_CASE("when_any handles stop requests from the environment correctly", "[adaptors][when_any]") + { + // Both children complete inside the stop request, and async_scope destroys + // when_any's operation state before the request returns. + auto snd = exec::when_any(completes_if(false), completes_if(false)); + + exec::async_scope scope; + scope.spawn(snd); + scope.request_stop(); + ex::sync_wait(scope.on_empty()); + } + + TEST_CASE("when_any completes once when a stop request races with its last child", + "[adaptors][when_any]") + { + impulse_scheduler sched; + racing_stop_source source; + std::atomic values{0}; + std::atomic stopped{0}; + + auto op = ex::connect(exec::when_any(ex::schedule(sched)), + racing_stop_receiver{&values, &stopped, source.get_token()}); + ex::start(op); + + std::thread requester{[&] { source.request_stop(); }}; + source.wait_for_request(); // the requester has taken when_any's stop callback + sched.start_next(); // the only child completes on this thread + requester.join(); + + CHECK(source.destructor_waited()); + CHECK(values == 0); + CHECK(stopped == 1); + } + TEST_CASE("when_any completion signatures", "[adaptors][when_any]") { struct move_throws diff --git a/test/stdexec/algos/adaptors/test_when_all.cpp b/test/stdexec/algos/adaptors/test_when_all.cpp index 49c085c0f..a698e3bf3 100644 --- a/test/stdexec/algos/adaptors/test_when_all.cpp +++ b/test/stdexec/algos/adaptors/test_when_all.cpp @@ -25,8 +25,16 @@ #include #include #include +#include #include +#if STDEXEC_USE_MODULES() +import std; +#else +# include +# include +#endif + namespace ex = STDEXEC; // For testing `when_all_with_variant`, we just check a couple of examples, @@ -601,6 +609,42 @@ namespace ex::sync_wait(scope.on_empty()); } + TEST_CASE("when_all leaves stop requests from the environment to its children", + "[adaptors][when_all]") + { + ex::inplace_stop_source source; + source.request_stop(); + auto env = ex::prop(ex::get_stop_token, source.get_token()); + // In this environment the children can complete with set_stopped, so + // when_all forwards the stop request to them, but they never do. + auto snd = ex::when_all(stop_sensitive_sender{}, stop_sensitive_sender{}); + auto op = ex::connect(std::move(snd), expect_void_receiver(env)); + ex::start(op); + } + + TEST_CASE("when_all completes once when a stop request races with its last child", + "[adaptors][when_all]") + { + impulse_scheduler sched; + racing_stop_source source; + std::atomic values{0}; + std::atomic stopped{0}; + + // just() completes in start(), so the impulse sender is the last child. + auto op = ex::connect(ex::when_all(ex::just(), ex::schedule(sched)), + racing_stop_receiver{&values, &stopped, source.get_token()}); + ex::start(op); + + std::thread requester{[&] { source.request_stop(); }}; + source.wait_for_request(); // the requester has taken when_all's stop callback + sched.start_next(); // the last child completes on this thread + requester.join(); + + CHECK(source.destructor_waited()); + CHECK(values == 1); + CHECK(stopped == 0); + } + TEST_CASE("regression test for #1988", "[adaptors][when_all]") { // just test that this compiles: diff --git a/test/test_common/stop_tokens.hpp b/test/test_common/stop_tokens.hpp new file mode 100644 index 000000000..a1f3564b8 --- /dev/null +++ b/test/test_common/stop_tokens.hpp @@ -0,0 +1,221 @@ +/* + * Copyright (c) 2026 stdexec contributors + * + * Licensed under the Apache License Version 2.0 with LLVM Exceptions + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * https://llvm.org/LICENSE.txt + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include + +#if STDEXEC_USE_MODULES() +import std; +import stdexec; +#else +# include + +# include +# include +# include +# include +# include +# include +#endif + +namespace +{ + //! A stop source for deterministically testing a stop request that races + //! with the completion of an operation. + //! + //! request_stop() takes the registered callback and then holds it, the way + //! a thread can be preempted inside inplace_stop_source::request_stop(), + //! until another thread is waiting in that callback's destructor. Only then + //! does it invoke the callback. As for any stop callback, the destructor + //! blocks while the callback runs on another thread and does not block when + //! the callback destroys itself. At most one callback can be registered. + class racing_stop_source + { + struct callback_base + { + void (*invoke_)(callback_base*) noexcept; + }; + + public: + class token; + + racing_stop_source() = default; + racing_stop_source(racing_stop_source&&) = delete; + + [[nodiscard]] + auto get_token() noexcept -> token; + + void request_stop() + { + std::unique_lock lock{mutex_}; + if (stop_requested_.exchange(true)) + { + return; + } + callback_base* callback = std::exchange(registered_, nullptr); + running_ = callback; + running_on_ = std::this_thread::get_id(); + cv_.notify_all(); + if (callback == nullptr) + { + return; + } + // The timeout only keeps a test from hanging if the destructor never + // runs while the callback is held. + (void) cv_.wait_for(lock, std::chrono::seconds(10), [this] { return destructor_waiting_; }); + lock.unlock(); + callback->invoke_(callback); + lock.lock(); + running_ = nullptr; + cv_.notify_all(); + } + + //! Blocks until request_stop() has taken the registered callback, if any. + void wait_for_request() + { + std::unique_lock lock{mutex_}; + cv_.wait(lock, [this] { return stop_requested_.load(); }); + } + + //! Whether the callback's destructor ran on another thread while + //! request_stop() was holding the callback. + [[nodiscard]] + auto destructor_waited() -> bool + { + std::lock_guard lock{mutex_}; + return destructor_waiting_; + } + + private: + std::mutex mutex_; + std::condition_variable cv_; + std::atomic stop_requested_{false}; + callback_base* registered_{nullptr}; + callback_base* running_{nullptr}; + std::thread::id running_on_{}; + bool destructor_waiting_{false}; + }; + + class racing_stop_source::token + { + public: + template + class callback_type : callback_base + { + public: + template + explicit callback_type(token tok, Init&& init) + : callback_base{&invoke} + , source_(tok.source_) + , fn_(static_cast(init)) + { + std::unique_lock lock{source_->mutex_}; + if (source_->stop_requested_.load()) + { + lock.unlock(); + fn_(); + } + else + { + source_->registered_ = this; + } + } + + callback_type(callback_type&&) = delete; + + ~callback_type() + { + std::unique_lock lock{source_->mutex_}; + if (source_->registered_ == this) + { + source_->registered_ = nullptr; + } + else if (source_->running_ == this && source_->running_on_ != std::this_thread::get_id()) + { + source_->destructor_waiting_ = true; + source_->cv_.notify_all(); + source_->cv_.wait(lock, [this] { return source_->running_ != this; }); + } + } + + private: + static void invoke(callback_base* self) noexcept + { + static_cast(self)->fn_(); + } + + racing_stop_source* source_; + Fn fn_; + }; + + [[nodiscard]] + auto stop_requested() const noexcept -> bool + { + return source_->stop_requested_.load(); + } + + [[nodiscard]] + auto stop_possible() const noexcept -> bool + { + return true; + } + + auto operator==(token const &) const noexcept -> bool = default; + + private: + friend racing_stop_source; + + explicit token(racing_stop_source* source) noexcept + : source_(source) + {} + + racing_stop_source* source_; + }; + + inline auto racing_stop_source::get_token() noexcept -> token + { + return token{this}; + } + + //! Counts its value and stopped completions. Its environment carries a + //! racing_stop_source's token. + struct racing_stop_receiver + { + using receiver_concept = STDEXEC::receiver_tag; + + template + void set_value(Values&&...) && noexcept + { + ++*values_; + } + + void set_stopped() && noexcept + { + ++*stopped_; + } + + [[nodiscard]] + auto get_env() const noexcept + { + return STDEXEC::prop{STDEXEC::get_stop_token, token_}; + } + + std::atomic* values_; + std::atomic* stopped_; + racing_stop_source::token token_; + }; +} // namespace