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 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,33 @@ public void OnAggregatedContextFlush(MetricType metricType, int count)
 
         public void Dispose()
         {
-            _optionalTimer?.Change(Timeout.Infinite, Timeout.Infinite);
-            _optionalTimer?.Dispose();
+            lock (_timerLock)
+            {
+                _disposed = true;
+                _optionalTimer?.Change(Timeout.Infinite, Timeout.Infinite);
+                _optionalTimer?.Dispose();
+            }
+        }
+
+        private void OnTimerFlush()
+        {
+            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

@@ -9,14 +9,17 @@ internal class NamedPipeTransport : ITransport
     {
         private readonly NamedPipeClientStream _namedPipe;
         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 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);
             _timeout = timeout ?? TimeSpan.FromSeconds(2);
+            _connectionCooldown = connectionCooldown ?? TimeSpan.FromSeconds(5);
         }
 
         public TransportType TransportType => TransportType.NamedPipe;
@@ -47,6 +50,16 @@ public void Dispose()
 
         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 +69,23 @@ private bool SendBuffer(byte[] buffer, int length, bool allowRetry)
             }
             catch (TimeoutException)
             {
+                _sendFailureTimer.Restart();
                 return false;
             }
 
             try
             {
                 // WriteAsync overload with a CancellationToken instance seems to not work.
-                return _namedPipe.WriteAsync(buffer, 0, length).Wait(_timeout);
+                if (_namedPipe.WriteAsync(buffer, 0, length).Wait(_timeout))
+                {
+                    _sendFailureTimer.Reset();
+                    return true;
+                }
+
+                // The write timed out. The pipe still reports connected, so cool down to
+                // avoid re-blocking for the full timeout on every subsequent send.
+                _sendFailureTimer.Restart();
+                return false;
             }
             catch (IOException)
             {
@@ -79,6 +102,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,94 @@ 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);
+
+            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");
+
+            // Let the in-flight flush finish and attempt to re-arm after Dispose; it must not throw.
+            releaseFlush.Set();
+            Thread.Sleep(50);
+
+            Assert.IsNull(caughtException);
+        }
+
         private void AssertTelemetryReceived(Dictionary<string, int> expectedResults)
         {
             _telemetry.Flush();

tests/StatsdClient.Tests/Transport/NamedPipeTransportTests.cs

@@ -1,5 +1,6 @@
 #if OS_WINDOWS
 using System;
+using System.Diagnostics;
 using System.IO.Pipes;
 using System.Threading;
 using System.Threading.Tasks;
@@ -79,6 +80,86 @@ public void Reconnection()
             }
         }
 
+        [Test]
+        public void ConnectionCooldownAvoidsRepeatedBlocking()
+        {
+            var connectTimeout = TimeSpan.FromSeconds(1);
+            var cooldown = TimeSpan.FromSeconds(10);
+
+            // No server listens on this pipe, so Connect blocks for the full timeout and fails.
+            using (var transport = new NamedPipeTransport("cooldownPipeNameTest", connectTimeout, cooldown))
+            {
+                var stopwatch = Stopwatch.StartNew();
+                Assert.False(transport.Send(_buffToSend, _buffToSend.Length));
+                var firstSendDuration = stopwatch.Elapsed;
+
+                stopwatch.Restart();
+                for (int i = 0; i < 5; i++)
+                {
+                    Assert.False(transport.Send(_buffToSend, _buffToSend.Length));
+                }
+
+                var cooldownSendsDuration = stopwatch.Elapsed;
+
+                // The first send pays the connect timeout; subsequent sends within the
+                // cooldown fail fast instead of each blocking on Connect again.
+                Assert.That(firstSendDuration, Is.GreaterThan(TimeSpan.FromMilliseconds(500)));
+                Assert.That(cooldownSendsDuration, Is.LessThan(TimeSpan.FromMilliseconds(500)));
+            }
+        }
+
+        [Test]
+        public void WriteTimeoutTriggersCooldown()
+        {
+            var timeout = TimeSpan.FromSeconds(1);
+            var cooldown = TimeSpan.FromSeconds(10);
+            var pipeName = "writeCooldownPipeNameTest";
+            var releaseServer = new ManualResetEventSlim(false);
+
+            // The server connects but never reads, so its buffer fills and the client's
+            // write stalls until it times out.
+            var serverTask = Task.Run(() =>
+            {
+                using (var serverStream = new NamedPipeServerStream(
+                            pipeName,
+                            PipeDirection.In,
+                            1,
+                            PipeTransmissionMode.Byte,
+                            PipeOptions.Asynchronous,
+                            _serverBufferSize,
+                            0))
+                {
+                    serverStream.WaitForConnection();
+                    releaseServer.Wait();
+                }
+            });
+
+            using (var transport = new NamedPipeTransport(pipeName, timeout, cooldown))
+            {
+                var buff = new byte[_serverBufferSize * 10];
+
+                var stopwatch = Stopwatch.StartNew();
+                Assert.False(transport.Send(buff, buff.Length));
+                var firstSendDuration = stopwatch.Elapsed;
+
+                stopwatch.Restart();
+                for (int i = 0; i < 5; i++)
+                {
+                    Assert.False(transport.Send(buff, buff.Length));
+                }
+
+                var cooldownSendsDuration = stopwatch.Elapsed;
+
+                // The first send pays the write timeout; subsequent sends within the cooldown
+                // fail fast instead of each blocking on the stalled write again.
+                Assert.That(firstSendDuration, Is.GreaterThan(TimeSpan.FromMilliseconds(500)));
+                Assert.That(cooldownSendsDuration, Is.LessThan(TimeSpan.FromMilliseconds(500)));
+            }
+
+            releaseServer.Set();
+            serverTask.Wait();
+        }
+
         private Task<byte[]> StartServerSingleRead(int bufferSize)
         {
             return StartServer(server =>