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..cf9459b8 --- /dev/null +++ b/src/ioxide/Connection/Tcp/IFlushListener.cs @@ -0,0 +1,23 @@ +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 +{ + /// + /// 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 new file mode 100644 index 00000000..fd560616 --- /dev/null +++ b/src/ioxide/Connection/Tcp/IRecvListener.cs @@ -0,0 +1,29 @@ +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 +{ + /// + /// 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 4486d1b9..93e9187a 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,75 @@ 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 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) + { + throw new InvalidOperationException("ReadAsync already armed."); + } + + // 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) + { + _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 +230,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 +261,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..2d154834 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,32 @@ 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."); + } + + return !ArmFlush(listener, out _); + } + // 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 +153,15 @@ 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(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, 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++) @@ -135,12 +171,17 @@ 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. + // 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."); @@ -158,7 +199,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 +210,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,10 +222,25 @@ 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. internal void CompleteFlush() { @@ -203,7 +259,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..bdccf97e 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,16 @@ 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. A park resets the core first (OnParking), + // so the version is read after it. 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); } - if (Ingest(pending.Result) || _connectionClosed || _cancelRequested) + if (Ingest(snapshot) || _connectionClosed || _cancelRequested) { if (LostToOverflow) { @@ -123,30 +117,46 @@ public override ValueTask ReadAsync(CancellationToken cancellationTo } } - // Completion of a parked conn.ReadAsync - runs inline on the reactor. - private void OnRecvReady() + // 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() { - RecvSnapshot snapshot = _pendingRead.GetResult(); + 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) + { if (!Ingest(snapshot) && !_connectionClosed && !_cancelRequested) { // Nothing new: re-arm without completing the caller. - while (true) + _reparking = true; + try { - ValueTask pending = _conn.ReadAsync(); - - if (!pending.IsCompletedSuccessfully) + while (true) { - _pendingRead = pending.GetAwaiter(); - _pendingRead.UnsafeOnCompleted(_onRecvReady); - return; - } - - if (Ingest(pending.Result) || _connectionClosed || _cancelRequested) - { - 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) @@ -190,8 +200,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..8ea202a9 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,25 @@ public override ValueTask FlushAsync(CancellationToken cancellation } _unflushed = 0; - ValueTask inner = _conn.FlushAsync(); - if (inner.IsCompletedSuccessfully) + if (_conn.TryFlushOrPark(this)) { return new ValueTask(new FlushResult(isCanceled: false, isCompleted: _completed || _conn.IsClosed)); } - _core.Reset(); - _pendingFlush = inner.GetAwaiter(); - _pendingFlush.UnsafeOnCompleted(_onFlushDone); + // The arm reset the core first (OnParking), so the version is read after it. return new ValueTask(this, _core.Version); } - // Completion of a parked flush - runs inline on the reactor. - private void OnFlushDone() - { - _pendingFlush.GetResult(); + // 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() + { 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..0ca97503 --- /dev/null +++ b/tests/Ioxide.Tests.Unit/AdapterParkTests.cs @@ -0,0 +1,356 @@ +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); + + // 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]; + + /// + /// 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(); + } + } + } + } + + // 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); + } + + private sealed class CountingFlushListener : IFlushListener + { + public int Calls; + + public void OnParking() { } + + 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(); }