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); + } +}