Table of Contents

Async reads, cancellation, and pipelines

Every stream operation has an awaitable twin - ParseAsync, ReadValueAsync, WriteAsync, UpdateAsync, ParseManyAsync, and the generated ParseAsync/WriteAsync/RecordsAsync. This guide explains what those forms do differently from the synchronous ones (less than you might think), how a CancellationToken reaches a read, and how bytes that arrive in pieces - a socket, a PipeReader - become records.

/// <summary>Parses a file asynchronously and verifies failure, cancellation and stream positioning.</summary>
private static async Task AsyncStream()
{
    // A file holds a six-byte header and a payload. Opened for asynchronous I/O, the header is read with
    // ParseAsync: the bytes arrive through ReadAsync while the thread is free, then the ordinary reader decodes them.
    var layout = new CStruct("struct header { uint16 kind; uint32 length; };");
    string path = Path.Combine(Path.GetTempPath(), $"cstructsharp-{Guid.NewGuid():N}.bin");
    await File.WriteAllBytesAsync(path, [0x02, 0x00, 0x06, 0x00, 0x00, 0x00, 0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]);
    try
    {
        await using var file = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read, 4096, useAsync: true);

        // A token bounds the wait: a stalled disk or network ends in OperationCanceledException, not a read failure.
        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
        StructValue header = await layout.ParseAsync(file, "header", cancellationToken: timeout.Token);
        Equal((ushort)2, header.Get<ushort>("kind"));
        Equal(6u, header.Get<uint>("length"));
        Equal(6L, file.Position);

        // The payload's length came from the header; the rest of the file is read the usual way.
        byte[] payload = new byte[header.Get<uint>("length")];
        await file.ReadExactlyAsync(payload, timeout.Token);
        Equal("AABBCCDDEEFF", Convert.ToHexString(payload));

        // Three bytes at the end are not a header: the non-throwing form reports the failure the throwing form
        // would raise, and a seekable stream is back where it started.
        file.Position = 9;
        ReadAttempt<StructValue> attempt = await layout.TryReadValueAsync<StructValue>(file, "header", cancellationToken: timeout.Token);
        True(!attempt.Succeeded, "three bytes are not a header");
        True(attempt.Failure is CStructReadException, "the failure is the read exception");
        Equal(9L, file.Position);

        // A cancelled token stops the read before any byte is taken.
        using var cancelled = new CancellationTokenSource();
        cancelled.Cancel();
        file.Position = 0;
        try
        {
            await layout.ParseAsync(file, "header", cancellationToken: cancelled.Token);
            True(false, "unreachable");
        }
        catch (OperationCanceledException)
        {
            Equal(0L, file.Position);
        }
    }
    finally
    {
        File.Delete(path);
    }
}

What async and await do here

A synchronous layout.Parse(file, "header") asks the file for bytes and waits: the thread that made the call sits idle until the disk or the network answers. await layout.ParseAsync(file, "header") asks for the same bytes but gives the thread back while they are on their way; when they arrive, the method continues where it left off. The value it produces is the same StructValue, read by the same code with the same limits and messages.

The awaitable forms return a ValueTask<T>. A ValueTask is a lightweight promise of a result: await it once. When the bytes are already in memory - a MemoryStream that exposes its buffer - the library reads them in place and the ValueTask is already complete when it is returned, so await costs nothing and nothing is allocated for the wait. A FileStream opened with useAsync: true or a network stream pays the real asynchronous I/O, which is the point.

The buffer-then-span rule

There is exactly one reader for a stream and one for a span; the awaitable forms add no third. Instead, each one follows a single rule:

  1. Read the stream with ReadAsync into a pooled buffer: a seekable stream up to its remaining length, any stream up to MaxTotalBytesRead, plus one byte.
  2. Run the span reader over the buffer.
  3. Set the stream's position and return the buffer to the pool.

The extra byte is what makes a value larger than the budget fail with the budget message rather than a short read, exactly as the synchronous stream reader would. The stream position follows from the rule:

seekable stream, success:            origin ─── value ───┤ position   (just after the value)
seekable stream, any failure:        origin ┤ position                (back where it started)
ResolveAddressAsync, GetArrayLength: origin ┤ position                (a query consumes nothing)
non-seekable stream:                 consumed by what was buffered, whatever the outcome

