diff --git a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Callbacks.cs b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Callbacks.cs index 9b6ff27b..ffe98d79 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) @@ -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 ef01d3d3..403abf10 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.RunBuffered.cs @@ -126,20 +126,6 @@ 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() - { - if (_readyThisPass.Count == 0) - { - return []; - } - - PendingRequest[] ready = _readyThisPass.ToArray(); - _readyThisPass.Clear(); - return ready; - } - // A request that outgrew MaxRequestBytes is reset instead of dispatched, whichever the response mode. private bool RefuseOverflowed(PendingRequest pending) { @@ -160,7 +146,7 @@ private bool RefuseOverflowed(PendingRequest pending) private void DispatchBuffered(Func> handler) { - foreach (PendingRequest pending in TakeReady()) + while (_readyThisPass.TryDequeue(out PendingRequest? pending)) { if (RefuseOverflowed(pending)) { @@ -194,7 +180,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 +220,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 a4ffd753..b3c501ac 100644 --- a/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs +++ b/src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs @@ -42,7 +42,7 @@ public Task RunAsync(Func private void DispatchStreamed(Func handler) { - foreach (PendingRequest pending in TakeReady()) + while (_readyThisPass.TryDequeue(out PendingRequest? pending)) { if (RefuseOverflowed(pending)) { @@ -92,7 +92,7 @@ private async Task ServeStreamedAsync(Func _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(); + private const int MaxPooledRequests = 32; + + 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 streamed body's stays in _pending. + private void Retire(PendingRequest pending) + { + pending.Dispose(); + + // 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(); + _pendingPool.Push(pending); + } + } /// /// Serve over an already-chosen transport. The pipe is the caller's to dispose - it may be a @@ -164,6 +188,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; @@ -229,19 +254,17 @@ 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; foreach ((int nameOffset, int nameLength, int valueOffset, int valueLength) in _fields) { @@ -267,5 +290,20 @@ 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; + + // 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; + } + + public bool IsSmall => _fields.Capacity <= 64; } } 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..61b8462f --- /dev/null +++ b/tests/Ioxide.Tests.Unit/Nghttp2RequestReuseTests.cs @@ -0,0 +1,307 @@ +using System.Collections; +using System.Reflection; +using System.Text; +using ioxide.nghttp2; + +namespace Ioxide.Tests; + +/// +/// ioxide.nghttp2 hands a retired request whose body arrived whole to the next stream on the connection. +/// +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", () => + { + 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 Ok; + }); + + Open(client); + client.Feed(Request(1, post: true, "/first", "a.example", [("x-one", "1"), ("x-two", "2")], body: "hello")); + client.Drain(); + // 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 / - [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 parked handler's request goes back once it answers, holding nothing", () => + { + using var client = new StrictClient(); + var connection = new Nghttp2Connection(client); + + 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(); + 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); + + 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"); + }); + + 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(); + 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 + }); + + 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 grow the ready queue and the pool. + const int perPass = 16; + client.Feed(Batch(1, perPass)); + client.Drain(); + client.Feed(Batch(1 + (2 * perPass), perPass)); + client.Drain(); + + 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); + + // 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 Open(StrictClient client) + { + client.SendRequests(); // preface and SETTINGS + client.Drain(); + } + + 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, 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) + { + 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); + } + + bool headersEnd = endStream && body is null; + var frames = new List(); + 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, 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 757af8f1..47faae5f 100644 --- a/tests/Ioxide.Tests.Unit/Program.cs +++ b/tests/Ioxide.Tests.Unit/Program.cs @@ -31,6 +31,7 @@ private static int Main() Http2PseudoHeaderTests.Register(runner); Http2RequestHeaderTests.Register(runner); Http2RequestReuseTests.Register(runner); + Nghttp2RequestReuseTests.Register(runner); Http2StreamedBodyTests.Register(runner); Http2FlowControlTests.Register(runner); Http2BufferedCreditTests.Register(runner);