Request pipe based data transfer to the application in H/3.

Author: ladeak

src/CHttpServer/CHttpServer/Http3/Http3Stream.QPackHeaderHandler.cs

@@ -14,6 +14,8 @@ internal sealed partial class Http3Stream : IQPackHeaderHandler, IHttpRequestFea
 
     private string _hostDecoded;
     private byte[] _hostEncoded;
+    private Pipe _requestDataToAppPipe;
+
 
     public string Protocol { get => "HTTP/3"; set => throw new PlatformNotSupportedException(); }
     public string PathBase { get => string.Empty; set => throw new PlatformNotSupportedException(); }
@@ -27,7 +29,7 @@ internal sealed partial class Http3Stream : IQPackHeaderHandler, IHttpRequestFea
     IHeaderDictionary IHttpRequestFeature.Headers { get => _requestHeaders; set => throw new PlatformNotSupportedException(); }
 #pragma warning restore CS9266 // Property accessor should use 'field' because the other accessor is using it.
 
-    public PipeReader Reader => PipeReader.Create(Stream.Null);
+    public PipeReader Reader => _requestDataToAppPipe.Reader;
 
     private readonly Http3RequestHeaderCollection _requestHeaders;
 

src/CHttpServer/CHttpServer/Http3/Http3Stream.cs

@@ -35,6 +35,9 @@ public Http3Stream(FeatureCollection features)
         Id = 0;
         _quicStream = null;
         _dataReader = PipeReader.Create(new ReadOnlySequence<byte>());
+        _requestDataToAppPipe = new(new PipeOptions(pauseWriterThreshold: 1024 * 1024));
+        _requestDataToAppPipe.Writer.Complete();
+        _requestDataToAppPipe.Reader.Complete();
         _responseDataWriter = new(Stream.Null, frameType: 0);
         _responseHeaderWriter = new(Stream.Null, frameType: 1);
         _qpackDecoder = new QPackDecoder();
@@ -75,13 +78,13 @@ public void Initialize(Http3Connection? connection, QuicStream quicStream)
         _connection = connection;
         _quicStream = quicStream;
         _dataReader = PipeReader.Create(quicStream);
+        _requestDataToAppPipe.Reset();
         _responseDataWriter.Reset(quicStream, StartAsync);
         _responseHeaderWriter.Reset(quicStream);
         Scheme = string.Empty;
         Method = string.Empty;
         QueryString = string.Empty;
         _isPathSet = false; // The actual Path is not reset.
-        _isHostSet = false; // The actual Host is not reset.
         _features.ResetCheckpoint();
         _requestHeaders.ResetHeaderCollection();
         _responseHeaders.ResetHeaderCollection();
