Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Sequence Number non-blocking #1675

Merged
merged 2 commits into from
Sep 17, 2024
Merged

Sequence Number non-blocking #1675

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a6b7e38
Allow semaphore to be awaited and cancelled
danielmarbach Sep 13, 2024
5326343
Acquiring a sequence number should be an async method
danielmarbach Sep 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion projects/RabbitMQ.Client/PublicAPI.Shipped.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -440,9 +440,9 @@ RabbitMQ.Client.IChannel.CurrentQueue.get -> string
RabbitMQ.Client.IChannel.DefaultConsumer.get -> RabbitMQ.Client.IAsyncBasicConsumer
RabbitMQ.Client.IChannel.DefaultConsumer.set -> void
RabbitMQ.Client.IChannel.FlowControl -> System.EventHandler<RabbitMQ.Client.Events.FlowControlEventArgs>
RabbitMQ.Client.IChannel.GetNextPublishSequenceNumberAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask<ulong>
RabbitMQ.Client.IChannel.IsClosed.get -> bool
RabbitMQ.Client.IChannel.IsOpen.get -> bool
RabbitMQ.Client.IChannel.NextPublishSeqNo.get -> ulong
RabbitMQ.Client.IChannelExtensions
RabbitMQ.Client.IConnection
RabbitMQ.Client.IConnection.CallbackException -> System.EventHandler<RabbitMQ.Client.Events.CallbackExceptionEventArgs>
Expand Down
10 changes: 5 additions & 5 deletions projects/RabbitMQ.Client/client/api/IChannel.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,11 +95,6 @@ public interface IChannel : IDisposable
/// </summary>
bool IsOpen { get; }

/// <summary>
/// When in confirm mode, return the sequence number of the next message to be published.
/// </summary>
ulong NextPublishSeqNo { get; }

/// <summary>
/// The name of the last queue declared on this channel.
/// </summary>
Expand Down Expand Up @@ -143,6 +138,11 @@ public interface IChannel : IDisposable
/// </remarks>
event EventHandler<ShutdownEventArgs> ChannelShutdown;

/// <summary>
/// When in confirm mode, return the sequence number of the next message to be published.
/// </summary>
ValueTask<ulong> GetNextPublishSequenceNumberAsync(CancellationToken cancellationToken = default);

/// <summary>Asynchronously acknknowledges one or more messages.</summary>
/// <param name="deliveryTag">The delivery tag.</param>
/// <param name="multiple">Ack all messages up to the delivery tag if set to <c>true</c>.</param>
Expand Down
6 changes: 2 additions & 4 deletions projects/RabbitMQ.Client/client/framing/Channel.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,13 +89,11 @@ protected override Task<bool> DispatchCommandAsync(IncomingCommand cmd, Cancella
}
case ProtocolCommandId.BasicAck:
{
HandleBasicAck(cmd);
return Task.FromResult(true);
return HandleBasicAck(cmd, cancellationToken);
}
case ProtocolCommandId.BasicNack:
{
HandleBasicNack(cmd);
return Task.FromResult(true);
return HandleBasicNack(cmd, cancellationToken);
}
case ProtocolCommandId.BasicReturn:
{
Expand Down
4 changes: 2 additions & 2 deletions projects/RabbitMQ.Client/client/impl/AutorecoveringChannel.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,8 +144,6 @@ public IAsyncBasicConsumer? DefaultConsumer

public bool IsOpen => !_disposed && _innerChannel.IsOpen;

public ulong NextPublishSeqNo => InnerChannel.NextPublishSeqNo;

public string? CurrentQueue => InnerChannel.CurrentQueue;

internal async Task<bool> AutomaticallyRecoverAsync(AutorecoveringConnection conn, bool recoverConsumers,
Expand Down Expand Up @@ -274,6 +272,8 @@ public void Dispose()
_disposed = true;
}

public ValueTask<ulong> GetNextPublishSequenceNumberAsync(CancellationToken cancellationToken = default) => InnerChannel.GetNextPublishSequenceNumberAsync(cancellationToken);

public ValueTask BasicAckAsync(ulong deliveryTag, bool multiple, CancellationToken cancellationToken)
=> InnerChannel.BasicAckAsync(deliveryTag, multiple, cancellationToken);

Expand Down
60 changes: 31 additions & 29 deletions projects/RabbitMQ.Client/client/impl/ChannelBase.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,29 +181,6 @@ public IAsyncBasicConsumer? DefaultConsumer
[MemberNotNullWhen(false, nameof(CloseReason))]
public bool IsOpen => CloseReason is null;

public ulong NextPublishSeqNo
{
get
{
if (ConfirmsAreEnabled)
{
_confirmSemaphore.Wait();
try
{
return _nextPublishSeqNo;
}
finally
{
_confirmSemaphore.Release();
}
}
else
{
return _nextPublishSeqNo;
}
}
}

public string? CurrentQueue { get; private set; }

public ISession Session { get; private set; }
Expand Down Expand Up @@ -589,7 +566,7 @@ public Task ConnectionTuneOkAsync(ushort channelMax, uint frameMax, ushort heart
return ModelSendAsync(method, cancellationToken).AsTask();
}

protected void HandleBasicAck(IncomingCommand cmd)
protected async Task<bool> HandleBasicAck(IncomingCommand cmd, CancellationToken cancellationToken)
{
var ack = new BasicAck(cmd.MethodSpan);
if (!_basicAcksWrapper.IsEmpty)
Expand All @@ -598,10 +575,12 @@ protected void HandleBasicAck(IncomingCommand cmd)
_basicAcksWrapper.Invoke(this, args);
}

HandleAckNack(ack._deliveryTag, ack._multiple, false);
await HandleAckNack(ack._deliveryTag, ack._multiple, false, cancellationToken)
.ConfigureAwait(false);
return true;
}

