dotnet-io-pipelines

dotnet-io-pipelines

使用 System.IO.Pipelines 实现高性能 I/O 模式。涵盖 PipeReader、PipeWriter、背压管理、协议解析器实现和 Kestrel 集成。Pipelines 解决了传统的基于流的网络代码常见的缓冲区管理、不完整读取和内存复制问题。

不在讨论范围内： Async/await 基础知识和 ValueTask 模式——请参阅 [skill:dotnet-csharp-async-patterns]。基准测试方法和 Span<T> 微优化——请参阅 [skill:dotnet-performance-patterns]。基于文件的 I/O（FileStream、RandomAccess、MemoryMappedFile）——请参阅 [skill:dotnet-file-io]。

交叉引用：[skill:dotnet-csharp-async-patterns] 介绍管道循环中使用的异步模式；[skill:dotnet-performance-patterns] 介绍 Span/Memory 优化技术；[skill:dotnet-file-io] 介绍基于文件的 I/O 模式（FileStream、RandomAccess、MemoryMappedFile）。

---

为什么选择 Pipelines 而不是 Streams

传统的基于 Stream 的 I/O 要求开发者手动管理缓冲区、处理部分读取，并在缓冲区之间复制数据。System.IO.Pipelines 解决了这些问题：

问题	Stream 方式	Pipeline 方式
缓冲区管理	分配 byte[]，手动调整大小	自动管理池化缓冲区
部分读取	跟踪位置、拼接片段	使用带有 SequencePosition 书签的 ReadResult
背压	无——写入速度可能超过读取速度	内置暂停/恢复阈值
内存复制	在每一层之间复制缓冲区	使用 ReadOnlySequence<byte> 进行零复制切片
生命周期管理	手动管理 byte[] 生命周期	调用 AdvanceTo 后归还池化内存

Pipe 类通过内部缓冲区池、流量控制和完成信号，将 PipeWriter（生产者）与 PipeReader（消费者）连接起来。

---

核心概念

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

PipeWriter——生成数据

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

关键规则：

• 写入前调用 GetMemory 或 GetSpan——调用 Advance 后，切勿再写入之前获取的缓冲区
• 使用实际写入的字节数调用 Advance
• 调用 FlushAsync 以使读取方可以读取数据并遵守背压机制

PipeReader——消费数据

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

关键规则：

• ReadAsync 后务必调用 AdvanceTo——否则会导致内存泄漏
• 同时传入 consumed 和 examined 位置：consumed 用于释放内存，examined 用于在缓冲区已扫描但不包含完整消息时避免忙等待
• 调用 AdvanceTo 后，切勿访问 ReadResult.Buffer——该内存可能已被回收

---

背压

背压可防止快速生产者压垮缓慢的消费者。当未读数据超过阈值时，管道会暂停写入方。

PipeOptions 配置

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));

选项	默认值	用途
PauseWriterThreshold	65,536	未读字节数超过此值时，FlushAsync 会暂停
ResumeWriterThreshold	32,768	未读字节数低于此值时，FlushAsync 会恢复
MinimumSegmentSize	4,096	缓冲区段的最小分配大小
UseSynchronizationContext	false	对于服务端代码，设为 false 以避免捕获上下文

背压的工作原理

1. 写入方在调用 Advance 后调用 FlushAsync
2. 如果缓冲的（未读）数据超过 PauseWriterThreshold，则 FlushAsync 不会完成，直到读取方消费足够的数据，使数据量降至 ResumeWriterThreshold 以下
3. 写入方实际上会暂停——不会忙等待，也不会抛出异常，只会等待读取方追上进度

当生产者（网络套接字、文件）快于消费者（解析器、业务逻辑）时，这可以防止内存无限增长。

---

协议解析

ReadOnlySequence<byte> 可以处理跨多个缓冲区段的分片数据而无需复制，因此 Pipelines 非常适合解析二进制协议。

长度前缀协议解析器

一种常见模式：每条消息以 4 字节的大端长度头开头，后接负载。

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

ReadOnlySequence<byte> 可能跨越多个不连续的内存段。需要处理以下两种情况：

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

---

流适配器

使用 PipeReader.Create 和 PipeWriter.Create，将 System.IO.Pipelines 与现有的基于 Stream 的 API 衔接起来。

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

---

Kestrel 集成

ASP.NET Core 的 Kestrel Web 服务器在内部使用 System.IO.Pipelines 处理 HTTP 请求和响应。自定义连接中间件可以直接访问传输层管道。

连接中间件

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

IDuplexPipe

Kestrel 将连接作为 IDuplexPipe 暴露出来，把 PipeReader 和 PipeWriter 组合为单一的传输抽象。此模式也适用于自定义 TCP 服务器、WebSocket 处理程序和命名管道协议。

public interface IDuplexPipe
{
    PipeReader Input { get; }
    PipeWriter Output { get; }
}

---

性能提示

1. 尽量减少复制——使用 ReadOnlySequence<byte> 切片，而不是复制到 byte[]。尽可能直接从序列中解析。
2. 正确使用 GetSpan/GetMemory——只请求所需的最小长度。管道可能会返回更大的缓冲区，这没有问题。不要在 Advance/FlushAsync 调用之间缓存返回的 Span/Memory。
3. 设置 useSynchronizationContext: false——服务器代码不应捕获同步上下文。PipeOptions 默认就是如此，但明确设置更清晰。
4. 调整暂停/恢复阈值——默认值（64 KB / 32 KB）适用于大多数场景。高吞吐量批量传输时可以调高；低延迟交互式协议则可以调低。
5. 优先使用 SequenceReader<byte>——对于复杂解析，SequenceReader<byte> 提供了 TryRead、TryReadBigEndian、AdvancePast 和 IsNext 方法，可透明地处理多段序列。

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;
}

---

Agent 注意事项

1. 调用 ReadAsync 后不要忘记调用 AdvanceTo——跳过 AdvanceTo 会泄漏池化内存，最终导致 OutOfMemoryException。每次 ReadAsync 都必须配对调用 AdvanceTo。
2. 调用 AdvanceTo 后不要访问 ReadResult.Buffer——底层内存段可能会被归还到池中。推进之前，先复制或解析所有需要的数据。
3. 没有找到完整消息时，不要将 consumed 设为与 examined 相同——这会造成忙等待循环。将 consumed 设为 buffer.Start（未消耗任何数据），将 examined 设为 buffer.End（已检查所有数据），以便管道等待新数据。
4. 不要忽略 FlushResult.IsCompleted——这表示读取器已停止读取。此后继续写入，数据会被悄然丢弃。
5. 不要用 Pipe 进行简单的流到流复制——Stream.CopyToAsync 更简单且同样高效。需要解析、背压或零复制切片时再使用管道。
6. 不要在长度不足的 span 上调用 BinaryPrimitives 方法——读取固定宽度的值前，始终检查 buffer.Length，以避免 ArgumentOutOfRangeException。

---

知识来源

• Stephen Toub，System.IO.Pipelines：.NET 中的高性能 I/O——关于管道设计、动机和使用模式的权威深入介绍

参考资料

• System.IO.Pipelines 概述
• Pipe 类 API 参考
• PipeReader API 参考
• PipeWriter API 参考
• SequenceReader
• Kestrel 连接中间件