From d864edb1f47b891f69bfe910b42f85cf05140e6c Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Sun, 25 Oct 2026 14:47:09 +0000 Subject: [PATCH 1/2] pipes: the TCP pipe reader and writer park on the connection itself, so a read or flush that waits has one value-task source, not two TcpConnectionPipeReader waited on the connection the way a handler does: it awaited ReadAsync, registering on the connection's value-task source, and its caller awaited the reader's own. Every read that parked made two continuation registrations (a CAS each), two Resets, and the IValueTaskSource calls and token checks between the layers; a flush through TcpConnectionPipeWriter paid the same. The adapters now park on the connection itself. IRecvListener and IFlushListener (internal, one method each) are what the connection calls when the read or flush an adapter parked on completes; the adapter then completes the one source its caller awaits. TryReadOrPark and TryFlushOrPark, the adapters' entry points next to ReadAsync and FlushAsync, keep their arm protocol, guards and sweep clocks, with one addition on the read side: the exchange that takes the arm back decides who delivers. ReadAsync's re-check after the arm can deliver twice when a completion races it, and the delivery into the never-registered source is lost harmlessly; with the adapter called directly, both would run it. The flush side already decided through _flushArmed. The listener is stored before the arm that publishes it, and each adapter resets its own core before it parks, since off the reactor the completion can run before the park returns. ReadAsync and FlushAsync keep their code; their completion sites gain one null check. wrk on Tcp/Pipe, two runs of 12 paired rounds against main: user-mode CPU per request -11.4% and -13.3% (0.164 -> 0.145 us), lower in every round; total CPU per request and req/s flat, as 20 ns is under 1% of a request whose CPU is ~92% the kernel's. Tcp/Raw, which does not use the adapters, stays inside its noise (-1.5% and -4.3%, rounds on both sides of zero); the main-vs-main control moved +0.6% and +3.0% on Pipe. Tests (Unit, AdapterParkTests): a read and a flush an adapter parks are delivered exactly once when a close races their arming, and the reader's ReadAsync and the writer's FlushAsync made off the reactor complete when a close races them - 200,000 rounds each, each asserting that at least 1,000 closes landed while the arm was running. Red-checked one break per variant: the re-check delivering without taking the arm back (123 rounds delivered twice), the read listener stored after the arm (36 and 58 lost), the flush listener stored after the arm (6,971 and 16,220 lost), and each adapter resetting its core after it parks (15 and 36,336 lost) fail exactly the tests that cover them; the unmodified tree passes all four in three runs. Docs: the BCL bridges and the lost-wakeup notes in dev-core. --- docs/learn/dev-core.html | 20 +- src/ioxide/Connection/Tcp/IFlushListener.cs | 15 ++ src/ioxide/Connection/Tcp/IRecvListener.cs | 20 ++ .../Connection/Tcp/TcpConnection.Read.cs | 73 ++++- .../Tcp/TcpConnection.Write.Flush.cs | 73 ++++- src/ioxide/Connection/Tcp/TcpConnection.cs | 5 +- .../Connection/Tcp/TcpConnectionPipeReader.cs | 52 ++-- .../Connection/Tcp/TcpConnectionPipeWriter.cs | 30 +-- tests/Ioxide.Tests.Unit/AdapterParkTests.cs | 249 ++++++++++++++++++ tests/Ioxide.Tests.Unit/Program.cs | 1 + 10 files changed, 473 insertions(+), 65 deletions(-) create mode 100644 src/ioxide/Connection/Tcp/IFlushListener.cs create mode 100644 src/ioxide/Connection/Tcp/IRecvListener.cs create mode 100644 tests/Ioxide.Tests.Unit/AdapterParkTests.cs diff --git a/docs/learn/dev-core.html b/docs/learn/dev-core.html index bf652f18..cc9deae7 100644 --- a/docs/learn/dev-core.html +++ b/docs/learn/dev-core.html @@ -311,7 +311,10 @@

