Avoid context switching

Author: kr-igor

tracer/src/Datadog.Trace/DataStreamsMonitoring/DataStreamsWriter.cs

@@ -24,14 +24,20 @@ internal sealed class DataStreamsWriter : IDataStreamsWriter
 {
     private const TaskCreationOptions TaskOptions = TaskCreationOptions.RunContinuationsAsynchronously;
 
+    // Drain the queues when this many items accumulate OR after DrainTimeoutMs — whichever comes first.
+    // At the queue limit of 10_000, this leaves 50% headroom even at 2× current peak throughput.
+    private const int DrainThreshold = 5_000;
+    private const int DrainTimeoutMs = 1_000;
+
     private static readonly IDatadogLogger Log = DatadogLogging.GetLoggerFor<DataStreamsWriter>();
 
     private readonly object _initLock = new();
     private readonly long _bucketDurationMs;
+
     private readonly BoundedConcurrentQueue<StatsPoint> _buffer = new(queueLimit: 10_000);
     private readonly BoundedConcurrentQueue<BacklogPoint> _backlogBuffer = new(queueLimit: 10_000);
     private readonly BoundedConcurrentQueue<DataStreamsTransactionInfo> _transactionBuffer = new(queueLimit: 10_000);
-    private readonly TimeSpan _waitTimeSpan = TimeSpan.FromMilliseconds(10);
+    private readonly ManualResetEventSlim _drainSignal = new(false);
     private readonly TimeSpan _flushSemaphoreWaitTime = TimeSpan.FromSeconds(1);
     private readonly DataStreamsAggregator _aggregator;
     private readonly IDiscoveryService _discoveryService;
@@ -125,6 +131,11 @@ public void Add(in StatsPoint point)
 
             if (_buffer.TryEnqueue(point))
             {
+                if (!_drainSignal.IsSet && _buffer.Count >= DrainThreshold)
+                {
+                    _drainSignal.Set();
+                }
+
                 return;
             }
         }
@@ -143,6 +154,11 @@ public void AddTransaction(in DataStreamsTransactionInfo transaction)
         {
             if (_transactionBuffer.TryEnqueue(transaction))
             {
+                if (!_drainSignal.IsSet && _transactionBuffer.Count >= DrainThreshold)
+                {
+                    _drainSignal.Set();
+                }
+
                 return;
             }
         }
@@ -181,6 +197,7 @@ public async Task DisposeAsync()
 #endif
         await FlushAndCloseAsync().ConfigureAwait(false);
         _flushSemaphore.Dispose();
+        _drainSignal.Dispose();
     }
 
     private async Task FlushAndCloseAsync()
@@ -190,6 +207,9 @@ private async Task FlushAndCloseAsync()
             return;
         }
 
+        // Wake ProcessQueueLoop immediately so it can observe _processExit and return.
+        _drainSignal.Set();
+
         // nothing else to do, since the writer was not fully initialized
         if (!Volatile.Read(ref _isInitialized) || _processTask == null || _flushTask == null)
         {
@@ -347,7 +367,8 @@ private void ProcessQueueLoop()
     {
         while (true)
         {
-            Thread.Sleep(_waitTimeSpan);
+            _drainSignal.Wait(DrainTimeoutMs);
+            _drainSignal.Reset();
 
             if (!_flushSemaphore.Wait(_flushSemaphoreWaitTime))
             {