use diagnostic events task scheduler so that we don't flood async event queue

src/Features/Core/Portable/Diagnostics/DiagnosticAnalyzerService_UpdateSource.cs

@@ -20,7 +20,10 @@ internal partial class DiagnosticAnalyzerService : IDiagnosticUpdateSource
         private DiagnosticAnalyzerService(IDiagnosticUpdateSourceRegistrationService registrationService) : this()
         {
             _eventMap = new EventMap();
-            _eventQueue = new SimpleTaskQueue(TaskScheduler.Default);
+
+            // use diagnostic event task scheduler so that we never flood async events queue with million of events.
+            // queue itself can handle huge number of events but we are seeing OOM due to captured data in pending events.
+            _eventQueue = new SimpleTaskQueue(new DiagnosticEventTaskScheduler(blockingUpperBound: 100));
 
             registrationService.Register(this);
         }

src/Features/Core/Portable/Diagnostics/DiagnosticEventTaskScheduler.cs

+// Copyright (c) Microsoft.  All Rights Reserved.  Licensed under the Apache License, Version 2.0.  See License.txt in the project root for license information.
+
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Roslyn.Utilities;
+
+namespace Microsoft.CodeAnalysis.Diagnostics
+{
+    /// <summary>
+    /// This task scheduler will block queuing new tasks if upper bound has met.
+    /// </summary>
+    internal class DiagnosticEventTaskScheduler : TaskScheduler
+    {
+        private readonly Task _mainTask;
+        private readonly BlockingCollection<Task> _tasks;
+
+        public DiagnosticEventTaskScheduler(int blockingUpperBound)
+        {
+            _tasks = new BlockingCollection<Task>(blockingUpperBound);
+
+            // portable layer doesnt support explicit thread creation. use long running task to create and hold onto a thread
+            _mainTask = Task.Factory.SafeStartNew(Start, CancellationToken.None,
+                TaskCreationOptions.DenyChildAttach | TaskCreationOptions.LongRunning, TaskScheduler.Default);
+        }
+
+        private void Start()
+        {
+            while (true)
+            {
+                var task = _tasks.Take();
+                bool ret = this.TryExecuteTask(task);
+            }
+        }
+
+        protected override void QueueTask(Task task)
+        {
+            _tasks.Add(task);
+        }
+
+        protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
+        {
+            // NOTE: TPL will ensure only one task ever run when running scheduled task. and since this is only used
+            // in diagnostic events, we know task will ever run sequencely. so no worry about reverted order here.
+            return this.TryExecuteTask(task);
+        }
+
+        protected override IEnumerable<Task> GetScheduledTasks()
+        {
+            // debugger will use this method to get scheduled tasks for this scheduler
+            return _tasks.ToArray();
+        }
+    }
+}
\ No newline at end of file

src/Features/Core/Portable/Diagnostics/DiagnosticService.cs

