Use object pool

Author: JamesNK

src/Servers/Kestrel/Transport.NamedPipes/src/Internal/NamedPipeConnection.cs

@@ -294,9 +294,13 @@ public override async ValueTask DisposeAsync()
             return;
         }
 
-        if (!_streamDisconnected || !_connectionListener.TryCacheStream(_stream))
+        if (!_streamDisconnected)
         {
             _stream.Dispose();
         }
+        else
+        {
+            _connectionListener.ReturnStream(_stream);
+        }
     }
 }

src/Servers/Kestrel/Transport.NamedPipes/src/Internal/NamedPipeConnectionListener.cs

@@ -2,14 +2,14 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 
 using System.Buffers;
-using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.IO.Pipelines;
 using System.IO.Pipes;
 using System.Net;
 using System.Threading.Channels;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.ObjectPool;
 using NamedPipeOptions = System.IO.Pipes.PipeOptions;
 using PipeOptions = System.IO.Pipelines.PipeOptions;
 
@@ -20,14 +20,14 @@ internal sealed class NamedPipeConnectionListener : IConnectionListener
     private readonly ILogger _log;
     private readonly NamedPipeEndPoint _endpoint;
     private readonly NamedPipeTransportOptions _options;
+    private readonly ObjectPool<NamedPipeServerStream> _namedPipeServerStreamPool;
     private readonly CancellationTokenSource _listeningTokenSource = new CancellationTokenSource();
     private readonly CancellationToken _listeningToken;
     private readonly Channel<ConnectionContext> _acceptedQueue;
     private readonly MemoryPool<byte> _memoryPool;
     private readonly PipeOptions _inputOptions;
     private readonly PipeOptions _outputOptions;
     private readonly Mutex _mutex;
-    private readonly ConcurrentQueue<NamedPipeServerStream> _streamsCache = new ConcurrentQueue<NamedPipeServerStream>();
     private Task[]? _listeningTasks;
     private int _disposed;
 
@@ -40,6 +40,7 @@ public NamedPipeConnectionListener(
         _log = loggerFactory.CreateLogger("Microsoft.AspNetCore.Server.Kestrel.Transport.NamedPipes");
         _endpoint = endpoint;
         _options = options;
+        _namedPipeServerStreamPool = new DefaultObjectPoolProvider().Create(new NamedPipeServerStreamPoolPolicy(this));
         _mutex = mutex;
         _memoryPool = options.MemoryPoolFactory();
         _listeningToken = _listeningTokenSource.Token;
@@ -56,17 +57,10 @@ public NamedPipeConnectionListener(
         _outputOptions = new PipeOptions(_memoryPool, PipeScheduler.Inline, PipeScheduler.ThreadPool, maxWriteBufferSize, maxWriteBufferSize / 2, useSynchronizationContext: false);
     }
 
-    internal bool TryCacheStream(NamedPipeServerStream namedPipeServerStream)
+    internal void ReturnStream(NamedPipeServerStream namedPipeServerStream)
     {
-        // Limit the number of cached named pipe server streams.
-        // This isn't thread safe so it's possible for Count and Enqueue to race and slightly exceed this limit.
-        if (_streamsCache.Count <= 50)
-        {
-            _streamsCache.Enqueue(namedPipeServerStream);
-            return true;
-        }
-
-        return false;
+        // The stream is automatically disposed if there isn't space in the pool.
+        _namedPipeServerStreamPool.Return(namedPipeServerStream);
     }
 
     public void Start()
@@ -78,7 +72,7 @@ public void Start()
         for (var i = 0; i < _listeningTasks.Length; i++)
         {
             // Start first stream inline to catch creation errors.
-            var initialStream = CreateServerStream();
+            var initialStream = _namedPipeServerStreamPool.Get();
 
             _listeningTasks[i] = Task.Run(() => StartAsync(initialStream));
         }
@@ -104,7 +98,7 @@ private async Task StartAsync(NamedPipeServerStream nextStream)
                     // Create the next stream before writing connected stream to the channel.
                     // This ensures there is always a created stream and another process can't
                     // create a stream with the same name with different a access policy.
-                    nextStream = GetOrCreateServerStream();
+                    nextStream = _namedPipeServerStreamPool.Get();
 
                     while (!_acceptedQueue.Writer.TryWrite(connection))
                     {
@@ -121,7 +115,7 @@ private async Task StartAsync(NamedPipeServerStream nextStream)
 
                     // Dispose existing pipe, create a new one and continue accepting.
                     nextStream.Dispose();
-                    nextStream = GetOrCreateServerStream();
+                    nextStream = _namedPipeServerStreamPool.Get();
                 }
                 catch (OperationCanceledException ex) when (_listeningToken.IsCancellationRequested)
                 {
@@ -140,52 +134,6 @@ private async Task StartAsync(NamedPipeServerStream nextStream)
         }
     }
 
