Using the TPL instead of dedicated threads. Results in a much more stable connection and consumers so far.

Code cleanups. Removing unneeded dependency.

Author: Stefán J. Sigurðarson

RabbitMQDotNetClient.sln

@@ -1,6 +1,6 @@
 Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio 15
-VisualStudioVersion = 15.0.26114.2
+# Visual Studio Version 16
+VisualStudioVersion = 16.0.29709.97
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "projects", "projects", "{068D7DC3-8E6E-4951-B9E3-272C641BF0DE}"
 EndProject

projects/client/RabbitMQ.Client/RabbitMQ.Client.csproj

@@ -64,4 +64,4 @@
     </AssemblyAttribute>
   </ItemGroup>
 
-</Project>
+</Project>

projects/client/RabbitMQ.Client/src/client/impl/AsyncConsumerDispatcher.cs

@@ -19,14 +19,12 @@ public void Quiesce()
 
         public void Shutdown()
         {
-            // necessary evil
-            this.workService.Stop().GetAwaiter().GetResult();
+            workService.Stop();
         }
 
         public void Shutdown(IModel model)
         {
-            // necessary evil
-            this.workService.Stop(model).GetAwaiter().GetResult();
+            workService.Stop(model);
         }
 
         public bool IsShutdown
@@ -66,10 +64,10 @@ public void HandleBasicCancel(IBasicConsumer consumer, string consumerTag)
         public void HandleModelShutdown(IBasicConsumer consumer, ShutdownEventArgs reason)
         {
             // the only case where we ignore the shutdown flag.
-            new ModelShutdown(consumer,reason).Execute(model).GetAwaiter().GetResult();
+            new ModelShutdown(consumer, reason).Execute(model).GetAwaiter().GetResult();
         }
 
-        private void ScheduleUnlessShuttingDown<TWork>(TWork work) 
+        private void ScheduleUnlessShuttingDown<TWork>(TWork work)
             where TWork : Work
         {
             if (!this.IsShutdown)

projects/client/RabbitMQ.Client/src/client/impl/AsyncConsumerWorkService.cs

@@ -1,69 +1,55 @@
-using System;
+using RabbitMQ.Client.Impl;
+
 using System.Collections.Concurrent;
 using System.Threading;
 using System.Threading.Tasks;
-using RabbitMQ.Client.Impl;
 
 namespace RabbitMQ.Client
 {
     internal sealed class AsyncConsumerWorkService : ConsumerWorkService
     {
-        readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();
+        private readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();
 
-        public void Schedule<TWork>(ModelBase model, TWork work)
-            where TWork : Work
+        public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work
         {
-            // two step approach is taken, as TryGetValue does not aquire locks
-            // if this fails, GetOrAdd is called, which takes a lock
-
-            WorkPool workPool;
-            if (workPools.TryGetValue(model, out workPool) == false)
-            {
-                var newWorkPool = new WorkPool(model);
-                workPool = workPools.GetOrAdd(model, newWorkPool);
-
-                // start if it's only the workpool that has been just created
-                if (newWorkPool == workPool)
-                {
-                    newWorkPool.Start();
-                }
-            }
+            workPools.GetOrAdd(model, StartNewWorkPool).Enqueue(work);
+        }
 
-            workPool.Enqueue(work);
+        private WorkPool StartNewWorkPool(IModel model)
+        {
+            var newWorkPool = new WorkPool(model as ModelBase);
+            newWorkPool.Start();
+            return newWorkPool;
         }
 
-        public async Task Stop(IModel model)
+        public void Stop(IModel model)
         {
-            WorkPool workPool;
-            if (workPools.TryRemove(model, out workPool))
+            if (workPools.TryRemove(model, out WorkPool workPool))
             {
-                await workPool.Stop().ConfigureAwait(false);
+                workPool.Stop();
             }
         }
 
-        public async Task Stop()
+        public void Stop()
         {
-            foreach (var model in workPools.Keys)
+            foreach (IModel model in workPools.Keys)
             {
-                await Stop(model).ConfigureAwait(false);
+                Stop(model);
             }
         }
 
         class WorkPool
         {
             readonly ConcurrentQueue<Work> workQueue;
-            readonly TimeSpan waitTime;
             readonly CancellationTokenSource tokenSource;
             readonly ModelBase model;
-            TaskCompletionSource<bool> messageArrived;
+            readonly SemaphoreSlim semaphore = new SemaphoreSlim(0);
             private Task task;
 
             public WorkPool(ModelBase model)
             {
                 this.model = model;
                 workQueue = new ConcurrentQueue<Work>();
-                messageArrived = new TaskCompletionSource<bool>();
-                waitTime = TimeSpan.FromMilliseconds(100);
                 tokenSource = new CancellationTokenSource();
             }
 
@@ -75,30 +61,33 @@ public void Start()
             public void Enqueue(Work work)
             {
                 workQueue.Enqueue(work);
-                messageArrived.TrySetResult(true);
+                semaphore.Release();
             }
 
             async Task Loop()
             {
                 while (tokenSource.IsCancellationRequested == false)
                 {
-                    Work work;
-                    while (workQueue.TryDequeue(out work))
+                    try
                     {
-                        await work.Execute(model).ConfigureAwait(false);
+                        await semaphore.WaitAsync(tokenSource.Token).ConfigureAwait(false);
+                    }
+                    catch (TaskCanceledException)
+                    {
+                        // Swallowing the task cancellation in case we are stopping work.
                     }
 
-                    await Task.WhenAny(Task.Delay(waitTime, tokenSource.Token), messageArrived.Task).ConfigureAwait(false);
-                    messageArrived.TrySetResult(true);
-                    messageArrived = new TaskCompletionSource<bool>();
+                    if (!tokenSource.IsCancellationRequested && workQueue.TryDequeue(out Work work))
+                    {
+                        await work.Execute(model).ConfigureAwait(false);
+                    }
                 }
             }
 
-            public Task Stop()
+            public void Stop()
             {
                 tokenSource.Cancel();
-                return task;
             }
         }
     }
-}
+}

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

