diff --git a/include/beman/execution/detail/allocator_of.hpp b/include/beman/execution/detail/allocator_of.hpp deleted file mode 100644 index 4e6d9010..00000000 --- a/include/beman/execution/detail/allocator_of.hpp +++ /dev/null @@ -1,39 +0,0 @@ -// include/beman/execution/detail/allocator_of.hpp -*-C++-*- -// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception - -#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_ALLOCATOR_OF -#define INCLUDED_BEMAN_EXECUTION_DETAIL_ALLOCATOR_OF - -#include -#include -#include - -// ---------------------------------------------------------------------------- - -namespace beman::execution::detail { -/*! - * \brief Utility to get an allocator type from a context - * \headerfile beman/execution/task.hpp - * \internal - */ -template -struct allocator_of { - using type = std::allocator; -}; -template - requires requires { typename Context::allocator_type; } -struct allocator_of { - using type = typename Context::allocator_type; - static_assert( - requires(type& a, std::size_t s, std::byte* ptr) { - { a.allocate(s) } -> std::same_as; - a.deallocate(ptr, s); - }, "The allocator_type needs to be an allocator of std::byte"); -}; -template -using allocator_of_t = typename allocator_of::type; -} // namespace beman::execution::detail - -// ---------------------------------------------------------------------------- - -#endif diff --git a/include/beman/execution/detail/error_types_of.hpp b/include/beman/execution/detail/error_types_of.hpp deleted file mode 100644 index b6e5c0fb..00000000 --- a/include/beman/execution/detail/error_types_of.hpp +++ /dev/null @@ -1,28 +0,0 @@ -// include/beman/execution/detail/error_types_of.hpp -*-C++-*- -// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception - -#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_CONTEXT_ERROR_TYPES_OF -#define INCLUDED_BEMAN_EXECUTION_DETAIL_CONTEXT_ERROR_TYPES_OF - -#include -#include - -// ---------------------------------------------------------------------------- - -namespace beman::execution::detail { -template -struct error_types_of { - using type = ::beman::execution::completion_signatures<::beman::execution::set_error_t(::std::exception_ptr)>; -}; -template - requires requires { typename Context::error_types; } -struct error_types_of { - using type = typename Context::error_types; -}; -template -using error_types_of_t = typename error_types_of::type; -} // namespace beman::execution::detail - -// ---------------------------------------------------------------------------- - -#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_CONTEXT_ERROR_TYPES_OF diff --git a/include/beman/execution/detail/find_allocator.hpp b/include/beman/execution/detail/find_allocator.hpp index 8ace784d..9f94ee31 100644 --- a/include/beman/execution/detail/find_allocator.hpp +++ b/include/beman/execution/detail/find_allocator.hpp @@ -4,39 +4,39 @@ #ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_FIND_ALLOCATOR #define INCLUDED_BEMAN_EXECUTION_DETAIL_FIND_ALLOCATOR +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else #include #include +#endif // ---------------------------------------------------------------------------- namespace beman::execution::detail { -/*! - * \brief Utility locating an allocator_arg/allocator pair - * \headerfile beman/execution/task.hpp - * \internal - */ template -Allocator find_allocator() { +Allocator find_allocator() noexcept { return Allocator(); } template -Allocator find_allocator(const std::allocator_arg_t&) { +Allocator find_allocator(const ::std::allocator_arg_t&) noexcept { static_assert( requires { - { Allocator() } -> std::same_as; + { Allocator() } -> ::std::same_as; }, "There needs to be an allocator argument following std::allocator_arg"); return Allocator(); } template -Allocator find_allocator(const std::allocator_arg_t&, const Alloc& alloc, const A&...) { +Allocator find_allocator(const ::std::allocator_arg_t&, const Alloc& alloc, const A&...) noexcept { static_assert( requires(const Alloc& a) { Allocator(a); }, "The allocator needs to be constructible from the argument following std::allocator"); return Allocator(alloc); } template -Allocator find_allocator(A0 const&, const A&... a) { +Allocator find_allocator(A0 const&, const A&... a) noexcept { return ::beman::execution::detail::find_allocator(a...); } diff --git a/include/beman/execution/detail/handle.hpp b/include/beman/execution/detail/handle.hpp deleted file mode 100644 index 2d09de8a..00000000 --- a/include/beman/execution/detail/handle.hpp +++ /dev/null @@ -1,49 +0,0 @@ -// include/beman/execution/detail/handle.hpp -*-C++-*- -// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception - -#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_HANDLE -#define INCLUDED_BEMAN_EXECUTION_DETAIL_HANDLE - -#include -#include -#include -#include -#include - -// ---------------------------------------------------------------------------- - -namespace beman::execution::detail { -template -class handle { - private: - struct deleter { - auto operator()(P* p) noexcept -> void { - if (p) { - std::coroutine_handle

::from_promise(*p).destroy(); - } - } - }; - std::unique_ptr h; - - public: - explicit handle(P* p) : h(p) {} - auto reset() -> void { this->h.reset(); } - template - auto start(A&&... a) noexcept -> auto { - return this->h->start(::std::forward(a)...); - } - auto release() -> ::std::coroutine_handle

{ - return ::std::coroutine_handle

::from_promise(*this->h.release()); - } - P* get() const noexcept { return this->h.get(); } - auto get_env() const noexcept { - assert(this->h.get()); - return ::beman::execution::get_env(*this->h); - } -}; - -} // namespace beman::execution::detail - -// ---------------------------------------------------------------------------- - -#endif diff --git a/include/beman/execution/detail/meta_index.hpp b/include/beman/execution/detail/meta_index.hpp new file mode 100644 index 00000000..4a810bd8 --- /dev/null +++ b/include/beman/execution/detail/meta_index.hpp @@ -0,0 +1,33 @@ +// include/beman/execution/detail/meta_index.hpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_META_INDEX +#define INCLUDED_BEMAN_EXECUTION_DETAIL_META_INDEX + +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include +#endif + +// ---------------------------------------------------------------------------- + +namespace beman::execution::detail::meta { +template <::std::size_t, typename, typename> +struct index { + static constexpr ::std::size_t value = static_cast<::std::size_t>(-1); +}; +template <::std::size_t I, template class L, typename First, typename... Rest, typename T> +struct index, T> : index, T> {}; +template <::std::size_t I, template class L, typename... Rest, typename T> +struct index, T> { + static constexpr ::std::size_t value = I; +}; +template +inline constexpr ::std::size_t index_v{::beman::execution::detail::meta::index<0uz, List, T>::value}; +} // namespace beman::execution::detail::meta + +// ---------------------------------------------------------------------------- + +#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_META_INDEX diff --git a/include/beman/execution/detail/new_object.hpp b/include/beman/execution/detail/new_object.hpp new file mode 100644 index 00000000..1c903839 --- /dev/null +++ b/include/beman/execution/detail/new_object.hpp @@ -0,0 +1,45 @@ +// include/beman/execution/detail/new_object.hpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#ifndef INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_NEW_OBJECT +#define INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_NEW_OBJECT + +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include +#include +#endif + +// ---------------------------------------------------------------------------- + +namespace beman::execution::detail { +template +[[nodiscard]] auto new_object(const Allocator& allocator, Args&&... args) -> T* { + using alloc_t = typename ::std::allocator_traits::template rebind_alloc; + using alloc_traits = ::std::allocator_traits; + alloc_t alloc(allocator); + auto ptr = alloc_traits::allocate(alloc, 1); + try { + alloc_traits::construct(alloc, ptr, ::std::forward(args)...); + return ptr; + } catch (...) { + alloc_traits::deallocate(alloc, ptr, 1); + throw; + } +} + +template +auto delete_object(const Allocator& allocator, T* ptr) noexcept -> void { + using alloc_t = typename ::std::allocator_traits::template rebind_alloc; + using alloc_traits = ::std::allocator_traits; + alloc_t alloc(allocator); + alloc_traits::destroy(alloc, ptr); + alloc_traits::deallocate(alloc, ptr, 1); +} +} // namespace beman::execution::detail + +// ---------------------------------------------------------------------------- + +#endif // INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_NEW_OBJECT diff --git a/include/beman/execution/detail/parallel_scheduler.hpp b/include/beman/execution/detail/parallel_scheduler.hpp index c1c2b492..2110f523 100644 --- a/include/beman/execution/detail/parallel_scheduler.hpp +++ b/include/beman/execution/detail/parallel_scheduler.hpp @@ -20,280 +20,57 @@ import std; #endif #ifdef BEMAN_HAS_MODULES import beman.execution.detail.bulk; -import beman.execution.detail.connect; -import beman.execution.detail.connect_result_t; import beman.execution.detail.completion_signatures; -import beman.execution.detail.decayed_tuple; import beman.execution.detail.env_of_t; import beman.execution.detail.execution_policy; import beman.execution.detail.forward_like; import beman.execution.detail.get_completion_domain; import beman.execution.detail.get_completion_scheduler; -import beman.execution.detail.get_completion_signatures; import beman.execution.detail.get_env; import beman.execution.detail.get_forward_progress_guarantee; import beman.execution.detail.get_stop_token; import beman.execution.detail.inplace_stop_source; -import beman.execution.detail.meta.combine; -import beman.execution.detail.meta.prepend; -import beman.execution.detail.meta.unique; import beman.execution.detail.operation_state; +import beman.execution.detail.psched_bulk_sender; +import beman.execution.detail.parallel_scheduler_replacement; import beman.execution.detail.receiver; -import beman.execution.detail.scheduler; import beman.execution.detail.scheduler_tag; import beman.execution.detail.sender; -import beman.execution.detail.start; import beman.execution.detail.tag_of_t; import beman.execution.detail.set_error; import beman.execution.detail.set_stopped; import beman.execution.detail.set_value; import beman.execution.detail.stop_propagator; import beman.execution.detail.stop_token_of_t; -import beman.execution.detail.unreachable; -import beman.execution.detail.value_types_of_t; #else #include -#include -#include #include -#include #include #include #include #include #include -#include #include #include #include #include -#include -#include -#include #include +#include +#include #include -#include #include #include #include #include #include -#include #include #include #include -#include -#include #endif // ---------------------------------------------------------------------------- -namespace beman::execution::parallel_scheduler_replacement { - -struct receiver_proxy { - virtual ~receiver_proxy() = default; - - virtual auto set_value() noexcept -> void = 0; - virtual auto set_error(::std::exception_ptr) noexcept -> void = 0; - virtual auto set_stopped() noexcept -> void = 0; - - template - requires(::std::same_as> && ::std::is_object_v

&& !::std::is_array_v

) - auto try_query(Query) const noexcept -> ::std::optional