-    private NamedPipeServerStream GetOrCreateServerStream()
-    {
-        if (!_streamsCache.TryDequeue(out var stream))
-        {
-            // Cache is empty. Create a new server stream.
-            stream = CreateServerStream();
-        }
-
-        return stream;
-    }
-
-    private NamedPipeServerStream CreateServerStream()
-    {
-        NamedPipeServerStream stream;
-        var pipeOptions = NamedPipeOptions.Asynchronous | NamedPipeOptions.WriteThrough;
-        if (_options.CurrentUserOnly)
-        {
-            pipeOptions |= NamedPipeOptions.CurrentUserOnly;
-        }
-
-        if (_options.PipeSecurity != null)
-        {
-            stream = NamedPipeServerStreamAcl.Create(
-                _endpoint.PipeName,
-                PipeDirection.InOut,
-                NamedPipeServerStream.MaxAllowedServerInstances,
-                PipeTransmissionMode.Byte,
-                pipeOptions,
-                inBufferSize: 0, // Buffer in System.IO.Pipelines
-                outBufferSize: 0, // Buffer in System.IO.Pipelines
-                _options.PipeSecurity);
-        }
-        else
-        {
-            stream = new NamedPipeServerStream(
-                _endpoint.PipeName,
-                PipeDirection.InOut,
-                NamedPipeServerStream.MaxAllowedServerInstances,
-                PipeTransmissionMode.Byte,
-                pipeOptions,
-                inBufferSize: 0,
-                outBufferSize: 0);
-        }
-        return stream;
-    }
-
     public async ValueTask<ConnectionContext?> AcceptAsync(CancellationToken cancellationToken = default)
     {
         while (await _acceptedQueue.Reader.WaitToReadAsync(cancellationToken))
@@ -217,5 +165,56 @@ public async ValueTask DisposeAsync()
         {
             await Task.WhenAll(_listeningTasks);
         }
+
+        // Dispose pool after listening tasks are complete so there is no chance a stream
+        // is fetched from the pool after the pool is disposed.
+        ((IDisposable)_namedPipeServerStreamPool).Dispose();
+    }
+
+    private sealed class NamedPipeServerStreamPoolPolicy : IPooledObjectPolicy<NamedPipeServerStream>
+    {
+        public NamedPipeConnectionListener _listener;
+
+        public NamedPipeServerStreamPoolPolicy(NamedPipeConnectionListener listener)
+        {
+            _listener = listener;
+        }
+
+        public NamedPipeServerStream Create()
+        {
+            NamedPipeServerStream stream;
+            var pipeOptions = NamedPipeOptions.Asynchronous | NamedPipeOptions.WriteThrough;
+            if (_listener._options.CurrentUserOnly)
+            {
+                pipeOptions |= NamedPipeOptions.CurrentUserOnly;
+            }
+
+            if (_listener._options.PipeSecurity != null)
+            {
+                stream = NamedPipeServerStreamAcl.Create(
+                    _listener._endpoint.PipeName,
+                    PipeDirection.InOut,
+                    NamedPipeServerStream.MaxAllowedServerInstances,
+                    PipeTransmissionMode.Byte,
+                    pipeOptions,
+                    inBufferSize: 0, // Buffer in System.IO.Pipelines
+                    outBufferSize: 0, // Buffer in System.IO.Pipelines
+                    _listener._options.PipeSecurity);
+            }
+            else
+            {
+                stream = new NamedPipeServerStream(
+                    _listener._endpoint.PipeName,
+                    PipeDirection.InOut,
+                    NamedPipeServerStream.MaxAllowedServerInstances,
+                    PipeTransmissionMode.Byte,
+                    pipeOptions,
+                    inBufferSize: 0,
+                    outBufferSize: 0);
+            }
+            return stream;
+        }
+
+        public bool Return(NamedPipeServerStream obj) => true;
     }
 }

src/Servers/Kestrel/Transport.NamedPipes/src/Microsoft.AspNetCore.Server.Kestrel.Transport.NamedPipes.csproj

@@ -26,6 +26,7 @@
   <ItemGroup>
     <Reference Include="Microsoft.AspNetCore.Hosting.Abstractions" />
     <Reference Include="Microsoft.AspNetCore.Connections.Abstractions" />
+    <Reference Include="Microsoft.Extensions.ObjectPool" />
     <Reference Include="Microsoft.Extensions.Options" />
   </ItemGroup>