claude-skills/engineering/boost-asio-pro/references/coroutines.md

C++20 Coroutine Style (Boost ≥ 1.77)

The preferred style when the toolchain allows it. Read SKILL.md first — the rules there (write queue, buffer lifetime, version floors) apply here and are not repeated.

Core Architecture

Boost.Asio uses the Proactor pattern: async operations run in the background, completion handlers are invoked with results.

Program → I/O Object → Execution Context → OS → (completion) → Handler

Execution contexts: io_context (single/multi-thread event loop), thread_pool, system_context

I/O objects: tcp::socket, tcp::acceptor, udp::socket, steady_timer, ssl::stream<>

Completion tokens: Control how async results are delivered — use_awaitable, deferred (default), detached, callbacks, futures.

C++20 Coroutines (Preferred Style)

#include <boost/asio.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/use_awaitable.hpp>

namespace asio = boost::asio;
using tcp = asio::ip::tcp;

asio::awaitable<void> echo_session(tcp::socket socket) {
    try {
        char data[1024];
        for (;;) {
            std::size_t n = co_await socket.async_read_some(asio::buffer(data));
            co_await async_write(socket, asio::buffer(data, n));
        }
    } catch (std::exception&) {
        // Connection closed or error — coroutine ends
    }
}

asio::awaitable<void> listener(tcp::acceptor acceptor) {
    for (;;) {
        auto socket = co_await acceptor.async_accept();
        co_spawn(acceptor.get_executor(), echo_session(std::move(socket)), asio::detached);
    }
}

int main() {
    asio::io_context io(1); // concurrency_hint=1 for single-threaded
    tcp::acceptor acceptor(io, {tcp::v4(), 8080});
    co_spawn(io, listener(std::move(acceptor)), asio::detached);
    io.run();
}

Key rules:

• co_spawn(executor, coroutine, completion_token) launches a coroutine
• Without explicit token, async ops use deferred (returns awaitable object for co_await)
• Errors become system_error exceptions by default inside coroutines
• Use asio::detached when you don't need the coroutine's result

Error Handling in Coroutines

Default: Errors throw boost::system::system_error.

Explicit error handling with as_tuple:

auto [ec, n] = co_await socket.async_read_some(
    asio::buffer(data), asio::as_tuple(asio::use_awaitable));
if (ec) { /* handle error, no exception */ }

Wrap, don't use bare as_tuple. Always write as_tuple(use_awaitable). Bare asio::as_tuple resolves against the operation's default completion token (often deferred), which compiles in some contexts but fails in others — wrapping an explicit base token is unambiguous everywhere.

With redirect_error:

boost::system::error_code ec;
std::size_t n = co_await socket.async_read_some(
    asio::buffer(data), asio::redirect_error(ec));

Strands (Thread Safety)

Rule: All async operations on a shared object MUST execute on the same strand.

// Per-connection strand
asio::strand<asio::io_context::executor_type> strand(io.get_executor());
co_spawn(strand, session(std::move(socket)), asio::detached);

// Bind handler to strand
socket.async_read_some(asio::buffer(data),
    asio::bind_executor(strand, [](error_code ec, size_t n) { /*...*/ }));

Implicit strands (no explicit strand needed):

• Single-threaded io_context::run() — all handlers are sequential
• Single chain of async ops on one connection (half-duplex)

Explicit strand required when:

• Multiple threads call io_context::run()
• Full-duplex read+write on same socket
• Shared state accessed from multiple async chains

Full-Duplex: Strand + Write Queue

A strand serializes handler execution, NOT whole composed operations. Two async_writes started "concurrently" on the same strand still overlap and interleave bytes on the wire — the strand only orders the intermediate handlers, not the byte stream. For full-duplex (a read loop plus pushes/replies writing at the same time on one socket), a strand alone is not enough: you must serialize outbound writes yourself with a queue.

// Give each accepted socket its OWN strand, then run every chain (read loop,
// pushes, replies) on that strand. Passing an executor to async_accept means you
// must ALSO pass an explicit completion token — the default-deferred shortcut on
// the zero-arg form no longer applies.
auto socket = co_await acceptor.async_accept(asio::make_strand(io), asio::use_awaitable);
std::make_shared<connection>(std::move(socket))->start();

class connection : public std::enable_shared_from_this<connection> {
    tcp::socket socket_;                 // bound to its own strand
    std::deque<std::string> outbox_;
    bool writing_ = false;
public:
    explicit connection(tcp::socket s) : socket_(std::move(s)) {}

    void start() {
        // Each chain captures `self` so the connection outlives all its coroutines.
        co_spawn(socket_.get_executor(),
                 [self = shared_from_this()] { return self->read_loop(); }, asio::detached);
    }

