Remove log packet aggregation (QuantConnect#6538)

Martin-Molinero · web-flow · commit 37dca9d45864 · 2022-08-11T19:06:11.000-03:00
- Remove log packet aggregation
- Clean up Queue api usage
diff --git a/Algorithm.Framework/Execution/SpreadExecutionModel.cs b/Algorithm.Framework/Execution/SpreadExecutionModel.cs
@@ -51,7 +51,7 @@ public override void Execute(QCAlgorithm algorithm, IPortfolioTarget[] targets)
             _targetsCollection.AddRange(targets);
 
             // for performance we check count value, OrderByMarginImpact and ClearFulfilled are expensive to call
-            if (_targetsCollection.Count > 0)
+            if (!_targetsCollection.IsEmpty)
             {
                 foreach (var target in _targetsCollection.OrderByMarginImpact(algorithm))
                 {
diff --git a/Algorithm.Framework/Execution/SpreadExecutionModel.py b/Algorithm.Framework/Execution/SpreadExecutionModel.py
@@ -35,7 +35,7 @@ def Execute(self, algorithm, targets):
         self.targetsCollection.AddRange(targets)
 
         # for performance we check count value, OrderByMarginImpact and ClearFulfilled are expensive to call
-        if self.targetsCollection.Count > 0:
+        if not self.targetsCollection.IsEmpty:
             for target in self.targetsCollection.OrderByMarginImpact(algorithm):
                 symbol = target.Symbol
                 
diff --git a/Algorithm.Framework/Execution/StandardDeviationExecutionModel.cs b/Algorithm.Framework/Execution/StandardDeviationExecutionModel.cs
@@ -1,4 +1,4 @@
-﻿/*
+/*
  * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
  * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
  *
@@ -74,7 +74,7 @@ public override void Execute(QCAlgorithm algorithm, IPortfolioTarget[] targets)
             _targetsCollection.AddRange(targets);
 
             // for performance we check count value, OrderByMarginImpact and ClearFulfilled are expensive to call
-            if (_targetsCollection.Count > 0)
+            if (!_targetsCollection.IsEmpty)
             {
                 foreach (var target in _targetsCollection.OrderByMarginImpact(algorithm))
                 {
@@ -213,4 +213,4 @@ public SymbolData(QCAlgorithm algorithm, Security security, int period, Resoluti
             }
         }
     }
-}
+}
diff --git a/Algorithm.Framework/Execution/StandardDeviationExecutionModel.py b/Algorithm.Framework/Execution/StandardDeviationExecutionModel.py
@@ -1,4 +1,4 @@
-﻿# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
+# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
 # Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -47,7 +47,7 @@ def Execute(self, algorithm, targets):
         self.targetsCollection.AddRange(targets)
 
         # for performance we check count value, OrderByMarginImpact and ClearFulfilled are expensive to call
-        if self.targetsCollection.Count > 0:
+        if not self.targetsCollection.IsEmpty:
             for target in self.targetsCollection.OrderByMarginImpact(algorithm):
                 symbol = target.Symbol
 
diff --git a/Algorithm.Framework/Execution/VolumeWeightedAveragePriceExecutionModel.cs b/Algorithm.Framework/Execution/VolumeWeightedAveragePriceExecutionModel.cs
@@ -1,4 +1,4 @@
-﻿/*
+/*
  * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
  * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
  *
@@ -53,7 +53,7 @@ public override void Execute(QCAlgorithm algorithm, IPortfolioTarget[] targets)
             _targetsCollection.AddRange(targets);
 
             // for performance we check count value, OrderByMarginImpact and ClearFulfilled are expensive to call
-            if (_targetsCollection.Count > 0)
+            if (!_targetsCollection.IsEmpty)
             {
                 foreach (var target in _targetsCollection.OrderByMarginImpact(algorithm))
                 {
@@ -183,4 +183,4 @@ public SymbolData(QCAlgorithm algorithm, Security security)
             }
         }
     }
-}
+}
diff --git a/Algorithm.Framework/Execution/VolumeWeightedAveragePriceExecutionModel.py b/Algorithm.Framework/Execution/VolumeWeightedAveragePriceExecutionModel.py
@@ -1,4 +1,4 @@
-﻿# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
+# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
 # Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -38,7 +38,7 @@ def Execute(self, algorithm, targets):
         self.targetsCollection.AddRange(targets)
 
         # for performance we check count value, OrderByMarginImpact and ClearFulfilled are expensive to call
-        if self.targetsCollection.Count > 0:
+        if not self.targetsCollection.IsEmpty:
             for target in self.targetsCollection.OrderByMarginImpact(algorithm):
                 symbol = target.Symbol
 
diff --git a/Algorithm.Framework/Selection/InceptionDateUniverseSelectionModel.cs b/Algorithm.Framework/Selection/InceptionDateUniverseSelectionModel.cs
@@ -1,4 +1,4 @@
-﻿/*
+/*
  * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
  * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
  *
@@ -59,7 +59,7 @@ public override IEnumerable<string> Select(QCAlgorithm algorithm, DateTime date)
         {
             // Move Symbols that are trading from the queue to a list 
             var added = new List<string>();
-            while (_queue.Count > 0 && _queue.First().Value <= date)
+            while (_queue.TryPeek(out var keyValuePair) && keyValuePair.Value <= date)
             {
                 added.Add(_queue.Dequeue().Key);
             }
@@ -75,4 +75,4 @@ public override IEnumerable<string> Select(QCAlgorithm algorithm, DateTime date)
             return _symbols;
         }
     }
-}
+}
diff --git a/Algorithm/Execution/ImmediateExecutionModel.cs b/Algorithm/Execution/ImmediateExecutionModel.cs
@@ -1,4 +1,4 @@
-﻿/*
+/*
  * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
  * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
  *
@@ -36,8 +36,8 @@ public class ImmediateExecutionModel : ExecutionModel
         public override void Execute(QCAlgorithm algorithm, IPortfolioTarget[] targets)
         {
             _targetsCollection.AddRange(targets);
-            // for performance we check count value, OrderByMarginImpact and ClearFulfilled are expensive to call
-            if (_targetsCollection.Count > 0)
+            // for performance we if empty, OrderByMarginImpact and ClearFulfilled are expensive to call
+            if (!_targetsCollection.IsEmpty)
             {
                 foreach (var target in _targetsCollection.OrderByMarginImpact(algorithm))
                 {
diff --git a/Algorithm/Execution/ImmediateExecutionModel.py b/Algorithm/Execution/ImmediateExecutionModel.py
@@ -1,4 +1,4 @@
-﻿# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
+# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
 # Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -28,7 +28,7 @@ def Execute(self, algorithm, targets):
 
         # for performance we check count value, OrderByMarginImpact and ClearFulfilled are expensive to call
         self.targetsCollection.AddRange(targets)
-        if self.targetsCollection.Count > 0:
+        if not self.targetsCollection.IsEmpty:
             for target in self.targetsCollection.OrderByMarginImpact(algorithm):
                 security = algorithm.Securities[target.Symbol]
                 # calculate remaining quantity to be ordered
diff --git a/Common/Algorithm/Framework/Portfolio/PortfolioTargetCollection.cs b/Common/Algorithm/Framework/Portfolio/PortfolioTargetCollection.cs
@@ -1,4 +1,4 @@
-﻿/*
+/*
  * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
  * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
  *
@@ -34,6 +34,11 @@ public class PortfolioTargetCollection : ICollection<IPortfolioTarget>, IDiction
         /// </summary>
         public int Count => _targets.Skip(0).Count();
 
+        /// <summary>
+        /// True if there is no target in the collection
+        /// </summary>
+        public bool IsEmpty => _targets.IsEmpty;
+
         /// <summary>
         /// Gets `false`. This collection is not read-only.
         /// </summary>
@@ -317,7 +322,11 @@ private T WithDictionary<T>(Func<IDictionary<Symbol, IPortfolioTarget>, T> func)
         /// <param name="algorithm">The algorithm instance</param>
         public IEnumerable<IPortfolioTarget> OrderByMarginImpact(IAlgorithm algorithm)
         {
+            if (IsEmpty)
+            {
+                return Enumerable.Empty<IPortfolioTarget>();
+            }
             return this.OrderTargetsByMarginImpact(algorithm);
         }
     }
-}
+}
diff --git a/Common/Packets/Packet.cs b/Common/Packets/Packet.cs
@@ -1,4 +1,4 @@
-﻿/*
+/*
  * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
  * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
  *
@@ -165,6 +165,9 @@ public enum PacketType
         OptimizationStatus,
 
         /// Optimization work result
-        OptimizationResult
+        OptimizationResult,
+
+        /// Aggregated packets
+        Aggregated
     }
 }
diff --git a/Common/Util/BusyCollection.cs b/Common/Util/BusyCollection.cs
@@ -1,4 +1,4 @@
-﻿/*
+/*
  * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
  * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
  *
@@ -46,7 +46,7 @@ public class BusyCollection<T> : IBusyCollection<T>
         /// <summary>
         /// Returns true if processing, false otherwise
         /// </summary>
-        public bool IsBusy => _collection.Count > 0 || !_processingCompletedEvent.IsSet;
+        public bool IsBusy => !_collection.IsEmpty || !_processingCompletedEvent.IsSet;
 
         /// <summary>
         /// Adds the items to this collection
@@ -129,4 +129,4 @@ public void CompleteAdding()
             _completedAdding = true;
         }
     }
-}
+}
diff --git a/Common/Util/FixedSizeHashQueue.cs b/Common/Util/FixedSizeHashQueue.cs
@@ -62,13 +62,7 @@ public bool Add(T item)
         /// </summary>
         public bool TryPeek(out T item)
         {
-            if (_queue.Count > 0)
-            {
-                item = _queue.Peek();
-                return true;
-            }
-            item = default(T);
-            return false;
+            return _queue.TryPeek(out item);
         }
 
         /// <summary>
@@ -113,4 +107,4 @@ IEnumerator IEnumerable.GetEnumerator()
             return GetEnumerator();
         }
     }
-}
+}
diff --git a/Engine/LeanEngineSystemHandlers.cs b/Engine/LeanEngineSystemHandlers.cs
@@ -131,9 +131,9 @@ public void Dispose()
         {
             Log.Trace("LeanEngineSystemHandlers.Dispose(): start...");
 
-            Api.DisposeSafely();
             LeanManager.DisposeSafely();
             Notify.DisposeSafely();
+            Api.DisposeSafely();
 
             Log.Trace("LeanEngineSystemHandlers.Dispose(): Disposed of system handlers.");
         }
diff --git a/Engine/Results/BacktestingResultHandler.cs b/Engine/Results/BacktestingResultHandler.cs
@@ -106,10 +106,10 @@ protected override void Run()
         {
             try
             {
-                while (!(ExitTriggered && Messages.Count == 0))
+                while (!(ExitTriggered && Messages.IsEmpty))
                 {
                     //While there's no work to do, go back to the algorithm:
-                    if (Messages.Count == 0)
+                    if (Messages.IsEmpty)
                     {
                         ExitEvent.WaitOne(50);
                     }
diff --git a/Engine/Results/BaseResultsHandler.cs b/Engine/Results/BaseResultsHandler.cs
@@ -38,6 +38,7 @@ namespace QuantConnect.Lean.Engine.Results
     /// </summary>
     public abstract class BaseResultsHandler
     {
+        private bool _packetDroppedWarning;
         // used for resetting out/error upon completion
         private static readonly TextWriter StandardOut = Console.Out;
         private static readonly TextWriter StandardError = Console.Error;
@@ -47,7 +48,7 @@ public abstract class BaseResultsHandler
         /// <summary>
         /// The main loop update interval
         /// </summary>
-        protected virtual TimeSpan MainUpdateInterval => TimeSpan.FromSeconds(3);
+        protected virtual TimeSpan MainUpdateInterval { get; } = TimeSpan.FromSeconds(3);
 
         /// <summary>
         /// The chart update interval
@@ -731,16 +732,14 @@ protected void ProcessAlgorithmLogs(int? messageQueueLimit = null)
 
         private void ProcessAlgorithmLogsImpl(ConcurrentQueue<string> concurrentQueue, PacketType packetType, int? messageQueueLimit = null)
         {
-            if (concurrentQueue.Count <= 0)
+            if (concurrentQueue.IsEmpty)
             {
                 return;
             }
 
-            var result = new List<string>();
             var endTime = DateTime.UtcNow.AddMilliseconds(250).Ticks;
-            string message;
             var currentMessageCount = -1;
-            while (DateTime.UtcNow.Ticks < endTime && concurrentQueue.TryDequeue(out message))
+            while (DateTime.UtcNow.Ticks < endTime && concurrentQueue.TryDequeue(out var message))
             {
                 if (messageQueueLimit.HasValue)
                 {
@@ -751,19 +750,17 @@ private void ProcessAlgorithmLogsImpl(ConcurrentQueue<string> concurrentQueue, P
                     }
                     if (currentMessageCount > messageQueueLimit)
                     {
+                        if (!_packetDroppedWarning)
+                        {
+                            _packetDroppedWarning = true;
+                            // this shouldn't happen in most cases, queue limit is high and consumed often but just in case let's not silently drop packets without a warning
+                            Messages.Enqueue(new HandledErrorPacket(AlgorithmId, "Your algorithm messaging has been rate limited to prevent browser flooding."));
+                        }
                         //if too many in the queue already skip the logging and drop the messages
                         continue;
                     }
                 }
-                AddToLogStore(message);
-                result.Add(message);
-                // increase count after we add
-                currentMessageCount++;
-            }
 
-            if (result.Count > 0)
-            {
-                message = string.Join(Environment.NewLine, result);
                 if (packetType == PacketType.Debug)
                 {
                     Messages.Enqueue(new DebugPacket(ProjectId, AlgorithmId, CompileId, message));
@@ -776,6 +773,10 @@ private void ProcessAlgorithmLogsImpl(ConcurrentQueue<string> concurrentQueue, P
                 {
                     Messages.Enqueue(new HandledErrorPacket(AlgorithmId, message));
                 }
+                AddToLogStore(message);
+
+                // increase count after we add
+                currentMessageCount++;
             }
         }
     }
diff --git a/Engine/Results/LiveTradingResultHandler.cs b/Engine/Results/LiveTradingResultHandler.cs
@@ -117,7 +117,7 @@ protected override void Run()
             ExitEvent.WaitOne(3000);
 
             // -> 1. Run Primary Sender Loop: Continually process messages from queue as soon as they arrive.
-            while (!(ExitTriggered && Messages.Count == 0))
+            while (!(ExitTriggered && Messages.IsEmpty))
             {
                 try
                 {
@@ -131,7 +131,7 @@ protected override void Run()
                     //2. Update the packet scanner:
                     Update();
 
-                    if (Messages.Count == 0)
+                    if (Messages.IsEmpty)
                     {
                         // prevent thread lock/tight loop when there's no work to be done
                         ExitEvent.WaitOne(Time.GetSecondUnevenWait(1000));
@@ -1070,8 +1070,8 @@ public virtual void ProcessSynchronousEvents(bool forceProcess = false)
             }
 
             //Send all the notification messages but timeout within a second, or if this is a force process, wait till its done.
-            var start = DateTime.UtcNow;
-            while (Algorithm.Notify.Messages.Count > 0 && (DateTime.UtcNow < start.AddSeconds(1) || forceProcess))
+            var timeout = DateTime.UtcNow.AddSeconds(1);
+            while (!Algorithm.Notify.Messages.IsEmpty && (DateTime.UtcNow < timeout || forceProcess))
             {
                 Notification message;
                 if (Algorithm.Notify.Messages.TryDequeue(out message))
diff --git a/Messaging/EventMessagingHandler.cs b/Messaging/EventMessagingHandler.cs
@@ -106,9 +106,9 @@ public void Send(Packet packet)
             }
 
             //Catch up if this is the first time
-            while (_queue.Count > 0)
+            while (_queue.TryDequeue(out var item))
             {
-                ProcessPacket(_queue.Dequeue());
+                ProcessPacket(item);
             }
 
             //Finally process this new packet
@@ -135,9 +135,9 @@ public void SendNotification(Notification notification)
         /// </summary>
         public void SendEnqueuedPackets()
         {
-            while (_queue.Count > 0 && _loaded)
+            while (_loaded && _queue.TryDequeue(out var item))
             {
-                ProcessPacket(_queue.Dequeue());
+                ProcessPacket(item);
             }
         }
 
diff --git a/Optimizer/LeanOptimizer.cs b/Optimizer/LeanOptimizer.cs
diff --git a/Tests/Engine/Setup/BrokerageSetupHandlerTests.cs b/Tests/Engine/Setup/BrokerageSetupHandlerTests.cs
diff --git a/Tests/Optimizer/Strategies/EulerSearchOptimizationStrategyTests.cs b/Tests/Optimizer/Strategies/EulerSearchOptimizationStrategyTests.cs
diff --git a/Tests/Optimizer/Strategies/OptimizationStrategyTests.cs b/Tests/Optimizer/Strategies/OptimizationStrategyTests.cs

Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,7 @@ public override void Execute(QCAlgorithm algorithm, IPortfolioTarget[] targets)`
`51`	`51`	`_targetsCollection.AddRange(targets);`
`52`	`52`
`53`	`53`	`// for performance we check count value, OrderByMarginImpact and ClearFulfilled are expensive to call`
`54`		`- if (_targetsCollection.Count > 0)`
	`54`	`+ if (!_targetsCollection.IsEmpty)`
`55`	`55`	`{`
`56`	`56`	`foreach (var target in _targetsCollection.OrderByMarginImpact(algorithm))`
`57`	`57`	`{`
Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-/*`
	`1`	`+/*`
`2`	`2`	`* QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.`
`3`	`3`	`* Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.`
`4`	`4`	`*`
`@@ -74,7 +74,7 @@ public override void Execute(QCAlgorithm algorithm, IPortfolioTarget[] targets)`
`74`	`74`	`_targetsCollection.AddRange(targets);`
`75`	`75`
`76`	`76`	`// for performance we check count value, OrderByMarginImpact and ClearFulfilled are expensive to call`
`77`		`- if (_targetsCollection.Count > 0)`
	`77`	`+ if (!_targetsCollection.IsEmpty)`
`78`	`78`	`{`
`79`	`79`	`foreach (var target in _targetsCollection.OrderByMarginImpact(algorithm))`
`80`	`80`	`{`
`@@ -213,4 +213,4 @@ public SymbolData(QCAlgorithm algorithm, Security security, int period, Resoluti`
`213`	`213`	`}`
`214`	`214`	`}`
`215`	`215`	`}`
`216`		`-}`
	`216`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -62,13 +62,7 @@ public bool Add(T item)`
`62`	`62`	`/// </summary>`
`63`	`63`	`public bool TryPeek(out T item)`
`64`	`64`	`{`
`65`		`- if (_queue.Count > 0)`
`66`		`- {`
`67`		`- item = _queue.Peek();`
`68`		`- return true;`
`69`		`- }`
`70`		`- item = default(T);`
`71`		`- return false;`
	`65`	`+ return _queue.TryPeek(out item);`
`72`	`66`	`}`
`73`	`67`
`74`	`68`	`/// <summary>`
`@@ -113,4 +107,4 @@ IEnumerator IEnumerable.GetEnumerator()`
`113`	`107`	`return GetEnumerator();`
`114`	`108`	`}`
`115`	`109`	`}`
`116`		`-}`
	`110`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -131,9 +131,9 @@ public void Dispose()`
`131`	`131`	`{`
`132`	`132`	`Log.Trace("LeanEngineSystemHandlers.Dispose(): start...");`
`133`	`133`
`134`		`- Api.DisposeSafely();`
`135`	`134`	`LeanManager.DisposeSafely();`
`136`	`135`	`Notify.DisposeSafely();`
	`136`	`+ Api.DisposeSafely();`
`137`	`137`
`138`	`138`	`Log.Trace("LeanEngineSystemHandlers.Dispose(): Disposed of system handlers.");`
`139`	`139`	`}`
Original file line number	Diff line number	Diff line change
`@@ -106,10 +106,10 @@ protected override void Run()`
`106`	`106`	`{`
`107`	`107`	`try`
`108`	`108`	`{`
`109`		`- while (!(ExitTriggered && Messages.Count == 0))`
	`109`	`+ while (!(ExitTriggered && Messages.IsEmpty))`
`110`	`110`	`{`
`111`	`111`	`//While there's no work to do, go back to the algorithm:`
`112`		`- if (Messages.Count == 0)`
	`112`	`+ if (Messages.IsEmpty)`
`113`	`113`	`{`
`114`	`114`	`ExitEvent.WaitOne(50);`
`115`	`115`	`}`
Original file line number	Diff line number	Diff line change
`@@ -38,6 +38,7 @@ namespace QuantConnect.Lean.Engine.Results`
`38`	`38`	`/// </summary>`
`39`	`39`	`public abstract class BaseResultsHandler`
`40`	`40`	`{`
	`41`	`+ private bool _packetDroppedWarning;`
`41`	`42`	`// used for resetting out/error upon completion`
`42`	`43`	`private static readonly TextWriter StandardOut = Console.Out;`
`43`	`44`	`private static readonly TextWriter StandardError = Console.Error;`
`@@ -47,7 +48,7 @@ public abstract class BaseResultsHandler`
`47`	`48`	`/// <summary>`
`48`	`49`	`/// The main loop update interval`
`49`	`50`	`/// </summary>`
`50`		`- protected virtual TimeSpan MainUpdateInterval => TimeSpan.FromSeconds(3);`
	`51`	`+ protected virtual TimeSpan MainUpdateInterval { get; } = TimeSpan.FromSeconds(3);`
`51`	`52`
`52`	`53`	`/// <summary>`
`53`	`54`	`/// The chart update interval`
`@@ -731,16 +732,14 @@ protected void ProcessAlgorithmLogs(int? messageQueueLimit = null)`
`731`	`732`
`732`	`733`	`private void ProcessAlgorithmLogsImpl(ConcurrentQueue<string> concurrentQueue, PacketType packetType, int? messageQueueLimit = null)`
`733`	`734`	`{`
`734`		`- if (concurrentQueue.Count <= 0)`
	`735`	`+ if (concurrentQueue.IsEmpty)`
`735`	`736`	`{`
`736`	`737`	`return;`
`737`	`738`	`}`
`738`	`739`
`739`		`- var result = new List<string>();`
`740`	`740`	`var endTime = DateTime.UtcNow.AddMilliseconds(250).Ticks;`
`741`		`- string message;`
`742`	`741`	`var currentMessageCount = -1;`
`743`		`- while (DateTime.UtcNow.Ticks < endTime && concurrentQueue.TryDequeue(out message))`
	`742`	`+ while (DateTime.UtcNow.Ticks < endTime && concurrentQueue.TryDequeue(out var message))`
`744`	`743`	`{`
`745`	`744`	`if (messageQueueLimit.HasValue)`
`746`	`745`	`{`
`@@ -751,19 +750,17 @@ private void ProcessAlgorithmLogsImpl(ConcurrentQueue<string> concurrentQueue, P`
`751`	`750`	`}`
`752`	`751`	`if (currentMessageCount > messageQueueLimit)`
`753`	`752`	`{`
	`753`	`+ if (!_packetDroppedWarning)`
	`754`	`+ {`
	`755`	`+ _packetDroppedWarning = true;`
	`756`	`+ // this shouldn't happen in most cases, queue limit is high and consumed often but just in case let's not silently drop packets without a warning`
	`757`	`+ Messages.Enqueue(new HandledErrorPacket(AlgorithmId, "Your algorithm messaging has been rate limited to prevent browser flooding."));`
	`758`	`+ }`
`754`	`759`	`//if too many in the queue already skip the logging and drop the messages`
`755`	`760`	`continue;`
`756`	`761`	`}`
`757`	`762`	`}`
`758`		`- AddToLogStore(message);`
`759`		`- result.Add(message);`
`760`		`- // increase count after we add`
`761`		`- currentMessageCount++;`
`762`		`- }`
`763`	`763`
`764`		`- if (result.Count > 0)`
`765`		`- {`
`766`		`- message = string.Join(Environment.NewLine, result);`
`767`	`764`	`if (packetType == PacketType.Debug)`
`768`	`765`	`{`
`769`	`766`	`Messages.Enqueue(new DebugPacket(ProjectId, AlgorithmId, CompileId, message));`
`@@ -776,6 +773,10 @@ private void ProcessAlgorithmLogsImpl(ConcurrentQueue<string> concurrentQueue, P`
`776`	`773`	`{`
`777`	`774`	`Messages.Enqueue(new HandledErrorPacket(AlgorithmId, message));`
`778`	`775`	`}`
	`776`	`+ AddToLogStore(message);`
	`777`	`+`
	`778`	`+ // increase count after we add`
	`779`	`+ currentMessageCount++;`
`779`	`780`	`}`
`780`	`781`	`}`
`781`	`782`	`}`
Original file line number	Diff line number	Diff line change
`@@ -117,7 +117,7 @@ protected override void Run()`
`117`	`117`	`ExitEvent.WaitOne(3000);`
`118`	`118`
`119`	`119`	`// -> 1. Run Primary Sender Loop: Continually process messages from queue as soon as they arrive.`
`120`		`- while (!(ExitTriggered && Messages.Count == 0))`
	`120`	`+ while (!(ExitTriggered && Messages.IsEmpty))`
`121`	`121`	`{`
`122`	`122`	`try`
`123`	`123`	`{`
`@@ -131,7 +131,7 @@ protected override void Run()`
`131`	`131`	`//2. Update the packet scanner:`
`132`	`132`	`Update();`
`133`	`133`
`134`		`- if (Messages.Count == 0)`
	`134`	`+ if (Messages.IsEmpty)`
`135`	`135`	`{`
`136`	`136`	`// prevent thread lock/tight loop when there's no work to be done`
`137`	`137`	`ExitEvent.WaitOne(Time.GetSecondUnevenWait(1000));`
`@@ -1070,8 +1070,8 @@ public virtual void ProcessSynchronousEvents(bool forceProcess = false)`
`1070`	`1070`	`}`
`1071`	`1071`
`1072`	`1072`	`//Send all the notification messages but timeout within a second, or if this is a force process, wait till its done.`
`1073`		`- var start = DateTime.UtcNow;`
`1074`		`- while (Algorithm.Notify.Messages.Count > 0 && (DateTime.UtcNow < start.AddSeconds(1) \|\| forceProcess))`
	`1073`	`+ var timeout = DateTime.UtcNow.AddSeconds(1);`
	`1074`	`+ while (!Algorithm.Notify.Messages.IsEmpty && (DateTime.UtcNow < timeout \|\| forceProcess))`
`1075`	`1075`	`{`
`1076`	`1076`	`Notification message;`
`1077`	`1077`	`if (Algorithm.Notify.Messages.TryDequeue(out message))`
Original file line number	Diff line number	Diff line change
`@@ -106,9 +106,9 @@ public void Send(Packet packet)`
`106`	`106`	`}`
`107`	`107`
`108`	`108`	`//Catch up if this is the first time`
`109`		`- while (_queue.Count > 0)`
	`109`	`+ while (_queue.TryDequeue(out var item))`
`110`	`110`	`{`
`111`		`- ProcessPacket(_queue.Dequeue());`
	`111`	`+ ProcessPacket(item);`
`112`	`112`	`}`
`113`	`113`
`114`	`114`	`//Finally process this new packet`
`@@ -135,9 +135,9 @@ public void SendNotification(Notification notification)`
`135`	`135`	`/// </summary>`
`136`	`136`	`public void SendEnqueuedPackets()`
`137`	`137`	`{`
`138`		`- while (_queue.Count > 0 && _loaded)`
	`138`	`+ while (_loaded && _queue.TryDequeue(out var item))`
`139`	`139`	`{`
`140`		`- ProcessPacket(_queue.Dequeue());`
	`140`	`+ ProcessPacket(item);`
`141`	`141`	`}`
`142`	`142`	`}`
`143`	`143`