Pr/cuda pipeline canonical as sender

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<class Promise>
     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<error_code>
+// wrapper the canonical bridge requires; do not copy it into examples.
 template<class IoAw>
-auto as_sender(IoAw&& aw)
+auto as_sender_lossy(IoAw&& aw)
 {
     return awaitable_sender<std::decay_t<IoAw>>{
         std::forward<IoAw>(aw)};

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 <class Stream>
@@ -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<ioaw_io_read_stream&>(
                             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
     {

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<error_code>
+// wrapper the canonical bridge requires; do not copy it into examples.
 template<class IoAw>
-auto as_sender(IoAw&& aw)
+auto as_sender_lossy(IoAw&& aw)
 {
     return awaitable_sender<std::decay_t<IoAw>>{
         std::forward<IoAw>(aw)};

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<ioaw_io_read_stream&>(
                                 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
     {

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
 {

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<io_result<size_t>>`). 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<size_t>`, 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<std::error_code>`
+   (`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