diff --git a/include/exec/start_now.hpp b/include/exec/start_now.hpp index cc1b26394..c216dfe4b 100644 --- a/include/exec/start_now.hpp +++ b/include/exec/start_now.hpp @@ -22,6 +22,8 @@ #include "../stdexec/__detail/__meta.hpp" #include "../stdexec/__detail/__receivers.hpp" #include "../stdexec/__detail/__senders.hpp" +#include "../stdexec/__detail/__stop_token.hpp" +#include "../stdexec/__detail/__stop_when.hpp" #include "async_scope.hpp" #include "env.hpp" @@ -180,7 +182,10 @@ namespace experimental::execution using __receiver_t = __receiver<_Env>; using __sender_t = __sender<_Env>; template - using __nested_t = nest_result_t<_AsyncScope, _Sender>; + using __stoppable_t = + STDEXEC::__call_result_t; + template + using __nested_t = nest_result_t<_AsyncScope, __stoppable_t<_Sender>>; STDEXEC::__tuple, __receiver_t>...> __opstates_; @@ -188,8 +193,12 @@ namespace experimental::execution public: constexpr __storage(_Env&& __env, _AsyncScope& __scope, _Senders&&... __sndr) : __storage_base<_Env>(static_cast<_Env&&>(__env), sizeof...(__sndr)) + // nest() gives the sender the scope's stop token in place of the receiver's, + // so the storage's token has to be fused in underneath it. , __opstates_{ - STDEXEC::connect(__scope.nest(static_cast<_Senders&&>(__sndr)), __receiver_t{this})...} + STDEXEC::connect(__scope.nest(STDEXEC::__stop_when(static_cast<_Senders&&>(__sndr), + this->__source_.get_token())), + __receiver_t{this})...} { // Start all of the child operations STDEXEC::__apply(STDEXEC::__for_each{STDEXEC::start}, __opstates_); diff --git a/test/exec/async_scope/test_start_now.cpp b/test/exec/async_scope/test_start_now.cpp index d4d1c1131..2c37703a7 100644 --- a/test/exec/async_scope/test_start_now.cpp +++ b/test/exec/async_scope/test_start_now.cpp @@ -83,4 +83,42 @@ namespace REQUIRE(executedB); } + TEST_CASE("start_now request_stop stops the senders", "[async_scope][start_now]") + { + bool executed{false}; + bool stopped{false}; + async_scope scope; + ex::run_loop loop; + + auto stg = start_now(scope, + ex::schedule(loop.get_scheduler()) + | ex::then([&]() noexcept { executed = true; }) + | ex::upon_stopped([&]() noexcept { stopped = true; })); + REQUIRE(stg.request_stop()); + loop.finish(); + loop.run(); + REQUIRE(stopped); + REQUIRE_FALSE(executed); + REQUIRE(sync_wait(stg.async_wait()).has_value()); + } + + TEST_CASE("start_now senders still stop with the scope", "[async_scope][start_now]") + { + bool executed{false}; + bool stopped{false}; + async_scope scope; + ex::run_loop loop; + + auto stg = start_now(scope, + ex::schedule(loop.get_scheduler()) + | ex::then([&]() noexcept { executed = true; }) + | ex::upon_stopped([&]() noexcept { stopped = true; })); + REQUIRE(scope.request_stop()); + loop.finish(); + loop.run(); + REQUIRE(stopped); + REQUIRE_FALSE(executed); + REQUIRE(sync_wait(stg.async_wait()).has_value()); + } + } // namespace