{ - if constexpr (::std::same_as && - ::std::same_as) { - return query_stop_token(); - } - // TODO: support more queries - return ::std::nullopt; - } - - private: - virtual auto query_stop_token() const noexcept -> ::std::optional<::beman::execution::inplace_stop_token> { - return std::nullopt; - } -}; - -struct bulk_item_receiver_proxy : receiver_proxy { - virtual auto execute(::std::size_t, ::std::size_t) noexcept -> void = 0; -}; - -struct parallel_scheduler_backend { - virtual ~parallel_scheduler_backend() = default; - - virtual auto schedule(receiver_proxy&, ::std::span<::std::byte>) noexcept -> void = 0; - virtual auto schedule_bulk_chunked(::std::size_t, bulk_item_receiver_proxy&, ::std::span<::std::byte>) noexcept - -> void = 0; - virtual auto schedule_bulk_unchunked(::std::size_t, bulk_item_receiver_proxy&, ::std::span<::std::byte>) noexcept - -> void = 0; -}; - -// TODO(P2079R10): provide the project-supported link-time replaceability hook. -auto query_parallel_scheduler_backend() -> ::std::shared_ptr; - -} // namespace beman::execution::parallel_scheduler_replacement - namespace beman::execution::detail { -template -struct psched_bulk_sender { - using sender_concept = ::beman::execution::sender_tag; - - static constexpr bool is_parallel_policy = ::std::same_as || - ::std::same_as; - template - struct rcvr_proxy : ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy { - using result_type = ::beman::execution::detail::meta:: - prepend<::std::monostate, ::beman::execution::value_types_of_t>>; - using stop_token_t = ::beman::execution::stop_token_of_t<::beman::execution::env_of_t>; - - rcvr_proxy(Rcvr rcvr, Policy policy, Shape shape, Fn fn) noexcept - : sp(::beman::execution::get_stop_token(::beman::execution::get_env(rcvr))), - rcvr(::std::move(rcvr)), - policy(::std::move(policy)), - shape(::std::move(shape)), - fn(::std::move(fn)) {} - - auto get_env() const noexcept { return ::beman::execution::get_env(rcvr); } - - auto execute(::std::size_t begin, ::std::size_t end) noexcept -> void final { - const Shape first = is_parallel_policy ? static_cast(begin) : Shape(0); - const Shape last = is_parallel_policy ? static_cast(end) : shape; - const auto call_fn = [=, this](const Args&... args) { - if constexpr (IsChunked) { - fn(first, last, args...); - } else { - for (Shape i = first; i < last; ++i) { - fn(i, args...); - } - } - }; - std::visit( - [&](T& tpl) { - if constexpr (::std::same_as) { - ::beman::execution::detail::unreachable(); - } else { - ::std::apply(call_fn, ::std::move(tpl)); - } - }, - result); - } - - auto set_value() noexcept -> void final { - const auto call_set_value = [this](Args&&... args) { - ::beman::execution::set_value(::std::move(rcvr), ::std::forward(args)...); - }; - try { - std::visit( - [&](T tpl) { - if constexpr (::std::same_as) { - ::beman::execution::detail::unreachable(); - } else { - ::std::apply(call_set_value, ::std::move(tpl)); - } - }, - ::std::move(result)); - } catch (...) { - this->set_error(::std::current_exception()); - } - } - - template - auto set_error(E e) noexcept -> void { - ::beman::execution::set_error(::std::move(rcvr), ::std::move(e)); - } - - auto set_error(::std::exception_ptr e) noexcept -> void final { this->set_error<>(::std::move(e)); } - - auto set_stopped() noexcept -> void final { ::beman::execution::set_stopped(::std::move(rcvr)); } - - private: - auto query_stop_token() const noexcept -> ::std::optional<::beman::execution::inplace_stop_token> final { - return sp.get_token(); - } - - public: - result_type result; - [[no_unique_address]] ::beman::execution::detail::stop_propagator sp; - Rcvr rcvr; - Policy policy; - Shape shape; - Fn fn; - }; - - template - struct state { - using operation_state_concept = ::beman::execution::operation_state_tag; - using result_type = ::beman::execution::detail::meta:: - prepend<::std::monostate, ::beman::execution::value_types_of_t>>; - - struct receiver_ref { - using receiver_concept = ::beman::execution::receiver_tag; - - template - auto set_value(Args&&... args) noexcept -> void try { - using arg_t = ::beman::execution::detail::decayed_tuple; - proxy->result.template emplace(::std::forward(args)...); - auto backend = ::beman::execution::parallel_scheduler_replacement::query_parallel_scheduler_backend(); - if (backend == nullptr) [[unlikely]] { - ::std::terminate(); - } - const ::std::size_t s = is_parallel_policy ? proxy->shape : 1uz; - alignas(void*)::std::byte storage[sizeof(void*) * 4uz]; - if constexpr (IsChunked) { - backend->schedule_bulk_chunked(s, *proxy, storage); - } else { - backend->schedule_bulk_unchunked(s, *proxy, storage); - } - } catch (...) { - proxy->set_error(::std::current_exception()); - } - - template - auto set_error(E e) noexcept -> void { - proxy->set_error(::std::move(e)); - } - - auto set_stopped() noexcept -> void { proxy->set_stopped(); } - - auto get_env() const noexcept { return proxy->get_env(); } - - rcvr_proxy* proxy; - }; - using sub_state_t = ::beman::execution::connect_result_t; - - state(Child child, Rcvr rcvr, Policy policy, Shape shape, Fn fn) - : proxy(::std::move(rcvr), ::std::move(policy), ::std::move(shape), ::std::move(fn)), - sub_state(::beman::execution::connect(::std::move(child), receiver_ref{&proxy})) {} - - auto start() & noexcept -> void { ::beman::execution::start(sub_state); } - - rcvr_proxy proxy; - sub_state_t sub_state; - }; - - template - static consteval auto get_completion_signatures() { - constexpr auto compl_sigs = ::beman::execution::get_completion_signatures(); - return ::beman::execution::detail::meta::unique<::beman::execution::detail::meta::combine< - ::std::remove_cvref_t, - ::beman::execution::completion_signatures<::beman::execution::set_error_t(::std::exception_ptr)>>>{}; - } - - template - auto connect(Rcvr rcvr) && noexcept { - return state{ - ::std::move(child), ::std::move(rcvr), ::std::move(policy), ::std::move(shape), ::std::move(fn)}; - } - - auto get_env() const noexcept { return ::beman::execution::get_env(child); } - - [[no_unique_address]] ::std::bool_constant _; - Policy policy; - Shape shape; - Fn fn; - Child child; -}; - struct parallel_scheduler_domain { template requires ::std::same_as<::beman::execution::tag_of_t, ::beman::execution::bulk_chunked_t> || @@ -304,6 +81,7 @@ struct parallel_scheduler_domain { return ::beman::execution::detail::psched_bulk_sender{ ::std::bool_constant< ::std::same_as<::beman::execution::tag_of_t, ::beman::execution::bulk_chunked_t>>{}, + ::beman::execution::parallel_scheduler_replacement::query_parallel_scheduler_backend(), ::std::move(policy), ::std::move(shape), ::std::move(fn), @@ -425,7 +203,8 @@ class parallel_scheduler::sender { private: ::std::shared_ptr backend_; rcvr_proxy<::std::remove_cvref_t> proxy_; - alignas(void*)::std::byte storage_[sizeof(void*) * 4]{}; + alignas(::beman::execution::detail::psched_storage_alignment)::std::byte + storage_[::beman::execution::detail::psched_storage_size]; }; explicit sender(::std::shared_ptr backend) noexcept : backend_(::std::move(backend)) {} diff --git a/include/beman/execution/detail/parallel_scheduler_replacement.hpp b/include/beman/execution/detail/parallel_scheduler_replacement.hpp new file mode 100644 index 00000000..b1951b31 --- /dev/null +++ b/include/beman/execution/detail/parallel_scheduler_replacement.hpp @@ -0,0 +1,77 @@ +// include/beman/execution/detail/parallel_scheduler_replacement.hpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_PARALLEL_SCHEDULER_REPLACEMENT +#define INCLUDED_BEMAN_EXECUTION_DETAIL_PARALLEL_SCHEDULER_REPLACEMENT + +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include +#include +#include +#include +#include +#include +#include +#include +#endif +#ifdef BEMAN_HAS_MODULES +import beman.execution.detail.get_stop_token; +import beman.execution.detail.inplace_stop_source; +#else +#include +#include +#endif + +// ---------------------------------------------------------------------------- + +namespace beman::execution::parallel_scheduler_replacement { + +struct receiver_proxy { + virtual ~receiver_proxy() = default; + + virtual auto set_value() noexcept -> void = 0; + virtual auto set_error(::std::exception_ptr) noexcept -> void = 0; + virtual auto set_stopped() noexcept -> void = 0; + + template + requires(::std::same_as> && ::std::is_object_v

&& !::std::is_array_v

) + auto try_query(Query) const noexcept -> ::std::optional

