diff --git a/include/stdexec/__detail/__continues_on.hpp b/include/stdexec/__detail/__continues_on.hpp index 2d3249b9e..c79a93989 100644 --- a/include/stdexec/__detail/__continues_on.hpp +++ b/include/stdexec/__detail/__continues_on.hpp @@ -164,6 +164,60 @@ namespace STDEXEC return false; } + //! @brief Returns `true` when a @c _SetTag completion of the predecessor + //! (or a failure to decay-copy its values) can be forwarded through the scheduler. + template + static consteval bool __forwards_completion() noexcept + { + return __sends<_SetTag, _Sender, __fwd_env_t<_Env>...> + || __has_decay_copy_errors<_SetTag, _Env...>(); + } + + template + static consteval auto __forwarded_domain() noexcept + { + using _SchSndr = schedule_result_t<_Scheduler>; + // Clang 16 substitutes the return type of `query` before checking its + // constraints, so this function can be instantiated with a _SetTag (e.g., + // `void`) that the query would reject. Bail out before touching anything + // that could hard-error. + if constexpr (!__one_of<_SetTag, set_error_t, set_stopped_t>) + { + return indeterminate_domain<>(); + } + else if constexpr (__forwards_completion<_SetTag, _Env...>()) + { + constexpr bool __own = __sends<_SetTag, _SchSndr, __env2_t<_Env>...>; + using _Fwd = __completion_domain_of_t...>; + if constexpr (__own) + { + return __common_domain_t< + _Fwd, + __completion_domain_of_t<_SetTag, _SchSndr, __env2_t<_Env>...>>(); + } + else + { + return _Fwd(); + } + } + else if constexpr (__sends<_SetTag, _SchSndr, __env2_t<_Env>...>) + { + return __completion_domain_of_t<_SetTag, _SchSndr, __env2_t<_Env>...>(); + } + else + { + // Neither the predecessor nor the scheduler's sender can complete with + // _SetTag. The query's constraints exclude this case; see above. + return indeterminate_domain<>(); + } + } + + template + static consteval bool __sch_sends() noexcept + { + return __sends<_SetTag, schedule_result_t<_Scheduler>, __env2_t<_Env>...>; + } + _Scheduler __sch_; _Sender const & __sndr_; @@ -181,38 +235,34 @@ namespace STDEXEC //! @note If @c _SetTag is @c set_value_t, then we are in the happy path: //! everything succeeded and execution continues on @c _Scheduler. //! - //! Otherwise, if @c _Sender never completes with @c _SetTag, and either - //! @c _SetTag is @c set_stopped_t or decay-copying @c _Sender's value - //! results cannot throw, then a @c _SetTag completion can only come from - //! the scheduler's sender. In this case, return the scheduler's - //! completion scheduler if it has one. - //! - //! Otherwise, if the scheduler's sender never completes with @c _SetTag, - //! then a @c _SetTag completion can only come from the original sender, - //! so return the original sender's completion scheduler. + //! Otherwise, completions of the predecessor sender are saved and replayed + //! from the @c set_value completion of the scheduler's sender, so they + //! complete on the scheduler's @c set_value completion scheduler. If the + //! scheduler's sender can also complete with @c _SetTag itself, there is no + //! way to know at compile time which scheduler a given completion will use + //! (they are runtime values), so the query is not provided. template requires(__same_as<_SetTag, set_value_t> - || __never_sends<_SetTag, _Sender, __fwd_env_t<_Env>...>) - && (!__has_decay_copy_errors<_SetTag, _Env...>()) + || (__forwards_completion<_SetTag, _Env...>() && !__sch_sends<_SetTag, _Env...>())) [[nodiscard]] constexpr auto query(get_completion_scheduler_t<_SetTag>, _Env const &... __env) const noexcept - -> __call_result_t, _Scheduler, __sch_env_t<_Env>...> + -> __call_result_t, _Scheduler, __sch_env_t<_Env>...> { - return get_completion_scheduler<_SetTag>(__sch_, __mk_sch_env(__env)...); + return get_completion_scheduler(__sch_, __mk_sch_env(__env)...); } //! @overload + //! Only the scheduler's sender can complete with @c _SetTag. template - requires __never_sends<_SetTag, schedule_result_t<_Scheduler>, __sch_env_t<_Env>...> + requires(!__same_as<_SetTag, set_value_t> && !__forwards_completion<_SetTag, _Env...>() + && __sch_sends<_SetTag, _Env...>()) [[nodiscard]] constexpr auto query(get_completion_scheduler_t<_SetTag>, _Env const &... __env) const noexcept - -> __call_result_t, - env_of_t<_Sender>, - __fwd_env_t<_Env>...> + -> __call_result_t, _Scheduler, __sch_env_t<_Env>...> { - return get_completion_scheduler<_SetTag>(get_env(__sndr_), __fwd_env(__env)...); + return get_completion_scheduler<_SetTag>(__sch_, __mk_sch_env(__env)...); } //! @brief Queries the completion domain for a given @c _SetTag. @@ -233,39 +283,25 @@ namespace STDEXEC //! from the sender's value completions. template <__same_as _SetTag, class... _Env> [[nodiscard]] - constexpr auto - query(get_completion_domain_t<_SetTag>, _Env const &...) const noexcept -> __unless_one_of_t< - __completion_domain_of_t<_SetTag, schedule_result_t<_Scheduler>, __env2_t<_Env>...>, - indeterminate_domain<>> + constexpr auto query(get_completion_domain_t<_SetTag>, _Env const &...) const noexcept + -> __ensure_valid_domain_t< + __completion_domain_of_t<_SetTag, schedule_result_t<_Scheduler>, __env2_t<_Env>...>> { return {}; } //! @overload + //! @note Completions of the predecessor sender are saved and replayed from + //! the @c set_value completion of the scheduler's sender, so they complete + //! in the scheduler's @c set_value completion domain. The scheduler's sender + //! may additionally complete with @c _SetTag itself, in which case that + //! domain contributes too. template <__one_of _SetTag, class... _Env> - requires(!__has_decay_copy_errors<_SetTag, _Env...>()) - [[nodiscard]] - constexpr auto - query(get_completion_domain_t<_SetTag>, _Env const &...) const noexcept -> __unless_one_of_t< - __common_domain_t< - __completion_domain_of_t<_SetTag, _Sender, __fwd_env_t<_Env>...>, - __completion_domain_of_t<_SetTag, schedule_result_t<_Scheduler>, __env2_t<_Env>...>>, - indeterminate_domain<>> - { - return {}; - } - - //! @overload - template - requires(__has_decay_copy_errors<_SetTag, _Env...>()) + requires(__forwards_completion<_SetTag, _Env...>() + || __sends<_SetTag, schedule_result_t<_Scheduler>, __env2_t<_Env>...>) [[nodiscard]] - constexpr auto - query(get_completion_domain_t<_SetTag>, _Env const &...) const noexcept -> __unless_one_of_t< - __common_domain_t< - __completion_domain_of_t<_SetTag, _Sender, __fwd_env_t<_Env>...>, - __completion_domain_of_t<_SetTag, schedule_result_t<_Scheduler>, __env2_t<_Env>...>, - __completion_domain_of_t...>>, - indeterminate_domain<>> + constexpr auto query(get_completion_domain_t<_SetTag>, _Env const &...) const noexcept + -> __ensure_valid_domain_t())> { return {}; } diff --git a/test/nvexec/variant.cpp b/test/nvexec/variant.cpp index fd63181f8..9c831daa0 100644 --- a/test/nvexec/variant.cpp +++ b/test/nvexec/variant.cpp @@ -18,7 +18,6 @@ #include #include -#include #include #include @@ -109,8 +108,13 @@ namespace TEST_CASE("nvexec variant emplaces alternative from GPU", "[cuda][stream][containers][variant]") { using variant_t = variant_t; - thrust::universal_vector variant_storage(1); - variant_t* v = thrust::raw_pointer_cast(variant_storage.data()); + // Note: avoid thrust::universal_vector here. Its default-construction kernel has + // `&cudaFree` as a template argument, which clang-cuda mangles differently in the + // host and device compilation passes, so the kernel cannot be found at runtime + // (cudaErrorSymbolNotFound). + variant_t* v = nullptr; + STDEXEC_TRY_CUDA_API(cudaMallocManaged(reinterpret_cast(&v), sizeof(variant_t))); + ::new (static_cast(v)) variant_t(); REQUIRE(v->index_ == 0); @@ -123,6 +127,9 @@ namespace STDEXEC_TRY_CUDA_API(cudaDeviceSynchronize()); visit([](auto alt) { REQUIRE(alt == 42); }, *v); + + v->~variant_t(); + STDEXEC_TRY_CUDA_API(cudaFree(v)); } TEST_CASE("nvexec variant works with cuda tuple", "[cuda][stream][containers][variant]") diff --git a/test/stdexec/algos/adaptors/test_continues_on.cpp b/test/stdexec/algos/adaptors/test_continues_on.cpp index 9a7139cbe..b5d43335b 100644 --- a/test/stdexec/algos/adaptors/test_continues_on.cpp +++ b/test/stdexec/algos/adaptors/test_continues_on.cpp @@ -37,6 +37,10 @@ using namespace std::chrono_literals; namespace { + template + using value_type_of_t = + ex::value_types_of_t, std::type_identity_t, std::type_identity_t>; + struct potentially_throwing_connect_scheduler { using scheduler_concept = ex::scheduler_tag; @@ -75,6 +79,20 @@ namespace operator==(potentially_throwing_connect_scheduler const &) const noexcept -> bool = default; }; + struct value_type + { + int val_; + }; + + struct continues_on_test_domain + { + template Sender> + static auto transform_sender(STDEXEC::set_value_t, Sender &&, auto const &...) + { + return ex::just(value_type{53}); + } + }; + TEST_CASE("continues_on returns a sender", "[adaptors][continues_on]") { auto snd = ex::continues_on(ex::just(13), inline_scheduler{}); @@ -296,25 +314,205 @@ namespace check_sends_stopped(ex::continues_on(ex::just(3), sched3)); } - struct value_type + TEST_CASE("continues_on keeps updates completion domain for error channel", + "[adaptors][continues_on]") { - int val_; - }; - - template - using value_type_of_t = - ex::value_types_of_t, std::type_identity_t, std::type_identity_t>; + // The following scheduler does not send errors. Any errors are forwarded + // from the predecessor sender on the scheduler's domain. + basic_inline_scheduler sched{}; + auto snd = ex::continues_on(ex::just_error(1), sched); + auto domain = ex::get_env(snd).query(ex::get_completion_domain); + static_assert(std::same_as); + } - struct continues_on_test_domain + //! A scheduler with runtime identity. Its value completions happen on a scheduler + //! with the same id. If `ErrorId != 0`, its sender can also complete with an + //! error, which happens on a scheduler with id `ErrorId`. + template + struct id_scheduler { - template Sender> - static auto transform_sender(STDEXEC::set_value_t, Sender &&, auto const &...) + using scheduler_concept = ex::scheduler_tag; + + int id_; + + auto operator==(id_scheduler const &) const noexcept -> bool = default; + + [[nodiscard]] + auto query(ex::get_completion_scheduler_t) const noexcept -> id_scheduler { - return ex::just(value_type{53}); + return *this; } + + [[nodiscard]] + auto query(ex::get_completion_scheduler_t) const noexcept -> id_scheduler + requires(ErrorId != 0) + { + return id_scheduler{id_ + ErrorId}; + } + + [[nodiscard]] + auto query(ex::get_completion_domain_t) const noexcept -> ValueDomain + { + return {}; + } + + [[nodiscard]] + auto schedule() const noexcept + { + return sender{*this}; + } + + struct attrs + { + id_scheduler sched_; + + [[nodiscard]] + auto query(ex::get_completion_scheduler_t) const noexcept -> id_scheduler + { + return sched_; + } + + [[nodiscard]] + auto query(ex::get_completion_scheduler_t) const noexcept -> id_scheduler + requires(ErrorId != 0) + { + return id_scheduler{sched_.id_ + ErrorId}; + } + + [[nodiscard]] + auto query(ex::get_completion_domain_t) const noexcept -> ValueDomain + { + return {}; + } + }; + + template + struct opstate : immovable + { + void start() & noexcept + { + ex::set_value(static_cast(rcvr_)); + } + + Receiver rcvr_; + }; + + struct sender + { + using sender_concept = STDEXEC::sender_tag; + using completion_signatures = + std::conditional_t, + ex::completion_signatures>; + + template + [[nodiscard]] + auto connect(Receiver rcvr) const -> opstate + { + return {{}, static_cast(rcvr)}; + } + + [[nodiscard]] + auto get_env() const noexcept -> attrs + { + return {sched_}; + } + + id_scheduler sched_; + }; + }; + + template + concept has_completion_scheduler = requires(Sender const &snd) { + ex::get_completion_scheduler(ex::get_env(snd)); }; - TEST_CASE("continues_on can be customized early", "[adaptors][continues_on]") + template + concept has_completion_domain = requires(Sender const &snd) { + ex::get_env(snd).query(ex::get_completion_domain); + }; + + TEST_CASE("continues_on: completion scheduler of the value channel is the scheduler", + "[adaptors][continues_on]") + { + id_scheduler sched{42}; + auto snd = ex::continues_on(ex::just(1), sched); + auto csched = ex::get_completion_scheduler(ex::get_env(snd)); + CHECK(csched.id_ == 42); + } + + TEST_CASE("continues_on: forwarded errors and stopped complete on the scheduler's value " + "completion scheduler", + "[adaptors][continues_on]") + { + id_scheduler sched{42}; + + auto err = ex::continues_on(ex::just_error(1), sched); + auto err_sched = ex::get_completion_scheduler(ex::get_env(err)); + CHECK(err_sched.id_ == 42); + + auto stopped = ex::continues_on(ex::just_stopped(), sched); + auto stopped_sched = ex::get_completion_scheduler(ex::get_env(stopped)); + CHECK(stopped_sched.id_ == 42); + + // ... and not on the predecessor's completion scheduler + id_scheduler other{7}; + auto on_other = ex::continues_on(ex::just_error(1), other); + auto snd = ex::continues_on(std::move(on_other), sched); + auto snd_sched = ex::get_completion_scheduler(ex::get_env(snd)); + CHECK(snd_sched.id_ == 42); + } + + TEST_CASE("continues_on: errors sent only by the scheduler's sender complete on the scheduler's " + "error completion scheduler", + "[adaptors][continues_on]") + { + id_scheduler sched{1}; + auto snd = ex::continues_on(ex::just(), sched); + auto err_sched = ex::get_completion_scheduler(ex::get_env(snd)); + CHECK(err_sched.id_ == 101); + // the value completion scheduler is unaffected + CHECK(ex::get_completion_scheduler(ex::get_env(snd)).id_ == 1); + } + + TEST_CASE("continues_on: no completion scheduler when the schedulers are ambiguous or the " + "channel is never used", + "[adaptors][continues_on]") + { + // Both the predecessor's errors (forwarded through the value scheduler) and the + // scheduler's own errors are possible. Which scheduler is used is only known + // at runtime, so there is no completion scheduler. + id_scheduler err_sched{1}; + auto both = ex::continues_on(ex::just_error(1), err_sched); + static_assert(!has_completion_scheduler); + static_assert(has_completion_scheduler); + + // Nothing can complete with an error. + id_scheduler sched{1}; + auto none = ex::continues_on(ex::just(), sched); + static_assert(!has_completion_scheduler); + static_assert(!has_completion_scheduler); + } + + TEST_CASE("continues_on: forwarded stopped completes in the scheduler's value domain", + "[adaptors][continues_on]") + { + id_scheduler sched{1}; + auto snd = ex::continues_on(ex::just_stopped(), sched); + auto domain = ex::get_env(snd).query(ex::get_completion_domain); + static_assert(std::same_as); + } + + TEST_CASE("continues_on: there is no error completion domain if nothing can send errors", + "[adaptors][continues_on]") + { + id_scheduler sched{1}; + auto snd = ex::continues_on(ex::just(), sched); + static_assert(!has_completion_domain); + static_assert(has_completion_domain); + } + + TEST_CASE("continues_on can be customized", "[adaptors][continues_on]") { // The customization will return a different value basic_inline_scheduler sched; diff --git a/test/test_common/schedulers.hpp b/test/test_common/schedulers.hpp index 2f67e6651..3cac6f974 100644 --- a/test/test_common/schedulers.hpp +++ b/test/test_common/schedulers.hpp @@ -132,6 +132,13 @@ namespace auto operator==(impulse_scheduler const &) const noexcept -> bool = default; + template Tag> + [[nodiscard]] + auto query(ex::get_completion_scheduler_t) const noexcept + { + return *this; + } + private: //! Command type that can store the action of firing up a sender using oper_command_t = std::function; @@ -221,13 +228,19 @@ namespace { using scheduler_concept = ex::scheduler_tag; + auto operator==(basic_inline_scheduler const &) const noexcept -> bool = default; + [[nodiscard]] auto schedule() const noexcept { return sender{}; } - auto operator==(basic_inline_scheduler const &) const noexcept -> bool = default; + [[nodiscard]] + auto query(ex::get_completion_scheduler_t) const noexcept + { + return *this; + } [[nodiscard]] auto query(ex::get_completion_domain_t) const noexcept -> Domain @@ -248,6 +261,23 @@ namespace Receiver rcvr_; }; + // The schedule sender's attributes mirror the scheduler's completion queries. + struct attrs + { + [[nodiscard]] + auto query(ex::get_completion_scheduler_t) const noexcept + { + return basic_inline_scheduler(); + } + + [[nodiscard]] + auto query(ex::get_completion_domain_t) const noexcept -> Domain + requires(!std::same_as) + { + return {}; + } + }; + struct sender { using sender_concept = STDEXEC::sender_tag; @@ -263,7 +293,7 @@ namespace [[nodiscard]] auto get_env() const noexcept { - return sched_attrs(basic_inline_scheduler(), ex::set_value); + return attrs{}; } }; };