dotnet-io-pipelines

name	dotnet-io-pipelines
description	Builds high-perf network I/O. PipeReader/PipeWriter, backpressure, protocol parsers, Kestrel.

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.

Scope

• PipeReader/PipeWriter patterns and backpressure management
• Protocol parser implementation with ReadOnlySequence
• Kestrel integration and custom transports
• Buffer management and SequencePosition bookmarks

Out of scope

• Async/await fundamentals and ValueTask patterns -- see [skill:dotnet-csharp-async-patterns]
• Benchmarking methodology and Span 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

// 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

```text

### 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();
}

```text

**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

 Task ()
{
     ()
    {
        ReadResult result =  reader.ReadAsync(ct);
        ReadOnlySequence<> buffer = result.Buffer;

        
         (TryParseMessage( buffer,   message))
        {
             ProcessMessageAsync(message, ct);
        }

        
        
        
        
        reader.AdvanceTo(buffer.Start, buffer.End);

         (result.IsCompleted)
            ;  
    }

     reader.CompleteAsync();
}

```text

**Critical rules:**

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

---



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



```csharp

 pipe =  Pipe( PipeOptions(
    pauseWriterThreshold:  * ,   
    resumeWriterThreshold:  * ,  
    minimumSegmentSize: ,
    useSynchronizationContext: ));

```text

| Option                      | Default | Purpose                                                |
| --------------------------- | ------- | ------------------------------------------------------ |
| `PauseWriterThreshold`      | ,  | `FlushAsync` pauses  unread bytes exceed       |
| `ResumeWriterThreshold`     | ,  | `FlushAsync` resumes  unread bytes drop below  |
| `MinimumSegmentSize`        | ,   | Minimum buffer segment allocation size                 |
| `UseSynchronizationContext` | `` | Set ``  server code to avoid context captures  |



 Writer calls `FlushAsync` after `Advance`
 
{
    payload = ;

    
     (buffer.Length < )
         ;

    
     length;
     (buffer.FirstSpan.Length >= )
    {
        length = BinaryPrimitives.ReadInt32BigEndian(buffer.FirstSpan);
    }
    
    {
        
        Span<> lengthBytes =  [];
        buffer.Slice(, ).CopyTo(lengthBytes);
        length = BinaryPrimitives.ReadInt32BigEndian(lengthBytes);
    }

    
     (length <  || length > _048_576)  
          ProtocolViolationException(
            );

    
     totalLength =  + length;
     (buffer.Length < totalLength)
         ;

    
    payload = buffer.Slice(, length);

    
    buffer = buffer.Slice(totalLength);
     ;
}

```text



```
{
    
    SequencePosition? position = buffer.PositionOf(());
     (position  )
    {
        line = ;
         ;
    }

    
    line = buffer.Slice(, position.Value);

    
    buffer = buffer.Slice(buffer.GetPosition(, position.Value));
     ;
}

```text



`ReadOnlySequence<>` may span multiple non-contiguous memory segments. Handle both paths:

```
{
    
     (sequence.IsSingleSegment)
    {
         Encoding.UTF8.GetString(sequence.FirstSpan);
    }

    
     length = ()sequence.Length;
    [] rented = ArrayPool<>.Shared.Rent(length);
    
    {
        sequence.CopyTo(rented);
         Encoding.UTF8.GetString(rented, , length);
    }
    
    {
        ArrayPool<>.Shared.Return(rented);
    }
}

```text

---



Bridge `System.IO.Pipelines`  existing `Stream`-based APIs  `PipeReader.Create`  `PipeWriter.Create`.

```csharp


   networkStream = tcpClient.GetStream();
 reader = PipeReader.Create(networkStream,  StreamPipeReaderOptions(
    bufferSize: ,
    minimumReadSize: ,
    leaveOpen: )); 


{
     ProcessProtocolAsync(reader, cancellationToken);
}

{
     reader.CompleteAsync();
}

```text

```csharp


 writer = PipeWriter.Create(networkStream,  StreamPipeWriterOptions(
    minimumBufferSize: ,
    leaveOpen: )); 


{
     WriteResponseAsync(writer, response, cancellationToken);
}

{
     writer.CompleteAsync();
}

```text

---



ASP.NET Core