http2: a streamed flush that waits for the write pump allocates nothing, and a connection reuses its request objects

src/protocols/ioxide.http2/Hpack.cs

@@ -20,10 +20,10 @@ internal sealed class HpackDecoder
     internal sealed class HpackException(string message) : Exception(message);
 
     /// <summary>
-    /// Decode one header block, invoking <paramref name="onHeader"/> per field. Names and values
-    /// point into <paramref name="scratch"/> or into the tables, and are valid until the next call.
+    /// Decode one header block, invoking <paramref name="onHeader"/> with <paramref name="state"/> per field.
+    /// Names and values point into <paramref name="scratch"/> or into the tables, and are valid until the next call.
     /// </summary>
-    internal void Decode(ReadOnlySpan<byte> block, Span<byte> scratch, HeaderSink onHeader)
+    internal void Decode<TState>(ReadOnlySpan<byte> block, Span<byte> scratch, TState state, HeaderSink<TState> onHeader)
     {
         int position = 0;
         int scratchUsed = 0;
@@ -37,7 +37,7 @@ internal void Decode(ReadOnlySpan<byte> block, Span<byte> scratch, HeaderSink on
                 // 1xxxxxxx - fully indexed: both name and value come from a table.
                 int index = (int)ReadInteger(block, ref position, 7);
                 Lookup(index, out ReadOnlySpan<byte> name, out ReadOnlySpan<byte> value);
-                onHeader(name, value);
+                onHeader(state, name, value);
                 continue;
             }
 
@@ -72,11 +72,11 @@ internal void Decode(ReadOnlySpan<byte> block, Span<byte> scratch, HeaderSink on
                 _table.Add(headerName, headerValue);
             }
 
-            onHeader(headerName, headerValue);
+            onHeader(state, headerName, headerValue);
         }
     }
 
-    internal delegate void HeaderSink(ReadOnlySpan<byte> name, ReadOnlySpan<byte> value);
+    internal delegate void HeaderSink<in TState>(TState state, ReadOnlySpan<byte> name, ReadOnlySpan<byte> value);
 
     // out parameters rather than a tuple: a tuple cannot hold ref structs, and these have to stay
     // spans so a table hit costs no copy.

src/protocols/ioxide.http2/Http2Connection.Frames.cs

@@ -165,15 +165,7 @@ private void HandleHeaders(in FrameHeader header, ReadOnlySpan<byte> payload)
                 return;
             }
 
-            // Opens at what the peer's SETTINGS advertised, not at the RFC default. Streams are
-            // created long after those SETTINGS arrive, so starting at 65535 and waiting for a
-            // WINDOW_UPDATE means waiting for one the peer has no reason to send - it believes we
-            // still have its whole window. That stalls any response longer than 65535 bytes.
-            pending = new PendingRequest
-            {
-                StreamId = header.StreamId,
-                SendWindow = _peerInitialStreamWindow,
-            };
+            pending = RentPending(header.StreamId);
             _streams[header.StreamId] = pending;
             _highestStream = header.StreamId;
         }
@@ -264,7 +256,7 @@ private void BeginStreamedBody(PendingRequest pending, bool ended)
 
         pending.BodyReader = new Http2BodyReader(this, pending.StreamId, ended);
         _responseWindows[pending.StreamId] = pending.SendWindow;
-        _ready.Add(pending);
+        _ready.Enqueue(pending);
     }
 
     private void HandleContinuation(in FrameHeader header, ReadOnlySpan<byte> payload)
@@ -315,9 +307,9 @@ private bool DecodeHeaderBlock(PendingRequest pending, bool discard = false)
                 _headerScratch = new byte[needed];
             }
 
-            PendingRequest target = pending;
-            _decoder.Decode(block, _headerScratch,
-                discard ? static (_, _) => { } : (name, value) => target.AddHeader(name, value));
+            // The request rides as state, so the sink is a static lambda and no closure is made per block.
+            _decoder.Decode(block, _headerScratch, pending,
+                discard ? static (_, _, _) => { } : static (target, name, value) => target.AddHeader(name, value));
             pending.ClearHeaderBlock();
             pending.HeadersDone = true;
             return true;
@@ -416,7 +408,7 @@ private void TryComplete(PendingRequest pending)
         // Its window moves with it: a WINDOW_UPDATE finds a stream by _streams only until here.
         _streams.Remove(pending.StreamId);
         _responseWindows[pending.StreamId] = pending.SendWindow;
-        _ready.Add(pending);
+        _ready.Enqueue(pending);
     }
 
     private void HandleSettings(in FrameHeader header, ReadOnlySpan<byte> payload)
