为什么这个 System.IO.Pipelines 代码比基于 Stream 的代码慢得多？

Question

我编写了一个小解析程序来比较 .NET Core 中较旧的 System.IO.Stream 和较新的 System.IO.Pipelines。我期望管道代码具有相同的速度或更快。但是，它慢了大约 40%。

程序很简单：它在 100Mb 的文本文件中搜索关键字，并返回关键字的行号。这是 Stream 版本：

public static async Task<int> GetLineNumberUsingStreamAsync(
    string file,
    string searchWord)
{
    using var fileStream = File.OpenRead(file);
    using var lines = new StreamReader(fileStream, bufferSize: 4096);

    int lineNumber = 1;
    // ReadLineAsync returns null on stream end, exiting the loop
    while (await lines.ReadLineAsync() is string line)
    {
        if (line.Contains(searchWord))
            return lineNumber;

        lineNumber++;
    }
    return -1;
}

我希望上面的流代码比下面的管道代码慢，因为流代码将字节编码为 StreamReader 中的字符串。管道代码通过对字节进行操作来避免这种情况：

public static async Task<int> GetLineNumberUsingPipeAsync(string file, string searchWord)
{
    var searchBytes = Encoding.UTF8.GetBytes(searchWord);
    using var fileStream = File.OpenRead(file);
    var pipe = PipeReader.Create(fileStream, new StreamPipeReaderOptions(bufferSize: 4096));

    var lineNumber = 1;
    while (true)
    {
        var readResult = await pipe.ReadAsync().ConfigureAwait(false);
        var buffer = readResult.Buffer;

        if(TryFindBytesInBuffer(ref buffer, searchBytes, ref lineNumber))
        {
            return lineNumber;
        }

        pipe.AdvanceTo(buffer.End);

        if (readResult.IsCompleted) break;
    }

    await pipe.CompleteAsync();

    return -1;
}

以下是相关的辅助方法：

/// <summary>
/// Look for `searchBytes` in `buffer`, incrementing the `lineNumber` every
/// time we find a new line.
/// </summary>
/// <returns>true if we found the searchBytes, false otherwise</returns>
static bool TryFindBytesInBuffer(
    ref ReadOnlySequence<byte> buffer,
    in ReadOnlySpan<byte> searchBytes,
    ref int lineNumber)
{
    var bufferReader = new SequenceReader<byte>(buffer);
    while (TryReadLine(ref bufferReader, out var line))
    {
        if (ContainsBytes(ref line, searchBytes))
            return true;

        lineNumber++;
    }
    return false;
}

static bool TryReadLine(
    ref SequenceReader<byte> bufferReader,
    out ReadOnlySequence<byte> line)
{
    var foundNewLine = bufferReader.TryReadTo(out line, (byte)'\n', advancePastDelimiter: true);
    if (!foundNewLine)
    {
        line = default;
        return false;
    }

    return true;
}

static bool ContainsBytes(
    ref ReadOnlySequence<byte> line,
    in ReadOnlySpan<byte> searchBytes)
{
    return new SequenceReader<byte>(line).TryReadTo(out var _, searchBytes);
}

我在上面使用SequenceReader<byte>，因为我的理解是它比ReadOnlySequence<byte> 更智能/更快；当它可以在单个 Span<byte> 上运行时，它有一条快速路径。

以下是基准测试结果 (.NET Core 3.1)。完整代码和 BenchmarkDotNet 结果可在in this repo 获得。

• GetLineNumberWithStreamAsync - 435.6 毫秒，同时分配 366.19 MB
• GetLineNumberUsingPipeAsync - 619.8 毫秒，同时分配 9.28 MB

我在管道代码中做错了吗？

更新：Evk 已经回答了这个问题。应用他的修复后，以下是新的基准数字：

• GetLineNumberWithStreamAsync - 452.2 毫秒，同时分配 366.19 MB
• GetLineNumberWithPipeAsync - 203.8 毫秒，但分配了 9.28 MB

Answer 1

这可能不是您要寻找的解释，但我希望它能提供一些见解：

浏览一下你所拥有的两种方法，它显示第二个解决方案在计算上比另一个解决方案更复杂，因为它有两个嵌套循环。

使用代码分析进行更深入的挖掘表明，第二个 (GetLineNumberUsingPipeAsync) 的 CPU 密集度几乎比使用 Stream 的那个高 21.5%（请查看屏幕截图，）并且它与我得到的基准测试结果足够接近：

• 解决方案#1：683.7 毫秒，365.84 MB

• 解决方案#2：777.5 毫秒，9.08 MB

Answer 2

我相信原因是SequenceReader.TryReadTo 的实现。 Here is the source code 这个方法。它使用非常简单的算法（读取到第一个字节的匹配，然后检查该匹配之后的所有后续字节，如果不是 - 向前推进 1 个字节并重复），并注意在这个实现中有很多方法称为“慢” （IsNextSlow、TryReadToSlow 等等），所以至少在某些情况下，在某些情况下，它会退回到一些缓慢的路径。它还必须处理可能包含多个段的事实序列，并保持位置。

在您的情况下，您可以避免使用 SequenceReader 专门用于搜索匹配项（但将其留给实际读取行），例如进行这些小的更改（TryReadTo 的重载在这种情况下也更有效）：

private static bool TryReadLine(ref SequenceReader<byte> bufferReader, out ReadOnlySpan<byte> line) {
    // note that both `match` and `line` are now `ReadOnlySpan` and not `ReadOnlySequence`
    var foundNewLine = bufferReader.TryReadTo(out ReadOnlySpan<byte> match, (byte) '\n', advancePastDelimiter: true);

    if (!foundNewLine) {
        line = default;
        return false;
    }

    line = match;
    return true;
}

然后：

private static bool ContainsBytes(ref ReadOnlySpan<byte> line, in ReadOnlySpan<byte> searchBytes) {
    // line is now `ReadOnlySpan` so we can use efficient `IndexOf` method
    return line.IndexOf(searchBytes) >= 0;
}

这将使您的管道代码比流代码运行得更快。