A stream that cannot seek (a socket, a pipe, a compressed stream) cannot be rewound, so the bytes the rule buffered are gone whether the value used them or not. A value-sized budget does not frame consecutive records: ParseAsync may consume the budget plus one byte. For example, with struct record { uint16 value; };, input 01 00 02 00 and a two-byte budget, the first call returns 1 but consumes 01 00 02. A second call has only 00 left, not the two bytes needed for 2. The budget limits decoding; it does not define stream message boundaries.

For fixed-size records, use ParseManyAsync (or generated RecordsAsync). Each two-byte record below starts at offset 0 within its own input slice: 01 00 gives 1 and 02 00 gives 2, in little-endian order. The record iterator consumes whole records without single-value read-ahead. A trailing incomplete record is an error.

/// <summary>Reads two packed uint16 records from a forward-only stream without consuming part of the next record.</summary>
private static async Task ForwardOnlyRecords()
{
    var layout = new CStruct("struct record { uint16 value; };");
    var pipe = new Pipe();
    await pipe.Writer.WriteAsync(new byte[] { 1, 0, 2, 0 });
    await pipe.Writer.CompleteAsync();
    using Stream source = pipe.Reader.AsStream(); // Forward-only, like a network stream.
    var values = new List<ushort>();
    await foreach (StructValue record in layout.ParseManyAsync(source, "record", options: new ReadOptions { MaxTotalBytesRead = 2, }))
    {
        values.Add(record.Get<ushort>("value"));
    }

    Equal("1,2", string.Join(",", values));
}

When the application owns message framing, keep unconsumed bytes in a PipeReader, as shown below.

One consequence of the rule is easy to miss. The buffer starts at the stream's current position, so a stored absolute pointer address counts from that origin - as it does for a span or memory input - while the synchronous stream form counts from the stream's first byte. A file whose addresses are absolute file positions is read asynchronously from position 0, or through the synchronous form.

Cancellation

A CancellationToken is a signal an operation checks: "has someone asked me to stop?" You create it from a CancellationTokenSource, which can be cancelled by a timer (new CancellationTokenSource(TimeSpan.FromSeconds(10))), by a shutdown handler, or by a call to Cancel(). Every awaitable form takes one as its last parameter, and every read and write option record has a CancellationToken property for the synchronous forms; when both are given they are linked, and either one stops the operation.

The library checks the token where a check is cheap and a stop is safe: before the first byte, at every composite, pointer, and array boundary, between the blocks of a large primitive array, between the chunks of a terminated string, and between records of a sequence. It never checks in the middle of decoding one value, so a cancelled read stops with a whole value read and nothing half-decoded.

Cancellation surfaces as OperationCanceledException and never as a read failure: the non-throwing forms (TryReadValue, TryParse, TryReadValueAsync) let it through instead of returning false, because a cancelled read says nothing about the bytes. Buffered runtime async reads and generated stream reads restore a seekable stream's origin even if acquisition fails after reading some bytes. Synchronous runtime reads can leave the cursor where the reader stopped. A forward-only source cannot restore consumed bytes.

Restoration assumes the stream's Position setter still works. If restoring the cursor also fails while handling an earlier read/update failure, the original exception is preserved and the cursor is no longer guaranteed.

Awaitable writes

private static async Task WriteAsyncExample()
{
    var layout = new CStruct("struct header { uint16 kind; uint32 length; };");
    var value = new Dictionary<string, object?> { ["kind"] = (ushort)2, ["length"] = 6u, };
    using var stream = new MemoryStream();

    // Validation happens before the write: a rejected value leaves the stream empty.
    await layout.WriteAsync(stream, "header", value);
    SequenceEqual([0x02, 0x00, 0x06, 0x00, 0x00, 0x00], stream.ToArray());
    try
    {
        await layout.WriteAsync(stream, "header", new Dictionary<string, object?> { ["kind"] = 70000, ["length"] = 6u, });
        True(false, "70000 does not fit uint16");
    }
    catch (CStructWriteException)
    {
        Equal(6L, stream.Length);
    }
}

WriteAsync serializes the value first, into an array of exactly the value's size, then writes it with one WriteAsync call. A validation failure therefore writes nothing: the destination is untouched. UpdateAsync needs a readable, writable, seekable stream; it buffers the region from the origin, applies the update to a copy, writes back only the runs of bytes that differ, and restores the origin. Its failure rule is the synchronous one: a replacement that does not fit, or that would move a later field, changes nothing.

