From 72fd61ebcfcebecc385c3158987647693c6c2b7a Mon Sep 17 00:00:00 2001 From: Robert Leahy Date: Sun, 18 Oct 2026 20:01:53 -0400 Subject: [PATCH] exec::started_by Implements started_by from P4394R0. --- include/exec/started_by.hpp | 340 ++++++++++++++++++++++++++++++++++ test/exec/CMakeLists.txt | 1 + test/exec/test_started_by.cpp | 295 +++++++++++++++++++++++++++++ 3 files changed, 636 insertions(+) create mode 100644 include/exec/started_by.hpp create mode 100644 test/exec/test_started_by.cpp diff --git a/include/exec/started_by.hpp b/include/exec/started_by.hpp new file mode 100644 index 000000000..99865c0ba --- /dev/null +++ b/include/exec/started_by.hpp @@ -0,0 +1,340 @@ +/* + * 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/execution.hpp" + +#include +#include +#include +#include +#include +#include + +namespace experimental::execution +{ + namespace detail::started_by + { + struct default_base + {}; + + template + struct BASE_MUST_BE_CONSTRUCTIBLE_FROM + {}; + + template + struct INITIATION_MUST_BE_INVOCABLE_WITH_OPERATION_STATE + {}; + + template + struct state_base; + + template + struct state_base<_Derived, _Base, ::STDEXEC::completion_signatures<>> : _Base + { + state_base() = default; + + using _Base::_Base; + + template + requires(sizeof...(_Args) != 0) && (!std::is_abstract_v<_Base>) + && std::is_constructible_v<_Base, _Args...> + constexpr explicit state_base(_Args&&... __args) + noexcept(std::is_nothrow_constructible_v<_Base, _Args...>) + : _Base(static_cast<_Args&&>(__args)...) + {} + }; + + template + struct state_base<_Derived, _Base, ::STDEXEC::completion_signatures<_Tag(_Args...)>> : _Base + { + state_base() = default; + + using _Base::_Base; + + template + requires(sizeof...(_BaseArgs) != 0) && (!std::is_abstract_v<_Base>) + && std::is_constructible_v<_Base, _BaseArgs...> + constexpr explicit state_base(_BaseArgs&&... __args) + noexcept(std::is_nothrow_constructible_v<_Base, _BaseArgs...>) + : _Base(static_cast<_BaseArgs&&>(__args)...) + {} + + constexpr void complete(_Tag __tag, _Args&&... __args) noexcept + { + static_cast<_Derived&>(*this).__complete(__tag, static_cast<_Args&&>(__args)...); + } + }; + + template + struct state_base< + _Derived, + _Base, + ::STDEXEC::completion_signatures<_Tag(_Args...), _NextSignature, _RemainingSignatures...>> + : state_base<_Derived, + _Base, + ::STDEXEC::completion_signatures<_NextSignature, _RemainingSignatures...>> + { + using __base_t = + state_base<_Derived, + _Base, + ::STDEXEC::completion_signatures<_NextSignature, _RemainingSignatures...>>; + + state_base() = default; + + using __base_t::__base_t; + + template + requires(sizeof...(_BaseArgs) != 0) && (!std::is_abstract_v<_Base>) + && std::is_constructible_v<_Base, _BaseArgs...> + constexpr explicit state_base(_BaseArgs&&... __args) + noexcept(std::is_nothrow_constructible_v<_Base, _BaseArgs...>) + : __base_t(static_cast<_BaseArgs&&>(__args)...) + {} + + using __base_t::complete; + + constexpr void complete(_Tag __tag, _Args&&... __args) noexcept + { + static_cast<_Derived&>(*this).__complete(__tag, static_cast<_Args&&>(__args)...); + } + }; + + template + struct base_probe : state_base, _Base, _Signatures> + { + using __base_t = state_base; + using __base_t::__base_t; + + template + constexpr void __complete(_Tag, _Args&&...) noexcept + {} + }; + + template + class state + : public state_base, _Base, _Signatures> + { + using __base_t = + state_base, _Base, _Signatures>; + + public: + using operation_state_concept = ::STDEXEC::operation_state_tag; + + template + constexpr explicit state(_Initiation __init, _Receiver __receiver, _Args&&... __args) + noexcept(std::is_nothrow_move_constructible_v<_Initiation> + && std::is_nothrow_constructible_v<__base_t, _Args...>) + : __base_t(static_cast<_Args&&>(__args)...) + , __receiver(static_cast<_Receiver&&>(__receiver)) + , __init(static_cast<_Initiation&&>(__init)) + {} + + STDEXEC_IMMOVABLE(state); + + constexpr void start() & noexcept + { + try + { + std::invoke(static_cast<_Initiation&&>(__init), *this); + } + catch (...) + { + if constexpr (!std::is_nothrow_invocable_v<_Initiation, state&>) + { + ::STDEXEC::set_error(static_cast<_Receiver&&>(__receiver), std::current_exception()); + } + } + } + + template + constexpr void __complete(_Tag __tag, _Args&&... __args) noexcept + { + __tag(static_cast<_Receiver&&>(__receiver), static_cast<_Args&&>(__args)...); + } + + private: + STDEXEC_ATTRIBUTE(no_unique_address) + _Receiver __receiver; + STDEXEC_ATTRIBUTE(no_unique_address) + _Initiation __init; + }; + + template + using data_t = std::remove_cvref_t<::STDEXEC::__data_of<_Sender>>; + + template + using initiation_t = std::tuple_element_t<0, data_t<_Sender>>; + + template + using data_element_t = decltype(std::get<_Index>( + std::declval<::STDEXEC::__data_of<_Sender>>())); + + template + struct impl : ::STDEXEC::__sexpr_defaults + { + template + using state_t = state<_Signatures, _Base, initiation_t<_Sender>, _Receiver>; + + template + static consteval bool __base_constructible(std::index_sequence<_Indices...>) + { + return std::is_constructible_v, + data_element_t<_Sender, _Indices + 1>...>; + } + + template + static constexpr bool __base_constructible_v = __base_constructible<_Sender>( + std::make_index_sequence> - 1>{}); + + template + static constexpr bool __connectable_v = [] + { + if constexpr (!__base_constructible_v<_Sender>) + { + return false; + } + else + { + using __state_t = state_t<_Sender, _Receiver>; + using __archetype_state_t = + state_t<_Sender, ::STDEXEC::__receiver_archetype<::STDEXEC::env_of_t<_Receiver>>>; + using __init_t = initiation_t<_Sender>; + + return std::is_invocable_v<__init_t, __state_t&> + && std::is_invocable_v<__init_t, __archetype_state_t&> + && (std::is_nothrow_invocable_v<__init_t, __state_t&> + == std::is_nothrow_invocable_v<__init_t, __archetype_state_t&>); + } + }(); + + template + static constexpr bool __nothrow_connect(std::index_sequence<_Indices...>) + { + return std::is_nothrow_constructible_v, + data_element_t<_Sender, 0>, + _Receiver, + data_element_t<_Sender, _Indices + 1>...>; + } + + template + static constexpr bool __nothrow_connect_v = __nothrow_connect<_Sender, _Receiver>( + std::make_index_sequence> - 1>{}); + + template + static constexpr auto + __connect_impl(_Sender&& __sender, _Receiver __receiver, std::index_sequence<_Indices...>) + noexcept(__nothrow_connect_v<_Sender, _Receiver>) -> state_t<_Sender, _Receiver> + { + auto& [__tag, __data] = __sender; + return state_t<_Sender, _Receiver>(::STDEXEC::__forward_like<_Sender>(std::get<0>(__data)), + static_cast<_Receiver&&>(__receiver), + ::STDEXEC::__forward_like<_Sender>( + std::get<_Indices + 1>(__data))...); + } + + template + static consteval auto __get_completion_signatures() + { + if constexpr (sizeof...(_Env) == 0) + { + return ::STDEXEC::transform_completion_signatures< + _Signatures, + ::STDEXEC::completion_signatures<::STDEXEC::set_error_t(std::exception_ptr)>>{}; + } + else if constexpr (!__base_constructible_v<_Sender>) + { + return [](std::index_sequence<_Indices...>) + { + return ::STDEXEC::__throw_compile_time_error< + BASE_MUST_BE_CONSTRUCTIBLE_FROM<_Base, data_element_t<_Sender, _Indices + 1>...>, + ::STDEXEC::_WITH_SENDER_<_Sender>>(); + }(std::make_index_sequence> - 1>{}); + } + else + { + using __archetype_state_t = state_t<_Sender, ::STDEXEC::__receiver_archetype<_Env...>>; + using __init_t = initiation_t<_Sender>; + + if constexpr (!std::is_invocable_v<__init_t, __archetype_state_t&>) + { + return ::STDEXEC::__throw_compile_time_error< + INITIATION_MUST_BE_INVOCABLE_WITH_OPERATION_STATE<__init_t, __archetype_state_t>, + ::STDEXEC::_WITH_SENDER_<_Sender>>(); + } + else + { + using __additional_signatures_t = std::conditional_t< + std::is_nothrow_invocable_v<__init_t, __archetype_state_t&>, + ::STDEXEC::completion_signatures<>, + ::STDEXEC::completion_signatures<::STDEXEC::set_error_t(std::exception_ptr)>>; + return ::STDEXEC::transform_completion_signatures<_Signatures, + __additional_signatures_t>{}; + } + } + } + + static constexpr auto __connect = + [](_Sender&& __sender, _Receiver __receiver) noexcept( + __nothrow_connect_v<_Sender, _Receiver>) + requires __connectable_v<_Sender, _Receiver> + { + return __connect_impl(static_cast<_Sender&&>(__sender), + static_cast<_Receiver&&>(__receiver), + std::make_index_sequence> - 1>{}); + }; + }; + } // namespace detail::started_by + + template <::STDEXEC::__valid_completion_signatures _Signatures, + class _Base = detail::started_by::default_base> + requires std::is_class_v<_Base> && (!std::is_final_v<_Base>) + struct started_by_t + { + template <::STDEXEC::__movable_value _Initiation, ::STDEXEC::__movable_value... _Args> + [[nodiscard]] + constexpr auto operator()(_Initiation&& __init, _Args&&... __args) const + noexcept(::STDEXEC::__nothrow_decay_copyable<_Initiation, _Args...>) + -> ::STDEXEC::__well_formed_sender auto + { + return ::STDEXEC::__sexpr{started_by_t{}, + std::tuple(static_cast<_Initiation&&>(__init), + static_cast<_Args&&>(__args)...)}; + } + }; + + template <::STDEXEC::__valid_completion_signatures _Signatures, + class _Base = detail::started_by::default_base> + requires std::is_class_v<_Base> && (!std::is_final_v<_Base>) + inline constexpr started_by_t<_Signatures, _Base> started_by{}; +} // namespace experimental::execution + +namespace exec = experimental::execution; + +namespace STDEXEC +{ + template + struct __sexpr_impl<::exec::started_by_t<_Signatures, _Base>> + : ::exec::detail::started_by::impl<_Signatures, _Base> + {}; +} // namespace STDEXEC diff --git a/test/exec/CMakeLists.txt b/test/exec/CMakeLists.txt index 1d967bd96..3ebbee7ba 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_started_by.cpp $<$:test_io_uring_context.cpp> $<$:test_windows_thread_pool_context.cpp> test_trampoline_scheduler.cpp diff --git a/test/exec/test_started_by.cpp b/test/exec/test_started_by.cpp new file mode 100644 index 000000000..6109e4b65 --- /dev/null +++ b/test/exec/test_started_by.cpp @@ -0,0 +1,295 @@ +/* + * 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 +#include + +namespace +{ + namespace ex = ::STDEXEC; + + using value_signatures = ex::completion_signatures; + + struct value_receiver + { + using receiver_concept = ex::receiver_tag; + + void set_value(int value) && noexcept + { + *result = value; + } + + void set_error(std::exception_ptr) && noexcept + { + FAIL_CHECK("Unexpected error completion"); + } + + constexpr auto get_env() const noexcept -> ex::env<> + { + return {}; + } + + int* result; + }; + + struct completion_control + { + void* operation{}; + void (*complete)(void*, int) noexcept {}; + + void set_value(int value) const noexcept + { + complete(operation, value); + } + }; + + struct continuation + { + explicit continuation(int value) noexcept + : value(value) + {} + + continuation(continuation&&) = delete; + + virtual void complete(ex::set_value_t, int&&) noexcept = 0; + + int value; + }; + + struct multi_continuation + { + virtual void complete(ex::set_value_t, int&&) noexcept = 0; + virtual void complete(ex::set_error_t, std::exception_ptr&&) noexcept = 0; + }; + + struct throwing_base + { + explicit throwing_base(int) {} + }; + + struct incomplete_base + { + explicit incomplete_base(int) noexcept {} + + virtual void missing() noexcept = 0; + }; + + struct aggregate_base + { + int value; + }; + + struct large_receiver + { + using receiver_concept = ex::receiver_tag; + + void set_value() && noexcept {} + + void set_error(std::exception_ptr) && noexcept {} + + constexpr auto get_env() const noexcept -> ex::env<> + { + return {}; + } + + char padding[256]{}; + }; + + struct receiver_dependent_noexcept + { + template + void operator()(_Operation&) noexcept(sizeof(_Operation) > 128) + {} + }; + + TEST_CASE("started_by advertises its explicit completion signatures", "[started_by]") + { + auto sender = ::exec::started_by([](auto&) noexcept {}); + + STATIC_REQUIRE(ex::sender_in); + STATIC_REQUIRE_FALSE(ex::dependent_sender); + STATIC_REQUIRE(ex::sender_in>); + STATIC_REQUIRE( + set_equivalent< + ex::completion_signatures_of_t, + ex::completion_signatures>); + STATIC_REQUIRE(set_equivalent>, + value_signatures>); + } + + TEST_CASE("started_by adds exception_ptr when initiation can throw", "[started_by]") + { + auto sender = ::exec::started_by([](auto&) {}); + + STATIC_REQUIRE(ex::sender_in>); + STATIC_REQUIRE( + set_equivalent< + ex::completion_signatures_of_t>, + ex::completion_signatures>); + } + + TEST_CASE("started_by initiates on start and completes asynchronously", "[started_by]") + { + completion_control control; + int result{}; + + auto sender = ::exec::started_by( + [&](_Operation& operation) noexcept + { + control.operation = &operation; + control.complete = [](void* pointer, int value) noexcept + { + static_cast<_Operation*>(pointer)->complete(ex::set_value, std::move(value)); + }; + }); + + auto operation = ex::connect(sender, value_receiver{&result}); + CHECK(control.operation == nullptr); + ex::start(operation); + REQUIRE(control.operation != nullptr); + CHECK(result == 0); + control.set_value(42); + CHECK(result == 42); + } + + TEST_CASE("started_by converts a thrown initiation into set_error", "[started_by]") + { + bool caught{}; + struct receiver + { + using receiver_concept = ex::receiver_tag; + + void set_value(int) && noexcept + { + FAIL_CHECK("Unexpected value completion"); + } + + void set_error(std::exception_ptr error) && noexcept + { + try + { + std::rethrow_exception(std::move(error)); + } + catch (int value) + { + CHECK(value == 17); + *caught = true; + } + catch (...) + { + FAIL_CHECK("Unexpected exception type"); + } + } + + constexpr auto get_env() const noexcept -> ex::env<> + { + return {}; + } + + bool* caught; + }; + + auto sender = ::exec::started_by([](auto&) { throw 17; }); + auto operation = ex::connect(sender, receiver{&caught}); + ex::start(operation); + CHECK(caught); + } + + TEST_CASE("started_by constructs and publicly exposes its operation base", "[started_by]") + { + continuation* erased{}; + int observed{}; + int result{}; + + auto sender = ::exec::started_by( + [&](auto& operation) noexcept + { + observed = operation.value; + erased = &operation; + }, + 23); + + auto operation = ex::connect(sender, value_receiver{&result}); + STATIC_REQUIRE(std::derived_from); + STATIC_REQUIRE(ex::__nothrow_connectable); + CHECK(operation.value == 23); + ex::start(operation); + CHECK(observed == 23); + REQUIRE(erased != nullptr); + erased->complete(ex::set_value, 29); + CHECK(result == 29); + } + + TEST_CASE("started_by checks base construction in the sender cvref", "[started_by]") + { + auto valid = ::exec::started_by([](auto&) noexcept {}, 1); + auto invalid = ::exec::started_by([](auto&) noexcept {}, + nullptr); + + STATIC_REQUIRE(ex::sender_in>); + STATIC_REQUIRE_FALSE(ex::sender_in>); + STATIC_REQUIRE_FALSE(ex::__nothrow_connectable); + } + + TEST_CASE("started_by checks construction and concreteness of an abstract base", "[started_by]") + { + auto invalid_args = ::exec::started_by([](auto&) noexcept {}, + nullptr); + auto incomplete = ::exec::started_by([](auto&) noexcept {}, + 1); + + STATIC_REQUIRE_FALSE(ex::sender_in>); + STATIC_REQUIRE_FALSE(ex::sender_in>); + } + + TEST_CASE("started_by constructs an aggregate operation base", "[started_by]") + { + auto sender = ::exec::started_by([](auto&) noexcept {}, 31); + auto operation = ex::connect(sender, value_receiver{}); + + CHECK(operation.value == 31); + } + + TEST_CASE("started_by implements every abstract completion overload", "[started_by]") + { + using signatures = + ex::completion_signatures; + + multi_continuation* erased{}; + int result{}; + auto sender = ::exec::started_by([&](auto& operation) noexcept + { erased = &operation; }); + auto operation = ex::connect(sender, value_receiver{&result}); + + ex::start(operation); + REQUIRE(erased != nullptr); + erased->complete(ex::set_value, 37); + CHECK(result == 37); + } + + TEST_CASE("started_by checks initiation against an archetype receiver", "[started_by]") + { + auto invalid = ::exec::started_by([](int) noexcept {}); + STATIC_REQUIRE(ex::sender_in); + STATIC_REQUIRE_FALSE(ex::sender_in>); + } + + TEST_CASE("started_by rejects receiver-dependent initiation throwingness", "[started_by]") + { + using signatures = ex::completion_signatures; + auto sender = ::exec::started_by(receiver_dependent_noexcept{}); + + STATIC_REQUIRE(ex::sender_in>); + STATIC_REQUIRE_FALSE(ex::sender_to); + } +} // namespace