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

docs/learn/dev-core.html

@@ -311,7 +311,10 @@ <h2>Connection &middot; <span class="dim">Connection/Connection.cs (+ .Read, .Wr
       <li><b>Lost-wakeup-free read</b> - <code>ReadAsync</code> uses an <code>_armed</code> flag plus
       a sticky <code>_pending</code>, with a re-check between arming and returning the task, so a
       completion that races the arm is never lost. Continuations are synchronous
-      (<code>RunContinuationsAsynchronously = false</code>) - that's what makes resume inline.</li>
+      (<code>RunContinuationsAsynchronously = false</code>) - that's what makes resume inline. An
+      adapter parked through <code>TryReadOrPark</code> / <code>TryFlushOrPark</code> 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.</li>
       <li><b>Read snapshot</b> - <code>ReadAsync</code> returns a <code>RecvSnapshot</code> (a tail
       marker + closed flag). The handler then drains slices up to that tail with
       <code>TryGetItem</code> / <code>GetSnapshotMemories</code>, and returns their buffers with
@@ -402,13 +405,16 @@ <h2>BCL bridges</h2>
     <ul>
       <li><b>TcpConnectionPipeReader</b> (<code>Connection/Tcp/TcpConnectionPipeReader.cs</code>) - adapts the
       raw read API to <code>System.IO.Pipelines.PipeReader</code>, 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 <code>AdvanceTo</code> trims the consumed
-      front and returns those buffers. It honors <code>examined</code> - when everything held has been
-      examined, it waits for new bytes instead of returning the same data.</li>
+      parked read is the connection's own read listener (<code>IRecvListener</code>) - 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 <code>AdvanceTo</code> trims the consumed front and returns
+      those buffers. It honors <code>examined</code> - when everything held has been examined, it
+      waits for new bytes instead of returning the same data.</li>
       <li><b>TcpConnectionPipeWriter</b> / <b>TcpConnectionStream</b> / <b>TcpConnectionDualPipe</b>
-      (<code>Connection/Tcp/</code>) - a <code>PipeWriter</code> over the write slab, a
-      <code>Stream</code> (so <code>SslStream</code> and other BCL code can ride a connection - see
+      (<code>Connection/Tcp/</code>) - a <code>PipeWriter</code> over the write slab, parked on the
+      connection's flush the same way (<code>IFlushListener</code>), a <code>Stream</code> (so
+      <code>SslStream</code> and other BCL code can ride a connection - see
       <a href="dev-tls.html">TLS internals</a>), and a duplex pipe pairing.</li>
     </ul>
 <p class="src"><b>TcpConnectionStream.cs</b> · 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)</p>

src/ioxide/Connection/Tcp/IFlushListener.cs

@@ -0,0 +1,23 @@
+namespace ioxide;
+
+/// <summary>
+/// An adapter parked on a connection's flush in place of an awaiter - the write side of
+/// <see cref="IRecvListener"/>, for <see cref="TcpConnectionPipeWriter"/>.
+/// </summary>
+/// <remarks>
+/// Parked through <see cref="TcpConnection.TryFlushOrPark"/>. <see cref="OnFlushDone"/> 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.
+/// </remarks>
+internal interface IFlushListener
+{
+    /// <summary>
+    /// The connection is about to arm this listener's flush: the in-progress guard has passed and
+    /// something is staged. As <see cref="IRecvListener.OnParking"/>: 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.
+    /// </summary>
+    void OnParking();
+
+    void OnFlushDone();
+}

src/ioxide/Connection/Tcp/IRecvListener.cs

@@ -0,0 +1,29 @@
+namespace ioxide;
+
+/// <summary>
+/// An adapter parked on a connection's read in place of an awaiter.
+/// </summary>
+/// <remarks>
+/// <see cref="TcpConnectionPipeReader"/> used to park on <see cref="TcpConnection.ReadAsync"/>'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 <see cref="TcpConnection.TryReadOrPark"/> 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 <see cref="TcpConnection.ReadAsync"/>'s, with one addition: whichever side
+/// takes the arm back delivers, so <see cref="OnRecvReady"/> runs exactly once per park, on the thread
+/// that completed the read - the reactor's for a recv, the closing one for a close.
+/// </remarks>
+internal interface IRecvListener
+{
+    /// <summary>
+    /// 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
+    /// <see cref="OnRecvReady"/> before the park returns. Not called for a read served at once.
+    /// </summary>
+    void OnParking();
+
+    void OnRecvReady(RecvSnapshot snapshot);
+}

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
 
     /// <summary>
@@ -83,6 +87,75 @@ public ValueTask<RecvSnapshot> ReadAsync()
         return new ValueTask<RecvSnapshot>(this, (short)gen);
     }
 
+    /// <summary>
+    /// <see cref="ReadAsync"/> 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 <paramref name="listener"/>, once, on the completing thread.
+    /// </summary>
+    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);
+    }
+
     /// <summary>
     /// Take the next item of a snapshot. This is a DEQUEUE and it transfers ownership: every item
     /// with <c>HasBuffer</c> must be handed back with <see cref="ReturnBuffer"/>, 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
         {

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;
 
     /// <summary>
@@ -105,6 +109,32 @@ internal ValueTask FlushIfIdleAsync()
         return FlushCore();
     }
 
+    /// <summary>
+    /// <see cref="FlushAsync"/> 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 <paramref name="listener"/>, once, on
+    /// its own thread.
+    /// </summary>
+    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. <paramref name="listener"/>
+    // is the adapter parking on the flush, or null for FlushAsync's awaiter; <paramref name="gen"/>
+    // 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();
         }
     }
 

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;