- Segments cache

- Initial tests for Http3DeframingPipeReader

Author: ladeak

src/CHttpServer/CHttpServer/Http3/Http3DeframingPipeReader.cs

@@ -2,8 +2,6 @@
 using System.IO.Pipelines;
 using CHttpServer.Http3;
 
-// TODO buffer segments
-
 internal sealed class Http3DeframingPipeReader : PipeReader
 {
     private enum StreamReadingStatus
@@ -15,17 +13,17 @@ private enum StreamReadingStatus
 
     private class Segment : ReadOnlySequenceSegment<byte>
     {
-        public Segment(ReadOnlyMemory<byte> data, object? source, int sourceOffset, long framePayloadRemaining)
+        public void Initialize(ReadOnlyMemory<byte> data, object? source, int sourceOffset, long framePayloadRemaining)
         {
             base.Memory = data;
             Source = source;
             SourceOffset = sourceOffset;
             FramePayloadRemaining = framePayloadRemaining;
         }
 
-        public object? Source { get; }
-        public int SourceOffset { get; }
-        public long FramePayloadRemaining { get; }
+        public object? Source { get; private set; }
+        public int SourceOffset { get; private set; }
+        public long FramePayloadRemaining { get; private set; }
 
         public Segment SetNext(Segment s)
         {
@@ -54,6 +52,15 @@ public Segment Slice(int start)
             throw new InvalidOperationException("Not a Segment");
         }
 
+        public void Reset()
+        {
+            Memory = ReadOnlyMemory<byte>.Empty;
+            Source = null;
+            SourceOffset = 0; FramePayloadRemaining = 0;
+            RunningIndex = 0;
+            Next = null;
+        }
+
         public SequencePosition End => new SequencePosition(this, Memory.Length);
     }
 
@@ -62,10 +69,12 @@ public Segment Slice(int start)
     private long _payloadRemainingLength = 0;
     private Segment? _head;
     private Segment? _tail;
+    private Stack<Segment> _segmentsBuffer;
 
     public Http3DeframingPipeReader(PipeReader pipeReader)
     {
         _pipeReader = pipeReader;
+        _segmentsBuffer = new Stack<Segment>(capacity: 10);
     }
 
     public void Reset(PipeReader pipeReader)
@@ -77,7 +86,7 @@ public void Reset(PipeReader pipeReader)
         _streamReadingState = StreamReadingStatus.ReadingFrameHeader;
     }
 
