SLES-2913: prevent named pipe failures from eating the whole thread pool

src/StatsdClient/Telemetry.cs

@@ -14,11 +14,14 @@ internal class Telemetry : IDisposable, ITelemetryCounters
     {
         private static string _telemetryPrefix = "datadog.dogstatsd.client.";
         private readonly Timer _optionalTimer;
+        private readonly TimeSpan _flushInterval;
+        private readonly object _timerLock = new object();
         private readonly string[] _optionalTags;
         private readonly MetricSerializer _optionalMetricSerializer;
         private readonly ITransport _optionalTransport;
         private readonly Dictionary<MetricType, ValueWithTags> _aggregatedContexts = new Dictionary<MetricType, ValueWithTags>();
         private readonly Action<Exception> _optionalExceptionHandler;
+        private volatile bool _disposed;
 
         private int _metricsSent;
         private int _eventsSent;
@@ -54,13 +57,18 @@ public Telemetry(
             _aggregatedContexts.Add(MetricType.Count, new ValueWithTags(_optionalTags, "metrics_type:count"));
             _aggregatedContexts.Add(MetricType.Set, new ValueWithTags(_optionalTags, "metrics_type:set"));
             _optionalExceptionHandler = optionalExceptionHandler;
+            _flushInterval = flushInterval;
             if (!synchronousMode)
             {
+                // One-shot timer re-armed at the end of each flush (see OnTimerFlush) so that
+                // at most one flush runs at a time. A periodic timer would keep dispatching
+                // callbacks onto the thread pool even while a flush is blocked on a stalled
+                // transport, growing the thread count without bound.
                 _optionalTimer = new Timer(
-                    _ => Flush(),
+                    _ => OnTimerFlush(),
                     null,
                     flushInterval,
-                    flushInterval);
+                    Timeout.InfiniteTimeSpan);
             }
         }
 
@@ -170,8 +178,46 @@ public void OnAggregatedContextFlush(MetricType metricType, int count)
 
         public void Dispose()
         {
-            _optionalTimer?.Change(Timeout.Infinite, Timeout.Infinite);
-            _optionalTimer?.Dispose();
+            lock (_timerLock)
+            {
+                if (_disposed)
+                {
+                    return;
+                }
+
+                _disposed = true;
+                _optionalTimer?.Change(Timeout.Infinite, Timeout.Infinite);
+                _optionalTimer?.Dispose();
+            }
+        }
+
+        private void OnTimerFlush()
+        {
+            // A one-shot callback can already be in flight when Dispose runs, so skip the flush
+            // if the timer has been disposed. The check runs without holding _timerLock so it
+            // never blocks Dispose.
+            if (_disposed)
+            {
+                return;
+            }
+
+            try
+            {
+                Flush();
+            }
+            finally
+            {
+                // Re-arm the one-shot timer only after this flush finishes, so flushes cannot overlap.
+                // Dispose sets _disposed and disposes the timer under the same lock, so while _disposed
+                // is false the timer is still alive.
+                lock (_timerLock)
+                {
+                    if (!_disposed)
+                    {
+                        _optionalTimer?.Change(_flushInterval, Timeout.InfiniteTimeSpan);
+                    }
+                }
+            }
         }
 
         private void SendMetricWithTags(string metricName, string[] tags, int value)

src/StatsdClient/Transport/NamedPipeTransport.cs

@@ -2,21 +2,27 @@
 using System.IO;
 using System.IO.Pipes;
 using System.Threading;
