Fixing tests.

Author: Stefán J. Sigurðarson

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

@@ -68,8 +68,16 @@ async Task Loop()
             {
                 while (tokenSource.IsCancellationRequested == false)
                 {
-                    await semaphore.WaitAsync().ConfigureAwait(false);
-                    if (workQueue.TryDequeue(out Work work))
+                    try
+                    {
+                        await semaphore.WaitAsync(tokenSource.Token).ConfigureAwait(false);
+                    }
+                    catch (TaskCanceledException)
+                    {
+                        // Swallowing the task cancellation in case we are stopping work.
+                    }
+
+                    if (!tokenSource.IsCancellationRequested && workQueue.TryDequeue(out Work work))
                     {
                         await work.Execute(model).ConfigureAwait(false);
                     }
@@ -79,7 +87,6 @@ async Task Loop()
             public void Stop()
             {
                 tokenSource.Cancel();
-                task.Wait();
             }
         }
     }

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

@@ -42,6 +42,7 @@
 using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Impl;
 using RabbitMQ.Util;
+
 using System;
 using System.Collections.Generic;
 using System.IO;
@@ -98,10 +99,6 @@ public class Connection : IConnection
         private Timer _heartbeatReadTimer;
         private AutoResetEvent m_heartbeatRead = new AutoResetEvent(false);
 
-        private readonly object _heartBeatReadLock = new object();
-        private readonly object _heartBeatWriteLock = new object();
-        private bool m_hasDisposedHeartBeatReadTimer;
-        private bool m_hasDisposedHeartBeatWriteTimer;
         private Task _mainLoopTask;
 
         private static string version = typeof(Connection).Assembly
