Optimizing the multiple-confirm logic, pulling it out into it's own task that's only triggered when needed.

Author: stebet

RabbitMQDotNetClient.sln

@@ -15,9 +15,10 @@ EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{34486CC0-D61E-46BA-9E5E-6E8EFA7C34B5}"
 	ProjectSection(SolutionItems) = preProject
 		.editorconfig = .editorconfig
+		docs\specs\amqp0-9-1.xml = docs\specs\amqp0-9-1.xml
 	EndProjectSection
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "RabbitMQTest", "RabbitMQTest\RabbitMQTest.csproj", "{3CFAC019-8281-48AD-8925-16CE8EC3CE50}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "RabbitMQTest", "RabbitMQTest\RabbitMQTest.csproj", "{3CFAC019-8281-48AD-8925-16CE8EC3CE50}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution

RabbitMQTest/Program.cs

@@ -15,7 +15,7 @@ class Program
         private static int itemsPerBatch = 500;
         static async Task Main(string[] args)
         {
-            ThreadPool.SetMinThreads(16 * Environment.ProcessorCount, 16 * Environment.ProcessorCount);
+            Console.ReadLine();
             var connectionString = new Uri("amqp://guest:guest@localhost/");
 
             var connectionFactory = new ConnectionFactory() { DispatchConsumersAsync = true, Uri = connectionString };
@@ -32,9 +32,9 @@ static async Task Main(string[] args)
             var asyncListener = new AsyncEventingBasicConsumer(subscriber);
             asyncListener.Received += AsyncListener_Received;
             subscriber.QueueBind("testqueue", "test", "myawesome.routing.key");
-            subscriber.BasicConsume("testqueue", true, "testconsumer", asyncListener);
+            subscriber.BasicConsume("testqueue", false, "testconsumer", asyncListener);
 
-            byte[] payload = new byte[16384];
+            byte[] payload = new byte[512];
             var batchPublish = Task.Run(async () =>
             {
                 while (messagesSent < batchesToSend * itemsPerBatch)
@@ -78,6 +78,7 @@ static async Task Main(string[] args)
             });
 
             await Task.WhenAll(sentTask, receivedTask);
+            Console.ReadLine();
         }
 
         private static Task AsyncListener_Received(object sender, BasicDeliverEventArgs @event)
@@ -90,6 +91,7 @@ private static Task AsyncListener_Received(object sender, BasicDeliverEventArgs
             // Moving to better synchronization constructs solves the issue, and using the ThreadPool
             // is standard practice as well to maximize core utilization and reduce overhead of Thread creation
             Interlocked.Increment(ref messagesReceived);
+            (sender as AsyncDefaultBasicConsumer).Model.BasicAck(@event.DeliveryTag, true);
             return Task.CompletedTask;
         }
     }

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

@@ -75,6 +75,10 @@ public abstract class ModelBase : IFullModel, IRecoverable
 
         //private readonly SynchronizedList<ulong> m_unconfirmedSet = new SynchronizedList<ulong>();
         private readonly ConcurrentDictionary<ulong, bool> m_unconfirmedSet = new ConcurrentDictionary<ulong, bool>();
+        private ulong _highestDeliveryTag = 0;
+        private Task _multipleConfirmCleanup;
+        private SemaphoreSlim _multipleConfirmLock = new SemaphoreSlim(1, 1);
+        private readonly CancellationTokenSource _connectionClosingCancellation = new CancellationTokenSource();
         private readonly SemaphoreSlim _outstandingSemaphore = new SemaphoreSlim(0);
 
         private EventHandler<EventArgs> m_basicRecoverOk;
@@ -114,6 +118,7 @@ protected void Initialise(ISession session)
             Session = session;
             Session.CommandReceived = HandleCommand;
             Session.SessionShutdown += OnSessionShutdown;
+            _multipleConfirmCleanup = Task.Run(CleanupUnconfirmedTagsAsync);
         }
 
         public TimeSpan HandshakeContinuationTimeout
@@ -268,6 +273,7 @@ public void Close(ShutdownEventArgs reason, bool abort)
         {
             var k = new ShutdownContinuation();
             ModelShutdown += k.OnConnectionShutdown;
+            _connectionClosingCancellation.Cancel();
 
             try
             {
@@ -278,6 +284,7 @@ public void Close(ShutdownEventArgs reason, bool abort)
                 }
                 k.Wait(TimeSpan.FromMilliseconds(10000));
                 ConsumerDispatcher.Shutdown(this);
+                _multipleConfirmCleanup.Wait();
             }
             catch (AlreadyClosedException)
             {
@@ -422,6 +429,13 @@ public MethodBase ModelRpc(MethodBase method, ContentHeaderBase header, byte[] b
             }
         }
 
+        public async ValueTask<MethodBase> ModelRpcAsync(MethodBase method, ContentHeaderBase header, byte[] body)
+        {
+            var k = new AsyncRpcContinuation();
+            TransmitAndEnqueue(new Command(method, header, body), k);
+            return (await k.GetReplyAsync(this.ContinuationTimeout).ConfigureAwait(false)).Method;
+        }
+
         public void ModelSend(MethodBase method, ContentHeaderBase header, byte[] body)
         {
             if (method.HasContent)
@@ -1574,23 +1588,63 @@ protected virtual void handleAckNack(ulong deliveryTag, bool multiple, bool isNa
         {
             if (multiple)
             {
-                foreach (ulong key in m_unconfirmedSet.Keys)
+                if (_highestDeliveryTag < deliveryTag)
                 {
-                    if (key <= deliveryTag)
+                    // Multiple confirms, let's trigger a cleanup of any older deliveries.
+                    _highestDeliveryTag = deliveryTag;
+                    try
                     {
-                        m_unconfirmedSet.TryRemove(key, out _);
+                        _multipleConfirmLock.Release();
+                    }
+                    catch (SemaphoreFullException)
+                    {
+                        // Ignore if we are trying to release often.
                     }
                 }
             }
-            else
-            {
-                m_unconfirmedSet.TryRemove(deliveryTag, out _);
-            }
+
+            m_unconfirmedSet.TryRemove(deliveryTag, out _);
 
             m_onlyAcksReceived = m_onlyAcksReceived && !isNack;
+            TriggerAllOutstandingCompleted();
+        }
+
+        private void TriggerAllOutstandingCompleted()
+        {
             if (m_unconfirmedSet.Count == 0)
             {
-                _outstandingSemaphore.Release();
+                try
+                {
+                    _outstandingSemaphore.Release();
+                }
+                catch (SemaphoreFullException)
+                {
+                    // Swallow the semaphore full exception.
+                }
+            }
+        }
+
+        public async Task CleanupUnconfirmedTagsAsync()
+        {
+            while (!_connectionClosingCancellation.IsCancellationRequested)
+            {
+                try
+                {
+                    await _multipleConfirmLock.WaitAsync(_connectionClosingCancellation.Token).ConfigureAwait(false);
+                    foreach (ulong key in m_unconfirmedSet.Keys)
+                    {
+                        if (key <= _highestDeliveryTag)
+                        {
+                            m_unconfirmedSet.TryRemove(key, out _);
+                        }
+                    }
+
+                    TriggerAllOutstandingCompleted();
+                }
+                catch (TaskCanceledException)
+                {
+                    // Swallow the task cancel exception since the model is being closed.
+                }
             }
         }