Improve control stream handling with HTTP/3 (#26644)

Author: jkotalik

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

@@ -5,23 +5,21 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Net;
-using System.Net.Http;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Connections.Experimental;
 using Microsoft.AspNetCore.Connections.Features;
 using Microsoft.AspNetCore.Hosting.Server;
 using Microsoft.AspNetCore.Http.Features;
-using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http3.QPack;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure;
 using Microsoft.Extensions.Logging;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http3
 {
     internal class Http3Connection : ITimeoutHandler
     {
-        public DynamicTable DynamicTable { get; set; }
+        public Http3ControlStream OutboundControlStream { get; set; }
 
         public Http3ControlStream? ControlStream { get; set; }
         public Http3ControlStream? EncoderStream { get; set; }
@@ -47,7 +45,6 @@ public Http3Connection(Http3ConnectionContext context)
         {
             _multiplexedContext = context.ConnectionContext;
             _context = context;
-            DynamicTable = new DynamicTable(0);
             _systemClock = context.ServiceContext.SystemClock;
             _timeoutControl = new TimeoutControl(this);
             _context.TimeoutControl ??= _timeoutControl;
@@ -198,10 +195,16 @@ public void OnTimeout(TimeoutReason reason)
 
         internal async Task InnerProcessRequestsAsync<TContext>(IHttpApplication<TContext> application) where TContext : notnull
         {
-            // Start other three unidirectional streams here.
+            // An endpoint MAY avoid creating an encoder stream if it's not going to
+            // be used(for example if its encoder doesn't wish to use the dynamic
+            // table, or if the maximum size of the dynamic table permitted by the
+            // peer is zero).
+
+            // An endpoint MAY avoid creating a decoder stream if its decoder sets
+            // the maximum capacity of the dynamic table to zero.
+
+            // Don't create Encoder and Decoder as they aren't used now.
             var controlTask = CreateControlStream(application);
-            var encoderTask = CreateEncoderStream(application);
-            var decoderTask = CreateDecoderStream(application);
 
             try
             {
@@ -271,39 +274,33 @@ internal async Task InnerProcessRequestsAsync<TContext>(IHttpApplication<TContex
                             stream.Abort(_abortedException);
                         }
                     }
+                    else
+                    {
+                        foreach (var stream in _streams.Values)
+                        {
+                            stream.OnInputOrOutputCompleted();
+                        }
+                    }
+
+                    _inboundControlStream?.OnInputOrOutputCompleted();
+                    _inboundEncoderStream?.OnInputOrOutputCompleted();
+                    _inboundDecoderStream?.OnInputOrOutputCompleted();
                 }
 
-                ControlStream?.Abort(new ConnectionAbortedException("Connection is shutting down."));
-                EncoderStream?.Abort(new ConnectionAbortedException("Connection is shutting down."));
-                DecoderStream?.Abort(new ConnectionAbortedException("Connection is shutting down."));
+                OutboundControlStream?.Abort(new ConnectionAbortedException("Connection is shutting down."));
 
                 await controlTask;
-                await encoderTask;
-                await decoderTask;
             }
         }
 
         private async ValueTask CreateControlStream<TContext>(IHttpApplication<TContext> application) where TContext : notnull
         {
             var stream = await CreateNewUnidirectionalStreamAsync(application);
-            ControlStream = stream;
+            OutboundControlStream = stream;
             await stream.SendStreamIdAsync(id: 0);
             await stream.SendSettingsFrameAsync();
         }
 
-        private async ValueTask CreateEncoderStream<TContext>(IHttpApplication<TContext> application) where TContext : notnull
-        {
-            var stream = await CreateNewUnidirectionalStreamAsync(application);
-            EncoderStream = stream;
-            await stream.SendStreamIdAsync(id: 2);
-        }
-
-        private async ValueTask CreateDecoderStream<TContext>(IHttpApplication<TContext> application) where TContext : notnull
-        {
-            var stream = await CreateNewUnidirectionalStreamAsync(application);
-            DecoderStream = stream;
-            await stream.SendStreamIdAsync(id: 3);
-        }
 
         private async ValueTask<Http3ControlStream> CreateNewUnidirectionalStreamAsync<TContext>(IHttpApplication<TContext> application) where TContext : notnull
         {
@@ -350,10 +347,10 @@ private void InnerAbort(ConnectionAbortedException ex)
 
             lock (_sync)
             {
-                if (ControlStream != null)
+                if (OutboundControlStream != null)
                 {
                     // TODO need to await this somewhere or allow this to be called elsewhere?
-                    ControlStream.SendGoAway(_highestOpenedStreamId).GetAwaiter().GetResult();
+                    OutboundControlStream.SendGoAway(_highestOpenedStreamId).GetAwaiter().GetResult();
                 }
 
                 _haveSentGoAway = true;
@@ -383,5 +380,44 @@ internal void RemoveStream(long streamId)
                 _streams.Remove(streamId);
             }
         }
+
+        public bool SetInboundControlStream(Http3ControlStream stream)
+        {
+            lock (_sync)
+            {
+                if (_inboundControlStream == null)
+                {
+                    _inboundControlStream = stream;
+                    return true;
+                }
+                return false;
+            }
+        }
+
+        public bool SetInboundEncoderStream(Http3ControlStream stream)
+        {
+            lock (_sync)
+            {
+                if (_inboundEncoderStream == null)
+                {
+                    _inboundEncoderStream = stream;
+                    return true;
+                }
+                return false;
+            }
+        }
+
+        public bool SetInboundDecoderStream(Http3ControlStream stream)
+        {
+            lock (_sync)
+            {
+                if (_inboundDecoderStream == null)
+                {
+                    _inboundDecoderStream = stream;
+                    return true;
+                }
+                return false;
+            }
+        }
     }
 }

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

