Changeset 7417 for branches/HiveHiveEngine/HeuristicLab.HiveEngine/3.3/HiveEngine.cs

Timestamp:

01/26/12 22:01:38 (13 years ago)

Author:

ascheibe

Message:

#1744 added an option so that the parent task can also wait on the slave for the child tasks to finish

File:

• branches/HiveHiveEngine/HeuristicLab.HiveEngine/3.3/HiveEngine.cs (modified) (9 diffs)

branches/HiveHiveEngine/HeuristicLab.HiveEngine/3.3/HiveEngine.cs

@@ -47 +47 @@
 
     [Storable]
+    public bool PauseWhileCalculatingChilds { get; set; }
+
+    [Storable]
     private IOperator currentOperator;
 
@@ -131 +134 @@
       username = cl.ClientCredentials.UserName.UserName;
       password = cl.ClientCredentials.UserName.Password;
+      PauseWhileCalculatingChilds = true;
     }
 
@@ -157 +161 @@
       this.nrOfSentRuns = original.nrOfSentRuns;
       this.ParentTaskId = original.ParentTaskId;
+      this.PauseWhileCalculatingChilds = original.PauseWhileCalculatingChilds;
       // do not clone jobs - otherwise they would be sent with every task
     }
@@ -200 +205 @@
         cancellationToken.ThrowIfCancellationRequested();
 
-        if (continueCollection != null && lastJobGuid != Guid.Empty) {
+        if (continueCollection != null && lastJobGuid != Guid.Empty && PauseWhileCalculatingChilds) {
           EngineTask[] eTasks = RetrieveResultsFromHive(lastJobGuid);
           RestoreStateFromExecutedHiveTasks(eTasks, continueCollection);
@@ -224 +229 @@
 
               var experiment = CreateJob();
-              ExecuteOnHive(experiment, tasks, parentScopeClone, new CancellationToken());
-              continueCollection = coll;
-              Pause();
+              var engineTasks = ExecuteOnHive(experiment, tasks, parentScopeClone, new CancellationToken());
+              if (PauseWhileCalculatingChilds) {
+                continueCollection = coll;
+                Pause();
+              } else {
+                RestoreStateFromExecutedHiveTasks(engineTasks, coll);
+              }
             }
             catch {
@@ -304 +313 @@
 
     /// <summary>
-    /// This method blocks until all jobs are finished
     /// TODO: Cancelation needs to be refined; all tasks currently stay in Semaphore.WaitOne after cancelation
     /// </summary>
     /// <param name="tasks"></param>
-    private void ExecuteOnHive(RefreshableJob refreshableJob, EngineTask[] tasks, IScope parentScopeClone, CancellationToken cancellationToken) {
+    private EngineTask[] ExecuteOnHive(RefreshableJob refreshableJob, EngineTask[] tasks, IScope parentScopeClone, CancellationToken cancellationToken) {
       log.LogMessage(string.Format("Executing {0} operations on the hive.", tasks.Length));
       IScope[] scopes = new Scope[tasks.Length];
@@ -319 +327 @@
           var engineHiveTask = new EngineHiveTask(tasks[i], parentScopeClone);
           engineHiveTask.Task.Priority = this.Priority;
-          if (ParentTaskId != Guid.Empty) {
+          if (ParentTaskId != Guid.Empty && PauseWhileCalculatingChilds) {
             engineHiveTask.Task.ParentTaskId = ParentTaskId;
           }
@@ -325 +333 @@
 
           // 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(tasks[i].InitialOperation as IExecutionContext);
-          if (random != null)
-            random.Reset(random.Next());
+          //IRandom random = FindRandomParameter(tasks[i].InitialOperation as IExecutionContext);
+          //if (random != null)
+          //random.Reset(random.Next());
         }
         HiveClient.StartJob((e) => { log.LogException(e); }, refreshableJob, cancellationToken);
@@ -338 +346 @@
         handle.WaitOne();
         refreshableJob.Progress.Finished -= new EventHandler(Progress_Finished);
+
+        if (!PauseWhileCalculatingChilds) {
+
+          while (!refreshableJob.AllJobsFinished()) {
+            Thread.Sleep(1000);
+          }
+
+          List<HiveTask> allHiveTasks = refreshableJob.HiveTasks.ToList();
+          allHiveTasks.ForEach(x => this.ExecutionTimeOnHive += x.Task.ExecutionTime);
+
+          var failedJobs = allHiveTasks.Where(x => x.Task.State != TaskState.Finished);
+          if (failedJobs.Count() > 0) {
+            throw new HiveEngineException("Task (" + failedJobs.First().Task.Id + ") failed: " + failedJobs.First().Task.StateLog.Last().Exception);
+          }
+
+          List<EngineTask> engineTasks = new List<EngineTask>();
+
+          foreach (var hTask in allHiveTasks) {
+            EngineTask ehTask = (EngineTask)hTask.ItemTask;
+            engineTasks.Add(ehTask);
+          }
+
+          jobs.Clear();
+          DeleteHiveExperiment(refreshableJob.Id);
+
+          return engineTasks.ToArray();
+        } else {
+          return null;
+        }
       }
       catch (OperationCanceledException e) {