From bcf6c9557c0ebfef6310b075007e0d5db2636ae8 Mon Sep 17 00:00:00 2001 From: Michael Vandeberg Date: Fri, 16 Oct 2026 13:54:05 -0600 Subject: [PATCH] docs: add sequence diagrams and a protocol anatomy to the introduction The Boost review asked for up-front mental models for each kind of reader and for sequence diagrams of the component interactions (#399). Protocol anatomy and diagrams: - The introduction (index.adoc) ends with a Protocol Anatomy section in place of its Next Steps list. It has a sequence diagram of a task starting, suspending on I/O and resuming through the executor. A short model and reading path follow for each of the three audiences that P4172R1 section 3 defines: application developers (with a subsection for protocol libraries on streams), framework authors and I/O library authors. Each audience gets the facilities the paper assigns it, and the paths name example programs where they exist. - 4f gains "The Cancellation Sequence": a sequence diagram of a stop request reaching a suspended task, the race between the stop callback and normal completion, and a warning about that race. - Each diagram has a numbered step list that matches its arrows, plus mermaid accTitle/accDescr, so the content survives without JS. - partials/diagram-style.adoc puts a +++++ diff --git a/test/doc/programs/index_page_trace.cpp b/test/doc/programs/index_page_trace.cpp new file mode 100644 index 000000000..c2c86d649 --- /dev/null +++ b/test/doc/programs/index_page_trace.cpp @@ -0,0 +1,244 @@ +// +// 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 +// + +// Full program shown in pages/index.adoc, section Protocol Anatomy. + +/* + Runs the sequence drawn in the "How a Task Runs" diagram and records one + trace entry per diagram arrow: the executor logs every dispatch, post and + resume, the I/O awaitable logs its protocol steps, and the tasks log + around each co_await. main() exits non-zero if the recorded order differs + from the diagram, so a library change that alters the sequence (an extra + executor hop, an inline resume) fails this test instead of silently + falsifying the page. + + Everything runs on the main thread. The executor never resumes inline: + dispatch() and post() both queue, and main() drains the queue. The fake + I/O backend completes an operation only when the queue is empty, which is + the "thread is free" state the diagram shows. +*/ + +// tag::full[] +#include +#include +#include +#include +#include +#include +#include + +namespace capy = boost::capy; + +std::vector trace; + +void step(std::string what) +{ + std::cout << trace.size() + 1 << ". " << what << "\n"; + trace.push_back(std::move(what)); +} + +// A fake I/O backend: pending operations complete when main() polls. +std::queue> backend; + +class trace_loop : public capy::execution_context +{ + std::queue queue_; + +public: + class executor_type; + + trace_loop() + : execution_context(this) + { + } + + // Nothing will run what is still queued: give it up, so its chain is + // unwound. Destroying a service, or unwinding a chain, can queue more + // work, so repeat until a drain finds nothing. + ~trace_loop() + { + auto drain = [this] + { + bool any = false; + while (!queue_.empty()) + { + auto c = queue_.front(); + queue_.pop(); + any = true; + capy::abandon(c); + } + return any; + }; + shutdown(); + drain(); + do + { + destroy(); + } + while (drain()); + } + + // A copy keeps the owner, which abandon needs. + void enqueue(capy::continuation const& c) + { + queue_.push(c); + } + + // Run queued coroutines. Poll the backend only when the queue is empty. + void run() + { + while (!queue_.empty() || !backend.empty()) + { + if (queue_.empty()) + { + auto complete = std::move(backend.front()); + backend.pop(); + complete(); + continue; + } + auto c = queue_.front(); + queue_.pop(); + step("executor: resume"); + capy::safe_resume(c.h); + } + } + + executor_type get_executor() noexcept; +}; + +class trace_loop::executor_type +{ + trace_loop* loop_ = nullptr; + +public: + executor_type() = default; + + explicit executor_type(trace_loop& loop) noexcept + : loop_(&loop) + { + } + + capy::execution_context& context() const noexcept + { + return *loop_; + } + + void on_work_started() const noexcept {} + void on_work_finished() const noexcept {} + + std::coroutine_handle<> dispatch(capy::continuation& c) const + { + step("executor: dispatch"); + loop_->enqueue(c); + return std::noop_coroutine(); + } + + void post(capy::continuation& c) const + { + step("executor: post"); + loop_->enqueue(c); + } + + bool operator==(executor_type const& other) const noexcept + { + return loop_ == other.loop_; + } +}; + +trace_loop::executor_type +trace_loop::get_executor() noexcept +{ + return executor_type{*this}; +} + +// An I/O awaitable for a read that the fake backend completes with 5 bytes. +struct read_awaitable +{ + capy::continuation cont_; + capy::executor_ref ex_; + std::size_t bytes_ = 0; + + bool await_ready() const noexcept + { + return false; + } + + std::coroutine_handle<> await_suspend( + std::coroutine_handle<> h, capy::await_context const* env) + { + cont_.h = h; + cont_.owner = env->owner; + ex_ = env->executor; + step("I/O awaitable: start the operation"); + backend.push([this] { + step("I/O backend: the operation completes"); + bytes_ = 5; + ex_.post(cont_); // never cont_.h.resume() + }); + step("I/O awaitable: suspend"); + return std::noop_coroutine(); + } + + capy::results await_resume() const noexcept + { + return {std::error_code(), bytes_}; + } +}; + +capy::task child() +{ + step("child: co_await read_some"); + auto [ec, n] = co_await read_awaitable{}; + step("child: co_return"); + co_return ec ? 0 : n; +} + +capy::task my_task() +{ + step("my_task: co_await child()"); + std::size_t n = co_await child(); + step("my_task: co_return"); + co_return n; +} + +int main() +{ + trace_loop loop; + + step("caller: run_async(ex)(my_task())"); + capy::run_async(loop.get_executor(), [](std::size_t n) { + step("run_async: call the completion handler with " + + std::to_string(n)); + })(my_task()); + + loop.run(); + + // One entry per arrow in the diagram, in the same order. + std::vector const diagram = { + "caller: run_async(ex)(my_task())", // 1 + "executor: dispatch", // 2 + "executor: resume", // 3 + "my_task: co_await child()", // 4 + "child: co_await read_some", // 5 + "I/O awaitable: start the operation", // 6 + "I/O awaitable: suspend", // 7 + "I/O backend: the operation completes", // 8 + "executor: post", // 9 + "executor: resume", // 10 + "child: co_return", // 11 + "my_task: co_return", // 12 + "run_async: call the completion handler with 5", // 13 + }; + if (trace != diagram) + { + std::cout << "trace does not match the diagram\n"; + return 1; + } +} +// end::full[] diff --git a/test/doc/snippets/4e_context_awaitable.cpp b/test/doc/snippets/4e_context_awaitable.cpp index 07c1fe6c7..eff7ad0b5 100644 --- a/test/doc/snippets/4e_context_awaitable.cpp +++ b/test/doc/snippets/4e_context_awaitable.cpp @@ -113,7 +113,6 @@ using result_type = int; // Stand-ins for the async machinery the page leaves unspecified. void start_operation() {} -void start_async_operation() {} void store_result() {} // tag::my_awaitable[] @@ -196,34 +195,6 @@ capy::task add_via_awaitable() } // end::runnable_awaitable[] -// tag::stoppable_awaitable[] -struct stoppable_awaitable -{ - mutable capy::continuation cont_; - std::optional>> stop_cb_; - - bool await_ready() { return false; } - - std::coroutine_handle<> await_suspend( - std::coroutine_handle<> h, capy::await_context const* env) - { - if (env->stop_token.stop_requested()) - return h; // Already cancelled, resume immediately - - // Post through executor when stop is requested - cont_.h = h; - auto ex = env->executor; - stop_cb_.emplace(env->stop_token, - [this, ex]() mutable noexcept { ex.post(cont_); }); - - start_async_operation(); - return std::noop_coroutine(); - } - - void await_resume() { /* ... */ } -}; -// end::stoppable_awaitable[] - // Compiles on purpose: the page shows this to warn that the broken // pattern is not caught by the compiler. struct wrong_stop_callback_demo diff --git a/test/doc/snippets/4f_cancellation.cpp b/test/doc/snippets/4f_cancellation.cpp index 986f637f8..477437bca 100644 --- a/test/doc/snippets/4f_cancellation.cpp +++ b/test/doc/snippets/4f_cancellation.cpp @@ -12,6 +12,13 @@ #include "../doc_warnings.hpp" +// GCC false positive at -O2/-O3: read_op::awaiter lives in a coroutine +// frame, and GCC cannot prove that std::optional +// is only destroyed after emplace() has initialized it. +#if defined(__GNUC__) && !defined(__clang__) +#pragma GCC diagnostic ignored "-Wmaybe-uninitialized" +#endif + #include #include #include @@ -27,15 +34,15 @@ #include #include -#include +#include #include -#include #include // tag::include_stop_token[] #include // end::include_stop_token[] #include #include +#include #include #include @@ -170,58 +177,200 @@ capy::task fetch_with_cancel() } // end::canceled_convention[] -// tag::inline_resume_wrong[] -// WRONG — causes use-after-free -std::optional>> stop_cb; +// Compile-only bug demo: running it would resume a coroutine twice, so +// nothing awaits it. start_read stands in for the I/O backend. +struct racy_read; +void start_read(racy_read*) {} -std::coroutine_handle<> await_suspend( - std::coroutine_handle<> h, capy::await_context const* env) +// tag::post_without_claim[] +// WRONG: the completion and the stop callback can both resume the +// coroutine. +struct racy_read { - stop_cb.emplace(env->stop_token, h); // Resumes inline! - return std::noop_coroutine(); -} -// end::inline_resume_wrong[] + struct cancel_fn + { + racy_read* self; + + void operator()() const noexcept + { + self->canceled_ = true; + self->ex_.post(self->cont_); // even if the read completed + } + }; -// Compile-only bug demo: firing the callback would resume the -// coroutine on the stopping thread, so nothing ever calls it. -[[maybe_unused]] std::coroutine_handle<> (* const wrong_pattern)( - std::coroutine_handle<>, capy::await_context const*) = &await_suspend; + capy::continuation cont_; + capy::executor_ref ex_; + bool canceled_ = false; + std::optional> stop_cb_; -void start_async_operation(std::coroutine_handle<>, capy::await_context const*) {} + bool await_ready() const noexcept { return false; } + + std::coroutine_handle<> await_suspend( + std::coroutine_handle<> h, capy::await_context const* env) + { + cont_.h = h; + cont_.owner = env->owner; + ex_ = env->executor; + stop_cb_.emplace(env->stop_token, cancel_fn{this}); + start_read(this); // the backend calls on_complete() later + return std::noop_coroutine(); + } + + // Called by the I/O backend, on any thread, when the read finishes. + void on_complete() noexcept + { + ex_.post(cont_); // even if the stop callback posted already + } + + void await_resume() noexcept { stop_cb_.reset(); } +}; +// end::post_without_claim[] // tag::stoppable_awaitable[] -struct my_stoppable_awaitable +// An I/O object with one pending read. The I/O backend calls complete() +// from any thread when the read finishes. +class read_op +{ +public: + struct awaiter; + + awaiter wait() noexcept; + + // Called by the I/O backend. Ignored if no wait is armed, or if a + // stop request already claimed the wait. + void complete(std::size_t n) noexcept; + + bool waiting() const noexcept + { + return armed_.load(std::memory_order_acquire); + } + +private: + // The claim: whoever moves armed_ from true to false owns the + // result. The loser does not touch the waiter. + bool claim() noexcept + { + bool expected = true; + return armed_.compare_exchange_strong( + expected, false, std::memory_order_acq_rel); + } + + std::atomic armed_{false}; + awaiter* waiter_ = nullptr; +}; + +struct read_op::awaiter { - using stop_cb_t = std::stop_callback>; + // Runs on the thread that calls request_stop(). + struct cancel_fn + { + awaiter* self; - mutable capy::continuation cont_; - std::unique_ptr stop_cb_; - // ... other members for the async operation ... + void operator()() const noexcept + { + if (!self->op_->claim()) + return; // complete() won, or nothing is armed yet + self->canceled_ = true; + self->arrive(); + } + }; + + read_op* op_; + capy::continuation cont_; + capy::executor_ref ex_; + std::size_t bytes_ = 0; + bool canceled_ = false; + // The winner and the end of await_suspend both arrive here. The + // second to arrive resumes the coroutine, so it cannot resume while + // await_suspend is still running. + std::atomic arrivals_{0}; + std::optional> stop_cb_; + + explicit awaiter(read_op* op) noexcept : op_(op) {} + + // co_await can move the awaiter, but only before await_suspend, + // when there is no stop callback yet. + awaiter(awaiter&& other) noexcept : op_(other.op_) {} + + // Called by the winner. If await_suspend has returned, post the + // coroutine; otherwise await_suspend resumes it. + void arrive() noexcept + { + if (arrivals_.fetch_add(1, std::memory_order_acq_rel) == 1) + ex_.post(cont_); // last action: *this can be destroyed now + } - bool await_ready() { return false; } + bool await_ready() const noexcept { return false; } std::coroutine_handle<> await_suspend( std::coroutine_handle<> h, capy::await_context const* env) { if (env->stop_token.stop_requested()) - return h; // Already cancelled + { + canceled_ = true; + return h; + } + // 1. Write everything the winner reads, before anyone can win. + // The owner lets a holder abandon the chain while it waits. cont_.h = h; - auto ex = env->executor; - stop_cb_ = std::make_unique(env->stop_token, - [this, ex]() mutable noexcept { ex.post(cont_); }); + cont_.owner = env->owner; + ex_ = env->executor; + op_->waiter_ = this; + + // 2. If a stop is requested now, the callback runs here and + // loses, because nothing is armed yet. Step 4 catches it. + stop_cb_.emplace(env->stop_token, cancel_fn{this}); - start_async_operation(h, env); + // 3. Arm. complete() or the stop callback can win from here on. + op_->armed_.store(true, std::memory_order_release); + + // 4. Claim a stop request that arrived during step 2. + if (env->stop_token.stop_requested() && op_->claim()) + { + canceled_ = true; + return h; + } + + // 5. Arrive. If the winner came first, resume now, on this + // thread. Otherwise the winner posts the coroutine later. + if (arrivals_.fetch_add(1, std::memory_order_acq_rel) == 1) + return h; return std::noop_coroutine(); } - void await_resume() { /* check result or throw */ } + capy::results await_resume() noexcept + { + // Waits for a stop callback that is still running to return. + stop_cb_.reset(); + if (canceled_) + return {capy::make_error_code(capy::error::canceled), 0}; + return {std::error_code(), bytes_}; + } }; + +inline read_op::awaiter read_op::wait() noexcept +{ + return awaiter{this}; +} + +inline void read_op::complete(std::size_t n) noexcept +{ + if (!claim()) + return; + waiter_->bytes_ = n; + waiter_->arrive(); +} // end::stoppable_awaitable[] -capy::task<> use_stoppable() +capy::task<> read_once( + read_op& op, std::atomic& resumed, std::atomic& canceled) { - co_await my_stoppable_awaitable{}; + auto [ec, n] = co_await op.wait(); + if (ec == capy::cond::canceled) + canceled.fetch_add(1); + else if (!ec && n == 5) + resumed.fetch_add(1); } // tag::racing_deadline[] @@ -453,15 +602,48 @@ struct cancellation_test void testStoppableAwaitable() { - // A pre-signaled token exercises the already-cancelled fast - // path; the armed path would wait on an operation that this - // sketch never starts. - capy::thread_pool pool(1); - std::stop_source source; - source.request_stop(); - capy::run_async( - pool.get_executor(), source.get_token())(use_stoppable()); + // Declared before the pool, so they outlive its work. + std::atomic resumed{0}; + std::atomic canceled{0}; + read_op stopped_op; + std::stop_source stopped; + int const rounds = 500; + + capy::thread_pool pool(2); + + // Stop requested before the wait: the fast path. + stopped.request_stop(); + capy::run_async(pool.get_executor(), stopped.get_token())( + read_once(stopped_op, resumed, canceled)); + + // complete() racing request_stop(). Exactly one of them wins + // each round, so every round resumes the task exactly once. + for (int i = 0; i < rounds; ++i) + { + read_op op; + std::stop_source source; + std::atomic done{false}; + // The backend completes the read the moment the wait is + // armed, which can be before await_suspend returns. + std::thread backend([&op, &done] { + while (!op.waiting() && !done.load()) + ; + op.complete(5); // a no-op if the stop request won + }); + capy::run_async(pool.get_executor(), source.get_token(), + [&done] { done.store(true); })( + read_once(op, resumed, canceled)); + while (!op.waiting() && !done.load()) + std::this_thread::yield(); + source.request_stop(); + backend.join(); + while (!done.load()) + std::this_thread::yield(); + } + pool.join(); + BOOST_TEST_EQ(resumed.load() + canceled.load(), rounds + 1); + BOOST_TEST(canceled.load() >= 1); } void