From 3ca34dfd76a2e898d6b1b1d6a5e7168a620cecf2 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Thu, 10 Sep 2026 05:36:18 +0100 Subject: [PATCH] Sync path must wait for completion, not merely for a fault ExecuteSyncImpl stopped waiting as soon as it observed IResultBox.IsFaulted. That is only safe when the message faulted inline on the calling thread during the write - in which case the completion's PulseAll happened under our own (reentrant) lock and is already gone. When another thread faults the box, the two steps are separate and unsynchronized: ResultProcessor.SetException publishes the exception with no lock, and the matching pulse only follows later, from Message.Complete -> ActivateContinuations. So a caller preempted between the write reaching the wire (inside TryPushMessageToBridgeSync) and the IsFaulted read can see the fault, skip the wait, and recycle the box into the [ThreadStatic] pool while the reader thread is still on its way to PulseAll on that very object. The next synchronous call on that thread borrows the same box, parks in Monitor.Wait, and is woken by that stale pulse: it returns with neither result nor exception. Its real reply then lands on the already-recycled box, so every later reply on the thread is shifted by one - and because the pool is thread-static, the damage survives new multiplexers and new servers. Track completion explicitly instead. ActivateContinuations sets _completed inside the lock, immediately before the pulse, and the waiter leaves only on _completed - so a box can never be recycled while a pulse is inbound, and a pulse the waiter did not cause no longer shortens the wait (the original deadline is preserved across such wakes). Recycling clears the flag. Reported against Garnet CI as microsoft/garnet#2110, where it surfaced as one test losing its exception and a later test in the same fixture receiving it. --- .../ConnectionMultiplexer.cs | 37 ++++-- src/StackExchange.Redis/ResultBox.cs | 40 ++++-- .../ResultBoxCompletionTests.cs | 116 ++++++++++++++++++ 3 files changed, 178 insertions(+), 15 deletions(-) create mode 100644 tests/StackExchange.Redis.Tests/ResultBoxCompletionTests.cs diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs index 903941c1f..939829e9b 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs @@ -2485,21 +2485,42 @@ internal static void ThrowFailed(TaskCompletionSource? source, Exception u #pragma warning disable CS0618 // Type or member is obsolete result = TryPushMessageToBridgeSync(message, processor, source, ref server); #pragma warning restore CS0618 - if (!source.IsFaulted) // if we faulted while writing, we don't need to wait + // Note this tests IsCompleted, *not* IsFaulted: the write path can fault and complete the + // message inline on this thread (the lock is ours, so its PulseAll had no waiter and is + // gone) - in which case there is nothing to wait for. But a fault published by *another* + // thread always still owes us a pulse, and leaving early on that would let the pulse + // arrive after we have recycled the box, landing on the next operation to borrow it. + if (!source.IsCompleted) { if (result != WriteResult.Success) { throw GetException(result, message, server); } - if (Monitor.Wait(source, TimeoutMilliseconds)) + // wait for the *completion*, not merely for a pulse; a pulse that we did not cause + // must not be allowed to shorten this wait, so keep the original deadline + var remaining = TimeoutMilliseconds; + var startedAt = Environment.TickCount; + while (true) { - Trace("Timely response to " + message); - } - else - { - Trace("Timeout performing " + message); - timeout = true; + if (!Monitor.Wait(source, remaining)) + { + Trace("Timeout performing " + message); + timeout = true; + break; + } + if (source.IsCompleted) + { + Trace("Timely response to " + message); + break; + } + remaining = TimeoutMilliseconds - unchecked(Environment.TickCount - startedAt); + if (remaining <= 0) + { + Trace("Timeout performing " + message); + timeout = true; + break; + } } } } diff --git a/src/StackExchange.Redis/ResultBox.cs b/src/StackExchange.Redis/ResultBox.cs index 160493cc1..eeacae0f0 100644 --- a/src/StackExchange.Redis/ResultBox.cs +++ b/src/StackExchange.Redis/ResultBox.cs @@ -9,6 +9,20 @@ internal interface IResultBox Exception? Fault { get; } bool IsAsync { get; } bool IsFaulted { get; } + + /// + /// Indicates that has completed, i.e. the completing thread + /// has finished with this box. + /// + /// + /// This - not - is the signal a synchronous waiter must use. The fault is + /// published (unlocked) by ResultProcessor.SetException, but the matching pulse only happens + /// later, in Message.Complete; a waiter that stops on can therefore + /// recycle the box while a pulse is still inbound, and that pulse then lands on whatever operation + /// next borrows the box from the thread-static pool. + /// + bool IsCompleted { get; } + void SetException(Exception ex); void ActivateContinuations(); void Cancel(); @@ -22,9 +36,11 @@ internal interface IResultBox : IResultBox internal abstract class SimpleResultBox : IResultBox { private volatile Exception? _exception; + private volatile bool _completed; bool IResultBox.IsAsync => false; bool IResultBox.IsFaulted => _exception != null; + bool IResultBox.IsCompleted => _completed; Exception? IResultBox.Fault => _exception; void IResultBox.SetException(Exception exception) => _exception = exception ?? CancelledException; void IResultBox.Cancel() => _exception = CancelledException; @@ -33,22 +49,28 @@ void IResultBox.ActivateContinuations() { lock (this) { // tell the waiting thread that we're done + // note this is set *inside* the lock, so a waiter that only leaves on _completed cannot + // exit (and recycle the box) while this pulse is still inbound + _completed = true; Monitor.PulseAll(this); } ConnectionMultiplexer.TraceWithoutContext("Pulsed", "Result"); } + /// Clears the per-operation state so the box can be handed to another operation. + protected void ResetForRecycle() + { + _exception = null; + _completed = false; + } + // in theory nobody should directly observe this; the only things // that call Cancel are transactions etc - TCS-based, and we detect // that and use TrySetCanceled instead // about any confusion in stack-trace internal static readonly Exception CancelledException = new TaskCanceledException(); - protected Exception? Exception - { - get => _exception; - set => _exception = value; - } + protected Exception? Exception => _exception; } internal sealed class SimpleResultBox : SimpleResultBox, IResultBox @@ -62,8 +84,10 @@ private SimpleResultBox() { } public static IResultBox Create() => new SimpleResultBox(); public static IResultBox Get() // includes recycled boxes; used from sync, so makes re-use easy { - var obj = _perThreadInstance ?? new SimpleResultBox(); + var obj = _perThreadInstance; + if (obj is null) return new SimpleResultBox(); _perThreadInstance = null; // in case of oddness; only set back when recycled + obj.ResetForRecycle(); // belt and braces; the recycle already did this return obj; } @@ -75,7 +99,8 @@ private SimpleResultBox() { } ex = Exception; if (canRecycle) { - Exception = null; + // only ever reached once the box is IsCompleted, so no other thread is still touching it + ResetForRecycle(); _value = default!; _perThreadInstance = this; } @@ -97,6 +122,7 @@ private TaskResultBox(object? asyncState, TaskCreationOptions creationOptions) : bool IResultBox.IsAsync => true; bool IResultBox.IsFaulted => _exception != null; + bool IResultBox.IsCompleted => Task.IsCompleted; Exception? IResultBox.Fault => _exception; void IResultBox.Cancel() => _exception = SimpleResultBox.CancelledException; diff --git a/tests/StackExchange.Redis.Tests/ResultBoxCompletionTests.cs b/tests/StackExchange.Redis.Tests/ResultBoxCompletionTests.cs new file mode 100644 index 000000000..1b7d436dc --- /dev/null +++ b/tests/StackExchange.Redis.Tests/ResultBoxCompletionTests.cs @@ -0,0 +1,116 @@ +using System; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// boxes are pooled per-thread, so a synchronous caller must not stop +/// waiting on one until the completing thread has finished with it. Faulting the box +/// (ResultProcessor.SetException) and pulsing it (Message.Complete -> +/// ActivateContinuations) are separate, unsynchronized steps: a waiter that leaves on the fault +/// alone can recycle the box while the pulse is still inbound, and that pulse then wakes whichever +/// operation next borrows the box - which returns with neither result nor exception, shifting every +/// later reply on that thread by one. Reported against Garnet CI as microsoft/garnet#2110. +/// +public class ResultBoxCompletionTests(ITestOutputHelper log) +{ + /// + /// A fault published by another thread must not be mistaken for completion: that thread still owes + /// us the pulse, and it is only safe to recycle the box once the pulse has been delivered. + /// + [Fact] + public void FaultAloneIsNotCompletion() + { + var box = SimpleResultBox.Get(); + using var faulted = new ManualResetEventSlim(); + var pulsedAt = 0L; + + var completer = Task.Run(() => + { + box.SetException(new InvalidOperationException("boom")); + faulted.Set(); + Thread.Sleep(200); // the real gap between ResultProcessor.SetException and Message.Complete + Volatile.Write(ref pulsedAt, Stopwatch.GetTimestamp()); + box.ActivateContinuations(); + }); + + lock (box) + { + Assert.True(faulted.Wait(5000)); + + // this is the state the old code bailed out on - and doing so is what recycles a box that + // is still owed a pulse + Assert.True(box.IsFaulted); + Assert.False(box.IsCompleted); + + while (!box.IsCompleted) + { + Assert.True(Monitor.Wait(box, 5000), "timed out waiting for completion"); + } + } + + var leftAt = Stopwatch.GetTimestamp(); + completer.Wait(5000); + log.WriteLine($"left the wait {(leftAt - Volatile.Read(ref pulsedAt)) * 1000.0 / Stopwatch.Frequency:F2}ms after the pulse"); + Assert.True(leftAt >= Volatile.Read(ref pulsedAt), "waiter left before the completing thread pulsed"); + + var value = box.GetResult(out var ex, canRecycle: true); + Assert.Equal(0, value); + Assert.IsType(ex); + } + + /// Recycling must hand back a box with no completion or fault left on it. + [Fact] + public void RecyclingResetsCompletionState() + { + var box = SimpleResultBox.Get(); + box.SetException(new InvalidOperationException("boom")); + box.ActivateContinuations(); + Assert.True(box.IsCompleted); + Assert.True(box.IsFaulted); + + box.GetResult(out _, canRecycle: true); + + var reused = SimpleResultBox.Get(); + Assert.Same(box, reused); // same thread, so we get the pooled instance back + Assert.False(reused.IsCompleted); + Assert.False(reused.IsFaulted); + } + + /// + /// A pulse the waiter did not cause must not shorten the wait - the sync path treats any wake as + /// "the reply is here", so a stale pulse would otherwise be reported as a successful (empty) result. + /// + [Fact] + public void StrayPulseDoesNotEndTheWait() + { + var box = SimpleResultBox.Get(); + using var strayed = new ManualResetEventSlim(); + + _ = Task.Run(() => + { + Thread.Sleep(100); + lock (box) { Monitor.PulseAll(box); } // a pulse with no completion behind it + strayed.Set(); + Thread.Sleep(100); + box.SetResult(42); + box.ActivateContinuations(); + }); + + lock (box) + { + while (!box.IsCompleted) + { + Assert.True(Monitor.Wait(box, 5000), "timed out waiting for completion"); + } + } + + Assert.True(strayed.IsSet); + var value = box.GetResult(out var ex, canRecycle: true); + Assert.Null(ex); + Assert.Equal(42, value); + } +}