From 009c2c6e62a9b7e37788b176baa52334187e15e9 Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Thu, 27 Aug 2026 17:01:07 +0200 Subject: [PATCH 1/2] refactor(example): use the canonical as_sender bridge in cuda/pipeline cuda/pipeline carried a copy of bench/stdexec's awaitable_sender.hpp, which splits a compound io_result at runtime into set_value(n) or set_error(ec). The canonical bridge in example/awaitable-sender rejects such awaitables at compile time because exclusive completion channels would drop a partial read's byte count. The tree therefore had two as_sender definitions that disagreed on the paper's central question. Include the canonical header instead and rewrite scene 2 around the sanctioned route: a task wrapper (read_into) that moves the byte count out through a side channel and hands the bridge only the error code. Both scene 2 paths still run; the error path shows the count reaching the caller while upon_error sees the error. The await_sender direction keeps its local stdexec copy, since the canonical sender-bridge copy targets beman.execution. --- example/cuda/pipeline/README.md | 24 +- example/cuda/pipeline/awaitable_sender.hpp | 568 --------------------- example/cuda/pipeline/cuda_pipeline.cu | 49 +- 3 files changed, 44 insertions(+), 597 deletions(-) delete mode 100644 example/cuda/pipeline/awaitable_sender.hpp diff --git a/example/cuda/pipeline/README.md b/example/cuda/pipeline/README.md index 6a465a759..6f8ca2cd2 100644 --- a/example/cuda/pipeline/README.md +++ b/example/cuda/pipeline/README.md @@ -11,18 +11,18 @@ built but not run (P4251R0): When the CUDA stream signals completion, the coroutine resumes on the capy executor with the kernel's result. -2. **Scene 2 (Direction 2).** `boost::capy::test::stream::read_some` is +2. **Scene 2 (Direction 2).** A read over `boost::capy::test::stream` is exposed as a stdexec sender via `boost::capy::as_sender`, composed with `stdexec::upon_error`, and driven by `stdexec::sync_wait`. Two runs: a happy-path read, and a peer-close that exercises the `upon_error` arm. - The example wraps `read_some` (a raw IoAwaitable) rather than - `boost::capy::read` (a `task>`). The bridge's `start()` - does not perform symmetric transfer to a wrapped task's own coroutine - handle, so wrapping a task in `as_sender` hangs. Wrapping a raw - IoAwaitable works because its `await_suspend` is either ready-with-data - or returns `noop_coroutine()` after stashing the continuation for the - peer to resume. + `read_some` returns `io_result`, which `as_sender` rejects at + compile time: sender completion channels are exclusive, so routing the + error through `set_error` would drop the byte count of a partial read. + The scene therefore wraps the read in a `task` + (`read_into`) that moves the count out through a side channel and + returns only the error code. On the error path the count still reaches + the caller while `upon_error` sees the error. 3. **Scene 3 (P4251R0), built but not run.** `handle_request` shows the inference-handler shape: a type-erased `any_read_stream` read, GPU @@ -34,9 +34,11 @@ built but not run (P4251R0): kernel and hopping `continues_on(cpu)` before the host-only bridge, and takes a CPU scheduler the paper's signature omits. -The bridge headers (`awaitable_sender.hpp`, `sender_awaitable.hpp`) are -copied verbatim from `bench/stdexec/`; the bridge in the bench was already -written against NVIDIA/stdexec. +`as_sender` comes from the canonical bridge in `example/awaitable-sender`, +included by relative path. `sender_awaitable.hpp` (the `await_sender` +direction) is a local copy of `bench/stdexec/sender_awaitable.hpp`, kept +here because the canonical `example/sender-bridge` copy targets +beman.execution and this example needs NVIDIA/stdexec for nvexec. ## Prerequisites diff --git a/example/cuda/pipeline/awaitable_sender.hpp b/example/cuda/pipeline/awaitable_sender.hpp deleted file mode 100644 index f13050bd5..000000000 --- a/example/cuda/pipeline/awaitable_sender.hpp +++ /dev/null @@ -1,568 +0,0 @@ -// -// Copyright (c) 2026 Vinnie Falco (vinnie.falco@gmail.com) -// Copyright (c) 2026 Steve Gerbino -// -// Distributed under the Boost Software License, Version 1.0. (See accompanying -// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) -// -// Official repository: https://github.com/cppalliance/capy -// - -#ifndef BOOST_CAPY_BENCH_STDEXEC_AWAITABLE_SENDER_HPP -#define BOOST_CAPY_BENCH_STDEXEC_AWAITABLE_SENDER_HPP - -#include -#include -#include -#include -#include - -#include - -#include -#include -#include -#include -#include -#include -#include - -namespace boost::capy { - -// Query CPO for obtaining a Capy-compatible executor -// from a P2300 environment. The returned object must -// satisfy Capy's Executor concept. Environments that -// host IoAwaitables via the as_sender bridge must -// answer this query. -struct get_io_executor_t -{ - static consteval auto query( - stdexec::forwarding_query_t) noexcept -> bool - { - return true; - } - - template - requires requires(Env const& env) { - env.query( - std::declval()); - } - auto operator()(Env const& env) const noexcept - { - return env.query(*this); - } -}; - -inline constexpr get_io_executor_t get_io_executor{}; - -namespace detail { - -template -struct has_tuple_protocol : std::false_type {}; - -template -struct has_tuple_protocol::type, - typename std::tuple_element<0, T>::type>> - : std::true_type {}; - -template::value> -struct is_ec_outcome : std::is_same {}; - -template -struct is_ec_outcome - : std::bool_constant< - std::tuple_size_v == 1 && - std::is_same_v< - std::tuple_element_t<0, T>, - std::error_code>> -{}; - -template -constexpr bool is_ec_outcome_v = - std::is_same_v || - is_ec_outcome::value; - -template::value> -struct is_compound_ec_result : std::false_type {}; - -template -struct is_compound_ec_result - : std::bool_constant< - std::tuple_size_v >= 2 && - std::is_same_v< - std::tuple_element_t<0, T>, - std::error_code>> -{}; - -template -constexpr bool is_compound_ec_result_v = - is_compound_ec_result::value; - -struct frame_cb -{ - void (*resume)(frame_cb*); - void (*destroy)(frame_cb*); - void* data; -}; - -// Return the concrete executor by value, trying get_io_executor -// on the env first, then falling back to the start scheduler. -template -auto resolve_executor(Env const& env) -{ - if constexpr (requires { get_io_executor(env); }) - return get_io_executor(env); - else - return stdexec::get_start_scheduler(env) - .query(get_io_executor_t{}); -} - -} // namespace detail - -/** Sender that wraps an IoAwaitable. - - When connected or co_awaited, the bridge queries - the receiver's or promise's environment for a - Capy-compatible executor via get_io_executor. - The executor is stored by value in the operation - state and used to construct the io_env passed to - the IoAwaitable's await_suspend. - - @tparam IoAw The IoAwaitable type. -*/ -template -struct awaitable_sender -{ - using sender_concept = stdexec::sender_tag; - - using result_type = decltype( - std::declval&>().await_resume()); - - static auto make_sigs() - { - if constexpr (std::is_void_v) - return stdexec::completion_signatures< - stdexec::set_value_t(), - stdexec::set_error_t(std::exception_ptr), - stdexec::set_stopped_t()>{}; - else if constexpr ( - detail::is_compound_ec_result_v) - return stdexec::completion_signatures< - stdexec::set_value_t( - std::tuple_element_t<1, result_type>), - stdexec::set_error_t(std::error_code), - stdexec::set_error_t(std::exception_ptr), - stdexec::set_stopped_t()>{}; - else if constexpr ( - detail::is_ec_outcome_v) - return stdexec::completion_signatures< - stdexec::set_value_t(), - stdexec::set_error_t(std::error_code), - stdexec::set_error_t(std::exception_ptr), - stdexec::set_stopped_t()>{}; - else - return stdexec::completion_signatures< - stdexec::set_value_t(result_type), - stdexec::set_error_t(std::exception_ptr), - stdexec::set_stopped_t()>{}; - } - - using completion_signatures = decltype(make_sigs()); - - IoAw aw_; - - template - struct op_state - { - using operation_state_concept = - stdexec::operation_state_tag; - - // Concrete executor type deduced from the receiver's - // environment. Stored by value to avoid the dangling - // pointer that executor_ref would produce when the - // source is a temporary (scheduler query or prop). - using executor_type = decltype( - detail::resolve_executor( - stdexec::get_env( - std::declval()))); - - IoAw aw_; - Receiver rcvr_; - executor_type ex_; - io_env env_; - detail::frame_cb cb_; - - op_state(IoAw aw, Receiver rcvr) - : aw_(std::move(aw)) - , rcvr_(std::move(rcvr)) - , ex_{} - , cb_{} - { - } - - op_state(op_state const&) = delete; - op_state(op_state&&) = delete; - op_state& operator=(op_state const&) = delete; - op_state& operator=(op_state&&) = delete; - - static void - on_resume(detail::frame_cb* p) noexcept - { - auto* self = static_cast(p->data); - self->complete(); - } - - static void - on_destroy(detail::frame_cb*) noexcept - { - } - - void complete() noexcept - { - try - { - if constexpr (std::is_void_v) - { - aw_.await_resume(); - if(env_.stop_token.stop_requested()) - stdexec::set_stopped( - std::move(rcvr_)); - else - stdexec::set_value( - std::move(rcvr_)); - } - else if constexpr ( - detail::is_compound_ec_result_v) - { - auto result = aw_.await_resume(); - if(env_.stop_token.stop_requested()) - { - stdexec::set_stopped( - std::move(rcvr_)); - } - else - { - auto ec = get<0>(result); - if(!ec) - stdexec::set_value( - std::move(rcvr_), - get<1>(std::move(result))); - else - stdexec::set_error( - std::move(rcvr_), ec); - } - } - else if constexpr ( - detail::is_ec_outcome_v) - { - auto result = aw_.await_resume(); - if(env_.stop_token.stop_requested()) - { - stdexec::set_stopped( - std::move(rcvr_)); - } - else - { - std::error_code ec; - if constexpr (std::is_same_v< - result_type, std::error_code>) - ec = result; - else - ec = get<0>(result); - if(!ec) - stdexec::set_value( - std::move(rcvr_)); - else - stdexec::set_error( - std::move(rcvr_), ec); - } - } - else - { - auto result = aw_.await_resume(); - if(env_.stop_token.stop_requested()) - stdexec::set_stopped( - std::move(rcvr_)); - else - stdexec::set_value( - std::move(rcvr_), - std::move(result)); - } - } - catch(...) - { - stdexec::set_error( - std::move(rcvr_), - std::current_exception()); - } - } - - void start() noexcept - { - auto renv = stdexec::get_env(rcvr_); - ex_ = detail::resolve_executor(renv); - - std::stop_token st; - if constexpr (requires { - { renv.query(stdexec::get_stop_token_t{}) } - -> std::convertible_to; }) - { - st = renv.query( - stdexec::get_stop_token_t{}); - } - - env_ = io_env{ex_, st, nullptr}; - - if(aw_.await_ready()) - { - complete(); - return; - } - - cb_.resume = &on_resume; - cb_.destroy = &on_destroy; - cb_.data = this; - - auto h = std::coroutine_handle<>::from_address( - static_cast(&cb_)); - - auto resumed = detail::call_await_suspend( - &aw_, h, &env_); - if(resumed == h) - complete(); - } - }; - - template - auto connect(Receiver rcvr) && - -> op_state - { - return op_state( - std::move(aw_), std::move(rcvr)); - } - - template - auto connect(Receiver rcvr) const& - -> op_state - { - return op_state(aw_, std::move(rcvr)); - } - - // Bypass stdexec's sender_awaitable when co_awaited - // from a coroutine that provides get_io_executor or - // a start scheduler with get_io_executor. Adapts the - // IoAwaitable's 2-arg await_suspend to the standard - // 1-arg protocol. - template - auto as_awaitable(Promise& promise) && - { - auto penv = promise.get_env(); - auto ex = detail::resolve_executor(penv); - - std::stop_token st; - if constexpr (requires { - { penv.query(stdexec::get_stop_token_t{}) } - -> std::convertible_to; }) - { - st = penv.query( - stdexec::get_stop_token_t{}); - } - - using executor_type = decltype(ex); - - struct aw - { - IoAw aw_; - executor_type ex_; - std::stop_token st_; - io_env env_; - - bool await_ready() noexcept - { - return aw_.await_ready(); - } - - std::coroutine_handle<> - await_suspend(std::coroutine_handle<> h) - { - env_ = io_env{ex_, st_, nullptr}; - return aw_.await_suspend(h, &env_); - } - - auto await_resume() - { - return aw_.await_resume(); - } - }; - - return aw{std::move(aw_), std::move(ex), st, {}}; - } -}; - -/** Create a stdexec sender from an IoAwaitable. - - The bridge routes the awaitable's result through sender - channels based on its type: - - - `void` - calls `set_value()`. - - `error_code` (or a single-element tuple-like whose - element 0 is `error_code`) - calls `set_value()` - when the code is zero, `set_error(ec)` otherwise. - - Any other single value `T` - calls `set_value(T)`. - - Compound results whose element 0 is `error_code` - with additional elements are rejected at compile - time. Wrap the operation in a `task` - that inspects the compound result and returns the - error code. - - When connected or co_awaited, the bridge queries the - receiver's or promise's environment for a Capy executor - via get_io_executor. The environment must answer this - query with an object satisfying Capy's Executor concept. - - @param aw The IoAwaitable to wrap. - @return A sender whose completion channels reflect - the awaitable's result type. -*/ -template -auto as_sender(IoAw&& aw) -{ - return awaitable_sender>{ - std::forward(aw)}; -} - -// split_ec: sender adapter that routes error_code to -// set_value() or set_error(ec) at runtime. - -namespace detail { - -template -struct split_ec_sender -{ - using sender_concept = stdexec::sender_tag; - - using completion_signatures = - stdexec::completion_signatures< - stdexec::set_value_t(), - stdexec::set_error_t(std::error_code), - stdexec::set_error_t(std::exception_ptr), - stdexec::set_stopped_t()>; - - Sender sndr_; - - template - struct ec_receiver - { - using receiver_concept = stdexec::receiver_tag; - - Receiver rcvr_; - - auto get_env() const noexcept - { - return stdexec::get_env(rcvr_); - } - - void set_value(std::error_code ec) && noexcept - { - if (!ec) - stdexec::set_value( - std::move(rcvr_)); - else - stdexec::set_error( - std::move(rcvr_), ec); - } - - void set_value() && noexcept - { - stdexec::set_value( - std::move(rcvr_)); - } - - template - void set_error(E&& e) && noexcept - { - stdexec::set_error( - std::move(rcvr_), - std::forward(e)); - } - - void set_stopped() && noexcept - { - stdexec::set_stopped( - std::move(rcvr_)); - } - }; - - template - struct op_state - { - using operation_state_concept = - stdexec::operation_state_tag; - - using inner_op_t = decltype( - stdexec::connect( - std::declval(), - std::declval>())); - - inner_op_t op_; - - op_state(Sender sndr, Receiver rcvr) - : op_(stdexec::connect( - std::move(sndr), - ec_receiver{std::move(rcvr)})) - { - } - - op_state(op_state const&) = delete; - op_state(op_state&&) = delete; - op_state& operator=(op_state const&) = delete; - op_state& operator=(op_state&&) = delete; - - void start() noexcept - { - stdexec::start(op_); - } - }; - - template - auto connect(Receiver rcvr) && - -> op_state - { - return op_state( - std::move(sndr_), std::move(rcvr)); - } - - template - auto connect(Receiver rcvr) const& - -> op_state - { - return op_state( - sndr_, std::move(rcvr)); - } -}; - -} // namespace detail - -/** Split an `error_code` value channel into success and error channels. - - Takes a sender that completes with `set_value(error_code)` and - routes it at runtime: `set_value()` when the code is zero, - `set_error(ec)` otherwise. No exceptions. - - @param sndr The predecessor sender. - @return A sender completing with `set_value()`, - `set_error(error_code)`, or `set_stopped()`. -*/ -template -auto split_ec(Sender&& sndr) -{ - return detail::split_ec_sender< - std::decay_t>{ - std::forward(sndr)}; -} - -} // namespace boost::capy - -#endif diff --git a/example/cuda/pipeline/cuda_pipeline.cu b/example/cuda/pipeline/cuda_pipeline.cu index 20dff38fd..9bd22d06b 100644 --- a/example/cuda/pipeline/cuda_pipeline.cu +++ b/example/cuda/pipeline/cuda_pipeline.cu @@ -12,14 +12,14 @@ // terminal action is a real CUDA __global__ kernel scheduled on // nvexec::stream_scheduler. // -// Scene 2 (Direction 2): a capy IoAwaitable (capy::read over a -// deterministic in-process stream pair) is exposed as a stdexec -// sender, then composed with stdexec::upon_error, and consumed -// via stdexec::sync_wait. Both the happy path and an injected-eof -// path are exercised. +// Scene 2 (Direction 2): a read over a deterministic in-process +// stream pair is exposed as a stdexec sender through the canonical +// as_sender bridge (example/awaitable-sender), composed with +// stdexec::upon_error, and consumed via stdexec::sync_wait. Both the +// happy path and an injected-eof path are exercised. // -#include "awaitable_sender.hpp" +#include "../../awaitable-sender/awaitable_sender.hpp" #include "sender_awaitable.hpp" #include @@ -212,14 +212,26 @@ run_scene1(capy::thread_pool& pool, float& out) std::rethrow_exception(err); } -// Scene 2: capy::read exposed as a stdexec sender, composed with +// Scene 2: a stream read exposed as a stdexec sender, composed with // stdexec::upon_error, driven by sync_wait. write_env injects the // capy executor that the as_sender bridge needs to drive the // underlying IoAwaitable. -// stream::read_some returns a raw IoAwaitable, which the bridge -// expects. (capy::read returns a task>, and the -// bridge's start() does not perform symmetric transfer to the -// task's own handle, so wrapping a task hangs.) +// +// read_some returns io_result, which as_sender rejects at +// compile time: sender completion channels are exclusive, so routing +// the error through set_error would drop the byte count of a partial +// read. The sanctioned route is a task wrapper that moves +// the count out through a side channel and hands the bridge only the +// error code. +capy::task +read_into(capy::test::stream& s, capy::mutable_buffer buf, + std::size_t& n_out) +{ + auto [ec, n] = co_await s.read_some(buf); + n_out = n; + co_return ec; +} + void scene2_happy_path(capy::thread_pool& pool) { @@ -229,11 +241,12 @@ scene2_happy_path(capy::thread_pool& pool) b.provide(payload); char buf[64]; + std::size_t n = 0; auto sndr = ex::write_env( capy::as_sender( - a.read_some(capy::mutable_buffer(buf, sizeof buf))), + read_into(a, capy::mutable_buffer(buf, sizeof buf), n)), ex::prop{capy::get_io_executor, pool.get_executor()}) - | ex::upon_error([](auto e) noexcept -> std::size_t { + | ex::upon_error([](auto e) noexcept { if constexpr (std::is_same_v< std::decay_t, std::error_code>) { @@ -246,7 +259,6 @@ scene2_happy_path(capy::thread_pool& pool) auto result = ex::sync_wait(std::move(sndr)); assert(result.has_value()); - auto const [n] = *result; assert(n == payload.size()); assert(std::string_view(buf, n) == payload); @@ -262,27 +274,28 @@ scene2_error_path(capy::thread_pool& pool) b.close(); char buf[64]; + std::size_t n = 0; bool fired = false; std::error_code observed; auto sndr = ex::write_env( capy::as_sender( - a.read_some(capy::mutable_buffer(buf, sizeof buf))), + read_into(a, capy::mutable_buffer(buf, sizeof buf), n)), ex::prop{capy::get_io_executor, pool.get_executor()}) - | ex::upon_error([&](auto e) noexcept -> std::size_t { + | ex::upon_error([&](auto e) noexcept { if constexpr (std::is_same_v< std::decay_t, std::error_code>) { fired = true; observed = e; } - return 0; }); auto result = ex::sync_wait(std::move(sndr)); assert(result.has_value()); - auto const [n] = *result; + // The error travelled on the error channel and the byte count + // still reached the caller through the wrapper's side channel. assert(fired); assert(observed); std::cout From 34d10460f74f2e376b362593cc29781a1317dfce Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Thu, 27 Aug 2026 17:22:56 +0200 Subject: [PATCH 2/2] refactor(bench): rename the benchmark bridges to as_sender_lossy The bench copies of the awaitable-to-sender bridge accept a compound io_result and split it at runtime, dropping the byte count that accompanies an error. The canonical as_sender in example/awaitable-sender rejects that at compile time. Two functions with one name and opposite contracts invite copying the wrong one, as cuda/pipeline did. Name the benchmark bridges as_sender_lossy and say why they exist: they measure the bridge without the task wrapper the canonical bridge requires. No behavior change. --- bench/beman/awaitable_sender.hpp | 12 +++++++++--- bench/beman/main.cpp | 18 +++++++++--------- bench/stdexec/awaitable_sender.hpp | 10 ++++++++-- bench/stdexec/main.cpp | 16 ++++++++-------- bench/stdexec/sender_io_env.hpp | 2 +- 5 files changed, 35 insertions(+), 23 deletions(-) diff --git a/bench/beman/awaitable_sender.hpp b/bench/beman/awaitable_sender.hpp index 0e1f3d8cc..07eb0f701 100644 --- a/bench/beman/awaitable_sender.hpp +++ b/bench/beman/awaitable_sender.hpp @@ -31,7 +31,7 @@ namespace boost::capy { // Query CPO for obtaining a Capy-compatible executor // from a P2300 environment. The returned object must // satisfy Capy's Executor concept. Environments that -// host IoAwaitables via the as_sender bridge must +// host IoAwaitables via the as_sender_lossy bridge must // answer this query. struct get_io_executor_t { @@ -339,7 +339,7 @@ struct awaitable_sender // Bypass beman's sender_awaitable when co_awaited // from a bex::task. Adapts the IoAwaitable's 2-arg // await_suspend to standard 1-arg protocol, avoiding - // the double bridge (as_sender + sender_awaitable). + // the double bridge (as_sender_lossy + sender_awaitable). template auto as_awaitable(Promise& promise) && { @@ -414,8 +414,14 @@ struct awaitable_sender @return A sender whose completion channels reflect the awaitable's result type. */ +// Benchmark-only bridge. Unlike the canonical as_sender in +// example/awaitable-sender, this one accepts a compound io_result and +// splits it at runtime: set_value(n) on success, set_error(ec) on +// failure, dropping the byte count that accompanied the error. The +// benchmark keeps it to measure the bridge without the task +// wrapper the canonical bridge requires; do not copy it into examples. template -auto as_sender(IoAw&& aw) +auto as_sender_lossy(IoAw&& aw) { return awaitable_sender>{ std::forward(aw)}; diff --git a/bench/beman/main.cpp b/bench/beman/main.cpp index 60d096f31..1fc73c626 100644 --- a/bench/beman/main.cpp +++ b/bench/beman/main.cpp @@ -195,10 +195,10 @@ auto bex_accept( } // =================================================================== -// Table 2: bex::task — Column B (awaitable via as_sender bridge) +// Table 2: bex::task — Column B (awaitable via as_sender_lossy bridge) // // The stream returns an IoAwaitable. bex::task consumes it by -// wrapping in as_sender which bridges the awaitable to a sender. +// wrapping in as_sender_lossy which bridges the awaitable to a sender. // =================================================================== template @@ -209,7 +209,7 @@ auto bex_session_ioaw( { char buf[64]; for (int i = 0; i < INNER_LOOPS; ++i) - (void)co_await capy::as_sender( + (void)co_await capy::as_sender_lossy( stream.read_some( capy::mutable_buffer(buf, sizeof(buf)))); } @@ -341,7 +341,7 @@ int main() after - before}; } - // Col B: Awaitable (via as_sender bridge) + // Col B: Awaitable (via as_sender_lossy bridge) // Native — ioaw_read_stream @@ -357,7 +357,7 @@ int main() bex::sync_wait(bex::starts_on(sched, repeat_until( bex::let_value(bex::just(), [&]() { - return capy::as_sender(stream.read_some( + return capy::as_sender_lossy(stream.read_some( capy::mutable_buffer(buf, sizeof(buf)))); }), [&count]() { return --count == 0; }))); @@ -385,7 +385,7 @@ int main() bex::sync_wait(bex::starts_on(sched, repeat_until( bex::let_value(bex::just(), [&]() { - return capy::as_sender( + return capy::as_sender_lossy( static_cast( stream).read_some( capy::mutable_buffer( @@ -417,7 +417,7 @@ int main() bex::sync_wait(bex::starts_on(sched, repeat_until( bex::let_value(bex::just(), [&]() { - return capy::as_sender(stream.read_some( + return capy::as_sender_lossy(stream.read_some( capy::mutable_buffer(buf, sizeof(buf)))); }), [&count]() { return --count == 0; }))); @@ -474,7 +474,7 @@ int main() bex::sync_wait(bex::starts_on(sched, repeat_until( bex::let_value(bex::just(), [&]() { - return capy::as_sender(stream.read_some( + return capy::as_sender_lossy(stream.read_some( capy::mutable_buffer(buf, sizeof(buf)))); }), [&count]() { return --count == 0; }))); @@ -638,7 +638,7 @@ int main() pool.join(); } - // Col B: Awaitable (via as_sender bridge) + // Col B: Awaitable (via as_sender_lossy bridge) // Native — ioaw_read_stream { diff --git a/bench/stdexec/awaitable_sender.hpp b/bench/stdexec/awaitable_sender.hpp index 5f54601f4..aaadc14ba 100644 --- a/bench/stdexec/awaitable_sender.hpp +++ b/bench/stdexec/awaitable_sender.hpp @@ -31,7 +31,7 @@ namespace boost::capy { // Query CPO for obtaining a Capy-compatible executor // from a P2300 environment. The returned object must // satisfy Capy's Executor concept. Environments that -// host IoAwaitables via the as_sender bridge must +// host IoAwaitables via the as_sender_lossy bridge must // answer this query. struct get_io_executor_t { @@ -425,8 +425,14 @@ struct awaitable_sender @return A sender whose completion channels reflect the awaitable's result type. */ +// Benchmark-only bridge. Unlike the canonical as_sender in +// example/awaitable-sender, this one accepts a compound io_result and +// splits it at runtime: set_value(n) on success, set_error(ec) on +// failure, dropping the byte count that accompanied the error. The +// benchmark keeps it to measure the bridge without the task +// wrapper the canonical bridge requires; do not copy it into examples. template -auto as_sender(IoAw&& aw) +auto as_sender_lossy(IoAw&& aw) { return awaitable_sender>{ std::forward(aw)}; diff --git a/bench/stdexec/main.cpp b/bench/stdexec/main.cpp index c1edc93ad..8ad5e6685 100644 --- a/bench/stdexec/main.cpp +++ b/bench/stdexec/main.cpp @@ -180,7 +180,7 @@ auto exec_accept(Stream& stream, cell_result& out) } // ----------------------------------------------------------- -// Table 3: exec::task - Column B (awaitable via as_sender) +// Table 3: exec::task - Column B (awaitable via as_sender_lossy) // // exec::task's promise env carries only get_start_scheduler // (type-erased __any_scheduler), which does not propagate @@ -197,7 +197,7 @@ auto exec_session_ioaw( char buf[64]; for (int i = 0; i < INNER_LOOPS; ++i) (void)co_await stdexec::write_env( - capy::as_sender( + capy::as_sender_lossy( stream.read_some( capy::mutable_buffer(buf, sizeof(buf)))), stdexec::prop{capy::get_io_executor, ex}); @@ -361,7 +361,7 @@ int main() after - before}; } - // Col B: Awaitable (via as_sender bridge) + // Col B: Awaitable (via as_sender_lossy bridge) // Native - ioaw_read_stream { @@ -378,7 +378,7 @@ int main() exec::repeat_until( stdexec::let_value(stdexec::just(), [&]() { return stdexec::write_env( - capy::as_sender(stream.read_some( + capy::as_sender_lossy(stream.read_some( capy::mutable_buffer(buf, sizeof(buf)))), stdexec::prop{capy::get_io_executor, adapter}); }) @@ -407,7 +407,7 @@ int main() exec::repeat_until( stdexec::let_value(stdexec::just(), [&]() { return stdexec::write_env( - capy::as_sender( + capy::as_sender_lossy( static_cast( stream).read_some( capy::mutable_buffer( @@ -440,7 +440,7 @@ int main() exec::repeat_until( stdexec::let_value(stdexec::just(), [&]() { return stdexec::write_env( - capy::as_sender(stream.read_some( + capy::as_sender_lossy(stream.read_some( capy::mutable_buffer(buf, sizeof(buf)))), stdexec::prop{capy::get_io_executor, adapter}); }) @@ -469,7 +469,7 @@ int main() exec::repeat_until( stdexec::let_value(stdexec::just(), [&]() { return stdexec::write_env( - capy::as_sender(stream.read_some( + capy::as_sender_lossy(stream.read_some( capy::mutable_buffer(buf, sizeof(buf)))), stdexec::prop{capy::get_io_executor, adapter}); }) @@ -658,7 +658,7 @@ int main() pool.request_stop(); } - // Col B: Awaitable (via as_sender bridge) + // Col B: Awaitable (via as_sender_lossy bridge) // Native - ioaw_read_stream { diff --git a/bench/stdexec/sender_io_env.hpp b/bench/stdexec/sender_io_env.hpp index 019edf8fe..de238db4b 100644 --- a/bench/stdexec/sender_io_env.hpp +++ b/bench/stdexec/sender_io_env.hpp @@ -149,7 +149,7 @@ struct pool_schedule_sender // Scheduler wrapper that delegates schedule() to exec::static_thread_pool // but answers boost::capy's get_io_executor_t query. Required by the -// capy::as_sender bridge in awaitable_sender.hpp, which queries the +// capy::as_sender_lossy bridge in awaitable_sender.hpp, which queries the // receiver-env scheduler for the capy executor at instantiation time. struct pool_scheduler {