@@ -562,7 +554,7 @@ private void HandleRstStream(in FrameHeader header)
     {
         if (_streams.Remove(header.StreamId, out PendingRequest? pending))
         {
-            pending.Abort();   // the peer gave up; there is nobody to answer
+            Discard(pending);   // the peer gave up; there is nobody to answer
         }
 
         // A response in flight may send nothing more on the stream (RFC 9113 5.1), and a writer parked
@@ -575,7 +567,8 @@ private void HandleRstStream(in FrameHeader header)
 
     /// <summary>
     /// One request being assembled: the raw header block until it is complete, then decoded fields
-    /// and body bytes in a pooled arena described by ranges.
+    /// and body bytes in a pooled arena described by ranges. Reused for later streams once retired,
+    /// with its request, header list and field ranges.
     /// </summary>
     private sealed class PendingRequest : IDisposable
     {
@@ -585,6 +578,7 @@ private sealed class PendingRequest : IDisposable
         private int _blockUsed;
         private readonly List<(int NameOffset, int NameLength, int ValueOffset, int ValueLength)> _fields = [];
         private (int Offset, int Length) _body;
+        private readonly Http2Request _request = new();
 
         public int StreamId;
         public bool HeadersDone;
@@ -598,6 +592,9 @@ private sealed class PendingRequest : IDisposable
 
         public int BodyLength => _body.Length;
 
+        // Grew nothing a pooled request should keep: a field list past this was a peer's thousands of header lines.
+        public bool IsSmall => _fields.Capacity <= 64;
+
         public (int Offset, int Length) Method;
         public (int Offset, int Length) Path;
         public (int Offset, int Length) Scheme;
@@ -687,16 +684,14 @@ public void AppendBody(ReadOnlySpan<byte> data)
 
         public Http2Request Freeze()
         {
-            var request = new Http2Request
-            {
-                StreamId   = StreamId,
-                Method     = Slice(Method),
-                Path       = Slice(Path),
-                Scheme     = Slice(Scheme),
-                Authority  = Slice(Authority),
-                Body       = Slice(_body),
-                BodyReader = BodyReader,
-            };
+            Http2Request request = _request;
+            request.StreamId   = StreamId;
+            request.Method     = Slice(Method);
+            request.Path       = Slice(Path);
+            request.Scheme     = Slice(Scheme);
+            request.Authority  = Slice(Authority);
+            request.Body       = Slice(_body);
+            request.BodyReader = BodyReader;
 
             foreach ((int nameOffset, int nameLength, int valueOffset, int valueLength) in _fields)
             {
@@ -741,6 +736,17 @@ public void Dispose()
                 _arena = [];
             }
             _used = 0;
+
+            // Its request still points into the arena that just went back, which the pool must not pin.
+            _request.Headers.Clear();
+            _request.Method = _request.Path = _request.Scheme = _request.Authority = _request.Body = default;
+            _request.BodyReader = null;
+
+            StreamId = 0;
+            HeadersDone = false;
+            RequestEnded = false;
+            SendWindow = 65535;
+            Method = Path = Scheme = Authority = _body = default;
         }
     }
 }

src/protocols/ioxide.http2/Http2Connection.Streamed.cs

@@ -1,4 +1,5 @@
 using System.Buffers;
+using System.IO.Pipelines;
 
 namespace ioxide.http2;
 
@@ -54,7 +55,7 @@ public Task RunAsync(Func<Http2Request, Http2BodyReader, Http2ResponseWriter, Va
     /// </summary>
     private void DispatchStreamed(Func<Http2Request, Http2ResponseWriter, ValueTask> handler)
     {
-        foreach (PendingRequest pending in TakeReady())
+        while (_ready.TryDequeue(out PendingRequest? pending))
         {
             try
             {
@@ -99,8 +100,7 @@ private async Task ServeStreamedAsync(Func<Http2Request, Http2ResponseWriter, Va
             // the connection grows without bound - 20 GB over eight seconds of load, measured.
             _creditWaiters.Remove(writer.StreamId);
             _responseWindows.Remove(writer.StreamId);
-            _streams.Remove(writer.StreamId);
-            pending.Dispose();
+            RetireStream(pending);
             _writerPool.Push(writer);
         }
     }
@@ -179,17 +179,17 @@ internal void SendStreamedData(int streamId, ReadOnlySpan<byte> body, bool endSt
     /// pump that carries them (Http2Connection.Write.cs) - so completions sharing a reactor turn
     /// still share a write.
     /// </summary>
-    internal ValueTask MaybeFlushAsync()
-        => _passFlushPending ? ValueTask.CompletedTask : FlushAsync();
+    internal ValueTask<FlushResult> MaybeFlushAsync()
+        => _passFlushPending ? default : FlushAsync();
 
-    /// <summary>A streamed writer's flush: out to the transport, or the write queue's next turn.</summary>
-    internal ValueTask FlushOutboundAsync() => FlushAsync();
+    /// <summary>A streamed writer's flush: out to the transport, or the write queue's next turn, waited for on the writer's own waiter.</summary>
+    internal ValueTask<FlushResult> FlushOutboundAsync(TurnWaiter turn) => FlushAsync(turn);
 
     /// <summary>A streamed response its handler could not finish: RST_STREAM, then out it goes.</summary>
-    internal ValueTask ResetStreamedAsync(int streamId)
+    internal ValueTask<FlushResult> ResetStreamedAsync(int streamId, TurnWaiter turn)
     {
         ResetStream(streamId, Http2Error.InternalError);
-        return InDispatchPass ? default : FlushOutboundAsync();
+        return InDispatchPass ? default : FlushOutboundAsync(turn);
     }
 
     /// <summary>