HTTP/3: Http3Stream pooling (builds on top of QUIC stream pooling)

src/Servers/Connections.Abstractions/src/ConnectionContext.cs

@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Collections.Generic;
 using System.IO.Pipelines;
 using Microsoft.AspNetCore.Connections.Features;
 
@@ -12,6 +13,30 @@ namespace Microsoft.AspNetCore.Connections
     /// </summary>
     public abstract class ConnectionContext : BaseConnectionContext, IAsyncDisposable
     {
+        internal IDictionary<object, object?>? _persistentState;
+
+        /// <summary>
+        /// Gets or sets a key/value collection that can be used to persist state between connections.
+        /// Whether a transport pools and reuses <see cref="ConnectionContext"/> instances and allows state to
+        /// be persisted depends on the transport implementation.
+        /// <para>
+        /// Because values added to persistent state can live in memory until a <see cref="ConnectionContext"/>
+        /// is no longer pooled, use caution with this collection to avoid excessive memory use.
+        /// </para>
+        /// </summary>
+        public virtual IDictionary<object, object?> PersistentState
+        {
+            get
+            {
+                // Lazily allocate connection metadata
+                return _persistentState ?? (_persistentState = new ConnectionItems());
+            }
+            set
+            {
+                _persistentState = value;
+            }
+        }
+
         /// <summary>
         /// Gets or sets the <see cref="IDuplexPipe"/> that can be used to read or write data on this connection.
         /// </summary>

src/Servers/Connections.Abstractions/src/PublicAPI.Unshipped.txt

@@ -25,3 +25,5 @@ Microsoft.AspNetCore.Connections.MultiplexedConnectionContext.MultiplexedConnect
 Microsoft.AspNetCore.Connections.MultiplexedConnectionDelegate
 abstract Microsoft.AspNetCore.Connections.MultiplexedConnectionContext.AcceptAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask<Microsoft.AspNetCore.Connections.ConnectionContext?>
 abstract Microsoft.AspNetCore.Connections.MultiplexedConnectionContext.ConnectAsync(Microsoft.AspNetCore.Http.Features.IFeatureCollection? features = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask<Microsoft.AspNetCore.Connections.ConnectionContext!>
+virtual Microsoft.AspNetCore.Connections.ConnectionContext.PersistentState.get -> System.Collections.Generic.IDictionary<object!, object?>!
+virtual Microsoft.AspNetCore.Connections.ConnectionContext.PersistentState.set -> void

src/Servers/Kestrel/Core/src/Internal/Http/Http1OutputProducer.cs

@@ -91,7 +91,8 @@ public Http1OutputProducer(
             _minResponseDataRateFeature = minResponseDataRateFeature;
             _outputAborter = outputAborter;
 
-            _flusher = new TimingPipeFlusher(_pipeWriter, timeoutControl, log);
+            _flusher = new TimingPipeFlusher(timeoutControl, log);
+            _flusher.Initialize(_pipeWriter);
         }
 
         public Task WriteDataAsync(ReadOnlySpan<byte> buffer, CancellationToken cancellationToken = default)

src/Servers/Kestrel/Core/src/Internal/Http2/Http2FrameWriter.cs

@@ -67,7 +67,8 @@ public Http2FrameWriter(
             _log = serviceContext.Log;
             _timeoutControl = timeoutControl;
             _minResponseDataRate = minResponseDataRate;
-            _flusher = new TimingPipeFlusher(_outputWriter, timeoutControl, serviceContext.Log);
+            _flusher = new TimingPipeFlusher(timeoutControl, serviceContext.Log);
+            _flusher.Initialize(_outputWriter);
             _outgoingFrame = new Http2Frame();
             _headerEncodingBuffer = new byte[_maxFrameSize];
 

src/Servers/Kestrel/Core/src/Internal/Http2/Http2OutputProducer.cs

@@ -72,7 +72,8 @@ public Http2OutputProducer(Http2Stream stream, Http2StreamContext context, Strea
 
             // No need to pass in timeoutControl here, since no minDataRates are passed to the TimingPipeFlusher.
             // The minimum output data rate is enforced at the connection level by Http2FrameWriter.
-            _flusher = new TimingPipeFlusher(_pipeWriter, timeoutControl: null, _log);
+            _flusher = new TimingPipeFlusher(timeoutControl: null, _log);
+            _flusher.Initialize(_pipeWriter);
 
             _dataWriteProcessingTask = ProcessDataWrites();
         }

src/Servers/Kestrel/Core/src/Internal/Http2/Http2StreamContext.cs

@@ -26,7 +26,7 @@ public Http2StreamContext(
             Http2PeerSettings serverPeerSettings,
             Http2FrameWriter frameWriter,
             InputFlowControl connectionInputFlowControl,
-            OutputFlowControl connectionOutputFlowControl) : base(connectionId, protocols, connectionContext: null!, serviceContext, connectionFeatures, memoryPool, localEndPoint, remoteEndPoint, transport: null!)
+            OutputFlowControl connectionOutputFlowControl) : base(connectionId, protocols, connectionContext: null!, serviceContext, connectionFeatures, memoryPool, localEndPoint, remoteEndPoint)
         {
             StreamId = streamId;
             StreamLifetimeHandler = streamLifetimeHandler;

src/Servers/Kestrel/Core/src/Internal/Http3/Http3Connection.cs

@@ -23,6 +23,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http3
 {
     internal class Http3Connection : IHttp3StreamLifetimeHandler, IRequestProcessor
     {
+        private static readonly object StreamPersistentStateKey = new object();
+
         // Internal for unit testing
         internal readonly Dictionary<long, IHttp3Stream> _streams = new Dictionary<long, IHttp3Stream>();
         internal IHttp3StreamLifetimeHandler _streamLifetimeHandler;
@@ -257,26 +259,10 @@ public async Task ProcessRequestsAsync<TContext>(IHttpApplication<TContext> appl
                         Debug.Assert(streamDirectionFeature != null);
                         Debug.Assert(streamIdFeature != null);
 
-                        var httpConnectionContext = new Http3StreamContext(
-                            streamContext.ConnectionId,
-                            protocols: default,
-                            connectionContext: null!, // TODO connection context is null here. Should we set it to anything?
-                            _context.ServiceContext,
-                            streamContext.Features,
-                            _context.MemoryPool,
-                            streamContext.LocalEndPoint as IPEndPoint,
-                            streamContext.RemoteEndPoint as IPEndPoint,
-                            streamContext.Transport,
-                            _streamLifetimeHandler,
-                            streamContext,
-                            _clientSettings,
-                            _serverSettings);
-                        httpConnectionContext.TimeoutControl = _context.TimeoutControl;
-
                         if (!streamDirectionFeature.CanWrite)
                         {
                             // Unidirectional stream
-                            var stream = new Http3ControlStream<TContext>(application, httpConnectionContext);
+                            var stream = new Http3ControlStream<TContext>(application, CreateHttpStreamContext(streamContext));
                             _streamLifetimeHandler.OnStreamCreated(stream);
 
                             ThreadPool.UnsafeQueueUserWorkItem(stream, preferLocal: false);
@@ -286,7 +272,21 @@ public async Task ProcessRequestsAsync<TContext>(IHttpApplication<TContext> appl
                             // Request stream
                             UpdateHighestStreamId(streamIdFeature.StreamId);
 
-                            var stream = new Http3Stream<TContext>(application, httpConnectionContext);
+                            Http3Stream<TContext> stream;
+
+                            // Check whether there is an existing HTTP/3 stream on the transport stream.
+                            // A stream will only be cached if the transport stream itself is reused.
+                            if (!streamContext.PersistentState.TryGetValue(StreamPersistentStateKey, out var s))
+                            {
+                                stream = new Http3Stream<TContext>(application, CreateHttpStreamContext(streamContext));
+                                streamContext.PersistentState.Add(StreamPersistentStateKey, stream);
+                            }
+                            else
+                            {
+                                stream = (Http3Stream<TContext>)s!;
+                                stream.InitializeWithExistingContext(streamContext.Transport);
+                            }
+
                             _streamLifetimeHandler.OnStreamCreated(stream);
 
                             KestrelEventSource.Log.RequestQueuedStart(stream, AspNetCore.Http.HttpProtocol.Http3);
@@ -371,6 +371,27 @@ public async Task ProcessRequestsAsync<TContext>(IHttpApplication<TContext> appl
             }
         }
 
+        private Http3StreamContext CreateHttpStreamContext(ConnectionContext streamContext)
+        {
+            var httpConnectionContext = new Http3StreamContext(
+                streamContext.ConnectionId,
+                protocols: default,
+                connectionContext: null!, // TODO connection context is null here. Should we set it to anything?
+                _context.ServiceContext,
+                streamContext.Features,
+                _context.MemoryPool,
+                streamContext.LocalEndPoint as IPEndPoint,
+                streamContext.RemoteEndPoint as IPEndPoint,
+                _streamLifetimeHandler,
+                streamContext,
+                _clientSettings,
+                _serverSettings);
+            httpConnectionContext.TimeoutControl = _context.TimeoutControl;
+            httpConnectionContext.Transport = streamContext.Transport;
+
+            return httpConnectionContext;
+        }
+
         private void UpdateConnectionState()
         {
             if (_isClosed != 0)
@@ -443,12 +464,12 @@ private async ValueTask<Http3ControlStream> CreateNewUnidirectionalStreamAsync<T
                 _context.MemoryPool,
                 streamContext.LocalEndPoint as IPEndPoint,
                 streamContext.RemoteEndPoint as IPEndPoint,
-                streamContext.Transport,
                 _streamLifetimeHandler,
                 streamContext,
                 _clientSettings,
                 _serverSettings);
             httpConnectionContext.TimeoutControl = _context.TimeoutControl;
+            httpConnectionContext.Transport = streamContext.Transport;
 
             return new Http3ControlStream<TContext>(application, httpConnectionContext);
         }

src/Servers/Kestrel/Core/src/Internal/Http3/Http3ControlStream.cs

@@ -48,16 +48,15 @@ public Http3ControlStream(Http3StreamContext context)
             _headerType = -1;
 
             _frameWriter = new Http3FrameWriter(
-                context.Transport.Output,
                 context.StreamContext,
                 context.TimeoutControl,
                 httpLimits.MinResponseDataRate,
-                context.ConnectionId,
                 context.MemoryPool,
                 context.ServiceContext.Log,
                 _streamIdFeature,
                 context.ClientPeerSettings,
                 this);
+            _frameWriter.Reset(context.Transport.Output, context.ConnectionId);
         }
 
         private void OnStreamClosed()

src/Servers/Kestrel/Core/src/Internal/Http3/Http3FrameWriter.cs

@@ -28,18 +28,19 @@ internal class Http3FrameWriter
         private readonly object _writeLock = new object();
 
         private readonly int _maxTotalHeaderSize;
-        private readonly PipeWriter _outputWriter;
         private readonly ConnectionContext _connectionContext;
         private readonly ITimeoutControl _timeoutControl;
         private readonly MinDataRate? _minResponseDataRate;
-        private readonly string _connectionId;
         private readonly MemoryPool<byte> _memoryPool;
         private readonly IKestrelTrace _log;
         private readonly IStreamIdFeature _streamIdFeature;
         private readonly IHttp3Stream _http3Stream;
         private readonly Http3RawFrame _outgoingFrame;
         private readonly TimingPipeFlusher _flusher;
 
+        private PipeWriter _outputWriter = default!;
+        private string _connectionId = default!;
+
         // HTTP/3 doesn't have a max frame size (peer can optionally specify a size).
         // Write headers to a buffer that can grow. Possible performance improvement
         // by writing directly to output writer (difficult as frame length is prefixed).
@@ -51,19 +52,17 @@ internal class Http3FrameWriter
         private bool _completed;
         private bool _aborted;
 
-        public Http3FrameWriter(PipeWriter output, ConnectionContext connectionContext, ITimeoutControl timeoutControl, MinDataRate? minResponseDataRate, string connectionId, MemoryPool<byte> memoryPool, IKestrelTrace log, IStreamIdFeature streamIdFeature, Http3PeerSettings clientPeerSettings, IHttp3Stream http3Stream)
+        public Http3FrameWriter(ConnectionContext connectionContext, ITimeoutControl timeoutControl, MinDataRate? minResponseDataRate, MemoryPool<byte> memoryPool, IKestrelTrace log, IStreamIdFeature streamIdFeature, Http3PeerSettings clientPeerSettings, IHttp3Stream http3Stream)
         {
-            _outputWriter = output;
             _connectionContext = connectionContext;
             _timeoutControl = timeoutControl;
             _minResponseDataRate = minResponseDataRate;
-            _connectionId = connectionId;
             _memoryPool = memoryPool;
             _log = log;
             _streamIdFeature = streamIdFeature;
             _http3Stream = http3Stream;
             _outgoingFrame = new Http3RawFrame();
-            _flusher = new TimingPipeFlusher(_outputWriter, timeoutControl, log);
+            _flusher = new TimingPipeFlusher(timeoutControl, log);
             _headerEncodingBuffer = new ArrayBufferWriter<byte>(HeaderBufferSize);
 
             // Note that max total header size value doesn't react to settings change during a stream.
@@ -75,6 +74,19 @@ public Http3FrameWriter(PipeWriter output, ConnectionContext connectionContext,
                 : (int)clientPeerSettings.MaxRequestHeaderFieldSectionSize;
         }
 
+        public void Reset(PipeWriter output, string connectionId)
+        {
+            _outputWriter = output;
+            _flusher.Initialize(output);
+            _connectionId = connectionId;
+
+            _headersTotalSize = 0;
+            _headerEncodingBuffer.Clear();
+            _unflushedBytes = 0;
+            _completed = false;
+            _aborted = false;
+        }
+
         internal Task WriteSettingsAsync(List<Http3PeerSetting> settings)
         {
             _outgoingFrame.PrepareSettings();