Simplify get_completion_signatures implementation for some sender adaptors

examples/CMakeLists.txt

@@ -30,6 +30,7 @@ set(EXAMPLES
     intro_5_consumer
     playground
     sender_demo
+    sleep_sort
     stackoverflow
     stopping
     stop_token
@@ -41,6 +42,10 @@ if(BEMAN_USE_MODULES)
     list(APPEND EXAMPLES modules_and_header)
 endif()
 
+if(BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND)
+    list(APPEND EXAMPLES async_inclusive_scan)
+endif()
+
 foreach(EXAMPLE ${EXAMPLES})
     set(EXAMPLE_TARGET ${PROJECT_NAME}.${EXAMPLE})
     add_executable(${EXAMPLE_TARGET})

examples/async_inclusive_scan.cpp

@@ -0,0 +1,69 @@
+// examples/async_inclusive_scan.cpp                       -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#include <beman/execution/detail/common.hpp>
+#ifdef BEMAN_HAS_IMPORT_STD
+import std;
+#else
+#include <cstddef>
+#include <algorithm>
+#include <iterator>
+#include <numeric>
+#include <ranges>
+#include <print>
+#include <vector>
+#endif
+#ifdef BEMAN_HAS_MODULES
+import beman.execution;
+#else
+#include <beman/execution.hpp>
+#endif
+
+namespace ex = beman::execution;
+
+namespace {
+template <ex::scheduler                                      Scheduler,
+          std::forward_iterator                              ForwardIt,
+          std::sentinel_for<ForwardIt>                       Sentinel,
+          std::output_iterator<std::iter_value_t<ForwardIt>> OutputIt>
+ex::sender auto
+async_inclusive_scan(Scheduler scheduler, std::size_t tile_count, ForwardIt first, Sentinel last, OutputIt dest) {
+    const std::size_t n         = std::ranges::distance(first, last);
+    const std::size_t tile_size = (n + tile_count - 1) / tile_count;
+    return ex::just(std::vector<int>(tile_count + 1)) | ex::let_value([=](std::vector<int>& partials) {
+               return ex::schedule(scheduler) |
+                      ex::bulk(ex::par,
+                               tile_count,
+                               [=, &partials](std::size_t i) {
+                                   auto start = i * tile_size;
+                                   auto end   = std::min(n, (i + 1) * tile_size);
+                                   partials[i + 1] =
+                                       *std::ranges::prev(std::inclusive_scan(std::ranges::next(first, start),
+                                                                              std::ranges::next(first, end),
+                                                                              std::ranges::next(dest, start)));
+                               }) |
+                      ex::then(
+                          [&partials] { std::inclusive_scan(partials.begin(), partials.end(), partials.begin()); }) |
+                      ex::bulk(ex::par,
+                               tile_count,
+                               [=, &partials](std::size_t i) {
+                                   auto start = i * tile_size;
+                                   auto end   = std::min(n, (i + 1) * tile_size);
+                                   std::for_each(std::ranges::next(dest, start),
+                                                 std::ranges::next(dest, end),
+                                                 [&](int& e) { e = partials[i] + e; });
+                               }) |
+                      ex::then([=] { return std::ranges::next(dest, n); });
+           });
+}
+} // namespace
+
+auto main() -> int {
+    constexpr std::size_t  n         = 32uz;
+    ex::parallel_scheduler scheduler = ex::get_parallel_scheduler();
+    std::vector<int>       input(n);
+    std::vector<int>       output(input.size());
+    std::iota(input.begin(), input.end(), 1); // NOLINT
+    auto _ = ex::sync_wait(async_inclusive_scan(scheduler, 4, input.begin(), input.end(), output.begin())).value();
+    std::println("{}", output);
+}

examples/intro_timer.hpp

@@ -123,8 +123,8 @@ struct intro::timer {
 
     auto run() { return run_sender{this}; }
 
-    template <typename T>
-    auto resume_after(std::chrono::duration<T> d) {
+    template <typename Rep, typename Period>
+    auto resume_after(std::chrono::duration<Rep, Period> d) {
         auto ms(std::chrono::duration_cast<std::chrono::milliseconds>(d));
         return sender{this, ms};
     }

examples/sleep_sort.cpp

@@ -0,0 +1,45 @@
+// examples/sleep_sort.cpp                                 -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#include <beman/execution/detail/common.hpp>
+#ifdef BEMAN_HAS_IMPORT_STD
+import std;
+#else
+#include <chrono>
+#include <print>
+#include <queue>
+#include <vector>
+#endif
+#ifdef BEMAN_HAS_MODULES
+import beman.execution;
+#else
+#include <beman/execution.hpp>
+#endif
+#include "intro_timer.hpp"
+
+namespace ex = beman::execution;
+
+namespace {
+auto sleep_sort(std::span<const int> input) -> ex::task<std::vector<int>> {
+    intro::timer     timer;
+    std::vector<int> result;
+    result.reserve(input.size());
+    ex::simple_counting_scope scope;
+    for (auto i : input) {
+        ex::spawn(timer.resume_after(std::chrono::milliseconds(i)) |
+                      ex::then([i, &result]() noexcept { result.push_back(i); }),
+                  scope.get_token());
+    }
+    auto sch = co_await ex::read_env(ex::get_start_scheduler);
+    co_await ex::when_all(timer.run() | ex::write_env(ex::prop{ex::get_delegation_scheduler, sch}), scope.join());
+    co_return result;
+}
+} // namespace
+
+auto main() -> int {
+    auto        tick = std::chrono::steady_clock::now();
+    std::vector input{5, 10, 1, 3, 4, 2, 9, 11, 8, 6, 7, 3, 4};
+    auto [result] = ex::sync_wait(sleep_sort(input)).value();
+    auto tock     = std::chrono::steady_clock::now();
+    std::println("result = {}, take {}", result, std::chrono::duration_cast<std::chrono::milliseconds>(tock - tick));
+}

include/beman/execution/detail/affine.hpp

@@ -78,7 +78,7 @@ namespace beman::execution::detail {
  * scheduler obtained the receiver's environment. The get_start_scheduler query is used
  * to obtain the scheduler on which the sender gets started.
  */
-struct affine_t : ::beman::execution::sender_adaptor_closure<affine_t> {
+struct affine_t {
     /**
      * @brief Adapt a sender with affine.
      *

include/beman/execution/detail/bulk.hpp

@@ -19,31 +19,39 @@ import beman.execution.detail.basic_sender;
 import beman.execution.detail.completion_signatures;
 import beman.execution.detail.completion_signatures_of_t;
 import beman.execution.detail.default_impls;
+import beman.execution.detail.env;
 import beman.execution.detail.execution_policy;
 import beman.execution.detail.forward_like;
+import beman.execution.detail.fwd_env;
+import beman.execution.detail.get_completion_signatures;
 import beman.execution.detail.make_sender;
 import beman.execution.detail.meta.combine;
 import beman.execution.detail.meta.unique;
 import beman.execution.detail.product_type;
 import beman.execution.detail.sender;
 import beman.execution.detail.sender_adaptor_closure;
 import beman.execution.detail.sender_for;
+import beman.execution.detail.sender_in;
 import beman.execution.detail.set_error;
 import beman.execution.detail.set_value;
 #else
 #include <beman/execution/detail/basic_sender.hpp>
 #include <beman/execution/detail/completion_signatures.hpp>
 #include <beman/execution/detail/completion_signatures_of_t.hpp>
 #include <beman/execution/detail/default_impls.hpp>
+#include <beman/execution/detail/env.hpp>
 #include <beman/execution/detail/execution_policy.hpp>
 #include <beman/execution/detail/forward_like.hpp>
+#include <beman/execution/detail/fwd_env.hpp>
+#include <beman/execution/detail/get_completion_signatures.hpp>
 #include <beman/execution/detail/make_sender.hpp>
 #include <beman/execution/detail/meta_combine.hpp>
 #include <beman/execution/detail/meta_unique.hpp>
 #include <beman/execution/detail/product_type.hpp>
 #include <beman/execution/detail/sender.hpp>
 #include <beman/execution/detail/sender_adaptor_closure.hpp>
 #include <beman/execution/detail/sender_for.hpp>
+#include <beman/execution/detail/sender_in.hpp>
 #include <beman/execution/detail/set_error.hpp>
 #include <beman/execution/detail/set_value.hpp>
 #endif
@@ -129,21 +137,26 @@ struct bulk_algo_t {
     template <typename, typename...>
     struct get_signatures;
     template <typename Policy, typename Shape, typename F, typename Sender, typename... Env>
+        requires ::beman::execution::sender_in<Sender, ::beman::execution::detail::fwd_env_t<Env>...>
     struct get_signatures<
         ::beman::execution::detail::
             basic_sender<bulk_algo_t, ::beman::execution::detail::product_type<Policy, Shape, F>, Sender>,
         Env...> {
-        using type = typename ::beman::execution::detail::bulk_transform_signatures<
-            is_chunked,
-            F,
-            Shape,
-            ::beman::execution::completion_signatures_of_t<Sender, Env...>>::type;
+        static consteval auto get() {
+            [[maybe_unused]] auto child_sigs =
+                ::beman::execution::get_completion_signatures<Sender, ::beman::execution::detail::fwd_env_t<Env>...>();
+            using child_sigs_t = decltype(child_sigs);
+            using type         = typename ::beman::execution::detail::
+                bulk_transform_signatures<is_chunked, F, Shape, child_sigs_t>::type;
+            return type{};
+        }
     };
 
   public:
     template <typename Sender, typename... Env>
-    static consteval auto get_completion_signatures() noexcept {
-        return typename get_signatures<::std::remove_cvref_t<Sender>, Env...>::type{};
+        requires requires { get_signatures<::std::remove_cvref_t<Sender>, Env...>{}; }
+    static consteval auto get_completion_signatures() {
+        return get_signatures<::std::remove_cvref_t<Sender>, Env...>::get();
     }
 
     struct impls_for : ::beman::execution::detail::default_impls {
@@ -212,7 +225,7 @@ struct bulk_t {
     }
 
     template <::beman::execution::detail::sender_for<bulk_t> Sender, typename Env>
-    auto transform_sender(::beman::execution::set_value_t, Sender&& sndr, const Env&) const {
+    static auto transform_sender(::beman::execution::set_value_t, Sender&& sndr, const Env&) {
         auto data  = ::beman::execution::detail::forward_like<Sender>(sndr.template get<1>());
         auto child = ::beman::execution::detail::forward_like<Sender>(sndr.template get<2>());
 
@@ -223,23 +236,10 @@ struct bulk_t {
         return bulk_chunked_t{}(::std::move(child),
                                 policy,
                                 shape,
-                                this->wrap_chunked<::std::remove_cvref_t<decltype(shape)>>(::std::move(f)));
+                                bulk_t::wrap_chunked<::std::remove_cvref_t<decltype(shape)>>(::std::move(f)));
     }
 
   private:
-    template <typename, typename...>
-    struct get_signatures;
-    template <typename Policy, typename Shape, typename F, typename Sender, typename... Env>
-    struct get_signatures<::beman::execution::detail::
-                              basic_sender<bulk_t, ::beman::execution::detail::product_type<Policy, Shape, F>, Sender>,
-                          Env...> {
-        using type = typename ::beman::execution::detail::bulk_transform_signatures<
-            false,
-            F,
-            Shape,
-            ::beman::execution::completion_signatures_of_t<Sender, Env...>>::type;
-    };
-
     template <typename Shape, typename Fn>
     static auto wrap_chunked(Fn f) noexcept {
         return [f = ::std::move(f)]<typename... Args>(Shape begin, Shape end, Args&&... args) noexcept(
@@ -250,10 +250,15 @@ struct bulk_t {
         };
     }
 
+    template <typename Sender>
+    using transformed_sender = decltype(bulk_t::transform_sender(
+        ::beman::execution::set_value, ::std::declval<Sender>(), ::std::declval<::beman::execution::env<>>()));
+
   public:
-    template <typename Sender, typename... Env>
-    static consteval auto get_completion_signatures() noexcept {
-        return typename get_signatures<::std::remove_cvref_t<Sender>, Env...>::type{};
+    template <::beman::execution::sender Sender>
+        requires ::beman::execution::sender_in<transformed_sender<Sender>>
+    static consteval auto get_completion_signatures() {
+        return ::beman::execution::get_completion_signatures<transformed_sender<Sender>>();
     }
 };
 

include/beman/execution/detail/into_variant.hpp

@@ -64,7 +64,7 @@ import beman.execution.detail.variant_or_empty;
 // ----------------------------------------------------------------------------
 
 namespace beman::execution::detail {
-struct into_variant_t : ::beman::execution::sender_adaptor_closure<into_variant_t> {
+struct into_variant_t {
     template <::beman::execution::sender Sender>
     auto operator()(Sender&& sender) const {
         return ::beman::execution::detail::make_sender(*this, {}, ::std::forward<Sender>(sender));

include/beman/execution/detail/on.hpp

@@ -72,7 +72,7 @@ import beman.execution.detail.transform_sender;
 // ----------------------------------------------------------------------------
 
 namespace beman::execution::detail {
-struct on_t : ::beman::execution::sender_adaptor_closure<on_t> {
+struct on_t {
     template <typename Env, typename OutSndr>
     static auto dry_run(OutSndr&& out_sndr) noexcept {
         auto& [_, data, child] = out_sndr;

include/beman/execution/detail/run_loop.hpp

@@ -117,7 +117,7 @@ class run_loop {
 
         run_loop* loop;
 
-        auto schedule() noexcept -> sender { return {this->loop}; }
+        auto schedule() const noexcept -> sender { return {this->loop}; }
         auto operator==(const scheduler&) const -> bool = default;
 
         static constexpr auto query(::beman::execution::get_forward_progress_guarantee_t) noexcept

include/beman/execution/detail/spawn.hpp

@@ -20,6 +20,9 @@ import beman.execution.detail.queryable;
 import beman.execution.detail.receiver;
 import beman.execution.detail.scope_token;
 import beman.execution.detail.sender;
+import beman.execution.detail.sender_in;
+import beman.execution.detail.sends;
+import beman.execution.detail.set_error;
 import beman.execution.detail.spawn_get_allocator;
 import beman.execution.detail.start;
 import beman.execution.detail.write_env;
@@ -31,6 +34,9 @@ import beman.execution.detail.write_env;
 #include <beman/execution/detail/receiver.hpp>
 #include <beman/execution/detail/scope_token.hpp>
 #include <beman/execution/detail/sender.hpp>
+#include <beman/execution/detail/sender_in.hpp>
+#include <beman/execution/detail/sends.hpp>
+#include <beman/execution/detail/set_error.hpp>
 #include <beman/execution/detail/spawn_get_allocator.hpp>
 #include <beman/execution/detail/start.hpp>
 #include <beman/execution/detail/write_env.hpp>
@@ -92,6 +98,11 @@ struct spawn_t {
         auto [all, senv] = ::beman::execution::detail::spawn_get_allocator(new_sender, env);
 
         using sender_t = decltype(::beman::execution::write_env(::std::move(new_sender), senv));
+        static_assert(::beman::execution::sender_in<sender_t, ::beman::execution::env<>>);
+        static_assert(
+            !::beman::execution::detail::sends<::beman::execution::set_error_t, sender_t, ::beman::execution::env<>>,
+            "`the sender must either succeed or be canceled; it cannot fail!");
+
         using state_t  = state<decltype(all), Token, sender_t>;
         using alloc_t  = typename ::std::allocator_traits<decltype(all)>::template rebind_alloc<state_t>;
         using traits_t = ::std::allocator_traits<alloc_t>;