diff --git a/include/beman/execution/detail/get_completion_domain.hpp b/include/beman/execution/detail/get_completion_domain.hpp index a150ba98..743c53aa 100644 --- a/include/beman/execution/detail/get_completion_domain.hpp +++ b/include/beman/execution/detail/get_completion_domain.hpp @@ -55,12 +55,10 @@ struct get_completion_domain_t : ::beman::execution::forwarding_query_t { ::beman::execution::detail::try_query( ::beman::execution::get_completion_scheduler(q, e...), get_completion_domain_t<::beman::execution::set_value_t>{}, - ::std::forward(q), ::std::forward(e)...); }) { return ::beman::execution::detail::try_query(::beman::execution::get_completion_scheduler(q, e...), get_completion_domain_t<::beman::execution::set_value_t>{}, - ::std::forward(q), ::std::forward(e)...); } else if constexpr (::beman::execution::scheduler && 0u != sizeof...(E)) { return ::beman::execution::default_domain{}; diff --git a/include/beman/execution/detail/parallel_scheduler.hpp b/include/beman/execution/detail/parallel_scheduler.hpp new file mode 100644 index 00000000..0e0843a2 --- /dev/null +++ b/include/beman/execution/detail/parallel_scheduler.hpp @@ -0,0 +1,420 @@ +// include/beman/execution/detail/parallel_scheduler.hpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_PARALLEL_SCHEDULER +#define INCLUDED_BEMAN_EXECUTION_DETAIL_PARALLEL_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.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.meta.combine; +import beman.execution.detail.meta.prepend; +import beman.execution.detail.meta.unique; +import beman.execution.detail.operation_state; +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.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 +#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) noexcept -> ::std::optional

{ + // TODO(P2079R10): forward supported receiver environment queries + // through this proxy, especially get_stop_token_t -> 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>>; + + rcvr_proxy(Rcvr rcvr, Policy policy, Shape shape, Fn fn) noexcept + : 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)); } + + result_type result; + 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> || + ::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&) noexcept { + auto&& [_, data, child] = ::std::forward(sndr); + auto [policy, shape, fn] = ::beman::execution::detail::forward_like(data); + return ::beman::execution::detail::psched_bulk_sender{ + ::std::bool_constant< + ::std::same_as<::beman::execution::tag_of_t, ::beman::execution::bulk_chunked_t>>{}, + ::std::move(policy), + ::std::move(shape), + ::std::move(fn), + ::beman::execution::detail::forward_like(child)}; + } +}; +} // namespace beman::execution::detail + +namespace beman::execution { + +class parallel_scheduler { + using backend_type = ::beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend; + + public: + using scheduler_concept = ::beman::execution::scheduler_tag; + + class sender; + + parallel_scheduler() = delete; + ~parallel_scheduler() = default; + + parallel_scheduler(const parallel_scheduler&) noexcept = default; + parallel_scheduler(parallel_scheduler&&) noexcept = default; + auto operator=(const parallel_scheduler&) noexcept -> parallel_scheduler& = default; + auto operator=(parallel_scheduler&&) noexcept -> parallel_scheduler& = default; + + auto operator==(const parallel_scheduler& other) const noexcept -> bool { + return this->backend_ == other.backend_; + } + + static constexpr auto query(::beman::execution::get_forward_progress_guarantee_t) noexcept + -> ::beman::execution::forward_progress_guarantee { + return ::beman::execution::forward_progress_guarantee::parallel; + } + + static constexpr auto query(::beman::execution::get_completion_domain_t<::beman::execution::set_value_t>) noexcept + -> ::beman::execution::detail::parallel_scheduler_domain { + return {}; + } + + auto schedule() const noexcept -> sender; + + private: + explicit parallel_scheduler(::std::shared_ptr backend) noexcept : backend_(::std::move(backend)) {} + + ::std::shared_ptr backend_; + + friend auto get_parallel_scheduler() -> parallel_scheduler; +}; + +class parallel_scheduler::sender { + using backend_type = ::beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend; + + public: + using sender_concept = ::beman::execution::sender_tag; + using completion_signatures = + ::beman::execution::completion_signatures<::beman::execution::set_value_t(), + ::beman::execution::set_error_t(::std::exception_ptr), + ::beman::execution::set_stopped_t()>; + + 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::parallel_scheduler { + return ::beman::execution::parallel_scheduler{this->backend_}; + } + + private: + ::std::shared_ptr backend_; + }; + + template <::beman::execution::receiver Rcvr> + class operation : public ::beman::execution::parallel_scheduler_replacement::receiver_proxy { + public: + using operation_state_concept = ::beman::execution::operation_state_tag; + + operation(::std::shared_ptr backend, + Rcvr&& rcvr) noexcept(::std::is_nothrow_constructible_v<::std::remove_cvref_t, Rcvr>) + : backend_(::std::move(backend)), rcvr_(::std::forward(rcvr)) {} + + auto start() & noexcept -> void { + // TODO(P2079R10): define backend storage sizing and stopped-before-start handling. + this->backend_->schedule(*this, ::std::span<::std::byte>{this->storage_}); + } + + private: + auto set_value() noexcept -> void override { ::beman::execution::set_value(::std::move(this->rcvr_)); } + + auto set_error(::std::exception_ptr error) noexcept -> void override { + ::beman::execution::set_error(::std::move(this->rcvr_), ::std::move(error)); + } + + auto set_stopped() noexcept -> void override { ::beman::execution::set_stopped(::std::move(this->rcvr_)); } + + ::std::shared_ptr backend_; + ::std::remove_cvref_t rcvr_; + alignas(void*)::std::byte storage_[sizeof(void*) * 4]{}; + }; + + explicit sender(::std::shared_ptr backend) noexcept : backend_(::std::move(backend)) {} + + template + static consteval auto get_completion_signatures() noexcept -> completion_signatures { + return {}; + } + + 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>) + -> operation { + return operation{this->backend_, ::std::forward(rcvr)}; + } + + template <::beman::execution::receiver Rcvr> + auto connect(Rcvr&& rcvr) && noexcept(::std::is_nothrow_constructible_v<::std::remove_cvref_t, Rcvr>) + -> operation { + return operation{::std::move(this->backend_), ::std::forward(rcvr)}; + } + + private: + ::std::shared_ptr backend_; +}; + +inline auto parallel_scheduler::schedule() const noexcept -> sender { return sender{this->backend_}; } + +[[nodiscard]] inline auto get_parallel_scheduler() -> parallel_scheduler { + auto backend = ::beman::execution::parallel_scheduler_replacement::query_parallel_scheduler_backend(); + if (backend == nullptr) [[unlikely]] { + ::std::terminate(); + } + return parallel_scheduler{::std::move(backend)}; +} + +} // namespace beman::execution + +// ---------------------------------------------------------------------------- + +#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_PARALLEL_SCHEDULER diff --git a/include/beman/execution/execution.hpp b/include/beman/execution/execution.hpp index 70a5f5c5..57656396 100644 --- a/include/beman/execution/execution.hpp +++ b/include/beman/execution/execution.hpp @@ -42,6 +42,7 @@ import beman.execution.detail.let; 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.prop; import beman.execution.detail.read_env; import beman.execution.detail.receiver; @@ -107,6 +108,7 @@ import beman.execution.detail.write_env; #include #include #include +#include #include #include #include diff --git a/include/beman/execution26/execution.hpp b/include/beman/execution26/execution.hpp index d414719b..5b685670 100644 --- a/include/beman/execution26/execution.hpp +++ b/include/beman/execution26/execution.hpp @@ -42,6 +42,7 @@ using ::beman::execution::get_domain; using ::beman::execution::get_domain_t; using ::beman::execution::get_env; using ::beman::execution::get_env_t; +using ::beman::execution::get_parallel_scheduler; using ::beman::execution::get_scheduler; using ::beman::execution::get_scheduler_t; using ::beman::execution::get_stop_token; @@ -62,6 +63,7 @@ using ::beman::execution::let_value; using ::beman::execution::let_value_t; using ::beman::execution::operation_state; using ::beman::execution::operation_state_tag; +using ::beman::execution::parallel_scheduler; using ::beman::execution::read_env; using ::beman::execution::receiver; using ::beman::execution::receiver_of; @@ -107,6 +109,13 @@ using ::beman::execution::when_all_with_variant; using ::beman::execution::when_all_with_variant_t; using ::beman::execution::with_awaitable_senders; +namespace parallel_scheduler_replacement { +using ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy; +using ::beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend; +using ::beman::execution::parallel_scheduler_replacement::query_parallel_scheduler_backend; +using ::beman::execution::parallel_scheduler_replacement::receiver_proxy; +} // namespace parallel_scheduler_replacement + } // namespace beman::execution26 // ---------------------------------------------------------------------------- diff --git a/src/beman/execution/CMakeLists.txt b/src/beman/execution/CMakeLists.txt index e3cbe916..6cedc55b 100644 --- a/src/beman/execution/CMakeLists.txt +++ b/src/beman/execution/CMakeLists.txt @@ -144,6 +144,7 @@ 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/parallel_scheduler.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 @@ -350,6 +351,7 @@ if(BEMAN_USE_MODULES) on.cppm operation_state_task.cppm operation_state.cppm + parallel_scheduler.cppm product_type.cppm prop.cppm query_with_default.cppm diff --git a/src/beman/execution/execution-detail.cppm b/src/beman/execution/execution-detail.cppm index 36f1328c..1792ed0c 100644 --- a/src/beman/execution/execution-detail.cppm +++ b/src/beman/execution/execution-detail.cppm @@ -59,6 +59,7 @@ export import beman.execution.detail.non_assignable; export import beman.execution.detail.nothrow_callable; export import beman.execution.detail.notify; export import beman.execution.detail.operation_state_task; +export import beman.execution.detail.parallel_scheduler; export import beman.execution.detail.product_type; export import beman.execution.detail.query_with_default; export import beman.execution.detail.queryable; diff --git a/src/beman/execution/execution.cppm b/src/beman/execution/execution.cppm index 9d0c91d7..a43dcf2e 100644 --- a/src/beman/execution/execution.cppm +++ b/src/beman/execution/execution.cppm @@ -46,6 +46,7 @@ import beman.execution.detail.never_stop_token; 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.prop; import beman.execution.detail.read_env; import beman.execution.detail.run_loop; @@ -242,6 +243,17 @@ export using ::beman::execution::stopped_as_error; // [exec.run.loop], run_loop export using ::beman::execution::run_loop; +// [exec.parallel.scheduler], parallel scheduler +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 + // [exec.consumers], consumers export using ::beman::execution::sync_wait_t; export using ::beman::execution::sync_wait_with_variant_t; diff --git a/src/beman/execution/parallel_scheduler.cppm b/src/beman/execution/parallel_scheduler.cppm new file mode 100644 index 00000000..8fcf75e4 --- /dev/null +++ b/src/beman/execution/parallel_scheduler.cppm @@ -0,0 +1,19 @@ +module; +// src/beman/execution/parallel_scheduler.cppm -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include + +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/tests/beman/execution/CMakeLists.txt b/tests/beman/execution/CMakeLists.txt index 59e58f6a..d6b49092 100644 --- a/tests/beman/execution/CMakeLists.txt +++ b/tests/beman/execution/CMakeLists.txt @@ -36,6 +36,7 @@ list( exec-just.test exec-let.test exec-on.test + exec-parallel-scheduler.test exec-opstate-start.test exec-opstate.test exec-prop.test diff --git a/tests/beman/execution/exec-associate.test.cpp b/tests/beman/execution/exec-associate.test.cpp index 9d2987ed..07144f32 100644 --- a/tests/beman/execution/exec-associate.test.cpp +++ b/tests/beman/execution/exec-associate.test.cpp @@ -119,7 +119,6 @@ TEST(exec_associate) { static_assert(std::same_as, test_std::completion_signatures_of_t>>); -#ifndef _MSC_VER //-dk:TODO MSVC++ struggles with more than one of these test using snd1_t = decltype(test_std::associate(test_std::just(std::string{}), null_token{})); static_assert(std::same_as, test_std::completion_signatures_of_t>>); @@ -136,10 +135,8 @@ TEST(exec_associate) { using snd4_t = decltype(test_std::associate(test_std::just(std::ref(i)), null_token{})); static_assert(std::same_as)>, test_std::completion_signatures_of_t>>); -#endif } -#ifndef _MSC_VER //-dk:TODO MSVC++ struggles with more than one of these test // Identity behavior with null_token for value path + piping works. { { @@ -241,5 +238,4 @@ TEST(exec_associate) { ASSERT(completes_with_value(test_std::just(1) | test_std::associate(null_token{}))); ASSERT(!completes_with_value(test_std::just(1) | test_std::associate(expired_token{}))); } -#endif } diff --git a/tests/beman/execution/exec-bulk.test.cpp b/tests/beman/execution/exec-bulk.test.cpp index 0febe816..01a8b0a7 100644 --- a/tests/beman/execution/exec-bulk.test.cpp +++ b/tests/beman/execution/exec-bulk.test.cpp @@ -391,10 +391,10 @@ struct pstl_for_each_sender { template auto set_value(Args&&... args) noexcept -> void { try { - auto iota = std::views::iota(Shape(0), shape); - std::for_each(policy, std::ranges::begin(iota), std::ranges::end(iota), [&](Shape i) { + auto indices = std::views::iota(Shape(0), shape); + std::for_each(policy, std::ranges::begin(indices), std::ranges::end(indices), [&](Shape i) { if constexpr (IsChunked) { - std::invoke(fn, i, i + 1, args...); + std::invoke(fn, i, i + Shape(1), args...); } else { std::invoke(fn, i, args...); }; @@ -430,7 +430,8 @@ struct pstl_for_each_sender { template auto connect(Rcvr rcvr) && noexcept { - return test_std::connect(child, receiver{std::move(rcvr), std::move(policy), std::move(shape), std::move(fn)}); + return test_std::connect(std::move(child), + receiver{std::move(rcvr), std::move(policy), std::move(shape), std::move(fn)}); } auto get_env() const noexcept { return test_std::get_env(child); } diff --git a/tests/beman/execution/exec-just.test.cpp b/tests/beman/execution/exec-just.test.cpp index 8feab9f9..3ce04652 100644 --- a/tests/beman/execution/exec-just.test.cpp +++ b/tests/beman/execution/exec-just.test.cpp @@ -191,16 +191,17 @@ auto test_just_allocator() -> void { ASSERT(resource.count == 0u); auto copy(std::make_obj_using_allocator(std::pmr::polymorphic_allocator<>(&resource), str)); test::use(copy); - ASSERT(resource.count == 1u); + auto old_count = resource.count; + ASSERT(old_count > 0u); auto env{test_std::get_env(receiver)}; auto alloc{test_std::get_allocator(env)}; test::use(alloc); - ASSERT(resource.count == 1u); + ASSERT(resource.count == old_count); auto state{test_std::connect(std::move(sender), memory_receiver{&resource})}; test::use(state); - ASSERT(resource.count == 2u); + ASSERT(resource.count > old_count); } auto test_completion_signatures() -> void { @@ -220,10 +221,7 @@ TEST(exec_just) { test_just_constraints(); test_just(); test_completion_signatures(); -#ifndef _MSC_VER - //-dk:TODO re-enable allocator test for MSVC++ test_just_allocator(); -#endif } catch (...) { // NOLINTNEXTLINE(cert-dcl03-c,hicpp-static-assert,misc-static-assert) ASSERT(nullptr == "the just tests shouldn't throw"); diff --git a/tests/beman/execution/exec-parallel-scheduler.test.cpp b/tests/beman/execution/exec-parallel-scheduler.test.cpp new file mode 100644 index 00000000..7758ebb3 --- /dev/null +++ b/tests/beman/execution/exec-parallel-scheduler.test.cpp @@ -0,0 +1,83 @@ +// src/beman/execution/tests/exec-parallel-scheduler.test.cpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#ifdef BEMAN_HAS_MODULES +import beman.execution; +import beman.execution.detail.schedule_result_t; +#else +#include +#endif + +namespace { +namespace replaceability = test_std::parallel_scheduler_replacement; + +struct proxy : replaceability::receiver_proxy { + auto set_value() noexcept -> void override {} + auto set_error(::std::exception_ptr) noexcept -> void override {} + auto set_stopped() noexcept -> void override {} +}; + +struct bulk_proxy : replaceability::bulk_item_receiver_proxy { + auto set_value() noexcept -> void override {} + auto set_error(::std::exception_ptr) noexcept -> void override {} + auto set_stopped() noexcept -> void override {} + auto execute(::std::size_t, ::std::size_t) noexcept -> void override {} +}; + +struct backend : replaceability::parallel_scheduler_backend { + auto schedule(replaceability::receiver_proxy&, ::std::span<::std::byte>) noexcept -> void override {} + auto schedule_bulk_chunked(::std::size_t, + replaceability::bulk_item_receiver_proxy&, + ::std::span<::std::byte>) noexcept -> void override {} + auto schedule_bulk_unchunked(::std::size_t, + replaceability::bulk_item_receiver_proxy&, + ::std::span<::std::byte>) noexcept -> void override {} +}; + +auto test_parallel_scheduler_synopsis() -> void { + static_assert(!::std::default_initializable); + static_assert(::std::copy_constructible); + static_assert(::std::move_constructible); + static_assert(test_std::scheduler); + + static_assert(::std::same_as); + static_assert(::std::same_as, + test_std::parallel_scheduler::sender>); + static_assert(test_std::sender); + static_assert(::std::same_as()), + test_std::completion_signatures>); + + static_assert( + noexcept(test_std::get_forward_progress_guarantee(::std::declval()))); + static_assert(::std::same_as())), + test_std::forward_progress_guarantee>); +} + +auto test_replaceability_synopsis() -> void { + static_assert(::std::is_abstract_v); + static_assert(::std::is_abstract_v); + static_assert(::std::is_abstract_v); + static_assert(::std::derived_from); + static_assert(::std::derived_from); + static_assert(::std::same_as().template try_query(0)), ::std::optional>); + static_assert(::std::same_as>); +} +} // namespace + +TEST(exec_parallel_scheduler) { + test_parallel_scheduler_synopsis(); + test_replaceability_synopsis(); +} diff --git a/tests/beman/execution/exec-then.test.cpp b/tests/beman/execution/exec-then.test.cpp index ec0abef7..101fde6a 100644 --- a/tests/beman/execution/exec-then.test.cpp +++ b/tests/beman/execution/exec-then.test.cpp @@ -58,6 +58,8 @@ struct sender { } }; +constexpr auto consume_all = [](auto&&...) {}; + template auto test_has(auto cpo, auto in_sender, auto fun) -> void { static_assert(test_std::receiver); @@ -70,14 +72,9 @@ auto test_has(auto cpo, auto in_sender, auto fun) -> void { static_assert(requires { { in_sender | cpo(fun) } -> test_std::sender; }); -#ifndef _MSC_VER - //-dk:TODO re-enable this test static_assert(requires { - { - in_sender | cpo(fun) | cpo([](auto&&...) {}) - } -> test_std::sender; + { in_sender | cpo(fun) | cpo(consume_all) } -> test_std::sender; }); -#endif auto sender{cpo(in_sender, fun)}; auto op{test_std::connect(::std::move(sender), receiver{})}; test_std::start(op);