[WIP] Implement P2079R10 parallel_scheduler

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<Tag>(q, e...),
                                      get_completion_domain_t<::beman::execution::set_value_t>{},
-                                     ::std::forward<Q>(q),
                                      ::std::forward<E>(e)...);
                              }) {
             return ::beman::execution::detail::try_query(::beman::execution::get_completion_scheduler<Tag>(q, e...),
                                                          get_completion_domain_t<::beman::execution::set_value_t>{},
-                                                         ::std::forward<Q>(q),
                                                          ::std::forward<E>(e)...);
         } else if constexpr (::beman::execution::scheduler<Q> && 0u != sizeof...(E)) {
             return ::beman::execution::default_domain{};

include/beman/execution/detail/stop_propagator.hpp

@@ -0,0 +1,77 @@
+// include/beman/execution/detail/stop_propagator.hpp                -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_STOP_PROPAGATOR
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_STOP_PROPAGATOR
+
+#include <beman/execution/detail/common.hpp>
+#ifdef BEMAN_HAS_IMPORT_STD
+import std;
+#else
+#include <functional>
+#include <optional>
+#include <type_traits>
+#include <utility>
+#endif
+#ifdef BEMAN_HAS_MODULES
+import beman.execution.detail.inplace_stop_source;
+import beman.execution.detail.stoppable_source;
+import beman.execution.detail.stoppable_token;
+import beman.execution.detail.stop_callback_for_t;
+import beman.execution.detail.unstoppable_token;
+#else
+#include <beman/execution/detail/inplace_stop_source.hpp>
+#include <beman/execution/detail/stoppable_source.hpp>
+#include <beman/execution/detail/stoppable_token.hpp>
+#include <beman/execution/detail/stop_callback_for_t.hpp>
+#include <beman/execution/detail/unstoppable_token.hpp>
+#endif
+
+// ----------------------------------------------------------------------------
+
+namespace beman::execution::detail {
+template <::beman::execution::stoppable_token StopToken>
+class stop_propagator {
+  public:
+    explicit stop_propagator(StopToken stop_token) noexcept
+        : stop_callback_(::std::move(stop_token), ::std::bind_front(&stop_propagator::propagate, this)) {}
+
+    auto get_token() const noexcept -> ::std::optional<::beman::execution::inplace_stop_token> {
+        return stop_source_.get_token();
+    }
+
+  private:
+    auto propagate() noexcept -> void { stop_source_.request_stop(); }
+
+  private:
+    using stop_cb_t = decltype(::std::bind_front(&stop_propagator::propagate, ::std::declval<stop_propagator*>()));
+
+    ::beman::execution::inplace_stop_source                       stop_source_;
+    ::beman::execution::stop_callback_for_t<StopToken, stop_cb_t> stop_callback_;
+};
+
+template <>
+class stop_propagator<::beman::execution::inplace_stop_token> {
+  public:
+    explicit stop_propagator(::beman::execution::inplace_stop_token stop_token) noexcept : stop_token_(stop_token) {}
+
+    auto get_token() const noexcept -> ::std::optional<::beman::execution::inplace_stop_token> { return stop_token_; }
+
+  private:
+    ::beman::execution::inplace_stop_token stop_token_;
+};
+
+template <::beman::execution::unstoppable_token StopToken>
+class stop_propagator<StopToken> {
+  public:
+    explicit stop_propagator(StopToken) noexcept {}
+
+    static auto get_token() noexcept -> ::std::optional<::beman::execution::inplace_stop_token> {
+        return ::std::nullopt;
+    }
+};
+} // namespace beman::execution::detail
+
+// ----------------------------------------------------------------------------
+
+#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_STOP_PROPAGATOR

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 <beman/execution/detail/matching_sig.hpp>
 #include <beman/execution/detail/on.hpp>
 #include <beman/execution/detail/operation_state.hpp>
+#include <beman/execution/detail/parallel_scheduler.hpp>
 #include <beman/execution/detail/prop.hpp>
 #include <beman/execution/detail/read_env.hpp>
 #include <beman/execution/detail/receiver.hpp>

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
 
 // ----------------------------------------------------------------------------

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
@@ -187,6 +188,7 @@ target_sources(
                 ${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
@@ -350,6 +352,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
@@ -391,6 +394,7 @@ if(BEMAN_USE_MODULES)
                     state_type.cppm
                     stoppable_callback_for.cppm
                     stop_callback_for_t.cppm
+                    stop_propagator.cppm
                     stop_source.cppm
                     stop_token_of_t.cppm
                     stop_when.cppm

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;

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;

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 <beman/execution/detail/parallel_scheduler.hpp>
+
+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

src/beman/execution/stop_propagator.cppm

@@ -0,0 +1,11 @@
+module;
+// src/beman/execution/stop_propagator.cppm                     -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#include <beman/execution/detail/stop_propagator.hpp>
+
+export module beman.execution.detail.stop_propagator;
+
+namespace beman::execution::detail {
+export using beman::execution::detail::stop_propagator;
+} // namespace beman::execution::detail

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

tests/beman/execution/exec-associate.test.cpp

@@ -119,7 +119,6 @@ TEST(exec_associate) {
         static_assert(std::same_as<test_std::completion_signatures<test_std::set_value_t()>,
                                    test_std::completion_signatures_of_t<snd0_t, test_std::env<>>>);
 
-#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<test_std::set_value_t(std::string)>,
                                    test_std::completion_signatures_of_t<snd1_t, test_std::env<>>>);
@@ -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<test_std::set_value_t(std::reference_wrapper<int>)>,
                                    test_std::completion_signatures_of_t<snd4_t, test_std::env<>>>);
-#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
 }

tests/beman/execution/exec-bulk.test.cpp

@@ -391,10 +391,10 @@ struct pstl_for_each_sender {
         template <typename... Args>
         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 <typename Rcvr>
     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); }

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::string>(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");