Investigate & fix SemaphoreFullException

Fixes #1818

Start by modifying test to concurrently publish to the same `IChannel` instance.

Author: lukebakken

projects/Test/Integration/TestFloodPublishing.cs

@@ -92,31 +92,37 @@ public async Task TestUnthrottledFloodPublishing()
                 return Task.CompletedTask;
             };
 
-            var publishTasks = new List<Task>();
             var stopwatch = Stopwatch.StartNew();
-            int i = 0;
             int publishCount = 0;
             try
             {
-                for (i = 0; i < 65535 * 64; i++)
+                var tasks = new List<Task>();
+                for (int j = 0; j < 64; j++)
                 {
-                    if (i % 65536 == 0)
+                    tasks.Add(Task.Run(async () =>
                     {
-                        if (stopwatch.Elapsed > FiveSeconds)
+                        var publishTasks = new List<Task>();
+                        for (int i = 0; i < 65536 * 2; i++)
                         {
-                            break;
-                        }
-                    }
+                            if (stopwatch.Elapsed > FiveSeconds)
+                            {
+                                await Task.WhenAll(publishTasks).WaitAsync(ShortSpan);
+                                publishTasks.Clear();
+                                break;
+                            }
 
-                    publishCount++;
-                    publishTasks.Add(_channel.BasicPublishAsync(CachedString.Empty, CachedString.Empty, _body).AsTask());
+                            Interlocked.Increment(ref publishCount);
+                            publishTasks.Add(_channel.BasicPublishAsync(CachedString.Empty, CachedString.Empty, _body).AsTask());
 
-                    if (i % 500 == 0)
-                    {
-                        await Task.WhenAll(publishTasks).WaitAsync(ShortSpan);
-                        publishTasks.Clear();
-                    }
+                            if (i % 500 == 0)
+                            {
+                                await Task.WhenAll(publishTasks).WaitAsync(ShortSpan);
+                                publishTasks.Clear();
+                            }
+                        }
+                    }));
                 }
+                await Task.WhenAll(tasks).WaitAsync(WaitSpan);
             }
             finally
             {