Only create a single publish channel

[danielmarbach]

Fixes #1621

Historically the underlying SDK used was dispatching synchronously the network requests and our code added some additional asynchronous parts using a custom scheduler that waited for the publisher confirms to return. The previous guidance of the RabbitMQ client suggested channels should not be re-used across concurrent operations. That's why the channel provider code was added to make sure during concurrent operations clients are never concurrently accessed. Given the synchronous nature of the SDK calls it was rarely ever possible for more than one or two channels being managed by the channel provider.

With the new async nature of the SDK that also contains an implementation to manage publisher confirms asynchronously (contributed by us) the SDK calls are now completely asynchronous. During the migration to the new client, this detail was overlooked. This leads to massive concurrent access to the channel provider and the channel provider trying to create N number of channels which leads to slow down of publish operations and channels lingering around until the endpoint is restarted or in the worst case the client operations reaching the maximum number of channels allowed.

This PR changes the channel provider to lazy create a single publish channel (since channels are now thread safe as long as they are not used for publishing and consumptions at the same time). The channel provider still applies a rent and return pattern to make sure a faulty channel is closed when returned and new publish operations acquire a fresh channel when needed.

I have played around with multiple internal implementations using AsyncManualResetEvent or CompareExchange/ Spin but ultimately concluded the semaphore version is best given channels are expensive to create.

Benchmark

using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Configs;
using BenchmarkDotNet.Diagnosers;
using BenchmarkDotNet.Jobs;
using NServiceBus.Routing;
using NServiceBus.Transport;

namespace RabbitMQPublishBenchmark;

[Config(typeof(Config))]
public class MessageSessionPublish
{
    private TransportInfrastructure infrastructure;
    private IMessageDispatcher messageDispatcher;

    class Config : ManualConfig
    {
        public Config()
        {
            var baseJob = Job.ShortRun;

            AddDiagnoser(MemoryDiagnoser.Default);

            AddJob(baseJob.WithNuGet("NServiceBus.RabbitMQ", "11.0.0-alpha.1.4").WithId("SemaphoreSlim"));
            AddJob(baseJob.WithNuGet("NServiceBus.RabbitMQ", "10.1.2").WithId("10.1.2"));
            AddJob(baseJob.WithNuGet("NServiceBus.RabbitMQ", "9.2.1").WithId("9.2.1"));
        }
    }

    [IterationSetup]
    public void Setup() // does not support async setup
    {
        var transport = new RabbitMQTransport(RoutingTopology.Conventional(QueueType.Quorum, useDurableEntities: true), "host=localhost;username=guest;password=guest");
        infrastructure = transport.Initialize(new HostSettings("RabbitMQPublishBenchmark", "Benchmark", new StartupDiagnosticEntries(),
            (s, exception, arg3) => { }, true), [], [ "destination"]).GetAwaiter().GetResult();
        messageDispatcher = infrastructure.Dispatcher;
    }

    [IterationCleanup]
    public void Cleanup() // does not support async cleanup
    {
        infrastructure.Shutdown().GetAwaiter().GetResult();
    }

    [Params(1000, 1500, 2000)]
    public int Concurrency { get; set; }

    [Benchmark]
    public Task Concurrent_Publish()
    {
        return Task.WhenAll(Enumerable.Range(0, Concurrency).Select(_ =>
        {
            var message = new OutgoingMessage(Guid.NewGuid().ToString(), new Dictionary<string, string>(), Array.Empty<byte>());
            var transportOperation = new TransportOperations(new TransportOperation(message, new UnicastAddressTag("destination")));

            return messageDispatcher.Dispatch(transportOperation, new TransportTransaction());
        }));
    }

    [Benchmark]
    public async Task Sequential_Publish()
    {
        for (var i = 0; i < Concurrency; i++)
        {
            var message = new OutgoingMessage(Guid.NewGuid().ToString(), new Dictionary<string, string>(), Array.Empty<byte>());
            var transportOperation = new TransportOperations(new TransportOperation(message, new UnicastAddressTag("destination")));

            await messageDispatcher.Dispatch(transportOperation, new TransportTransaction());
        }
    }
}

Results

[danielmarbach]

I have also tested another implementation using a Lazy that roughly looks something like that

        public async ValueTask<ConfirmsAwareChannel> GetPublishChannel(CancellationToken cancellationToken = default)
        {
            var maybePublishChannel = publishChannel;
            if (maybePublishChannel is not null)
            {
                var channel = await maybePublishChannel.Value.ConfigureAwait(false);
                if (channel.IsOpen)
                {
                    return channel;
                }

                Interlocked.CompareExchange(ref publishChannel, null, maybePublishChannel);
            }

            var maybeNewChannel = new Lazy<Task<ConfirmsAwareChannel>>(() => Task.Run(async () =>
            {
                var newChannel = new ConfirmsAwareChannel(connection!, routingTopology);
                await newChannel.Initialize(cancellationToken).ConfigureAwait(false);
                return newChannel;
            }));

            var actualLazy = Interlocked.CompareExchange(ref publishChannel, maybeNewChannel, null) ?? publishChannel!;

            return await actualLazy.Value.ConfigureAwait(false);
        }

        public async ValueTask ReturnPublishChannel(ConfirmsAwareChannel channel, CancellationToken cancellationToken = default)
        {
            if (channel.IsOpen)
            {
                return;
            }

            var maybePublishChannel = publishChannel;
            if (maybePublishChannel is not null)
            {
                var currentChannel = await maybePublishChannel.Value.ConfigureAwait(false);
                if (ReferenceEquals(currentChannel, channel))
                {
                    Interlocked.CompareExchange(ref publishChannel, null, maybePublishChannel);
                    await channel.DisposeAsync().ConfigureAwait(false);
                }
            }
        }

#pragma warning disable PS0018
        public async ValueTask DisposeAsync()
#pragma warning restore PS0018
        {
            if (disposed)
            {
                return;
            }

            await stoppingTokenSource.CancelAsync().ConfigureAwait(false);
            stoppingTokenSource.Dispose();

            var oldConnection = Interlocked.Exchange(ref connection, null);
            oldConnection?.Dispose();

            var maybeOldChannel = Interlocked.Exchange(ref publishChannel, null);
            if (maybeOldChannel is { IsValueCreated: true })
            {
                var oldChannel = await maybeOldChannel.Value.ConfigureAwait(false);
                await oldChannel.DisposeAsync().ConfigureAwait(false);
            }

            disposed = true;
        }

        readonly CancellationTokenSource stoppingTokenSource = new();
        volatile IConnection? connection;
        volatile Lazy<Task<ConfirmsAwareChannel>>? publishChannel;
        bool disposed;

but there is no additional benefit and the complexity is much higher.