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]、Span/Memory の最適化手法については [skill:dotnet-performance-patterns]、ファイルベースの I/O パターン（FileStream、RandomAccess、MemoryMappedFile）については [skill:dotnet-file-io] を参照してください。

---

ストリームより Pipelines を選ぶ理由

従来の Stream ベースの I/O では、開発者がバッファーを手動で管理し、部分読み取りに対処し、バッファー間でデータをコピーする必要があります。System.IO.Pipelines はこれらの問題を解決します。

問題	ストリームのアプローチ	パイプラインのアプローチ
バッファー管理	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 を超えると、読み取り側が十分なデータを消費して ResumeWriterThreshold を下回るまで、FlushAsync は完了しません
3. 書き込み側は実質的に一時停止します。ビジーウェイトも例外もなく、読み取り側が追いつくと完了する await 可能な処理だけです

これにより、プロデューサー（ネットワークソケット、ファイル）がコンシューマー（パーサー、ビジネスロジック）より速い場合に、メモリが無制限に増加するのを防ぎます。

---

プロトコルの解析

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 サーバーは、HTTP リクエスト／レスポンスの処理に System.IO.Pipelines を内部的に使用しています。カスタム接続ミドルウェアから、トランスポートレベルのパイプに直接アクセスできます。

接続ミドルウェア

// 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. コピーを最小限にする -- byte[] にコピーする代わりに、ReadOnlySequence<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;
}

---

エージェントが注意すべき点

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. 必要な長さより短いスパンに対して BinaryPrimitives のメソッドを使用しないでください -- 固定幅の値を読み取る前に、必ず buffer.Length を確認して ArgumentOutOfRangeException を防いでください。

---

参考資料

• Stephen Toub 著、System.IO.Pipelines: .NET における高パフォーマンス I/O -- パイプラインの設計、動機、使用パターンについての決定版の詳細解説

参考資料

• System.IO.Pipelines の概要
• Pipe クラスの API リファレンス
• PipeReader の API リファレンス
• PipeWriter の API リファレンス
• SequenceReader
• Kestrel の接続ミドルウェア