@@ -30,7 +30,10 @@ public DiagnosticService([ImportMany] IEnumerable<Lazy<IAsynchronousOperationLis
         {
             // queue to serialize events.
             _eventMap = new EventMap();
-            _eventQueue = new SimpleTaskQueue(TaskScheduler.Default);
+
+            // use diagnostic event task scheduler so that we never flood async events queue with million of events.
+            // queue itself can handle huge number of events but we are seeing OOM due to captured data in pending events.
+            _eventQueue = new SimpleTaskQueue(new DiagnosticEventTaskScheduler(blockingUpperBound: 100));
 
             _listener = new AggregateAsynchronousOperationListener(asyncListeners, FeatureAttribute.DiagnosticService);
 

src/Features/Core/Portable/Diagnostics/EngineV1/DiagnosticIncrementalAnalyzer.DiagnosticState.cs

@@ -35,12 +35,18 @@ public DiagnosticState(string stateName, VersionStamp version, string language)
                 _language = language;
             }
 
-            internal string Name
+            /// <summary>
+            /// Only For Testing
+            /// </summary>
+            internal string Name_TestingOnly
             {
                 get { return _stateName; }
             }
 
-            internal string Language
+            /// <summary>
+            /// Only For Testing
+            /// </summary>
+            internal string Language_TestingOnly
             {
                 get { return _language; }
             }

src/Features/Core/Portable/Diagnostics/EngineV1/DiagnosticIncrementalAnalyzer.StateManager.cs

@@ -178,7 +178,7 @@ private static void VerifyDiagnosticStates(IEnumerable<StateSet> stateSets)
                     {
                         var state = stateSet.GetState((StateType)i);
 
-                        if (!(set.Add(ValueTuple.Create(state.Language, state.Name))))
+                        if (!(set.Add(ValueTuple.Create(state.Language_TestingOnly, state.Name_TestingOnly))))
                         {
                             Contract.Fail();
                         }

src/Features/Core/Portable/Diagnostics/EngineV1/DiagnosticIncrementalAnalyzer.cs

@@ -181,7 +181,7 @@ private async Task AnalyzeSyntaxAsync(Document document, ImmutableHashSet<string
                         var data = await _executor.GetSyntaxAnalysisDataAsync(userDiagnosticDriver, stateSet, versions).ConfigureAwait(false);
                         if (data.FromCache)
                         {
-                            RaiseDiagnosticsCreated(StateType.Syntax, document.Id, stateSet, new SolutionArgument(document), data.Items);
+                            RaiseDiagnosticCreatedFromCacheIfNeeded(StateType.Syntax, document, stateSet, data.Items);
                             continue;
                         }
 
@@ -269,7 +269,7 @@ private async Task AnalyzeBodyDocumentAsync(Document document, SyntaxNode member
 
                         if (data.FromCache)
                         {
-                            RaiseDiagnosticsCreated(StateType.Document, document.Id, stateSet, new SolutionArgument(document), data.Items);
+                            RaiseDiagnosticCreatedFromCacheIfNeeded(StateType.Document, document, stateSet, data.Items);
                             continue;
                         }
 
@@ -306,7 +306,7 @@ private async Task AnalyzeDocumentAsync(Document document, VersionArgument versi
                         var data = await _executor.GetDocumentAnalysisDataAsync(userDiagnosticDriver, stateSet, versions).ConfigureAwait(false);
                         if (data.FromCache)
                         {
-                            RaiseDiagnosticsCreated(StateType.Document, document.Id, stateSet, new SolutionArgument(document), data.Items);
+                            RaiseDiagnosticCreatedFromCacheIfNeeded(StateType.Document, document, stateSet, data.Items);
                             continue;
                         }
 
@@ -363,7 +363,7 @@ private async Task AnalyzeProjectAsync(Project project, CancellationToken cancel
                         var data = await _executor.GetProjectAnalysisDataAsync(analyzerDriver, stateSet, versions).ConfigureAwait(false);
                         if (data.FromCache)
                         {
-                            RaiseProjectDiagnosticsUpdated(project, stateSet, data.Items);
+                            RaiseProjectDiagnosticsUpdatedIfNeeded(project, stateSet, ImmutableArray<DiagnosticData>.Empty, data.Items);
                             continue;
                         }
 
@@ -617,6 +617,11 @@ private static bool CheckSemanticVersions(Project project, AnalysisData existing
                    project.CanReusePersistedDependentSemanticVersion(versions.ProjectVersion, versions.DataVersion, existingData.DataVersion);
         }
 
+        private void RaiseDiagnosticCreatedFromCacheIfNeeded(StateType type, Document document, StateSet stateSet, ImmutableArray<DiagnosticData> items)
+        {
+            RaiseDocumentDiagnosticsUpdatedIfNeeded(type, document, stateSet, ImmutableArray<DiagnosticData>.Empty, items);
+        }
+
         private void RaiseDocumentDiagnosticsUpdatedIfNeeded(
             StateType type, Document document, StateSet stateSet, ImmutableArray<DiagnosticData> existingItems, ImmutableArray<DiagnosticData> newItems)
         {

src/Features/Core/Portable/Features.csproj

@@ -177,6 +177,7 @@
     <Compile Include="Diagnostics\AnalyzerHelper.cs" />
     <Compile Include="Diagnostics\AbstractHostDiagnosticUpdateSource.cs" />
     <Compile Include="Diagnostics\AnalyzerUpdateArgsId.cs" />
+    <Compile Include="Diagnostics\DiagnosticEventTaskScheduler.cs" />
     <Compile Include="Diagnostics\DiagnosticService_UpdateSourceRegistrationService.cs" />
     <Compile Include="Common\UpdatedEventArgs.cs" />
     <Compile Include="Diagnostics\HostDiagnosticAnalyzerPackage.cs" />