@@ -115,7 +118,15 @@ public async Task ProcessStreamAsync<TContext>(IHttpApplication<TContext> applic
 
                 while (!buffer.IsEmpty)
                 {
-                    long bufferConsumed = ProcessStreamAsync(application, ref payloadRemainingLength, ref applicationProcessing, ref streamReadingState, buffer, token);
+                    long bufferConsumed = 0;
+                    if (streamReadingState == StreamReadingStatus.ReadingFrameHeader)
+                        bufferConsumed = ReadFrameHeader(ref payloadRemainingLength, ref streamReadingState, applicationProcessing, buffer);
+                    else if (streamReadingState == StreamReadingStatus.ReadingPayloadData)
+                        (bufferConsumed, payloadRemainingLength, streamReadingState) = await ProcessDataFrameAsync(payloadRemainingLength, streamReadingState, buffer);
+                    else if (streamReadingState == StreamReadingStatus.ReadingPayloadHeader)
+                        bufferConsumed = ReadHeaderFrame(application, ref payloadRemainingLength, ref applicationProcessing, ref streamReadingState, buffer, token);
+                    else if (streamReadingState == StreamReadingStatus.ReadingPayloadReserved)
+                        bufferConsumed = ReadReservedFrame(ref payloadRemainingLength, ref streamReadingState, buffer);
 
                     // Could not further process. Break the inner loop to read more data
                     if (bufferConsumed == 0)
@@ -148,56 +159,75 @@ public async Task ProcessStreamAsync<TContext>(IHttpApplication<TContext> applic
         }
     }
 
-    private long ProcessStreamAsync<TContext>(IHttpApplication<TContext> application,
+    private long ReadFrameHeader(
         ref long payloadRemainingLength,
-        ref Task? applicationProcessing,
         ref StreamReadingStatus streamReadingState,
-        ReadOnlySequence<byte> buffer,
-        CancellationToken token) where TContext : notnull
+        Task? applicationProcessing,
+        ReadOnlySequence<byte> buffer)
     {
-        long bufferConsumed = 0;
+        if (!VariableLenghtIntegerDecoder.TryRead(buffer.FirstSpan, out var frameType, out int bytesReadFrameType))
+            return 0; // Not enough data to read payload length.
 
-        if (streamReadingState == StreamReadingStatus.ReadingFrameHeader)
-        {
-            // FrameType is a single byte.
-            if (!VariableLenghtIntegerDecoder.TryRead(buffer.FirstSpan, out var frameType, out int bytesReadFrameType))
-            {
-                // Not enough data to read payload length.
-                return 0;
-            }
-            buffer = buffer.Slice(bytesReadFrameType);
-            if (!VariableLenghtIntegerDecoder.TryRead(buffer, out var payloadLength, out int bytesReadPayloadLength))
-            {
-                // Not enough data to read payload length.
-                return 0;
-            }
-            payloadRemainingLength = checked((long)payloadLength);
-            streamReadingState = NextRequestReadingState(applicationProcessing, frameType);
-            buffer = buffer.Slice(bytesReadPayloadLength);
-            bufferConsumed += bytesReadFrameType + bytesReadPayloadLength;
-            return bufferConsumed;
-        }
+        buffer = buffer.Slice(bytesReadFrameType);
+        if (!VariableLenghtIntegerDecoder.TryRead(buffer, out var payloadLength, out int bytesReadPayloadLength))
+            return 0; // Not enough data to read payload length.
 
+        payloadRemainingLength = checked((long)payloadLength);
+        streamReadingState = NextRequestReadingState(applicationProcessing, frameType);
+        buffer = buffer.Slice(bytesReadPayloadLength);
+        var bufferConsumed = bytesReadFrameType + bytesReadPayloadLength;
+        return bufferConsumed;
+    }
+
+    private long ReadReservedFrame(
+        ref long payloadRemainingLength,
+        ref StreamReadingStatus streamReadingState,
+        ReadOnlySequence<byte> buffer)
+    {
         if (payloadRemainingLength < buffer.Length)
             buffer = buffer.Slice(0, payloadRemainingLength);
+        var bufferConsumed = buffer.Length; // Read the complete reserved frame.
+        payloadRemainingLength -= bufferConsumed;
+        if (payloadRemainingLength == 0)
+            streamReadingState = StreamReadingStatus.ReadingFrameHeader;
+        return bufferConsumed;
+    }
 
-        if (streamReadingState == StreamReadingStatus.ReadingPayloadHeader)
+    private long ReadHeaderFrame<TContext>(IHttpApplication<TContext> application,
+        ref long payloadRemainingLength,
+        ref Task? applicationProcessing,
+        ref StreamReadingStatus streamReadingState,
+        ReadOnlySequence<byte> buffer,
+        CancellationToken token) where TContext : notnull
+    {
+        if (payloadRemainingLength < buffer.Length)
+            buffer = buffer.Slice(0, payloadRemainingLength);
+        var bufferConsumed = ProcessHeaderFrame(buffer);
+        payloadRemainingLength -= bufferConsumed;
+        if (payloadRemainingLength == 0)
         {
-            bufferConsumed = ProcessHeaderFrame(buffer);
-            if (payloadRemainingLength == bufferConsumed)
-                applicationProcessing = Task.Run(() => StartApplicationProcessing(application, token), token);
+            streamReadingState = StreamReadingStatus.ReadingFrameHeader;
+            applicationProcessing = Task.Run(() => StartApplicationProcessing(application, token), token);
         }
+        return bufferConsumed;
+    }
 
-        if (streamReadingState == StreamReadingStatus.ReadingPayloadData)
-            bufferConsumed = ProcessDataFrame(buffer);
+    private async Task<(long Consumed, long PayloadRemaining, StreamReadingStatus Status)> ProcessDataFrameAsync(
+        long payloadRemainingLength,
+        StreamReadingStatus streamReadingState,
+        ReadOnlySequence<byte> buffer)
+    {
+        if (payloadRemainingLength < buffer.Length)
+            buffer = buffer.Slice(0, payloadRemainingLength);
 
-        if (streamReadingState == StreamReadingStatus.ReadingPayloadReserved)
-            bufferConsumed = buffer.Length; // Read the complete reserved frame.
+        foreach (var segment in buffer)
+            await _requestDataToAppPipe.Writer.WriteAsync(segment);
+        var bufferConsumed = buffer.Length;
 
         payloadRemainingLength -= bufferConsumed;
         if (payloadRemainingLength == 0)
             streamReadingState = StreamReadingStatus.ReadingFrameHeader;
-        return bufferConsumed;
+        return (bufferConsumed, payloadRemainingLength, streamReadingState);
     }
 
     private StreamReadingStatus NextRequestReadingState(Task? applicationProcessing, ulong frameType)
@@ -227,6 +257,8 @@ private StreamReadingStatus NextRequestReadingState(Task? applicationProcessing,
 
     private async Task CloseStreamAsync()
     {
+        await _requestDataToAppPipe.Writer.CompleteAsync();
+        await _requestDataToAppPipe.Reader.CompleteAsync();
         await _dataReader.CompleteAsync();
         await _responseHeaderWriter.CompleteAsync();
         await _responseDataWriter.CompleteAsync();
@@ -235,13 +267,6 @@ private async Task CloseStreamAsync()
         _connection?.StreamClosed(this);
     }
 
-    private long ProcessDataFrame(ReadOnlySequence<byte> buffer)
-    {
-        // 'Copy' to Body
-        return buffer.Length;
-
-    }
-
     private long ProcessHeaderFrame(ReadOnlySequence<byte> buffer)
     {
         _qpackDecoder.DecodeHeader(buffer, this, out long consumed);