Rename repeat_effect_until to repeat_until with trampoline and sync row

sgerbino · sgerbino · commit 21ca572a2f4d · 2026-04-04T00:21:02.000+02:00
Align with stdexec's naming: repeat_effect_until → repeat_until.
Add depth-bounded trampoline (max_depth = 19) that converts deep
recursion into an iterative drain loop for synchronous completions.

Enable the synchronous benchmark row for Table 1 (sender pipeline),
previously N/A due to stack overflow without a trampoline.

Fix awaitable_sender bridge to handle synchronous completion from
await_suspend returning the caller's handle — call complete()
instead of discarding the returned handle.
diff --git a/bench/beman/awaitable_sender.hpp b/bench/beman/awaitable_sender.hpp
@@ -314,7 +314,10 @@ struct awaitable_sender
             auto h = std::coroutine_handle<>::from_address(
                 static_cast<void*>(&cb_));
 
-            detail::call_await_suspend(&aw_, h, &env_);
+            auto resumed = detail::call_await_suspend(
+                &aw_, h, &env_);
+            if(resumed == h)
+                complete();
         }
     };
 
diff --git a/bench/beman/main.cpp b/bench/beman/main.cpp
@@ -11,7 +11,7 @@
 // I/O Read Stream Benchmark
 //
 // Compares three execution models across three stream abstraction
-// levels. 100M read_some calls per cell, single thread.
+// levels. 20M read_some calls per cell, single thread.
 //
 // Table 1: sender pipeline   (connect/start)
 // Table 2: capy::task        (capy::thread_pool)
@@ -27,7 +27,7 @@
 #include "awaitable_sender.hpp"
 #include "ioaw_read_stream.hpp"
 #include "ioaw_io_read_stream.hpp"
-#include "repeat_effect_until.hpp"
+#include "repeat_until.hpp"
 #include "sender_awaitable.hpp"
 #include "sndr_any_read_stream.hpp"
 #include "sndr_io_read_stream.hpp"