@@ -39,11 +39,11 @@
 //---------------------------------------------------------------------------
 
 using RabbitMQ.Client.Events;
-using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Impl;
+
 using System;
-using System.Collections.Generic;
 using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -93,7 +93,7 @@ public AutorecoveringConnection(ConnectionFactory factory, string clientProvided
             m_factory = factory;
             this.ClientProvidedName = clientProvidedName;
         }
-        
+
         public event EventHandler<EventArgs> RecoverySucceeded
         {
             add
@@ -551,12 +551,8 @@ private void Init(IFrameHandler fh)
         {
             m_delegate = new Connection(m_factory, false,
                 fh, this.ClientProvidedName);
-            
-            m_recoveryThread = new Thread(MainRecoveryLoop)
-            {
-                IsBackground = true
-            };
-            m_recoveryThread.Start();
+
+            m_recoveryTask = Task.Run(MainRecoveryLoop);
 
             EventHandler<ShutdownEventArgs> recoveryListener = (_, args) =>
             {
@@ -587,7 +583,7 @@ public void UpdateSecret(string newSecret, string reason)
         public void Abort()
         {
             StopRecoveryLoop();
-            if(m_delegate.IsOpen)
+            if (m_delegate.IsOpen)
                 m_delegate.Abort();
         }
 
@@ -606,7 +602,7 @@ public void Abort(int timeout)
             if (m_delegate.IsOpen)
                 m_delegate.Abort(timeout);
         }
-        
+
         ///<summary>API-side invocation of connection abort with timeout.</summary>
         public void Abort(TimeSpan timeout)
         {
@@ -622,7 +618,7 @@ public void Abort(ushort reasonCode, string reasonText, int timeout)
             if (m_delegate.IsOpen)
                 m_delegate.Abort(reasonCode, reasonText, timeout);
         }
-        
+
         ///<summary>API-side invocation of connection abort with timeout.</summary>
         public void Abort(ushort reasonCode, string reasonText, TimeSpan timeout)
         {
@@ -990,7 +986,7 @@ private bool ShouldTriggerConnectionRecovery(ShutdownEventArgs args)
                     // connectivity loss or abrupt shutdown
                     args.Initiator == ShutdownInitiator.Library);
         }
-        
+
         private enum RecoveryCommand
         {
             /// <summary>
@@ -1017,7 +1013,7 @@ private enum RecoveryConnectionState
             Recovering
         }
 
-        private Thread m_recoveryThread;
+        private Task m_recoveryTask;
         private RecoveryConnectionState m_recoveryLoopState = RecoveryConnectionState.Connected;
 
         private readonly BlockingCollection<RecoveryCommand> m_recoveryLoopCommandQueue = new BlockingCollection<RecoveryCommand>();
@@ -1027,7 +1023,7 @@ private enum RecoveryConnectionState
         /// <summary>
         /// This is the main loop for the auto-recovery thread.
         /// </summary>
-        private void MainRecoveryLoop()
+        private async Task MainRecoveryLoop()
         {
             try
             {
@@ -1036,10 +1032,10 @@ private void MainRecoveryLoop()
                     switch (m_recoveryLoopState)
                     {
                         case RecoveryConnectionState.Connected:
-                            RecoveryLoopConnectedHandler(command);
+                            await RecoveryLoopConnectedHandler(command).ConfigureAwait(false);
                             break;
                         case RecoveryConnectionState.Recovering:
-                            RecoveryLoopRecoveringHandler(command);
+                            await RecoveryLoopRecoveringHandler(command).ConfigureAwait(false);
                             break;
                         default:
                             ESLog.Warn("RecoveryLoop state is out of range.");
@@ -1055,6 +1051,7 @@ private void MainRecoveryLoop()
             {
                 ESLog.Error("Main recovery loop threw unexpected exception.", e);
             }
+
             m_recoveryLoopComplete.SetResult(0);
         }
 
@@ -1075,7 +1072,7 @@ private void StopRecoveryLoop()
         /// Handles commands when in the Recovering state.
         /// </summary>
         /// <param name="command"></param>
-        private void RecoveryLoopRecoveringHandler(RecoveryCommand command)
+        private async Task RecoveryLoopRecoveringHandler(RecoveryCommand command)
         {
             switch (command)
             {
@@ -1089,7 +1086,8 @@ private void RecoveryLoopRecoveringHandler(RecoveryCommand command)
                     }
                     else
                     {
-                        Task.Delay(m_factory.NetworkRecoveryInterval).ContinueWith(t => { m_recoveryLoopCommandQueue.TryAdd(RecoveryCommand.PerformAutomaticRecovery); });
+                        await Task.Delay(m_factory.NetworkRecoveryInterval);
+                        m_recoveryLoopCommandQueue.TryAdd(RecoveryCommand.PerformAutomaticRecovery);
                     }
 
                     break;
@@ -1103,7 +1101,7 @@ private void RecoveryLoopRecoveringHandler(RecoveryCommand command)
         /// Handles commands when in the Connected state.
         /// </summary>
         /// <param name="command"></param>
-        private void RecoveryLoopConnectedHandler(RecoveryCommand command)
+        private async Task RecoveryLoopConnectedHandler(RecoveryCommand command)
         {
             switch (command)
             {
@@ -1112,7 +1110,8 @@ private void RecoveryLoopConnectedHandler(RecoveryCommand command)
                     break;
                 case RecoveryCommand.BeginAutomaticRecovery:
                     m_recoveryLoopState = RecoveryConnectionState.Recovering;
-                    Task.Delay(m_factory.NetworkRecoveryInterval).ContinueWith(t => { m_recoveryLoopCommandQueue.TryAdd(RecoveryCommand.PerformAutomaticRecovery); });
+                    await Task.Delay(m_factory.NetworkRecoveryInterval).ConfigureAwait(false);
+                    m_recoveryLoopCommandQueue.TryAdd(RecoveryCommand.PerformAutomaticRecovery);
                     break;
                 default:
                     ESLog.Warn($"RecoveryLoop command {command} is out of range.");