-    private void AddSegment(Segment s)
+    private void AppendDataSegment(Segment s)
     {
         if (_tail == null)
             _head = _tail = s;
@@ -110,6 +119,9 @@ public override void AdvanceTo(SequencePosition consumed, SequencePosition exami
         sourceObject = segmentExamined.Source;
         examined = new SequencePosition(sourceObject, sourceOffset);
 
+        while (_head != segmentConsumed && _head != null)
+            _head = ReturnSegment(_head);
+
         _payloadRemainingLength = segmentConsumed.FramePayloadRemaining - segmentOffset;
         if (_payloadRemainingLength == 0)
             _streamReadingState = StreamReadingStatus.ReadingFrameHeader;
@@ -155,6 +167,8 @@ public override async ValueTask<ReadResult> ReadAsync(CancellationToken token =
     // at the beginning of the segment.
     private bool ProcessReadResult(ReadOnlySequence<byte> buffer, out ReadOnlySequence<byte> sequence)
     {
+        while (_head != null)
+            _head = ReturnSegment(_head);
         _head = _tail = null;
         sequence = ReadOnlySequence<byte>.Empty;
         long bufferConsumed = 0;
@@ -173,9 +187,10 @@ private bool ProcessReadResult(ReadOnlySequence<byte> buffer, out ReadOnlySequen
                 foreach (var s in dataPayload)
                 {
                     var position = buffer.GetPosition(currentPosition);
-                    var segment = new Segment(s, position.GetObject(), position.GetInteger(), currentPayloadRemainingLength);
+                    var segment = RentSegment();
+                    segment.Initialize(s, position.GetObject(), position.GetInteger(), currentPayloadRemainingLength);
                     currentPayloadRemainingLength -= s.Length;
-                    AddSegment(segment);
+                    AppendDataSegment(segment);
                     currentPosition += s.Length;
                 }
             }
@@ -259,6 +274,21 @@ private static StreamReadingStatus NextRequestReadingState(ulong frameType)
         return streamReadingState;
     }
 
+    private Segment RentSegment()
+    {
+        if (!_segmentsBuffer.TryPop(out var segment))
+            segment = new Segment();
+        return segment;
+    }
+
+    private Segment? ReturnSegment(Segment segment)
+    {
+        var next = segment.Next;
+        segment.Reset();
+        _segmentsBuffer.Push(segment);
+        return next as Segment;
+    }
+
     public override async Task CopyToAsync(PipeWriter destination, CancellationToken cancellationToken = default)
     {
         StreamReadingStatus streamState = StreamReadingStatus.ReadingFrameHeader;

tests/CHttpServer.Tests/Http3/Http3DeframingPipeReaderTests.cs

@@ -0,0 +1,28 @@
+using System.IO.Pipelines;
+using CHttpServer.Http3;
+
+namespace CHttpServer.Tests.Http3;
+
+public class Http3DeframingPipeReaderTests
+{
+    [Fact]
+    public async Task HeadersFrameThrows()
+    {
+        var pipe = new Pipe();
+        var sut = new Http3DeframingPipeReader(pipe.Reader);
+        await pipe.Writer.WriteAsync(Http3FrameFixture.GetHeadersFrame(), TestContext.Current.CancellationToken);
+        await Assert.ThrowsAsync<Http3ConnectionException>(async () => await sut.ReadAsync(TestContext.Current.CancellationToken));
+    }
+
+    [Fact]
+    public async Task ReservedFrameIgnored()
+    {
+        var pipe = new Pipe();
+        var sut = new Http3DeframingPipeReader(pipe.Reader);
+        await pipe.Writer.WriteAsync(Http3FrameFixture.GetReservedFrame(100), TestContext.Current.CancellationToken);
+        pipe.Writer.Complete();
+        var result = await sut.ReadAsync(TestContext.Current.CancellationToken);
+        Assert.True(result.IsCompleted);
+        Assert.True(result.Buffer.IsEmpty);
+    }
+}

tests/CHttpServer.Tests/Http3/Http3FrameFixture.cs

@@ -0,0 +1,101 @@
+using System.IO.Pipelines;
+using CHttpServer.Http3;
+
+namespace CHttpServer.Tests.Http3;
+
+internal class TestPipeWriter : PipeWriter
+{
+    private byte[] _buffer = new byte[4096];
+    private int _consumedLength = 0;
+
+    public ReadOnlySpan<byte> WrittenData => _buffer.AsSpan(0, _consumedLength);
+
+    public override void Advance(int bytes)
+    {
+        _consumedLength += bytes;
+    }
+
+    public override void CancelPendingFlush()
+    {
+    }
+
+    public override void Complete(Exception? exception = null)
+    {
+    }
+
+    public override ValueTask<FlushResult> FlushAsync(CancellationToken cancellationToken = default)
+    {
+        return ValueTask.FromResult(new FlushResult());
+    }
+
+    public override Memory<byte> GetMemory(int sizeHint = 0)
+    {
+        if (_buffer == null || sizeHint < _buffer.Length - _consumedLength)
+            Grow(sizeHint);
+        return _buffer.AsMemory(_consumedLength);
+    }
+
+    public override Span<byte> GetSpan(int sizeHint = 0)
+    {
+        if (_buffer == null || sizeHint < _buffer.Length - _consumedLength)
+            Grow(sizeHint);
+        return _buffer.AsSpan(_consumedLength);
+    }
+
+    private void Grow(int sizeHint)
+    {
+        var newSize = Math.Max(sizeHint + _consumedLength, _buffer.Length * 2);
+        Array.Resize(ref _buffer, newSize);
+    }
+}
+
+internal class Http3FrameFixture
+{
+    public static byte[] GetHeadersFrame()
+    {
+        var encoder = new QPackDecoder();
+        var headers = new Http3ResponseHeaderCollection
+        {
+            { ":path", "/" },
+            { ":authority", "localhost" },
+            { ":method", "GET" },
+            { ":scheme", "https" }
+        };
+        var writer = new TestPipeWriter();
+        encoder.Encode(headers, writer);
+        var payloadLength = VariableLenghtIntegerDecoder.Write(writer.WrittenData.Length);
+        return [1, .. payloadLength.Span, .. writer.WrittenData];
+    }
+
+    public static byte[] GetLargeHeadersFrame()
+    {
+        var encoder = new QPackDecoder();
+        var headers = new Http3ResponseHeaderCollection
+        {
+            { ":path", "/" },
+            { ":authority", "localhost" },
+            { ":method", "GET" },
+            { ":scheme", "https" },
+            { "x-custom-header", new string('a', 4096*2) }
+        };
+        var writer = new TestPipeWriter();
+        encoder.Encode(headers, writer);
+        var payloadLength = VariableLenghtIntegerDecoder.Write(writer.WrittenData.Length);
+        return [1, .. payloadLength.Span, .. writer.WrittenData];
+    }
+
+    public static byte[] GetData(int length)
+    {
+        var payloadLength = VariableLenghtIntegerDecoder.Write(length);
+        var payload = Enumerable.Sequence(0, length - 1, 1).Select(x => (byte)x);
+        return [0, .. payloadLength.Span, .. payload];
+    }
+
+    public static byte[] GetReservedFrame(int length, int seed = 2)
+    {
+        var frameType = VariableLenghtIntegerDecoder.Write(seed * 0x1f + 0x21);
+        var payloadLength = VariableLenghtIntegerDecoder.Write(length);
+        var payload = Enumerable.Sequence(0, length - 1, 1).Select(x => (byte)x);
+        return [.. frameType.Span, .. payloadLength.Span, .. payload];
+    }
+}