src/SignalR/perf/Microbenchmarks/HubConnectionReceiveBenchmark.cs - main - githubmirror/aspnetcore

src/SignalR/perf/Microbenchmarks/HubConnectionReceiveBenchmark.cs
159 строк5 KB
William Godbe
Rename Microsoft.Aspnetcore.Testing to Microsoft.AspNetCore.InternalTesting (#51713)
02 ноя 2023, 03:15
02 ноя 2023, 03:15fc8deca
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.

using System.Buffers;
using System.IO.Pipelines;
using BenchmarkDotNet.Attributes;
using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Connections.Features;
using Microsoft.AspNetCore.Internal;
using Microsoft.AspNetCore.SignalR.Client;
using Microsoft.AspNetCore.SignalR.Microbenchmarks.Shared;
using Microsoft.AspNetCore.SignalR.Protocol;
using Microsoft.AspNetCore.SignalR.Tests;
using Microsoft.AspNetCore.InternalTesting;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;

namespace Microsoft.AspNetCore.SignalR.Microbenchmarks;

public class HubConnectionReceiveBenchmark
{
    private const string MethodName = "TestMethodName";
    private static readonly object _invokeLock = new object();

    private HubConnection _hubConnection;
    private TestDuplexPipe _pipe;
    private ReadOnlyMemory<byte> _invocationMessageBytes;

    private int _currentInterationMessageCount;
    private TaskCompletionSource<ReadResult> _tcs;
    private TaskCompletionSource<ReadResult> _nextReadTcs;
    private TaskCompletionSource _waitTcs;

    [GlobalSetup]
    public void GlobalSetup()
    {
        var arguments = new object[ArgumentCount];
        for (var i = 0; i < arguments.Length; i++)
        {
            arguments[i] = "Hello world!";
        }

        var writer = MemoryBufferWriter.Get();
        try
        {
            HandshakeProtocol.WriteResponseMessage(HandshakeResponseMessage.Empty, writer);
            var handshakeResponseResult = new ReadResult(new ReadOnlySequence<byte>(writer.ToArray()), false, false);

            _pipe = new TestDuplexPipe();
            _pipe.AddReadResult(new ValueTask<ReadResult>(handshakeResponseResult));
        }
        finally
        {
            MemoryBufferWriter.Return(writer);
        }

        _nextReadTcs = new TaskCompletionSource<ReadResult>();
        _pipe.AddReadResult(new ValueTask<ReadResult>(_nextReadTcs.Task));

        IHubProtocol hubProtocol;

        var hubConnectionBuilder = new HubConnectionBuilder();
        if (Protocol == "json")
        {
            hubProtocol = new NewtonsoftJsonHubProtocol();
        }
        else
        {
            hubProtocol = new MessagePackHubProtocol();
        }

        hubConnectionBuilder.Services.TryAddEnumerable(ServiceDescriptor.Singleton(typeof(IHubProtocol), hubProtocol));
        hubConnectionBuilder.WithUrl("http://doesntmatter");

        _invocationMessageBytes = hubProtocol.GetMessageBytes(new InvocationMessage(MethodName, arguments));

        var delegateConnectionFactory = new DelegateConnectionFactory(endPoint =>
        {
            var connection = new DefaultConnectionContext();
            // prevents keep alive time being activated
            connection.Features.Set<IConnectionInherentKeepAliveFeature>(new TestConnectionInherentKeepAliveFeature());
            connection.Transport = _pipe;
            return new ValueTask<ConnectionContext>(connection);
        });
        hubConnectionBuilder.Services.AddSingleton<IConnectionFactory>(delegateConnectionFactory);

        _hubConnection = hubConnectionBuilder.Build();
        _hubConnection.On(MethodName, arguments.Select(v => v.GetType()).ToArray(), OnInvoke);
        _hubConnection.StartAsync().GetAwaiter().GetResult();
    }

    private Task OnInvoke(object[] args)
    {
        // HubConnection now runs this callback serially but just in case
        // add a lock in case of future experimentation
        lock (_invokeLock)
        {
            _currentInterationMessageCount++;

            if (_currentInterationMessageCount == MessageCount)
            {
                _currentInterationMessageCount = 0;
                _waitTcs.SetResult();
            }
            else if (_currentInterationMessageCount > MessageCount)
            {
                throw new InvalidOperationException("Should never happen.");
            }
        }

        return Task.CompletedTask;
    }

    [Params(0, 1, 10, 100)]
    public int ArgumentCount;

    [Params(1, 100)]
    public int MessageCount;

    [Params("json", "messagepack")]
    public string Protocol;

    [GlobalCleanup]
    public void GlobalCleanup()
    {
        _nextReadTcs.SetResult(new ReadResult(default, false, true));
        _hubConnection.StopAsync().GetAwaiter().GetResult();
    }

    public void OperationSetup()
    {
        _tcs = _nextReadTcs;

        // Add the results for additional messages (minus 1 because 1 result has already been added)
        for (int i = 0; i < MessageCount - 1; i++)
        {
            _pipe.AddReadResult(new ValueTask<ReadResult>(new ReadResult(new ReadOnlySequence<byte>(_invocationMessageBytes), false, false)));
        }

        // The receive task that will be waited on once messages are read
        _nextReadTcs = new TaskCompletionSource<ReadResult>();
        _pipe.AddReadResult(new ValueTask<ReadResult>(_nextReadTcs.Task));

        _waitTcs = new TaskCompletionSource();
    }

    [Benchmark]
    public async Task ReceiveAsync()
    {
        // Setup messages
        OperationSetup();

        // Start receive of the next batch of messages
        _tcs.SetResult(new ReadResult(new ReadOnlySequence<byte>(_invocationMessageBytes), false, false));

        // Wait for all messages to be read and invoked
        await _waitTcs.Task.DefaultTimeout();
    }
}