+using System.Threading.Tasks;
 
 namespace StatsdClient.Transport
 {
     internal class NamedPipeTransport : ITransport
     {
-        private readonly NamedPipeClientStream _namedPipe;
+        private readonly string _pipeName;
         private readonly TimeSpan _timeout;
+        private readonly TimeSpan _connectionCooldown;
+        private readonly System.Diagnostics.Stopwatch _sendFailureTimer = new System.Diagnostics.Stopwatch();
         private readonly object _lock = new object();
 
+        private NamedPipeClientStream _namedPipe;
         private byte[] _internalbuffer = Array.Empty<byte>();
 
-        public NamedPipeTransport(string pipeName, TimeSpan? timeout = null)
+        public NamedPipeTransport(string pipeName, TimeSpan? timeout = null, TimeSpan? connectionCooldown = null)
         {
-            _namedPipe = new NamedPipeClientStream(".", pipeName, PipeDirection.Out, PipeOptions.Asynchronous);
+            _pipeName = pipeName;
+            _namedPipe = CreatePipe();
             _timeout = timeout ?? TimeSpan.FromSeconds(2);
+            _connectionCooldown = connectionCooldown ?? TimeSpan.FromSeconds(5);
         }
 
         public TransportType TransportType => TransportType.NamedPipe;
@@ -42,11 +48,43 @@ public bool Send(byte[] buffer, int length)
 
         public void Dispose()
         {
+            lock (_lock)
+            {
+                _namedPipe.Dispose();
+            }
+        }
+
+        private NamedPipeClientStream CreatePipe()
+        {
+            return new NamedPipeClientStream(".", _pipeName, PipeDirection.Out, PipeOptions.Asynchronous);
+        }
+
+        private void ResetPipeAfterAbandonedWrite(Task abandonedWrite)
+        {
+            // The abandoned write is fire-and-forget now. Observe its eventual fault so tearing
+            // down the disposed pipe does not surface as an unobserved task exception.
+            abandonedWrite.ContinueWith(
+                t => { _ = t.Exception; },
+                CancellationToken.None,
+                TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
+                TaskScheduler.Default);
+
             _namedPipe.Dispose();
+            _namedPipe = CreatePipe();
         }
 
         private bool SendBuffer(byte[] buffer, int length, bool allowRetry)
         {
+            // After a failed send, fail fast until the cooldown elapses instead of blocking
+            // again on Connect or Write. Otherwise every send re-blocks for the full timeout
+            // while the pipe is unavailable, starving the worker and (through the telemetry
+            // timer) inflating the thread count. The gate sits ahead of the connection check
+            // so a connected-but-stalled pipe (writes timing out) is throttled too.
+            if (_sendFailureTimer.IsRunning && _sendFailureTimer.Elapsed < _connectionCooldown)
+            {
+                return false;
+            }
+
             try
             {
                 if (!_namedPipe.IsConnected)
@@ -56,13 +94,43 @@ private bool SendBuffer(byte[] buffer, int length, bool allowRetry)
             }
             catch (TimeoutException)
             {
+                _sendFailureTimer.Restart();
+                return false;
+            }
+            catch (IOException)
+            {
+                // A missing or busy pipe fails the connect with IOException rather than
+                // TimeoutException. Start the cooldown so subsequent sends fail fast instead
+                // of re-blocking on Connect.
+                _sendFailureTimer.Restart();
+                return false;
+            }
+            catch (AggregateException e) when (e.InnerException is IOException)
+            {
+                // dotnet6.0 raises AggregateException when an IOException occurs.
+                _sendFailureTimer.Restart();
                 return false;
             }
 
             try
             {
-                // WriteAsync overload with a CancellationToken instance seems to not work.
-                return _namedPipe.WriteAsync(buffer, 0, length).Wait(_timeout);
+                // TODO: Blocking on an async Task with Wait() is bad practice and can deadlock;
+                // this should move to an async send path. The WriteAsync overload with a
+                // CancellationToken instance seems to not work, so the write cannot be cancelled.
+                var writeTask = _namedPipe.WriteAsync(buffer, 0, length);
+                if (writeTask.Wait(_timeout))
+                {
+                    _sendFailureTimer.Reset();
+                    return true;
+                }
+
+                // The write timed out. WriteAsync keeps running against the pipe with no way to
+                // cancel it, so abandon the stalled write and recreate the pipe: the next send
+                // reconnects cleanly instead of racing the orphaned write. Cool down first to
+                // avoid re-blocking for the full timeout on every subsequent send.
+                _sendFailureTimer.Restart();
+                ResetPipeAfterAbandonedWrite(writeTask);
+                return false;
             }
             catch (IOException)
             {
@@ -79,6 +147,7 @@ private bool SendBuffer(byte[] buffer, int length, bool allowRetry)
                 return SendBuffer(buffer, length, allowRetry: false);
             }
 
+            _sendFailureTimer.Restart();
             return false;
         }
     }

tests/StatsdClient.Tests/TelemetryTests.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Text;
+using System.Threading;
 using Moq;
 using NUnit.Framework;
 using StatsdClient;
@@ -131,6 +132,106 @@ public void CheckTags()
                 "globalTagKey:globalTagValue", _metrics[0]);
         }
 
