diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index fdfe1b514..aeabc84ab 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -68,6 +68,8 @@ jobs:
with:
name: packages
path: .nupkgs/*.nupkg
+ include-hidden-files: true # .nupkgs is a dot-folder, which upload-artifact skips by default
+ if-no-files-found: error
- name: NuGet login (OIDC to temp API key)
if: github.event_name == 'release'
uses: NuGet/login@v1
diff --git a/docs/SyncOverAsync.md b/docs/SyncOverAsync.md
index a1b2dd7a2..746f691e0 100644
--- a/docs/SyncOverAsync.md
+++ b/docs/SyncOverAsync.md
@@ -123,9 +123,10 @@ service while the real fix is made. It is not a licence to keep the blocking cal
Two caveats worth knowing before you enable it:
-- it costs a reader and a writer thread **for each node you connect to** (not RESP2 pub/sub connections, which
- stay on the thread-pool; RESP3 does not use separate pub/sub connections), so think about it before enabling
- it against a very wide cluster, where that scales with the number of shards;
+- it costs three threads **for each node you connect to** — two readers (one pulling bytes off the socket, one
+ parsing and dispatching them) and a writer (not RESP2 pub/sub connections, which stay on the thread-pool; RESP3
+ does not use separate pub/sub connections), so think about it before enabling it against a very wide cluster,
+ where that scales with the number of shards;
- it is deliberately opt-in, and set process-wide at startup rather than per-connection.
Neither is meant to be permanent, and the first one especially. Work is in progress on dedicated readers built
diff --git a/src/RESPite/Buffers/CycleBuffer.cs b/src/RESPite/Buffers/CycleBuffer.cs
index c674e6bec..c01cbb8bf 100644
--- a/src/RESPite/Buffers/CycleBuffer.cs
+++ b/src/RESPite/Buffers/CycleBuffer.cs
@@ -571,6 +571,15 @@ public void Recycle()
Memory = default;
RunningIndex = 0;
_flags = Flags.None;
+
+ // This object becomes available - via the *shared, static* _spare slot below - to any
+ // CycleBuffer instance in the process, handed out by Segment.Create() as a pristine segment.
+ // StartTrimCount must not survive that: some callers reach Recycle() without having gone
+ // through Untrim() first (e.g. AppendOrRecycle's search-exhausted path), and Init() (called
+ // from Create()) never touches it either. A stale nonzero value here previously surfaced as
+ // Debug.Assert(leasedStart == 0, "should be zero for a new segment") failing in
+ // GetUncommittedMemory for what looked like a brand new segment.
+ StartTrimCount = 0;
Interlocked.Exchange(ref _spare, this);
DebugAssertValidChain();
}
diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.FeatureFlags.cs b/src/StackExchange.Redis/ConnectionMultiplexer.FeatureFlags.cs
index 73180546f..e361efdd1 100644
--- a/src/StackExchange.Redis/ConnectionMultiplexer.FeatureFlags.cs
+++ b/src/StackExchange.Redis/ConnectionMultiplexer.FeatureFlags.cs
@@ -25,8 +25,9 @@ private enum FeatureFlags
/// out of that queue. It does not *fix* the thread-pool, and nothing here can: it means only that redis
/// traffic keeps flowing while the real problem is found. See docs/SyncOverAsync.md.
///
- /// Costs a reader and a writer thread per connection, so it is worth thinking about before enabling it
- /// against a very wide cluster, where connection counts scale with the number of shards.
+ /// Costs three threads per connection - two readers (one pulling bytes off the socket, one parsing and
+ /// dispatching them) and a writer - so it is worth thinking about before enabling it against a very wide
+ /// cluster, where connection counts scale with the number of shards.
///
///
DedicatedThreads = 2,
diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs
index 8aca7e41c..af7aa18b3 100644
--- a/src/StackExchange.Redis/PhysicalConnection.Read.cs
+++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs
@@ -5,6 +5,7 @@
using System.Diagnostics.CodeAnalysis;
using System.IO;
using System.Net;
+using System.Runtime.ExceptionServices;
using System.Threading;
using System.Threading.Tasks;
using RESPite;
@@ -19,6 +20,45 @@ internal sealed partial class PhysicalConnection
{
private long totalBytesReceived;
+ // Guards every touch of _readBuffer once a background filler is in play (see ReadAllAsync): the filler
+ // and the parse loop run as two independent tasks, and CycleBuffer itself has no internal thread-safety
+ // (by design - see its own remarks), so external mutual exclusion is what makes that safe. Only ever held
+ // for short, synchronous sections - never across an await (Monitor requires the same thread to exit that
+ // entered, which an await's continuation cannot guarantee), and never across frame dispatch (see
+ // ParseAvailableFrames for why that matters).
+ private readonly object _readBufferLock = new();
+
+ // Signals from the filler to the parse loop that there is new committed data (or that the filler has
+ // stopped, successfully or not) worth waking up for. Bounded to a single pending signal - multiple
+ // commits that land before the parser gets back around to waiting just coalesce into one wake, which
+ // is fine: the parser always re-checks the buffer's actual state rather than trusting the signal count.
+ private SemaphoreSlim? _fillSignal;
+ private Exception? _fillerFault;
+
+ // Filler lifecycle flags. Every *decision* based on them is taken under _readBufferLock (the filler's
+ // lease/commit gates, and the stop handshake in OnParseLoopStopped/OnFillerStopped), which is what makes
+ // "whichever of the two stops last recycles the buffer" race-free; volatile so that the cheap reads
+ // outside the lock see a fresh value too.
+ //
+ // _fillerDone: the filler has genuinely, fully stopped - no read in flight, no lease it will still
+ // commit. Set in the filler's own finally, so it is a stronger statement than "was asked to stop".
+ private volatile bool _fillerDone;
+
+ // _readLoopDoomed: the parse loop has given up, for any reason (parse fault, ForceReconnect, the
+ // filler's own clean EOF or fault), and _readBuffer is on its way out. The filler checks this before
+ // taking a lease and again before committing a completed read - once doomed, a read that completes is
+ // simply dropped: the connection is being torn down, and nothing will ever parse it. The read itself is
+ // left to complete naturally rather than raced/cancelled - it doesn't touch shared state until the commit
+ // step this flag gates.
+ private volatile bool _readLoopDoomed;
+
+ // _fillerHandoff: ReadAllSync is transitioning to ReadAllAsync mid-flight and needs its filler thread to
+ // stop *without losing anything*: a read that completes after this is set is still committed - the
+ // connection is alive, and every byte of it is a reply somebody is waiting for - and only then does the
+ // filler stop, leaving _readBuffer intact for ReadAllAsync's own filler to inherit. Dooming instead would
+ // drop that read on the floor, and with it desynchronise every later reply on the connection.
+ private volatile bool _fillerHandoff;
+
internal static PhysicalConnection Dummy(Stream stream, BufferedStreamWriter.WriteMode writeMode = BufferedStreamWriter.WriteMode.Default)
=> new(ioStream: stream, writeMode: writeMode);
@@ -88,44 +128,52 @@ private void StartReadAllAsync(CancellationToken cancellationToken)
private async Task ReadAllAsync(CancellationToken cancellationToken)
{
var tail = _ioStream ?? Stream.Null;
- if (_readStatus is not ReadStatus.TransitioningToAsync)
+ bool transitioning = _readStatus is ReadStatus.TransitioningToAsync;
+ if (!transitioning) // otherwise: preserve the state (and buffer contents) inherited from ReadAllSync
{
- // preserve existing state if transitioning
_readStatus = ReadStatus.Init;
_readState = default;
_readBuffer = CycleBuffer.Create(pool: ReaderBufferPool);
}
+
+ _fillerDone = false;
+ _fillerFault = null;
+ _readLoopDoomed = false;
+ _fillerHandoff = false;
+
+ // When inheriting a buffer from ReadAllSync, whatever its filler committed on the way out was signalled
+ // to *that* loop's semaphore, not this one; start signalled so the first pass parses it (a no-op if
+ // there was nothing) rather than sitting on it until the next reply happens to arrive.
+ var fillSignal = _fillSignal = new SemaphoreSlim(transitioning ? 1 : 0, 1);
+
+ // A background filler keeps reading into the shared buffer independently of the parse loop below,
+ // so a connection that's busy parsing/dispatching still keeps accumulating data underneath it -
+ // this is what lets a big batch build up per parse cycle under load, instead of every cycle
+ // capping out at whatever one physical read returns (which is what a single, self-overlapping
+ // read loop is still limited to: it can hide the *wait* for the next read behind parsing, but not
+ // grow past one read's worth of buffer per cycle, because nothing keeps reading *while* parsing).
+ // _readBufferLock is what makes sharing _readBuffer between the two tasks safe; see its remarks.
+ var fillerTask = Task.Run(() => FillBufferAsync(tail, sync: false, cancellationToken), cancellationToken);
try
{
- int read;
- do
+ while (true)
{
_readStatus = ReadStatus.ReadAsync;
- var buffer = _readBuffer.GetUncommittedMemory();
- var pending = tail.ReadAsync(buffer, cancellationToken);
-#if DEBUG
- bool inline = pending.IsCompleted;
-#endif
- read = await pending.ConfigureAwait(false);
- _readStatus = ReadStatus.UpdateWriteTime;
- UpdateLastReadTime();
-#if DEBUG
- DebugCounters.OnAsyncRead(read, inline);
-#endif
+ await fillSignal.WaitAsync(cancellationToken).ConfigureAwait(false);
+
_readStatus = ReadStatus.TryParseResult;
+ ParseAvailableFrames();
+
+ if (ForceReconnect) break;
+ if (_fillerDone)
+ {
+ if (_fillerFault is { } fault) ExceptionDispatchInfo.Capture(fault).Throw(); // keep the read site's trace
+ break; // clean EOF; the filler already established there is nothing more coming
+ }
}
- // another formatter glitch
- while (CommitAndParseFrames(read) && !ForceReconnect);
_readStatus = ReadStatus.ProcessBufferComplete;
-
- // Volatile.Write(ref _readStatus, ReaderCompleted);
- _readBuffer.Release(); // clean exit, we can recycle
- _readStatus = ReadStatus.RanToCompletion;
- RecordConnectionFailed(ConnectionFailureType.SocketClosed);
- }
- catch (EndOfStreamException) when (_readStatus is ReadStatus.ReadAsync)
- {
+ OnParseLoopStopped(); // stops the filler; recycles the buffer now, or leaves that to the filler
_readStatus = ReadStatus.RanToCompletion;
RecordConnectionFailed(ConnectionFailureType.SocketClosed);
}
@@ -141,50 +189,268 @@ private async Task ReadAllAsync(CancellationToken cancellationToken)
}
finally
{
- _readBuffer = default; // wipe, however we exited
+ // Redundant with the clean-exit path above (harmless: it is idempotent), and the only exit for
+ // every other path - keeps each of them self-contained rather than relying on that having run.
+ OnParseLoopStopped();
+
+ // the filler also observes the same cancellationToken (via the disposing/reconnecting
+ // machinery that drives it elsewhere) and will wind itself down on its own; this just makes
+ // sure a late fault doesn't surface as an unobserved task exception, matching the
+ // fire-and-forget convention used elsewhere on this teardown path.
+ fillerTask.RedisFireAndForget();
}
}
- private void ReadAllSync(CancellationToken cancellationToken)
+ ///
+ /// Independently keeps full: reads, commits under ,
+ /// signals , repeats - with no knowledge of (or dependency on) whether the parse
+ /// loop has caught up. Never has more than one outstanding, uncommitted read at a time, which is what keeps
+ /// this safe without CycleBuffer needing any concurrency-awareness of its own: the existing single-writer
+ /// lease/commit bookkeeping (see CycleBuffer.Commit's remarks on CopyDueToDiscardDuringWrite) already
+ /// tolerates a discard landing on the parse-loop side between this method's lease and its own commit of
+ /// that same lease - that guarantee doesn't depend on which thread the discard happens on, only on nothing
+ /// *else* taking a second, overlapping lease meanwhile, and nothing here ever does.
+ ///
+ ///
+ /// One body serves both readers. With it runs on 's
+ /// dedicated filler thread and issues blocking reads; there is then no await on the path at all, so the
+ /// whole loop runs to completion on that thread and the returned task is already complete. Without it, it
+ /// is 's filler task. Everything else - the lease/commit gates, the stop
+ /// handshake, the signalling - is identical, and deliberately exists exactly once.
+ ///
+ /// A second thread/task is what makes this decoupling possible today: the parse loop's own thread is
+ /// busy dispatching (result matching, TCS completion, user callbacks) while this one keeps reading, and
+ /// that dispatch work is exactly what a single thread can't do *and* keep reading at the same time. If a
+ /// future architecture (e.g. a v4 that moves parsing and dispatch out of the read loop entirely, onto
+ /// whatever consumes the parsed results) shrinks the read loop's own per-cycle cost down to "read plus
+ /// minimal bookkeeping," it may be worth re-testing whether a single thread can then sustain the same
+ /// batching depth without this filler split at all - the mechanism that made the second thread necessary
+ /// here is specifically the competition between dispatch and re-issuing reads on one thread; remove that
+ /// competition and the calculus might change. Untested hypothesis, not a finding - flagged for whoever
+ /// next revisits this once that groundwork exists.
+ ///
+ ///
+ private async Task FillBufferAsync(Stream tail, bool sync, CancellationToken cancellationToken)
{
- var tail = _ioStream ?? Stream.Null;
- _readStatus = ReadStatus.Init;
- _readState = default;
- _readBuffer = CycleBuffer.Create(pool: ReaderBufferPool);
try
{
- int read;
- do
+ while (true)
{
- _readStatus = ReadStatus.ReadSync;
- var buffer = _readBuffer.GetUncommittedMemory();
- cancellationToken.ThrowIfCancellationRequested();
+ Memory buffer;
+ lock (_readBufferLock)
+ {
+ // Checked *inside* the same lock that guards every _readBuffer touch (here and in the
+ // parse loop's own recycle), not just before it: checking outside the lock leaves a
+ // window between "saw false" and "took the lock" for the parse loop to doom-and-recycle
+ // in between, so this thread would then touch a buffer that's gone - or, worse, still get
+ // a lease into a segment that's just been handed back to the shared spare pool for some
+ // *other* connection's CycleBuffer to reuse. Checking under the same lock makes the two
+ // mutually exclusive.
+ if (_readLoopDoomed | _fillerHandoff) return; // nothing in flight: stop before taking a lease
+ buffer = _readBuffer.GetUncommittedMemory();
+ }
+
+ int read;
+ try
+ {
+ if (sync)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
#if NET
- read = tail.Read(buffer.Span);
+ read = tail.Read(buffer.Span);
#else
- read = tail.Read(buffer);
+ read = tail.Read(buffer);
#endif
+ }
+ else
+ {
+ read = await tail.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
+ }
+ }
+ catch (EndOfStreamException)
+ {
+ read = 0; // some streams throw rather than returning 0; treat identically
+ }
- _readStatus = ReadStatus.UpdateWriteTime;
- UpdateLastReadTime();
+ if (read <= 0)
+ {
+ return; // clean EOF - _fillerDone is set in the finally block below
+ }
- DebugCounters.OnSyncRead(read);
- _readStatus = ReadStatus.TryParseResult;
+ bool handoff;
+ lock (_readBufferLock)
+ {
+ // as above - checked again now the read has had time to run, under the same lock as
+ // the Commit() it guards
+ if (_readLoopDoomed) return; // torn down while the read was in flight; nothing will ever parse it
+ _readBuffer.Commit(read);
+
+ // counted here (rather than by the parse loop) because this is where the bytes arrive, and
+ // inside the lock so that the count is never behind a reply the parser has already handed
+ // out; read from arbitrary threads by the timeout/failure diagnostics (see GetBytes),
+ // hence interlocked rather than a plain add
+ Interlocked.Add(ref totalBytesReceived, read);
+ handoff = _fillerHandoff; // decided under the same lock as the commit it must follow
+ }
+
+ UpdateLastReadTime();
+#if DEBUG
+ if (sync)
+ {
+ DebugCounters.OnSyncRead(read);
+ }
+ else
+ {
+ DebugCounters.OnAsyncRead(read, inline: false);
+ }
+#endif
+ SignalFillProgress();
+ if (handoff) return; // committed, so nothing is lost; ReadAllAsync's filler takes over from here
}
- // another formatter glitch
- while (CommitAndParseFrames(read) && !ForceReconnect && !ShouldTransitionToAsync());
+ }
+ catch (OperationCanceledException)
+ {
+ // normal shutdown path (connection tearing down) - not a fault worth reporting
+ }
+ catch (Exception ex)
+ {
+ _fillerFault = ex;
+ }
+ finally
+ {
+ OnFillerStopped(); // recycles the buffer if the parse loop has already gone; else leaves that to it
+ SignalFillProgress(); // wake the parse loop even if nothing new arrived, so it notices completion
+ }
+ }
+
+ // The parse loop and the filler stop independently, and _readBuffer must not be recycled (Release() hands
+ // segments back to the shared static spare, and their leases back to the pool) while either could still
+ // touch it: the filler may have a read in flight into a segment, and the parse loop may still be
+ // dispatching frames out of one. So whichever of the two stops *last* is the one that recycles, decided
+ // under _readBufferLock so that exactly one of them observes the other as already stopped. The filler
+ // stopping is what makes this safe on the common teardown shape: on ForceReconnect it is nearly always
+ // blocked in a read that only RecordConnectionFailed's socket shutdown will unblock - so the parse loop
+ // dooms it, tears the socket down, and the filler recycles on its way out. Merely dropping the field
+ // instead (as an earlier iteration did whenever the filler was still mid-read) leaks the leases of a
+ // custom ConfigurationOptions.ResponseBufferPool on every reconnect.
+ private void OnParseLoopStopped()
+ {
+ lock (_readBufferLock)
+ {
+ _readLoopDoomed = true;
+ if (_fillerDone) ReleaseReadBufferInsideLock();
+ }
+ }
- if (_readStatus is ReadStatus.TransitioningToAsync) return;
+ private void OnFillerStopped()
+ {
+ lock (_readBufferLock)
+ {
+ _fillerDone = true;
+ if (_readLoopDoomed) ReleaseReadBufferInsideLock();
+ }
+ }
- _readStatus = ReadStatus.ProcessBufferComplete;
+ private void ReleaseReadBufferInsideLock()
+ {
+ Debug.Assert(Monitor.IsEntered(_readBufferLock), "must hold _readBufferLock");
+ _readBuffer.Release(); // a no-op once already wiped, so running twice (clean exit + finally) is harmless
+ _readBuffer = default;
+ }
- // Volatile.Write(ref _readStatus, ReaderCompleted);
- _readBuffer.Release(); // clean exit, we can recycle
- _readStatus = ReadStatus.RanToCompletion;
- RecordConnectionFailed(ConnectionFailureType.SocketClosed);
+ // Only ever called from the filler's own single logical thread of control (FillBufferAsync's task, or
+ // ReadAllSync's filler thread running that same method) - never concurrently with itself - which is what
+ // makes the CurrentCount==0-then-Release() check below race-free. That's a property of the call sites,
+ // not of this method, so a future second caller needs to be reasoned about afresh rather than assumed
+ // safe. The semaphore is never disposed (it holds no unmanaged resources), just abandoned when the loop
+ // it belongs to exits, so an old filler's last signal after a handoff simply lands nowhere.
+ private void SignalFillProgress()
+ {
+ var sem = _fillSignal;
+ if (sem is not null && sem.CurrentCount == 0)
+ {
+ try
+ {
+ sem.Release();
+ }
+ catch (SemaphoreFullException)
+ {
+ // would mean the single-caller invariant above was violated by a second, concurrent
+ // caller racing the check-then-Release above; benign here (the semaphore ends up
+ // signaled either way, which was the only goal), but Debug.Fail so a future violation
+ // surfaces immediately in debug/test runs instead of silently vanishing.
+ Debug.Fail("SignalFillProgress: concurrent Release race - single-caller invariant violated");
+ }
}
- catch (EndOfStreamException) when (_readStatus is ReadStatus.ReadSync)
+ }
+
+ ///
+ /// Sync-mode counterpart of the async filler/parser split in : a dedicated
+ /// filler thread (running the same body, in blocking mode) keeps reading into
+ /// independently of this (parser) thread, synchronized the same way
+ /// (, , ). This pays for
+ /// the same batching-depth recovery with a second dedicated thread, on top of the reader thread that
+ /// already spins up for sync/
+ /// connections - a cost callers of that mode already accept in exchange for not risking ThreadPool starvation.
+ ///
+ ///
+ /// Unlike the earlier single-threaded version this replaced, transitioning to async mid-flight (see
+ /// ) needs the filler thread to have genuinely stopped touching
+ /// before starts its own filler on the same buffer -
+ /// two independent fillers racing the same would violate the single-writer
+ /// assumption both designs otherwise rely on. But it must stop *gracefully*: the connection is alive, so a
+ /// read that is in flight at that moment is a real reply that must still be committed, not dropped - hence
+ /// rather than for that path, and the explicit
+ /// filler.Join() to sequence the handoff after it. The stream stays open throughout, so a filler
+ /// blocked in a read unblocks the normal way, on the next reply; this thread waits it out, which is the
+ /// (rare, and accepted) cost of a transition on an idle connection.
+ ///
+ private void ReadAllSync(CancellationToken cancellationToken)
+ {
+ _readStatus = ReadStatus.Init;
+ _readState = default;
+ _readBuffer = CycleBuffer.Create(pool: ReaderBufferPool);
+ _fillerDone = false;
+ _fillerFault = null;
+ _readLoopDoomed = false;
+ _fillerHandoff = false;
+ var fillSignal = _fillSignal = new SemaphoreSlim(0, 1);
+
+ var tail = _ioStream ?? Stream.Null;
+ Thread filler = new Thread(() => FillBufferAsync(tail, sync: true, cancellationToken).GetAwaiter().GetResult())
+ {
+ IsBackground = true,
+
+ // Deliberately Normal, not AboveNormal like the parser thread below: the parser does the
+ // higher-value, latency-sensitive work (result matching, TCS completion, user callbacks), so
+ // under CPU contention it should win any priority-based scheduling contest against this
+ // thread, whose job - reading ahead - can afford to lose a beat without anyone waiting on it.
+ Priority = ThreadPriority.Normal,
+ Name = "SE.Redis Sync Filler",
+ };
+ filler.Start();
+
+ try
{
+ while (true)
+ {
+ _readStatus = ReadStatus.ReadSync;
+ fillSignal.Wait(cancellationToken);
+
+ _readStatus = ReadStatus.TryParseResult;
+ ParseAvailableFrames();
+
+ if (ForceReconnect) break;
+ if (ShouldTransitionToAsync()) return; // finally below hands off once the filler has stopped
+ if (_fillerDone)
+ {
+ if (_fillerFault is { } fault) ExceptionDispatchInfo.Capture(fault).Throw(); // keep the read site's trace
+ break; // clean EOF; the filler already established there is nothing more coming
+ }
+ }
+
+ _readStatus = ReadStatus.ProcessBufferComplete;
+ OnParseLoopStopped(); // stops the filler; recycles the buffer now, or leaves that to the filler
_readStatus = ReadStatus.RanToCompletion;
RecordConnectionFailed(ConnectionFailureType.SocketClosed);
}
@@ -202,11 +468,14 @@ private void ReadAllSync(CancellationToken cancellationToken)
{
if (_readStatus is ReadStatus.TransitioningToAsync)
{
+ // hand off rather than tear down; see this method's remarks
+ _fillerHandoff = true;
+ filler.Join(); // fully stopped - not merely asked to - before another filler touches the buffer
StartReadAllAsync(cancellationToken);
}
else
{
- _readBuffer = default; // wipe, however we exited
+ OnParseLoopStopped(); // as in ReadAllAsync: idempotent, and the only exit for the non-clean paths
}
}
}
@@ -241,41 +510,68 @@ private long GetReadCommittedLength()
}
}
+ // Push-mode (transport) entry point; the pull loops commit from their filler and parse from their own
+ // loop instead. No filler exists in push mode, so no lock is needed around the commit here.
private bool CommitAndParseFrames(int bytesRead)
{
if (bytesRead <= 0)
{
return false;
}
- ref RespScanState state = ref _readState; // avoid a ton of ldarg0
totalBytesReceived += bytesRead;
-#if PARSE_DETAIL
- string src = $"parse {bytesRead}";
- try
-#endif
+ Debug.Assert(
+ bytesRead <= _readBuffer.UncommittedAvailable,
+ $"Insufficient bytes in {nameof(CommitAndParseFrames)}; got {bytesRead}, Available={_readBuffer.UncommittedAvailable}");
+ _readBuffer.Commit(bytesRead);
+
+ ParseAvailableFrames();
+ return true;
+ }
+
+ ///
+ /// Parses and dispatches as many complete RESP frames as are currently committed in ,
+ /// discarding what it consumes. A no-op if there is nothing new since the last call (which can legitimately
+ /// happen when called from the background-filler design in - e.g. a wake that
+ /// turns out to carry no new data, such as the filler's final "I'm done" signal).
+ ///
+ private void ParseAvailableFrames()
+ {
+ ref RespScanState state = ref _readState; // avoid a ton of ldarg0
+
+ // Snapshot the committed region under the lock, then parse and dispatch *outside* it. Holding the lock
+ // across dispatch (result matching, TCS completion, user callbacks) would stall the filler's next
+ // Commit - and so its next read - for the whole pass, capping read-ahead at one outstanding read and
+ // leaving the kernel's socket buffer, not ours, in charge of batching: the opposite of the point of
+ // the split. Parsing outside is safe because the filler only ever *appends*: committed bytes are never
+ // moved or rewritten by it (a discard-during-write copy targets bytes already discarded), and the
+ // sequence/span handed out here stays valid until the discard below - which, like every other touch
+ // of the buffer's own bookkeeping, is back under the lock. The transport push path has no filler and
+ // simply takes the lock uncontended.
+ bool single;
+ ReadOnlySpan fullSpan;
+ ReadOnlySequence fullSequence = default;
+ lock (_readBufferLock)
{
Debug.Assert(_readBuffer.GetCommittedLength() >= 0, "multi-segment running-indices are corrupt");
+ single = _readBuffer.TryGetCommitted(out fullSpan);
+ if (!single) fullSequence = _readBuffer.GetAllCommitted();
+ }
#if PARSE_DETAIL
- src += $" ({_readBuffer.GetCommittedLength()}+{bytesRead}-{state.TotalBytes})";
-#endif
- Debug.Assert(
- bytesRead <= _readBuffer.UncommittedAvailable,
- $"Insufficient bytes in {nameof(CommitAndParseFrames)}; got {bytesRead}, Available={_readBuffer.UncommittedAvailable}");
- _readBuffer.Commit(bytesRead);
-#if PARSE_DETAIL
- src += $",total {_readBuffer.GetCommittedLength()}";
+ string src = $"parse ({(single ? fullSpan.Length : fullSequence.Length)}-{state.TotalBytes})";
+ try
#endif
+ {
var scanner = RespFrameScanner.Default;
OperationStatus status = OperationStatus.NeedMoreData;
- if (_readBuffer.TryGetCommitted(out var fullSpan))
+ if (single)
{
int fullyConsumed = 0;
var toParse = fullSpan.Slice((int)state.TotalBytes); // skip what we've already parsed
OnDetailLog($"parsing {toParse.Length} bytes, single buffer");
- Debug.Assert(!toParse.IsEmpty);
+ if (toParse.IsEmpty) return; // nothing new since the last call
while (true)
{
#if PARSE_DETAIL
@@ -312,19 +608,21 @@ state is
status = OperationStatus.NeedMoreData;
}
OnDetailLog($"discarding {fullyConsumed} bytes");
- _readBuffer.DiscardCommitted(fullyConsumed);
+ lock (_readBufferLock)
+ {
+ _readBuffer.DiscardCommitted(fullyConsumed);
+ }
}
else // the same thing again, but this time with multi-segment sequence
{
- var fullSequence = _readBuffer.GetAllCommitted();
- Debug.Assert(
- fullSequence is { IsEmpty: false, IsSingleSegment: false },
- "non-trivial sequence expected");
+ if (fullSequence.IsEmpty) return; // nothing committed at all yet
+ Debug.Assert(!fullSequence.IsSingleSegment, "non-trivial sequence expected");
long fullyConsumed = 0;
var toParse = fullSequence.Slice((int)state.TotalBytes); // skip what we've already parsed
OnDetailLog($"parsing {toParse.Length} bytes, multi-buffer");
+ if (toParse.IsEmpty) return; // nothing new since the last call
while (true)
{
#if PARSE_DETAIL
@@ -362,7 +660,10 @@ state is
}
OnDetailLog($"discarding {fullyConsumed} bytes");
- _readBuffer.DiscardCommitted(fullyConsumed);
+ lock (_readBufferLock)
+ {
+ _readBuffer.DiscardCommitted(fullyConsumed);
+ }
}
if (status != OperationStatus.NeedMoreData)
@@ -372,13 +673,11 @@ state is
static void ThrowStatus(OperationStatus status) =>
throw new InvalidOperationException($"Unexpected operation status: {status}");
}
-
- return true;
}
#if PARSE_DETAIL
catch (Exception ex)
{
- OnDetailLog($"{nameof(CommitAndParseFrames)}: {ex.Message}");
+ OnDetailLog($"{nameof(ParseAvailableFrames)}: {ex.Message}");
OnDetailLog(src);
if (Debugger.IsAttached) Debugger.Break();
throw new InvalidOperationException($"{src} lead to {ex.Message}", ex);
diff --git a/src/StackExchange.Redis/PhysicalConnection.Write.cs b/src/StackExchange.Redis/PhysicalConnection.Write.cs
index eafc6b5e2..d178f54d8 100644
--- a/src/StackExchange.Redis/PhysicalConnection.Write.cs
+++ b/src/StackExchange.Redis/PhysicalConnection.Write.cs
@@ -73,6 +73,12 @@ internal static BufferedStreamWriter.WriteMode ResolveWriteMode(
///
internal bool IsSyncWriter => _output is { IsSync: true };
+ ///
+ /// Asks a switchable writer to leave sync mode; the reader follows (see ReadAllSync's hand-off to
+ /// ReadAllAsync). Returns false if the writer is not in sync mode, or cannot switch.
+ ///
+ internal bool TransitionToAsync() => _output?.TransitionToAsync() ?? false;
+
private void InitOutput(Stream? stream)
{
if (stream is null) return;
diff --git a/tests/RESPite.Tests/CycleBufferTests.cs b/tests/RESPite.Tests/CycleBufferTests.cs
index df45420c6..499e35cc3 100644
--- a/tests/RESPite.Tests/CycleBufferTests.cs
+++ b/tests/RESPite.Tests/CycleBufferTests.cs
@@ -1,6 +1,7 @@
using System;
using System.Buffers;
using System.Linq;
+using System.Threading;
using RESPite.Buffers;
using Xunit;
@@ -128,4 +129,143 @@ public void CanDiscardSafely(Timing timing)
Assert.Equal(0, buffer.GetCommittedLength());
}
+
+ ///
+ /// Stress test for the filler/parser split used by PhysicalConnection.ReadAllAsync: one real
+ /// thread keeps writing (GetUncommittedMemory+Commit) while another concurrently reads and discards
+ /// (TryGetCommitted/GetAllCommitted+DiscardCommitted), synchronized only by an external lock - exactly
+ /// the pattern the production code relies on, since CycleBuffer itself has no internal thread-safety.
+ /// A running byte pattern makes any loss, duplication, reordering, or corruption immediately visible.
+ ///
+ [Fact]
+ public void ConcurrentFillAndParse_PreservesDataUnderStress()
+ {
+ var buffer = CycleBuffer.Create();
+ var bufferLock = new object();
+ const long TargetBytes = 500_000; // several dozen 8KB segments' worth
+ long totalWritten = 0, totalVerified = 0;
+ Exception? fillerError = null, parserError = null;
+
+ var filler = new Thread(() =>
+ {
+ try
+ {
+ var rng = new Random(12345);
+ long written = 0;
+ while (written < TargetBytes)
+ {
+ Memory mem;
+ lock (bufferLock)
+ {
+ mem = buffer.GetUncommittedMemory();
+ }
+
+ var chunkLen = Math.Min(Math.Min(mem.Length, rng.Next(1, 4001)), (int)Math.Min(TargetBytes - written, int.MaxValue));
+ var span = mem.Span.Slice(0, chunkLen);
+ for (int i = 0; i < chunkLen; i++)
+ {
+ span[i] = unchecked((byte)(written + i));
+ }
+
+ lock (bufferLock)
+ {
+ buffer.Commit(chunkLen);
+ }
+ written += chunkLen;
+ Volatile.Write(ref totalWritten, written);
+ }
+ }
+ catch (Exception ex)
+ {
+ fillerError = ex;
+ }
+ });
+
+ var parser = new Thread(() =>
+ {
+ try
+ {
+ long verified = 0;
+ while (verified < TargetBytes)
+ {
+ bool single;
+ ReadOnlySpan span = default;
+ ReadOnlySequence seq = default;
+ lock (bufferLock)
+ {
+ single = buffer.TryGetCommitted(out span);
+ if (!single) seq = buffer.GetAllCommitted();
+ }
+
+ long consumed;
+ if (single)
+ {
+ consumed = VerifyAndCount(span, verified);
+ }
+ else
+ {
+ consumed = 0;
+ foreach (var segment in seq)
+ {
+ consumed += VerifyAndCount(segment.Span, verified + consumed);
+ }
+ }
+
+ if (consumed > 0)
+ {
+ lock (bufferLock)
+ {
+ buffer.DiscardCommitted(consumed);
+ }
+ verified += consumed;
+ Volatile.Write(ref totalVerified, verified);
+ }
+ else
+ {
+ Thread.Sleep(0); // nothing new yet; yield rather than hot-spin
+ }
+ }
+ }
+ catch (Exception ex)
+ {
+ parserError = ex;
+ }
+
+ static long VerifyAndCount(ReadOnlySpan span, long expectedStart)
+ {
+ for (int i = 0; i < span.Length; i++)
+ {
+ var expectedByte = unchecked((byte)(expectedStart + i));
+ if (span[i] != expectedByte)
+ {
+ throw new InvalidOperationException(
+ $"Data mismatch at offset {expectedStart + i}: expected {expectedByte}, got {span[i]}");
+ }
+ }
+ return span.Length;
+ }
+ });
+
+ filler.Start();
+ parser.Start();
+ var fillerDone = filler.Join(TimeSpan.FromSeconds(10));
+ var parserDone = parser.Join(TimeSpan.FromSeconds(10));
+ if (!fillerDone || !parserDone)
+ {
+ long committedNow;
+ lock (bufferLock)
+ {
+ committedNow = buffer.GetCommittedLength();
+ }
+ Assert.Fail(
+ $"stalled: fillerDone={fillerDone} parserDone={parserDone} totalWritten={Volatile.Read(ref totalWritten)} " +
+ $"totalVerified={Volatile.Read(ref totalVerified)} committedLength={committedNow} " +
+ $"fillerError={fillerError} parserError={parserError}");
+ }
+
+ Assert.Null(fillerError);
+ Assert.Null(parserError);
+ Assert.Equal(TargetBytes, totalWritten);
+ Assert.Equal(TargetBytes, totalVerified);
+ }
}
diff --git a/tests/StackExchange.Redis.Tests/ReadBufferPoolTests.cs b/tests/StackExchange.Redis.Tests/ReadBufferPoolTests.cs
new file mode 100644
index 000000000..dcdd59fb8
--- /dev/null
+++ b/tests/StackExchange.Redis.Tests/ReadBufferPoolTests.cs
@@ -0,0 +1,96 @@
+using System;
+using System.Buffers;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace StackExchange.Redis.Tests;
+
+///
+/// The reader's CycleBuffer rents its segments from ;
+/// this checks that they go back when a connection is torn down, whichever of the parse loop and the
+/// background filler happens to stop last (see PhysicalConnection.OnParseLoopStopped).
+///
+[Collection(NonParallelCollection.Name)]
+public class ReadBufferPoolTests(ITestOutputHelper output) : TestBase(output)
+{
+ protected override string GetConfiguration() => TestConfig.Current.PrimaryServerAndPort + "," + TestConfig.Current.ReplicaServerAndPort;
+
+ [Fact]
+ [Trait(TestCategories.Category, TestCategories.SimulatedConnectionFailure)]
+ public async Task ReadBuffersAreReturnedOnReconnectAndDispose()
+ {
+ var pool = new TrackingMemoryPool();
+ var config = ConfigurationOptions.Parse(GetConfiguration());
+ config.ResponseBufferPool = pool;
+ config.AllowAdmin = true;
+ config.AllowSimulateConnectionFailure = true;
+ config.KeepAlive = 1;
+ config.ReconnectRetryPolicy = new LinearRetry(200);
+
+ try
+ {
+ var conn = await ConnectionMultiplexer.ConnectAsync(config, Writer);
+ await using (conn)
+ {
+ var db = conn.GetDatabase();
+ await db.PingAsync();
+ var steady = pool.Outstanding;
+ Log($"outstanding after connect: {steady} (rented {pool.Rented})");
+ Assert.True(steady > 0, "the reader should be renting from the configured pool");
+
+ var server = conn.GetServer(conn.GetEndPoints()[0]);
+ Assert.SkipUnless(server.CanSimulateConnectionFailure(), "server cannot simulate connection failure");
+
+ // tear the connection down under the reader: its buffers must come back, and the replacement
+ // connection's rentals must not stack on top of leaked ones
+ server.SimulateConnectionFailure(SimulatedFailureType.All);
+ await UntilConditionAsync(TimeSpan.FromSeconds(5), () => server.IsConnected).ForAwait();
+ Assert.True(server.IsConnected, "expected reconnect");
+ await db.PingAsync();
+
+ await UntilConditionAsync(TimeSpan.FromSeconds(5), () => pool.Outstanding <= steady).ForAwait();
+ Log($"outstanding after reconnect: {pool.Outstanding} (rented {pool.Rented})");
+ Assert.True(pool.Outstanding <= steady, $"read buffers leaked across reconnect: {steady} -> {pool.Outstanding}");
+ }
+
+ // and once everything is disposed, nothing should still be out
+ await UntilConditionAsync(TimeSpan.FromSeconds(5), () => pool.Outstanding == 0).ForAwait();
+ Log($"outstanding after dispose: {pool.Outstanding} (rented {pool.Rented})");
+ Assert.Equal(0, pool.Outstanding);
+ }
+ finally
+ {
+ ClearAmbientFailures();
+ }
+ }
+
+ private sealed class TrackingMemoryPool : MemoryPool
+ {
+ private int _rented, _outstanding;
+ public int Rented => Volatile.Read(ref _rented);
+ public int Outstanding => Volatile.Read(ref _outstanding);
+ public override int MaxBufferSize => Shared.MaxBufferSize;
+ public override IMemoryOwner Rent(int minBufferSize = -1)
+ {
+ Interlocked.Increment(ref _rented);
+ Interlocked.Increment(ref _outstanding);
+ return new Lease(this, Shared.Rent(minBufferSize));
+ }
+ protected override void Dispose(bool disposing) { }
+
+ private sealed class Lease(TrackingMemoryPool owner, IMemoryOwner inner) : IMemoryOwner
+ {
+ private IMemoryOwner? _inner = inner;
+ public Memory Memory => _inner?.Memory ?? default;
+ public void Dispose()
+ {
+ if (Interlocked.Exchange(ref _inner, null) is { } tmp)
+ {
+ tmp.Dispose();
+ Interlocked.Decrement(ref owner._outstanding);
+ }
+ }
+ }
+ }
+}
diff --git a/tests/StackExchange.Redis.Tests/RoundTripUnitTests/ReaderLifecycleRoundTrip.cs b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/ReaderLifecycleRoundTrip.cs
new file mode 100644
index 000000000..ba9d80087
--- /dev/null
+++ b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/ReaderLifecycleRoundTrip.cs
@@ -0,0 +1,72 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace StackExchange.Redis.Tests.RoundTripUnitTests;
+
+///
+/// Exercises the filler/parser split in PhysicalConnection.ReadAllAsync/ReadAllSync through
+/// the normal request/response machinery, for behaviours that a simple one-shot round trip does not cover.
+///
+public class ReaderLifecycleRoundTrip(ITestOutputHelper log)
+{
+ private static readonly TimeSpan Patience = TimeSpan.FromSeconds(5);
+
+ [Theory(Timeout = 15_000)]
+ [InlineData(WriteMode.Default)]
+ [InlineData(WriteMode.Sync)]
+ public async Task ReceivedBytesAreCounted(WriteMode mode)
+ {
+ // the received-byte counter feeds the "in:"/"recd=" diagnostics in timeout and connection-failure
+ // messages; it is maintained by the filler, so it must keep working in both reader modes
+ using var conn = new TestConnection(startReading: true, writeMode: mode, log: log);
+ Assert.Equal(0, conn.GetBytesReceived());
+
+ Assert.True(await Bounded(conn.PingAsync()));
+ Assert.Equal("+PONG\r\n".Length, conn.GetBytesReceived());
+
+ Assert.True(await Bounded(conn.PingAsync()));
+ Assert.Equal(2 * "+PONG\r\n".Length, conn.GetBytesReceived());
+ }
+
+ [Fact(Timeout = 30_000)]
+ public async Task SyncToAsyncHandoffLosesNoReplies()
+ {
+ using var conn = new TestConnection(startReading: true, writeMode: WriteMode.Sync, log: log);
+ Assert.True(conn.IsSyncWriter);
+ Assert.True(conn.IsSyncReader);
+ Assert.True(await Bounded(conn.PingAsync(), "sync round trip"));
+
+ // flip the writer; the reader follows on its next wake, i.e. after it has parsed the *next* reply
+ Assert.True(conn.TransitionToAsync(), "writer should accept the transition");
+ SpinUntil(() => !conn.IsSyncWriter, "writer should leave sync mode");
+ Assert.True(await Bounded(conn.PingAsync(), "reply that triggers the transition"));
+
+ // The sync parse loop has now returned and is waiting for its filler thread to stop - and that filler
+ // is blocked in a read, with nothing to read. The next reply lands in *that* read: it must be
+ // committed and parsed, not dropped, or this command never completes and every later reply on the
+ // connection pairs with the wrong command.
+ SpinUntil(() => conn.GetReadStatus() == PhysicalConnection.ReadStatus.TransitioningToAsync, "reader should be handing off");
+ Assert.True(await Bounded(conn.PingAsync(), "reply that arrives mid-handoff"));
+
+ // and from here on the async reader owns the connection
+ SpinUntil(() => conn.GetReadStatus() == PhysicalConnection.ReadStatus.ReadAsync, "async reader should take over");
+ Assert.False(conn.IsSyncReader);
+ Assert.True(await Bounded(conn.PingAsync(), "async round trip"));
+ Assert.True(await Bounded(conn.PingAsync(), "second async round trip"));
+ Assert.Equal(5 * "+PONG\r\n".Length, conn.GetBytesReceived());
+ }
+
+ private static void SpinUntil(Func condition, string what)
+ => Assert.True(SpinWait.SpinUntil(condition, Patience), $"timed out: {what}");
+
+ private static async Task Bounded(Task task, string? what = null)
+ {
+ // Task.WaitAsync is not available on every test TFM, so: race against a delay, but fail with a
+ // message rather than just hanging until the test-level timeout fires
+ var completed = await Task.WhenAny(task, Task.Delay(Patience));
+ Assert.True(ReferenceEquals(completed, task), $"timed out waiting for reply{(what is null ? "" : ": " + what)}");
+ return await task;
+ }
+}
diff --git a/tests/StackExchange.Redis.Tests/RoundTripUnitTests/TestConnection.cs b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/TestConnection.cs
index dd144349a..eb1b3091f 100644
--- a/tests/StackExchange.Redis.Tests/RoundTripUnitTests/TestConnection.cs
+++ b/tests/StackExchange.Redis.Tests/RoundTripUnitTests/TestConnection.cs
@@ -83,6 +83,35 @@ public TestConnection(
public void StartReading() => _physical.StartReading(TestContext.Current.CancellationToken);
+ internal bool IsSyncReader => _physical.IsSyncReader;
+ internal bool IsSyncWriter => _physical.IsSyncWriter;
+ internal PhysicalConnection.ReadStatus GetReadStatus() => _physical.GetReadStatus();
+ internal bool TransitionToAsync() => _physical.TransitionToAsync();
+ internal long GetBytesReceived()
+ {
+ _physical.GetBytes(out _, out var received);
+ return received;
+ }
+
+ ///
+ /// Enqueues a PING (without flushing; the reader does not need the request on the wire to pair the
+ /// reply), feeds inbound, and returns the task for the paired result.
+ ///
+ internal Task PingAsync(string responseResp = "+PONG\r\n")
+ {
+ var box = TaskResultBox.Create(out var tcs, null);
+ var message = Message.Create(-1, CommandFlags.None, RedisCommand.PING);
+ message.SetSource(box, ResultProcessor.DemandPONG);
+ WriteOutbound(message);
+ return Complete(this, tcs.Task, responseResp);
+
+ static async Task Complete(TestConnection conn, Task pending, string responseResp)
+ {
+ await conn.AddInboundAsync(responseResp);
+ return await pending;
+ }
+ }
+
public ReadOnlySpan GetOutboundData() => _stream.GetOutboundData();
public void FlushOutboundData() => _stream.FlushOutboundData();
public ValueTask AddInboundAsync(ReadOnlyMemory data, CancellationToken cancellationToken = default)