protected void HandleBasicNack(IncomingCommand cmd)
protected async Task<bool> HandleBasicNack(IncomingCommand cmd, CancellationToken cancellationToken)
{
var nack = new BasicNack(cmd.MethodSpan);
if (!_basicNacksWrapper.IsEmpty)
Expand All @@ -611,7 +590,9 @@ protected void HandleBasicNack(IncomingCommand cmd)
_basicNacksWrapper.Invoke(this, args);
}

HandleAckNack(nack._deliveryTag, nack._multiple, true);
await HandleAckNack(nack._deliveryTag, nack._multiple, true, cancellationToken)
.ConfigureAwait(false);
return true;
}

protected async Task<bool> HandleBasicCancelAsync(IncomingCommand cmd, CancellationToken cancellationToken)
Expand Down Expand Up @@ -801,6 +782,26 @@ protected void HandleConnectionUnblocked()
Session.Connection.HandleConnectionUnblocked();
}

public async ValueTask<ulong> GetNextPublishSequenceNumberAsync(CancellationToken cancellationToken = default)
{
if (ConfirmsAreEnabled)
{
await _confirmSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
return _nextPublishSeqNo;
}
finally
{
_confirmSemaphore.Release();
}
}
else
{
return _nextPublishSeqNo;
}
}

public abstract ValueTask BasicAckAsync(ulong deliveryTag, bool multiple,
CancellationToken cancellationToken);

Expand Down Expand Up @@ -1829,7 +1830,7 @@ await tokenRegistration.DisposeAsync()