+        [Test]
+        public void FlushesDoNotOverlap()
+        {
+            var concurrencyLock = new object();
+            int concurrentSends = 0;
+            int maxConcurrentSends = 0;
+
+            var transport = new Mock<ITransport>();
+            transport.SetupGet(s => s.TelemetryClientTransport).Returns("uds");
+            transport.Setup(s => s.Send(It.IsAny<byte[]>(), It.IsAny<int>()))
+                .Callback<byte[], int>((bytes, l) =>
+                {
+                    lock (concurrencyLock)
+                    {
+                        concurrentSends++;
+                        maxConcurrentSends = Math.Max(maxConcurrentSends, concurrentSends);
+                    }
+
+                    // Make a flush slow relative to the interval so overlapping callbacks would
+                    // be observable if the timer were still periodic.
+                    Thread.Sleep(20);
+
+                    lock (concurrencyLock)
+                    {
+                        concurrentSends--;
+                    }
+                });
+
+            using (new Telemetry(
+                new MetricSerializer(new SerializerHelper(null, null), string.Empty),
+                "1.0.0.0",
+                TimeSpan.FromMilliseconds(50),
+                transport.Object,
+                Array.Empty<string>(),
+                Tools.ExceptionHandler))
+            {
+                Thread.Sleep(TimeSpan.FromSeconds(1));
+            }
+
+            // A single flush sends sequentially on one thread, so concurrency stays 1 unless
+            // two flushes ran at once.
+            Assert.AreEqual(1, maxConcurrentSends);
+        }
+
+        [Test]
+        public void DisposeDoesNotDeadlockDuringInFlightFlush()
+        {
+            Exception caughtException = null;
+            int firstSend = 1;
+            var flushStarted = new ManualResetEventSlim(false);
+            var releaseFlush = new ManualResetEventSlim(false);
+
+            var transport = new Mock<ITransport>();
+            transport.SetupGet(s => s.TelemetryClientTransport).Returns("uds");
+            transport.Setup(s => s.Send(It.IsAny<byte[]>(), It.IsAny<int>()))
+                .Callback<byte[], int>((bytes, l) =>
+                {
+                    // Hold the first flush open so Dispose runs while a flush is in flight.
+                    if (Interlocked.Exchange(ref firstSend, 0) == 1)
+                    {
+                        flushStarted.Set();
+                        releaseFlush.Wait();
+                    }
+                });
+
+            var telemetry = new Telemetry(
+                new MetricSerializer(new SerializerHelper(null, null), string.Empty),
+                "1.0.0.0",
+                TimeSpan.FromMilliseconds(20),
+                transport.Object,
+                Array.Empty<string>(),
+                e => caughtException = e);
+
+            try
+            {
+                Assert.True(flushStarted.Wait(TimeSpan.FromSeconds(5)), "timer never dispatched a flush");
+
+                // Dispose must complete without waiting for the in-flight flush: the flush holds no lock
+                // while blocked in the transport, and the re-arm only takes _timerLock after the flush ends.
+                var disposeThread = new Thread(() => telemetry.Dispose());
+                disposeThread.Start();
+                Assert.True(disposeThread.Join(TimeSpan.FromSeconds(5)), "Dispose deadlocked while a flush was in flight");
+            }
+            finally
+            {
+                // Always release the blocked flush callback so a failed assertion above does not
+                // leave a thread-pool thread parked on releaseFlush for the rest of the run.
+                releaseFlush.Set();
+
+                // Dispose here too so a failed assertion before the in-test Dispose does not leave
+                // the timer running and leak into subsequent tests. Dispose is idempotent.
+                telemetry.Dispose();
+            }
+
+            // Let the in-flight flush finish and attempt to re-arm after Dispose; it must not throw.
+            Thread.Sleep(50);
+
+            Assert.IsNull(caughtException);
+        }
+
         private void AssertTelemetryReceived(Dictionary<string, int> expectedResults)
         {
             _telemetry.Flush();