From dotnet-skills
Building high-perf network I/O. PipeReader/PipeWriter, backpressure, protocol parsers, Kestrel.
npx claudepluginhub wshaddix/dotnet-skillsThis skill uses the workspace's default tool permissions.
Searches prompts.chat for AI prompt templates by keyword or category, retrieves by ID with variable handling, and improves prompts via AI. Use for discovering or enhancing prompts.
Searches, retrieves, and installs Agent Skills from prompts.chat registry using MCP tools like search_skills and get_skill. Activates for finding skills, browsing catalogs, or extending Claude.
Compares coding agents like Claude Code and Aider on custom YAML-defined codebase tasks using git worktrees, measuring pass rate, cost, time, and consistency.
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).
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.
// 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
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:
GetMemory or GetSpan before writing -- never write to a previously obtained buffer after AdvanceAdvance with the exact number of bytes writtenFlushAsync to make data available to the reader and to respect backpressureasync 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:
AdvanceTo after ReadAsync -- failing to do so leaks memoryconsumed and examined positions: consumed frees memory, examined prevents busy-wait when the buffer has been scanned but does not contain a complete messageReadResult.Buffer after calling AdvanceTo -- the memory may be recycledBackpressure prevents fast producers from overwhelming slow consumers. The pipe pauses the writer when unread data exceeds a threshold.
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 |
FlushAsync after AdvancePauseWriterThreshold, FlushAsync does not complete until the reader consumes enough data to drop below ResumeWriterThresholdThis prevents unbounded memory growth when a producer (network socket, file) is faster than the consumer (parser, business logic).
Pipelines excel at parsing binary protocols because ReadOnlySequence<byte> handles fragmented data across multiple buffer segments without copying.
A common pattern: each message starts with a 4-byte big-endian length header followed by the payload.
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;
}
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.Slice(buffer.GetPosition(1, position.Value));
return true;
}
ReadOnlySequence<byte> may span multiple non-contiguous memory segments. Handle both paths:
static string DecodeUtf8(ReadOnlySequence<byte> sequence)
{
// Fast path: single contiguous segment
if (sequence.IsSingleSegment)
{
return Encoding.UTF8.GetString(sequence.FirstSpan);
}
// Slow path: multi-segment -- rent a contiguous buffer
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);
}
}
Bridge System.IO.Pipelines with existing Stream-based APIs using PipeReader.Create and PipeWriter.Create.
// Wrap a NetworkStream for pipeline-based reading
await using var networkStream = tcpClient.GetStream();
var reader = PipeReader.Create(networkStream, new StreamPipeReaderOptions(
bufferSize: 4096,
minimumReadSize: 1024,
leaveOpen: true)); // Caller manages networkStream lifetime
try
{
await ProcessProtocolAsync(reader, cancellationToken);
}
finally
{
await reader.CompleteAsync();
}
// Wrap a stream for pipeline-based writing
var writer = PipeWriter.Create(networkStream, new StreamPipeWriterOptions(
minimumBufferSize: 4096,
leaveOpen: true)); // Caller manages networkStream lifetime
try
{
await WriteResponseAsync(writer, response, cancellationToken);
}
finally
{
await writer.CompleteAsync();
}
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.
// 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();
}
}
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.
public interface IDuplexPipe
{
PipeReader Input { get; }
PipeWriter Output { get; }
}
ReadOnlySequence<byte> slicing instead of copying to byte[]. Parse directly from the sequence when possible.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.useSynchronizationContext: false -- server code should never capture the synchronization context. This is the default for PipeOptions but explicit is clearer.SequenceReader<byte> -- for complex parsing, SequenceReader<byte> provides TryRead, TryReadBigEndian, AdvancePast, and IsNext methods that handle multi-segment sequences transparently.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;
}
AdvanceTo after ReadAsync -- skipping AdvanceTo leaks pooled memory and eventually causes OutOfMemoryException. Every ReadAsync must be paired with an AdvanceTo.ReadResult.Buffer after calling AdvanceTo -- the underlying memory segments may be returned to the pool. Copy or parse all needed data before advancing.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.FlushResult.IsCompleted -- it means the reader has stopped consuming. Continue writing after this and data will be silently discarded.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.BinaryPrimitives methods on spans shorter than required -- always check buffer.Length before reading fixed-width values to avoid ArgumentOutOfRangeException.