diff --git a/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.SendStream.cs b/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.SendStream.cs index 4b377a19..5e4797b3 100644 --- a/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.SendStream.cs +++ b/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.SendStream.cs @@ -23,9 +23,20 @@ private sealed class OutStream public bool FinSent; public bool Dead; // reset/refused: stop feeding; chunks freed at stream close public bool Pending; // in the replay list + + public void Reset() + { + Base = Sent = End = 0; + Fin = FinSent = Dead = Pending = false; + } } private readonly Dictionary _outStreams = new(); + + // Closed streams' send state, for the next stream that sends; capped, so a burst leaves little on an idle connection. + private readonly Stack _outStreamPool = new(); + private const int MaxPooledStreams = 32; + private const int MaxPooledChunks = 64; // slots a pooled queue keeps private readonly List _outPending = []; private long _outRetained; @@ -76,7 +87,7 @@ public override void SendStream(long streamId, ReadOnlySpan data, bool fin if (!_outStreams.TryGetValue(streamId, out OutStream? os)) { - os = new OutStream(); + os = _outStreamPool.TryPop(out OutStream? pooled) ? pooled : new OutStream(); _outStreams[streamId] = os; } if (os.Dead || (os.Fin && !fin && data.Length > 0)) @@ -354,5 +365,16 @@ private void PurgeOutStream(long sid) NativeMemory.Free((void*)p); _outRetained -= l; } + + if (_outStreamPool.Count >= MaxPooledStreams) + { + return; + } + os.Reset(); // SendStream and ReplayOut can still hold it after PumpOut's purge; reset, it has nothing to send + if (os.Chunks.Capacity > MaxPooledChunks) + { + os.Chunks.TrimExcess(MaxPooledChunks); + } + _outStreamPool.Push(os); } } diff --git a/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.cs b/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.cs index f9293919..0fc2257e 100644 --- a/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.cs +++ b/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.cs @@ -449,6 +449,7 @@ private void Destroy() } } _outStreams.Clear(); + _outStreamPool.Clear(); _outPending.Clear(); if (_conn != 0) { diff --git a/tests/Ioxide.Tests.E2E/Program.cs b/tests/Ioxide.Tests.E2E/Program.cs index e0c353db..77fe72cd 100644 --- a/tests/Ioxide.Tests.E2E/Program.cs +++ b/tests/Ioxide.Tests.E2E/Program.cs @@ -47,6 +47,7 @@ private static int Main() QuicUnansweredStreamTests.Register(runner); QuicDeferredFaultTests.Register(runner); QuicStreamAllowanceTests.Register(runner); + QuicSendStreamTests.Register(runner); QuicTimerTests.Register(runner); QuicSniHostileTests.Register(runner); QuicDemuxRoutingTests.Register(runner); diff --git a/tests/Ioxide.Tests.E2E/Protocols/QuicSendStreamTests.cs b/tests/Ioxide.Tests.E2E/Protocols/QuicSendStreamTests.cs new file mode 100644 index 00000000..55b1f7de --- /dev/null +++ b/tests/Ioxide.Tests.E2E/Protocols/QuicSendStreamTests.cs @@ -0,0 +1,127 @@ +using System.Collections; +using System.Reflection; +using ioxide; +using ioxide.http3; +using ioxide.nghttp3; +using ioxide.ngtcp2; + +namespace Ioxide.Tests; + +/// +/// A QUIC connection hands a closed stream's send state to the next stream that sends. +/// +internal static class QuicSendStreamTests +{ + public static void Register(Runner runner) + { + runner.Test("quic: a warm connection's responses allocate no send-stream state (nghttp3)", () => + { + (string certPath, string keyPath) = TestCert.Ensure(); + var engine = TestServer.Own(new QuicEngine(certPath, keyPath, cidLength: 8, alpn: ["h3"])); + + // Few requests: the test client spends about 1.5 s on each. + const int warm = 4; + const int measured = 4; + const int oneStreamState = 184; // an OutStream, its Queue and the queue's first array + Nghttp3Response response = Nghttp3Response.Text("ok"); // shared, so the handler allocates nothing + long[] marks = new long[2]; + int[] threads = new int[2]; + int served = 0; + + (_, int udpPort) = TestServer.StartDatagram( + onDatagram: null, + quicFactory: engine.CreateFactory(), + quicHandle: (_, conn) => new Nghttp3Connection(conn).RunAsync(_ => + { + // On the reactor's thread: what it allocated between these two requests. + int n = ++served; + int mark = n == warm ? 0 : n == warm + measured ? 1 : -1; + if (mark >= 0) + { + marks[mark] = GC.GetAllocatedBytesForCurrentThread(); + threads[mark] = Environment.CurrentManagedThreadId; + } + return response; + })); + + using var client = new H3TestClient("127.0.0.1", udpPort); + client.Connect(); + Assert.True(client.CompleteHandshake(timeoutMs: 5000), "handshake did not complete"); + + // One stream after another on one connection, so each reuses the state the one before it left. + for (int i = 0; i < warm + measured; i++) + { + (int status, string body) = client.Get($"/{i}", timeoutMs: 5000); + Assert.Equal(200, status); + Assert.Equal("ok", body); + } + + Assert.True(threads[0] == threads[1] && marks[1] >= marks[0], "both marks come from the reactor's thread"); + + long allocated = marks[1] - marks[0]; // nghttp3 makes no other garbage per request + Assert.True(allocated < oneStreamState, + $"{measured} responses on a warm connection allocated {allocated} B on the reactor, a stream's send state or more"); + }); + + runner.Test("quic: a burst leaves the send state of at most 32 streams, with no queue past 64 slots", () => + { + (string certPath, string keyPath) = TestCert.Ensure(); + var engine = TestServer.Own(new QuicEngine(certPath, keyPath, cidLength: 8, alpn: ["h3"])); + + const int burst = 40; + const int chunks = 100; + (int Count, int LargestQueue) kept = default; + + (_, int udpPort) = TestServer.StartDatagram( + onDatagram: null, + quicFactory: engine.CreateFactory(), + quicHandle: (_, conn) => new Http3Connection(conn).RunAsync(async (request, writer) => + { + bool after = request.Path.Span.SequenceEqual("/after"u8); + if (after) + { + kept = Pooled(conn); // before this stream's headers take one + } + writer.WriteHeaders(new Http3Response { Status = 200 }); + for (int i = 0; i < (after ? 0 : chunks); i++) // no ack can land before this returns, so each stays in the queue + { + writer.GetSpan(1)[0] = (byte)'x'; + writer.Advance(1); + await writer.FlushAsync(); + } + })); + + using var client = new H3TestClient("127.0.0.1", udpPort); + client.Connect(); + Assert.True(client.CompleteHandshake(timeoutMs: 5000), "handshake did not complete"); + + string[] paths = [.. Enumerable.Range(0, burst).Select(i => $"/{i}")]; + foreach ((int status, long length) in client.GetConcurrent(paths, timeoutMs: 15000)) + { + Assert.Equal(200, status); + Assert.Equal((long)chunks, length); + } + Assert.Equal(200, client.Get("/after", timeoutMs: 5000).Status); // long enough for the burst's acks to close its streams + + Assert.Equal(32, kept.Count); + Assert.Equal(64, kept.LargestQueue); // each grew past it, to 128 + }); + } + + // On the reactor's thread. OutStream is private to the connection, so by reflection. + private static (int Count, int LargestQueue) Pooled(QuicConnection conn) + { + var pool = (IEnumerable)typeof(QuicEngineConnection) + .GetField("_outStreamPool", BindingFlags.NonPublic | BindingFlags.Instance)!.GetValue(conn)!; + + int count = 0; + int largest = 0; + foreach (object state in pool) + { + object queue = state.GetType().GetField("Chunks")!.GetValue(state)!; + largest = Math.Max(largest, (int)queue.GetType().GetProperty("Capacity")!.GetValue(queue)!); + count++; + } + return (count, largest); + } +}