{ + if constexpr (::std::same_as && + ::std::same_as) { + return query_stop_token(); + } + // TODO: support more queries + return ::std::nullopt; + } + + private: + virtual auto query_stop_token() const noexcept -> ::std::optional<::beman::execution::inplace_stop_token> { + return std::nullopt; + } +}; + +struct bulk_item_receiver_proxy : receiver_proxy { + virtual auto execute(::std::size_t, ::std::size_t) noexcept -> void = 0; +}; + +struct parallel_scheduler_backend { + virtual ~parallel_scheduler_backend() = default; + + virtual auto schedule(receiver_proxy&, ::std::span<::std::byte>) noexcept -> void = 0; + virtual auto schedule_bulk_chunked(::std::size_t, bulk_item_receiver_proxy&, ::std::span<::std::byte>) noexcept + -> void = 0; + virtual auto schedule_bulk_unchunked(::std::size_t, bulk_item_receiver_proxy&, ::std::span<::std::byte>) noexcept + -> void = 0; +}; + +// TODO(P2079R10): provide the project-supported link-time replaceability hook. +auto query_parallel_scheduler_backend() -> ::std::shared_ptr; + +} // namespace beman::execution::parallel_scheduler_replacement + +// ---------------------------------------------------------------------------- + +#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_PARALLEL_SCHEDULER_REPLACEMENT diff --git a/include/beman/execution/detail/psched_bulk_sender.hpp b/include/beman/execution/detail/psched_bulk_sender.hpp new file mode 100644 index 00000000..19360fda --- /dev/null +++ b/include/beman/execution/detail/psched_bulk_sender.hpp @@ -0,0 +1,250 @@ +// include/beman/execution/detail/psched_bulk_sender.hpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_PSCHED_BULK_SENDER +#define INCLUDED_BEMAN_EXECUTION_DETAIL_PSCHED_BULK_SENDER + +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include +#include +#include +#include +#include +#include +#include +#include +#include +#endif +#ifdef BEMAN_HAS_MODULES +import beman.execution.detail.connect; +import beman.execution.detail.connect_result_t; +import beman.execution.detail.completion_signatures; +import beman.execution.detail.decayed_tuple; +import beman.execution.detail.env_of_t; +import beman.execution.detail.execution_policy; +import beman.execution.detail.get_completion_signatures; +import beman.execution.detail.get_env; +import beman.execution.detail.get_stop_token; +import beman.execution.detail.inplace_stop_source; +import beman.execution.detail.meta.combine; +import beman.execution.detail.meta.prepend; +import beman.execution.detail.meta.unique; +import beman.execution.detail.operation_state; +import beman.execution.detail.parallel_scheduler_replacement; +import beman.execution.detail.receiver; +import beman.execution.detail.sender; +import beman.execution.detail.start; +import beman.execution.detail.set_error; +import beman.execution.detail.set_stopped; +import beman.execution.detail.set_value; +import beman.execution.detail.stop_propagator; +import beman.execution.detail.stop_token_of_t; +import beman.execution.detail.unreachable; +import beman.execution.detail.value_types_of_t; +#else +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#endif + +// ---------------------------------------------------------------------------- + +namespace beman::execution::detail { +inline constexpr ::std::size_t psched_storage_alignment = alignof(void*); +inline constexpr ::std::size_t psched_storage_size = 6uz * sizeof(void*); +template +struct psched_bulk_sender { + using sender_concept = ::beman::execution::sender_tag; + using backend_type = ::beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend; + + static constexpr bool is_parallel_policy = ::std::same_as || + ::std::same_as; + template + struct rcvr_proxy : ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy { + using result_type = ::beman::execution::detail::meta:: + prepend<::std::monostate, ::beman::execution::value_types_of_t>>; + using stop_token_t = ::beman::execution::stop_token_of_t<::beman::execution::env_of_t>; + + rcvr_proxy(::std::shared_ptr backend, Rcvr rcvr, Policy policy, Shape shape, Fn fn) noexcept + : backend(::std::move(backend)), + sp(::beman::execution::get_stop_token(::beman::execution::get_env(rcvr))), + rcvr(::std::move(rcvr)), + policy(::std::move(policy)), + shape(::std::move(shape)), + fn(::std::move(fn)) {} + + auto get_env() const noexcept { return ::beman::execution::get_env(rcvr); } + + auto execute(::std::size_t begin, ::std::size_t end) noexcept -> void final { + const Shape first = is_parallel_policy ? static_cast(begin) : Shape(0); + const Shape last = is_parallel_policy ? static_cast(end) : shape; + const auto call_fn = [=, this](const Args&... args) { + if constexpr (IsChunked) { + fn(first, last, args...); + } else { + for (Shape i = first; i < last; ++i) { + fn(i, args...); + } + } + }; + ::std::visit( + [&](T& tpl) { + if constexpr (::std::same_as) { + ::beman::execution::detail::unreachable(); + } else { + ::std::apply(call_fn, ::std::move(tpl)); + } + }, + result); + } + + auto set_value() noexcept -> void final { + const auto call_set_value = [this](Args&&... args) { + ::beman::execution::set_value(::std::move(rcvr), ::std::forward(args)...); + }; + try { + ::std::visit( + [&](T tpl) { + if constexpr (::std::same_as) { + ::beman::execution::detail::unreachable(); + } else { + ::std::apply(call_set_value, ::std::move(tpl)); + } + }, + ::std::move(result)); + } catch (...) { + this->set_error(::std::current_exception()); + } + } + + template + auto set_error(E e) noexcept -> void { + ::beman::execution::set_error(::std::move(rcvr), ::std::move(e)); + } + + auto set_error(::std::exception_ptr e) noexcept -> void final { this->set_error<>(::std::move(e)); } + + auto set_stopped() noexcept -> void final { ::beman::execution::set_stopped(::std::move(rcvr)); } + + private: + auto query_stop_token() const noexcept -> ::std::optional<::beman::execution::inplace_stop_token> final { + return sp.get_token(); + } + + public: + alignas(psched_storage_alignment)::std::byte storage[psched_storage_size]; + ::std::shared_ptr backend; + result_type result; + [[no_unique_address]] ::beman::execution::detail::stop_propagator sp; + Rcvr rcvr; + Policy policy; + Shape shape; + Fn fn; + }; + + template + struct state { + using operation_state_concept = ::beman::execution::operation_state_tag; + using result_type = ::beman::execution::detail::meta:: + prepend<::std::monostate, ::beman::execution::value_types_of_t>>; + + struct receiver_ref { + using receiver_concept = ::beman::execution::receiver_tag; + + template + auto set_value(Args&&... args) noexcept -> void try { + using arg_t = ::beman::execution::detail::decayed_tuple; + proxy->result.template emplace(::std::forward(args)...); + if (!proxy->backend) [[unlikely]] { + ::std::terminate(); + } + const ::std::size_t s = is_parallel_policy ? proxy->shape : 1uz; + if constexpr (IsChunked) { + proxy->backend->schedule_bulk_chunked(s, *proxy, proxy->storage); + } else { + proxy->backend->schedule_bulk_unchunked(s, *proxy, proxy->storage); + } + } catch (...) { + proxy->set_error(::std::current_exception()); + } + + template + auto set_error(E e) noexcept -> void { + proxy->set_error(::std::move(e)); + } + + auto set_stopped() noexcept -> void { proxy->set_stopped(); } + + auto get_env() const noexcept { return proxy->get_env(); } + + rcvr_proxy* proxy; + }; + using sub_state_t = ::beman::execution::connect_result_t; + + state(::std::shared_ptr backend, Child child, Rcvr rcvr, Policy policy, Shape shape, Fn fn) + : proxy(std::move(backend), ::std::move(rcvr), ::std::move(policy), ::std::move(shape), ::std::move(fn)), + sub_state(::beman::execution::connect(::std::move(child), receiver_ref{&proxy})) {} + + auto start() & noexcept -> void { ::beman::execution::start(sub_state); } + + rcvr_proxy proxy; + sub_state_t sub_state; + }; + + template + static consteval auto get_completion_signatures() { + constexpr auto compl_sigs = ::beman::execution::get_completion_signatures(); + return ::beman::execution::detail::meta::unique<::beman::execution::detail::meta::combine< + ::std::remove_cvref_t, + ::beman::execution::completion_signatures<::beman::execution::set_error_t(::std::exception_ptr)>>>{}; + } + + template + auto connect(Rcvr rcvr) && noexcept { + return state{::std::move(backend), + ::std::move(child), + ::std::move(rcvr), + ::std::move(policy), + ::std::move(shape), + ::std::move(fn)}; + } + + auto get_env() const noexcept { return ::beman::execution::get_env(child); } + + [[no_unique_address]] ::std::bool_constant _; + ::std::shared_ptr backend; + Policy policy; + Shape shape; + Fn fn; + Child child; +}; +} // namespace beman::execution::detail + +// ---------------------------------------------------------------------------- + +#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_PSCHED_BULK_SENDER diff --git a/include/beman/execution/detail/spawn_future.hpp b/include/beman/execution/detail/spawn_future.hpp index af2e02ec..df9aa835 100644 --- a/include/beman/execution/detail/spawn_future.hpp +++ b/include/beman/execution/detail/spawn_future.hpp @@ -156,22 +156,22 @@ struct spawn_future_state using traits_t = ::std::allocator_traits; using spawned_sender_t = ::beman::execution::detail::future_spawned_sender; using sigs_t = ::beman::execution::detail::spawn_future_sigs; - using receiver_tag = ::beman::execution::detail::spawn_future_receiver; + using receiver_t = ::beman::execution::detail::spawn_future_receiver; static_assert(::beman::execution::sender); - static_assert(::beman::execution::receiver); - using op_t = ::beman::execution::connect_result_t; + static_assert(::beman::execution::receiver); + using op_t = ::beman::execution::connect_result_t; template <::beman::execution::sender S> spawn_future_state(auto a, S&& s, const Token& tok, Env env) : alloc(::std::move(a)), op(::beman::execution::write_env( ::beman::execution::detail::stop_when(::std::forward(s), source.get_token()), env), - receiver_tag{this}), + receiver_t{this}), assoc(tok.try_associate()) { if (this->assoc) { ::beman::execution::start(this->op); } else { - ::beman::execution::set_stopped(receiver_tag{this}); + ::beman::execution::set_stopped(receiver_t{this}); } } auto complete() noexcept -> void override { diff --git a/include/beman/execution/detail/state_rep.hpp b/include/beman/execution/detail/state_rep.hpp deleted file mode 100644 index da25b7e9..00000000 --- a/include/beman/execution/detail/state_rep.hpp +++ /dev/null @@ -1,49 +0,0 @@ -// include/beman/execution/detail/state_rep.hpp -*-C++-*- -// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception - -#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_STATE_REP -#define INCLUDED_BEMAN_EXECUTION_DETAIL_STATE_REP - -#include -#include -#include - -// ---------------------------------------------------------------------------- - -namespace beman::execution::detail { -template -struct state_rep { - std::remove_cvref_t receiver; - C context; - template - state_rep(R&& r) : receiver(std::forward(r)), context() {} -}; -template - requires requires { C(::beman::execution::get_env(std::declval&>())); } && - (not requires(const Receiver& receiver) { - typename C::template env_type; - }) -struct state_rep { - std::remove_cvref_t receiver; - C context; - template - state_rep(R&& r) : receiver(std::forward(r)), context(::beman::execution::get_env(this->receiver)) {} -}; -template - requires requires(const Receiver& receiver) { - typename C::template env_type; - } -struct state_rep { - using upstream_env = decltype(::beman::execution::get_env(std::declval&>())); - std::remove_cvref_t receiver; - typename C::template env_type own_env; - C context; - template - state_rep(R&& r) - : receiver(std::forward(r)), own_env(::beman::execution::get_env(this->receiver)), context(this->own_env) {} -}; -} // namespace beman::execution::detail - -// ---------------------------------------------------------------------------- - -#endif diff --git a/include/beman/execution/detail/stop_source_of.hpp b/include/beman/execution/detail/stop_source_of.hpp deleted file mode 100644 index 8c8b8de8..00000000 --- a/include/beman/execution/detail/stop_source_of.hpp +++ /dev/null @@ -1,27 +0,0 @@ -// include/beman/execution/detail/stop_source_of.hpp -*-C++-*- -// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception - -#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_STOP_SOURCE_OF -#define INCLUDED_BEMAN_EXECUTION_DETAIL_STOP_SOURCE_OF - -#include - -// ---------------------------------------------------------------------------- - -namespace beman::execution::detail { -template -struct stop_source_of { - using type = ::beman::execution::inplace_stop_source; -}; -template - requires requires { typename Context::stop_source_type; } -struct stop_source_of { - using type = typename Context::stop_source_type; -}; -template -using stop_source_of_t = typename stop_source_of::type; -} // namespace beman::execution::detail - -// ---------------------------------------------------------------------------- - -#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_STOP_SOURCE_OF diff --git a/include/beman/execution/detail/sub_visit.hpp b/include/beman/execution/detail/sub_visit.hpp index 19b9cb59..f46d0d31 100644 --- a/include/beman/execution/detail/sub_visit.hpp +++ b/include/beman/execution/detail/sub_visit.hpp @@ -1,37 +1,29 @@ // include/beman/execution/detail/sub_visit.hpp -*-C++-*- -// ---------------------------------------------------------------------------- // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception -// ---------------------------------------------------------------------------- #ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_SUB_VISIT #define INCLUDED_BEMAN_EXECUTION_DETAIL_SUB_VISIT - +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include #include #include -#include +#endif // ---------------------------------------------------------------------------- namespace beman::execution::detail { -/* - * \brief Helper function creatig thunks for a variant visit. - * \headerfile beman/execution/task.hpp - * \internal - */ -template -void sub_visit_thunks(Fun& fun, Var& var, std::index_sequence) { +template <::std::size_t Start, typename Fun, typename Var, ::std::size_t... I> +auto sub_visit_thunks(Fun& fun, Var& var, ::std::index_sequence) -> void { using thunk_t = void (*)(Fun&, Var&); - static constexpr thunk_t thunks[]{(+[](Fun& f, Var& v) { f(std::get(v)); })...}; + static constexpr thunk_t thunks[]{(+[](Fun& f, Var& v) { f(::std::get(v)); })...}; thunks[var.index() - Start](fun, var); } -/* - * \brief Helper function visiting a suffix of variant options - * \headerfile beman/execution/task.hpp - * \internal - */ -template -void sub_visit(auto&& fun, std::variant& v) { +template <::std::size_t Start, typename... T> +auto sub_visit(const auto& fun, std::variant& v) -> void { if (v.index() < Start) return; sub_visit_thunks(fun, v, std::make_index_sequence{}); @@ -41,4 +33,4 @@ void sub_visit(auto&& fun, std::variant& v) { // ---------------------------------------------------------------------------- -#endif +#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_SUB_VISIT diff --git a/include/beman/execution/detail/task.hpp b/include/beman/execution/detail/task.hpp new file mode 100644 index 00000000..49976b11 --- /dev/null +++ b/include/beman/execution/detail/task.hpp @@ -0,0 +1,808 @@ +// include/beman/execution/detail/task.hpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_TASK +#define INCLUDED_BEMAN_EXECUTION_DETAIL_TASK + +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#endif +#ifdef BEMAN_HAS_MODULES +import beman.execution.detail.affine; +import beman.execution.detail.as_awaitable; +import beman.execution.detail.completion_signatures; +import beman.execution.detail.decayed_same_as; +import beman.execution.detail.env; +import beman.execution.detail.env_of_t; +import beman.execution.detail.find_allocator; +import beman.execution.detail.forwarding_query; +import beman.execution.detail.get_allocator; +import beman.execution.detail.get_env; +import beman.execution.detail.get_scheduler; +import beman.execution.detail.get_start_scheduler; +import beman.execution.detail.get_stop_token; +import beman.execution.detail.inplace_stop_source; +import beman.execution.detail.meta.combine; +import beman.execution.detail.meta.contains; +import beman.execution.detail.meta.index; +import beman.execution.detail.operation_state; +import beman.execution.detail.receiver; +import beman.execution.detail.scheduler; +import beman.execution.detail.sender; +import beman.execution.detail.set_error; +import beman.execution.detail.set_stopped; +import beman.execution.detail.set_value; +import beman.execution.detail.simple_allocator; +import beman.execution.detail.stoppable_source; +import beman.execution.detail.stoppable_token; +import beman.execution.detail.stop_callback_for_t; +import beman.execution.detail.stop_token_of_t; +import beman.execution.detail.sub_visit; +import beman.execution.detail.task_scheduler; +import beman.execution.detail.type_list; +import beman.execution.detail.with_error; +import beman.execution.detail.unreachable; +#else +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#endif + +// ---------------------------------------------------------------------------- + +namespace beman::execution::detail::task { +template +class handle { + private: + struct deleter { + auto operator()(Promise* p) noexcept -> void { + if (p) { + ::std::coroutine_handle::from_promise(*p).destroy(); + } + } + }; + ::std::unique_ptr h; + + public: + explicit handle(Promise* promise) noexcept : h(promise) {} + + auto reset() noexcept -> void { this->h.reset(); } + + template + auto start(A&&... a) noexcept { + return this->h->start(::std::forward(a)...); + } + + auto release() noexcept -> ::std::coroutine_handle { + return ::std::coroutine_handle::from_promise(*this->h.release()); + } + + auto get() const noexcept -> Promise* { return this->h.get(); } + + auto get_env() const noexcept { return ::beman::execution::get_env(*this->h); } +}; + +template +struct allocator_support { + using allocator_traits = ::std::allocator_traits; + + static auto offset(::std::size_t size) noexcept -> ::std::size_t { + return (size + alignof(Allocator) - 1u) & ~(alignof(Allocator) - 1u); + } + + static auto get_allocator(void* ptr, ::std::size_t size) noexcept -> Allocator* { + ptr = static_cast<::std::byte*>(ptr) + offset(size); + return ::std::launder(reinterpret_cast(ptr)); + } + + template + auto operator new(std::size_t size, const Args&... args) -> void* { + if constexpr (::std::same_as>) { + Allocator alloc{}; + return allocator_traits::allocate(alloc, size); + } else { + Allocator alloc{::beman::execution::detail::find_allocator(args...)}; + void* ptr{allocator_traits::allocate(alloc, allocator_support::offset(size) + sizeof(Allocator))}; + try { + new (allocator_support::get_allocator(ptr, size)) Allocator(alloc); + } catch (...) { + allocator_traits::deallocate( + alloc, static_cast<::std::byte*>(ptr), allocator_support::offset(size) + sizeof(Allocator)); + throw; + } + return ptr; + } + } + + auto operator delete(void* ptr, ::std::size_t size) noexcept -> void { + if constexpr (::std::same_as>) { + Allocator alloc{}; + allocator_traits::deallocate(alloc, static_cast<::std::byte*>(ptr), size); + } else { + Allocator* aptr{allocator_support::get_allocator(ptr, size)}; + Allocator alloc{*aptr}; + aptr->~Allocator(); + allocator_traits::deallocate( + alloc, static_cast<::std::byte*>(ptr), allocator_support::offset(size) + sizeof(Allocator)); + } + } + + auto operator delete(void* ptr, ::std::size_t size, const auto&...) noexcept -> void { + allocator_support::operator delete(ptr, size); + } +}; + +template +struct allocator_of { + using type = ::std::allocator<::std::byte>; +}; +template + requires requires { typename Context::allocator_type; } +struct allocator_of { + using alloc_t = typename Context::allocator_type; + static_assert(::beman::execution::detail::simple_allocator, + "The type alias allocator_type needs to refer to an allocator"); + using type = ::std::allocator_traits::template rebind_alloc<::std::byte>; +}; +template +using allocator_of_t = typename allocator_of::type; + +template +struct value_signature { + using type = ::beman::execution::set_value_t(R); +}; +template <> +struct value_signature { + using type = ::beman::execution::set_value_t(); +}; + +template +using value_signature_t = typename value_signature::type; + +template +struct error_types_of { + using type = ::beman::execution::completion_signatures<::beman::execution::set_error_t(::std::exception_ptr)>; +}; +template + requires requires { typename Context::error_types; } +struct error_types_of { + using type = typename Context::error_types; +}; +template +using error_types_of_t = typename error_types_of::type; + +template +struct has_exception_signature; + +template +struct has_exception_signature<::beman::execution::completion_signatures> + : ::std::bool_constant< + ::beman::execution::detail::meta::contains<::beman::execution::set_error_t(::std::exception_ptr), Sigs...>> { +}; + +template +struct start_scheduler_of { + using type = ::beman::execution::task_scheduler; +}; +template + requires requires { typename Context::start_scheduler_type; } +struct start_scheduler_of { + using type = typename Context::start_scheduler_type; + static_assert(::beman::execution::scheduler, + "The type alias start_scheduler_type needs to refer to a scheduler"); +}; +template +using start_scheduler_of_t = typename start_scheduler_of::type; + +template +struct stop_source_of { + using type = ::beman::execution::inplace_stop_source; +}; +template + requires requires { typename Context::stop_source_type; } +struct stop_source_of { + using type = typename Context::stop_source_type; + static_assert(::beman::execution::detail::stoppable_source, + "The type alias stop_source_type needs to refer to a stoppable-source"); +}; +template +using stop_source_of_t = typename stop_source_of::type; + +struct void_type {}; + +template +class result_type; + +template +class result_type> { + private: + using value_type = ::std::conditional_t<::std::same_as, void_type, Value>; + + public: + template + auto set_value(T&& value) -> void { + this->result.template emplace<1uz>(::std::forward(value)); + } + + template + auto set_error(E&& error) -> void { + static constexpr ::std::size_t index = ::beman::execution::detail::meta:: + index_v<::beman::execution::detail::type_list<::std::remove_cvref_t...>, ::std::remove_cvref_t>; + static_assert(index != static_cast<::std::size_t>(-1), + "Error type not supported: 'E' is not among the defined Errors."); + this->result.template emplace<2uz + index>(::std::forward(error)); + } + + auto no_completion_set() const noexcept -> bool { return this->result.index() == 0uz; } + + template <::beman::execution::receiver Receiver> + auto result_complete(Receiver rcvr) noexcept -> void { + switch (this->result.index()) { + case 0: + ::beman::execution::set_stopped(::std::move(rcvr)); + break; + case 1: + if constexpr (::std::same_as) + ::beman::execution::set_value(::std::move(rcvr)); + else + ::beman::execution::set_value(::std::move(rcvr), ::std::move(::std::get<1uz>(this->result))); + break; + default: + if constexpr (0uz < sizeof...(Errors)) { + ::beman::execution::detail::sub_visit<2uz>( + [&rcvr](auto& error) { ::beman::execution::set_error(::std::move(rcvr), ::std::move(error)); }, + this->result); + } + break; + } + } + auto result_resume() { + switch (this->result.index()) { + case 0: + ::beman::execution::detail::unreachable(); + break; + case 1: + break; + default: + if constexpr (0uz < sizeof...(Errors)) + ::beman::execution::detail::sub_visit<2uz>( + [](E& error) { + if constexpr (::std::same_as<::std::remove_cvref_t, ::std::exception_ptr>) { + std::rethrow_exception(::std::move(error)); + } else if constexpr (::std::same_as<::std::remove_cvref_t, ::std::error_code>) { + throw ::std::system_error{::std::move(error)}; + } else { + throw ::std::move(error); + } + }, + this->result); + ::beman::execution::detail::unreachable(); + break; + } + if constexpr (::std::same_as) { + return; + } else { + return ::std::move(::std::get<1uz>(this->result)); + } + } + + private: + ::std::variant<::std::monostate, value_type, ::std::remove_cvref_t...> result; +}; +template +class result_type> { + private: + using value_type = ::std::conditional_t<::std::same_as, void_type, Value>; + + ::std::variant result; + + public: + template + auto set_value(T&& value) -> void { + this->result.template emplace<1uz>(::std::forward(value)); + } + + auto no_completion_set() const noexcept -> bool { return this->result.index() == 0; } + + template <::beman::execution::receiver Receiver> + auto result_complete(Receiver rcvr) noexcept -> void { + switch (this->result.index()) { + case 0: + ::beman::execution::set_stopped(::std::move(rcvr)); + break; + case 1: + if constexpr (::std::same_as) { + ::beman::execution::set_value(::std::move(rcvr)); + } else { + ::beman::execution::set_value(::std::move(rcvr), ::std::move(::std::get<1uz>(this->result))); + } + break; + default: + ::beman::execution::detail::unreachable(); + break; + } + } + + auto result_resume() noexcept { + if (this->result.index() != 1) [[unlikely]] { + ::beman::execution::detail::unreachable(); + } + if constexpr (::std::same_as) { + return; + } else { + return ::std::move(::std::get<1uz>(this->result)); + } + } +}; + +template +struct own_env { + using type = ::beman::execution::env<>; +}; + +template + requires requires { typename Environment::template env_type; } +struct own_env { + using type = Environment::template env_type; +}; + +template +struct env_holder { + using own_env_t = typename own_env::type; + + static auto make_own_env(const RcvrEnv& rcvr_env) noexcept -> own_env_t { + if constexpr (::std::constructible_from) { + return own_env_t(rcvr_env); + } else { + return own_env_t(); + } + } + + static auto make_environment(const RcvrEnv& rcvr_env, const own_env_t& own_env) noexcept -> Environment { + if constexpr (::std::constructible_from) { + return Environment(own_env); + } else if constexpr (::std::constructible_from) { + return Environment(rcvr_env); + } else { + return Environment(); + } + } + + explicit env_holder(const RcvrEnv& rcvr_env) noexcept + : own_env(env_holder::make_own_env(rcvr_env)), environment(env_holder::make_environment(rcvr_env, own_env)) {} + + own_env_t own_env; + Environment environment; +}; + +template +class propagator { + public: + template + requires ::std::constructible_from + explicit propagator(StopToken stop_token, Args&&... args) + : stop_source_(::std::forward(args)...), + stop_callback_(::std::move(stop_token), ::std::bind_front(&propagator::propagate, this)) {} + + auto get_token() const noexcept { return stop_source_.get_token(); } + + private: + auto propagate() -> void { stop_source_.request_stop(); } + + private: + using stop_cb_t = decltype(::std::bind_front(&propagator::propagate, ::std::declval())); + + StopSource stop_source_; + ::beman::execution::stop_callback_for_t stop_callback_; +}; + +template + requires ::std::same_as<::std::decay_t().get_token())>, StopToken> +class propagator { + public: + explicit propagator(StopToken stop_token) noexcept : stop_token_(::std::move(stop_token)) {} + + auto get_token() const noexcept -> StopToken { return stop_token_; } + + private: + StopToken stop_token_; +}; + +template +// ReSharper disable once CppPolymorphicClassWithNonVirtualPublicDestructor +class state_base : public result_type> { + public: + using allocator_type = allocator_of_t; + using stop_source_type = stop_source_of_t; + using stop_token_type = decltype(::std::declval().get_token()); + using scheduler_type = start_scheduler_of_t; + + state_base() = default; + + state_base(const state_base&) = delete; + + state_base(state_base&&) = delete; + + ~state_base() = default; + + auto operator=(const state_base&) -> state_base& = delete; + + auto operator=(state_base&&) -> state_base& = delete; + + auto complete() noexcept -> std::coroutine_handle<> { return this->do_complete(); } + auto get_allocator() noexcept -> allocator_type { return this->do_get_allocator(); } + auto get_stop_token() noexcept -> stop_token_type { return this->do_get_stop_token(); } + auto get_environment() noexcept -> Environment& { return this->do_get_environment(); } + auto get_start_scheduler() noexcept -> scheduler_type { return this->do_get_start_scheduler(); } + + protected: + template + static auto make_sched(const Env& env, const Alloc& alloc) noexcept -> scheduler_type { + if constexpr (requires { + ::std::make_obj_using_allocator( + alloc, ::beman::execution::get_start_scheduler(env)); + }) { + return ::std::make_obj_using_allocator(alloc, + ::beman::execution::get_start_scheduler(env)); + } else if constexpr (requires { + ::std::make_obj_using_allocator( + alloc, ::beman::execution::get_scheduler(env)); + }) { + return ::std::make_obj_using_allocator(alloc, ::beman::execution::get_scheduler(env)); + } else { + return scheduler_type(); + } + } + + private: + virtual auto do_complete() noexcept -> ::std::coroutine_handle<> = 0; + virtual auto do_get_allocator() noexcept -> allocator_type = 0; + virtual auto do_get_stop_token() noexcept -> stop_token_type = 0; + virtual auto do_get_environment() noexcept -> Environment& = 0; + virtual auto do_get_start_scheduler() noexcept -> scheduler_type = 0; +}; + +template +// ReSharper disable once CppPolymorphicClassWithNonVirtualPublicDestructor +struct state : state_base { + using operation_state_concept = ::beman::execution::operation_state_tag; + using scheduler_type = typename state_base::scheduler_type; + using allocator_type = typename state_base::allocator_type; + using stop_source_type = typename state_base::stop_source_type; + using stop_token_type = typename state_base::stop_token_type; + using upstream_stop_token_t = ::beman::execution::stop_token_of_t<::beman::execution::env_of_t>; + + template + state(Receiver r, H h, const Env& env) noexcept + : rcvr(::std::move(r)), + handle(std::move(h)), + holder(env), + scheduler(this->make_sched(env, do_get_allocator())), + propagator(::beman::execution::get_stop_token(env)) {} + + auto start() & noexcept -> void { this->handle.start(this).resume(); } + + private: + auto do_complete() noexcept -> ::std::coroutine_handle<> final { + this->handle.reset(); + this->result_complete(::std::move(this->rcvr)); + return ::std::noop_coroutine(); + } + + auto do_get_allocator() noexcept -> allocator_type final { + if constexpr (requires { + allocator_type(::beman::execution::get_allocator(::beman::execution::get_env(this->rcvr))); + }) { + return allocator_type(::beman::execution::get_allocator(::beman::execution::get_env(this->rcvr))); + } else { + return allocator_type(); + } + } + + auto do_get_start_scheduler() noexcept -> scheduler_type final { return this->scheduler; } + + auto do_get_stop_token() noexcept -> stop_token_type final { return this->propagator.get_token(); } + + auto do_get_environment() noexcept -> Environment& final { return this->holder.environment; } + + Receiver rcvr; + ::beman::execution::detail::task::handle handle; + env_holder> holder; + scheduler_type scheduler; + ::beman::execution::detail::task::propagator propagator; +}; + +template +// ReSharper disable once CppPolymorphicClassWithNonVirtualPublicDestructor +class awaiter : public state_base { + public: + using allocator_type = typename state_base::allocator_type; + using scheduler_type = typename state_base::scheduler_type; + using stop_source_type = typename state_base::stop_source_type; + using stop_token_type = typename state_base::stop_token_type; + using upstream_stop_token_t = ::beman::execution::stop_token_of_t<::beman::execution::env_of_t>; + + explicit awaiter(::std::coroutine_handle parent, handle h, const auto& env) noexcept + : parent(parent), + handle(std::move(h)), + holder(env), + scheduler(this->make_sched(env, do_get_allocator())), + propagator(::beman::execution::get_stop_token(env)) {} + + static constexpr auto await_ready() noexcept -> bool { return false; } + + auto await_suspend([[maybe_unused]] ::std::coroutine_handle p) noexcept + -> ::std::coroutine_handle<> { + assert(p == this->parent); + return this->handle.start(this); + } + + auto await_resume() -> Value { return this->result_resume(); } + + private: + auto do_complete() noexcept -> std::coroutine_handle<> final { + if (this->no_completion_set()) { + return this->parent.promise().unhandled_stopped(); + } + return this->parent; + } + + auto do_get_allocator() noexcept -> allocator_type final { + if constexpr (requires { + ::beman::execution::get_allocator(::beman::execution::get_env(this->parent.promise())); + }) { + return ::beman::execution::get_allocator(::beman::execution::get_env(this->parent.promise())); + } else { + return allocator_type(); + } + } + + auto do_get_start_scheduler() noexcept -> scheduler_type final { return this->scheduler; } + + auto do_get_stop_token() noexcept -> stop_token_type final { return this->propagator.get_token(); } + + auto do_get_environment() noexcept -> Environment& final { return this->holder.environment; } + + ::std::coroutine_handle parent; + ::beman::execution::detail::task::handle handle; + env_holder> holder; + scheduler_type scheduler; + ::beman::execution::detail::task::propagator propagator; +}; + +template +struct promise_env { + auto query(const ::beman::execution::get_scheduler_t&) const noexcept -> typename Promise::scheduler_type { + return this->promise->get_start_scheduler(); + } + + auto query(const ::beman::execution::get_start_scheduler_t&) const noexcept -> typename Promise::scheduler_type { + return this->promise->get_start_scheduler(); + } + + auto query(const ::beman::execution::get_allocator_t&) const noexcept -> typename Promise::allocator_type { + return this->promise->get_allocator(); + } + + auto query(const ::beman::execution::get_stop_token_t&) const noexcept -> typename Promise::stop_token_type { + return this->promise->get_stop_token(); + } + + template + requires(::beman::execution::forwarding_query(Q())) && requires(const Promise* p, Q q, Args&&... args) { + q(p->get_environment(), ::std::forward(args)...); + } + auto query(Q q, Args&&... args) const noexcept { + return q(this->promise->get_environment(), ::std::forward(args)...); + } + + const Promise* promise; +}; + +struct final_awaiter { + static constexpr auto await_ready() noexcept -> bool { return false; } + + template + static auto await_suspend(::std::coroutine_handle handle) noexcept { + return handle.promise().notify_complete(); + } + + static constexpr auto await_resume() noexcept -> void {} +}; + +template +class promise_base { + public: + template + auto return_value(T&& value) -> void { + this->get_state()->set_value(::std::forward(value)); + } + + public: + auto set_state(state_base* s) noexcept -> void { this->state_ = s; } + auto get_state() const noexcept -> state_base* { return this->state_; } + + private: + state_base* state_{}; +}; + +template +class promise_base { + public: + auto return_void() -> void { this->get_state()->set_value(void_type{}); } + + public: + auto set_state(state_base* s) noexcept -> void { this->state_ = s; } + + auto get_state() const noexcept -> state_base* { return this->state_; } + + private: + state_base* state_{}; +}; + +template +class promise_type : public ::beman::execution::detail::task::promise_base<::std::remove_cvref_t, Environment>, + public ::beman::execution::detail::task::allocator_support< + ::beman::execution::detail::task::allocator_of_t> { + public: + using allocator_type = allocator_of_t; + using scheduler_type = start_scheduler_of_t; + using stop_source_type = stop_source_of_t; + using stop_token_type = decltype(::std::declval().get_token()); + + static auto initial_suspend() noexcept -> ::std::suspend_always { return {}; } + + static auto final_suspend() noexcept -> ::beman::execution::detail::task::final_awaiter { return {}; } + + auto unhandled_exception() noexcept -> void { + using error_types = error_types_of_t; + if constexpr (::beman::execution::detail::task::has_exception_signature::value) { + this->get_state()->set_error(::std::current_exception()); + } else { + ::std::terminate(); + } + } + + auto unhandled_stopped() -> std::coroutine_handle<> { return this->get_state()->complete(); } + + auto get_return_object() noexcept { + return Coroutine(::beman::execution::detail::task::handle(this)); + } + + template <::beman::execution::sender Expr> + auto await_transform(Expr&& expr) -> decltype(auto) { + return ::beman::execution::as_awaitable(::beman::execution::affine(::std::forward(expr)), *this); + } + + template + auto yield_value(::beman::execution::with_error with) noexcept + -> ::beman::execution::detail::task::final_awaiter { + this->get_state()->set_error(::std::move(with.error)); + return {}; + } + + auto get_env() const noexcept -> ::beman::execution::detail::task::promise_env { return {this}; } + + auto start(::beman::execution::detail::task::state_base* state) -> ::std::coroutine_handle<> { + this->set_state(state); + return ::std::coroutine_handle::from_promise(*this); + } + + auto notify_complete() noexcept -> ::std::coroutine_handle<> { return this->get_state()->complete(); } + + auto get_start_scheduler() const noexcept -> scheduler_type { return this->get_state()->get_start_scheduler(); } + + auto get_allocator() const noexcept -> allocator_type { return this->get_state()->get_allocator(); } + + auto get_stop_token() const noexcept -> stop_token_type { return this->get_state()->get_stop_token(); } + + auto get_environment() const noexcept -> const Environment& { return this->get_state()->get_environment(); } +}; + +} // namespace beman::execution::detail::task + +namespace beman::execution { +template > +class task { + friend ::beman::execution::detail::task::promise_type; + + public: + using sender_concept = ::beman::execution::sender_tag; + using promise_type = ::beman::execution::detail::task::promise_type; + using allocator_type = ::beman::execution::detail::task::allocator_of_t; + using start_scheduler_type = ::beman::execution::detail::task::start_scheduler_of_t; + using stop_source_type = ::beman::execution::detail::task::stop_source_of_t; + using stop_token_type = decltype(::std::declval().get_token()); + using completion_signatures = ::beman::execution::detail::meta::combine< + ::beman::execution::completion_signatures<::beman::execution::detail::task::value_signature_t, + ::beman::execution::set_stopped_t()>, + ::beman::execution::detail::task::error_types_of_t>; + + private: + template + using state = ::beman::execution::detail::task::state; + + template + using awaiter = ::beman::execution::detail::task::awaiter; + + public: + task(const task&) = delete; + + task(task&&) noexcept = default; + + task& operator=(const task&) = delete; + + task& operator=(task&&) noexcept = default; + + ~task() = default; + + template <::beman::execution::detail::decayed_same_as, typename...> + static consteval auto get_completion_signatures() noexcept -> completion_signatures { + return {}; + } + + template <::beman::execution::receiver Receiver> + auto connect(Receiver receiver) && noexcept -> state { + auto env = ::beman::execution::get_env(receiver); + return state(::std::move(receiver), ::std::move(this->handle), env); + } + + template + auto as_awaitable(ParentPromise& parent) && noexcept -> awaiter { + auto env = ::beman::execution::get_env(parent); + return awaiter(parent, ::std::move(this->handle), env); + } + + private: + explicit task(::beman::execution::detail::task::handle h) noexcept : handle(::std::move(h)) {} + + ::beman::execution::detail::task::handle handle; +}; +} // namespace beman::execution + +// ---------------------------------------------------------------------------- + +#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_TASK diff --git a/include/beman/execution/detail/task_scheduler.hpp b/include/beman/execution/detail/task_scheduler.hpp new file mode 100644 index 00000000..42172845 --- /dev/null +++ b/include/beman/execution/detail/task_scheduler.hpp @@ -0,0 +1,501 @@ +// include/beman/execution/detail/task_scheduler.hpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_TASK_SCHEDULER +#define INCLUDED_BEMAN_EXECUTION_DETAIL_TASK_SCHEDULER + +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include +#include +#include +#include +#include +#include +#include +#include +#include +#endif +#ifdef BEMAN_HAS_MODULES +import beman.execution.detail.bulk; +import beman.execution.detail.completion_signatures; +import beman.execution.detail.connect; +import beman.execution.detail.connect_result_t; +import beman.execution.detail.env; +import beman.execution.detail.env_of_t; +import beman.execution.detail.execution_policy; +import beman.execution.detail.fwd_env; +import beman.execution.detail.forward_like; +import beman.execution.detail.get_completion_domain; +import beman.execution.detail.get_completion_scheduler; +import beman.execution.detail.get_env; +import beman.execution.detail.get_forward_progress_guarantee; +import beman.execution.detail.get_stop_token; +import beman.execution.detail.infallible_scheduler; +import beman.execution.detail.inplace_stop_source; +import beman.execution.detail.new_object; +import beman.execution.detail.operation_state; +import beman.execution.detail.psched_bulk_sender; +import beman.execution.detail.parallel_scheduler_replacement; +import beman.execution.detail.prop; +import beman.execution.detail.query_with_default; +import beman.execution.detail.receiver; +import beman.execution.detail.schedule; +import beman.execution.detail.schedule_result_t; +import beman.execution.detail.scheduler; +import beman.execution.detail.scheduler_tag; +import beman.execution.detail.sender; +import beman.execution.detail.set_error; +import beman.execution.detail.set_stopped; +import beman.execution.detail.set_value; +import beman.execution.detail.simple_allocator; +import beman.execution.detail.start; +import beman.execution.detail.stop_propagator; +import beman.execution.detail.stop_token_of_t; +import beman.execution.detail.tag_of_t; +import beman.execution.detail.unreachable; +import beman.execution.detail.unstoppable_token; +#else +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#endif + +// ---------------------------------------------------------------------------- + +namespace beman::execution::detail { +struct task_scheduler_backend : ::beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend { + // ReSharper disable once CppPolymorphicClassWithNonVirtualPublicDestructor + struct state_holder_base { + virtual auto destroy() noexcept -> void = 0; + }; + + struct rcvr_wrapper { + using receiver_concept = ::beman::execution::receiver_tag; + + auto set_value() noexcept -> void { + auto& proxy_ref = proxy; + holder.destroy(); + proxy_ref.set_value(); + } + + auto set_error(::std::exception_ptr error) noexcept -> void { + auto& proxy_ref = proxy; + holder.destroy(); + proxy_ref.set_error(::std::move(error)); + } + + auto set_stopped() noexcept -> void { + auto& proxy_ref = proxy; + holder.destroy(); + proxy_ref.set_stopped(); + } + + auto get_env() const noexcept { + return ::beman::execution::prop{ + ::beman::execution::get_stop_token, + proxy.try_query<::beman::execution::inplace_stop_token>(::beman::execution::get_stop_token) + .value_or(::beman::execution::inplace_stop_token{})}; + } + + ::beman::execution::parallel_scheduler_replacement::receiver_proxy& proxy; + state_holder_base& holder; + }; + + template + // ReSharper disable once CppPolymorphicClassWithNonVirtualPublicDestructor + struct allocated_state_holder : state_holder_base { + using op_t = ::beman::execution::connect_result_t; + + allocated_state_holder(Sndr sndr, + ::beman::execution::parallel_scheduler_replacement::receiver_proxy& proxy, + Alloc alloc) + : alloc(std::move(alloc)), + op(::beman::execution::connect(::std::move(sndr), rcvr_wrapper{proxy, *this})) {} + + auto destroy() noexcept -> void override { ::beman::execution::detail::delete_object(alloc, this); } + + auto do_start() noexcept -> void { ::beman::execution::start(op); } + + [[no_unique_address]] Alloc alloc; + op_t op; + }; + + template + // ReSharper disable once CppPolymorphicClassWithNonVirtualPublicDestructor + struct inlined_state_holder : state_holder_base { + using op_t = ::beman::execution::connect_result_t; + + inlined_state_holder(Sndr sndr, ::beman::execution::parallel_scheduler_replacement::receiver_proxy& proxy) + : op(::beman::execution::connect(::std::move(sndr), rcvr_wrapper{proxy, *this})) {} + + auto destroy() noexcept -> void override { ::std::destroy_at(this); } + + auto do_start() noexcept -> void { ::beman::execution::start(op); } + + op_t op; + }; + + virtual auto get_forward_progress_guarantee() const noexcept -> ::beman::execution::forward_progress_guarantee = 0; + + virtual auto do_equals(const task_scheduler_backend&) const noexcept -> bool = 0; +}; + +template +struct task_scheduler_backend_sched : task_scheduler_backend { + struct just_sndr_like { + using sender_concept = ::beman::execution::sender_tag; + using completion_signatures = ::beman::execution::completion_signatures<::beman::execution::set_value_t()>; + struct attr { + template + auto query(::beman::execution::get_completion_scheduler_t<::beman::execution::set_value_t>, + const Args&... args) const noexcept { + return ::beman::execution::get_completion_scheduler<::beman::execution::set_value_t>(sched, args...); + } + + Sched sched; + }; + + template + struct state { + using operation_state_concept = ::beman::execution::operation_state_tag; + + auto start() & noexcept -> void { ::beman::execution::set_value(::std::move(rcvr)); } + + Rcvr rcvr; + }; + + template + static consteval auto get_completion_signatures() noexcept { + return completion_signatures{}; + } + + auto get_env() const noexcept -> attr { return attr{sched}; } + + template <::beman::execution::receiver Rcvr> + auto connect(Rcvr rcvr) noexcept -> state { + return state(::std::move(rcvr)); + } + + Sched sched; + }; + + explicit task_scheduler_backend_sched(Sched sched) noexcept : sched(::std::move(sched)) {} + + auto get_forward_progress_guarantee() const noexcept -> ::beman::execution::forward_progress_guarantee override { + return ::beman::execution::get_forward_progress_guarantee(sched); + } + + auto do_equals(const task_scheduler_backend& other) const noexcept -> bool override { + if (auto o = dynamic_cast(&other)) { + return o->sched == sched; + } + return false; + } + + Sched sched; +}; + +template +struct task_scheduler_backend_for : task_scheduler_backend_sched { + using base = task_scheduler_backend_sched; + using typename base::just_sndr_like; + + task_scheduler_backend_for(Sched sched, Alloc alloc) noexcept + : base(::std::move(sched)), alloc(::std::move(alloc)) {} + + task_scheduler_backend_for(const task_scheduler_backend_for&) = delete; + + ~task_scheduler_backend_for() override = default; + + auto operator=(const task_scheduler_backend_for&) -> task_scheduler_backend_for& = delete; + + auto schedule(::beman::execution::parallel_scheduler_replacement::receiver_proxy& proxy, + ::std::span<::std::byte> storage) noexcept -> void override { + this->fire_and_forget(storage, ::beman::execution::schedule(this->sched), proxy); + } + + auto schedule_bulk_chunked(::std::size_t shape, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::span<::std::byte> storage) noexcept -> void override { + const ::std::size_t chunk_size = ::std::min(shape, 8uz); + const ::std::size_t num_chunks = (shape + chunk_size - 1uz) / chunk_size; + this->fire_and_forget(storage, + ::beman::execution::bulk(just_sndr_like{this->sched}, + ::beman::execution::par, + num_chunks, + [=, &proxy](::std::size_t i) noexcept { + const auto begin = i * chunk_size; + const auto end = shape - begin <= chunk_size + ? shape + : begin + chunk_size; + proxy.execute(begin, end); + }), + proxy); + } + + auto schedule_bulk_unchunked(::std::size_t shape, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::span<::std::byte> storage) noexcept -> void override { + this->fire_and_forget( + storage, + ::beman::execution::bulk(just_sndr_like{this->sched}, + ::beman::execution::par, + shape, + [&proxy](::std::size_t i) noexcept { proxy.execute(i, i + 1uz); }), + proxy); + } + + template + auto fire_and_forget(::std::span<::std::byte> storage, + Sndr sndr, + ::beman::execution::parallel_scheduler_replacement::receiver_proxy& proxy) const noexcept + -> void { + using inlined_holder = typename base::template inlined_state_holder; + using allocated_holder = typename base::template allocated_state_holder; + constexpr bool inlined = + sizeof(inlined_holder) <= psched_storage_size && alignof(inlined_holder) <= psched_storage_alignment; + try { + if constexpr (inlined) { + auto holder = ::new (static_cast(storage.data())) inlined_holder(::std::move(sndr), proxy); + holder->do_start(); + } else { + auto holder = + ::beman::execution::detail::new_object(alloc, ::std::move(sndr), proxy, alloc); + holder->do_start(); + } + } catch (...) { + proxy.set_error(::std::current_exception()); + } + } + + [[no_unique_address]] Alloc alloc; +}; +} // namespace beman::execution::detail + +namespace beman::execution { +class task_scheduler { + public: + using scheduler_concept = ::beman::execution::scheduler_tag; + class sender; + struct domain { + struct not_a_scheduler {}; + template + requires ::std::same_as<::beman::execution::tag_of_t, ::beman::execution::bulk_chunked_t> || + ::std::same_as<::beman::execution::tag_of_t, ::beman::execution::bulk_unchunked_t> + static auto transform_sender(::beman::execution::set_value_t, Sndr&& sndr, const auto& env) noexcept { + auto&& [_, data, child] = ::std::forward(sndr); + auto [policy, shape, fn] = ::beman::execution::detail::forward_like(data); + auto sch = ::beman::execution::detail::call_with_default( + ::beman::execution::get_completion_scheduler<::beman::execution::set_value_t>, + not_a_scheduler{}, + ::beman::execution::get_env(child), + ::beman::execution::detail::fwd_env(env)); + + static_assert(::std::same_as, + "the completion scheduler of `sndr` shall be a `task_scheduler`."); + + return ::beman::execution::detail::psched_bulk_sender{ + ::std::bool_constant< + ::std::same_as<::beman::execution::tag_of_t, ::beman::execution::bulk_chunked_t>>{}, + sch.backend_, + ::std::move(policy), + ::std::move(shape), + ::std::move(fn), + ::beman::execution::detail::forward_like(child)}; + } + }; + + template > + requires(!::std::same_as>) && + ::beman::execution::detail::infallible_scheduler> + explicit task_scheduler(Sched&& sched, Alloc alloc = {}) + : backend_(::std::allocate_shared< + ::beman::execution::detail::task_scheduler_backend_for<::std::remove_cvref_t, Alloc>>( + alloc, ::std::forward(sched), ::std::move(alloc))) {} + + template <::beman::execution::detail::simple_allocator Alloc> + explicit task_scheduler(task_scheduler sched, const Alloc&) noexcept : task_scheduler(::std::move(sched)) {} + + auto schedule() const noexcept -> sender; + + auto query(::beman::execution::get_forward_progress_guarantee_t) const noexcept { + return backend_->get_forward_progress_guarantee(); + } + + static auto query(::beman::execution::get_completion_domain_t<::beman::execution::set_value_t>) noexcept { + return domain{}; + } + + friend auto operator==(const task_scheduler& lhs, const task_scheduler& rhs) noexcept -> bool { + if (lhs.backend_ == nullptr) { + return rhs.backend_ == nullptr; + } + if (rhs.backend_ == nullptr) { + return false; + } + return lhs.backend_->do_equals(*rhs.backend_); + } + + template + requires(!::std::same_as) && ::beman::execution::scheduler + auto operator==(const Sched& other) const noexcept -> bool { + auto b = dynamic_cast*>(backend_.get()); + return b && b->sched == other; + } + + private: + explicit task_scheduler(::std::shared_ptr<::beman::execution::detail::task_scheduler_backend> backend) noexcept + : backend_(std::move(backend)) {} + + ::std::shared_ptr<::beman::execution::detail::task_scheduler_backend> backend_; +}; + +class task_scheduler::sender { + using backend_type = ::beman::execution::detail::task_scheduler_backend; + + public: + using sender_concept = ::beman::execution::sender_tag; + + class env { + public: + explicit env(::std::shared_ptr backend) noexcept : backend_(::std::move(backend)) {} + + auto + query(const ::beman::execution::get_completion_scheduler_t<::beman::execution::set_value_t>&) const noexcept + -> ::beman::execution::task_scheduler { + return ::beman::execution::task_scheduler{this->backend_}; + } + + private: + ::std::shared_ptr backend_; + }; + + template + struct rcvr_proxy : ::beman::execution::parallel_scheduler_replacement::receiver_proxy { + using stop_token_t = ::beman::execution::stop_token_of_t<::beman::execution::env_of_t>; + + explicit rcvr_proxy(Rcvr rcvr) noexcept + : sp_(::beman::execution::get_stop_token(::beman::execution::get_env(rcvr))), rcvr_(::std::move(rcvr)) {} + + auto set_value() noexcept -> void final { ::beman::execution::set_value(::std::move(this->rcvr_)); } + + auto set_error(::std::exception_ptr) noexcept -> void final { ::beman::execution::detail::unreachable(); } + + auto set_stopped() noexcept -> void final { + if constexpr (::beman::execution::unstoppable_token) { + ::beman::execution::detail::unreachable(); + } else { + ::beman::execution::set_stopped(::std::move(this->rcvr_)); + } + } + + private: + auto query_stop_token() const noexcept -> ::std::optional<::beman::execution::inplace_stop_token> final { + return sp_.get_token(); + } + + [[no_unique_address]] ::beman::execution::detail::stop_propagator sp_; + Rcvr rcvr_; + }; + + template <::beman::execution::receiver Rcvr> + class state { + public: + using operation_state_concept = ::beman::execution::operation_state_tag; + using stop_token_t = ::beman::execution::stop_token_of_t<::beman::execution::env_of_t>; + + state(::std::shared_ptr backend, + Rcvr&& rcvr) noexcept(::std::is_nothrow_constructible_v<::std::remove_cvref_t, Rcvr>) + : backend_(::std::move(backend)), proxy_(::std::forward(rcvr)) {} + + auto start() & noexcept -> void { + auto stop_token = + proxy_.template try_query<::beman::execution::inplace_stop_token>(::beman::execution::get_stop_token); + if (stop_token.has_value() && stop_token->stop_requested()) { + proxy_.set_stopped(); + return; + } + this->backend_->schedule(proxy_, ::std::span<::std::byte>{this->storage_}); + } + + private: + ::std::shared_ptr backend_; + rcvr_proxy<::std::remove_cvref_t> proxy_; + alignas(::beman::execution::detail::psched_storage_alignment)::std::byte + storage_[::beman::execution::detail::psched_storage_size]; + }; + + explicit sender(::std::shared_ptr backend) noexcept : backend_(::std::move(backend)) {} + + template + static consteval auto get_completion_signatures() noexcept { + if constexpr (::beman::execution::unstoppable_token<::beman::execution::stop_token_of_t>) { + return ::beman::execution::completion_signatures<::beman::execution::set_value_t()>{}; + } else { + return ::beman::execution::completion_signatures<::beman::execution::set_value_t(), + ::beman::execution::set_stopped_t()>{}; + } + } + + auto get_env() const noexcept -> env { return env{this->backend_}; } + + template <::beman::execution::receiver Rcvr> + auto connect(Rcvr&& rcvr) && noexcept(::std::is_nothrow_constructible_v<::std::remove_cvref_t, Rcvr>) + -> state { + return state{::std::move(this->backend_), ::std::forward(rcvr)}; + } + + private: + ::std::shared_ptr backend_; +}; + +inline auto task_scheduler::schedule() const noexcept -> sender { return sender{this->backend_}; } + +} // namespace beman::execution + +#ifndef BEMAN_HAS_MODULES +template +struct std::uses_allocator<::beman::execution::task_scheduler, Alloc> : ::std::true_type {}; +#endif + +// ---------------------------------------------------------------------------- + +#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_TASK_SCHEDULER diff --git a/include/beman/execution/detail/with_error.hpp b/include/beman/execution/detail/with_error.hpp index c6bf423f..be67fbc5 100644 --- a/include/beman/execution/detail/with_error.hpp +++ b/include/beman/execution/detail/with_error.hpp @@ -4,17 +4,16 @@ #ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_WITH_ERROR #define INCLUDED_BEMAN_EXECUTION_DETAIL_WITH_ERROR -#include +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else #include -#include +#endif // ---------------------------------------------------------------------------- -/* - * \brief Tag type used to indicate an error is produced. - * \headerfile beman/execution/task.hpp - * \internal - */ -namespace beman::execution::detail { + +namespace beman::execution { template struct with_error { using type = ::std::remove_cvref_t; @@ -23,7 +22,7 @@ struct with_error { template with_error(E&&) -> with_error; -} // namespace beman::execution::detail +} // namespace beman::execution // ---------------------------------------------------------------------------- diff --git a/include/beman/execution/execution.hpp b/include/beman/execution/execution.hpp index 46860df0..0f10f604 100644 --- a/include/beman/execution/execution.hpp +++ b/include/beman/execution/execution.hpp @@ -48,6 +48,7 @@ import beman.execution.detail.matching_sig; import beman.execution.detail.on; import beman.execution.detail.operation_state; import beman.execution.detail.parallel_scheduler; +import beman.execution.detail.parallel_scheduler_replacement; import beman.execution.detail.prop; import beman.execution.detail.read_env; import beman.execution.detail.receiver; @@ -72,12 +73,15 @@ import beman.execution.detail.stopped_as_error; import beman.execution.detail.stopped_as_optional; import beman.execution.detail.sync_wait; import beman.execution.detail.sync_wait_with_variant; +import beman.execution.detail.task; +import beman.execution.detail.task_scheduler; import beman.execution.detail.then; import beman.execution.detail.valid_completion_for; import beman.execution.detail.valid_completion_signatures; import beman.execution.detail.when_all_with_variant; import beman.execution.detail.when_all; import beman.execution.detail.with_awaitable_senders; +import beman.execution.detail.with_error; import beman.execution.detail.within; import beman.execution.detail.write_env; #else @@ -120,6 +124,7 @@ import beman.execution.detail.write_env; #include #include #include +#include #include #include #include @@ -144,12 +149,15 @@ import beman.execution.detail.write_env; #include #include #include +#include +#include #include #include #include #include #include #include +#include #include #include #endif diff --git a/src/beman/execution/CMakeLists.txt b/src/beman/execution/CMakeLists.txt index 0a5ddf05..6e7a3191 100644 --- a/src/beman/execution/CMakeLists.txt +++ b/src/beman/execution/CMakeLists.txt @@ -27,7 +27,6 @@ target_sources( FILES ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/affine.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/allocator_aware_move.hpp - ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/allocator_of.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/almost_scheduler.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/apply_sender.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/as_awaitable.hpp @@ -85,7 +84,6 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/env_of_t.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/env_promise.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/env_type.hpp - ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/error_types_of.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/error_types_of_t.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/exit_scope_sender.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/exit_scope_sender_in.hpp @@ -109,7 +107,6 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/get_scheduler.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/get_start_scheduler.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/get_stop_token.hpp - ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/handle.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/has_as_awaitable.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/has_completions.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/hide_sched.hpp @@ -138,6 +135,7 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/meta_contain_same.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/meta_contains.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/meta_filter.hpp + ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/meta_index.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/meta_prepend.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/meta_size.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/meta_to.hpp @@ -146,6 +144,7 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/movable_value.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/nested_sender_has_affine.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/never_stop_token.hpp + ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/new_object.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/non_assignable.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/nostopstate.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/nothrow_callable.hpp @@ -155,7 +154,9 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/on_stop_request.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/operation_state.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/operation_state_task.hpp + ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/psched_bulk_sender.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/parallel_scheduler.hpp + ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/parallel_scheduler_replacement.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/product_type.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/poly.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/promise_env.hpp @@ -196,12 +197,10 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/spawn_get_allocator.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/start.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/starts_on.hpp - ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/state_rep.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/state_type.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/stop_callback_for_t.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/stop_propagator.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/stop_source.hpp - ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/stop_source_of.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/stop_token_of_t.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/stop_when.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/stoppable_source.hpp @@ -217,6 +216,8 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/sync_wait.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/sync_wait_with_variant.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/tag_of_t.hpp + ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/task.hpp + ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/task_scheduler.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/then.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/transform_sender.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/try_query.hpp @@ -311,6 +312,7 @@ if(BEMAN_USE_MODULES) exit_scope_sender.cppm exit_scope_sender_in.cppm exit_scope_sender_of_t.cppm + find_allocator.cppm forward_like.cppm forwarding_query.cppm fwd_env.cppm @@ -357,6 +359,7 @@ if(BEMAN_USE_MODULES) meta_contains.cppm meta_contain_same.cppm meta_filter.cppm + meta_index.cppm meta_prepend.cppm meta_size.cppm meta_to.cppm @@ -365,6 +368,7 @@ if(BEMAN_USE_MODULES) movable_value.cppm nested_sender_has_affine.cppm never_stop_token.cppm + new_object.cppm notify.cppm non_assignable.cppm nostopstate.cppm @@ -375,8 +379,10 @@ if(BEMAN_USE_MODULES) operation_state_task.cppm operation_state.cppm parallel_scheduler.cppm + parallel_scheduler_replacement.cppm product_type.cppm prop.cppm + psched_bulk_sender.cppm query_with_default.cppm queryable.cppm read_env.cppm @@ -426,10 +432,13 @@ if(BEMAN_USE_MODULES) stopped_as_error.cppm stopped_as_optional.cppm store_receiver.cppm + sub_visit.cppm suspend_complete.cppm sync_wait.cppm sync_wait_with_variant.cppm tag_of_t.cppm + task.cppm + task_scheduler.cppm then.cppm transform_sender.cppm try_query.cppm @@ -448,6 +457,7 @@ if(BEMAN_USE_MODULES) when_all_with_variant.cppm with_await_transform.cppm with_awaitable_senders.cppm + with_error.cppm within.cppm write_env.cppm ) diff --git a/src/beman/execution/execution.cppm b/src/beman/execution/execution.cppm index 83f71505..a1770da6 100644 --- a/src/beman/execution/execution.cppm +++ b/src/beman/execution/execution.cppm @@ -52,6 +52,7 @@ import beman.execution.detail.nostopstate; import beman.execution.detail.on; export import beman.execution.detail.operation_state; // [exec.opstate], operation states import beman.execution.detail.parallel_scheduler; +import beman.execution.detail.parallel_scheduler_replacement; import beman.execution.detail.prop; import beman.execution.detail.read_env; import beman.execution.detail.run_loop; @@ -108,6 +109,11 @@ export import beman.execution.detail.sender_to; export import beman.execution.detail.dependent_sender; export import beman.execution.detail.sends_stopped; +// [exec.task] +export import beman.execution.detail.task; +export import beman.execution.detail.task_scheduler; +export import beman.execution.detail.with_error; + namespace beman::execution { export using ::beman::execution::nostopstate_t; diff --git a/src/beman/execution/find_allocator.cppm b/src/beman/execution/find_allocator.cppm new file mode 100644 index 00000000..65d418ea --- /dev/null +++ b/src/beman/execution/find_allocator.cppm @@ -0,0 +1,11 @@ +module; +// src/beman/execution/find_allocator.cppm -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include + +export module beman.execution.detail.find_allocator; + +namespace beman::execution::detail { +export using beman::execution::detail::find_allocator; +} // namespace beman::execution::detail diff --git a/src/beman/execution/meta_index.cppm b/src/beman/execution/meta_index.cppm new file mode 100644 index 00000000..d5cca563 --- /dev/null +++ b/src/beman/execution/meta_index.cppm @@ -0,0 +1,12 @@ +module; +// src/beman/execution/meta_index.cppm -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include + +export module beman.execution.detail.meta.index; + +namespace beman::execution::detail::meta { +export using beman::execution::detail::meta::index; +export using beman::execution::detail::meta::index_v; +} // namespace beman::execution::detail::meta diff --git a/src/beman/execution/new_object.cppm b/src/beman/execution/new_object.cppm new file mode 100644 index 00000000..ccd1a664 --- /dev/null +++ b/src/beman/execution/new_object.cppm @@ -0,0 +1,12 @@ +module; +// src/beman/execution/new_object.cppm -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include + +export module beman.execution.detail.new_object; + +namespace beman::execution::detail { +export using beman::execution::detail::new_object; +export using beman::execution::detail::delete_object; +} // namespace beman::execution::detail diff --git a/src/beman/execution/parallel_scheduler.cppm b/src/beman/execution/parallel_scheduler.cppm index 8fcf75e4..14076f84 100644 --- a/src/beman/execution/parallel_scheduler.cppm +++ b/src/beman/execution/parallel_scheduler.cppm @@ -9,11 +9,4 @@ export module beman.execution.detail.parallel_scheduler; namespace beman::execution { export using beman::execution::parallel_scheduler; export using beman::execution::get_parallel_scheduler; - -namespace parallel_scheduler_replacement { -export using beman::execution::parallel_scheduler_replacement::receiver_proxy; -export using beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy; -export using beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend; -export using beman::execution::parallel_scheduler_replacement::query_parallel_scheduler_backend; -} // namespace parallel_scheduler_replacement } // namespace beman::execution diff --git a/src/beman/execution/parallel_scheduler_replacement.cppm b/src/beman/execution/parallel_scheduler_replacement.cppm new file mode 100644 index 00000000..252df8ee --- /dev/null +++ b/src/beman/execution/parallel_scheduler_replacement.cppm @@ -0,0 +1,14 @@ +module; +// src/beman/execution/parallel_scheduler_replacement.cppm -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include + +export module beman.execution.detail.parallel_scheduler_replacement; + +namespace beman::execution::parallel_scheduler_replacement { +export using beman::execution::parallel_scheduler_replacement::receiver_proxy; +export using beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy; +export using beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend; +export using beman::execution::parallel_scheduler_replacement::query_parallel_scheduler_backend; +} // namespace beman::execution::parallel_scheduler_replacement diff --git a/src/beman/execution/psched_bulk_sender.cppm b/src/beman/execution/psched_bulk_sender.cppm new file mode 100644 index 00000000..2bd440ad --- /dev/null +++ b/src/beman/execution/psched_bulk_sender.cppm @@ -0,0 +1,13 @@ +module; +// src/beman/execution/psched_bulk_sender.cppm -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include + +export module beman.execution.detail.psched_bulk_sender; + +namespace beman::execution::detail { +export using beman::execution::detail::psched_storage_alignment; +export using beman::execution::detail::psched_storage_size; +export using beman::execution::detail::psched_bulk_sender; +} // namespace beman::execution::detail diff --git a/src/beman/execution/sub_visit.cppm b/src/beman/execution/sub_visit.cppm new file mode 100644 index 00000000..a032a081 --- /dev/null +++ b/src/beman/execution/sub_visit.cppm @@ -0,0 +1,11 @@ +module; +// src/beman/execution/sub_visit.cppm -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include + +export module beman.execution.detail.sub_visit; + +namespace beman::execution::detail { +export using beman::execution::detail::sub_visit; +} // namespace beman::execution::detail diff --git a/src/beman/execution/task.cppm b/src/beman/execution/task.cppm new file mode 100644 index 00000000..62dd6fe3 --- /dev/null +++ b/src/beman/execution/task.cppm @@ -0,0 +1,11 @@ +module; +// src/beman/execution/task.cppm -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include + +export module beman.execution.detail.task; + +namespace beman::execution { +export using beman::execution::task; +} // namespace beman::execution diff --git a/src/beman/execution/task_scheduler.cppm b/src/beman/execution/task_scheduler.cppm new file mode 100644 index 00000000..7a8b2af8 --- /dev/null +++ b/src/beman/execution/task_scheduler.cppm @@ -0,0 +1,16 @@ +module; +// src/beman/execution/task_scheduler.cppm -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include +#include +#include + +export module beman.execution.detail.task_scheduler; + +namespace beman::execution { +export using beman::execution::task_scheduler; +} // namespace beman::execution + +template +struct std::uses_allocator<::beman::execution::task_scheduler, Alloc> : ::std::true_type {}; diff --git a/src/beman/execution/with_error.cppm b/src/beman/execution/with_error.cppm new file mode 100644 index 00000000..f5982921 --- /dev/null +++ b/src/beman/execution/with_error.cppm @@ -0,0 +1,11 @@ +module; +// src/beman/execution/with_error.cppm -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include + +export module beman.execution.detail.with_error; + +namespace beman::execution { +export using beman::execution::with_error; +} // namespace beman::execution diff --git a/tests/beman/execution/CMakeLists.txt b/tests/beman/execution/CMakeLists.txt index 91bcd74e..337f4769 100644 --- a/tests/beman/execution/CMakeLists.txt +++ b/tests/beman/execution/CMakeLists.txt @@ -69,6 +69,8 @@ list( exec-stopped-as-optional.test exec-sync-wait.test exec-sync-wait-with-variant.test + exec-task.test + exec-task-scheduler.test exec-then.test exec-utils-cmplsigs.test exec-when-all.test diff --git a/tests/beman/execution/exec-task-scheduler.test.cpp b/tests/beman/execution/exec-task-scheduler.test.cpp new file mode 100644 index 00000000..f0b41c86 --- /dev/null +++ b/tests/beman/execution/exec-task-scheduler.test.cpp @@ -0,0 +1,165 @@ +// tests/beman/execution/exec-task-scheduler.test.cpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include +#include +#include +#include +#include +#include +#include +#include + +#ifdef BEMAN_HAS_MODULES +import beman.execution; +#else +#include +#include +#include +#include +#include +#include +#include +#include +#include +#endif + +// ---------------------------------------------------------------------------- + +namespace { + +struct async_scheduler { + using scheduler_concept = test_std::scheduler_tag; + + int id{}; + + template + struct operation { + using operation_state_concept = test_std::operation_state_tag; + Rcvr rcvr; + + auto start() & noexcept -> void { + ::std::thread([receiver = ::std::move(rcvr)]() mutable noexcept { + test_std::set_value(::std::move(receiver)); + }).detach(); + } + }; + + struct sender { + using sender_concept = test_std::sender_tag; + using completion_signatures = test_std::completion_signatures; + + template + static consteval auto get_completion_signatures() noexcept -> completion_signatures { + return {}; + } + static auto get_env() noexcept -> test_std::env<> { return {}; } + + template + auto connect(Rcvr rcvr) const -> operation { + return {::std::move(rcvr)}; + } + }; + + static auto schedule() noexcept -> sender { return {}; } + auto operator==(const async_scheduler&) const noexcept -> bool = default; +}; + +struct stopped_receiver_env { + test_std::inplace_stop_token token; + + auto query(const test_std::get_stop_token_t&) const noexcept -> test_std::inplace_stop_token { return token; } +}; + +struct stopped_receiver { + using receiver_concept = test_std::receiver_tag; + + test_std::inplace_stop_token token; + bool* stopped; + + auto set_value() && noexcept -> void { ASSERT(nullptr == "scheduler completed with an unexpected value"); } + auto set_error(auto&&) && noexcept -> void { ASSERT(nullptr == "scheduler completed with an unexpected error"); } + auto set_stopped() && noexcept -> void { *stopped = true; } + auto get_env() const noexcept -> stopped_receiver_env { return {token}; } +}; + +template +concept connectable_as_lvalue = + requires(Sender sender, stopped_receiver receiver) { sender.connect(std::move(receiver)); }; + +auto test_task_scheduler_interface() -> void { + using schedule_sender = decltype(test_std::schedule(std::declval())); + + static_assert(!::std::default_initializable); + static_assert(::std::copy_constructible); + static_assert(test_std::scheduler); + static_assert(::std::uses_allocator_v>); + static_assert(requires(schedule_sender sender, stopped_receiver receiver) { + std::move(sender).connect(std::move(receiver)); + }); + static_assert(!connectable_as_lvalue); + + const test_std::inline_scheduler base{}; + const test_std::task_scheduler scheduler{base}; + const test_std::task_scheduler same{base}; + + ASSERT(scheduler == base); + ASSERT(base == scheduler); + ASSERT(scheduler == same); + ASSERT(scheduler != test_std::task_scheduler{async_scheduler{1}}); + ASSERT(test_std::get_forward_progress_guarantee(scheduler) == + test_std::forward_progress_guarantee::weakly_parallel); + static_assert( + ::std::same_as(scheduler)), + test_std::get_completion_domain_t::domain_of>); + + auto sender = test_std::schedule(scheduler); + ASSERT(test_std::get_completion_scheduler(test_std::get_env(sender)) == scheduler); +} + +auto test_task_scheduler_schedule_and_bulk() -> void { + const test_std::task_scheduler scheduler{test_std::inline_scheduler{}}; + int value{}; + test_std::sync_wait(test_std::schedule(scheduler) | test_std::then([&] { value = 42; })); + ASSERT(value == 42); + + std::vector values(8uz); + std::iota(values.begin(), values.end(), 0); + test_std::sync_wait(test_std::schedule(scheduler) | + test_std::bulk(test_std::par, values.size(), [&values](std::size_t i) { values[i] *= 2; })); + for (std::size_t i = 0; i != values.size(); ++i) { + ASSERT(values[i] == static_cast(2uz * i)); + } + + test_std::sync_wait( + test_std::schedule(scheduler) | + test_std::bulk_unchunked(test_std::par, values.size(), [&values](std::size_t i) { ++values[i]; })); + for (std::size_t i = 0; i != values.size(); ++i) { + ASSERT(values[i] == static_cast(2uz * i + 1uz)); + } + + const auto caller = ::std::this_thread::get_id(); + auto worker = caller; + test_std::sync_wait(test_std::schedule(test_std::task_scheduler{async_scheduler{}}) | + test_std::then([&] { worker = ::std::this_thread::get_id(); })); + ASSERT(worker != caller); +} + +auto test_task_scheduler_stops_before_scheduling() -> void { + test_std::inplace_stop_source source; + ASSERT(source.request_stop()); + + bool stopped{}; + auto operation = test_std::connect(test_std::schedule(test_std::task_scheduler{test_std::inline_scheduler{}}), + stopped_receiver{source.get_token(), &stopped}); + test_std::start(operation); + ASSERT(stopped); +} + +} // namespace + +TEST(exec_task_scheduler) { + test_task_scheduler_interface(); + test_task_scheduler_schedule_and_bulk(); + test_task_scheduler_stops_before_scheduling(); +} diff --git a/tests/beman/execution/exec-task.test.cpp b/tests/beman/execution/exec-task.test.cpp new file mode 100644 index 00000000..c24149e5 --- /dev/null +++ b/tests/beman/execution/exec-task.test.cpp @@ -0,0 +1,230 @@ +// tests/beman/execution/exec-task.test.cpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#ifdef BEMAN_HAS_MODULES +import beman.execution; +#else +#include +#endif + +// ---------------------------------------------------------------------------- + +namespace { + +auto value_task(bool& started) -> test_std::task { + started = true; + co_return 42; +} + +auto void_task(bool& completed) -> test_std::task<> { + completed = true; + co_return; +} + +auto sender_awaiting_task() -> test_std::task { + const auto value = co_await test_std::just(21); + co_return 2 * value; +} + +auto nested_task() -> test_std::task { co_return co_await sender_awaiting_task(); } + +auto stopped_task(bool& reached_after_stop) -> test_std::task<> { + co_await test_std::just_stopped(); + reached_after_stop = true; +} + +auto nested_stopped_task(bool& reached_after_stop) -> test_std::task<> { + co_await stopped_task(reached_after_stop); + reached_after_stop = true; +} + +auto throwing_task() -> test_std::task { + throw std::runtime_error("task failure"); + co_return 0; +} + +struct integer_error_environment { + using error_types = test_std::completion_signatures; +}; + +auto integer_error_task() -> test_std::task { co_yield test_std::with_error{17}; } + +struct pmr_environment { + using allocator_type = std::pmr::polymorphic_allocator<>; +}; + +struct counting_memory_resource final : std::pmr::memory_resource { + std::size_t allocations{}; + std::size_t deallocations{}; + + auto do_allocate(std::size_t bytes, std::size_t alignment) -> void* override { + ++allocations; + return ::operator new(bytes, std::align_val_t{alignment}); + } + + auto do_deallocate(void* pointer, std::size_t, std::size_t alignment) -> void override { + ++deallocations; + ::operator delete(pointer, std::align_val_t{alignment}); + } + + auto do_is_equal(const std::pmr::memory_resource& other) const noexcept -> bool override { return this == &other; } +}; + +auto pmr_allocator_task(std::allocator_arg_t, std::pmr::polymorphic_allocator<> alloc) + -> test_std::task { + auto frame_alloc = co_await test_std::read_env(test_std::get_allocator); + static_assert(std::same_as>); + ASSERT(frame_alloc == alloc); + co_return 42; +} + +auto scheduler_task() -> test_std::task { + const auto scheduler = co_await test_std::read_env(test_std::get_scheduler); + static_assert(std::same_as, test_std::task_scheduler>); + co_await test_std::schedule(scheduler); + co_return 42; +} + +struct stop_token_env { + test_std::inplace_stop_token token; + + auto query(const test_std::get_stop_token_t&) const noexcept -> test_std::inplace_stop_token { return token; } + auto query(const test_std::get_scheduler_t&) const noexcept -> test_std::inline_scheduler { return {}; } + auto query(const test_std::get_start_scheduler_t&) const noexcept -> test_std::inline_scheduler { return {}; } +}; + +struct stop_token_receiver { + using receiver_concept = test_std::receiver_tag; + + test_std::inplace_stop_token token; + bool* result; + + auto set_value(bool value) && noexcept -> void { *result = value; } + auto set_error(auto&&) && noexcept -> void { ASSERT(nullptr == "task completed with an unexpected error"); } + auto set_stopped() && noexcept -> void { ASSERT(nullptr == "task completed unexpectedly stopped"); } + auto get_env() const noexcept -> stop_token_env { return {token}; } +}; + +auto stop_token_task() -> test_std::task { + const auto token = co_await test_std::read_env(test_std::get_stop_token); + co_return token.stop_requested(); +} + +auto test_task_interface() -> void { + using int_task = test_std::task; + using void_task = test_std::task<>; + + static_assert(test_std::sender); + static_assert(!std::default_initializable); + static_assert(!std::copy_constructible); + static_assert(std::movable); + static_assert(std::same_as>); + static_assert(std::same_as>); +} + +auto test_task_is_lazy_and_completes_with_values() -> void { + bool started{}; + auto task = value_task(started); + ASSERT(!started); + + auto result = test_std::sync_wait(std::move(task)); + ASSERT(started); + ASSERT(result.has_value()); + ASSERT(std::get<0>(*result) == 42); + + bool completed{}; + auto void_result = test_std::sync_wait(void_task(completed)); + ASSERT(completed); + ASSERT(void_result.has_value()); +} + +auto test_task_awaits_senders_and_tasks() -> void { + auto result = test_std::sync_wait(nested_task()); + ASSERT(result.has_value()); + ASSERT(std::get<0>(*result) == 42); + + auto scheduler_result = test_std::sync_wait(scheduler_task()); + ASSERT(scheduler_result.has_value()); + ASSERT(std::get<0>(*scheduler_result) == 42); +} + +auto test_task_stopped_completion() -> void { + bool reached_after_stop{}; + auto result = test_std::sync_wait(nested_stopped_task(reached_after_stop)); + ASSERT(!result.has_value()); + ASSERT(!reached_after_stop); +} + +auto test_task_errors() -> void { + bool caught{}; + try { + test::use(test_std::sync_wait(throwing_task())); + } catch (const std::runtime_error&) { + caught = true; + } + ASSERT(caught); + + caught = false; + try { + test::use(test_std::sync_wait(integer_error_task())); + } catch (int value) { + ASSERT(value == 17); + caught = true; + } + ASSERT(caught); +} + +auto test_task_uses_pmr_allocator_arg() -> void { + counting_memory_resource resource; + std::pmr::polymorphic_allocator<> allocator{&resource}; + + auto task = pmr_allocator_task(std::allocator_arg, allocator) | + test_std::write_env(test_std::prop{test_std::get_allocator, allocator}); + ASSERT(resource.allocations == 1uz); + + auto result = test_std::sync_wait(std::move(task)); + ASSERT(result.has_value()); + ASSERT(std::get<0>(*result) == 42); + ASSERT(resource.deallocations == resource.allocations); +} + +auto test_task_propagates_stop_token() -> void { + test_std::inplace_stop_source source; + ASSERT(source.request_stop()); + + bool value{}; + auto operation = test_std::connect(stop_token_task(), stop_token_receiver{source.get_token(), &value}); + test_std::start(operation); + ASSERT(value); +} + +} // namespace + +TEST(exec_task) { + test_task_interface(); + test_task_is_lazy_and_completes_with_values(); + test_task_awaits_senders_and_tasks(); + test_task_stopped_completion(); + test_task_errors(); + test_task_uses_pmr_allocator_arg(); + test_task_propagates_stop_token(); +}