var pipe = new Pipe();
PipeWriter writer = pipe.Writer;
PipeReader reader = pipe.Reader;
```text
### PipeWriter -- Producing Data
```csharp
async Task FillPipeAsync(Stream source, PipeWriter writer,
CancellationToken ct)
{
const int minimumBufferSize = 512;
while (true)
{
Memory<byte> memory = writer.GetMemory(minimumBufferSize);
int bytesRead = await source.ReadAsync(memory, ct);
if (bytesRead == 0)
break;
writer.Advance(bytesRead);
FlushResult result = await writer.FlushAsync(ct);
if (result.IsCompleted)
break;
}
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
async Task ReadPipeAsync(PipeReader reader, CancellationToken ct)
{
while (true)
{
ReadResult result = await reader.ReadAsync(ct);
ReadOnlySequence<byte> buffer = result.Buffer;
while (TryParseMessage(ref buffer, out var message))
{
await ProcessMessageAsync(message, ct);
}
reader.AdvanceTo(buffer.Start, buffer.End);
if (result.IsCompleted)
break;
}
await reader.CompleteAsync();
}
```text
**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,
resumeWriterThreshold: 32 * 1024,
minimumSegmentSize: 4096,
useSynchronizationContext: false));
```text
| 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;
if (buffer.Length < 4)
return false;
int length;
if (buffer.FirstSpan.Length >= 4)
{
length = BinaryPrimitives.ReadInt32BigEndian(buffer.FirstSpan);
}
else
{
Span<byte> lengthBytes = stackalloc byte[4];
buffer.Slice(0, 4).CopyTo(lengthBytes);
length = BinaryPrimitives.ReadInt32BigEndian(lengthBytes);
}
if (length < 0 || length > 1_048_576)
throw new ProtocolViolationException(
$"Invalid message length: {length}");
long totalLength = 4 + length;
if (buffer.Length < totalLength)
return false;
payload = buffer.Slice(4, length);
buffer = buffer.Slice(totalLength);
return true;
}
```text
### Delimiter-Based Protocol Parser (Line Protocol)
```csharp
static bool TryReadLine(
ref ReadOnlySequence<byte> buffer,
out ReadOnlySequence<byte> line)
{
SequencePosition? position = buffer.PositionOf((byte)'\n');
if (position is null)
{
line = default;
return false;
}
line = buffer.Slice(0, position.Value);
buffer = buffer.Slice(buffer.GetPosition(1, position.Value));
return true;
}
```text
### Working with ReadOnlySequence<byte>
`ReadOnlySequence<byte>` may span multiple non-contiguous memory segments. Handle both paths:
```csharp
static string DecodeUtf8(ReadOnlySequence<byte> sequence)
{
if (sequence.IsSingleSegment)
{
return Encoding.UTF8.GetString(sequence.FirstSpan);
}
int length = (int)sequence.Length;
byte[] rented = ArrayPool<byte>.Shared.Rent(length);
try
{
sequence.CopyTo(rented);
return Encoding.UTF8.GetString(rented, 0, length);
}
finally
{
ArrayPool<byte>.Shared.Return(rented);
}
}
```text
---
## Stream Adapter
Bridge `System.IO.Pipelines` with existing `Stream`-based APIs using `PipeReader.Create` and `PipeWriter.Create`.
```csharp
await using var networkStream = tcpClient.GetStream();
var reader = PipeReader.Create(networkStream, new StreamPipeReaderOptions(
bufferSize: 4096,
minimumReadSize: 1024,
leaveOpen: true));
try
{
await ProcessProtocolAsync(reader, cancellationToken);
}
finally
{
await reader.CompleteAsync();
}
```text
```csharp
var writer = PipeWriter.Create(networkStream, new StreamPipeWriterOptions(
minimumBufferSize: 4096,
leaveOpen: true));
try
{
await WriteResponseAsync(writer, response, cancellationToken);
}
finally
{
await writer.CompleteAsync();
}
```text
---
## Kestrel Integration
ASP.NET Core's Kestrel web server uses `System.IO.Pipelines` internally for HTTP request/response processing. Custom
connection middleware can access the transport-level pipe directly.
### Connection Middleware
```csharp
// Custom connection middleware for protocol-level processing
builder.WebHost.ConfigureKestrel(options =>
{
options.ListenLocalhost(9000, listenOptions =>
{
listenOptions.UseConnectionHandler<MyProtocolHandler>();
});
});
public sealed class MyProtocolHandler : ConnectionHandler
{
public override async Task OnConnectedAsync(
ConnectionContext connection)
{
var reader = connection.Transport.Input;
var writer = connection.Transport.Output;
var ct = connection.ConnectionClosed;
try
{
while (true)
{
ReadResult result = await reader.ReadAsync(ct);
ReadOnlySequence<byte> buffer = result.Buffer;
while (TryParseMessage(ref buffer, out var payload))
{
var response = ProcessRequest(payload);
await WriteResponseAsync(writer, response);
}
reader.AdvanceTo(buffer.Start, buffer.End);
if (result.IsCompleted)
break;
}
}
finally
{
await reader.CompleteAsync();
await writer.CompleteAsync();
}
}
private static async Task WriteResponseAsync(
PipeWriter writer, ReadOnlyMemory<byte> response)
{
// Write length prefix + payload
var memory = writer.GetMemory(4 + response.Length);
BinaryPrimitives.WriteInt32BigEndian(
memory.Span, response.Length);
response.CopyTo(memory[4..]);
writer.Advance(4 + response.Length);
await writer.FlushAsync();
}
}
```text
### IDuplexPipe
Kestrel exposes connections as `IDuplexPipe`, combining `PipeReader` and `PipeWriter` into a single transport
abstraction. This pattern also works for custom TCP servers, WebSocket handlers, and named-pipe protocols.
```csharp
public interface IDuplexPipe
{
PipeReader Input { get; }
PipeWriter Output { get; }
}
```text
---
## Performance Tips
1. **Minimize copies** -- use `ReadOnlySequence<byte>` slicing instead of copying to `byte[]`. Parse directly from the
sequence when possible.
2. **Use `GetSpan`/`GetMemory` correctly** -- request the minimum size you need. The pipe may return a larger buffer,
which is fine. Do not cache the returned `Span`/`Memory` across `Advance`/`FlushAsync` calls.
3. **Set `useSynchronizationContext: false`** -- server code should never capture the synchronization context. This is
the default for `PipeOptions` but explicit is clearer.
4. **Tune pause/resume thresholds** -- the defaults (64 KB / 32 KB) work for most scenarios. Increase for
high-throughput bulk transfer; decrease for low-latency interactive protocols.
5. **Prefer `SequenceReader<byte>`** -- for complex parsing, `SequenceReader<byte>` provides `TryRead`,
`TryReadBigEndian`, `AdvancePast`, and `IsNext` methods that handle multi-segment sequences transparently.
```csharp
static bool TryParseHeader(
ref ReadOnlySequence<byte> buffer,
out int messageType,
out int length)
{
var reader = new SequenceReader<byte>(buffer);
if (!reader.TryRead(out byte typeByte) ||
!reader.TryReadBigEndian(out int len))
{
messageType = 0;
length = 0;
return false;
}
messageType = typeByte;
length = len;
buffer = buffer.Slice(reader.Position);
return true;
}
```text
---
## Agent Gotchas
1. **Do not forget to call `AdvanceTo` after `ReadAsync`** -- skipping `AdvanceTo` leaks pooled memory and eventually
causes `OutOfMemoryException`. Every `ReadAsync` must be paired with an `AdvanceTo`.
2. **Do not access `ReadResult.Buffer` after calling `AdvanceTo`** -- the underlying memory segments may be returned to
the pool. Copy or parse all needed data before advancing.
3. **Do not set `consumed` equal to `examined` when no complete message was found** -- this creates a busy-wait loop.
Set `consumed` to `buffer.Start` (nothing consumed) and `examined` to `buffer.End` (everything examined) so the pipe
waits for new data.
4. **Do not ignore `FlushResult.IsCompleted`** -- it means the reader has stopped consuming. Continue writing after this
and data will be silently discarded.
5. **Do not use `Pipe` for simple stream-to-stream copying** -- `Stream.CopyToAsync` is simpler and equally efficient.
Use pipelines when you need parsing, backpressure, or zero-copy slicing.
6. **Do not use `BinaryPrimitives` methods on spans shorter than required** -- always check `buffer.Length` before
reading fixed-width values to avoid `ArgumentOutOfRangeException`.
---
## Knowledge Sources
- Stephen Toub,
[System.IO.Pipelines: High performance IO in .NET](https://devblogs.microsoft.com/dotnet/system-io-pipelines-high-performance-io-in-net/)
-- canonical deep dive on pipeline design, motivation, and usage patterns
## References
- [System.IO.Pipelines overview](https://learn.microsoft.com/en-us/dotnet/standard/io/pipelines)
- [Pipe class API reference](https://learn.microsoft.com/en-us/dotnet/api/system.io.pipelines.pipe)
- [PipeReader API reference](https://learn.microsoft.com/en-us/dotnet/api/system.io.pipelines.pipereader)
- [PipeWriter API reference](https://learn.microsoft.com/en-us/dotnet/api/system.io.pipelines.pipewriter)
- [SequenceReader<T>](https://learn.microsoft.com/en-us/dotnet/api/system.buffers.sequencereader-1)
- [Kestrel connection middleware](https://learn.microsoft.com/en-us/aspnet/core/fundamentals/servers/kestrel/endpoints)