* Add delay if outstanding confirms is more than 50% of max.

Author: lukebakken

projects/RabbitMQ.Client/CreateChannelOptions.cs

@@ -16,7 +16,8 @@ public sealed class CreateChannelOptions
         public bool PublisherConfirmationTrackingEnabled { get; set; } = false;
 
         /// <summary>
-        /// The number of allowed outstanding publisher confirmations before publishing is blocked.
+        /// If publisher confirmation tracking is enabled, this represents the number of allowed
+        /// outstanding publisher confirmations before publishing is blocked.
         ///
         /// Defaults to <c>128</c>
         ///

projects/RabbitMQ.Client/Impl/Channel.PublisherConfirms.cs

@@ -125,7 +125,7 @@ private void ConfigurePublisherConfirmations(bool publisherConfirmationsEnabled,
             _publisherConfirmationTrackingEnabled = publisherConfirmationTrackingEnabled;
             _maxOutstandingPublisherConfirmations = maxOutstandingPublisherConfirmations;
 
-            if (_maxOutstandingPublisherConfirmations is not null)
+            if (_publisherConfirmationTrackingEnabled && _maxOutstandingPublisherConfirmations is not null)
             {
                 _maxOutstandingConfirmationsSemaphore = new SemaphoreSlim(
                     (int)_maxOutstandingPublisherConfirmations,
@@ -282,6 +282,18 @@ await _confirmSemaphore.WaitAsync(reason.CancellationToken)
         {
             if (_publisherConfirmationsEnabled)
             {
+                if (_publisherConfirmationTrackingEnabled)
+                {
+                    if (_maxOutstandingPublisherConfirmations is not null)
+                    {
+                        int percentOfMax = _confirmsTaskCompletionSources.Count / (int)_maxOutstandingPublisherConfirmations;
+                        if (percentOfMax > 0.5)
+                        {
+                            await Task.Delay(1000 * percentOfMax).ConfigureAwait(false);
+                        }
+                    }
+                }
+
                 await _confirmSemaphore.WaitAsync(cancellationToken)
                     .ConfigureAwait(false);
 
@@ -292,12 +304,12 @@ await _confirmSemaphore.WaitAsync(cancellationToken)
                 {
                     publisherConfirmationTcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
                     _confirmsTaskCompletionSources[publishSequenceNumber] = publisherConfirmationTcs;
-                }
 
-                if (_maxOutstandingConfirmationsSemaphore is not null)
-                {
-                    await _maxOutstandingConfirmationsSemaphore.WaitAsync(cancellationToken)
-                        .ConfigureAwait(false);
+                    if (_maxOutstandingConfirmationsSemaphore is not null)
+                    {
+                        await _maxOutstandingConfirmationsSemaphore.WaitAsync(cancellationToken)
+                            .ConfigureAwait(false);
+                    }
                 }
 
                 _nextPublishSeqNo++;
@@ -346,7 +358,8 @@ await publisherConfirmationInfo.MaybeWaitForConfirmationAsync(cancellationToken)
                         .ConfigureAwait(false);
                 }
 
-                if (_maxOutstandingConfirmationsSemaphore is not null)
+                if (_publisherConfirmationTrackingEnabled &&
+                    _maxOutstandingConfirmationsSemaphore is not null)
                 {
                     _maxOutstandingConfirmationsSemaphore.Release();
                 }

projects/Test/Applications/PublisherConfirms/PublisherConfirms.cs

@@ -38,9 +38,18 @@
 using System.Threading.Tasks;
 using RabbitMQ.Client;
 
+const ushort MAX_OUTSTANDING_CONFIRMS = 256;
+
 const int MESSAGE_COUNT = 50_000;
 bool debug = false;
 
+var channelOpts = new CreateChannelOptions
+{
+    PublisherConfirmationsEnabled = true,
+    PublisherConfirmationTrackingEnabled = true,
+    MaxOutstandingPublisherConfirmations = MAX_OUTSTANDING_CONFIRMS
+};
+
 #pragma warning disable CS8321 // Local function is declared but never used
 
 await PublishMessagesIndividuallyAsync();
@@ -53,12 +62,12 @@ static Task<IConnection> CreateConnectionAsync()
     return factory.CreateConnectionAsync();
 }
 
-static async Task PublishMessagesIndividuallyAsync()
+async Task PublishMessagesIndividuallyAsync()
 {
     Console.WriteLine($"{DateTime.Now} [INFO] publishing {MESSAGE_COUNT:N0} messages and handling confirms per-message");
 
     await using IConnection connection = await CreateConnectionAsync();
-    await using IChannel channel = await connection.CreateChannelAsync(new CreateChannelOptions { PublisherConfirmationsEnabled = true, PublisherConfirmationTrackingEnabled = true });
+    await using IChannel channel = await connection.CreateChannelAsync(channelOpts);
 
     // declare a server-named queue
     QueueDeclareOk queueDeclareResult = await channel.QueueDeclareAsync();
@@ -85,18 +94,18 @@ static async Task PublishMessagesIndividuallyAsync()
     Console.WriteLine($"{DateTime.Now} [INFO] published {MESSAGE_COUNT:N0} messages individually in {sw.ElapsedMilliseconds:N0} ms");
 }
 
-static async Task PublishMessagesInBatchAsync()
+async Task PublishMessagesInBatchAsync()
 {
     Console.WriteLine($"{DateTime.Now} [INFO] publishing {MESSAGE_COUNT:N0} messages and handling confirms in batches");
 
     await using IConnection connection = await CreateConnectionAsync();
-    await using IChannel channel = await connection.CreateChannelAsync(new CreateChannelOptions { PublisherConfirmationsEnabled = true, PublisherConfirmationTrackingEnabled = true });
+    await using IChannel channel = await connection.CreateChannelAsync(channelOpts);
 
     // declare a server-named queue
     QueueDeclareOk queueDeclareResult = await channel.QueueDeclareAsync();
     string queueName = queueDeclareResult.QueueName;
 
-    int batchSize = 1000;
+    int batchSize = MAX_OUTSTANDING_CONFIRMS;
     int outstandingMessageCount = 0;
 
     var sw = new Stopwatch();
@@ -154,12 +163,8 @@ async Task HandlePublishConfirmsAsynchronously()
 
     await using IConnection connection = await CreateConnectionAsync();
 
-    var channelOptions = new CreateChannelOptions
-    {
-        PublisherConfirmationsEnabled = true,
-        PublisherConfirmationTrackingEnabled = false
-    };
-    await using IChannel channel = await connection.CreateChannelAsync(channelOptions);
+    channelOpts.PublisherConfirmationTrackingEnabled = false;
+    await using IChannel channel = await connection.CreateChannelAsync(channelOpts);
 
     // declare a server-named queue
     QueueDeclareOk queueDeclareResult = await channel.QueueDeclareAsync();