Add algorithm thread

- Adding `WorkerThread` class, wrapper for a worker thread that will
execute given `Actions`.
    - Algorithm related code (`Construction`, `Initialization`,
    `Execution` will be executed by the same `WorkerThread` instance,
    this is required for `Python` debugging.

Author: Martin-Molinero

AlgorithmFactory/Loader.cs

@@ -24,6 +24,7 @@
 using QuantConnect.Logging;
 using QuantConnect.AlgorithmFactory.Python.Wrappers;
 using QuantConnect.Python;
+using QuantConnect.Util;
 
 namespace QuantConnect.AlgorithmFactory
 {
@@ -42,6 +43,9 @@ public class Loader : MarshalByRefObject
         // Defines how we resolve a list of type names into a single type name to be instantiated
         private readonly Func<List<string>, string> _multipleTypeNameResolverFunction;
 
+        // The worker thread instance the loader will use if not null
+        private readonly WorkerThread _workerThread;
+
         /// <summary>
         /// Memory space of the user algorithm
         /// </summary>
@@ -84,10 +88,11 @@ public Loader()
         /// for the QuantConnect.Algorithm assembly in this solution.  In order to pick the correct type, consumers must specify how to pick the type,
         /// that's what this function does, it picks the correct type from the list of types found within the assembly.
         /// </param>
-        public Loader(Language language, TimeSpan loaderTimeLimit, Func<List<string>, string> multipleTypeNameResolverFunction)
+        /// <param name="workerThread">The worker thread instance the loader should use</param>
+        public Loader(Language language, TimeSpan loaderTimeLimit, Func<List<string>, string> multipleTypeNameResolverFunction, WorkerThread workerThread = null)
         {
             _language = language;
-
+            _workerThread = workerThread;
             if (multipleTypeNameResolverFunction == null)
             {
                 throw new ArgumentNullException("multipleTypeNameResolverFunction");
@@ -348,7 +353,7 @@ public bool TryCreateAlgorithmInstanceWithIsolator(string assemblyPath, int ramL
             var complete = isolator.ExecuteWithTimeLimit(_loaderTimeLimit, () =>
             {
                 success = TryCreateAlgorithmInstance(assemblyPath, out instance, out error);
-            }, ramLimit, sleepIntervalMillis:50);
+            }, ramLimit, sleepIntervalMillis:50, workerThread:_workerThread);
 
             algorithmInstance = instance;
             errorMessage = error;

Common/Isolator.cs

@@ -17,6 +17,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using QuantConnect.Logging;
+using QuantConnect.Util;
 
 namespace QuantConnect
 {
@@ -67,8 +68,38 @@ public Isolator()
         /// <param name="codeBlock">Action codeblock to execute</param>
         /// <param name="memoryCap">Maximum memory allocation, default 1024Mb</param>
         /// <param name="sleepIntervalMillis">Sleep interval between each check in ms</param>
+        /// <param name="workerThread">The worker thread instance that will execute the provided action, if null
+        /// will use a <see cref="Task"/></param>
         /// <returns>True if algorithm exited successfully, false if cancelled because it exceeded limits.</returns>
-        public bool ExecuteWithTimeLimit(TimeSpan timeSpan, Func<IsolatorLimitResult> withinCustomLimits, Action codeBlock, long memoryCap = 1024, int sleepIntervalMillis = 1000)
+        public bool ExecuteWithTimeLimit(TimeSpan timeSpan, Func<IsolatorLimitResult> withinCustomLimits, Action codeBlock, long memoryCap = 1024, int sleepIntervalMillis = 1000, WorkerThread workerThread = null)
+        {
+            workerThread?.Add(codeBlock);
+
+            var task = workerThread == null
+                //Launch task
+                ? Task.Factory.StartNew(codeBlock, CancellationTokenSource.Token)
+                // wrapper task so we can reuse MonitorTask
+                : Task.Factory.StartNew(() => workerThread.FinishedWorkItem.WaitOne(), CancellationTokenSource.Token);
+            try
+            {
+                return MonitorTask(task, timeSpan, withinCustomLimits, memoryCap, sleepIntervalMillis);
+            }
+            catch (Exception)
+            {
+                if (!task.IsCompleted)
+                {
+                    // lets free the wrapper task even if the worker thread didn't finish
+                    workerThread?.FinishedWorkItem.Set();
+                }
+                throw;
+            }
+        }
+
+        private bool MonitorTask(Task task,
+            TimeSpan timeSpan,
+            Func<IsolatorLimitResult> withinCustomLimits,
+            long memoryCap = 1024,
+            int sleepIntervalMillis = 1000)
         {
             // default to always within custom limits
             withinCustomLimits = withinCustomLimits ?? (() => new IsolatorLimitResult(TimeSpan.Zero, string.Empty));
@@ -84,9 +115,6 @@ public bool ExecuteWithTimeLimit(TimeSpan timeSpan, Func<IsolatorLimitResult> wi
             memoryCap *= 1024 * 1024;
             var spikeLimit = memoryCap*2;
 
-            //Launch task
-            var task = Task.Factory.StartNew(codeBlock, CancellationTokenSource.Token);
-
             // give some granularity to the sleep interval if >= 1000ms
             var sleepGranularity = sleepIntervalMillis >= 1000 ? 5 : 1;
             var granularSleepIntervalMillis = sleepIntervalMillis / sleepGranularity;
@@ -151,7 +179,7 @@ public bool ExecuteWithTimeLimit(TimeSpan timeSpan, Func<IsolatorLimitResult> wi
             {
                 CancellationTokenSource.Cancel();
                 Log.Error("Security.ExecuteWithTimeLimit(): " + message);
-                throw new Exception(message);
+                throw new TimeoutException(message);
             }
             return task.IsCompleted;
         }
@@ -163,10 +191,12 @@ public bool ExecuteWithTimeLimit(TimeSpan timeSpan, Func<IsolatorLimitResult> wi
         /// <param name="codeBlock">Action codeblock to execute</param>
         /// <param name="memoryCap">Maximum memory allocation, default 1024Mb</param>
         /// <param name="sleepIntervalMillis">Sleep interval between each check in ms</param>
+        /// <param name="workerThread">The worker thread instance that will execute the provided action, if null
+        /// will use a <see cref="Task"/></param>
         /// <returns>True if algorithm exited successfully, false if cancelled because it exceeded limits.</returns>
-        public bool ExecuteWithTimeLimit(TimeSpan timeSpan, Action codeBlock, long memoryCap, int sleepIntervalMillis = 1000)
+        public bool ExecuteWithTimeLimit(TimeSpan timeSpan, Action codeBlock, long memoryCap, int sleepIntervalMillis = 1000, WorkerThread workerThread = null)
         {
-            return ExecuteWithTimeLimit(timeSpan, null, codeBlock, memoryCap, sleepIntervalMillis);
+            return ExecuteWithTimeLimit(timeSpan, null, codeBlock, memoryCap, sleepIntervalMillis, workerThread);
         }
 
         /// <summary>

Common/QuantConnect.csproj

@@ -697,6 +697,7 @@
     <Compile Include="Util\Ref.cs" />
     <Compile Include="Util\SingleValueListConverter.cs" />
     <Compile Include="Util\VersionHelper.cs" />
+    <Compile Include="Util\WorkerThread.cs" />
     <Compile Include="Util\XElementExtensions.cs" />
   </ItemGroup>
   <ItemGroup>

Common/Util/WorkerThread.cs

@@ -0,0 +1,86 @@
+/*
+ * 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");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+using System;
+using System.Collections.Concurrent;
+using System.Threading;
+
+namespace QuantConnect.Util
+{
+    /// <summary>
+    /// This worker tread is required to guarantee all python operations are
+    /// executed by the same thread, to enable complete debugging functionality.
+    /// We don't use the main thread, to avoid any chance of blocking the process
+    /// </summary>
+    public class WorkerThread : IDisposable
+    {
+        private readonly BlockingCollection<Action> _blockingCollection;
+        private readonly CancellationTokenSource _threadCancellationTokenSource;
+
+        /// <summary>
+        /// Will be set when the worker thread finishes a work item
+        /// </summary>
+        public AutoResetEvent FinishedWorkItem { get; }
+
+        /// <summary>
+        /// Creates a new instance, which internally launches a new worker thread
+        /// </summary>
+        /// <remarks><see cref="Dispose"/></remarks>
+        public WorkerThread()
+        {
+            _threadCancellationTokenSource = new CancellationTokenSource();
+            FinishedWorkItem = new AutoResetEvent(false);
+            _blockingCollection = new BlockingCollection<Action>();
+            var thread = new Thread(() =>
+            {
+                try
+                {
+                    foreach (var action in _blockingCollection.GetConsumingEnumerable(_threadCancellationTokenSource.Token))
+                    {
+                        FinishedWorkItem.Reset();
+                        action();
+                        FinishedWorkItem.Set();
+                    }
+                }
+                catch (OperationCanceledException)
+                {
+                    // pass, when the token gets cancelled
+                }
+            })
+            { IsBackground = true, Name = "Isolator Thread" };
+            thread.Start();
+        }
+
+        /// <summary>
+        /// Adds a new item of work
+        /// </summary>
+        /// <param name="action">The work item to add</param>
+        public void Add(Action action)
+        {
+            _blockingCollection.Add(action);
+        }
+
+        /// <summary>
+        /// Disposes the worker thread.
+        /// </summary>
+        /// <remarks>Note that the worker thread is a background thread,
+        /// so it won't block the process from terminating even if not disposed</remarks>
+        public void Dispose()
+        {
+            _blockingCollection.CompleteAdding();
+            _threadCancellationTokenSource.Cancel();
+        }
+    }
+}

Engine/Engine.cs

@@ -104,6 +104,7 @@ public void Run(AlgorithmNodePacket job, AlgorithmManager manager, string assemb
                 Thread threadResults = null;
                 Thread threadRealTime = null;
                 Thread threadAlphas = null;
+                WorkerThread workerThread = null;
 
                 //-> Initialize messaging system
                 _systemHandlers.Notify.SetAuthentication(job);
@@ -123,6 +124,10 @@ public void Run(AlgorithmNodePacket job, AlgorithmManager manager, string assemb
                     // since the algorithm constructor will use it
                     var marketHoursDatabase = marketHoursDatabaseTask.Result;
 
+                    // start worker thread
+                    workerThread = new WorkerThread();
+                    _algorithmHandlers.Setup.WorkerThread = workerThread;
+
                     // Save algorithm to cache, load algorithm instance:
                     algorithm = _algorithmHandlers.Setup.CreateAlgorithmInstance(job, assemblyPath);
 
@@ -342,7 +347,7 @@ public void Run(AlgorithmNodePacket job, AlgorithmManager manager, string assemb
                             }
 
                             Log.Trace("Engine.Run(): Exiting Algorithm Manager");
-                        }, job.Controls.RamAllocation);
+                        }, job.Controls.RamAllocation, workerThread:workerThread);
 
                         if (!complete)
                         {
@@ -447,6 +452,7 @@ public void Run(AlgorithmNodePacket job, AlgorithmManager manager, string assemb
                     _algorithmHandlers.RealTime.Exit();
                     _algorithmHandlers.Alphas.Exit();
                     dataManager?.RemoveAllSubscriptions();
+                    workerThread?.Dispose();
                 }
 
                 //Close result handler:

Engine/Results/BacktestingResultHandler.cs

@@ -262,7 +262,11 @@ public void Update()
                 runtimeStatistics.Add("Unrealized", "$" + Algorithm.Portfolio.TotalUnrealizedProfit.ToString("N2"));
                 runtimeStatistics.Add("Fees", "-$" + Algorithm.Portfolio.TotalFees.ToString("N2"));
                 runtimeStatistics.Add("Net Profit", "$" + (Algorithm.Portfolio.TotalProfit - Algorithm.Portfolio.TotalFees).ToString("N2"));
-                runtimeStatistics.Add("Return", ((Algorithm.Portfolio.TotalPortfolioValue - _setupHandler.StartingPortfolioValue) / _setupHandler.StartingPortfolioValue).ToString("P"));
+                // when there is an initialization error StartingPortfolioValue is 0, want to avoid dividing by zero
+                if (_setupHandler.StartingPortfolioValue != 0)
+                {
+                    runtimeStatistics.Add("Return", ((Algorithm.Portfolio.TotalPortfolioValue - _setupHandler.StartingPortfolioValue) / _setupHandler.StartingPortfolioValue).ToString("P"));
+                }
                 runtimeStatistics.Add("Equity", "$" + Algorithm.Portfolio.TotalPortfolioValue.ToString("N2"));
 
                 //Profit Loss Changes:

Engine/Setup/BacktestingSetupHandler.cs

@@ -27,6 +27,7 @@
 using QuantConnect.Data;
 using QuantConnect.Lean.Engine.DataFeeds;
 using QuantConnect.Securities;
+using QuantConnect.Util;
 
 namespace QuantConnect.Lean.Engine.Setup
 {
@@ -39,6 +40,11 @@ public class BacktestingSetupHandler : ISetupHandler
         private int _maxOrders = 0;
         private DateTime _startingDate = new DateTime(1998, 01, 01);
 
+        /// <summary>
+        /// The worker thread instance the setup handler should use
+        /// </summary>
+        public WorkerThread WorkerThread { get; set; }
+
         /// <summary>
         /// Internal errors list from running the setup proceedures.
         /// </summary>
@@ -111,7 +117,7 @@ public virtual IAlgorithm CreateAlgorithmInstance(AlgorithmNodePacket algorithmN
             IAlgorithm algorithm;
 
             // limit load times to 60 seconds and force the assembly to have exactly one derived type
-            var loader = new Loader(algorithmNodePacket.Language, TimeSpan.FromSeconds(60), names => names.SingleOrAlgorithmTypeName(Config.Get("algorithm-type-name")));
+            var loader = new Loader(algorithmNodePacket.Language, TimeSpan.FromSeconds(60), names => names.SingleOrAlgorithmTypeName(Config.Get("algorithm-type-name")), WorkerThread);
             var complete = loader.TryCreateAlgorithmInstanceWithIsolator(assemblyPath, algorithmNodePacket.RamAllocation, out algorithm, out error);
             if (!complete) throw new AlgorithmSetupException($"During the algorithm initialization, the following exception has occurred: {error}");
 
@@ -198,7 +204,8 @@ public bool Setup(SetupHandlerParameters parameters)
                     Errors.Add(new AlgorithmSetupException("During the algorithm initialization, the following exception has occurred: ", err));
                 }
             }, controls.RamAllocation,
-                sleepIntervalMillis:50); // entire system is waiting on this, so be as fast as possible
+                sleepIntervalMillis:50,  // entire system is waiting on this, so be as fast as possible
+                workerThread: WorkerThread);
 
             //Before continuing, detect if this is ready:
             if (!initializeComplete) return false;

Engine/Setup/BrokerageSetupHandler.cs

@@ -40,6 +40,11 @@ namespace QuantConnect.Lean.Engine.Setup
     /// </summary>
     public class BrokerageSetupHandler : ISetupHandler
     {
+        /// <summary>
+        /// The worker thread instance the setup handler should use
+        /// </summary>
+        public WorkerThread WorkerThread { get; set; }
+
         /// <summary>
         /// Any errors from the initialization stored here:
         /// </summary>
@@ -91,7 +96,7 @@ public IAlgorithm CreateAlgorithmInstance(AlgorithmNodePacket algorithmNodePacke
             IAlgorithm algorithm;
 
             // limit load times to 10 seconds and force the assembly to have exactly one derived type
-            var loader = new Loader(algorithmNodePacket.Language, TimeSpan.FromSeconds(60), names => names.SingleOrAlgorithmTypeName(Config.Get("algorithm-type-name")));
+            var loader = new Loader(algorithmNodePacket.Language, TimeSpan.FromSeconds(60), names => names.SingleOrAlgorithmTypeName(Config.Get("algorithm-type-name")), WorkerThread);
             var complete = loader.TryCreateAlgorithmInstanceWithIsolator(assemblyPath, algorithmNodePacket.RamAllocation, out algorithm, out error);
             if (!complete) throw new AlgorithmSetupException($"During the algorithm initialization, the following exception has occurred: {error}");
 

Engine/Setup/ConsoleSetupHandler.cs

@@ -25,6 +25,7 @@
 using QuantConnect.Logging;
 using QuantConnect.Packets;
 using QuantConnect.Lean.Engine.DataFeeds;
+using QuantConnect.Util;
 
 namespace QuantConnect.Lean.Engine.Setup
 {
@@ -33,6 +34,11 @@ namespace QuantConnect.Lean.Engine.Setup
     /// </summary>
     public class ConsoleSetupHandler : ISetupHandler
     {
+        /// <summary>
+        /// The worker thread instance the setup handler should use
+        /// </summary>
+        public WorkerThread WorkerThread { get; set; }
+
         /// <summary>
         /// Error which occured during setup may appear here.
         /// </summary>
@@ -84,7 +90,7 @@ public IAlgorithm CreateAlgorithmInstance(AlgorithmNodePacket algorithmNodePacke
 
             // don't force load times to be fast here since we're running locally, this allows us to debug
             // and step through some code that may take us longer than the default 10 seconds
-            var loader = new Loader(algorithmNodePacket.Language, TimeSpan.FromHours(1), names => names.SingleOrDefault(name => MatchTypeName(name, algorithmName)));
+            var loader = new Loader(algorithmNodePacket.Language, TimeSpan.FromHours(1), names => names.SingleOrDefault(name => MatchTypeName(name, algorithmName)), WorkerThread);
             var complete = loader.TryCreateAlgorithmInstanceWithIsolator(assemblyPath, algorithmNodePacket.RamAllocation, out algorithm, out error);
             if (!complete) throw new AlgorithmSetupException($"During the algorithm initialization, the following exception has occurred: {error}");
 
@@ -144,8 +150,15 @@ public bool Setup(SetupHandlerParameters parameters)
                     // set the future chain provider
                     algorithm.SetFutureChainProvider(new CachingFutureChainProvider(new BacktestingFutureChainProvider()));
 
-                    //Setup Base Algorithm:
-                    algorithm.Initialize();
+                    var isolator = new Isolator();
+                    isolator.ExecuteWithTimeLimit(TimeSpan.FromMinutes(5),
+                        () =>
+                        {
+                            //Setup Base Algorithm:
+                            algorithm.Initialize();
+                        }, baseJob.Controls.RamAllocation,
+                        sleepIntervalMillis: 50,
+                        workerThread: WorkerThread);
 
                     //Finalize Initialization
                     algorithm.PostInitialize();

Engine/Setup/ISetupHandler.cs

@@ -19,6 +19,7 @@
 using System.ComponentModel.Composition;
 using QuantConnect.Interfaces;
 using QuantConnect.Packets;
+using QuantConnect.Util;
 
 namespace QuantConnect.Lean.Engine.Setup
 {
@@ -28,6 +29,14 @@ namespace QuantConnect.Lean.Engine.Setup
     [InheritedExport(typeof(ISetupHandler))]
     public interface ISetupHandler : IDisposable
     {
+        /// <summary>
+        /// The worker thread instance the setup handler should use
+        /// </summary>
+        WorkerThread WorkerThread
+        {
+            set;
+        }
+
         /// <summary>
         /// Any errors from the initialization stored here:
         /// </summary>