From 547a6ac34368907624019b3f3769b16363b0ef23 Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Wed, 8 Jul 2026 18:52:07 +0200 Subject: [PATCH] refactor!: remove delay, timeout, and the timer thread; add async_waker Delete delay(), timeout(), and detail::timer_service, the background timer thread capy spawned per execution_context. Capy no longer creates any thread the user did not ask for; timed operations belong to the I/O layer, which owns a reactor and a clock. Add async_waker, a single-waiter notification point that replaces the timer thread as the escape hatch: the user provides the thread and the clock, capy provides the suspension point. wake() is callable from any thread and only ever posts through the waiter's executor; a notification with no waiter is latched as a single token so the notify-before-wait race is benign. A three-state atomic is the sole arbiter of the waiter's resume, which makes cross-thread notify safe without locks. wait() requires an executor that resumes continuations on a single thread, the same model as async_event and async_mutex. The quitter tests and the timeout-cancellation, when-any, and awaitable-sender examples are reworked around the new primitive with std::thread. Also fixes two pre-existing awaitable-sender bugs: a dangling executor_ref bound to a temporary executor, and a discarded symmetric-transfer handle that left a wrapped task never running. error::timeout remains; corosio maps its deadline results onto it. --- doc/as_sender.md | 16 +- .../ROOT/pages/3.concurrency/3d.patterns.adoc | 2 +- .../pages/4.coroutines/4d.io-awaitable.adoc | 2 +- .../pages/4.coroutines/4e.cancellation.adoc | 49 ++- .../pages/4.coroutines/4f.composition.adoc | 63 ++-- .../8.examples/8f.timeout-cancellation.adoc | 244 +++++++++--- .../ROOT/pages/9.design/9b.Separation.adoc | 2 +- doc/modules/ROOT/pages/why-capy.adoc | 6 +- example/awaitable-sender/awaitable_sender.cpp | 47 ++- example/awaitable-sender/awaitable_sender.hpp | 13 +- example/quitter-shutdown/quitter_shutdown.cpp | 56 ++- .../timeout_cancellation.cpp | 152 ++++++-- include/boost/capy.hpp | 3 +- include/boost/capy/delay.hpp | 245 ------------ include/boost/capy/ex/async_waker.hpp | 357 ++++++++++++++++++ .../boost/capy/ex/detail/timer_service.hpp | 128 ------- include/boost/capy/test/stream.hpp | 10 +- include/boost/capy/timeout.hpp | 228 ----------- src/ex/detail/timer_service.cpp | 125 ------ src/ex/thread_pool.cpp | 6 +- test/unit/delay.cpp | 288 -------------- test/unit/ex/async_waker.cpp | 272 +++++++++++++ test/unit/ex/detail/timer_service.cpp | 287 -------------- test/unit/quitter.cpp | 20 +- test/unit/timeout.cpp | 332 ---------------- 25 files changed, 1142 insertions(+), 1811 deletions(-) delete mode 100644 include/boost/capy/delay.hpp create mode 100644 include/boost/capy/ex/async_waker.hpp delete mode 100644 include/boost/capy/ex/detail/timer_service.hpp delete mode 100644 include/boost/capy/timeout.hpp delete mode 100644 src/ex/detail/timer_service.cpp delete mode 100644 test/unit/delay.cpp create mode 100644 test/unit/ex/async_waker.cpp delete mode 100644 test/unit/ex/detail/timer_service.cpp delete mode 100644 test/unit/timeout.cpp diff --git a/doc/as_sender.md b/doc/as_sender.md index 85859da02..98417d50f 100644 --- a/doc/as_sender.md +++ b/doc/as_sender.md @@ -147,8 +147,9 @@ produce - at zero allocation cost. namespace capy = boost::capy; namespace ex = beman::execution; -// A Capy IoAwaitable - a 500ms timer -auto sndr = capy::as_sender(capy::delay(500ms)); +// A Capy IoAwaitable - a single-slot waker wait +capy::async_waker waker; +auto sndr = capy::as_sender(waker.wait()); // Connect a receiver whose environment carries a Capy executor auto op = ex::connect( @@ -159,12 +160,15 @@ auto op = ex::connect( // Start the operation - no coroutine frame allocated ex::start(op); + +// A user thread wakes the waiter +waker.wake(); ``` The receiver's environment provides the executor and stop token. The -bridge threads them into the `io_env` that the awaitable expects. The -timer fires, the executor resumes the handle, the receiver gets -`set_value()`. Twenty-four bytes of `frame_cb` on the operation state. -That is the entire cost. +bridge threads them into the `io_env` that the awaitable expects. When +`waker.wake()` runs, the executor resumes the handle, the receiver +gets `set_value()`. Twenty-four bytes of `frame_cb` on the operation +state. That is the entire cost. Welcome to the awaitable universe. The door is open. diff --git a/doc/modules/ROOT/pages/3.concurrency/3d.patterns.adoc b/doc/modules/ROOT/pages/3.concurrency/3d.patterns.adoc index d1f8ca2e4..868a14b03 100644 --- a/doc/modules/ROOT/pages/3.concurrency/3d.patterns.adoc +++ b/doc/modules/ROOT/pages/3.concurrency/3d.patterns.adoc @@ -290,4 +290,4 @@ You have seen the dangers—race conditions, deadlocks—and the tools to avoid Concurrency is challenging. Bugs hide until the worst moment. Testing is hard because timing varies. But the rewards are substantial: responsive applications, full hardware utilization, and elegant solutions to naturally parallel problems. -This foundation prepares you for understanding Capy's concurrency facilities: `thread_pool`, `strand`, `when_all`, and `async_event`. These build on standard primitives to provide coroutine-friendly concurrent programming. +This foundation prepares you for understanding Capy's concurrency facilities: `thread_pool`, `strand`, `when_all`, `async_event`, `async_mutex`, and `async_waker`. These build on standard primitives to provide coroutine-friendly concurrent programming. diff --git a/doc/modules/ROOT/pages/4.coroutines/4d.io-awaitable.adoc b/doc/modules/ROOT/pages/4.coroutines/4d.io-awaitable.adoc index c306f425e..041c94afc 100644 --- a/doc/modules/ROOT/pages/4.coroutines/4d.io-awaitable.adoc +++ b/doc/modules/ROOT/pages/4.coroutines/4d.io-awaitable.adoc @@ -226,7 +226,7 @@ stop_cb_.emplace(env->stop_token, h); // h is a raw coroutine_handle See xref:4.coroutines/4e.cancellation.adoc#stoppable-awaitables[Implementing Stoppable Awaitables] for a complete example. -For a production implementation of this exact pattern, read the source of `delay_awaitable` (xref:reference:boost/capy/delay_awaitable.adoc[`delay_awaitable`]): it schedules a timer, registers a stop callback that posts the resume through the executor, and arbitrates between the timer and cancellation with a single atomic claim. +For a production implementation of this exact pattern, read the source of `async_waker::wait_awaiter` (xref:reference:boost/capy/async_waker/wait_awaiter.adoc[`async_waker::wait_awaiter`]): it registers a stop callback that posts the resume through the executor, and arbitrates between wakeup and cancellation with a single atomic claim. [#bridging-a-foreign-awaitable] == Bridging a Foreign Awaitable diff --git a/doc/modules/ROOT/pages/4.coroutines/4e.cancellation.adoc b/doc/modules/ROOT/pages/4.coroutines/4e.cancellation.adoc index 9510c7dc7..0b1d589af 100644 --- a/doc/modules/ROOT/pages/4.coroutines/4e.cancellation.adoc +++ b/doc/modules/ROOT/pages/4.coroutines/4e.cancellation.adoc @@ -364,29 +364,54 @@ NOTE: Capy's built-in I/O awaitables (via Corosio) already use the post-back pat == Part 9: Patterns -=== Timeout Pattern +=== Racing a Deadline -Capy ships a first-class `timeout()` combinator (``) that races an `io_result`-returning awaitable against a deadline. The first to complete wins and cancels the other; if the timer fires first, the result carries `cond::timeout`: +A timeout is expressed as a race: the operation you care about against +a deadline. `async_waker` provides the suspension point for both sides +of that race; a user thread supplies the clock: [source,cpp] ---- -#include - -using namespace std::chrono_literals; +struct fetch_channel +{ + capy::async_waker fetch_ready; + std::atomic cancelled{false}; + std::string result; +}; -task read_with_timeout(socket& sock, mutable_buffer buf) +// One side of the race: completes when the fetch worker thread +// wakes fetch_ready. If the deadline wins first, when_any's stop +// request cancels the wait; flag the worker so it stops early. +capy::io_task await_fetch(fetch_channel& ch) { - auto [ec, n] = co_await capy::timeout(sock.read_some(buf), 50ms); - if (ec == cond::timeout) + auto [ec] = co_await ch.fetch_ready.wait(); + if (ec) { - // deadline elapsed before the read completed - co_return; + ch.cancelled.store(true); + co_return capy::io_result{ec, {}}; } - // ... use the n bytes read + co_return capy::io_result{{}, std::move(ch.result)}; +} + +// The other side of the race: completes once whatever plays the +// clock wakes the waker. +capy::io_task<> deadline(capy::async_waker& waker) +{ + auto [ec] = co_await waker.wait(); + co_return capy::io_result<>{ec}; } ---- -The deadline itself is built on `delay()` (``), an awaitable that suspends for a duration and resumes with `cond::canceled` if its stop token is activated. Reach for `timeout()` rather than wiring a timer to a `std::stop_source` by hand. +`when_any(await_fetch(ch), deadline(waker))` returns as soon as either side +wakes; the loser's wait resolves with `cond::canceled`. See +xref:8.examples/8f.timeout-cancellation.adoc[Timeout with Cancellation] +for the full runnable demo, including the two `std::thread` objects that +play fetch worker and clock. + +When the work you are racing is Corosio I/O rather than coroutine-internal +work, Corosio's own timed operations take the place of the deadline +thread, scheduling against the platform event loop instead of a +`std::this_thread::sleep_for`. === User Cancellation diff --git a/doc/modules/ROOT/pages/4.coroutines/4f.composition.adoc b/doc/modules/ROOT/pages/4.coroutines/4f.composition.adoc index 574015222..f0fe6c842 100644 --- a/doc/modules/ROOT/pages/4.coroutines/4f.composition.adoc +++ b/doc/modules/ROOT/pages/4.coroutines/4f.composition.adoc @@ -268,47 +268,43 @@ task process_all(std::vector const& items) } ---- -=== Asynchronous Sleep +=== Racing a Deadline -`delay` is the awaitable counterpart to `std::this_thread::sleep_for`. Instead of blocking the thread, it suspends the current coroutine until the duration elapses, leaving the thread free to run other coroutines in the meantime: +`when_any` is the tool for expressing a deadline race directly. Two waker waits, one completed by the operation itself, the other by whatever plays the clock, race as ordinary `when_any` children: [source,cpp] ---- -#include - -task<> example() +capy::io_task await_fetch(fetch_channel& ch) { - auto [ec] = co_await delay(100ms); - // 100ms have elapsed; other coroutines ran on this thread while we waited + auto [ec] = co_await ch.fetch_ready.wait(); + if (ec) + { + ch.cancelled.store(true); + co_return capy::io_result{ec, {}}; + } + co_return capy::io_result{{}, std::move(ch.result)}; } ----- - -[NOTE] -==== -A thread is *not* consumed per sleeping coroutine. All concurrently sleeping coroutines on the same execution context share a single timer thread, so a thousand simultaneous `delay()` calls cost one thread, not a thousand. -==== - -`delay` is cancellable. If the environment's stop token is activated before the deadline, the coroutine resumes early with `ec` set to `error::canceled` (compare with `cond::canceled`); otherwise `ec` is clear. A zero or negative duration completes synchronously without scheduling a timer. - -=== Timeout - -The `timeout` combinator races an awaitable against a deadline. It is built directly on `delay` — the inner awaitable is run against a `delay` of the given duration, and whichever completes first cancels the other: -[source,cpp] ----- -#include - -task<> example() +capy::io_task<> deadline(capy::async_waker& waker) { - auto [ec, n] = co_await timeout(sock.read_some(buf), 50ms); - if (ec == cond::timeout) - { - // deadline expired before read completed - } + auto [ec] = co_await waker.wait(); + co_return capy::io_result<>{ec}; } + +auto race = [&]() -> capy::task<> +{ + auto result = co_await capy::when_any( + await_fetch(fetch_ch), + deadline(deadline_waker)); + winner = result.index(); + if (result.index() == 1) + fetch_result = std::get<1>(std::move(result)); +}; ---- -`timeout` returns the same `io_result` type as the inner awaitable. On timeout, `ec` is set to `error::timeout` and payload values are default-initialized. Unlike `when_any`, exceptions from the inner awaitable are always propagated and never swallowed by the timer. +`deadline_waker.wake()` can come from any thread, a `std::thread` that sleeps for the deadline, hardware, or a completion callback from another library. See xref:8.examples/8f.timeout-cancellation.adoc[Timeout with Cancellation] for the full example, including the `fetch_channel` type and both `std::thread` objects that play fetch worker and clock. + +For timed I/O against a socket, prefer Corosio's own timed operations: they schedule the deadline on the platform event loop instead of a dedicated thread. == Implementation Notes @@ -343,11 +339,8 @@ This design ensures proper context propagation to all children. | `` | First-completion racing with when_any -| `` -| Asynchronous sleep that suspends instead of blocking the thread - -| `` -| Race an awaitable against a deadline +| `` +| Single-waiter notification point for deadlines and other external events |=== You have now learned how to compose tasks concurrently with `when_all` and `when_any`. In the next section, you will learn about frame allocators for customizing coroutine memory allocation. diff --git a/doc/modules/ROOT/pages/8.examples/8f.timeout-cancellation.adoc b/doc/modules/ROOT/pages/8.examples/8f.timeout-cancellation.adoc index 4bd6d9a3d..54dc9f2d0 100644 --- a/doc/modules/ROOT/pages/8.examples/8f.timeout-cancellation.adoc +++ b/doc/modules/ROOT/pages/8.examples/8f.timeout-cancellation.adoc @@ -1,12 +1,15 @@ = Timeout with Cancellation -Using stop tokens to implement operation timeouts. +Racing a slow operation against a deadline, and cancelling work directly +with a stop token. == What You Will Learn -* Creating and using `std::stop_source` +* Where timed operations belong in the layering +* Using `async_waker` as the escape hatch for external timing +* Racing a fetch against a deadline with `when_any` * Checking `stop_requested()` in coroutines -* Cancellation patterns +* Direct cancellation with `std::stop_source` == Prerequisites @@ -17,10 +20,21 @@ Using stop tokens to implement operation timeouts. [source,cpp] ---- +// +// Copyright (c) 2026 Mungo Gill +// 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 +// + #include #include #include #include +#include #include #include #include @@ -33,7 +47,7 @@ capy::task slow_fetch(int steps) { auto token = co_await capy::this_coro::stop_token; // std::stop_token std::string result; - + for (int i = 0; i < steps; ++i) { // Check cancellation before each step @@ -43,26 +57,57 @@ capy::task slow_fetch(int steps) throw std::system_error( make_error_code(std::errc::operation_canceled)); } - + result += "step" + std::to_string(i) + " "; - + // Simulate slow work (in real code, this would be I/O) std::this_thread::sleep_for(std::chrono::milliseconds(10)); std::cout << " Completed step " << i << std::endl; - + // Yield to allow stop request to be processed before next check // Extra 5ms ensures print completes before main thread prints std::this_thread::sleep_for(std::chrono::milliseconds(15)); } - + co_return result; } -// Run with timeout (conceptual - real implementation needs timer) -capy::task> fetch_with_timeout() +// Shared between the fetch worker thread and the coroutine side of +// the race in demo_timeout below. +struct fetch_channel +{ + capy::async_waker fetch_ready; + std::atomic cancelled{false}; + std::string result; +}; + +// One side of the race: completes when the fetch worker thread +// wakes fetch_ready. If the deadline wins first, when_any's stop +// request cancels the wait; flag the worker so it stops early. +capy::io_task await_fetch(fetch_channel& ch) +{ + auto [ec] = co_await ch.fetch_ready.wait(); + if (ec) + { + ch.cancelled.store(true); + co_return capy::io_result{ec, {}}; + } + co_return capy::io_result{{}, std::move(ch.result)}; +} + +// The other side of the race: completes once a user thread wakes +// the waker. This is the escape hatch: the user supplies the +// thread and the clock. +capy::io_task<> deadline(capy::async_waker& waker) +{ + auto [ec] = co_await waker.wait(); + co_return capy::io_result<>{ec}; +} + +// Wraps slow_fetch, translating a cancellation exception into +// std::nullopt for the manual stop_token demo below. +capy::task> fetch_or_cancelled() { - auto token = co_await capy::this_coro::stop_token; // std::stop_token - try { auto result = co_await slow_fetch(5); // std::string @@ -76,26 +121,79 @@ capy::task> fetch_with_timeout() } } -void demo_normal_completion() +void demo_timeout() { - std::cout << "Demo: Normal completion\n"; - - capy::thread_pool pool; - std::stop_source source; - std::latch done(1); // std::latch - wait for 1 task + std::cout << "Demo: Fetch races a deadline\n"; + + // Both when_any children are plain waker waits, so the whole + // race runs on one thread, as async_waker requires. The + // blocking work happens on user threads that wake them. + capy::thread_pool pool(1); + std::latch done(1); + + fetch_channel fetch_ch; + capy::async_waker deadline_waker; + + // Worker thread does the slow fetch and wakes fetch_ready on + // completion, checking the cancel flag between steps so a lost race stops + // the work promptly. Mirrors slow_fetch's step loop, kept + // separate since one polls an atomic and the other a stop token. + std::thread fetch_thread([&fetch_ch] { + for (int i = 0; i < 5; ++i) + { + if (fetch_ch.cancelled.load()) + { + std::cout << " Cancelled at step " << i << std::endl; + return; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + fetch_ch.result += "step" + std::to_string(i) + " "; + std::cout << " Completed step " << i << std::endl; + std::this_thread::sleep_for(std::chrono::milliseconds(15)); + } + fetch_ch.fetch_ready.wake(); + }); - capy::run_async(pool.get_executor(), source.get_token(), - [&done](std::optional result) { - if (result) - std::cout << "Result: " << *result << "\n"; - else - std::cout << "Cancelled\n"; - done.count_down(); - }, - [&done](std::exception_ptr) { done.count_down(); } - )(fetch_with_timeout()); + // Deadline thread: a user thread plays the clock, waking the + // waker after the allotted time. + std::thread deadline_thread([&deadline_waker] { + std::this_thread::sleep_for(std::chrono::milliseconds(30)); + deadline_waker.wake(); + }); - done.wait(); // Block until task completes + // Index 1 = fetch won, index 2 = deadline won, index 0 = error. + std::size_t winner = 0; + std::string fetch_result; + + auto race = [&]() -> capy::task<> + { + auto result = co_await capy::when_any( + await_fetch(fetch_ch), + deadline(deadline_waker)); + winner = result.index(); + if (result.index() == 1) + fetch_result = std::get<1>(std::move(result)); + }; + + capy::run_async(pool.get_executor(), + [&done]() { done.count_down(); }, + [&done](std::exception_ptr) { done.count_down(); } + )(race()); + + done.wait(); // Block until the race completes + fetch_thread.join(); + deadline_thread.join(); + + // Report after the joins so the worker's final line lands + // before the verdict. + if (winner == 1) + // This demo's timing always favors the deadline; this branch + // shows how a fetch win would be consumed if it ever happened. + std::cout << "Result: " << fetch_result << "\n"; + else if (winner == 2) + std::cout << "Timed out waiting for fetch\n"; + else + std::cout << "Error\n"; } void demo_cancellation() @@ -116,8 +214,8 @@ void demo_cancellation() done.count_down(); }, [&done](std::exception_ptr) { done.count_down(); } - )(fetch_with_timeout()); - + )(fetch_or_cancelled()); + // Simulate timeout: cancel after 2 steps complete // Timing: each step is 10ms work + 15ms yield = 25ms total // Step 1 prints at 35ms, step 2 check at 50ms @@ -125,7 +223,7 @@ void demo_cancellation() std::this_thread::sleep_for(std::chrono::milliseconds(42)); std::cout << " Requesting stop..." << std::endl; source.request_stop(); - + done.wait(); // Block until task completes (after cancellation) } @@ -134,7 +232,7 @@ capy::task process_items(std::vector const& items) { auto token = co_await capy::this_coro::stop_token; // std::stop_token int sum = 0; - + for (auto item : items) // int { if (token.stop_requested()) @@ -142,18 +240,18 @@ capy::task process_items(std::vector const& items) std::cout << "Processing cancelled, partial sum: " << sum << "\n"; co_return sum; // Return partial result } - + sum += item; } - + co_return sum; } int main() { - demo_normal_completion(); + demo_timeout(); demo_cancellation(); - + return 0; } ---- @@ -168,6 +266,68 @@ target_link_libraries(timeout_cancellation PRIVATE capy) == Walkthrough +=== Where Timing Lives + +Capy is the coroutine layer: no reactor, no clock, no hidden threads. +Every thread Capy touches is one you handed it, a `thread_pool` worker, +or the thread that calls `run`. Timing needs a clock and something to +sleep on it, and that belongs to the I/O layer, where a reactor and +platform clock already exist for other reasons: Corosio provides +`delay()` and `timeout()` built on its event loop. + +In pure Capy, with no Corosio in the picture, the user's own thread +plays the clock, and `async_waker` gives it a suspension point to wake. + +=== The Escape Hatch: async_waker + +`async_waker` is Capy's answer to "how do I wait for something that isn't +a coroutine?" It hands a single wakeup from any thread to one waiting +coroutine: one coroutine suspends in `wait()`; any thread, including one +you spun up yourself to play the role of a clock, wakes it with `wake()`: + +[source,cpp] +---- +struct fetch_channel +{ + capy::async_waker fetch_ready; + std::atomic cancelled{false}; + std::string result; +}; +---- + +`wait()` returns an `io_result<>` that is empty on wakeup and carries +`cond::canceled` if the environment's stop token fires first. A wakeup +with no waiter present is latched as one pending token, so the +wake-before-wait race is benign; extra wakes collapse into that single +token rather than queuing up. Because `wake()` is the only operation +callable from a foreign thread, the pattern generalizes to anything +external: a timer, a hardware interrupt, a completion callback from +another library. + +=== Racing a Fetch Against a Deadline + +`demo_timeout` builds two `io_task` children, each just a waker wait, and +races them with `when_any`: + +[source,cpp] +---- +auto result = co_await capy::when_any( + await_fetch(fetch_ch), + deadline(deadline_waker)); +---- + +`await_fetch` completes when the fetch worker thread finishes and wakes +`fetch_ch.fetch_ready`. `deadline` completes when a second user thread, +playing the clock, sleeps for the allotted duration and wakes +`deadline_waker`. Whichever wakes first wins; +`when_any` requests stop on the loser, which is exactly how `await_fetch` +learns to set `cancelled` and let the fetch worker thread bail out early. + +`async_waker::wait()` must only be awaited on a single-threaded +executor, so both children run on `thread_pool(1)`. The slow, blocking +work itself still happens off that thread, on the two `std::thread` +objects the demo owns directly. + === Getting the Stop Token [source,cpp] @@ -217,13 +377,11 @@ Cancellation doesn't have to throw. You can return partial results or a sentinel == Output ---- -Demo: Normal completion +Demo: Fetch races a deadline Completed step 0 Completed step 1 - Completed step 2 - Completed step 3 - Completed step 4 -Result: step0 step1 step2 step3 step4 + Cancelled at step 2 +Timed out waiting for fetch Demo: Cancellation after 2 steps Completed step 0 @@ -235,7 +393,7 @@ Cancelled (returned nullopt) == Exercises -1. Implement a retry-with-timeout pattern +1. Change the deadline to 60ms and observe the fetch winning the race instead 2. Add cancellation support to the echo session from the previous example 3. Create a task that cancels itself after processing N items diff --git a/doc/modules/ROOT/pages/9.design/9b.Separation.adoc b/doc/modules/ROOT/pages/9.design/9b.Separation.adoc index c639ef827..dcd397de4 100644 --- a/doc/modules/ROOT/pages/9.design/9b.Separation.adoc +++ b/doc/modules/ROOT/pages/9.design/9b.Separation.adoc @@ -20,7 +20,7 @@ This document applies well-established physical design principles to show why th **Capy** provides the foundational abstractions for coroutine-based I/O. Tasks. Buffers. Stream concepts. Executors. The IoAwaitable protocol. Type-erased streams. Composition primitives like `when_all` and `when_any`. It is pure {cpp}20. It does not include a single line of platform-specific code. No sockets. No file descriptors. No `#ifdef _WIN32`. -**Corosio** provides platform networking. TCP sockets. TLS streams. DNS resolution. Timers. Signal handling. It implements four platform-specific event loop backends: IOCP on Windows, epoll on Linux, kqueue on macOS/BSD, and POSIX select as a fallback. Corosio depends on Capy. Capy does not depend on Corosio. +**Corosio** provides platform networking. TCP sockets. TLS streams. DNS resolution. Delays and timeouts. Signal handling. It implements four platform-specific event loop backends: IOCP on Windows, epoll on Linux, kqueue on macOS/BSD, and POSIX select as a fallback. Corosio depends on Capy. Capy does not depend on Corosio. The dependency arrow points in one direction. That is not an accident. diff --git a/doc/modules/ROOT/pages/why-capy.adoc b/doc/modules/ROOT/pages/why-capy.adoc index d687b9e2a..98e11c402 100644 --- a/doc/modules/ROOT/pages/why-capy.adoc +++ b/doc/modules/ROOT/pages/why-capy.adoc @@ -200,7 +200,7 @@ Neither Asio nor `std::execution` offers this combination of forward-flow alloca * `IoAwaitable`, `IoRunnable` — taxonomy of awaitable concepts * `task` — concrete task type implementing the protocol (user-defined tasks also supported) * `run`, `run_async` — launch functions with forward-flow allocator control -* `strand`, `thread_pool`, `async_mutex`, `async_event` — concurrency primitives +* `strand`, `thread_pool`, `async_mutex`, `async_event`, `async_waker`: concurrency primitives * `frame_allocator`, `recycling_memory_resource` — coroutine-optimized allocation === Comparison @@ -265,6 +265,10 @@ Neither Asio nor `std::execution` offers this combination of forward-flow alloca ^| - ^| - +| `async_waker` +^| - +^| - + | `stop_token` propagation ^| - | `stop_token`*** diff --git a/example/awaitable-sender/awaitable_sender.cpp b/example/awaitable-sender/awaitable_sender.cpp index fdeaf685e..1ddef26a7 100644 --- a/example/awaitable-sender/awaitable_sender.cpp +++ b/example/awaitable-sender/awaitable_sender.cpp @@ -1,5 +1,6 @@ // // 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) @@ -75,41 +76,60 @@ int main() << "main thread: " << std::this_thread::get_id() << "\n"; - // Capy execution context (provides timer service, etc.) - capy::thread_pool pool; + // Capy execution context. Single-threaded: async_waker's + // wait() requires resumption on a single thread. + capy::thread_pool pool(1); + + // Named so io_sender_env's executor_ref (which stores a + // pointer to whatever it is given) points at a stable object + // instead of a temporary that dies at the end of the + // full-expression that constructs each demo_receiver. + auto pool_ex = pool.get_executor(); + + // Escape-hatch timing: a helper thread plays the clock and + // wakes waker_1 and waker_3 after a short delay. + // waker_2 is deliberately never woken, so its demo below + // completes only via stop-token cancellation, not a wakeup. + capy::async_waker waker_1, waker_2, waker_3; + std::thread waker_thread([&] { + std::this_thread::sleep_for(50ms); + waker_1.wake(); + std::this_thread::sleep_for(50ms); + waker_3.wake(); + }); std::latch done(1); // Build a sender from a Capy IoAwaitable - auto sndr = capy::as_sender(capy::delay(500ms)); + auto sndr = capy::as_sender(waker_1.wait()); // Connect with a receiver whose environment carries // the Capy thread_pool executor auto op = ex::connect( std::move(sndr), demo_receiver{ - {pool.get_executor(), std::stop_token{}}, + {pool_ex, std::stop_token{}}, &done}); - std::cout << " starting delay...\n"; + std::cout << " starting wait...\n"; ex::start(op); done.wait(); - std::cout << " delay completed\n"; + std::cout << " wait completed\n"; // Test cancellation via stop token std::cout << "\n--- cancellation test ---\n"; std::stop_source ss; std::latch done2(1); - auto sndr2 = capy::as_sender(capy::delay(5s)); + auto sndr2 = capy::as_sender(waker_2.wait()); auto op2 = ex::connect( std::move(sndr2), demo_receiver{ - {pool.get_executor(), ss.get_token()}, + {pool_ex, ss.get_token()}, &done2}); - std::cout << " starting 5s delay...\n"; + std::cout << " starting wait (never woken)...\n"; ex::start(op2); std::this_thread::sleep_for(100ms); @@ -124,11 +144,11 @@ int main() std::latch done3(1); auto sndr3 = capy::split_ec( - capy::as_sender(capy::delay(100ms))); + capy::as_sender(waker_3.wait())); auto op3 = ex::connect( std::move(sndr3), demo_receiver{ - {pool.get_executor(), std::stop_token{}}, + {pool_ex, std::stop_token{}}, &done3}); ex::start(op3); @@ -153,10 +173,13 @@ int main() auto op4 = ex::connect( std::move(sndr4), demo_receiver{ - {pool.get_executor(), std::stop_token{}}, + {pool_ex, std::stop_token{}}, &done4}); ex::start(op4); done4.wait(); std::cout << " split_ec error test done\n"; + + // All demos have drained; safe to join the waker thread now. + waker_thread.join(); } diff --git a/example/awaitable-sender/awaitable_sender.hpp b/example/awaitable-sender/awaitable_sender.hpp index 0e3fb9852..f942605be 100644 --- a/example/awaitable-sender/awaitable_sender.hpp +++ b/example/awaitable-sender/awaitable_sender.hpp @@ -1,5 +1,6 @@ // // 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) @@ -291,7 +292,14 @@ struct awaitable_sender auto h = std::coroutine_handle<>::from_address( static_cast(&cb_)); - detail::call_await_suspend(&aw_, h, &env_); + // Not a real coroutine caller, so symmetric transfer + // must be driven by hand: any non-noop handle (our own + // frame on immediate completion, or a wrapped task's + // handle that still needs to run) has to be resumed + // explicitly or nothing ever completes. + auto resumed = detail::call_await_suspend(&aw_, h, &env_); + if(resumed != std::noop_coroutine()) + resumed.resume(); } }; @@ -329,8 +337,7 @@ struct awaitable_sender @par Example @code - auto sndr = as_sender(capy::delay( - std::chrono::milliseconds(100))); + auto sndr = as_sender(waker.wait()); @endcode @param aw The IoAwaitable to wrap. diff --git a/example/quitter-shutdown/quitter_shutdown.cpp b/example/quitter-shutdown/quitter_shutdown.cpp index ef77f6e29..afa4bec67 100644 --- a/example/quitter-shutdown/quitter_shutdown.cpp +++ b/example/quitter-shutdown/quitter_shutdown.cpp @@ -1,5 +1,6 @@ // // Copyright (c) 2026 Michael Vandeberg +// 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) @@ -12,9 +13,12 @@ Demonstrates quitter for responsive application shutdown. Four workers simulate a batch file-processing pipeline: each - "downloads" data (delay), "transforms" it, and "writes" the - result (delay). Workers are quitter<> coroutines — their - bodies contain zero cancellation-handling code. + "downloads" data, "transforms" it, and "writes" the result. + A single "ticker" thread plays the role of the clock: it wakes + each worker's async_waker on an interval, and the worker's + co_await waker.wait() is the suspension point. Workers are + quitter<> coroutines: + their bodies contain zero cancellation-handling code. Press Ctrl+C to request shutdown. Every in-flight worker exits at its next co_await, RAII cleanup runs (each worker @@ -23,7 +27,7 @@ Contrast with task<>: With task<>, every co_await that touches I/O needs: - auto [ec] = co_await delay(dur); + auto [ec] = co_await waker.wait(); if(ec) co_return; // <-- cancellation boilerplate This is repeated at every suspension point. @@ -33,6 +37,7 @@ #include +#include #include #include #include @@ -40,6 +45,8 @@ #include #include #include +#include +#include namespace capy = boost::capy; using namespace std::chrono_literals; @@ -90,6 +97,7 @@ struct resource_guard // No cancellation code. quitter handles it. capy::quitter<> worker( int id, + capy::async_waker& waker, std::atomic& items_processed, std::atomic& cleanup_count) { @@ -97,9 +105,8 @@ capy::quitter<> worker( for(int item = 0; ; ++item) { - // Simulate download (200-400ms depending on worker) - auto download_time = 200ms + 50ms * id; - (void) co_await capy::delay(download_time); + // Simulate download: suspend until the ticker wakes us. + (void) co_await waker.wait(); // Simulate transform (CPU work — no co_await needed) { @@ -109,8 +116,8 @@ capy::quitter<> worker( std::cout << oss.str(); } - // Simulate write (100ms) - (void) co_await capy::delay(100ms); + // Simulate write: suspend for another wakeup. + (void) co_await waker.wait(); ++items_processed; } @@ -133,21 +140,48 @@ int main() std::atomic items_processed{0}; std::atomic cleanup_count{0}; + // One waker per worker (single-waiter precondition); each + // runs on its own strand so the pool's num_workers OS threads + // still keep every waker's resumption single-threaded. + std::array wakers; + std::vector> strands; + strands.reserve(num_workers); + for(int i = 0; i < num_workers; ++i) + strands.emplace_back(pool.get_executor()); + + // Ticker thread paces the workers: it periodically wakes + // every worker's waker. + std::atomic ticker_stop{false}; + std::thread ticker([&] { + while(!ticker_stop.load(std::memory_order_relaxed)) + { + std::this_thread::sleep_for(50ms); + for(auto& waker : wakers) + waker.wake(); + } + }); + std::cout << "Starting " << num_workers << " workers. Press Ctrl+C to quit.\n\n"; for(int i = 0; i < num_workers; ++i) { capy::run_async( - pool.get_executor(), + strands[i], g_stop.get_token(), [&]() { done.count_down(); }, [&](std::exception_ptr) { done.count_down(); })( - worker(i, items_processed, cleanup_count)); + worker(i, wakers[i], items_processed, cleanup_count)); } done.wait(); + // Stop and join the ticker now that the pool has drained, so + // it cannot wake a waker (or post to a strand) after the + // pool starts tearing down. + ticker_stop.store(true, std::memory_order_relaxed); + ticker.join(); + auto stop_at = g_stop_time.load(std::memory_order_relaxed); auto now = std::chrono::steady_clock::now(); diff --git a/example/timeout-cancellation/timeout_cancellation.cpp b/example/timeout-cancellation/timeout_cancellation.cpp index b1c791bba..be5abb445 100644 --- a/example/timeout-cancellation/timeout_cancellation.cpp +++ b/example/timeout-cancellation/timeout_cancellation.cpp @@ -1,5 +1,6 @@ // // Copyright (c) 2026 Mungo Gill +// 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) @@ -11,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -23,7 +25,7 @@ capy::task slow_fetch(int steps) { auto token = co_await capy::this_coro::stop_token; // std::stop_token std::string result; - + for (int i = 0; i < steps; ++i) { // Check cancellation before each step @@ -33,26 +35,57 @@ capy::task slow_fetch(int steps) throw std::system_error( make_error_code(std::errc::operation_canceled)); } - + result += "step" + std::to_string(i) + " "; - + // Simulate slow work (in real code, this would be I/O) std::this_thread::sleep_for(std::chrono::milliseconds(10)); std::cout << " Completed step " << i << std::endl; - + // Yield to allow stop request to be processed before next check // Extra 5ms ensures print completes before main thread prints std::this_thread::sleep_for(std::chrono::milliseconds(15)); } - + co_return result; } -// Run with timeout (conceptual - real implementation needs timer) -capy::task> fetch_with_timeout() +// Shared between the fetch worker thread and the coroutine side of +// the race in demo_timeout below. +struct fetch_channel +{ + capy::async_waker fetch_ready; + std::atomic cancelled{false}; + std::string result; +}; + +// One side of the race: completes when the fetch worker thread +// wakes fetch_ready. If the deadline wins first, when_any's stop +// request cancels the wait; flag the worker so it stops early. +capy::io_task await_fetch(fetch_channel& ch) +{ + auto [ec] = co_await ch.fetch_ready.wait(); + if (ec) + { + ch.cancelled.store(true); + co_return capy::io_result{ec, {}}; + } + co_return capy::io_result{{}, std::move(ch.result)}; +} + +// The other side of the race: completes once a user thread wakes +// the waker. This is the escape hatch: the user supplies the +// thread and the clock. +capy::io_task<> deadline(capy::async_waker& waker) +{ + auto [ec] = co_await waker.wait(); + co_return capy::io_result<>{ec}; +} + +// Wraps slow_fetch, translating a cancellation exception into +// std::nullopt for the manual stop_token demo below. +capy::task> fetch_or_cancelled() { - auto token = co_await capy::this_coro::stop_token; // std::stop_token - try { auto result = co_await slow_fetch(5); // std::string @@ -66,26 +99,79 @@ capy::task> fetch_with_timeout() } } -void demo_normal_completion() +void demo_timeout() { - std::cout << "Demo: Normal completion\n"; - - capy::thread_pool pool; - std::stop_source source; - std::latch done(1); // std::latch - wait for 1 task + std::cout << "Demo: Fetch races a deadline\n"; - capy::run_async(pool.get_executor(), source.get_token(), - [&done](std::optional result) { - if (result) - std::cout << "Result: " << *result << "\n"; - else - std::cout << "Cancelled\n"; - done.count_down(); - }, + // Both when_any children are plain waker waits, so the whole + // race runs on one thread, as async_waker requires. The + // blocking work happens on user threads that wake them. + capy::thread_pool pool(1); + std::latch done(1); + + fetch_channel fetch_ch; + capy::async_waker deadline_waker; + + // Worker thread does the slow fetch and wakes fetch_ready on + // completion, checking the cancel flag between steps so a lost race stops + // the work promptly. Mirrors slow_fetch's step loop, kept + // separate since one polls an atomic and the other a stop token. + std::thread fetch_thread([&fetch_ch] { + for (int i = 0; i < 5; ++i) + { + if (fetch_ch.cancelled.load()) + { + std::cout << " Cancelled at step " << i << std::endl; + return; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + fetch_ch.result += "step" + std::to_string(i) + " "; + std::cout << " Completed step " << i << std::endl; + std::this_thread::sleep_for(std::chrono::milliseconds(15)); + } + fetch_ch.fetch_ready.wake(); + }); + + // Deadline thread: a user thread plays the clock, waking the + // waker after the allotted time. + std::thread deadline_thread([&deadline_waker] { + std::this_thread::sleep_for(std::chrono::milliseconds(30)); + deadline_waker.wake(); + }); + + // Index 1 = fetch won, index 2 = deadline won, index 0 = error. + std::size_t winner = 0; + std::string fetch_result; + + auto race = [&]() -> capy::task<> + { + auto result = co_await capy::when_any( + await_fetch(fetch_ch), + deadline(deadline_waker)); + winner = result.index(); + if (result.index() == 1) + fetch_result = std::get<1>(std::move(result)); + }; + + capy::run_async(pool.get_executor(), + [&done]() { done.count_down(); }, [&done](std::exception_ptr) { done.count_down(); } - )(fetch_with_timeout()); + )(race()); - done.wait(); // Block until task completes + done.wait(); // Block until the race completes + fetch_thread.join(); + deadline_thread.join(); + + // Report after the joins so the worker's final line lands + // before the verdict. + if (winner == 1) + // This demo's timing always favors the deadline; this branch + // shows how a fetch win would be consumed if it ever happened. + std::cout << "Result: " << fetch_result << "\n"; + else if (winner == 2) + std::cout << "Timed out waiting for fetch\n"; + else + std::cout << "Error\n"; } void demo_cancellation() @@ -106,8 +192,8 @@ void demo_cancellation() done.count_down(); }, [&done](std::exception_ptr) { done.count_down(); } - )(fetch_with_timeout()); - + )(fetch_or_cancelled()); + // Simulate timeout: cancel after 2 steps complete // Timing: each step is 10ms work + 15ms yield = 25ms total // Step 1 prints at 35ms, step 2 check at 50ms @@ -115,7 +201,7 @@ void demo_cancellation() std::this_thread::sleep_for(std::chrono::milliseconds(42)); std::cout << " Requesting stop..." << std::endl; source.request_stop(); - + done.wait(); // Block until task completes (after cancellation) } @@ -124,7 +210,7 @@ capy::task process_items(std::vector const& items) { auto token = co_await capy::this_coro::stop_token; // std::stop_token int sum = 0; - + for (auto item : items) // int { if (token.stop_requested()) @@ -132,17 +218,17 @@ capy::task process_items(std::vector const& items) std::cout << "Processing cancelled, partial sum: " << sum << "\n"; co_return sum; // Return partial result } - + sum += item; } - + co_return sum; } int main() { - demo_normal_completion(); + demo_timeout(); demo_cancellation(); - + return 0; } diff --git a/include/boost/capy.hpp b/include/boost/capy.hpp index dd84839ec..9a26eafc2 100644 --- a/include/boost/capy.hpp +++ b/include/boost/capy.hpp @@ -26,10 +26,8 @@ #include // Algorithms -#include #include #include -#include #include #include #include @@ -65,6 +63,7 @@ #include #include #include +#include #include #include #include diff --git a/include/boost/capy/delay.hpp b/include/boost/capy/delay.hpp deleted file mode 100644 index f4e524554..000000000 --- a/include/boost/capy/delay.hpp +++ /dev/null @@ -1,245 +0,0 @@ -// -// Copyright (c) 2026 Michael Vandeberg -// -// 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_DELAY_HPP -#define BOOST_CAPY_DELAY_HPP - -#include -#include -#include -#include -#include -#include -#include - -#include -#include -#include -#include -#include -#include - -namespace boost { -namespace capy { - -/** IoAwaitable returned by @ref delay. - - Suspends the calling coroutine until the deadline elapses - or the environment's stop token is activated, whichever - comes first. Resumption is always posted through the - executor, never inline on the timer thread. - - Not intended to be named directly; use the @ref delay - factory function instead. - - @par Return Value - - Returns `io_result<>{}` (no error) when the timer fires - normally, or `io_result<>{error::canceled}` when - cancellation claims the resume before the deadline. - - @par Cancellation - - If `stop_requested()` is true before suspension, the - coroutine resumes immediately without scheduling a timer - and returns `io_result<>{error::canceled}`. If stop is - requested while suspended, the stop callback claims the - resume and posts it through the executor; the pending - timer is cancelled on the next `await_resume` or - destructor call. - - @par Thread Safety - - A single `delay_awaitable` must not be awaited concurrently. - Multiple independent `delay()` calls on the same - execution_context are safe and share one timer thread. - - @see delay, timeout -*/ -class delay_awaitable -{ - std::chrono::nanoseconds dur_; - - detail::timer_service* ts_ = nullptr; - detail::timer_service::timer_id tid_ = 0; - - // Declared before stop_cb_buf_: the callback - // accesses these members, so they must still be - // alive if the stop_cb_ destructor blocks. - continuation cont_; - std::atomic claimed_{false}; - bool canceled_ = false; - bool stop_cb_active_ = false; - - struct cancel_fn - { - delay_awaitable* self_; - executor_ref ex_; - - void operator()() const noexcept - { - if(!self_->claimed_.exchange( - true, std::memory_order_acq_rel)) - { - self_->canceled_ = true; - ex_.post(self_->cont_); - } - } - }; - - using stop_cb_t = std::stop_callback; - - // Aligned storage for the stop callback. - // Declared last: its destructor may block while - // the callback accesses the members above. - BOOST_CAPY_MSVC_WARNING_PUSH - BOOST_CAPY_MSVC_WARNING_DISABLE(4324) - alignas(stop_cb_t) - unsigned char stop_cb_buf_[sizeof(stop_cb_t)]; - BOOST_CAPY_MSVC_WARNING_POP - - stop_cb_t& stop_cb_() noexcept - { - return *reinterpret_cast(stop_cb_buf_); - } - -public: - /// Construct an awaitable that waits for `dur` nanoseconds. - /// Prefer the @ref delay factory over constructing directly. - explicit delay_awaitable(std::chrono::nanoseconds dur) noexcept - : dur_(dur) - { - } - - /// @pre The stop callback must not be active - /// (i.e. the object has not yet been awaited). - delay_awaitable(delay_awaitable&& o) noexcept - : dur_(o.dur_) - , ts_(o.ts_) - , tid_(o.tid_) - , cont_(o.cont_) - , claimed_(o.claimed_.load(std::memory_order_relaxed)) - , canceled_(o.canceled_) - , stop_cb_active_(std::exchange(o.stop_cb_active_, false)) - { - } - - /// Tear down any registered stop callback and cancel the - /// pending timer if one is still scheduled. - ~delay_awaitable() - { - if(stop_cb_active_) - stop_cb_().~stop_cb_t(); - if(ts_) - ts_->cancel(tid_); - } - - delay_awaitable(delay_awaitable const&) = delete; - delay_awaitable& operator=(delay_awaitable const&) = delete; - delay_awaitable& operator=(delay_awaitable&&) = delete; - - /// Return true for zero or negative durations, completing - /// synchronously without scheduling a timer. - bool await_ready() const noexcept - { - return dur_.count() <= 0; - } - - /// Suspend the coroutine, scheduling the timer and a stop - /// callback on the environment's executor and stop token. - /// Resumes `h` immediately if stop was already requested. - std::coroutine_handle<> - await_suspend( - std::coroutine_handle<> h, - io_env const* env) noexcept - { - // Already stopped: resume immediately - if(env->stop_token.stop_requested()) - { - canceled_ = true; - return h; - } - - cont_.h = h; - ts_ = &env->executor.context().use_service(); - - // Schedule timer (won't fire inline since deadline is in the future) - tid_ = ts_->schedule_after(dur_, - [this, ex = env->executor]() - { - if(!claimed_.exchange( - true, std::memory_order_acq_rel)) - { - ex.post(cont_); - } - }); - - // Register stop callback (may fire inline) - ::new(stop_cb_buf_) stop_cb_t( - env->stop_token, - cancel_fn{this, env->executor}); - stop_cb_active_ = true; - - return std::noop_coroutine(); - } - - /// Clean up the stop callback and timer, then return - /// `io_result<>{error::canceled}` if cancellation claimed - /// the resume, or an empty `io_result<>` otherwise. - io_result<> await_resume() noexcept - { - if(stop_cb_active_) - { - stop_cb_().~stop_cb_t(); - stop_cb_active_ = false; - } - if(ts_) - ts_->cancel(tid_); - if(canceled_) - return io_result<>{make_error_code(error::canceled)}; - return io_result<>{}; - } -}; - -/** Suspend the current coroutine for a duration. - - Returns an IoAwaitable that completes at or after the - specified duration, or earlier if the environment's stop - token is activated. - - Zero or negative durations complete synchronously without - scheduling a timer. - - @par Example - @code - auto [ec] = co_await delay(std::chrono::milliseconds(100)); - @endcode - - @param dur The duration to wait. - - @return A @ref delay_awaitable whose `await_resume` - returns `io_result<>`. On normal completion, `ec` - is clear. On cancellation, `ec == error::canceled`. - - @throws Nothing. - - @see timeout, delay_awaitable -*/ -template -delay_awaitable -delay(std::chrono::duration dur) noexcept -{ - return delay_awaitable{ - std::chrono::duration_cast(dur)}; -} - -} // capy -} // boost - -#endif diff --git a/include/boost/capy/ex/async_waker.hpp b/include/boost/capy/ex/async_waker.hpp new file mode 100644 index 000000000..3f47af66a --- /dev/null +++ b/include/boost/capy/ex/async_waker.hpp @@ -0,0 +1,357 @@ +// +// 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_EX_ASYNC_WAKER_HPP +#define BOOST_CAPY_EX_ASYNC_WAKER_HPP + +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +/* async_waker implementation notes + =================================== + + wake() must be callable from foreign threads (that is the whole + point: the user's thread provides the timing). A waiter-side + claimed_ flag is not enough there -- the + waker has to dereference the waiter, and nothing would pin the + waiter's frame between reading the pointer and claiming it. + + So the three-state st_ atomic is the single arbiter: + + empty --arm--> armed --wake/cancel CAS--> empty + empty --wake--> token --wait consumes--> empty + + Whoever wins the armed->empty CAS owns the resume and may + dereference waiter_: the frame cannot die underneath the + winner because the coroutine only resumes when the winner + posts it. The loser never touches the waiter. When the stop + callback wins, a concurrent wake retries, finds empty, and + latches a token -- a racing wakeup is deferred, never lost. + + Serialized resumption is required: await_suspend keeps + writing after the publishing armed-CAS (the stop_cb + placement-new and active_ = true), so a wake/cancel winner + can post the continuation while that tail is still running. + The posted resume must be ordered after await_suspend's + return, which holds on a single-threaded executor (the one + thread is still inside await_suspend) and on a strand (the + resume is a later turn, synchronized with the current one). + A raw multi-threaded executor lets another worker run + await_resume against those in-flight writes. async_event and + async_mutex make the same assumption; it is stated explicitly + here because wake() invites foreign threads into the picture. +*/ + +namespace boost { +namespace capy { + +/** A single-slot waker that hands one wakeup to a waiting coroutine. + + This is the escape hatch for timing and other external events: + the user provides the thread and the clock, capy provides the + suspension point. One coroutine suspends in `wait()`; any + thread wakes it with `wake()`. + + A wakeup with no waiter present is latched as a single pending + token, and the next `wait()` consumes it immediately. This + makes the wake-before-wait race benign without any lock + protocol. Multiple wakes collapse into one token. + + @par Cancellation + + If the environment's stop token is triggered while suspended, + the wait completes with `error::canceled`. A wake that loses + the race against cancellation is latched for the next `wait()` + rather than dropped. + + @par Zero Allocation + + No heap allocation occurs for wait or wake operations. + + @par Thread Safety + + Distinct objects: Safe.@n + Shared objects: `wake()` may be called from any thread. + `wait()` must only be awaited by one coroutine at a time, and + only on an executor that never runs the coroutine's + continuations concurrently: a single-threaded executor or a + strand over a multi-threaded one (the same threading model as + `async_event` and `async_mutex`). Awaiting `wait()` directly + on a multi-threaded executor is undefined. + + This type is non-copyable and non-movable because a suspended + waiter holds a pointer into the object. + + @par Example + @code + async_waker waker; + + // user-provided timing thread + std::thread th([&waker] { + std::this_thread::sleep_for(100ms); + waker.wake(); + }); + + task<> waiter() { + auto [ec] = co_await waker.wait(); + // resumed on the executor after ~100ms + } + // ... th.join() after the pool drains + @endcode +*/ +class async_waker +{ +public: + class wait_awaiter; + +private: + static constexpr int state_empty = 0; // no token, no waiter + static constexpr int state_token = 1; // latched wakeup + static constexpr int state_armed = 2; // waiter suspended + + std::atomic st_{state_empty}; + wait_awaiter* waiter_ = nullptr; + +public: + /** Awaiter returned by wait(). + */ + class wait_awaiter + { + friend class async_waker; + + async_waker* waker_; + continuation cont_; + executor_ref ex_; + + // Declared before stop_cb_buf_: the callback accesses + // these members, so they must still be alive if the + // stop_cb_ destructor blocks. + bool canceled_ = false; + bool active_ = false; + bool published_ = false; + + struct cancel_fn + { + wait_awaiter* self_; + + void operator()() const noexcept + { + int expected = state_armed; + if(self_->waker_->st_.compare_exchange_strong( + expected, state_empty, + std::memory_order_acq_rel, + std::memory_order_acquire)) + { + self_->canceled_ = true; + self_->ex_.post(self_->cont_); + } + } + }; + + using stop_cb_t = std::stop_callback; + + // Aligned storage for stop_cb_t. Declared last: its + // destructor may block while the callback accesses the + // members above. + BOOST_CAPY_MSVC_WARNING_PUSH + BOOST_CAPY_MSVC_WARNING_DISABLE(4324) + alignas(stop_cb_t) + unsigned char stop_cb_buf_[sizeof(stop_cb_t)]; + BOOST_CAPY_MSVC_WARNING_POP + + stop_cb_t& stop_cb_() noexcept + { + return *reinterpret_cast(stop_cb_buf_); + } + + public: + ~wait_awaiter() + { + if(active_) + stop_cb_().~stop_cb_t(); + if(published_) + { + // Destroyed while still armed (frame torn down + // without resuming): deregister so a later + // wake cannot touch the dead frame. + int expected = state_armed; + waker_->st_.compare_exchange_strong( + expected, state_empty, + std::memory_order_acq_rel, + std::memory_order_acquire); + } + } + + explicit wait_awaiter(async_waker* waker) noexcept + : waker_(waker) + { + } + + wait_awaiter(wait_awaiter&& o) noexcept + : waker_(o.waker_) + , cont_(o.cont_) + , ex_(o.ex_) + , canceled_(o.canceled_) + , active_(std::exchange(o.active_, false)) + , published_(std::exchange(o.published_, false)) + { + } + + wait_awaiter(wait_awaiter const&) = delete; + wait_awaiter& operator=(wait_awaiter const&) = delete; + wait_awaiter& operator=(wait_awaiter&&) = delete; + + /// Consume a latched token, completing synchronously. + bool await_ready() noexcept + { + int expected = state_token; + return waker_->st_.compare_exchange_strong( + expected, state_empty, + std::memory_order_acq_rel, + std::memory_order_acquire); + } + + /** IoAwaitable protocol overload. */ + std::coroutine_handle<> + await_suspend( + std::coroutine_handle<> h, + io_env const* env) noexcept + { + if(env->stop_token.stop_requested()) + { + canceled_ = true; + return h; + } + cont_.h = h; + ex_ = env->executor; + waker_->waiter_ = this; + + int expected = state_empty; + if(!waker_->st_.compare_exchange_strong( + expected, state_armed, + std::memory_order_acq_rel, + std::memory_order_acquire)) + { + // Single-waiter precondition: a second concurrent + // wait would find the slot armed. + BOOST_CAPY_ASSERT(expected == state_token); + + // A wake latched between await_ready and here; + // consume it and resume inline. + waker_->st_.store( + state_empty, std::memory_order_release); + return h; + } + published_ = true; + + ::new(stop_cb_buf_) stop_cb_t( + env->stop_token, cancel_fn{this}); + active_ = true; + return std::noop_coroutine(); + } + + io_result<> await_resume() noexcept + { + if(active_) + { + stop_cb_().~stop_cb_t(); + active_ = false; + } + published_ = false; + if(canceled_) + return {make_error_code(error::canceled)}; + return {{}}; + } + }; + + /// Construct with no token latched. + async_waker() = default; + + /// Copy constructor (deleted). + async_waker(async_waker const&) = delete; + + /// Copy assignment (deleted). + async_waker& operator=(async_waker const&) = delete; + + /// Move constructor (deleted). + async_waker(async_waker&&) = delete; + + /// Move assignment (deleted). + async_waker& operator=(async_waker&&) = delete; + + /** Asynchronously wait until woken. + + If a token is latched, completes immediately and consumes + it. Otherwise suspends until `wake()` or the stop token + fires. + + @par Preconditions + No other coroutine is currently waiting on this object. + + @return An awaitable that await-returns `io_result<>`; + empty on wakeup, `error::canceled` if the stop + token wins. + */ + wait_awaiter wait() noexcept + { + return wait_awaiter{this}; + } + + /** Wake the waiter, or latch the wakeup if none waits. + + Callable from any thread. The waiter's resumption is + posted through its executor; this call never resumes a + coroutine inline. Multiple calls without an intervening + `wait()` collapse into a single token. + */ + void wake() noexcept + { + for(;;) + { + int s = st_.load(std::memory_order_acquire); + if(s == state_token) + return; + if(s == state_empty) + { + if(st_.compare_exchange_weak( + s, state_token, + std::memory_order_acq_rel, + std::memory_order_acquire)) + return; + continue; + } + // armed: winning this CAS claims the waiter, whose + // frame is pinned until we post its resumption. + if(st_.compare_exchange_weak( + s, state_empty, + std::memory_order_acq_rel, + std::memory_order_acquire)) + { + auto* w = waiter_; + w->ex_.post(w->cont_); + return; + } + } + } +}; + +} // namespace capy +} // namespace boost + +#endif diff --git a/include/boost/capy/ex/detail/timer_service.hpp b/include/boost/capy/ex/detail/timer_service.hpp deleted file mode 100644 index c383c539e..000000000 --- a/include/boost/capy/ex/detail/timer_service.hpp +++ /dev/null @@ -1,128 +0,0 @@ -// -// Copyright (c) 2026 Michael Vandeberg -// -// 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_EX_TIMER_SERVICE_HPP -#define BOOST_CAPY_EX_TIMER_SERVICE_HPP - -#include -#include - -#include -#include -#include -#include -#include -#include -#include -#include -#include - -namespace boost { -namespace capy { -namespace detail { - -/* Shared timer thread for an execution_context. - - One background std::thread per execution_context. All timeouts - scheduled through this context share the same thread, which sleeps - on a condition variable until the next deadline. - - The timer thread never touches coroutine frames or executors - directly — callbacks are responsible for posting work through - the appropriate executor. -*/ - -class BOOST_CAPY_DECL - timer_service - : public execution_context::service -{ -public: - using timer_id = std::uint64_t; - - explicit timer_service(execution_context& ctx); - - // Calls shutdown() to join the background thread. - // Handles the discard path in use_service_impl where - // a duplicate service is deleted without shutdown(). - ~timer_service(); - - /** Schedule a callback to fire after a duration. - - The callback is invoked on the timer service's background - thread. It must not block for extended periods. - - @return An id that can be passed to cancel(). - */ - template - timer_id schedule_after( - std::chrono::duration dur, - std::function cb) - { - auto deadline = std::chrono::steady_clock::now() + dur; - return schedule_at(deadline, std::move(cb)); - } - - /** Cancel a pending timer. - - After this function returns, the callback is guaranteed - not to be running and will never be invoked. If the - callback is currently executing on the timer thread, - this call blocks until it completes. - - Safe to call with any id, including ids that have - already fired, been cancelled, or were never issued. - */ - void cancel(timer_id id); - -protected: - void shutdown() override; - -private: - void stop_and_join(); - struct entry - { - std::chrono::steady_clock::time_point deadline; - timer_id id; - std::function callback; - - bool operator>(entry const& o) const noexcept - { - return deadline > o.deadline; - } - }; - - timer_id schedule_at( - std::chrono::steady_clock::time_point deadline, - std::function cb); - - void run(); - -// warning C4251: std types need to have dll-interface - BOOST_CAPY_MSVC_WARNING_PUSH - BOOST_CAPY_MSVC_WARNING_DISABLE(4251) - std::mutex mutex_; - std::condition_variable cv_; - std::condition_variable cancel_cv_; - std::priority_queue< - entry, - std::vector, - std::greater<>> queue_; - std::unordered_set active_ids_; - timer_id next_id_ = 0; - timer_id executing_id_ = 0; - bool stopped_ = false; - std::thread thread_; - BOOST_CAPY_MSVC_WARNING_POP -}; - -} // detail -} // capy -} // boost - -#endif diff --git a/include/boost/capy/test/stream.hpp b/include/boost/capy/test/stream.hpp index 21ce1f918..c1a6d2c52 100644 --- a/include/boost/capy/test/stream.hpp +++ b/include/boost/capy/test/stream.hpp @@ -233,8 +233,8 @@ class stream // The read suspends when no data is available, parking its // continuation on the side until the peer writes/closes. To // support cancellation it follows the same pattern as - // delay_awaitable: a stop callback claims the resume (racing - // the peer wake via an atomic) and posts the continuation + // async_waker::wait_awaiter: a stop callback claims the resume + // (racing the peer wake via an atomic) and posts the continuation // through the executor. Because it owns a std::atomic and a // std::stop_callback, the awaitable needs explicit move and // destruction (the task promise moves it into its @@ -274,9 +274,9 @@ class stream // accesses the members above. A union gives correct alignment // for stop_cb_t without an alignas specifier, which avoids // MSVC's C4324 padding warning on this function-local class - // (the member-level pragma used by delay_awaitable does not - // suppress it here). Lifetime is managed manually: placement - // new in await_suspend, explicit destruction once done. + // (the member-level pragma used by async_waker::wait_awaiter + // does not suppress it here). Lifetime is managed manually: + // placement new in await_suspend, explicit destruction once done. union { stop_cb_t stop_cb_; }; awaitable(stream* self, MB buffers) noexcept diff --git a/include/boost/capy/timeout.hpp b/include/boost/capy/timeout.hpp deleted file mode 100644 index a1b4aaad4..000000000 --- a/include/boost/capy/timeout.hpp +++ /dev/null @@ -1,228 +0,0 @@ -// -// Copyright (c) 2026 Michael Vandeberg -// -// 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_TIMEOUT_HPP -#define BOOST_CAPY_TIMEOUT_HPP - -#include -#include -#include -#include -#include -#include -#include -#include - -#include -#include -#include -#include - -namespace boost { -namespace capy { -namespace detail { - -template -struct timeout_state -{ - when_all_core core_; - std::atomic winner_{-1}; // -1=none, 0=inner, 1=delay - std::optional inner_result_; - std::exception_ptr inner_exception_; - std::array runner_handles_{}; - - timeout_state() - : core_(2) - { - } -}; - -template -when_all_runner> -make_timeout_inner_runner( - Awaitable inner, timeout_state* state) -{ - try - { - auto result = co_await std::move(inner); - state->inner_result_.emplace(std::move(result)); - } - catch(...) - { - state->inner_exception_ = std::current_exception(); - } - - int expected = -1; - if(state->winner_.compare_exchange_strong( - expected, 0, std::memory_order_relaxed)) - state->core_.stop_source_.request_stop(); -} - -template -when_all_runner> -make_timeout_delay_runner( - DelayAw d, timeout_state* state) -{ - auto result = co_await std::move(d); - - if(!result.ec) - { - int expected = -1; - if(state->winner_.compare_exchange_strong( - expected, 1, std::memory_order_relaxed)) - state->core_.stop_source_.request_stop(); - } -} - -template -class timeout_launcher -{ - Inner* inner_; - DelayAw* delay_; - timeout_state* state_; - -public: - timeout_launcher( - Inner* inner, DelayAw* delay, - timeout_state* state) - : inner_(inner) - , delay_(delay) - , state_(state) - { - } - - bool await_ready() const noexcept { return false; } - - std::coroutine_handle<> await_suspend( - std::coroutine_handle<> continuation, - io_env const* caller_env) - { - state_->core_.continuation_.h = continuation; - state_->core_.caller_env_ = caller_env; - - if(caller_env->stop_token.stop_possible()) - { - state_->core_.parent_stop_callback_.emplace( - caller_env->stop_token, - when_all_core::stop_callback_fn{ - &state_->core_.stop_source_}); - - if(caller_env->stop_token.stop_requested()) - state_->core_.stop_source_.request_stop(); - } - - auto token = state_->core_.stop_source_.get_token(); - - auto r0 = make_timeout_inner_runner( - std::move(*inner_), state_); - auto h0 = r0.release(); - h0.promise().state_ = state_; - h0.promise().env_ = io_env{ - caller_env->executor, token, - caller_env->frame_allocator}; - state_->runner_handles_[0].h = - std::coroutine_handle<>{h0}; - - auto r1 = make_timeout_delay_runner( - std::move(*delay_), state_); - auto h1 = r1.release(); - h1.promise().state_ = state_; - h1.promise().env_ = io_env{ - caller_env->executor, token, - caller_env->frame_allocator}; - state_->runner_handles_[1].h = - std::coroutine_handle<>{h1}; - - caller_env->executor.post( - state_->runner_handles_[0]); - caller_env->executor.post( - state_->runner_handles_[1]); - - return std::noop_coroutine(); - } - - void await_resume() const noexcept {} -}; - -} // namespace detail - -/** Race an io_result-returning awaitable against a deadline. - - Starts the awaitable and a timer concurrently. The first to - complete wins and cancels the other. If the awaitable finishes - first, its result is returned as-is (success, error, or - exception). If the timer fires first, an `io_result` with - `ec == error::timeout` is produced. The timeout fires at or - after the specified duration. - - Unlike @ref when_any, exceptions from the inner awaitable - are always propagated — they are never swallowed by the timer. - - @par Cancellation - - If the parent's stop token is activated, both children are - cancelled. The inner awaitable's cancellation result is - returned. - - @par Example - @code - auto [ec, n] = co_await timeout(sock.read_some(buf), 50ms); - if (ec == cond::timeout) { - // handle timeout - } - @endcode - - @tparam A An IoAwaitable returning `io_result`. - - @param a The awaitable to race against the deadline. - @param dur The maximum duration to wait. - - @return An `io_result` matching the inner awaitable's - result type. On normal completion the inner awaitable's - result is returned unchanged, whether it indicates success - or sets `ec` to an error. On timeout, `ec` is set to - `error::timeout` and the payload values are - default-initialized. - - @throws Rethrows any exception thrown by the inner awaitable, - regardless of whether the timer has fired. - - @see delay, cond::timeout -*/ -template - requires detail::is_io_result_v> -auto timeout(A a, std::chrono::duration dur) - -> task> -{ - using T = awaitable_result_t; - - auto d = delay(dur); - detail::timeout_state state; - - co_await detail::timeout_launcher< - A, decltype(d), T>(&a, &d, &state); - - if(state.core_.first_exception_) - std::rethrow_exception(state.core_.first_exception_); - if(state.inner_exception_) - std::rethrow_exception(state.inner_exception_); - - if(state.winner_.load(std::memory_order_relaxed) == 0) - co_return std::move(*state.inner_result_); - - // Delay fired first: timeout - T r{}; - r.ec = make_error_code(error::timeout); - co_return r; -} - -} // capy -} // boost - -#endif diff --git a/src/ex/detail/timer_service.cpp b/src/ex/detail/timer_service.cpp deleted file mode 100644 index 995a6d3e0..000000000 --- a/src/ex/detail/timer_service.cpp +++ /dev/null @@ -1,125 +0,0 @@ -// -// Copyright (c) 2026 Michael Vandeberg -// -// 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 -// - -#include - -namespace boost { -namespace capy { -namespace detail { - -timer_service:: -timer_service(execution_context& ctx) - : thread_([this] { run(); }) -{ - (void)ctx; -} - -timer_service:: -~timer_service() -{ - stop_and_join(); -} - -timer_service::timer_id -timer_service:: -schedule_at( - std::chrono::steady_clock::time_point deadline, - std::function cb) -{ - std::lock_guard lock(mutex_); - auto id = ++next_id_; - active_ids_.insert(id); - queue_.push(entry{deadline, id, std::move(cb)}); - cv_.notify_one(); - return id; -} - -void -timer_service:: -cancel(timer_id id) -{ - std::unique_lock lock(mutex_); - if(!active_ids_.contains(id)) - return; - if(executing_id_ == id) - { - // Callback is running — wait for it to finish. - // run() erases from active_ids_ after execution. - while(executing_id_ == id) - cancel_cv_.wait(lock); - return; - } - active_ids_.erase(id); -} - -void -timer_service:: -stop_and_join() -{ - { - std::lock_guard lock(mutex_); - stopped_ = true; - } - cv_.notify_one(); - if(thread_.joinable()) - thread_.join(); -} - -void -timer_service:: -shutdown() -{ - stop_and_join(); -} - -void -timer_service:: -run() -{ - std::unique_lock lock(mutex_); - for(;;) - { - if(stopped_) - return; - - if(queue_.empty()) - { - cv_.wait(lock); - continue; - } - - auto deadline = queue_.top().deadline; - auto now = std::chrono::steady_clock::now(); - if(deadline > now) - { - cv_.wait_until(lock, deadline); - continue; - } - - // Pop the entry (const_cast needed because priority_queue::top is const) - auto e = std::move(const_cast(queue_.top())); - queue_.pop(); - - // Skip if cancelled (no longer in active set) - if(!active_ids_.contains(e.id)) - continue; - - executing_id_ = e.id; - lock.unlock(); - e.callback(); - lock.lock(); - active_ids_.erase(e.id); - executing_id_ = 0; - cancel_cv_.notify_all(); - } -} - -} // detail -} // capy -} // boost diff --git a/src/ex/thread_pool.cpp b/src/ex/thread_pool.cpp index 14beea8e5..62e85cd62 100644 --- a/src/ex/thread_pool.cpp +++ b/src/ex/thread_pool.cpp @@ -116,9 +116,9 @@ class thread_pool::impl // Destroy abandoned coroutine frames. Must be called // before execution_context::shutdown()/destroy() so - // that suspended-frame destructors (e.g. delay_awaitable - // calling timer_service::cancel()) run while services - // are still valid. + // that suspended-frame destructors touching services + // (e.g. cancelling registrations) run while those + // services are still valid. void drain_abandoned() noexcept { diff --git a/test/unit/delay.cpp b/test/unit/delay.cpp deleted file mode 100644 index cc53d6cd4..000000000 --- a/test/unit/delay.cpp +++ /dev/null @@ -1,288 +0,0 @@ -// -// Copyright (c) 2026 Michael Vandeberg -// -// 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 -// - -// Test that header file is self-contained. -#include - -#include -#include -#include - -#include "test_helpers.hpp" -#include "test_suite.hpp" - -#include -#include -#include - -namespace boost { -namespace capy { - -using namespace std::chrono_literals; - -struct delay_test -{ - // Test: delay completes after duration - void - testDelayCompletes() - { - thread_pool pool(1); - std::latch done(1); - bool completed = false; - - auto delay_task = [&]() -> task - { - (void) co_await delay(10ms); - completed = true; - }; - - run_async(pool.get_executor(), - [&]() { - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(delay_task()); - - done.wait(); - BOOST_TEST(completed); - } - - // Test: delay waits at least the specified duration - void - testDelayMinimumDuration() - { - thread_pool pool(1); - std::latch done(1); - - auto delay_task = [&]() -> task - { - (void) co_await delay(50ms); - }; - - auto start = std::chrono::steady_clock::now(); - - run_async(pool.get_executor(), - [&]() { - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(delay_task()); - - done.wait(); - auto elapsed = std::chrono::steady_clock::now() - start; - BOOST_TEST(elapsed >= 50ms); - } - - // Test: stop requested before delay suspends (early-out path) - void - testDelayCancellationEarlyOut() - { - thread_pool pool(1); - std::latch done(1); - std::stop_source source; - - auto delay_task = [&]() -> task - { - (void) co_await delay(10s); - }; - - auto start = std::chrono::steady_clock::now(); - - run_async(pool.get_executor(), source.get_token(), - [&]() { - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(delay_task()); - - // Cancel immediately — likely before delay suspends - source.request_stop(); - - done.wait(); - auto elapsed = std::chrono::steady_clock::now() - start; - BOOST_TEST(elapsed < 1s); - } - - // Test: stop requested after delay is fully suspended - // (exercises cancel_fn stop callback path) - void - testDelayCancellationWhileSuspended() - { - thread_pool pool(1); - std::latch done(1); - std::latch suspended(1); - std::stop_source source; - - auto delay_task = [&]() -> task - { - // Signal that we're about to suspend on delay - suspended.count_down(); - (void) co_await delay(10s); - }; - - auto start = std::chrono::steady_clock::now(); - - run_async(pool.get_executor(), source.get_token(), - [&]() { - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(delay_task()); - - // Wait for the task to reach the delay point - suspended.wait(); - // Small sleep to ensure delay_awaitable::await_suspend - // has fully completed (stop callback registered) - std::this_thread::sleep_for(10ms); - source.request_stop(); - - done.wait(); - auto elapsed = std::chrono::steady_clock::now() - start; - BOOST_TEST(elapsed < 1s); - } - - // Test: zero-duration delay completes immediately - void - testZeroDuration() - { - thread_pool pool(1); - std::latch done(1); - bool completed = false; - - auto delay_task = [&]() -> task - { - (void) co_await delay(0ms); - completed = true; - }; - - run_async(pool.get_executor(), - [&]() { - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(delay_task()); - - done.wait(); - BOOST_TEST(completed); - } - - // Test: multiple sequential delays - void - testSequentialDelays() - { - thread_pool pool(1); - std::latch done(1); - int step = 0; - - auto delay_task = [&]() -> task - { - (void) co_await delay(5ms); - step = 1; - (void) co_await delay(5ms); - step = 2; - (void) co_await delay(5ms); - step = 3; - }; - - run_async(pool.get_executor(), - [&]() { - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(delay_task()); - - done.wait(); - BOOST_TEST_EQ(step, 3); - } - - // Test: destroying delay_awaitable while suspended - // cleans up both stop callback and timer - void - testDestroyWhileSuspended() - { -// GCC emits a false -Wmaybe-uninitialized when it inlines -// the stop_callback destructor through the alignas buffer. -#if defined(__GNUC__) && !defined(__clang__) -# pragma GCC diagnostic push -# pragma GCC diagnostic ignored "-Wmaybe-uninitialized" -#endif - thread_pool pool(1); - auto ex = pool.get_executor(); - std::stop_source source; - io_env env{ex, source.get_token(), - std::pmr::get_default_resource()}; - - { - delay_awaitable da(std::chrono::seconds(10)); - // Manually suspend — registers timer and stop callback - da.await_suspend(std::noop_coroutine(), &env); - // da destroyed here without calling await_resume - } - - // If cleanup was incomplete, requesting stop or waiting - // for the timer would access freed memory (UB/crash). - source.request_stop(); - std::this_thread::sleep_for(20ms); - BOOST_TEST(true); -#if defined(__GNUC__) && !defined(__clang__) -# pragma GCC diagnostic pop -#endif - } - - // Test: concurrent delays on a multi-threaded pool - // exercises use_service race and shared timer_service - void - testConcurrentDelays() - { - constexpr int N = 10; - thread_pool pool(4); - std::latch done(N); - - auto delay_task = [](int i) -> task - { - (void) co_await delay(10ms * i); - }; - - for(int i = 0; i < N; ++i) - { - run_async(pool.get_executor(), - [&]() { done.count_down(); }, - [&](std::exception_ptr) { - done.count_down(); - })(delay_task(i)); - } - - done.wait(); - BOOST_TEST(true); - } - - void - run() - { - testDelayCompletes(); - testDelayMinimumDuration(); - testDelayCancellationEarlyOut(); - testDelayCancellationWhileSuspended(); - testZeroDuration(); - testSequentialDelays(); - testDestroyWhileSuspended(); - testConcurrentDelays(); - } -}; - -TEST_SUITE(delay_test, "capy.delay"); - -} // capy -} // boost diff --git a/test/unit/ex/async_waker.cpp b/test/unit/ex/async_waker.cpp new file mode 100644 index 000000000..97e2b2638 --- /dev/null +++ b/test/unit/ex/async_waker.cpp @@ -0,0 +1,272 @@ +// +// 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 +// + +// Test that header file is self-contained. +#include + +#include +#include +#include +#include + +#include +#include +#include + +#include "test_suite.hpp" + +namespace boost { +namespace capy { + +struct async_waker_test +{ + // wake() before wait(): the token is latched and the + // wait completes immediately (the race-free escape hatch). + void testLatchedTokenBeforeWait() + { + thread_pool pool(1); + async_waker waker; + bool ok = false; + + waker.wake(); + + auto t = [](async_waker& waker, bool& ok_out) -> task<> { + auto [ec] = co_await waker.wait(); + ok_out = !ec; + }; + run_async(pool.get_executor())(t(waker, ok)); + pool.join(); + BOOST_TEST(ok); + } + + void testWakeAfterWaitResumes() + { + thread_pool pool(1); + async_waker waker; + bool ok = false; + + auto waiter = [](async_waker& waker, bool& ok_out) -> task<> { + auto [ec] = co_await waker.wait(); + ok_out = !ec; + }; + auto wake_task = [](async_waker& waker) -> task<> { + waker.wake(); + co_return; + }; + run_async(pool.get_executor())(waiter(waker, ok)); + run_async(pool.get_executor())(wake_task(waker)); + pool.join(); + BOOST_TEST(ok); + } + + // The core use case: a user-provided thread wakes into the + // executor. The waiter must resume on the executor, not on + // the waking thread. + void testCrossThreadWake() + { + thread_pool pool(1); + async_waker waker; + bool ok = false; + std::thread::id resume_id; + std::thread::id waker_id; + + auto t = [](async_waker& waker, bool& ok_out, + std::thread::id& resume_id_out) -> task<> { + auto [ec] = co_await waker.wait(); + resume_id_out = std::this_thread::get_id(); + ok_out = !ec; + }; + run_async(pool.get_executor())(t(waker, ok, resume_id)); + + std::thread th([&waker, &waker_id] { + waker_id = std::this_thread::get_id(); + std::this_thread::sleep_for( + std::chrono::milliseconds(20)); + waker.wake(); + }); + pool.join(); + th.join(); + BOOST_TEST(ok); + // Pins the resume locus: the pool has one thread, so a + // mismatch here would mean the wait resumed inline on the + // waking thread instead of being posted through the executor. + BOOST_TEST(resume_id != waker_id); + } + + void testStopCancelsWait() + { + thread_pool pool(1); + async_waker waker; + std::stop_source src; + bool canceled = false; + + auto t = [](async_waker& waker, bool& out) -> task<> { + auto [ec] = co_await waker.wait(); + out = (ec == cond::canceled); + }; + run_async(pool.get_executor(), src.get_token())(t(waker, canceled)); + + std::thread th([&src] { + std::this_thread::sleep_for( + std::chrono::milliseconds(20)); + src.request_stop(); + }); + pool.join(); + th.join(); + BOOST_TEST(canceled); + } + + void testAlreadyStoppedCompletesCanceled() + { + thread_pool pool(1); + async_waker waker; + std::stop_source src; + src.request_stop(); + bool canceled = false; + + auto t = [](async_waker& waker, bool& out) -> task<> { + auto [ec] = co_await waker.wait(); + out = (ec == cond::canceled); + }; + run_async(pool.get_executor(), src.get_token())(t(waker, canceled)); + pool.join(); + BOOST_TEST(canceled); + } + + // Token latched while nobody waits survives a completed wait: + // wake, wait (consumes), wake, wait (consumes again). + void testReuseAfterCompletion() + { + thread_pool pool(1); + async_waker waker; + int count = 0; + + auto t = [](async_waker& waker, int& count_out) -> task<> { + waker.wake(); + { + auto [ec] = co_await waker.wait(); + if(!ec) + ++count_out; + } + waker.wake(); + { + auto [ec] = co_await waker.wait(); + if(!ec) + ++count_out; + } + }; + run_async(pool.get_executor())(t(waker, count)); + pool.join(); + BOOST_TEST_EQ(count, 2); + } + + // Extra wakes collapse into one token (single-token latch, + // not a counting semaphore). With no token left, the second + // wait can only complete via cancellation; a leftover token + // would instead resume it with success, and quickly. + void testExtraWakesCollapse() + { + thread_pool pool(1); + async_waker waker; + bool first_ok = false; + bool second_canceled = false; + + waker.wake(); + waker.wake(); + waker.wake(); + + auto t = [](async_waker& waker, + bool& ok1, bool& canceled2) -> task<> { + auto [ec1] = co_await waker.wait(); // consumes the one token + ok1 = !ec1; + auto [ec2] = co_await waker.wait(); // no token left + canceled2 = (ec2 == cond::canceled); + }; + + std::stop_source src; + run_async(pool.get_executor(), src.get_token())( + t(waker, first_ok, second_canceled)); + + // Give the pool time to reach the second wait, then stop + // it so join() returns. + std::thread th([&src] { + std::this_thread::sleep_for( + std::chrono::milliseconds(50)); + src.request_stop(); + }); + pool.join(); + th.join(); + BOOST_TEST(first_ok); + BOOST_TEST(second_canceled); + } + + // wake() racing request_stop: exactly one side must win; + // if cancel wins, the token is re-latched, not lost. + void testWakeCancelRace() + { + for(int i = 0; i < 100; ++i) + { + thread_pool pool(1); + async_waker waker; + std::stop_source src; + bool resumed = false; + bool canceled = false; + + auto t = [](async_waker& waker, bool& r, bool& c) -> task<> { + auto [ec] = co_await waker.wait(); + if(ec == cond::canceled) + c = true; + else + r = true; + }; + run_async(pool.get_executor(), src.get_token())(t(waker, resumed, canceled)); + + std::thread t1([&waker] { waker.wake(); }); + std::thread t2([&src] { src.request_stop(); }); + t1.join(); + t2.join(); + pool.join(); + + BOOST_TEST(resumed != canceled); + if(canceled) + { + // Cancel claimed the waiter, so the wake must + // have latched instead of vanishing: a fresh wait + // completes immediately with no further wake. + thread_pool pool2(1); + bool token_latched = false; + auto probe = [](async_waker& waker, + bool& out) -> task<> { + auto [ec] = co_await waker.wait(); + out = !ec; + }; + run_async(pool2.get_executor())(probe(waker, token_latched)); + pool2.join(); + BOOST_TEST(token_latched); + } + } + } + + void run() + { + testLatchedTokenBeforeWait(); + testWakeAfterWaitResumes(); + testCrossThreadWake(); + testStopCancelsWait(); + testAlreadyStoppedCompletesCanceled(); + testReuseAfterCompletion(); + testExtraWakesCollapse(); + testWakeCancelRace(); + } +}; + +TEST_SUITE(async_waker_test, "boost.capy.ex.async_waker"); + +} // namespace capy +} // namespace boost diff --git a/test/unit/ex/detail/timer_service.cpp b/test/unit/ex/detail/timer_service.cpp deleted file mode 100644 index 9da009769..000000000 --- a/test/unit/ex/detail/timer_service.cpp +++ /dev/null @@ -1,287 +0,0 @@ -// -// Copyright (c) 2026 Michael Vandeberg -// -// 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 -// - -// Test that header file is self-contained. -#include - -#include - -#include "test_helpers.hpp" - -#include -#include - -namespace boost { -namespace capy { - -using namespace std::chrono_literals; - -struct timer_service_test -{ - // Test: timer fires after duration - void - testBasicFire() - { - thread_pool pool(1); - auto& ts = pool.get_executor().context() - .use_service(); - - std::latch done(1); - bool fired = false; - - ts.schedule_after(1ms, [&] { - fired = true; - done.count_down(); - }); - - done.wait(); - BOOST_TEST(fired); - } - - // Test: cancel prevents callback from firing - void - testCancelBeforeFire() - { - thread_pool pool(1); - auto& ts = pool.get_executor().context() - .use_service(); - - bool fired = false; - - auto id = ts.schedule_after(1s, [&] { - fired = true; - }); - - ts.cancel(id); - - // Give some time to confirm it doesn't fire - std::this_thread::sleep_for(20ms); - BOOST_TEST(!fired); - } - - // Test: cancel on already-fired timer is safe - void - testCancelAfterFire() - { - thread_pool pool(1); - auto& ts = pool.get_executor().context() - .use_service(); - - std::latch done(1); - - auto id = ts.schedule_after(1ms, [&] { - done.count_down(); - }); - - done.wait(); - // Should not block or crash - ts.cancel(id); - } - - // Test: multiple timers fire in deadline order - void - testFiringOrder() - { - thread_pool pool(1); - auto& ts = pool.get_executor().context() - .use_service(); - - std::vector order; - std::mutex mu; - std::latch done(3); - - auto const scale = failsafe_scale; - - ts.schedule_after(30ms * scale, [&] { - std::lock_guard lock(mu); - order.push_back(3); - done.count_down(); - }); - ts.schedule_after(10ms * scale, [&] { - std::lock_guard lock(mu); - order.push_back(1); - done.count_down(); - }); - ts.schedule_after(20ms * scale, [&] { - std::lock_guard lock(mu); - order.push_back(2); - done.count_down(); - }); - - done.wait(); - BOOST_TEST_EQ(order.size(), 3u); - BOOST_TEST_EQ(order[0], 1); - BOOST_TEST_EQ(order[1], 2); - BOOST_TEST_EQ(order[2], 3); - } - - // Test: zero duration fires promptly - void - testZeroDuration() - { - thread_pool pool(1); - auto& ts = pool.get_executor().context() - .use_service(); - - std::latch done(1); - auto start = std::chrono::steady_clock::now(); - - ts.schedule_after(0ms, [&] { - done.count_down(); - }); - - done.wait(); - auto elapsed = std::chrono::steady_clock::now() - start; - BOOST_TEST(elapsed < 100ms); - } - - // Test: many timers scheduled concurrently - void - testManyConcurrent() - { - thread_pool pool(1); - auto& ts = pool.get_executor().context() - .use_service(); - - constexpr int N = 100; - std::atomic count{0}; - std::latch done(N); - - for(int i = 0; i < N; ++i) - { - ts.schedule_after(1ms, [&] { - count.fetch_add(1, std::memory_order_relaxed); - done.count_down(); - }); - } - - done.wait(); - BOOST_TEST_EQ(count.load(), N); - } - - // Test: cancel subset of timers - void - testCancelSubset() - { - thread_pool pool(1); - auto& ts = pool.get_executor().context() - .use_service(); - - std::atomic count{0}; - std::latch done(1); - - auto id1 = ts.schedule_after(10ms, [&] { - count.fetch_add(1, std::memory_order_relaxed); - }); - ts.schedule_after(10ms, [&] { - count.fetch_add(1, std::memory_order_relaxed); - done.count_down(); - }); - auto id3 = ts.schedule_after(10ms, [&] { - count.fetch_add(1, std::memory_order_relaxed); - }); - - ts.cancel(id1); - ts.cancel(id3); - - // Wait for the uncancelled timer to fire - done.wait(); - // Give time for any incorrectly-uncancelled timers - std::this_thread::sleep_for(20ms); - BOOST_TEST_EQ(count.load(), 1); - } - - // Test: shutdown with pending timers doesn't crash - void - testShutdownWithPending() - { - { - thread_pool pool(1); - auto& ts = pool.get_executor().context() - .use_service(); - - // Schedule timers far in the future - ts.schedule_after(10s, [] {}); - ts.schedule_after(10s, [] {}); - ts.schedule_after(10s, [] {}); - - // pool destructor calls shutdown — should not hang - } - BOOST_TEST(true); - } - - // Test: timer fires at or after the specified duration - void - testFiresAtOrAfter() - { - thread_pool pool(1); - auto& ts = pool.get_executor().context() - .use_service(); - - std::latch done(1); - auto start = std::chrono::steady_clock::now(); - auto dur = 50ms; - - ts.schedule_after(dur, [&] { - done.count_down(); - }); - - done.wait(); - auto elapsed = std::chrono::steady_clock::now() - start; - BOOST_TEST(elapsed >= dur); - } - - // Test: cancel blocks while callback is executing - void - testCancelBlocksDuringExecution() - { - thread_pool pool(1); - auto& ts = pool.get_executor().context() - .use_service(); - - std::atomic callback_started{false}; - std::atomic callback_finished{false}; - std::latch started(1); - - auto id = ts.schedule_after(1ms, [&] { - callback_started.store(true); - started.count_down(); - std::this_thread::sleep_for(50ms); - callback_finished.store(true); - }); - - // Wait for callback to start executing - started.wait(); - BOOST_TEST(callback_started.load()); - - // cancel() must block until callback finishes - ts.cancel(id); - BOOST_TEST(callback_finished.load()); - } - - void - run() - { - testBasicFire(); - testCancelBeforeFire(); - testCancelAfterFire(); - testFiringOrder(); - testZeroDuration(); - testManyConcurrent(); - testCancelSubset(); - testShutdownWithPending(); - testFiresAtOrAfter(); - testCancelBlocksDuringExecution(); - } -}; - -TEST_SUITE(timer_service_test, "capy.ex.timer_service"); - -} // capy -} // boost diff --git a/test/unit/quitter.cpp b/test/unit/quitter.cpp index a0778cdea..0ab30446b 100644 --- a/test/unit/quitter.cpp +++ b/test/unit/quitter.cpp @@ -11,8 +11,8 @@ #include #include -#include #include +#include #include #include #include @@ -721,24 +721,26 @@ struct quitter_test } //---------------------------------------------------------- - // 14. Timer cancellation + // 14. Waker cancellation //---------------------------------------------------------- void - testTimerCancellation() + testWakerCancellation() { - using namespace std::chrono_literals; - thread_pool pool(1); std::latch done(1); std::latch suspended(1); std::stop_source source; bool reached_end = false; + // Never woken: an external wait that only cancellation + // can complete. + async_waker waker; + auto q = [&]() -> quitter<> { suspended.count_down(); - auto [ec] = co_await delay(10s); + auto [ec] = co_await waker.wait(); (void)ec; reached_end = true; }; @@ -756,8 +758,8 @@ struct quitter_test done.wait(); auto elapsed = std::chrono::steady_clock::now() - start; - // Should complete promptly, well under 10s - BOOST_TEST(elapsed < 1s); + // Should complete promptly + BOOST_TEST(elapsed < std::chrono::seconds(1)); // Quitter intercepted the stop — body did not continue BOOST_TEST(!reached_end); } @@ -877,7 +879,7 @@ struct quitter_test testCoAwaitThrowingQuitter(); testWhenAllWithStop(); testWhenAnyWithStop(); - testTimerCancellation(); + testWakerCancellation(); testEchoWithShutdown(); } }; diff --git a/test/unit/timeout.cpp b/test/unit/timeout.cpp deleted file mode 100644 index 0ec2d43de..000000000 --- a/test/unit/timeout.cpp +++ /dev/null @@ -1,332 +0,0 @@ -// -// Copyright (c) 2026 Michael Vandeberg -// -// 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 -// - -// Test that header file is self-contained. -#include - -#include -#include -#include -#include -#include - -#include "test_helpers.hpp" -#include "test_suite.hpp" - -#include -#include - -namespace boost { -namespace capy { - -using namespace std::chrono_literals; - -//---------------------------------------------------------- -// Helper tasks for timeout testing -//---------------------------------------------------------- - -// Returns an io_result immediately -inline io_task -returns_io_int(int value) -{ - co_return io_result{{}, value}; -} - -// Returns an io_result immediately -inline io_task -returns_io_string(std::string value) -{ - co_return io_result{{}, std::move(value)}; -} - -// Returns io_result<> immediately (void equivalent) -inline io_task<> -returns_io_void() -{ - co_return io_result<>{}; -} - -// Returns io_result after stop is requested -inline io_task -slow_io_result(std::size_t n) -{ - co_await stop_only_awaitable{}; - co_return io_result{{}, n}; -} - -// Returns io_result after stop is requested -inline io_task -slow_io_int(int value) -{ - co_await stop_only_awaitable{}; - co_return io_result{{}, value}; -} - -// Returns io_result<> after stop is requested -inline io_task<> -slow_io_void() -{ - co_await stop_only_awaitable{}; - co_return io_result<>{}; -} - -// io_task that throws an exception immediately -inline io_task -io_immediate_throw(char const* msg) -{ - throw test_exception(msg); - co_return io_result{{}, 0}; -} - -//---------------------------------------------------------- -// Tests -//---------------------------------------------------------- - -struct timeout_test -{ - // Test: io_result completes before timeout - void - testTaskCompletesBeforeTimeout() - { - thread_pool pool(1); - std::latch done(1); - io_result result{}; - - run_async(pool.get_executor(), - [&](io_result r) { - result = r; - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(timeout(returns_io_int(42), 5s)); - - done.wait(); - BOOST_TEST(!result.ec); - BOOST_TEST_EQ(std::get<0>(result.values), 42); - } - - // Test: io_result completes before timeout - void - testTaskCompletesWithString() - { - thread_pool pool(1); - std::latch done(1); - io_result result{}; - - run_async(pool.get_executor(), - [&](io_result r) { - result = std::move(r); - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(timeout(returns_io_string("hello"), 5s)); - - done.wait(); - BOOST_TEST(!result.ec); - BOOST_TEST_EQ(std::get<0>(result.values), "hello"); - } - - // Test: io_result<> completes before timeout - void - testVoidTaskCompletes() - { - thread_pool pool(1); - std::latch done(1); - io_result<> result{make_error_code(error::timeout)}; - - run_async(pool.get_executor(), - [&](io_result<> r) { - result = r; - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(timeout(returns_io_void(), 5s)); - - done.wait(); - BOOST_TEST(!result.ec); - } - - // Test: Timeout fires - io_result path returns error::timeout - void - testTimeoutIoResult() - { - thread_pool pool(1); - std::latch done(1); - std::error_code ec; - std::size_t n = 999; - - run_async(pool.get_executor(), - [&](io_result r) { - ec = r.ec; - n = std::get<0>(r.values); - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(timeout(slow_io_result(100), 1ms)); - - done.wait(); - BOOST_TEST(ec == error::timeout); - BOOST_TEST(ec == cond::timeout); - BOOST_TEST_EQ(n, 0u); - } - - // Test: Timeout fires - io_result reports error::timeout - void - testTimeoutReportsErrorForInt() - { - thread_pool pool(1); - std::latch done(1); - std::error_code ec; - - run_async(pool.get_executor(), - [&](io_result r) { - ec = r.ec; - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(timeout(slow_io_int(42), 1ms)); - - done.wait(); - BOOST_TEST(ec == error::timeout); - } - - // Test: Timeout fires - io_result<> reports error::timeout - void - testTimeoutReportsErrorForVoid() - { - thread_pool pool(1); - std::latch done(1); - std::error_code ec; - - run_async(pool.get_executor(), - [&](io_result<> r) { - ec = r.ec; - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(timeout(slow_io_void(), 1ms)); - - done.wait(); - BOOST_TEST(ec == error::timeout); - } - - // Test: Zero duration times out immediately - void - testZeroDuration() - { - thread_pool pool(1); - std::latch done(1); - std::error_code ec; - - run_async(pool.get_executor(), - [&](io_result r) { - ec = r.ec; - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(timeout(slow_io_int(42), 0ms)); - - done.wait(); - BOOST_TEST(ec == error::timeout); - } - - // Test: cond::timeout equivalence - void - testCondEquivalence() - { - auto ec = make_error_code(error::timeout); - BOOST_TEST(ec == cond::timeout); - BOOST_TEST(!(ec == cond::canceled)); - BOOST_TEST(!(ec == cond::eof)); - - auto cond_ec = make_error_condition(cond::timeout); - BOOST_TEST(cond_ec.message() == "operation timed out"); - } - - // Inner task throws before delay fires. - // Exception propagates to caller, not swallowed by timer. - void - testThrowPropagatesBeforeTimeout() - { - thread_pool pool(1); - std::latch done(1); - bool caught = false; - std::string msg; - - run_async(pool.get_executor(), - [&](io_result) { - done.count_down(); - }, - [&](std::exception_ptr ep) { - try { std::rethrow_exception(ep); } - catch (test_exception const& e) { - caught = true; - msg = e.what(); - } - done.count_down(); - })(timeout(io_immediate_throw("boom"), 5s)); - - done.wait(); - BOOST_TEST(caught); - BOOST_TEST_EQ(msg, "boom"); - } - - // A caller stop token already requested when timeout launches makes - // timeout propagate the stop to its source. The stop-aware inner - // resumes immediately, so completion is deterministic: the long - // (5s) timer never participates. - void - testStopAlreadyRequested() - { - thread_pool pool(1); - std::latch done(1); - bool completed = false; - std::stop_source src; - src.request_stop(); - - run_async(pool.get_executor(), src.get_token(), - [&](io_result) { - completed = true; - done.count_down(); - }, - [&](std::exception_ptr) { - done.count_down(); - })(timeout(slow_io_int(42), 5s)); - - done.wait(); - BOOST_TEST(completed); - } - - void - run() - { - testTaskCompletesBeforeTimeout(); - testTaskCompletesWithString(); - testVoidTaskCompletes(); - testTimeoutIoResult(); - testTimeoutReportsErrorForInt(); - testTimeoutReportsErrorForVoid(); - testZeroDuration(); - testCondEquivalence(); - testThrowPropagatesBeforeTimeout(); - testStopAlreadyRequested(); - } -}; - -TEST_SUITE(timeout_test, "capy.timeout"); - -} // capy -} // boost