Changeset 6357 for branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.HiveEngine/3.4

Timestamp:

06/05/11 22:35:40 (13 years ago)

Author:

cneumuel

Message:

#1233

• refactoring of slave core
• created JobManager, which is responsible for managing jobs without knowing anything about the service. this class is easier testable than slave core
• lots of cleanup
• created console test project for slave

File:

• branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.HiveEngine/3.4/HiveEngine.cs (modified) (7 diffs)

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.HiveEngine/3.4/HiveEngine.cs

@@ -37 +37 @@
   [Item("Hive Engine", "Engine for parallel execution on the hive. You need enable `Parallel` for at least one operator in your operator graph to have all childoperations parallelized. Also those childoperations must not have sideeffects on a higher scope.")]
   public class HiveEngine : Engine {
+    private static object locker = new object();
     private static object logLocker = new object();
     private CancellationToken cancellationToken;
-    
+    private bool firstRun = true;
+
     [Storable]
     private IOperator currentOperator;
@@ -136 +138 @@
       OperationCollection coll;
       IAtomicOperation operation;
-      TaskScheduler.UnobservedTaskException += new EventHandler<UnobservedTaskExceptionEventArgs>(TaskScheduler_UnobservedTaskException);
-
-      this.OnlinePlugins = ServiceLocator.Instance.CallHiveService(s => s.GetPlugins()).Where(x => x.IsLocal == false).ToList();
-      this.AlreadyUploadedPlugins = new List<Plugin>();
-
-      while (ExecutionStack.Count > 0) {
+
+      if (firstRun) {
+        TaskScheduler.UnobservedTaskException += new EventHandler<UnobservedTaskExceptionEventArgs>(TaskScheduler_UnobservedTaskException);
+        this.OnlinePlugins = ServiceLocator.Instance.CallHiveService(s => s.GetPlugins()).Where(x => x.IsLocal == false).ToList();
+        this.AlreadyUploadedPlugins = new List<Plugin>();
+        firstRun = false;
+      }
+
+      while (executionStack.Count > 0) {
         cancellationToken.ThrowIfCancellationRequested();
 
-        next = ExecutionStack.Pop();
+        next = executionStack.Pop();
+        bool isOpCollection = next is OperationCollection;
+        int collCount = isOpCollection ? ((OperationCollection)next).Count : 0;
+        string opName = !isOpCollection ? ((IAtomicOperation)next).Operator.Name : "OpCollection";
+
         if (next is OperationCollection) {
           coll = (OperationCollection)next;
-          if (coll.Parallel) {
+
+          bool isPMOEvaluator = coll.Count > 0 && coll.First() is HeuristicLab.Core.ExecutionContext && ((HeuristicLab.Core.ExecutionContext)coll.First()).Operator.GetType().Name == "PMOEvaluator";
+          bool isAlgorithmEvaluator = coll.Count > 0 && coll.First() is HeuristicLab.Core.ExecutionContext && ((HeuristicLab.Core.ExecutionContext)coll.First()).Operator.GetType().Name == "AlgorithmEvaluator";
+
+          if (coll.Parallel && isPMOEvaluator) {
+            Task[] tasks = new Task[coll.Count];
+            Stack<IOperation>[] stacks = new Stack<IOperation>[coll.Count];
+            for (int i = 0; i < coll.Count; i++) {
+              stacks[i] = new Stack<IOperation>();
+              stacks[i].Push(coll[i]);
+              tasks[i] = Task.Factory.StartNew(Run, stacks[i], cancellationToken);
+            }
+            try {
+              Task.WaitAll(tasks);
+            }
+            catch (AggregateException ex) {
+              OperationCollection remaining = new OperationCollection() { Parallel = true };
+              for (int i = 0; i < stacks.Length; i++) {
+                if (stacks[i].Count == 1)
+                  remaining.Add(stacks[i].Pop());
+                if (stacks[i].Count > 1) {
+                  OperationCollection ops = new OperationCollection();
+                  while (stacks[i].Count > 0)
+                    ops.Add(stacks[i].Pop());
+                  remaining.Add(ops);
+                }
+              }
+              if (remaining.Count > 0) executionStack.Push(remaining);
+              throw ex;
+            }
+          } else if (coll.Parallel) {
             // clone the parent scope here and reuse it for each operation. otherwise for each job the whole scope-tree first needs to be copied and then cleaned, which causes a lot of work for the Garbage Collector
             IScope parentScopeClone = (IScope)((IAtomicOperation)coll.First()).Scope.Parent.Clone();
@@ -158 +197 @@
             }
 
-            IScope[] scopes = ExecuteOnHive(jobs, parentScopeClone, cancellationToken);
-            //IScope[] scopes = ExecuteLocally(jobs, parentScopeClone, cancellationToken);
+            var experiment = CreateHiveExperiment();
+            IScope[] scopes = ExecuteOnHive(experiment, jobs, parentScopeClone, cancellationToken);
+            DisposeHiveExperiment(experiment);
 
             for (int i = 0; i < coll.Count; i++) {
@@ -178 +218 @@
           }
           catch (Exception ex) {
-            ExecutionStack.Push(operation);
+            executionStack.Push(operation);
             if (ex is OperationCanceledException) throw ex;
             else throw new OperatorExecutionException(operation.Operator, ex);
           }
-          if (next != null) ExecutionStack.Push(next);
+          if (next != null) executionStack.Push(next);
 
           if (operation.Operator.Breakpoint) {
@@ -227 +267 @@
     /// </summary>
     /// <param name="jobs"></param>
-    private IScope[] ExecuteOnHive(EngineJob[] jobs, IScope parentScopeClone, CancellationToken cancellationToken) {
+    private IScope[] ExecuteOnHive(RefreshableHiveExperiment refreshableHiveExperiment, EngineJob[] jobs, IScope parentScopeClone, CancellationToken cancellationToken) {
       LogMessage(string.Format("Executing {0} operations on the hive.", jobs.Length));
       IScope[] scopes = new Scope[jobs.Length];
       object locker = new object();
       IDictionary<Guid, int> jobIndices = new Dictionary<Guid, int>();
-      var hiveExperiment = new HiveExperiment();
+      var hiveExperiment = refreshableHiveExperiment.HiveExperiment;
 
       try {
         List<Guid> remainingJobIds = new List<Guid>();
 
-        // create hive experiment
+        // create upload-tasks
+        var uploadTasks = new List<Task<Job>>();
+        for (int i = 0; i < jobs.Length; i++) {
+          hiveExperiment.HiveJobs.Add(new EngineHiveJob(jobs[i], parentScopeClone));
+
+          // shuffle random variable to avoid the same random sequence in each operation; todo: does not yet work (it cannot find the random variable)
+          IRandom random = FindRandomParameter(jobs[i].InitialOperation as IExecutionContext);
+          if (random != null)
+            random.Reset(random.Next());
+        }
+        ExperimentManagerClient.StartExperiment((e) => {
+          LogException(e);
+        }, refreshableHiveExperiment);
+
+        // do polling until experiment is finished and all jobs are downloaded
+        while (!refreshableHiveExperiment.AllJobsFinished()) {
+          Thread.Sleep(500);
+          this.ExecutionTimeOnHive = TimeSpan.FromMilliseconds(hiveExperiments.Sum(x => x.HiveExperiment.ExecutionTime.TotalMilliseconds));
+          cancellationToken.ThrowIfCancellationRequested();
+        }
+        LogMessage(string.Format("{0} finished (TotalExecutionTime: {1}).", refreshableHiveExperiment.ToString(), refreshableHiveExperiment.HiveExperiment.ExecutionTime));
+
+        // get scopes
+        int j = 0;
+        foreach (var hiveJob in hiveExperiment.HiveJobs) {
+          if (hiveJob.Job.State != JobState.Finished)
+            throw new HiveEngineException("Job failed: " + hiveJob.Job.StateLog.Last().Exception);
+
+          var scope = ((IAtomicOperation)((EngineJob)hiveJob.ItemJob).InitialOperation).Scope;
+          scopes[j++] = scope;
+        }
+        return scopes;
+      }
+      catch (OperationCanceledException e) {
+        lock (locker) {
+          if (jobIndices != null) DeleteHiveExperiment(hiveExperiment.Id);
+        }
+        throw e;
+      }
+      catch (Exception e) {
+        lock (locker) {
+          if (jobIndices != null) DeleteHiveExperiment(hiveExperiment.Id);
+        }
+        LogException(e);
+        throw e;
+      }
+    }
+
+    private RefreshableHiveExperiment CreateHiveExperiment() {
+      lock (locker) {
+        var hiveExperiment = new HiveExperiment();
         hiveExperiment.Name = "HiveEngine Run " + hiveExperiments.Count;
         hiveExperiment.DateCreated = DateTime.Now;
@@ -245 +335 @@
         refreshableHiveExperiment.IsControllable = false;
         hiveExperiments.Add(refreshableHiveExperiment);
-
-        // create upload-tasks
-        var uploadTasks = new List<Task<Job>>();
-        for (int i = 0; i < jobs.Length; i++) {
-          hiveExperiment.HiveJobs.Add(new EngineHiveJob(jobs[i], parentScopeClone));
-
-          // shuffle random variable to avoid the same random sequence in each operation; todo: does not yet work (it cannot find the random variable)
-          IRandom random = FindRandomParameter(jobs[i].InitialOperation as IExecutionContext);
-          if (random != null)
-            random.Reset(random.Next());
-        }
-        ExperimentManagerClient.StartExperiment((e) => {
-          LogException(e);
-        }, refreshableHiveExperiment);
-
-        // do polling until experiment is finished and all jobs are downloaded
-        while (!refreshableHiveExperiment.AllJobsFinished()) {
-          Thread.Sleep(500);
-          this.ExecutionTimeOnHive = TimeSpan.FromMilliseconds(hiveExperiments.Sum(x => x.HiveExperiment.ExecutionTime.TotalMilliseconds));
-          cancellationToken.ThrowIfCancellationRequested();
-        }
-        LogMessage(string.Format("{0} finished (TotalExecutionTime: {1}).", refreshableHiveExperiment.ToString(), refreshableHiveExperiment.HiveExperiment.ExecutionTime));
-
-        // get scopes
-        int j = 0;
-        foreach (var hiveJob in hiveExperiment.HiveJobs) {
-          var scope = ((IAtomicOperation) ((EngineJob)hiveJob.ItemJob).InitialOperation).Scope;
-          scopes[j++] = scope;
-        }
-        refreshableHiveExperiment.RefreshAutomatically = false;
-        DeleteHiveExperiment(hiveExperiment.Id);
-        ClearData(refreshableHiveExperiment);
-        return scopes;
-      }
-      catch (OperationCanceledException e) {
-        lock (locker) {
-          if (jobIndices != null) DeleteHiveExperiment(hiveExperiment.Id);
-        }
-        throw e;
-      }
-      catch (Exception e) {
-        lock (locker) {
-          if (jobIndices != null) DeleteHiveExperiment(hiveExperiment.Id);
-        }
-        LogException(e);
-        throw e;
-      }
+        return refreshableHiveExperiment;
+      }
+    }
+
+    private void DisposeHiveExperiment(RefreshableHiveExperiment refreshableHiveExperiment) {
+      refreshableHiveExperiment.RefreshAutomatically = false;
+      DeleteHiveExperiment(refreshableHiveExperiment.HiveExperiment.Id);
+      ClearData(refreshableHiveExperiment);
     }
 
@@ -306 +357 @@
       }, 5, string.Format("Could not delete jobs"));
     }
-    
+
     private List<Guid> GetResourceIds() {
       return ServiceLocator.Instance.CallHiveService(service => {