# Dotnet Io Pipelines > Builds high-perf network I/O. PipeReader/PipeWriter, backpressure, protocol parsers, Kestrel. - Skill: `rudironsoni/dotnet-io-pipelines` (Agent Skill, multi-file: 2 files) - Install (CLI): `npx skillmds@latest add rudironsoni/dotnet-io-pipelines` - Raw SKILL.md: https://api.skillmd.com/api/skills/rudironsoni/dotnet-io-pipelines/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools, AI & ML - License: MIT - Author: rudironsoni (https://skillmd.com/u/rudironsoni) - Updated: 2026-09-10 - Page: https://skillmd.com/skills/rudironsoni/dotnet-io-pipelines --- # 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` | | 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 ```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 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 async Task ReadPipeAsync(PipeReader reader, CancellationToken ct) { while (true) { ReadResult result = await reader.ReadAsync(ct); ReadOnlySequence 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(); } ```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, // Pause writer at 64 KB buffered resumeWriterThreshold: 32 * 1024, // Resume writer when buffer drops to 32 KB 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` 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 buffer, out ReadOnlySequence 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 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; } ```text ### Delimiter-Based Protocol Parser (Line Protocol) ```csharp static bool TryReadLine( ref ReadOnlySequence buffer, out ReadOnlySequence 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; } ```text ### Working with ReadOnlySequence `ReadOnlySequence` may span multiple non-contiguous memory segments. Handle both paths: ```csharp static string DecodeUtf8(ReadOnlySequence 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.Shared.Rent(length); try { sequence.CopyTo(rented); return Encoding.UTF8.GetString(rented, 0, length); } finally { ArrayPool.Shared.Return(rented); } } ```text --- ## Stream Adapter Bridge `System.IO.Pipelines` with existing `Stream`-based APIs using `PipeReader.Create` and `PipeWriter.Create`. ```csharp // 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(); } ```text ```csharp // 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(); } ```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(); }); }); 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 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; } } ## Detailed Examples See [references/detailed-examples.md](references/detailed-examples.md) for complete code samples and advanced patterns.