private static async Task UpdateAsyncExample()
{
    var layout = new CStruct("struct header { uint16 kind; uint32 length; };");
    using var stream = new MemoryStream([0x02, 0x00, 0x06, 0x00, 0x00, 0x00]);

    // Only the two bytes of 'kind' are written back; the position returns to the origin.
    await layout.UpdateAsync(stream, "header.kind", (ushort)3);
    SequenceEqual([0x03, 0x00, 0x06, 0x00, 0x00, 0x00], stream.ToArray());
    Equal(0L, stream.Position);
}

Records from a pipe

A socket delivers bytes in whatever pieces the network chose. A PipeReader (System.IO.Pipelines) manages that: ReadAsync hands you everything that has arrived as a ReadOnlySequence<byte> (a chain of buffers), and AdvanceTo(consumed, examined) tells it what you used and what you looked at, so the next ReadAsync returns new bytes appended to the ones you did not consume.

This small example first receives 01, keeps that incomplete record, then receives 00 02 00. It parses exactly two bytes at a time from the retained four-byte sequence, yielding 1 and 2 without losing the next record's bytes.

/// <summary>Keeps an incomplete uint16 in a PipeReader until the next bytes arrive, then consumes only whole records.</summary>
private static async Task RetainedRecord()
{
    var layout = new CStruct("struct record { uint16 value; };");
    var pipe = new Pipe();
    await pipe.Writer.WriteAsync(new byte[] { 1 });
    System.IO.Pipelines.ReadResult first = await pipe.Reader.ReadAsync();
    Equal(1L, first.Buffer.Length);

    // Examined but not consumed: the first byte must survive the next read.
    pipe.Reader.AdvanceTo(first.Buffer.Start, first.Buffer.End);
    await pipe.Writer.WriteAsync(new byte[] { 0, 2, 0 });
    await pipe.Writer.CompleteAsync();

    System.IO.Pipelines.ReadResult next = await pipe.Reader.ReadAsync();
    ReadOnlySequence<byte> retained = next.Buffer;
    Equal(4L, retained.Length);
    var values = new List<ushort>();
    while (retained.Length >= 2)
    {
        StructValue record = layout.Parse(retained.Slice(0, 2), "record");
        values.Add(record.Get<ushort>("value"));
        retained = retained.Slice(2);
    }

    Equal("1,2", string.Join(",", values));
    True(retained.IsEmpty, "no truncated record at end of input");
    pipe.Reader.AdvanceTo(retained.Start, retained.End);
    await pipe.Reader.CompleteAsync();
}

Run both two-record examples from the repository root:

dotnet run --project docs/examples/CStructSharp.Docs.Examples.csproj -c Release -- forward-only-records retained-record

/// <summary>Frames fixed-size and count-prefixed messages, retaining incomplete input across pipe reads.</summary>
private static async Task PipeReaderFraming()
{
    // Four-byte frames arrive on a pipe in chunks that do not line up with the frame boundaries - exactly what a
    // socket delivers. A PipeReader over a NetworkStream (PipeReader.Create(stream)) is used the same way.
    var layout = new CStruct("struct frame { uint16 id; uint16 value; };");
    int size = layout.GetStructSizeInBytes("frame");
    var pipe = new Pipe();

    // Supply chunks independently of the consumer so record boundaries need not match read boundaries.
    Task producer = Task.Run(async () =>
    {
        byte[] frames = [1, 0, 10, 0, 2, 0, 20, 0, 3, 0, 30, 0, 4, 0, 40, 0, 5, 0, 50, 0];
        foreach (int[] chunk in new[] { new[] { 0, 6 }, new[] { 6, 7 }, new[] { 13, 7 } })
        {
            await pipe.Writer.WriteAsync(frames.AsMemory(chunk[0], chunk[1]));
        }

        await pipe.Writer.CompleteAsync();
    });

    var ids = new List<ushort>();
    while (true)
    {
        var result = await pipe.Reader.ReadAsync();
        ReadOnlySequence<byte> buffer = result.Buffer;

        // Parse the whole frames the buffer holds - ParseMany reads a ReadOnlySequence<byte> directly, one frame per
        // step - and hand the partial frame at the end back to the pipe: consumed up to the last whole frame,
        // examined to the end, so the next ReadAsync waits for more bytes instead of returning the same ones.
        long whole = buffer.Length / size * size;
        foreach (StructValue frame in layout.ParseMany(buffer.Slice(0, whole), "frame"))
        {
            ids.Add(frame.Get<ushort>("id"));
        }

        pipe.Reader.AdvanceTo(buffer.GetPosition(whole), buffer.End);
        if (result.IsCompleted)
        {
            True(buffer.Length == whole, "the producer ended on a frame boundary");
            break;
        }
    }

    await producer;
    await pipe.Reader.CompleteAsync();
    Equal("1,2,3,4,5", string.Join(",", ids));

    // A count-prefixed message has no fixed size: read the count first (a whole header, or wait), work out the
    // message's length from it, and parse the message only when every byte of it is there.
    var messages = new CStruct("struct message { uint8 count; uint8 payload[count]; };");
    var chunks = new Pipe();
    await chunks.Writer.WriteAsync(new byte[] { 2, 0xAA, 0xBB, 3, 0xCC });
    await chunks.Writer.WriteAsync(new byte[] { 0xDD, 0xEE, 0 });
    await chunks.Writer.CompleteAsync();
    var lengths = new List<int>();
    while (true)
    {
        var result = await chunks.Reader.ReadAsync();
        ReadOnlySequence<byte> buffer = result.Buffer;
        while (messages.TryReadValue(buffer, "message.count", out byte count) && buffer.Length >= 1 + count)
        {
            StructValue message = messages.Parse(buffer.Slice(0, 1 + count), "message");
            lengths.Add(message.Get<byte[]>("payload").Length);
            buffer = buffer.Slice(1 + count);
        }

        chunks.Reader.AdvanceTo(buffer.Start, buffer.End);
        if (result.IsCompleted)
        {
            break;
        }
    }

    await chunks.Reader.CompleteAsync();
    Equal("2,3,0", string.Join(",", lengths));
}