    // Call ONLY from the connection's strand (e.g. from its own coroutines).
    // From another thread/strand: asio::dispatch(socket_.get_executor(), ...).
    void send(std::string frame) {
        outbox_.push_back(std::move(frame));
        if (!writing_)
            co_spawn(socket_.get_executor(),
                     [self = shared_from_this()] { return self->write_loop(); }, asio::detached);
    }
private:
    asio::awaitable<void> write_loop() {
        writing_ = true;
        while (!outbox_.empty()) {
            co_await async_write(socket_, asio::buffer(outbox_.front()));
            outbox_.pop_front();         // pop only AFTER the write completes
        }
        writing_ = false;
    }
    asio::awaitable<void> read_loop();   // reads frames, calls send() for replies
};

Why each rule matters:

• One strand per connection → read loop and write loop never run their handlers concurrently.
• Write queue + writing_ flag → at most one async_write in flight, so frames never interleave.
• enable_shared_from_this + capturing self in every co_spawn → the connection survives until all of its read/write/timer chains finish.
• The accepted socket from async_accept(make_strand(...)) is basic_stream_socket<tcp, strand<...>>, not tcp::socket. Take it by value (connection(tcp::socket s), store tcp::socket socket_) — the strand executor type-erases into any_io_executor on the move. Passing that accepted socket to a tcp::socket& (by reference) instead will fail to compile — use auto or accept by value.

