diff --git a/src/StatsdClient/Telemetry.cs b/src/StatsdClient/Telemetry.cs index 3366b1db..0b3b8f4c 100644 --- a/src/StatsdClient/Telemetry.cs +++ b/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 _aggregatedContexts = new Dictionary(); private readonly Action _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) diff --git a/src/StatsdClient/Transport/NamedPipeTransport.cs b/src/StatsdClient/Transport/NamedPipeTransport.cs index d826e2b7..2fbd62e3 100644 --- a/src/StatsdClient/Transport/NamedPipeTransport.cs +++ b/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(); - 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; } } diff --git a/tests/StatsdClient.Tests/TelemetryTests.cs b/tests/StatsdClient.Tests/TelemetryTests.cs index 384293ea..d598be59 100644 --- a/tests/StatsdClient.Tests/TelemetryTests.cs +++ b/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(); + transport.SetupGet(s => s.TelemetryClientTransport).Returns("uds"); + transport.Setup(s => s.Send(It.IsAny(), It.IsAny())) + .Callback((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(), + 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(); + transport.SetupGet(s => s.TelemetryClientTransport).Returns("uds"); + transport.Setup(s => s.Send(It.IsAny(), It.IsAny())) + .Callback((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(), + 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 expectedResults) { _telemetry.Flush(); diff --git a/tests/StatsdClient.Tests/Transport/NamedPipeTransportTests.cs b/tests/StatsdClient.Tests/Transport/NamedPipeTransportTests.cs index 1a1bf5af..e9ab3f44 100644 --- a/tests/StatsdClient.Tests/Transport/NamedPipeTransportTests.cs +++ b/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 StartServerSingleRead(int bufferSize) { return StartServer(server =>