QUWI_Pipelines.cs

using System;
using System.Buffers;
using System.IO.Pipelines;
using System.Threading.Tasks;
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Running;
using Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal;

[MemoryDiagnoser]
public class PipeThroughputBenchmark
{
    // 128 (min Length) * InnerLoopCount > _memoryPool.MaxBufferSize
    private const int InnerLoopCount = 33;

    private PipeReader _reader;
    private PipeWriter _writer;
    private MemoryPool<byte> _memoryPool;

    [GlobalSetup]
    public void Setup()
    {
        _memoryPool = KestrelMemoryPool.Create();

        var chunkLength = Length / Chunks;

        // Intentional throws for N/A in the summary
        if (chunkLength > _memoryPool.MaxBufferSize)
        {
            // Parallel test will deadlock if too large (waiting for second Task to complete), so N/A that run
            throw new InvalidOperationException();
        }

        if (Length != chunkLength * Chunks)
        {
            // Test will deadlock waiting for data so N/A that run
            throw new InvalidOperationException();
        }

        var pipe = new Pipe(new PipeOptions(_memoryPool));
        _reader = pipe.Reader;
        _writer = pipe.Writer;
    }

    [Params(1, 2, 4, 8)]
    public int Chunks { get; set; }

    [Params(128, 256, 1024, 2048, 4096)]
    public int Length { get; set; }

    [Benchmark(OperationsPerInvoke = InnerLoopCount)]
    public Task Parse_ParallelAsync()
    {
        return Parse_ParallelAsyncImpl();
    }

    private Task Parse_ParallelAsyncImpl()
    {
        var writing = Task.Run(async () =>
        {
            var chunks = Chunks;
            var chunkLength = Length / chunks;
            for (int i = 0; i < InnerLoopCount; i++)
            {
                for (var c = 0; c < chunks; c++)
                {
                    _writer.GetMemory(chunkLength);
                    _writer.Advance(chunkLength);
                    await _writer.FlushAsync();
                }
            }
        });

        var reading = Task.Run(async () =>
        {
            long remaining = InnerLoopCount * Length;
            while (remaining != 0)
            {
                var result = await _reader.ReadAsync();
                remaining -= result.Buffer.Length;
                _reader.AdvanceTo(result.Buffer.End, result.Buffer.End);
            }
        });

        return Task.WhenAll(writing, reading);
    }

    [GlobalCleanup]
    public void Cleanup()
    {
        _memoryPool.Dispose();
    }

    public static void Main(string[] args)
    {
        var summary = BenchmarkSwitcher.FromAssembly(typeof(PipeThroughputBenchmark).Assembly).Run(args);
    }
}