Strand from inside a coroutine (when io isn't a captured local): get the executor from the coroutine and make a strand off it — no io_context& needed:

auto ex = co_await asio::this_coro::executor;
auto socket = co_await acceptor.async_accept(asio::make_strand(ex), asio::use_awaitable);

Run the read loop and idle watch together — two awaitable<void> branches; don't inspect the result, the first to finish unwinds the other:

using namespace asio::experimental::awaitable_operators;
co_await (read_loop() || idle_watch(socket_, timer_));   // either returning tears down the connection

Stopping a detached side-coroutine (e.g. a per-symbol ticker that must end on unsubscribe/close): a detached co_spawn won't stop itself. Either (a) have its loop re-check a flag each iteration and co_return when gone:

while (subscriptions_.contains(symbol) && socket_.is_open()) {
    timer.expires_after(250ms);
    co_await timer.async_wait(asio::as_tuple(asio::use_awaitable));
    if (/* still subscribed */) send(make_tick(symbol));
}

or (b) spawn it with a cancellation_signal and emit() cancellation on unsubscribe. The flag approach is simpler for per-subscription tickers.

Timers and Timeouts

asio::awaitable<void> with_timeout(tcp::socket& socket) {
    asio::steady_timer timer(co_await asio::this_coro::executor);
    timer.expires_after(std::chrono::seconds(30));

    // Race: read vs timeout (requires awaitable_operators)
    using namespace asio::experimental::awaitable_operators;

    auto result = co_await (
        socket.async_read_some(asio::buffer(data), asio::use_awaitable)
        || timer.async_wait(asio::use_awaitable)
    );

    if (result.index() == 0) { /* read completed */ }
    else { /* timeout — cancel the socket */ socket.close(); }
}

Re-armable idle timeout (reset on every received frame — the common server pattern):

// Run as a long-lived parallel branch. Calling expires_after() again cancels the
// pending wait, resolving the in-flight async_wait with operation_aborted — that
// is the signal to keep waiting, NOT an error. Genuine expiry resolves with no error.
asio::awaitable<void> idle_watch(tcp::socket& sock, asio::steady_timer& timer) {
    for (;;) {
        auto [ec] = co_await timer.async_wait(asio::as_tuple(asio::use_awaitable));
        if (ec == asio::error::operation_aborted) continue;  // re-armed → keep waiting
        if (ec) co_return;                                   // timer error
        sock.close();                                        // real timeout fired
        co_return;
    }
}
// On every frame received from the peer: timer.expires_after(30s);

Parallel operations (&& and ||):

#include <boost/asio/experimental/awaitable_operators.hpp>
using namespace asio::experimental::awaitable_operators;

// Wait for both (AND) — cancels other on failure
auto [read_n, write_n] = co_await (
    async_read(sock, in_buf, use_awaitable) &&
    async_write(sock, out_buf, use_awaitable)
);

// Wait for first (OR) — cancels other on success
auto result = co_await (
    async_read(sock, buf, use_awaitable) ||
    timer.async_wait(use_awaitable)
);

Note: || and && operators require explicit use_awaitable token, and the awaitable_operators.hpp header (Boost ≥ 1.77 — see the version floors in SKILL.md).

Void branches: when a branch returns void (e.g. two awaitable<void> chains), that arm contributes std::monostate to the result variant. If both branches are void the result is variant<monostate, monostate> — don't inspect .index(); just co_await the expression and let whichever finishes first unwind the other.

Cancellation

asio::awaitable<void> cancellable_work() {
    // Check cancellation state
    auto cs = co_await asio::this_coro::cancellation_state;
    if (cs.cancelled() != asio::cancellation_type::none) {
        co_return;
    }

    // Enable cancellation types
    co_await asio::this_coro::reset_cancellation_state(
        asio::enable_total_cancellation());
}

TCP Server Pattern

asio::awaitable<void> server(asio::io_context& io, unsigned short port) {
    tcp::acceptor acceptor(io, {tcp::v4(), port});
    acceptor.set_option(tcp::acceptor::reuse_address(true));

    for (;;) {
        auto socket = co_await acceptor.async_accept();
        co_spawn(
            io.get_executor(),  // or a strand for multi-threaded
            handle_client(std::move(socket)),
            [](std::exception_ptr ep) {
                if (ep) std::rethrow_exception(ep);
            }
        );
    }
}

Buffers

Type	Use
asio::buffer(data, size)	Wrap existing memory (no ownership)
asio::dynamic_buffer(vec)	Growable buffer over vector/string
asio::streambuf	Legacy stream buffer
asio::const_buffer	Read-only view
asio::mutable_buffer	Writable view

Critical: asio::buffer() does NOT own memory. The underlying storage must outlive the async operation.

Resolver (DNS)

asio::awaitable<void> connect_to(asio::io_context& io,
                                  std::string host, std::string port) {
    tcp::resolver resolver(io);
    auto endpoints = co_await resolver.async_resolve(host, port);

    tcp::socket socket(io);
    co_await asio::async_connect(socket, endpoints);
    // socket is now connected
}

Multi-Threaded io_context

asio::io_context io;
std::vector<std::thread> threads;

for (int i = 0; i < std::thread::hardware_concurrency(); ++i) {
    threads.emplace_back([&io] { io.run(); });
}

// All handlers MUST be strand-protected when sharing state
for (auto& t : threads) t.join();

Composed Async Operations (Custom)

template <typename CompletionToken>
auto async_echo(tcp::socket& socket, CompletionToken&& token) {
    return asio::async_initiate<CompletionToken, void(boost::system::error_code)>(
        asio::co_composed<void(boost::system::error_code)>(
            [](auto state, tcp::socket& socket) -> void {
                state.throw_if_cancelled(true);
                state.reset_cancellation_state(asio::enable_terminal_cancellation());
                try {
                    char data[1024];
                    for (;;) {
                        std::size_t n = co_await socket.async_read_some(asio::buffer(data));
                        co_await async_write(socket, asio::buffer(data, n));
                    }
                } catch (const boost::system::system_error& e) {
                    co_return {e.code()};
                }
            }, socket),
        token, std::ref(socket));
}

Line-Based Protocols

For newline-delimited protocols, prefer async_read_until over manual async_read_some + buffer parsing:

asio::awaitable<void> line_echo(tcp::socket socket) {
    asio::streambuf buf;
    for (;;) {
        std::size_t n = co_await asio::async_read_until(socket, buf, '\n');
        std::string line(asio::buffers_begin(buf.data()),
                         asio::buffers_begin(buf.data()) + n);
        buf.consume(n);
        co_await async_write(socket, asio::buffer(line));
    }
}

Or with dynamic_buffer over a std::string:

std::string buf;
std::size_t n = co_await asio::async_read_until(socket, asio::dynamic_buffer(buf), '\n');
std::string line = buf.substr(0, n);
buf.erase(0, n);

Length-Prefixed Binary Framing

For binary protocols, read the fixed-size header fully, then the body fully — two sequential composed reads (async_read fills the whole buffer, handling short reads). Do NOT use async_read_some for framing.

// Frame: [4-byte big-endian length N][N-byte body]
asio::awaitable<std::string> read_frame(tcp::socket& sock) {
    uint32_t len_be = 0;
    co_await async_read(sock, asio::buffer(&len_be, sizeof len_be));  // exactly 4 bytes
    uint32_t n = ntohl(len_be);                    // <arpa/inet.h>; or hand-roll endian swap
    std::string body(n, '\0');
    co_await async_read(sock, asio::buffer(body));  // exactly n bytes
    co_return body;
}

asio::awaitable<void> write_frame(tcp::socket& sock, std::string_view body) {
    uint32_t len_be = htonl(static_cast<uint32_t>(body.size()));
    std::array<asio::const_buffer, 2> bufs{
        asio::buffer(&len_be, sizeof len_be), asio::buffer(body)};
    co_await async_write(sock, bufs);              // gather-write header + body atomically
    // len_be and body must outlive the write — they do here (co_await suspends in-frame).
}

Graceful Shutdown (signal_set)

asio::signal_set signals(io, SIGINT, SIGTERM);
signals.async_wait([&](const boost::system::error_code&, int /*signo*/) {
    acceptor.close();   // stop accepting; let in-flight sessions drain, then io.run() returns
    // or, for an immediate stop: io.stop();
});

For coroutine-style shutdown, co_await signals.async_wait() in a dedicated coroutine instead of a callback.

Quick Reference

Operation	Function
Launch coroutine	co_spawn(executor, coro, token)
Accept connection	co_await acceptor.async_accept()
Read some bytes	co_await socket.async_read_some(buffer)
Read exact/until	co_await async_read(stream, buf) / async_read_until(stream, buf, delim)
Write all	co_await async_write(stream, buffer)
Connect	co_await async_connect(socket, endpoints)
Resolve DNS	co_await resolver.async_resolve(host, port)
Wait timer	co_await timer.async_wait()
TLS handshake	co_await stream.async_handshake(type)
Get executor	co_await asio::this_coro::executor