Various small fixes to align with LEWG feedback.

Author: lucteo

include/exec/__detail/__system_context_default_impl.hpp

@@ -29,7 +29,6 @@ namespace exec::__system_context_default_impl {
   using namespace stdexec::tags;
   using system_context_replaceability::receiver;
   using system_context_replaceability::bulk_item_receiver;
-  using system_context_replaceability::storage;
   using system_context_replaceability::parallel_scheduler_backend;
   using system_context_replaceability::__parallel_scheduler_backend_factory;
 
@@ -104,17 +103,18 @@ namespace exec::__system_context_default_impl {
   };
 
   /// Ensure that `__storage` is aligned to `__alignment`. Shrinks the storage, if needed, to match desired alignment.
-  inline storage __ensure_alignment(storage __storage, size_t __alignment) noexcept {
-    auto __pn = reinterpret_cast<uintptr_t>(__storage.__data);
+  inline std::span<std::byte>
+    __ensure_alignment(std::span<std::byte> __storage, size_t __alignment) noexcept {
+    auto __pn = reinterpret_cast<uintptr_t>(__storage.data());
     if (__pn % __alignment == 0) {
       return __storage;
-    } else if (__storage.__size < __alignment) {
-      return {nullptr, 0};
+    } else if (__storage.size() < __alignment) {
+      return {};
     } else {
       auto __new_pn = (__pn + __alignment - 1) & ~(__alignment - 1);
       return {
-        reinterpret_cast<void*>(__new_pn),
-        static_cast<uint32_t>(__storage.__size - (__new_pn - __pn))};
+        reinterpret_cast<std::byte*>(__new_pn),
+        static_cast<size_t>(__storage.size() - (__new_pn - __pn))};
     }
   }
 
@@ -126,13 +126,15 @@ namespace exec::__system_context_default_impl {
     bool __on_heap_;
 
     /// Try to construct the operation in the preallocated memory if it fits, otherwise allocate a new operation.
-    static __operation*
-      __construct_maybe_alloc(storage __storage, receiver* __completion, _Sender __sndr) {
+    static __operation* __construct_maybe_alloc(
+      std::span<std::byte> __storage,
+      receiver* __completion,
+      _Sender __sndr) {
       __storage = __ensure_alignment(__storage, alignof(__operation));
-      if (__storage.__data == nullptr || __storage.__size < sizeof(__operation)) {
+      if (__storage.data() == nullptr || __storage.size() < sizeof(__operation)) {
         return new __operation(std::move(__sndr), __completion, true);
       } else {
-        return new (__storage.__data) __operation(std::move(__sndr), __completion, false);
+        return new (__storage.data()) __operation(std::move(__sndr), __completion, false);
       }
     }
 
@@ -174,7 +176,7 @@ namespace exec::__system_context_default_impl {
       bulk_item_receiver* __r_;
 
       void operator()(unsigned long __idx) const noexcept {
-        __r_->start(static_cast<uint32_t>(__idx));
+        __r_->execute(static_cast<uint32_t>(__idx));
       }
     };
 
@@ -187,27 +189,29 @@ namespace exec::__system_context_default_impl {
       std::declval<__bulk_functor>()))>;
 
    public:
-    void schedule(storage __storage, receiver* __r) noexcept override {
+    void schedule(std::span<std::byte> __storage, receiver& __r) noexcept override {
       try {
         auto __sndr = stdexec::schedule(__pool_scheduler_);
         auto __os =
-          __schedule_operation_t::__construct_maybe_alloc(__storage, __r, std::move(__sndr));
+          __schedule_operation_t::__construct_maybe_alloc(__storage, &__r, std::move(__sndr));
         __os->start();
       } catch (std::exception& __e) {
-        __r->set_error(std::current_exception());
+        __r.set_error(std::current_exception());
       }
     }
 
-    void
-      bulk_schedule(uint32_t __size, storage __storage, bulk_item_receiver* __r) noexcept override {
+    void bulk_schedule(
+      uint32_t __size,
+      std::span<std::byte> __storage,
+      bulk_item_receiver& __r) noexcept override {
       try {
         auto __sndr =
-          stdexec::bulk(stdexec::schedule(__pool_scheduler_), __size, __bulk_functor{__r});
+          stdexec::bulk(stdexec::schedule(__pool_scheduler_), __size, __bulk_functor{&__r});
         auto __os =
-          __bulk_schedule_operation_t::__construct_maybe_alloc(__storage, __r, std::move(__sndr));
+          __bulk_schedule_operation_t::__construct_maybe_alloc(__storage, &__r, std::move(__sndr));
         __os->start();
       } catch (std::exception& __e) {
-        __r->set_error(std::current_exception());
+        __r.set_error(std::current_exception());
       }
     }
   };

include/exec/__detail/__system_context_replaceability_api.hpp

@@ -20,9 +20,11 @@
 #include "stdexec/__detail/__execution_fwd.hpp"
 
 #include <cstdint>
+#include <cstddef>
 #include <exception>
 #include <optional>
 #include <memory>
+#include <span>
 
 struct __uuid {
   std::uint64_t __parts1;
@@ -111,14 +113,7 @@ namespace exec::system_context_replaceability {
   /// Receiver for bulk sheduling operations.
   struct bulk_item_receiver : receiver {
     /// Called for each item of a bulk operation, possible on different threads.
-    virtual void start(std::uint32_t) noexcept = 0;
-  };
-
-  /// Describes a storage space.
-  /// Used to pass preallocated storage from the frontend to the backend.
-  struct storage {
-    void* __data;
-    std::uint32_t __size;
+    virtual void execute(std::uint32_t) noexcept = 0;
   };
 
   /// Interface for the parallel scheduler backend.
@@ -129,11 +124,13 @@ namespace exec::system_context_replaceability {
 
     /// Schedule work on parallel scheduler, calling `__r` when done and using `__s` for preallocated
     /// memory.
-    virtual void schedule(storage __s, receiver* __r) noexcept = 0;
+    virtual void schedule(std::span<std::byte> __s, receiver& __r) noexcept = 0;
     /// Schedule bulk work of size `__n` on parallel scheduler, calling `__r` for each item and then
     /// when done, and using `__s` for preallocated memory.
-    virtual void
-      bulk_schedule(std::uint32_t __n, storage __s, bulk_item_receiver* __r) noexcept = 0;
+    virtual void bulk_schedule(
+      std::uint32_t __n,
+      std::span<std::byte> __s,
+      bulk_item_receiver& __r) noexcept = 0;
   };
 
 } // namespace exec::system_context_replaceability

include/exec/system_context.hpp

@@ -121,8 +121,8 @@ namespace exec {
     struct __aligned_storage {
       alignas(_Align) unsigned char __data_[_Size];
 
-      system_context_replaceability::storage __as_storage() noexcept {
-        return {__data_, _Size};
+      std::span<std::byte> __as_storage() noexcept {
+        return {reinterpret_cast<std::byte*>(__data_), _Size};
       }
 
       template <class _T>
@@ -192,7 +192,7 @@ namespace exec {
         auto& __scheduler_impl = __preallocated_.__as<__backend_ptr>();
         auto __impl = std::move(__scheduler_impl);
         std::destroy_at(&__scheduler_impl);
-        __impl->schedule(__preallocated_.__as_storage(), &__rcvr_);
+        __impl->schedule(__preallocated_.__as_storage(), __rcvr_);
       }
 
       /// Object that receives completion from the work described by the sender.
@@ -228,6 +228,10 @@ namespace exec {
       return {__scheduler_};
     }
 
+    /// Value completion happens on the parallel scheduler.
+    parallel_scheduler
+      query(stdexec::get_completion_scheduler_t<stdexec::set_value_t>) const noexcept;
+
     /// Connects `__self` to `__rcvr`, returning the operation state containing the work to be done.
     template <stdexec::receiver _Rcvr>
     auto connect(_Rcvr __rcvr) && noexcept(stdexec::__nothrow_move_constructible<_Rcvr>) //
@@ -354,7 +358,7 @@ namespace exec {
       }
 
       /// Calls the bulk functor passing `__index` and the values from the previous sender.
-      void start(uint32_t __index) noexcept override {
+      void execute(uint32_t __index) noexcept override {
         auto __state = reinterpret_cast<_BulkState*>(this);
         std::apply(
           [&](auto&&... __args) { __state->__fun_(__index, __args...); },
@@ -382,8 +386,7 @@ namespace exec {
       _Rcvr __rcvr_;
 
       /// Function that prepares the preallocated storage for calling the backend.
-      system_context_replaceability::storage (*__prepare_storage_for_backend)(__bulk_state_base*){
-        nullptr};
+      std::span<std::byte> (*__prepare_storage_for_backend)(__bulk_state_base*){nullptr};
 
       __bulk_state_base(_Fn&& __fun, _Rcvr&& __rcvr)
         : __fun_{std::move(__fun)}
@@ -426,7 +429,7 @@ namespace exec {
 
         // Schedule the bulk work on the system scheduler.
         // This will invoke `start` on our receiver multiple times, and then a completion signal (e.g., `set_value`).
-        __scheduler->bulk_schedule(__size, __storage, __r);
+        __scheduler->bulk_schedule(__size, __storage, *__r);
       }
 
       /// Invoked when the previous sender completes with "stopped" to stop the entire work.
@@ -469,8 +472,7 @@ namespace exec {
       __aligned_storage<_PreallocatedSize, _PreallocatedAlign> __preallocated_;
 
       /// Destroys the inner operation state object, and returns the preallocated storage for it to be used by the backend.
-      static system_context_replaceability::storage
-        __prepare_storage_for_backend_impl(__bulk_state_base_t* __base) {
+      static std::span<std::byte> __prepare_storage_for_backend_impl(__bulk_state_base_t* __base) {
         auto* __self = static_cast<__system_bulk_op*>(__base);
         // We don't need anymore the storage for the previous operation state.
         __self->__preallocated_.template __as<__inner_op_state>().~__inner_op_state();
@@ -583,6 +585,11 @@ namespace exec {
     return parallel_scheduler{std::move(__impl)};
   }
 
+  inline parallel_scheduler __parallel_sender::query(
+    stdexec::get_completion_scheduler_t<stdexec::set_value_t>) const noexcept {
+    return __detail::__make_parallel_scheduler_from(stdexec::set_value_t{}, __scheduler_);
+  }
+
   inline auto parallel_scheduler::query(stdexec::get_forward_progress_guarantee_t) const noexcept
     -> stdexec::forward_progress_guarantee {
     return stdexec::forward_progress_guarantee::parallel;

test/exec/test_system_context.cpp

@@ -208,7 +208,7 @@ struct my_parallel_scheduler_backend_impl
     return count_schedules_;
   }
 
-  void schedule(scr::storage __s, scr::receiver* __r) noexcept override {
+  void schedule(std::span<std::byte> __s, scr::receiver& __r) noexcept override {
     count_schedules_++;
     base_t::schedule(__s, __r);
   }
@@ -219,14 +219,15 @@ struct my_parallel_scheduler_backend_impl
 };
 
 struct my_inline_scheduler_backend_impl : scr::parallel_scheduler_backend {
-  void schedule(scr::storage s, scr::receiver* r) noexcept override {
-    r->set_value();
+  void schedule(std::span<std::byte> s, scr::receiver& r) noexcept override {
+    r.set_value();
   }
 
-  void bulk_schedule(uint32_t count, scr::storage s, scr::bulk_item_receiver* r) noexcept override {
+  void bulk_schedule(uint32_t count, std::span<std::byte> s, scr::bulk_item_receiver& r) noexcept
+    override {
     for (uint32_t i = 0; i < count; ++i)
-      r->start(i);
-    r->set_value();
+      r.execute(i);
+    r.set_value();
   }
 };
 

test/exec/test_system_context_replaceability.cpp

@@ -32,7 +32,7 @@ namespace {
 
     my_parallel_scheduler_backend_impl() = default;
 
-    void schedule(scr::storage __s, scr::receiver* __r) noexcept override {
+    void schedule(std::span<std::byte> __s, scr::receiver& __r) noexcept override {
       count_schedules++;
       base_t::schedule(__s, __r);
     }

test_package/test.cpp

@@ -4,7 +4,7 @@
 #include <cstdlib>
 
 int main() {
-  auto x = stdexec::starts_on(exec::system_context().get_scheduler(), stdexec::just(42));
+  auto x = stdexec::starts_on(exec::get_parallel_scheduler(), stdexec::just(42));
   auto [a] = stdexec::sync_wait(std::move(x)).value();
   return a == 42 ? EXIT_SUCCESS : EXIT_FAILURE;
 }