From 7e7c59a7f61ddeb4649103953002636fba14d88e Mon Sep 17 00:00:00 2001 From: Robert Leahy Date: Thu, 30 Jul 2026 13:09:50 -0400 Subject: [PATCH 1/5] Include __storage.hpp From execution.hpp Several headers in exec depend on __storage.hpp. --- include/stdexec/execution.hpp | 1 + 1 file changed, 1 insertion(+) diff --git a/include/stdexec/execution.hpp b/include/stdexec/execution.hpp index 2a767544a..a0413cdec 100644 --- a/include/stdexec/execution.hpp +++ b/include/stdexec/execution.hpp @@ -63,6 +63,7 @@ import stdexec; # include "__detail/__starts_on.hpp" # include "__detail/__stopped_as_error.hpp" # include "__detail/__stopped_as_optional.hpp" +# include "__detail/__storage.hpp" # include "__detail/__submit.hpp" # include "__detail/__sync_wait.hpp" # include "__detail/__task.hpp" From ff5ca5281025242969447e8fff17f54487c0fea8 Mon Sep 17 00:00:00 2001 From: Robert Leahy Date: Mon, 29 Jun 2026 14:00:30 -0400 Subject: [PATCH 2/5] matching_completion_signature_t Allows for template arguments deduced from a completion, for example in: template void set_value(Args&&...); To be used to determine which completion signature such a completion corresponds to. See P4288. --- .../__matching_completion_signature.hpp | 104 ++++++++++++++++++ include/stdexec/execution.hpp | 1 + test/CMakeLists.txt | 5 +- .../test_matching_completion_signature.cpp | 81 ++++++++++++++ 4 files changed, 190 insertions(+), 1 deletion(-) create mode 100644 include/stdexec/__detail/__matching_completion_signature.hpp create mode 100644 test/stdexec/types/test_matching_completion_signature.cpp diff --git a/include/stdexec/__detail/__matching_completion_signature.hpp b/include/stdexec/__detail/__matching_completion_signature.hpp new file mode 100644 index 000000000..83d3678be --- /dev/null +++ b/include/stdexec/__detail/__matching_completion_signature.hpp @@ -0,0 +1,104 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * Copyright (c) 2026 Robert Leahy. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + * + * 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 "__completion_signatures.hpp" + +#include +#include + +namespace STDEXEC +{ + + namespace __matching_completion_signature + { + + template + struct __canonicalize; + + template + struct __canonicalize<_Tag(_Args...)> + { + using type = _Tag(_Args&&...); + }; + + template + using __canonicalize_t = typename __canonicalize<_Signature>::type; + + template + inline constexpr bool __is_canonical_match = + std::is_same_v<__canonicalize_t<_Needle>, __canonicalize_t<_Signature>>; + + template + struct __canonical_match_count; + + template + struct __canonical_match_count<_Needle, completion_signatures<_Signatures...>> + : std::integral_constant + ... + 0)> + {}; + + template + struct __find_canonical_match; + + template + struct __find_canonical_match<_Needle> + {}; + + template + struct __find_canonical_match<_Needle, _Signature, _Rest...> + : std::conditional_t<__is_canonical_match<_Needle, _Signature>, + std::type_identity<_Signature>, + __find_canonical_match<_Needle, _Rest...>> + {}; + + template ::value> + struct __matching_completion_signature + {}; + + template + struct __matching_completion_signature, _Signature, 1> + : __find_canonical_match<_Signature, _Signatures...> + {}; + + } // namespace __matching_completion_signature + + template + using matching_completion_signature_t = + typename __matching_completion_signature::__matching_completion_signature<_Signatures, + _Tag(_Args...)>::type; + + template + inline constexpr bool has_matching_completion_signature_v = requires { + typename matching_completion_signature_t<_Signatures, _Tag, _Args...>; + }; + + template + struct matching_completion_signature + { + using type = matching_completion_signature_t<_Signatures, _Tag, _Args...>; + }; + + template + struct has_matching_completion_signature + : std::bool_constant> + {}; + +} // namespace STDEXEC diff --git a/include/stdexec/execution.hpp b/include/stdexec/execution.hpp index a0413cdec..802044f2e 100644 --- a/include/stdexec/execution.hpp +++ b/include/stdexec/execution.hpp @@ -45,6 +45,7 @@ import stdexec; # include "__detail/__intrusive_slist.hpp" # include "__detail/__just.hpp" # include "__detail/__let.hpp" +# include "__detail/__matching_completion_signature.hpp" # include "__detail/__meta.hpp" # include "__detail/__on.hpp" # include "__detail/__operation_states.hpp" diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index f504bab24..7a0cb6b3a 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -94,7 +94,10 @@ if(NOT STDEXEC_BUILD_MODULES) stdexec/cpos/test_cpo_upon_error.cpp stdexec/detail/test_any_allocator.cpp stdexec/schedulers/test_parallel_scheduler.cpp - ) + stdexec/schedulers/test_task_scheduler.cpp + stdexec/types/test_counting_scopes.cpp + stdexec/types/test_matching_completion_signature.cpp + stdexec/types/test_task.cpp) endif() add_library(common_test_settings INTERFACE) diff --git a/test/stdexec/types/test_matching_completion_signature.cpp b/test/stdexec/types/test_matching_completion_signature.cpp new file mode 100644 index 000000000..09ad99e8a --- /dev/null +++ b/test/stdexec/types/test_matching_completion_signature.cpp @@ -0,0 +1,81 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * Copyright (c) 2026 Robert Leahy. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + * + * 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 +#include + +#include +#include + +namespace +{ + + struct base + {}; + + struct derived : base + {}; + + using signatures = ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int), + ::STDEXEC::set_value_t(base), + ::STDEXEC::set_value_t(derived&), + ::STDEXEC::set_error_t(std::exception_ptr), + ::STDEXEC::set_stopped_t()>; + + static_assert(std::is_same_v< + ::STDEXEC::matching_completion_signature_t, + ::STDEXEC::set_value_t(int)>); + + static_assert(std::is_same_v<::STDEXEC::matching_completion_signature_t, + ::STDEXEC::set_error_t(std::exception_ptr)>); + + static_assert( + std::is_same_v<::STDEXEC::matching_completion_signature_t, + ::STDEXEC::set_stopped_t()>); + + static_assert( + std::is_same_v< + ::STDEXEC::matching_completion_signature_t, + ::STDEXEC::set_value_t(derived&)>); + + static_assert( + ::STDEXEC::has_matching_completion_signature_v); + static_assert( + ::STDEXEC::has_matching_completion_signature::value); + + static_assert( + !::STDEXEC::has_matching_completion_signature_v); + static_assert(!::STDEXEC::has_matching_completion_signature::value); + + static_assert( + !::STDEXEC::has_matching_completion_signature_v); + + static_assert( + !::STDEXEC::has_matching_completion_signature_v); + + TEST_CASE("matching_completion_signature compile-time checks", "[matching_completion_signature]") + { + CHECK(true); + } + +} // unnamed namespace From b2e18e19de3ddcffe80b34a6ef8ca394516d44c2 Mon Sep 17 00:00:00 2001 From: Robert Leahy Date: Thu, 30 Jul 2026 21:45:39 -0400 Subject: [PATCH 3/5] __tuple: Reference Aware --- include/stdexec/__detail/__tuple.hpp | 32 +++++++++++++++++++--------- 1 file changed, 22 insertions(+), 10 deletions(-) diff --git a/include/stdexec/__detail/__tuple.hpp b/include/stdexec/__detail/__tuple.hpp index cb69b5015..1fb6ad821 100644 --- a/include/stdexec/__detail/__tuple.hpp +++ b/include/stdexec/__detail/__tuple.hpp @@ -81,7 +81,8 @@ namespace STDEXEC { return static_cast<_Fn&&>(__fn)( static_cast<_Us&&>(__us)..., - static_cast<_Self&&>(__self).STDEXEC_CWG1835_TEMPLATE __box<_Ts, _Index>::__value...); + static_cast<__copy_cvref_t<_Self&&, _Ts>>( + static_cast<_Self&&>(__self).STDEXEC_CWG1835_TEMPLATE __box<_Ts, _Index>::__value)...); } }; @@ -233,7 +234,10 @@ namespace STDEXEC template STDEXEC_HOST_DEVICE_DEDUCTION_GUIDE __tuple(_Ts...) -> __tuple<_Ts...>; -# define STDEXEC_TUPLE_GET(_Idx) , static_cast<_Tuple&&>(__tupl).__val##_Idx +# define STDEXEC_TUPLE_GET(_Idx) \ + , static_cast< \ + __copy_cvref_t<_Tuple&&, decltype(static_cast<_Tuple&&>(__tupl).__val##_Idx)>>( \ + static_cast<_Tuple&&>(__tupl).__val##_Idx) // // __apply(fn, tuple, extra...) @@ -383,35 +387,43 @@ namespace STDEXEC } else if constexpr (_Index == 0) { - return static_cast<_Tuple&&>(__tupl).__val0; + return static_cast<__copy_cvref_t<_Tuple&&, decltype(static_cast<_Tuple&&>(__tupl).__val0)>>( + static_cast<_Tuple&&>(__tupl).__val0); } else if constexpr (_Index == 1) { - return static_cast<_Tuple&&>(__tupl).__val1; + return static_cast<__copy_cvref_t<_Tuple&&, decltype(static_cast<_Tuple&&>(__tupl).__val1)>>( + static_cast<_Tuple&&>(__tupl).__val1); } else if constexpr (_Index == 2) { - return static_cast<_Tuple&&>(__tupl).__val2; + return static_cast<__copy_cvref_t<_Tuple&&, decltype(static_cast<_Tuple&&>(__tupl).__val2)>>( + static_cast<_Tuple&&>(__tupl).__val2); } else if constexpr (_Index == 3) { - return static_cast<_Tuple&&>(__tupl).__val3; + return static_cast<__copy_cvref_t<_Tuple&&, decltype(static_cast<_Tuple&&>(__tupl).__val3)>>( + static_cast<_Tuple&&>(__tupl).__val3); } else if constexpr (_Index == 4) { - return static_cast<_Tuple&&>(__tupl).__val4; + return static_cast<__copy_cvref_t<_Tuple&&, decltype(static_cast<_Tuple&&>(__tupl).__val4)>>( + static_cast<_Tuple&&>(__tupl).__val4); } else if constexpr (_Index == 5) { - return static_cast<_Tuple&&>(__tupl).__val5; + return static_cast<__copy_cvref_t<_Tuple&&, decltype(static_cast<_Tuple&&>(__tupl).__val5)>>( + static_cast<_Tuple&&>(__tupl).__val5); } else if constexpr (_Index == 6) { - return static_cast<_Tuple&&>(__tupl).__val6; + return static_cast<__copy_cvref_t<_Tuple&&, decltype(static_cast<_Tuple&&>(__tupl).__val6)>>( + static_cast<_Tuple&&>(__tupl).__val6); } else if constexpr (_Index == 7) { - return static_cast<_Tuple&&>(__tupl).__val7; + return static_cast<__copy_cvref_t<_Tuple&&, decltype(static_cast<_Tuple&&>(__tupl).__val7)>>( + static_cast<_Tuple&&>(__tupl).__val7); } else { From ae173b0f45ca397d1f781bce4cc7050faf9d1e4f Mon Sep 17 00:00:00 2001 From: Robert Leahy Date: Mon, 29 Jun 2026 14:00:39 -0400 Subject: [PATCH 4/5] storage_for_completion_signatures Provides a turnkey, reference-aware way to store completion signatures for later examination or transmission. See P4288. Note that this implementation diverges from the implementation described in P4288. In order to remain device compatible this implementation stores a __tuple rather than a std::tuple. Therefore storage_for_completion_signature::arguments cannot be provided (since it returns a reference to the directly stored std::tuple, which doesn't exist in this implementation). Moreover to support algorithms which aim to be usable on host and device the implementation in this commit provides both storage_for_completion_signature::forward_arguments and ::__forward_arguments where the latter returns a __tuple and can therefore be marked as being device compatible. --- .../__storage_for_completion_signatures.hpp | 750 ++++++++++++++ include/stdexec/execution.hpp | 1 + test/CMakeLists.txt | 1 + ...test_storage_for_completion_signatures.cpp | 961 ++++++++++++++++++ 4 files changed, 1713 insertions(+) create mode 100644 include/stdexec/__detail/__storage_for_completion_signatures.hpp create mode 100644 test/stdexec/types/test_storage_for_completion_signatures.cpp diff --git a/include/stdexec/__detail/__storage_for_completion_signatures.hpp b/include/stdexec/__detail/__storage_for_completion_signatures.hpp new file mode 100644 index 000000000..a999a4bf6 --- /dev/null +++ b/include/stdexec/__detail/__storage_for_completion_signatures.hpp @@ -0,0 +1,750 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * Copyright (c) 2025 Robert Leahy. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + * + * 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 "../functional.hpp" +#include "__config.hpp" +#include "__matching_completion_signature.hpp" +#include "__receivers.hpp" +#include "__transform_completion_signatures.hpp" +#include "__tuple.hpp" +#include "__variant.hpp" + +#include +#include +#include +#include + +namespace STDEXEC +{ + + template + struct storage_for_completion_signature; + + template + struct storage_for_completion_signature<_Tag(_Args...)> + { + using tag_type = _Tag; + using signature_type = _Tag(_Args...); + using __normalized_signature_t = _Tag(_Args&&...); + + private: + using __tuple_t = __tuple<_Args...>; + + struct __forward_as_tuple_t + { + template + constexpr std::tuple<_Ts&&...> operator()(_Ts&&... __args) const noexcept + { + return {static_cast<_Ts&&>(__args)...}; + } + }; + + struct __forward_as_internal_tuple_t + { + template + STDEXEC_ATTRIBUTE(host, device) + constexpr __tuple<_Ts&&...> operator()(_Ts&&... __args) const noexcept + { + return {static_cast<_Ts&&>(__args)...}; + } + }; + + public: + template + requires(sizeof...(_OtherArgs) == sizeof...(_Args)) + && (std::is_constructible_v<_Args, _OtherArgs> && ...) + STDEXEC_ATTRIBUTE(host, device) + constexpr explicit storage_for_completion_signature(_Tag, _OtherArgs&&... __args) + noexcept((std::is_nothrow_constructible_v<_Args, _OtherArgs> && ...)) + : __args_{static_cast<_OtherArgs&&>(__args)...} + {} + + STDEXEC_ATTRIBUTE(host, device) + static constexpr auto tag() noexcept -> _Tag + { + return {}; + } + + constexpr auto forward_arguments() & noexcept + { + return STDEXEC::__apply(__forward_as_tuple_t{}, __args_); + } + + constexpr auto forward_arguments() const & noexcept + { + return STDEXEC::__apply(__forward_as_tuple_t{}, __args_); + } + + constexpr auto forward_arguments() && noexcept + { + return STDEXEC::__apply(__forward_as_tuple_t{}, static_cast<__tuple_t&&>(__args_)); + } + + constexpr auto forward_arguments() const && noexcept + { + return STDEXEC::__apply(__forward_as_tuple_t{}, static_cast<__tuple_t const &&>(__args_)); + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr auto __forward_arguments() & noexcept + { + return STDEXEC::__apply(__forward_as_internal_tuple_t{}, __args_); + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr auto __forward_arguments() const & noexcept + { + return STDEXEC::__apply(__forward_as_internal_tuple_t{}, __args_); + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr auto __forward_arguments() && noexcept + { + return STDEXEC::__apply(__forward_as_internal_tuple_t{}, static_cast<__tuple_t&&>(__args_)); + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr auto __forward_arguments() const && noexcept + { + return STDEXEC::__apply(__forward_as_internal_tuple_t{}, + static_cast<__tuple_t const &&>(__args_)); + } + + private: + __tuple_t __args_; + }; + + namespace __completion_storage + { + + template + using __storage_for_arrival_t = storage_for_completion_signature< + matching_completion_signature_t<_Signatures, _Tag, _Args...>>; + + template + struct __variant_for_signatures; + + template + struct __variant_for_signatures> + { + using __t = __uniqued_variant...>; + }; + + template + struct __nothrow_storable; + + template + struct __nothrow_storable<_Tag(_Args...)> + { + static constexpr bool value = (std::is_nothrow_constructible_v<_Args, _Args> && ...); + }; + + template + struct __arrival_storable; + + template + struct __arrival_storable, _Args...> + { + static constexpr bool value = + sizeof...(_StoredArgs) == sizeof...(_Args) + && ((std::is_reference_v<_StoredArgs> || !std::is_lvalue_reference_v<_Args>) && ...); + }; + + struct __ambiguous_normalized_completion_signatures; + struct __non_persistable_completion_signature; + + template + struct __persistable_completion_signature; + + template + struct __persistable_completion_signature<_Tag(_Args...)> + { + static constexpr bool value = + ((std::is_reference_v<_Args> || std::is_move_constructible_v<_Args>) && ...); + }; + + template + struct __persistable_completion_signatures; + + template + struct __persistable_completion_signatures> + { + static constexpr bool value = (__persistable_completion_signature<_Signatures>::value && ...); + }; + + template + struct __normalized_completion_signatures; + + template + struct __normalized_completion_signatures> + { + using __t = transform_completion_signatures::__normalized_signature_t...>>; + }; + + template + using __normalized_completion_signatures_t = + typename __normalized_completion_signatures<_Signatures>::__t; + + template + inline constexpr bool __has_unique_normalized_completion_signatures = + __mapply<__msize, _Signatures>::value + == __mapply<__msize, __normalized_completion_signatures_t<_Signatures>>::value; + + template + consteval auto __get_completion_signatures() + { + if constexpr (!__has_unique_normalized_completion_signatures<_InputSignatures>) + { + return __throw_compile_time_error<__ambiguous_normalized_completion_signatures>(); + } + else if constexpr (!__persistable_completion_signatures<_InputSignatures>::value) + { + return __throw_compile_time_error<__non_persistable_completion_signature>(); + } + else + { + return _OutputSignatures{}; + } + } + + template + using __variant_for_storage_t = decltype(std::declval<_Storage>().__get_variant()); + + template + struct __variant_alternatives; + + template + struct __variant_alternatives<__variant<_Alternatives...>> + { + template + using __with_cvref_t = __mlist...>; + }; + + template + using __variant_alternatives_t = typename __variant_alternatives< + std::remove_cvref_t<_Variant>>::template __with_cvref_t<_Variant>; + + template + struct __nothrow_visit_stored_completion; + + template + struct __nothrow_visit_stored_completion<_Visitor, __mlist<_Alternatives...>> + { + static constexpr bool value = __nothrow_invocable<_Visitor, _Alternatives...>; + }; + + template + struct __nothrow_visit_stored_completion<_Visitor, + __mlist<_Alternatives...>, + __mlist<_CurrentAlternatives...>, + _Variants...> + { + static constexpr bool value = + (__nothrow_visit_stored_completion<_Visitor, + __mlist<_Alternatives..., _CurrentAlternatives>, + _Variants...>::value + && ...); + }; + + template + constexpr bool __nothrow_visit_stored_completion_for_variants_v = + __nothrow_invocable<_Visitor> + && __nothrow_visit_stored_completion<_Visitor, + __mlist<>, + __variant_alternatives_t<_Variants>...>::value; + + template + constexpr bool __nothrow_visit_stored_completion_v = + __nothrow_visit_stored_completion_for_variants_v<_Visitor, + __variant_for_storage_t<_Storages>...>; + + template + struct __nothrow_visit_stored_completion_r; + + template + struct __nothrow_visit_stored_completion_r<_Result, _Visitor, __mlist<_Alternatives...>> + { + static constexpr bool value = + std::is_nothrow_invocable_r_v<_Result, _Visitor, _Alternatives...>; + }; + + template + struct __nothrow_visit_stored_completion_r<_Result, + _Visitor, + __mlist<_Alternatives...>, + __mlist<_CurrentAlternatives...>, + _Variants...> + { + static constexpr bool value = + (__nothrow_visit_stored_completion_r<_Result, + _Visitor, + __mlist<_Alternatives..., _CurrentAlternatives>, + _Variants...>::value + && ...); + }; + + template + constexpr bool __nothrow_visit_stored_completion_r_for_variants_v = + std::is_nothrow_invocable_r_v<_Result, _Visitor> + && __nothrow_visit_stored_completion_r<_Result, + _Visitor, + __mlist<>, + __variant_alternatives_t<_Variants>...>::value; + + template + constexpr bool __nothrow_visit_stored_completion_r_v = + __nothrow_visit_stored_completion_r_for_variants_v<_Result, + _Visitor, + __variant_for_storage_t<_Storages>...>; + + template + STDEXEC_ATTRIBUTE(host, device) + constexpr decltype(auto) __visit_variant_completion(_Visitor&& __visitor, _Variant&& __variant) + { + return STDEXEC::__visit(static_cast<_Visitor&&>(__visitor), + static_cast<_Variant&&>(__variant)); + } + + template + STDEXEC_ATTRIBUTE(host, device) + constexpr decltype(auto) + __visit_present_stored_completion(_Visitor&& __visitor, _Storage&& __storage) + { + return __completion_storage::__visit_variant_completion( + static_cast<_Visitor&&>(__visitor), + static_cast<_Storage&&>(__storage).__get_variant()); + } + + template + STDEXEC_ATTRIBUTE(host, device) + constexpr decltype(auto) __visit_present_stored_completion(_Visitor&& __visitor, + _Storage&& __storage, + _Storages&&... __storages) + { + return __completion_storage::__visit_variant_completion( + [&](auto&& __completion) -> decltype(auto) + { + return __completion_storage::__visit_present_stored_completion( + [&](auto&&... __completions) -> decltype(auto) + { + return STDEXEC::__invoke(static_cast<_Visitor&&>(__visitor), + static_cast(__completion), + static_cast(__completions)...); + }, + static_cast<_Storages&&>(__storages)...); + }, + static_cast<_Storage&&>(__storage).__get_variant()); + } + + template + struct __complete_stored_completion + { + template + STDEXEC_ATTRIBUTE(host, device) + constexpr void operator()(_Complete&& __complete, _Receiver&&) const noexcept + { + static_cast<_Complete&&>(__complete)(); + } + }; + + template <> + struct __complete_stored_completion + { + template + STDEXEC_ATTRIBUTE(host, device) + constexpr void operator()(_Complete&& __complete, _Receiver&& __rcvr) const noexcept + { + STDEXEC_TRY + { + static_cast<_Complete&&>(__complete)(); + } + STDEXEC_CATCH_ALL + { + set_error(static_cast<_Receiver&&>(__rcvr), std::current_exception()); + } + } + }; + + } // namespace __completion_storage + + template + STDEXEC_ATTRIBUTE(host, device) + constexpr decltype(auto) + visit_stored_completion(_Visitor&& __visitor, _Storage&& __storage, _Storages&&... __storages) + noexcept( + __completion_storage::__nothrow_visit_stored_completion_v<_Visitor, _Storage, _Storages...>) + { + if (!(__storage.has_completion() && ... && __storages.has_completion())) + { + return STDEXEC::__invoke(static_cast<_Visitor&&>(__visitor)); + } + return __completion_storage::__visit_present_stored_completion( + static_cast<_Visitor&&>(__visitor), + static_cast<_Storage&&>(__storage), + static_cast<_Storages&&>(__storages)...); + } + + template + STDEXEC_ATTRIBUTE(host, device) + constexpr _Result + visit_stored_completion(_Visitor&& __visitor, _Storage&& __storage, _Storages&&... __storages) + noexcept(__completion_storage::__nothrow_visit_stored_completion_r_v<_Result, + _Visitor, + _Storage, + _Storages...>) + { + if (!(__storage.has_completion() && ... && __storages.has_completion())) + { + if constexpr (std::is_void_v<_Result>) + { + STDEXEC::__invoke(static_cast<_Visitor&&>(__visitor)); + return; + } + else + { + return STDEXEC::__invoke(static_cast<_Visitor&&>(__visitor)); + } + } + if constexpr (std::is_void_v<_Result>) + { + __completion_storage::__visit_present_stored_completion(static_cast<_Visitor&&>(__visitor), + static_cast<_Storage&&>(__storage), + static_cast<_Storages&&>( + __storages)...); + return; + } + else + { + return __completion_storage::__visit_present_stored_completion( + static_cast<_Visitor&&>(__visitor), + static_cast<_Storage&&>(__storage), + static_cast<_Storages&&>(__storages)...); + } + } + + enum class storage_for_completion_signatures_error_policy + { + internalize, + propagate + }; + + template + class storage_for_completion_signatures; + + template <> + class storage_for_completion_signatures< + completion_signatures<>, + storage_for_completion_signatures_error_policy::internalize> + { + public: + using completion_signatures = STDEXEC::completion_signatures<>; + static constexpr bool nothrow_arrive = true; + + static consteval auto get_completion_signatures() noexcept + { + return completion_signatures{}; + } + + STDEXEC_ATTRIBUTE(host, device) + static constexpr __variant<> __get_variant() noexcept + { + return __variant<>(__no_init); + } + + STDEXEC_ATTRIBUTE(host, device) + static constexpr bool has_completion() noexcept + { + return false; + } + + template + STDEXEC_ATTRIBUTE(host, device) + constexpr bool complete(_Receiver&&) && noexcept + { + return false; + } + }; + + template <> + class storage_for_completion_signatures, + storage_for_completion_signatures_error_policy::propagate> + : public storage_for_completion_signatures< + completion_signatures<>, + storage_for_completion_signatures_error_policy::internalize> + {}; + + template + class storage_for_completion_signatures< + completion_signatures<_Signatures...>, + storage_for_completion_signatures_error_policy::internalize> + { + using __input_completion_signatures_t = + transform_completion_signatures>; + static constexpr auto __nothrow = (__completion_storage::__nothrow_storable<_Signatures>::value + && ...); + using __maybe_throwing_signature_t = + std::conditional_t<__nothrow, + STDEXEC::completion_signatures<>, + STDEXEC::completion_signatures>; + public: + using completion_signatures = transform_completion_signatures<__input_completion_signatures_t, + __maybe_throwing_signature_t>; + static constexpr bool nothrow_arrive = true; + + static consteval auto get_completion_signatures() + { + return __completion_storage::__get_completion_signatures<__input_completion_signatures_t, + completion_signatures>(); + } + + private: + template + using __storage_for_arrival_t = + __completion_storage::__storage_for_arrival_t<__input_completion_signatures_t, _Tag, _Args...>; + using __storage_t = + typename __completion_storage::__variant_for_signatures::__t; + __storage_t __storage_; + public: + STDEXEC_ATTRIBUTE(host, device) + constexpr storage_for_completion_signatures() noexcept + : __storage_(__no_init) + {} + + template + requires std::is_constructible_v<__storage_for_arrival_t<_Tag, _Args...>, _Tag, _Args...> + && __completion_storage::__arrival_storable<__storage_for_arrival_t<_Tag, _Args...>, + _Args...>::value + STDEXEC_ATTRIBUTE(host, device) + constexpr void arrive(_Tag __tag, _Args&&... __args) noexcept + { + STDEXEC_ASSERT(__storage_.__is_valueless()); + constexpr auto __nothrow = + std::is_nothrow_constructible_v<__storage_for_arrival_t<_Tag, _Args...>, _Tag, _Args...>; + auto const __impl = [&]() noexcept(__nothrow) + { + __storage_.template emplace<__storage_for_arrival_t<_Tag, _Args...>>((_Tag&&) __tag, + (_Args&&) __args...); + }; + if constexpr (__nothrow) + { + __impl(); + } + else + { + STDEXEC_TRY + { + __impl(); + } + STDEXEC_CATCH_ALL + { + __storage_ + .template emplace>( + set_error, + std::current_exception()); + } + } + } + STDEXEC_ATTRIBUTE(host, device) + constexpr bool has_completion() const noexcept + { + return !__storage_.__is_valueless(); + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr __storage_t& __get_variant() & noexcept + { + return __storage_; + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr __storage_t const & __get_variant() const & noexcept + { + return __storage_; + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr __storage_t&& __get_variant() && noexcept + { + return static_cast<__storage_t&&>(__storage_); + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr __storage_t const && __get_variant() const && noexcept + { + return static_cast<__storage_t const &&>(__storage_); + } + + template _Receiver> + STDEXEC_ATTRIBUTE(host, device) + constexpr bool complete(_Receiver&& __rcvr) && noexcept + { + if (!has_completion()) + { + return false; + } + STDEXEC::__visit( + [&](auto&& __active_alternative) noexcept + { + using __completion_t = std::remove_cvref_t; + constexpr auto __nothrow = + std::is_nothrow_constructible_v<__completion_t, __completion_t&&>; + auto const __complete = [&]() noexcept(__nothrow) + { + auto __completion = __completion_t(std::move(__active_alternative)); + STDEXEC::__apply( + [&](auto&&... __args) noexcept + { + typename __completion_t::tag_type{}((_Receiver&&) __rcvr, + (decltype(__args)&&) __args...); + }, + // Odds are this is inside an operation state, which means that + // sending the completion signal may end our lifetime, which means + // we shouldn't send references into ourselves, therefore we move + // all the non-references onto the stack + std::move(__completion).__forward_arguments()); + }; + __completion_storage::__complete_stored_completion<__nothrow>{}(__complete, + (_Receiver&&) __rcvr); + }, + (__storage_t&&) __storage_); + return true; + } + }; + + template + class storage_for_completion_signatures, + storage_for_completion_signatures_error_policy::propagate> + { + using __input_completion_signatures_t = + transform_completion_signatures>; + public: + using completion_signatures = __input_completion_signatures_t; + static constexpr bool nothrow_arrive = + (__completion_storage::__nothrow_storable<_Signatures>::value && ...); + + static consteval auto get_completion_signatures() + { + return __completion_storage::__get_completion_signatures<__input_completion_signatures_t, + completion_signatures>(); + } + + private: + template + using __storage_for_arrival_t = + __completion_storage::__storage_for_arrival_t<__input_completion_signatures_t, _Tag, _Args...>; + using __storage_t = + typename __completion_storage::__variant_for_signatures::__t; + __storage_t __storage_; + + public: + STDEXEC_ATTRIBUTE(host, device) + constexpr storage_for_completion_signatures() noexcept + : __storage_(__no_init) + {} + + template + requires std::is_constructible_v<__storage_for_arrival_t<_Tag, _Args...>, _Tag, _Args...> + && __completion_storage::__arrival_storable<__storage_for_arrival_t<_Tag, _Args...>, + _Args...>::value + STDEXEC_ATTRIBUTE(host, device) + constexpr void arrive(_Tag __tag, _Args&&... __args) noexcept( + std::is_nothrow_constructible_v<__storage_for_arrival_t<_Tag, _Args...>, _Tag, _Args...>) + { + STDEXEC_ASSERT(__storage_.__is_valueless()); + STDEXEC_TRY + { + __storage_.template emplace<__storage_for_arrival_t<_Tag, _Args...>>((_Tag&&) __tag, + (_Args&&) __args...); + } + STDEXEC_CATCH_ALL + { + STDEXEC_THROW(); + } + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr bool has_completion() const noexcept + { + return !__storage_.__is_valueless(); + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr __storage_t& __get_variant() & noexcept + { + return __storage_; + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr __storage_t const & __get_variant() const & noexcept + { + return __storage_; + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr __storage_t&& __get_variant() && noexcept + { + return static_cast<__storage_t&&>(__storage_); + } + + STDEXEC_ATTRIBUTE(host, device) + constexpr __storage_t const && __get_variant() const && noexcept + { + return static_cast<__storage_t const &&>(__storage_); + } + + template _Receiver> + STDEXEC_ATTRIBUTE(host, device) + constexpr bool complete(_Receiver&& __rcvr) && noexcept(nothrow_arrive) + { + if (!has_completion()) + { + return false; + } + STDEXEC::__visit( + [&](auto&& __active_alternative) noexcept(nothrow_arrive) + { + using __completion_t = std::remove_cvref_t; + auto __completion = __completion_t(std::move(__active_alternative)); + STDEXEC::__apply( + [&](auto&&... __args) noexcept( + noexcept(typename __completion_t::tag_type{}((_Receiver&&) __rcvr, + (decltype(__args)&&) __args...))) + { + typename __completion_t::tag_type{}((_Receiver&&) __rcvr, + (decltype(__args)&&) __args...); + }, + std::move(__completion).__forward_arguments()); + }, + (__storage_t&&) __storage_); + return true; + } + }; + +} // namespace STDEXEC diff --git a/include/stdexec/execution.hpp b/include/stdexec/execution.hpp index 802044f2e..fae512c2b 100644 --- a/include/stdexec/execution.hpp +++ b/include/stdexec/execution.hpp @@ -65,6 +65,7 @@ import stdexec; # include "__detail/__stopped_as_error.hpp" # include "__detail/__stopped_as_optional.hpp" # include "__detail/__storage.hpp" +# include "__detail/__storage_for_completion_signatures.hpp" # include "__detail/__submit.hpp" # include "__detail/__sync_wait.hpp" # include "__detail/__task.hpp" diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 7a0cb6b3a..8cbb5bb77 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -97,6 +97,7 @@ if(NOT STDEXEC_BUILD_MODULES) stdexec/schedulers/test_task_scheduler.cpp stdexec/types/test_counting_scopes.cpp stdexec/types/test_matching_completion_signature.cpp + stdexec/types/test_storage_for_completion_signatures.cpp stdexec/types/test_task.cpp) endif() diff --git a/test/stdexec/types/test_storage_for_completion_signatures.cpp b/test/stdexec/types/test_storage_for_completion_signatures.cpp new file mode 100644 index 000000000..2d6b71e8a --- /dev/null +++ b/test/stdexec/types/test_storage_for_completion_signatures.cpp @@ -0,0 +1,961 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * Copyright (c) 2025 Robert Leahy. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + * + * 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 +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +namespace +{ + + template + concept stored_completion = requires { typename std::remove_cvref_t::tag_type; }; + + template + concept completion_with_tag = + stored_completion + && std::is_same_v::tag_type, Tag>; + + template + struct overloaded : Fns... + { + using Fns::operator()...; + }; + + template + overloaded(Fns...) -> overloaded; + + template + concept tuple_size_defined_for = requires { typename std::tuple_size::type; }; + + template + concept tuple_element_defined_for = requires { typename std::tuple_element<0, T>::type; }; + + struct monostate_may_throw + { + void operator()(std::monostate) const {} + + template + requires completion_with_tag + void operator()(Completion&&) const noexcept + {} + }; + + static_assert(std::is_same_v<::STDEXEC::set_value_t(int), ::STDEXEC::set_value_t(int const)>); + static_assert(std::is_same_v<::STDEXEC::set_value_t(int*), ::STDEXEC::set_value_t(int[])>); + static_assert( + std::is_same_v<::STDEXEC::set_value_t(int (*)(int)), ::STDEXEC::set_value_t(int(int))>); + + TEST_CASE("A single stored completion exposes its tag and arguments", + "[storage_for_completion_signatures]") + { + using completion = + ::STDEXEC::storage_for_completion_signature<::STDEXEC::set_value_t(int, int&)>; + + static_assert(!std::derived_from>); + static_assert(!tuple_size_defined_for); + static_assert(!tuple_element_defined_for); + static_assert(std::is_same_v); + static_assert(std::is_same_v); + static_assert( + std::is_same_v); + static_assert(std::is_constructible_v); + static_assert(!std::is_constructible_v); + static_assert(std::is_same_v); + static_assert(noexcept(completion::tag())); + static_assert(std::is_same_v().forward_arguments()), + std::tuple>); + static_assert(std::is_same_v().forward_arguments()), + std::tuple>); + static_assert(std::is_same_v().forward_arguments()), + std::tuple>); + static_assert(std::is_same_v().forward_arguments()), + std::tuple>); + static_assert(std::is_same_v().__forward_arguments()), + ::STDEXEC::__tuple>); + static_assert(std::is_same_v().__forward_arguments()), + ::STDEXEC::__tuple>); + static_assert(std::is_same_v().__forward_arguments()), + ::STDEXEC::__tuple>); + static_assert( + std::is_same_v().__forward_arguments()), + ::STDEXEC::__tuple>); + using internal_arguments = decltype(std::declval().__forward_arguments()); + static_assert( + std::is_same_v(std::declval())), int&&>); + static_assert( + std::is_same_v(std::declval())), int&>); + + int referenced = 42; + completion c{::STDEXEC::set_value, 13, referenced}; + auto args = c.forward_arguments(); + CHECK(std::get<0>(args) == 13); + CHECK(&std::get<1>(args) == &referenced); + + auto const & cc = c; + auto cc_args = cc.forward_arguments(); + std::get<1>(cc_args) = 99; + CHECK(referenced == 99); + + ::STDEXEC::__apply( + [](auto&& first, auto&& second) + { + static_assert(std::is_same_v); + static_assert(std::is_same_v); + }, + std::move(c).__forward_arguments()); + + int other = 7; + ::STDEXEC::storage_for_completion_signature<::STDEXEC::set_value_t(int&)> c2{ + ::STDEXEC::set_value, + other}; + bool invoked = false; + std::apply( + [&](auto&&... values) + { + CHECK(!invoked); + invoked = true; + CHECK(sizeof...(values) == 3); + }, + std::tuple_cat(std::move(c).forward_arguments(), c2.forward_arguments())); + CHECK(invoked); + } + + TEST_CASE("An internal argument tuple forwards more than eight values", + "[storage_for_completion_signatures]") + { + using completion = ::STDEXEC::storage_for_completion_signature< + ::STDEXEC::set_value_t(int, int, int, int, int, int, int, int, int)>; + + completion c{::STDEXEC::set_value, 0, 1, 2, 3, 4, 5, 6, 7, 8}; + ::STDEXEC::__apply( + [](_Ts&&... values) + { + static_assert(sizeof...(_Ts) == 9); + static_assert((std::is_same_v<_Ts, int> && ...)); + CHECK((values + ...) == 36); + }, + std::move(c).__forward_arguments()); + } + + template + concept can_arrive_set_value_with = requires(Storage& s, Value&& value) { + s.arrive(::STDEXEC::set_value, static_cast(value)); + }; + + template + concept can_arrive_set_error_with = requires(Storage& s, Error&& error) { + s.arrive(::STDEXEC::set_error, static_cast(error)); + }; + + struct copyable_value + { + copyable_value() = default; + copyable_value(copyable_value const &) noexcept = default; + copyable_value(copyable_value&&) noexcept = default; + }; + + using copyable_value_storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(copyable_value)>>; + + static_assert(!can_arrive_set_value_with); + static_assert(can_arrive_set_value_with); + + TEST_CASE("Storing no completion signatures works", "[storage_for_completion_signatures]") + { + ::STDEXEC::storage_for_completion_signatures<::STDEXEC::completion_signatures<>> storage; + static_assert(std::is_empty_v); + static_assert( + std::is_same_v>); + static_assert(noexcept(storage.has_completion())); + CHECK(!storage.has_completion()); + bool visited_empty = false; + ::STDEXEC::visit_stored_completion([&] { visited_empty = true; }, std::move(storage)); + CHECK(visited_empty); + struct receiver + { + using receiver_concept = ::STDEXEC::receiver_t; + explicit receiver(bool& moved) noexcept + : moved_(&moved) + {} + receiver(receiver&& other) noexcept + : moved_(std::exchange(other.moved_, nullptr)) + { + *moved_ = true; + } + void set_value() noexcept + { + FAIL("Unexpected value"); + } + bool* moved_; + }; + bool moved = false; + CHECK(!std::move(storage).complete(receiver{moved})); + CHECK(!moved); + } + + TEST_CASE("Storing simple completion signatures and then visiting them works", + "[storage_for_completion_signatures]") + { + using completion_signatures = + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int), + ::STDEXEC::set_stopped_t(), + ::STDEXEC::set_error_t(std::error_code)>; + using storage = ::STDEXEC::storage_for_completion_signatures; + static_assert(set_equivalent); + { + storage s; + CHECK(!s.has_completion()); + bool visited_empty = false; + ::STDEXEC::visit_stored_completion(overloaded{[&] { visited_empty = true; }, + [](auto&&) { FAIL("Unexpected completion"); }}, + std::move(s)); + CHECK(visited_empty); + } + { + storage s; + CHECK(!s.has_completion()); + static_assert(noexcept(s.arrive(::STDEXEC::set_stopped))); + s.arrive(::STDEXEC::set_stopped); + CHECK(s.has_completion()); + static_assert(noexcept( + ::STDEXEC::visit_stored_completion(overloaded{[]() noexcept {}, [](auto&&) noexcept {}}, + std::move(s)))); + static_assert(!noexcept( + ::STDEXEC::visit_stored_completion(overloaded{[] {}, [](auto&&) {}}, std::move(s)))); + bool invoked = false; + ::STDEXEC::visit_stored_completion( + overloaded{ + [] { FAIL("Unexpected completion"); }, + [&](auto&& completion) + { + CHECK(!invoked); + invoked = true; + if constexpr (completion_with_tag) + { + CHECK(std::tuple_size_v == 0); + } + else + { + FAIL("Unexpected completion"); + } + }}, + std::move(s)); + CHECK(invoked); + } + { + storage s; + CHECK(!s.has_completion()); + static_assert(noexcept(s.arrive(::STDEXEC::set_value, 5))); + s.arrive(::STDEXEC::set_value, 5); + CHECK(s.has_completion()); + static_assert(!noexcept( + ::STDEXEC::visit_stored_completion(overloaded{[] {}, [](auto&&) {}}, std::move(s)))); + bool invoked = false; + ::STDEXEC::visit_stored_completion( + overloaded{ + [] { FAIL("Unexpected completion"); }, + [&](auto&& completion) + { + CHECK(!invoked); + invoked = true; + if constexpr (completion_with_tag) + { + auto args = static_cast(completion).forward_arguments(); + CHECK(std::get<0>(args) == 5); + } + else + { + FAIL("Unexpected completion"); + } + }}, + std::move(s)); + CHECK(invoked); + } + { + storage s; + CHECK(!s.has_completion()); + static_assert(noexcept(s.arrive(::STDEXEC::set_error, std::error_code{}))); + s.arrive(::STDEXEC::set_error, make_error_code(std::errc::no_such_file_or_directory)); + CHECK(s.has_completion()); + static_assert(!noexcept( + ::STDEXEC::visit_stored_completion(overloaded{[] {}, [](auto&&) {}}, std::move(s)))); + bool invoked = false; + ::STDEXEC::visit_stored_completion( + overloaded{ + [] { FAIL("Unexpected completion"); }, + [&](auto&& completion) + { + CHECK(!invoked); + invoked = true; + if constexpr (completion_with_tag) + { + auto args = static_cast(completion).forward_arguments(); + CHECK(std::get<0>(args) == make_error_code(std::errc::no_such_file_or_directory)); + } + else + { + FAIL("Unexpected completion"); + } + }}, + std::move(s)); + CHECK(invoked); + } + } + + TEST_CASE("Stored completions can be visited as completion objects", + "[storage_for_completion_signatures]") + { + using storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int, int&)>>; + + int referenced = 42; + storage s; + s.arrive(::STDEXEC::set_value, 13, referenced); + + bool invoked = false; + ::STDEXEC::visit_stored_completion( + overloaded{[] { FAIL("Unexpected completion"); }, + [&](auto&& completion) + { + CHECK(!invoked); + invoked = true; + if constexpr (completion_with_tag) + { + auto args = + static_cast(completion).forward_arguments(); + CHECK(std::get<0>(args) == 13); + CHECK(&std::get<1>(args) == &referenced); + } + else + { + FAIL("Unexpected completion"); + } + }}, + s); + CHECK(invoked); + + auto const & cs = s; + ::STDEXEC::visit_stored_completion( + overloaded{[] { FAIL("Unexpected completion"); }, + [&](auto&& completion) + { + if constexpr (completion_with_tag) + { + static_assert(std::is_const_v>); + auto args = completion.forward_arguments(); + std::get<1>(args) = 99; + } + else + { + FAIL("Unexpected completion"); + } + }}, + cs); + CHECK(referenced == 99); + } + + TEST_CASE("Multiple stored completions can be visited together", + "[storage_for_completion_signatures]") + { + using first_storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int)>>; + using second_storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_error_t(std::exception_ptr)>>; + using third_storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int&)>>; + + int referenced = 42; + first_storage first; + second_storage second; + third_storage third; + first.arrive(::STDEXEC::set_value, 13); + second.arrive(::STDEXEC::set_error, std::exception_ptr{}); + third.arrive(::STDEXEC::set_value, referenced); + + bool invoked = false; + ::STDEXEC::visit_stored_completion( + overloaded{ + [] { FAIL("Unexpected completion"); }, + [&](auto&& first_completion, auto&& second_completion, auto&& third_completion) + { + CHECK(!invoked); + invoked = true; + if constexpr (completion_with_tag + && completion_with_tag + && completion_with_tag) + { + auto args = std::tuple_cat( + static_cast(first_completion).forward_arguments(), + static_cast(second_completion).forward_arguments(), + static_cast(third_completion).forward_arguments()); + CHECK(std::get<0>(args) == 13); + CHECK(&std::get<2>(args) == &referenced); + } + else + { + FAIL("Unexpected completion"); + } + }}, + std::move(first), + std::move(second), + std::move(third)); + CHECK(invoked); + } + + TEST_CASE("Stored completion visitation computes noexcept", "[storage_for_completion_signatures]") + { + using storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int)>>; + + storage s; + static_assert(noexcept( + ::STDEXEC::visit_stored_completion(overloaded{[]() noexcept {}, [](auto&&) noexcept {}}, s))); + static_assert( + !noexcept(::STDEXEC::visit_stored_completion(overloaded{[] {}, [](auto&&) noexcept {}}, s))); + static_assert(!noexcept( + ::STDEXEC::visit_stored_completion(overloaded{[]() noexcept {}, [](auto&&) {}}, s))); + + using empty_storage = + ::STDEXEC::storage_for_completion_signatures<::STDEXEC::completion_signatures<>>; + empty_storage empty; + static_assert(noexcept(::STDEXEC::visit_stored_completion([]() noexcept {}, empty))); + } + + TEST_CASE("Stored completion visitation collapses empty storage to nullary invocation", + "[storage_for_completion_signatures]") + { + using storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int)>>; + + storage first; + storage second; + second.arrive(::STDEXEC::set_value, 13); + + bool visited_empty = false; + ::STDEXEC::visit_stored_completion(overloaded{[&] { visited_empty = true; }, + [](auto&&) { FAIL("Unexpected completion"); }}, + first); + CHECK(visited_empty); + + bool visited_empty_from_mixed = false; + ::STDEXEC::visit_stored_completion(overloaded{[&] { visited_empty_from_mixed = true; }, + [](auto&&...) { FAIL("Unexpected completion"); }}, + second, + first); + CHECK(visited_empty_from_mixed); + } + + TEST_CASE("Stored completion visitation returns visitor results", + "[storage_for_completion_signatures]") + { + using storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int)>>; + + storage empty; + storage value; + value.arrive(::STDEXEC::set_value, 42); + + auto const classify = overloaded{ + []() noexcept -> int { return 0; }, + [](auto&& completion) noexcept -> int + { + if constexpr (completion_with_tag) + { + return std::get<0>(static_cast(completion).forward_arguments()); + } + else + { + return -1; + } + }}; + + static_assert( + std::is_same_v); + static_assert(noexcept(::STDEXEC::visit_stored_completion(classify, empty))); + CHECK(::STDEXEC::visit_stored_completion(classify, empty) == 0); + CHECK(::STDEXEC::visit_stored_completion(classify, value) == 42); + + static_assert( + std::is_same_v(classify, value)), long>); + static_assert(noexcept(::STDEXEC::visit_stored_completion(classify, value))); + CHECK(::STDEXEC::visit_stored_completion(classify, value) == 42L); + } + + TEST_CASE("Storing simple completion signatures and then completing a receiver therewith works", + "[storage_for_completion_signatures]") + { + using storage_type = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(), + ::STDEXEC::set_stopped_t(), + ::STDEXEC::set_error_t(std::exception_ptr)>>; + { + storage_type storage; + storage.arrive(::STDEXEC::set_value); + CHECK(std::move(storage).complete(expect_void_receiver{})); + } + { + std::optional storage(std::in_place); + storage->arrive(::STDEXEC::set_error, std::make_exception_ptr(std::logic_error("TEST"))); + std::exception_ptr ex; + struct receiver + { + using receiver_concept = ::STDEXEC::receiver_t; + void set_value() noexcept + { + FAIL("Unexpected value invocation"); + } + void set_stopped() noexcept + { + FAIL("Unexpected stopped invocation"); + } + void set_error(std::exception_ptr&& ex) noexcept + { + // This ensures that the exception_ptr is moved onto the stack + CHECK(storage_); + storage_.reset(); + CHECK(!ex_); + ex_ = std::move(ex); + } + std::optional& storage_; + std::exception_ptr& ex_; + }; + CHECK(std::move(*storage).complete(receiver{storage, ex})); + REQUIRE(ex); +#if !STDEXEC_NO_STDCPP_EXCEPTIONS() + bool threw = false; + try + { + std::rethrow_exception(std::move(ex)); + } + catch (std::logic_error const & ex) + { + threw = true; + CHECK(ex.what() == std::string_view("TEST")); + } + CHECK(threw); +#endif // !STDEXEC_NO_STDCPP_EXCEPTIONS() + } + } + +#if !STDEXEC_NO_STDCPP_EXCEPTIONS() + TEST_CASE("When moving a stored value onto the stack throws complete reports std::exception_ptr", + "[storage_for_completion_signatures]") + { + struct throws_on_second_move + { + explicit throws_on_second_move(int& moves) noexcept + : moves_(&moves) + {} + throws_on_second_move(throws_on_second_move&& other) + { + moves_ = other.moves_; + ++*moves_; + if (*moves_ == 2) + { + throw std::runtime_error("Throwing as requested"); + } + } + int* moves_; + }; + + using storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(throws_on_second_move)>>; + + struct receiver + { + using receiver_concept = ::STDEXEC::receiver_t; + void set_value(throws_on_second_move&&) noexcept + { + FAIL("Unexpected value invocation"); + } + void set_error(std::exception_ptr&& ex) noexcept + { + CHECK(!ex_); + ex_ = std::move(ex); + } + std::exception_ptr& ex_; + }; + + int moves = 0; + storage s; + s.arrive(::STDEXEC::set_value, throws_on_second_move{moves}); + CHECK(moves == 1); + + std::exception_ptr ex; + CHECK(std::move(s).complete(receiver{ex})); + REQUIRE(ex); + bool threw = false; + try + { + std::rethrow_exception(std::move(ex)); + } + catch (std::runtime_error const & ex) + { + threw = true; + CHECK(ex.what() == std::string_view("Throwing as requested")); + } + CHECK(threw); + } +#endif // !STDEXEC_NO_STDCPP_EXCEPTIONS() + + TEST_CASE("When storing a completion signature would throw it is simply coalesced to " + "std::exception_ptr", + "[storage_for_completion_signatures]") + { + struct maybe_throws_on_move + { + maybe_throws_on_move() = default; +#if !STDEXEC_NO_STDCPP_EXCEPTIONS() + maybe_throws_on_move(maybe_throws_on_move&& other) + { + if (other.throws) + { + throw std::runtime_error("Throwing as requested"); + } + } +#else + maybe_throws_on_move(maybe_throws_on_move&&) = delete; +#endif // !STDEXEC_NO_STDCPP_EXCEPTIONS() + bool throws{false}; + }; +#if !STDEXEC_NO_STDCPP_EXCEPTIONS() + { + using storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(maybe_throws_on_move)>>; + static_assert(storage::nothrow_arrive); + static_assert(set_equivalent< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(maybe_throws_on_move), + ::STDEXEC::set_error_t(std::exception_ptr)>, + storage::completion_signatures>); + { + storage s; + static_assert(noexcept(s.arrive(::STDEXEC::set_value, maybe_throws_on_move{}))); + maybe_throws_on_move obj; + obj.throws = true; + s.arrive(::STDEXEC::set_value, std::move(obj)); + bool invoked = false; + ::STDEXEC::visit_stored_completion( + overloaded{ + [] { FAIL("Unexpected completion"); }, + [&](auto&& completion) + { + CHECK(!invoked); + invoked = true; + if constexpr (completion_with_tag) + { + auto args = static_cast(completion).forward_arguments(); + REQUIRE(std::get<0>(args)); + // TODO? + } + else + { + FAIL("Unexpected completion"); + } + }}, + std::move(s)); + CHECK(invoked); + } + { + storage s; + s.arrive(::STDEXEC::set_value, maybe_throws_on_move{}); + bool invoked = false; + ::STDEXEC::visit_stored_completion( + overloaded{ + [] { FAIL("Unexpected completion"); }, + [&](auto&& completion) + { + CHECK(!invoked); + invoked = true; + if constexpr (!completion_with_tag) + { + FAIL("Unexpected completion"); + } + }}, + std::move(s)); + CHECK(invoked); + } + } +#endif // !STDEXEC_NO_STDCPP_EXCEPTIONS() + // Important that the below cases don't add the std::exception_ptr completion + // since propagating a reference can't throw + { + using signatures = + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(maybe_throws_on_move&)>; + using storage = ::STDEXEC::storage_for_completion_signatures; + static_assert(std::is_same_v); + maybe_throws_on_move obj; + storage s; + s.arrive(::STDEXEC::set_value, obj); + bool invoked = false; + ::STDEXEC::visit_stored_completion( + overloaded{ + [] { FAIL("Unexpected completion"); }, + [&](auto&& completion) + { + CHECK(!invoked); + invoked = true; + if constexpr (completion_with_tag) + { + auto args = static_cast(completion).forward_arguments(); + CHECK(&obj == &std::get<0>(args)); + } + else + { + FAIL("Unexpected completion"); + } + }}, + std::move(s)); + CHECK(invoked); + } + { + using signatures = + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(maybe_throws_on_move&&)>; + using storage = ::STDEXEC::storage_for_completion_signatures; + static_assert(std::is_same_v); + maybe_throws_on_move obj; + storage s; + s.arrive(::STDEXEC::set_value, std::move(obj)); + bool invoked = false; + ::STDEXEC::visit_stored_completion( + overloaded{ + [] { FAIL("Unexpected completion"); }, + [&](auto&& completion) + { + CHECK(!invoked); + invoked = true; + if constexpr (completion_with_tag) + { + auto args = static_cast(completion).forward_arguments(); + CHECK(&obj == &std::get<0>(args)); + } + else + { + FAIL("Unexpected completion"); + } + }}, + std::move(s)); + CHECK(invoked); + } + } + +#if !STDEXEC_NO_STDCPP_EXCEPTIONS() + TEST_CASE("Storage can propagate persistence exceptions instead of coalescing them", + "[storage_for_completion_signatures]") + { + struct maybe_throws_on_move + { + maybe_throws_on_move() = default; + maybe_throws_on_move(maybe_throws_on_move&& other) + { + if (other.throws) + { + throw std::runtime_error("Throwing as requested"); + } + other.moved = true; + } + bool throws{false}; + bool moved{false}; + }; + + using signatures = + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(maybe_throws_on_move)>; + using storage = ::STDEXEC::storage_for_completion_signatures< + signatures, + ::STDEXEC::storage_for_completion_signatures_error_policy::propagate>; + + static_assert(!storage::nothrow_arrive); + static_assert(std::is_same_v); + static_assert(!noexcept( + std::declval().arrive(::STDEXEC::set_value, std::declval()))); + + storage s; + maybe_throws_on_move obj; + obj.throws = true; + CHECK_THROWS_AS(s.arrive(::STDEXEC::set_value, std::move(obj)), std::runtime_error); + CHECK(!obj.moved); + bool visited_empty = false; + ::STDEXEC::visit_stored_completion(overloaded{[&] { visited_empty = true; }, + [](auto&&) { FAIL("Unexpected completion"); }}, + std::move(s)); + CHECK(visited_empty); + } +#endif // !STDEXEC_NO_STDCPP_EXCEPTIONS() + + TEST_CASE("Propagating storage remains noexcept when persistence is noexcept", + "[storage_for_completion_signatures]") + { + using signatures = ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int)>; + using storage = ::STDEXEC::storage_for_completion_signatures< + signatures, + ::STDEXEC::storage_for_completion_signatures_error_policy::propagate>; + + static_assert(storage::nothrow_arrive); + static_assert(std::is_same_v); + storage s; + static_assert(noexcept(s.arrive(::STDEXEC::set_value, 42))); + s.arrive(::STDEXEC::set_value, 42); + + bool invoked = false; + ::STDEXEC::visit_stored_completion( + overloaded{[]() noexcept { FAIL("Unexpected completion"); }, + [&](auto&& completion) noexcept + { + CHECK(!invoked); + invoked = true; + if constexpr (completion_with_tag) + { + auto args = + static_cast(completion).forward_arguments(); + CHECK(std::get<0>(args) == 42); + } + else + { + FAIL("Unexpected completion"); + } + }}, + std::move(s)); + CHECK(invoked); + } + + TEST_CASE("Storage deduplicates exact duplicate completion signatures", + "[storage_for_completion_signatures]") + { + using duplicate_signatures = + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int), ::STDEXEC::set_value_t(int)>; + using deduplicated_signatures = ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int)>; + + using internalizing_storage = + ::STDEXEC::storage_for_completion_signatures; + using propagating_storage = ::STDEXEC::storage_for_completion_signatures< + duplicate_signatures, + ::STDEXEC::storage_for_completion_signatures_error_policy::propagate>; + + static_assert( + std::is_same_v); + static_assert( + std::is_same_v); + + propagating_storage storage; + storage.arrive(::STDEXEC::set_value, 42); + + bool invoked = false; + ::STDEXEC::visit_stored_completion( + overloaded{[] { FAIL("Unexpected empty storage"); }, + [&](auto&& completion) + { + CHECK(!invoked); + invoked = true; + if constexpr (completion_with_tag) + { + CHECK(std::get<0>( + static_cast(completion).forward_arguments()) + == 42); + } + else + { + FAIL("Unexpected completion"); + } + }}, + std::move(storage)); + CHECK(invoked); + } + + TEST_CASE("Internalized storage errors are not external arrivals", + "[storage_for_completion_signatures]") + { + struct maybe_throws_on_move + { + maybe_throws_on_move() = default; + maybe_throws_on_move(maybe_throws_on_move&&) {} + }; + + using storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(maybe_throws_on_move)>>; + + static_assert( + set_equivalent>); + static_assert(!can_arrive_set_error_with); + } + + TEST_CASE("Storage computes completion signatures through a consteval function", + "[storage_for_completion_signatures]") + { + { + using storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int), ::STDEXEC::set_value_t(int)>>; + + static_assert(std::is_same_v); + } + { + struct maybe_throws_on_move + { + maybe_throws_on_move() = default; + maybe_throws_on_move(maybe_throws_on_move&&) {} + }; + + using storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(maybe_throws_on_move)>>; + + static_assert(set_equivalent); + } + } + + TEST_CASE("Storage rejects ambiguous or non-persistable completion signatures", + "[storage_for_completion_signatures]") + { +#if STDEXEC_NO_STDCPP_CONSTEXPR_EXCEPTIONS() + { + using storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(int), + ::STDEXEC::set_value_t(int&&)>>; + + static_assert(::STDEXEC::__merror); + } + { + struct not_move_constructible + { + not_move_constructible() = default; + not_move_constructible(not_move_constructible&&) = delete; + }; + + using storage = ::STDEXEC::storage_for_completion_signatures< + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t(not_move_constructible)>>; + + static_assert(::STDEXEC::__merror); + } +#endif + } + +} // unnamed namespace From 244d6aa000897523f4842435c332b6a906c93c8d Mon Sep 17 00:00:00 2001 From: Robert Leahy Date: Sat, 17 Oct 2026 23:18:14 -0400 Subject: [PATCH 5/5] exec::stopped_by Implements stopped_by from P4390R0. --- include/exec/stopped_by.hpp | 517 ++++++++++++++++++++++++++++++++++ test/exec/CMakeLists.txt | 1 + test/exec/test_stopped_by.cpp | 433 ++++++++++++++++++++++++++++ 3 files changed, 951 insertions(+) create mode 100644 include/exec/stopped_by.hpp create mode 100644 test/exec/test_stopped_by.cpp diff --git a/include/exec/stopped_by.hpp b/include/exec/stopped_by.hpp new file mode 100644 index 000000000..1ae07122b --- /dev/null +++ b/include/exec/stopped_by.hpp @@ -0,0 +1,517 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 Robert Leahy. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + * + * 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 "../stdexec/__detail/__manual_lifetime.hpp" +#include "../stdexec/execution.hpp" + +#include +#include +#include +#include +#include +#include +#include + +namespace experimental::execution +{ + struct stopped_by_t; + + namespace detail::stopped_by + { + struct no_classifier + {}; + + template + struct STOP_FACTORY_MUST_BE_NOTHROW_INVOCABLE + {}; + + template + struct STOP_FACTORY_MUST_RETURN_A_SENDER_IN + {}; + + template + struct STOP_SENDER_MUST_BE_NOTHROW_CONNECTABLE + {}; + + template + struct STOP_SENDER_MUST_SEND_EXACTLY_NULLARY_SET_VALUE + {}; + + template + struct CLASSIFIER_MUST_BE_NOTHROW_INVOCABLE_FOR_EACH_COMPLETION + {}; + + template + struct classifier_for_signature; + + template + struct classifier_for_signature<_Classifier, _Tag(_Args...)> + : std::bool_constant< + std::same_as<_Tag, ::STDEXEC::set_stopped_t> + || std::is_nothrow_invocable_r_v< + bool, + _Classifier&&, + _Tag, + std::add_lvalue_reference_t>>...>> + {}; + + template + struct classifier_for; + + template + struct classifier_for<_Classifier, ::STDEXEC::completion_signatures<_Signatures...>> + : std::bool_constant<(classifier_for_signature<_Classifier, _Signatures>::value && ...)> + {}; + + template + inline constexpr bool classifier_for_v = classifier_for<_Classifier, _Signatures>::value; + + template + using data_t = std::remove_cvref_t<::STDEXEC::__data_of<_Sender>>; + + template + using factory_t = std::tuple_element_t<0, data_t<_Sender>>; + + template + using classifier_t = std::tuple_element_t<1, data_t<_Sender>>; + + template + inline constexpr bool has_classifier_v = !std::same_as, no_classifier>; + + template + using stored_signatures_t = std::conditional_t< + has_classifier_v<_Sender>, + ::STDEXEC::transform_completion_signatures< + _ChildSignatures, + ::STDEXEC::completion_signatures<::STDEXEC::set_stopped_t()>>, + _ChildSignatures>; + + template + using storage_t = ::STDEXEC::storage_for_completion_signatures< + stored_signatures_t<_Sender, _ChildSignatures>>; + + template + inline constexpr bool stoppable_v = + !::STDEXEC::unstoppable_token<::STDEXEC::stop_token_of_t<_Env>>; + + template + class state + { + static constexpr bool has_classifier = !std::same_as<_Classifier, no_classifier>; + + struct receiver + { + using receiver_concept = ::STDEXEC::receiver_tag; + + template + constexpr void set_value(_Values&&... __values) && noexcept + { + __state.complete(::STDEXEC::set_value, static_cast<_Values&&>(__values)...); + } + + template + constexpr void set_error(_Error&& __error) && noexcept + { + __state.complete(::STDEXEC::set_error, static_cast<_Error&&>(__error)); + } + + constexpr void set_stopped() && noexcept + { + __state.complete(::STDEXEC::set_stopped); + } + + constexpr auto get_env() const noexcept -> ::STDEXEC::env_of_t<_Receiver> + { + return ::STDEXEC::get_env(__state.__receiver); + } + + state& __state; + }; + + struct stop_receiver + { + using receiver_concept = ::STDEXEC::receiver_tag; + + constexpr void set_value() && noexcept + { + // Save the back-pointer before destroying the operation state that owns *this. + auto& __saved_state = __state; + __saved_state.__stop_operation.__destroy(); + if (__saved_state.release_ref()) [[unlikely]] + { + __saved_state.complete_stored(); + } + } + + constexpr auto get_env() const noexcept -> ::STDEXEC::env_of_t<_Receiver> + { + return ::STDEXEC::get_env(__state.__receiver); + } + + state& __state; + }; + + using child_operation_t = ::STDEXEC::connect_result_t<_Child, receiver>; + using stop_sender_t = std::invoke_result_t<_Factory, child_operation_t&>; + using stop_operation_t = ::STDEXEC::connect_result_t; + using child_signatures_t = + ::STDEXEC::completion_signatures_of_t<_Child, ::STDEXEC::env_of_t<_Receiver>>; + using stored_signatures_t = + std::conditional_t>, + child_signatures_t>; + using storage_t = ::STDEXEC::storage_for_completion_signatures; + + struct on_stop_request + { + constexpr void operator()() && noexcept + { + __state.request_stop(); + } + + state& __state; + }; + + using stop_token_t = ::STDEXEC::stop_token_of_t<::STDEXEC::env_of_t<_Receiver>>; + using stop_callback_t = ::STDEXEC::stop_callback_for_t; + + constexpr bool release_ref() noexcept + { + // Each participant publishes its work before dropping its reference. + return __outstanding.fetch_sub(1, std::memory_order_release) == 1; + } + + constexpr void complete_stored() noexcept + { + // The last participant acquires those publications and delivers the stored + // completion. + std::atomic_thread_fence(std::memory_order_acquire); + __stop_callback.__destroy(); + std::move(__storage).complete(static_cast<_Receiver&&>(__receiver)); + } + + constexpr void request_stop() noexcept + { + if (__outstanding.fetch_add(1, std::memory_order_relaxed) == 0) [[unlikely]] + { + // Completion has already been claimed. Destroying the callback synchronizes + // with this invocation, so it is enough to let the invocation return. + return; + } + + auto& __operation = __stop_operation.__construct_from( + ::STDEXEC::connect, + std::invoke(static_cast<_Factory&&>(__factory), __child_operation), + stop_receiver{*this}); + ::STDEXEC::start(__operation); + } + + template + constexpr void complete_unclassified(_Tag __tag, _Args&&... __args) noexcept + { + if (__outstanding.load(std::memory_order_relaxed) == 1) [[likely]] + { + // Usually no stop was requested. Retire the callback and, after checking that + // it did not race with us, forward the completion without storing it. + std::atomic_thread_fence(std::memory_order_acquire); + __stop_callback.__destroy(); + if (__outstanding.load(std::memory_order_acquire) == 1) [[likely]] + { + __tag(static_cast<_Receiver&&>(__receiver), static_cast<_Args&&>(__args)...); + return; + } + } + + __storage.arrive(__tag, static_cast<_Args&&>(__args)...); + if (release_ref()) [[unlikely]] + { + complete_stored(); + } + } + + public: + using operation_state_concept = ::STDEXEC::operation_state_tag; + + constexpr explicit state(_Child&& __child, + _Factory __factory, + _Classifier __classifier, + _Receiver __receiver) + noexcept(std::is_nothrow_move_constructible_v<_Factory> + && std::is_nothrow_move_constructible_v<_Classifier> + && ::STDEXEC::__nothrow_connectable<_Child, receiver>) + : __receiver(static_cast<_Receiver&&>(__receiver)) + , __factory(static_cast<_Factory&&>(__factory)) + , __classifier(static_cast<_Classifier&&>(__classifier)) + , __child_operation(::STDEXEC::connect(static_cast<_Child&&>(__child), receiver{*this})) + {} + + STDEXEC_IMMOVABLE(state); + + constexpr void start() & noexcept + { + ::STDEXEC::start(__child_operation); + __stop_callback.__construct(::STDEXEC::get_stop_token(::STDEXEC::get_env(__receiver)), + on_stop_request{*this}); + if (release_ref()) [[unlikely]] + { + complete_stored(); + } + } + + template + constexpr void complete(_Tag __tag, _Args&&... __args) noexcept + { + if constexpr (has_classifier && !std::same_as<_Tag, ::STDEXEC::set_stopped_t>) + { + if (bool(std::invoke(static_cast<_Classifier&&>(__classifier), + _Tag{}, + std::as_const(__args)...))) + { + __storage.arrive(::STDEXEC::set_stopped); + if (release_ref()) [[likely]] + { + complete_stored(); + } + return; + } + } + + complete_unclassified(__tag, static_cast<_Args&&>(__args)...); + } + + private: + _Receiver __receiver; + STDEXEC_ATTRIBUTE(no_unique_address) + _Factory __factory; + STDEXEC_ATTRIBUTE(no_unique_address) + _Classifier __classifier; + ::STDEXEC::__manual_lifetime __stop_operation; + storage_t __storage; + ::STDEXEC::__manual_lifetime __stop_callback; + child_operation_t __child_operation; + // One reference for start() and one for the child operation. A stop callback + // temporarily adds a third for the asynchronous stop operation. Keeping the + // byte-sized counter in the low-alignment tail gives a compact child operation + // the best chance of reducing the enclosing operation state's size. + std::atomic __outstanding{2}; + }; + + template + using state_t = + state<::STDEXEC::__child_of<_Sender>, factory_t<_Sender>, classifier_t<_Sender>, _Receiver>; + + struct impl : ::STDEXEC::__sexpr_defaults + { + template + static consteval auto __get_completion_signatures() + { + if constexpr (sizeof...(_Env) == 0) + { + using __child_t = ::STDEXEC::__child_of<_Sender>; + auto __child_signatures = ::STDEXEC::get_completion_signatures<__child_t>(); + using __child_signatures_t = decltype(__child_signatures); + + if constexpr (!::STDEXEC::__valid_completion_signatures<__child_signatures_t>) + { + return __child_signatures; + } + else + { + // Without an environment, conservatively assume that storage is needed. + using __storage_t = storage_t<_Sender, __child_signatures_t>; + return __storage_t::get_completion_signatures(); + } + } + else + { + using __env_t = std::tuple_element_t<0, std::tuple<_Env...>>; + using __child_t = ::STDEXEC::__child_of<_Sender>; + auto __child_signatures = + ::STDEXEC::get_completion_signatures<__child_t, _Env...>(); + using __child_signatures_t = decltype(__child_signatures); + + if constexpr (!::STDEXEC::__valid_completion_signatures<__child_signatures_t>) + { + return __child_signatures; + } + else if constexpr (!stoppable_v<__env_t>) + { + return __child_signatures_t{}; + } + else + { + using __factory_t = factory_t<_Sender>; + using __classifier_t = classifier_t<_Sender>; + using __archetype_receiver_t = ::STDEXEC::__receiver_archetype<_Env...>; + using __child_operation_t = + ::STDEXEC::connect_result_t<__child_t, __archetype_receiver_t>; + + if constexpr (!std::is_nothrow_invocable_v<__factory_t, __child_operation_t&>) + { + return ::STDEXEC::__throw_compile_time_error< + STOP_FACTORY_MUST_BE_NOTHROW_INVOCABLE<__factory_t, __child_operation_t>, + ::STDEXEC::_WITH_SENDER_<_Sender>>(); + } + else + { + using __stop_sender_t = std::invoke_result_t<__factory_t, __child_operation_t&>; + + if constexpr (!::STDEXEC::sender_in<__stop_sender_t, _Env...>) + { + return ::STDEXEC::__throw_compile_time_error< + STOP_FACTORY_MUST_RETURN_A_SENDER_IN<__stop_sender_t, __env_t>, + ::STDEXEC::_WITH_SENDER_<_Sender>>(); + } + else if constexpr (!::STDEXEC::__nothrow_connectable<__stop_sender_t, + __archetype_receiver_t>) + { + return ::STDEXEC::__throw_compile_time_error< + STOP_SENDER_MUST_BE_NOTHROW_CONNECTABLE<__stop_sender_t, __env_t>, + ::STDEXEC::_WITH_SENDER_<_Sender>>(); + } + else + { + using __stop_signatures_t = + ::STDEXEC::completion_signatures_of_t<__stop_sender_t, _Env...>; + + if constexpr (!std::same_as< + __stop_signatures_t, + ::STDEXEC::completion_signatures<::STDEXEC::set_value_t()>>) + { + return ::STDEXEC::__throw_compile_time_error< + STOP_SENDER_MUST_SEND_EXACTLY_NULLARY_SET_VALUE<__stop_sender_t, + __stop_signatures_t>, + ::STDEXEC::_WITH_SENDER_<_Sender>>(); + } + else if constexpr (has_classifier_v<_Sender> + && !classifier_for_v<__classifier_t, __child_signatures_t>) + { + return ::STDEXEC::__throw_compile_time_error< + CLASSIFIER_MUST_BE_NOTHROW_INVOCABLE_FOR_EACH_COMPLETION<__classifier_t, + __child_signatures_t>, + ::STDEXEC::_WITH_SENDER_<_Sender>>(); + } + else + { + using __storage_t = storage_t<_Sender, __child_signatures_t>; + return __storage_t::get_completion_signatures(); + } + } + } + } + } + } + + static constexpr auto __connect = + [](_Sender&& __sender, _Receiver __receiver) noexcept( + []() constexpr + { + if constexpr (stoppable_v<::STDEXEC::env_of_t<_Receiver>>) + { + return std::is_nothrow_constructible_v, + ::STDEXEC::__child_of<_Sender>, + factory_t<_Sender>, + classifier_t<_Sender>, + _Receiver>; + } + else + { + return ::STDEXEC::__nothrow_connectable<::STDEXEC::__child_of<_Sender>, _Receiver>; + } + }()) + { + auto& [__tag, __data, __child] = __sender; + if constexpr (stoppable_v<::STDEXEC::env_of_t<_Receiver>>) + { + auto& [__factory, __classifier] = __data; + return state_t<_Sender, _Receiver>(::STDEXEC::__forward_like<_Sender>(__child), + ::STDEXEC::__forward_like<_Sender>(__factory), + ::STDEXEC::__forward_like<_Sender>(__classifier), + static_cast<_Receiver&&>(__receiver)); + } + else + { + return ::STDEXEC::connect(::STDEXEC::__forward_like<_Sender>(__child), + static_cast<_Receiver&&>(__receiver)); + } + }; + }; + } // namespace detail::stopped_by + + struct stopped_by_t + { + template <::STDEXEC::sender _Sender, + ::STDEXEC::__movable_value _Factory, + ::STDEXEC::__movable_value _Classifier> + [[nodiscard]] + constexpr auto + operator()(_Sender&& __sender, _Factory&& __factory, _Classifier&& __classifier) const + noexcept(::STDEXEC::__nothrow_decay_copyable<_Sender, _Factory, _Classifier>) + -> ::STDEXEC::__well_formed_sender auto + { + return ::STDEXEC::__sexpr{stopped_by_t{}, + std::tuple(static_cast<_Factory&&>(__factory), + static_cast<_Classifier&&>(__classifier)), + static_cast<_Sender&&>(__sender)}; + } + + template <::STDEXEC::sender _Sender, ::STDEXEC::__movable_value _Factory> + [[nodiscard]] + constexpr auto operator()(_Sender&& __sender, _Factory&& __factory) const + noexcept(::STDEXEC::__nothrow_decay_copyable<_Sender, _Factory>) + -> ::STDEXEC::__well_formed_sender auto + { + return (*this)(static_cast<_Sender&&>(__sender), + static_cast<_Factory&&>(__factory), + detail::stopped_by::no_classifier{}); + } + + template <::STDEXEC::__movable_value _Factory> + [[nodiscard]] + constexpr auto + operator()(_Factory&& __factory) const noexcept(::STDEXEC::__nothrow_decay_copyable<_Factory>) + { + return ::STDEXEC::__closure(*this, static_cast<_Factory&&>(__factory)); + } + + template <::STDEXEC::__movable_value _Factory, ::STDEXEC::__movable_value _Classifier> + requires(!::STDEXEC::sender>) + [[nodiscard]] + constexpr auto operator()(_Factory&& __factory, _Classifier&& __classifier) const + noexcept(::STDEXEC::__nothrow_decay_copyable<_Factory, _Classifier>) + { + return ::STDEXEC::__closure(*this, + static_cast<_Factory&&>(__factory), + static_cast<_Classifier&&>(__classifier)); + } + }; + + inline constexpr stopped_by_t stopped_by{}; +} // namespace experimental::execution + +namespace exec = experimental::execution; + +namespace STDEXEC +{ + template <> + struct __sexpr_impl<::exec::stopped_by_t> : ::exec::detail::stopped_by::impl + {}; +} // namespace STDEXEC diff --git a/test/exec/CMakeLists.txt b/test/exec/CMakeLists.txt index 1d967bd96..2b63d9511 100644 --- a/test/exec/CMakeLists.txt +++ b/test/exec/CMakeLists.txt @@ -39,6 +39,7 @@ set(exec_test_sources test_when_any.cpp test_at_coroutine_exit.cpp test_materialize.cpp + test_stopped_by.cpp $<$:test_io_uring_context.cpp> $<$:test_windows_thread_pool_context.cpp> test_trampoline_scheduler.cpp diff --git a/test/exec/test_stopped_by.cpp b/test/exec/test_stopped_by.cpp new file mode 100644 index 000000000..f347ec25d --- /dev/null +++ b/test/exec/test_stopped_by.cpp @@ -0,0 +1,433 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 Robert Leahy. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +#include + +#include +#include +#include + +#include +#include +#include + +namespace +{ + namespace ex = ::STDEXEC; + + using stop_env = ex::prop; + + struct main_control + { + void* operation{}; + void (*set_value)(void*, int) noexcept {}; + bool started{}; + + void complete(int value) const noexcept + { + set_value(operation, value); + } + }; + + struct main_sender + { + using sender_concept = ex::sender_tag; + + template + static consteval auto + get_completion_signatures() noexcept -> ex::completion_signatures + { + return {}; + } + + template + struct operation + { + using operation_state_concept = ex::operation_state_tag; + + void start() & noexcept + { + control->started = true; + control->operation = this; + control->set_value = [](void* pointer, int value) noexcept + { + auto& self = *static_cast(pointer); + ex::set_value(static_cast<_Receiver&&>(self.receiver), std::move(value)); + }; + } + + main_control* control; + _Receiver receiver; + }; + + template + auto connect(_Receiver receiver) const noexcept -> operation<_Receiver> + { + return {control, static_cast<_Receiver&&>(receiver)}; + } + + main_control* control; + }; + + struct stop_control + { + void* operation{}; + void (*set_value)(void*) noexcept {}; + bool started{}; + bool destroyed{}; + + void complete() const noexcept + { + set_value(operation); + } + }; + + struct stop_sender + { + using sender_concept = ex::sender_tag; + + template + static consteval auto + get_completion_signatures() noexcept -> ex::completion_signatures + { + return {}; + } + + template + struct operation + { + using operation_state_concept = ex::operation_state_tag; + + void start() & noexcept + { + control->started = true; + control->operation = this; + control->set_value = [](void* pointer) noexcept + { + auto& self = *static_cast(pointer); + ex::set_value(static_cast<_Receiver&&>(self.receiver)); + }; + } + + ~operation() + { + control->destroyed = true; + } + + stop_control* control; + _Receiver receiver; + }; + + template + auto connect(_Receiver receiver) const noexcept -> operation<_Receiver> + { + return {control, static_cast<_Receiver&&>(receiver)}; + } + + stop_control* control; + }; + + struct invalid_stop_factory + {}; + + struct throwing_move + { + throwing_move() = default; + throwing_move(throwing_move&&) noexcept(false); + }; + + struct throwing_value_sender + { + using sender_concept = ex::sender_tag; + + template + static consteval auto get_completion_signatures() noexcept + -> ex::completion_signatures + { + return {}; + } + }; + + inline constexpr auto stop_sender_sends_stopped = + ::exec::stopped_by(main_sender{nullptr}, [](auto&) noexcept { return ex::just_stopped(); }); + + inline constexpr auto stop_factory_can_throw = ::exec::stopped_by(main_sender{nullptr}, + [](auto&) + { return ex::just(); }); + + inline constexpr auto stop_factory_returns_non_sender = ::exec::stopped_by(main_sender{nullptr}, + [](auto&) noexcept + { return 42; }); + + inline constexpr auto classifier_can_throw = ::exec::stopped_by( + main_sender{nullptr}, + [](auto&) noexcept { return ex::just(); }, + [](ex::set_value_t, int const &) { return false; }); + + TEST_CASE("stopped_by validates its stop factory and classifier", "[stopped_by]") + { + STATIC_REQUIRE(ex::sender_in); + STATIC_REQUIRE(ex::sender_in); + STATIC_REQUIRE(ex::sender_in); + STATIC_REQUIRE(ex::sender_in); + STATIC_REQUIRE_FALSE(ex::sender_in); + STATIC_REQUIRE_FALSE(ex::sender_in); + STATIC_REQUIRE_FALSE(ex::sender_in); + STATIC_REQUIRE_FALSE(ex::sender_in); + } + + TEST_CASE("stopped_by has conservative environment-independent signatures", "[stopped_by]") + { + auto unclassified = ::exec::stopped_by(main_sender{nullptr}, + [](auto&) noexcept { return ex::just(); }); + auto classified = ::exec::stopped_by( + main_sender{nullptr}, + [](auto&) noexcept { return ex::just(); }, + [](ex::set_value_t, int const&) noexcept { return false; }); + auto throwing_storage = ::exec::stopped_by(throwing_value_sender{}, + [](auto&) noexcept { return ex::just(); }); + + STATIC_REQUIRE(ex::sender_in); + STATIC_REQUIRE_FALSE(ex::dependent_sender); + STATIC_REQUIRE(set_equivalent, + ex::completion_signatures>); + STATIC_REQUIRE( + set_equivalent< + ex::completion_signatures_of_t, + ex::completion_signatures>); + STATIC_REQUIRE( + set_equivalent< + ex::completion_signatures_of_t, + ex::completion_signatures>); + } + + enum class result_kind + { + none, + value, + stopped + }; + + struct result_receiver + { + using receiver_concept = ex::receiver_tag; + + void set_value(int value) && noexcept + { + *result = result_kind::value; + *result_value = value; + } + + void set_error(std::exception_ptr) && noexcept + { + FAIL_CHECK("Unexpected error completion"); + } + + void set_stopped() && noexcept + { + *result = result_kind::stopped; + } + + auto get_env() const noexcept -> stop_env + { + return {ex::get_stop_token, source->get_token()}; + } + + result_kind* result; + int* result_value; + ex::inplace_stop_source* source; + }; + + TEST_CASE("stopped_by is transparent in an unstoppable environment", "[stopped_by]") + { + auto sender = ex::just(42) | ::exec::stopped_by(invalid_stop_factory{}); + + STATIC_REQUIRE(set_equivalent>, + ex::completion_signatures>); + } + + TEST_CASE("stopped_by forwards an ordinary completion", "[stopped_by]") + { + main_control main; + stop_control stop; + ex::inplace_stop_source source; + result_kind result = result_kind::none; + int value{}; + + auto sender = ::exec::stopped_by(main_sender{&main}, + [&](auto&) noexcept { return stop_sender{&stop}; }); + STATIC_REQUIRE(set_equivalent, + ex::completion_signatures>); + + auto operation = ex::connect(sender, result_receiver{&result, &value, &source}); + ex::start(operation); + CHECK(main.started); + CHECK_FALSE(stop.started); + + main.complete(42); + CHECK(result == result_kind::value); + CHECK(value == 42); + } + + TEST_CASE("stopped_by coordinates a child that completes inline during start", "[stopped_by]") + { + ex::inplace_stop_source source; + result_kind result = result_kind::none; + int value{}; + bool stop_started{}; + + auto sender = ::exec::stopped_by(ex::just(43), + [&](auto&) noexcept + { + stop_started = true; + return ex::just(); + }); + + auto operation = ex::connect(sender, result_receiver{&result, &value, &source}); + ex::start(operation); + + CHECK_FALSE(stop_started); + CHECK(result == result_kind::value); + CHECK(value == 43); + } + + TEST_CASE("stopped_by waits for both operations after stop is requested", "[stopped_by]") + { + main_control main; + stop_control stop; + ex::inplace_stop_source source; + result_kind result = result_kind::none; + int value{}; + void* target{}; + + auto sender = main_sender{&main} + | ::exec::stopped_by( + [&](auto& operation) noexcept + { + target = &operation; + return stop_sender{&stop}; + }); + + auto operation = ex::connect(sender, result_receiver{&result, &value, &source}); + ex::start(operation); + + source.request_stop(); + CHECK(stop.started); + CHECK(target == main.operation); + + main.complete(17); + CHECK(result == result_kind::none); + stop.complete(); + CHECK(result == result_kind::value); + CHECK(value == 17); + } + + TEST_CASE("stopped_by completes when the main operation is last", "[stopped_by]") + { + main_control main; + stop_control stop; + ex::inplace_stop_source source; + result_kind result = result_kind::none; + int value{}; + + auto sender = ::exec::stopped_by(main_sender{&main}, + [&](auto&) noexcept { return stop_sender{&stop}; }); + + auto operation = ex::connect(sender, result_receiver{&result, &value, &source}); + ex::start(operation); + + source.request_stop(); + CHECK_FALSE(stop.destroyed); + stop.complete(); + CHECK(result == result_kind::none); + CHECK(stop.destroyed); + main.complete(29); + CHECK(result == result_kind::value); + CHECK(value == 29); + } + + TEST_CASE("stopped_by supports an inline stop operation", "[stopped_by]") + { + main_control main; + ex::inplace_stop_source source; + result_kind result = result_kind::none; + int value{}; + bool stop_started{}; + + auto sender = ::exec::stopped_by(main_sender{&main}, + [&](auto&) noexcept + { + stop_started = true; + return ex::just(); + }); + + auto operation = ex::connect(sender, result_receiver{&result, &value, &source}); + ex::start(operation); + + source.request_stop(); + CHECK(stop_started); + CHECK(result == result_kind::none); + main.complete(31); + CHECK(result == result_kind::value); + CHECK(value == 31); + } + + TEST_CASE("stopped_by starts the main operation before servicing a pending stop request", + "[stopped_by]") + { + main_control main; + stop_control stop; + ex::inplace_stop_source source; + result_kind result = result_kind::none; + int value{}; + bool main_was_started{}; + source.request_stop(); + + auto sender = ::exec::stopped_by(main_sender{&main}, + [&](auto&) noexcept + { + main_was_started = main.started; + return stop_sender{&stop}; + }); + + auto operation = ex::connect(sender, result_receiver{&result, &value, &source}); + ex::start(operation); + + CHECK(main_was_started); + CHECK(stop.started); + main.complete(3); + stop.complete(); + CHECK(result == result_kind::value); + } + + TEST_CASE("stopped_by invokes its classifier for every completion", "[stopped_by]") + { + main_control main; + ex::inplace_stop_source source; + result_kind result = result_kind::none; + int value{}; + int classifications{}; + + auto sender = ::exec::stopped_by( + main_sender{&main}, + [](auto&) noexcept { return ex::just(); }, + [&](ex::set_value_t, int const & completed_value) noexcept + { + ++classifications; + return completed_value == 42; + }); + STATIC_REQUIRE( + set_equivalent, + ex::completion_signatures>); + + auto operation = ex::connect(sender, result_receiver{&result, &value, &source}); + ex::start(operation); + main.complete(42); + + CHECK(classifications == 1); + CHECK(result == result_kind::stopped); + } +} // namespace