Partition more ConcurrentQueues in Kestrel (#42237)

Author: halter73

src/Servers/Kestrel/Transport.Sockets/src/SocketConnectionContextFactory.cs

@@ -15,7 +15,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets;
 /// </summary>
 public sealed class SocketConnectionContextFactory : IDisposable
 {
-    private readonly MemoryPool<byte> _memoryPool;
     private readonly SocketConnectionFactoryOptions _options;
     private readonly ILogger _logger;
     private readonly int _settingsCount;
@@ -43,7 +42,6 @@ public SocketConnectionContextFactory(SocketConnectionFactoryOptions options, IL
 
         _options = options;
         _logger = logger;
-        _memoryPool = _options.MemoryPoolFactory();
         _settingsCount = _options.IOQueueCount;
 
         var maxReadBufferSize = _options.MaxReadBufferSize ?? 0;
@@ -56,33 +54,37 @@ public SocketConnectionContextFactory(SocketConnectionFactoryOptions options, IL
 
             for (var i = 0; i < _settingsCount; i++)
             {
+                var memoryPool = _options.MemoryPoolFactory();
                 var transportScheduler = options.UnsafePreferInlineScheduling ? PipeScheduler.Inline : new IOQueue();
                 // https://github.com/aspnet/KestrelHttpServer/issues/2573
                 var awaiterScheduler = OperatingSystem.IsWindows() ? transportScheduler : PipeScheduler.Inline;
 
                 _settings[i] = new QueueSettings()
                 {
                     Scheduler = transportScheduler,
-                    InputOptions = new PipeOptions(_memoryPool, applicationScheduler, transportScheduler, maxReadBufferSize, maxReadBufferSize / 2, useSynchronizationContext: false),
-                    OutputOptions = new PipeOptions(_memoryPool, transportScheduler, applicationScheduler, maxWriteBufferSize, maxWriteBufferSize / 2, useSynchronizationContext: false),
-                    SocketSenderPool = new SocketSenderPool(awaiterScheduler)
+                    InputOptions = new PipeOptions(memoryPool, applicationScheduler, transportScheduler, maxReadBufferSize, maxReadBufferSize / 2, useSynchronizationContext: false),
+                    OutputOptions = new PipeOptions(memoryPool, transportScheduler, applicationScheduler, maxWriteBufferSize, maxWriteBufferSize / 2, useSynchronizationContext: false),
+                    SocketSenderPool = new SocketSenderPool(awaiterScheduler),
+                    MemoryPool = memoryPool,
                 };
             }
         }
         else
         {
+            var memoryPool = _options.MemoryPoolFactory();
             var transportScheduler = options.UnsafePreferInlineScheduling ? PipeScheduler.Inline : PipeScheduler.ThreadPool;
             // https://github.com/aspnet/KestrelHttpServer/issues/2573
             var awaiterScheduler = OperatingSystem.IsWindows() ? transportScheduler : PipeScheduler.Inline;
             _settings = new QueueSettings[]
             {
-                    new QueueSettings()
-                    {
-                        Scheduler = transportScheduler,
-                        InputOptions = new PipeOptions(_memoryPool, applicationScheduler, transportScheduler, maxReadBufferSize, maxReadBufferSize / 2, useSynchronizationContext: false),
-                        OutputOptions = new PipeOptions(_memoryPool, transportScheduler, applicationScheduler, maxWriteBufferSize, maxWriteBufferSize / 2, useSynchronizationContext: false),
-                        SocketSenderPool = new SocketSenderPool(awaiterScheduler)
-                    }
+                new QueueSettings()
+                {
+                    Scheduler = transportScheduler,
+                    InputOptions = new PipeOptions(memoryPool, applicationScheduler, transportScheduler, maxReadBufferSize, maxReadBufferSize / 2, useSynchronizationContext: false),
+                    OutputOptions = new PipeOptions(memoryPool, transportScheduler, applicationScheduler, maxWriteBufferSize, maxWriteBufferSize / 2, useSynchronizationContext: false),
+                    SocketSenderPool = new SocketSenderPool(awaiterScheduler),
+                    MemoryPool = memoryPool,
+                }
             };
             _settingsCount = 1;
         }
@@ -98,7 +100,7 @@ public ConnectionContext Create(Socket socket)
         var setting = _settings[Interlocked.Increment(ref _settingsIndex) % _settingsCount];
 
         var connection = new SocketConnection(socket,
-            _memoryPool,
+            setting.MemoryPool,
             setting.Scheduler,
             _logger,
             setting.SocketSenderPool,
@@ -113,13 +115,11 @@ public ConnectionContext Create(Socket socket)
     /// <inheritdoc />
     public void Dispose()
     {
-        // Dispose the memory pool
-        _memoryPool.Dispose();
-
-        // Dispose any pooled senders
+        // Dispose any pooled senders and memory pools
         foreach (var setting in _settings)
         {
             setting.SocketSenderPool.Dispose();
+            setting.MemoryPool.Dispose();
         }
     }
 
@@ -129,5 +129,6 @@ private sealed class QueueSettings
         public PipeOptions InputOptions { get; init; } = default!;
         public PipeOptions OutputOptions { get; init; } = default!;
         public SocketSenderPool SocketSenderPool { get; init; } = default!;
+        public MemoryPool<byte> MemoryPool { get; init; } = default!;
     }
 }