* Make quiescing thread safe

* Add warning log when processing work while quiescing

Author: lukebakken

projects/RabbitMQ.Client/ConsumerDispatching/AsyncConsumerDispatcher.cs

@@ -26,6 +26,11 @@ protected override async Task ProcessChannelAsync()
                         {
                             try
                             {
+                                if (IsQuiescing)
+                                {
+                                    ESLog.Warn($"processing consumer work while quiescing: {work.WorkType}");
+                                }
+
                                 switch (work.WorkType)
                                 {
                                     case WorkType.Deliver:

projects/RabbitMQ.Client/ConsumerDispatching/ConsumerDispatcherChannelBase.cs

@@ -44,7 +44,7 @@ internal abstract class ConsumerDispatcherChannelBase : ConsumerDispatcherBase,
         private readonly System.Threading.Channels.ChannelWriter<WorkStruct> _writer;
         private readonly Task _worker;
         private readonly ushort _concurrency;
-        private bool _quiesce = false;
+        private long _isQuiescing;
         private bool _disposed;
 
         internal ConsumerDispatcherChannelBase(Impl.Channel channel, ushort concurrency)
@@ -79,15 +79,15 @@ internal ConsumerDispatcherChannelBase(Impl.Channel channel, ushort concurrency)
             }
         }
 
-        public bool IsShutdown => _quiesce;
+        public bool IsShutdown => IsQuiescing;
 
         public ushort Concurrency => _concurrency;
 
         public async ValueTask HandleBasicConsumeOkAsync(IAsyncBasicConsumer consumer, string consumerTag, CancellationToken cancellationToken)
         {
             cancellationToken.ThrowIfCancellationRequested();
 
-            if (false == _disposed && false == _quiesce)
+            if (false == _disposed && false == IsQuiescing)
             {
                 try
                 {
@@ -110,7 +110,7 @@ public async ValueTask HandleBasicDeliverAsync(string consumerTag, ulong deliver
         {
             cancellationToken.ThrowIfCancellationRequested();
 
-            if (false == _disposed && false == _quiesce)
+            if (false == _disposed && false == IsQuiescing)
             {
                 IAsyncBasicConsumer consumer = GetConsumerOrDefault(consumerTag);
                 var work = WorkStruct.CreateDeliver(consumer, consumerTag, deliveryTag, redelivered, exchange, routingKey, basicProperties, body);
@@ -123,7 +123,7 @@ public async ValueTask HandleBasicCancelOkAsync(string consumerTag, Cancellation
         {
             cancellationToken.ThrowIfCancellationRequested();
 
-            if (false == _disposed && false == _quiesce)
+            if (false == _disposed && false == IsQuiescing)
             {
                 IAsyncBasicConsumer consumer = GetAndRemoveConsumer(consumerTag);
                 WorkStruct work = WorkStruct.CreateCancelOk(consumer, consumerTag);
@@ -136,7 +136,7 @@ public async ValueTask HandleBasicCancelAsync(string consumerTag, CancellationTo
         {
             cancellationToken.ThrowIfCancellationRequested();
 
-            if (false == _disposed && false == _quiesce)
+            if (false == _disposed && false == IsQuiescing)
             {
                 IAsyncBasicConsumer consumer = GetAndRemoveConsumer(consumerTag);
                 WorkStruct work = WorkStruct.CreateCancel(consumer, consumerTag);
@@ -147,7 +147,7 @@ await _writer.WriteAsync(work, cancellationToken)
 
         public void Quiesce()
         {
-            _quiesce = true;
+            Interlocked.Exchange(ref _isQuiescing, 1);
         }
 
         public async Task WaitForShutdownAsync()
@@ -157,7 +157,7 @@ public async Task WaitForShutdownAsync()
                 return;
             }
 
-            if (_quiesce)
+            if (IsQuiescing)
             {
                 try
                 {
@@ -193,6 +193,19 @@ await _worker
             }
         }
 
+        protected bool IsQuiescing
+        {
+            get
+            {
+                if (Interlocked.Read(ref _isQuiescing) == 1)
+                {
+                    return true;
+                }
+
+                return false;
+            }
+        }
+
         protected sealed override void ShutdownConsumer(IAsyncBasicConsumer consumer, ShutdownEventArgs reason)
         {
             _writer.TryWrite(WorkStruct.CreateShutdown(consumer, reason));