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..111be240 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; } @@ -264,7 +256,7 @@ private void BeginStreamedBody(PendingRequest pending, bool ended) pending.BodyReader = new Http2BodyReader(this, pending.StreamId, ended); _responseWindows[pending.StreamId] = pending.SendWindow; - _ready.Add(pending); + _ready.Enqueue(pending); } private void HandleContinuation(in FrameHeader header, ReadOnlySpan payload) @@ -315,9 +307,9 @@ private bool DecodeHeaderBlock(PendingRequest pending, bool discard = false) _headerScratch = new byte[needed]; } - PendingRequest target = pending; - _decoder.Decode(block, _headerScratch, - discard ? static (_, _) => { } : (name, value) => target.AddHeader(name, value)); + // The request rides as state, so the sink is a static lambda and no closure is made per block. + _decoder.Decode(block, _headerScratch, pending, + discard ? static (_, _, _) => { } : static (target, name, value) => target.AddHeader(name, value)); pending.ClearHeaderBlock(); pending.HeadersDone = true; return true; @@ -416,7 +408,7 @@ private void TryComplete(PendingRequest pending) // Its window moves with it: a WINDOW_UPDATE finds a stream by _streams only until here. _streams.Remove(pending.StreamId); _responseWindows[pending.StreamId] = pending.SendWindow; - _ready.Add(pending); + _ready.Enqueue(pending); } private void HandleSettings(in FrameHeader header, ReadOnlySpan payload) @@ -562,7 +554,7 @@ private void HandleRstStream(in FrameHeader header) { if (_streams.Remove(header.StreamId, out PendingRequest? pending)) { - pending.Abort(); // the peer gave up; there is nobody to answer + Discard(pending); // the peer gave up; there is nobody to answer } // A response in flight may send nothing more on the stream (RFC 9113 5.1), and a writer parked @@ -575,7 +567,8 @@ private void HandleRstStream(in FrameHeader header) /// /// 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; @@ -598,6 +592,9 @@ private sealed class PendingRequest : IDisposable public int BodyLength => _body.Length; + // Grew nothing a pooled request should keep: a field list past this was a peer's thousands of header lines. + public bool IsSmall => _fields.Capacity <= 64; + public (int Offset, int Length) Method; public (int Offset, int Length) Path; public (int Offset, int Length) Scheme; @@ -687,16 +684,14 @@ public void AppendBody(ReadOnlySpan data) public Http2Request Freeze() { - var request = new Http2Request - { - StreamId = StreamId, - Method = Slice(Method), - Path = Slice(Path), - Scheme = Slice(Scheme), - Authority = Slice(Authority), - Body = Slice(_body), - BodyReader = BodyReader, - }; + Http2Request request = _request; + request.StreamId = StreamId; + request.Method = Slice(Method); + request.Path = Slice(Path); + request.Scheme = Slice(Scheme); + request.Authority = Slice(Authority); + request.Body = Slice(_body); + request.BodyReader = BodyReader; foreach ((int nameOffset, int nameLength, int valueOffset, int valueLength) in _fields) { @@ -741,6 +736,17 @@ public void Dispose() _arena = []; } _used = 0; + + // Its request still points into the arena that just went back, which the pool must not pin. + _request.Headers.Clear(); + _request.Method = _request.Path = _request.Scheme = _request.Authority = _request.Body = default; + _request.BodyReader = null; + + StreamId = 0; + HeadersDone = false; + RequestEnded = false; + SendWindow = 65535; + Method = Path = Scheme = Authority = _body = default; } } } diff --git a/src/protocols/ioxide.http2/Http2Connection.Streamed.cs b/src/protocols/ioxide.http2/Http2Connection.Streamed.cs index f77ee723..cf1fc761 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,7 +55,7 @@ public Task RunAsync(Func private void DispatchStreamed(Func handler) { - foreach (PendingRequest pending in TakeReady()) + while (_ready.TryDequeue(out PendingRequest? pending)) { try { @@ -99,8 +100,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..464758a2 100644 --- a/src/protocols/ioxide.http2/Http2Connection.Write.cs +++ b/src/protocols/ioxide.http2/Http2Connection.Write.cs @@ -28,10 +28,15 @@ 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; + + // 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) { @@ -51,56 +56,52 @@ private void Stage(ReadOnlySpan bytes) } } - private async ValueTask FlushAsync() + // 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) { - 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, default); + turn = null; if (_queuedUsed == 0 || _disposed) { - return; + return default; } _pipe.Output.Write(_queued.AsSpan(0, _queuedUsed)); @@ -119,29 +120,32 @@ 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()) + FlushResult result = _writeDead || _disposed ? PeerGone : default; + Wake(faulted, result); + Wake(TakeTurn(), result); + if (_writeDead) { - waiter.TrySetResult(); + ReleaseAllCreditWaiters(); // the fault reached a writer, so no teardown is on its way to them } } - private async Task EndTurnOnReactorAsync() + private async Task EndTurnOnReactorAsync(TurnWaiter? faulted) { try { @@ -151,29 +155,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, FlushResult result) + { + while (turn is not null) { - return []; + TurnWaiter? next = turn.Next; + turn.Next = null; + turn.Enlisted = false; + turn.Release(result); + turn = next; } - - TaskCompletionSource[] turn = [.. _turnWaiters]; - _turnWaiters.Clear(); - return turn; } private void Enqueue(ReadOnlySpan bytes) @@ -238,7 +261,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 ae46bc72..4b6fd719 100644 --- a/src/protocols/ioxide.http2/Http2Connection.cs +++ b/src/protocols/ioxide.http2/Http2Connection.cs @@ -42,7 +42,11 @@ 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(); + 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. @@ -134,10 +138,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(), PeerGone); // Last: a handler parked on its body resumes into its end once everything above is released. FireBodyWakes(); @@ -288,22 +289,9 @@ 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() - { - if (_ready.Count == 0) - { - return []; - } - - PendingRequest[] ready = _ready.ToArray(); - _ready.Clear(); - return ready; - } - private void DispatchBuffered(Func> handler) { - foreach (PendingRequest pending in TakeReady()) + while (_ready.TryDequeue(out PendingRequest? pending)) { ValueTask inFlight; @@ -379,12 +367,47 @@ 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); + Recycle(pending); + } + + // 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(); + if (_pendingPool.Count < MaxPooledRequests && pending.IsSmall && !OffReactor) + { + _pendingPool.Push(pending); + } + } + + // 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) + { + Recycle(pending); + } + else + { + pending.Abort(); + } + } + + // 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..1f73372f 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; @@ -160,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); /// @@ -211,13 +215,34 @@ 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 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); + if (!_headersSent || !onReactor.IsCompleted) + { + return FlushCoreAsync(endStream); + } + + 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) { if (!_headersSent) { @@ -226,13 +251,35 @@ private async ValueTask FlushCore(bool endStream) await new ReactorSwitch(_connection.Reactor); + SendUnsentHeaders(); + return await Send(endStream); + } + + private void SendUnsentHeaders() + { if (_unsent is Http2Response unsent) { _unsent = null; SendHeaders(unsent); } + } + 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) { // Bounded by the frame size AND by both windows: exceeding either is a connection @@ -241,29 +288,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 Progress.Gone; } 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 Progress.Stalled; } int chunk = Math.Min(credit, _staged - sent); @@ -272,14 +301,19 @@ private async ValueTask FlushCore(bool endStream) _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; // 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 +331,47 @@ private async ValueTask FlushCore(bool endStream) } _sinceRealFlush = 0; - await _connection.FlushOutboundAsync(); - return default; + return _connection.FlushOutboundAsync(_turn); + } + + // 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) + { + // 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; + if ((await _connection.FlushOutboundAsync(_turn)).IsCompleted) + { + _staged = 0; // the write side is gone, so no credit is coming either + return PeerGone; + } + + 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 new file mode 100644 index 00000000..ce003412 --- /dev/null +++ b/src/protocols/ioxide.http2/TurnWaiter.cs @@ -0,0 +1,36 @@ +using System.IO.Pipelines; +using System.Threading.Tasks.Sources; + +namespace ioxide.http2; + +/// 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() + { + RunContinuationsAsynchronously = false, + }; + + /// The waiter parked after this one, for the same turn. + internal TurnWaiter? Next; + + /// Parked and not woken yet: a writer whose wait went unawaited waits for that same wake again. + internal bool Enlisted; + + internal ValueTask Arm() + { + _core.Reset(); + return new ValueTask(this, _core.Version); + } + + internal void Release(FlushResult result) => _core.SetResult(result); + + 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..a00dcd2d 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,6 +129,206 @@ 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"); + woke.SetResult(await FlushOutcome(writer)); // queued behind it: its turn is the flush after + }); + + 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.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); + }); + + // 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(); + + 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 }); + 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; + waited[i] = !flush.IsCompleted; + + 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(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"); + } + }); + + // 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(); + + 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]; + + 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) diff --git a/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs b/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs new file mode 100644 index 00000000..70b892ad --- /dev/null +++ b/tests/Ioxide.Tests.Unit/Http2RequestReuseTests.cs @@ -0,0 +1,299 @@ +using System.Collections; +using System.Reflection; +using System.Text; +using ioxide.http2; + +namespace Ioxide.Tests; + +/// +/// A connection hands a retired request, with its header list and field ranges, to the next stream. +/// +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(); + // 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(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.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(); + 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); + + 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 = Pooled(client.Connection); + 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 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(); + + Http2Response response = Http2Response.Text("ok"); // shared, so the handler allocates nothing + 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 grow the ready queue and the pool. + const int perPass = 16; + client.Feed(Batch(1, perPass)); + client.ReleaseFlush(); + client.Feed(Batch(1 + (2 * perPass), perPass)); + client.ReleaseFlush(); + + 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); + + // 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"); + }); + } + + 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(); + foreach (KeyValuePair, ReadOnlyMemory> field in request.Headers.AsSpan()) + { + fields.Add($"{Ascii(field.Key)}={Ascii(field.Value)}"); + } + + 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); + } + + 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; 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, bool split = false, bool endStream = true) + { + var block = new List { post ? (byte)0x83 : (byte)0x82, https ? (byte)0x87 : (byte)0x86 }; // :method, :scheme + 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); + } + + byte end = body is null && endStream ? (byte)0x1 : (byte)0x0; + var frames = new List(); + 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, endStream ? (byte)0x1 : (byte)0x0, 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);