diff --git a/include/nvexec/stream/when_all.cuh b/include/nvexec/stream/when_all.cuh index db1864f42..420a143ad 100644 --- a/include/nvexec/stream/when_all.cuh +++ b/include/nvexec/stream/when_all.cuh @@ -39,6 +39,9 @@ namespace nv::execution::_strm { namespace _when_all { + using STDEXEC::__when_all::_THE_GIVEN_SENDER_CAN_COMPLETE_SUCCESSFULLY_IN_MORE_THAN_ONE_WAY_; + using STDEXEC::__when_all::_USE_WHEN_ALL_WITH_VARIANT_INSTEAD_; + enum disposition : std::uint32_t { started, @@ -75,14 +78,11 @@ namespace nv::execution::_strm requires(too_many_completions_sender || ...) struct completions<__mlist, Senders...> { - static constexpr auto position_of() noexcept -> std::size_t - { - constexpr bool which[] = {too_many_completions_sender...}; - return __pos_of(which, which + sizeof...(Senders)); - } - - using invalid_arg_t = __m_at_c; - using __t = STDEXEC::__when_all::__too_many_value_completions_error_t; + using __t = + __mexception<_WHAT_(_INVALID_ARGUMENT_), + _WHERE_(_IN_ALGORITHM_, when_all_t), + _WHY_(_THE_GIVEN_SENDER_CAN_COMPLETE_SUCCESSFULLY_IN_MORE_THAN_ONE_WAY_), + _TO_FIX_THIS_ERROR_(_USE_WHEN_ALL_WITH_VARIANT_INSTEAD_)>; }; template diff --git a/include/stdexec/__detail/__basic_sender.hpp b/include/stdexec/__detail/__basic_sender.hpp index 0a8137d98..d2906307f 100644 --- a/include/stdexec/__detail/__basic_sender.hpp +++ b/include/stdexec/__detail/__basic_sender.hpp @@ -112,9 +112,6 @@ namespace STDEXEC template struct __connect; - template - using __connect_t = __connect, __child_indices_t<_Sexpr>>; - template struct __connect<_Tag, __indices<_Idx...>> { @@ -129,9 +126,13 @@ namespace STDEXEC } }; + template + using __connect_t = __connect, __child_indices_t<_Sexpr>>; + template concept __connectable_to = - __applicable<__connect_t<_Sexpr>, _Sexpr, __state_type_t<_Sexpr, _Receiver>&>; + sender_in<_Sexpr, env_of_t<_Receiver>> + && __applicable<__connect_t<_Sexpr>, _Sexpr, __state_type_t<_Sexpr, _Receiver>&>; STDEXEC_MODULE_EXPORT_AUTHORING struct __defaults @@ -283,8 +284,8 @@ namespace STDEXEC struct __opstate { using __tag_t = __decay_t<_Sexpr>::__tag_t; + using __connect_t = __detail::__connect_t<_Sexpr>; using __state_t = __detail::__state_type_t<_Sexpr, _Receiver>; - using __connect_t = __detail::__connect<__tag_t, __detail::__child_indices_t<_Sexpr>>; using __child_ops_t = __apply_result_t<__connect_t, _Sexpr, __state_t&>; constexpr explicit __opstate(_Sexpr&& __sndr, _Receiver __rcvr) diff --git a/include/stdexec/__detail/__continues_on.hpp b/include/stdexec/__detail/__continues_on.hpp index cf96b8c60..fc167aa94 100644 --- a/include/stdexec/__detail/__continues_on.hpp +++ b/include/stdexec/__detail/__continues_on.hpp @@ -332,7 +332,6 @@ namespace STDEXEC __nothrow_constructible_from<__state_for_t<_Sender, _Receiver>, __data_of<_Sender>&, _Receiver>) -> __state_for_t<_Sender, _Receiver> - requires sender_in<__child_of<_Sender>, __fwd_env_t>> { static_assert(__sender_for<_Sender, continues_on_t>); auto& [__tag, __sched, __child] = __sndr; diff --git a/include/stdexec/__detail/__transform_completion_signatures.hpp b/include/stdexec/__detail/__transform_completion_signatures.hpp index f2bd2ebad..f015713a0 100644 --- a/include/stdexec/__detail/__transform_completion_signatures.hpp +++ b/include/stdexec/__detail/__transform_completion_signatures.hpp @@ -372,11 +372,14 @@ namespace STDEXEC template consteval auto operator()() const noexcept { - if constexpr (__decay_copyable<_Args...>) + if constexpr (__nothrow_decay_copyable<_Args...>) { - return __concat_completion_signatures( - completion_signatures<_SetTag(__decay_t<_Args>...)>(), - __eptr_completion_unless_t<__nothrow_decay_copyable_t<_Args...>>()); + return completion_signatures<_SetTag(__decay_t<_Args>...)>(); + } + else if constexpr (__decay_copyable<_Args...>) + { + return completion_signatures<_SetTag(__decay_t<_Args>...), + set_error_t(std::exception_ptr)>(); } else { diff --git a/include/stdexec/__detail/__tuple.hpp b/include/stdexec/__detail/__tuple.hpp index 24ca3415d..03367de32 100644 --- a/include/stdexec/__detail/__tuple.hpp +++ b/include/stdexec/__detail/__tuple.hpp @@ -328,6 +328,32 @@ namespace STDEXEC } }; + struct __fold_right_t + { + template <__std::move_constructible _Init> + STDEXEC_ATTRIBUTE(always_inline, host, device) + constexpr auto operator()(__ignore, _Init&& __init) const + noexcept(__nothrow_move_constructible<_Init>) -> _Init + { + return static_cast<_Init&&>(__init); + } + + template + requires __callable<_Fn&, _Init, _Head> + && __callable<__fold_right_t, _Fn&, __call_result_t<_Fn&, _Init, _Head>, _Tail...> + STDEXEC_ATTRIBUTE(always_inline, host, device) + constexpr auto + operator()(_Fn&& __fn, _Init&& __init, _Head&& __head, _Tail&&... __tail) const // + noexcept( + __nothrow_callable<_Fn&, _Init, _Head> + && __nothrow_callable<__fold_right_t, _Fn&, __call_result_t<_Fn&, _Init, _Head>, _Tail...>) + { + return (*this)(__fn, + __fn(static_cast<_Init&&>(__init), static_cast<_Head&&>(__head)), + static_cast<_Tail&&>(__tail)...); + } + }; + # undef STDEXEC_TUPLE_GET } // namespace __tup diff --git a/include/stdexec/__detail/__when_all.hpp b/include/stdexec/__detail/__when_all.hpp index 597f8e20d..4a1eb7a72 100644 --- a/include/stdexec/__detail/__when_all.hpp +++ b/include/stdexec/__detail/__when_all.hpp @@ -428,96 +428,20 @@ namespace STDEXEC using __stoppable_env_t = decltype(__when_all::__mk_env(__declval<_Env>(), __declval())); - template - concept __max1_sender = - sender_in<_Sender, _Env> - && __minvocable_q<__value_types_of_t, _Sender, _Env, __mconst, __msingle_or>; - - struct _THE_GIVEN_SENDER_CAN_COMPLETE_SUCCESSFULLY_IN_MORE_THAN_ONE_WAY_ - {}; - struct _USE_WHEN_ALL_WITH_VARIANT_INSTEAD_ - {}; - - template - using __too_many_value_completions_error_t = - __mexception<_WHAT_(_INVALID_ARGUMENT_), - _WHERE_(_IN_ALGORITHM_, when_all_t), - _WHY_(_THE_GIVEN_SENDER_CAN_COMPLETE_SUCCESSFULLY_IN_MORE_THAN_ONE_WAY_), - _TO_FIX_THIS_ERROR_(_USE_WHEN_ALL_WITH_VARIANT_INSTEAD_), - _WITH_SENDER_<_Sender>, - __fn_t<_WITH_ENVIRONMENT_, _Env>...>; - - template - using __set_error_t = completion_signatures)>; - - template - using __nothrow_decay_copyable_results_t = - STDEXEC::__nothrow_decay_copyable_results_t<__completion_signatures_of_t<_Sender, _Env...>>; - - template - inline constexpr bool __can_fail = !__never_sends - || sends_stopped<_Sender, _Env> - || !__nothrow_decay_copyable_results_t<_Sender, _Env>::value; - - template - inline constexpr bool __uses_stop_source = (__can_fail<_Senders, _Env> || ...); - - template - using __env_t = __if_c<__uses_stop_source<_Env, _Senders...>, __stoppable_env_t<_Env>, _Env>; - - template - struct __completions - { - // TODO(ericniebler): check that all senders have a common completion domain - template - using __all_nothrow_decay_copyable_results_t = - __mand<__nothrow_decay_copyable_results_t<_Senders, _Env...>...>; - - template - using __value_tuple_t = __minvoke<__if_c<(0 == sizeof...(_Rest)), - __mconst<_ValueTuple>, - __q<__too_many_value_completions_error_t>>, - _Sender, - _Env...>; - - template - using __single_values_of_t = __value_types_t<__completion_signatures_of_t<_Sender, _Env...>, - __mtransform<__q<__decay_t>, __q<__mlist>>, - __mbind_front_q<__value_tuple_t, _Sender>>; - - template - using __set_values_sig_t = - __minvoke_q>, __single_values_of_t<_Senders>...>>; - - template - using __f = __minvoke_q< - __concat_completion_signatures_t, - __minvoke_q<__eptr_completion_unless_t, __all_nothrow_decay_copyable_results_t<_Senders...>>, - __minvoke<__mwith_default<__qq<__set_values_sig_t>, completion_signatures<>>, _Senders...>, - __transform_reduce_completion_signatures_t<__completion_signatures_of_t<_Senders, _Env...>, - __mconst>::__f, - __set_error_t, - completion_signatures, - __concat_completion_signatures_t>...>; - }; + struct _THE_GIVEN_SENDER_CAN_COMPLETE_SUCCESSFULLY_IN_MORE_THAN_ONE_WAY_; + struct _USE_WHEN_ALL_WITH_VARIANT_INSTEAD_; - template - struct __completions_for; + template + concept __can_fail = _CmplSigs::__count(set_error) != 0 // + || _CmplSigs::__count(set_stopped) != 0 // + || !__nothrow_decay_copyable_results_t<_CmplSigs>(); - template <> - struct __completions_for<> - { - template - using __f = __completions<>::template __f<_Senders...>; - }; + template + using __uses_stop_source_t = + __mbool<(__can_fail> || ...)>; - template - struct __completions_for<_Env> - { - template - using __f = __completions<__env_t<_Env, _Senders...>>::template __f<_Senders...>; - }; + template + using __env_t = __if<__uses_stop_source_t<_Env, _CvSenders...>, __stoppable_env_t<_Env>, _Env>; template constexpr void __set_values(_Receiver& __rcvr, _ValuesTuple& __values) noexcept @@ -531,38 +455,6 @@ namespace STDEXEC static_cast<_ValuesTuple&&>(__values)); } - template - using __values_opt_tuple_t = value_types_of_t<_Sender, _ChildEnv, __decayed_tuple, __optional>; - - template - requires(__max1_sender<_Senders, __env_t<_Env, _Senders...>> && ...) - struct __traits - { - using __child_env = __env_t<_Env, _Senders...>; - - // tuple>, optional>, ...> - using __values_tuple = - __minvoke<__mwith_default< - __mtransform<__mbind_front_q<__values_opt_tuple_t, __child_env>, __q<__tuple>>, - __ignore>, - _Senders...>; - - using __collect_errors = __mbind_front_q<__mset_insert, __mset<>>; - - using __errors_list = - __minvoke<__mconcat<>, - __if<__mand<__nothrow_decay_copyable_results_t<_Senders, __child_env>...>, - __mlist<>, - __mlist>, - __error_types_of_t<_Senders, __child_env, __q<__mlist>>...>; - - using __errors_variant = __mapply<__q<__uniqued_variant>, __errors_list>; - static constexpr bool __uses_stop_source = __when_all::__uses_stop_source<_Env, _Senders...>; - }; - - struct _INVALID_ARGUMENTS_TO_WHEN_ALL_ - {}; - template -# if !STDEXEC_USE_MODULES() - static -# endif - constexpr auto __mk_state_fn(_Receiver&& __rcvr) noexcept + struct __when_all_impl : __sexpr_defaults { - return [&](__ignore, __ignore, _Child&&...) noexcept - requires(__max1_sender<_Child, __env_t, _Child...>> && ...) + static constexpr auto __mk_accumulate_sigs_fn() noexcept { - using _Traits = __traits, _Child...>; - using _ErrorsVariant = _Traits::__errors_variant; - using _ValuesTuple = _Traits::__values_tuple; - using _ChildEnv = _Traits::__child_env; - using _State = __state<_ErrorsVariant, - _ValuesTuple, - _Receiver, - (sends_stopped<_Child, _ChildEnv> || ...), - _Traits::__uses_stop_source>; - return _State{static_cast<_Receiver&&>(__rcvr), sizeof...(_Child)}; - }; - } + return [](_State __state, _Item __item) + { + STDEXEC_IF_OK(__item) + STDEXEC_IF_OK(__state) + { + static_assert(_State::__count(set_value) <= 1); + auto __ignore = __ignore_completion(); + auto __vdecay = __decay_arguments(); + auto __edecay = __decay_arguments(); - template - using __mk_state_fn_t = decltype(__when_all::__mk_state_fn(__declval<_Receiver>())); + auto __item2 = STDEXEC::__transform_completion_signatures(__item, __vdecay, __edecay); + auto __item3 = STDEXEC::__transform_completion_signatures(__item2, __ignore); + auto __state2 = STDEXEC::__transform_completion_signatures(__state, __ignore); - struct __when_all_impl : __sexpr_defaults - { - template - using __completions_t = __children_of<_Self, __when_all::__completions_for<_Env...>>; + if constexpr (_Item::__count(set_value) > 1) + { + return STDEXEC::__throw_compile_time_error< + _WHAT_(_INVALID_ARGUMENT_), + _WHERE_(_IN_ALGORITHM_, when_all_t), + _WHY_(_THE_GIVEN_SENDER_CAN_COMPLETE_SUCCESSFULLY_IN_MORE_THAN_ONE_WAY_), + _TO_FIX_THIS_ERROR_(_USE_WHEN_ALL_WITH_VARIANT_INSTEAD_)>(); + } + else if constexpr (_Item::__count(set_value) == 0 || _State::__count(set_value) == 0) + { + return STDEXEC::__concat_completion_signatures(__state2, __item3); + } + else + { + using __vdecay_fn = __mtransform<__q1<__decay_t>, __qq<__mlist>>; + using __mk_vsig_fn = __mconcat<__qf>; + using __ivalues_t = __value_types_t<_Item, __vdecay_fn, __q<__msingle>>; + using __svalues_t = __value_types_t<_State, __qq<__mlist>, __q<__msingle>>; + using __vsig_t = __mcall<__mk_vsig_fn, __svalues_t, __ivalues_t>; + + auto __values = completion_signatures<__vsig_t>(); + return STDEXEC::__concat_completion_signatures(__values, __state2, __item3); + } + } + }; + } + + template + static constexpr auto __fold_cmplsigs(_CmplSigs... __cmplsigs) + { + auto __init = completion_signatures(); + return __tup::__fold_right_t()(__mk_accumulate_sigs_fn(), __init, __cmplsigs...); + } + + template + static constexpr auto __mk_fold_cmplsigs_fn() noexcept + { + return [](__sexpr const *) + { + return __fold_cmplsigs( + STDEXEC::get_completion_signatures<__mcall1<_CpCvref, _Child>>()...); + }; + } + + template + static constexpr auto __mk_fold_cmplsigs_fn() noexcept + { + return [](__sexpr const *) + { + using __env_t = __when_all::__env_t<_Env, __mcall1<_CpCvref, _Child>...>; + return __fold_cmplsigs( + STDEXEC::get_completion_signatures<__mcall1<_CpCvref, _Child>, __env_t>()...); + }; + } static constexpr auto __get_attrs = [](__ignore, __ignore, _Child const &...) noexcept @@ -725,21 +660,8 @@ namespace STDEXEC static consteval auto __get_completion_signatures() { static_assert(__sender_for<_Self, when_all_t>); - if constexpr (__minvocable_q<__completions_t, _Self, _Env...>) - { - // TODO: update this to use constant evaluation: - return __completions_t<_Self, _Env...>{}; - } - else if constexpr (sizeof...(_Env) == 0) - { - return STDEXEC::__throw_dependent_sender_error<_Self>(); - } - else - { - return STDEXEC::__throw_compile_time_error<_INVALID_ARGUMENTS_TO_WHEN_ALL_, - __children_of<_Self, __qq<_WITH_SENDERS_>>, - __fn_t<_WITH_ENVIRONMENT_, _Env>...>(); - } + return __mk_fold_cmplsigs_fn<__copy_cvref_fn<_Self>, _Env...>()( + std::add_pointer_t<_Self>()); } static constexpr auto __get_env = [](__ignore, _State const & __state) noexcept @@ -757,12 +679,38 @@ namespace STDEXEC } }; + template + using __values_opt_tuple_t = + value_types_of_t<_CvSender, _ChildEnv, __decayed_tuple, __optional>; + static constexpr auto __get_state = - [](_Self&& __self, _Receiver&& __rcvr) noexcept - -> __apply_result_t<__mk_state_fn_t<_Receiver>, _Self> + [](_Self&&, _Receiver&& __rcvr) noexcept { - return __apply(__when_all::__mk_state_fn(static_cast<_Receiver&&>(__rcvr)), - static_cast<_Self&&>(__self)); + using __env_t = env_of_t<_Receiver>; + using __env_fn = __mbind_front_q<__when_all::__env_t, __env_t>; + using __cmplsigs_t = completion_signatures_of_t<_Self, __env_t>; + using __child_env_t = __children_of<_Self, __env_fn>; + + using __values_tuple_t = // + __children_of< + _Self, + __mwith_default< + __mtransform<__mbind_front_q<__values_opt_tuple_t, __child_env_t>, __q<__tuple>>, + __ignore>>; + + using __errors_variant_t = __error_types_t<__cmplsigs_t, __q<__uniqued_variant>>; + + using __uses_stop_source_fn = __mbind_front_q<__uses_stop_source_t, __env_t>; + constexpr bool __uses_stop_source = __children_of<_Self, __uses_stop_source_fn>(); + constexpr bool __sends_stopped = __cmplsigs_t::__count(set_stopped) != 0; + + using __state_t = __state<__errors_variant_t, + __values_tuple_t, + _Receiver, + __sends_stopped, + __uses_stop_source>; + + return __state_t{static_cast<_Receiver&&>(__rcvr), __nbr_children_of<_Self>}; }; static constexpr auto __start = @@ -777,6 +725,8 @@ namespace STDEXEC __state.__on_stop_.emplace(get_stop_token(STDEXEC::get_env(__state.__rcvr_)), __forward_stop_request{__state.__stop_source_}); } + + // Start all the child operations: (STDEXEC::start(__child_ops), ...); }; @@ -841,8 +791,8 @@ namespace STDEXEC } else if constexpr (!__same_as<_ValuesTuple, __ignore>) { - // We only need to bother recording the completion values if we're - // not already in the "error" or "stopped" state. + // 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) { auto& __opt_values = STDEXEC::__get<_Index::value>(__state.__values_); @@ -850,19 +800,19 @@ namespace STDEXEC 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> && ...)) + STDEXEC_TRY { __opt_values.emplace(_Tuple{static_cast<_Args&&>(__args)...}); } - else + STDEXEC_CATCH_ALL { - STDEXEC_TRY + if constexpr (!__nothrow_decay_copyable<_Args...>) { - __opt_values.emplace(_Tuple{static_cast<_Args&&>(__args)...}); + __set_error(__state, std::current_exception()); } - STDEXEC_CATCH_ALL + else { - __set_error(__state, std::current_exception()); + __std::unreachable(); } } } diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 9920fd7f5..86db83bcb 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -234,6 +234,10 @@ if(NOT STDEXEC_BUILD_MODULES) NAME test_on_fail1 SOURCE stdexec/algos/adaptors/test_on_fail1.cpp LIBRARIES stdexec) + stdexec_add_build_failure_test( + NAME test_when_all_fail1 SOURCE stdexec/algos/adaptors/test_when_all_fail1.cpp + LIBRARIES stdexec) + if(STDEXEC_BUILD_RELACY_TESTS) add_subdirectory(rrd) endif() diff --git a/test/stdexec/algos/adaptors/test_when_all_fail1.cpp b/test/stdexec/algos/adaptors/test_when_all_fail1.cpp new file mode 100644 index 000000000..36c23398a --- /dev/null +++ b/test/stdexec/algos/adaptors/test_when_all_fail1.cpp @@ -0,0 +1,45 @@ +/* + * Copyright (c) 2023 NVIDIA Corporation + * + * 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. + */ + +#include + +namespace ex = STDEXEC; + +template +struct a_sender_of +{ + using sender_concept = ex::sender_tag; + using completion_signatures = ex::completion_signatures; + + struct operation + { + void start() & noexcept {} + }; + + template + auto connect(Receiver) const noexcept + { + return operation{}; + } +}; + +auto main() -> int +{ + using invalid_arg_t = a_sender_of; + ex::sender auto snd = ex::when_all(ex::just(42), invalid_arg_t()); + // build error: _THE_GIVEN_SENDER_CAN_COMPLETE_SUCCESSFULLY_IN_MORE_THAN_ONE_WAY_ + STDEXEC::sync_wait(std::move(snd)); +}