Metrics: Use dedicated thread instead of the thread pool for the defa… (

#1028) * Metrics: Use dedicated thread instead of the thread pool for the default aggregation cycle (code only, does not build not NetStandard). * Add missing reference to System.Threading.Thread on netstandard1.3 * changelog
microsoft · Dec 28, 2018 · 43feef1 · 43feef1
1 parent 1dc2fb4
commit 43feef1
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 15 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -5,6 +5,7 @@ This changelog will be used to generate documentation on [release notes page](ht
 ## Version 2.9.0-beta3
 - [Flatten IExtension and Unknown ITelemetry implementations for Rich Payload Event Source consumption](https://github.com/Microsoft/ApplicationInsights-dotnet/pull/1017)
 - [Fix: Start/StopOperation with W3C distributed tracing enabled does not track telemetry](https://github.com/Microsoft/ApplicationInsights-dotnet/pull/1031)
+- [Fix: Do not run metric aggregation on a thread pool thread](https://github.com/Microsoft/ApplicationInsights-dotnet/pull/1028)
 
 ## Version 2.9.0-beta2
 - [Remove unused reference to System.Web.Extensions](https://github.com/Microsoft/ApplicationInsights-dotnet/pull/956)

diff --git a/src/Microsoft.ApplicationInsights/Metrics/Implementation/DefaultAggregationPeriodCycle.cs b/src/Microsoft.ApplicationInsights/Metrics/Implementation/DefaultAggregationPeriodCycle.cs
@@ -16,26 +16,25 @@ internal class DefaultAggregationPeriodCycle
         [System.Diagnostics.CodeAnalysis.SuppressMessage("Naming Rules", "SA1310: C# Field must not contain an underscore", Justification = "By design: Structured name.")]
         private const int RunningState_Stopped = 2;
 
-        private readonly Action workerMethod;
-
         private readonly MetricAggregationManager aggregationManager;
         private readonly MetricManager metricManager;
 
+        private readonly TaskCompletionSource<bool> workerTaskCompletionControl;
+
         private int runningState;
-        private Task workerTask;
+        private Thread aggregationThread;
 
         public DefaultAggregationPeriodCycle(MetricAggregationManager aggregationManager, MetricManager metricManager)
         {
             Util.ValidateNotNull(aggregationManager, nameof(aggregationManager));
             Util.ValidateNotNull(metricManager, nameof(metricManager));
 
-            this.workerMethod = this.Run;
-
             this.aggregationManager = aggregationManager;
             this.metricManager = metricManager;
 
             this.runningState = RunningState_NotStarted;
-            this.workerTask = null;
+            this.workerTaskCompletionControl = new TaskCompletionSource<bool>();
+            this.aggregationThread = null;
         }
 
         ~DefaultAggregationPeriodCycle()
@@ -52,21 +51,28 @@ public bool Start()
                 return false; // Was already running or stopped.
             }
 
-            this.workerTask = Task.Run(this.workerMethod)
-                              .ContinueWith(
-                                        (t) => { this.workerTask = null; },
-                                        TaskContinuationOptions.ExecuteSynchronously);
+            // We create a new thread rather than using the thread pool.
+            // This is because inside of the main loop in the Run() method we use a synchronous wait.
+            // The reason for that is to prevent aggregation from being affected by potential thread pool starvation.
+            // As a result, Run() is a very long running task that occupies a thread forever.
+            // If we were to schedule Run() on the thread pool it would be possible that the thread chosen by the
+            // pool had run user code before. Such user code may be doing an asynchronous wait scheduled to
+            // continue on the same thread(e.g. this can occur when using a custom synchronization context or a 
+            // custom task scheduler). If such case the waiting user code will never continue.
+            // By creating our own thread, we guarantee no interactions with potentially incorrectly written async user code.
+
+            this.aggregationThread = new Thread(this.Run);
+            this.aggregationThread.Name = nameof(DefaultAggregationPeriodCycle);
+
+            this.aggregationThread.Start();
             return true;
         }
 
         public Task StopAsync()
         {
             Interlocked.Exchange(ref this.runningState, RunningState_Stopped);
-
-            // Benign race on being called very soon after start. Will miss a cycle but eventually complete correctly.
 
-            Task workerTask = this.workerTask;
-            return workerTask ?? Task.FromResult(true);
+            return this.workerTaskCompletionControl.Task;
         }
 
         public void FetchAndTrackMetrics()
@@ -139,11 +145,13 @@ private void Run()
                 DateTimeOffset now = DateTimeOffset.Now;
                 TimeSpan waitPeriod = GetNextCycleTargetTime(now) - now;
 
-                Task.Delay(waitPeriod).ConfigureAwait(continueOnCapturedContext: false).GetAwaiter().GetResult();
+                Thread.Sleep(waitPeriod);
 
                 int shouldBeRunning = Volatile.Read(ref this.runningState);
                 if (shouldBeRunning != RunningState_Running)
                 {
+                    this.aggregationThread = null;
+                    this.workerTaskCompletionControl.TrySetResult(true);
                     return;
                 }
 

diff --git a/src/Microsoft.ApplicationInsights/Microsoft.ApplicationInsights.csproj b/src/Microsoft.ApplicationInsights/Microsoft.ApplicationInsights.csproj
@@ -48,6 +48,7 @@
   <ItemGroup Condition="'$(TargetFramework)' == 'netstandard1.3'">
     <PackageReference Include="System.Diagnostics.StackTrace" Version="4.3.0" />
     <PackageReference Include="System.Net.Requests" Version="4.3.0" />
+    <PackageReference Include="System.Threading.Thread" Version="4.3.0" />
   </ItemGroup>
 
   <ItemGroup>