From c2a9d29f97d28cfa6996ae3864168b1c5854245c Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Sat, 3 Oct 2026 19:05:34 +0100 Subject: [PATCH] reactor: wake for a due QUIC deadline instead of waiting for the 250 ms tick The loop's wait is bounded by the earliest engine deadline, so an idle reactor fires a retransmit, ACK or pacing step when it is due. The 250 ms timer and the sweeps are unchanged, and a reactor with no QUIC deadline makes the same unbounded wait as before. --- src/ioxide/Native/Native.IoUring.cs | 15 +- .../Reactor/Loop/Reactor.Loop.Incremental.cs | 6 +- .../Reactor/Loop/Reactor.Loop.SharedRing.cs | 6 +- src/ioxide/Reactor/Loop/Reactor.Timer.cs | 3 +- src/ioxide/Reactor/Reactor.cs | 3 +- .../Reactor/Transport/Quic/Reactor.Quic.cs | 21 ++- src/ioxide/io_uring/Ring.cs | 18 ++- .../QuicEngineConnection.KeepAlive.cs | 3 +- .../Connection/QuicEngineConnection.cs | 4 +- .../Protocols/QuicTimerTests.cs | 149 +++++++++++++++++- 10 files changed, 203 insertions(+), 25 deletions(-) diff --git a/src/ioxide/Native/Native.IoUring.cs b/src/ioxide/Native/Native.IoUring.cs index a2c01ed..d35afc0 100644 --- a/src/ioxide/Native/Native.IoUring.cs +++ b/src/ioxide/Native/Native.IoUring.cs @@ -31,6 +31,7 @@ public static unsafe partial class Native { // CQE (IORING_CQE_F_NOTIF) once the buffer can be reused. public const byte IORING_OP_SEND_ZC = 47; public const uint IORING_ENTER_GETEVENTS = 1u << 0; + public const uint IORING_ENTER_EXT_ARG = 1u << 3; // the enter's arg is an io_uring_getevents_arg: a wait with a timeout public const long IORING_OFF_SQ_RING = 0; public const long IORING_OFF_SQES = 0x10000000; @@ -99,8 +100,11 @@ public static int io_uring_setup(uint entries, IoUringParams* p) } public static int io_uring_enter(int fd, uint toSubmit, uint minComplete, uint flags) + => io_uring_enter(fd, toSubmit, minComplete, flags, null, 0); + + public static int io_uring_enter(int fd, uint toSubmit, uint minComplete, uint flags, void* arg, nuint argSize) { - long rc = syscall6(SYS_IO_URING_ENTER, (uint)fd, toSubmit, minComplete, flags, null, 0); + long rc = syscall6(SYS_IO_URING_ENTER, (uint)fd, toSubmit, minComplete, flags, arg, argSize); return rc < 0 ? -Marshal.GetLastPInvokeError() : (int)rc; } @@ -178,4 +182,13 @@ public struct io_uring_buf_reg { // include/uapi/linux/time_types.h - the timespec IORING_OP_TIMEOUT reads. [StructLayout(LayoutKind.Sequential)] public struct __kernel_timespec { public long tv_sec; public long tv_nsec; } + + // io_uring_enter's argument under IORING_ENTER_EXT_ARG; ts points at a __kernel_timespec. + [StructLayout(LayoutKind.Sequential)] + public struct io_uring_getevents_arg { + public ulong sigmask; + public uint sigmask_sz; + public uint min_wait_usec; + public ulong ts; + } } diff --git a/src/ioxide/Reactor/Loop/Reactor.Loop.Incremental.cs b/src/ioxide/Reactor/Loop/Reactor.Loop.Incremental.cs index ca734f2..1920141 100644 --- a/src/ioxide/Reactor/Loop/Reactor.Loop.Incremental.cs +++ b/src/ioxide/Reactor/Loop/Reactor.Loop.Incremental.cs @@ -185,8 +185,10 @@ private void LoopIncremental() // Anything else is a lifecycle or programming error. Throwing rather than breaking, // because a reactor that vanishes while the process reports healthy is the worst of // both; whether the process then dies is the host's call, via Reactor.OnFault. - int rc = _ring.SubmitAndWait(1); - if (rc < 0 && rc != -EINTR && rc != -EAGAIN && rc != -EBUSY) + // + // ETIME is not an error at all: the wait's own bound ran out (WaitForCompletions). + int rc = WaitForCompletions(); + if (rc < 0 && rc != -EINTR && rc != -EAGAIN && rc != -EBUSY && rc != -ETIME) { throw new InvalidOperationException( $"[r{_id}] io_uring_enter failed with errno {-rc}; this reactor cannot continue"); diff --git a/src/ioxide/Reactor/Loop/Reactor.Loop.SharedRing.cs b/src/ioxide/Reactor/Loop/Reactor.Loop.SharedRing.cs index 4fc1282..f80c961 100644 --- a/src/ioxide/Reactor/Loop/Reactor.Loop.SharedRing.cs +++ b/src/ioxide/Reactor/Loop/Reactor.Loop.SharedRing.cs @@ -62,8 +62,10 @@ private void LoopSharedRing() // Anything else is a lifecycle or programming error. Throwing rather than breaking, // because a reactor that vanishes while the process reports healthy is the worst of // both; whether the process then dies is the host's call, via Reactor.OnFault. - int rc = _ring.SubmitAndWait(1); - if (rc < 0 && rc != -EINTR && rc != -EAGAIN && rc != -EBUSY) + // + // ETIME is not an error at all: the wait's own bound ran out (WaitForCompletions). + int rc = WaitForCompletions(); + if (rc < 0 && rc != -EINTR && rc != -EAGAIN && rc != -EBUSY && rc != -ETIME) { throw new InvalidOperationException( $"[r{_id}] io_uring_enter failed with errno {-rc}; this reactor cannot continue"); diff --git a/src/ioxide/Reactor/Loop/Reactor.Timer.cs b/src/ioxide/Reactor/Loop/Reactor.Timer.cs index fa00a88..d4d8068 100644 --- a/src/ioxide/Reactor/Loop/Reactor.Timer.cs +++ b/src/ioxide/Reactor/Loop/Reactor.Timer.cs @@ -13,8 +13,7 @@ public sealed unsafe partial class Reactor private readonly List _tickers = []; /// - /// The ticker's interval: the granularity of every sweep, and on an otherwise idle reactor of - /// QUIC engine timers too, which fire only when the loop wakes. + /// The ticker's interval: the granularity of every sweep. /// public const int TickMs = 250; diff --git a/src/ioxide/Reactor/Reactor.cs b/src/ioxide/Reactor/Reactor.cs index c6e39fd..654e639 100644 --- a/src/ioxide/Reactor/Reactor.cs +++ b/src/ioxide/Reactor/Reactor.cs @@ -97,10 +97,11 @@ internal void ReturnWriteSlab(nint p) private readonly int _connBufRingEntries; private readonly uint _incRecvBufferSize; - // Transient io_uring_enter errnos. + // Transient io_uring_enter errnos, and ETIME: a bounded wait that ran out. private const int EINTR = 4; private const int EAGAIN = 11; private const int EBUSY = 16; + private const int ETIME = 62; public Reactor(int id, ServerConfig config) { diff --git a/src/ioxide/Reactor/Transport/Quic/Reactor.Quic.cs b/src/ioxide/Reactor/Transport/Quic/Reactor.Quic.cs index 00d96b6..0b39851 100644 --- a/src/ioxide/Reactor/Transport/Quic/Reactor.Quic.cs +++ b/src/ioxide/Reactor/Transport/Quic/Reactor.Quic.cs @@ -270,7 +270,7 @@ public void QuicRemoveConnection(QuicConnection conn) } // Ticker callback (~250 ms): evict quiet connections. Engine deadlines are fired by - // QuicFireDueTimers at loop-pass granularity; this ticker's loop wake doubles as its floor. + // QuicFireDueTimers, and bound the loop's wait (WaitForCompletions). private void QuicSweep() { long now = Environment.TickCount64; @@ -294,8 +294,8 @@ private void QuicSweep() } // Earliest engine deadline across live conns; long.MaxValue = none. Checked at the top of every - // loop pass, so loss/PTO timers fire at completion-batch granularity (~RTT under load) instead - // of the 250 ms ticker - a retransmit that waits 250 ms per loss makes storms self-sustaining. + // loop pass and bounding the wait, so loss/PTO timers fire when due instead of on the 250 ms + // ticker - a retransmit that waits 250 ms per loss makes storms self-sustaining. private long _quicNextTimeoutMs = long.MaxValue; private void QuicFireDueTimers() @@ -362,6 +362,21 @@ internal void QuicArmTimer(QuicConnection conn) } } + // The loop's wait, bounded by the earliest engine deadline: otherwise only a completion wakes an + // idle reactor, and the one it can count on is the 250 ms ticker. + private int WaitForCompletions() + { + // No connections left: the tracked deadline is the last one's leftover. + long due = _quicConnSet.Count == 0 ? long.MaxValue : _quicNextTimeoutMs; + if (due == long.MaxValue) + { + return _ring.SubmitAndWait(1); + } + + // +1: the ms clock can read just short of the deadline when the kernel's timer wakes us. + return _ring.SubmitAndWait(1, Math.Max(0, due - NowMs) + 1); + } + private void TeardownQuic() { if (_quicOptions == null && _quicConnSet.Count == 0) diff --git a/src/ioxide/io_uring/Ring.cs b/src/ioxide/io_uring/Ring.cs index cc296b8..4ed9211 100644 --- a/src/ioxide/io_uring/Ring.cs +++ b/src/ioxide/io_uring/Ring.cs @@ -164,7 +164,13 @@ public static Ring Create(uint entries) return &_sqes[slot]; } - public int SubmitAndWait(uint waitFor) + public int SubmitAndWait(uint waitFor) => SubmitAndWait(waitFor, -1); + + /// + /// As , with the wait bounded by + /// (negative: unbounded). A wait that runs out with nothing submitted returns -ETIME. + /// + public int SubmitAndWait(uint waitFor, long timeoutMs) { // liburing-style accounting: derive the submit count from the kernel-consumed head, so // SQEs published by an enter that consumed nothing (-EBUSY under CQ-overflow pressure) @@ -181,7 +187,15 @@ public int SubmitAndWait(uint waitFor) uint flags = waitFor > 0 ? IORING_ENTER_GETEVENTS : 0; - return io_uring_enter(_fd, toSubmit, waitFor, flags); + if (timeoutMs < 0) + { + return io_uring_enter(_fd, toSubmit, waitFor, flags); + } + + // The kernel copies both during the call, so the stack is fine. + var ts = new __kernel_timespec { tv_sec = timeoutMs / 1000, tv_nsec = timeoutMs % 1000 * 1_000_000 }; + var arg = new io_uring_getevents_arg { ts = (ulong)&ts }; + return io_uring_enter(_fd, toSubmit, waitFor, flags | IORING_ENTER_EXT_ARG, &arg, (nuint)sizeof(io_uring_getevents_arg)); } [MethodImpl(MethodImplOptions.AggressiveInlining)] diff --git a/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.KeepAlive.cs b/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.KeepAlive.cs index 3efbc20..67990e5 100644 --- a/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.KeepAlive.cs +++ b/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.KeepAlive.cs @@ -37,8 +37,7 @@ private void ApplyKeepAlive() _keepAliveOn = want; _keepAliveStale = false; - // The sweep reaps on a tick, and an idle reactor sends the ping - and the ACK before it, - // which restarts ngtcp2's keep-alive clock - only on a tick. Keep two ticks clear. + // The sweep reaps on a tick; keep two ticks clear for the ping and the peer's ACK of it. int readMs = Math.Max(0, _reactor.QuicReadTimeoutMs); int boundMs = readMs - Math.Min(2 * Reactor.TickMs, readMs / 2); Ngtcp2.iq_conn_set_keep_alive(_conn, want ? 1 : 0, (ulong)boundMs * 1_000_000UL); diff --git a/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.cs b/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.cs index e4ce8b7..42ae9dd 100644 --- a/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.cs +++ b/src/protocols/ioxide.ngtcp2/Connection/QuicEngineConnection.cs @@ -7,8 +7,8 @@ namespace ioxide.ngtcp2; /// /// A live ngtcp2 server connection, bridging the reactor's QUIC transport to the native engine. /// Datagrams routed by CID arrive at and are fed to ngtcp2; the engine's -/// output is flushed back through the transport's Send; loss/idle deadlines ride the reactor -/// ticker via / . Everything runs on the owning +/// output is flushed back through the transport's Send; loss/idle deadlines reach the reactor +/// loop via / . Everything runs on the owning /// reactor thread, so the whole connection - transport half and engine half - is single-threaded. /// /// Application bytes flow through the read surface on : each decrypted diff --git a/tests/Ioxide.Tests.E2E/Protocols/QuicTimerTests.cs b/tests/Ioxide.Tests.E2E/Protocols/QuicTimerTests.cs index 5475683..86edf14 100644 --- a/tests/Ioxide.Tests.E2E/Protocols/QuicTimerTests.cs +++ b/tests/Ioxide.Tests.E2E/Protocols/QuicTimerTests.cs @@ -27,6 +27,7 @@ public static void Register(Runner runner) RegisterLossRecovery(runner); RegisterTimerFaults(runner); RegisterOutboundTimers(runner); + RegisterIdleReactor(runner); } // --- the expiry arithmetic itself ------------------------------------------------------- @@ -37,8 +38,8 @@ private static void RegisterExpiryArithmetic(Runner runner) { // GetNextTimeout converts ngtcp2's ns expiry into the sweep's TickCount64 ms frame by // subtracting the current ns clock. Both are UNSIGNED, and by the time the loop looks, - // the expiry has normally already passed - the loop parks in io_uring until a - // completion arrives, so it reads the clock milliseconds late, every time. Unguarded, + // the expiry has normally already passed - the loop wakes for it at millisecond + // granularity, or for some other completion, so it reads the clock after it. Unguarded, // that subtraction underflows and the connection's next deadline comes back roughly 584 // years out. It does not look like a crash: the connection works perfectly until the // first packet is lost, and then never recovers, for the rest of its life. @@ -72,16 +73,15 @@ private static void RegisterExpiryArithmetic(Runner runner) // Keep it turning over until the transport has actually been asked for a deadline that // had already passed - the state the guard exists for, and the only state in which this // test means anything. It happens within a second; the budget is generous only so a - // loaded machine cannot make it flaky. "Already passed" carries 2 ms of margin, so the - // answer cannot turn on the microseconds between the probe's sample and the engine's - // own re-read inside GetNextTimeout. + // loaded machine cannot make it flaky. Any amount past counts: the clock only moves + // forward, so an expiry the probe saw pass has passed for GetNextTimeout's re-read too. TimerProbeConnection.Observation[] alreadyDue; long deadline = Environment.TickCount64 + 20_000; do { client.Pump(100); alreadyDue = TimerProbeConnection.Snapshot() - .Where(o => o.ExpiryNs != ulong.MaxValue && o.ExpiryNs + 2_000_000 <= o.NowNs) + .Where(o => o.ExpiryNs != ulong.MaxValue && o.ExpiryNs < o.NowNs) .ToArray(); } while (alreadyDue.Length == 0 && Environment.TickCount64 < deadline); @@ -427,6 +427,99 @@ private static void RegisterOutboundTimers(Runner runner) }); } + // --- an idle reactor --------------------------------------------------------------------- + + private static void RegisterIdleReactor(Runner runner) + { + runner.Test("quic/timer: an idle reactor fires an engine deadline when it is due, not at the next tick", () => + { + // Nothing wakes an idle reactor but a completion, and the only one it can count on is the + // 250 ms ticker - so a retransmit due ~25 ms after a loss went out at the next tick. Here + // the answer is lost and the peer stays silent, so the retransmit timer fires again and + // again with backoff, each firing timed against the engine's own deadline. + (string certPath, string keyPath) = TestCert.Ensure(); + using var engine = new QuicEngine(certPath, keyPath, cidLength: 8); + var answered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + (_, int udpPort) = TestServer.StartDatagram( + onDatagram: null, + quicFactory: engine.CreateFactory(r => new TimerLatenessConnection(engine)), + quicHandle: RingDelayedEchoHandler(300, answered)); + + using var client = new LossyQuicClient(udpPort); + client.Connect(); + Assert.True(client.CompleteHandshake(5_000), "handshake did not complete"); + + client.Pump(500); + client.SendRequest(Encoding.ASCII.GetBytes("on-time")); + client.Pump(100); + + // From here the reactor has nothing to do but the answer, then its retransmits. + TimerLatenessConnection.Reset(); + client.Blackout(3_000); + + Assert.True(answered.Task.IsCompleted, "the handler had not answered by the end of the blackout"); + double[] lateMs = TimerLatenessConnection.Snapshot(); + Assert.True(lateMs.Length >= 3, + $"the engine timer fired {lateMs.Length} times in the blackout, too few to judge its timing"); + Assert.True(lateMs.Max() < 50, + $"engine deadlines fired up to {lateMs.Max():F0} ms late " + + $"([{string.Join(", ", lateMs.Select(l => l.ToString("F0")))}] ms): the reactor slept past " + + "them until the next tick"); + }); + + runner.Test("quic/timer: a reactor whose last connection is gone does not keep waking for it", () => + { + // The reactor tracks only the earliest engine deadline, and nothing resets it when the last + // connection leaves: the loop stops looking once there are none. A wait bounded by that + // leftover would wake an empty reactor every millisecond, for good. + (string certPath, string keyPath) = TestCert.Ensure(); + using var engine = new QuicEngine(certPath, keyPath, cidLength: 8); + + (_, int udpPort) = TestServer.StartDatagram( + onDatagram: null, + quicFactory: engine.CreateFactory(), + quicReadMs: 1_000, + quicHandle: EchoHandler); + + using (var client = new LossyQuicClient(udpPort)) + { + client.Connect(); + Assert.True(client.CompleteHandshake(5_000), "handshake did not complete"); + + // The answer is lost and the peer goes silent, so the connection is evicted for its read + // timeout with a retransmit still scheduled - a deadline that then passes. + client.SendRequest(Encoding.ASCII.GetBytes("leave-a-deadline")); + client.Blackout(3_000); + Assert.True(client.DroppedInbound > 0, "the server never answered, so it scheduled no retransmit"); + } + + string reactorThread = $"test-reactor-udp-{udpPort}"; + long before = VoluntarySwitches(reactorThread); + Thread.Sleep(1_000); + long wakes = VoluntarySwitches(reactorThread) - before; + + Assert.True(wakes < 50, + $"a reactor with no connections woke {wakes} times in a second; its ticker accounts for 4"); + }); + } + + // How often a thread has parked: each io_uring wait that sleeps is one voluntary switch. + private static long VoluntarySwitches(string threadName) + { + string comm = threadName.Length > 15 ? threadName[..15] : threadName; // the kernel keeps 15 bytes + foreach (string task in Directory.GetDirectories("/proc/self/task")) + { + if (File.ReadAllText(Path.Combine(task, "comm")).TrimEnd('\n') != comm) + { + continue; + } + string line = File.ReadLines(Path.Combine(task, "status")).First(l => l.StartsWith("voluntary_ctxt_switches:")); + return long.Parse(line["voluntary_ctxt_switches:".Length..].Trim()); + } + throw new InvalidOperationException($"no thread named {comm}"); + } + // --- handler ----------------------------------------------------------------------------- private static async Task EchoHandler(Reactor reactor, QuicConnection conn) @@ -519,7 +612,7 @@ private sealed class TimerProbeConnection(QuicEngine engine) : QuicEngineConnect { internal readonly record struct Observation(ulong ExpiryNs, ulong NowNs, long NowMs, long Deadline); - private static readonly System.Reflection.FieldInfo ConnHandle = + internal static readonly System.Reflection.FieldInfo ConnHandle = typeof(QuicEngineConnection).GetField("_conn", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance) ?? throw new Exception("could not reflect QuicEngineConnection._conn"); @@ -545,7 +638,7 @@ public static Observation[] Snapshot() } // Same clock and the same call the engine makes; reactor thread, like everything else here. - private static ulong NowNs() => (ulong)(System.Diagnostics.Stopwatch.GetTimestamp() * + internal static ulong NowNs() => (ulong)(System.Diagnostics.Stopwatch.GetTimestamp() * (1_000_000_000.0 / System.Diagnostics.Stopwatch.Frequency)); public override long GetNextTimeout(long nowMs) @@ -570,6 +663,46 @@ public override void OnTimer(long nowMs) } } + /// + /// Records how late each firing was: the engine's own ns expiry against the ns clock, on entry to + /// OnTimer. A firing whose expiry is still ahead (the ms deadline rounds down) is not a late one. + /// + private sealed class TimerLatenessConnection(QuicEngine engine) : QuicEngineConnection(engine) + { + private static readonly List LateMs = []; + + public static void Reset() + { + lock (LateMs) + { + LateMs.Clear(); + } + } + + public static double[] Snapshot() + { + lock (LateMs) + { + return LateMs.ToArray(); + } + } + + public override void OnTimer(long nowMs) + { + nint conn = (nint)TimerProbeConnection.ConnHandle.GetValue(this)!; + ulong expiry = conn == 0 ? ulong.MaxValue : LossyQuicClient.Expiry(conn); + ulong now = TimerProbeConnection.NowNs(); + if (expiry <= now) + { + lock (LateMs) + { + LateMs.Add((now - expiry) / 1_000_000.0); + } + } + base.OnTimer(nowMs); + } + } + /// An engine connection whose deadlines are still armed but never acted on. private sealed class SuppressedTimerConnection(QuicEngine engine) : QuicEngineConnection(engine) {