dotnet-io-pipelines

What this skill does

# dotnet-io-pipelines

High-performance I/O patterns using `System.IO.Pipelines`. Covers `PipeReader`, `PipeWriter`, backpressure management, protocol parser implementation, and Kestrel integration. Pipelines solve the classic problems of buffer management, incomplete reads, and memory copying that plague traditional stream-based network code.

**Out of scope:** Async/await fundamentals and `ValueTask` patterns -- see [skill:dotnet-csharp-async-patterns]. Benchmarking methodology and `Span<T>` micro-optimization -- see [skill:dotnet-performance-patterns]. File-based I/O (FileStream, RandomAccess, MemoryMappedFile) -- see [skill:dotnet-file-io].

Cross-references: [skill:dotnet-csharp-async-patterns] for async patterns used in pipeline loops, [skill:dotnet-performance-patterns] for Span/Memory optimization techniques, [skill:dotnet-file-io] for file-based I/O patterns (FileStream, RandomAccess, MemoryMappedFile).

---

## Why Pipelines Over Streams

Traditional `Stream`-based I/O forces developers to manage buffers manually, handle partial reads, and copy data between buffers. `System.IO.Pipelines` solves these problems:

| Problem | Stream Approach | Pipeline Approach |
|---------|----------------|-------------------|
| Buffer management | Allocate `byte[]`, resize manually | Automatic pooled buffer management |
| Partial reads | Track position, concatenate fragments | `ReadResult` with `SequencePosition` bookmarks |
| Backpressure | None -- writer can outpace reader | Built-in pause/resume thresholds |
| Memory copies | Copy between buffers at each layer | Zero-copy slicing with `ReadOnlySequence<byte>` |
| Lifetime management | Manual `byte[]` lifecycle | Pooled memory returned on `AdvanceTo` |

The `Pipe` class connects a `PipeWriter` (producer) and a `PipeReader` (consumer) with an internal buffer pool, flow control, and completion signaling.

---

## Core Concepts

### Pipe, PipeReader, PipeWriter

```csharp
// Create a pipe with default options (uses ArrayPool internally)
var pipe = new Pipe();

PipeWriter writer = pipe.Writer;  // Producer side
PipeReader reader = pipe.Reader;  // Consumer side
```

### PipeWriter -- Producing Data

```csharp
async Task FillPipeAsync(Stream source, PipeWriter writer,
    CancellationToken ct)
{
    const int minimumBufferSize = 512;

    while (true)
    {
        // Request a buffer from the pipe's memory pool
        Memory<byte> memory = writer.GetMemory(minimumBufferSize);

        int bytesRead = await source.ReadAsync(memory, ct);
        if (bytesRead == 0)
            break;  // End of stream

        // Tell the pipe how many bytes were written
        writer.Advance(bytesRead);

        // Flush makes data available to the reader.
        // FlushAsync may pause here if the reader is slow (backpressure).
        FlushResult result = await writer.FlushAsync(ct);
        if (result.IsCompleted)
            break;  // Reader stopped consuming
    }

    // Signal completion -- reader will see IsCompleted = true
    await writer.CompleteAsync();
}
```

**Critical rules:**
- Call `GetMemory` or `GetSpan` before writing -- never write to a previously obtained buffer after `Advance`
- Call `Advance` with the exact number of bytes written
- Call `FlushAsync` to make data available to the reader and to respect backpressure

### PipeReader -- Consuming Data

```csharp
async Task ReadPipeAsync(PipeReader reader, CancellationToken ct)
{
    while (true)
    {
        ReadResult result = await reader.ReadAsync(ct);
        ReadOnlySequence<byte> buffer = result.Buffer;

        // Try to parse messages from the buffer
        while (TryParseMessage(ref buffer, out var message))
        {
            await ProcessMessageAsync(message, ct);
        }

        // Tell the pipe how much was consumed and how much was examined.
        // consumed: data that has been fully processed (will be freed)
        // examined: data that has been looked at (won't trigger re-read
        //           until new data arrives)
        reader.AdvanceTo(buffer.Start, buffer.End);

        if (result.IsCompleted)
            break;  // Writer finished and all data consumed
    }

    await reader.CompleteAsync();
}
```

