From 53dbe79f78a6a42acf9d51a720aab8a0ee9fd5f9 Mon Sep 17 00:00:00 2001 From: Daniel Marbach Date: Fri, 6 Jun 2025 12:06:40 +0200 Subject: [PATCH 1/5] Use a single publish channel with atomic swaps --- .../Connection/ChannelProvider.cs | 51 +++++++++++-------- 1 file changed, 30 insertions(+), 21 deletions(-) diff --git a/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs b/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs index 14102f98b..fcc5ccaa3 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs @@ -3,7 +3,6 @@ namespace NServiceBus.Transport.RabbitMQ { using System; - using System.Collections.Concurrent; using System.Threading; using System.Threading.Tasks; using global::RabbitMQ.Client; @@ -18,8 +17,6 @@ public ChannelProvider(ConnectionFactory connectionFactory, TimeSpan retryDelay, this.retryDelay = retryDelay; this.routingTopology = routingTopology; - - channels = new ConcurrentQueue(); } public async Task Initialize(CancellationToken cancellationToken = default) => connection = await CreateConnectionWithShutdownListener(cancellationToken).ConfigureAwait(false); @@ -60,7 +57,7 @@ async Task ReconnectSwallowingExceptions(string? connectionName, CancellationTok var newConnection = await CreateConnectionWithShutdownListener(cancellationToken).ConfigureAwait(false); - // A race condition is possible where CreatePublishConnection is invoked during Dispose + // A race condition is possible where CreatePublishConnection is invoked during Dispose // where the returned connection isn't disposed so invoking Dispose to be sure if (cancellationToken.IsCancellationRequested) { @@ -94,32 +91,43 @@ protected virtual void FireAndForget(Func action, Cance public async ValueTask GetPublishChannel(CancellationToken cancellationToken = default) { - if (channels.TryDequeue(out var channel) && !channel.IsClosed) + while (true) { - return channel; - } + var existing = Interlocked.CompareExchange(ref publishChannel, null, null); + if (existing is not null && !existing.IsClosed) + { + return existing; + } - if (channel is not null) - { - await channel.DisposeAsync() - .ConfigureAwait(false); - } + var newChannel = new ConfirmsAwareChannel(connection!, routingTopology); + await newChannel.Initialize(cancellationToken).ConfigureAwait(false); - channel = new ConfirmsAwareChannel(connection, routingTopology); - await channel.Initialize(cancellationToken).ConfigureAwait(false); + var expected = existing; + var prev = Interlocked.CompareExchange(ref publishChannel, newChannel, expected); - return channel; + if (ReferenceEquals(prev, expected)) + { + if (expected is not null) + { + await expected.DisposeAsync().ConfigureAwait(false); + } + + return newChannel; + } + + await newChannel.DisposeAsync().ConfigureAwait(false); + } } public ValueTask ReturnPublishChannel(ConfirmsAwareChannel channel, CancellationToken cancellationToken = default) { - if (channel.IsOpen) + if (!channel.IsClosed) { - channels.Enqueue(channel); return ValueTask.CompletedTask; } - return channel.DisposeAsync(); + var prev = Interlocked.CompareExchange(ref publishChannel, null, channel); + return ReferenceEquals(prev, channel) ? channel.DisposeAsync() : ValueTask.CompletedTask; } #pragma warning disable PS0018 @@ -137,9 +145,10 @@ public async ValueTask DisposeAsync() var oldConnection = Interlocked.Exchange(ref connection, null); oldConnection?.Dispose(); - foreach (var channel in channels) + var oldChannel = Interlocked.Exchange(ref publishChannel, null); + if (oldChannel is not null) { - await channel.DisposeAsync().ConfigureAwait(false); + await oldChannel.DisposeAsync().ConfigureAwait(false); } disposed = true; @@ -148,9 +157,9 @@ public async ValueTask DisposeAsync() readonly ConnectionFactory connectionFactory; readonly TimeSpan retryDelay; readonly IRoutingTopology routingTopology; - readonly ConcurrentQueue channels; readonly CancellationTokenSource stoppingTokenSource = new(); volatile IConnection? connection; + ConfirmsAwareChannel? publishChannel; bool disposed; static readonly ILog Logger = LogManager.GetLogger(typeof(ChannelProvider)); From 52c757feb0f69f4393cc14bfd5a2073f8eb4fa2e Mon Sep 17 00:00:00 2001 From: Daniel Marbach Date: Fri, 6 Jun 2025 12:07:01 +0200 Subject: [PATCH 2/5] Cosmetics --- .../Connection/ChannelProvider.cs | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs b/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs index fcc5ccaa3..a88cff7fe 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs @@ -9,16 +9,9 @@ namespace NServiceBus.Transport.RabbitMQ using global::RabbitMQ.Client.Events; using Logging; - class ChannelProvider : IAsyncDisposable + class ChannelProvider(ConnectionFactory connectionFactory, TimeSpan retryDelay, IRoutingTopology routingTopology) + : IAsyncDisposable { - public ChannelProvider(ConnectionFactory connectionFactory, TimeSpan retryDelay, IRoutingTopology routingTopology) - { - this.connectionFactory = connectionFactory; - this.retryDelay = retryDelay; - - this.routingTopology = routingTopology; - } - public async Task Initialize(CancellationToken cancellationToken = default) => connection = await CreateConnectionWithShutdownListener(cancellationToken).ConfigureAwait(false); async Task CreateConnectionWithShutdownListener(CancellationToken cancellationToken) @@ -154,9 +147,6 @@ public async ValueTask DisposeAsync() disposed = true; } - readonly ConnectionFactory connectionFactory; - readonly TimeSpan retryDelay; - readonly IRoutingTopology routingTopology; readonly CancellationTokenSource stoppingTokenSource = new(); volatile IConnection? connection; ConfirmsAwareChannel? publishChannel; From 50fb15cacb2b79236bfa582402d932c89e488048 Mon Sep 17 00:00:00 2001 From: Daniel Marbach Date: Fri, 6 Jun 2025 13:41:26 +0200 Subject: [PATCH 3/5] Few tweaks --- .../Connection/ChannelProvider.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs b/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs index a88cff7fe..1569a01f8 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs @@ -87,7 +87,7 @@ public async ValueTask GetPublishChannel(CancellationToken while (true) { var existing = Interlocked.CompareExchange(ref publishChannel, null, null); - if (existing is not null && !existing.IsClosed) + if (existing is { IsOpen: true }) { return existing; } @@ -114,7 +114,7 @@ public async ValueTask GetPublishChannel(CancellationToken public ValueTask ReturnPublishChannel(ConfirmsAwareChannel channel, CancellationToken cancellationToken = default) { - if (!channel.IsClosed) + if (channel.IsOpen) { return ValueTask.CompletedTask; } From 993abbb60ba3c0ff8a10ff6357ce1d9966772abe Mon Sep 17 00:00:00 2001 From: Daniel Marbach Date: Fri, 6 Jun 2025 17:59:46 +0200 Subject: [PATCH 4/5] Simplify implementation --- .../Connection/ChannelProvider.cs | 61 ++++++++++++------- 1 file changed, 38 insertions(+), 23 deletions(-) diff --git a/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs b/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs index 1569a01f8..71176eda7 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs @@ -84,43 +84,57 @@ protected virtual void FireAndForget(Func action, Cance public async ValueTask GetPublishChannel(CancellationToken cancellationToken = default) { - while (true) + if (publishChannel is { IsOpen: true }) { - var existing = Interlocked.CompareExchange(ref publishChannel, null, null); - if (existing is { IsOpen: true }) + return publishChannel; + } + + try + { + await publishChannelSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false); + if (publishChannel is { IsOpen: true }) { - return existing; + return publishChannel; } - var newChannel = new ConfirmsAwareChannel(connection!, routingTopology); - await newChannel.Initialize(cancellationToken).ConfigureAwait(false); - - var expected = existing; - var prev = Interlocked.CompareExchange(ref publishChannel, newChannel, expected); - - if (ReferenceEquals(prev, expected)) + var oldChannel = publishChannel; + if (oldChannel is not null) { - if (expected is not null) - { - await expected.DisposeAsync().ConfigureAwait(false); - } - - return newChannel; + await oldChannel.DisposeAsync().ConfigureAwait(false); } - await newChannel.DisposeAsync().ConfigureAwait(false); + var newChannel = new ConfirmsAwareChannel(connection!, routingTopology); + await newChannel.Initialize(cancellationToken).ConfigureAwait(false); + publishChannel = newChannel; + return newChannel; + } + finally + { + publishChannelSemaphore.Release(); } } - public ValueTask ReturnPublishChannel(ConfirmsAwareChannel channel, CancellationToken cancellationToken = default) + public async ValueTask ReturnPublishChannel(ConfirmsAwareChannel channel, CancellationToken cancellationToken = default) { if (channel.IsOpen) { - return ValueTask.CompletedTask; + return; } - var prev = Interlocked.CompareExchange(ref publishChannel, null, channel); - return ReferenceEquals(prev, channel) ? channel.DisposeAsync() : ValueTask.CompletedTask; + try + { + await publishChannelSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false); + + if (ReferenceEquals(publishChannel, channel)) + { + await channel.DisposeAsync().ConfigureAwait(false); + publishChannel = null; + } + } + finally + { + publishChannelSemaphore.Release(); + } } #pragma warning disable PS0018 @@ -149,7 +163,8 @@ public async ValueTask DisposeAsync() readonly CancellationTokenSource stoppingTokenSource = new(); volatile IConnection? connection; - ConfirmsAwareChannel? publishChannel; + readonly SemaphoreSlim publishChannelSemaphore = new(1, 1); + volatile ConfirmsAwareChannel? publishChannel; bool disposed; static readonly ILog Logger = LogManager.GetLogger(typeof(ChannelProvider)); From 0fc847399fe7f528cd9014d88799322a419e3ccf Mon Sep 17 00:00:00 2001 From: Daniel Marbach Date: Tue, 10 Jun 2025 17:57:21 +0200 Subject: [PATCH 5/5] Remove null-forgiving Co-authored-by: Brandon Ording --- .../Connection/ChannelProvider.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs b/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs index 71176eda7..9cd91dfc2 100644 --- a/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs +++ b/src/NServiceBus.Transport.RabbitMQ/Connection/ChannelProvider.cs @@ -103,7 +103,7 @@ public async ValueTask GetPublishChannel(CancellationToken await oldChannel.DisposeAsync().ConfigureAwait(false); } - var newChannel = new ConfirmsAwareChannel(connection!, routingTopology); + var newChannel = new ConfirmsAwareChannel(connection, routingTopology); await newChannel.Initialize(cancellationToken).ConfigureAwait(false); publishChannel = newChannel; return newChannel;