From aeefc7d0a77531f6417bcac15ea508863b393297 Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Wed, 4 Nov 2026 15:32:04 +0000 Subject: [PATCH 1/4] http2: a streamed flush that waits for the write pump allocates nothing, and a connection reuses its request objects A writer whose flush only waits for the pump's next turn now returns that wait on a waiter it keeps (TurnWaiter, an IValueTaskSource), from a flush that is no longer an async method on the reactor: no state-machine boxes, no TaskCompletionSource, and the turn is a chain linked through its waiters instead of a list copied into an array. The turn whose own flush faulted is woken too. A retired PendingRequest goes back to its connection with its Http2Request, header list and field ranges, and the next stream reuses them. The HPACK sink takes its request as state instead of a closure, and a dispatch swaps two ready lists instead of copying one. --- src/protocols/ioxide.http2/Hpack.cs | 12 +- .../ioxide.http2/Http2Connection.Frames.cs | 46 +++--- .../ioxide.http2/Http2Connection.Streamed.cs | 44 +++--- .../ioxide.http2/Http2Connection.Write.cs | 116 ++++++++------ src/protocols/ioxide.http2/Http2Connection.cs | 108 +++++++------ src/protocols/ioxide.http2/Http2Request.cs | 3 +- .../ioxide.http2/Http2ResponseWriter.cs | 84 ++++++---- src/protocols/ioxide.http2/TurnWaiter.cs | 42 +++++ .../Http2OutputQueueTests.cs | 79 ++++++++++ .../Http2RequestReuseTests.cs | 147 ++++++++++++++++++ tests/Ioxide.Tests.Unit/Program.cs | 1 + 11 files changed, 514 insertions(+), 168 deletions(-) create mode 100644 src/protocols/ioxide.http2/TurnWaiter.cs create mode 100644 tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs diff --git a/src/protocols/ioxide.http2/Hpack.cs b/src/protocols/ioxide.http2/Hpack.cs index 431af2c4..9012297b 100644 --- a/src/protocols/ioxide.http2/Hpack.cs +++ b/src/protocols/ioxide.http2/Hpack.cs @@ -20,10 +20,10 @@ internal sealed class HpackDecoder internal sealed class HpackException(string message) : Exception(message); /// - /// Decode one header block, invoking per field. Names and values - /// point into or into the tables, and are valid until the next call. + /// Decode one header block, invoking with per field. + /// Names and values point into or into the tables, and are valid until the next call. /// - internal void Decode(ReadOnlySpan block, Span scratch, HeaderSink onHeader) + internal void Decode(ReadOnlySpan block, Span scratch, TState state, HeaderSink onHeader) { int position = 0; int scratchUsed = 0; @@ -37,7 +37,7 @@ internal void Decode(ReadOnlySpan block, Span 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 name, out ReadOnlySpan value); - onHeader(name, value); + onHeader(state, name, value); continue; } @@ -72,11 +72,11 @@ internal void Decode(ReadOnlySpan block, Span scratch, HeaderSink on _table.Add(headerName, headerValue); } - onHeader(headerName, headerValue); + onHeader(state, headerName, headerValue); } } - internal delegate void HeaderSink(ReadOnlySpan name, ReadOnlySpan value); + internal delegate void HeaderSink(TState state, ReadOnlySpan name, ReadOnlySpan 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. diff --git a/src/protocols/ioxide.http2/Http2Connection.Frames.cs b/src/protocols/ioxide.http2/Http2Connection.Frames.cs index 222c4586..ea63fa5d 100644 --- a/src/protocols/ioxide.http2/Http2Connection.Frames.cs +++ b/src/protocols/ioxide.http2/Http2Connection.Frames.cs @@ -165,15 +165,7 @@ private void HandleHeaders(in FrameHeader header, ReadOnlySpan 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; } @@ -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; @@ -575,7 +567,8 @@ private void HandleRstStream(in FrameHeader header) /// /// 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. /// 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; @@ -687,17 +681,16 @@ public void AppendBody(ReadOnlySpan 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; + request.Headers.Clear(); // keeps the array a previous stream grew foreach ((int nameOffset, int nameLength, int valueOffset, int valueLength) in _fields) { request.Headers.Add(_arena.AsMemory(nameOffset, nameLength), @@ -741,6 +734,13 @@ public void Dispose() _arena = []; } _used = 0; + + // Fresh for the next stream that rents it. + StreamId = 0; + HeadersDone = false; + RequestEnded = false; + SendWindow = 65535; + Method = Path = Scheme = Authority = _body = default; } } } diff --git a/src/protocols/ioxide.http2/Http2Connection.Streamed.cs b/src/protocols/ioxide.http2/Http2Connection.Streamed.cs index f77ee723..df098d5a 100644 --- a/src/protocols/ioxide.http2/Http2Connection.Streamed.cs +++ b/src/protocols/ioxide.http2/Http2Connection.Streamed.cs @@ -1,4 +1,5 @@ using System.Buffers; +using System.IO.Pipelines; namespace ioxide.http2; @@ -54,20 +55,28 @@ public Task RunAsync(Func private void DispatchStreamed(Func handler) { - foreach (PendingRequest pending in TakeReady()) + List ready = TakeReady(); + try { - try - { - Http2Request request = pending.Freeze(); - Http2ResponseWriter writer = RentWriter(request.StreamId); - _ = ServeStreamedAsync(handler, request, writer, pending); // the writer owns this stream, and retires it when done - } - catch + foreach (PendingRequest pending in ready) { - pending.Dispose(); - throw; + try + { + Http2Request request = pending.Freeze(); + Http2ResponseWriter writer = RentWriter(request.StreamId); + _ = ServeStreamedAsync(handler, request, writer, pending); // the writer owns this stream, and retires it when done + } + catch + { + pending.Dispose(); + throw; + } } } + finally + { + ready.Clear(); + } } private async Task ServeStreamedAsync(Func handler, @@ -99,8 +108,7 @@ private async Task ServeStreamedAsync(Func body, bool endSt /// pump that carries them (Http2Connection.Write.cs) - so completions sharing a reactor turn /// still share a write. /// - internal ValueTask MaybeFlushAsync() - => _passFlushPending ? ValueTask.CompletedTask : FlushAsync(); + internal ValueTask MaybeFlushAsync() + => _passFlushPending ? default : FlushAsync(); - /// A streamed writer's flush: out to the transport, or the write queue's next turn. - internal ValueTask FlushOutboundAsync() => FlushAsync(); + /// A streamed writer's flush: out to the transport, or the write queue's next turn, waited for on the writer's own waiter. + internal ValueTask FlushOutboundAsync(TurnWaiter turn) => FlushAsync(turn); /// A streamed response its handler could not finish: RST_STREAM, then out it goes. - internal ValueTask ResetStreamedAsync(int streamId) + internal ValueTask ResetStreamedAsync(int streamId, TurnWaiter turn) { ResetStream(streamId, Http2Error.InternalError); - return InDispatchPass ? default : FlushOutboundAsync(); + return InDispatchPass ? default : FlushOutboundAsync(turn); } /// diff --git a/src/protocols/ioxide.http2/Http2Connection.Write.cs b/src/protocols/ioxide.http2/Http2Connection.Write.cs index 01a5fecd..980a9ce8 100644 --- a/src/protocols/ioxide.http2/Http2Connection.Write.cs +++ b/src/protocols/ioxide.http2/Http2Connection.Write.cs @@ -28,10 +28,12 @@ public sealed partial class Http2Connection private byte[] _queued = []; private int _queuedUsed; - // Completed when the flush AFTER the current one finishes - the one that carries the queue. - // Callers who staged into the queue await this, which keeps the two things a real flush - // provided: backpressure, and the yield that hands the reactor back. - private readonly List _turnWaiters = []; + // Callers parked on the pump's next turn, in the order they parked: woken when the flush AFTER the + // current one finishes - the one that carries the queue. Callers who staged into the queue wait for + // it, which keeps the two things a real flush provided: backpressure, and the yield that hands the + // reactor back. Linked through the waiters themselves, so parking allocates nothing. + private TurnWaiter? _turnHead; + private TurnWaiter? _turnTail; private void Stage(ReadOnlySpan bytes) { @@ -51,56 +53,53 @@ private void Stage(ReadOnlySpan bytes) } } - private async ValueTask FlushAsync() + // Not async: a caller that only waits for the pump's next turn gets that wait as it is. A streamed + // writer passes the waiter it keeps, so its flush allocates nothing; anyone else parks a new one. + private ValueTask FlushAsync(TurnWaiter? waiter = null) { if (_disposed || _writeDead) { - return; + return default; } if (_flushing) { // The pump owns the pipe. If this caller queued bytes they leave on the pump's next - // turn - await it, so the caller keeps real backpressure. With nothing queued there + // turn - wait for it, so the caller keeps real backpressure. With nothing queued there // is nothing to wait for. - if (_queuedUsed > 0) - { - await TurnAsync(); - } - return; + return _queuedUsed > 0 ? AwaitTurn(waiter ?? new TurnWaiter()) : default; } - if (!_staged) - { - return; - } + return _staged ? PumpAsync() : default; + } - // This caller becomes the pump: it drives turn after turn until a flush completes with - // the queue empty. Everyone else who staged meanwhile is parked on TurnAsync. + // This caller becomes the pump: it drives turn after turn until a flush completes with the + // queue empty. Everyone else who staged meanwhile waits for their turn. + private async ValueTask PumpAsync() + { _flushing = true; + TurnWaiter? turn = null; try { while (true) { _staged = false; - // Waiters captured BEFORE the flush are the ones whose bytes are in it; anyone - // arriving during the await parks a fresh waiter for the turn after. - TaskCompletionSource[] turn = TakeTurn(); + // Waiters taken BEFORE the flush are the ones whose bytes are in it; anyone + // arriving during the await parks for the turn after. + turn = TakeTurn(); await _pipe.Output.FlushAsync(); await new ReactorSwitch(_reactor); // a pipe may complete its flush elsewhere: a default Pipe on the thread pool // May resume writers inline; _flushing is still true, so anything they stage // lands in the queue and is picked up by the check just below. - foreach (TaskCompletionSource waiter in turn) - { - waiter.TrySetResult(); - } + Wake(turn); + turn = null; if (_queuedUsed == 0 || _disposed) { - return; + return default; } _pipe.Output.Write(_queued.AsSpan(0, _queuedUsed)); @@ -119,29 +118,27 @@ private async ValueTask FlushAsync() { if (_writeDead && OffReactor) { - _ = EndTurnOnReactorAsync(); // a faulted flush can end the pump on the thread that completed it + _ = EndTurnOnReactorAsync(turn); // a faulted flush can end the pump on the thread that completed it } else { - EndTurn(); + EndTurn(turn); } } } // Never strand a waiter: on a fault the turn it is waiting for will not come, and it has to wake to see - // the connection is broken rather than hang forever. Out of line: a loop in a hot async method's finally - // costs it its optimized code. + // the connection is broken rather than hang forever - the turn whose own flush faulted included. Out of + // line: a loop in a hot async method's finally costs it its optimized code. [MethodImpl(MethodImplOptions.NoInlining)] - private void EndTurn() + private void EndTurn(TurnWaiter? faulted) { _flushing = false; - foreach (TaskCompletionSource waiter in TakeTurn()) - { - waiter.TrySetResult(); - } + Wake(faulted); + Wake(TakeTurn()); } - private async Task EndTurnOnReactorAsync() + private async Task EndTurnOnReactorAsync(TurnWaiter? faulted) { try { @@ -151,29 +148,48 @@ private async Task EndTurnOnReactorAsync() { // The reactor has stopped, so nothing races the wake: it goes on here. } - EndTurn(); + EndTurn(faulted); } - // One completion per waiter: a task awaited by several resumes only the first inline and sends - // the rest to the thread pool. - private Task TurnAsync() + private ValueTask AwaitTurn(TurnWaiter waiter) { - var waiter = new TaskCompletionSource(); - _turnWaiters.Add(waiter); - return waiter.Task; + ValueTask turn = waiter.Arm(); + if (!waiter.Enlisted) + { + waiter.Enlisted = true; + if (_turnTail is null) + { + _turnHead = waiter; + } + else + { + _turnTail.Next = waiter; + } + _turnTail = waiter; + } + return turn; } // Taken out before any is woken: one resumed inline that parks again belongs to the next turn. - private TaskCompletionSource[] TakeTurn() + private TurnWaiter? TakeTurn() { - if (_turnWaiters.Count == 0) + TurnWaiter? turn = _turnHead; + _turnHead = null; + _turnTail = null; + return turn; + } + + // Each link is read before its waiter wakes: one resumed inline may park again, relinking itself. + private static void Wake(TurnWaiter? turn) + { + while (turn is not null) { - return []; + TurnWaiter? next = turn.Next; + turn.Next = null; + turn.Enlisted = false; + turn.Release(); + turn = next; } - - TaskCompletionSource[] turn = [.. _turnWaiters]; - _turnWaiters.Clear(); - return turn; } private void Enqueue(ReadOnlySpan bytes) diff --git a/src/protocols/ioxide.http2/Http2Connection.cs b/src/protocols/ioxide.http2/Http2Connection.cs index ae46bc72..303f48bf 100644 --- a/src/protocols/ioxide.http2/Http2Connection.cs +++ b/src/protocols/ioxide.http2/Http2Connection.cs @@ -42,7 +42,14 @@ public sealed partial class Http2Connection : IDisposable private byte[] _headerScratch = new byte[16 * 1024]; private readonly Dictionary _streams = new(); - private readonly List _ready = []; + + // Requests ready for their handler, and the list a dispatch is walking: swapped each pass, so the + // parser never adds to the list being walked and no snapshot is copied. + private List _ready = []; + private List _dispatching = []; + + // Retired requests, each keeping its Http2Request, header list and field ranges for the next stream. + private readonly Stack _pendingPool = new(); // Readers whose parked ReadAsync has something to hand over. Collected during parsing and // fired once it has unwound, so a resumed handler cannot re-enter the parser mid-frame. @@ -134,10 +141,7 @@ public void Dispose() // A writer parked on its turn of the write pump has to wake into IsBroken, not hang: the // flush that would have completed its turn is never coming. - foreach (TaskCompletionSource waiter in TakeTurn()) - { - waiter.TrySetResult(); - } + Wake(TakeTurn()); // Last: a handler parked on its body resumes into its end once everything above is released. FireBodyWakes(); @@ -288,58 +292,61 @@ private void Append(ReadOnlySpan data) _inboundUsed += data.Length; } - // Snapshot and clear first: the list must not be mutated while it is walked. - private PendingRequest[] TakeReady() + // Taken whole before the walk: the list must not be mutated while it is walked. The caller clears + // it once done. + private List TakeReady() { - if (_ready.Count == 0) - { - return []; - } - - PendingRequest[] ready = _ready.ToArray(); - _ready.Clear(); - return ready; + (_ready, _dispatching) = (_dispatching, _ready); + return _dispatching; } private void DispatchBuffered(Func> handler) { - foreach (PendingRequest pending in TakeReady()) + List ready = TakeReady(); + try { - ValueTask inFlight; - - try + foreach (PendingRequest pending in ready) { - Http2Request request = pending.Freeze(); - inFlight = handler(request); - } - catch (Exception exception) - { - // A synchronous handler throws before there is a task. It gets what a faulted task - // gets, a 500 on its own stream, rather than taking the connection down. - inFlight = ValueTask.FromException(exception); - } + ValueTask inFlight; - // Answered synchronously, which nearly every handler does. Stay inline: this response - // is staged in time for the pass flush, so it still leaves with every other one, and - // there is no Task to allocate. - if (inFlight.IsCompletedSuccessfully) - { try { - WriteResponse(pending.StreamId, inFlight.Result); + Http2Request request = pending.Freeze(); + inFlight = handler(request); } - finally + catch (Exception exception) { - RetireStream(pending); + // A synchronous handler throws before there is a task. It gets what a faulted task + // gets, a 500 on its own stream, rather than taking the connection down. + inFlight = ValueTask.FromException(exception); + } + + // Answered synchronously, which nearly every handler does. Stay inline: this response + // is staged in time for the pass flush, so it still leaves with every other one, and + // there is no Task to allocate. + if (inFlight.IsCompletedSuccessfully) + { + try + { + WriteResponse(pending.StreamId, inFlight.Result); + } + finally + { + RetireStream(pending); + } + continue; } - continue; - } - // It parked - a database, an upstream, a disk. Awaiting here would hold every OTHER - // stream on this connection behind it, including responses already staged and ready to - // go, because they all share this one dispatch loop and one TCP connection. So it - // finishes on its own and writes its own bytes when it does. - _ = CompleteBufferedAsync(inFlight, pending); + // It parked - a database, an upstream, a disk. Awaiting here would hold every OTHER + // stream on this connection behind it, including responses already staged and ready to + // go, because they all share this one dispatch loop and one TCP connection. So it + // finishes on its own and writes its own bytes when it does. + _ = CompleteBufferedAsync(inFlight, pending); + } + } + finally + { + ready.Clear(); } } @@ -379,12 +386,25 @@ private async Task CompleteBufferedAsync(ValueTask inFlight, Pend /// /// Done with a stream. The buffered path already took it out of _streams when it became /// ready; a streamed request is still in there, because DATA frames were arriving the whole - /// time the handler ran. + /// time the handler ran. Once per dispatched request: it goes back for the next stream. /// private void RetireStream(PendingRequest pending) { _streams.Remove(pending.StreamId); pending.Dispose(); + _pendingPool.Push(pending); + } + + // 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. + private PendingRequest RentPending(int streamId) + { + PendingRequest pending = _pendingPool.TryPop(out PendingRequest? pooled) ? pooled : new PendingRequest(); + pending.StreamId = streamId; + pending.SendWindow = _peerInitialStreamWindow; + return pending; } /// Return a consumed chunk's credit to the peer, on both windows it was charged to. diff --git a/src/protocols/ioxide.http2/Http2Request.cs b/src/protocols/ioxide.http2/Http2Request.cs index 54ef3b42..4b8e0a12 100644 --- a/src/protocols/ioxide.http2/Http2Request.cs +++ b/src/protocols/ioxide.http2/Http2Request.cs @@ -6,7 +6,8 @@ namespace ioxide.http2; /// most handlers only ever compare a path or read a header. /// /// The memories point into an arena owned by the connection and are valid for the duration of the -/// handler call. Copy anything you keep past it. +/// handler call. Copy anything you keep past it: the request itself is reused for a later stream on +/// the same connection once its handler is done. /// public sealed class Http2Request { diff --git a/src/protocols/ioxide.http2/Http2ResponseWriter.cs b/src/protocols/ioxide.http2/Http2ResponseWriter.cs index f9d6d37f..73eacd47 100644 --- a/src/protocols/ioxide.http2/Http2ResponseWriter.cs +++ b/src/protocols/ioxide.http2/Http2ResponseWriter.cs @@ -43,6 +43,9 @@ public sealed class Http2ResponseWriter : IBufferWriter // The names and values of _unsent: the caller's memories are its own again once WriteHeaders returns. private byte[] _unsentFields = []; + // What a flush waits on while its bytes are queued behind the write pump's flush in flight. + private readonly TurnWaiter _turn = new(); + internal Http2ResponseWriter(Http2Connection connection, int streamId) { _connection = connection; @@ -211,13 +214,28 @@ internal async ValueTask FailAsync() // Not in answer to the peer's own reset (RFC 9113 5.4.2), and not on a connection that is gone. if (_connection.IsResponseLive(_streamId)) { - await _connection.ResetStreamedAsync(_streamId); + await _connection.ResetStreamedAsync(_streamId, _turn); } } private static readonly FlushResult PeerGone = new(isCanceled: false, isCompleted: true); - private async ValueTask FlushCore(bool endStream) + // Not async on the reactor: a flush that sends and then only waits for the write pump's turn returns that + // wait as it is, on this writer's own waiter, so it allocates nothing. The rest goes the async way. + private ValueTask FlushCore(bool endStream) + { + var onReactor = new ReactorSwitch(_connection.Reactor); + if (!_headersSent || !onReactor.IsCompleted) + { + return FlushCoreAsync(endStream); + } + + onReactor.GetResult(); // throws once the reactor has stopped, as awaiting it does + SendUnsentHeaders(); + return Send(endStream, sent: 0); + } + + private async ValueTask FlushCoreAsync(bool endStream) { if (!_headersSent) { @@ -226,13 +244,23 @@ private async ValueTask FlushCore(bool endStream) await new ReactorSwitch(_connection.Reactor); + SendUnsentHeaders(); + return await Send(endStream, sent: 0); + } + + private void SendUnsentHeaders() + { if (_unsent is Http2Response unsent) { _unsent = null; SendHeaders(unsent); } + } - int sent = 0; + // Everything staged from `sent` on, as far as credit allows, then the flush. Out of credit, the rest goes on in + // SendAfterCreditAsync. + private ValueTask Send(bool endStream, int sent) + { while (sent < _staged) { // Bounded by the frame size AND by both windows: exceeding either is a connection @@ -241,29 +269,11 @@ private async ValueTask FlushCore(bool endStream) if (credit < 0) { _staged = 0; // reset by the peer, or the connection is gone: nobody to send it to - return PeerGone; + return new ValueTask(PeerGone); } if (credit == 0) { - // Take the wait BEFORE flushing. The flush below is an await, and the WINDOW_UPDATE - // it exists to provoke can arrive while this writer is still inside it - at which - // point a release has no waiter to find, is dropped, and the writer then parks on a - // message that has already been and gone. Registering first means that credit - // completes this task instead, and the await below returns at once. - Task credited = _connection.WaitForSendCreditAsync(_streamId); - - // Everything staged so far has to REACH the peer before waiting on it to open the - // window. WINDOW_UPDATE is what a peer sends once it has consumed DATA, so parking - // with those bytes still in the connection's buffer waits for a message that the - // wait itself prevents - and any body larger than the initial 65535-byte window - // deadlocks until the client times out. The flush is unconditional because inside a - // pass the pass flush would otherwise carry them, and this stream is about to stop - // making progress rather than return to it. - _sinceRealFlush = 0; - await _connection.FlushOutboundAsync(); - - await credited; - continue; + return SendAfterCreditAsync(endStream, sent); } int chunk = Math.Min(credit, _staged - sent); @@ -279,7 +289,7 @@ private async ValueTask FlushCore(bool endStream) // Sending proved the stream live; with nothing sent it still has to be asked. if (sent == 0 && !_connection.IsResponseLive(_streamId)) { - return PeerGone; + return new ValueTask(PeerGone); } if (endStream && sent == 0) @@ -297,8 +307,30 @@ private async ValueTask FlushCore(bool endStream) } _sinceRealFlush = 0; - await _connection.FlushOutboundAsync(); - return default; + return _connection.FlushOutboundAsync(_turn); + } + + private async ValueTask SendAfterCreditAsync(bool endStream, int sent) + { + // Take the wait BEFORE flushing. The flush below is an await, and the WINDOW_UPDATE + // it exists to provoke can arrive while this writer is still inside it - at which + // point a release has no waiter to find, is dropped, and the writer then parks on a + // message that has already been and gone. Registering first means that credit + // completes this task instead, and the await below returns at once. + Task credited = _connection.WaitForSendCreditAsync(_streamId); + + // Everything staged so far has to REACH the peer before waiting on it to open the + // window. WINDOW_UPDATE is what a peer sends once it has consumed DATA, so parking + // with those bytes still in the connection's buffer waits for a message that the + // wait itself prevents - and any body larger than the initial 65535-byte window + // deadlocks until the client times out. The flush is unconditional because inside a + // pass the pass flush would otherwise carry them, and this stream is about to stop + // making progress rather than return to it. + _sinceRealFlush = 0; + await _connection.FlushOutboundAsync(_turn); + + await credited; + return await Send(endStream, sent); } private void EnsureStaging(int sizeHint) diff --git a/src/protocols/ioxide.http2/TurnWaiter.cs b/src/protocols/ioxide.http2/TurnWaiter.cs new file mode 100644 index 00000000..56aef598 --- /dev/null +++ b/src/protocols/ioxide.http2/TurnWaiter.cs @@ -0,0 +1,42 @@ +using System.IO.Pipelines; +using System.Threading.Tasks.Sources; + +namespace ioxide.http2; + +/// +/// A flush waiting for the write pump's next turn (Http2Connection.Write.cs). Reusable: a streamed writer +/// keeps one for its whole life, so a flush that only waits for the turn allocates nothing. +/// +internal sealed class TurnWaiter : IValueTaskSource +{ + private ManualResetValueTaskSourceCore _core = new() + { + RunContinuationsAsynchronously = false, + }; + + /// The waiter parked after this one, for the same turn. + internal TurnWaiter? Next; + + /// + /// Parked and not woken yet. A flush whose wait was never awaited leaves its writer in the turn, and the + /// writer's next flush then waits for that same wake instead of joining a second time. + /// + internal bool Enlisted; + + internal ValueTask Arm() + { + _core.Reset(); + return new ValueTask(this, _core.Version); + } + + internal void Release() => _core.SetResult(default); + + FlushResult IValueTaskSource.GetResult(short token) => _core.GetResult(token); + + ValueTaskSourceStatus IValueTaskSource.GetStatus(short token) => _core.GetStatus(token); + + // Woken by the pump on the reactor: resume there rather than post. + void IValueTaskSource.OnCompleted(Action continuation, object? state, short token, + ValueTaskSourceOnCompletedFlags flags) + => _core.OnCompleted(continuation, state, token, flags & ~ValueTaskSourceOnCompletedFlags.UseSchedulingContext); +} diff --git a/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs b/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs index fe1705a7..c2b3fa51 100644 --- a/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs +++ b/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs @@ -1,5 +1,6 @@ +using System.IO.Pipelines; using ioxide.http2; namespace Ioxide.Tests; @@ -128,8 +129,86 @@ public static void Register(Runner runner) Assert.True(run.Wait(5_000), "connection wound down after the transport fault"); }); + + runner.Test("h2 queue: a faulted flush wakes the writer whose bytes it carried", () => + { + using var client = new StrictClient(); + + var gate = new TaskCompletionSource(); + var woke = new TaskCompletionSource(); + Task run = client.Connection.RunAsync(async (_, writer) => + { + writer.WriteHeaders(new Http2Response { Status = 200 }); + await gate.Task; // resumed outside the pass, its flush in flight + Push(writer, "lost"); + try + { + await writer.FlushAsync(); // queued behind it: its turn is the flush after + } + catch + { + // The transport is gone; waking at all is what this test is about. + } + woke.SetResult(); + }); + + client.ReleaseFlush(); // server SETTINGS + client.SendRequests(1); // the pass flush is in flight + gate.SetResult(); // the chunk queues, and the writer waits for its turn + client.ReleaseFlush(); // the pump sends the chunk: the writer's turn + client.FaultFlush(); // and that flush faults + + Assert.True(woke.Task.Wait(5_000), "the writer whose chunk the faulted flush carried never woke"); + Assert.True(run.Wait(5_000), "connection wound down after the transport fault"); + }); + + runner.Test("h2 queue: a streamed flush that waits for the pump's turn allocates nothing", () => + { + using var client = new StrictClient(); + + TaskCompletionSource[] gates = [new(), new(), new(), new()]; + long[] allocated = new long[gates.Length]; + Task run = client.Connection.RunAsync(async (_, writer) => + { + writer.WriteHeaders(new Http2Response { Status = 200 }); + for (int i = 0; i < gates.Length; i++) + { + await gates[i].Task; // resumed outside the pass, its flush in flight + Push(writer, "x"); + + long before = GC.GetAllocatedBytesForCurrentThread(); + ValueTask flush = writer.FlushAsync(); // queued behind it: waits for its turn + allocated[i] = GC.GetAllocatedBytesForCurrentThread() - before; + + await flush; + } + }); + + client.ReleaseFlush(); // server SETTINGS + client.SendRequests(1); // the pass flush is in flight + for (int i = 0; i < gates.Length; i++) + { + if (i > 0) + { + client.Feed(Ping()); // a pass whose flush, the PING ack, is in flight + } + gates[i].SetResult(); + client.ReleaseFlush(); // the flush in flight; the pump then sends the chunk + client.ReleaseFlush(); // the chunk's flush: the writer's turn + } + client.Drain(); // END_STREAM + client.Close(run); + + // The first wait grows the connection's queue; every wait after it reuses everything. + for (int i = 1; i < allocated.Length; i++) + { + Assert.True(allocated[i] == 0, $"flush {i} allocated {allocated[i]} B while it waited for its turn"); + } + }); } + private static byte[] Ping() => [0, 0, 8, 0x6, 0, 0, 0, 0, 0, 1, 2, 3, 4, 5, 6, 7, 8]; + private static void Push(Http2ResponseWriter writer, string text) { Span span = writer.GetSpan(text.Length); diff --git a/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs b/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs new file mode 100644 index 00000000..4fa71f88 --- /dev/null +++ b/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs @@ -0,0 +1,147 @@ +using System.Text; +using ioxide.http2; + +namespace Ioxide.Tests; + +/// +/// A connection keeps a retired request, with its header list and field ranges, for the next stream: the next +/// request must come back holding nothing of the one before, and a warm connection's request allocates none +/// of the parser's objects. +/// +internal static class Http2RequestReuseTests +{ + public static void Register(Runner runner) + { + runner.Test("h2 request: the next stream's request holds nothing of the one before", () => + { + using var client = new StrictClient(); + + var seen = new List<(Http2Request Request, string Fields)>(); + Task run = client.Connection.RunAsync(request => + { + seen.Add((request, Describe(request))); + return Http2Response.Text("ok"); + }); + + client.ReleaseFlush(); // server SETTINGS + client.SendRequests(); // preface and SETTINGS + client.ReleaseFlush(); // their ack + client.Feed(Request(1, post: true, "/first", "a.example", [("x-one", "1"), ("x-two", "2")], body: "hello")); + client.ReleaseFlush(); + client.Feed(Request(3, post: false, "/", authority: null, [], body: null)); + client.ReleaseFlush(); + client.Close(run); + + Assert.Equal(2, seen.Count); + Assert.Equal("1 POST /first a.example [x-one=1 x-two=2] hello", seen[0].Fields); + Assert.Equal("3 GET / - [] -", seen[1].Fields); + Assert.True(ReferenceEquals(seen[0].Request, seen[1].Request), "the second stream reuses the first one's request"); + }); + + runner.Test("h2 request: a warm connection's request allocates none of the parser's objects", () => + { + using var client = new StrictClient(); + + Http2Response response = Http2Response.Text("ok"); // shared, so the handler allocates nothing + Task run = client.Connection.RunAsync(_ => response); + + client.ReleaseFlush(); // server SETTINGS + client.SendRequests(); // preface and SETTINGS + client.ReleaseFlush(); // their ack + + const int perPass = 16; + client.Feed(Batch(1, perPass)); // warm-up: the requests the next batch reuses + client.ReleaseFlush(); + + byte[] batch = Batch(1 + (2 * perPass), perPass); + long before = GC.GetAllocatedBytesForCurrentThread(); + client.Feed(batch); // parsed, dispatched and answered inline + long perRequest = (GC.GetAllocatedBytesForCurrentThread() - before) / perPass; + + client.ReleaseFlush(); + client.Close(run); + + // Before reuse, every request allocated about 700 B of parser objects. What is left is the test + // transport's own flush, shared by the pass. + Assert.True(perRequest < 200, $"a request on a warm connection allocated {perRequest} B"); + }); + } + + private static string Describe(Http2Request request) + { + var fields = new List(); + foreach (KeyValuePair, ReadOnlyMemory> field in request.Headers.AsSpan()) + { + fields.Add($"{Ascii(field.Key)}={Ascii(field.Value)}"); + } + + return $"{request.StreamId} {Ascii(request.Method)} {Ascii(request.Path)} {Or(request.Authority)} " + + $"[{string.Join(' ', fields)}] {Or(request.Body)}"; + + static string Or(ReadOnlyMemory bytes) => bytes.IsEmpty ? "-" : Ascii(bytes); + } + + private static string Ascii(ReadOnlyMemory bytes) => Encoding.ASCII.GetString(bytes.Span); + + // Each a GET for / with one header of its own, in one HEADERS frame with END_STREAM. + private static byte[] Batch(int firstStream, int count) + { + var bytes = new List(); + for (int i = 0; i < count; i++) + { + bytes.AddRange(Request(firstStream + (2 * i), post: false, "/", authority: null, [("x-n", $"{i}")], body: null)); + } + return bytes.ToArray(); + } + + // HEADERS, with END_STREAM unless a DATA frame carrying the body follows. HPACK from the static table and + // literals without indexing only, so the request needs no encoder of its own. + private static byte[] Request(int streamId, bool post, string path, string? authority, + (string Name, string Value)[] fields, string? body) + { + var block = new List { post ? (byte)0x83 : (byte)0x82, 0x86 }; // :method, :scheme http + if (path == "/") + { + block.Add(0x84); // :path / + } + else + { + block.Add(0x04); // :path, literal value + AddString(block, path); + } + if (authority is not null) + { + block.Add(0x01); // :authority, literal value + AddString(block, authority); + } + foreach ((string name, string value) in fields) + { + block.Add(0x00); // literal name and value + AddString(block, name); + AddString(block, value); + } + + var frames = new List(); + frames.AddRange(FrameHeader(block.Count, 0x1, body is null ? (byte)0x5 : (byte)0x4, streamId)); + frames.AddRange(block); + if (body is not null) + { + frames.AddRange(FrameHeader(body.Length, 0x0, 0x1, streamId)); + frames.AddRange(Encoding.ASCII.GetBytes(body)); + } + return frames.ToArray(); + } + + private static void AddString(List block, string value) + { + block.Add((byte)value.Length); // no Huffman; every value here is under 127 bytes + block.AddRange(Encoding.ASCII.GetBytes(value)); + } + + private static byte[] FrameHeader(int length, byte type, byte flags, int streamId) => + [ + (byte)(length >> 16), (byte)(length >> 8), (byte)length, + type, flags, + (byte)(streamId >> 24), (byte)(streamId >> 16), (byte)(streamId >> 8), (byte)streamId, + ]; +} diff --git a/tests/Ioxide.Tests.Unit/Program.cs b/tests/Ioxide.Tests.Unit/Program.cs index fe2963c8..21ddba0b 100644 --- a/tests/Ioxide.Tests.Unit/Program.cs +++ b/tests/Ioxide.Tests.Unit/Program.cs @@ -29,6 +29,7 @@ private static int Main() Http2StreamedRequestTests.Register(runner); Http2TrailerTests.Register(runner); Http2RequestHeaderTests.Register(runner); + Http2RequestReuseTests.Register(runner); Http2StreamedBodyTests.Register(runner); Http2FlowControlTests.Register(runner); Http2BufferedCreditTests.Register(runner); From 7dea7facedf4165c43e4c0076889eec86c3a172e Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Wed, 4 Nov 2026 19:05:16 +0000 Subject: [PATCH 2/4] http2: a flush that waits for credit again and again loops in one frame; the request pool is capped and lets go of what it held From review: - SendAfterCreditAsync called back into Send, so every credit stall inside one flush nested another async frame, kept until the flush ended: 20,000 one-byte WINDOW_UPDATEs on the connection retained 4.1 MB, and the deep unwind moved the writer onto the thread pool. It loops again, as the original method did. - A turn that ended in a fault or a disposed connection now wakes its writers with IsCompleted, rather than with a flush that looked as if it had gone out. - The fast path hands exceptions back through the returned task, as the async method did. - The request pool keeps at most 32, never one whose field list grew past 64, and only on the reactor. A retired request lets go of the arena's memories. A stream reset before its request completed goes back to the pool too, instead of costing a new one. - Tests: the credit-stall case; the turn test asserts that its flush waited; the warm-request test warms both ready lists and counts its handlers; the reuse test changes the scheme as well. --- .../ioxide.http2/Http2Connection.Frames.cs | 12 ++- .../ioxide.http2/Http2Connection.Write.cs | 16 ++-- src/protocols/ioxide.http2/Http2Connection.cs | 33 ++++++- .../ioxide.http2/Http2ResponseWriter.cs | 94 +++++++++++++------ src/protocols/ioxide.http2/TurnWaiter.cs | 2 +- .../Http2OutputQueueTests.cs | 48 ++++++++++ .../Http2RequestReuseTests.cs | 28 ++++-- 7 files changed, 184 insertions(+), 49 deletions(-) diff --git a/src/protocols/ioxide.http2/Http2Connection.Frames.cs b/src/protocols/ioxide.http2/Http2Connection.Frames.cs index ea63fa5d..7b1d3d10 100644 --- a/src/protocols/ioxide.http2/Http2Connection.Frames.cs +++ b/src/protocols/ioxide.http2/Http2Connection.Frames.cs @@ -554,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 @@ -592,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; @@ -690,7 +693,6 @@ public Http2Request Freeze() request.Body = Slice(_body); request.BodyReader = BodyReader; - request.Headers.Clear(); // keeps the array a previous stream grew foreach ((int nameOffset, int nameLength, int valueOffset, int valueLength) in _fields) { request.Headers.Add(_arena.AsMemory(nameOffset, nameLength), @@ -735,6 +737,12 @@ public void Dispose() } _used = 0; + // Its request still points into the arena that just went back: let go of it, so the pool does not pin + // the array, and keep the header list's capacity for the next stream. + _request.Headers.Clear(); + _request.Method = _request.Path = _request.Scheme = _request.Authority = _request.Body = default; + _request.BodyReader = null; + // Fresh for the next stream that rents it. StreamId = 0; HeadersDone = false; diff --git a/src/protocols/ioxide.http2/Http2Connection.Write.cs b/src/protocols/ioxide.http2/Http2Connection.Write.cs index 980a9ce8..55a53828 100644 --- a/src/protocols/ioxide.http2/Http2Connection.Write.cs +++ b/src/protocols/ioxide.http2/Http2Connection.Write.cs @@ -35,6 +35,9 @@ public sealed partial class Http2Connection private TurnWaiter? _turnHead; private TurnWaiter? _turnTail; + // What a waiter woken by a turn that will never carry its bytes is told: as a TCP pipe writer reports a closed peer. + private static readonly FlushResult PeerGone = new(isCanceled: false, isCompleted: true); + private void Stage(ReadOnlySpan bytes) { if (_disposed || _writeDead) @@ -94,7 +97,7 @@ private async ValueTask PumpAsync() // May resume writers inline; _flushing is still true, so anything they stage // lands in the queue and is picked up by the check just below. - Wake(turn); + Wake(turn, default); turn = null; if (_queuedUsed == 0 || _disposed) @@ -134,8 +137,9 @@ private async ValueTask PumpAsync() private void EndTurn(TurnWaiter? faulted) { _flushing = false; - Wake(faulted); - Wake(TakeTurn()); + FlushResult result = _writeDead || _disposed ? PeerGone : default; + Wake(faulted, result); + Wake(TakeTurn(), result); } private async Task EndTurnOnReactorAsync(TurnWaiter? faulted) @@ -180,14 +184,14 @@ private ValueTask AwaitTurn(TurnWaiter waiter) } // Each link is read before its waiter wakes: one resumed inline may park again, relinking itself. - private static void Wake(TurnWaiter? turn) + private static void Wake(TurnWaiter? turn, FlushResult result) { while (turn is not null) { TurnWaiter? next = turn.Next; turn.Next = null; turn.Enlisted = false; - turn.Release(); + turn.Release(result); turn = next; } } @@ -254,7 +258,7 @@ private void ResetStream(int streamId, uint errorCode) { if (_streams.Remove(streamId, out PendingRequest? pending)) { - pending.Abort(); + Discard(pending); } Span frame = stackalloc byte[FrameHeader.Size + 4]; diff --git a/src/protocols/ioxide.http2/Http2Connection.cs b/src/protocols/ioxide.http2/Http2Connection.cs index 303f48bf..eaae34c0 100644 --- a/src/protocols/ioxide.http2/Http2Connection.cs +++ b/src/protocols/ioxide.http2/Http2Connection.cs @@ -48,8 +48,10 @@ public sealed partial class Http2Connection : IDisposable private List _ready = []; private List _dispatching = []; - // Retired requests, each keeping its Http2Request, header list and field ranges for the next stream. + // Retired requests, each keeping its Http2Request, header list and field ranges for the next stream. Capped, + // so a burst of streams leaves no more than this on an idle connection. private readonly Stack _pendingPool = new(); + private const int MaxPooledRequests = 32; // Readers whose parked ReadAsync has something to hand over. Collected during parsing and // fired once it has unwound, so a resumed handler cannot re-enter the parser mid-frame. @@ -141,7 +143,7 @@ public void Dispose() // A writer parked on its turn of the write pump has to wake into IsBroken, not hang: the // flush that would have completed its turn is never coming. - Wake(TakeTurn()); + Wake(TakeTurn(), PeerGone); // Last: a handler parked on its body resumes into its end once everything above is released. FireBodyWakes(); @@ -391,8 +393,33 @@ private async Task CompleteBufferedAsync(ValueTask inFlight, Pend private void RetireStream(PendingRequest pending) { _streams.Remove(pending.StreamId); + Recycle(pending); + } + + // Disposed, and kept for the next stream when the pool has room and the request grew nothing large: a peer + // sending thousands of header fields must not leave lists that size on the connection. Only on the reactor, + // which is the one thread that rents. + private void Recycle(PendingRequest pending) + { pending.Dispose(); - _pendingPool.Push(pending); + if (_pendingPool.Count < MaxPooledRequests && pending.IsSmall && !OffReactor) + { + _pendingPool.Push(pending); + } + } + + // A stream that ended before its request did: one without a body reader was never dispatched, so nothing else + // holds it. One with a reader has a handler running on it, which retires it when done. + private void Discard(PendingRequest pending) + { + if (pending.BodyReader is null) + { + Recycle(pending); + } + else + { + pending.Abort(); + } } // Opens at what the peer's SETTINGS advertised, not at the RFC default. Streams are created diff --git a/src/protocols/ioxide.http2/Http2ResponseWriter.cs b/src/protocols/ioxide.http2/Http2ResponseWriter.cs index 73eacd47..eee91bb1 100644 --- a/src/protocols/ioxide.http2/Http2ResponseWriter.cs +++ b/src/protocols/ioxide.http2/Http2ResponseWriter.cs @@ -230,9 +230,16 @@ private ValueTask FlushCore(bool endStream) return FlushCoreAsync(endStream); } - onReactor.GetResult(); // throws once the reactor has stopped, as awaiting it does - SendUnsentHeaders(); - return Send(endStream, sent: 0); + try + { + onReactor.GetResult(); // throws once the reactor has stopped, as awaiting it does + SendUnsentHeaders(); + return Send(endStream); + } + catch (Exception exception) + { + return ValueTask.FromException(exception); // through the task, as an async method would + } } private async ValueTask FlushCoreAsync(bool endStream) @@ -245,7 +252,7 @@ private async ValueTask FlushCoreAsync(bool endStream) await new ReactorSwitch(_connection.Reactor); SendUnsentHeaders(); - return await Send(endStream, sent: 0); + return await Send(endStream); } private void SendUnsentHeaders() @@ -257,9 +264,21 @@ private void SendUnsentHeaders() } } - // Everything staged from `sent` on, as far as credit allows, then the flush. Out of credit, the rest goes on in - // SendAfterCreditAsync. - private ValueTask Send(bool endStream, int sent) + private enum Progress : byte { Sent, Stalled, Gone } + + private ValueTask Send(bool endStream) + { + int sent = 0; + return SendStaged(endStream, ref sent) switch + { + Progress.Sent => Finish(endStream, sent), + Progress.Stalled => SendAfterCreditAsync(endStream, sent), + _ => new ValueTask(PeerGone), + }; + } + + // Everything staged from `sent` on, as far as credit allows. + private Progress SendStaged(bool endStream, ref int sent) { while (sent < _staged) { @@ -269,11 +288,11 @@ private ValueTask Send(bool endStream, int sent) if (credit < 0) { _staged = 0; // reset by the peer, or the connection is gone: nobody to send it to - return new ValueTask(PeerGone); + return Progress.Gone; } if (credit == 0) { - return SendAfterCreditAsync(endStream, sent); + return Progress.Stalled; } int chunk = Math.Min(credit, _staged - sent); @@ -282,7 +301,12 @@ private ValueTask Send(bool endStream, int sent) _connection.SendStreamedData(_streamId, _staging.AsSpan(sent, chunk), last); sent += chunk; } + return Progress.Sent; + } + // The staged bytes are out: the stream's end if it is owed one, then the flush. + private ValueTask Finish(bool endStream, int sent) + { _staged = 0; _sinceRealFlush += sent; @@ -310,27 +334,41 @@ private ValueTask Send(bool endStream, int sent) return _connection.FlushOutboundAsync(_turn); } + // Out of credit. It loops rather than calling back into Send: one frame for the whole flush, however many + // times the peer opens the window by a byte. private async ValueTask SendAfterCreditAsync(bool endStream, int sent) { - // Take the wait BEFORE flushing. The flush below is an await, and the WINDOW_UPDATE - // it exists to provoke can arrive while this writer is still inside it - at which - // point a release has no waiter to find, is dropped, and the writer then parks on a - // message that has already been and gone. Registering first means that credit - // completes this task instead, and the await below returns at once. - Task credited = _connection.WaitForSendCreditAsync(_streamId); - - // Everything staged so far has to REACH the peer before waiting on it to open the - // window. WINDOW_UPDATE is what a peer sends once it has consumed DATA, so parking - // with those bytes still in the connection's buffer waits for a message that the - // wait itself prevents - and any body larger than the initial 65535-byte window - // deadlocks until the client times out. The flush is unconditional because inside a - // pass the pass flush would otherwise carry them, and this stream is about to stop - // making progress rather than return to it. - _sinceRealFlush = 0; - await _connection.FlushOutboundAsync(_turn); - - await credited; - return await Send(endStream, sent); + while (true) + { + // Take the wait BEFORE flushing. The flush below is an await, and the WINDOW_UPDATE + // it exists to provoke can arrive while this writer is still inside it - at which + // point a release has no waiter to find, is dropped, and the writer then parks on a + // message that has already been and gone. Registering first means that credit + // completes this task instead, and the await below returns at once. + Task credited = _connection.WaitForSendCreditAsync(_streamId); + + // Everything staged so far has to REACH the peer before waiting on it to open the + // window. WINDOW_UPDATE is what a peer sends once it has consumed DATA, so parking + // with those bytes still in the connection's buffer waits for a message that the + // wait itself prevents - and any body larger than the initial 65535-byte window + // deadlocks until the client times out. The flush is unconditional because inside a + // pass the pass flush would otherwise carry them, and this stream is about to stop + // making progress rather than return to it. + _sinceRealFlush = 0; + await _connection.FlushOutboundAsync(_turn); + + await credited; + + Progress progress = SendStaged(endStream, ref sent); + if (progress == Progress.Gone) + { + return PeerGone; + } + if (progress == Progress.Sent) + { + return await Finish(endStream, sent); + } + } } private void EnsureStaging(int sizeHint) diff --git a/src/protocols/ioxide.http2/TurnWaiter.cs b/src/protocols/ioxide.http2/TurnWaiter.cs index 56aef598..dbc2bbd5 100644 --- a/src/protocols/ioxide.http2/TurnWaiter.cs +++ b/src/protocols/ioxide.http2/TurnWaiter.cs @@ -29,7 +29,7 @@ internal ValueTask Arm() return new ValueTask(this, _core.Version); } - internal void Release() => _core.SetResult(default); + internal void Release(FlushResult result) => _core.SetResult(result); FlushResult IValueTaskSource.GetResult(short token) => _core.GetResult(token); diff --git a/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs b/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs index c2b3fa51..797d4661 100644 --- a/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs +++ b/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs @@ -168,6 +168,7 @@ public static void Register(Runner runner) TaskCompletionSource[] gates = [new(), new(), new(), new()]; long[] allocated = new long[gates.Length]; + bool[] waited = new bool[gates.Length]; Task run = client.Connection.RunAsync(async (_, writer) => { writer.WriteHeaders(new Http2Response { Status = 200 }); @@ -179,6 +180,7 @@ public static void Register(Runner runner) long before = GC.GetAllocatedBytesForCurrentThread(); ValueTask flush = writer.FlushAsync(); // queued behind it: waits for its turn allocated[i] = GC.GetAllocatedBytesForCurrentThread() - before; + waited[i] = !flush.IsCompleted; await flush; } @@ -202,9 +204,55 @@ public static void Register(Runner runner) // The first wait grows the connection's queue; every wait after it reuses everything. for (int i = 1; i < allocated.Length; i++) { + Assert.True(waited[i], $"flush {i} did not wait for its turn, so it measured nothing"); Assert.True(allocated[i] == 0, $"flush {i} allocated {allocated[i]} B while it waited for its turn"); } }); + + // A stream window at zero while the peer opens the connection a byte at a time: the writer wakes on each + // WINDOW_UPDATE, still has no credit, and waits again, all inside one flush. Those waits have to be one loop, + // not one more async frame each, kept until the flush ends and unwound off this thread when it does. + runner.Test("h2 queue: a flush that waits for credit again and again keeps one frame, and resumes inline", () => + { + using var client = new StrictClient(); + + int thread = Environment.CurrentManagedThreadId; + var resumed = new TaskCompletionSource(); + Task run = client.Connection.RunAsync(async (_, writer) => + { + writer.WriteHeaders(new Http2Response { Status = 200 }); + Push(writer, "x"); + await writer.FlushAsync(); + resumed.SetResult(Environment.CurrentManagedThreadId); + }); + + client.ReleaseFlush(); // server SETTINGS + client.Feed([.. Http2FlowControlTests.Open((Http2FlowControlTests.InitialWindowSize, 0)), .. Http2FlowControlTests.Get(1)]); + client.Drain(); + + byte[] connectionByte = Http2FlowControlTests.WindowUpdate(0, 1); + for (int i = 0; i < 1_000; i++) + { + client.Feed(connectionByte); + client.Drain(); + } + + long before = GC.GetTotalMemory(forceFullCollection: true); + for (int i = 0; i < 20_000; i++) + { + client.Feed(connectionByte); + client.Drain(); + } + long retained = GC.GetTotalMemory(forceFullCollection: true) - before; + + client.Feed(Http2FlowControlTests.WindowUpdate(1, 1)); // the stream's credit: the chunk goes, the flush ends + client.Drain(); + + Assert.True(resumed.Task.Wait(5_000), "the flush never completed"); + Assert.True(retained < 256 * 1024, $"20,000 waits for credit inside one flush retained {retained} B"); + Assert.Equal(thread, resumed.Task.Result); + client.Close(run); + }); } private static byte[] Ping() => [0, 0, 8, 0x6, 0, 0, 0, 0, 0, 1, 2, 3, 4, 5, 6, 7, 8]; diff --git a/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs b/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs index 4fa71f88..851bd56f 100644 --- a/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs +++ b/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs @@ -28,13 +28,13 @@ public static void Register(Runner runner) client.ReleaseFlush(); // their ack client.Feed(Request(1, post: true, "/first", "a.example", [("x-one", "1"), ("x-two", "2")], body: "hello")); client.ReleaseFlush(); - client.Feed(Request(3, post: false, "/", authority: null, [], body: null)); + client.Feed(Request(3, post: false, "/", authority: null, [], body: null, https: true)); client.ReleaseFlush(); client.Close(run); Assert.Equal(2, seen.Count); - Assert.Equal("1 POST /first a.example [x-one=1 x-two=2] hello", seen[0].Fields); - Assert.Equal("3 GET / - [] -", seen[1].Fields); + Assert.Equal("1 POST http /first a.example [x-one=1 x-two=2] hello", seen[0].Fields); + Assert.Equal("3 GET https / - [] -", seen[1].Fields); Assert.True(ReferenceEquals(seen[0].Request, seen[1].Request), "the second stream reuses the first one's request"); }); @@ -43,20 +43,30 @@ public static void Register(Runner runner) using var client = new StrictClient(); Http2Response response = Http2Response.Text("ok"); // shared, so the handler allocates nothing - Task run = client.Connection.RunAsync(_ => response); + int served = 0; + Task run = client.Connection.RunAsync(_ => + { + served++; + return response; + }); client.ReleaseFlush(); // server SETTINGS client.SendRequests(); // preface and SETTINGS client.ReleaseFlush(); // their ack + // Two warm-up passes: the ready lists swap each pass, so each grows in a pass of its own. const int perPass = 16; - client.Feed(Batch(1, perPass)); // warm-up: the requests the next batch reuses + client.Feed(Batch(1, perPass)); + client.ReleaseFlush(); + client.Feed(Batch(1 + (2 * perPass), perPass)); client.ReleaseFlush(); - byte[] batch = Batch(1 + (2 * perPass), perPass); + byte[] batch = Batch(1 + (4 * perPass), perPass); + int servedBefore = served; long before = GC.GetAllocatedBytesForCurrentThread(); client.Feed(batch); // parsed, dispatched and answered inline long perRequest = (GC.GetAllocatedBytesForCurrentThread() - before) / perPass; + Assert.Equal(perPass, served - servedBefore); // all of them inside the window client.ReleaseFlush(); client.Close(run); @@ -75,7 +85,7 @@ private static string Describe(Http2Request request) fields.Add($"{Ascii(field.Key)}={Ascii(field.Value)}"); } - return $"{request.StreamId} {Ascii(request.Method)} {Ascii(request.Path)} {Or(request.Authority)} " + return $"{request.StreamId} {Ascii(request.Method)} {Ascii(request.Scheme)} {Ascii(request.Path)} {Or(request.Authority)} " + $"[{string.Join(' ', fields)}] {Or(request.Body)}"; static string Or(ReadOnlyMemory bytes) => bytes.IsEmpty ? "-" : Ascii(bytes); @@ -97,9 +107,9 @@ private static byte[] Batch(int firstStream, int count) // HEADERS, with END_STREAM unless a DATA frame carrying the body follows. HPACK from the static table and // literals without indexing only, so the request needs no encoder of its own. private static byte[] Request(int streamId, bool post, string path, string? authority, - (string Name, string Value)[] fields, string? body) + (string Name, string Value)[] fields, string? body, bool https = false) { - var block = new List { post ? (byte)0x83 : (byte)0x82, 0x86 }; // :method, :scheme http + var block = new List { post ? (byte)0x83 : (byte)0x82, https ? (byte)0x87 : (byte)0x86 }; // :method, :scheme if (path == "/") { block.Add(0x84); // :path / From bcc0bb795646b7a3548df2aa7a59519eb1e437f2 Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Thu, 5 Nov 2026 00:32:56 +0000 Subject: [PATCH 3/4] http2: a credit wait hears the peer gone from a faulted turn, a dispatch walks the ready list in place, and the tests guard the review fixes From the second review: - SendAfterCreditAsync ignored what its flush's turn returned. A writer short of credit whose turn faulted (the pump being another writer, so the read loop does not tear down) went on to wait for a WINDOW_UPDATE that cannot come, until the connection's next read. It returns PeerGone now. - The second ready list guarded against something that cannot happen: only ParseAvailable adds to the list, and it never runs during the walk. _dispatching and TakeReady are gone. - FlushAsync's doc says each call has to complete before the next. Tests: - the faulted-turn test checks the flush returned IsCompleted, and a new one covers the credit wait; - the reuse test splits two header blocks over HEADERS + CONTINUATION, so stale HeadersDone or RequestEnded would show, and new ones cover the parked handler and a reset stream's Discard; - the warm-request bound is 100 B, which main's HPACK closure (145 B) fails. Each fails against a mutation of the code it guards. New comments are cut to one line. --- .../ioxide.http2/Http2Connection.Frames.cs | 4 +- .../ioxide.http2/Http2Connection.Streamed.cs | 6 +- .../ioxide.http2/Http2Connection.Write.cs | 3 +- src/protocols/ioxide.http2/Http2Connection.cs | 31 ++---- .../ioxide.http2/Http2ResponseWriter.cs | 13 ++- src/protocols/ioxide.http2/TurnWaiter.cs | 10 +- .../Http2OutputQueueTests.cs | 64 ++++++++--- .../Http2RequestReuseTests.cs | 105 +++++++++++++++--- 8 files changed, 163 insertions(+), 73 deletions(-) diff --git a/src/protocols/ioxide.http2/Http2Connection.Frames.cs b/src/protocols/ioxide.http2/Http2Connection.Frames.cs index 7b1d3d10..8f594e32 100644 --- a/src/protocols/ioxide.http2/Http2Connection.Frames.cs +++ b/src/protocols/ioxide.http2/Http2Connection.Frames.cs @@ -737,13 +737,11 @@ public void Dispose() } _used = 0; - // Its request still points into the arena that just went back: let go of it, so the pool does not pin - // the array, and keep the header list's capacity for the next stream. + // 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; - // Fresh for the next stream that rents it. StreamId = 0; HeadersDone = false; RequestEnded = false; diff --git a/src/protocols/ioxide.http2/Http2Connection.Streamed.cs b/src/protocols/ioxide.http2/Http2Connection.Streamed.cs index df098d5a..73f4df3f 100644 --- a/src/protocols/ioxide.http2/Http2Connection.Streamed.cs +++ b/src/protocols/ioxide.http2/Http2Connection.Streamed.cs @@ -55,11 +55,11 @@ public Task RunAsync(Func private void DispatchStreamed(Func handler) { - List ready = TakeReady(); try { - foreach (PendingRequest pending in ready) + for (int i = 0; i < _ready.Count; i++) { + PendingRequest pending = _ready[i]; try { Http2Request request = pending.Freeze(); @@ -75,7 +75,7 @@ private void DispatchStreamed(Func } finally { - ready.Clear(); + _ready.Clear(); } } diff --git a/src/protocols/ioxide.http2/Http2Connection.Write.cs b/src/protocols/ioxide.http2/Http2Connection.Write.cs index 55a53828..0446b770 100644 --- a/src/protocols/ioxide.http2/Http2Connection.Write.cs +++ b/src/protocols/ioxide.http2/Http2Connection.Write.cs @@ -56,8 +56,7 @@ private void Stage(ReadOnlySpan bytes) } } - // Not async: a caller that only waits for the pump's next turn gets that wait as it is. A streamed - // writer passes the waiter it keeps, so its flush allocates nothing; anyone else parks a new one. + // Not async: a wait for the pump's next turn is returned as it is, on the caller's own waiter if it keeps one. private ValueTask FlushAsync(TurnWaiter? waiter = null) { if (_disposed || _writeDead) diff --git a/src/protocols/ioxide.http2/Http2Connection.cs b/src/protocols/ioxide.http2/Http2Connection.cs index eaae34c0..0f598551 100644 --- a/src/protocols/ioxide.http2/Http2Connection.cs +++ b/src/protocols/ioxide.http2/Http2Connection.cs @@ -42,14 +42,9 @@ public sealed partial class Http2Connection : IDisposable private byte[] _headerScratch = new byte[16 * 1024]; private readonly Dictionary _streams = new(); + private readonly List _ready = []; - // Requests ready for their handler, and the list a dispatch is walking: swapped each pass, so the - // parser never adds to the list being walked and no snapshot is copied. - private List _ready = []; - private List _dispatching = []; - - // Retired requests, each keeping its Http2Request, header list and field ranges for the next stream. Capped, - // so a burst of streams leaves no more than this on an idle connection. + // Retired requests, kept for the next stream; capped, so a burst leaves little on an idle connection. private readonly Stack _pendingPool = new(); private const int MaxPooledRequests = 32; @@ -294,21 +289,14 @@ private void Append(ReadOnlySpan data) _inboundUsed += data.Length; } - // Taken whole before the walk: the list must not be mutated while it is walked. The caller clears - // it once done. - private List TakeReady() - { - (_ready, _dispatching) = (_dispatching, _ready); - return _dispatching; - } - private void DispatchBuffered(Func> handler) { - List ready = TakeReady(); try { - foreach (PendingRequest pending in ready) + // Only ParseAvailable adds to the list, and it never runs during the walk. + for (int i = 0; i < _ready.Count; i++) { + PendingRequest pending = _ready[i]; ValueTask inFlight; try @@ -348,7 +336,7 @@ private void DispatchBuffered(Func> handl } finally { - ready.Clear(); + _ready.Clear(); } } @@ -396,9 +384,7 @@ private void RetireStream(PendingRequest pending) Recycle(pending); } - // Disposed, and kept for the next stream when the pool has room and the request grew nothing large: a peer - // sending thousands of header fields must not leave lists that size on the connection. Only on the reactor, - // which is the one thread that rents. + // Kept unless the pool is full or a peer's header lines grew its lists; only the reactor rents. private void Recycle(PendingRequest pending) { pending.Dispose(); @@ -408,8 +394,7 @@ private void Recycle(PendingRequest pending) } } - // A stream that ended before its request did: one without a body reader was never dispatched, so nothing else - // holds it. One with a reader has a handler running on it, which retires it when done. + // Ended before its request did: without a body reader it was never dispatched; with one, its handler retires it. private void Discard(PendingRequest pending) { if (pending.BodyReader is null) diff --git a/src/protocols/ioxide.http2/Http2ResponseWriter.cs b/src/protocols/ioxide.http2/Http2ResponseWriter.cs index eee91bb1..1f73372f 100644 --- a/src/protocols/ioxide.http2/Http2ResponseWriter.cs +++ b/src/protocols/ioxide.http2/Http2ResponseWriter.cs @@ -163,6 +163,7 @@ public void Advance(int count) /// once the peer has reset the stream or the connection is /// gone, as a TCP pipe writer reports a closed peer: nothing written from then on reaches anyone. /// + /// Each call has to complete before the next one is made. public ValueTask FlushAsync() => FlushCore(endStream: false); /// @@ -220,8 +221,7 @@ internal async ValueTask FailAsync() private static readonly FlushResult PeerGone = new(isCanceled: false, isCompleted: true); - // Not async on the reactor: a flush that sends and then only waits for the write pump's turn returns that - // wait as it is, on this writer's own waiter, so it allocates nothing. The rest goes the async way. + // Not async on the reactor: a flush that only waits for the pump's turn returns that wait, on this writer's waiter. private ValueTask FlushCore(bool endStream) { var onReactor = new ReactorSwitch(_connection.Reactor); @@ -334,8 +334,7 @@ private ValueTask Finish(bool endStream, int sent) return _connection.FlushOutboundAsync(_turn); } - // Out of credit. It loops rather than calling back into Send: one frame for the whole flush, however many - // times the peer opens the window by a byte. + // Out of credit: one loop, so one async frame, however many times the peer opens the window by a byte. private async ValueTask SendAfterCreditAsync(bool endStream, int sent) { while (true) @@ -355,7 +354,11 @@ private async ValueTask SendAfterCreditAsync(bool endStream, int se // pass the pass flush would otherwise carry them, and this stream is about to stop // making progress rather than return to it. _sinceRealFlush = 0; - await _connection.FlushOutboundAsync(_turn); + if ((await _connection.FlushOutboundAsync(_turn)).IsCompleted) + { + _staged = 0; // the write side is gone, so no credit is coming either + return PeerGone; + } await credited; diff --git a/src/protocols/ioxide.http2/TurnWaiter.cs b/src/protocols/ioxide.http2/TurnWaiter.cs index dbc2bbd5..ce003412 100644 --- a/src/protocols/ioxide.http2/TurnWaiter.cs +++ b/src/protocols/ioxide.http2/TurnWaiter.cs @@ -3,10 +3,7 @@ namespace ioxide.http2; -/// -/// A flush waiting for the write pump's next turn (Http2Connection.Write.cs). Reusable: a streamed writer -/// keeps one for its whole life, so a flush that only waits for the turn allocates nothing. -/// +/// A flush waiting for the write pump's next turn; a streamed writer keeps one for its whole life. internal sealed class TurnWaiter : IValueTaskSource { private ManualResetValueTaskSourceCore _core = new() @@ -17,10 +14,7 @@ internal sealed class TurnWaiter : IValueTaskSource /// The waiter parked after this one, for the same turn. internal TurnWaiter? Next; - /// - /// Parked and not woken yet. A flush whose wait was never awaited leaves its writer in the turn, and the - /// writer's next flush then waits for that same wake instead of joining a second time. - /// + /// Parked and not woken yet: a writer whose wait went unawaited waits for that same wake again. internal bool Enlisted; internal ValueTask Arm() diff --git a/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs b/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs index 797d4661..085d3b17 100644 --- a/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs +++ b/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs @@ -135,21 +135,13 @@ public static void Register(Runner runner) using var client = new StrictClient(); var gate = new TaskCompletionSource(); - var woke = new TaskCompletionSource(); + var woke = new TaskCompletionSource(); Task run = client.Connection.RunAsync(async (_, writer) => { writer.WriteHeaders(new Http2Response { Status = 200 }); await gate.Task; // resumed outside the pass, its flush in flight Push(writer, "lost"); - try - { - await writer.FlushAsync(); // queued behind it: its turn is the flush after - } - catch - { - // The transport is gone; waking at all is what this test is about. - } - woke.SetResult(); + woke.SetResult(await FlushOutcome(writer)); // queued behind it: its turn is the flush after }); client.ReleaseFlush(); // server SETTINGS @@ -159,9 +151,45 @@ public static void Register(Runner runner) client.FaultFlush(); // and that flush faults Assert.True(woke.Task.Wait(5_000), "the writer whose chunk the faulted flush carried never woke"); + Assert.Equal("peer gone", woke.Task.Result); Assert.True(run.Wait(5_000), "connection wound down after the transport fault"); }); + // The fault reaches the writer that pumps, not the read loop, so no teardown releases stream 1's wait for credit. + runner.Test("h2 queue: a flush waiting for credit hears the peer gone when the turn it waits on faults", () => + { + using var client = new StrictClient(); + + var gates = new Dictionary { [1] = new(), [3] = new() }; + var outcome = new TaskCompletionSource(); + Task run = client.Connection.RunAsync(async (request, writer) => + { + int id = request.StreamId; + writer.WriteHeaders(new Http2Response { Status = 200 }); + await gates[id].Task; + Push(writer, id == 1 ? "xy" : "x"); + string result = await FlushOutcome(writer); + if (id == 1) + { + outcome.TrySetResult(result); + } + }); + + client.ReleaseFlush(); // server SETTINGS + client.Feed([.. Http2FlowControlTests.Open((Http2FlowControlTests.InitialWindowSize, 0)), + .. Http2FlowControlTests.Get(1), .. Http2FlowControlTests.Get(3), + .. Http2FlowControlTests.WindowUpdate(3, 1_000), .. Http2FlowControlTests.WindowUpdate(1, 1)]); + client.Drain(); // the pass is out; the read loop waits for input + gates[3].SetResult(); // stream 3 flushes first, so it is the pump + gates[1].SetResult(); // stream 1 queues its one byte of credit, then waits for more + client.ReleaseFlush(); // stream 3's flush; the pump's next turn carries stream 1's byte + client.FaultFlush(); // and that flush faults + + Assert.True(outcome.Task.IsCompleted, "stream 1's flush still waited for credit after its turn faulted"); + Assert.Equal("peer gone", outcome.Task.Result); + client.Close(run); + }); + runner.Test("h2 queue: a streamed flush that waits for the pump's turn allocates nothing", () => { using var client = new StrictClient(); @@ -209,9 +237,7 @@ public static void Register(Runner runner) } }); - // A stream window at zero while the peer opens the connection a byte at a time: the writer wakes on each - // WINDOW_UPDATE, still has no credit, and waits again, all inside one flush. Those waits have to be one loop, - // not one more async frame each, kept until the flush ends and unwound off this thread when it does. + // The stream window stays shut while the connection opens a byte at a time: every wait for credit is one flush's. runner.Test("h2 queue: a flush that waits for credit again and again keeps one frame, and resumes inline", () => { using var client = new StrictClient(); @@ -257,6 +283,18 @@ public static void Register(Runner runner) private static byte[] Ping() => [0, 0, 8, 0x6, 0, 0, 0, 0, 0, 1, 2, 3, 4, 5, 6, 7, 8]; + private static async Task FlushOutcome(Http2ResponseWriter writer) + { + try + { + return (await writer.FlushAsync()).IsCompleted ? "peer gone" : "flushed"; + } + catch (Exception e) + { + return e.GetType().Name; + } + } + private static void Push(Http2ResponseWriter writer, string text) { Span span = writer.GetSpan(text.Length); diff --git a/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs b/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs index 851bd56f..7fb9773c 100644 --- a/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs +++ b/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs @@ -1,12 +1,12 @@ +using System.Collections; +using System.Reflection; using System.Text; using ioxide.http2; namespace Ioxide.Tests; /// -/// A connection keeps a retired request, with its header list and field ranges, for the next stream: the next -/// request must come back holding nothing of the one before, and a warm connection's request allocates none -/// of the parser's objects. +/// A connection hands a retired request, with its header list and field ranges, to the next stream. /// internal static class Http2RequestReuseTests { @@ -28,14 +28,76 @@ public static void Register(Runner runner) client.ReleaseFlush(); // their ack client.Feed(Request(1, post: true, "/first", "a.example", [("x-one", "1"), ("x-two", "2")], body: "hello")); client.ReleaseFlush(); - client.Feed(Request(3, post: false, "/", authority: null, [], body: null, https: true)); + // Split over HEADERS and CONTINUATION, so a header-block flag left set by the request before would show. + client.Feed(Request(3, post: false, "/", authority: null, [], body: null, https: true, split: true)); + client.ReleaseFlush(); + client.Feed(Request(5, post: true, "/again", "c.example", [], body: "world", split: true)); client.ReleaseFlush(); client.Close(run); - Assert.Equal(2, seen.Count); + Assert.Equal(3, seen.Count); Assert.Equal("1 POST http /first a.example [x-one=1 x-two=2] hello", seen[0].Fields); Assert.Equal("3 GET https / - [] -", seen[1].Fields); - Assert.True(ReferenceEquals(seen[0].Request, seen[1].Request), "the second stream reuses the first one's request"); + Assert.Equal("5 POST http /again c.example [] world", seen[2].Fields); + Assert.True(ReferenceEquals(seen[0].Request, seen[1].Request) && ReferenceEquals(seen[1].Request, seen[2].Request), + "each stream reuses the request of the one before"); + }); + + runner.Test("h2 request: a parked handler's request goes back once it answers", () => + { + using var client = new StrictClient(); + + var parked = new TaskCompletionSource(); + var seen = new List<(Http2Request Request, string Fields)>(); + Task run = client.Connection.RunAsync(request => + { + seen.Add((request, Describe(request))); + return request.StreamId == 1 + ? new ValueTask(parked.Task) // answered later, through CompleteBufferedAsync + : new ValueTask(Http2Response.Text("ok")); + }); + + client.ReleaseFlush(); // server SETTINGS + client.SendRequests(); // preface and SETTINGS + client.Drain(); + client.Feed(Request(1, post: false, "/park", "a.example", [("x-one", "1")], body: null)); + client.Drain(); + parked.SetResult(Http2Response.Text("late")); + client.Drain(); + client.Feed(Request(3, post: false, "/", authority: null, [], body: null)); + client.Drain(); + client.Close(run); + + Assert.Equal("3 GET http / - [] -", seen[1].Fields); + Assert.True(ReferenceEquals(seen[0].Request, seen[1].Request), "the parked handler's request was reused"); + }); + + runner.Test("h2 request: a request reset before it ends comes back for the next stream holding nothing", () => + { + using var client = new StrictClient(); + + var seen = new List(); + Task run = client.Connection.RunAsync(request => + { + seen.Add(Describe(request)); + return Http2Response.Text("ok"); + }); + + client.ReleaseFlush(); // server SETTINGS + client.SendRequests(); // preface and SETTINGS + client.Drain(); + // Its headers and part of its body, then the reset: never dispatched, so the parser takes it back. + client.Feed([.. Request(1, post: true, "/first", "a.example", [("x-one", "1")], body: "hel", endStream: false), + .. Http2FlowControlTests.RstStream(1, 8)]); + client.Drain(); + int pooled = ((ICollection)typeof(Http2Connection) + .GetField("_pendingPool", BindingFlags.NonPublic | BindingFlags.Instance)!.GetValue(client.Connection)!).Count; + client.Feed(Request(3, post: false, "/", authority: null, [], body: null)); + client.Drain(); + client.Close(run); + + Assert.Equal(1, pooled); + Assert.True(seen.SequenceEqual(["3 GET http / - [] -"]), $"the handler was handed: {string.Join(", ", seen)}"); }); runner.Test("h2 request: a warm connection's request allocates none of the parser's objects", () => @@ -54,7 +116,7 @@ public static void Register(Runner runner) client.SendRequests(); // preface and SETTINGS client.ReleaseFlush(); // their ack - // Two warm-up passes: the ready lists swap each pass, so each grows in a pass of its own. + // Two warm-up passes grow the ready list and the pool. const int perPass = 16; client.Feed(Batch(1, perPass)); client.ReleaseFlush(); @@ -71,9 +133,8 @@ public static void Register(Runner runner) client.ReleaseFlush(); client.Close(run); - // Before reuse, every request allocated about 700 B of parser objects. What is left is the test - // transport's own flush, shared by the pass. - Assert.True(perRequest < 200, $"a request on a warm connection allocated {perRequest} B"); + // What is left is the test transport's own flush, shared by the pass: about 65 B. + Assert.True(perRequest < 100, $"a request on a warm connection allocated {perRequest} B"); }); } @@ -104,10 +165,10 @@ private static byte[] Batch(int firstStream, int count) return bytes.ToArray(); } - // HEADERS, with END_STREAM unless a DATA frame carrying the body follows. HPACK from the static table and - // literals without indexing only, so the request needs no encoder of its own. + // HEADERS, with END_STREAM unless a DATA frame carrying the body follows; split, a CONTINUATION ends the block. + // HPACK from the static table and literals without indexing only, so the request needs no encoder of its own. private static byte[] Request(int streamId, bool post, string path, string? authority, - (string Name, string Value)[] fields, string? body, bool https = false) + (string Name, string Value)[] fields, string? body, bool https = false, bool split = false, bool endStream = true) { var block = new List { post ? (byte)0x83 : (byte)0x82, https ? (byte)0x87 : (byte)0x86 }; // :method, :scheme if (path == "/") @@ -131,12 +192,24 @@ private static byte[] Request(int streamId, bool post, string path, string? auth AddString(block, value); } + byte end = body is null && endStream ? (byte)0x1 : (byte)0x0; var frames = new List(); - frames.AddRange(FrameHeader(block.Count, 0x1, body is null ? (byte)0x5 : (byte)0x4, streamId)); - frames.AddRange(block); + if (split) + { + int half = block.Count / 2; + frames.AddRange(FrameHeader(half, 0x1, end, streamId)); // no END_HEADERS + frames.AddRange(block.Take(half)); + frames.AddRange(FrameHeader(block.Count - half, 0x9, 0x4, streamId)); // CONTINUATION, END_HEADERS + frames.AddRange(block.Skip(half)); + } + else + { + frames.AddRange(FrameHeader(block.Count, 0x1, (byte)(end | 0x4), streamId)); + frames.AddRange(block); + } if (body is not null) { - frames.AddRange(FrameHeader(body.Length, 0x0, 0x1, streamId)); + frames.AddRange(FrameHeader(body.Length, 0x0, endStream ? (byte)0x1 : (byte)0x0, streamId)); frames.AddRange(Encoding.ASCII.GetBytes(body)); } return frames.ToArray(); From 7cdc212ea9a6b707910f78b3cdac64a044e289ab Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Thu, 5 Nov 2026 20:34:58 +0000 Subject: [PATCH 4/4] http2: the ready list is a queue, so it holds only requests not yet dispatched, and a dead write side releases credit waits A dispatch that walked the list in place left dispatched requests in it, so a handler disposing the connection mid-walk disposed a parked handler's request. A writer already parked on credit when another writer's flush faulted waited for a teardown the read loop had no reason to start. Tests guard the request pool's cap, its header-line limit, and a retired request letting go of its arena. --- .../ioxide.http2/Http2Connection.Frames.cs | 4 +- .../ioxide.http2/Http2Connection.Streamed.cs | 28 +++---- .../ioxide.http2/Http2Connection.Write.cs | 4 + src/protocols/ioxide.http2/Http2Connection.cs | 67 +++++++---------- .../Http2OutputQueueTests.cs | 36 +++++++++ .../Http2RequestReuseTests.cs | 75 ++++++++++++++++++- 6 files changed, 153 insertions(+), 61 deletions(-) diff --git a/src/protocols/ioxide.http2/Http2Connection.Frames.cs b/src/protocols/ioxide.http2/Http2Connection.Frames.cs index 8f594e32..111be240 100644 --- a/src/protocols/ioxide.http2/Http2Connection.Frames.cs +++ b/src/protocols/ioxide.http2/Http2Connection.Frames.cs @@ -256,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 payload) @@ -408,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 payload) diff --git a/src/protocols/ioxide.http2/Http2Connection.Streamed.cs b/src/protocols/ioxide.http2/Http2Connection.Streamed.cs index 73f4df3f..cf1fc761 100644 --- a/src/protocols/ioxide.http2/Http2Connection.Streamed.cs +++ b/src/protocols/ioxide.http2/Http2Connection.Streamed.cs @@ -55,27 +55,19 @@ public Task RunAsync(Func private void DispatchStreamed(Func handler) { - try + while (_ready.TryDequeue(out PendingRequest? pending)) { - for (int i = 0; i < _ready.Count; i++) + try { - PendingRequest pending = _ready[i]; - try - { - Http2Request request = pending.Freeze(); - Http2ResponseWriter writer = RentWriter(request.StreamId); - _ = ServeStreamedAsync(handler, request, writer, pending); // the writer owns this stream, and retires it when done - } - catch - { - pending.Dispose(); - throw; - } + Http2Request request = pending.Freeze(); + Http2ResponseWriter writer = RentWriter(request.StreamId); + _ = ServeStreamedAsync(handler, request, writer, pending); // the writer owns this stream, and retires it when done + } + catch + { + pending.Dispose(); + throw; } - } - finally - { - _ready.Clear(); } } diff --git a/src/protocols/ioxide.http2/Http2Connection.Write.cs b/src/protocols/ioxide.http2/Http2Connection.Write.cs index 0446b770..464758a2 100644 --- a/src/protocols/ioxide.http2/Http2Connection.Write.cs +++ b/src/protocols/ioxide.http2/Http2Connection.Write.cs @@ -139,6 +139,10 @@ private void EndTurn(TurnWaiter? faulted) FlushResult result = _writeDead || _disposed ? PeerGone : default; Wake(faulted, result); Wake(TakeTurn(), result); + if (_writeDead) + { + ReleaseAllCreditWaiters(); // the fault reached a writer, so no teardown is on its way to them + } } private async Task EndTurnOnReactorAsync(TurnWaiter? faulted) diff --git a/src/protocols/ioxide.http2/Http2Connection.cs b/src/protocols/ioxide.http2/Http2Connection.cs index 0f598551..4b6fd719 100644 --- a/src/protocols/ioxide.http2/Http2Connection.cs +++ b/src/protocols/ioxide.http2/Http2Connection.cs @@ -42,7 +42,7 @@ public sealed partial class Http2Connection : IDisposable private byte[] _headerScratch = new byte[16 * 1024]; private readonly Dictionary _streams = new(); - private readonly List _ready = []; + private readonly Queue _ready = new(); // Retired requests, kept for the next stream; capped, so a burst leaves little on an idle connection. private readonly Stack _pendingPool = new(); @@ -291,52 +291,43 @@ private void Append(ReadOnlySpan data) private void DispatchBuffered(Func> handler) { - try + while (_ready.TryDequeue(out PendingRequest? pending)) { - // Only ParseAvailable adds to the list, and it never runs during the walk. - for (int i = 0; i < _ready.Count; i++) + ValueTask inFlight; + + try + { + Http2Request request = pending.Freeze(); + inFlight = handler(request); + } + catch (Exception exception) { - PendingRequest pending = _ready[i]; - ValueTask inFlight; + // A synchronous handler throws before there is a task. It gets what a faulted task + // gets, a 500 on its own stream, rather than taking the connection down. + inFlight = ValueTask.FromException(exception); + } + // Answered synchronously, which nearly every handler does. Stay inline: this response + // is staged in time for the pass flush, so it still leaves with every other one, and + // there is no Task to allocate. + if (inFlight.IsCompletedSuccessfully) + { try { - Http2Request request = pending.Freeze(); - inFlight = handler(request); - } - catch (Exception exception) - { - // A synchronous handler throws before there is a task. It gets what a faulted task - // gets, a 500 on its own stream, rather than taking the connection down. - inFlight = ValueTask.FromException(exception); + WriteResponse(pending.StreamId, inFlight.Result); } - - // Answered synchronously, which nearly every handler does. Stay inline: this response - // is staged in time for the pass flush, so it still leaves with every other one, and - // there is no Task to allocate. - if (inFlight.IsCompletedSuccessfully) + finally { - try - { - WriteResponse(pending.StreamId, inFlight.Result); - } - finally - { - RetireStream(pending); - } - continue; + RetireStream(pending); } - - // It parked - a database, an upstream, a disk. Awaiting here would hold every OTHER - // stream on this connection behind it, including responses already staged and ready to - // go, because they all share this one dispatch loop and one TCP connection. So it - // finishes on its own and writes its own bytes when it does. - _ = CompleteBufferedAsync(inFlight, pending); + continue; } - } - finally - { - _ready.Clear(); + + // It parked - a database, an upstream, a disk. Awaiting here would hold every OTHER + // stream on this connection behind it, including responses already staged and ready to + // go, because they all share this one dispatch loop and one TCP connection. So it + // finishes on its own and writes its own bytes when it does. + _ = CompleteBufferedAsync(inFlight, pending); } } diff --git a/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs b/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs index 085d3b17..a00dcd2d 100644 --- a/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs +++ b/tests/Ioxide.Tests.Unit/Http2OutputQueueTests.cs @@ -190,6 +190,42 @@ .. Http2FlowControlTests.Get(1), .. Http2FlowControlTests.Get(3), client.Close(run); }); + // Stream 1's own flush is long done, and the fault reaches stream 3, so only the turn's end can wake stream 1. + runner.Test("h2 queue: a flush parked on credit hears the peer gone when another writer's flush faults", () => + { + using var client = new StrictClient(); + + var gates = new Dictionary { [1] = new(), [3] = new() }; + var outcome = new TaskCompletionSource(); + Task run = client.Connection.RunAsync(async (request, writer) => + { + int id = request.StreamId; + writer.WriteHeaders(new Http2Response { Status = 200 }); + await gates[id].Task; + Push(writer, id == 1 ? "xy" : "x"); + string result = await FlushOutcome(writer); + if (id == 1) + { + outcome.TrySetResult(result); + } + }); + + client.ReleaseFlush(); // server SETTINGS + client.Feed([.. Http2FlowControlTests.Open((Http2FlowControlTests.InitialWindowSize, 0)), + .. Http2FlowControlTests.Get(1), .. Http2FlowControlTests.Get(3), + .. Http2FlowControlTests.WindowUpdate(3, 1_000), .. Http2FlowControlTests.WindowUpdate(1, 1)]); + client.Drain(); // the pass is out; the read loop waits for input + gates[1].SetResult(); // stream 1 sends its one byte of credit + client.ReleaseFlush(); // that flush ends, and it parks for more + Assert.True(!outcome.Task.IsCompleted, "stream 1 was meant to be waiting for credit"); + gates[3].SetResult(); // stream 3 flushes + client.FaultFlush(); // and that flush faults + + Assert.True(outcome.Task.IsCompleted, "stream 1's flush still waited for credit after the write side died"); + Assert.Equal("peer gone", outcome.Task.Result); + client.Close(run); + }); + runner.Test("h2 queue: a streamed flush that waits for the pump's turn allocates nothing", () => { using var client = new StrictClient(); diff --git a/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs b/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs index 7fb9773c..70b892ad 100644 --- a/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs +++ b/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs @@ -64,6 +64,9 @@ public static void Register(Runner runner) client.Drain(); parked.SetResult(Http2Response.Text("late")); client.Drain(); + Http2Request retired = seen[0].Request; // in the pool now, which must not pin the arena it read from + Assert.True(retired.Method.IsEmpty && retired.Path.IsEmpty && retired.Headers.AsSpan().IsEmpty, + "a retired request still points into the arena that went back"); client.Feed(Request(3, post: false, "/", authority: null, [], body: null)); client.Drain(); client.Close(run); @@ -90,8 +93,7 @@ public static void Register(Runner runner) client.Feed([.. Request(1, post: true, "/first", "a.example", [("x-one", "1")], body: "hel", endStream: false), .. Http2FlowControlTests.RstStream(1, 8)]); client.Drain(); - int pooled = ((ICollection)typeof(Http2Connection) - .GetField("_pendingPool", BindingFlags.NonPublic | BindingFlags.Instance)!.GetValue(client.Connection)!).Count; + int pooled = Pooled(client.Connection); client.Feed(Request(3, post: false, "/", authority: null, [], body: null)); client.Drain(); client.Close(run); @@ -100,6 +102,69 @@ public static void Register(Runner runner) Assert.True(seen.SequenceEqual(["3 GET http / - [] -"]), $"the handler was handed: {string.Join(", ", seen)}"); }); + runner.Test("h2 request: a handler that disposes the connection leaves a parked handler's request alone", () => + { + using var client = new StrictClient(); + + var parked = new TaskCompletionSource(); + Http2Request? held = null; + Task run = client.Connection.RunAsync(request => + { + if (request.StreamId == 1) + { + held = request; + return new ValueTask(parked.Task); + } + client.Connection.Dispose(); // in the same dispatch as stream 1 + return new ValueTask(Http2Response.Text("ok")); + }); + + client.ReleaseFlush(); // server SETTINGS + client.SendRequests(1, 3); + string path = Ascii(held!.Path); + parked.SetResult(Http2Response.Text("late")); + client.Close(run); + + Assert.Equal("/", path); + }); + + runner.Test("h2 request: a burst leaves at most 32 requests on the connection", () => + { + using var client = new StrictClient(); + Task run = client.Connection.RunAsync(_ => Http2Response.Text("ok")); + + client.ReleaseFlush(); // server SETTINGS + client.SendRequests(); // preface and SETTINGS + client.Drain(); + client.Feed(Batch(1, 40)); // one read, so all 40 are open before the first retires + client.Drain(); + int pooled = Pooled(client.Connection); + client.Close(run); + + Assert.Equal(32, pooled); + }); + + foreach ((int lines, int kept) in new[] { (64, 1), (65, 0) }) + { + int count = lines, expected = kept; + runner.Test($"h2 request: a request with {count} header lines {(expected == 1 ? "is" : "is not")} kept for the next stream", () => + { + using var client = new StrictClient(); + Task run = client.Connection.RunAsync(_ => Http2Response.Text("ok")); + + client.ReleaseFlush(); // server SETTINGS + client.SendRequests(); // preface and SETTINGS + client.Drain(); + client.Feed(Request(1, post: false, "/", authority: null, + Enumerable.Range(0, count).Select(i => ($"x-{i}", "1")).ToArray(), body: null)); + client.Drain(); + int pooled = Pooled(client.Connection); + client.Close(run); + + Assert.Equal(expected, pooled); + }); + } + runner.Test("h2 request: a warm connection's request allocates none of the parser's objects", () => { using var client = new StrictClient(); @@ -116,7 +181,7 @@ public static void Register(Runner runner) client.SendRequests(); // preface and SETTINGS client.ReleaseFlush(); // their ack - // Two warm-up passes grow the ready list and the pool. + // Two warm-up passes grow the ready queue and the pool. const int perPass = 16; client.Feed(Batch(1, perPass)); client.ReleaseFlush(); @@ -138,6 +203,10 @@ public static void Register(Runner runner) }); } + private static int Pooled(Http2Connection connection) + => ((ICollection)typeof(Http2Connection) + .GetField("_pendingPool", BindingFlags.NonPublic | BindingFlags.Instance)!.GetValue(connection)!).Count; + private static string Describe(Http2Request request) { var fields = new List();