From 0ce150be41615611cef8baf722165d5c1625e4dd Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Tue, 22 Sep 2026 09:40:20 +0100 Subject: [PATCH 1/8] Decouple the async read loop's socket fill from parsing StackExchange.Redis v3's async PhysicalConnection read loop reads, then parses/dispatches, then reads again - all on one path. Under concurrent load this caps how much data accumulates per parse pass at whatever a single physical read returns (~5KB observed), versus the pre-RESPite Pipe-backed reader's ~29-31KB, because nothing kept reading while the old design's consumer was busy. That gap is the root cause of a concurrent bulk-string (GET) throughput ceiling that integer replies don't hit, reported against 3.3.0 vs 2.9.11. ReadAllAsync now runs a background filler task that keeps reading into the shared CycleBuffer independently of the parse loop, synchronized by a lock plus a bounded semaphore wake signal. This needed no changes to CycleBuffer's own single-writer semantics: its existing lease/discard tolerance (already covered by CanDiscardSafely) is sufficient once the filler and parser are genuinely mutually excluded and the filler never holds more than one outstanding, uncommitted read at a time. Also fixes a latent bug this surfaced: Segment.Recycle() never reset StartTrimCount before returning the segment to the shared static spare slot, so a segment recycled via AppendOrRecycle's search-exhausted path (the only path that skips Untrim()) could later be handed out by Segment.Create() as a "fresh" segment with a stale trim offset. Pre- existing and unrelated to this change; only the new stress test's heavy segment churn hit it reliably enough to notice. Bytes committed per parse cycle: ~5KB -> ~23-28KB, closing most of the gap to the pre-RESPite ~29-31KB. Verified via a new CycleBuffer concurrency stress test (two real threads, lock-synchronized, running byte-pattern verification), the full RESPite.Tests and StackExchange.Redis.Tests suites, and a continuous byte-for-byte correctness check built into ad hoc load testing (zero corruption across millions of real GETs). --- src/RESPite/Buffers/CycleBuffer.cs | 9 + .../PhysicalConnection.Read.cs | 189 ++++++++++++++---- tests/RESPite.Tests/CycleBufferTests.cs | 140 +++++++++++++ 3 files changed, 302 insertions(+), 36 deletions(-) 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/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index 8aca7e41c..7cc6985a0 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -19,6 +19,21 @@ internal sealed partial class PhysicalConnection { private long totalBytesReceived; + // Guards every touch of _readBuffer/_readState 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). + 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 volatile bool _fillerDone; + private Exception? _fillerFault; + internal static PhysicalConnection Dummy(Stream stream, BufferedStreamWriter.WriteMode writeMode = BufferedStreamWriter.WriteMode.Default) => new(ioStream: stream, writeMode: writeMode); @@ -95,32 +110,41 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) _readState = default; _readBuffer = CycleBuffer.Create(pool: ReaderBufferPool); } + _fillerDone = false; + _fillerFault = null; + var fillSignal = _fillSignal = new SemaphoreSlim(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, 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; + lock (_readBufferLock) + { + ParseAvailableFrames(); + } + + if (ForceReconnect) break; + if (_fillerDone) + { + if (_fillerFault is { } fault) throw fault; + 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 + lock (_readBufferLock) { _readBuffer.Release(); } // clean exit, we can recycle _readStatus = ReadStatus.RanToCompletion; RecordConnectionFailed(ConnectionFailureType.SocketClosed); } @@ -141,10 +165,95 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) } finally { + // the filler observes the same cancellationToken (via the disposing/reconnecting machinery + // that also drives it elsewhere) and will wind itself down; just make 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(); _readBuffer = default; // wipe, however we exited } } + /// + /// Independently keeps full: reads, commits under , + /// signals , repeats - with no knowledge of (or dependency on) whether the parse + /// loop in 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. + /// + private async Task FillBufferAsync(Stream tail, CancellationToken cancellationToken) + { + try + { + while (true) + { + Memory buffer; + lock (_readBufferLock) + { + buffer = _readBuffer.GetUncommittedMemory(); + } + + int read; + try + { + read = await tail.ReadAsync(buffer, cancellationToken).ConfigureAwait(false); + } + catch (EndOfStreamException) + { + read = 0; // some streams throw rather than returning 0; treat identically + } + + if (read <= 0) + { + return; // clean EOF - _fillerDone is set in the finally block below + } + + lock (_readBufferLock) + { + _readBuffer.Commit(read); + } + UpdateLastReadTime(); +#if DEBUG + DebugCounters.OnAsyncRead(read, inline: false); +#endif + SignalFillProgress(); + } + } + catch (OperationCanceledException) + { + // normal shutdown path (connection tearing down) - not a fault worth reporting + } + catch (Exception ex) + { + _fillerFault = ex; + } + finally + { + _fillerDone = true; + SignalFillProgress(); // wake the parse loop even if nothing new arrived, so it notices completion + } + } + + private void SignalFillProgress() + { + var sem = _fillSignal; + if (sem is not null && sem.CurrentCount == 0) + { + try + { + sem.Release(); + } + catch (ObjectDisposedException) + { + // torn down from under us during shutdown; the parse loop has already stopped caring + } + } + } + private void ReadAllSync(CancellationToken cancellationToken) { var tail = _ioStream ?? Stream.Null; @@ -241,31 +350,41 @@ private long GetReadCommittedLength() } } - private bool CommitAndParseFrames(int bytesRead) + private bool CommitAndParseFrames(int bytesRead, bool alreadyCommitted = false) { 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 + if (!alreadyCommitted) { - Debug.Assert(_readBuffer.GetCommittedLength() >= 0, "multi-segment running-indices are corrupt"); -#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); + } + + 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 + Debug.Assert(_readBuffer.GetCommittedLength() >= 0, "multi-segment running-indices are corrupt"); #if PARSE_DETAIL - src += $",total {_readBuffer.GetCommittedLength()}"; + string src = $"parse ({_readBuffer.GetCommittedLength()}-{state.TotalBytes})"; + try #endif + { var scanner = RespFrameScanner.Default; OperationStatus status = OperationStatus.NeedMoreData; @@ -275,7 +394,7 @@ private bool CommitAndParseFrames(int bytesRead) 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 @@ -317,14 +436,14 @@ state is 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 @@ -372,13 +491,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/tests/RESPite.Tests/CycleBufferTests.cs b/tests/RESPite.Tests/CycleBufferTests.cs index df45420c6..7de728ef2 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[..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); + } } From 250df8e418edb539c6e5a665ab361c7b9f559838 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Tue, 22 Sep 2026 14:52:28 +0100 Subject: [PATCH 2/8] Guard the filler against a torn-down _readBuffer on parser fault If the parse loop gives up (a parse fault, ForceReconnect, or the filler's own clean EOF), the filler could still be mid-cycle: it might commit a read into _readBuffer after the parse loop's finally block has already wiped it, or race that unsynchronized wipe outright, since _readBuffer is a multi-field struct. _readLoopDoomed is set before the wipe (now itself lock-protected) and checked by the filler both before requesting its next buffer and after its current read completes, before combining (committing) it - so it stops cleanly rather than touching a buffer that's gone or racing the wipe itself. --- .../PhysicalConnection.Read.cs | 34 ++++++++++++++++--- 1 file changed, 29 insertions(+), 5 deletions(-) diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index 7cc6985a0..11e4c4f15 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -34,6 +34,14 @@ internal sealed partial class PhysicalConnection private volatile bool _fillerDone; private Exception? _fillerFault; + // Set (before _readBuffer is touched again) the moment the parse loop gives up, for any reason - a + // parse fault, ForceReconnect, or the filler's own clean EOF. The filler checks this after each read + // completes, before combining (committing) it into _readBuffer: once doomed, _readBuffer may already + // be torn down (see the finally block in ReadAllAsync), so the filler must stop rather than touch 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; + internal static PhysicalConnection Dummy(Stream stream, BufferedStreamWriter.WriteMode writeMode = BufferedStreamWriter.WriteMode.Default) => new(ioStream: stream, writeMode: writeMode); @@ -112,6 +120,7 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) } _fillerDone = false; _fillerFault = null; + _readLoopDoomed = false; var fillSignal = _fillSignal = new SemaphoreSlim(0, 1); // A background filler keeps reading into the shared buffer independently of the parse loop below, @@ -165,12 +174,23 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) } finally { - // the filler observes the same cancellationToken (via the disposing/reconnecting machinery - // that also drives it elsewhere) and will wind itself down; just make sure a late fault - // doesn't surface as an unobserved task exception, matching the fire-and-forget convention - // used elsewhere on this teardown path. + // set *before* touching _readBuffer below: the filler checks this after each read completes, + // before committing into _readBuffer, so it never combines into (or races) a buffer that's + // about to be - or already has been - wiped here. + _readLoopDoomed = true; + + // 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(); - _readBuffer = default; // wipe, however we exited + + // lock-protected because the filler's own reads/writes of _readBuffer are - an unsynchronized + // write here would race a multi-field struct against them, not just risk a missed signal. + lock (_readBufferLock) + { + _readBuffer = default; // wipe, however we exited + } } } @@ -191,6 +211,8 @@ private async Task FillBufferAsync(Stream tail, CancellationToken cancellationTo { while (true) { + if (_readLoopDoomed) return; // the parse loop already gave up; _readBuffer may be gone + Memory buffer; lock (_readBufferLock) { @@ -207,6 +229,8 @@ private async Task FillBufferAsync(Stream tail, CancellationToken cancellationTo read = 0; // some streams throw rather than returning 0; treat identically } + if (_readLoopDoomed) return; // as above - checked again now the read has had time to run + if (read <= 0) { return; // clean EOF - _fillerDone is set in the finally block below From 0ffc81aa28cd3c03da6e12f611a8359164b98912 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Tue, 22 Sep 2026 15:01:17 +0100 Subject: [PATCH 3/8] Fix net481 build: avoid the range operator in the new stress test System.Range/System.Index aren't available on net481, and RESPite.Tests multi-targets it. mem.Span[..chunkLen] -> mem.Span.Slice(0, chunkLen). Verified: dotnet build Build.csproj -c Release /p:CI=true (all TFMs, whole solution) now succeeds; previously this only surfaced on CI (net10.0-only local runs don't build net481). --- tests/RESPite.Tests/CycleBufferTests.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/RESPite.Tests/CycleBufferTests.cs b/tests/RESPite.Tests/CycleBufferTests.cs index 7de728ef2..499e35cc3 100644 --- a/tests/RESPite.Tests/CycleBufferTests.cs +++ b/tests/RESPite.Tests/CycleBufferTests.cs @@ -161,7 +161,7 @@ public void ConcurrentFillAndParse_PreservesDataUnderStress() } var chunkLen = Math.Min(Math.Min(mem.Length, rng.Next(1, 4001)), (int)Math.Min(TargetBytes - written, int.MaxValue)); - var span = mem.Span[..chunkLen]; + var span = mem.Span.Slice(0, chunkLen); for (int i = 0; i < chunkLen; i++) { span[i] = unchecked((byte)(written + i)); From 1ef66b37baef6d909b2400222bbaa9f3afcf8120 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Tue, 22 Sep 2026 15:45:09 +0100 Subject: [PATCH 4/8] Extend the filler/parser split to the sync (DedicatedThreads) reader ReadAllSync had the same single-threaded batching ceiling as the pre-fix async loop, for the same reason: nothing kept reading while this thread was busy parsing/dispatching. Give it its own background filler thread, mirroring ReadAllAsync/FillBufferAsync, so DedicatedThreads connections get the same batching-depth recovery - on top of the reader thread that mode already pays for, not layered as a separate cost. Preserves the existing mid-flight sync->async transition (ShouldTransitionToAsync): the filler thread is explicitly joined before handing the buffer off to ReadAllAsync's own filler, since two independent fillers racing the same CycleBuffer would violate the single-writer assumption both designs rely on. Leaves a note on FillBufferAsync flagging an open question for a future architecture that moves parsing/dispatch out of the read loop entirely: that might recover the same batching depth without needing a second thread at all, since the second thread exists specifically to stop dispatch work from competing with re-issuing reads on one thread. --- .../PhysicalConnection.Read.cs | 153 +++++++++++++++--- 1 file changed, 131 insertions(+), 22 deletions(-) diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index 11e4c4f15..b7da1341d 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -205,6 +205,18 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) /// the discard happens on, only on nothing *else* taking a second, overlapping lease meanwhile, and /// nothing here ever does. /// + /// + /// 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, CancellationToken cancellationToken) { try @@ -278,41 +290,67 @@ private void SignalFillProgress() } } + /// + /// Sync-mode counterpart of the async filler/parser split in and + /// : a dedicated filler thread 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. Hence the explicit filler.Join() below, which is + /// otherwise unnecessary (the lock plus already make the plain wipe-on-exit + /// path safe, matching 's finally block, which doesn't wait for its filler task). + /// private void ReadAllSync(CancellationToken cancellationToken) { - var tail = _ioStream ?? Stream.Null; _readStatus = ReadStatus.Init; _readState = default; _readBuffer = CycleBuffer.Create(pool: ReaderBufferPool); + _fillerDone = false; + _fillerFault = null; + _readLoopDoomed = false; + var fillSignal = _fillSignal = new SemaphoreSlim(0, 1); + + var tail = _ioStream ?? Stream.Null; + Thread filler = new Thread(() => FillBufferSync(tail, cancellationToken)) + { + IsBackground = true, + Priority = ThreadPriority.AboveNormal, + Name = "SE.Redis Sync Filler", + }; + filler.Start(); + try { - int read; - do + while (true) { _readStatus = ReadStatus.ReadSync; - var buffer = _readBuffer.GetUncommittedMemory(); - cancellationToken.ThrowIfCancellationRequested(); -#if NET - read = tail.Read(buffer.Span); -#else - read = tail.Read(buffer); -#endif - - _readStatus = ReadStatus.UpdateWriteTime; - UpdateLastReadTime(); + fillSignal.Wait(cancellationToken); - DebugCounters.OnSyncRead(read); _readStatus = ReadStatus.TryParseResult; - } - // another formatter glitch - while (CommitAndParseFrames(read) && !ForceReconnect && !ShouldTransitionToAsync()); + lock (_readBufferLock) + { + ParseAvailableFrames(); + } - if (_readStatus is ReadStatus.TransitioningToAsync) return; + if (ForceReconnect) break; + if (ShouldTransitionToAsync()) return; // finally below hands off once the filler has stopped + if (_fillerDone) + { + if (_fillerFault is { } fault) throw fault; + break; // clean EOF; the filler already established there is nothing more coming + } + } _readStatus = ReadStatus.ProcessBufferComplete; - - // Volatile.Write(ref _readStatus, ReaderCompleted); - _readBuffer.Release(); // clean exit, we can recycle + lock (_readBufferLock) { _readBuffer.Release(); } // clean exit, we can recycle _readStatus = ReadStatus.RanToCompletion; RecordConnectionFailed(ConnectionFailureType.SocketClosed); } @@ -333,15 +371,86 @@ private void ReadAllSync(CancellationToken cancellationToken) } finally { + _readLoopDoomed = true; if (_readStatus is ReadStatus.TransitioningToAsync) { + // must be fully stopped - not just signalled - before ReadAllAsync starts its own filler + // on the same _readBuffer; see this method's remarks. + filler.Join(); StartReadAllAsync(cancellationToken); } else { - _readBuffer = default; // wipe, however we exited + lock (_readBufferLock) + { + _readBuffer = default; // wipe, however we exited + } + } + } + } + + /// + /// Blocking-thread counterpart of - see its remarks, which apply unchanged + /// (single outstanding read at a time, no reservation needed). + /// + private void FillBufferSync(Stream tail, CancellationToken cancellationToken) + { + try + { + while (true) + { + if (_readLoopDoomed) return; // the parse loop already gave up; _readBuffer may be gone + + Memory buffer; + lock (_readBufferLock) + { + buffer = _readBuffer.GetUncommittedMemory(); + } + + int read; + try + { + cancellationToken.ThrowIfCancellationRequested(); +#if NET + read = tail.Read(buffer.Span); +#else + read = tail.Read(buffer); +#endif + } + catch (EndOfStreamException) + { + read = 0; // some streams throw rather than returning 0; treat identically + } + + if (_readLoopDoomed) return; // as above - checked again now the read has had time to run + + if (read <= 0) + { + return; // clean EOF - _fillerDone is set in the finally block below + } + + lock (_readBufferLock) + { + _readBuffer.Commit(read); + } + UpdateLastReadTime(); + DebugCounters.OnSyncRead(read); + SignalFillProgress(); } } + catch (OperationCanceledException) + { + // normal shutdown path (connection tearing down) - not a fault worth reporting + } + catch (Exception ex) + { + _fillerFault = ex; + } + finally + { + _fillerDone = true; + SignalFillProgress(); // wake the parse loop even if nothing new arrived, so it notices completion + } } private bool ShouldTransitionToAsync() From 0e2d5f75ea286724c917f9cc0f5f993337583d24 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Tue, 22 Sep 2026 16:43:43 +0100 Subject: [PATCH 5/8] Address review: close a real TOCTOU race, and a scheduling priority mismatch _readLoopDoomed was checked outside _readBufferLock before touching _readBuffer in both fillers. Between the check and taking the lock, the parse loop's own exit path could doom-and-Release() (or doom-and-wipe) in between, so the filler could then get a lease from an already-released buffer - and since CycleBuffer.Release() hands segments back to the shared *static* spare slot regardless of any outstanding lease, that lease could point at memory another connection's CycleBuffer now owns. Moving the check inside the same lock that guards the touch it gates makes the two mutually exclusive. Also closed a related gap the ForceReconnect exit path had on both readers: Release() was called before _readLoopDoomed was set (only finally set it), so a ForceReconnect break - unlike the filler's own clean-EOF break, where the filler has already stopped by construction - could still race a genuinely-running filler. Doom is now set before Release() on that path too. SignalFillProgress's CurrentCount==0-then-Release() is safe today only because it's never called concurrently with itself; documented that invariant and added a SemaphoreFullException catch with Debug.Fail, so a future second caller fails loudly in debug/test instead of the exception silently vanishing into _fillerFault. The sync filler thread was inheriting ThreadPriority.AboveNormal from the parser thread it was copied from. The parser does the higher-value, latency-sensitive work (dispatch, TCS completion, user callbacks) and should win any priority-based scheduling contest under load; dropped the filler to Normal. --- .../PhysicalConnection.Read.cs | 63 +++++++++++++++---- 1 file changed, 51 insertions(+), 12 deletions(-) diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index b7da1341d..54f778ac0 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -152,6 +152,11 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) } } + // ForceReconnect can land here with the filler still genuinely mid-cycle (unlike the + // _fillerDone break above, where the filler has already fully stopped by construction) - set + // this *before* Release() below, not just in finally, or the filler could still be holding an + // outstanding lease into a segment Release() is about to hand back to the shared spare pool. + _readLoopDoomed = true; _readStatus = ReadStatus.ProcessBufferComplete; lock (_readBufferLock) { _readBuffer.Release(); } // clean exit, we can recycle _readStatus = ReadStatus.RanToCompletion; @@ -174,9 +179,12 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) } finally { - // set *before* touching _readBuffer below: the filler checks this after each read completes, - // before committing into _readBuffer, so it never combines into (or races) a buffer that's - // about to be - or already has been - wiped here. + // set *before* touching _readBuffer below: the filler checks this (now inside the same lock + // as its own touches - see FillBufferAsync) before combining a read into _readBuffer, so it + // never combines into (or races) a buffer that's about to be - or already has been - wiped + // here. Redundant with the clean-exit path above, which sets it earlier for the same reason; + // harmless to repeat, and keeps every exit path (including ones that throw straight into a + // catch below, skipping that code) self-contained rather than relying on it having run. _readLoopDoomed = true; // the filler also observes the same cancellationToken (via the disposing/reconnecting @@ -223,11 +231,17 @@ private async Task FillBufferAsync(Stream tail, CancellationToken cancellationTo { while (true) { - if (_readLoopDoomed) return; // the parse loop already gave up; _readBuffer may be gone - Memory buffer; lock (_readBufferLock) { + // Checked *inside* the same lock that guards every _readBuffer touch (here and in the + // parse loop's own wipe/Release), 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-wipe + // (or doom-and-Release) in between, so this thread would then touch a buffer that's + // gone - or, worse for Release(), 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) return; // the parse loop already gave up; _readBuffer may be gone buffer = _readBuffer.GetUncommittedMemory(); } @@ -241,8 +255,6 @@ private async Task FillBufferAsync(Stream tail, CancellationToken cancellationTo read = 0; // some streams throw rather than returning 0; treat identically } - if (_readLoopDoomed) return; // as above - checked again now the read has had time to run - if (read <= 0) { return; // clean EOF - _fillerDone is set in the finally block below @@ -250,6 +262,9 @@ private async Task FillBufferAsync(Stream tail, CancellationToken cancellationTo 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; _readBuffer.Commit(read); } UpdateLastReadTime(); @@ -274,6 +289,10 @@ private async Task FillBufferAsync(Stream tail, CancellationToken cancellationTo } } + // Only ever called from the filler's own single logical thread of control (FillBufferAsync's task, + // or FillBufferSync's thread) - 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. private void SignalFillProgress() { var sem = _fillSignal; @@ -287,6 +306,14 @@ private void SignalFillProgress() { // torn down from under us during shutdown; the parse loop has already stopped caring } + 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"); + } } } @@ -322,7 +349,12 @@ private void ReadAllSync(CancellationToken cancellationToken) Thread filler = new Thread(() => FillBufferSync(tail, cancellationToken)) { IsBackground = true, - Priority = ThreadPriority.AboveNormal, + + // 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(); @@ -349,6 +381,11 @@ private void ReadAllSync(CancellationToken cancellationToken) } } + // ForceReconnect can land here with the filler still genuinely mid-cycle (unlike the + // _fillerDone break above, where the filler has already fully stopped by construction) - set + // this *before* Release() below, not just in finally, or the filler could still be holding an + // outstanding lease into a segment Release() is about to hand back to the shared spare pool. + _readLoopDoomed = true; _readStatus = ReadStatus.ProcessBufferComplete; lock (_readBufferLock) { _readBuffer.Release(); } // clean exit, we can recycle _readStatus = ReadStatus.RanToCompletion; @@ -399,11 +436,12 @@ private void FillBufferSync(Stream tail, CancellationToken cancellationToken) { while (true) { - if (_readLoopDoomed) return; // the parse loop already gave up; _readBuffer may be gone - Memory buffer; lock (_readBufferLock) { + // see FillBufferAsync's matching check for why this must be inside the lock, not + // just before it + if (_readLoopDoomed) return; // the parse loop already gave up; _readBuffer may be gone buffer = _readBuffer.GetUncommittedMemory(); } @@ -422,8 +460,6 @@ private void FillBufferSync(Stream tail, CancellationToken cancellationToken) read = 0; // some streams throw rather than returning 0; treat identically } - if (_readLoopDoomed) return; // as above - checked again now the read has had time to run - if (read <= 0) { return; // clean EOF - _fillerDone is set in the finally block below @@ -431,6 +467,9 @@ private void FillBufferSync(Stream tail, CancellationToken cancellationToken) 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; _readBuffer.Commit(read); } UpdateLastReadTime(); From f442bde80e6db3a66c7150b81d29cb2a9ca25aaa Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Tue, 22 Sep 2026 17:33:42 +0100 Subject: [PATCH 6/8] Fix the deeper Release()-vs-in-flight-read race, without deadlocking The prior fix closed "take a new lease after Release()" by moving the _readLoopDoomed check inside the lock, but not "a lease taken before Release() runs, still being written into": Release() hands segments back to the shared static spare regardless of any outstanding lease, so a filler read still in flight when Release() runs can go on writing into memory another connection's CycleBuffer has since taken over. The post-read doom check stops the *commit*, not the write that already landed. My first attempt at a fix - awaiting/joining the filler before Release() - was itself broken: the filler can only unblock via new data or the stream being disposed, and that disposal is exactly what RecordConnectionFailed (called right after) is about to do. Waiting for the filler first deadlocks against the very teardown that would unblock it - confirmed by 8 MovedUnitTests failures (SocketClosed, 0-read) once that landed, all in the ForceReconnect-into-retry path. Fixed properly instead: only call Release() when _fillerDone proves the filler has already, definitely, fully stopped (set in its own finally, right before it returns - a stronger guarantee than doomed, which only stops a *new* lease or commit). When it isn't yet safe, skip the recycle and let the existing plain wipe in finally - which only drops the field's value rather than returning anything to the shared pool - handle cleanup instead, same as the fault/cancellation paths already do. No waiting, no deadlock risk, and the corruption window closes without a behavioural cost on the common path. Verified: RESPite.Tests, StackExchange.Redis.Tests (multiple full runs, including the specific tests that caught the deadlock), full all-TFM build. --- .../PhysicalConnection.Read.cs | 54 +++++++++++++++---- 1 file changed, 45 insertions(+), 9 deletions(-) diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index 54f778ac0..614a6bba6 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -154,11 +154,27 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) // ForceReconnect can land here with the filler still genuinely mid-cycle (unlike the // _fillerDone break above, where the filler has already fully stopped by construction) - set - // this *before* Release() below, not just in finally, or the filler could still be holding an - // outstanding lease into a segment Release() is about to hand back to the shared spare pool. + // this *before* touching _readBuffer below, or the filler could still be holding an + // outstanding lease into a segment that's about to be recycled or wiped. _readLoopDoomed = true; _readStatus = ReadStatus.ProcessBufferComplete; - lock (_readBufferLock) { _readBuffer.Release(); } // clean exit, we can recycle + + // _fillerDone is only ever set (in FillBufferAsync's own finally, right before it returns) once + // the filler has genuinely, fully stopped - no outstanding lease, no read in flight. That's a + // stronger guarantee than "doomed": doom alone stops a *new* lease or commit, but not a read + // already in flight into memory captured before doom was set, and Release() hands segments + // back to the shared static spare regardless of any outstanding lease - so recycling here is + // only safe when _fillerDone proves there's nothing still writing into one. Waiting for that + // instead (e.g. joining the filler task/thread here) would deadlock: the filler can only + // unblock via new data or the stream being disposed, and that disposal is exactly what + // RecordConnectionFailed below is about to do. When it isn't yet safe, skip the recycle - the + // plain wipe in finally (which only drops the field's value, not returning anything to the + // shared pool) covers cleanup instead, same as the fault/cancellation paths already rely on. + if (_fillerDone) + { + lock (_readBufferLock) { _readBuffer.Release(); } // filler confirmed stopped - safe to recycle + } + _readStatus = ReadStatus.RanToCompletion; RecordConnectionFailed(ConnectionFailureType.SocketClosed); } @@ -331,9 +347,12 @@ private void SignalFillProgress() /// ) 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. Hence the explicit filler.Join() below, which is - /// otherwise unnecessary (the lock plus already make the plain wipe-on-exit - /// path safe, matching 's finally block, which doesn't wait for its filler task). + /// assumption both designs otherwise rely on. Hence the explicit filler.Join() below for that path + /// specifically: the stream stays open and connected throughout a transition, so the filler is going to + /// unblock the normal way (the next reply arriving) regardless, and joining just sequences the handoff + /// after that happens. The clean-exit Release() call further down deliberately does *not* join + /// the filler first, even though it has the same-shaped exposure - see the comment there for why that + /// would deadlock instead of just cost a wait. /// private void ReadAllSync(CancellationToken cancellationToken) { @@ -383,11 +402,28 @@ private void ReadAllSync(CancellationToken cancellationToken) // ForceReconnect can land here with the filler still genuinely mid-cycle (unlike the // _fillerDone break above, where the filler has already fully stopped by construction) - set - // this *before* Release() below, not just in finally, or the filler could still be holding an - // outstanding lease into a segment Release() is about to hand back to the shared spare pool. + // this *before* touching _readBuffer below, or the filler could still be holding an + // outstanding lease into a segment that's about to be recycled or wiped. _readLoopDoomed = true; _readStatus = ReadStatus.ProcessBufferComplete; - lock (_readBufferLock) { _readBuffer.Release(); } // clean exit, we can recycle + + // _fillerDone is only ever set (in FillBufferSync's own finally, right before it returns) once + // the filler has genuinely, fully stopped - no outstanding lease, no read in flight. That's a + // stronger guarantee than "doomed": doom alone stops a *new* lease or commit, but not a read + // already in flight into memory captured before doom was set, and Release() hands segments + // back to the shared static spare regardless of any outstanding lease - so recycling here is + // only safe when _fillerDone proves there's nothing still writing into one. filler.Join() here + // instead (unlike the sync-to-async transition's above, where the stream stays open and the + // filler unblocks the normal way, on the next reply) would deadlock: the filler can only + // unblock via new data or the stream being disposed, and that disposal is exactly what + // RecordConnectionFailed below is about to do. When it isn't yet safe, skip the recycle - the + // plain wipe in finally (which only drops the field's value, not returning anything to the + // shared pool) covers cleanup instead, same as the fault/cancellation paths already rely on. + if (_fillerDone) + { + lock (_readBufferLock) { _readBuffer.Release(); } // filler confirmed stopped - safe to recycle + } + _readStatus = ReadStatus.RanToCompletion; RecordConnectionFailed(ConnectionFailureType.SocketClosed); } From a5309cd6b3234ab739191cb9e246115b3e534276 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 23 Sep 2026 08:27:01 +0100 Subject: [PATCH 7/8] Enable hidden file upload in release workflow Allow uploading of hidden files from .nupkgs directory. --- .github/workflows/release.yml | 2 ++ 1 file changed, 2 insertions(+) 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 From d0ce01122ad2ef24ade49b07e0d2f6ec38741583 Mon Sep 17 00:00:00 2001 From: mgravell Date: Wed, 23 Sep 2026 14:52:26 +0100 Subject: [PATCH 8/8] Address review: hand-off drops replies, byte counter, buffer leak, lock scope - Sync-to-async transition: stop the filler via a dedicated _fillerHandoff flag that commits the in-flight read before stopping, instead of dooming it (which dropped that read and desynchronised every later reply). ReadAllAsync starts pre-signalled when inheriting the buffer so anything already committed is parsed without waiting for the next reply. - totalBytesReceived is now maintained by the filler (under the commit lock), restoring the in:/recd= timeout and failure diagnostics on the pull paths. - Whichever of the parse loop and the filler stops last recycles _readBuffer, so a custom ResponseBufferPool no longer leaks its leases on every reconnect. - ParseAvailableFrames snapshots under _readBufferLock and parses/dispatches outside it, so the filler can keep reading during dispatch. - Rethrow filler faults via ExceptionDispatchInfo to keep the read site's trace. - One FillBufferAsync body for both readers (sync flag), and drop the dead alreadyCommitted parameter, unreachable EndOfStream catches, and the never-disposed semaphore's ObjectDisposedException catch. - Docs: DedicatedThreads now costs two reader threads plus a writer. - Tests: ReaderLifecycleRoundTrip (byte counter in both modes; mid-hand-off reply is delivered) and ReadBufferPoolTests (pool leases return on reconnect and dispose). All three fail on the previous revision. --- docs/SyncOverAsync.md | 7 +- .../ConnectionMultiplexer.FeatureFlags.cs | 5 +- .../PhysicalConnection.Read.cs | 422 ++++++++---------- .../PhysicalConnection.Write.cs | 6 + .../ReadBufferPoolTests.cs | 96 ++++ .../ReaderLifecycleRoundTrip.cs | 72 +++ .../RoundTripUnitTests/TestConnection.cs | 29 ++ 7 files changed, 408 insertions(+), 229 deletions(-) create mode 100644 tests/StackExchange.Redis.Tests/ReadBufferPoolTests.cs create mode 100644 tests/StackExchange.Redis.Tests/RoundTripUnitTests/ReaderLifecycleRoundTrip.cs 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/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 614a6bba6..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,11 +20,12 @@ internal sealed partial class PhysicalConnection { private long totalBytesReceived; - // Guards every touch of _readBuffer/_readState 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). + // 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 @@ -31,17 +33,32 @@ internal sealed partial class PhysicalConnection // 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 volatile bool _fillerDone; private Exception? _fillerFault; - // Set (before _readBuffer is touched again) the moment the parse loop gives up, for any reason - a - // parse fault, ForceReconnect, or the filler's own clean EOF. The filler checks this after each read - // completes, before combining (committing) it into _readBuffer: once doomed, _readBuffer may already - // be torn down (see the finally block in ReadAllAsync), so the filler must stop rather than touch 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. + // 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); @@ -111,17 +128,23 @@ 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; - var fillSignal = _fillSignal = new SemaphoreSlim(0, 1); + _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 - @@ -130,7 +153,7 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) // 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, cancellationToken), cancellationToken); + var fillerTask = Task.Run(() => FillBufferAsync(tail, sync: false, cancellationToken), cancellationToken); try { while (true) @@ -139,47 +162,18 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) await fillSignal.WaitAsync(cancellationToken).ConfigureAwait(false); _readStatus = ReadStatus.TryParseResult; - lock (_readBufferLock) - { - ParseAvailableFrames(); - } + ParseAvailableFrames(); if (ForceReconnect) break; if (_fillerDone) { - if (_fillerFault is { } fault) throw fault; + 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 } } - // ForceReconnect can land here with the filler still genuinely mid-cycle (unlike the - // _fillerDone break above, where the filler has already fully stopped by construction) - set - // this *before* touching _readBuffer below, or the filler could still be holding an - // outstanding lease into a segment that's about to be recycled or wiped. - _readLoopDoomed = true; _readStatus = ReadStatus.ProcessBufferComplete; - - // _fillerDone is only ever set (in FillBufferAsync's own finally, right before it returns) once - // the filler has genuinely, fully stopped - no outstanding lease, no read in flight. That's a - // stronger guarantee than "doomed": doom alone stops a *new* lease or commit, but not a read - // already in flight into memory captured before doom was set, and Release() hands segments - // back to the shared static spare regardless of any outstanding lease - so recycling here is - // only safe when _fillerDone proves there's nothing still writing into one. Waiting for that - // instead (e.g. joining the filler task/thread here) would deadlock: the filler can only - // unblock via new data or the stream being disposed, and that disposal is exactly what - // RecordConnectionFailed below is about to do. When it isn't yet safe, skip the recycle - the - // plain wipe in finally (which only drops the field's value, not returning anything to the - // shared pool) covers cleanup instead, same as the fault/cancellation paths already rely on. - if (_fillerDone) - { - lock (_readBufferLock) { _readBuffer.Release(); } // filler confirmed stopped - safe to 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); } @@ -195,41 +189,35 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) } finally { - // set *before* touching _readBuffer below: the filler checks this (now inside the same lock - // as its own touches - see FillBufferAsync) before combining a read into _readBuffer, so it - // never combines into (or races) a buffer that's about to be - or already has been - wiped - // here. Redundant with the clean-exit path above, which sets it earlier for the same reason; - // harmless to repeat, and keeps every exit path (including ones that throw straight into a - // catch below, skipping that code) self-contained rather than relying on it having run. - _readLoopDoomed = true; + // 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(); - - // lock-protected because the filler's own reads/writes of _readBuffer are - an unsynchronized - // write here would race a multi-field struct against them, not just risk a missed signal. - lock (_readBufferLock) - { - _readBuffer = default; // wipe, however we exited - } } } /// /// Independently keeps full: reads, commits under , /// signals , repeats - with no knowledge of (or dependency on) whether the parse - /// loop in 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. + /// 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 @@ -240,8 +228,9 @@ private async Task ReadAllAsync(CancellationToken cancellationToken) /// 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, CancellationToken cancellationToken) + private async Task FillBufferAsync(Stream tail, bool sync, CancellationToken cancellationToken) { try { @@ -251,20 +240,32 @@ private async Task FillBufferAsync(Stream tail, CancellationToken cancellationTo lock (_readBufferLock) { // Checked *inside* the same lock that guards every _readBuffer touch (here and in the - // parse loop's own wipe/Release), 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-wipe - // (or doom-and-Release) in between, so this thread would then touch a buffer that's - // gone - or, worse for Release(), 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) return; // the parse loop already gave up; _readBuffer may be gone + // 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 { - read = await tail.ReadAsync(buffer, cancellationToken).ConfigureAwait(false); + if (sync) + { + cancellationToken.ThrowIfCancellationRequested(); +#if NET + read = tail.Read(buffer.Span); +#else + read = tail.Read(buffer); +#endif + } + else + { + read = await tail.ReadAsync(buffer, cancellationToken).ConfigureAwait(false); + } } catch (EndOfStreamException) { @@ -276,18 +277,35 @@ private async Task FillBufferAsync(Stream tail, CancellationToken cancellationTo return; // clean EOF - _fillerDone is set in the finally block below } + 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; + 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 - DebugCounters.OnAsyncRead(read, inline: false); + 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 } } catch (OperationCanceledException) @@ -300,15 +318,52 @@ private async Task FillBufferAsync(Stream tail, CancellationToken cancellationTo } finally { - _fillerDone = true; + 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 } } - // Only ever called from the filler's own single logical thread of control (FillBufferAsync's task, - // or FillBufferSync's thread) - 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 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(); + } + } + + private void OnFillerStopped() + { + lock (_readBufferLock) + { + _fillerDone = true; + if (_readLoopDoomed) ReleaseReadBufferInsideLock(); + } + } + + 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; + } + + // 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; @@ -318,10 +373,6 @@ private void SignalFillProgress() { sem.Release(); } - catch (ObjectDisposedException) - { - // torn down from under us during shutdown; the parse loop has already stopped caring - } catch (SemaphoreFullException) { // would mean the single-caller invariant above was violated by a second, concurrent @@ -334,25 +385,25 @@ private void SignalFillProgress() } /// - /// Sync-mode counterpart of the async filler/parser split in and - /// : a dedicated filler thread 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. + /// 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. Hence the explicit filler.Join() below for that path - /// specifically: the stream stays open and connected throughout a transition, so the filler is going to - /// unblock the normal way (the next reply arriving) regardless, and joining just sequences the handoff - /// after that happens. The clean-exit Release() call further down deliberately does *not* join - /// the filler first, even though it has the same-shaped exposure - see the comment there for why that - /// would deadlock instead of just cost a wait. + /// 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) { @@ -362,10 +413,11 @@ private void ReadAllSync(CancellationToken cancellationToken) _fillerDone = false; _fillerFault = null; _readLoopDoomed = false; + _fillerHandoff = false; var fillSignal = _fillSignal = new SemaphoreSlim(0, 1); var tail = _ioStream ?? Stream.Null; - Thread filler = new Thread(() => FillBufferSync(tail, cancellationToken)) + Thread filler = new Thread(() => FillBufferAsync(tail, sync: true, cancellationToken).GetAwaiter().GetResult()) { IsBackground = true, @@ -386,49 +438,19 @@ private void ReadAllSync(CancellationToken cancellationToken) fillSignal.Wait(cancellationToken); _readStatus = ReadStatus.TryParseResult; - lock (_readBufferLock) - { - ParseAvailableFrames(); - } + ParseAvailableFrames(); if (ForceReconnect) break; if (ShouldTransitionToAsync()) return; // finally below hands off once the filler has stopped if (_fillerDone) { - if (_fillerFault is { } fault) throw fault; + 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 } } - // ForceReconnect can land here with the filler still genuinely mid-cycle (unlike the - // _fillerDone break above, where the filler has already fully stopped by construction) - set - // this *before* touching _readBuffer below, or the filler could still be holding an - // outstanding lease into a segment that's about to be recycled or wiped. - _readLoopDoomed = true; _readStatus = ReadStatus.ProcessBufferComplete; - - // _fillerDone is only ever set (in FillBufferSync's own finally, right before it returns) once - // the filler has genuinely, fully stopped - no outstanding lease, no read in flight. That's a - // stronger guarantee than "doomed": doom alone stops a *new* lease or commit, but not a read - // already in flight into memory captured before doom was set, and Release() hands segments - // back to the shared static spare regardless of any outstanding lease - so recycling here is - // only safe when _fillerDone proves there's nothing still writing into one. filler.Join() here - // instead (unlike the sync-to-async transition's above, where the stream stays open and the - // filler unblocks the normal way, on the next reply) would deadlock: the filler can only - // unblock via new data or the stream being disposed, and that disposal is exactly what - // RecordConnectionFailed below is about to do. When it isn't yet safe, skip the recycle - the - // plain wipe in finally (which only drops the field's value, not returning anything to the - // shared pool) covers cleanup instead, same as the fault/cancellation paths already rely on. - if (_fillerDone) - { - lock (_readBufferLock) { _readBuffer.Release(); } // filler confirmed stopped - safe to recycle - } - - _readStatus = ReadStatus.RanToCompletion; - RecordConnectionFailed(ConnectionFailureType.SocketClosed); - } - catch (EndOfStreamException) when (_readStatus is ReadStatus.ReadSync) - { + OnParseLoopStopped(); // stops the filler; recycles the buffer now, or leaves that to the filler _readStatus = ReadStatus.RanToCompletion; RecordConnectionFailed(ConnectionFailureType.SocketClosed); } @@ -444,88 +466,18 @@ private void ReadAllSync(CancellationToken cancellationToken) } finally { - _readLoopDoomed = true; if (_readStatus is ReadStatus.TransitioningToAsync) { - // must be fully stopped - not just signalled - before ReadAllAsync starts its own filler - // on the same _readBuffer; see this method's remarks. - filler.Join(); + // 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 { - lock (_readBufferLock) - { - _readBuffer = default; // wipe, however we exited - } - } - } - } - - /// - /// Blocking-thread counterpart of - see its remarks, which apply unchanged - /// (single outstanding read at a time, no reservation needed). - /// - private void FillBufferSync(Stream tail, CancellationToken cancellationToken) - { - try - { - while (true) - { - Memory buffer; - lock (_readBufferLock) - { - // see FillBufferAsync's matching check for why this must be inside the lock, not - // just before it - if (_readLoopDoomed) return; // the parse loop already gave up; _readBuffer may be gone - buffer = _readBuffer.GetUncommittedMemory(); - } - - int read; - try - { - cancellationToken.ThrowIfCancellationRequested(); -#if NET - read = tail.Read(buffer.Span); -#else - read = tail.Read(buffer); -#endif - } - catch (EndOfStreamException) - { - read = 0; // some streams throw rather than returning 0; treat identically - } - - if (read <= 0) - { - return; // clean EOF - _fillerDone is set in the finally block below - } - - 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; - _readBuffer.Commit(read); - } - UpdateLastReadTime(); - DebugCounters.OnSyncRead(read); - SignalFillProgress(); + OnParseLoopStopped(); // as in ReadAllAsync: idempotent, and the only exit for the non-clean paths } } - catch (OperationCanceledException) - { - // normal shutdown path (connection tearing down) - not a fault worth reporting - } - catch (Exception ex) - { - _fillerFault = ex; - } - finally - { - _fillerDone = true; - SignalFillProgress(); // wake the parse loop even if nothing new arrived, so it notices completion - } } private bool ShouldTransitionToAsync() @@ -558,7 +510,9 @@ private long GetReadCommittedLength() } } - private bool CommitAndParseFrames(int bytesRead, bool alreadyCommitted = false) + // 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) { @@ -566,13 +520,10 @@ private bool CommitAndParseFrames(int bytesRead, bool alreadyCommitted = false) } totalBytesReceived += bytesRead; - if (!alreadyCommitted) - { - Debug.Assert( - bytesRead <= _readBuffer.UncommittedAvailable, - $"Insufficient bytes in {nameof(CommitAndParseFrames)}; got {bytesRead}, Available={_readBuffer.UncommittedAvailable}"); - _readBuffer.Commit(bytesRead); - } + Debug.Assert( + bytesRead <= _readBuffer.UncommittedAvailable, + $"Insufficient bytes in {nameof(CommitAndParseFrames)}; got {bytesRead}, Available={_readBuffer.UncommittedAvailable}"); + _readBuffer.Commit(bytesRead); ParseAvailableFrames(); return true; @@ -587,16 +538,34 @@ private bool CommitAndParseFrames(int bytesRead, bool alreadyCommitted = false) private void ParseAvailableFrames() { ref RespScanState state = ref _readState; // avoid a ton of ldarg0 - Debug.Assert(_readBuffer.GetCommittedLength() >= 0, "multi-segment running-indices are corrupt"); + + // 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 - string src = $"parse ({_readBuffer.GetCommittedLength()}-{state.TotalBytes})"; + 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 @@ -639,11 +608,13 @@ 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(); if (fullSequence.IsEmpty) return; // nothing committed at all yet Debug.Assert(!fullSequence.IsSingleSegment, "non-trivial sequence expected"); @@ -689,7 +660,10 @@ state is } OnDetailLog($"discarding {fullyConsumed} bytes"); - _readBuffer.DiscardCommitted(fullyConsumed); + lock (_readBufferLock) + { + _readBuffer.DiscardCommitted(fullyConsumed); + } } if (status != OperationStatus.NeedMoreData) 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/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)