From 107aba2a0be7d3e242d7997318c96f8a1482effb Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Wed, 4 Nov 2026 16:30:36 +0000 Subject: [PATCH 1/4] nghttp2: a connection reuses a finished request's objects for its next stream, so a warm request allocates none of the parser's Every stream got a new PendingRequest, Nghttp2Request, header KeyValueList and array, and a field-range List with its array: about 650 B per request. A request whose body arrived whole has left _pending by the time it is dispatched, so once its handler is done it goes back to the connection with all of them, and the next stream reuses it. A streamed body's request stays in _pending until its stream ends, so it is only disposed, as before. A dispatch swaps two ready lists instead of copying one. --- .../Connection/Nghttp2Connection.Callbacks.cs | 2 +- .../Nghttp2Connection.RunBuffered.cs | 35 ++-- .../Connection/Nghttp2Connection.Streamed.cs | 17 +- .../Connection/Nghttp2Connection.cs | 63 ++++++-- .../ioxide.nghttp2/Http/Nghttp2Request.cs | 3 +- .../Nghttp2RequestReuseTests.cs | 150 ++++++++++++++++++ tests/Ioxide.Tests.Unit/Program.cs | 1 + 7 files changed, 239 insertions(+), 32 deletions(-) create mode 100644 tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs diff --git a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Callbacks.cs b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Callbacks.cs index 9b6ff27b..4f2eb057 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Callbacks.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Callbacks.cs @@ -47,7 +47,7 @@ private static unsafe void CallbackBeginHeaders(void* user, int streamId) // the request we have rather than replacing it and losing everything accumulated. if (!connection._pending.ContainsKey(streamId)) { - connection._pending[streamId] = new PendingRequest { StreamId = streamId }; + connection._pending[streamId] = connection.RentPending(streamId); } } catch (Exception e) diff --git a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs index ef01d3d3..a7cc8c38 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs @@ -126,18 +126,12 @@ private unsafe void Feed(in ReadOnlySequence buffer) } } - // Snapshot and clear first: a handler that awaits can let another read pass land more - // requests here, and 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, and the next + // read pass lands its requests in the other one. The caller clears it once done. + private List TakeReady() { - if (_readyThisPass.Count == 0) - { - return []; - } - - PendingRequest[] ready = _readyThisPass.ToArray(); - _readyThisPass.Clear(); - return ready; + (_readyThisPass, _dispatching) = (_dispatching, _readyThisPass); + return _dispatching; } // A request that outgrew MaxRequestBytes is reset instead of dispatched, whichever the response mode. @@ -160,7 +154,20 @@ private bool RefuseOverflowed(PendingRequest pending) private void DispatchBuffered(Func> handler) { - foreach (PendingRequest pending in TakeReady()) + List ready = TakeReady(); + try + { + DispatchBuffered(handler, ready); + } + finally + { + ready.Clear(); + } + } + + private void DispatchBuffered(Func> handler, List ready) + { + foreach (PendingRequest pending in ready) { if (RefuseOverflowed(pending)) { @@ -194,7 +201,7 @@ private void DispatchBuffered(Func> h { // The arena backs the request's memories, so it can only go back once the // handler has returned and the response is submitted (which copies). - pending.Dispose(); + Retire(pending); } continue; } @@ -234,7 +241,7 @@ private async Task CompleteAsync(ValueTask inFlight, PendingReq } finally { - pending.Dispose(); + Retire(pending); await FlushEgressAsync(); } } diff --git a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs index 48a76f23..57ffe1d4 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs @@ -42,7 +42,20 @@ public Task RunAsync(Func private void DispatchStreamed(Func handler) { - foreach (PendingRequest pending in TakeReady()) + List ready = TakeReady(); + try + { + DispatchStreamed(handler, ready); + } + finally + { + ready.Clear(); + } + } + + private void DispatchStreamed(Func handler, List ready) + { + foreach (PendingRequest pending in ready) { if (RefuseOverflowed(pending)) { @@ -92,7 +105,7 @@ private async Task ServeStreamedAsync(Func _pending = new(); // Requests whose streams ended during the current ih2_read. The callbacks only record them; - // the loop dispatches after mem_recv returns. - private readonly List _readyThisPass = []; + // the loop dispatches after mem_recv returns, walking the other list: the two swap each pass. + private List _readyThisPass = []; + private List _dispatching = []; + + // Retired requests with a buffered body, each keeping its Nghttp2Request, header list and field + // ranges for the next stream (#489). + private readonly Stack _pendingPool = new(); + + private PendingRequest RentPending(int streamId) + { + PendingRequest pending = _pendingPool.TryPop(out PendingRequest? pooled) ? pooled : new PendingRequest(); + pending.StreamId = streamId; + return pending; + } + + /// + /// Done with a dispatched request, once its handler has returned. A buffered body's request left + /// _pending before it was dispatched, so nothing can reach it any more and it goes back for the + /// next stream. A streamed body's stays in _pending until its stream ends, so it is only disposed. + /// + private void Retire(PendingRequest pending) + { + pending.Dispose(); + if (pending.BodyReader is null) + { + pending.Reset(); + _pendingPool.Push(pending); + } + } /// /// Serve over an already-chosen transport. The pipe is the caller's to dispose - it may be a @@ -163,6 +190,7 @@ private sealed class PendingRequest : IDisposable private int _used; private readonly List<(int NameOffset, int NameLength, int ValueOffset, int ValueLength)> _fields = []; private (int Offset, int Length) _body = (0, 0); + private readonly Nghttp2Request _request = new(); public int StreamId; @@ -228,20 +256,19 @@ public void AppendBody(ReadOnlySpan data) } } - /// Materialize the public request. Valid until . + /// Materialize the public request, this one's own and reused. Valid until . public Nghttp2Request Freeze() { - var request = new Nghttp2Request - { - StreamId = StreamId, - Method = Slice(Method), - Path = Slice(Path), - Scheme = Slice(Scheme), - Authority = Slice(Authority), - Body = Slice(_body), - BodyReader = BodyReader, - }; - + Nghttp2Request 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), @@ -266,5 +293,13 @@ public void Dispose() } _used = 0; } + + /// Fresh for the next stream that rents it, after . + public void Reset() + { + StreamId = 0; + Overflowed = false; + Method = Path = Scheme = Authority = _body = default; + } } } diff --git a/src/protocols/ioxide.nghttp2/Http/Nghttp2Request.cs b/src/protocols/ioxide.nghttp2/Http/Nghttp2Request.cs index 04758551..a28fb83d 100644 --- a/src/protocols/ioxide.nghttp2/Http/Nghttp2Request.cs +++ b/src/protocols/ioxide.nghttp2/Http/Nghttp2Request.cs @@ -6,7 +6,8 @@ namespace ioxide.nghttp2; /// 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: a request whose body arrived whole is itself reused +/// for a later stream on the same connection once its handler is done. /// public sealed class Nghttp2Request { diff --git a/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs b/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs new file mode 100644 index 00000000..6286eea5 --- /dev/null +++ b/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs @@ -0,0 +1,150 @@ +using System.Text; +using ioxide.nghttp2; + +namespace Ioxide.Tests; + +/// +/// ioxide.nghttp2 keeps a retired request whose body arrived whole, 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. Driven over the strict test transport, like the managed stack. +/// +internal static class Nghttp2RequestReuseTests +{ + public static void Register(Runner runner) + { + runner.Test("nghttp2 request: the next stream's request holds nothing of the one before", () => + { + using var client = new StrictClient(); + var connection = new Nghttp2Connection(client); + + var seen = new List<(Nghttp2Request Request, string Fields)>(); + Task run = connection.RunAsync(request => + { + seen.Add((request, Describe(request))); + return Nghttp2Response.Text("ok"); + }); + + client.SendRequests(); // preface and SETTINGS + client.Drain(); + client.Feed(Request(1, post: true, "/first", "a.example", [("x-one", "1"), ("x-two", "2")], body: "hello")); + client.Drain(); + client.Feed(Request(3, post: false, "/", "b.example", [], body: null)); // nghttp2 refuses a request with no authority + client.Drain(); + Close(client, connection, 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 / b.example [] -", seen[1].Fields); + Assert.True(ReferenceEquals(seen[0].Request, seen[1].Request), "the second stream reuses the first one's request"); + }); + + runner.Test("nghttp2 request: a warm connection's request allocates none of the parser's objects", () => + { + using var client = new StrictClient(); + var connection = new Nghttp2Connection(client); + + Nghttp2Response response = Nghttp2Response.Text("ok"); // shared, so the handler allocates nothing + Task run = connection.RunAsync(_ => response); + + client.SendRequests(); // preface and SETTINGS + client.Drain(); + + const int perPass = 16; + client.Feed(Batch(1, perPass)); // warm-up: the requests the next batch reuses + client.Drain(); + + 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.Drain(); + Close(client, connection, run); + + // Before reuse, every request allocated about 650 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 void Close(StrictClient client, Nghttp2Connection connection, Task run) + { + client.Close(run); + connection.Dispose(); + } + + private static string Describe(Nghttp2Request 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 an authority and 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, "/", "localhost", [("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); + } + 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..90a2cf90 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); + Nghttp2RequestReuseTests.Register(runner); Http2StreamedBodyTests.Register(runner); Http2FlowControlTests.Register(runner); Http2BufferedCreditTests.Register(runner); From 87f5da26795b5410639b8539199e83647071a6c7 Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Wed, 4 Nov 2026 19:11:58 +0000 Subject: [PATCH 2/4] nghttp2: the request pool is capped, keeps nothing large and lets go of the arena; tests cover every retire path From review: - A retired request still pointed into the arena Dispose had just returned to ArrayPool, so a pooled entry could pin it (with a body, up to 16 MiB). Reset now lets go of the request's memories and headers. - The pool keeps at most 32, never one whose field list grew past 64 (a peer's thousands of header lines; nghttp2 does not limit their number), and only on the reactor, the one thread that rents. - Tests: the reuse test sends a host instead of :authority, so a stale authority would show; a parked handler's request goes back once it answers (CompleteAsync); a request whose body is still arriving is never handed on (fails with the BodyReader guard removed: stream 3 is never dispatched); the warm-request test warms both ready lists and counts its handlers. --- .../Connection/Nghttp2Connection.cs | 18 ++- .../Nghttp2RequestReuseTests.cs | 122 +++++++++++++++--- 2 files changed, 116 insertions(+), 24 deletions(-) diff --git a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs index 313f7aa4..6c61425d 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs @@ -75,8 +75,10 @@ public sealed partial class Nghttp2Connection : IDisposable private List _dispatching = []; // Retired requests with a buffered body, each keeping its Nghttp2Request, header list and field - // ranges for the next stream (#489). + // ranges for the next stream (#489). 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; private PendingRequest RentPending(int streamId) { @@ -93,7 +95,10 @@ private PendingRequest RentPending(int streamId) private void Retire(PendingRequest pending) { pending.Dispose(); - if (pending.BodyReader is null) + + // Only on the reactor, the one thread that rents, and never one whose field list a peer's thousands of + // header lines grew: it would stay that size on the connection. + if (pending.BodyReader is null && _pendingPool.Count < MaxPooledRequests && !OffReactor && pending.IsSmall) { pending.Reset(); _pendingPool.Push(pending); @@ -268,7 +273,6 @@ public Nghttp2Request 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), @@ -300,6 +304,14 @@ public void Reset() StreamId = 0; Overflowed = false; Method = Path = Scheme = Authority = _body = default; + + // Its request still points into the arena Dispose returned: 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; } + + public bool IsSmall => _fields.Capacity <= 64; } } diff --git a/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs b/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs index 6286eea5..23deb0e4 100644 --- a/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs +++ b/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs @@ -5,11 +5,14 @@ namespace Ioxide.Tests; /// /// ioxide.nghttp2 keeps a retired request whose body arrived whole, 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. Driven over the strict test transport, like the managed stack. +/// the next stream: the next request must come back holding nothing of the one before, a request still reading its +/// streamed body must never be handed on, and a warm connection's request allocates none of the parser's objects. +/// Driven over the strict test transport, like the managed stack. /// internal static class Nghttp2RequestReuseTests { + private static readonly Nghttp2Response Ok = Nghttp2Response.Text("ok"); // shared, so handlers allocate nothing + public static void Register(Runner runner) { runner.Test("nghttp2 request: the next stream's request holds nothing of the one before", () => @@ -21,42 +24,109 @@ public static void Register(Runner runner) Task run = connection.RunAsync(request => { seen.Add((request, Describe(request))); - return Nghttp2Response.Text("ok"); + return Ok; }); - client.SendRequests(); // preface and SETTINGS - client.Drain(); + Open(client); client.Feed(Request(1, post: true, "/first", "a.example", [("x-one", "1"), ("x-two", "2")], body: "hello")); client.Drain(); - client.Feed(Request(3, post: false, "/", "b.example", [], body: null)); // nghttp2 refuses a request with no authority + // nghttp2 wants an authority or a host; a host alone leaves :authority unset, so a stale one would show. + client.Feed(Request(3, post: false, "/", authority: null, [("host", "b.example")], body: null)); client.Drain(); Close(client, connection, 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 / b.example [] -", seen[1].Fields); + Assert.Equal("3 GET / - [host=b.example] -", seen[1].Fields); Assert.True(ReferenceEquals(seen[0].Request, seen[1].Request), "the second stream reuses the first one's request"); }); - runner.Test("nghttp2 request: a warm connection's request allocates none of the parser's objects", () => + runner.Test("nghttp2 request: a parked handler's request goes back once it answers, holding nothing", () => { using var client = new StrictClient(); var connection = new Nghttp2Connection(client); - Nghttp2Response response = Nghttp2Response.Text("ok"); // shared, so the handler allocates nothing - Task run = connection.RunAsync(_ => response); + var parked = new TaskCompletionSource(); + var seen = new Dictionary(); + Task run = connection.RunAsync(request => + { + seen[request.StreamId] = (request, Describe(request)); + return request.Path.Span.SequenceEqual("/park"u8) + ? new ValueTask(parked.Task) // answered later, through CompleteAsync + : new ValueTask(Ok); + }); + + Open(client); + client.Feed(Request(1, post: true, "/park", "a.example", [("x-one", "1")], body: "hello")); + client.Drain(); + client.Feed(Request(3, post: false, "/quick", "a.example", [], body: null)); + client.Drain(); + parked.SetResult(Ok); // stream 1 answers now, and its request is the last to go back + client.Drain(); + client.Feed(Request(5, post: false, "/", authority: null, [("host", "c.example")], body: null)); + client.Drain(); + Close(client, connection, run); + + Assert.Equal("5 GET / - [host=c.example] -", seen[5].Fields); + Assert.True(ReferenceEquals(seen[1].Request, seen[5].Request), "the parked handler's request was reused"); + }); + + // A handler that takes a body reader is dispatched at the headers, and its request stays in _pending until + // the stream ends, because the body's DATA and trailers still find it there. Handed to the next stream, that + // stream's headers would land on it as trailers and its request would never be dispatched. + runner.Test("nghttp2 request: a request whose body is still arriving is not handed to the next stream", () => + { + using var client = new StrictClient(); + var connection = new Nghttp2Connection(client); + + var seen = new List<(Nghttp2Request Request, string Fields)>(); + Task run = connection.RunAsync((request, body) => + { + seen.Add((request, Describe(request))); + return new ValueTask(Ok); // done before its body ends + }); - client.SendRequests(); // preface and SETTINGS + Open(client); + client.Feed(Request(1, post: true, "/upload", "a.example", [("x-one", "1")], body: null, endStream: false)); + client.Drain(); + client.Feed(Request(3, post: false, "/", "b.example", [], body: null)); + client.Drain(); + client.Feed([.. FrameHeader(5, 0x0, 0x1, 1), .. "hello"u8.ToArray()]); // stream 1's body, and its end client.Drain(); + Close(client, connection, run); + + Assert.Equal(2, seen.Count); + Assert.Equal("3 GET / b.example [] -", seen[1].Fields); + Assert.True(!ReferenceEquals(seen[0].Request, seen[1].Request), "a request still in _pending was handed on"); + }); + runner.Test("nghttp2 request: a warm connection's request allocates none of the parser's objects", () => + { + using var client = new StrictClient(); + var connection = new Nghttp2Connection(client); + + int served = 0; + Task run = connection.RunAsync(_ => + { + served++; + return Ok; + }); + + Open(client); + + // 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.Drain(); + client.Feed(Batch(1 + (2 * perPass), perPass)); client.Drain(); - 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.Drain(); Close(client, connection, run); @@ -67,6 +137,12 @@ public static void Register(Runner runner) }); } + private static void Open(StrictClient client) + { + client.SendRequests(); // preface and SETTINGS + client.Drain(); + } + private static void Close(StrictClient client, Nghttp2Connection connection, Task run) { client.Close(run); @@ -100,10 +176,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. - private static byte[] Request(int streamId, bool post, string path, string authority, - (string Name, string Value)[] fields, string? body) + // HEADERS, then a DATA frame with the body if there is one; the last frame carries END_STREAM unless told not + // to. 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 endStream = true) { var block = new List { post ? (byte)0x83 : (byte)0x82, 0x86 }; // :method, :scheme http if (path == "/") @@ -115,8 +191,11 @@ private static byte[] Request(int streamId, bool post, string path, string autho block.Add(0x04); // :path, literal value AddString(block, path); } - block.Add(0x01); // :authority, literal value - AddString(block, authority); + 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 @@ -124,12 +203,13 @@ private static byte[] Request(int streamId, bool post, string path, string autho AddString(block, value); } + bool headersEnd = endStream && body is null; var frames = new List(); - frames.AddRange(FrameHeader(block.Count, 0x1, body is null ? (byte)0x5 : (byte)0x4, streamId)); + frames.AddRange(FrameHeader(block.Count, 0x1, headersEnd ? (byte)0x5 : (byte)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 55b6f55657fafc24eb33ec2a5f4767434526520a Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Thu, 5 Nov 2026 00:24:34 +0000 Subject: [PATCH 3/4] nghttp2: a dispatch walks the ready list in place, and the tests cover the streamed path, the size check and the cap From review: the second ready list guarded against something that cannot happen. Only nghttp2's callbacks add to it, and they run inside ih2_read, which never runs during the walk. The walk now goes over _readyThisPass by index and clears it after; _dispatching and TakeReady are gone. Three tests close the gaps that let a retire path or a bound regress unseen: - a streamed response's request goes back once its handler is done (ServeStreamedAsync); - a request with 65 header lines is not kept, and a small one after it is; - 40 parked handlers leave 32 requests in the pool. Each fails against a mutation of the code it guards. Comments are cut to one line. --- .../Nghttp2Connection.RunBuffered.cs | 17 +--- .../Connection/Nghttp2Connection.Streamed.cs | 8 +- .../Connection/Nghttp2Connection.cs | 21 ++--- .../Nghttp2RequestReuseTests.cs | 93 +++++++++++++++++-- 4 files changed, 99 insertions(+), 40 deletions(-) diff --git a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs index a7cc8c38..47770b4f 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs @@ -126,14 +126,6 @@ private unsafe void Feed(in ReadOnlySequence buffer) } } - // Taken whole before the walk: the list must not be mutated while it is walked, and the next - // read pass lands its requests in the other one. The caller clears it once done. - private List TakeReady() - { - (_readyThisPass, _dispatching) = (_dispatching, _readyThisPass); - return _dispatching; - } - // A request that outgrew MaxRequestBytes is reset instead of dispatched, whichever the response mode. private bool RefuseOverflowed(PendingRequest pending) { @@ -154,21 +146,22 @@ private bool RefuseOverflowed(PendingRequest pending) private void DispatchBuffered(Func> handler) { - List ready = TakeReady(); try { - DispatchBuffered(handler, ready); + DispatchBuffered(handler, _readyThisPass); } finally { - ready.Clear(); + _readyThisPass.Clear(); } } + // Callbacks only run inside ih2_read, which never runs during this walk. private void DispatchBuffered(Func> handler, List ready) { - foreach (PendingRequest pending in ready) + for (int i = 0; i < ready.Count; i++) { + PendingRequest pending = ready[i]; if (RefuseOverflowed(pending)) { continue; diff --git a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs index 57ffe1d4..012b7921 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs @@ -42,21 +42,21 @@ public Task RunAsync(Func private void DispatchStreamed(Func handler) { - List ready = TakeReady(); try { - DispatchStreamed(handler, ready); + DispatchStreamed(handler, _readyThisPass); } finally { - ready.Clear(); + _readyThisPass.Clear(); } } private void DispatchStreamed(Func handler, List ready) { - foreach (PendingRequest pending in ready) + for (int i = 0; i < ready.Count; i++) { + PendingRequest pending = ready[i]; if (RefuseOverflowed(pending)) { continue; diff --git a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs index 6c61425d..2e15f280 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs @@ -70,13 +70,10 @@ public sealed partial class Nghttp2Connection : IDisposable private readonly Dictionary _pending = new(); // Requests whose streams ended during the current ih2_read. The callbacks only record them; - // the loop dispatches after mem_recv returns, walking the other list: the two swap each pass. - private List _readyThisPass = []; - private List _dispatching = []; + // the loop dispatches after mem_recv returns. + private readonly List _readyThisPass = []; - // Retired requests with a buffered body, each keeping its Nghttp2Request, header list and field - // ranges for the next stream (#489). Capped, so a burst of streams leaves no more than this on an idle - // connection. + // Retired buffered 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; @@ -87,17 +84,12 @@ private PendingRequest RentPending(int streamId) return pending; } - /// - /// Done with a dispatched request, once its handler has returned. A buffered body's request left - /// _pending before it was dispatched, so nothing can reach it any more and it goes back for the - /// next stream. A streamed body's stays in _pending until its stream ends, so it is only disposed. - /// + /// Done with a dispatched request once its handler has returned; a streamed body's stays in _pending. private void Retire(PendingRequest pending) { pending.Dispose(); - // Only on the reactor, the one thread that rents, and never one whose field list a peer's thousands of - // header lines grew: it would stay that size on the connection. + // Only the reactor rents, and a field list a peer's header lines grew would stay that size. if (pending.BodyReader is null && _pendingPool.Count < MaxPooledRequests && !OffReactor && pending.IsSmall) { pending.Reset(); @@ -305,8 +297,7 @@ public void Reset() Overflowed = false; Method = Path = Scheme = Authority = _body = default; - // Its request still points into the arena Dispose returned: 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 Dispose returned, which the pool must not pin. _request.Headers.Clear(); _request.Method = _request.Path = _request.Scheme = _request.Authority = _request.Body = default; _request.BodyReader = null; diff --git a/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs b/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs index 23deb0e4..bea6a66a 100644 --- a/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs +++ b/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs @@ -1,13 +1,12 @@ +using System.Collections; +using System.Reflection; using System.Text; using ioxide.nghttp2; namespace Ioxide.Tests; /// -/// ioxide.nghttp2 keeps a retired request whose body arrived whole, with its header list and field ranges, for -/// the next stream: the next request must come back holding nothing of the one before, a request still reading its -/// streamed body must never be handed on, and a warm connection's request allocates none of the parser's objects. -/// Driven over the strict test transport, like the managed stack. +/// ioxide.nghttp2 hands a retired request whose body arrived whole to the next stream on the connection. /// internal static class Nghttp2RequestReuseTests { @@ -71,9 +70,86 @@ public static void Register(Runner runner) Assert.True(ReferenceEquals(seen[1].Request, seen[5].Request), "the parked handler's request was reused"); }); - // A handler that takes a body reader is dispatched at the headers, and its request stays in _pending until - // the stream ends, because the body's DATA and trailers still find it there. Handed to the next stream, that - // stream's headers would land on it as trailers and its request would never be dispatched. + runner.Test("nghttp2 request: a streamed response's request goes back once its handler is done", () => + { + using var client = new StrictClient(); + var connection = new Nghttp2Connection(client); + + var seen = new List<(Nghttp2Request Request, string Fields)>(); + Task run = connection.RunAsync((request, writer) => + { + seen.Add((request, Describe(request))); + return ValueTask.CompletedTask; // the connection sends a 200 with no body + }); + + Open(client); + client.Feed(Request(1, post: false, "/first", "a.example", [("x-one", "1")], body: null)); + client.Drain(); + client.Feed(Request(3, post: false, "/", authority: null, [("host", "b.example")], body: null)); + client.Drain(); + Close(client, connection, run); + + Assert.Equal("3 GET / - [host=b.example] -", seen[1].Fields); + Assert.True(ReferenceEquals(seen[0].Request, seen[1].Request), "the second stream reuses the first one's request"); + }); + + runner.Test("nghttp2 request: a request a peer sent 65 header lines is not kept for the next stream", () => + { + using var client = new StrictClient(); + var connection = new Nghttp2Connection(client); + + var seen = new List(); + Task run = connection.RunAsync(request => + { + seen.Add(request); + return Ok; + }); + + Open(client); + (string, string)[] many = [.. Enumerable.Range(0, 65).Select(i => ($"x-{i}", "v"))]; + client.Feed(Request(1, post: false, "/", "a.example", many, body: null)); + client.Drain(); + client.Feed(Request(3, post: false, "/", "a.example", [], body: null)); + client.Drain(); + client.Feed(Request(5, post: false, "/", "a.example", [], body: null)); + client.Drain(); + Close(client, connection, run); + + Assert.Equal(3, seen.Count); + Assert.True(!ReferenceEquals(seen[0], seen[1]), "a request whose field list grew past 64 entries was kept"); + Assert.True(ReferenceEquals(seen[1], seen[2]), "control: a small request is kept"); + }); + + runner.Test("nghttp2 request: a connection keeps at most 32 requests for later streams", () => + { + using var client = new StrictClient(); + var connection = new Nghttp2Connection(client); + + var parked = new List>(); // one each: a shared one resumes off-thread + Task run = connection.RunAsync(_ => + { + var answer = new TaskCompletionSource(); + parked.Add(answer); + return new ValueTask(answer.Task); + }); + + Open(client); + client.Feed(Batch(1, 40)); + client.Drain(); + foreach (TaskCompletionSource answer in parked) + { + answer.SetResult(Ok); // each retires inline, on this thread + } + client.Drain(); + int pooled = ((ICollection)typeof(Nghttp2Connection) + .GetField("_pendingPool", BindingFlags.NonPublic | BindingFlags.Instance)!.GetValue(connection)!).Count; + Close(client, connection, run); + + Assert.Equal(40, parked.Count); + Assert.Equal(32, pooled); + }); + + // A body reader's request stays in _pending until its stream ends; handed on, the next stream's headers land on it. runner.Test("nghttp2 request: a request whose body is still arriving is not handed to the next stream", () => { using var client = new StrictClient(); @@ -131,8 +207,7 @@ public static void Register(Runner runner) client.Drain(); Close(client, connection, run); - // Before reuse, every request allocated about 650 B of parser objects. What is left is the test - // transport's own flush, shared by the pass. + // 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"); }); } From df9298ff353d6bf295a1f825d04b57200ceb7276 Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Thu, 5 Nov 2026 20:36:10 +0000 Subject: [PATCH 4/4] nghttp2: the ready list is a queue the dispatch drains, and a test sees a retired request let go of its arena No wrapper and no claim about what can add to the list during a walk: a drain's on_stream_close runs inside it, and a queue doesn't care. --- .../Connection/Nghttp2Connection.Callbacks.cs | 6 +++--- .../Connection/Nghttp2Connection.RunBuffered.cs | 16 +--------------- .../Connection/Nghttp2Connection.Streamed.cs | 15 +-------------- .../Connection/Nghttp2Connection.cs | 2 +- .../Nghttp2RequestReuseTests.cs | 8 +++++--- 5 files changed, 11 insertions(+), 36 deletions(-) diff --git a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Callbacks.cs b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Callbacks.cs index 4f2eb057..ffe98d79 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Callbacks.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Callbacks.cs @@ -108,7 +108,7 @@ private static unsafe void CallbackEndHeaders(void* user, int streamId) connection._pending.TryGetValue(streamId, out PendingRequest? pending) && pending.BodyReader is null) { pending.BodyReader = new Nghttp2BodyReader(connection, streamId); - connection._readyThisPass.Add(pending); + connection._readyThisPass.Enqueue(pending); } } catch (Exception e) @@ -141,7 +141,7 @@ private static unsafe void CallbackData(void* user, int streamId, byte* data, nu { // Left pending, so a trailer later in this read cannot start a fresh request on the stream. pending.Overflowed = true; - connection._readyThisPass.Add(pending); + connection._readyThisPass.Enqueue(pending); return; } pending.AppendBody(new ReadOnlySpan(data, (int)dataLength)); @@ -170,7 +170,7 @@ private static unsafe void CallbackEndStream(void* user, int streamId) } else if (!pending.Overflowed) { - connection._readyThisPass.Add(pending); + connection._readyThisPass.Enqueue(pending); } } } diff --git a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs index 47770b4f..403abf10 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs @@ -146,22 +146,8 @@ private bool RefuseOverflowed(PendingRequest pending) private void DispatchBuffered(Func> handler) { - try - { - DispatchBuffered(handler, _readyThisPass); - } - finally - { - _readyThisPass.Clear(); - } - } - - // Callbacks only run inside ih2_read, which never runs during this walk. - private void DispatchBuffered(Func> handler, List ready) - { - for (int i = 0; i < ready.Count; i++) + while (_readyThisPass.TryDequeue(out PendingRequest? pending)) { - PendingRequest pending = ready[i]; if (RefuseOverflowed(pending)) { continue; diff --git a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs index 012b7921..3c290192 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs @@ -42,21 +42,8 @@ public Task RunAsync(Func private void DispatchStreamed(Func handler) { - try - { - DispatchStreamed(handler, _readyThisPass); - } - finally - { - _readyThisPass.Clear(); - } - } - - private void DispatchStreamed(Func handler, List ready) - { - for (int i = 0; i < ready.Count; i++) + while (_readyThisPass.TryDequeue(out PendingRequest? pending)) { - PendingRequest pending = ready[i]; if (RefuseOverflowed(pending)) { continue; diff --git a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs index 2e15f280..c38998c9 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs @@ -71,7 +71,7 @@ public sealed partial class Nghttp2Connection : IDisposable // Requests whose streams ended during the current ih2_read. The callbacks only record them; // the loop dispatches after mem_recv returns. - private readonly List _readyThisPass = []; + private readonly Queue _readyThisPass = new(); // Retired buffered requests, kept for the next stream; capped, so a burst leaves little on an idle connection. private readonly Stack _pendingPool = new(); diff --git a/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs b/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs index bea6a66a..61b8462f 100644 --- a/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs +++ b/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs @@ -62,6 +62,9 @@ public static void Register(Runner runner) client.Drain(); parked.SetResult(Ok); // stream 1 answers now, and its request is the last to go back client.Drain(); + Nghttp2Request retired = seen[1].Request; // in the pool now, which must not pin the arena it read from + Assert.True(retired.Method.IsEmpty && retired.Path.IsEmpty && retired.Body.IsEmpty && retired.Headers.AsSpan().IsEmpty, + "a retired request still points into the arena that went back"); client.Feed(Request(5, post: false, "/", authority: null, [("host", "c.example")], body: null)); client.Drain(); Close(client, connection, run); @@ -190,7 +193,7 @@ public static void Register(Runner runner) Open(client); - // 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 queue and the pool. const int perPass = 16; client.Feed(Batch(1, perPass)); client.Drain(); @@ -251,8 +254,7 @@ private static byte[] Batch(int firstStream, int count) return bytes.ToArray(); } - // HEADERS, then a DATA frame with the body if there is one; the last frame carries END_STREAM unless told not - // to. HPACK from the static table and literals without indexing only, so the request needs no encoder of its own. + // HEADERS, then DATA with the body if any; the last carries END_STREAM unless told not to. Static table and literals only. private static byte[] Request(int streamId, bool post, string path, string? authority, (string Name, string Value)[] fields, string? body, bool endStream = true) {