Cache and reuse NamedPipeServerStream instances

Author: JamesNK

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

@@ -13,15 +13,17 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.NamedPipes.Internal;
 
 internal sealed class NamedPipeConnection : TransportConnection, IConnectionNamedPipeFeature
 {
+    private static readonly ConnectionAbortedException SendGracefullyCompletedException = new ConnectionAbortedException("The named pipe transport's send loop completed gracefully.");
     private const int MinAllocBufferSize = 4096;
-
+    private readonly NamedPipeConnectionListener _connectionListener;
     private readonly NamedPipeServerStream _stream;
     private readonly ILogger _log;
     private readonly IDuplexPipe _originalTransport;
 
     private readonly CancellationTokenSource _connectionClosedTokenSource = new CancellationTokenSource();
     private bool _connectionClosed;
-    private bool _connectionDisposed;
+    private bool _connectionShutdown;
+    private bool _streamDisconnected;
     private Exception? _shutdownReason;
     private readonly object _shutdownLock = new object();
 
@@ -30,13 +32,15 @@ internal sealed class NamedPipeConnection : TransportConnection, IConnectionName
     internal Task _sendingTask = Task.CompletedTask;
 
     public NamedPipeConnection(
+        NamedPipeConnectionListener connectionListener,
         NamedPipeServerStream stream,
         NamedPipeEndPoint endPoint,
         ILogger logger,
         MemoryPool<byte> memoryPool,
         PipeOptions inputOptions,
         PipeOptions outputOptions)
     {
+        _connectionListener = connectionListener;
         _stream = stream;
         _log = logger;
         MemoryPool = memoryPool;
@@ -120,7 +124,7 @@ private async Task DoReceiveAsync()
             // This exception should always be ignored because _shutdownReason should be set.
             error = ex;
 
-            if (!_connectionDisposed)
+            if (!_connectionShutdown)
             {
                 // This is unexpected if the socket hasn't been disposed yet.
                 NamedPipeLog.ConnectionError(_log, this, error);
@@ -206,33 +210,32 @@ private void Shutdown(Exception? shutdownReason)
     {
         lock (_shutdownLock)
         {
-            if (_connectionDisposed)
+            if (_connectionShutdown)
             {
                 return;
             }
 
             // Make sure to close the connection only after the _aborted flag is set.
             // Without this, the RequestsCanBeAbortedMidRead test will sometimes fail when
             // a BadHttpRequestException is thrown instead of a TaskCanceledException.
-            _connectionDisposed = true;
+            _connectionShutdown = true;
 
             // shutdownReason should only be null if the output was completed gracefully, so no one should ever
             // ever observe the nondescript ConnectionAbortedException except for connection middleware attempting
             // to half close the connection which is currently unsupported.
-            _shutdownReason = shutdownReason ?? new ConnectionAbortedException("The Socket transport's send loop completed gracefully.");
+            _shutdownReason = shutdownReason ?? SendGracefullyCompletedException;
             NamedPipeLog.ConnectionDisconnect(_log, this, _shutdownReason.Message);
 
             try
             {
                 // Try to gracefully close the socket even for aborts to match libuv behavior.
                 _stream.Disconnect();
+                _streamDisconnected = true;
             }
             catch
             {
                 // Ignore any errors from NamedPipeStream.Disconnect() since we're tearing down the connection anyway.
             }
-
-            _stream.Dispose();
         }
     }
 
@@ -287,8 +290,13 @@ public override async ValueTask DisposeAsync()
         catch (Exception ex)
         {
             _log.LogError(0, ex, $"Unexpected exception in {nameof(NamedPipeConnection)}.{nameof(Start)}.");
+            _stream.Dispose();
+            return;
+        }
+
+        if (!_streamDisconnected || !_connectionListener.TryCacheStream(_stream))
+        {
+            _stream.Dispose();
         }
-        
-        await _stream.DisposeAsync();
     }
 }

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

@@ -2,6 +2,7 @@
 // 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;
@@ -26,6 +27,7 @@ internal sealed class NamedPipeConnectionListener : IConnectionListener
     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;
 
@@ -54,6 +56,19 @@ public NamedPipeConnectionListener(
         _outputOptions = new PipeOptions(_memoryPool, PipeScheduler.Inline, PipeScheduler.ThreadPool, maxWriteBufferSize, maxWriteBufferSize / 2, useSynchronizationContext: false);
     }
 
+    internal bool TryCacheStream(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;
+    }
+
     public void Start()
     {
         Debug.Assert(_listeningTasks == null, "Already started");
@@ -83,13 +98,13 @@ private async Task StartAsync(NamedPipeServerStream nextStream)
 
                     await stream.WaitForConnectionAsync(_listeningToken);
 
-                    var connection = new NamedPipeConnection(stream, _endpoint, _log, _memoryPool, _inputOptions, _outputOptions);
+                    var connection = new NamedPipeConnection(this, stream, _endpoint, _log, _memoryPool, _inputOptions, _outputOptions);
                     connection.Start();
 
                     // 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 = CreateServerStream();
+                    nextStream = GetOrCreateServerStream();
 
                     while (!_acceptedQueue.Writer.TryWrite(connection))
                     {
@@ -106,7 +121,7 @@ private async Task StartAsync(NamedPipeServerStream nextStream)
 
                     // Dispose existing pipe, create a new one and continue accepting.
                     nextStream.Dispose();
-                    nextStream = CreateServerStream();
+                    nextStream = GetOrCreateServerStream();
                 }
                 catch (OperationCanceledException ex) when (_listeningToken.IsCancellationRequested)
                 {
@@ -125,6 +140,17 @@ 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;