[IO]System.IO.Pipelines

　　System.IO.Pipelines 是一个新库，旨在使在 .NET 中执行高性能 I/O 更加容易。 该库的目标为适用于所有 .NET 实现的 .NET Standard。

System.IO.Pipelines 解决什么问题

　　System.IO.Pipelines 已构建为：

• 具有高性能的流数据分析功能。
• 减少代码复杂性。

下面的代码是典型的 TCP 服务器，它从客户机接收行分隔的消息（由 ' ' 分隔）：

async Task ProcessLinesAsync(NetworkStream stream)
{
    var buffer = new byte[1024];
    await stream.ReadAsync(buffer, 0, buffer.Length);
    
    // Process a single line from the buffer
    ProcessLine(buffer);
}

前面的代码有几个问题：

• 单次调用 ReadAsync 可能无法接收整条消息（行尾）。
• 忽略了 stream.ReadAsync 的结果。 stream.ReadAsync 返回读取的数据量。
• 它不能处理在单个 ReadAsync 调用中读取多行的情况。
• 它为每次读取分配一个 byte 数组。

要解决上述问题，需要进行以下更改：

• 缓冲传入的数据，直到找到新行。

• 分析缓冲区中返回的所有行。

• 该行可能大于 1KB（1024 字节）。 找到需要调整输入缓冲区大小的代码（一行完整的代码）。

  • 如果调整缓冲区的大小，当输入中出现较长的行时，将生成更多缓冲区副本。
  • 压缩用于读取行的缓冲区，以减少空余。

• 请考虑使用缓冲池来避免重复分配内存。

下面的代码解决了其中一些问题：

async Task ProcessLinesAsync(NetworkStream stream)
{
    byte[] buffer = ArrayPool<byte>.Shared.Rent(1024);
    var bytesBuffered = 0;
    var bytesConsumed = 0;

    while (true)
    {
        // Calculate the amount of bytes remaining in the buffer.
        var bytesRemaining = buffer.Length - bytesBuffered;

        if (bytesRemaining == 0)
        {
            // Double the buffer size and copy the previously buffered data into the new buffer.
            var newBuffer = ArrayPool<byte>.Shared.Rent(buffer.Length * 2);
            Buffer.BlockCopy(buffer, 0, newBuffer, 0, buffer.Length);
            // Return the old buffer to the pool.
            ArrayPool<byte>.Shared.Return(buffer);
            buffer = newBuffer;
            bytesRemaining = buffer.Length - bytesBuffered;
        }

        var bytesRead = await stream.ReadAsync(buffer, bytesBuffered, bytesRemaining);
        if (bytesRead == 0)
        {
            // EOF
            break;
        }

        // Keep track of the amount of buffered bytes.
        bytesBuffered += bytesRead;
        var linePosition = -1;

        do
        {
            // Look for a EOL in the buffered data.
            linePosition = Array.IndexOf(buffer, (byte)'
', bytesConsumed,
                                         bytesBuffered - bytesConsumed);

            if (linePosition >= 0)
            {
                // Calculate the length of the line based on the offset.
                var lineLength = linePosition - bytesConsumed;

                // Process the line.
                ProcessLine(buffer, bytesConsumed, lineLength);

                // Move the bytesConsumed to skip past the line consumed (including 
).
                bytesConsumed += lineLength + 1;
            }
        }
        while (linePosition >= 0);
    }
}

Pipe

 Pipe 类可用于创建 PipeWriter/PipeReader 对。 写入 PipeWriter 的所有数据都可用于 PipeReader：

var pipe = new Pipe();
PipeReader reader = pipe.Reader;
PipeWriter writer = pipe.Writer;

Pipe基本用法

async Task ProcessLinesAsync(Socket socket)
{
    var pipe = new Pipe();
    Task writing = FillPipeAsync(socket, pipe.Writer);
    Task reading = ReadPipeAsync(pipe.Reader);

    await Task.WhenAll(reading, writing);
}