@@ -519,8 +516,6 @@ public void EnsureIsOpen()
         // Only call at the end of the Mainloop or HeartbeatLoop
         public void FinishClose()
         {
-            // Notify hearbeat loops that they can leave
-            m_heartbeatRead.Set();
             m_closed = true;
             MaybeStopHeartbeatTimers();
 
@@ -948,19 +943,15 @@ public void MaybeStartHeartbeatTimers()
         {
             if (Heartbeat != 0)
             {
-                m_hasDisposedHeartBeatReadTimer = false;
-                m_hasDisposedHeartBeatWriteTimer = false;
-                lock (_heartBeatWriteLock)
+                if (_heartbeatWriteTimer == null)
                 {
                     _heartbeatWriteTimer = new Timer(HeartbeatWriteTimerCallback, null, Timeout.Infinite, Timeout.Infinite);
-
                     _heartbeatWriteTimer.Change(200, Timeout.Infinite);
                 }
 
-                lock (_heartBeatReadLock)
+                if (_heartbeatReadTimer == null)
                 {
                     _heartbeatReadTimer = new Timer(HeartbeatReadTimerCallback, null, Timeout.Infinite, Timeout.Infinite);
-
                     _heartbeatReadTimer.Change(300, Timeout.Infinite);
                 }
             }
@@ -973,75 +964,69 @@ public void StartMainLoop(bool useBackgroundThread)
 
         public void HeartbeatReadTimerCallback(object state)
         {
-            lock (_heartBeatReadLock)
+            if (_heartbeatReadTimer == null)
             {
-                if (m_hasDisposedHeartBeatReadTimer)
-                {
-                    return;
-                }
+                return;
+            }
 
-                bool shouldTerminate = false;
+            bool shouldTerminate = false;
 
-                try
+            try
+            {
+                if (!m_closed)
                 {
-                    if (!m_closed)
+                    if (!m_heartbeatRead.WaitOne(0))
                     {
-                        if (!m_heartbeatRead.WaitOne(0))
-                        {
-                            m_missedHeartbeats++;
-                        }
-                        else
-                        {
-                            m_missedHeartbeats = 0;
-                        }
-
-                        // We check against 8 = 2 * 4 because we need to wait for at
-                        // least two complete heartbeat setting intervals before
-                        // complaining, and we've set the socket timeout to a quarter
-                        // of the heartbeat setting in setHeartbeat above.
-                        if (m_missedHeartbeats > 2 * 4)
-                        {
-                            String description = String.Format("Heartbeat missing with heartbeat == {0} seconds", m_heartbeat);
-                            var eose = new EndOfStreamException(description);
-                            ESLog.Error(description, eose);
-                            m_shutdownReport.Add(new ShutdownReportEntry(description, eose));
-                            HandleMainLoopException(
-                                new ShutdownEventArgs(ShutdownInitiator.Library, 0, "End of stream", eose));
-                            shouldTerminate = true;
-                        }
+                        m_missedHeartbeats++;
                     }
-
-                    if (shouldTerminate)
+                    else
                     {
-                        TerminateMainloop();
-                        FinishClose();
+                        m_missedHeartbeats = 0;
                     }
-                    else if (_heartbeatReadTimer != null)
+
+                    // We check against 8 = 2 * 4 because we need to wait for at
+                    // least two complete heartbeat setting intervals before
+                    // complaining, and we've set the socket timeout to a quarter
+                    // of the heartbeat setting in setHeartbeat above.
+                    if (m_missedHeartbeats > 2 * 4)
                     {
-                        _heartbeatReadTimer.Change(Heartbeat * 1000, Timeout.Infinite);
+                        String description = String.Format("Heartbeat missing with heartbeat == {0} seconds", m_heartbeat);
+                        var eose = new EndOfStreamException(description);
+                        ESLog.Error(description, eose);
+                        m_shutdownReport.Add(new ShutdownReportEntry(description, eose));
+                        HandleMainLoopException(
+                            new ShutdownEventArgs(ShutdownInitiator.Library, 0, "End of stream", eose));
+                        shouldTerminate = true;
                     }
                 }
-                catch (ObjectDisposedException)
+
+                if (shouldTerminate)
                 {
-                    // timer is already disposed,
-                    // e.g. due to shutdown
+                    TerminateMainloop();
+                    FinishClose();
                 }
-                catch (NullReferenceException)
+                else if (_heartbeatReadTimer != null)
                 {
-                    // timer has already been disposed from a different thread after null check
-                    // this event should be rare
+                    _heartbeatReadTimer.Change(Heartbeat * 1000, Timeout.Infinite);
                 }
             }
+            catch (ObjectDisposedException)
+            {
+                // timer is already disposed,
+                // e.g. due to shutdown
+            }
+            catch (NullReferenceException)
+            {
+                // timer has already been disposed from a different thread after null check
+                // this event should be rare
+            }
         }
 
         public void HeartbeatWriteTimerCallback(object state)
         {
-            lock (_heartBeatWriteLock)
+            if (_heartbeatWriteTimer == null)
             {
-                if (m_hasDisposedHeartBeatWriteTimer)
-                {
-                    return;
-                }
+                return;
             }
 
             try
@@ -1062,51 +1047,17 @@ public void HeartbeatWriteTimerCallback(object state)
                 // peer unavailability. See rabbitmq/rabbitmq-dotnet-client#638 for details.
             }
 
-            lock(_heartBeatWriteLock)
+            if (m_closed == false)
             {
-                if(m_closed == false && _heartbeatWriteTimer != null)
-                {
-                    _heartbeatWriteTimer.Change((int)m_heartbeatTimeSpan.TotalMilliseconds, Timeout.Infinite);
-                }
+                _heartbeatWriteTimer?.Change((int)m_heartbeatTimeSpan.TotalMilliseconds, Timeout.Infinite);
             }
         }
 
         void MaybeStopHeartbeatTimers()
         {
-            lock (_heartBeatReadLock)
-            {
-                MaybeDisposeTimer(ref _heartbeatReadTimer);
-                m_hasDisposedHeartBeatReadTimer = true;
-            }
-
-            lock (_heartBeatWriteLock)
-            {
-                MaybeDisposeTimer(ref _heartbeatWriteTimer);
-                m_hasDisposedHeartBeatWriteTimer = true;
-            }
-        }
-
-        private void MaybeDisposeTimer(ref Timer timer)
-        {
-            // capture the timer to reduce chance of a null ref exception
-            var captured = timer;
-            if (captured != null)
-            {
-                try
-                {
-                    captured.Change(Timeout.Infinite, Timeout.Infinite);
-                    captured.Dispose();
-                    timer = null;
-                }
-                catch (ObjectDisposedException)
-                {
-                    // we are shutting down, ignore
-                }
-                catch (NullReferenceException)
-                {
-                    // this should be very rare but could occur from a race condition
-                }
-            }
+            NotifyHeartbeatListener();
+            _heartbeatReadTimer?.Dispose();
+            _heartbeatWriteTimer?.Dispose();
         }
 
         ///<remarks>
@@ -1231,9 +1182,8 @@ private void Dispose(bool disposing)
                 // dispose managed resources
                 try
                 {
-                    _mainLoopTask.Wait();
-                    MaybeStopHeartbeatTimers();
                     Abort();
+                    _mainLoopTask.Wait();
                 }
                 catch (OperationInterruptedException)
                 {

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

@@ -65,9 +65,16 @@ async Task Loop()
             {
                 while (tokenSource.IsCancellationRequested == false)
                 {
-                    await semaphore.WaitAsync().ConfigureAwait(false);
+                    try
+                    {
+                        await semaphore.WaitAsync(tokenSource.Token).ConfigureAwait(false);
+                    }
+                    catch (TaskCanceledException)
+                    {
+                        // Swallowing the task cancellation exception for the semaphore in case we are stopping.
+                    }
 
-                    if (actions.TryDequeue(out Action action))
+                    if (!tokenSource.IsCancellationRequested && actions.TryDequeue(out Action action))
                     {
                         try
                         {
@@ -77,13 +84,13 @@ async Task Loop()
                         {
                         }
                     }
+
                 }
             }
 
             public void Stop()
             {
                 tokenSource.Cancel();
-                worker.Wait();
             }
         }
     }

projects/client/Unit/src/unit/TestConcurrentAccessWithSharedConnection.cs

@@ -41,15 +41,16 @@
 using NUnit.Framework;
 
 using System;
-using System.Text;
 using System.Linq;
+using System.Text;
 using System.Threading;
+using System.Threading.Tasks;
 
-using RabbitMQ.Client;
-
-namespace RabbitMQ.Client.Unit {
+namespace RabbitMQ.Client.Unit
+{
     [TestFixture]
-    public class TestConcurrentAccessWithSharedConnection : IntegrationFixture {
+    public class TestConcurrentAccessWithSharedConnection : IntegrationFixture
+    {
 
         internal const int threads = 32;
         internal CountdownEvent latch;
@@ -59,6 +60,7 @@ public class TestConcurrentAccessWithSharedConnection : IntegrationFixture {
         public override void Init()
         {
             base.Init();
+            ThreadPool.SetMinThreads(threads, threads);
             latch = new CountdownEvent(threads);
         }
 
@@ -150,7 +152,8 @@ public void TestConcurrentChannelOpenAndPublishingCase12()
         [Test]
         public void TestConcurrentChannelOpenCloseLoop()
         {
-            TestConcurrentChannelOperations((conn) => {
+            TestConcurrentChannelOperations((conn) =>
+            {
                 var ch = conn.CreateModel();
                 ch.Close();
             }, 50);
@@ -164,7 +167,7 @@ internal void TestConcurrentChannelOpenAndPublishingWithBodyOfSize(int length)
         internal void TestConcurrentChannelOpenAndPublishingWithBodyOfSize(int length, int iterations)
         {
             string s = "payload";
-            if(length > s.Length)
+            if (length > s.Length)
             {
                 s.PadRight(length);
             }
@@ -174,7 +177,8 @@ internal void TestConcurrentChannelOpenAndPublishingWithBodyOfSize(int length, i
 
         internal void TestConcurrentChannelOpenAndPublishingWithBody(byte[] body, int iterations)
         {
-            TestConcurrentChannelOperations((conn) => {
+            TestConcurrentChannelOperations((conn) =>
+            {
                 // publishing on a shared channel is not supported
                 // and would missing the point of this test anyway
                 var ch = Conn.CreateModel();
@@ -196,18 +200,20 @@ internal void TestConcurrentChannelOperations(Action<IConnection> actions,
         internal void TestConcurrentChannelOperations(Action<IConnection> actions,
             int iterations, TimeSpan timeout)
         {
-            foreach (var i in Enumerable.Range(0, threads))
+            var tasks = Enumerable.Range(0, threads).Select(x =>
             {
-                var t = new Thread(() => {
+                return Task.Run(() =>
+                {
                     foreach (var j in Enumerable.Range(0, iterations))
                     {
                         actions(Conn);
                     }
+
                     latch.Signal();
                 });
-                t.Start();
-            }
+            }).ToArray();
 
+            //Task.WhenAll(tasks).Wait();
             Assert.IsTrue(latch.Wait(timeout));
             // incorrect frame interleaving in these tests will result
             // in an unrecoverable connection-level exception, thus

projects/client/Unit/src/unit/TestHeartbeats.cs

@@ -98,6 +98,8 @@ public void TestHundredsOfConnectionsWithRandomHeartbeatInterval()
         {
             var rnd = new Random();
             List<IConnection> xs = new List<IConnection>();
+            // Since we are using the ThreadPool, let's set MinThreads to a high-enough value.
+            ThreadPool.SetMinThreads(200, 200);
             for (var i = 0; i < 200; i++)
             {
                 var n = Convert.ToUInt16(rnd.Next(2, 6));