@@ -248,12 +248,14 @@ int main()
     for (int run = 0; run <= NUM_RUNS; ++run)
     {
 
+
     // ---------------------------------------------------------------
-    // Table 1: sender/receiver pipeline (repeat_effect_until)
+    // Table 1: sender/receiver pipeline (repeat_until)
     // ---------------------------------------------------------------
 
     // Col A: Sender (native)
 
+
     // Native — sndr_read_stream
     {
         sender_thread_pool pool(1);
@@ -265,7 +267,7 @@ int main()
             std::memory_order_relaxed);
         auto start = std::chrono::steady_clock::now();
         bex::sync_wait(bex::starts_on(sched,
-            repeat_effect_until(
+            repeat_until(
                 bex::let_value(bex::just(), [&]() {
                     return stream.read_some(
                         capy::mutable_buffer(buf, sizeof(buf)));
@@ -293,7 +295,7 @@ int main()
             std::memory_order_relaxed);
         auto start = std::chrono::steady_clock::now();
         bex::sync_wait(bex::starts_on(sched,
-            repeat_effect_until(
+            repeat_until(
                 bex::let_value(bex::just(), [&]() {
                     return static_cast<sndr_io_read_stream&>(
                         stream).read_some(
@@ -322,7 +324,7 @@ int main()
             std::memory_order_relaxed);
         auto start = std::chrono::steady_clock::now();
         bex::sync_wait(bex::starts_on(sched,
-            repeat_effect_until(
+            repeat_until(
                 bex::let_value(bex::just(), [&]() {
                     return stream.read_some(
                         capy::mutable_buffer(buf, sizeof(buf)));
@@ -341,6 +343,7 @@ int main()
 
     // Col B: Awaitable (via as_sender bridge)
 
+
     // Native — ioaw_read_stream
     {
         sender_thread_pool pool(1);
@@ -352,7 +355,7 @@ int main()
             std::memory_order_relaxed);
         auto start = std::chrono::steady_clock::now();
         bex::sync_wait(bex::starts_on(sched,
-            repeat_effect_until(
+            repeat_until(
                 bex::let_value(bex::just(), [&]() {
                     return capy::as_sender(stream.read_some(
                         capy::mutable_buffer(buf, sizeof(buf))));
@@ -380,7 +383,7 @@ int main()
             std::memory_order_relaxed);
         auto start = std::chrono::steady_clock::now();
         bex::sync_wait(bex::starts_on(sched,
-            repeat_effect_until(
+            repeat_until(
                 bex::let_value(bex::just(), [&]() {
                     return capy::as_sender(
                         static_cast<ioaw_io_read_stream&>(
@@ -412,7 +415,7 @@ int main()
             std::memory_order_relaxed);
         auto start = std::chrono::steady_clock::now();
         bex::sync_wait(bex::starts_on(sched,
-            repeat_effect_until(
+            repeat_until(
                 bex::let_value(bex::just(), [&]() {
                     return capy::as_sender(stream.read_some(
                         capy::mutable_buffer(buf, sizeof(buf))));
@@ -429,16 +432,68 @@ int main()
             after - before};
     }
 
-    // Synchronous — sender pipeline cannot run synchronous
-    // senders. repeat_effect_until requires a trampoline for
-    // synchronous completions, and the trampoline interacts
-    // poorly with the let_value/starts_on sender layering.
-    // Table 1 SYNC_STREAM cells are left at zero.
+
+    // Synchronous — sndr_sync_read_stream (Col A)
+    {
+        sender_thread_pool pool(1);
+        sndr_sync_read_stream stream;
+        auto sched = pool.get_scheduler();
+        int count = OPS_PER_CELL;
+        char buf[64];
+        auto before = g_alloc_count.load(
+            std::memory_order_relaxed);
+        auto start = std::chrono::steady_clock::now();
+        bex::sync_wait(bex::starts_on(sched,
+            repeat_until(
+                bex::let_value(bex::just(), [&]() {
+                    return stream.read_some(
+                        capy::mutable_buffer(buf, sizeof(buf)));
+                }),
+                [&count]() { return --count == 0; })));
+        pool.join();
+        auto elapsed =
+            std::chrono::steady_clock::now() - start;
+        auto after = g_alloc_count.load(
+            std::memory_order_relaxed);
+        grid[run][SENDER_RECEIVER][SYNC_STREAM][NATIVE_EXEC_MODEL] = {
+            std::chrono::duration_cast<
+                std::chrono::nanoseconds>(elapsed).count(),
+            after - before};
+    }
+
+    // Synchronous — ioaw_sync_read_stream (Col B)
+    {
+        sender_thread_pool pool(1);
+        ioaw_sync_read_stream stream;
+        auto sched = pool.get_scheduler();
+        int count = OPS_PER_CELL;
+        char buf[64];
+        auto before = g_alloc_count.load(
+            std::memory_order_relaxed);
+        auto start = std::chrono::steady_clock::now();
+        bex::sync_wait(bex::starts_on(sched,
+            repeat_until(
+                bex::let_value(bex::just(), [&]() {
+                    return capy::as_sender(stream.read_some(
+                        capy::mutable_buffer(buf, sizeof(buf))));
+                }),
+                [&count]() { return --count == 0; })));
+        pool.join();
+        auto elapsed =
+            std::chrono::steady_clock::now() - start;
+        auto after = g_alloc_count.load(
+            std::memory_order_relaxed);
+        grid[run][SENDER_RECEIVER][SYNC_STREAM][BRIDGED_EXEC_MODEL] = {
+            std::chrono::duration_cast<
+                std::chrono::nanoseconds>(elapsed).count(),
+            after - before};
+    }
 
     // ---------------------------------------------------------------
     // Table 2: capy::task (capy::thread_pool)
     // ---------------------------------------------------------------
 
+
     // Native — ioaw_read_stream
     {
         capy::thread_pool pool(1);
@@ -525,6 +580,7 @@ int main()
     // Table 3: beman::execution::task (bex::task<void, io_env>)
     // ---------------------------------------------------------------
 
+
     // Native — sndr_read_stream
     {
         sender_thread_pool pool(1);
@@ -676,18 +732,6 @@ int main()
 
         for (int s = 0; s < NUM_STREAMS; ++s)
         {
-            if (s == SYNC_STREAM &&
-                table == SENDER_RECEIVER)
-            {
-                std::printf(
-                    "  %-18s"
-                    "  %-30s  %-30s\n",
-                    row_labels[s],
-                    "              N/A",
-                    "              N/A");
-                continue;
-            }
-
             double sum[NUM_COLUMNS]{};
             double sum2[NUM_COLUMNS]{};
             double al[NUM_COLUMNS]{};
diff --git a/bench/beman/repeat_until.hpp b/bench/beman/repeat_until.hpp
@@ -1,10 +1,10 @@
 //
-// Adapted from beman execution examples (Apache-2.0 WITH LLVM-exception)
+// Adapted from stdexec (Apache-2.0 WITH LLVM-exception)
 // for benchmark use.
 //
 
-#ifndef BOOST_CAPY_BENCH_REPEAT_EFFECT_UNTIL_HPP
-#define BOOST_CAPY_BENCH_REPEAT_EFFECT_UNTIL_HPP
+#ifndef BOOST_CAPY_BENCH_REPEAT_UNTIL_HPP
+#define BOOST_CAPY_BENCH_REPEAT_UNTIL_HPP
 
 #include <beman/execution/execution.hpp>
 
@@ -29,7 +29,13 @@ struct repeat_connector
     auto start() & noexcept -> void { bex::start(op); }
 };
 
-inline constexpr struct repeat_effect_until_t
+/// Sender algorithm that repeats a child sender until
+/// a predicate returns true. Predicate is called with
+/// no arguments (child values are discarded).
+///
+/// Includes a trampoline that bounds recursion depth
+/// for synchronous completions (max_depth = 19).
+inline constexpr struct repeat_until_t
 {
     template <bex::sender Child, typename Pred>
     struct sender
@@ -47,6 +53,8 @@ inline constexpr struct repeat_effect_until_t
             using operation_state_concept =
                 bex::operation_state_t;
 
+            static constexpr std::size_t max_depth = 19;
+
             struct own_receiver
             {
                 using receiver_concept = bex::receiver_t;
@@ -97,24 +105,58 @@ inline constexpr struct repeat_effect_until_t
             std::optional<repeat_connector<
                 std::remove_cvref_t<Child>,
                 own_receiver>> child_op;
+            std::size_t depth_ = 0;
+            bool draining_ = false;
+            bool again_ = false;
 
             auto start() & noexcept -> void
             {
-                run_one();
+                drain();
             }
 
-            auto run_one() & noexcept -> void
+            // Iterative trampoline that bounds stack
+            // depth for synchronous completions
+            auto drain() & noexcept -> void
             {
-                child_op.emplace(child, own_receiver{this});
-                child_op->start();
+                draining_ = true;
+                do
+                {
+                    again_ = false;
+                    depth_ = 0;
+                    child_op.emplace(
+                        child, own_receiver{this});
+                    child_op->start();
+                }
+                while (again_);
+                draining_ = false;
             }
 
             auto next() & noexcept -> void
             {
                 if (pred())
+                {
                     bex::set_value(std::move(receiver));
-                else
-                    run_one();
+                    return;
+                }
+
+                if (!draining_)
+                {
+                    // Async completion — enter drain loop
+                    drain();
+                    return;
+                }
+
+                if (++depth_ >= max_depth)
+                {
+                    // Hit depth limit — trampoline
+                    again_ = true;
+                    return;
+                }
+
+                // Within limit — recurse inline
+                child_op.emplace(
+                    child, own_receiver{this});
+                child_op->start();
             }
         };
 
@@ -137,6 +179,6 @@ inline constexpr struct repeat_effect_until_t
         return {std::forward<Child>(child),
             std::forward<Pred>(pred)};
     }
-} repeat_effect_until{};
+} repeat_until{};
 
 #endif
diff --git a/bench/beman/sndr_sync_read_stream.hpp b/bench/beman/sndr_sync_read_stream.hpp
@@ -15,11 +15,9 @@
 // as_awaitable path returns the coroutine handle
 // for symmetric transfer.
 //
-// WARNING: Using this sender in a loop algorithm
-// like repeat_effect_until will stack overflow —
-// there is no trampoline for synchronous
-// completions. Coroutines handle this via symmetric
-// transfer; sender pipelines cannot.
+// repeat_until's trampoline bounds stack depth for
+// sender pipelines. Coroutines handle this via
+// symmetric transfer.
 //
 
 #ifndef BOOST_CAPY_BENCH_SNDR_SYNC_READ_STREAM_HPP