// NOTE: this method is internal for its use in this test:
// TestWaitForConfirmsWithTimeoutAsync_MessageNacked_WaitingHasTimedout_ReturnFalse
internal void HandleAckNack(ulong deliveryTag, bool multiple, bool isNack)
internal async Task HandleAckNack(ulong deliveryTag, bool multiple, bool isNack, CancellationToken cancellationToken = default)
{
// Only do this if confirms are enabled *and* the library is tracking confirmations
if (ConfirmsAreEnabled && _trackConfirmations)
Expand All @@ -1839,7 +1840,8 @@ internal void HandleAckNack(ulong deliveryTag, bool multiple, bool isNack)
throw new InvalidOperationException(InternalConstants.BugFound);
}
// let's take a lock so we can assume that deliveryTags are unique, never duplicated and always sorted
_confirmSemaphore.Wait();
await _confirmSemaphore.WaitAsync(cancellationToken)
.ConfigureAwait(false);
try
{
// No need to do anything if there are no delivery tags in the list
Expand Down
2 changes: 1 addition & 1 deletion projects/Test/Applications/PublisherConfirms/PublisherConfirms.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,7 +173,7 @@ void CleanOutstandingConfirms(ulong deliveryTag, bool multiple)
{
string msg = i.ToString();
byte[] body = Encoding.UTF8.GetBytes(msg);
ulong nextPublishSeqNo = channel.NextPublishSeqNo;
ulong nextPublishSeqNo = await channel.GetNextPublishSequenceNumberAsync();
if ((ulong)(i + 1) != nextPublishSeqNo)
{
Console.WriteLine($"{DateTime.Now} [WARNING] i {i + 1} does not equal next sequence number: {nextPublishSeqNo}");
Expand Down
18 changes: 9 additions & 9 deletions projects/Test/Integration/TestConfirmSelect.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,19 +53,19 @@ ValueTask PublishAsync()
}

await _channel.ConfirmSelectAsync();
Assert.Equal(1ul, _channel.NextPublishSeqNo);
Assert.Equal(1ul, await _channel.GetNextPublishSequenceNumberAsync());
await PublishAsync();
Assert.Equal(2ul, _channel.NextPublishSeqNo);
Assert.Equal(2ul, await _channel.GetNextPublishSequenceNumberAsync());
await PublishAsync();
Assert.Equal(3ul, _channel.NextPublishSeqNo);
Assert.Equal(3ul, await _channel.GetNextPublishSequenceNumberAsync());

await _channel.ConfirmSelectAsync();
await PublishAsync();
Assert.Equal(4ul, _channel.NextPublishSeqNo);
Assert.Equal(4ul, await _channel.GetNextPublishSequenceNumberAsync());
await PublishAsync();
Assert.Equal(5ul, _channel.NextPublishSeqNo);
Assert.Equal(5ul, await _channel.GetNextPublishSequenceNumberAsync());
await PublishAsync();
Assert.Equal(6ul, _channel.NextPublishSeqNo);
Assert.Equal(6ul, await _channel.GetNextPublishSequenceNumberAsync());
}

[Theory]
Expand All @@ -80,7 +80,7 @@ public async Task TestDeliveryTagDiverged_GH1043(ushort correlationIdLength)
await _channel.ConfirmSelectAsync();

var properties = new BasicProperties();
// _output.WriteLine("Client delivery tag {0}", _channel.NextPublishSeqNo);
// _output.WriteLine("Client delivery tag {0}", await _channel.GetNextPublishSequenceNumberAsync());
await _channel.BasicPublishAsync(exchange: "sample", routingKey: string.Empty,
mandatory: false, basicProperties: properties, body: body);
await _channel.WaitForConfirmsOrDieAsync();
Expand All @@ -91,7 +91,7 @@ await _channel.BasicPublishAsync(exchange: "sample", routingKey: string.Empty,
{
CorrelationId = new string('o', correlationIdLength)
};
// _output.WriteLine("Client delivery tag {0}", _channel.NextPublishSeqNo);
// _output.WriteLine("Client delivery tag {0}", await _channel.GetNextPublishSequenceNumberAsync());
await _channel.BasicPublishAsync("sample", string.Empty, false, properties, body);
await _channel.WaitForConfirmsOrDieAsync();
}
Expand All @@ -101,7 +101,7 @@ await _channel.BasicPublishAsync(exchange: "sample", routingKey: string.Empty,
}

properties = new BasicProperties();
// _output.WriteLine("Client delivery tag {0}", _channel.NextPublishSeqNo);
// _output.WriteLine("Client delivery tag {0}", await _channel.GetNextPublishSequenceNumberAsync());
await _channel.BasicPublishAsync("sample", string.Empty, false, properties, body);
await _channel.WaitForConfirmsOrDieAsync();
// _output.WriteLine("I'm done...");
Expand Down
12 changes: 6 additions & 6 deletions projects/Test/Integration/TestConfirmSelectAsync.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,19 +49,19 @@ public TestConfirmSelectAsync(ITestOutputHelper output) : base(output)
public async Task TestConfirmSelectIdempotency()
{
await _channel.ConfirmSelectAsync();
Assert.Equal(1ul, _channel.NextPublishSeqNo);
Assert.Equal(1ul, await _channel.GetNextPublishSequenceNumberAsync());
await Publish();
Assert.Equal(2ul, _channel.NextPublishSeqNo);
Assert.Equal(2ul, await _channel.GetNextPublishSequenceNumberAsync());
await Publish();
Assert.Equal(3ul, _channel.NextPublishSeqNo);
Assert.Equal(3ul, await _channel.GetNextPublishSequenceNumberAsync());

await _channel.ConfirmSelectAsync();
await Publish();
Assert.Equal(4ul, _channel.NextPublishSeqNo);
Assert.Equal(4ul, await _channel.GetNextPublishSequenceNumberAsync());
await Publish();
Assert.Equal(5ul, _channel.NextPublishSeqNo);
Assert.Equal(5ul, await _channel.GetNextPublishSequenceNumberAsync());
await Publish();
Assert.Equal(6ul, _channel.NextPublishSeqNo);
Assert.Equal(6ul, await _channel.GetNextPublishSequenceNumberAsync());
}

private ValueTask Publish()
Expand Down
2 changes: 1 addition & 1 deletion projects/Test/Integration/TestPublisherConfirms.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,7 @@ public Task TestWaitForConfirmsWithTimeoutAsync_MessageNacked_WaitingHasTimedout
return TestWaitForConfirmsAsync(2000, async (ch) =>
{
RecoveryAwareChannel actualChannel = ((AutorecoveringChannel)ch).InnerChannel;
actualChannel.HandleAckNack(10UL, false, true);
await actualChannel.HandleAckNack(10UL, false, true);
using (var cts = new CancellationTokenSource(ShortSpan))
{
Assert.False(await ch.WaitForConfirmsAsync(cts.Token));
Expand Down
Loading