HTTP/2 output processing make over (#40925)

This changes the HTTP/2's output processing to be use queues instead of locks (a channel to be precise). We're manually scheduling the Http2OutputProducer's intent to write instead of a write operation. This removes locks and allows the thread to do other useful work while processing in line for processing in a low allocation way.

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

@@ -36,13 +36,10 @@ internal partial class Http2Connection : IHttp2StreamLifetimeHandler, IHttpStrea
     private readonly HttpConnectionContext _context;
     private readonly Http2FrameWriter _frameWriter;
     private readonly Pipe _input;
-    private readonly Pipe _output;
     private readonly Task _inputTask;
-    private readonly Task _outputTask;
     private readonly int _minAllocBufferSize;
     private readonly HPackDecoder _hpackDecoder;
     private readonly InputFlowControl _inputFlowControl;
-    private readonly OutputFlowControl _outputFlowControl = new OutputFlowControl(new MultipleAwaitableProvider(), Http2PeerSettings.DefaultInitialWindowSize);
 
     private readonly Http2PeerSettings _serverSettings = new Http2PeerSettings();
     private readonly Http2PeerSettings _clientSettings = new Http2PeerSettings();
@@ -74,6 +71,7 @@ internal partial class Http2Connection : IHttp2StreamLifetimeHandler, IHttpStrea
     internal readonly Http2KeepAlive? _keepAlive;
     internal readonly Dictionary<int, Http2Stream> _streams = new Dictionary<int, Http2Stream>();
     internal PooledStreamStack<Http2Stream> StreamPool;
+    internal Action<Http2Stream>? _onStreamCompleted;
 
     public Http2Connection(HttpConnectionContext context)
     {
@@ -86,18 +84,6 @@ public Http2Connection(HttpConnectionContext context)
         _context.InitialExecutionContext = ExecutionContext.Capture();
 
         _input = new Pipe(GetInputPipeOptions());
-        _output = new Pipe(GetOutputPipeOptions());
-
-        _frameWriter = new Http2FrameWriter(
-            _output.Writer,
-            context.ConnectionContext,
-            this,
-            _outputFlowControl,
-            context.TimeoutControl,
-            httpLimits.MinResponseDataRate,
-            context.ConnectionId,
-            context.MemoryPool,
-            context.ServiceContext);
 
         _minAllocBufferSize = context.MemoryPool.GetMinimumAllocSize();
 
@@ -126,7 +112,17 @@ public Http2Connection(HttpConnectionContext context)
         _scheduleInline = context.ServiceContext.Scheduler == PipeScheduler.Inline;
 
         _inputTask = CopyPipeAsync(_context.Transport.Input, _input.Writer);
-        _outputTask = CopyPipeAsync(_output.Reader, _context.Transport.Output);
+
+        _frameWriter = new Http2FrameWriter(
+            context.Transport.Output,
+            context.ConnectionContext,
+            this,
+            (int)Math.Min(MaxTrackedStreams, int.MaxValue),
+            context.TimeoutControl,
+            httpLimits.MinResponseDataRate,
+            context.ConnectionId,
+            context.MemoryPool,
+            context.ServiceContext);
     }
 
     public string ConnectionId => _context.ConnectionId;
@@ -378,10 +374,9 @@ public async Task ProcessRequestsAsync<TContext>(IHttpApplication<TContext> appl
             finally
             {
                 Input.Complete();
-                _output.Writer.Complete();
                 _context.Transport.Input.CancelPendingRead();
                 await _inputTask;
-                await _outputTask;
+                await _frameWriter.ShutdownAsync();
             }
         }
     }
@@ -762,8 +757,7 @@ private Http2StreamContext CreateHttp2StreamContext()
             _clientSettings,
             _serverSettings,
             _frameWriter,
-            _inputFlowControl,
-            _outputFlowControl);
+            _inputFlowControl);
         streamContext.TimeoutControl = _context.TimeoutControl;
         streamContext.InitialExecutionContext = _context.InitialExecutionContext;
 
@@ -1236,6 +1230,7 @@ void IHttp2StreamLifetimeHandler.OnStreamCompleted(Http2Stream stream)
     {
         _completedStreams.Enqueue(stream);
         _streamCompletionAwaitable.Complete();
+        _onStreamCompleted?.Invoke(stream);
     }
 
     private void UpdateCompletedStreams()
@@ -1662,38 +1657,6 @@ public void DecrementActiveClientStreamCount()
             minimumSegmentSize: _context.MemoryPool.GetMinimumSegmentSize(),
             useSynchronizationContext: false);
 
-    private PipeOptions GetOutputPipeOptions()
-    {
-        // Never write inline because we do not want to hold Http2FramerWriter._writeLock for potentially expensive TLS
-        // write operations. This essentially doubles the MaxResponseBufferSize for HTTP/2 connections compared to
-        // HTTP/1.x. This seems reasonable given HTTP/2's support for many concurrent streams per connection. We don't
-        // want every write to return an incomplete ValueTask now that we're dispatching TLS write operations which
-        // would likely happen with a pauseWriterThreshold of 1, but we still need to respect connection back pressure.
-        var pauseWriterThreshold = _context.ServiceContext.ServerOptions.Limits.MaxResponseBufferSize switch
-        {
-            // null means that we have no back pressure
-            null => 0,
-            // 0 = no buffering so we need to configure the pipe so the writer waits on the reader directly
-            0 => 1,
-            long limit => limit,
-        };
-
-        var resumeWriterThreshold = pauseWriterThreshold switch
-        {
-            // The resumeWriterThreshold must be at least 1 to ever resume after pausing.
-            1 => 1,
-            long limit => limit / 2,
-        };
-
-        return new PipeOptions(pool: _context.MemoryPool,
-            readerScheduler: _context.ServiceContext.Scheduler,
-            writerScheduler: PipeScheduler.Inline,
-            pauseWriterThreshold: pauseWriterThreshold,
-            resumeWriterThreshold: resumeWriterThreshold,
-            minimumSegmentSize: _context.MemoryPool.GetMinimumSegmentSize(),
-            useSynchronizationContext: false);
-    }
-
     private async Task CopyPipeAsync(PipeReader reader, PipeWriter writer)
     {
         Exception? error = null;