From af6ba40d8ffe5541faf44e2de89c35d1278a0287 Mon Sep 17 00:00:00 2001 From: Eric Niebler Date: Mon, 20 Jul 2026 04:57:03 +0000 Subject: [PATCH 1/4] computing completion signatures with consteval instead of TMP --- include/exec/async_scope.hpp | 42 +- include/exec/at_coroutine_exit.hpp | 19 +- include/exec/env.hpp | 57 +- include/exec/fork_join.hpp | 8 +- include/exec/libdispatch_queue.hpp | 57 +- include/exec/linux/io_uring_context.hpp | 3 +- include/exec/static_thread_pool.hpp | 621 ++++++++---------- include/exec/variant_sender.hpp | 11 +- include/stdexec/__detail/__associate.hpp | 10 +- include/stdexec/__detail/__bulk.hpp | 140 ++-- include/stdexec/__detail/__continues_on.hpp | 33 +- include/stdexec/__detail/__diagnostics.hpp | 2 +- include/stdexec/__detail/__finally.hpp | 6 +- include/stdexec/__detail/__into_variant.hpp | 35 +- include/stdexec/__detail/__let.hpp | 2 +- .../stdexec/__detail/__parallel_scheduler.hpp | 13 +- include/stdexec/__detail/__sender_ref.hpp | 3 +- .../stdexec/__detail/__transform_sender.hpp | 27 +- include/stdexec/functional.hpp | 74 ++- 19 files changed, 563 insertions(+), 600 deletions(-) diff --git a/include/exec/async_scope.hpp b/include/exec/async_scope.hpp index 0ade14061..7635dbb0b 100644 --- a/include/exec/async_scope.hpp +++ b/include/exec/async_scope.hpp @@ -19,6 +19,7 @@ #include "../stdexec/__detail/__optional.hpp" #include "../stdexec/execution.hpp" #include "../stdexec/stop_token.hpp" +#include "completion_signatures.hpp" #include "env.hpp" #include "../stdexec/__detail/__atomic.hpp" @@ -38,6 +39,9 @@ namespace experimental::execution struct __impl; struct async_scope; + template + struct __future; + template concept __async_scope = requires(_A& __a) { { __a.nest(STDEXEC::just()) } -> sender_of; @@ -130,9 +134,9 @@ namespace experimental::execution template <__decays_to<__when_empty_sender> _Self, class... _Env> static consteval auto get_completion_signatures() - -> __completion_signatures_of_t<__copy_cvref_t<_Self, _Constrained>, __env_t<_Env>...> { - return {}; + return STDEXEC::get_completion_signatures<__copy_cvref_t<_Self, _Constrained>, + __env_t<_Env>...>(); } __impl const * __scope_; @@ -256,9 +260,9 @@ namespace experimental::execution template <__decays_to<__nest_sender> _Self, class... _Env> static consteval auto get_completion_signatures() - -> __completion_signatures_of_t<__copy_cvref_t<_Self, _Constrained>, __env_t<_Env>...> { - return {}; + return STDEXEC::get_completion_signatures<__copy_cvref_t<_Self, _Constrained>, + __env_t<_Env>...>(); } __impl const * __scope_; @@ -442,22 +446,11 @@ namespace experimental::execution static_cast<_Fn*>(nullptr))); #endif - template - using __decay_values_t = completion_signatures...)>; - - template - using __decay_error_t = completion_signatures)>; - template - using __future_completions_t = __transform_completion_signatures_of_t< - _Sender, - __env_t<_Env>, - completion_signatures, - __decay_values_t, - __decay_error_t>; + using __future_completions_t = STDEXEC::completion_signatures_of_t<__future<_Sender, _Env>>; template - using __completions_as_variant = __mapply< + using __completions_as_variant_t = __mapply< __mtransform<__q<__completion_as_tuple_t>, __mbind_front_q>, _Completions>; @@ -532,7 +525,7 @@ namespace experimental::execution std::mutex __mutex_; __future_step __step_ = __future_step::__created; std::unique_ptr<__future_state_base, __dynamic_delete<__future_state_base>> __no_future_; - __completions_as_variant<_Completions> __data_; + __completions_as_variant_t<_Completions> __data_; __intrusive_queue<&__subscription::__next_> __subscribers_; __env_t<_Env> __env_; }; @@ -652,9 +645,6 @@ namespace experimental::execution struct __future { private: - template - using __completions_t = __future_completions_t<__mfront<_Sender, _Self>, _Env>; - template using __future_opstate_t = __future_opstate<_Sender, _Env, _Receiver>; @@ -684,7 +674,6 @@ namespace experimental::execution } template <__decays_to<__future> _Self, receiver _Receiver> - requires receiver_of<_Receiver, __completions_t<_Self>> STDEXEC_EXPLICIT_THIS_BEGIN(auto connect)(this _Self&& __self, _Receiver __rcvr) -> __future_opstate_t<_Receiver> { @@ -694,9 +683,14 @@ namespace experimental::execution STDEXEC_EXPLICIT_THIS_END(connect) template <__decays_to<__future> _Self, class... _OtherEnv> - static consteval auto get_completion_signatures() -> __completions_t<_Self> + static consteval auto get_completion_signatures() { - return {}; + return exec::transform_completion_signatures( + STDEXEC::get_completion_signatures<_Sender, __env_t<_Env>>(), + exec::decay_arguments(), + exec::decay_arguments(), + {}, + completion_signatures()); } private: diff --git a/include/exec/at_coroutine_exit.hpp b/include/exec/at_coroutine_exit.hpp index 247b33b70..72ce9f189 100644 --- a/include/exec/at_coroutine_exit.hpp +++ b/include/exec/at_coroutine_exit.hpp @@ -21,6 +21,7 @@ #include "../stdexec/execution.hpp" #include "any_sender_of.hpp" +#include "completion_signatures.hpp" #include #include @@ -79,23 +80,23 @@ namespace experimental::execution { using sender_concept = STDEXEC::sender_tag; - template - using __completions_t = __mapply<__mremove>, - __completion_signatures_of_t<_Sender, _Env...>>; - template - requires sender_to<_Sender, __receiver<_Receiver>> - auto connect(_Receiver __rcvr) && noexcept // - -> connect_result_t<_Sender, __receiver<_Receiver>> + auto connect(_Receiver __rcvr) && // + noexcept(__nothrow_connectable<_Sender, __receiver<_Receiver>>) + -> connect_result_t<_Sender, __receiver<_Receiver>> { return STDEXEC::connect(static_cast<_Sender&&>(__sender_), __receiver<_Receiver>{static_cast<_Receiver&&>(__rcvr)}); } template <__same_as<__sender> _Self, class... _Env> - static consteval auto get_completion_signatures() -> __completions_t<_Env...> + static consteval auto get_completion_signatures() { - return {}; + return exec::transform_completion_signatures( + STDEXEC::get_completion_signatures<_Sender, _Env...>(), + {}, + {}, + exec::ignore_completion()); } auto get_env() const noexcept -> env_of_t<_Sender> diff --git a/include/exec/env.hpp b/include/exec/env.hpp index c7544a9f8..595f9ecda 100644 --- a/include/exec/env.hpp +++ b/include/exec/env.hpp @@ -122,10 +122,11 @@ namespace experimental::execution struct read_with_default_t; - template + template struct __opstate { constexpr explicit __opstate(_Default&& __default, _Receiver&& __rcvr) + noexcept(__nothrow_move_constructible<_Default>) : __default_(static_cast<_Default&&>(__default)) , __rcvr_(static_cast<_Receiver&&>(__rcvr)) {} @@ -136,10 +137,10 @@ namespace experimental::execution { STDEXEC_TRY { - if constexpr (__callable<_Tag, env_of_t<_Receiver>>) + if constexpr (__callable<_Query, env_of_t<_Receiver>>) { - auto const & __env = get_env(__rcvr_); - STDEXEC::set_value(std::move(__rcvr_), _Tag{}(__env)); + auto const & __env = STDEXEC::get_env(__rcvr_); + STDEXEC::set_value(std::move(__rcvr_), _Query()(__env)); } else { @@ -148,7 +149,10 @@ namespace experimental::execution } STDEXEC_CATCH_ALL { - STDEXEC::set_error(std::move(__rcvr_), std::current_exception()); + if constexpr (!__nothrow_callable<_Query, env_of_t<_Receiver>>) + { + STDEXEC::set_error(std::move(__rcvr_), std::current_exception()); + } } } @@ -157,37 +161,41 @@ namespace experimental::execution _Receiver __rcvr_; }; - template + template struct __sender { using sender_concept = STDEXEC::sender_tag; template - using __value_t = - __minvoke<__mwith_default<__mbind_back_q<__call_result_t, _Env>, _Default>, _Tag>; - template - using __default_t = __if_c<__callable<_Tag, _Env>, __ignore, _Default>; - - template - using __completions_t = - completion_signatures), set_error_t(std::exception_ptr)>; + using __default_t = __if_c<__callable<_Query, _Env>, __ignore, _Default>; template <__decays_to<__sender> _Self, class _Receiver> - requires receiver_of<_Receiver, __completions_t>> constexpr STDEXEC_EXPLICIT_THIS_BEGIN(auto connect)(this _Self&& __self, _Receiver __rcvr) noexcept(std::is_nothrow_move_constructible_v<_Receiver>) - -> __opstate<_Tag, __default_t>, _Receiver> + -> __opstate<_Query, __default_t>, _Receiver> { - using __opstate_t = __opstate<_Tag, __default_t>, _Receiver>; + using __opstate_t = __opstate<_Query, __default_t>, _Receiver>; return __opstate_t{static_cast<_Self&&>(__self).__default_, static_cast<_Receiver&&>(__rcvr)}; } STDEXEC_EXPLICIT_THIS_END(connect) template - static consteval auto get_completion_signatures() -> __completions_t<_Env> + static consteval auto get_completion_signatures() noexcept { - return {}; + if constexpr (__nothrow_callable<_Query, _Env>) + { + return STDEXEC::completion_signatures)>(); + } + else if constexpr (__callable<_Query, _Env>) + { + return STDEXEC::completion_signatures), + set_error_t(std::exception_ptr)>(); + } + else + { + return STDEXEC::completion_signatures(); + } } STDEXEC_ATTRIBUTE(no_unique_address) @@ -196,11 +204,11 @@ namespace experimental::execution struct __read_with_default_t { - template - constexpr auto - operator()(_Tag, _Default&& __default) const -> __sender<_Tag, __decay_t<_Default>> + template + constexpr auto operator()(_Query, _Default&& __default) const + noexcept(__nothrow_decay_copyable<_Default>) { - return {static_cast<_Default&&>(__default)}; + return __sender<_Query, __decay_t<_Default>>{static_cast<_Default&&>(__default)}; } }; } // namespace __read_with_default @@ -229,9 +237,8 @@ namespace experimental::execution template <__decays_to<__sender> _Self, class... _Env> static consteval auto get_completion_signatures() - -> __completion_signatures_of_t<__copy_cvref_t<_Self, _Sender>, _Env...> { - return {}; + return STDEXEC::get_completion_signatures<__copy_cvref_t<_Self, _Sender>, _Env...>(); } template <__decays_to<__sender> _Self, class _Receiver> diff --git a/include/exec/fork_join.hpp b/include/exec/fork_join.hpp index 33fa9b495..d4e37afae 100644 --- a/include/exec/fork_join.hpp +++ b/include/exec/fork_join.hpp @@ -25,11 +25,7 @@ namespace experimental::execution { struct fork_join_t; - struct PREDECESSOR_RESULTS_ARE_NOT_DECAY_COPYABLE - {}; - - struct INVALID_ARGUMENTS_TO_FORK_JOIN - {}; + struct INVALID_ARGUMENTS_TO_FORK_JOIN; namespace _fork_join { @@ -247,7 +243,7 @@ namespace experimental::execution else if constexpr (!__decay_copyable_results_t::value) { return STDEXEC::__throw_compile_time_error< // - _WHAT_(PREDECESSOR_RESULTS_ARE_NOT_DECAY_COPYABLE), + _WHAT_(_PREDECESSOR_RESULTS_ARE_NOT_DECAY_COPYABLE_), _IN_ALGORITHM_(exec::fork_join_t)>(); } else diff --git a/include/exec/libdispatch_queue.hpp b/include/exec/libdispatch_queue.hpp index 55d3d582f..32c92cdac 100644 --- a/include/exec/libdispatch_queue.hpp +++ b/include/exec/libdispatch_queue.hpp @@ -28,7 +28,10 @@ # endif # include "../stdexec/execution.hpp" + +# include "completion_signatures.hpp" # include "sender_for.hpp" + # include namespace experimental::execution @@ -253,30 +256,11 @@ namespace experimental::execution { using sender_concept = STDEXEC::sender_tag; - template - using with_error_invoke_t = STDEXEC::__if_c< - STDEXEC::__value_types_t, - STDEXEC::__mbind_front_q, - STDEXEC::__q>::value, - STDEXEC::completion_signatures<>, - STDEXEC::__eptr_completion_t>; - - template - using set_value_t = - STDEXEC::completion_signatures...)>; - - template - using _completions_t = STDEXEC::__transform_completion_signatures_t< - STDEXEC::__completion_signatures_of_t, Env...>, - with_error_invoke_t, Env...>, - set_value_t>; - template using bulk_op_state_t = bulk_op_state, Receiver, Shape, Fun>; template Self, STDEXEC::receiver Receiver> - requires STDEXEC::receiver_of>> STDEXEC_EXPLICIT_THIS_BEGIN(auto connect)(this Self &&self, Receiver rcvr) noexcept(STDEXEC::__nothrow_constructible_from, libdispatch_queue &, @@ -294,11 +278,42 @@ namespace experimental::execution STDEXEC_EXPLICIT_THIS_END(connect) template Self, class... Env> - static consteval auto get_completion_signatures() -> _completions_t + static consteval auto get_completion_signatures() { - return {}; + return exec::transform_completion_signatures( + STDEXEC::get_completion_signatures, Env...>(), + []() + { + if constexpr (!__decay_copyable) + { + return exec::throw_compile_time_error< + _WHAT_(_PREDECESSOR_RESULTS_ARE_NOT_DECAY_COPYABLE_), + _WHERE_(_IN_ALGORITHM_, bulk_t), + _WITH_PRETTY_SENDER_<__copy_cvref_t>, + _WITH_ENVIRONMENT_(Env...)>(); + } + else if constexpr (!__callable &...>) + { + return exec::throw_compile_time_error< + _WHAT_(_FUNCTION_IS_NOT_CALLABLE_WITH_THE_GIVEN_ARGUMENTS_), + _WHERE_(_IN_ALGORITHM_, bulk_t), + _WITH_FUNCTION_(Fun &), + _WITH_ARGUMENTS_(__decay_t & ...)>(); + } + else if constexpr (__nothrow_callable &...> + && __nothrow_decay_copyable) + { + return completion_signatures...)>(); + } + else + { + return completion_signatures...), + set_error_t(std::exception_ptr)>(); + } + }); } + [[nodiscard]] auto get_env() const noexcept -> STDEXEC::env_of_t { return STDEXEC::get_env(sndr_); diff --git a/include/exec/linux/io_uring_context.hpp b/include/exec/linux/io_uring_context.hpp index a030081f9..011cf1fb6 100644 --- a/include/exec/linux/io_uring_context.hpp +++ b/include/exec/linux/io_uring_context.hpp @@ -1298,8 +1298,7 @@ namespace experimental::execution } template - [[nodiscard]] - static consteval auto get_completion_signatures() noexcept -> __completions_t + static consteval auto get_completion_signatures() { return __completions_t{}; } diff --git a/include/exec/static_thread_pool.hpp b/include/exec/static_thread_pool.hpp index ae664af8c..739b461a6 100644 --- a/include/exec/static_thread_pool.hpp +++ b/include/exec/static_thread_pool.hpp @@ -1,7 +1,7 @@ /* * Copyright (c) 2021-2022 Facebook, Inc. and its affiliates. - * Copyright (c) 2021-2024 NVIDIA Corporation * Copyright (c) 2023 Maikel Nadolski + * Copyright (c) 2026 NVIDIA Corporation * * Licensed under the Apache License Version 2.0 with LLVM Exceptions * (the "License"); you may not use this file except in compliance with @@ -31,7 +31,6 @@ #include "../stdexec/__detail/__meta.hpp" #include "../stdexec/__detail/__optional.hpp" #include "../stdexec/__detail/__receivers.hpp" -#include "../stdexec/__detail/__transform_completion_signatures.hpp" #include "../stdexec/__detail/__tuple.hpp" #include "../stdexec/__detail/__type_traits.hpp" #include "../stdexec/__detail/__variant.hpp" @@ -41,6 +40,7 @@ #include "detail/numa.hpp" #include "detail/xorshift.hpp" +#include "completion_signatures.hpp" #include "sender_for.hpp" #include "sequence/iterate.hpp" #include "sequence_senders.hpp" @@ -112,8 +112,8 @@ namespace experimental::execution struct task_base { - task_base* next = nullptr; void (*execute_)(task_base*, std::uint32_t tid) noexcept = nullptr; + task_base* next_ = nullptr; }; struct remote_queue @@ -127,9 +127,9 @@ namespace experimental::execution , queues_(nthreads) {} - remote_queue* next_{}; - std::vector<__atomic_intrusive_queue<&task_base::next>> queues_{}; - std::thread::id id_{std::this_thread::get_id()}; + remote_queue* next_{}; + std::vector<__atomic_intrusive_queue<&task_base::next_>> queues_{}; + std::thread::id id_{std::this_thread::get_id()}; // This marks whether the submitter is a thread in the pool or not. std::size_t index_{(std::numeric_limits::max)()}; }; @@ -160,10 +160,10 @@ namespace experimental::execution } } - auto pop_all_reversed(std::size_t tid) noexcept -> __intrusive_queue<&task_base::next> + auto pop_all_reversed(std::size_t tid) noexcept -> __intrusive_queue<&task_base::next_> { - remote_queue* head = head_.load(__std::memory_order_acquire); - __intrusive_queue<&task_base::next> tasks{}; + remote_queue* head = head_.load(__std::memory_order_acquire); + __intrusive_queue<&task_base::next_> tasks{}; while (head != nullptr) { tasks.append(head->queues_[tid].pop_all_reversed()); @@ -202,34 +202,10 @@ namespace experimental::execution template struct _bulk_sender; - template - struct _is_nothrow_bulk_fn - { - template - requires __callable&...> - using __f = __mbool< - // If function invocation doesn't throw ... - __nothrow_callable&...> && - // ... and decay-copying the arguments doesn't throw ... - __nothrow_decay_copyable - // ... then there is no need to advertise completion with `exception_ptr` - >; - }; - - template + template struct _bulk_shared_state; - template + template struct _bulk_receiver; template @@ -249,7 +225,7 @@ namespace experimental::execution if constexpr (__same_as) { - // Turn a bulk_unchunked into a bulk_chunked operation + // Turn a bulk_unchunked into a bulk_chunked opstate using fun_t = STDEXEC::__bulk::__as_bulk_chunked_fn; using sender_t = _bulk_sender>; return sender_t{pool_, static_cast(sndr), shape, fun_t(std::move(fun))}; @@ -339,9 +315,9 @@ namespace experimental::execution public: _static_thread_pool(); - _static_thread_pool(std::uint32_t threadCount, - bwos_params params = {}, - numa_policy numa = get_numa_policy()); + explicit _static_thread_pool(std::uint32_t threadCount, + bwos_params params = {}, + numa_policy numa = get_numa_policy()); ~_static_thread_pool(); struct scheduler @@ -352,11 +328,8 @@ namespace experimental::execution class _sender { - struct env + struct attrs { - _static_thread_pool& pool_; - remote_queue* queue_; - template auto query(get_completion_scheduler_t, __ignore = {}) const noexcept -> _static_thread_pool::scheduler @@ -369,12 +342,13 @@ namespace experimental::execution { return {}; } + + _static_thread_pool& pool_; + remote_queue* queue_; }; public: using sender_concept = sender_tag; - template - using _opstate_t = _opstate; template static consteval auto get_completion_signatures() noexcept @@ -390,19 +364,15 @@ namespace experimental::execution } [[nodiscard]] - auto get_env() const noexcept -> env + auto get_env() const noexcept -> attrs { - return env{.pool_ = pool_, .queue_ = queue_}; + return attrs{.pool_ = pool_, .queue_ = queue_}; } template - auto connect(Receiver rcvr) const -> _opstate_t + auto connect(Receiver rcvr) const -> _opstate { - return _opstate_t{pool_, - queue_, - static_cast(rcvr), - threadIndex_, - constraints_}; + return _opstate{pool_, queue_, std::move(rcvr), threadIndex_, constraints_}; } private: @@ -543,10 +513,10 @@ namespace experimental::execution //! This is O(n_threads) on the calling thread. template Task> void bulk_enqueue(std::span tasks) noexcept; - void bulk_enqueue(remote_queue& queue, - __intrusive_queue<&task_base::next> tasks, - std::size_t tasks_size, - nodemask const & constraints = nodemask::any()) noexcept; + void bulk_enqueue(remote_queue& queue, + __intrusive_queue<&task_base::next_> tasks, + std::size_t tasks_size, + nodemask const & constraints = nodemask::any()) noexcept; private: class workstealing_victim @@ -621,7 +591,7 @@ namespace experimental::execution auto pop() -> pop_result; void push_local(task_base* task); - void push_local(__intrusive_queue<&task_base::next>&& tasks); + void push_local(__intrusive_queue<&task_base::next_>&& tasks); auto notify() -> bool; void request_stop(); @@ -682,7 +652,7 @@ namespace experimental::execution void clear_sleeping(); bwos::lifo_queue> local_queue_; - __intrusive_queue<&task_base::next> pending_queue_{}; + __intrusive_queue<&task_base::next_> pending_queue_{}; std::mutex mut_{}; std::condition_variable cv_{}; bool stop_requested_{false}; @@ -945,9 +915,9 @@ namespace experimental::execution // See `if (is_last_thread)` above. } - inline void _static_thread_pool::bulk_enqueue(remote_queue& queue, - __intrusive_queue<&task_base::next> tasks, - std::size_t tasks_size, + inline void _static_thread_pool::bulk_enqueue(remote_queue& queue, + __intrusive_queue<&task_base::next_> tasks, + std::size_t tasks_size, nodemask const & constraints) noexcept { static thread_local std::thread::id const this_id = std::this_thread::get_id(); @@ -971,7 +941,7 @@ namespace experimental::execution { continue; } - __intrusive_queue<&task_base::next> tmp{}; + __intrusive_queue<&task_base::next_> tmp{}; for (auto j = begin; j < end; ++j) { tmp.push_back(tasks.pop_front()); @@ -982,11 +952,11 @@ namespace experimental::execution } inline void - move_pending_to_local(__intrusive_queue<&task_base::next>& pending_queue, + move_pending_to_local(__intrusive_queue<&task_base::next_>& pending_queue, bwos::lifo_queue>& local_queue) { auto const last = local_queue.push_back(pending_queue.begin(), pending_queue.end()); - __intrusive_queue<&task_base::next> tmp{}; + __intrusive_queue<&task_base::next_> tmp{}; tmp.splice(tmp.begin(), pending_queue, pending_queue.begin(), last); tmp.clear(); } @@ -994,8 +964,8 @@ namespace experimental::execution inline auto _static_thread_pool::thread_state::try_remote() -> _static_thread_pool::thread_state::pop_result { - pop_result result{.task = nullptr, .queue_index = index_}; - __intrusive_queue<&task_base::next> remotes = pool_->remotes_.pop_all_reversed(index_); + pop_result result{.task = nullptr, .queue_index = index_}; + __intrusive_queue<&task_base::next_> remotes = pool_->remotes_.pop_all_reversed(index_); pending_queue_.append(std::move(remotes)); if (!pending_queue_.empty()) { @@ -1054,7 +1024,7 @@ namespace experimental::execution } inline void - _static_thread_pool::thread_state::push_local(__intrusive_queue<&task_base::next>&& tasks) + _static_thread_pool::thread_state::push_local(__intrusive_queue<&task_base::next_>&& tasks) { pending_queue_.prepend(std::move(tasks)); } @@ -1195,6 +1165,11 @@ namespace experimental::execution template struct _static_thread_pool::_opstate : task_base { + void start() & noexcept + { + enqueue_(this); + } + private: friend class ::exec::_pool_::_static_thread_pool::scheduler::_sender; @@ -1202,34 +1177,14 @@ namespace experimental::execution remote_queue* queue, Receiver rcvr, std::size_t tid, - nodemask const & constraints) - : pool_(pool) + nodemask const & constraints) noexcept + : task_base{&_opstate::do_execute_} + , pool_(pool) , queue_(queue) , rcvr_(static_cast(rcvr)) , thread_index_{tid} , constraints_{constraints} - { - this->execute_ = [](task_base* t, - [[maybe_unused]] - std::uint32_t const tid) noexcept - { - auto& op = *static_cast<_opstate*>(t); - auto stoken = get_stop_token(get_env(op.rcvr_)); - - if constexpr (STDEXEC::unstoppable_token) - { // NOLINT(bugprone-branch-clone) - STDEXEC::set_value(static_cast(op.rcvr_)); - } - else if (stoken.stop_requested()) - { - STDEXEC::set_stopped(static_cast(op.rcvr_)); - } - else - { - STDEXEC::set_value(static_cast(op.rcvr_)); - } - }; - } + {} void enqueue_(task_base* op) const { @@ -1243,17 +1198,30 @@ namespace experimental::execution } } + static void do_execute_(task_base* t, [[maybe_unused]] std::uint32_t const tid) noexcept + { + auto& op = *static_cast<_opstate*>(t); + auto stoken = get_stop_token(get_env(op.rcvr_)); + + if constexpr (STDEXEC::unstoppable_token) + { // NOLINT(bugprone-branch-clone) + STDEXEC::set_value(static_cast(op.rcvr_)); + } + else if (stoken.stop_requested()) + { + STDEXEC::set_stopped(static_cast(op.rcvr_)); + } + else + { + STDEXEC::set_value(static_cast(op.rcvr_)); + } + } + _static_thread_pool& pool_; remote_queue* queue_; Receiver rcvr_; std::size_t thread_index_{}; nodemask constraints_{}; - - public: - void start() & noexcept - { - enqueue_(this); - } }; ////////////////////////////////////////////////////////////////////////////////////////////////// @@ -1263,23 +1231,6 @@ namespace experimental::execution { using sender_concept = sender_tag; - template - using _with_error_invoke_t = - __if_c<__value_types_t<__completion_signatures_of_t, - _is_nothrow_bulk_fn, - __q<__mand>>::value, - completion_signatures<>, - __eptr_completion_t>; - - template - using _set_value_t = completion_signatures...)>; - - template - using _completions_t = STDEXEC::__transform_completion_signatures_t< - __completion_signatures_of_t<__copy_cvref_t, Env...>, - _with_error_invoke_t<__copy_cvref_t, Env...>, - _set_value_t>; - template using _bulk_opstate_t = _static_thread_pool::_bulk_opstate; @@ -1293,7 +1244,6 @@ namespace experimental::execution {} template <__decays_to<_bulk_sender> Self, receiver Receiver> - requires receiver_of>> STDEXEC_EXPLICIT_THIS_BEGIN(auto connect)(this Self&& self, Receiver rcvr) noexcept(__nothrow_constructible_from<_bulk_opstate_t, _static_thread_pool&, @@ -1311,11 +1261,41 @@ namespace experimental::execution STDEXEC_EXPLICIT_THIS_END(connect) template <__decays_to<_bulk_sender> Self, class... Env> - static consteval auto get_completion_signatures() -> _completions_t + static consteval auto get_completion_signatures() { - return {}; + return exec::transform_completion_signatures( + STDEXEC::get_completion_signatures<__copy_cvref_t, __fwd_env_t...>(), + []() + { + if constexpr (!__nothrow_decay_copyable) + { + return exec::throw_compile_time_error< + _WHAT_(_PREDECESSOR_RESULTS_ARE_NOT_DECAY_COPYABLE_), + _WHERE_(_IN_ALGORITHM_, bulk_t), + _WITH_PRETTY_SENDER_<__copy_cvref_t>, + _WITH_ENVIRONMENT_(Env...)>(); + } + else if constexpr (!__callable&...>) + { + return exec::throw_compile_time_error< + _WHAT_(_FUNCTION_IS_NOT_CALLABLE_WITH_THE_GIVEN_ARGUMENTS_), + _WHERE_(_IN_ALGORITHM_, bulk_t), + _WITH_FUNCTION_(Fun&), + _WITH_ARGUMENTS_(Shape, Shape, __decay_t & ...)>(); + } + else if constexpr (__nothrow_callable&...>) + { + return completion_signatures...)>(); + } + else + { + return completion_signatures...), + set_error_t(std::exception_ptr)>(); + } + }); } + [[nodiscard]] auto get_env() const noexcept -> env_of_t { return STDEXEC::get_env(sndr_); @@ -1328,97 +1308,111 @@ namespace experimental::execution Fun fun_; }; - //! The customized operation state for `STDEXEC::bulk` operations - template + consteval auto _get_apply_fn() noexcept + { + return [](Fun& fun, Tuple& tupl) noexcept( + __nothrow_applicable) -> void + { + __apply(std::move(fun), tupl); + }; + } + + //! The customized opstate state for `STDEXEC::bulk` opstates + template struct _static_thread_pool::_bulk_shared_state { + using variant_t = + __value_types_of_t, __qq<__decayed_tuple>, __qq<__variant>>; + + //! Construct from a pool, receiver, shape, and function. + //! Allocates O(min(shape, available_parallelism())) memory. + explicit _bulk_shared_state(_static_thread_pool& pool, Receiver rcvr, Shape shape, Fun fun) + : pool_{pool} + , rcvr_{static_cast(rcvr)} + , shape_{shape} + , fun_{fun} + , thread_with_exception_{num_agents_required()} + , tasks_{num_agents_required(), bulk_task{this}} + {} + + //! The number of agents required is the minimum of `shape_` and the available parallelism. + //! That is, we don't need an agent for each of the shape values. + [[nodiscard]] + auto num_agents_required() const noexcept -> std::uint32_t + { + if constexpr (Parallelize) + { + return static_cast( + __umin({std::size_t(shape_), std::size_t(pool_.available_parallelism())})); + } + else + { + return static_cast(1); + } + } + + private: + friend struct _bulk_receiver; + //! The actual `bulk_task` holds a pointer to the shared state //! and its `execute_` function reads from that shared state. struct bulk_task : task_base { - _bulk_shared_state* sh_state_; + explicit bulk_task(_bulk_shared_state* sh_state) noexcept + : task_base{&bulk_task::do_execute_} + , sh_state_(sh_state) + {} - bulk_task(_bulk_shared_state* sh_state) - : sh_state_(sh_state) - { - this->execute_ = [](task_base* t, std::uint32_t const tid) noexcept + private: + static void do_execute_(task_base* t, std::uint32_t const tid) noexcept + { + auto& sh_state = *static_cast(t)->sh_state_; + auto total_threads = sh_state.num_agents_required(); + // Each computation does one or more call to the the bulk function. In the + // case that the shape is much larger than the total number of threads, then + // each call to computation will call the function many times. + auto const [begin, end] = _pool_::even_share(sh_state.shape_, tid, total_threads); + auto const applicator = std::bind_front(std::ref(sh_state.fun_), begin, end); + auto const computation = std::bind_front(__apply, applicator); + auto const completion = STDEXEC::__flip(std::bind_front(__apply, STDEXEC::set_value)); + + if constexpr (noexcept(__visit(computation, sh_state.data_))) { - auto& sh_state = *static_cast(t)->sh_state_; - auto total_threads = sh_state.num_agents_required(); - - auto computation = [&](auto&... args) - { - // Each computation does one or more call to the the bulk function. - // In the case that the shape is much larger than the total number of threads, - // then each call to computation will call the function many times. - auto [begin, end] = even_share(sh_state.shape_, tid, total_threads); - sh_state.fun_(begin, end, args...); - }; - - auto completion = [&](auto&... args) + __visit(computation, sh_state.data_); + if (sh_state.finished_threads_.fetch_add(1) + 1 == total_threads) // last thread? + __visit(completion, std::move(sh_state.data_), std::move(sh_state.rcvr_)); + } + else + { + STDEXEC_TRY { - STDEXEC::set_value(static_cast(sh_state.rcvr_), std::move(args)...); - }; - - if constexpr (MayThrow) + __visit(computation, sh_state.data_); + } + STDEXEC_CATCH_ALL { - STDEXEC_TRY + std::uint32_t expected = total_threads; + if (sh_state.thread_with_exception_.compare_exchange_strong( + expected, + tid, + __std::memory_order_relaxed, + __std::memory_order_relaxed)) { - sh_state.apply(computation); - } - STDEXEC_CATCH_ALL - { - std::uint32_t expected = total_threads; - - if (sh_state.thread_with_exception_.compare_exchange_strong( - expected, - tid, - __std::memory_order_relaxed, - __std::memory_order_relaxed)) - { - sh_state.exception_ = std::current_exception(); - } - } - - bool const is_last_thread = sh_state.finished_threads_.fetch_add(1) - == (total_threads - 1); - - if (is_last_thread) - { - if (sh_state.exception_) - { - STDEXEC::set_error(static_cast(sh_state.rcvr_), - std::move(sh_state.exception_)); - } - else - { - sh_state.apply(completion); - } + sh_state.exception_ = std::current_exception(); } } - else - { - sh_state.apply(computation); - - bool const is_last_thread = sh_state.finished_threads_.fetch_add(1) - == (total_threads - 1); - if (is_last_thread) - { - sh_state.apply(completion); - } + if (sh_state.finished_threads_.fetch_add(1) + 1 == total_threads) // last thread? + { + if (sh_state.exception_) + STDEXEC::set_error(std::move(sh_state.rcvr_), std::move(sh_state.exception_)); + else + __visit(completion, std::move(sh_state.data_), std::move(sh_state.rcvr_)); } - }; + } } - }; - using variant_t = - __value_types_of_t, __q<__decayed_tuple>, __q<__variant>>; + _bulk_shared_state* sh_state_; + }; variant_t data_{STDEXEC::__no_init}; _static_thread_pool& pool_; @@ -1430,56 +1424,14 @@ namespace experimental::execution __std::atomic thread_with_exception_{0}; std::exception_ptr exception_; std::vector tasks_; - - //! The number of agents required is the minimum of `shape_` and the available parallelism. - //! That is, we don't need an agent for each of the shape values. - [[nodiscard]] - auto num_agents_required() const noexcept -> std::uint32_t - { - if constexpr (Parallelize) - { - return static_cast( - __umin({std::size_t(shape_), std::size_t(pool_.available_parallelism())})); - } - else - { - return static_cast(1); - } - } - - template - void apply(F f) - { - STDEXEC_ASSERT(!data_.__is_valueless()); - __visit([&](auto& tupl) -> void { __apply(std::move(f), tupl); }, data_); - } - - //! Construct from a pool, receiver, shape, and function. - //! Allocates O(min(shape, available_parallelism())) memory. - _bulk_shared_state(_static_thread_pool& pool, Receiver rcvr, Shape shape, Fun fun) - : pool_{pool} - , rcvr_{static_cast(rcvr)} - , shape_{shape} - , fun_{fun} - , thread_with_exception_{num_agents_required()} - , tasks_{num_agents_required(), {this}} - {} }; - //! A customized receiver to allow parallel execution of `STDEXEC::bulk` operations: - template + //! A customized receiver to allow parallel execution of `STDEXEC::bulk` opstates: + template struct _static_thread_pool::_bulk_receiver { using receiver_concept = receiver_tag; - using shared_state = - _bulk_shared_state; - void enqueue() noexcept { STDEXEC_ASSERT(shared_state_.tasks_.size() == shared_state_.num_agents_required()); @@ -1489,45 +1441,41 @@ namespace experimental::execution template void set_value(As&&... as) noexcept { - using tuple_t = __decayed_tuple; - - shared_state& state = shared_state_; - STDEXEC_TRY { - state.data_.template emplace(static_cast(as)...); + shared_state_.data_.template emplace<__decayed_tuple>(static_cast(as)...); } STDEXEC_CATCH_ALL { - if constexpr (MayThrow) + if constexpr (!__nothrow_decay_copyable) { - STDEXEC::set_error(std::move(state.rcvr_), std::current_exception()); + STDEXEC::set_error(static_cast(shared_state_.rcvr_), + std::current_exception()); return; } } - if (state.shape_) + if (shared_state_.shape_) { enqueue(); } else { - state.apply([&](auto&... args) noexcept -> void - { STDEXEC::set_value(std::move(state.rcvr_), std::move(args)...); }); + __visit(STDEXEC::__flip(std::bind_front(__apply, STDEXEC::set_value)), + std::move(shared_state_.data_), + static_cast(shared_state_.rcvr_)); } } template void set_error(Error&& error) noexcept { - shared_state& state = shared_state_; - STDEXEC::set_error(static_cast(state.rcvr_), static_cast(error)); + STDEXEC::set_error(std::move(shared_state_.rcvr_), static_cast(error)); } void set_stopped() noexcept { - shared_state& state = shared_state_; - STDEXEC::set_stopped(static_cast(state.rcvr_)); + STDEXEC::set_stopped(std::move(shared_state_.rcvr_)); } auto get_env() const noexcept -> env_of_t @@ -1535,34 +1483,32 @@ namespace experimental::execution return STDEXEC::get_env(shared_state_.rcvr_); } - shared_state& shared_state_; + _bulk_shared_state& shared_state_; }; template struct _static_thread_pool::_bulk_opstate { - static constexpr bool may_throw = !__value_types_of_t, - _is_nothrow_bulk_fn, - __q<__mand>>::value; - - using receiver_t = _bulk_receiver; - using shared_state_t = - _bulk_shared_state; + using receiver_t = _bulk_receiver; + using shared_state_t = _bulk_shared_state; using inner_opstate_t = connect_result_t; - shared_state_t shared_state_; - inner_opstate_t inner_op_; - void start() & noexcept { STDEXEC::start(inner_op_); } - _bulk_opstate(_static_thread_pool& pool, Shape shape, Fun fun, CvSender&& sndr, Receiver rcvr) + explicit _bulk_opstate(_static_thread_pool& pool, + Shape shape, + Fun fun, + CvSender&& sndr, + Receiver rcvr) : shared_state_(pool, static_cast(rcvr), shape, fun) , inner_op_{STDEXEC::connect(static_cast(sndr), receiver_t{shared_state_})} {} + + shared_state_t shared_state_; + inner_opstate_t inner_op_; }; namespace schedule_all_ @@ -1584,35 +1530,24 @@ namespace experimental::execution using allocator_of_t = decltype(get_allocator(__declval())); template - struct operation_base - { - Range range_; - _static_thread_pool& pool_; - std::mutex start_mutex_{}; - bool has_started_{false}; - __intrusive_queue<&task_base::next> tasks_{}; - std::size_t tasks_size_{}; - __std::atomic countdown_{std::ranges::size(range_)}; + struct opstate_base + { + Range range_; + _static_thread_pool& pool_; + std::mutex start_mutex_{}; + bool has_started_{false}; + __intrusive_queue<&task_base::next_> tasks_{}; + std::size_t tasks_size_{}; + __std::atomic countdown_{std::ranges::size(range_)}; }; template - class item_operation : task_base + struct item_opstate : task_base { - static void execute_(task_base* base, std::uint32_t /* tid */) noexcept - { - auto op = static_cast(base); - STDEXEC::set_value(static_cast(op->item_receiver_), *op->it_); - } - - ItemReceiver item_receiver_; - std::ranges::iterator_t it_; - operation_base* parent_; - - public: - item_operation(ItemReceiver&& item_receiver, - std::ranges::iterator_t it, - operation_base* parent) - : task_base{.execute_ = execute_} + explicit item_opstate(ItemReceiver&& item_receiver, + std::ranges::iterator_t it, + opstate_base* parent) + : task_base{&item_opstate::do_execute_} , item_receiver_(static_cast(item_receiver)) , it_(it) , parent_(parent) @@ -1632,6 +1567,17 @@ namespace experimental::execution parent_->pool_.enqueue(static_cast(this)); } } + + private: + static void do_execute_(task_base* base, std::uint32_t /* tid */) noexcept + { + auto op = static_cast(base); + STDEXEC::set_value(static_cast(op->item_receiver_), *op->it_); + } + + ItemReceiver item_receiver_; + std::ranges::iterator_t it_; + opstate_base* parent_; }; template @@ -1641,9 +1587,6 @@ namespace experimental::execution using completion_signatures = STDEXEC::completion_signatures)>; - operation_base* op_; - std::ranges::iterator_t it_; - struct attrs { _static_thread_pool* pool_; @@ -1662,21 +1605,24 @@ namespace experimental::execution template requires receiver_of - auto connect(ItemReceiver rcvr) const noexcept -> item_operation + auto connect(ItemReceiver rcvr) const noexcept { - return {static_cast(rcvr), it_, op_}; + return item_opstate{static_cast(rcvr), it_, op_}; } + + opstate_base* op_; + std::ranges::iterator_t it_; }; template - struct operation_base_with_receiver : operation_base + struct opstate_base_with_receiver : opstate_base { - Receiver rcvr_; - - operation_base_with_receiver(Range range, _static_thread_pool& pool, Receiver rcvr) - : operation_base{range, pool} + explicit opstate_base_with_receiver(Range range, _static_thread_pool& pool, Receiver rcvr) + : opstate_base{range, pool} , rcvr_(static_cast(rcvr)) {} + + Receiver rcvr_; }; template @@ -1707,33 +1653,21 @@ namespace experimental::execution return STDEXEC::get_env(op_->rcvr_); } - operation_base_with_receiver* op_; + opstate_base_with_receiver* op_; }; template - class operation : operation_base_with_receiver + class opstate : opstate_base_with_receiver { - using allocator_t = allocator_of_t; - using item_sender_t = item_sender; - using next_sender_t = next_sender_of_t; - using next_receiver_t = next_receiver; - using item_operation_t = connect_result_t; - - using item_allocator_t = std::allocator_traits::template rebind_alloc< - STDEXEC::__manual_lifetime>; - - std::vector<__manual_lifetime, item_allocator_t> items_; - std::size_t items_constructed_{}; - public: - operation(Range range, _static_thread_pool& pool, Receiver rcvr) - : operation_base_with_receiver{std::move(range), - pool, - static_cast(rcvr)} + explicit opstate(Range range, _static_thread_pool& pool, Receiver rcvr) + : opstate_base_with_receiver{std::move(range), + pool, + static_cast(rcvr)} , items_(std::ranges::size(this->range_), item_allocator_t(get_allocator(this->rcvr_))) {} - ~operation() + ~opstate() { for (std::size_t i = 0; i < items_constructed_; ++i) { @@ -1746,7 +1680,7 @@ namespace experimental::execution std::size_t size = items_.size(); if (size == 0) { - STDEXEC::set_value(static_cast(this->rcvr_)); + STDEXEC::set_value(std::move(this->rcvr_)); return; } @@ -1759,6 +1693,7 @@ namespace experimental::execution auto& remote_queue = *this->pool_.get_remote_queue(); auto it = std::ranges::begin(this->range_); std::size_t i0 = 0; + STDEXEC_TRY { for (std::size_t i = 0; i < size; ++i) @@ -1771,7 +1706,7 @@ namespace experimental::execution } STDEXEC_CATCH_ALL { - STDEXEC::set_error(static_cast(this->rcvr_), std::current_exception()); + STDEXEC::set_error(std::move(this->rcvr_), std::current_exception()); return; } @@ -1788,49 +1723,62 @@ namespace experimental::execution lock.unlock(); i0 += chunk_size; } + for (std::size_t i = i0; i < size; ++i) { STDEXEC::start(items_[i].__get()); } + std::unique_lock lock{this->start_mutex_}; this->has_started_ = true; this->pool_.bulk_enqueue(remote_queue, std::move(this->tasks_), this->tasks_size_); } + + private: + using allocator_t = allocator_of_t; + using item_sender_t = item_sender; + using next_sender_t = next_sender_of_t; + using next_receiver_t = next_receiver; + using item_opstate_t = connect_result_t; + + using item_allocator_t = std::allocator_traits::template rebind_alloc< + STDEXEC::__manual_lifetime>; + + std::vector<__manual_lifetime, item_allocator_t> items_; + std::size_t items_constructed_{}; }; template - class sequence + struct sequence { - Range range_; - _static_thread_pool* pool_; - - public: using sender_concept = sequence_sender_tag; - + using item_types = exec::item_types>; using completion_signatures = STDEXEC::completion_signatures; - using item_types = exec::item_types>; - - sequence(Range range, _static_thread_pool& pool) + explicit sequence(Range range, _static_thread_pool& pool) : range_(static_cast(range)) , pool_(&pool) {} template Receiver> - auto subscribe(Receiver rcvr) && noexcept -> operation + auto subscribe(Receiver rcvr) && noexcept -> opstate { - return {static_cast(range_), *pool_, static_cast(rcvr)}; + return opstate{std::move(range_), *pool_, std::move(rcvr)}; } template Receiver> requires __decay_copyable - auto subscribe(Receiver rcvr) const & noexcept -> operation + auto subscribe(Receiver rcvr) const & noexcept -> opstate { - return {range_, *pool_, static_cast(rcvr)}; + return opstate{range_, *pool_, std::move(rcvr)}; } + + private: + Range range_; + _static_thread_pool* pool_; }; } // namespace schedule_all_ @@ -1844,9 +1792,9 @@ namespace experimental::execution static_thread_pool() = default; - static_thread_pool(std::uint32_t thread_count, - bwos_params params = {}, - numa_policy numa = get_numa_policy()) + explicit static_thread_pool(std::uint32_t thread_count, + bwos_params params = {}, + numa_policy numa = get_numa_policy()) : _pool_::_static_thread_pool(thread_count, params, std::move(numa)) {} @@ -1878,9 +1826,8 @@ namespace experimental::execution { template auto operator()(static_thread_pool& pool, Range&& range) const - -> schedule_all_::sequence<__decay_t> { - return {static_cast(range), pool}; + return schedule_all_::sequence<__decay_t>{static_cast(range), pool}; } }; } // namespace _pool_ diff --git a/include/exec/variant_sender.hpp b/include/exec/variant_sender.hpp index fc465482b..02bdea6cf 100644 --- a/include/exec/variant_sender.hpp +++ b/include/exec/variant_sender.hpp @@ -19,6 +19,8 @@ #include "../stdexec/__detail/__variant.hpp" #include "../stdexec/execution.hpp" +#include "completion_signatures.hpp" + namespace experimental::execution { namespace __var @@ -60,10 +62,6 @@ namespace experimental::execution template struct variant_sender { - template - using __completions_t = STDEXEC::__mtry_q::__f< - STDEXEC::__completion_signatures_of_t, _Env...>...>; - template using __nth_t = STDEXEC::__m_at_c<_Index, _Senders...>; @@ -140,9 +138,10 @@ namespace experimental::execution STDEXEC_EXPLICIT_THIS_END(connect) template _Self, class... _Env> - static consteval auto get_completion_signatures() -> __completions_t<_Self, _Env...> + static consteval auto get_completion_signatures() { - return {}; + return exec::concat_completion_signatures( + STDEXEC::get_completion_signatures, _Env...>()...); } }; } // namespace experimental::execution diff --git a/include/stdexec/__detail/__associate.hpp b/include/stdexec/__detail/__associate.hpp index 6a9d92779..b4aeebcbc 100644 --- a/include/stdexec/__detail/__associate.hpp +++ b/include/stdexec/__detail/__associate.hpp @@ -36,7 +36,6 @@ import stdexec; # include "__sender_concepts.hpp" # include "__sender_introspection.hpp" # include "__senders.hpp" -# include "__transform_completion_signatures.hpp" # include "__type_traits.hpp" # if !STDEXEC_USE_MODULES() @@ -256,13 +255,12 @@ namespace STDEXEC __copy_cvref_t<_Sender, typename __data_of>::__wrap_sender_t>; template - static consteval auto __get_completion_signatures() // - -> __transform_completion_signatures_t< - __completion_signatures_of_t<__wrap_sender_of_t<_Sender>, _Env...>, - completion_signatures> + static consteval auto __get_completion_signatures() { static_assert(__sender_for<_Sender, associate_t>); - return {}; + return STDEXEC::__concat_completion_signatures( + STDEXEC::get_completion_signatures<__wrap_sender_of_t<_Sender>, _Env...>(), + completion_signatures()); }; static constexpr auto __get_state = diff --git a/include/stdexec/__detail/__bulk.hpp b/include/stdexec/__detail/__bulk.hpp index 0da2eeb30..63b0e54eb 100644 --- a/include/stdexec/__detail/__bulk.hpp +++ b/include/stdexec/__detail/__bulk.hpp @@ -120,62 +120,6 @@ namespace STDEXEC STDEXEC_HOST_DEVICE_DEDUCTION_GUIDE __data(_Pol const &, _Shape, _Fun) -> __data<_Pol, _Shape, _Fun>; - template - struct __bulk_traits; - - template <> - struct __bulk_traits - { - using __on_not_callable = __mbind_front_q<__callable_error_t, bulk_t>; - - // Curried function, after passing the required indices. - template - using __fun_curried = - __mbind_front<__mtry_catch_q<__nothrow_invocable_t, __on_not_callable>, _Fun, _Shape>; - }; - - template <> - struct __bulk_traits - { - using __on_not_callable = __mbind_front_q<__callable_error_t, bulk_chunked_t>; - - // Curried function, after passing the required indices. - template - using __fun_curried = __mbind_front<__mtry_catch_q<__nothrow_invocable_t, __on_not_callable>, - _Fun, - _Shape, - _Shape>; - }; - - template <> - struct __bulk_traits - { - using __on_not_callable = __mbind_front_q<__callable_error_t, bulk_unchunked_t>; - - // Curried function, after passing the required indices. - template - using __fun_curried = - __mbind_front<__mtry_catch_q<__nothrow_invocable_t, __on_not_callable>, _Fun, _Shape>; - }; - - template - using __decay_ref = __decay_t<_Ty>&; - - template - using __with_error_invoke_t = - __if<__value_types_t< - __completion_signatures_of_t<_CvSender, _Env...>, - __mtransform<__q<__decay_ref>, - typename __bulk_traits<_AlgoTag>::template __fun_curried<_Fun, _Shape>>, - __q<__mand>>, - completion_signatures<>, - __eptr_completion_t>; - - template - using __completion_signatures = __transform_completion_signatures_t< - __completion_signatures_of_t<_CvSender, _Env...>, - __with_error_invoke_t<_AlgoTag, _Fun, _Shape, _CvSender, _Env...>>; - template struct __generic_bulk_t // NOLINT(bugprone-crtp-constructor-accessibility) { @@ -231,7 +175,7 @@ namespace STDEXEC template constexpr void operator()(_Shape __begin, _Shape __end, _Args&... __args) - noexcept(__nothrow_callable<_Fun&, _Shape, decltype(__args)...>) + noexcept(__nothrow_callable<_Fun&, _Shape, _Args&...>) { for (; __begin != __end; ++__begin) { @@ -264,12 +208,6 @@ namespace STDEXEC template struct __impl_base : __sexpr_defaults { - template - using __fun_t = decltype(__decay_t<__data_of<_Sender>>::__fun_); - - template - using __shape_t = decltype(__decay_t<__data_of<_Sender>>::__shape_); - // Forward the child sender's environment (which contains completion scheduler) static constexpr auto __get_attrs = // [](__ignore, __ignore, _Child const & __child) noexcept @@ -277,25 +215,49 @@ namespace STDEXEC return __attrs{__child}; }; - template + template static consteval auto __get_completion_signatures() { - static_assert(__sender_for<_Sender, _AlgoTag>); - // TODO: port this to use constant evaluation - return __completion_signatures<_AlgoTag, - __fun_t<_Sender>, - __shape_t<_Sender>, - __child_of<_Sender>, - _Env...>{}; + static_assert(__sender_for<_CvSender, _AlgoTag>); + using __state_t = __decay_t<__data_of<_CvSender>>; + using __fun_t = decltype(__state_t::__fun_); + using __shape_t = decltype(__state_t::__shape_); + + return STDEXEC::__transform_completion_signatures( + STDEXEC::get_completion_signatures<__child_of<_CvSender>, __fwd_env_t<_Env>...>(), + []() + { + using __value_sig_t = set_value_t(_Args...); + using __arg_pack_t = __if_c<__same_as<_AlgoTag, bulk_chunked_t>, + __tuple<__shape_t, __shape_t, _Args&...>, + __tuple<__shape_t, _Args&...>>; + + if constexpr (__nothrow_applicable<__fun_t&, __arg_pack_t>) + { + return completion_signatures<__value_sig_t>(); + } + else if constexpr (__applicable<__fun_t&, __arg_pack_t>) + { + return completion_signatures<__value_sig_t, set_error_t(std::exception_ptr)>(); + } + else + { + return STDEXEC::__throw_compile_time_error< + _WHAT_(_FUNCTION_IS_NOT_CALLABLE_WITH_THE_GIVEN_ARGUMENTS_), + _WHERE_(_IN_ALGORITHM_, _AlgoTag), + _WITH_FUNCTION_(__fun_t&), + __mapply<__qf<_WITH_ARGUMENTS_>, __arg_pack_t>>(); + } + }); }; }; struct __chunked_impl : __impl_base { - //! This implements the core default behavior for `bulk_chunked`: - //! When setting value, it calls the function with the entire range. - //! Note: This is not done in parallel. That is customized by the scheduler. - //! See, e.g., static_thread_pool::bulk_receiver::__t. + //! This implements the core default behavior for `bulk_chunked`: When setting + //! value, it calls the function with the entire range. Note: This is not done in + //! parallel. That is customized by the scheduler. See, e.g., + //! `static_thread_pool::bulk_receiver`. static constexpr auto __complete = [](__ignore, _State& __state, @@ -305,17 +267,18 @@ namespace STDEXEC if constexpr (__std::same_as<_Tag, set_value_t>) { // Intercept set_value and dispatch to the bulk operation. - using __shape_t = decltype(__state.__data_.__shape_); - using __fun_t = decltype(__state.__data_.__fun_); - constexpr bool __is_nothrow = __nothrow_callable<__fun_t, __shape_t, __shape_t, _Args...>; + using __shape_t = decltype(__state.__data_.__shape_); + using __fun_t = decltype(__state.__data_.__fun_); + STDEXEC_TRY { - __state.__data_.__fun_(static_cast<__shape_t>(0), __state.__data_.__shape_, __args...); - _Tag()(static_cast<_State&&>(__state).__rcvr_, static_cast<_Args&&>(__args)...); + __state.__data_.__fun_(__shape_t(), __shape_t(__state.__data_.__shape_), __args...); + STDEXEC::set_value(static_cast<_State&&>(__state).__rcvr_, + static_cast<_Args&&>(__args)...); } STDEXEC_CATCH_ALL { - if constexpr (!__is_nothrow) + if constexpr (!__nothrow_callable<__fun_t&, __shape_t, __shape_t, _Args&...>) { STDEXEC::set_error(static_cast<_State&&>(__state).__rcvr_, std::current_exception()); } @@ -341,21 +304,22 @@ namespace STDEXEC { if constexpr (__std::same_as<_Tag, set_value_t>) { - using __shape_t = decltype(__state.__data_.__shape_); - using __fun_t = decltype(__state.__data_.__fun_); - constexpr bool __is_nothrow = __nothrow_callable<__fun_t, __shape_t, _Args...>; - auto const __shape = __state.__data_.__shape_; + using __shape_t = decltype(__state.__data_.__shape_); + using __fun_t = decltype(__state.__data_.__fun_); + auto const __shape = __state.__data_.__shape_; + STDEXEC_TRY { for (__shape_t __i{}; __i != __shape; ++__i) { - __state.__data_.__fun_(__i, __args...); + __state.__data_.__fun_(__shape_t(__i), __args...); } - _Tag()(static_cast<_State&&>(__state).__rcvr_, static_cast<_Args&&>(__args)...); + STDEXEC::set_value(static_cast<_State&&>(__state).__rcvr_, + static_cast<_Args&&>(__args)...); } STDEXEC_CATCH_ALL { - if constexpr (!__is_nothrow) + if constexpr (!__nothrow_callable<__fun_t&, __shape_t, _Args&...>) { STDEXEC::set_error(static_cast<_State&&>(__state).__rcvr_, std::current_exception()); } diff --git a/include/stdexec/__detail/__continues_on.hpp b/include/stdexec/__detail/__continues_on.hpp index d9fb2866b..4d663276a 100644 --- a/include/stdexec/__detail/__continues_on.hpp +++ b/include/stdexec/__detail/__continues_on.hpp @@ -291,31 +291,22 @@ namespace STDEXEC template static consteval auto __get_child_completions() { - auto __child_completions = - STDEXEC::get_completion_signatures<_Child, __fwd_env_t<_Env>...>(); - STDEXEC_IF_OK(__child_completions) - { - // continues_on has the completions of the child sender, but with value and - // error result types decayed. - return __transform_completion_signatures( - __child_completions, - __decay_arguments(), - __decay_arguments()); - } + // continues_on has the completions of the child sender, but with value and error + // result types decayed. + return STDEXEC::__transform_completion_signatures( + STDEXEC::get_completion_signatures<_Child, __fwd_env_t<_Env>...>(), + __decay_arguments(), + __decay_arguments()); } template static consteval auto __get_scheduler_completions() { - using __sndr_t = schedule_result_t<_Scheduler>; - auto __sched_completions = - STDEXEC::get_completion_signatures<__sndr_t, __fwd_env_t<_Env>...>(); - STDEXEC_IF_OK(__sched_completions) - { - // The scheduler contributes only error and stopped completions; we ignore value - // completions here - return __transform_completion_signatures(__sched_completions, __ignore_completion()); - } + // The scheduler contributes only error and stopped completions; we ignore value + // completions here + return STDEXEC::__transform_completion_signatures( + STDEXEC::get_completion_signatures, __fwd_env_t<_Env>...>(), + __ignore_completion()); } template @@ -337,7 +328,7 @@ namespace STDEXEC using __scheduler_t = __decay_t<__data_of<_Sender>>; using __child_t = __child_of<_Sender>; auto __child_completions = __get_child_completions<__child_t, _Env...>(); - return __concat_completion_signatures( + return STDEXEC::__concat_completion_signatures( __child_completions, __get_scheduler_completions<__scheduler_t, _Env...>(), __eptr_completion_unless_t< diff --git a/include/stdexec/__detail/__diagnostics.hpp b/include/stdexec/__detail/__diagnostics.hpp index 6f24c56ba..74ac61eb0 100644 --- a/include/stdexec/__detail/__diagnostics.hpp +++ b/include/stdexec/__detail/__diagnostics.hpp @@ -103,7 +103,7 @@ namespace STDEXEC struct _SENDER_TYPE_IS_NOT_DECAY_COPYABLE_ {}; - struct _SENDER_RESULTS_ARE_NOT_DECAY_COPYABLE_ + struct _PREDECESSOR_RESULTS_ARE_NOT_DECAY_COPYABLE_ {}; struct _TYPE_IS_NOT_DECAY_COPYABLE_ diff --git a/include/stdexec/__detail/__finally.hpp b/include/stdexec/__detail/__finally.hpp index 994bcb2c4..1bff782d0 100644 --- a/include/stdexec/__detail/__finally.hpp +++ b/include/stdexec/__detail/__finally.hpp @@ -361,9 +361,9 @@ namespace STDEXEC template <__decays_to<__sender> _Self, class... _Env> static consteval auto get_completion_signatures() { - return __final ::__get_completion_signatures<__copy_cvref_t<_Self, _InitialSender>, - __copy_cvref_t<_Self, _FinalSender>, - _Env...>(); + return __final::__get_completion_signatures<__copy_cvref_t<_Self, _InitialSender>, + __copy_cvref_t<_Self, _FinalSender>, + _Env...>(); } _InitialSender __initial_sndr_; diff --git a/include/stdexec/__detail/__into_variant.hpp b/include/stdexec/__detail/__into_variant.hpp index b8ad6d189..5f6987865 100644 --- a/include/stdexec/__detail/__into_variant.hpp +++ b/include/stdexec/__detail/__into_variant.hpp @@ -30,7 +30,7 @@ import stdexec; # include "__concepts.hpp" # include "__meta.hpp" # include "__sender_adaptor_closure.hpp" -# include "__senders.hpp" // IWYU pragma: keep for __well_formed_sender +# include "__senders.hpp" # include "__transform_completion_signatures.hpp" # include "__utility.hpp" @@ -48,25 +48,6 @@ namespace STDEXEC // [exec.into.variant] namespace __into_variant { - template - requires sender_in<_Sender, _Env> - using __into_variant_result_t = value_types_of_t<_Sender, _Env>; - - template - using __variant_t = __value_types_t<__completion_signatures_of_t<_Sender, _Env...>, - __qq<__decayed_std_tuple>, - __qq<__std_variant>>; - - template - using __variant_completions = - completion_signatures; - - template - using __completions = __transform_completion_signatures_t< - __completion_signatures_of_t<_Sender, _Env...>, - __minvoke_q<__variant_completions, __variant_t<_Sender, _Env...>>, - __mconst>::__f>; - template struct __state { @@ -114,7 +95,19 @@ namespace STDEXEC static consteval auto __get_completion_signatures() { static_assert(__sender_for<_Self, into_variant_t>); - return __completions<__child_of<_Self>, _Env...>{}; + auto __completions = STDEXEC::get_completion_signatures<__child_of<_Self>, _Env...>(); + STDEXEC_IF_OK(__completions) + { + using __completions_t = decltype(__completions); + using __variant_t = + __value_types_t<__completions_t, __qq<__decayed_std_tuple>, __qq<__std_variant>>; + return STDEXEC::__transform_completion_signatures( + __completions, + __ignore_completion(), + {}, + {}, + completion_signatures()); + } }; }; } // namespace __into_variant diff --git a/include/stdexec/__detail/__let.hpp b/include/stdexec/__detail/__let.hpp index be4baf44d..45b2950a6 100644 --- a/include/stdexec/__detail/__let.hpp +++ b/include/stdexec/__detail/__let.hpp @@ -120,7 +120,7 @@ namespace STDEXEC template using __not_decay_copyable_error_t = - __mexception<_WHAT_(_SENDER_RESULTS_ARE_NOT_DECAY_COPYABLE_), + __mexception<_WHAT_(_PREDECESSOR_RESULTS_ARE_NOT_DECAY_COPYABLE_), _WHERE_(_IN_ALGORITHM_, _LetTag), _WITH_ARGUMENTS_(_Args...)>; diff --git a/include/stdexec/__detail/__parallel_scheduler.hpp b/include/stdexec/__detail/__parallel_scheduler.hpp index 6c904a166..234e21e87 100644 --- a/include/stdexec/__detail/__parallel_scheduler.hpp +++ b/include/stdexec/__detail/__parallel_scheduler.hpp @@ -33,7 +33,6 @@ import stdexec; # include "__schedulers.hpp" # include "__sender_introspection.hpp" # include "__senders.hpp" -# include "__transform_completion_signatures.hpp" # include "__transform_sender.hpp" # if !STDEXEC_USE_MODULES() @@ -644,12 +643,6 @@ namespace STDEXEC template class __parallel_bulk_sender { - /// Meta-function that returns the completion signatures of `this`. - template - using __completions_t = __transform_completion_signatures_t< - __completion_signatures_of_t<__copy_cvref_t<_Self, _Previous>, _Env...>, - completion_signatures>; - template friend struct __detail::__system_bulk_op; @@ -704,9 +697,11 @@ namespace STDEXEC /// Gets the completion signatures for this sender. template <__decays_to<__parallel_bulk_sender> _Self, class... _Env> - static consteval auto get_completion_signatures() -> __completions_t<_Self, _Env...> + static consteval auto get_completion_signatures() { - return {}; + using __cv_previous_t = __copy_cvref_t<_Self, _Previous>; + auto __completions = STDEXEC::get_completion_signatures<__cv_previous_t, _Env...>(); + return STDEXEC::__concat_completion_signatures(__completions, __eptr_completion_t()); } private: diff --git a/include/stdexec/__detail/__sender_ref.hpp b/include/stdexec/__detail/__sender_ref.hpp index 6aad681ed..500c106fa 100644 --- a/include/stdexec/__detail/__sender_ref.hpp +++ b/include/stdexec/__detail/__sender_ref.hpp @@ -37,8 +37,7 @@ namespace STDEXEC using __sender_t = __copy_cvref_t<_Self, std::remove_cv_t<_Sender>>; template - static consteval auto get_completion_signatures() // - -> __completion_signatures_of_t<__sender_t<_Self>, _Env...> + static consteval auto get_completion_signatures() { return STDEXEC::get_completion_signatures<__sender_t<_Self>, _Env...>(); } diff --git a/include/stdexec/__detail/__transform_sender.hpp b/include/stdexec/__detail/__transform_sender.hpp index 55157b1e9..4ca58bb99 100644 --- a/include/stdexec/__detail/__transform_sender.hpp +++ b/include/stdexec/__detail/__transform_sender.hpp @@ -34,7 +34,7 @@ namespace STDEXEC namespace __detail { template - struct __transform_sender_t + struct __transform_sender { template using __domain_for_t = @@ -65,7 +65,7 @@ namespace STDEXEC else { using __transform_recurse_t = - __transform_sender_t<__completing_domain_t, set_value_t>; + __transform_sender<__completing_domain_t, set_value_t>; return __transform_recurse_t::template __get_declfn<__result_t, _Env, (_Nothrow && __is_nothrow)>(); @@ -102,7 +102,7 @@ namespace STDEXEC else { using __transform_recurse_t = - __transform_sender_t<__completing_domain_t, set_value_t>; + __transform_sender<__completing_domain_t, set_value_t>; return __transform_recurse_t()(__domain_t().transform_sender(_OpTag(), static_cast<_Sndr&&>(__sndr), __env), @@ -133,28 +133,27 @@ namespace STDEXEC // 2. Starting domain transformation (where the operation state starts) template using __impl_fn_t = - __compose<__detail::__transform_sender_t<__detail::__starting_domain_t<_Env>, start_t>, - __detail::__transform_sender_t<__detail::__completing_domain_t, - set_value_t>>; + __compose<__detail::__transform_sender<__detail::__starting_domain_t<_Env>, start_t>, + __detail::__transform_sender<__detail::__completing_domain_t, + set_value_t>>; public: // NOT TO SPEC: template STDEXEC_ATTRIBUTE(nodiscard, host, device, always_inline) - constexpr auto - operator()(_Sndr&& __sndr) const noexcept(__nothrow_move_constructible<_Sndr>) // - -> _Sndr + constexpr auto operator()(_Sndr&& __sndr) const // + noexcept(__nothrow_move_constructible<_Sndr>) -> _Sndr { return static_cast<_Sndr&&>(__sndr); } - template {}> + template > STDEXEC_ATTRIBUTE(nodiscard, host, device, always_inline) - constexpr auto operator()(_Sndr && __sndr, _Env const & __env) const - noexcept(noexcept(_ImplFn(static_cast<_Sndr&&>(__sndr), __env))) - -> decltype(_ImplFn(static_cast<_Sndr&&>(__sndr), __env)) + constexpr auto operator()(_Sndr&& __sndr, _Env const & __env) const + noexcept(__nothrow_callable<_ImplFn, _Sndr, _Env const &>) + -> __call_result_t<_ImplFn, _Sndr, _Env const &> { - return _ImplFn(static_cast<_Sndr&&>(__sndr), __env); + return _ImplFn()(static_cast<_Sndr&&>(__sndr), __env); } }; diff --git a/include/stdexec/functional.hpp b/include/stdexec/functional.hpp index 50e79cd5b..80e894cf3 100644 --- a/include/stdexec/functional.hpp +++ b/include/stdexec/functional.hpp @@ -357,12 +357,16 @@ namespace STDEXEC struct __back_binder { template - requires __callable<_Fn, _Args..., __copy_cvref_t<_Self, _BoundArgs>...> + requires __callable<__copy_cvref_t<_Self, _Fn>, _Args..., __copy_cvref_t<_Self, _BoundArgs>...> STDEXEC_ATTRIBUTE(host, device) constexpr STDEXEC_EXPLICIT_THIS_BEGIN(auto operator())(this _Self &&__self, - _Args &&...__args) // - noexcept(__nothrow_callable<_Fn, _Args..., __copy_cvref_t<_Self, _BoundArgs>...>) // - -> __call_result_t<_Fn, _Args..., __copy_cvref_t<_Self, _BoundArgs>...> + _Args &&...__args) // + noexcept(__nothrow_callable<__copy_cvref_t<_Self, _Fn>, + _Args..., + __copy_cvref_t<_Self, _BoundArgs>...>) + -> __call_result_t<__copy_cvref_t<_Self, _Fn>, + _Args..., + __copy_cvref_t<_Self, _BoundArgs>...> { return STDEXEC::__apply(static_cast<_Self &&>(__self).__fn_, static_cast<_Self &&>(__self).__bound_args_, @@ -375,11 +379,73 @@ namespace STDEXEC }; template + STDEXEC_ATTRIBUTE(host, device) constexpr auto __bind_back(_Fn &&__fn, _BoundArgs... __bound_args) noexcept(__nothrow_move_constructible<_BoundArgs...> && __nothrow_decay_copyable<_Fn>) { using __binder_t = __back_binder<__decay_t<_Fn>, _BoundArgs...>; return __binder_t{static_cast<_Fn &&>(__fn), static_cast<_BoundArgs &&>(__bound_args)...}; }; + + //! \brief A binary callable that wraps another binary callable and calls it with its + //! two arguments in reverse order. + //! + //! \tparam _Fn The wrapped binary callable. + //! + //! \see __flip + template + struct __flipped + { + //! \brief Calls the wrapped callable with the arguments transposed. + //! + //! The cv- and reference-qualifiers of `*this` are propagated to the wrapped + //! callable, so an rvalue `__flipped` invokes `_Fn` as an rvalue. + //! + //! \tparam _Self The deduced type of `*this`, including cvref-qualifiers. + //! \tparam _T0 The type of the first argument, forwarded as the second. + //! \tparam _T1 The type of the second argument, forwarded as the first. + //! + //! \param __self The `__flipped` object itself. + //! \param __t0 The first argument; forwarded to `_Fn` as its second argument. + //! \param __t1 The second argument; forwarded to `_Fn` as its first argument. + //! + //! \returns The result of `__fn_(__t1, __t0)`. + template + requires __callable<__copy_cvref_t<_Self, _Fn>, _T1, _T0> + STDEXEC_ATTRIBUTE(host, device) + constexpr STDEXEC_EXPLICIT_THIS_BEGIN(auto operator())(this _Self &&__self, + _T0 &&__t0, + _T1 &&__t1) + noexcept(__nothrow_callable<__copy_cvref_t<_Self, _Fn>, _T1, _T0>) + -> __call_result_t<__copy_cvref_t<_Self, _Fn>, _T1, _T0> + { + return static_cast<_Self &&>(__self).__fn_(static_cast<_T1 &&>(__t1), + static_cast<_T0 &&>(__t0)); + } + STDEXEC_EXPLICIT_THIS_END(operator()) + + //! \brief The wrapped binary callable. + _Fn __fn_; + }; + + template + STDEXEC_HOST_DEVICE_DEDUCTION_GUIDE __flipped(_Fn) -> __flipped<_Fn>; + + //! \brief Wraps a binary callable in a `__flipped` object that reverses the order of + //! its two arguments. + //! + //! \tparam _Fn The type of the callable to wrap. + //! + //! \param __fn The binary callable to wrap. It is decay-copied into the result. + //! + //! \returns A `__flipped<__decay_t<_Fn>>` object such that calling it with `(a, b)` + //! calls `__fn(b, a)`. + template + STDEXEC_ATTRIBUTE(host, device) + constexpr auto __flip(_Fn &&__fn) noexcept(__nothrow_decay_copyable<_Fn>) + { + return __flipped{static_cast<_Fn &&>(__fn)}; + } } // namespace STDEXEC + #endif // !STDEXEC_USE_MODULES() || defined(STDEXEC_IN_MODULE_PURVIEW) From 1231f6ec1fad1a7f91340b7a1e976a430c649112 Mon Sep 17 00:00:00 2001 From: Eric Niebler Date: Fri, 7 Aug 2026 11:11:49 -0700 Subject: [PATCH 2/4] try to fix libdispatch_queue build --- include/exec/libdispatch_queue.hpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/include/exec/libdispatch_queue.hpp b/include/exec/libdispatch_queue.hpp index 32c92cdac..e1cee1ef5 100644 --- a/include/exec/libdispatch_queue.hpp +++ b/include/exec/libdispatch_queue.hpp @@ -280,8 +280,9 @@ namespace experimental::execution template Self, class... Env> static consteval auto get_completion_signatures() { + using namespace STDEXEC; return exec::transform_completion_signatures( - STDEXEC::get_completion_signatures, Env...>(), + STDEXEC::get_completion_signatures<__copy_cvref_t, Env...>(), []() { if constexpr (!__decay_copyable) From 4dcc9df3c320fdc9ac3f082796291917e7bd7d17 Mon Sep 17 00:00:00 2001 From: Eric Niebler Date: Fri, 7 Aug 2026 11:16:46 -0700 Subject: [PATCH 3/4] clang-format --- include/exec/env.hpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/include/exec/env.hpp b/include/exec/env.hpp index 595f9ecda..c9c079123 100644 --- a/include/exec/env.hpp +++ b/include/exec/env.hpp @@ -205,7 +205,7 @@ namespace experimental::execution struct __read_with_default_t { template - constexpr auto operator()(_Query, _Default&& __default) const + constexpr auto operator()(_Query, _Default&& __default) const // noexcept(__nothrow_decay_copyable<_Default>) { return __sender<_Query, __decay_t<_Default>>{static_cast<_Default&&>(__default)}; From 2fd76b8755a5318cb5073fe58109d7a752f02954 Mon Sep 17 00:00:00 2001 From: Eric Niebler Date: Fri, 7 Aug 2026 12:14:15 -0700 Subject: [PATCH 4/4] fix libdispatch bulk sender's `get_completion_signatures` --- include/exec/libdispatch_queue.hpp | 26 +++++++++++++++----------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/include/exec/libdispatch_queue.hpp b/include/exec/libdispatch_queue.hpp index e1cee1ef5..462340d6f 100644 --- a/include/exec/libdispatch_queue.hpp +++ b/include/exec/libdispatch_queue.hpp @@ -285,31 +285,35 @@ namespace experimental::execution STDEXEC::get_completion_signatures<__copy_cvref_t, Env...>(), []() { + using value_sig_t = set_value_t(__decay_t...); + using arg_pack_t = __tuple &...>; + // using arg_pack_t = __if_c<__same_as<_AlgoTag, bulk_chunked_t>, + // __tuple, + // __tuple>; if constexpr (!__decay_copyable) { return exec::throw_compile_time_error< _WHAT_(_PREDECESSOR_RESULTS_ARE_NOT_DECAY_COPYABLE_), _WHERE_(_IN_ALGORITHM_, bulk_t), + _WITH_ARGUMENTS_(Args...), _WITH_PRETTY_SENDER_<__copy_cvref_t>, _WITH_ENVIRONMENT_(Env...)>(); } - else if constexpr (!__callable &...>) + else if constexpr (__nothrow_applicable) { - return exec::throw_compile_time_error< - _WHAT_(_FUNCTION_IS_NOT_CALLABLE_WITH_THE_GIVEN_ARGUMENTS_), - _WHERE_(_IN_ALGORITHM_, bulk_t), - _WITH_FUNCTION_(Fun &), - _WITH_ARGUMENTS_(__decay_t & ...)>(); + return completion_signatures(); } - else if constexpr (__nothrow_callable &...> - && __nothrow_decay_copyable) + else if constexpr (__applicable) { - return completion_signatures...)>(); + return completion_signatures(); } else { - return completion_signatures...), - set_error_t(std::exception_ptr)>(); + return STDEXEC::__throw_compile_time_error< + _WHAT_(_FUNCTION_IS_NOT_CALLABLE_WITH_THE_GIVEN_ARGUMENTS_), + _WHERE_(_IN_ALGORITHM_, bulk_t), + _WITH_FUNCTION_(Fun &), + __mapply<__qf<_WITH_ARGUMENTS_>, arg_pack_t>>(); } }); }