Changeset 5213 for branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine/3.3/HiveEngine.cs

Timestamp:

01/05/11 02:05:05 (13 years ago)

Author:

cneumuel

Message:

#1260

• changed HiveEngine to be compatible with recent changes of engine

File:

• branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine/3.3/HiveEngine.cs (modified) (4 diffs)

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

@@ -14 +14 @@
 using System.Threading;
 using HeuristicLab.Random;
+using System.Threading.Tasks;
 
 namespace HeuristicLab.HiveEngine {
@@ -22 +23 @@
   [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 CancellationToken cancellationToken;
+
     [Storable]
     private IOperator currentOperator;
@@ -44 +47 @@
     #endregion
 
-    protected override void ProcessNextOperation() {
-      currentOperator = null;
-      IOperation next = ExecutionStack.Pop();
-      OperationCollection coll = next as OperationCollection;
+    protected override void Run(CancellationToken cancellationToken) {
+      this.cancellationToken = cancellationToken;
+      Run(ExecutionStack);
+    }
 
-      while (coll != null) {
-        if (coll.Parallel) {
-          IDictionary<IOperation, OperationJob> jobs = new Dictionary<IOperation, OperationJob>();
-          foreach (IOperation op in coll) {
-            jobs.Add(op, new OperationJob(op));
+    private void Run(object state) {
+      Stack<IOperation> executionStack = (Stack<IOperation>)state;
+      IOperation next;
+      OperationCollection coll;
+      IAtomicOperation operation;
+
+      while (ExecutionStack.Count > 0) {
+        cancellationToken.ThrowIfCancellationRequested();
+
+        next = ExecutionStack.Pop();
+        if (next is OperationCollection) {
+          coll = (OperationCollection)next;
+          if (coll.Parallel) {
+            IDictionary<IOperation, OperationJob> jobs = new Dictionary<IOperation, OperationJob>();
+            foreach (IOperation op in coll) {
+              jobs.Add(op, new OperationJob(op));
+            }
+
+            ExecuteOnHive(jobs);
+
+            foreach (var kvp in jobs) {
+              if (kvp.Key is IAtomicOperation) {
+                ReIntegrateScope(kvp.Value.Operation as IAtomicOperation, kvp.Key as IAtomicOperation);
+              } else if (kvp.Key is OperationCollection) {
+                OperationCollection ocoll = (OperationCollection)kvp.Key;
+                for (int i = ocoll.Count - 1; i >= 0; i--)
+                  if (ocoll[i] != null) executionStack.Push(ocoll[i]);
+              }
+            }
+          } else {
+            for (int i = coll.Count - 1; i >= 0; i--)
+              if (coll[i] != null) executionStack.Push(coll[i]);
           }
+        } else if (next is IAtomicOperation) {
+          operation = (IAtomicOperation)next;
+          try {
+            next = operation.Operator.Execute((IExecutionContext)operation, cancellationToken);
+          }
+          catch (Exception ex) {
+            ExecutionStack.Push(operation);
+            if (ex is OperationCanceledException) throw ex;
+            else throw new OperatorExecutionException(operation.Operator, ex);
+          }
+          if (next != null) ExecutionStack.Push(next);
 
-          ExecuteOnHive(jobs);
-
-          foreach (var kvp in jobs) {
-            if (kvp.Key is IAtomicOperation) {
-              ReIntegrateScope(kvp.Value.Operation as IAtomicOperation, kvp.Key as IAtomicOperation);
-            } else if (kvp.Key is OperationCollection) {
-              // todo
-            }
+          if (operation.Operator.Breakpoint) {
+            Log.LogMessage(string.Format("Breakpoint: {0}", operation.Operator.Name != string.Empty ? operation.Operator.Name : operation.Operator.ItemName));
+            Pause();
           }
-        } else {
-          for (int i = coll.Count - 1; i >= 0; i--)
-            ExecutionStack.Push(coll[i]);
-        }
-        next = ExecutionStack.Count > 0 ? ExecutionStack.Pop() : null;
-        coll = next as OperationCollection;
-      }
-
-      IAtomicOperation operation = next as IAtomicOperation;
-      if (operation != null) {
-        try {
-          currentOperator = operation.Operator;
-          ExecutionStack.Push(operation.Operator.Execute((IExecutionContext)operation));
-        }
-        catch (Exception ex) {
-          ExecutionStack.Push(operation);
-          OnExceptionOccurred(new OperatorExecutionException(operation.Operator, ex));
-          Pause();
-        }
-        if (operation.Operator.Breakpoint) {
-          Log.LogMessage(string.Format("Breakpoint: {0}", operation.Operator.Name != string.Empty ? operation.Operator.Name : operation.Operator.ItemName));
-          Pause();
         }
       }
@@ -123 +137 @@
     private void ExecuteOnHive(IDictionary<IOperation, OperationJob> jobDict) {
       Log.LogMessage(string.Format("Executing {0} operations on the hive.", jobDict.Count));
-      IDictionary<Guid, IOperation> jobIds = new Dictionary<Guid, IOperation>();
-      JobResultList results;
+      try {
+        IDictionary<Guid, IOperation> jobIds = new Dictionary<Guid, IOperation>();
+        JobResultList results;
 
-      using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
-        List<JobDto> jobs = new List<JobDto>();
-        foreach (var kvp in jobDict) {
-          // shuffle random variable to avoid the same random sequence in each operation
-          IRandom random = FindRandomParameter(kvp.Key as IExecutionContext);
-          if (random != null)
-            random.Reset(random.Next());
+        using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
+          List<JobDto> jobs = new List<JobDto>();
+          foreach (var kvp in jobDict) {
+            // shuffle random variable to avoid the same random sequence in each operation
+            IRandom random = FindRandomParameter(kvp.Key as IExecutionContext);
+            if (random != null)
+              random.Reset(random.Next());
 
-          var groups = ResourceIds.Split(';');
-          SerializedJob serializedJob = new SerializedJob();
-          serializedJob.SerializedJobData = SerializedJob.Serialize(kvp.Value);
-          serializedJob.JobInfo = new JobDto();
-          serializedJob.JobInfo.State = JobState.Offline;
-          serializedJob.JobInfo.CoresNeeded = 1;
-          serializedJob.JobInfo.PluginsNeeded = ApplicationManager.Manager.Plugins.Select(x => new HivePluginInfoDto { Name = x.Name, Version = x.Version }).ToList();
-          ResponseObject<JobDto> response = service.Obj.AddJobWithGroupStrings(serializedJob, groups);
-          jobs.Add(response.Obj);
-          jobIds.Add(response.Obj.Id, kvp.Key);
-        }
-        results = service.Obj.GetJobResults(jobIds.Keys).Obj;
-      }
-
-      while (!results.All(
-          x => x.State == JobState.Finished ||
-          x.State == JobState.Failed ||
-          x.State == JobState.Aborted)) {
-        Thread.Sleep(5000);
-        using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
+            var groups = ResourceIds.Split(';');
+            SerializedJob serializedJob = new SerializedJob();
+            serializedJob.SerializedJobData = SerializedJob.Serialize(kvp.Value);
+            serializedJob.JobInfo = new JobDto();
+            serializedJob.JobInfo.State = JobState.Offline;
+            serializedJob.JobInfo.CoresNeeded = 1;
+            serializedJob.JobInfo.PluginsNeeded = ApplicationManager.Manager.Plugins.Select(x => new HivePluginInfoDto { Name = x.Name, Version = x.Version }).ToList();
+            ResponseObject<JobDto> response = service.Obj.AddJobWithGroupStrings(serializedJob, groups);
+            jobs.Add(response.Obj);
+            jobIds.Add(response.Obj.Id, kvp.Key);
+          }
           results = service.Obj.GetJobResults(jobIds.Keys).Obj;
         }
+
+        while (!results.All(
+            x => x.State == JobState.Finished ||
+            x.State == JobState.Failed ||
+            x.State == JobState.Aborted)) {
+          Thread.Sleep(5000);
+          using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
+            results = service.Obj.GetJobResults(jobIds.Keys).Obj;
+          }
+        }
+
+        // all finished
+        using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
+          foreach (Guid jobId in jobIds.Keys) {
+            SerializedJob serializedJob = service.Obj.GetLastSerializedResult(jobId).Obj;
+            OperationJob operationJob = SerializedJob.Deserialize<OperationJob>(serializedJob.SerializedJobData);
+            jobDict[jobIds[jobId]] = operationJob;
+          }
+        }
+
+        // delete jobs
+        using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
+          foreach (Guid jobId in jobIds.Keys) {
+            service.Obj.DeleteJob(jobId);
+          }
+        }
+
+        Log.LogMessage(string.Format("Operations on the hive finished.", jobDict.Count));
       }
-
-      // all finished
-      using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
-        foreach (Guid jobId in jobIds.Keys) {
-          SerializedJob serializedJob = service.Obj.GetLastSerializedResult(jobId).Obj;
-          OperationJob operationJob = SerializedJob.Deserialize<OperationJob>(serializedJob.SerializedJobData);
-          jobDict[jobIds[jobId]] = operationJob;
-        }
+      catch (Exception e) {
+        Log.LogException(e);
+        throw e;
       }
-
-      // delete jobs
-      using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
-        foreach (Guid jobId in jobIds.Keys) {
-          service.Obj.DeleteJob(jobId);
-        }
-      }
-
-      Log.LogMessage(string.Format("Operations on the hive finished.", jobDict.Count));
     }
-
-    public override void Pause() {
-      base.Pause();
-      if (currentOperator != null) currentOperator.Abort();
-    }
-    public override void Stop() {
-      base.Stop();
-      if (currentOperator != null) currentOperator.Abort();
-    }
-
   }
 }