diff --git a/src/StatsdClient/Telemetry.cs b/src/StatsdClient/Telemetry.cs index 3366b1db..9101fdad 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 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) diff --git a/src/StatsdClient/Transport/NamedPipeTransport.cs b/src/StatsdClient/Transport/NamedPipeTransport.cs index d826e2b7..8b654de2 100644 --- a/src/StatsdClient/Transport/NamedPipeTransport.cs +++ b/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(); - 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; } } diff --git a/tests/StatsdClient.Tests/TelemetryTests.cs b/tests/StatsdClient.Tests/TelemetryTests.cs index 384293ea..da188ee9 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,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(); + 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); + + 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 expectedResults) { _telemetry.Flush(); diff --git a/tests/StatsdClient.Tests/Transport/NamedPipeTransportTests.cs b/tests/StatsdClient.Tests/Transport/NamedPipeTransportTests.cs index 1a1bf5af..fb2bc61d 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,93 @@ 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-" + Guid.NewGuid(), connectTimeout, cooldown)) + { + var stopwatch = Stopwatch.StartNew(); + Assert.False(transport.Send(_buffToSend, _buffToSend.Length)); + + stopwatch.Restart(); + for (int i = 0; i < 5; i++) + { + Assert.False(transport.Send(_buffToSend, _buffToSend.Length)); + } + + var cooldownSendsDuration = stopwatch.Elapsed; + + // Subsequent sends within the cooldown must fail fast instead of each blocking + // on Connect again. Assert the whole batch completes in a small fraction of a + // single connect timeout so the test fails if sends keep blocking, without + // relying on Connect actually blocking the full timeout (some Windows/.NET + // combinations fail a missing pipe quickly). + Assert.That(cooldownSendsDuration, Is.LessThan(TimeSpan.FromMilliseconds(connectTimeout.TotalMilliseconds * 0.5))); + } + } + + [Test] + public void WriteTimeoutTriggersCooldown() + { + var timeout = TimeSpan.FromSeconds(1); + var cooldown = TimeSpan.FromSeconds(10); + var pipeName = "writeCooldownPipeNameTest-" + Guid.NewGuid(); + 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(); + } + }); + + try + { + using (var transport = new NamedPipeTransport(pipeName, timeout, cooldown)) + { + // The OS may round the server's requested buffer size up, so use a payload + // far larger than the requested buffer to make the stalled write reliable. + var buff = new byte[_serverBufferSize * 1000]; + + var stopwatch = Stopwatch.StartNew(); + Assert.False(transport.Send(buff, buff.Length)); + + stopwatch.Restart(); + for (int i = 0; i < 5; i++) + { + Assert.False(transport.Send(buff, buff.Length)); + } + + var cooldownSendsDuration = stopwatch.Elapsed; + + // Subsequent sends within the cooldown must fail fast instead of each blocking + // on the stalled write again. Assert the whole batch completes in a small + // fraction of a single write timeout so the test fails if sends keep blocking. + Assert.That(cooldownSendsDuration, Is.LessThan(TimeSpan.FromMilliseconds(timeout.TotalMilliseconds * 0.5))); + } + } + finally + { + releaseServer.Set(); + Assert.True(serverTask.Wait(TimeSpan.FromSeconds(5)), "server task did not complete"); + } + } + private Task StartServerSingleRead(int bufferSize) { return StartServer(server =>