Replace sender_executor with P2300-realistic scheduler model

Remove sender_executor (no P2300 equivalent) and replace with
pool_scheduler obtained via sender_thread_pool::get_scheduler(),
matching beman::net's io_context pattern. Streams hold
sender_thread_pool* directly. Bridge posting now goes through
schedule → connect → start via heap-allocated scheduled_resume,
since P2300 has no post(coroutine_handle<>).

Author: sgerbino

bench/beman/main.cpp

@@ -246,9 +246,8 @@ int main()
     // Native — sndr_read_stream
     {
         sender_thread_pool pool(1);
-        auto ex = pool.get_executor();
-        sndr_read_stream stream{ex};
-        pool_scheduler sched{ex};
+        sndr_read_stream stream{&pool};
+        auto sched = pool.get_scheduler();
         int count = OPS_PER_CELL;
         char buf[64];
         auto before = g_alloc_count.load(
@@ -275,9 +274,8 @@ int main()
     // Abstract — sndr_io_read_stream
     {
         sender_thread_pool pool(1);
-        auto ex = pool.get_executor();
-        sndr_io_read_stream_impl stream{ex};
-        pool_scheduler sched{ex};
+        sndr_io_read_stream_impl stream{&pool};
+        auto sched = pool.get_scheduler();
         int count = OPS_PER_CELL;
         char buf[64];
         auto before = g_alloc_count.load(
@@ -305,9 +303,8 @@ int main()
     // Type erased — sndr_any_read_stream
     {
         sender_thread_pool pool(1);
-        auto ex = pool.get_executor();
-        sndr_any_read_stream stream(sndr_read_stream{ex});
-        pool_scheduler sched{ex};
+        sndr_any_read_stream stream(sndr_read_stream{&pool});
+        auto sched = pool.get_scheduler();
         int count = OPS_PER_CELL;
         char buf[64];
         auto before = g_alloc_count.load(
@@ -337,7 +334,7 @@ int main()
     {
         sender_thread_pool pool(1);
         ioaw_read_stream stream;
-        pool_scheduler sched{pool.get_executor()};
+        auto sched = pool.get_scheduler();
         int count = OPS_PER_CELL;
         char buf[64];
         auto before = g_alloc_count.load(
@@ -365,7 +362,7 @@ int main()
     {
         sender_thread_pool pool(1);
         ioaw_io_read_stream_impl stream;
-        pool_scheduler sched{pool.get_executor()};
+        auto sched = pool.get_scheduler();
         int count = OPS_PER_CELL;
         char buf[64];
         auto before = g_alloc_count.load(
@@ -397,7 +394,7 @@ int main()
         sender_thread_pool pool(1);
         ioaw_read_stream concrete;
         capy::any_read_stream stream(&concrete);
-        pool_scheduler sched{pool.get_executor()};
+        auto sched = pool.get_scheduler();
         int count = OPS_PER_CELL;
         char buf[64];
         auto before = g_alloc_count.load(
@@ -459,9 +456,8 @@ int main()
     // Native — sndr_read_stream
     {
         sender_thread_pool pool(1);
-        auto ex = pool.get_executor();
-        sender_as_capy_executor adapter{ex};
-        sndr_read_stream stream{ex};
+        sender_as_capy_executor adapter{&pool};
+        sndr_read_stream stream{&pool};
         capy::run_async(adapter)(
             capy_accept_sndr(stream, grid[run][CAPY_TASK][NATIVE_STREAM][BRIDGED_EXEC_MODEL]));
         pool.join();
@@ -470,9 +466,8 @@ int main()
     // Abstract — sndr_io_read_stream
     {
         sender_thread_pool pool(1);
-        auto ex = pool.get_executor();
-        sender_as_capy_executor adapter{ex};
-        sndr_io_read_stream_impl stream{ex};
+        sender_as_capy_executor adapter{&pool};
+        sndr_io_read_stream_impl stream{&pool};
         capy::run_async(adapter)(
             capy_accept_sndr(
                 static_cast<sndr_io_read_stream&>(stream),
@@ -483,9 +478,8 @@ int main()
     // Type erased — sndr_any_read_stream
     {
         sender_thread_pool pool(1);
-        auto ex = pool.get_executor();
-        sender_as_capy_executor adapter{ex};
-        sndr_any_read_stream stream(sndr_read_stream{ex});
+        sender_as_capy_executor adapter{&pool};
+        sndr_any_read_stream stream(sndr_read_stream{&pool});
         capy::run_async(adapter)(
             capy_accept_sndr(stream, grid[run][CAPY_TASK][TYPE_ERASED_STREAM][BRIDGED_EXEC_MODEL]));
         pool.join();
@@ -498,9 +492,8 @@ int main()
     // Native — sndr_read_stream
     {
         sender_thread_pool pool(1);
-        auto ex = pool.get_executor();
-        sndr_read_stream stream{ex};
-        pool_scheduler sched{ex};
+        sndr_read_stream stream{&pool};
+        auto sched = pool.get_scheduler();
         auto* mr = capy::get_recycling_memory_resource();
         bex::sync_wait(bex::starts_on(sched,
             bex_accept(
@@ -513,9 +506,8 @@ int main()
     // Abstract — sndr_io_read_stream
     {
         sender_thread_pool pool(1);
-        auto ex = pool.get_executor();
-        sndr_io_read_stream_impl stream{ex};
-        pool_scheduler sched{ex};
+        sndr_io_read_stream_impl stream{&pool};
+        auto sched = pool.get_scheduler();
         auto* mr = capy::get_recycling_memory_resource();
         bex::sync_wait(bex::starts_on(sched,
             bex_accept(
@@ -529,9 +521,8 @@ int main()
     // Type erased — sndr_any_read_stream
     {
         sender_thread_pool pool(1);
-        auto ex = pool.get_executor();
-        sndr_any_read_stream stream(sndr_read_stream{ex});
-        pool_scheduler sched{ex};
+        sndr_any_read_stream stream(sndr_read_stream{&pool});
+        auto sched = pool.get_scheduler();
         auto* mr = capy::get_recycling_memory_resource();
         bex::sync_wait(bex::starts_on(sched,
             bex_accept(
@@ -546,9 +537,8 @@ int main()
     // Native — ioaw_read_stream
     {
         sender_thread_pool pool(1);
-        auto ex = pool.get_executor();
         ioaw_read_stream stream;
-        pool_scheduler sched{ex};
+        auto sched = pool.get_scheduler();
         auto* mr = capy::get_recycling_memory_resource();
         bex::sync_wait(bex::starts_on(sched,
             bex_accept_ioaw(
@@ -561,9 +551,8 @@ int main()
     // Abstract — ioaw_io_read_stream
     {
         sender_thread_pool pool(1);
-        auto ex = pool.get_executor();
         ioaw_io_read_stream_impl stream;
-        pool_scheduler sched{ex};
+        auto sched = pool.get_scheduler();
         auto* mr = capy::get_recycling_memory_resource();
         bex::sync_wait(bex::starts_on(sched,
             bex_accept_ioaw(
@@ -577,10 +566,9 @@ int main()
     // Type erased — capy::any_read_stream
     {
         sender_thread_pool pool(1);
-        auto ex = pool.get_executor();
         ioaw_read_stream concrete;
         capy::any_read_stream stream(&concrete);
-        pool_scheduler sched{ex};
+        auto sched = pool.get_scheduler();
         auto* mr = capy::get_recycling_memory_resource();
         bex::sync_wait(bex::starts_on(sched,
             bex_accept_ioaw(

bench/beman/sender_io_env.hpp

@@ -10,9 +10,9 @@
 //
 // Beman execution environment for benchmarks.
 //
-// Provides a minimal scheduler, executor, and io_env that bridge
-// sender_thread_pool to beman::execution::task. Used by any
-// benchmark that co_awaits senders from an io_task.
+// Provides pool_scheduler (the P2300 scheduler for
+// sender_thread_pool), the capy executor adapter, and
+// the io_env for beman::execution::task.
 //
 
 #ifndef BOOST_CAPY_BENCH_SENDER_IO_ENV_HPP
@@ -32,31 +32,28 @@
 #include <type_traits>
 #include <utility>
 
-// Adapter making sender_executor satisfy capy's Executor
-// concept so capy::task can run on sender_thread_pool.
+// Adapter making sender_thread_pool satisfy capy's
+// Executor concept so capy::task can run on it.
 struct sender_as_capy_executor
 {
-    sender_executor ex_;
+    sender_thread_pool* pool_;
 
     boost::capy::execution_context& context() const noexcept
     {
-        return *ex_.pool_;
+        return *pool_;
     }
 
     void on_work_started() const noexcept
     {
-        ex_.pool_->on_work_started();
+        pool_->on_work_started();
     }
 
     void on_work_finished() const noexcept
     {
-        ex_.pool_->on_work_finished();
+        pool_->on_work_finished();
     }
 
-    void post(boost::capy::continuation& c) const
-    {
-        ex_.post(c.h);
-    }
+    void post(boost::capy::continuation& c) const;
 
     // Return the handle for symmetric transfer so the
     // caller resumes the coroutine inline. Posting would
@@ -78,17 +75,17 @@ struct pool_scheduler
 {
     using scheduler_concept = ex::scheduler_t;
 
-    sender_executor ex_;
+    sender_thread_pool* pool_;
 
     struct env
     {
-        sender_executor ex_;
+        sender_thread_pool* pool_;
 
         auto query(
             ex::get_completion_scheduler_t<ex::set_value_t> const&
         ) const noexcept
         {
-            return pool_scheduler{ex_};
+            return pool_scheduler{pool_};
         }
     };
 
@@ -98,11 +95,11 @@ struct pool_scheduler
         using operation_state_concept = ex::operation_state_t;
 
         std::remove_cvref_t<Receiver> rcvr_;
-        sender_executor ex_;
+        sender_thread_pool* pool_;
 
-        op_state(Receiver rcvr, sender_executor ex)
+        op_state(Receiver rcvr, sender_thread_pool* pool)
             : rcvr_(std::move(rcvr))
-            , ex_(ex)
+            , pool_(pool)
         {}
 
         op_state(op_state const&) = delete;
@@ -117,7 +114,7 @@ struct pool_scheduler
 
         void start() & noexcept
         {
-            ex_.enqueue(this);
+            pool_->enqueue(this);
         }
     };
 
@@ -127,35 +124,96 @@ struct pool_scheduler
         using completion_signatures =
             ex::completion_signatures<ex::set_value_t()>;
 
-        sender_executor ex_;
+        sender_thread_pool* pool_;
 
-        auto get_env() const noexcept { return env{ex_}; }
+        auto get_env() const noexcept { return env{pool_}; }
 
         template <ex::receiver Receiver>
         auto connect(Receiver&& rcvr)
             -> op_state<std::remove_cvref_t<Receiver>>
         {
-            return {std::forward<Receiver>(rcvr), ex_};
+            return {std::forward<Receiver>(rcvr), pool_};
         }
     };
 
     auto query(
         boost::capy::get_io_executor_t const&
     ) const noexcept -> sender_as_capy_executor
     {
-        return sender_as_capy_executor{ex_};
+        return sender_as_capy_executor{pool_};
     }
 
-    auto schedule() -> sender { return {ex_}; }
+    auto schedule() -> sender { return {pool_}; }
     bool operator==(pool_scheduler const&) const = default;
 };
 
+inline pool_scheduler
+sender_thread_pool::get_scheduler() noexcept
+{
+    return pool_scheduler{this};
+}
+
+// P2300 has no post(coroutine_handle<>). To resume a
+// coroutine on a scheduler you must go through
+// schedule → connect → start. The operation state
+// must be heap-allocated because the coroutine is
+// suspended and cannot host it.
+struct scheduled_resume
+{
+    struct receiver
+    {
+        using receiver_concept = ex::receiver_t;
+
+        scheduled_resume* self_;
+
+        void set_value() && noexcept
+        {
+            auto h = self_->h_;
+            delete self_;
+            h.resume();
+        }
+
+        void set_error(auto&&) && noexcept
+        {
+            std::terminate();
+        }
+
+        void set_stopped() && noexcept
+        {
+            std::terminate();
+        }
+    };
+
+    using op_state_t =
+        pool_scheduler::op_state<receiver>;
+
+    std::coroutine_handle<> h_;
+    op_state_t op_;
+
+    scheduled_resume(
+        pool_scheduler sched,
+        std::coroutine_handle<> h)
+        : h_(h)
+        , op_(ex::connect(
+            sched.schedule(),
+            receiver{this}))
+    {}
+};
+
+inline void sender_as_capy_executor::post(
+    boost::capy::continuation& c) const
+{
+    auto* p = new scheduled_resume(
+        pool_scheduler{pool_}, c.h);
+    ex::start(p->op_);
+}
+
 struct io_env
 {
     using scheduler_type = pool_scheduler;
     using allocator_type = std::pmr::polymorphic_allocator<std::byte>;
 
-    sender_executor executor;
+    sender_thread_pool* pool_ = nullptr;
 
     io_env() = default;
 
@@ -164,7 +222,7 @@ struct io_env
             pool_scheduler{ex::get_scheduler(e)};
         }
     io_env(Env const& e)
-        : executor(pool_scheduler{ex::get_scheduler(e)}.ex_)
+        : pool_(pool_scheduler{ex::get_scheduler(e)}.pool_)
     {}
 };