Composed Operations

Corosio provides composed operations that build on the primitive read_some() and write_some() functions to provide higher-level guarantees.

Code snippets assume:

#include <boost/capy/read.hpp>
#include <boost/capy/write.hpp>
#include <boost/capy/buffers.hpp>

namespace corosio = boost::corosio;
namespace capy = boost::capy;

The Problem with Primitives

The primitive operations read_some() and write_some() provide no guarantees about how much data is transferred:

char buf[1024];
auto [ec, n] = co_await s.read_some(
    capy::mutable_buffer(buf, sizeof(buf)));
// n could be 1, 100, 500, or 1024 - no guarantee

For many use cases, you need to transfer a specific amount of data. Composed operations provide these guarantees.

capy::read()

The read() function reads until the buffer is full or an error occurs:

char buf[1024];
auto [ec, n] = co_await capy::read(
    stream, capy::mutable_buffer(buf, sizeof(buf)));

// Either:
// - n == 1024 and ec is default (success)
// - ec == capy::cond::eof and n < 1024 (reached end of stream)
// - ec is some other error

Signature

auto
read(
    ReadStream auto& stream,
    MutableBufferSequence auto buffers) ->
        capy::io_task<std::size_t>;

Behavior

  1. Calls read_some() repeatedly until all buffers are filled

  2. If read_some() reports an error (including cond::eof) before the buffers are filled, returns immediately with bytes read so far

  3. On success, returns total bytes (equals buffer_size(buffers))

Reading Until EOF

To consume a stream of unknown length, loop over read_some() into a fixed-size chunk and grow a std::string as data arrives:

std::string content;
char chunk[2048];
for (;;)
{
    auto [ec, n] = co_await stream.read_some(
        capy::mutable_buffer(chunk, sizeof(chunk)));
    content.append(chunk, n);
    if (ec == capy::cond::eof)
        break;  // success: the whole stream was consumed
    if (ec)
        throw std::system_error(ec);
}

Behavior

  1. Each read_some() returns as soon as any data is available

  2. On cond::eof: the loop ends with content holding the entire stream

  3. On any other error: the error is reported with the bytes read so far already appended

capy::write()

The write() function writes all data or fails:

std::string msg = "Hello, World!";
auto [ec, n] = co_await capy::write(
    stream, capy::const_buffer(msg.data(), msg.size()));

// Either:
// - n == msg.size() and ec is default (all data written)
// - ec is an error

Signature

auto
write(
    WriteStream auto& stream,
    ConstBufferSequence auto buffers) ->
        capy::io_task<std::size_t>;

Behavior

  1. Calls write_some() repeatedly until all buffers are written

  2. If an error occurs, returns immediately with bytes written so far

  3. On success, returns total bytes (equals buffer_size(buffers))

buffer_slice Helper

Both read() and write() use capy::consuming_buffers internally to track progress through a buffer sequence. A related helper, capy::buffer_slice, returns a view over a byte range of the underlying sequence. The result is itself a buffer sequence covering only that range, so it can be passed directly to any I/O operation:

#include <boost/capy/buffers/buffer_slice.hpp>

        std::array<capy::mutable_buffer, 2> bufs = {
            capy::mutable_buffer(header, 16),
            capy::mutable_buffer(body, 1024)
        };

        // After reading 20 bytes, slice off what was consumed:
        auto rest = capy::buffer_slice(bufs, 20);
        // rest covers: 4 bytes of header remaining + full body

Interface

template<class BufferSequence>
    requires MutableBufferSequence<BufferSequence>
          || ConstBufferSequence<BufferSequence>
slice_type<BufferSequence>
buffer_slice(
    BufferSequence const& seq,
    std::size_t offset = 0,
    std::size_t length = (std::numeric_limits<std::size_t>::max)());

The returned slice_type models the same buffer-sequence concept as seq (mutable when seq is a mutable buffer sequence): slicing a single buffer yields an adjusted buffer of the same kind, while any other sequence yields a borrowed view. offset is clamped to the total size, and length defaults to the end of the sequence.

Except for the single-buffer case, the result borrows seq: it stores iterators into the sequence, not a copy, so seq must outlive it. Passing a temporary buffer sequence is rejected at compile time; hoist it into a named variable first.

Error Handling Patterns

Structured Bindings with EOF Check

auto [ec, n] = co_await capy::read(stream, buf);
if (ec)
{
    if (ec == capy::cond::eof)
        std::cout << "End of stream, read " << n << " bytes\n";
    else
        std::cerr << "Error: " << ec.message() << "\n";
}

Exception Pattern

// For write (EOF doesn't apply)
auto [wec, n] = co_await capy::write(stream, buf);
if (wec)
    capy::detail::throw_system_error(wec);

// For read (need to handle EOF)
auto [rec, rn] = co_await capy::read(stream, buf);
if (rec && rec != capy::cond::eof)
    capy::detail::throw_system_error(rec);

Cancellation

Composed operations support cancellation through the affine protocol. When cancelled, they return with cond::canceled and the partial byte count.

auto [ec, n] = co_await capy::read(stream, large_buffer);
if (ec == capy::cond::canceled)
    std::cout << "Cancelled after reading " << n << " bytes\n";

Performance Considerations

Single vs. Multiple Buffers

For optimal performance with multiple buffers:

// Efficient: single system call per read_some()
std::array<capy::mutable_buffer, 2> bufs = {
    capy::mutable_buffer(header, 16),
    capy::mutable_buffer(body, 1024)};
co_await capy::read(stream, bufs);

// Less efficient: may require more system calls
co_await capy::read(stream, buf1);
co_await capy::read(stream, buf2);

Buffer Sizing

Choose buffer sizes that match your expected data:

  • Too small: More system calls

  • Too large: Memory waste

For unknown-length data (like HTTP responses), use the read-until-EOF loop shown earlier.

Example: HTTP Response Reading

capy::task<std::string> read_http_response(corosio::io_stream& stream)
{
    std::string response;
    char chunk[2048];
    for (;;)
    {
        auto [ec, n] = co_await stream.read_some(
            capy::mutable_buffer(chunk, sizeof(chunk)));
        response.append(chunk, n);
        if (ec == capy::cond::eof)
            break;
        if (ec)
            throw std::system_error(ec);
    }
    co_return response;
}

Next Steps

Edit this Page