Skip to content

Commit 0682db6

Browse files
lukebakkenlukebakken
committed
* Make handling of server-originated Basic.Cancel async to try to address this failure - https://github.com/rabbitmq/rabbitmq-dotnet-client/actions/runs/8083493731/job/22086889719
* Make more code async * Fix when `Dispose` is called for subclasses of `AsyncRpcContinuation` I'm not exactly sure why this test sometimes fails 🤔
1 parent de8e1e4 commit 0682db6

File tree

11 files changed

+320
-129
lines changed

11 files changed

+320
-129
lines changed

‎projects/Benchmarks/ConsumerDispatching/ConsumerDispatcher.cs‎

Lines changed: 27 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
using System;
22
using System.Threading;
3+
using System.Threading.Tasks;
34
using BenchmarkDotNet.Attributes;
45
using RabbitMQ.Client;
56
using RabbitMQ.Client.ConsumerDispatching;
@@ -8,7 +9,7 @@ namespace RabbitMQ.Benchmarks
89
{
910
[Config(typeof(Config))]
1011
[BenchmarkCategory("ConsumerDispatcher")]
11-
internal class ConsumerDispatcherBase
12+
public class ConsumerDispatcherBase
1213
{
1314
protected static readonly ManualResetEventSlim _autoResetEvent = new ManualResetEventSlim(false);
1415

@@ -19,18 +20,16 @@ internal class ConsumerDispatcherBase
1920
protected readonly string _exchange = "Exchange";
2021
protected readonly string _routingKey = "RoutingKey";
2122
protected readonly ReadOnlyBasicProperties _properties = new ReadOnlyBasicProperties();
22-
protected readonly RentedMemory _body;
23+
protected readonly byte[] _body = new byte[512];
2324

2425
public ConsumerDispatcherBase()
2526
{
2627
var r = new Random();
27-
byte[] body = new byte[512];
28-
r.NextBytes(body);
29-
_body = new RentedMemory(body);
28+
r.NextBytes(_body);
3029
}
3130
}
3231

33-
internal class BasicDeliverConsumerDispatching : ConsumerDispatcherBase
32+
public class BasicDeliverConsumerDispatching : ConsumerDispatcherBase
3433
{
3534
[Params(1, 30)]
3635
public int Count { get; set; }
@@ -39,41 +38,49 @@ internal class BasicDeliverConsumerDispatching : ConsumerDispatcherBase
3938
public int Concurrency { get; set; }
4039

4140
[GlobalSetup(Target = nameof(AsyncConsumerDispatcher))]
42-
public void SetUpAsyncConsumer()
41+
public async Task SetUpAsyncConsumer()
4342
{
4443
_consumer.Count = Count;
4544
_dispatcher = new AsyncConsumerDispatcher(null, Concurrency);
46-
_dispatcher.HandleBasicConsumeOk(_consumer, _consumerTag);
45+
await _dispatcher.HandleBasicConsumeOkAsync(_consumer, _consumerTag, CancellationToken.None);
4746
}
4847

4948
[Benchmark]
50-
public void AsyncConsumerDispatcher()
49+
public async Task AsyncConsumerDispatcher()
5150
{
52-
for (int i = 0; i < Count; i++)
51+
using (RentedMemory body = new RentedMemory(_body))
5352
{
54-
_dispatcher.HandleBasicDeliver(_consumerTag, _deliveryTag, false, _exchange, _routingKey, _properties, _body);
53+
for (int i = 0; i < Count; i++)
54+
{
55+
await _dispatcher.HandleBasicDeliverAsync(_consumerTag, _deliveryTag, false, _exchange, _routingKey, _properties, body,
56+
CancellationToken.None);
57+
}
58+
_autoResetEvent.Wait();
59+
_autoResetEvent.Reset();
5560
}
56-
_autoResetEvent.Wait();
57-
_autoResetEvent.Reset();
5861
}
5962

6063
[GlobalSetup(Target = nameof(ConsumerDispatcher))]
61-
public void SetUpConsumer()
64+
public async Task SetUpConsumer()
6265
{
6366
_consumer.Count = Count;
6467
_dispatcher = new ConsumerDispatcher(null, Concurrency);
65-
_dispatcher.HandleBasicConsumeOk(_consumer, _consumerTag);
68+
await _dispatcher.HandleBasicConsumeOkAsync(_consumer, _consumerTag, CancellationToken.None);
6669
}
6770

6871
[Benchmark]
69-
public void ConsumerDispatcher()
72+
public async Task ConsumerDispatcher()
7073
{
71-
for (int i = 0; i < Count; i++)
74+
using (RentedMemory body = new RentedMemory(_body))
7275
{
73-
_dispatcher.HandleBasicDeliver(_consumerTag, _deliveryTag, false, _exchange, _routingKey, _properties, _body);
76+
for (int i = 0; i < Count; i++)
77+
{
78+
await _dispatcher.HandleBasicDeliverAsync(_consumerTag, _deliveryTag, false, _exchange, _routingKey, _properties, body,
79+
CancellationToken.None);
80+
}
81+
_autoResetEvent.Wait();
82+
_autoResetEvent.Reset();
7483
}
75-
_autoResetEvent.Wait();
76-
_autoResetEvent.Reset();
7784
}
7885
}
7986
}

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

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -78,13 +78,13 @@ protected override Task<bool> DispatchCommandAsync(IncomingCommand cmd, Cancella
7878
{
7979
case ProtocolCommandId.BasicCancel:
8080
{
81-
HandleBasicCancel(in cmd);
82-
return Task.FromResult(true);
81+
// Note: always returns true
82+
return HandleBasicCancelAsync(cmd, cancellationToken);
8383
}
8484
case ProtocolCommandId.BasicDeliver:
8585
{
86-
HandleBasicDeliver(in cmd);
87-
return Task.FromResult(true);
86+
// Note: always returns true
87+
return HandleBasicDeliverAsync(cmd, cancellationToken);
8888
}
8989
case ProtocolCommandId.BasicAck:
9090
{
@@ -108,8 +108,8 @@ protected override Task<bool> DispatchCommandAsync(IncomingCommand cmd, Cancella
108108
}
109109
case ProtocolCommandId.ChannelCloseOk:
110110
{
111-
HandleChannelCloseOk(in cmd);
112-
return Task.FromResult(true);
111+
// Note: always returns true
112+
return HandleChannelCloseOkAsync(cmd, cancellationToken);
113113
}
114114
case ProtocolCommandId.ChannelFlow:
115115
{
@@ -128,8 +128,8 @@ protected override Task<bool> DispatchCommandAsync(IncomingCommand cmd, Cancella
128128
}
129129
case ProtocolCommandId.ConnectionSecure:
130130
{
131-
HandleConnectionSecure(in cmd);
132-
return Task.FromResult(true);
131+
// Note: always returns true
132+
return HandleConnectionSecureAsync(cmd);
133133
}
134134
case ProtocolCommandId.ConnectionStart:
135135
{
@@ -138,8 +138,8 @@ protected override Task<bool> DispatchCommandAsync(IncomingCommand cmd, Cancella
138138
}
139139
case ProtocolCommandId.ConnectionTune:
140140
{
141-
HandleConnectionTune(in cmd);
142-
return Task.FromResult(true);
141+
// Note: always returns true
142+
return HandleConnectionTuneAsync(cmd);
143143
}
144144
case ProtocolCommandId.ConnectionUnblocked:
145145
{

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

Lines changed: 26 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@
4141

4242
namespace RabbitMQ.Client.Impl
4343
{
44-
internal abstract class AsyncRpcContinuation<T> : IRpcContinuation, IDisposable
44+
internal abstract class AsyncRpcContinuation<T> : IRpcContinuation
4545
{
4646
private readonly CancellationTokenSource _cancellationTokenSource;
4747
private readonly CancellationTokenRegistration _cancellationTokenRegistration;
@@ -101,7 +101,7 @@ public ConfiguredTaskAwaitable<T>.ConfiguredTaskAwaiter GetAwaiter()
101101
return _tcsConfiguredTaskAwaitable.GetAwaiter();
102102
}
103103

104-
public abstract void HandleCommand(in IncomingCommand cmd);
104+
public abstract Task HandleCommandAsync(IncomingCommand cmd);
105105

106106
public virtual void HandleChannelShutdown(ShutdownEventArgs reason)
107107
{
@@ -135,7 +135,7 @@ public ConnectionSecureOrTuneAsyncRpcContinuation(TimeSpan continuationTimeout)
135135
{
136136
}
137137

138-
public override void HandleCommand(in IncomingCommand cmd)
138+
public override Task HandleCommandAsync(IncomingCommand cmd)
139139
{
140140
try
141141
{
@@ -156,6 +156,8 @@ public override void HandleCommand(in IncomingCommand cmd)
156156
{
157157
_tcs.SetException(new InvalidOperationException($"Received unexpected command of type {cmd.CommandId}!"));
158158
}
159+
160+
return Task.CompletedTask;
159161
}
160162
finally
161163
{
@@ -173,7 +175,7 @@ public SimpleAsyncRpcContinuation(ProtocolCommandId expectedCommandId, TimeSpan
173175
_expectedCommandId = expectedCommandId;
174176
}
175177

176-
public override void HandleCommand(in IncomingCommand cmd)
178+
public override Task HandleCommandAsync(IncomingCommand cmd)
177179
{
178180
try
179181
{
@@ -185,6 +187,8 @@ public override void HandleCommand(in IncomingCommand cmd)
185187
{
186188
_tcs.SetException(new InvalidOperationException($"Received unexpected command of type {cmd.CommandId}!"));
187189
}
190+
191+
return Task.CompletedTask;
188192
}
189193
finally
190194
{
@@ -205,7 +209,7 @@ public BasicCancelAsyncRpcContinuation(string consumerTag, IConsumerDispatcher c
205209
_consumerDispatcher = consumerDispatcher;
206210
}
207211

208-
public override void HandleCommand(in IncomingCommand cmd)
212+
public override async Task HandleCommandAsync(IncomingCommand cmd)
209213
{
210214
try
211215
{
@@ -214,7 +218,8 @@ public override void HandleCommand(in IncomingCommand cmd)
214218
var method = new Client.Framing.Impl.BasicCancelOk(cmd.MethodSpan);
215219
_tcs.TrySetResult(true);
216220
Debug.Assert(_consumerTag == method._consumerTag);
217-
_consumerDispatcher.HandleBasicCancelOk(_consumerTag);
221+
await _consumerDispatcher.HandleBasicCancelOkAsync(_consumerTag, CancellationToken)
222+
.ConfigureAwait(false);
218223
}
219224
else
220225
{
@@ -240,15 +245,16 @@ public BasicConsumeAsyncRpcContinuation(IBasicConsumer consumer, IConsumerDispat
240245
_consumerDispatcher = consumerDispatcher;
241246
}
242247

243-
public override void HandleCommand(in IncomingCommand cmd)
248+
public override async Task HandleCommandAsync(IncomingCommand cmd)
244249
{
245250
try
246251
{
247252
if (cmd.CommandId == ProtocolCommandId.BasicConsumeOk)
248253
{
249254
var method = new Client.Framing.Impl.BasicConsumeOk(cmd.MethodSpan);
250255
_tcs.TrySetResult(method._consumerTag);
251-
_consumerDispatcher.HandleBasicConsumeOk(_consumer, method._consumerTag);
256+
await _consumerDispatcher.HandleBasicConsumeOkAsync(_consumer, method._consumerTag, CancellationToken)
257+
.ConfigureAwait(false);
252258
}
253259
else
254260
{
@@ -272,7 +278,7 @@ public BasicGetAsyncRpcContinuation(Func<ulong, ulong> adjustDeliveryTag, TimeSp
272278
_adjustDeliveryTag = adjustDeliveryTag;
273279
}
274280

275-
public override void HandleCommand(in IncomingCommand cmd)
281+
public override Task HandleCommandAsync(IncomingCommand cmd)
276282
{
277283
try
278284
{
@@ -300,6 +306,8 @@ public override void HandleCommand(in IncomingCommand cmd)
300306
{
301307
_tcs.SetException(new InvalidOperationException($"Received unexpected command of type {cmd.CommandId}!"));
302308
}
309+
310+
return Task.CompletedTask;
303311
}
304312
finally
305313
{
@@ -389,7 +397,7 @@ public QueueDeclareAsyncRpcContinuation(TimeSpan continuationTimeout) : base(con
389397
{
390398
}
391399

392-
public override void HandleCommand(in IncomingCommand cmd)
400+
public override Task HandleCommandAsync(IncomingCommand cmd)
393401
{
394402
try
395403
{
@@ -403,6 +411,8 @@ public override void HandleCommand(in IncomingCommand cmd)
403411
{
404412
_tcs.SetException(new InvalidOperationException($"Received unexpected command of type {cmd.CommandId}!"));
405413
}
414+
415+
return Task.CompletedTask;
406416
}
407417
finally
408418
{
@@ -433,7 +443,7 @@ public QueueDeleteAsyncRpcContinuation(TimeSpan continuationTimeout) : base(cont
433443
{
434444
}
435445

436-
public override void HandleCommand(in IncomingCommand cmd)
446+
public override Task HandleCommandAsync(IncomingCommand cmd)
437447
{
438448
try
439449
{
@@ -446,6 +456,8 @@ public override void HandleCommand(in IncomingCommand cmd)
446456
{
447457
_tcs.SetException(new InvalidOperationException($"Received unexpected command of type {cmd.CommandId}!"));
448458
}
459+
460+
return Task.CompletedTask;
449461
}
450462
finally
451463
{
@@ -460,7 +472,7 @@ public QueuePurgeAsyncRpcContinuation(TimeSpan continuationTimeout) : base(conti
460472
{
461473
}
462474

463-
public override void HandleCommand(in IncomingCommand cmd)
475+
public override Task HandleCommandAsync(IncomingCommand cmd)
464476
{
465477
try
466478
{
@@ -473,6 +485,8 @@ public override void HandleCommand(in IncomingCommand cmd)
473485
{
474486
_tcs.SetException(new InvalidOperationException($"Received unexpected command of type {cmd.CommandId}!"));
475487
}
488+
489+
return Task.CompletedTask;
476490
}
477491
finally
478492
{

0 commit comments

Comments
 (0)