async Task FillPipeAsync(Socket socket, PipeWriter writer)
{
    const int minimumBufferSize = 512;

    while (true)
    {
        // Allocate at least 512 bytes from the PipeWriter.
        Memory<byte> memory = writer.GetMemory(minimumBufferSize);
        try
        {
            int bytesRead = await socket.ReceiveAsync(memory, SocketFlags.None);
            if (bytesRead == 0)
            {
                break;
            }
            // Tell the PipeWriter how much was read from the Socket.
            writer.Advance(bytesRead);
        }
        catch (Exception ex)
        {
            LogError(ex);
            break;
        }

        // Make the data available to the PipeReader.
        FlushResult result = await writer.FlushAsync();

        if (result.IsCompleted)
        {
            break;
        }
    }

     // By completing PipeWriter, tell the PipeReader that there's no more data coming.
    await writer.CompleteAsync();
}

async Task ReadPipeAsync(PipeReader reader)
{
    while (true)
    {
        ReadResult result = await reader.ReadAsync();
        ReadOnlySequence<byte> buffer = result.Buffer;

        while (TryReadLine(ref buffer, out ReadOnlySequence<byte> line))
        {
            // Process the line.
            ProcessLine(line);
        }

        // Tell the PipeReader how much of the buffer has been consumed.
        reader.AdvanceTo(buffer.Start, buffer.End);

        // Stop reading if there's no more data coming.
        if (result.IsCompleted)
        {
            break;
        }
    }

    // Mark the PipeReader as complete.
    await reader.CompleteAsync();
}

bool TryReadLine(ref ReadOnlySequence<byte> buffer, out ReadOnlySequence<byte> line)
{
    // Look for a EOL in the buffer.
    SequencePosition? position = buffer.PositionOf((byte)'
');

    if (position == null)
    {
        line = default;
        return false;
    }

    // Skip the line + the 
.
    line = buffer.Slice(0, position.Value);
    buffer = buffer.Slice(buffer.GetPosition(1, position.Value));
    return true;
}

有两个循环：

• FillPipeAsync 从 Socket 读取并写入 PipeWriter。
• ReadPipeAsync 从 PipeReader 读取并分析传入的行。

没有分配显式缓冲区。 所有缓冲区管理都委托给 PipeReader 和 PipeWriter 实现。 委派缓冲区管理使使用代码更容易集中关注业务逻辑。

在第一个循环中：

• 调用 PipeWriter.GetMemory(Int32) 从基础编写器获取内存。
• 调用 PipeWriter.Advance(Int32) 以告知 PipeWriter 有多少数据已写入缓冲区。
• 调用 PipeWriter.FlushAsync 以使数据可用于 PipeReader。

在第二个循环中，PipeReader 使用由 PipeWriter 写入的缓冲区。 缓冲区来自套接字。 对 PipeReader.ReadAsync 的调用：

• 返回包含两条重要信息的 ReadResult：

  • 以 ReadOnlySequence<byte> 形式读取的数据。
  • 布尔值 IsCompleted，指示是否已到达数据结尾 (EOF)。

找到行尾 (EOL) 分隔符并分析该行后：

• 该逻辑处理缓冲区以跳过已处理的内容。
• 调用 PipeReader.AdvanceTo 以告知 PipeReader 已消耗和检查了多少数据。

读取器和编写器循环通过调用 Complete 结束。 Complete 使基础管道释放其分配的内存。

反压和流量控制

理想情况下，读取和分析可协同工作：

• 写入线程使用来自网络的数据并将其放入缓冲区。
• 分析线程负责构造适当的数据结构。

通常，分析所花费的时间比仅从网络复制数据块所用时间更长：

• 读取线程领先于分析线程。
• 读取线程必须减缓或分配更多内存来存储用于分析线程的数据。

为了获得最佳性能，需要在频繁暂停和分配更多内存之间取得平衡。

为解决上述问题，Pipe 提供了两个设置来控制数据流：

• PauseWriterThreshold：确定在调用 FlushAsync 暂停之前应缓冲多少数据。
• ResumeWriterThreshold：确定在恢复对 PipeWriter.FlushAsync 的调用之前，读取器必须观察多少数据。

PipeWriter.FlushAsync：

• 当 Pipe 中的数据量超过 PauseWriterThreshold 时，返回不完整的 ValueTask<FlushResult>。
• 低于 ResumeWriterThreshold 时，返回完整的 ValueTask<FlushResult>。

使用两个值可防止快速循环，如果只使用一个值，则可能发生这种循环。