@@ -82,46 +82,14 @@ private bool TryClose()
         {
             if (Interlocked.Exchange(ref _isClosed, 1) == 0)
             {
+                Input.Complete();
                 return true;
             }
 
             // TODO make this actually close the Http3Stream by telling quic to close the stream.
             return false;
         }
 
-        private async ValueTask HandleEncodingTask()
-        {
-            var encoder = new EncoderStreamReader(10000); // TODO get value from limits
-            while (_isClosed == 0)
-            {
-                var result = await Input.ReadAsync();
-                var readableBuffer = result.Buffer;
-                if (!readableBuffer.IsEmpty)
-                {
-                    // This should always read all bytes in the input no matter what.
-                    encoder.Read(readableBuffer);
-                }
-                Input.AdvanceTo(readableBuffer.End);
-            }
-        }
-
-        private async ValueTask HandleDecodingTask()
-        {
-            var decoder = new DecoderStreamReader();
-            while (_isClosed == 0)
-            {
-                var result = await Input.ReadAsync();
-                var readableBuffer = result.Buffer;
-                var consumed = readableBuffer.Start;
-                var examined = readableBuffer.Start;
-                if (!readableBuffer.IsEmpty)
-                {
-                    decoder.Read(readableBuffer);
-                }
-                Input.AdvanceTo(readableBuffer.End);
-            }
-        }
-
         internal async ValueTask SendStreamIdAsync(long id)
         {
             await _frameWriter.WriteStreamIdAsync(id);
@@ -182,25 +150,26 @@ public async Task ProcessRequestAsync<TContext>(IHttpApplication<TContext> appli
 
             if (streamType == ControlStream)
             {
-                if (_http3Connection.ControlStream != null)
+                if (!_http3Connection.SetInboundControlStream(this))
                 {
+                    // TODO propagate these errors to connection.
                     throw new Http3ConnectionException("HTTP_STREAM_CREATION_ERROR");
                 }
 
                 await HandleControlStream();
             }
             else if (streamType == EncoderStream)
             {
-                if (_http3Connection.EncoderStream != null)
+                if (!_http3Connection.SetInboundEncoderStream(this))
                 {
                     throw new Http3ConnectionException("HTTP_STREAM_CREATION_ERROR");
                 }
+
                 await HandleEncodingTask();
-                return;
             }
             else if (streamType == DecoderStream)
             {
-                if (_http3Connection.DecoderStream != null)
+                if (!_http3Connection.SetInboundDecoderStream(this))
                 {
                     throw new Http3ConnectionException("HTTP_STREAM_CREATION_ERROR");
                 }
@@ -210,7 +179,6 @@ public async Task ProcessRequestAsync<TContext>(IHttpApplication<TContext> appli
             {
                 // TODO Close the control stream as it's unexpected.
             }
-            return;
         }
 
         private async Task HandleControlStream()
@@ -250,6 +218,33 @@ private async Task HandleControlStream()
             }
         }
 
+        private async ValueTask HandleEncodingTask()
+        {
+            // Noop encoding task. Settings make it so we don't need to read content of encoder and decoder.
+            // An endpoint MUST allow its peer to create an encoder stream and a
+            // decoder stream even if the connection's settings prevent their use.
+
+            while (_isClosed == 0)
+            {
+                var result = await Input.ReadAsync();
+                var readableBuffer = result.Buffer;
+                Input.AdvanceTo(readableBuffer.End);
+            }
+        }
+
+        private async ValueTask HandleDecodingTask()
+        {
+            // Noop encoding task. Settings make it so we don't need to read content of encoder and decoder.
+            // An endpoint MUST allow its peer to create an encoder stream and a
+            // decoder stream even if the connection's settings prevent their use.
+            while (_isClosed == 0)
+            {
+                var result = await Input.ReadAsync();
+                var readableBuffer = result.Buffer;
+                Input.AdvanceTo(readableBuffer.End);
+            }
+        }
+
         private ValueTask ProcessHttp3ControlStream(in ReadOnlySequence<byte> payload)
         {
             // Two things:
@@ -283,7 +278,7 @@ private ValueTask ProcessSettingsFrameAsync(ReadOnlySequence<byte> payload)
             }
 
             _haveReceivedSettingsFrame = true;
-            using var closedRegistration = _context.ConnectionContext.ConnectionClosed.Register(state => ((Http3ControlStream)state!).OnStreamClosed(), this);
+            using var closedRegistration = _context.StreamContext.ConnectionClosed.Register(state => ((Http3ControlStream)state!).OnStreamClosed(), this);
 
             while (true)
             {