From ded1f7cb37843f2d2b2cb2095f4ca718c17ca119 Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Tue, 3 Nov 2026 16:33:20 +0100 Subject: [PATCH 1/2] feat: add a quitter example built on the public coroutine protocol The example quitter ends at the co_await it is suspended in when a stop is requested, without resuming its body and without throwing. The quitter that sees the stop stores canceled as the result of the topmost quitter in its run and resumes whatever awaited it, which then destroys the suspended frames from the top. Operations resume a gate frame, which applies the operation-decides rule before the body runs. It needs no hooks in the library: a launched or task-awaited quitter must have T = results<...> or std::error_code, enforced by the quitter itself. Includes the shutdown and deep call chain demos and a self-checking program for the stop, race and teardown cases. --- example/CMakeLists.txt | 1 + example/early-quitter/CMakeLists.txt | 22 + example/early-quitter/checks.cpp | 612 +++++++++++++++++++++++++++ example/early-quitter/main.cpp | 141 ++++++ example/early-quitter/quitter.hpp | 560 ++++++++++++++++++++++++ example/early-quitter/shutdown.cpp | 222 ++++++++++ 6 files changed, 1558 insertions(+) create mode 100644 example/early-quitter/CMakeLists.txt create mode 100644 example/early-quitter/checks.cpp create mode 100644 example/early-quitter/main.cpp create mode 100644 example/early-quitter/quitter.hpp create mode 100644 example/early-quitter/shutdown.cpp diff --git a/example/CMakeLists.txt b/example/CMakeLists.txt index b3bb11d2f..d6f309941 100644 --- a/example/CMakeLists.txt +++ b/example/CMakeLists.txt @@ -11,6 +11,7 @@ add_subdirectory(async-mutex) add_subdirectory(buffer-composition) add_subdirectory(custom-executor) +add_subdirectory(early-quitter) add_subdirectory(gui-integration) add_subdirectory(hello-task) add_subdirectory(mock-stream-testing) diff --git a/example/early-quitter/CMakeLists.txt b/example/early-quitter/CMakeLists.txt new file mode 100644 index 000000000..b38608448 --- /dev/null +++ b/example/early-quitter/CMakeLists.txt @@ -0,0 +1,22 @@ +# +# 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 +# + +add_executable(capy_example_early_quitter main.cpp quitter.hpp) +set_property(TARGET capy_example_early_quitter PROPERTY FOLDER "examples") +target_link_libraries(capy_example_early_quitter Boost::capy) + +add_executable(capy_example_early_quitter_checks checks.cpp quitter.hpp) +set_property(TARGET capy_example_early_quitter_checks PROPERTY FOLDER "examples") +target_link_libraries(capy_example_early_quitter_checks Boost::capy) +add_test(NAME capy_example_early_quitter_checks + COMMAND capy_example_early_quitter_checks) + +add_executable(capy_example_early_quitter_shutdown shutdown.cpp quitter.hpp) +set_property(TARGET capy_example_early_quitter_shutdown PROPERTY FOLDER "examples") +target_link_libraries(capy_example_early_quitter_shutdown Boost::capy) diff --git a/example/early-quitter/checks.cpp b/example/early-quitter/checks.cpp new file mode 100644 index 000000000..b33ca4257 --- /dev/null +++ b/example/early-quitter/checks.cpp @@ -0,0 +1,612 @@ +// +// 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 +// + +// Self-checks for the early quitter: stops at each depth, completions +// racing a stop, a task between quitters, and teardown. + +#include "quitter.hpp" + +#include + +#include +#include +#include +#include +#include +#include + +namespace capy = boost::capy; +using early::quitter; +using capy::results; + +static int failures = 0; + +#define CHECK(e) \ + do { if(!(e)) { ++failures; std::printf("%s:%d: CHECK(%s) failed\n", \ + __FILE__, __LINE__, #e); } } while(0) + +struct queue_context : capy::execution_context +{ + queue_context() : execution_context(this) {} +}; + +static queue_context g_ctx; + +// Queues every post; pump() runs them in order. +struct queue_executor +{ + std::queue* q; + bool* closed = nullptr; + + bool operator==(queue_executor const& o) const noexcept { return q == o.q; } + capy::execution_context& context() const noexcept { return g_ctx; } + void on_work_started() const noexcept {} + void on_work_finished() const noexcept {} + + std::coroutine_handle<> dispatch(capy::continuation& c) const + { + post(c); + return std::noop_coroutine(); + } + + // A closed executor gives up what is posted to it. + void post(capy::continuation& c) const + { + if(closed && *closed) + capy::abandon(c); + else + q->push(c); + } +}; + +static_assert(capy::Executor); +static_assert(capy::ContextRunnable>>); +static_assert(capy::ContextRunnable>); + +static void pump(std::queue& q) +{ + while(!q.empty()) + { + auto c = q.front(); + q.pop(); + c.h.resume(); + } +} + +struct counter +{ + int* n; + explicit counter(int& c) : n(&c) {} + counter(counter&& o) noexcept : n(std::exchange(o.n, nullptr)) {} + ~counter() { if(n) ++*n; } +}; + +struct yield_awaitable +{ + capy::continuation c_; + bool await_ready() const noexcept { return false; } + void await_suspend(std::coroutine_handle<> h, capy::await_context const* env) + { + c_.h = h; + c_.owner = env->owner; + env->executor.post(c_); + } + void await_resume() const noexcept {} +}; + +// Completes at once with `r`; optionally requests a stop first. +template +struct ready_awaitable +{ + R r; + std::stop_source* stop = nullptr; + bool await_ready() noexcept + { + if(stop) + stop->request_stop(); + return true; + } + void await_suspend(std::coroutine_handle<>, capy::await_context const*) noexcept {} + R await_resume() { return r; } +}; + +struct throwing_ready +{ + std::stop_source* stop; + bool await_ready() noexcept { stop->request_stop(); return true; } + void await_suspend(std::coroutine_handle<>, capy::await_context const*) noexcept {} + results<> await_resume() { throw std::runtime_error("early"); } +}; + +// What run_async's value handler got. +struct outcome +{ + int calls = 0; + std::error_code ec; + int value = 0; +}; + +template +static void launch(std::queue& q, std::stop_token st, + outcome& o, Q quitter_, bool* closed = nullptr) +{ + queue_executor ex{&q, closed}; + capy::run_async(ex, st, std::pmr::new_delete_resource(), + [&o](auto r) { + ++o.calls; + if constexpr (std::is_same_v) + o.ec = r; + else + { + o.ec = std::get<0>(r); + if constexpr (std::tuple_size_v == 2) + o.value = std::get<1>(r); + } + }, + [&o](std::exception_ptr) { o.calls += 100; })(std::move(quitter_)); +} + +//------------------------------------------------ + +static quitter> two_yields() +{ + co_await yield_awaitable{}; + co_await yield_awaitable{}; + co_return results{std::error_code(), 42}; +} + +static void value_without_stop() +{ + std::queue q; + std::stop_source ss; + outcome o; + launch(q, ss.get_token(), o, two_yields()); + pump(q); + CHECK(o.calls == 1); + CHECK(!o.ec); + CHECK(o.value == 42); +} + +//------------------------------------------------ + +struct reach +{ + bool leaf = false, mid = false, top = false; +}; + +static quitter leaf(capy::async_waker& w, int& down, reach& r) +{ + counter c(down); + std::ignore = co_await w.wait(); + r.leaf = true; + co_return 1; +} + +static quitter<> mid(capy::async_waker& w, int& down, reach& r) +{ + counter c(down); + std::ignore = co_await leaf(w, down, r); + r.mid = true; +} + +static quitter> top(capy::async_waker& w, int& down, reach& r) +{ + counter c(down); + co_await mid(w, down, r); + r.top = true; + co_return results<>{}; +} + +// The stop ends three quitters at once; none resumes, all unwind. +static void deep_stop_while_suspended() +{ + std::queue q; + std::stop_source ss; + capy::async_waker w; + outcome o; + int down = 0; + reach r; + launch(q, ss.get_token(), o, top(w, down, r)); + pump(q); + CHECK(o.calls == 0); + ss.request_stop(); + pump(q); + CHECK(o.calls == 1); + CHECK(o.ec == capy::cond::canceled); + CHECK(down == 3); + CHECK(!r.leaf && !r.mid && !r.top); +} + +//------------------------------------------------ + +static quitter starts_then_waits( + capy::async_waker& w, bool& started, bool& reached) +{ + started = true; + std::ignore = co_await w.wait(); + reached = true; + co_return std::error_code{}; +} + +// With no hook in the launcher, the body runs up to its first co_await. +static void stop_before_start() +{ + std::queue q; + std::stop_source ss; + ss.request_stop(); + capy::async_waker w; + outcome o; + bool started = false, reached = false; + launch(q, ss.get_token(), o, starts_then_waits(w, started, reached)); + pump(q); + CHECK(o.calls == 1); + CHECK(o.ec == capy::cond::canceled); + CHECK(started); + CHECK(!reached); +} + +//------------------------------------------------ + +static quitter> ready_then_yield( + std::stop_source& ss, results r, int& got, bool& reached) +{ + auto [ec, v] = co_await ready_awaitable>{r, &ss}; + got = ec ? -1 : v; + co_await yield_awaitable{}; + reached = true; + co_return results<>{}; +} + +// A result that completed is delivered; the stop lands at the next await. +static void ready_success_then_stop_delivers() +{ + std::queue q; + std::stop_source ss; + outcome o; + int got = 0; + bool reached = false; + launch(q, ss.get_token(), o, + ready_then_yield(ss, results{std::error_code(), 7}, got, reached)); + pump(q); + CHECK(got == 7); + CHECK(!reached); + CHECK(o.calls == 1); + CHECK(o.ec == capy::cond::canceled); +} + +// A canceled result with the stop pending ends the quitter there. +static void ready_canceled_then_stop_stops() +{ + std::queue q; + std::stop_source ss; + outcome o; + int got = 0; + bool reached = false; + launch(q, ss.get_token(), o, ready_then_yield(ss, + results{capy::error::canceled, 0}, got, reached)); + pump(q); + CHECK(got == 0); + CHECK(!reached); + CHECK(o.calls == 1); + CHECK(o.ec == capy::cond::canceled); +} + +//------------------------------------------------ + +// Records whether it was started at all. +struct probe_awaitable +{ + bool* touched; + bool await_ready() noexcept { *touched = true; return true; } + void await_suspend(std::coroutine_handle<>, capy::await_context const*) noexcept {} + void await_resume() const noexcept {} +}; + +static quitter> stop_then_probe(std::stop_source& ss, bool& touched) +{ + co_await ready_awaitable>{results<>{}, &ss}; + co_await probe_awaitable{&touched}; + co_return results<>{}; +} + +// With a stop pending at a co_await, the operation is never started, +// so nothing (a lock, a buffer) is acquired only to be dropped. +static void stop_pending_leaves_operation_untouched() +{ + std::queue q; + std::stop_source ss; + outcome o; + bool touched = false; + launch(q, ss.get_token(), o, stop_then_probe(ss, touched)); + pump(q); + CHECK(!touched); + CHECK(o.calls == 1); + CHECK(o.ec == capy::cond::canceled); +} + +//------------------------------------------------ + +static quitter> early_throw(std::stop_source& ss, bool& caught, + bool& reached) +{ + try + { + co_await throwing_ready{&ss}; + } + catch(std::runtime_error const&) + { + caught = true; + } + co_await yield_awaitable{}; + reached = true; + co_return results<>{}; +} + +// An exception from an early await_resume is not a stop: the body gets it. +static void early_exception_reaches_body() +{ + std::queue q; + std::stop_source ss; + outcome o; + bool caught = false, reached = false; + launch(q, ss.get_token(), o, early_throw(ss, caught, reached)); + pump(q); + CHECK(caught); + CHECK(!reached); + CHECK(o.calls == 1); + CHECK(o.ec == capy::cond::canceled); +} + +//------------------------------------------------ + +static quitter> wake_then_wait( + capy::async_waker& w, int& down, bool& woken, bool& reached) +{ + counter c(down); + auto [ec] = co_await w.wait(); + woken = !ec; + co_await w.wait(); + reached = true; + co_return results<>{}; +} + +// The wake has queued the gate; the stop arrives before it runs. +static void stop_after_wake_before_gate() +{ + std::queue q; + std::stop_source ss; + capy::async_waker w; + outcome o; + int down = 0; + bool woken = false, reached = false; + launch(q, ss.get_token(), o, wake_then_wait(w, down, woken, reached)); + pump(q); + w.wake(); + CHECK(!q.empty()); + ss.request_stop(); + pump(q); + CHECK(woken); + CHECK(!reached); + CHECK(down == 1); + CHECK(o.calls == 1); + CHECK(o.ec == capy::cond::canceled); +} + +//------------------------------------------------ + +static quitter> inner_quitter(capy::async_waker& w, int& down) +{ + counter c(down); + co_await w.wait(); + co_return results<>{}; +} + +static capy::task> middle_task( + capy::async_waker& w, int& down, std::error_code& seen) +{ + counter c(down); + auto [ec] = co_await inner_quitter(w, down); + seen = ec; + co_return results<>{ec}; +} + +static quitter> outer_quitter( + capy::async_waker& w, int& down, std::error_code& seen, bool& reached) +{ + counter c(down); + co_await middle_task(w, down, seen); + reached = true; + co_return results<>{}; +} + +// A task between quitters receives canceled as a value and decides. +static void task_between_quitters() +{ + std::queue q; + std::stop_source ss; + capy::async_waker w; + outcome o; + int down = 0; + std::error_code seen; + bool reached = false; + launch(q, ss.get_token(), o, outer_quitter(w, down, seen, reached)); + pump(q); + ss.request_stop(); + pump(q); + CHECK(seen == capy::cond::canceled); + CHECK(!reached); + CHECK(down == 3); + CHECK(o.calls == 1); + CHECK(o.ec == capy::cond::canceled); +} + +//------------------------------------------------ + +static quitter> thrower() +{ + co_await yield_awaitable{}; + throw std::runtime_error("boom"); +} + +static quitter> catches_child(bool& caught) +{ + try + { + co_await thrower(); + } + catch(std::runtime_error const&) + { + caught = true; + } + co_return results<>{}; +} + +static void exception_propagates() +{ + std::queue q; + std::stop_source ss; + outcome o; + bool caught = false; + launch(q, ss.get_token(), o, catches_child(caught)); + pump(q); + CHECK(caught); + CHECK(o.calls == 1); + CHECK(!o.ec); +} + +//------------------------------------------------ + +// The waker's stop posts the gate to a closed executor, which abandons +// it: the chain is torn down from its owner, and no handler runs. +static void teardown_through_closed_executor() +{ + std::queue q; + std::stop_source ss; + capy::async_waker w; + outcome o; + int down = 0; + reach r; + bool closed = false; + launch(q, ss.get_token(), o, top(w, down, r), &closed); + pump(q); + closed = true; + ss.request_stop(); + pump(q); + CHECK(o.calls == 0); + CHECK(down == 3); + CHECK(!r.leaf && !r.mid && !r.top); +} + +//------------------------------------------------ + +static quitter> waits_once(capy::async_waker& w, int& down) +{ + counter c(down); + co_await w.wait(); + co_return results<>{}; +} + +static capy::task> join_two( + capy::async_waker& a, capy::async_waker& b, int& down, bool& resumed) +{ + std::ignore = co_await capy::when_all(waits_once(a, down), waits_once(b, down)); + resumed = true; + co_return results<>{}; +} + +static void when_all_branches_stop() +{ + std::queue q; + std::stop_source ss; + capy::async_waker a, b; + int down = 0; + bool resumed = false; + bool value = false; + queue_executor ex{&q}; + capy::run_async(ex, ss.get_token(), std::pmr::new_delete_resource(), + [&](results<>) { value = true; })(join_two(a, b, down, resumed)); + pump(q); + ss.request_stop(); + pump(q); + CHECK(resumed); + CHECK(value); + CHECK(down == 2); +} + +//------------------------------------------------ + +struct atomic_counter +{ + std::atomic* n; + explicit atomic_counter(std::atomic& c) : n(&c) {} + ~atomic_counter() + { + ++*n; + n->notify_one(); + } +}; + +static quitter> race_body( + capy::async_waker& w, std::atomic& down, bool& reached) +{ + atomic_counter c(down); + co_await w.wait(); + co_await w.wait(); + reached = true; + co_return results<>{}; +} + +// A wake and a stop from two threads, against a pool thread running +// the quitter: exactly one completion, and the frame always unwinds. +static void stop_races_wake() +{ + capy::thread_pool pool(2); + // Outlives every iteration: the frame notifies it after the + // increment that wakes this thread. + std::atomic down{0}; + for(int i = 0; i < 2000; ++i) + { + capy::async_waker w; + std::stop_source ss; + std::atomic calls{0}; + down = 0; + bool reached = false; + capy::run_async(pool.get_executor(), ss.get_token(), + [&](results<>) { ++calls; }, + [&](std::exception_ptr) { calls += 100; })( + race_body(w, down, reached)); + std::thread waker([&] { w.wake(); }); + std::thread stopper([&] { ss.request_stop(); }); + waker.join(); + stopper.join(); + // The frame is destroyed after the handler runs. + down.wait(0); + CHECK(calls == 1); + CHECK(!reached); + } +} + +int main() +{ + value_without_stop(); + deep_stop_while_suspended(); + stop_before_start(); + ready_success_then_stop_delivers(); + ready_canceled_then_stop_stops(); + stop_pending_leaves_operation_untouched(); + early_exception_reaches_body(); + stop_after_wake_before_gate(); + task_between_quitters(); + exception_propagates(); + teardown_through_closed_executor(); + when_all_branches_stop(); + stop_races_wake(); + std::printf("%s (%d failures)\n", failures ? "FAILED" : "passed", failures); + return failures ? 1 : 0; +} diff --git a/example/early-quitter/main.cpp b/example/early-quitter/main.cpp new file mode 100644 index 000000000..7403caca2 --- /dev/null +++ b/example/early-quitter/main.cpp @@ -0,0 +1,141 @@ +// +// 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 +// + +// Deep call chain with quitter: a stop ends the whole chain at +// whichever co_await it is suspended in, with no checks at any level. +// +// The quitter here is example code (quitter.hpp), built only on capy's +// public coroutine protocol: no exceptions, no hooks in the library. + +#include "quitter.hpp" + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace capy = boost::capy; +using namespace std::chrono_literals; + +static std::stop_source g_stop; + +// request_stop runs the stop callbacks, which post to executors and +// allocate, so the signal handler only sets a lock-free flag and the +// ticker thread requests the stop. +static std::atomic g_signaled{false}; +static_assert(std::atomic::is_always_lock_free); + +extern "C" void on_signal(int) { g_signaled.store(true); } + +// Logs its release, standing in for a socket, file, or buffer. +struct resource_guard +{ + std::string what; + ~resource_guard() + { + std::ostringstream oss; + oss << " released " << what << "\n"; + std::cout << oss.str(); + } +}; + +std::string transform(std::string s) +{ + for(auto& c : s) + c = static_cast(std::toupper(static_cast(c))); + return s; +} + +early::quitter +fetch(int id, int item, capy::async_waker& waker) +{ + resource_guard conn{"connection for item " + std::to_string(item)}; + + std::ignore = co_await waker.wait(); // connect + std::ignore = co_await waker.wait(); // read + + co_return "worker " + std::to_string(id) + + " item " + std::to_string(item); +} + +early::quitter<> +store(std::string const& data, capy::async_waker& waker) +{ + resource_guard file{"output file"}; + + std::ignore = co_await waker.wait(); // open + std::ignore = co_await waker.wait(); // write + + std::ostringstream oss; + oss << " stored " << data << "\n"; + std::cout << oss.str(); +} + +early::quitter<> +process_item(int id, int item, capy::async_waker& waker) +{ + co_await store(transform(co_await fetch(id, item, waker)), waker); +} + +early::quitter worker(int id, capy::async_waker& waker) +{ + resource_guard guard{"worker " + std::to_string(id)}; + for(int item = 0; ; ++item) + co_await process_item(id, item, waker); +} + +int main() +{ + std::signal(SIGINT, on_signal); + + constexpr int n = 4; + capy::thread_pool pool(n); + std::latch done(n); + std::array wakers; + std::vector> strands; + for(int i = 0; i < n; ++i) + strands.emplace_back(pool.get_executor()); + + // Each wake stands in for one I/O operation completing. + std::atomic ticker_stop{false}; + std::thread ticker([&] { + while(!ticker_stop.load()) + { + std::this_thread::sleep_for(20ms); + if(g_signaled.load() && !g_stop.stop_requested()) + g_stop.request_stop(); + for(auto& w : wakers) + w.wake(); + } + }); + + std::cout << "Press Ctrl+C to quit.\n"; + for(int i = 0; i < n; ++i) + capy::run_async(strands[i], g_stop.get_token(), + // A stop arrives here as canceled. + [&](std::error_code) { done.count_down(); })( + worker(i, wakers[i])); + + done.wait(); + ticker_stop = true; + ticker.join(); + std::cout << "Shutdown complete.\n"; +} diff --git a/example/early-quitter/quitter.hpp b/example/early-quitter/quitter.hpp new file mode 100644 index 000000000..c84fce408 --- /dev/null +++ b/example/early-quitter/quitter.hpp @@ -0,0 +1,560 @@ +// +// 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 EARLY_QUITTER_HPP +#define EARLY_QUITTER_HPP + +/* A quitter built only on capy's public coroutine protocol. + + A stop ends a quitter at the co_await it is suspended in, without + resuming its body and without throwing. The quitter that sees the + stop walks up its run of quitters to the topmost one, stores the + stopped value (canceled) as that quitter's result, and resumes + whatever awaited it: a task, run, run_async, or a combinator. To + that awaiter the quitter simply completed. Destroying the top + quitter then destroys the suspended frames below it, so RAII runs + at every level. + + Unlike a task, whose locals are gone by the time it completes, a + stopped quitter's locals live until its owner destroys the frame. + run_async calls its handler first, so a handler that signals "done" + runs before the quitter's RAII cleanup: wait for the executor's + work to finish (thread_pool::join, for example) before touching + state those destructors use. + + Each quitter hands its operations a gate (a small coroutine frame) + instead of its own handle. A completion resumes the gate, which + decides before the body runs: with a stop pending, the operation's + result is taken early, and a canceled results or error_code (or any + result that cannot carry the stop) ends the quitter; any other + result is delivered, and the stop lands at the next co_await. + + Because the stop is a value, the first awaiter that is not a + quitter must receive a T of results<...> or std::error_code. +*/ + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace early { + +namespace capy = boost::capy; + +/// Test whether `T` can carry a stop as a value. +template +struct carries_stop : std::false_type {}; + +template<> +struct carries_stop : std::true_type {}; + +template +struct carries_stop> : std::true_type {}; + +template +inline constexpr bool carries_stop_v = carries_stop::value; + +template +class quitter; + +namespace detail { + +template +struct is_quitter : std::false_type {}; + +template +struct is_quitter> : std::true_type {}; + +template +bool is_canceled(T const& v) noexcept +{ + if constexpr (std::is_same_v) + return v == capy::cond::canceled; + else + return std::get<0>(v) == capy::cond::canceled; +} + +template +T stopped_value() +{ + std::error_code ec = capy::error::canceled; + if constexpr (std::is_same_v) + return ec; + else + return [&](std::tuple*) { + return T{ec, Ts{}...}; + }(static_cast(nullptr)); +} + +class promise_base; + +// What a quitter's operations resume instead of its body. +struct gate +{ + struct promise_type : capy::frame_alloc_mixin + { + promise_base* owner = nullptr; + + gate get_return_object() noexcept + { + return {std::coroutine_handle::from_promise(*this)}; + } + + std::suspend_always initial_suspend() noexcept { return {}; } + std::suspend_always final_suspend() noexcept { return {}; } + void return_void() noexcept {} + void unhandled_exception() noexcept { std::terminate(); } + }; + + std::coroutine_handle h; +}; + +// The parts of a quitter's promise that do not depend on T. +class promise_base +{ +protected: + ~promise_base() + { + if(gate_) + gate_.destroy(); + } + +public: + promise_base* parent_ = nullptr; + std::coroutine_handle<> body_; + void* awaiter_ = nullptr; + bool (*stops_)(void*) noexcept = nullptr; + bool stopped_ = false; + capy::await_context const* stop_env_ = nullptr; + + virtual void store_stopped() noexcept = 0; + virtual std::coroutine_handle<> take_continuation() noexcept = 0; + + bool stop_pending() const noexcept + { + return stop_env_->stop_token.stop_requested(); + } + + std::coroutine_handle<> gate_handle(); + + // End the run of quitters this one belongs to, delivering the + // stopped value from the top. + std::coroutine_handle<> stop() noexcept + { + promise_base* top = this; + top->stopped_ = true; + while(top->parent_) + { + top = top->parent_; + top->stopped_ = true; + } + top->store_stopped(); + return top->take_continuation(); + } + + std::coroutine_handle<> decide() noexcept + { + if(stop_pending() && stops_(awaiter_)) + return stop(); + return body_; + } + +private: + std::coroutine_handle gate_; +}; + +struct decide_awaiter +{ + bool await_ready() const noexcept { return false; } + + std::coroutine_handle<> await_suspend( + std::coroutine_handle h) const noexcept + { + return h.promise().owner->decide(); + } + + void await_resume() const noexcept {} +}; + +inline gate make_gate() +{ + for(;;) + co_await decide_awaiter{}; +} + +inline std::coroutine_handle<> promise_base::gate_handle() +{ + if(!gate_) + { + gate_ = make_gate().h; + gate_.promise().owner = this; + } + return gate_; +} + +template +struct return_base +{ + std::optional result_; + + void return_value(T v) { result_.emplace(std::move(v)); } + T&& result() noexcept { return std::move(*result_); } +}; + +template<> +struct return_base +{ + void return_void() noexcept {} +}; + +} // namespace detail + +/** A coroutine that a stop ends at its current `co_await`. + + The body needs no cancellation checks: with a stop requested, the + quitter ends at the next `co_await` that cannot complete without + it, and its frames are destroyed without resuming. Awaited by + another quitter, it ends that one too. Awaited by anything else, + it completes with `canceled`, so `T` must then be `results<...>` + or `std::error_code`. + + A stopped quitter's locals are destroyed when its owner destroys + it, not when it completes. Under `run_async` that is after the + handler returns. + + @tparam T The result type. +*/ +template +class [[nodiscard]] quitter +{ +public: + struct promise_type + : capy::context_awaitable_promise_base + , detail::promise_base + , detail::return_base + { + using base = capy::context_awaitable_promise_base; + + std::exception_ptr ep_; + + quitter get_return_object() noexcept + { + return quitter{ + std::coroutine_handle::from_promise(*this)}; + } + + auto initial_suspend() noexcept + { + struct awaiter + { + promise_type* p_; + bool await_ready() const noexcept { return false; } + void await_suspend(std::coroutine_handle<>) const noexcept {} + void await_resume() const noexcept + { + capy::set_current_frame_allocator( + p_->environment()->frame_allocator); + } + }; + return awaiter{this}; + } + + auto final_suspend() noexcept + { + struct awaiter + { + promise_type* p_; + bool await_ready() const noexcept { return false; } + std::coroutine_handle<> await_suspend( + std::coroutine_handle<>) const noexcept + { + return p_->base::continuation(); + } + void await_resume() const noexcept {} + }; + return awaiter{this}; + } + + void unhandled_exception() noexcept + { + ep_ = std::current_exception(); + } + + std::exception_ptr exception() const noexcept { return ep_; } + + // Only a launcher calls this, and it reads the result, which + // must be able to hold the stop. + void set_continuation(std::coroutine_handle<> h) noexcept + { + static_assert(carries_stop_v, + "a launched quitter must have T = results<...> or " + "std::error_code"); + base::set_continuation(h); + } + + // Also kept in the base, so the stop check needs no virtual call. + void set_environment(capy::await_context const* env) noexcept + { + base::set_environment(env); + this->stop_env_ = env; + } + + void store_stopped() noexcept override + { + if constexpr (carries_stop_v) + this->result_.emplace(detail::stopped_value()); + else + std::terminate(); // the top of a run always carries the stop + } + + std::coroutine_handle<> take_continuation() noexcept override + { + return base::continuation(); + } + + template + struct transform_awaiter + { + using A = std::decay_t; + using R = decltype(std::declval().await_resume()); + + // Hold a stop-carrying result taken early by value; any + // other result cannot carry the stop, so a stop wins. + static constexpr bool holds = + !std::is_reference_v && carries_stop_v; + using held_type = std::conditional_t; + + A a_; + promise_type* p_; + std::optional held_{}; + std::exception_ptr ep_{}; + bool stop_here_ = false; + + bool stops_on_completion() noexcept + { + try + { + if constexpr (holds) + { + held_.emplace(a_.await_resume()); + return detail::is_canceled(*held_); + } + else + { + // Called so anything handed over is destroyed. + static_cast(a_.await_resume()); + return true; + } + } + catch(...) + { + ep_ = std::current_exception(); + return false; + } + } + + static bool stops_at_gate(void* self) noexcept + { + return static_cast( + self)->stops_on_completion(); + } + + bool await_ready() + { + if(p_->stop_pending()) + { + stop_here_ = true; + return false; + } + if(!a_.await_ready()) + return false; + stop_here_ = p_->stop_pending() && stops_on_completion(); + return !stop_here_; + } + + std::coroutine_handle<> await_suspend( + std::coroutine_handle h) + { + if(stop_here_) + return p_->stop(); + p_->body_ = h; + p_->awaiter_ = this; + p_->stops_ = &stops_at_gate; + auto gate = p_->gate_handle(); + auto const* env = p_->environment(); + if constexpr (detail::is_quitter::value) + { + return a_.start(gate, env, p_); + } + else + { + using S = decltype(a_.await_suspend(gate, env)); + if constexpr (std::is_void_v) + { + a_.await_suspend(gate, env); + return std::noop_coroutine(); + } + else if constexpr (std::is_same_v) + { + if(a_.await_suspend(gate, env)) + return std::noop_coroutine(); + return p_->decide(); + } + else + { + return a_.await_suspend(gate, env); + } + } + } + + decltype(auto) await_resume() + { + capy::set_current_frame_allocator( + p_->environment()->frame_allocator); + if(ep_) + std::rethrow_exception(std::exchange(ep_, nullptr)); + if constexpr (holds) + { + if(held_) + return R(std::move(*held_)); + } + return a_.await_resume(); + } + }; + + template + auto transform_awaitable(Awaitable&& a) + { + static_assert(capy::ContextAwaitable>, + "requires ContextAwaitable"); + return transform_awaiter{ + std::forward(a), this}; + } + }; + + /// Destroy the quitter and its frame. + ~quitter() { destroy_frame(); } + + /// Construct by taking the frame of `other`. + quitter(quitter&& other) noexcept + : h_(std::exchange(other.h_, nullptr)) + , lent_(std::exchange(other.lent_, false)) + { + } + + /// Destroy the held frame, then take the frame of `other`. + quitter& operator=(quitter&& other) noexcept + { + if(this != &other) + { + destroy_frame(); + h_ = std::exchange(other.h_, nullptr); + lent_ = std::exchange(other.lent_, false); + } + return *this; + } + + /// Return false: a quitter always starts by suspending its awaiter. + bool await_ready() const noexcept { return false; } + + /** Start the quitter, awaited by something that is not a quitter. + + @param cont The awaiter, resumed when the quitter completes or + stops. + @param env The environment the quitter runs in. + + @return The quitter's frame, to resume. + */ + std::coroutine_handle<> await_suspend( + std::coroutine_handle<> cont, capy::await_context const* env) + { + static_assert(carries_stop_v, + "a quitter awaited outside a quitter must have " + "T = results<...> or std::error_code"); + return start(cont, env, nullptr); + } + + /** Return the quitter's result, or its stopped value. + + @throws Any exception the body let escape. + */ + auto await_resume() + { + lent_ = false; + auto& p = h_.promise(); + if(p.ep_) + std::rethrow_exception(p.ep_); + if constexpr (!std::is_void_v) + return std::move(*p.result_); + } + + /// Return the coroutine handle; the quitter keeps the frame. + std::coroutine_handle handle() const noexcept { return h_; } + + /// Give up the frame without destroying it. + void release() noexcept { h_ = nullptr; } + +private: + template + friend class quitter; + + std::coroutine_handle h_; + bool lent_ = false; + + explicit quitter(std::coroutine_handle h) noexcept + : h_(h) + { + } + + std::coroutine_handle<> start( + std::coroutine_handle<> cont, + capy::await_context const* env, + detail::promise_base* parent) noexcept + { + auto& p = h_.promise(); + p.parent_ = parent; + p.promise_type::base::set_continuation(cont); + p.set_environment(env); + lent_ = true; + return h_; + } + + // A lent frame is destroyed only by a stop of its run, or by the + // unwind of an abandoned chain. + void destroy_frame() noexcept + { + if(!h_) + return; + if(lent_ && !h_.done()) + { + auto& p = h_.promise(); + auto* owner = p.environment()->owner; + if(!p.stopped_ && !(owner && owner->abandoned())) + std::terminate(); + } + std::exchange(h_, nullptr).destroy(); + lent_ = false; + } +}; + +} // namespace early + +#endif diff --git a/example/early-quitter/shutdown.cpp b/example/early-quitter/shutdown.cpp new file mode 100644 index 000000000..d6c88455c --- /dev/null +++ b/example/early-quitter/shutdown.cpp @@ -0,0 +1,222 @@ +// +// 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) +// +// Official repository: https://github.com/cppalliance/capy +// + +/* Quitter Shutdown Example + + Demonstrates quitter for responsive application shutdown. + + Four workers simulate a batch file-processing pipeline: each + "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 (the example quitter in + quitter.hpp): 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 + holds a resource_guard that logs its cleanup), and the + application prints a summary and exits. + + Contrast with task<>: + With task<>, every co_await that touches I/O needs: + auto [ec] = co_await waker.wait(); + if(ec) co_return; // <-- cancellation boilerplate + This is repeated at every suspension point. + + With a quitter, a stop request ends the worker without + resuming it: the launcher's handler receives canceled as the + worker's result, then the frame is destroyed and its RAII + guards run. The worker body is pure business logic. +*/ + +#include "quitter.hpp" + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace capy = boost::capy; +using namespace std::chrono_literals; + +// Global stop source; the ticker requests the stop after Ctrl+C. +static std::stop_source g_stop; +static std::chrono::steady_clock::time_point g_stop_time{}; + +// request_stop runs the stop callbacks, which post to executors and +// allocate, so it cannot be called from a signal handler. The handler +// only sets a lock-free flag, which the ticker thread turns into the +// stop request. +static std::atomic g_signaled{false}; +static_assert(std::atomic::is_always_lock_free); + +extern "C" void signal_handler(int) +{ + g_signaled.store(true, std::memory_order_relaxed); +} + +// RAII resource that logs construction and destruction. +// Simulates holding a file handle, socket, or temp buffer +// that must be released on shutdown. +struct resource_guard +{ + int id; + std::atomic& cleanup_count; + + resource_guard(int id_, std::atomic& count) + : id(id_) + , cleanup_count(count) + { + std::ostringstream oss; + oss << " [worker " << id << "] acquired resources\n"; + std::cout << oss.str(); + } + + ~resource_guard() + { + ++cleanup_count; + std::ostringstream oss; + oss << " [worker " << id << "] released resources " + << "(cleanup)\n"; + std::cout << oss.str(); + } + + resource_guard(resource_guard const&) = delete; + resource_guard& operator=(resource_guard const&) = delete; +}; + +// A single worker: download → transform → write, repeated. +// No cancellation code. quitter handles it. +early::quitter worker( + int id, + capy::async_waker& waker, + std::atomic& items_processed, + std::atomic& cleanup_count) +{ + resource_guard guard(id, cleanup_count); + + for(int item = 0; ; ++item) + { + // Simulate download: suspend until the ticker wakes us. + std::ignore = co_await waker.wait(); + + // Simulate transform (CPU work — no co_await needed) + { + std::ostringstream oss; + oss << " [worker " << id << "] processing item " + << item << "\n"; + std::cout << oss.str(); + } + + // Simulate write: suspend for another wakeup. + std::ignore = co_await waker.wait(); + + ++items_processed; + } + + // Never reached — the loop is infinite. + // A stop request ends the worker at the co_await it is waiting in. +} + +int main() +{ + std::signal(SIGINT, signal_handler); +#ifdef SIGTERM + std::signal(SIGTERM, signal_handler); +#endif + + constexpr int num_workers = 4; + capy::thread_pool pool(num_workers); + std::latch done(num_workers); + + 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, and requests the stop once Ctrl+C has + // been pressed. + std::atomic ticker_stop{false}; + std::thread ticker([&] { + while(!ticker_stop.load(std::memory_order_relaxed)) + { + std::this_thread::sleep_for(50ms); + if(g_signaled.load(std::memory_order_relaxed) && + !g_stop.stop_requested()) + { + g_stop_time = std::chrono::steady_clock::now(); + g_stop.request_stop(); + } + 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( + strands[i], + g_stop.get_token(), + // A stop arrives here as canceled. + [&](std::error_code) { done.count_down(); })( + worker(i, wakers[i], items_processed, cleanup_count)); + } + + done.wait(); + + // Stop and join the ticker now that every worker has finished, + // 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(); + + // A stopped worker's frame, and the resource_guard in it, is + // destroyed after its handler runs: wait for that before reading + // the counters it updates. + pool.join(); + + // The ticker wrote g_stop_time before it was joined. + auto stop_at = g_stop_time; + auto now = std::chrono::steady_clock::now(); + + std::cout << "\nShutdown complete.\n" + << " Items processed: " << items_processed << "\n" + << " Workers cleaned up: " << cleanup_count + << "/" << num_workers << "\n"; + + if(stop_at != std::chrono::steady_clock::time_point{}) + { + auto us = std::chrono::duration_cast< + std::chrono::microseconds>(now - stop_at).count(); + std::cout << " Shutdown latency: " << us << " us\n"; + } +} From a0dea4e017938fa792019768e292ca3af9387eb4 Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Tue, 3 Nov 2026 16:33:33 +0100 Subject: [PATCH 2/2] feat!: remove quitter from the library quitter now lives in example/early-quitter, which needs nothing from the library beyond the public coroutine protocol. Removing it also removes stop_requested_exception, so the default error handler no longer catches a cancellation exception, and await_ready_then_suspend loses the out-parameter only quitter used. The quitter-shutdown example moves to example/early-quitter. --- doc/continuation-rationale.md | 2 +- doc/lint/baseline.json | 11 - .../pages/4.coroutines/4f.cancellation.adoc | 4 +- doc/modules/ROOT/pages/glossary.adoc | 6 - doc/mrdocs.yml | 5 +- doc/quitter.md | 481 ------ example/CMakeLists.txt | 1 - example/quitter-shutdown/CMakeLists.txt | 22 - example/quitter-shutdown/quitter_shutdown.cpp | 199 --- include/boost/capy.hpp | 1 - .../boost/capy/concept/context_awaitable.hpp | 2 +- .../capy/detail/await_suspend_helper.hpp | 40 +- include/boost/capy/detail/config.hpp | 2 +- include/boost/capy/detail/run_callbacks.hpp | 15 +- .../capy/detail/stop_requested_exception.hpp | 28 - include/boost/capy/ex/run_async.hpp | 5 +- include/boost/capy/quitter.hpp | 795 --------- include/boost/capy/task.hpp | 3 +- test/unit/quitter.cpp | 1496 ----------------- test/unit/test_helpers.hpp | 8 - 20 files changed, 19 insertions(+), 3107 deletions(-) delete mode 100644 doc/quitter.md delete mode 100644 example/quitter-shutdown/CMakeLists.txt delete mode 100644 example/quitter-shutdown/quitter_shutdown.cpp delete mode 100644 include/boost/capy/detail/stop_requested_exception.hpp delete mode 100644 include/boost/capy/quitter.hpp delete mode 100644 test/unit/quitter.cpp diff --git a/doc/continuation-rationale.md b/doc/continuation-rationale.md index 84c893943..eba8bce53 100644 --- a/doc/continuation-rationale.md +++ b/doc/continuation-rationale.md @@ -221,7 +221,7 @@ wraps it in the embedded `continuation`, and passes that to `co_await`). 3. Aligns with corosio's pattern, where I/O services already embed their operation state (`scheduler_op`) in the awaitable. -4. Zero burden on task authors. `task`, `quitter`, and future +4. Zero burden on task authors. `task` and future task types are unchanged. 5. Cancel callbacks store `continuation*` pointing into the awaitable, which outlives the suspension. diff --git a/doc/lint/baseline.json b/doc/lint/baseline.json index dd90586ad..478ccc3aa 100644 --- a/doc/lint/baseline.json +++ b/doc/lint/baseline.json @@ -263,17 +263,6 @@ "lint/.docstrings/io/any_write_stream.hpp.adoc:#4:Vale.Spelling", "lint/.docstrings/io/write_now.hpp.adoc:#1:Vale.Spelling", "lint/.docstrings/io/write_now.hpp.adoc:#2:Vale.Spelling", - "lint/.docstrings/quitter.hpp.adoc:#10:Vale.Spelling", - "lint/.docstrings/quitter.hpp.adoc:#1:Google.OxfordComma", - "lint/.docstrings/quitter.hpp.adoc:#1:Vale.Spelling", - "lint/.docstrings/quitter.hpp.adoc:#2:Vale.Spelling", - "lint/.docstrings/quitter.hpp.adoc:#3:Vale.Spelling", - "lint/.docstrings/quitter.hpp.adoc:#4:Vale.Spelling", - "lint/.docstrings/quitter.hpp.adoc:#5:Vale.Spelling", - "lint/.docstrings/quitter.hpp.adoc:#6:Vale.Spelling", - "lint/.docstrings/quitter.hpp.adoc:#7:Vale.Spelling", - "lint/.docstrings/quitter.hpp.adoc:#8:Vale.Spelling", - "lint/.docstrings/quitter.hpp.adoc:#9:Vale.Spelling", "lint/.docstrings/task.hpp.adoc:#1:Google.OxfordComma", "lint/.docstrings/task.hpp.adoc:#1:Vale.Spelling", "lint/.docstrings/task.hpp.adoc:#2:Vale.Spelling", diff --git a/doc/modules/ROOT/pages/4.coroutines/4f.cancellation.adoc b/doc/modules/ROOT/pages/4.coroutines/4f.cancellation.adoc index 22f65b69a..b19b53a12 100644 --- a/doc/modules/ROOT/pages/4.coroutines/4f.cancellation.adoc +++ b/doc/modules/ROOT/pages/4.coroutines/4f.cancellation.adoc @@ -159,12 +159,12 @@ include::example$snippets/4f_cancellation.cpp[tag=access_stop_token,indent=0] `std::coroutine_handle::destroy()` is the {cpp}20 primitive that frees a xref:glossary.adoc#coroutine-frame[coroutine frame]. It is not a cancellation mechanism. It has the same flaw as forceful thread interruption. The coroutine is torn down with no opportunity to complete pending I/O, release locks, or run RAII destructors in the expected order. -Capy exposes cpp:task::handle[handle]`()` and cpp:quitter::handle[handle]`()` so that Capy's own launchers (cpp:run_async[], cpp:run[]) and custom integrations can dispatch coroutines through xref:glossary.adoc#executor[executors]. Calling `destroy()` on such a handle while the coroutine is being awaited by a parent produces undefined behavior: the parent still holds the frame and will destroy it again. A holder that must give up a suspended coroutine for good, such as an xref:glossary.adoc#execution-context[execution context] being torn down, passes its xref:glossary.adoc#continuation[continuation] to cpp:abandon[] instead, which unwinds the chain from the top on its owner's executor. +Capy exposes cpp:task::handle[handle]`()` so that Capy's own launchers (cpp:run_async[], cpp:run[]) and custom integrations can dispatch coroutines through xref:glossary.adoc#executor[executors]. Calling `destroy()` on such a handle while the coroutine is being awaited by a parent produces undefined behavior: the parent still holds the frame and will destroy it again. A holder that must give up a suspended coroutine for good, such as an xref:glossary.adoc#execution-context[execution context] being torn down, passes its xref:glossary.adoc#continuation[continuation] to cpp:abandon[] instead, which unwinds the chain from the top on its owner's executor. The rule: * To cancel work, request a stop on a `std::stop_source` whose token the work observes. The work unwinds cleanly through `final_suspend` and any RAII guards run in the correct order. -* Do not call `destroy()` on a handle returned by cpp:task::handle[handle]`()` or cpp:quitter::handle[handle]`()` while the coroutine is being awaited. +* Do not call `destroy()` on a handle returned by cpp:task::handle[handle]`()` while the coroutine is being awaited. == Responding to Cancellation diff --git a/doc/modules/ROOT/pages/glossary.adoc b/doc/modules/ROOT/pages/glossary.adoc index 31e6c05e3..ea029f9f1 100644 --- a/doc/modules/ROOT/pages/glossary.adoc +++ b/doc/modules/ROOT/pages/glossary.adoc @@ -202,12 +202,6 @@ any amount of data. The object inside a coroutine frame that controls the coroutine's behavior, such as how it starts, finishes, and stores its result. -== Q - -[[quitter]]Quitter:: -A task-like coroutine type that unwinds early when its stop token is -triggered, instead of completing its body. See cpp:quitter[]. - == R [[result]]Result:: diff --git a/doc/mrdocs.yml b/doc/mrdocs.yml index 2bacbf45c..c0d34085f 100644 --- a/doc/mrdocs.yml +++ b/doc/mrdocs.yml @@ -19,14 +19,13 @@ exclude-symbols: - 'boost::capy::test::bufgrind::next_awaitable' # Private members of anonymous unions. `inaccessible-members: never` does not # reach variant members of an anonymous union in this MrDocs build, so they -# leak into the reference as "[variant member]" rows. In the task and quitter -# promises, `ep_` is the exception_ptr held in a union so it is not constructed +# leak into the reference as "[variant member]" rows. In the task promise, +# `ep_` is the exception_ptr held in a union so it is not constructed # or destroyed automatically. Its sibling `has_ep_`/`state_` -- private but not # in a union -- is correctly filtered, which is what identifies the union as # the cause. The `**` spans the unnamed union's own scope, which sits between # the class and the members. - 'boost::capy::task::**::ep_' - - 'boost::capy::quitter::**::ep_' implementation-defined: - 'boost::capy::detail' - 'boost::capy::*::detail' diff --git a/doc/quitter.md b/doc/quitter.md deleted file mode 100644 index 3a4130a4c..000000000 --- a/doc/quitter.md +++ /dev/null @@ -1,481 +0,0 @@ -# `quitter` - Stop-Aware Task Type - -## Summary - -`quitter` is a task type that satisfies `ContextRunnable` and provides -transparent cancellation. When the stop token is triggered, the -coroutine body never sees the cancellation. The promise intercepts -it on resume and short-circuits to `final_suspend`. RAII cleanup -runs normally. The parent sees a "stopped" completion. The coroutine -author writes zero cancellation-handling code. - -## Motivation - -Ville Voutilainen (the emeritus chair of EWG in WG21) identified a gap in Capy's `task`: -when an application needs to shut down (e.g. Qt "close all windows"), -every in-flight coroutine should exit cleanly without the programmer -writing `if (ec) co_return;` after every `co_await`. The sender model -achieves this via `set_stopped` - the coroutine never resumes. Capy's -`task` resumes with `operation_aborted` and requires explicit -handling. - -`quitter` closes this gap. It is a policy-level task type: same -ContextAwaitable protocol, same await_context propagation, same frame allocator, -same symmetric transfer - but with automatic stop-on-resume behavior. - -## Behavior - -### Normal operation (stop not requested) - -Identical to `task`. The coroutine runs, co_awaits I/O operations, -returns a value or throws. No observable difference. - -### When stop is requested - -1. The stop token in `await_context` is triggered (from a parent, a scope, - or application shutdown). -2. The in-flight I/O operation observes the token and completes - (typically with `operation_aborted`). -3. The I/O awaitable resumes the coroutine via symmetric transfer. -4. **Before the coroutine body sees the result**, the promise's - `transform_awaiter::await_resume()` checks the stop token. -5. If `env_->stop_token.stop_requested()` is true, the awaiter - throws a sentinel exception (`detail::stop_requested_exception`). -6. `unhandled_exception()` catches it and sets a `stopped_` flag - instead of storing the exception. -7. `final_suspend` returns the continuation via symmetric transfer. -8. The parent's `await_resume()` observes the stopped flag. - -### Why throw internally? - -Throwing inside the coroutine is the only portable mechanism to -jump from an arbitrary suspension point to `final_suspend` while -running destructors for all in-scope locals. The exception never -escapes the coroutine - `unhandled_exception()` catches it. The -user never sees it. No `exception_ptr` is stored. The cost is one -throw + catch per cancellation, which happens once per coroutine -lifetime (not per operation). - -Alternative: a compiler-level `coroutine_handle<>::destroy()` from -outside would also run destructors, but the parent loses the -completion signal. The internal-throw approach preserves structured -completion. - -## Interface - -### Header - -``` - -``` - -### Synopsis - -```cpp -namespace boost::capy { - -template -struct [[nodiscard]] quitter -{ - struct promise_type; - - // ContextAwaitable - bool await_ready() const noexcept; - auto await_resume(); - std::coroutine_handle<> await_suspend( - std::coroutine_handle<> cont, - await_context const* env); - - // ContextRunnable - std::coroutine_handle handle() const noexcept; - void release() noexcept; - - ~quitter(); - quitter(quitter&&) noexcept; - quitter& operator=(quitter&&) noexcept; - quitter(quitter const&) = delete; - quitter& operator=(quitter const&) = delete; -}; - -} // namespace boost::capy -``` - -### `promise_type` - -```cpp -struct quitter::promise_type - : context_awaitable_promise_base - , detail::task_return_base -{ - // Completion state - enum class completion { running, value, exception, stopped }; - - completion state_ = completion::running; - union { std::exception_ptr ep_; }; - - promise_type() noexcept; - ~promise_type(); - - quitter get_return_object(); - auto initial_suspend() noexcept; // same as task - auto final_suspend() noexcept; // same as task - - void unhandled_exception(); - std::exception_ptr exception() const noexcept; - bool stopped() const noexcept; - - template - auto transform_awaitable(Awaitable&& a); -}; -``` - -## Implementation Details - -### `detail::stop_requested_exception` - -```cpp -namespace boost::capy::detail { - -struct stop_requested_exception {}; - -} // namespace boost::capy::detail -``` - -A lightweight sentinel type. Not derived from `std::exception`. -Never escapes the coroutine. Never stored in `exception_ptr`. - -### `promise_type::unhandled_exception()` - -```cpp -void unhandled_exception() -{ - try { - throw; - } - catch (detail::stop_requested_exception const&) { - state_ = completion::stopped; - } - catch (...) { - new (&ep_) std::exception_ptr(std::current_exception()); - state_ = completion::exception; - } -} -``` - -### `promise_type::exception()` and `stopped()` - -```cpp -std::exception_ptr exception() const noexcept -{ - if (state_ == completion::exception) - return ep_; - return {}; -} - -bool stopped() const noexcept -{ - return state_ == completion::stopped; -} -``` - -### `transform_awaiter` (the key difference from `task`) - -```cpp -template -struct transform_awaiter -{ - std::decay_t a_; - promise_type* p_; - - bool await_ready() noexcept - { - return a_.await_ready(); - } - - template - auto await_suspend(std::coroutine_handle h) noexcept - { - using R = decltype( - a_.await_suspend(h, p_->environment())); - if constexpr (std::is_same_v< - R, std::coroutine_handle<>>) - return detail::symmetric_transfer( - a_.await_suspend(h, p_->environment())); - else - return a_.await_suspend( - h, p_->environment()); - } - - decltype(auto) await_resume() - { - set_current_frame_allocator( - p_->environment()->frame_allocator); - - // THE KEY DIFFERENCE: check stop token before - // returning to the coroutine body - if (p_->environment()->stop_token.stop_requested()) - throw detail::stop_requested_exception{}; - - return a_.await_resume(); - } -}; -``` - -### `quitter::await_resume()` - -```cpp -auto await_resume() -{ - if (h_.promise().stopped()) - throw detail::stop_requested_exception{}; - if (h_.promise().state_ == promise_type::completion::exception) - std::rethrow_exception(h_.promise().ep_); - if constexpr (!std::is_void_v) - return std::move(*h_.promise().result_); -} -``` - -When a `quitter` is awaited inside another `quitter`, the stopped -state propagates upward: the parent's `await_resume` throws -`stop_requested_exception`, which the parent's `unhandled_exception` -catches and sets `stopped_`. The chain unwinds automatically. - -When a `quitter` is awaited inside a regular `task`, the -`stop_requested_exception` becomes an unhandled exception in the -`task`. The `task`'s `await_resume` rethrows it. The caller must -handle it. This is by design: `task` does not opt into automatic -cancellation. - -### `quitter::await_suspend()` - -Identical to `task::await_suspend()`: - -```cpp -std::coroutine_handle<> await_suspend( - std::coroutine_handle<> cont, - await_context const* env) -{ - h_.promise().set_continuation(cont); - h_.promise().set_environment(env); - return h_; -} -``` - -### Everything else - -`initial_suspend`, `final_suspend`, `get_return_object`, -`return_value`/`return_void`, frame allocation (`operator -new`/`delete`), `handle()`, `release()`, move operations, -destructor - all identical to `task`. - -## Concept Satisfaction - -`quitter` satisfies: - -- `ContextAwaitable` - has `await_suspend(coroutine_handle<>, await_context const*)` -- `ContextRunnable` - has `promise_type`, `handle()`, `release()`, - `exception()`, `set_continuation()`, `set_environment()`, - `result()` (non-void) - -`quitter` is usable with `run`, `run_async`, `when_all`, -`when_any`, and any launcher that accepts `ContextRunnable`. - -## Interaction with `when_all` / `when_any` - -`when_all` and `when_any` create a child `stop_source` and register -a `stop_callback` on the parent token. When the parent requests -stop, the child token is triggered. Each child `quitter` observes -the token on its next resume and short-circuits. All children exit -cleanly. The parent combinator sees the stopped completions and -propagates accordingly. - -No changes to `when_all` or `when_any` are needed. The stop -propagation already works through `await_context`. The only new behavior -is inside `quitter`'s `transform_awaiter::await_resume()`. - -## Usage - -### Basic - -```cpp -quitter<> session(tcp_socket& sock) -{ - char buf[1024]; - auto [ec, n] = co_await sock.read_some(buf); - // If stop was requested, we never reach here. - // RAII cleanup for buf (trivial) and sock (if owned) - // runs automatically. - if (ec) - co_return; - co_await process(buf, n); -} -``` - -### Application shutdown - -```cpp -quitter<> run_server(tcp_acceptor& acc) -{ - while (true) - { - auto [ec, sock] = co_await acc.accept(); - if (ec) - co_return; - co_await scope.spawn(session(sock)); - } - // When the application calls stop_source.request_stop(), - // the next co_await in this coroutine (or any child) - // short-circuits. All sessions exit. All sockets close - // via RAII. The server exits cleanly. -} -``` - -### Mixing with `task` - -```cpp -task<> outer() -{ - // If the quitter is stopped, its await_resume throws - // stop_requested_exception. The task sees it as an - // unhandled exception. The task's caller must handle it. - co_await some_quitter_function(); -} -``` - -This is intentional: `task` does not opt into automatic -cancellation. The programmer who uses `task` has chosen -explicit error handling. - -## Test Plan - -### Unit tests (in `test/unit/quitter.cpp`) - -1. **Normal completion** - `quitter` returns a value. - Verify `await_resume()` returns the value. Verify - `stopped()` is false. - -2. **Void completion** - `quitter<>` completes normally. - -3. **Exception propagation** - `quitter<>` throws. Verify - `exception()` is non-null. Verify `stopped()` is false. - -4. **Stop before first co_await** - Request stop before the - quitter starts its first I/O operation. Verify the first - `co_await` short-circuits. Verify `stopped()` is true. - Verify RAII destructors ran (use a counting guard). - -5. **Stop during I/O** - Start an I/O operation, request stop - while it is in flight. Verify the coroutine does not resume - into the body. Verify `stopped()` is true. Verify RAII - destructors ran. - -6. **Stop propagation through chain** - `quitter` awaits - `quitter` awaits `quitter`. Request stop. Verify all three - report `stopped()`. Verify all RAII destructors ran. - -7. **Stop propagation with when_all** - `when_all` of two - `quitter` tasks. Request stop. Verify both stop. Verify - the `when_all` completes (does not hang). - -8. **Stop propagation with when_any** - `when_any` of two - `quitter` tasks. One completes normally. Verify the other - is stopped. - -9. **Mixing quitter and task** - `task<>` awaits `quitter<>`. - Request stop. Verify the `quitter` stops. Verify the - `task` sees `stop_requested_exception` as an unhandled - exception. - -10. **No stop requested** - Run a `quitter` to completion - without ever requesting stop. Verify identical behavior - to `task`. - -11. **RAII verification** - Use a struct with a destructor - that increments a counter. Declare it in the coroutine - body before a `co_await`. Request stop. Verify the - counter was incremented (destructor ran). - -12. **Multiple co_await** - `quitter` does three sequential - I/O operations. Request stop after the second completes. - Verify the third `co_await` short-circuits. Verify the - first two results were processed normally. - -### Integration tests - -13. **TCP echo with shutdown** - Start a TCP echo server - using `quitter`. Connect a client. Exchange data. Request - stop. Verify the server exits cleanly and the socket is - closed. - -14. **Timer cancellation** - `quitter` awaits a long timer. - Request stop. Verify the timer is cancelled and the - coroutine exits promptly. - -## Files to Create - -| File | Contents | -|------|----------| -| `include/boost/capy/quitter.hpp` | `quitter` class template | -| `include/boost/capy/detail/stop_requested_exception.hpp` | Sentinel exception type | -| `test/unit/quitter.cpp` | Unit tests | - -## Files to Modify - -| File | Change | -|------|--------| -| `include/boost/capy.hpp` | Add `#include ` | -| `test/unit/CMakeLists.txt` | Add `quitter.cpp` to test sources | - -## Design Decisions - -**Why a separate type instead of a flag on `task`?** - -The task type is a policy point. Users who want explicit error -handling use `task`. Users who want transparent cancellation -use `quitter`. The ContextAwaitable protocol does not mandate a -task. This is the design P4003 enables. - -**Why throw internally instead of `goto final_suspend`?** - -C++ coroutines have no mechanism to jump from an arbitrary -suspension point to `final_suspend` other than throwing. The -throw runs destructors for all in-scope locals. The exception -is caught in `unhandled_exception()` and never escapes. The -cost is one throw per cancellation per coroutine lifetime. - -**Why check in `await_resume` instead of `await_suspend`?** - -`await_suspend` has already committed to suspending. The I/O -operation is in flight. Checking the stop token in `await_suspend` -would race with the I/O completion. Checking in `await_resume` -is safe: the operation has completed, the coroutine is about to -resume, and we intercept before the body sees the result. - -**Why not check in `initial_suspend::await_resume`?** - -We could, and should. If the stop token is already triggered -when the quitter starts, the first `initial_suspend::await_resume` -should throw `stop_requested_exception` immediately. Add this -check to `initial_suspend`'s awaiter: - -```cpp -void await_resume() const noexcept(false) -{ - set_current_frame_allocator( - p_->environment()->frame_allocator); - if (p_->environment()->stop_token.stop_requested()) - throw detail::stop_requested_exception{}; -} -``` - -Note: this changes `initial_suspend::await_resume` from `noexcept` -to potentially-throwing. This is safe because `unhandled_exception()` -will catch it. - -## Execution Protocol - -Save output after each complete semantic unit or to-do item (never -mid-paragraph). Always save output BEFORE marking plan items done - -never the reverse. On resumption: read the plan and last ~30 lines -of the output file. Repair any truncated tail. Continue from where -output ends, matching existing style. Never rewrite prior content. diff --git a/example/CMakeLists.txt b/example/CMakeLists.txt index d6f309941..b65379a2c 100644 --- a/example/CMakeLists.txt +++ b/example/CMakeLists.txt @@ -18,7 +18,6 @@ add_subdirectory(mock-stream-testing) add_subdirectory(parallel-fetch) add_subdirectory(parallel-tasks) add_subdirectory(producer-consumer) -add_subdirectory(quitter-shutdown) add_subdirectory(strand-serialization) add_subdirectory(timeout-cancellation) add_subdirectory(type-erased-echo) diff --git a/example/quitter-shutdown/CMakeLists.txt b/example/quitter-shutdown/CMakeLists.txt deleted file mode 100644 index 14d6f164d..000000000 --- a/example/quitter-shutdown/CMakeLists.txt +++ /dev/null @@ -1,22 +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 -# - -file(GLOB_RECURSE PFILES CONFIGURE_DEPENDS *.cpp *.hpp - CMakeLists.txt - Jamfile) - -source_group(TREE ${CMAKE_CURRENT_SOURCE_DIR} PREFIX "" FILES ${PFILES}) - -add_executable(capy_example_quitter_shutdown ${PFILES}) - -set_property(TARGET capy_example_quitter_shutdown - PROPERTY FOLDER "examples") - -target_link_libraries(capy_example_quitter_shutdown - Boost::capy) diff --git a/example/quitter-shutdown/quitter_shutdown.cpp b/example/quitter-shutdown/quitter_shutdown.cpp deleted file mode 100644 index afa4bec67..000000000 --- a/example/quitter-shutdown/quitter_shutdown.cpp +++ /dev/null @@ -1,199 +0,0 @@ -// -// 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) -// -// Official repository: https://github.com/cppalliance/capy -// - -/* Quitter Shutdown Example - - Demonstrates quitter for responsive application shutdown. - - Four workers simulate a batch file-processing pipeline: each - "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 - holds a resource_guard that logs its cleanup), and the - application prints a summary and exits. - - Contrast with task<>: - With task<>, every co_await that touches I/O needs: - auto [ec] = co_await waker.wait(); - if(ec) co_return; // <-- cancellation boilerplate - This is repeated at every suspension point. - - With quitter<>, the promise intercepts the stop token - automatically. The worker body is pure business logic. -*/ - -#include - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -namespace capy = boost::capy; -using namespace std::chrono_literals; - -// Global stop source wired to Ctrl+C. -static std::stop_source g_stop; -static std::atomic - g_stop_time{std::chrono::steady_clock::time_point{}}; - -extern "C" void signal_handler(int) -{ - g_stop_time.store(std::chrono::steady_clock::now(), - std::memory_order_relaxed); - g_stop.request_stop(); -} - -// RAII resource that logs construction and destruction. -// Simulates holding a file handle, socket, or temp buffer -// that must be released on shutdown. -struct resource_guard -{ - int id; - std::atomic& cleanup_count; - - resource_guard(int id_, std::atomic& count) - : id(id_) - , cleanup_count(count) - { - std::ostringstream oss; - oss << " [worker " << id << "] acquired resources\n"; - std::cout << oss.str(); - } - - ~resource_guard() - { - ++cleanup_count; - std::ostringstream oss; - oss << " [worker " << id << "] released resources " - << "(cleanup)\n"; - std::cout << oss.str(); - } - - resource_guard(resource_guard const&) = delete; - resource_guard& operator=(resource_guard const&) = delete; -}; - -// A single worker: download → transform → write, repeated. -// No cancellation code. quitter handles it. -capy::quitter<> worker( - int id, - capy::async_waker& waker, - std::atomic& items_processed, - std::atomic& cleanup_count) -{ - resource_guard guard(id, cleanup_count); - - for(int item = 0; ; ++item) - { - // Simulate download: suspend until the ticker wakes us. - (void) co_await waker.wait(); - - // Simulate transform (CPU work — no co_await needed) - { - std::ostringstream oss; - oss << " [worker " << id << "] processing item " - << item << "\n"; - std::cout << oss.str(); - } - - // Simulate write: suspend for another wakeup. - (void) co_await waker.wait(); - - ++items_processed; - } - - // Never reached — the loop is infinite. - // quitter exits at the next co_await after stop is requested. -} - -int main() -{ - std::signal(SIGINT, signal_handler); -#ifdef SIGTERM - std::signal(SIGTERM, signal_handler); -#endif - - constexpr int num_workers = 4; - capy::thread_pool pool(num_workers); - std::latch done(num_workers); - - 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( - strands[i], - g_stop.get_token(), - [&]() { done.count_down(); }, - [&](std::exception_ptr) { done.count_down(); })( - 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(); - - std::cout << "\nShutdown complete.\n" - << " Items processed: " << items_processed << "\n" - << " Workers cleaned up: " << cleanup_count - << "/" << num_workers << "\n"; - - if(stop_at != std::chrono::steady_clock::time_point{}) - { - auto us = std::chrono::duration_cast< - std::chrono::microseconds>(now - stop_at).count(); - std::cout << " Shutdown latency: " << us << " us\n"; - } -} diff --git a/include/boost/capy.hpp b/include/boost/capy.hpp index 5429efffd..e49e77fa0 100644 --- a/include/boost/capy.hpp +++ b/include/boost/capy.hpp @@ -20,7 +20,6 @@ // Core types #include #include -#include #include #include diff --git a/include/boost/capy/concept/context_awaitable.hpp b/include/boost/capy/concept/context_awaitable.hpp index 02de16529..989d166b6 100644 --- a/include/boost/capy/concept/context_awaitable.hpp +++ b/include/boost/capy/concept/context_awaitable.hpp @@ -142,7 +142,7 @@ concept AwaitSuspendResult = @par Models - General-purpose class templates that model `ContextAwaitable`: @ref task, @ref quitter, @ref immediate. + General-purpose class templates that model `ContextAwaitable`: @ref task, @ref immediate. @see @ref AwaitSuspendResult, @ref ContextRunnable, @ref await_context, @ref executor_ref diff --git a/include/boost/capy/detail/await_suspend_helper.hpp b/include/boost/capy/detail/await_suspend_helper.hpp index 46f457ca0..dffd9b552 100644 --- a/include/boost/capy/detail/await_suspend_helper.hpp +++ b/include/boost/capy/detail/await_suspend_helper.hpp @@ -77,7 +77,7 @@ namespace detail { The void-returning transfer alone does not cover every case on 19.51. See `await_ready_then_suspend` below for the second - safeguard that `task` and `quitter` apply where this one is on. + safeguard that `task` applies where this one is on. `_M_ARM64EC` is included conservatively. It generates ARM64 code and has not been tested here; keeping the workaround on @@ -124,7 +124,7 @@ symmetric_transfer(std::coroutine_handle<> h) noexcept #if BOOST_CAPY_SYMMETRIC_TRANSFER_WORKAROUND /** Ask whether an awaitable is ready, and suspend on it if not. - Used by the `transform_awaiter` of `task` and `quitter` where the + Used by the `transform_awaiter` of `task` where the symmetric-transfer workaround is on. There the awaiter's own `await_ready` returns `false`, and its `await_suspend` calls this instead, so the wrapped awaitable's `await_ready` runs inside @@ -139,9 +139,7 @@ symmetric_transfer(std::coroutine_handle<> h) noexcept above in place. Observed on 19.51; not on 19.44. A plain or atomic load in `await_ready` was not affected. The regression tests are `testCatchAfterRmwAwaitReady` and - `testCatchAroundWakerWait` in test/unit/task.cpp, and - `testCatchAfterRmwAwaitReady` and - `testCatchStopAfterRmwAwaitReady` in test/unit/quitter.cpp. + `testCatchAroundWakerWait` in test/unit/task.cpp. The awaiting coroutine continues without suspending when the wrapped awaitable is ready, or when its `await_suspend` returns @@ -150,14 +148,6 @@ symmetric_transfer(std::coroutine_handle<> h) noexcept coroutine stays suspended. Returning `false` instead of resuming inline keeps immediate completions from growing the stack. - When `completed` is given, a returned handle equal to `h` is - treated as a synchronous completion as well: it is not resumed, - `*completed` is set, and `false` is returned. `*completed` is set - on every synchronous completion and is never written after a real - suspension, when the frame may already be running elsewhere. - `quitter` uses this to hand over a synchronous result even if a - stop arrives. `task` passes nothing and keeps the inline resume. - @param a The wrapped awaitable. @param h The coroutine performing the `co_await`. @@ -165,9 +155,6 @@ symmetric_transfer(std::coroutine_handle<> h) noexcept @param env The execution environment to pass to the wrapped awaitable's `await_suspend`. - @param completed If not null, set to `true` when the wrapped - awaitable completes synchronously. - @return `false` to continue without suspending, otherwise `true`. */ template @@ -175,24 +162,15 @@ bool await_ready_then_suspend( Awaitable& a, std::coroutine_handle<> h, - await_context const* env, - bool* completed = nullptr) noexcept + await_context const* env) noexcept { // Called exactly once: await_ready may consume state. if(a.await_ready()) - { - if(completed) - *completed = true; return false; - } using R = decltype(a.await_suspend(h, env)); if constexpr (std::is_same_v) { - if(a.await_suspend(h, env)) - return true; - if(completed) - *completed = true; - return false; + return a.await_suspend(h, env); } else if constexpr (std::is_void_v) { @@ -201,13 +179,7 @@ await_ready_then_suspend( } else { - std::coroutine_handle<> next = a.await_suspend(h, env); - if(completed && next == h) - { - *completed = true; - return false; - } - symmetric_transfer(next); + symmetric_transfer(a.await_suspend(h, env)); return true; } } diff --git a/include/boost/capy/detail/config.hpp b/include/boost/capy/detail/config.hpp index 95461f03c..a5b2ee143 100644 --- a/include/boost/capy/detail/config.hpp +++ b/include/boost/capy/detail/config.hpp @@ -81,7 +81,7 @@ the suspend point loses its handler. Reproduced on 19.44 and 19.51; the first clause already covers 19.44. - Where it is on, task's and quitter's transform_awaiter also defer + Where it is on, task's transform_awaiter also defers the wrapped await_ready into await_suspend, through detail::await_ready_then_suspend. That works around a related ARM64 release defect the void-returning transfer does not cover. diff --git a/include/boost/capy/detail/run_callbacks.hpp b/include/boost/capy/detail/run_callbacks.hpp index db8206135..b622b2c54 100644 --- a/include/boost/capy/detail/run_callbacks.hpp +++ b/include/boost/capy/detail/run_callbacks.hpp @@ -11,7 +11,6 @@ #define BOOST_CAPY_DETAIL_RUN_CALLBACKS_HPP #include -#include #include #include @@ -33,20 +32,12 @@ struct default_handler { } + // An unhandled exception propagates to the trampoline's + // unhandled_exception, which terminates. void operator()(std::exception_ptr ep) const { - if(!ep) - return; - try - { + if(ep) std::rethrow_exception(ep); - } - catch(stop_requested_exception const&) - { - // Cancellation is a normal completion, not an error. - } - // A real unhandled exception propagates to the trampoline's - // unhandled_exception, which terminates. } }; diff --git a/include/boost/capy/detail/stop_requested_exception.hpp b/include/boost/capy/detail/stop_requested_exception.hpp deleted file mode 100644 index f5bc47e98..000000000 --- a/include/boost/capy/detail/stop_requested_exception.hpp +++ /dev/null @@ -1,28 +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_DETAIL_STOP_REQUESTED_EXCEPTION_HPP -#define BOOST_CAPY_DETAIL_STOP_REQUESTED_EXCEPTION_HPP - -namespace boost { -namespace capy { -namespace detail { - -/* Lightweight sentinel thrown inside quitter when the stop token - is triggered. Not derived from std::exception. Never escapes the - coroutine — unhandled_exception() catches it and sets the stopped - flag. The cost is one throw+catch per cancellation per coroutine - lifetime. */ -struct stop_requested_exception {}; - -} // namespace detail -} // namespace capy -} // namespace boost - -#endif diff --git a/include/boost/capy/ex/run_async.hpp b/include/boost/capy/ex/run_async.hpp index 533132d7f..64bb4a317 100644 --- a/include/boost/capy/ex/run_async.hpp +++ b/include/boost/capy/ex/run_async.hpp @@ -218,9 +218,8 @@ struct run_async_trampoline // An exception reaches here only by escaping a handler: a handler // that threw, or the default handler rethrowing an otherwise - // unhandled task exception. Cancellation is filtered out earlier - // by default_handler, so this is always a genuine error with no - // owner to receive it: fail fast. + // unhandled task exception. It has no owner to receive it: fail + // fast. void unhandled_exception() noexcept { std::terminate(); } // LCOV_EXCL_LINE }; diff --git a/include/boost/capy/quitter.hpp b/include/boost/capy/quitter.hpp deleted file mode 100644 index f6b8341aa..000000000 --- a/include/boost/capy/quitter.hpp +++ /dev/null @@ -1,795 +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_QUITTER_HPP -#define BOOST_CAPY_QUITTER_HPP - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include -#include -#include -#include -#include - -/* Stop-aware coroutine task. - - quitter is identical to task except that a stop request ends - the body without the body checking an error code. The promise's - transform_awaiter checks the stop token before it touches an awaited - operation, and again when an operation that completed - asynchronously resumes the body. A pending stop there throws a - sentinel exception that unwinds through RAII destructors to - final_suspend, and the parent sees a "stopped" completion. An - operation that completes synchronously always has its await_resume - called, because its await_ready or await_suspend may already have - acquired something. A results<> it returns holding cond::canceled - while the stop is pending acquired nothing, so the stop throws there; - any other result reaches the body, and the stop takes effect at the - next co_await. - - See doc/quitter.md for the full design rationale. */ - -namespace boost { -namespace capy { - -namespace detail { - -// Reuse the same return-value storage as task. -// task_return_base is defined in task.hpp, but quitter needs its own -// copy to avoid a header dependency on task.hpp. -template -struct quitter_return_base -{ - std::optional result_; - - void return_value(T value) - { - result_ = std::move(value); - } - - T&& result() noexcept - { - return std::move(*result_); - } -}; - -template<> -struct quitter_return_base -{ - void return_void() - { - } -}; - -} // namespace detail - -/** Defers a coroutine body until awaited, then unwinds it early on a stop request. - - When the stop token is triggered, the body is unwound at a - `co_await` instead of resuming, and RAII destructors run normally. - Await-effects below says at which one. The parent observes a "stopped" - completion via @ref promise_type::stopped. - - Everything else — frame allocation, environment propagation, - symmetric transfer, move semantics — is identical to @ref task. - - @par Await-effects - - Let `q` be a `quitter`. `co_await q` always suspends the awaiting - coroutine, then transfers control directly into the quitter's - coroutine body on the current thread; no executor operation is - posted. The quitter records the caller's environment (executor, stop - token, and frame allocator) by pointer rather than copying it. It - propagates that environment to every `co_await` inside the body. - - Unlike @ref task, the stop token is checked before the body's first - statement, before each awaited operation is started, and when an - operation that completed asynchronously resumes the body. If a stop - request is pending at those points, the body is not resumed. An - internal sentinel exception unwinds it instead, so RAII destructors - run, and the coroutine completes as stopped. - - An awaited operation that completes synchronously always has its - `await_resume` called, even if a stop request arrives at the same - time. Its `await_ready` or `await_suspend` may already have acquired - a resource, such as `async_mutex`'s lock, that only its - `await_resume` hands over. If that result is a @ref results holding - `cond::canceled` while a stop is pending, the operation saw the stop - first and acquired nothing, so the body is unwound there. Any other - result is handed to the body, and the stop takes effect at the next - `co_await`. An operation whose result is not a @ref results can - therefore hand the body one result it produced after seeing the - stop. - - The body runs until it returns, exits via an exception, or is unwound - by a stop request. Control then transfers directly back to the - awaiting coroutine, again without an executor operation. - - @par Await-returns - The value the body passed to `co_return`, moved out of the quitter, - or nothing when `T` is `void`. - - If the body exits via an unhandled exception, that exception is - rethrown instead. - - If the coroutine completed as stopped, the internal sentinel - exception is thrown instead of await-returning. Awaiting a stopped - `quitter` from another `quitter` therefore stops that one too. A - @ref task awaiting it sees the sentinel as an unhandled exception in - its own body. When a quitter is started by `run_async`, a stopped - completion reaches the error handler as the sentinel - `std::exception_ptr`, not the value handler. - - @par Await-postcondition - The quitter's coroutine has run to completion and is suspended at its - final suspend point; the body's RAII destructors have run. Exactly - one of the following holds: the body returned a value; the body - exited via an exception; or `handle().promise().stopped()` returns - `true`. When the body returned a value, the await moved it out, so a - quitter must not be awaited twice. - - @par Remarks - Supports _ContextAwaitable cancellation_. - - @tparam T The result type. Use `quitter<>` for `quitter`. - - @see @ref task, @ref ContextRunnable, @ref ContextAwaitable -*/ -template -struct [[nodiscard]] BOOST_CAPY_CORO_AWAIT_ELIDABLE - quitter -{ - /** Stores `quitter`'s result and unwinds the body when the stop token fires. - - This is the promise object the compiler associates with a - `quitter` coroutine. It satisfies the coroutine promise - requirements and participates in the I/O awaitable protocol via - @ref context_awaitable_promise_base. Unlike @ref task::promise_type, - its `transform_awaitable` checks the stop token before each - awaited result reaches the body. A pending stop request throws an - internal sentinel exception that unwinds to a "stopped" - completion. It is part of the coroutine machinery and is not - intended to be used directly by callers. - - Result storage and `return_value`/`return_void` are provided by - `detail::quitter_return_base`. - - @see @ref context_awaitable_promise_base, @ref ContextRunnable - */ - struct promise_type - : context_awaitable_promise_base - , detail::quitter_return_base - { - private: - friend quitter; - - enum class completion { running, value, exception, stopped }; - - union { std::exception_ptr ep_; }; - completion state_; - - public: - /// Construct the promise in the running state. - promise_type() noexcept - : state_(completion::running) - { - } - - /// Destroy the promise, releasing any stored exception. - ~promise_type() - { - if(state_ == completion::exception || - state_ == completion::stopped) - ep_.~exception_ptr(); - } - - /** Return a non-null exception_ptr when the coroutine threw - or was stopped. - - Stopped quitters report the sentinel - stop_requested_exception so that run_async routes to - the error handler instead of accessing a non-existent - result. - - @return The stored exception if the coroutine exited via an - exception or was stopped, otherwise a null - `std::exception_ptr`. - */ - std::exception_ptr exception() const noexcept - { - if(state_ == completion::exception || - state_ == completion::stopped) - return ep_; - return {}; - } - - /** True when the coroutine was stopped via the stop token. - - @return `true` if the body was unwound by a stop request; - `false` if it returned a value or exited via any other - exception. - */ - bool stopped() const noexcept - { - return state_ == completion::stopped; - } - - /** Return the owning `quitter` for this coroutine. - - Called by the compiler to produce the object returned to the - caller when the coroutine is created. - - @return A `quitter` owning the coroutine frame. - */ - quitter get_return_object() - { - return quitter{ - std::coroutine_handle::from_promise(*this)}; - } - - /** Return the initial-suspend awaiter. - - The coroutine always suspends at the initial suspend point, - so the body does not start until the quitter is awaited. When - the body is resumed, the awaiter restores the thread-local - frame allocator. It then throws the internal sentinel - exception if stop is already requested, so the body never - runs and the coroutine completes as stopped. - - @return An awaiter that suspends unconditionally. - */ - auto initial_suspend() noexcept - { - struct awaiter - { - promise_type* p_; - - bool await_ready() const noexcept - { - return false; - } - - void await_suspend(std::coroutine_handle<>) const noexcept - { - } - - // Potentially-throwing: checks the stop token before - // the coroutine body executes its first statement. - void await_resume() const - { - set_current_frame_allocator( - p_->environment()->frame_allocator); - if(p_->environment()->stop_token.stop_requested()) - throw detail::stop_requested_exception{}; - } - }; - return awaiter{this}; - } - - /** Return the final-suspend awaiter. - - The coroutine always suspends at the final suspend point. The - awaiter's `await_suspend` performs symmetric transfer to the - stored continuation, resuming the awaiting coroutine. - - @return An awaiter that suspends and transfers to the - continuation. - */ - auto final_suspend() noexcept - { - struct awaiter - { - promise_type* p_; - - bool await_ready() const noexcept - { - return false; - } - - std::coroutine_handle<> await_suspend( - std::coroutine_handle<>) const noexcept - { - return p_->continuation(); - } - - void await_resume() const noexcept {} // LCOV_EXCL_LINE final_suspend awaiter, never resumed - }; - return awaiter{this}; - } - - /** Capture the in-flight exception from the coroutine body. - - Called by the compiler when the coroutine body exits via an - unhandled exception. The internal stop sentinel is recorded as - a stopped completion; any other exception is recorded as an - exception completion. The stored exception is surfaced (or - routed to the error handler) when the quitter is awaited or run. - */ - void unhandled_exception() - { - try - { - throw; - } - catch(detail::stop_requested_exception const&) - { - // Store the exception_ptr so that run_async's - // invoke_impl routes to the error handler - // instead of accessing a non-existent result. - new (&ep_) std::exception_ptr( - std::current_exception()); - state_ = completion::stopped; - } - catch(...) - { - new (&ep_) std::exception_ptr( - std::current_exception()); - state_ = completion::exception; - } - } - - //------------------------------------------------------ - // transform_awaitable — the key difference from task - //------------------------------------------------------ - - /** Awaiter wrapping a nested `co_await` of an @ref ContextAwaitable. - - Forwards the environment to the inner awaitable's - environment-taking `await_suspend` and restores the - thread-local frame allocator before the body resumes. Unlike - `task`'s, it also checks the stop token. A stop pending before - the wrapped awaitable is touched, or on resumption after an - asynchronous completion, throws the internal sentinel. The - body then unwinds before it observes the operation's result. - A synchronous completion is always handed over through the - wrapped awaitable's `await_resume`. - - @tparam Awaitable The awaitable being transformed. - */ - template - struct transform_awaiter - { - /// The wrapped awaitable, decayed and stored by value. - std::decay_t a_; - - /// The promise of the coroutine performing the `co_await`. - promise_type* p_; - - /// True when a stop was pending at the `co_await`, so the - /// wrapped awaitable was never touched. - bool skipped_ = false; - - /// True when the wrapped awaitable completed synchronously, so - /// its `await_resume` must run to hand over what it acquired. - bool completed_ = false; - - /** Report whether the `co_await` can continue without - suspending. - - The stop token is checked first. If a stop is pending, the - wrapped awaitable is never touched: this returns `true`, and - @ref await_resume throws. Otherwise the wrapped awaitable's - own `await_ready` decides. A `true` result there means it - completed synchronously, so @ref await_resume calls its - `await_resume` even if a stop arrives afterwards. That hands - over anything it acquired, such as `async_mutex`'s lock. - - Where `BOOST_CAPY_SYMMETRIC_TRANSFER_WORKAROUND` is on, - this always returns `false`, and @ref await_suspend does both - steps instead. See `detail::await_ready_then_suspend`. - - @return `true` if no suspension is needed. Always `false` - where the workaround is on. - */ - bool await_ready() noexcept - { -#if BOOST_CAPY_SYMMETRIC_TRANSFER_WORKAROUND - return false; -#else - if(p_->environment()->stop_token.stop_requested()) - { - skipped_ = true; - return true; - } - completed_ = a_.await_ready(); - return completed_; -#endif - } - - /** Restore the frame allocator, check for stop, then resume the - wrapped awaitable. - - Reinstalls the thread-local frame allocator from the stored - environment. The internal sentinel exception is then thrown - in three cases: a stop was pending before the wrapped - awaitable was touched; a stop is pending now and the - wrapped awaitable completed asynchronously; or a stop is - pending now and the wrapped awaitable's result is a - @ref results holding `cond::canceled`. The body therefore - never observes that operation's result, and unwinds through - its RAII destructors to a stopped completion. A wrapped - awaitable that completed synchronously always has its - `await_resume` called, and any other result it gives is - returned, with the stop taking effect at the next - `co_await`. This is the one place `quitter` differs from - @ref task::promise_type::transform_awaiter. - - After an asynchronous completion with a stop pending, the - operation's result is dropped. Releasing anything it held is - then the wrapped awaiter's destructor's job: `async_mutex` - releases a lock it granted, and `async_waker` drops the - wakeup. The body is unwinding, so nothing observes the loss. - - @return The wrapped awaitable's await-result, forwarded - unchanged, when no sentinel is thrown. - - @par Exception Safety - Throws the library's internal stop sentinel in the three - cases above. The wrapped awaitable's `await_resume` is not - called in the first two; in the third, its canceled result - is dropped. - */ - decltype(auto) await_resume() - { - set_current_frame_allocator( - p_->environment()->frame_allocator); - if(skipped_ || - (!completed_ && - p_->environment()->stop_token.stop_requested())) - throw detail::stop_requested_exception{}; - using R = decltype(a_.await_resume()); - if constexpr (detail::is_results_v) - { - // A result canceled by our own stop acquired - // nothing, so the stop can take effect here. - R r = a_.await_resume(); - if(std::get<0>(r) == cond::canceled && - p_->environment()->stop_token.stop_requested()) - throw detail::stop_requested_exception{}; - return r; - } - else - { - return a_.await_resume(); - } - } - - /** Suspend by calling the wrapped awaitable with the - environment. - - This is the plain `await_suspend` the compiler calls for the - nested `co_await`. It forwards to the wrapped awaitable's - @ref ContextAwaitable overload, supplying the promise's stored - environment as the second argument. It keeps the wrapped - awaitable's suspension decision, whatever form it takes. - When that decision is to continue without suspending, the - wrapped awaitable completed synchronously, and - @ref await_resume will call its `await_resume`. Returning - `h` itself counts as continuing without suspending. - - @param h The coroutine performing the `co_await`. - - @return Whatever the wrapped awaitable's `await_suspend` - returns. When that is a `std::coroutine_handle<>`, the - handle is routed through `detail::symmetric_transfer`. - Where `BOOST_CAPY_SYMMETRIC_TRANSFER_WORKAROUND` is on, - this function instead returns `bool`. It first checks the - stop token, as @ref await_ready does where the workaround is - off, and returns `false` without touching the wrapped - awaitable if a stop is pending. Otherwise it returns the - result of `detail::await_ready_then_suspend`. Where the - workaround is off, a returned handle is passed back - unchanged for symmetric transfer. - */ - template - auto await_suspend( - std::coroutine_handle h) noexcept - { - // completed_ is written only where nothing else can have - // resumed this frame: a frame that really suspended may - // already be running, or gone, on another thread. -#if BOOST_CAPY_SYMMETRIC_TRANSFER_WORKAROUND - if(p_->environment()->stop_token.stop_requested()) - { - skipped_ = true; - return false; - } - return detail::await_ready_then_suspend( - a_, h, p_->environment(), &completed_); -#else - using R = decltype( - a_.await_suspend(h, p_->environment())); - if constexpr (std::is_same_v< - R, std::coroutine_handle<>>) - { - std::coroutine_handle<> next = - a_.await_suspend(h, p_->environment()); - // Returning h means no one else can resume this - // frame: a synchronous completion. -// GCC falsely warns that writing completed_ overruns the noop -// coroutine's frame: when the wrapped await_suspend returns -// std::noop_coroutine(), it assumes the branch below can be taken -// and treats this frame as the noop frame. A noop handle never -// equals h, so the branch is dead on that path. -#if defined(__GNUC__) && !defined(__clang__) -#pragma GCC diagnostic push -#pragma GCC diagnostic ignored "-Warray-bounds" -#endif - if(next == h) - completed_ = true; -#if defined(__GNUC__) && !defined(__clang__) -#pragma GCC diagnostic pop -#endif - return detail::symmetric_transfer(next); - } - else if constexpr (std::is_same_v) - { - if(!a_.await_suspend(h, p_->environment())) - { - completed_ = true; - return false; - } - return true; - } - else - { - return a_.await_suspend( - h, p_->environment()); - } -#endif - } - }; - - /** Transform a nested awaitable before `co_await`. - - Wraps an @ref ContextAwaitable in a @ref transform_awaiter so the - coroutine's environment is propagated into it and the stop - token is checked on resumption. A diagnostic is emitted if the - awaitable does not satisfy @ref ContextAwaitable. - - @param a The awaitable expression from `co_await a`. - - @return A @ref transform_awaiter wrapping `a`. - */ - template - auto transform_awaitable(Awaitable&& a) - { - using A = std::decay_t; - if constexpr (ContextAwaitable) - { - return transform_awaiter{ - std::forward(a), this}; - } - else - { - static_assert(sizeof(A) == 0, - "requires ContextAwaitable"); - } - } - }; - - /** Handle to the owned coroutine frame. - - Null when the quitter is empty (for example after a move or after - @ref release). Prefer @ref handle to read this; the member is - public for use by the coroutine machinery. - */ - std::coroutine_handle h_; - - /** Destroy the quitter and its coroutine frame if owned. - - @par Preconditions - The frame is not lent: no parent is suspended awaiting it, - unless its chain is being unwound after an @ref abandon. - Destroying a lent frame otherwise calls `std::terminate`; end - a running chain by requesting stop, or by destroying its - execution context. - */ - ~quitter() - { - destroy_frame(); - } - - /** Return false; quitters are never immediately ready. - - A quitter is lazy and has not started when it is awaited, so the - awaiting coroutine always suspends. - - @return `false`. - */ - bool await_ready() const noexcept - { - return false; - } - - /** Return the result, rethrow exception, or propagate stop. - - When stopped, throws stop_requested_exception so that a - parent quitter also stops. A parent task sees this - as an unhandled exception — by design. - - @return The result value for non-void `T`, moved out of the - quitter; otherwise `void`. - - @par Exception Safety - If the coroutine was stopped, the library's internal stop sentinel - is thrown. If the body exited via any other exception, that - exception is rethrown. - */ - auto await_resume() - { - lent_ = false; - if(h_.promise().stopped()) - throw detail::stop_requested_exception{}; - if(h_.promise().state_ == promise_type::completion::exception) - std::rethrow_exception(h_.promise().ep_); - if constexpr (! std::is_void_v) - return std::move(*h_.promise().result_); - else - return; - } - - /** Start execution with the caller's context. - - Stores `cont` as the continuation to resume on completion. - Stores `env` as the execution environment propagated to nested - `co_await` expressions. Then transfers control into the quitter's - coroutine body via the returned handle. - - @param cont The awaiting coroutine to resume when the quitter - completes. - - @param env The execution environment (executor, stop token, and - frame allocator). It must outlive the quitter. - - @return The quitter's coroutine handle, for symmetric transfer. - */ - std::coroutine_handle<> await_suspend( - std::coroutine_handle<> cont, - await_context const* env) - { - h_.promise().set_continuation(cont); - h_.promise().set_environment(env); - lent_ = true; - return h_; - } - - /** Return the coroutine handle. - - @note Do not call `destroy()` on the returned handle while the - quitter is being awaited: the awaiting frame still holds it and - destroys it again, which has undefined behavior. A holder that - gives the coroutine up passes its continuation to @ref abandon - instead. (Destroying or - move-assigning over the quitter object itself while its frame is - lent is a different operation, and calls `std::terminate`.) - The quitter's lifetime is normally managed by `run_async`, `run`, - or the awaiting parent. For cooperative cancellation, use - `std::stop_token`. - - @note The operation, queue or primitive holding a suspended - coroutine gives it up by passing the continuation it keeps to - @ref abandon. `handle()` belongs to the owning quitter object. - - @return The coroutine handle. - */ - std::coroutine_handle handle() const noexcept - { - return h_; - } - - /** Release ownership of the coroutine frame. - - @note The caller may call `destroy()` on the released handle - only when the quitter has not started or has fully completed. - Destroying the released handle of a quitter that is being - awaited is not supported and has undefined behavior. - - @par Postconditions - `handle()` returns a null handle. Callers needing the - original handle must save it, via @ref handle, before - calling this. - */ - void release() noexcept - { - h_ = nullptr; - } - - /** Copy construction is disabled; a quitter uniquely owns its frame. - - @param other The quitter that would be copied. - */ - quitter(quitter const& other) = delete; - - /** Copy assignment is disabled; a quitter uniquely owns its frame. - - @param other The quitter that would be assigned from. - - @return A reference to `*this`. - */ - quitter& operator=(quitter const& other) = delete; - - /** Construct by moving, transferring ownership. - - @par Postconditions - `other` is empty and must not be awaited. - - @param other The quitter to move from. - */ - quitter(quitter&& other) noexcept - : h_(std::exchange(other.h_, nullptr)) - { - } - - /** Assign by moving, transferring ownership. - - If this quitter already owns a coroutine frame, that frame is - destroyed first. Self-assignment is a no-op. - - @par Preconditions - The frame being replaced is not lent: no parent is suspended - awaiting it. Replacing a lent frame calls `std::terminate`. - - @par Postconditions - `other` is empty and must not be awaited. - - @param other The quitter to move from. - - @return A reference to `*this`. - */ - quitter& operator=(quitter&& other) noexcept - { - if(this != &other) - { - destroy_frame(); - h_ = std::exchange(other.h_, nullptr); - } - return *this; - } - -private: - // Set while a parent is suspended awaiting this quitter. - bool lent_ = false; - - explicit quitter(std::coroutine_handle h) - : h_(h) - { - } - - void destroy_frame() noexcept - { - if(!h_) - return; - if(lent_ && !h_.done()) - { - // Only the unwind of an abandoned chain destroys a frame - // that is still awaited. - auto* owner = h_.promise().environment()->owner; - if(!owner || !owner->abandoned()) - std::terminate(); // LCOV_EXCL_LINE reached only in a death test's child process - } - std::exchange(h_, nullptr).destroy(); - lent_ = false; - } -}; - -} // namespace capy -} // namespace boost - -#endif diff --git a/include/boost/capy/task.hpp b/include/boost/capy/task.hpp index 57d0fedef..91b01e9a2 100644 --- a/include/boost/capy/task.hpp +++ b/include/boost/capy/task.hpp @@ -85,8 +85,7 @@ struct task_return_base `task` never inspects the stop token; it only propagates it. A task body observes a stop request through the results of the operations it - awaits, or by reading the token itself. See @ref quitter for a task - that stops its own body. + awaits, or by reading the token itself. @par Await-returns The value the body passed to `co_return`, moved out of the task, or diff --git a/test/unit/quitter.cpp b/test/unit/quitter.cpp deleted file mode 100644 index 7dd9447bd..000000000 --- a/test/unit/quitter.cpp +++ /dev/null @@ -1,1496 +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 -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include "death_test.hpp" -#include "test_helpers.hpp" - -#include - -#include -#include -#include -#include -#include -#include -#include -#include - -// GCC gives false positive -Wmaybe-uninitialized on structured bindings -// via the tuple protocol inside coroutine frames. -#if defined(__GNUC__) && !defined(__clang__) -#pragma GCC diagnostic ignored "-Wmaybe-uninitialized" -#endif - -namespace boost { -namespace capy { - -static_assert(ContextAwaitable>); -static_assert(ContextAwaitable>); -static_assert(ContextRunnable>); -static_assert(ContextRunnable>); - -struct quitter_test -{ - // Assigning over a quitter that is not awaited destroys the frame - // it held. - void testMoveAssignDestroysHeldFrame() - { - int a_down = 0, b_down = 0; - auto body = [](destroy_count_t) -> quitter { co_return; }; - { - auto t = body(destroy_count_t(a_down)); - t = body(destroy_count_t(b_down)); - BOOST_TEST_EQ(a_down, 1); - BOOST_TEST_EQ(b_down, 0); - } - BOOST_TEST_EQ(b_down, 1); - } - -#if !defined(_WIN32) - void testDestroyLentTaskTerminates() - { - BOOST_TEST(terminates([] { - parked slot; - int n = 0; - test_executor ex(n); - await_context env{executor_ref(ex), {}, nullptr}; - auto t = [](parked* s) -> quitter { - co_await park_awaitable{s}; - }(&slot); - lend(t, env); - // destructor runs with the frame lent and suspended - })); - } - - void testMoveAssignOverLentTaskTerminates() - { - BOOST_TEST(terminates([] { - parked slot; - int n = 0; - test_executor ex(n); - await_context env{executor_ref(ex), {}, nullptr}; - auto body = [](parked* s) -> quitter { - co_await park_awaitable{s}; - }; - auto t = body(&slot); - lend(t, env); - t = body(&slot); - std::_Exit(0); // must not be reached - })); - } - - void testDestroyUnstartedOrFinishedTaskIsFine() - { - BOOST_TEST(exits_cleanly([] { - auto body = []() -> quitter { co_return; }; - { auto t = body(); } // never started - int n = 0; - test_executor ex(n); - await_context env{executor_ref(ex), {}, nullptr}; - auto t = body(); - lend(t, env); // runs to final_suspend - return t.handle().done(); - })); // finished frame destroyed - } -#endif - - // A quitter in the middle of a chain comes down with it when the - // leaf below it is torn down. - void testTeardownThroughQuitter() - { - alloc_counts st; - int leaf_down = 0, mid_down = 0, root_down = 0; - parked slot; - { - auto leaf = [](parked* s, destroy_count_t) - -> task - { - co_await park_awaitable{s}; - }; - auto mid = [&](destroy_count_t) -> quitter - { - co_await leaf(&slot, destroy_count_t(leaf_down)); - }; - auto root = [&](destroy_count_t) -> task - { - co_await mid(destroy_count_t(mid_down)); - }; - int n = 0; - test_executor ex(n); - run_async(ex, counting_allocator(st))( - root(destroy_count_t(root_down))); - BOOST_TEST(slot); - capy::abandon(slot); - } - BOOST_TEST_EQ(leaf_down, 1); - BOOST_TEST_EQ(mid_down, 1); - BOOST_TEST_EQ(root_down, 1); - BOOST_TEST(st.balanced()); - } - - //---------------------------------------------------------- - // 1. Normal completion — quitter returns a value - //---------------------------------------------------------- - - static quitter - returns_int() - { - co_return 42; - } - - void - testNormalCompletion() - { - int result = 0; - test::run_blocking([&](int v) { result = v; })(returns_int()); - BOOST_TEST_EQ(result, 42); - } - - //---------------------------------------------------------- - // 2. Void completion - //---------------------------------------------------------- - - static quitter<> - void_quitter() - { - co_return; - } - - void - testVoidCompletion() - { - test::run_blocking()(void_quitter()); - } - - //---------------------------------------------------------- - // 3. Exception propagation - //---------------------------------------------------------- - - static quitter<> - throws_quitter() - { - throw test_exception("quitter exception"); - co_return; - } - - void - testExceptionPropagation() - { - BOOST_TEST_THROWS( - test::run_blocking()(throws_quitter()), - test_exception); - } - - //---------------------------------------------------------- - // 4. Stop before first co_await - //---------------------------------------------------------- - - struct raii_counter - { - int* count; - raii_counter(int& c) : count(&c) {} - raii_counter(raii_counter&& o) noexcept - : count(std::exchange(o.count, nullptr)) {} - ~raii_counter() { if(count) ++(*count); } - }; - - static quitter<> - quitter_with_raii(int& dtor_count) - { - raii_counter guard(dtor_count); - co_await stop_only_awaitable{}; - } - - void - testStopBeforeFirstAwait() - { - // When stop is already requested, initial_suspend throws - // before the coroutine body starts. Body locals are never - // constructed, so dtor_count stays 0. The stopped state - // routes to the error handler via exception(). - int dispatch_count = 0; - test_executor ex(dispatch_count); - std::stop_source source; - source.request_stop(); - - int dtor_count = 0; - bool got_stopped = false; - - run_async(ex, source.get_token(), - [](){ BOOST_TEST(false); }, - [&](std::exception_ptr ep) { - got_stopped = (ep != nullptr); - })(quitter_with_raii(dtor_count)); - - BOOST_TEST(got_stopped); - // Body never started, so no destructors ran - BOOST_TEST_EQ(dtor_count, 0); - } - - //---------------------------------------------------------- - // 5. Stop during I/O - //---------------------------------------------------------- - - // stop_only_awaitable as the tests below need it. - // - // A stop that stop_only_awaitable sees in await_suspend makes it - // return the awaiting coroutine: a synchronous completion, whose - // result quitter inspects. So await_resume reports that stop as - // canceled, as capy's own awaitables do, and quitter unwinds the - // body there, just as for a stop that arrives while suspended. - // - // `registered`, when given, is counted down once the awaitable has - // registered for the stop, so a stop the test requests after - // waiting on it always finds the frame suspended. - struct parked_until_stop : stop_only_awaitable - { - std::latch* registered; - bool canceled = false; - - explicit parked_until_stop( - std::latch* r = nullptr) noexcept - : registered(r) - { - } - - std::coroutine_handle<> - await_suspend( - std::coroutine_handle<> h, await_context const* env) - { - // Copied out: once the latch lets the stop in, this - // frame may be resumed and destroyed before - // await_suspend returns. - std::latch* r = registered; - auto next = stop_only_awaitable::await_suspend(h, env); - if(next == h) - canceled = true; - if(r) - r->count_down(); - return next; - } - - results<> await_resume() noexcept - { - if(canceled) - return {make_error_code(error::canceled)}; - return {{}}; - } - }; - - void - testStopDuringIO() - { - std::atomic state = 0; - std::latch suspended(1); - - int dtor_count = 0; - - auto q = [&]() -> quitter<> - { - raii_counter guard(dtor_count); - state = 1; - co_await parked_until_stop{&suspended}; - // Should never reach here when stopped - state = 99; - }; - - { - std::jthread jt( - [&](std::stop_token st) - { - int dc = 0; - test_executor ex(dc); - run_async(ex, st)(q()); - } - ); - suspended.wait(); - BOOST_TEST(state == 1); - // jthread destructor calls request_stop() then join() - } - - // The coroutine was stopped; it should NOT have reached state=99 - BOOST_TEST(state != 99); - BOOST_TEST_EQ(dtor_count, 1); - } - - //---------------------------------------------------------- - // 6. Stop propagation through chain - //---------------------------------------------------------- - - static quitter<> - inner_quitter( - int& dtor_count, - bool& reached_end, - std::latch& suspended) - { - raii_counter guard(dtor_count); - co_await parked_until_stop{&suspended}; - reached_end = true; - } - - static quitter<> - middle_quitter( - int& dtor_count, - bool& reached_end, - std::latch& suspended) - { - raii_counter guard(dtor_count); - co_await inner_quitter(dtor_count, reached_end, suspended); - reached_end = true; - } - - static quitter<> - outer_quitter( - int& dtor_count, - bool& reached_end, - std::latch& suspended) - { - raii_counter guard(dtor_count); - co_await middle_quitter(dtor_count, reached_end, suspended); - reached_end = true; - } - - void - testStopPropagationChain() - { - std::latch suspended(1); - int dtor_count = 0; - bool reached_end = false; - - auto top = [&]() -> quitter<> - { - co_await outer_quitter( - dtor_count, reached_end, suspended); - reached_end = true; - }; - - { - std::jthread jt( - [&](std::stop_token st) - { - int dc = 0; - test_executor ex(dc); - run_async(ex, st)(top()); - } - ); - suspended.wait(); - } - - // 3 guards from outer/middle/inner - BOOST_TEST_EQ(dtor_count, 3); - // No coroutine body continued past its co_await - BOOST_TEST(!reached_end); - } - - //---------------------------------------------------------- - // 9. Mixing quitter and task — task awaits quitter - //---------------------------------------------------------- - - static quitter - quitter_returns_42() - { - co_return 42; - } - - static task - task_awaits_quitter() - { - int v = co_await quitter_returns_42(); - co_return v + 1; - } - - void - testMixingQuitterAndTask() - { - // Normal case: task awaits quitter that completes normally - { - int result = 0; - test::run_blocking( - [&](int v) { result = v; })(task_awaits_quitter()); - BOOST_TEST_EQ(result, 43); - } - - // Stopped case: task awaits stopped quitter — sees exception - { - std::stop_source source; - source.request_stop(); - - int dispatch_count = 0; - test_executor ex(dispatch_count); - bool got_exception = false; - - auto t = []() -> task - { - co_return co_await quitter_returns_42(); - }; - - run_async(ex, source.get_token(), - [](int) { BOOST_TEST(false); }, - [&](std::exception_ptr ep) { - got_exception = (ep != nullptr); - })(t()); - - BOOST_TEST(got_exception); - } - } - - //---------------------------------------------------------- - // 10. No stop requested — identical to task - //---------------------------------------------------------- - - static quitter - quitter_chain() - { - auto inner = []() -> quitter { - co_return 10; - }; - - auto middle = [inner]() -> quitter { - int v = co_await inner(); - co_return v * 2; - }; - - int v = co_await middle(); - co_return v + 5; - } - - void - testNoStopRequested() - { - int result = 0; - test::run_blocking( - [&](int v) { result = v; })(quitter_chain()); - BOOST_TEST_EQ(result, 25); - } - - //---------------------------------------------------------- - // 11. RAII verification - //---------------------------------------------------------- - - void - testRAIIVerification() - { - // Body starts, constructs guards, then stop is requested - // during the co_await. All guard destructors must run. - // The body must NOT continue past co_await. - std::latch suspended(1); - int dtor_count = 0; - bool reached_end = false; - - auto q = [&]() -> quitter<> - { - raii_counter g1(dtor_count); - raii_counter g2(dtor_count); - raii_counter g3(dtor_count); - co_await parked_until_stop{&suspended}; - reached_end = true; - }; - - { - std::jthread jt( - [&](std::stop_token st) - { - int dc = 0; - test_executor ex(dc); - run_async(ex, st)(q()); - } - ); - suspended.wait(); - } - - BOOST_TEST_EQ(dtor_count, 3); - BOOST_TEST(!reached_end); - } - - //---------------------------------------------------------- - // 12. Multiple co_await — stop after second - //---------------------------------------------------------- - - static quitter - quitter_multi_await( - std::atomic& progress, - std::latch& sem) - { - progress = 1; - co_await yield_awaitable{}; - progress = 2; - co_await parked_until_stop{&sem}; - // Should not reach here - progress = 3; - co_return 0; - } - - void - testMultipleCoAwait() - { - std::atomic progress{0}; - std::latch sem(1); - - { - std::jthread jt( - [&](std::stop_token st) - { - int dc = 0; - test_executor ex(dc); - run_async(ex, st)( - quitter_multi_await(progress, sem)); - } - ); - sem.wait(); - BOOST_TEST(progress == 2); - // jthread destructor requests stop - } - - // The third await should have been short-circuited - BOOST_TEST(progress != 3); - } - - //---------------------------------------------------------- - // Move operations - //---------------------------------------------------------- - - void - testMoveOperations() - { - int dispatch_count = 0; - test_executor ex(dispatch_count); - await_context env{executor_ref(ex), {}, nullptr}; - - // move constructor - { - auto q1 = returns_int(); - auto h1 = q1.handle(); - q1.release(); - BOOST_TEST(h1); - - quitter q2(std::move(q1)); - BOOST_TEST(!q2.handle()); - - h1.promise().set_environment(&env); - while(!h1.done()) - h1.resume(); - BOOST_TEST_EQ(*h1.promise().result_, 42); - h1.destroy(); - } - - // release() - { - auto q = returns_int(); - auto h = q.handle(); - q.release(); - BOOST_TEST(h); - BOOST_TEST(!q.handle()); - - h.promise().set_environment(&env); - while(!h.done()) - h.resume(); - BOOST_TEST(h.promise().result_.has_value()); - BOOST_TEST_EQ(*h.promise().result_, 42); - h.destroy(); - } - } - - //---------------------------------------------------------- - // Quitter returning string - //---------------------------------------------------------- - - static quitter - returns_string() - { - co_return "hello"; - } - - void - testReturnString() - { - std::string result; - test::run_blocking( - [&](std::string v) { result = std::move(v); })( - returns_string()); - BOOST_TEST_EQ(result, "hello"); - } - - //---------------------------------------------------------- - // Exception in quitter - //---------------------------------------------------------- - - static quitter - quitter_throws_int() - { - throw test_exception("quitter int exception"); - co_return 0; - } - - void - testExceptionInValueQuitter() - { - BOOST_TEST_THROWS( - test::run_blocking()(quitter_throws_int()), - test_exception); - } - - // A ContextAwaitable whose await_suspend returns bool, to drive the - // non-handle branch of the quitter transform_awaiter. - struct bool_resume_awaitable - { - bool await_ready() const noexcept { return false; } - bool await_suspend(std::coroutine_handle<>, await_context const*) noexcept - { - return false; // do not suspend - } - int await_resume() noexcept { return 7; } - }; - - static quitter - co_awaits_bool_suspend() - { - co_return co_await bool_resume_awaitable{}; - } - - void - testBoolSuspendAwaitable() - { - int result = 0; - test::run_blocking([&](int v) { result = v; })( - co_awaits_bool_suspend()); - BOOST_TEST_EQ(result, 7); - } - - // A ContextAwaitable whose bool await_suspend really suspends, - // parking the awaiting coroutine for the test to resume. - struct bool_park_awaitable - { - std::coroutine_handle<>& parked; - bool& resumed; - - bool await_ready() const noexcept { return false; } - bool await_suspend( - std::coroutine_handle<> h, await_context const*) noexcept - { - parked = h; - return true; - } - void await_resume() noexcept { resumed = true; } - }; - - static quitter<> - co_awaits_bool_park( - std::coroutine_handle<>& parked, bool& resumed) - { - co_await bool_park_awaitable{parked, resumed}; - } - - // A stop that arrives while the wrapped awaitable is suspended - // takes effect at the resumption, as for a handle await_suspend. - void - testBoolSuspendThenStop() - { - int dispatch_count = 0; - test_executor ex(dispatch_count); - - std::stop_source source; - std::coroutine_handle<> parked; - bool resumed = false; - bool completed = false; - int stopped = 0; - - run_async(ex, source.get_token(), - [&]() { completed = true; }, - [&](std::exception_ptr) { ++stopped; })( - co_awaits_bool_park(parked, resumed)); - - BOOST_TEST(parked); - source.request_stop(); - parked.resume(); - - BOOST_TEST(!resumed); - BOOST_TEST(!completed); - BOOST_TEST_EQ(stopped, 1); - } - - static quitter<> - plain_void_quitter() - { - co_return; - } - - static quitter - co_awaits_void_quitter() - { - co_await plain_void_quitter(); - co_return 5; - } - - void - testCoAwaitVoidQuitter() - { - // Co-await a normally-completing void quitter: exercises the - // void path of await_resume. - int result = 0; - test::run_blocking([&](int v) { result = v; })( - co_awaits_void_quitter()); - BOOST_TEST_EQ(result, 5); - } - - static quitter - co_awaits_throwing_quitter() - { - co_await quitter_throws_int(); - co_return 0; - } - - void - testCoAwaitThrowingQuitter() - { - // Co-await a throwing quitter: exercises the rethrow path of - // await_resume. - BOOST_TEST_THROWS( - test::run_blocking()(co_awaits_throwing_quitter()), - test_exception); - } - - //---------------------------------------------------------- - // 7. Stop propagation with when_all - // - // Two quitter> children inside when_all. - // Both block on parked_until_stop. when_all creates a - // child stop_source; when the parent stop fires, when_all - // propagates it. Each quitter child intercepts the stop - // in transform_awaiter::await_resume and short-circuits - // via stop_requested_exception — it never reaches the - // co_return. Verify both stop and when_all completes. - //---------------------------------------------------------- - - static quitter> - quitter_pending_size( - bool& reached_co_return, - std::latch* registered = nullptr) - { - co_await parked_until_stop{registered}; - // If quitter's transform_awaiter intercepted the stop, - // we never reach here. - reached_co_return = true; - co_return results{ - make_error_code(error::canceled), 0}; - } - - void - testWhenAllWithStop() - { - thread_pool pool(2); - std::latch done(1); - std::latch registered(2); - std::stop_source source; - - bool child1_returned = false; - bool child2_returned = false; - - auto outer = [&]() -> task<> - { - auto result = co_await when_all( - quitter_pending_size(child1_returned, ®istered), - quitter_pending_size(child2_returned, ®istered)); - (void)result; - }; - - run_async(pool.get_executor(), source.get_token(), - [&]() { done.count_down(); }, - [&](std::exception_ptr) { done.count_down(); })( - outer()); - - registered.wait(); - source.request_stop(); - - done.wait(); - // Both quitter children were stopped by transform_awaiter - // before reaching co_return. - BOOST_TEST(!child1_returned); - BOOST_TEST(!child2_returned); - } - - //---------------------------------------------------------- - // 8. Stop propagation with when_any - // - // Two quitter children. One succeeds immediately, - // when_any stops the sibling. The sibling quitter - // intercepts the stop and exits cleanly. - //---------------------------------------------------------- - - static quitter> - quitter_success_size(std::size_t n) - { - co_return results{std::error_code(), n}; - } - - void - testWhenAnyWithStop() - { - // One child succeeds immediately. when_any stops - // the pending sibling quitter. The sibling must - // be intercepted by transform_awaiter (never reach - // co_return). - { - thread_pool pool(2); - std::latch done(1); - bool sibling_returned = false; - - auto outer = [&]() -> task<> - { - auto result = co_await when_any( - quitter_success_size(42), - quitter_pending_size(sibling_returned)); - // Variadic when_any returns - // variant. - // Index 1 = first child won. - BOOST_TEST(result.index() == 1); - if(result.index() == 1) - BOOST_TEST_EQ(std::get<1>(result), - std::size_t(42)); - }; - - run_async(pool.get_executor(), - [&]() { done.count_down(); }, - [&](std::exception_ptr) { - done.count_down(); - })(outer()); - - done.wait(); - BOOST_TEST(!sibling_returned); - } - - // Both children pending. Parent stop fires. - // when_any propagates stop to children. Both - // quitter children short-circuit. - { - thread_pool pool(2); - std::latch done(1); - std::latch registered(2); - std::stop_source source; - - bool child1_returned = false; - bool child2_returned = false; - - auto outer = [&]() -> task<> - { - auto result = co_await when_any( - quitter_pending_size(child1_returned, ®istered), - quitter_pending_size(child2_returned, ®istered)); - (void)result; - }; - - run_async(pool.get_executor(), source.get_token(), - [&]() { done.count_down(); }, - [&](std::exception_ptr) { - done.count_down(); - })(outer()); - - registered.wait(); - source.request_stop(); - - done.wait(); - BOOST_TEST(!child1_returned); - BOOST_TEST(!child2_returned); - } - } - - //---------------------------------------------------------- - // 14. Waker cancellation - //---------------------------------------------------------- - - void - testWakerCancellation() - { - 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 waker.wait(); - (void)ec; - reached_end = true; - }; - - auto start = std::chrono::steady_clock::now(); - - run_async(pool.get_executor(), source.get_token(), - [&]() { done.count_down(); }, - [&](std::exception_ptr) { done.count_down(); })(q()); - - suspended.wait(); - std::this_thread::sleep_for( - std::chrono::milliseconds(10 * failsafe_scale)); - source.request_stop(); - - done.wait(); - auto elapsed = std::chrono::steady_clock::now() - start; - // Should complete promptly - BOOST_TEST(elapsed < std::chrono::seconds(1)); - // Quitter intercepted the stop — body did not continue - BOOST_TEST(!reached_end); - } - - //---------------------------------------------------------- - // 15. Catch around a co_await whose await_ready does an atomic - // read-modify-write (MSVC 14.51 ARM64 release regression) - // - // quitter's transform_awaiter gets the same safeguard as - // task's: where the symmetric-transfer workaround is on, the - // wrapped await_ready is asked from inside await_suspend. See - // testCatchAfterRmwAwaitReady in test/unit/task.cpp. Both - // tests below pass on every other build with or without it. - //---------------------------------------------------------- - - // An opaque call before the try; the defect needs one there. - static BOOST_CAPY_NOINLINE void - call_before_try(bool& marker) - { - marker = true; - } - - // Parks; await_ready performs a compare-exchange that never - // succeeds here, as async_waker's does. await_resume throws when - // throw_on_resume is set. - struct rmw_ready_awaitable - { - std::coroutine_handle<>& slot; - std::atomic& state; - bool throw_on_resume; - - bool await_ready() noexcept - { - int expected = 1; - return state.compare_exchange_strong( - expected, 0, - std::memory_order_acq_rel, - std::memory_order_acquire); - } - - std::coroutine_handle<> - await_suspend( - std::coroutine_handle<> h, await_context const*) noexcept - { - slot = h; - return std::noop_coroutine(); - } - - void await_resume() const - { - if(throw_on_resume) - throw test_exception("rmw ready"); - } - }; - - static quitter<> - rmw_ready_quitter( - std::coroutine_handle<>& slot, - std::atomic& state, - bool throw_on_resume, - bool& marker, - bool& caught) - { - call_before_try(marker); - try - { - co_await rmw_ready_awaitable{slot, state, throw_on_resume}; - } - catch(...) - { - caught = true; - } - } - - // The awaitable's own await_resume throws. - void - testCatchAfterRmwAwaitReady() - { - int dispatch_count = 0; - test_executor ex(dispatch_count); - - std::coroutine_handle<> parked; - std::atomic state{0}; - bool marker = false; - bool caught = false; - bool escaped = false; - bool completed = false; - - run_async(ex, - [&]() { completed = true; }, - [&](std::exception_ptr) { escaped = true; })( - rmw_ready_quitter(parked, state, true, marker, caught)); - - BOOST_TEST(static_cast(parked)); - BOOST_TEST(marker); - if(parked) - parked.resume(); - - BOOST_TEST(caught); - BOOST_TEST(!escaped); - BOOST_TEST(completed); - } - - // A stop request while parked: quitter's await_resume throws its - // stop sentinel, which the catch(...) in the body takes, as in - // the documentation's catch(...) pitfall. - void - testCatchStopAfterRmwAwaitReady() - { - int dispatch_count = 0; - test_executor ex(dispatch_count); - - std::stop_source source; - std::coroutine_handle<> parked; - std::atomic state{0}; - bool marker = false; - bool caught = false; - bool escaped = false; - bool completed = false; - - run_async(ex, source.get_token(), - [&]() { completed = true; }, - [&](std::exception_ptr) { escaped = true; })( - rmw_ready_quitter(parked, state, false, marker, caught)); - - BOOST_TEST(static_cast(parked)); - BOOST_TEST(marker); - source.request_stop(); - if(parked) - parked.resume(); - - BOOST_TEST(caught); - BOOST_TEST(!escaped); - BOOST_TEST(completed); - } - - //---------------------------------------------------------- - // 13. Echo server with shutdown - // - // A quitter echo loop over a mock stream pair. - // The client exchanges data, then requests stop. - // The echo quitter exits cleanly via stop interception, - // RAII runs, and the echoed data was correct. - // - // The mock stream's read_some is not stop-aware, so we - // wrap each read in when_any with a stop_only_awaitable - // to make it cancellable — mirroring how a real server - // would have the OS cancel an in-flight read. - //---------------------------------------------------------- - - void - testEchoWithShutdown() - { - // Echo server over a mock stream pair. The server - // reads pre-provided data, echoes it back, then - // waits for shutdown via stop_only_awaitable. When - // stop fires, the quitter intercepts at the co_await - // and exits cleanly. All stream access is on the - // jthread's synchronous executor — no cross-thread - // stream use. - test::fuse f; - auto [server_end, client_end] = - test::make_stream_pair(f); - - client_end.provide("hello"); - - int dtor_count = 0; - bool reached_end = false; - std::size_t total_echoed = 0; - std::latch suspended(1); - - auto echo_server = [&]() -> quitter<> - { - raii_counter guard(dtor_count); - char buf[64]; - - // Echo loop: process all available data - auto [ec, n] = co_await server_end.read_some( - make_buffer(buf)); - if(ec) - co_return; - total_echoed += n; - auto [ec2, n2] = co_await server_end.write_some( - make_buffer(buf, n)); - if(ec2) - co_return; - - // Wait for shutdown, signalling once suspended - // until the stop token fires. - co_await parked_until_stop{&suspended}; - - // Should never reach here — quitter intercepts - reached_end = true; - }; - - { - std::jthread jt( - [&](std::stop_token st) - { - int dc = 0; - test_executor ex(dc); - run_async(ex, st)(echo_server()); - } - ); - suspended.wait(); - - // Verify the echo happened - BOOST_TEST_EQ(total_echoed, std::size_t(5)); - - // Read back the echoed data from client side - // (synchronous — data is already in the buffer) - // Not possible here since we're on the main - // thread and streams are single-threaded. The - // echo write went to client_end's read buffer - // which we can't access cross-thread. The - // echo itself is verified by total_echoed. - - // jthread destructor requests stop and joins - } - - BOOST_TEST_EQ(dtor_count, 1); - BOOST_TEST(!reached_end); - } - - //---------------------------------------------------------- - // 16. A stop never discards what the inner await_ready acquired - // - // async_mutex's await_ready takes the lock when the mutex is - // free. A stop pending at the co_await must leave the inner - // awaitable untouched, and once the inner awaitable reports - // ready, its await_resume must run so that what it acquired is - // handed over. Only a result canceled by the stop, which - // acquired nothing, unwinds the body there. - //---------------------------------------------------------- - - static quitter<> - lock_after_stop(async_mutex& mtx, std::stop_source& source) - { - source.request_stop(); - auto [ec, guard] = co_await mtx.scoped_lock(); - (void)ec; - } - - // The stop is already pending when the body reaches the co_await. - void - testStopPendingLeavesMutexUntouched() - { - int dispatch_count = 0; - test_executor ex(dispatch_count); - - async_mutex mtx; - std::stop_source source; - bool completed = false; - int stopped = 0; - - run_async(ex, source.get_token(), - [&]() { completed = true; }, - [&](std::exception_ptr) { ++stopped; })( - lock_after_stop(mtx, source)); - - BOOST_TEST(!mtx.is_locked()); - BOOST_TEST(!completed); - BOOST_TEST_EQ(stopped, 1); - } - - // Reports ready, and requests the stop from inside await_ready, so - // the stop arrives after the inner awaitable has completed. - struct stop_while_ready_awaitable - { - std::stop_source& source; - bool& resumed; - - bool await_ready() noexcept - { - source.request_stop(); - return true; - } - - std::coroutine_handle<> - await_suspend( - std::coroutine_handle<> h, await_context const*) noexcept - { - return h; // LCOV_EXCL_LINE never called: await_ready is true - } - - void await_resume() noexcept - { - resumed = true; - } - }; - - static quitter<> - ready_then_stop(std::stop_source& source, bool& resumed) - { - co_await stop_while_ready_awaitable{source, resumed}; - } - - void - testReadyThenStopRunsAwaitResume() - { - int dispatch_count = 0; - test_executor ex(dispatch_count); - - std::stop_source source; - bool resumed = false; - bool completed = false; - int stopped = 0; - - run_async(ex, source.get_token(), - [&]() { completed = true; }, - [&](std::exception_ptr) { ++stopped; })( - ready_then_stop(source, resumed)); - - BOOST_TEST(resumed); - // Nothing suspends after the co_await, so the body finishes. - BOOST_TEST(completed); - BOOST_TEST_EQ(stopped, 0); - } - - // Not ready, and await_suspend returns the awaiting coroutine - // itself, after requesting the stop: a synchronous completion - // through the returned handle, as async_waker's await_suspend does - // when it consumes a wakeup that was latched after its await_ready. - struct stop_then_return_self_awaitable - { - std::stop_source& source; - bool& resumed; - - bool await_ready() const noexcept - { - return false; - } - - std::coroutine_handle<> - await_suspend( - std::coroutine_handle<> h, await_context const*) noexcept - { - source.request_stop(); - return h; - } - - void await_resume() noexcept - { - resumed = true; - } - }; - - static quitter<> - return_self_then_stop(std::stop_source& source, bool& resumed) - { - co_await stop_then_return_self_awaitable{source, resumed}; - } - - void - testReturnSelfThenStopRunsAwaitResume() - { - int dispatch_count = 0; - test_executor ex(dispatch_count); - - std::stop_source source; - bool resumed = false; - bool completed = false; - int stopped = 0; - - run_async(ex, source.get_token(), - [&]() { completed = true; }, - [&](std::exception_ptr) { ++stopped; })( - return_self_then_stop(source, resumed)); - - BOOST_TEST(resumed); - // Nothing suspends after the co_await, so the body finishes. - BOOST_TEST(completed); - BOOST_TEST_EQ(stopped, 0); - } - - // Reports ready, after requesting the stop when given a source, - // with a results holding `ec`: a synchronous completion like - // async_mutex's or async_event's when they see the stop first. - // The payload counts its destruction, as a lock_guard releases. - struct ready_results_awaitable - { - std::stop_source* source; - std::error_code ec; - bool& resumed; - int& destroyed; - - bool await_ready() noexcept - { - if(source) - source->request_stop(); - return true; - } - - std::coroutine_handle<> - await_suspend( - std::coroutine_handle<> h, await_context const*) noexcept - { - return h; // LCOV_EXCL_LINE never called: await_ready is true - } - - results await_resume() noexcept - { - resumed = true; - return {ec, raii_counter(destroyed)}; - } - }; - - struct ready_results_outcome - { - bool resumed = false; - bool continued = false; - std::error_code seen; - int stopped = 0; - int destroyed = 0; - }; - - static quitter<> - ready_results_body( - std::stop_source* source, - std::error_code ec, - ready_results_outcome& out) - { - auto [seen, n] = co_await ready_results_awaitable{ - source, ec, out.resumed, out.destroyed}; - (void)n; - out.seen = seen; - out.continued = true; - } - - static ready_results_outcome - run_ready_results(bool stop, std::error_code ec) - { - int dispatch_count = 0; - test_executor ex(dispatch_count); - - std::stop_source source; - ready_results_outcome out; - - run_async(ex, source.get_token(), - []() {}, - [&](std::exception_ptr) { ++out.stopped; })( - ready_results_body( - stop ? &source : nullptr, ec, out)); - return out; - } - - // A canceled result under our stop acquired nothing, so the body - // is unwound at that co_await. - void - testReadyCanceledThenStopStops() - { - auto out = run_ready_results( - true, make_error_code(error::canceled)); - BOOST_TEST(out.resumed); - BOOST_TEST(!out.continued); - BOOST_TEST_EQ(out.stopped, 1); - // The dropped result is destroyed, not leaked. - BOOST_TEST_EQ(out.destroyed, 1); - } - - // A successful result is handed over even with a stop pending. - void - testReadySuccessThenStopDelivers() - { - auto out = run_ready_results(true, {}); - BOOST_TEST(out.continued); - BOOST_TEST(!out.seen); - BOOST_TEST_EQ(out.stopped, 0); - BOOST_TEST_EQ(out.destroyed, 1); - } - - // A canceled result with no stop of ours is the body's to handle. - void - testReadyCanceledWithoutStopDelivers() - { - auto out = run_ready_results( - false, make_error_code(error::canceled)); - BOOST_TEST(out.continued); - BOOST_TEST(out.seen == cond::canceled); - BOOST_TEST_EQ(out.stopped, 0); - BOOST_TEST_EQ(out.destroyed, 1); - } - - //---------------------------------------------------------- - // run() - //---------------------------------------------------------- - - void - run() - { - testMoveAssignDestroysHeldFrame(); -#if !defined(_WIN32) - testDestroyLentTaskTerminates(); - testMoveAssignOverLentTaskTerminates(); - testDestroyUnstartedOrFinishedTaskIsFine(); -#endif - testTeardownThroughQuitter(); - testNormalCompletion(); - testVoidCompletion(); - testExceptionPropagation(); - testStopBeforeFirstAwait(); - testStopDuringIO(); - testStopPropagationChain(); - testMixingQuitterAndTask(); - testNoStopRequested(); - testRAIIVerification(); - testMultipleCoAwait(); - testMoveOperations(); - testReturnString(); - testExceptionInValueQuitter(); - testBoolSuspendAwaitable(); - testBoolSuspendThenStop(); - testCoAwaitVoidQuitter(); - testCoAwaitThrowingQuitter(); - testWhenAllWithStop(); - testWhenAnyWithStop(); - testWakerCancellation(); - testCatchAfterRmwAwaitReady(); - testCatchStopAfterRmwAwaitReady(); - testStopPendingLeavesMutexUntouched(); - testReadyThenStopRunsAwaitResume(); - testReturnSelfThenStopRunsAwaitResume(); - testReadyCanceledThenStopStops(); - testReadySuccessThenStopDelivers(); - testReadyCanceledWithoutStopDelivers(); - testEchoWithShutdown(); - } -}; - -TEST_SUITE( - quitter_test, - "boost.capy.quitter"); - -} // capy -} // boost diff --git a/test/unit/test_helpers.hpp b/test/unit/test_helpers.hpp index c0369430d..a508adfbf 100644 --- a/test/unit/test_helpers.hpp +++ b/test/unit/test_helpers.hpp @@ -39,14 +39,6 @@ #include #include -// Valgrind slows execution ~10-20x; scale timing-sensitive -// durations to avoid false failures. -#ifdef BOOST_NO_STRESS_TEST -inline constexpr int failsafe_scale = 20; -#else -inline constexpr int failsafe_scale = 1; -#endif - #if defined(__linux__) || defined(__FreeBSD__) || defined(__NetBSD__) || defined(__APPLE__) #include #define BOOST_CAPY_TEST_CAN_GET_THREAD_NAME 1