Skip to content

Commit 735bbca

Browse files
lukebakkenlukebakken
authored
Merge pull request #1671 from rabbitmq/rabbitmq-dotnet-client-1669-followup
Follow-up to #1669 - per-channel dispatch concurrency
2 parents 624cf2e + 601f501 commit 735bbca

15 files changed

+87
-44
lines changed

‎projects/RabbitMQ.Client/PublicAPI.Shipped.txt

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -892,7 +892,6 @@ static RabbitMQ.Client.IChannelExtensions.BasicPublishAsync(this RabbitMQ.Client
892892
static RabbitMQ.Client.IChannelExtensions.BasicPublishAsync(this RabbitMQ.Client.IChannel! channel, string! exchange, string! routingKey, bool mandatory, System.ReadOnlyMemory<byte> body, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
893893
static RabbitMQ.Client.IChannelExtensions.BasicPublishAsync(this RabbitMQ.Client.IChannel! channel, string! exchange, string! routingKey, System.ReadOnlyMemory<byte> body, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
894894
RabbitMQ.Client.IChannel.ConfirmSelectAsync(bool trackConfirmations = true, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
895-
const RabbitMQ.Client.ConnectionFactory.DefaultConsumerDispatchConcurrency = 1 -> ushort
896-
RabbitMQ.Client.IConnection.CreateChannelAsync(ushort consumerDispatchConcurrency, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IChannel!>!
897-
static RabbitMQ.Client.IConnectionExtensions.CreateChannelAsync(this RabbitMQ.Client.IConnection! connection, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IChannel!>!
895+
const RabbitMQ.Client.Constants.DefaultConsumerDispatchConcurrency = 1 -> ushort
898896
readonly RabbitMQ.Client.ConnectionConfig.ConsumerDispatchConcurrency -> ushort
897+
RabbitMQ.Client.IConnection.CreateChannelAsync(ushort? consumerDispatchConcurrency = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IChannel!>!

‎projects/RabbitMQ.Client/client/api/ConnectionFactory.cs

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -92,11 +92,6 @@ namespace RabbitMQ.Client
9292
///hosts with an empty name are not addressable. </para></remarks>
9393
public sealed class ConnectionFactory : ConnectionFactoryBase, IConnectionFactory
9494
{
95-
/// <summary>
96-
/// Default value for consumer dispatch concurrency.
97-
/// </summary>
98-
public const ushort DefaultConsumerDispatchConcurrency = 1;
99-
10095
/// <summary>
10196
/// Default value for the desired maximum channel number. Default: 2047.
10297
/// </summary>
@@ -180,7 +175,7 @@ public sealed class ConnectionFactory : ConnectionFactoryBase, IConnectionFactor
180175
/// </summary>
181176
/// <remarks>For concurrency greater than one this removes the guarantee that consumers handle messages in the order they receive them.
182177
/// In addition to that consumers need to be thread/concurrency safe.</remarks>
183-
public ushort ConsumerDispatchConcurrency { get; set; } = DefaultConsumerDispatchConcurrency;
178+
public ushort ConsumerDispatchConcurrency { get; set; } = Constants.DefaultConsumerDispatchConcurrency;
184179

185180
/// <summary>The host to connect to.</summary>
186181
public string HostName { get; set; } = "localhost";

‎projects/RabbitMQ.Client/client/api/IConnection.cs

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -239,12 +239,13 @@ Task CloseAsync(ushort reasonCode, string reasonText, TimeSpan timeout, bool abo
239239
/// will be offloaded to the worker thread pool so it is important to choose the value for the concurrency wisely to avoid thread pool overloading.
240240
/// <see cref="IAsyncBasicConsumer"/> can handle concurrency much more efficiently due to the non-blocking nature of the consumer.
241241
///
242-
/// Defaults to <see cref="IConnectionFactory.ConsumerDispatchConcurrency"/>.
242+
/// Defaults to <c>null</c>, which will use the value from <see cref="IConnectionFactory.ConsumerDispatchConcurrency"/>
243243
///
244244
/// For concurrency greater than one this removes the guarantee that consumers handle messages in the order they receive them.
245245
/// In addition to that consumers need to be thread/concurrency safe.
246246
/// </param>
247247
/// <param name="cancellationToken">Cancellation token</param>
248-
Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency, CancellationToken cancellationToken = default);
248+
Task<IChannel> CreateChannelAsync(ushort? consumerDispatchConcurrency = null,
249+
CancellationToken cancellationToken = default);
249250
}
250251
}

‎projects/RabbitMQ.Client/client/api/IConnectionExtensions.cs

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -7,12 +7,6 @@ namespace RabbitMQ.Client
77
{
88
public static class IConnectionExtensions
99
{
10-
/// <summary>
11-
/// Asynchronously create and return a fresh channel, session, and channel.
12-
/// </summary>
13-
public static Task<IChannel> CreateChannelAsync(this IConnection connection, CancellationToken cancellationToken = default) =>
14-
connection.CreateChannelAsync(ConnectionFactory.DefaultConsumerDispatchConcurrency, cancellationToken);
15-
1610
/// <summary>
1711
/// Asynchronously close this connection and all its channels.
1812
/// </summary>

‎projects/RabbitMQ.Client/client/framing/Channel.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ namespace RabbitMQ.Client.Framing.Impl
3838
{
3939
internal class Channel : ChannelBase
4040
{
41-
public Channel(ConnectionConfig config, ISession session, ushort consumerDispatchConcurrency)
41+
public Channel(ConnectionConfig config, ISession session, ushort? consumerDispatchConcurrency = null)
4242
: base(config, session, consumerDispatchConcurrency)
4343
{
4444
}

‎projects/RabbitMQ.Client/client/framing/Constants.cs

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,5 +83,13 @@ public static class Constants
8383
public const int NotImplemented = 540;
8484
///<summary>(= 541)</summary>
8585
public const int InternalError = 541;
86+
87+
/// <summary>
88+
/// The default consumer dispatch concurrency. See <see cref="IConnectionFactory.ConsumerDispatchConcurrency"/>
89+
/// to set this value for every channel created on a connection,
90+
/// and <see cref="IConnection.CreateChannelAsync(ushort?, System.Threading.CancellationToken)"/>
91+
/// for setting this value for a particular channel.
92+
/// </summary>
93+
public const ushort DefaultConsumerDispatchConcurrency = 1;
8694
}
8795
}

‎projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -240,12 +240,14 @@ await CloseInnerConnectionAsync()
240240
}
241241
}
242242

243-
public async Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency, CancellationToken cancellationToken = default)
243+
public async Task<IChannel> CreateChannelAsync(ushort? consumerDispatchConcurrency = null,
244+
CancellationToken cancellationToken = default)
244245
{
245246
EnsureIsOpen();
246-
RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(consumerDispatchConcurrency, cancellationToken)
247+
ushort cdc = consumerDispatchConcurrency.GetValueOrDefault(_config.ConsumerDispatchConcurrency);
248+
RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(cdc, cancellationToken)
247249
.ConfigureAwait(false);
248-
AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel, consumerDispatchConcurrency);
250+
AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel, cdc);
249251
await RecordChannelAsync(channel, channelsSemaphoreHeld: false, cancellationToken: cancellationToken)
250252
.ConfigureAwait(false);
251253
return channel;

‎projects/RabbitMQ.Client/client/impl/ChannelBase.cs

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -73,10 +73,12 @@ internal abstract class ChannelBase : IChannel, IRecoverable
7373

7474
internal readonly IConsumerDispatcher ConsumerDispatcher;
7575

76-
protected ChannelBase(ConnectionConfig config, ISession session, ushort consumerDispatchConcurrency)
76+
protected ChannelBase(ConnectionConfig config, ISession session,
77+
ushort? perChannelConsumerDispatchConcurrency = null)
7778
{
7879
ContinuationTimeout = config.ContinuationTimeout;
79-
ConsumerDispatcher = new AsyncConsumerDispatcher(this, consumerDispatchConcurrency);
80+
ConsumerDispatcher = new AsyncConsumerDispatcher(this,
81+
perChannelConsumerDispatchConcurrency.GetValueOrDefault(config.ConsumerDispatchConcurrency));
8082
Action<Exception, string> onException = (exception, context) =>
8183
OnCallbackException(CallbackExceptionEventArgs.Build(exception, context));
8284
_basicAcksWrapper = new EventingWrapper<BasicAckEventArgs>("OnBasicAck", onException);

‎projects/RabbitMQ.Client/client/impl/Connection.cs

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,7 @@ internal Connection(ConnectionConfig config, IFrameHandler frameHandler)
7272

7373
_sessionManager = new SessionManager(this, 0, config.MaxInboundMessageBodySize);
7474
_session0 = new MainSession(this, config.MaxInboundMessageBodySize);
75-
_channel0 = new Channel(_config, _session0, ConnectionFactory.DefaultConsumerDispatchConcurrency); ;
75+
_channel0 = new Channel(_config, _session0);
7676

7777
ClientProperties = new Dictionary<string, object?>(_config.ClientProperties)
7878
{
@@ -253,7 +253,8 @@ await CloseAsync(ea, true,
253253
}
254254
}
255255

256-
public Task<IChannel> CreateChannelAsync(ushort consumerDispatchConcurrency, CancellationToken cancellationToken = default)
256+
public Task<IChannel> CreateChannelAsync(ushort? consumerDispatchConcurrency = null,
257+
CancellationToken cancellationToken = default)
257258
{
258259
EnsureIsOpen();
259260
ISession session = CreateSession();

‎projects/RabbitMQ.Client/client/impl/ConsumerDispatching/ConsumerDispatcherChannelBase.cs

Lines changed: 9 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -14,44 +14,42 @@ internal abstract class ConsumerDispatcherChannelBase : ConsumerDispatcherBase,
1414
protected readonly ChannelReader<WorkStruct> _reader;
1515
private readonly ChannelWriter<WorkStruct> _writer;
1616
private readonly Task _worker;
17+
private readonly ushort _concurrency;
1718
private bool _quiesce = false;
1819
private bool _disposed;
1920

2021
internal ConsumerDispatcherChannelBase(ChannelBase channel, ushort concurrency)
2122
{
2223
_channel = channel;
24+
_concurrency = concurrency;
2325
var workChannel = Channel.CreateUnbounded<WorkStruct>(new UnboundedChannelOptions
2426
{
25-
SingleReader = concurrency == 1,
27+
SingleReader = _concurrency == 1,
2628
SingleWriter = false,
2729
AllowSynchronousContinuations = false
2830
});
2931
_reader = workChannel.Reader;
3032
_writer = workChannel.Writer;
3133

3234
Func<Task> loopStart = ProcessChannelAsync;
33-
if (concurrency == 1)
35+
if (_concurrency == 1)
3436
{
3537
_worker = Task.Run(loopStart);
3638
}
3739
else
3840
{
39-
var tasks = new Task[concurrency];
40-
for (int i = 0; i < concurrency; i++)
41+
var tasks = new Task[_concurrency];
42+
for (int i = 0; i < _concurrency; i++)
4143
{
4244
tasks[i] = Task.Run(loopStart);
4345
}
4446
_worker = Task.WhenAll(tasks);
4547
}
4648
}
4749

48-
public bool IsShutdown
49-
{
50-
get
51-
{
52-
return _quiesce;
53-
}
54-
}
50+
public bool IsShutdown => _quiesce;
51+
52+
public ushort Concurrency => _concurrency;
5553

5654
public ValueTask HandleBasicConsumeOkAsync(IAsyncBasicConsumer consumer, string consumerTag, CancellationToken cancellationToken)
5755
{

‎projects/RabbitMQ.Client/client/impl/ConsumerDispatching/IConsumerDispatcher.cs

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,8 @@ internal interface IConsumerDispatcher : IDisposable
4141

4242
bool IsShutdown { get; }
4343

44+
ushort Concurrency { get; }
45+
4446
IAsyncBasicConsumer GetAndRemoveConsumer(string tag);
4547

4648
ValueTask HandleBasicConsumeOkAsync(IAsyncBasicConsumer consumer, string consumerTag, CancellationToken cancellationToken);

‎projects/Test/Common/IntegrationFixture.cs

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -71,7 +71,7 @@ public abstract class IntegrationFixture : IAsyncLifetime
7171
protected readonly ITestOutputHelper _output;
7272
protected readonly string _testDisplayName;
7373

74-
protected readonly ushort _consumerDispatchConcurrency = 1;
74+
protected readonly ushort _consumerDispatchConcurrency = Constants.DefaultConsumerDispatchConcurrency;
7575
protected readonly bool _openChannel = true;
7676

7777
public static readonly TimeSpan ShortSpan;
@@ -109,7 +109,7 @@ static IntegrationFixture()
109109
}
110110

111111
public IntegrationFixture(ITestOutputHelper output,
112-
ushort consumerDispatchConcurrency = 1,
112+
ushort consumerDispatchConcurrency = Constants.DefaultConsumerDispatchConcurrency,
113113
bool openChannel = true)
114114
{
115115
_consumerDispatchConcurrency = consumerDispatchConcurrency;
@@ -143,8 +143,7 @@ public virtual async Task InitializeAsync()
143143
*/
144144
if (_connFactory == null)
145145
{
146-
_connFactory = CreateConnectionFactory();
147-
_connFactory.ConsumerDispatchConcurrency = _consumerDispatchConcurrency;
146+
_connFactory = CreateConnectionFactory(_consumerDispatchConcurrency);
148147
}
149148

150149
if (_conn == null)
@@ -517,13 +516,15 @@ protected static async Task WaitAsync(TaskCompletionSource<bool> tcs, TimeSpan t
517516
}
518517
}
519518

520-
protected ConnectionFactory CreateConnectionFactory()
519+
protected ConnectionFactory CreateConnectionFactory(
520+
ushort consumerDispatchConcurrency = Constants.DefaultConsumerDispatchConcurrency)
521521
{
522522
return new ConnectionFactory
523523
{
524524
ClientProvidedName = $"{_testDisplayName}:{Util.Now}:{GetConnectionIdx()}",
525525
ContinuationTimeout = WaitSpan,
526526
HandshakeContinuationTimeout = WaitSpan,
527+
ConsumerDispatchConcurrency = consumerDispatchConcurrency
527528
};
528529
}
529530

‎projects/Test/Integration/TestAsyncConsumer.cs

Lines changed: 34 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,23 +36,28 @@
3636
using System.Threading.Tasks;
3737
using RabbitMQ.Client;
3838
using RabbitMQ.Client.Events;
39+
using RabbitMQ.Client.Impl;
3940
using Xunit;
4041
using Xunit.Abstractions;
4142

4243
namespace Test.Integration
4344
{
4445
public class TestAsyncConsumer : IntegrationFixture
4546
{
47+
private const ushort ConsumerDispatchConcurrency = 2;
48+
4649
private readonly ShutdownEventArgs _closeArgs = new ShutdownEventArgs(ShutdownInitiator.Application, Constants.ReplySuccess, "normal shutdown");
4750

4851
public TestAsyncConsumer(ITestOutputHelper output)
49-
: base(output, consumerDispatchConcurrency: 2)
52+
: base(output, consumerDispatchConcurrency: ConsumerDispatchConcurrency)
5053
{
5154
}
5255

5356
[Fact]
5457
public async Task TestBasicRoundtripConcurrent()
5558
{
59+
await ValidateConsumerDispatchConcurrency();
60+
5661
AddCallbackExceptionHandlers();
5762
_channel.DefaultConsumer = new DefaultAsyncConsumer(_channel, "_channel,", _output);
5863

@@ -146,6 +151,8 @@ public async Task TestBasicRoundtripConcurrent()
146151
[Fact]
147152
public async Task TestBasicRoundtripConcurrentManyMessages()
148153
{
154+
await ValidateConsumerDispatchConcurrency();
155+
149156
AddCallbackExceptionHandlers();
150157
_channel.DefaultConsumer = new DefaultAsyncConsumer(_channel, "_channel,", _output);
151158

@@ -323,6 +330,8 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
323330
[Fact]
324331
public async Task TestBasicRejectAsync()
325332
{
333+
await ValidateConsumerDispatchConcurrency();
334+
326335
string queueName = GenerateQueueName();
327336

328337
var publishSyncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -421,6 +430,8 @@ await _channel.BasicConsumeAsync(queue: queueName, autoAck: false,
421430
[Fact]
422431
public async Task TestBasicAckAsync()
423432
{
433+
await ValidateConsumerDispatchConcurrency();
434+
424435
string queueName = GenerateQueueName();
425436

426437
const int messageCount = 1024;
@@ -488,6 +499,8 @@ await _channel.BasicConsumeAsync(queue: queueName, autoAck: false,
488499
[Fact]
489500
public async Task TestBasicNackAsync()
490501
{
502+
await ValidateConsumerDispatchConcurrency();
503+
491504
var publishSyncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
492505

493506
_conn.ConnectionShutdown += (o, ea) =>
@@ -561,6 +574,8 @@ await _channel.BasicConsumeAsync(queue: queueName, autoAck: false,
561574
[Fact]
562575
public async Task TestDeclarationOfManyAutoDeleteQueuesWithTransientConsumer()
563576
{
577+
await ValidateConsumerDispatchConcurrency();
578+
564579
AssertRecordedQueues((RabbitMQ.Client.Framing.Impl.AutorecoveringConnection)_conn, 0);
565580
var tasks = new List<Task>();
566581
for (int i = 0; i < 256; i++)
@@ -581,6 +596,8 @@ public async Task TestDeclarationOfManyAutoDeleteQueuesWithTransientConsumer()
581596
[Fact]
582597
public async Task TestCreateChannelWithinAsyncConsumerCallback_GH650()
583598
{
599+
await ValidateConsumerDispatchConcurrency();
600+
584601
string exchangeName = GenerateExchangeName();
585602
string queue1Name = GenerateQueueName();
586603
string queue2Name = GenerateQueueName();
@@ -650,6 +667,8 @@ await innerChannel.BasicPublishAsync(exchangeName, queue2Name,
650667
[Fact]
651668
public async Task TestCloseWithinEventHandler_GH1567()
652669
{
670+
await ValidateConsumerDispatchConcurrency();
671+
653672
var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
654673

655674
QueueDeclareOk q = await _channel.QueueDeclareAsync();
@@ -679,6 +698,20 @@ await _channel.BasicPublishAsync(exchange: string.Empty, routingKey: queueName,
679698
Assert.True(await tcs.Task);
680699
}
681700

701+
private async Task ValidateConsumerDispatchConcurrency()
702+
{
703+
ushort expectedConsumerDispatchConcurrency = (ushort)S_Random.Next(3, 10);
704+
AutorecoveringChannel autorecoveringChannel = (AutorecoveringChannel)_channel;
705+
Assert.Equal(ConsumerDispatchConcurrency, autorecoveringChannel.ConsumerDispatcher.Concurrency);
706+
Assert.Equal(_consumerDispatchConcurrency, autorecoveringChannel.ConsumerDispatcher.Concurrency);
707+
using (IChannel ch = await _conn.CreateChannelAsync(
708+
consumerDispatchConcurrency: expectedConsumerDispatchConcurrency))
709+
{
710+
AutorecoveringChannel ach = (AutorecoveringChannel)ch;
711+
Assert.Equal(expectedConsumerDispatchConcurrency, ach.ConsumerDispatcher.Concurrency);
712+
}
713+
}
714+
682715
private static void SetException(Exception ex, params TaskCompletionSource<bool>[] tcsAry)
683716
{
684717
foreach (TaskCompletionSource<bool> tcs in tcsAry)

‎projects/Test/Integration/TestAsyncEventingBasicConsumer.cs

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,13 +34,16 @@
3434
using System.Threading.Tasks;
3535
using RabbitMQ.Client;
3636
using RabbitMQ.Client.Events;
37+
using RabbitMQ.Client.Impl;
3738
using Xunit;
3839
using Xunit.Abstractions;
3940

4041
namespace Test.Integration
4142
{
4243
public class TestAsyncEventingBasicConsumer : IntegrationFixture
4344
{
45+
private const ushort ConsumerDispatchConcurrency = 2;
46+
4447
private readonly CancellationTokenSource _cts = new CancellationTokenSource(ShortSpan);
4548
private readonly CancellationTokenRegistration _ctr;
4649
private readonly TaskCompletionSource<bool> _onCallbackExceptionTcs =
@@ -49,7 +52,7 @@ public class TestAsyncEventingBasicConsumer : IntegrationFixture
4952
new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
5053

5154
public TestAsyncEventingBasicConsumer(ITestOutputHelper output)
52-
: base(output, consumerDispatchConcurrency: 2)
55+
: base(output, consumerDispatchConcurrency: ConsumerDispatchConcurrency)
5356
{
5457
_ctr = _cts.Token.Register(OnTokenCanceled);
5558
}
@@ -81,6 +84,10 @@ private Task AsyncConsumerOnReceived(object sender, BasicDeliverEventArgs @event
8184
[Fact]
8285
public async Task TestAsyncEventingBasicConsumer_GH1038()
8386
{
87+
AutorecoveringChannel autorecoveringChannel = (AutorecoveringChannel)_channel;
88+
Assert.Equal(ConsumerDispatchConcurrency, autorecoveringChannel.ConsumerDispatcher.Concurrency);
89+
Assert.Equal(_consumerDispatchConcurrency, autorecoveringChannel.ConsumerDispatcher.Concurrency);
90+
8491
string exchangeName = GenerateExchangeName();
8592
string queueName = GenerateQueueName();
8693
string routingKey = string.Empty;

0 commit comments

Comments
 (0)