**Critical rules:**
- Always call `AdvanceTo` after `ReadAsync` -- failing to do so leaks memory
- Pass both `consumed` and `examined` positions: `consumed` frees memory, `examined` prevents busy-wait when the buffer has been scanned but does not contain a complete message
- Never access `ReadResult.Buffer` after calling `AdvanceTo` -- the memory may be recycled

---

## Backpressure

Backpressure prevents fast producers from overwhelming slow consumers. The pipe pauses the writer when unread data exceeds a threshold.

### PipeOptions Configuration

```csharp
var pipe = new Pipe(new PipeOptions(
    pauseWriterThreshold: 64 * 1024,   // Pause writer at 64 KB buffered
    resumeWriterThreshold: 32 * 1024,  // Resume writer when buffer drops to 32 KB
    minimumSegmentSize: 4096,
    useSynchronizationContext: false));
```

| Option | Default | Purpose |
|--------|---------|---------|
| `PauseWriterThreshold` | 65,536 | `FlushAsync` pauses when unread bytes exceed this |
| `ResumeWriterThreshold` | 32,768 | `FlushAsync` resumes when unread bytes drop below this |
| `MinimumSegmentSize` | 4,096 | Minimum buffer segment allocation size |
| `UseSynchronizationContext` | `false` | Set `false` for server code to avoid context captures |

### How Backpressure Works

1. Writer calls `FlushAsync` after `Advance`
2. If buffered (unread) data exceeds `PauseWriterThreshold`, `FlushAsync` does not complete until the reader consumes enough data to drop below `ResumeWriterThreshold`
3. The writer is effectively paused -- no busy-waiting, no exceptions, just an awaitable that completes when the reader catches up

This prevents unbounded memory growth when a producer (network socket, file) is faster than the consumer (parser, business logic).

---

## Protocol Parsing

Pipelines excel at parsing binary protocols because `ReadOnlySequence<byte>` handles fragmented data across multiple buffer segments without copying.

### Length-Prefixed Protocol Parser

A common pattern: each message starts with a 4-byte big-endian length header followed by the payload.

```csharp
static bool TryParseMessage(
    ref ReadOnlySequence<byte> buffer,
    out ReadOnlySequence<byte> payload)
{
    payload = default;

    // Need at least 4 bytes for the length prefix
    if (buffer.Length < 4)
        return false;

    // Read length from first 4 bytes
    int length;
    if (buffer.FirstSpan.Length >= 4)
    {
        length = BinaryPrimitives.ReadInt32BigEndian(buffer.FirstSpan);
    }
    else
    {
        // Slow path: length header spans multiple segments
        Span<byte> lengthBytes = stackalloc byte[4];
        buffer.Slice(0, 4).CopyTo(lengthBytes);
        length = BinaryPrimitives.ReadInt32BigEndian(lengthBytes);
    }

    // Validate length to prevent allocation attacks
    if (length < 0 || length > 1_048_576)  // 1 MB max
        throw new ProtocolViolationException(
            $"Invalid message length: {length}");

    // Check if the full message is available
    long totalLength = 4 + length;
    if (buffer.Length < totalLength)
        return false;

    // Extract the payload (zero-copy slice)
    payload = buffer.Slice(4, length);

    // Advance the buffer past this message
    buffer = buffer.Slice(totalLength);
    return true;
}
```

### Delimiter-Based Protocol Parser (Line Protocol)

```csharp
static bool TryReadLine(
    ref ReadOnlySequence<byte> buffer,
    out ReadOnlySequence<byte> line)
{
    // Look for the newline delimiter
    SequencePosition? position = buffer.PositionOf((byte)'\n');
    if (position is null)
    {
        line = default;
        return false;
    }

    // Slice up to (not including) the delimiter
    line = buffer.Slice(0, position.Value);

    // Advance past the delimiter
    buffer = buffer.S