Fix Semaphore Disposed Exception in AsyncConsumerWorkService

[ashneilson]

Proposed Changes

Fixes a System.ObjectDisposedException thrown when the SemaphoreSlim is Disposed on a AsyncConsumerWorkService with a Concurrency greater than 1.

Current State

When under load, the ChannelReader can return a Work item after the _limiter has been Disposed resulting in the Exception being thrown.

rabbitmq-dotnet-client/projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs

Lines 131 to 139 in d39bb71

	while (_channel.Reader.TryRead(out Work work))
	{
	// Do a quick synchronous check before we resort to async/await with the state-machine overhead.
	if (!_limiter.Wait(0))
	{
	await _limiter.WaitAsync(cancellationToken).ConfigureAwait(false);
	}
	_ = HandleConcurrent(work, _model, _limiter);

Proposed Fix

We check if the cancellationToken has been cancelled. If so, we avoid attempting to wait on the _limiter and will not see an Exception thrown.
https://github.com/ricado-group/rabbitmq-dotnet-client/blob/08854f1a83b3c4a1214f41518bbfa380ec673e1c/projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs#L130-L139

Types of Changes

• Bug fix (non-breaking change which fixes issue Semaphore Disposed Exception thrown in AsyncConsumerWorkService #1014)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause an observable behavior change in existing systems)
• Documentation improvements (corrections, new content, etc)
• Cosmetic change (whitespace, formatting, etc)

Checklist

• I have read the CONTRIBUTING.md document
• I have signed the CA (see https://cla.pivotal.io/sign/rabbitmq)
• All tests pass locally with my changes
• I have added tests that prove my fix is effective or that my feature works
• I have added necessary documentation (if appropriate)
• Any dependent changes have been merged and published in related repositories

[pivotal-issuemaster]

@ashneilson Please sign the Contributor License Agreement!

Click here to manually synchronize the status of this Pull Request.

See the FAQ for frequently asked questions.

[pivotal-issuemaster]

@ashneilson Thank you for signing the Contributor License Agreement!

[michaelklishin]

Thank you!

[michaelklishin]

Backported to 6.x.

[bollhals]

Hmm probably irrelevant but now the remaining work items won't be processed. (kind of also didn't before due to the dispose exception), but what would we expect here? Shouldn't all the items be processed?

[ashneilson]

Thanks for raising this point @bollhals.

The behavior is still as before given the _tokenSource is Cancelled here:

rabbitmq-dotnet-client/projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs

Lines 185 to 191 in cff0c3d

	public Task Stop()
	{
	_channel.Writer.Complete();
	_tokenSource?.Cancel();
	_limiter?.Dispose();
	return _worker;
	}

Under typical circumstances, an OperationCanceledException would be thrown on the _channel.Reader.WaitToReadAsync call which prevents any further processing of work items anyhow.

rabbitmq-dotnet-client/projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs

Lines 125 to 147 in cff0c3d

	private async Task LoopWithConcurrency(CancellationToken cancellationToken)
	{
	try
	{
	while (await _channel.Reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
	{
	while (_channel.Reader.TryRead(out Work work) && !cancellationToken.IsCancellationRequested)
	{
	// Do a quick synchronous check before we resort to async/await with the state-machine overhead.
	if (!_limiter.Wait(0))
	{
	await _limiter.WaitAsync(cancellationToken).ConfigureAwait(false);
	}
	_ = HandleConcurrent(work, _model, _limiter);
	}
	}
	}
	catch (OperationCanceledException)
	{
	// ignored
	}
	}

One would assume that if the processing of work items (e.g. BasicDeliver) was critical to the user, they would use .ConfirmSelect() on the Channel so the messages would be redelivered upon a Connection Recovery or a new Connection.

[michaelklishin]

Publisher confirms have no effect on the redelivery behaviour or consumer acknowledgements.

It would be nice to process the backlog of work items that were already submitted, in case anyone has more cycles to spend improving this client :)

[bollhals]

It would be nice to process the backlog of work items that were already submitted, in case anyone has more cycles to spend improving this client :)

#997 does it.

[bollhals]

But we could also consider a change to it so that on close it waits but on abort it does not.

[ashneilson]

Publisher confirms have no effect on the redelivery behaviour or consumer acknowledgements.

Sorry @michaelklishin! Thanks for correcting me. I had meant to suggest with AutoAck disabled, using a .BasicAck() or .BasicNack() for Consumer Acknowledgements would mean redelivery.

A simple solution could be to await _worker before calling _limiter?.Dispose() and like @bollhals suggests a bool abort could be added as an argument so we avoid this on an Abort.

rabbitmq-dotnet-client/projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs

Lines 185 to 191 in cff0c3d

	public Task Stop()
	{
	_channel.Writer.Complete();
	_tokenSource?.Cancel();
	_limiter?.Dispose();
	return _worker;
	}

I noted that during Connection Recovery, this WorkService is destroyed and re-created. Would we still want to process the backlog given this could delay recovery quite significantly?