Speed up contended HTTP/2 frame writing

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

@@ -36,7 +36,9 @@ 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;
@@ -83,8 +85,11 @@ public Http2Connection(HttpConnectionContext context)
         // Capture the ExecutionContext before dispatching HTTP/2 middleware. Will be restored by streams when processing request
         _context.InitialExecutionContext = ExecutionContext.Capture();
 
+        _input = new Pipe(GetInputPipeOptions());
+        _output = new Pipe(GetOutputPipeOptions());
+
         _frameWriter = new Http2FrameWriter(
-            context.Transport.Output,
+            _output.Writer,
             context.ConnectionContext,
             this,
             _outputFlowControl,
@@ -94,15 +99,6 @@ public Http2Connection(HttpConnectionContext context)
             context.MemoryPool,
             context.ServiceContext);
 
-        var inputOptions = new PipeOptions(pool: context.MemoryPool,
-            readerScheduler: context.ServiceContext.Scheduler,
-            writerScheduler: PipeScheduler.Inline,
-            pauseWriterThreshold: 1,
-            resumeWriterThreshold: 1,
-            minimumSegmentSize: context.MemoryPool.GetMinimumSegmentSize(),
-            useSynchronizationContext: false);
-
-        _input = new Pipe(inputOptions);
         _minAllocBufferSize = context.MemoryPool.GetMinimumAllocSize();
 
         _hpackDecoder = new HPackDecoder(http2Limits.HeaderTableSize, http2Limits.MaxRequestHeaderFieldSize);
@@ -129,7 +125,8 @@ public Http2Connection(HttpConnectionContext context)
 
         _scheduleInline = context.ServiceContext.Scheduler == PipeScheduler.Inline;
 
-        _inputTask = ReadInputAsync();
+        _inputTask = CopyPipeAsync(_context.Transport.Input, _input.Writer);
+        _outputTask = CopyPipeAsync(_output.Reader, _context.Transport.Output);
     }
 
     public string ConnectionId => _context.ConnectionId;
@@ -381,8 +378,10 @@ public async Task ProcessRequestsAsync<TContext>(IHttpApplication<TContext> appl
             finally
             {
                 Input.Complete();
+                _output.Writer.Complete();
                 _context.Transport.Input.CancelPendingRead();
                 await _inputTask;
+                await _outputTask;
             }
         }
     }
@@ -1655,16 +1654,53 @@ public void DecrementActiveClientStreamCount()
         Interlocked.Decrement(ref _clientActiveStreamCount);
     }
 
-    private async Task ReadInputAsync()
+    private PipeOptions GetInputPipeOptions() => new PipeOptions(pool: _context.MemoryPool,
+            readerScheduler: _context.ServiceContext.Scheduler,
+            writerScheduler: PipeScheduler.Inline,
+            pauseWriterThreshold: 1,
+            resumeWriterThreshold: 1,
+            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;
         try
         {
             while (true)
             {
-                var reader = _context.Transport.Input;
-                var writer = _input.Writer;
-
                 var readResult = await reader.ReadAsync();
 
                 if ((readResult.IsCompleted && readResult.Buffer.Length == 0) || readResult.IsCanceled)
@@ -1680,11 +1716,12 @@ private async Task ReadInputAsync()
 
                 bufferSlice.CopyTo(outputBuffer.Span);
 
-                reader.AdvanceTo(bufferSlice.End);
                 writer.Advance(copyAmount);
 
                 var result = await writer.FlushAsync();
 
+                reader.AdvanceTo(bufferSlice.End);
+
                 if (result.IsCompleted || result.IsCanceled)
                 {
                     // flushResult should not be canceled.
@@ -1699,8 +1736,8 @@ private async Task ReadInputAsync()
         }
         finally
         {
-            await _context.Transport.Input.CompleteAsync();
-            _input.Writer.Complete(error);
+            await reader.CompleteAsync(error);
+            await writer.CompleteAsync(error);
         }
     }