Connection · Connection/Connection.cs (+ .Read, .Wr
  • Lost-wakeup-free read - ReadAsync uses an _armed flag plus a sticky _pending, with a re-check between arming and returning the task, so a completion that races the arm is never lost. Continuations are synchronous - (RunContinuationsAsynchronously = false) - that's what makes resume inline.
    • + (RunContinuationsAsynchronously = false) - that's what makes resume inline. An + adapter parked through TryReadOrPark / TryFlushOrPark gets the same + protocol, plus one rule: whichever side takes the arm back delivers, so the adapter is called + exactly once - there is no unregistered value-task source to absorb a second delivery.
  • Read snapshot - ReadAsync returns a RecvSnapshot (a tail marker + closed flag). The handler then drains slices up to that tail with TryGetItem / GetSnapshotMemories, and returns their buffers with @@ -402,13 +405,16 @@

    BCL bridges

    • TcpConnectionPipeReader (Connection/Tcp/TcpConnectionPipeReader.cs) - adapts the raw read API to System.IO.Pipelines.PipeReader, allocation-free at steady state: a - parked read chains onto the connection's value-task source (no async state machine), recv slices - live in pooled segments on a persistent chain, and AdvanceTo trims the consumed - front and returns those buffers. It honors examined - when everything held has been - examined, it waits for new bytes instead of returning the same data.
      • + parked read is the connection's own read listener (IRecvListener) - the recv + completion calls the reader, which completes the one value task its caller awaits, so there is + no async state machine and no second value-task source in between. Recv slices live in pooled + segments on a persistent chain, and AdvanceTo trims the consumed front and returns + those buffers. It honors examined - when everything held has been examined, it + waits for new bytes instead of returning the same data.
    • TcpConnectionPipeWriter / TcpConnectionStream / TcpConnectionDualPipe - (Connection/Tcp/) - a PipeWriter over the write slab, a - Stream (so SslStream and other BCL code can ride a connection - see + (Connection/Tcp/) - a PipeWriter over the write slab, parked on the + connection's flush the same way (IFlushListener), a Stream (so + SslStream and other BCL code can ride a connection - see TLS internals), and a duplex pipe pairing.

    TcpConnectionStream.cs ยท the Stream write bridge - stage into the slab, then chain onto the connection's flush source (no async state machine, so an SslStream write still resumes inline)

    diff --git a/src/ioxide/Connection/Tcp/IFlushListener.cs b/src/ioxide/Connection/Tcp/IFlushListener.cs new file mode 100644 index 00000000..9b81eec7 --- /dev/null +++ b/src/ioxide/Connection/Tcp/IFlushListener.cs @@ -0,0 +1,15 @@ +namespace ioxide; + +/// +/// An adapter parked on a connection's flush in place of an awaiter - the write side of +/// , for . +/// +/// +/// Parked through . runs exactly +/// once per flush that had to wait, from whichever path disarms it: the send's completion on the +/// reactor, or the release of a flush whose connection closed before the kernel had it. +/// +internal interface IFlushListener +{ + void OnFlushDone(); +} diff --git a/src/ioxide/Connection/Tcp/IRecvListener.cs b/src/ioxide/Connection/Tcp/IRecvListener.cs new file mode 100644 index 00000000..10c9652a --- /dev/null +++ b/src/ioxide/Connection/Tcp/IRecvListener.cs @@ -0,0 +1,20 @@ +namespace ioxide; + +/// +/// An adapter parked on a connection's read in place of an awaiter. +/// +/// +/// used to park on 's +/// value-task source and complete its own from that source's continuation: two sources, two +/// continuation registrations and the interface calls between them, on every read that waits. +/// Parked through instead, the connection calls the reader +/// directly, and the reader's own source - the one its caller awaits - is the only one left. +/// +/// The arm protocol is 's, with one addition: whichever side +/// takes the arm back delivers, so runs exactly once per park, on the thread +/// that completed the read - the reactor's for a recv, the closing one for a close. +/// +internal interface IRecvListener +{ + void OnRecvReady(RecvSnapshot snapshot); +} diff --git a/src/ioxide/Connection/Tcp/TcpConnection.Read.cs b/src/ioxide/Connection/Tcp/TcpConnection.Read.cs index 4486d1b9..02121d80 100644 --- a/src/ioxide/Connection/Tcp/TcpConnection.Read.cs +++ b/src/ioxide/Connection/Tcp/TcpConnection.Read.cs @@ -26,6 +26,10 @@ internal TcpConnection SetFd(int fd) private int _pending; private int _closed; + // The adapter parked on the read in place of an awaiter (TryReadOrPark), or null for ReadAsync's. + // Set before the arm, so whoever takes the arm back sees it. + private IRecvListener? _recvListener; + private readonly SpscRecvRing _recv; // sized by TcpOptions.RecvQueueEntries /// @@ -83,6 +87,71 @@ public ValueTask ReadAsync() return new ValueTask(this, (short)gen); } + /// + /// for an adapter that parks itself instead of an awaiter: true with the + /// snapshot when one is ready now, false when the read was armed - and then the completion that + /// disarms it calls , once, on the completing thread. + /// + internal bool TryReadOrPark(IRecvListener listener, out RecvSnapshot snapshot) + { + if (!_recv.IsEmpty() || Volatile.Read(ref _pending) == 1) + { + Volatile.Write(ref _pending, 0); + snapshot = new RecvSnapshot(_recv.SnapshotTail(), Volatile.Read(ref _closed) != 0); + return true; + } + + if (Volatile.Read(ref _closed) != 0) + { + snapshot = RecvSnapshot.Closed(); + return true; + } + + // Before the listener is stored, so a second read refused here leaves the parked one's alone. + if (Volatile.Read(ref _armed) == 1) + { + throw new InvalidOperationException("ReadAsync already armed."); + } + + ReadParkedMs = _reactor.NowMs; // the read clock; before arming, so the sweep sees both + _recvListener = listener; // published by the exchange below + + if (Interlocked.Exchange(ref _armed, 1) == 1) + { + throw new InvalidOperationException("ReadAsync already armed."); + } + + // Race recovery, as in ReadAsync, with one difference: there a second delivery lands in a + // value-task source nobody registered on, and is harmless; here both would call the listener. + // So the exchange that takes the arm back decides who delivers - this call, or the completion. + if ((!_recv.IsEmpty() || Volatile.Read(ref _pending) == 1 || Volatile.Read(ref _closed) != 0) + && Interlocked.Exchange(ref _armed, 0) == 1) + { + _recvListener = null; + Volatile.Write(ref _pending, 0); + snapshot = new RecvSnapshot(_recv.SnapshotTail(), Volatile.Read(ref _closed) != 0); + return true; + } + + snapshot = default; + return false; + } + + // The one delivery of a read whose arm the caller took back: to the adapter parked on it, or to + // ReadAsync's awaiter. The slot is cleared first, as the listener may park again inline. + private void DeliverRead(RecvSnapshot snapshot) + { + IRecvListener? listener = _recvListener; + if (listener != null) + { + _recvListener = null; + listener.OnRecvReady(snapshot); + return; + } + + _readSignal.SetResult(snapshot); + } + /// /// Take the next item of a snapshot. This is a DEQUEUE and it transfers ownership: every item /// with HasBuffer must be handed back with , exactly once. @@ -157,7 +226,7 @@ internal void WakeRead() { if (Interlocked.Exchange(ref _armed, 0) == 1) { - _readSignal.SetResult(new RecvSnapshot(_recv.SnapshotTail(), Volatile.Read(ref _closed) != 0)); + DeliverRead(new RecvSnapshot(_recv.SnapshotTail(), Volatile.Read(ref _closed) != 0)); } else { @@ -188,7 +257,7 @@ public bool Complete(int res, ushort bid, bool hasBuffer, byte* ptr) if (Interlocked.Exchange(ref _armed, 0) == 1) { - _readSignal.SetResult(new RecvSnapshot(_recv.SnapshotTail(), Volatile.Read(ref _closed) != 0)); + DeliverRead(new RecvSnapshot(_recv.SnapshotTail(), Volatile.Read(ref _closed) != 0)); } else { diff --git a/src/ioxide/Connection/Tcp/TcpConnection.Write.Flush.cs b/src/ioxide/Connection/Tcp/TcpConnection.Write.Flush.cs index e51d5013..40844c61 100644 --- a/src/ioxide/Connection/Tcp/TcpConnection.Write.Flush.cs +++ b/src/ioxide/Connection/Tcp/TcpConnection.Write.Flush.cs @@ -20,6 +20,10 @@ public sealed unsafe partial class TcpConnection : IValueTaskSource private int _flushArmed; private int _flushInProgress; + // The adapter parked on the flush in place of an awaiter (TryFlushOrPark), or null for FlushAsync's. + // Set past the in-progress guard and before the arm, so whoever disarms the flush sees it. + private IFlushListener? _flushListener; + internal bool FlushSubmitted => Volatile.Read(ref _flushArmed) == 2; /// @@ -105,6 +109,41 @@ internal ValueTask FlushIfIdleAsync() return FlushCore(); } + /// + /// for an adapter that parks itself instead of an awaiter: true when the + /// flush is already over (nothing staged, or the connection closed), false when it was handed to + /// the reactor - and then whichever path disarms it calls , once, on + /// its own thread. + /// + internal bool TryFlushOrPark(IFlushListener listener) + { + if (Volatile.Read(ref _closed) == 1) + { + if (!FlushSubmitted) + { + DropStaged(); + } + + return true; + } + + if (Interlocked.Exchange(ref _flushInProgress, 1) == 1) + { + throw new InvalidOperationException("FlushAsync already in progress."); + } + + // Past the guard this is the connection's one flush, so the slot is free. Stored before the + // arm in ArmFlush, which publishes it to the path that disarms. + _flushListener = listener; + if (ArmFlush(out _)) + { + return false; + } + + _flushListener = null; // nothing was staged, so nothing was armed + return true; + } + // The connection is already torn down, so the staged bytes are dropped rather than kept: // nothing will ever send them, and leaving the tail where it was made every later write append // to a slab that only ever grew - a writer that keeps producing against a peer that has gone @@ -123,9 +162,13 @@ private void DropStaged() } } - // Both entry points past their guards: whoever gets here owns _flushInProgress and is the one - // flush the connection is allowed to have outstanding. private ValueTask FlushCore() + => ArmFlush(out int gen) ? new ValueTask(this, (short)gen) : default; + + // Every entry point past its guard: whoever gets here owns _flushInProgress and is the one flush + // the connection is allowed to have outstanding. False when nothing was staged - the flush is + // over before it started. is the generation the flush was queued under. + private bool ArmFlush(out int gen) { int target = WriteTail; for (int i = 0; i < _ovCount; i++) @@ -135,8 +178,9 @@ private ValueTask FlushCore() if (target == 0) { Volatile.Write(ref _flushInProgress, 0); + gen = 0; - return default; + return false; } // Before arming: once armed, a close on the reactor can complete the flush, and a Reset after it would lose that. @@ -158,7 +202,7 @@ private ValueTask FlushCore() // The generation lets the reactor drop a flush whose connection closed (or // whose fd was reused) before the queue drained. - int gen = Volatile.Read(ref _generation); + gen = Volatile.Read(ref _generation); _reactor.EnqueueFlush(ClientFd, gen); @@ -169,7 +213,7 @@ private ValueTask FlushCore() ReleaseParkedFlush(); } - return new ValueTask(this, (short)gen); + return true; } // The reactor takes a queued flush to the kernel. Fails if it was released first. @@ -181,8 +225,23 @@ private void ReleaseParkedFlush() if (Interlocked.CompareExchange(ref _flushArmed, 0, 1) == 1) { Volatile.Write(ref _flushInProgress, 0); - _flushSignal.SetResult(true); + DeliverFlush(); + } + } + + // The one signal of a flush whose arm the caller took back: to the adapter parked on it, or to + // FlushAsync's awaiter. The slot is cleared first, as the listener may flush again inline. + private void DeliverFlush() + { + IFlushListener? listener = _flushListener; + if (listener != null) + { + _flushListener = null; + listener.OnFlushDone(); + return; } + + _flushSignal.SetResult(true); } // Called by the reactor's send-completion path. @@ -203,7 +262,7 @@ internal void CompleteFlush() // reactor thread and crashes the process. if (Interlocked.Exchange(ref _flushArmed, 0) != 0) { - _flushSignal.SetResult(true); + DeliverFlush(); } } diff --git a/src/ioxide/Connection/Tcp/TcpConnection.cs b/src/ioxide/Connection/Tcp/TcpConnection.cs index f3610891..d40e3a60 100644 --- a/src/ioxide/Connection/Tcp/TcpConnection.cs +++ b/src/ioxide/Connection/Tcp/TcpConnection.cs @@ -122,7 +122,7 @@ public void MarkClosed() if (Interlocked.Exchange(ref _armed, 0) == 1) { - _readSignal.SetResult(new RecvSnapshot(_recv.SnapshotTail(), isClosed: true)); + DeliverRead(new RecvSnapshot(_recv.SnapshotTail(), isClosed: true)); } else { @@ -262,6 +262,9 @@ internal void Clear() Volatile.Write(ref _recvOverflowed, 0); Volatile.Write(ref _flushArmed, 0); Volatile.Write(ref _flushInProgress, 0); + // Normally empty already: the teardown's MarkClosed delivered to any adapter parked here. + _recvListener = null; + _flushListener = null; Volatile.Write(ref FlushArmedMs, 0); Volatile.Write(ref ReadParkedMs, 0); FlushSeenMs = 0; diff --git a/src/ioxide/Connection/Tcp/TcpConnectionPipeReader.cs b/src/ioxide/Connection/Tcp/TcpConnectionPipeReader.cs index 1b1b2eae..c07874dd 100644 --- a/src/ioxide/Connection/Tcp/TcpConnectionPipeReader.cs +++ b/src/ioxide/Connection/Tcp/TcpConnectionPipeReader.cs @@ -1,6 +1,5 @@ using System.Buffers; using System.IO.Pipelines; -using System.Runtime.CompilerServices; using System.Threading.Tasks.Sources; using ioxide.utils; @@ -8,12 +7,13 @@ namespace ioxide; /// /// Adapts the raw read API to a , allocation-free -/// at steady state: a parked read chains onto the connection's value-task source (no async state -/// machine), recv slices live in pooled segments on a persistent chain, and consumption trims the -/// chain's front. Honors examined: when everything held has been examined, ReadAsync waits -/// for new bytes instead of returning the same data again. +/// at steady state: a parked read is the connection's own read listener (no async state machine, and +/// no second value-task source in between - see ), recv slices live in +/// pooled segments on a persistent chain, and consumption trims the chain's front. Honors +/// examined: when everything held has been examined, ReadAsync waits for new bytes instead of +/// returning the same data again. /// -public sealed unsafe class TcpConnectionPipeReader : PipeReader, IValueTaskSource, IRecvBufferHolder +public sealed unsafe class TcpConnectionPipeReader : PipeReader, IValueTaskSource, IRecvBufferHolder, IRecvListener { // One pooled object per held recv slice: sequence segment + reusable memory // manager + the original ring item (needed to return the buffer). @@ -54,13 +54,12 @@ public void Link(Slice next) private readonly Stack _pool = new(); private ReadOnlySequence _lastSequence; - // Parked-read plumbing: chain onto the connection's IVTS, complete our own. + // Parked-read plumbing: the connection calls OnRecvReady when its read completes, and that + // completes the one value task the caller awaits - this core's. private ManualResetValueTaskSourceCore _core = new() { RunContinuationsAsynchronously = false, }; - private ValueTaskAwaiter _pendingRead; - private readonly Action _onRecvReady; private bool _completed; private bool _cancelRequested; @@ -72,7 +71,6 @@ public void Link(Slice next) public TcpConnectionPipeReader(TcpConnection connection) { _conn = connection ?? throw new ArgumentNullException(nameof(connection)); - _onRecvReady = OnRecvReady; // So the reactor can reclaim at recycle what this reader dequeued and never gave back. _conn.BufferHolder = this; @@ -96,20 +94,19 @@ public override ValueTask ReadAsync(CancellationToken cancellationTo : new ValueTask(BuildResult(isCanceled: false)); } - // Everything held was examined - wait for new bytes. + // Everything held was examined - wait for new bytes. The core is reset before the read can + // be armed: off the reactor, the completion may run OnRecvReady before this returns. + _core.Reset(); + short version = _core.Version; + while (true) { - ValueTask pending = _conn.ReadAsync(); - - if (!pending.IsCompletedSuccessfully) + if (!_conn.TryReadOrPark(this, out RecvSnapshot snapshot)) { - _core.Reset(); - _pendingRead = pending.GetAwaiter(); - _pendingRead.UnsafeOnCompleted(_onRecvReady); - return new ValueTask(this, _core.Version); + return new ValueTask(this, version); } - if (Ingest(pending.Result) || _connectionClosed || _cancelRequested) + if (Ingest(snapshot) || _connectionClosed || _cancelRequested) { if (LostToOverflow) { @@ -123,26 +120,21 @@ public override ValueTask ReadAsync(CancellationToken cancellationTo } } - // Completion of a parked conn.ReadAsync - runs inline on the reactor. - private void OnRecvReady() + // The connection's read completed while this reader was parked on it - inline on the reactor for + // a recv, on the closing thread for a close. + void IRecvListener.OnRecvReady(RecvSnapshot snapshot) { - RecvSnapshot snapshot = _pendingRead.GetResult(); - if (!Ingest(snapshot) && !_connectionClosed && !_cancelRequested) { // Nothing new: re-arm without completing the caller. while (true) { - ValueTask pending = _conn.ReadAsync(); - - if (!pending.IsCompletedSuccessfully) + if (!_conn.TryReadOrPark(this, out snapshot)) { - _pendingRead = pending.GetAwaiter(); - _pendingRead.UnsafeOnCompleted(_onRecvReady); return; } - if (Ingest(pending.Result) || _connectionClosed || _cancelRequested) + if (Ingest(snapshot) || _connectionClosed || _cancelRequested) { break; } @@ -190,8 +182,6 @@ private bool Ingest(in RecvSnapshot snapshot) any = true; } - _conn.ResetRead(); - if (snapshot.IsClosed) { _connectionClosed = true; diff --git a/src/ioxide/Connection/Tcp/TcpConnectionPipeWriter.cs b/src/ioxide/Connection/Tcp/TcpConnectionPipeWriter.cs index 97ddb36d..eecd2b1a 100644 --- a/src/ioxide/Connection/Tcp/TcpConnectionPipeWriter.cs +++ b/src/ioxide/Connection/Tcp/TcpConnectionPipeWriter.cs @@ -1,5 +1,4 @@ using System.IO.Pipelines; -using System.Runtime.CompilerServices; using System.Threading.Tasks.Sources; // ReSharper disable SuggestVarOrType_BuiltInTypes @@ -7,10 +6,11 @@ namespace ioxide; /// /// Adapts the write API to a . Writes go straight -/// into the connection's slab; a parked flush chains onto the connection's value-task source -/// instead of an async state machine, so the adapter allocates nothing at steady state. +/// into the connection's slab; a parked flush is the connection's own flush listener instead of an +/// async state machine or a second value-task source (see ), so the +/// adapter allocates nothing at steady state. /// -public sealed class TcpConnectionPipeWriter : PipeWriter, IValueTaskSource +public sealed class TcpConnectionPipeWriter : PipeWriter, IValueTaskSource, IFlushListener { private readonly TcpConnection _conn; @@ -18,8 +18,6 @@ public sealed class TcpConnectionPipeWriter : PipeWriter, IValueTaskSource true; @@ -53,24 +50,23 @@ public override ValueTask FlushAsync(CancellationToken cancellation } _unflushed = 0; - ValueTask inner = _conn.FlushAsync(); - if (inner.IsCompletedSuccessfully) + // Reset before the flush is handed over: off the reactor, it can complete before this returns. + _core.Reset(); + short version = _core.Version; + + if (_conn.TryFlushOrPark(this)) { return new ValueTask(new FlushResult(isCanceled: false, isCompleted: _completed || _conn.IsClosed)); } - _core.Reset(); - _pendingFlush = inner.GetAwaiter(); - _pendingFlush.UnsafeOnCompleted(_onFlushDone); - return new ValueTask(this, _core.Version); + return new ValueTask(this, version); } - // Completion of a parked flush - runs inline on the reactor. - private void OnFlushDone() + // The connection's flush completed while this writer was parked on it - inline on the reactor + // for a send, on the releasing thread for a flush its connection closed under. + void IFlushListener.OnFlushDone() { - _pendingFlush.GetResult(); - bool canceled = _cancelRequested; _cancelRequested = false; diff --git a/tests/Ioxide.Tests.Unit/AdapterParkTests.cs b/tests/Ioxide.Tests.Unit/AdapterParkTests.cs new file mode 100644 index 00000000..0df5a359 --- /dev/null +++ b/tests/Ioxide.Tests.Unit/AdapterParkTests.cs @@ -0,0 +1,249 @@ +using System.IO.Pipelines; +using System.Runtime.CompilerServices; +using ioxide; + +namespace Ioxide.Tests; + +/// +/// The pipe adapters park on the connection itself - and +/// - instead of on its value tasks, so the connection's arm protocol now +/// has to deliver exactly once on its own. Before, a delivery that raced the arm landed in a +/// value-task source nobody had registered on and went nowhere; with the adapter called directly, +/// that same delivery runs it a second time. +/// +/// +/// Each test races an arm made off the reactor against a close made on it, the shape a handler that +/// hops threads produces. The "reactor" is a thread of the test's own on a reactor that never runs, +/// which plays its part: it closes, and drains the queued flushes, whose fd -1 finds no table entry, +/// as a closed connection's does. The window is a few instructions wide, so every test also asserts +/// that closes landed inside it before trusting a clean count. +/// +internal static class AdapterParkTests +{ + private const int Rounds = 200_000; + + // Of the rounds, how many must have had the close complete while the arm was still running. + private const int MinOverlapped = 1_000; + + public static void Register(Runner runner) + { + runner.Test("park: a read an adapter parks is delivered exactly once when a close races its arming", () => + { + var listener = new CountingRecvListener(); + (int lost, int twice, int overlapped) = Race( + static (conn, state) => + { + var l = (CountingRecvListener)state!; + Volatile.Write(ref l.Calls, 0); + return conn.TryReadOrPark(l, out _) ? 1 : 0; + }, + static state => Volatile.Read(ref ((CountingRecvListener)state!).Calls), + listener); + + Assert.True(overlapped >= MinOverlapped, $"only {overlapped} closes landed while a read was arming: the race never ran"); + Assert.True(lost == 0, $"{lost} of {Rounds} parked reads were never delivered"); + Assert.True(twice == 0, $"{twice} of {Rounds} parked reads were delivered twice"); + }, skip: Environment.ProcessorCount < 2); + + runner.Test("park: a flush an adapter parks is delivered exactly once when a close races its arming", () => + { + var listener = new CountingFlushListener(); + (int lost, int twice, int overlapped) = Race( + static (conn, state) => + { + var l = (CountingFlushListener)state!; + Volatile.Write(ref l.Calls, 0); + conn.Write(One); + return conn.TryFlushOrPark(l) ? 1 : 0; + }, + static state => Volatile.Read(ref ((CountingFlushListener)state!).Calls), + listener); + + Assert.True(overlapped >= MinOverlapped, $"only {overlapped} closes landed while a flush was arming: the race never ran"); + Assert.True(lost == 0, $"{lost} of {Rounds} parked flushes were never delivered"); + Assert.True(twice == 0, $"{twice} of {Rounds} parked flushes were delivered twice"); + }, skip: Environment.ProcessorCount < 2); + + runner.Test("pipereader: a read from off the reactor that races the connection's close completes", () => + { + // The reader has to have its own value task ready before it parks: off the reactor, the + // close can complete the park before ReadAsync has returned. + var reads = new ValueTask[1]; + (int lost, _, int overlapped) = Race( + static (conn, state) => + { + ((ValueTask[])state!)[0] = new TcpConnectionPipeReader(conn).ReadAsync(); + return 0; + }, + static state => + { + ValueTask read = ((ValueTask[])state!)[0]; + return read.IsCompleted && read.Result.IsCompleted ? 1 : 0; + }, + reads); + + Assert.True(overlapped >= MinOverlapped, $"only {overlapped} closes landed while a read was arming: the race never ran"); + Assert.True(lost == 0, $"{lost} of {Rounds} reads never completed, or completed without the close, after it raced their arming"); + }, skip: Environment.ProcessorCount < 2); + + runner.Test("pipewriter: a flush from off the reactor that races the connection's close completes", () => + { + var flushes = new ValueTask[1]; + (int lost, _, int overlapped) = Race( + static (conn, state) => + { + conn.Write(One); + ((ValueTask[])state!)[0] = new TcpConnectionPipeWriter(conn).FlushAsync(); + return 0; + }, + static state => + { + ValueTask flush = ((ValueTask[])state!)[0]; + return flush.IsCompleted && flush.Result.IsCompleted ? 1 : 0; + }, + flushes); + + Assert.True(overlapped >= MinOverlapped, $"only {overlapped} closes landed while a flush was arming: the race never ran"); + Assert.True(lost == 0, $"{lost} of {Rounds} flushes never completed, or completed without the close, after it raced their arming"); + }, skip: Environment.ProcessorCount < 2); + } + + private static readonly byte[] One = [1]; + + /// + /// on this thread against a close on the "reactor" thread, once per round, + /// with the two sides skewed differently each time. returns the deliveries + /// it made itself and the ones the close made, read once both sides + /// are done: a round that adds up to 0 is lost, one past 1 is twice. + /// + private static (int Lost, int Twice, int Overlapped) Race( + Func arm, Func delivered, object? state) + { + var reactor = new Reactor(0, new ServerConfig { ReactorCount = 1, Tcp = null, Udp = null }); + var side = new CloseSide(reactor); + var thread = new Thread(side.Run) { IsBackground = true, Name = "park-race-reactor" }; + thread.Start(); + side.WaitStarted(); + + int lost = 0, twice = 0, overlapped = 0; + try + { + for (int i = 1; i <= Rounds; i++) + { + var conn = new TcpConnection(reactor, -1, 64, 8); + side.Conn = conn; + side.Skew = i / 8 * 3 % 8; + Volatile.Write(ref side.Go, i); + for (int s = i * 7 % 8; s > 0; s--) + { + Thread.SpinWait(1); + } + + bool closedBefore = Volatile.Read(ref side.Closed) == i; + int deliveries = arm(conn, state); + if (!closedBefore && Volatile.Read(ref side.Closed) == i) + { + overlapped++; // the close completed while the arm was running + } + CloseSide.WaitFor(ref side.Closed, i); + + deliveries += delivered(state); + if (deliveries == 0) + { + lost++; + } + else if (deliveries > 1) + { + twice++; + } + conn.Dispose(); + } + } + finally + { + Volatile.Write(ref side.Go, -1); + thread.Join(); + } + + return (lost, twice, overlapped); + } + + // The reactor's part: a close per round, and the drain of the flushes queued to it meanwhile. + private sealed class CloseSide(Reactor reactor) + { + public TcpConnection? Conn; + public int Go, Closed, Skew; + private int _started; + + public void Run() + { + ReactorThreadId(reactor) = Environment.CurrentManagedThreadId; // the arms are off it + Volatile.Write(ref _started, 1); + + for (int i = 1; ; i++) + { + if (!WaitFor(ref Go, i)) + { + return; + } + for (int s = Skew; s > 0; s--) + { + Thread.SpinWait(1); + } + + Conn!.MarkClosed(); + DrainFlushQ(reactor); + Volatile.Write(ref Closed, i); + } + } + + public void WaitStarted() => WaitFor(ref _started, 1); + + // Spins, so the two sides meet within nanoseconds; yields past that, so a busy machine still + // gets through. False when the round counter is -1: the race is over. + public static bool WaitFor(ref int field, int value) + { + for (int spins = 0; ; spins++) + { + int seen = Volatile.Read(ref field); + if (seen == value) + { + return true; + } + if (seen == -1) + { + return false; + } + + if (spins < 1_000) + { + Thread.SpinWait(1); + } + else + { + Thread.Yield(); + } + } + } + } + + private sealed class CountingRecvListener : IRecvListener + { + public int Calls; + + public void OnRecvReady(RecvSnapshot snapshot) => Interlocked.Increment(ref Calls); + } + + private sealed class CountingFlushListener : IFlushListener + { + public int Calls; + + public void OnFlushDone() => Interlocked.Increment(ref Calls); + } + + [UnsafeAccessor(UnsafeAccessorKind.Field, Name = "_reactorThreadId")] + private static extern ref int ReactorThreadId(Reactor reactor); + + [UnsafeAccessor(UnsafeAccessorKind.Method, Name = "DrainFlushQ")] + private static extern void DrainFlushQ(Reactor reactor); +} diff --git a/tests/Ioxide.Tests.Unit/Program.cs b/tests/Ioxide.Tests.Unit/Program.cs index d564bf63..92222f17 100644 --- a/tests/Ioxide.Tests.Unit/Program.cs +++ b/tests/Ioxide.Tests.Unit/Program.cs @@ -29,6 +29,7 @@ private static int Main() HpackEncodeTests.Register(runner); QpackStaticEncodeTests.Register(runner); SendWaitAllTests.Register(runner); + AdapterParkTests.Register(runner); return runner.Summary(); } From 651ff0678a733d41cdb6bf5338ab0377c439375b Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Sun, 25 Oct 2026 17:42:51 +0000 Subject: [PATCH 2/2] pipes: the adapters ready their value task only once the connection has decided to park them, so an overlapping read or flush it refuses leaves the parked one to complete From review of the first commit. Each adapter reset its own value-task source before calling into the connection, so a second read or flush made while one was parked threw the first one's continuation away before the connection refused it - or, on a closed connection with the first flush still with the kernel, returned at once and left the first never resuming. On main the guard ran first. A flush with nothing staged likewise invalidated an earlier flush's unread result. The connection now calls IRecvListener.OnParking / IFlushListener.OnParking past every guard, right before the arm that publishes the park, and only when it does park: never for a read served at once, a refused one, a flush on a closed connection, or one with nothing staged. The adapters reset their source there. The reader's re-park from inside OnRecvReady, after a wake that brought nothing new, keeps the value task its caller already waits on. The flush listener slot is now written only when a flush is armed, so ArmFlush's nothing-staged path no longer leaves it set while the in-progress guard is already released. Tests (Unit, AdapterParkTests): a second read refused while one is parked, a second flush refused while one is parked, a flush on a closed connection while the kernel holds an earlier one, and a flush with nothing staged after an unread completed one each leave the first operation to complete or readable; a wake with nothing new parks the read again without losing its caller. --- src/ioxide/Connection/Tcp/IFlushListener.cs | 8 ++ src/ioxide/Connection/Tcp/IRecvListener.cs | 9 ++ .../Connection/Tcp/TcpConnection.Read.cs | 12 +- .../Tcp/TcpConnection.Write.Flush.cs | 23 ++-- .../Connection/Tcp/TcpConnectionPipeReader.cs | 46 +++++--- .../Connection/Tcp/TcpConnectionPipeWriter.cs | 12 +- tests/Ioxide.Tests.Unit/AdapterParkTests.cs | 107 ++++++++++++++++++ 7 files changed, 181 insertions(+), 36 deletions(-) diff --git a/src/ioxide/Connection/Tcp/IFlushListener.cs b/src/ioxide/Connection/Tcp/IFlushListener.cs index 9b81eec7..cf9459b8 100644 --- a/src/ioxide/Connection/Tcp/IFlushListener.cs +++ b/src/ioxide/Connection/Tcp/IFlushListener.cs @@ -11,5 +11,13 @@ namespace ioxide; /// internal interface IFlushListener { + /// + /// The connection is about to arm this listener's flush: the in-progress guard has passed and + /// something is staged. As : the listener readies its value + /// task here, so a flush refused for overlapping one in flight, or one that sends nothing, leaves + /// an earlier flush's source alone. + /// + void OnParking(); + void OnFlushDone(); } diff --git a/src/ioxide/Connection/Tcp/IRecvListener.cs b/src/ioxide/Connection/Tcp/IRecvListener.cs index 10c9652a..fd560616 100644 --- a/src/ioxide/Connection/Tcp/IRecvListener.cs +++ b/src/ioxide/Connection/Tcp/IRecvListener.cs @@ -16,5 +16,14 @@ namespace ioxide; /// internal interface IRecvListener { + /// + /// The connection is about to park this listener: every guard has passed, and the arm that + /// publishes the park comes next. The listener readies the value task its caller will wait on + /// here and nowhere earlier - a read refused for overlapping a parked one must leave that one's + /// source alone - and not later either, since off the reactor the completion can run + /// before the park returns. Not called for a read served at once. + /// + void OnParking(); + void OnRecvReady(RecvSnapshot snapshot); } diff --git a/src/ioxide/Connection/Tcp/TcpConnection.Read.cs b/src/ioxide/Connection/Tcp/TcpConnection.Read.cs index 02121d80..93e9187a 100644 --- a/src/ioxide/Connection/Tcp/TcpConnection.Read.cs +++ b/src/ioxide/Connection/Tcp/TcpConnection.Read.cs @@ -107,13 +107,15 @@ internal bool TryReadOrPark(IRecvListener listener, out RecvSnapshot snapshot) return true; } - // Before the listener is stored, so a second read refused here leaves the parked one's alone. + // Before the listener readies anything or is stored, so a second read refused here leaves the + // parked one, and the value task its caller waits on, alone. if (Volatile.Read(ref _armed) == 1) { throw new InvalidOperationException("ReadAsync already armed."); } ReadParkedMs = _reactor.NowMs; // the read clock; before arming, so the sweep sees both + listener.OnParking(); // its value task, ready before the arm the completion can follow at once _recvListener = listener; // published by the exchange below if (Interlocked.Exchange(ref _armed, 1) == 1) @@ -121,9 +123,11 @@ internal bool TryReadOrPark(IRecvListener listener, out RecvSnapshot snapshot) throw new InvalidOperationException("ReadAsync already armed."); } - // Race recovery, as in ReadAsync, with one difference: there a second delivery lands in a - // value-task source nobody registered on, and is harmless; here both would call the listener. - // So the exchange that takes the arm back decides who delivers - this call, or the completion. + // Race recovery, as in ReadAsync, with one difference: there the exchange's result is not + // looked at, so when a completion has already taken the arm both sides deliver - the second + // time into the connection's own value-task source, which can then look complete to the next + // read. Here the second delivery would run the listener again. So the exchange that takes the + // arm back decides who delivers: this call, or the completion. if ((!_recv.IsEmpty() || Volatile.Read(ref _pending) == 1 || Volatile.Read(ref _closed) != 0) && Interlocked.Exchange(ref _armed, 0) == 1) { diff --git a/src/ioxide/Connection/Tcp/TcpConnection.Write.Flush.cs b/src/ioxide/Connection/Tcp/TcpConnection.Write.Flush.cs index 40844c61..2d154834 100644 --- a/src/ioxide/Connection/Tcp/TcpConnection.Write.Flush.cs +++ b/src/ioxide/Connection/Tcp/TcpConnection.Write.Flush.cs @@ -132,16 +132,7 @@ internal bool TryFlushOrPark(IFlushListener listener) throw new InvalidOperationException("FlushAsync already in progress."); } - // Past the guard this is the connection's one flush, so the slot is free. Stored before the - // arm in ArmFlush, which publishes it to the path that disarms. - _flushListener = listener; - if (ArmFlush(out _)) - { - return false; - } - - _flushListener = null; // nothing was staged, so nothing was armed - return true; + return !ArmFlush(listener, out _); } // The connection is already torn down, so the staged bytes are dropped rather than kept: @@ -163,12 +154,14 @@ private void DropStaged() } private ValueTask FlushCore() - => ArmFlush(out int gen) ? new ValueTask(this, (short)gen) : default; + => ArmFlush(null, out int gen) ? new ValueTask(this, (short)gen) : default; // Every entry point past its guard: whoever gets here owns _flushInProgress and is the one flush // the connection is allowed to have outstanding. False when nothing was staged - the flush is - // over before it started. is the generation the flush was queued under. - private bool ArmFlush(out int gen) + // over before it started, and the listener slot was never touched. + // is the adapter parking on the flush, or null for FlushAsync's awaiter; + // is the generation the flush was queued under. + private bool ArmFlush(IFlushListener? listener, out int gen) { int target = WriteTail; for (int i = 0; i < _ovCount; i++) @@ -184,7 +177,11 @@ private bool ArmFlush(out int gen) } // Before arming: once armed, a close on the reactor can complete the flush, and a Reset after it would lose that. + // The same goes for the listener's own value task, and the slot the disarming path reads: both + // are set here, past the guard and with something to send, and published by the arm below. _flushSignal.Reset(); + listener?.OnParking(); + _flushListener = listener; if (Interlocked.CompareExchange(ref _flushArmed, 1, 0) != 0) { throw new InvalidOperationException("FlushAsync already armed."); diff --git a/src/ioxide/Connection/Tcp/TcpConnectionPipeReader.cs b/src/ioxide/Connection/Tcp/TcpConnectionPipeReader.cs index c07874dd..bdccf97e 100644 --- a/src/ioxide/Connection/Tcp/TcpConnectionPipeReader.cs +++ b/src/ioxide/Connection/Tcp/TcpConnectionPipeReader.cs @@ -94,16 +94,13 @@ public override ValueTask ReadAsync(CancellationToken cancellationTo : new ValueTask(BuildResult(isCanceled: false)); } - // Everything held was examined - wait for new bytes. The core is reset before the read can - // be armed: off the reactor, the completion may run OnRecvReady before this returns. - _core.Reset(); - short version = _core.Version; - + // Everything held was examined - wait for new bytes. A park resets the core first (OnParking), + // so the version is read after it. while (true) { if (!_conn.TryReadOrPark(this, out RecvSnapshot snapshot)) { - return new ValueTask(this, version); + return new ValueTask(this, _core.Version); } if (Ingest(snapshot) || _connectionClosed || _cancelRequested) @@ -120,6 +117,19 @@ public override ValueTask ReadAsync(CancellationToken cancellationTo } } + // The connection is parking this reader, past its guards: the caller's value task, ready before + // the arm. Not for a re-park from inside OnRecvReady, whose caller is still waiting on the one it has. + void IRecvListener.OnParking() + { + if (!_reparking) + { + _core.Reset(); + } + } + + // OnRecvReady's re-park for the same caller, on the delivering thread. + private bool _reparking; + // The connection's read completed while this reader was parked on it - inline on the reactor for // a recv, on the closing thread for a close. void IRecvListener.OnRecvReady(RecvSnapshot snapshot) @@ -127,18 +137,26 @@ void IRecvListener.OnRecvReady(RecvSnapshot snapshot) if (!Ingest(snapshot) && !_connectionClosed && !_cancelRequested) { // Nothing new: re-arm without completing the caller. - while (true) + _reparking = true; + try { - if (!_conn.TryReadOrPark(this, out snapshot)) - { - return; - } - - if (Ingest(snapshot) || _connectionClosed || _cancelRequested) + while (true) { - break; + if (!_conn.TryReadOrPark(this, out snapshot)) + { + return; + } + + if (Ingest(snapshot) || _connectionClosed || _cancelRequested) + { + break; + } } } + finally + { + _reparking = false; // before the caller resumes: its next read is a new one + } } if (LostToOverflow) diff --git a/src/ioxide/Connection/Tcp/TcpConnectionPipeWriter.cs b/src/ioxide/Connection/Tcp/TcpConnectionPipeWriter.cs index eecd2b1a..8ea202a9 100644 --- a/src/ioxide/Connection/Tcp/TcpConnectionPipeWriter.cs +++ b/src/ioxide/Connection/Tcp/TcpConnectionPipeWriter.cs @@ -51,18 +51,20 @@ public override ValueTask FlushAsync(CancellationToken cancellation _unflushed = 0; - // Reset before the flush is handed over: off the reactor, it can complete before this returns. - _core.Reset(); - short version = _core.Version; - if (_conn.TryFlushOrPark(this)) { return new ValueTask(new FlushResult(isCanceled: false, isCompleted: _completed || _conn.IsClosed)); } - return new ValueTask(this, version); + // The arm reset the core first (OnParking), so the version is read after it. + return new ValueTask(this, _core.Version); } + // The connection is arming this writer's flush, past its guards and with something staged: the + // caller's value task, ready before the arm - off the reactor, the flush can complete before + // FlushAsync returns. + void IFlushListener.OnParking() => _core.Reset(); + // The connection's flush completed while this writer was parked on it - inline on the reactor // for a send, on the releasing thread for a flush its connection closed under. void IFlushListener.OnFlushDone() diff --git a/tests/Ioxide.Tests.Unit/AdapterParkTests.cs b/tests/Ioxide.Tests.Unit/AdapterParkTests.cs index 0df5a359..0ca97503 100644 --- a/tests/Ioxide.Tests.Unit/AdapterParkTests.cs +++ b/tests/Ioxide.Tests.Unit/AdapterParkTests.cs @@ -106,6 +106,108 @@ public static void Register(Runner runner) Assert.True(overlapped >= MinOverlapped, $"only {overlapped} closes landed while a flush was arming: the race never ran"); Assert.True(lost == 0, $"{lost} of {Rounds} flushes never completed, or completed without the close, after it raced their arming"); }, skip: Environment.ProcessorCount < 2); + + // Overlapping calls break the PipeReader/PipeWriter contract, and the second one is refused or + // returns at once - but the first, already waited on, must still complete, as it did when the + // adapters waited on the connection's value tasks. An adapter that readied its own value task + // before the connection had decided anything threw the first one's continuation away. + + runner.Test("pipereader: a second read refused while one is parked leaves the parked one to complete", () => + { + (TcpConnection conn, _) = Connection(); + var reader = new TcpConnectionPipeReader(conn); + ValueTask first = reader.ReadAsync(); + bool resumed = false; + first.GetAwaiter().UnsafeOnCompleted(() => resumed = true); + + Assert.Throws(() => reader.ReadAsync(), "already armed"); + conn.MarkClosed(); + + Assert.True(resumed, "the parked read was never resumed after the second one was refused"); + Assert.True(first.Result.IsCompleted, "the parked read resumed without the close"); + conn.Dispose(); + }); + + runner.Test("pipereader: a wake with nothing new parks the read again without losing its caller", () => + { + // WakeRead delivers whatever is queued, possibly nothing, and with nothing new the reader + // parks again for the same caller - who is still waiting on the value task it was handed, + // so that park must not ready a new one. + (TcpConnection conn, _) = Connection(); + var reader = new TcpConnectionPipeReader(conn); + ValueTask read = reader.ReadAsync(); + bool resumed = false; + read.GetAwaiter().UnsafeOnCompleted(() => resumed = true); + + conn.WakeRead(); + Assert.True(!resumed, "a wake with nothing new completed the read"); + + conn.MarkClosed(); + Assert.True(resumed, "the read parked again after the wake was never resumed"); + Assert.True(read.Result.IsCompleted, "the read resumed without the close"); + conn.Dispose(); + }); + + runner.Test("pipewriter: a second flush refused while one is parked leaves the parked one to complete", () => + { + (TcpConnection conn, _) = Connection(); + var writer = new TcpConnectionPipeWriter(conn); + conn.Write(One); + ValueTask first = writer.FlushAsync(); + bool resumed = false; + first.GetAwaiter().UnsafeOnCompleted(() => resumed = true); + + Assert.Throws(() => writer.FlushAsync(), "already in progress"); + conn.MarkClosed(); // the kernel never took it, so the close releases it + + Assert.True(resumed, "the parked flush was never resumed after the second one was refused"); + Assert.True(first.Result.IsCompleted, "the parked flush resumed without the close"); + conn.Dispose(); + }); + + runner.Test("pipewriter: a flush made after the peer closed, while the kernel holds an earlier one, leaves that one to complete", () => + { + // The second flush is not even refused here: on a closed connection it returns at once. + (TcpConnection conn, _) = Connection(); + var writer = new TcpConnectionPipeWriter(conn); + conn.Write(One); + ValueTask first = writer.FlushAsync(); + bool resumed = false; + first.GetAwaiter().UnsafeOnCompleted(() => resumed = true); + Assert.True(conn.TryClaimFlush(), "the flush was not armed for the reactor to claim"); // the kernel has it now + + conn.MarkClosed(); // which does not release a flush the kernel holds + ValueTask second = writer.FlushAsync(); + Assert.True(second.IsCompleted && second.Result.IsCompleted, "the flush on a closed connection did not return at once, completed"); + + conn.CompleteFlush(); // the send's completion + Assert.True(resumed, "the flush the kernel held was never resumed after its send completed"); + conn.Dispose(); + }); + + runner.Test("pipewriter: a flush with nothing staged leaves an earlier flush's unread result readable", () => + { + (TcpConnection conn, _) = Connection(); + var writer = new TcpConnectionPipeWriter(conn); + conn.Write(One); + ValueTask first = writer.FlushAsync(); + Assert.True(conn.TryClaimFlush(), "the flush was not armed for the reactor to claim"); + conn.CompleteFlush(); // completed, and not yet read + + ValueTask second = writer.FlushAsync(); // nothing staged: over at once + Assert.True(second.IsCompleted, "a flush with nothing staged did not return at once"); + + Assert.True(first.IsCompleted, "the earlier flush was no longer complete"); + Assert.True(!first.Result.IsCompleted, "the earlier flush reported the connection closed"); + conn.Dispose(); + }); + } + + // A connection on a reactor that never runs, as the races above use. + private static (TcpConnection Conn, Reactor Reactor) Connection() + { + var reactor = new Reactor(0, new ServerConfig { ReactorCount = 1, Tcp = null, Udp = null }); + return (new TcpConnection(reactor, -1, 64, 8), reactor); } private static readonly byte[] One = [1]; @@ -227,10 +329,13 @@ public static bool WaitFor(ref int field, int value) } } + // No value task of their own to ready, so OnParking has nothing to do. private sealed class CountingRecvListener : IRecvListener { public int Calls; + public void OnParking() { } + public void OnRecvReady(RecvSnapshot snapshot) => Interlocked.Increment(ref Calls); } @@ -238,6 +343,8 @@ private sealed class CountingFlushListener : IFlushListener { public int Calls; + public void OnParking() { } + public void OnFlushDone() => Interlocked.Increment(ref Calls); }