feat(membership): detect local stalls during direct probes (#9979)

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: ReubenBond

src/Orleans.Runtime/MembershipService/LocalSiloHealthEventHistory.cs

@@ -242,7 +242,7 @@ private bool Overlaps(LocalSiloHealthEvent healthEvent, long startTimestamp, lon
     private static bool IsStateful(LocalSiloHealthCheckKind kind)
         => kind is LocalSiloHealthCheckKind.MembershipStatus
             or LocalSiloHealthCheckKind.HealthCheckParticipant
-            or LocalSiloHealthCheckKind.ThreadPoolQueueDelay
+            or LocalSiloHealthCheckKind.ThreadPoolStall
             or LocalSiloHealthCheckKind.ProbeRequests
             or LocalSiloHealthCheckKind.ProbeResponses;
 

src/Orleans.Runtime/MembershipService/LocalSiloHealthMonitor.cs

@@ -27,7 +27,7 @@ internal enum LocalSiloHealthCheckKind
         MembershipStatus,
         SiloSuspected,
         HealthCheckParticipant,
-        ThreadPoolQueueDelay,
+        ThreadPoolStall,
         ProbeRequests,
         ProbeResponses,
         GarbageCollectionPause,
@@ -53,6 +53,24 @@ public ImmutableArray<string> Complaints
 
     internal interface ILocalSiloHealthMonitor
     {
+        /// <summary>
+        /// Returns a timestamp from the stall detector's time source.
+        /// </summary>
+        /// <returns>A timestamp suitable for use with <see cref="GetStallDurationAsync"/>.</returns>
+        long GetTimestamp();
+
+        /// <summary>
+        /// Waits for the stall detector to sample past the end of the interval, then returns the detected stall duration.
+        /// </summary>
+        /// <param name="startTimestamp">The start of the interval.</param>
+        /// <param name="endTimestamp">The end of the interval.</param>
+        /// <param name="cancellationToken">A token which cancels the wait.</param>
+        /// <returns>The detected stall duration.</returns>
+        ValueTask<TimeSpan> GetStallDurationAsync(
+            long startTimestamp,
+            long endTimestamp,
+            CancellationToken cancellationToken);
+
         /// <summary>
         /// Returns the aggregate local health status over the provided interval.
         /// </summary>
@@ -101,15 +119,16 @@ void RecordHealthEvent(
     ///   <item><description>Check that no other silo suspects this silo.</description></item>
     ///   <item><description>Check for recently received successful ping responses (via <see cref="IProbeHealthMonitor"/>).</description></item>
     ///   <item><description>Check for recently received ping requests (via <see cref="IProbeHealthMonitor"/>).</description></item>
-    ///   <item><description>Check that the .NET Thread Pool is able to process work items within one second.</description></item>
+    ///   <item><description>Check that the .NET Thread Pool executes periodic timer callbacks on schedule.</description></item>
     ///   <item><description>Check that local async timers have been firing on-time (within 3 seconds of their due time).</description></item>
     /// </list>
     /// </remarks>
     internal partial class LocalSiloHealthMonitor :
         ILifecycleParticipant<ISiloLifecycle>,
         ILifecycleObserver,
         ILocalSiloHealthMonitor,
-        ILocalSiloHealthEventRecorder
+        ILocalSiloHealthEventRecorder,
+        IDisposable
     {
         internal const int MaxScore = 8;
         private static readonly TimeSpan HistoryDuration = TimeSpan.FromMinutes(1);
@@ -122,7 +141,7 @@ internal partial class LocalSiloHealthMonitor :
         private readonly ILogger<LocalSiloHealthMonitor> _log;
         private readonly ClusterMembershipOptions _clusterMembershipOptions;
         private readonly IAsyncTimer _degradationCheckTimer;
-        private readonly ThreadPoolMonitor _threadPoolMonitor;
+        private readonly ThreadPoolStallDetector _stallDetector;
         private readonly TimeProvider _timeProvider;
 #if NET9_0_OR_GREATER
         private readonly Lock _historyLock = new();
@@ -166,12 +185,32 @@ public LocalSiloHealthMonitor(
                 MinimumCheckPeriod,
                 nameof(LocalSiloHealthMonitor),
                 timeProvider);
-            _threadPoolMonitor = new ThreadPoolMonitor(loggerFactory.CreateLogger<ThreadPoolMonitor>(), timeProvider);
+            var stallRetentionPeriod = _clusterMembershipOptions.MaxProbeTimeout + ThreadPoolStallDetector.DetectionPeriod;
+            if (stallRetentionPeriod < HistoryDuration)
+            {
+                stallRetentionPeriod = HistoryDuration;
+            }
+
+            _stallDetector = new(
+                loggerFactory.CreateLogger<ThreadPoolStallDetector>(),
+                timeProvider,
+                ThreadPoolStallDetector.DetectionPeriod,
+                stallRetentionPeriod);
         }
 
         /// <inheritdoc />
         public ImmutableArray<string> Complaints { get; private set; } = [];
 
+        /// <inheritdoc />
+        public long GetTimestamp() => _timeProvider.GetTimestamp();
+
+        /// <inheritdoc />
+        public ValueTask<TimeSpan> GetStallDurationAsync(
+            long startTimestamp,
+            long endTimestamp,
+            CancellationToken cancellationToken)
+            => _stallDetector.GetStallDurationAsync(startTimestamp, endTimestamp, cancellationToken);
+
         /// <inheritdoc />
         public LocalSiloHealthStatus GetLocalHealthStatus(
             long startTimestamp,
@@ -265,7 +304,7 @@ private LocalSiloHealthStatus EnsureHealthCheck(
                 var events = new List<LocalSiloHealthEvent>(_healthCheckParticipants.Count + 1);
                 var complaints = new List<string>();
                 CheckLocalHealthCheckParticipants(now.UtcDateTime, timestamp, events, complaints);
-                CheckThreadPoolQueueDelay(timestamp, events, complaints);
+                CheckThreadPoolStalls(timestamp, events, complaints);
                 AddEvents(timestamp, events);
 
                 var score = GetScore(events);
@@ -350,28 +389,30 @@ private void AddEvents(long timestamp, List<LocalSiloHealthEvent> events)
             }
         }
 
-        private void CheckThreadPoolQueueDelay(
+        private void CheckThreadPoolStalls(
             long timestamp,
             List<LocalSiloHealthEvent> events,
             List<string> complaints)
         {
-            var delay = _threadPoolMonitor.MeasureQueueDelay();
-            var score = (int)delay.TotalSeconds;
+            var stallDuration = _stallDetector.GetMaximumStallDuration(
+                SubtractTimestamp(timestamp, MinimumCheckPeriod),
+                timestamp);
+            var score = (int)stallDuration.TotalSeconds;
             string? complaint = null;
             if (score >= 1)
             {
-                complaint = $".NET Thread Pool is exhibiting delays of {delay.TotalSeconds}s. This can indicate .NET Thread Pool starvation, very long .NET GC pauses, or other runtime or machine pauses.";
+                complaint = $".NET Thread Pool execution stalled for {stallDuration.TotalSeconds}s. This can indicate .NET Thread Pool starvation, very long .NET GC pauses, or other runtime or machine pauses.";
                 complaints.Add(complaint);
             }
 
             events.Add(new(
                 timestamp,
-                LocalSiloHealthCheckKind.ThreadPoolQueueDelay,
-                GetCategory(LocalSiloHealthCheckKind.ThreadPoolQueueDelay),
+                LocalSiloHealthCheckKind.ThreadPoolStall,
+                GetCategory(LocalSiloHealthCheckKind.ThreadPoolStall),
                 Source: null,
                 score,
                 complaint,
-                delay,
+                stallDuration,
                 score >= 10 ? LogLevel.Error : LogLevel.Warning));
         }
 
@@ -610,9 +651,9 @@ private void LogHealthDetails(ImmutableArray<LocalSiloHealthEvent> events)
                     continue;
                 }
 
-                if (healthEvent.Kind == LocalSiloHealthCheckKind.ThreadPoolQueueDelay)
+                if (healthEvent.Kind == LocalSiloHealthCheckKind.ThreadPoolStall)
                 {
-                    LogThreadPoolDelay(healthEvent.LogLevel, healthEvent.Duration?.TotalSeconds ?? 0);
+                    LogThreadPoolStall(healthEvent.LogLevel, healthEvent.Duration?.TotalSeconds ?? 0);
                 }
                 else
                 {
@@ -644,79 +685,16 @@ public async Task OnStop(CancellationToken ct)
             }
         }
 
-        /// <summary>
-        /// Measures queue delay on the .NET <see cref="ThreadPool"/>.
-        /// </summary>
-        private class ThreadPoolMonitor
+        public void Dispose()
         {
-            private static readonly WaitCallback Callback = state => ((ThreadPoolMonitor)state!).Execute();
-#if NET9_0_OR_GREATER
-            private readonly Lock _lockObj = new();
-#else
-            private readonly object _lockObj = new();
-#endif
-            private readonly ILogger<ThreadPoolMonitor> _log;
-            private readonly TimeProvider _timeProvider;
-            private bool _scheduled;
-            private TimeSpan _lastQueueDelay;
-            private long _queueDelayTimestamp;
-
-            public ThreadPoolMonitor(ILogger<ThreadPoolMonitor> log, TimeProvider timeProvider)
-            {
-                _log = log;
-                _timeProvider = timeProvider;
-            }
-
-            public TimeSpan MeasureQueueDelay()
-            {
-                bool shouldSchedule;
-                TimeSpan delay;
-                lock (_lockObj)
-                {
-                    var currentQueueDelay = _scheduled ? _timeProvider.GetElapsedTime(_queueDelayTimestamp) : TimeSpan.Zero;
-                    delay = currentQueueDelay > _lastQueueDelay ? currentQueueDelay : _lastQueueDelay;
-
-                    if (!_scheduled)
-                    {
-                        _scheduled = true;
-                        shouldSchedule = true;
-                        _queueDelayTimestamp = _timeProvider.GetTimestamp();
-                    }
-                    else
-                    {
-                        shouldSchedule = false;
-                    }
-                }
-
-                if (shouldSchedule)
-                {
-                    _ = ThreadPool.UnsafeQueueUserWorkItem(Callback, this);
-                }
-
-                return delay;
-            }
-
-            private void Execute()
-            {
-                try
-                {
-                    lock (_lockObj)
-                    {
-                        _scheduled = false;
-                        _lastQueueDelay = _timeProvider.GetElapsedTime(_queueDelayTimestamp);
-                    }
-                }
-                catch (Exception exception)
-                {
-                    LocalSiloHealthMonitor.LogThreadPoolDelayMonitorError(_log, exception);
-                }
-            }
+            _degradationCheckTimer.Dispose();
+            _stallDetector.Dispose();
         }
 
         [LoggerMessage(
-            Message = ".NET Thread Pool is exhibiting delays of {ThreadPoolQueueDelaySeconds}s. This can indicate .NET Thread Pool starvation, very long .NET GC pauses, or other runtime or machine pauses."
+            Message = ".NET Thread Pool execution stalled for {ThreadPoolStallSeconds}s. This can indicate .NET Thread Pool starvation, very long .NET GC pauses, or other runtime or machine pauses."
         )]
-        private partial void LogThreadPoolDelay(LogLevel logLevel, double threadPoolQueueDelaySeconds);
+        private partial void LogThreadPoolStall(LogLevel logLevel, double threadPoolStallSeconds);
 
         [LoggerMessage(
             Message = "{Kind} health check for {Source} reported: {Complaint}"
@@ -763,12 +741,6 @@ private partial void LogRecordedHealthIssue(
         )]
         private partial void LogSelfMonitoringDegraded(int score, int maxScore, string complaints);
 
-        [LoggerMessage(
-            Level = LogLevel.Error,
-            Message = "Exception monitoring .NET thread pool delay"
-        )]
-        private static partial void LogThreadPoolDelayMonitorError(ILogger logger, Exception exception);
-
         [LoggerMessage(
             Level = LogLevel.Error,
             Message = "Error while monitoring local silo health"

src/Orleans.Runtime/MembershipService/SiloHealthMonitor.cs

@@ -207,12 +207,15 @@ private async Task Run()
                     localDegradationScore = GetLocalDegradationScore(previousProbePeriod);
                     var timeout = CalculateProbeTimeout(failureDetector, options, localDegradationScore, isDirectProbe, Debugger.IsAttached);
                     probeStartTimestamp = _timeProvider.GetTimestamp();
+                    var stallStartTimestamp = isDirectProbe
+                        ? _localSiloHealthMonitor.GetTimestamp()
+                        : 0;
                     using var cancellation = new CancellationTokenSource(timeout, _timeProvider);
 
                     if (isDirectProbe)
                     {
                         // Probe the silo directly.
-                        probeResult = await this.ProbeDirectly(cancellation.Token).ConfigureAwait(false);
+                        probeResult = await this.ProbeDirectly(cancellation.Token, timeout, stallStartTimestamp).ConfigureAwait(false);
                     }
                     else
                     {
@@ -321,22 +324,26 @@ internal static TimeSpan CalculateIndirectProbeTargetTimeout(TimeSpan timeout, i
         /// Probes the remote silo.
         /// </summary>
         /// <param name="cancellation">A token to cancel and fail the probe attempt.</param>
+        /// <param name="probeTimeout">The timeout used for this probe, for local stall evaluation.</param>
+        /// <param name="stallStartTimestamp">The start of the local stall collection interval.</param>
         /// <returns>The number of failed probes since the last successful probe.</returns>
-        private async Task<ProbeResult> ProbeDirectly(CancellationToken cancellation)
+        private async Task<ProbeResult> ProbeDirectly(CancellationToken cancellation, TimeSpan probeTimeout, long stallStartTimestamp)
         {
             var id = ++_nextProbeId;
             LogTraceGoingToSendPing(_log, id, TargetSiloAddress);
 
             var roundTripTimer = TimeProviderValueStopwatch.StartNew(_timeProvider);
             ProbeResult probeResult;
             Exception? failureException;
+            var probeTimedOut = false;
             try
             {
                 await _prober.Probe(TargetSiloAddress, id, cancellation).WaitAsync(cancellation);
                 failureException = null;
             }
             catch (OperationCanceledException exception)
             {
+                probeTimedOut = cancellation.IsCancellationRequested;
                 failureException = new OperationCanceledException(
                     $"The ping attempt was cancelled after {roundTripTimer.GetElapsedTime(out _)}. Ping #{id}",
                     exception);
@@ -346,7 +353,15 @@ private async Task<ProbeResult> ProbeDirectly(CancellationToken cancellation)
                 failureException = exception;
             }
             var roundTripTime = roundTripTimer.GetElapsedTime(out _);
-
+            var stallDurationDuring = TimeSpan.Zero;
+            if (probeTimedOut)
+            {
+                var probeEndTimestamp = _localSiloHealthMonitor.GetTimestamp();
+                stallDurationDuring = await _localSiloHealthMonitor.GetStallDurationAsync(
+                    stallStartTimestamp,
+                    probeEndTimestamp,
+                    _stoppingCancellation.Token).ConfigureAwait(false);
+            }
             if (failureException is null)
             {
                 _messagingInstruments?.OnPingReplyReceived(TargetSiloAddress);
@@ -361,12 +376,22 @@ private async Task<ProbeResult> ProbeDirectly(CancellationToken cancellation)
             }
             else
             {
-                _messagingInstruments?.OnPingReplyMissed(TargetSiloAddress);
-
-                var failedProbes = ++_failedProbes;
-                LogWarningDidNotGetResponseForProbe(_log, failureException, id, TargetSiloAddress, roundTripTime, failedProbes);
+                // Check if a local stall consumed a significant portion of the probe timeout.
+                // If so, the local silo may have been unable to process the response in time,
+                // so we treat this as an inconclusive result rather than a failure.
+                if (probeTimedOut && stallDurationDuring >= probeTimeout.Multiply(0.25))
+                {
+                    LogWarningProbeFailureDuringLocalStall(_log, id, TargetSiloAddress, roundTripTime, stallDurationDuring, _failedProbes);
+                    probeResult = ProbeResult.CreateDirect(_failedProbes, ProbeResultStatus.Unknown);
+                }
+                else
+                {
+                    _messagingInstruments?.OnPingReplyMissed(TargetSiloAddress);
+                    var failedProbes = ++_failedProbes;
+                    LogWarningDidNotGetResponseForProbe(_log, failureException, id, TargetSiloAddress, roundTripTime, failedProbes);
 
-                probeResult = ProbeResult.CreateDirect(failedProbes, ProbeResultStatus.Failed);
+                    probeResult = ProbeResult.CreateDirect(failedProbes, ProbeResultStatus.Failed);
+                }
             }
 
             return probeResult;
@@ -536,5 +561,11 @@ public enum ProbeResultStatus
             Message = "Exception monitoring silo {SiloAddress}"
         )]
         private static partial void LogErrorExceptionMonitoringSilo(ILogger logger, Exception exception, SiloAddress siloAddress);
+
+        [LoggerMessage(
+            Level = LogLevel.Warning,
+            Message = "Probe #{Id} to silo {SiloAddress} failed after {Elapsed}, but local execution stalled for {StallDuration} during the probe. Treating as inconclusive. Consecutive failed probe count remains at {FailedProbeCount}."
+        )]
+        private static partial void LogWarningProbeFailureDuringLocalStall(ILogger logger, int id, SiloAddress siloAddress, TimeSpan elapsed, TimeSpan stallDuration, int failedProbeCount);
     }
 }