Two patterns cover most protocols:

  • Fixed-size records. The number of whole records in the buffer is buffer.Length / size. ParseMany reads a ReadOnlySequence<byte> directly - one record per step, a single-segment sequence in place and a chain through one pooled copy - so hand it the whole records and advance past them; the partial record at the end stays in the pipe as examined, which makes ReadAsync wait for more bytes instead of returning the same ones.
  • Count-prefixed messages. Read the count first with TryReadValue (it fails harmlessly while the header is incomplete), compute the message's length from it, and parse the message only when every byte of it is there.

Both loops end when result.IsCompleted says the writer has finished; bytes left over then are a truncated record, which the application decides how to report.

Options as records

ReadOptions, WriteOptions, and UpdateOptions are C# record types, so a variant of a shared options object is a with expression: var quick = defaults with { MaxTotalBytesRead = 4096, CancellationToken = token };. Two option records with the same values are equal, which makes them safe keys and easy to assert on.

private static void OptionsWith()
{
    var layout = new CStruct("struct sample { uint8 count; uint16 values[count]; char name[4]; };");
    byte[] bytes = [1, 0x34, 0x12, (byte)'a', (byte)'b', 0, 0];

    // One shared policy, and a variation that changes a single member.
    var strict = new ReadOptions { MaxArrayElements = 8, MaxStringBytes = 64, };
    ReadOptions trimmed = strict with { TrimFixedText = true, };
    Equal(8, trimmed.MaxArrayElements);
    Equal("ab\0\0", layout.Parse(bytes, "sample", options: strict).Get<string>("name"));
    Equal("ab", layout.Parse(bytes, "sample", options: trimmed).Get<string>("name"));

    // Records compare by their members, so an equal policy is the same policy.
    True(strict == new ReadOptions { MaxArrayElements = 8, MaxStringBytes = 64, }, "equal members, equal options");
    True(strict != trimmed, "one member differs");
}

Check yourself

  1. Which reader decodes the bytes ParseAsync buffered?
  2. A ParseAsync on a FileStream at position 100 fails. Where is the stream afterwards?
  3. Why does TryReadValueAsync throw OperationCanceledException instead of returning false?
  4. In the pipe loop, why is the partial frame passed as examined but not as consumed?
Answers
  1. The span reader - the same one Parse(ReadOnlySpan<byte>) uses. There is no separate asynchronous decoder.
  2. At position 100: a seekable stream returns to its origin on any failure.
  3. A cancelled read says nothing about the bytes; false would claim the bytes were wrong.
  4. Consumed bytes are gone; examined-but-not-consumed bytes stay in the pipe and are returned again with the next chunk, so the frame completes when the rest arrives.

Exercise

Change the pipe producer to write the twenty bytes one at a time. The reader loop should still count five frames: each ReadAsync returns a buffer that holds at most one new whole frame, and often none, and AdvanceTo keeps the incomplete one waiting. Then remove the examined argument (pass buffer.GetPosition(whole) alone) and watch the loop spin: without examined, the pipe reports the same unconsumed bytes as new data every time.

Edit this page