Changeset 8939 for trunk/sources/HeuristicLab.Clients.Hive

Timestamp:

11/26/12 11:12:57 (11 years ago)

Author:

ascheibe

Message:

#1950

• added more aggressive locking so that the views don't read run collections that get modified in the meantime
• start downloading of tasks after the job has been uploaded completely
• fixed exceptions that got thrown when waiting for the threads that upload the tasks

Location:

trunk/sources/HeuristicLab.Clients.Hive/3.3

Files:

• HiveClient.cs (modified) (2 diffs)
• HiveTasks/HiveTask.cs (modified) (2 diffs)
• HiveTasks/OptimizerHiveTask.cs (modified) (6 diffs)
• RefreshableJob.cs (modified) (2 diffs)

trunk/sources/HeuristicLab.Clients.Hive/3.3/HiveClient.cs

@@ -294 +294 @@
         cancellationToken.ThrowIfCancellationRequested();
 
+        // upload tasks
+        refreshableJob.Progress.Status = "Uploading tasks...";
+
+        var tasks = new List<TS.Task>();
+        foreach (HiveTask hiveTask in refreshableJob.HiveTasks) {
+          var task = TS.Task.Factory.StartNew((hj) => {
+            UploadTaskWithChildren(refreshableJob.Progress, (HiveTask)hj, null, resourceIds, jobCount, totalJobCount, configFilePlugin.Id, refreshableJob.Job.Id, refreshableJob.Log, refreshableJob.Job.IsPrivileged, cancellationToken);
+          }, hiveTask);
+          task.ContinueWith((x) => refreshableJob.Log.LogException(x.Exception), TaskContinuationOptions.OnlyOnFaulted);
+          tasks.Add(task);
+        }
+        TS.Task.WaitAll(tasks.ToArray());
+      }
+      finally {
         refreshableJob.RefreshAutomatically = true;
         refreshableJob.StartResultPolling();
-
-        // upload tasks
-        refreshableJob.Progress.Status = "Uploading tasks...";
-
-        var tasks = new List<TS.Task>();
-        foreach (HiveTask hiveTask in refreshableJob.HiveTasks) {
-          tasks.Add(TS.Task.Factory.StartNew((hj) => {
-            UploadTaskWithChildren(refreshableJob.Progress, (HiveTask)hj, null, resourceIds, jobCount, totalJobCount, configFilePlugin.Id, refreshableJob.Job.Id, refreshableJob.Log, refreshableJob.Job.IsPrivileged, cancellationToken);
-          }, hiveTask)
-          .ContinueWith((x) => refreshableJob.Log.LogException(x.Exception), TaskContinuationOptions.OnlyOnFaulted));
-        }
-        try {
-          TS.Task.WaitAll(tasks.ToArray());
-        }
-        catch (AggregateException ae) {
-          if (!ae.InnerExceptions.All(e => e is TaskCanceledException)) throw ae; // for some reason the WaitAll throws a AggregateException containg a TaskCanceledException. i don't know where it comes from, however the tasks all finish properly, so for now just ignore it
-        }
         refreshableJob.Job.Modified = false;
-      }
-      finally {
         refreshableJob.IsProgressing = false;
         refreshableJob.Progress.Finish();
@@ -405 +400 @@
         var tasks = new List<TS.Task>();
         foreach (HiveTask child in hiveTask.ChildHiveTasks) {
-          tasks.Add(TS.Task.Factory.StartNew((tuple) => {
+          var task = TS.Task.Factory.StartNew((tuple) => {
             var arguments = (Tuple<HiveTask, HiveTask>)tuple;
             UploadTaskWithChildren(progress, arguments.Item1, arguments.Item2, groups, taskCount, totalJobCount, configPluginId, jobId, log, isPrivileged, cancellationToken);
-          }, new Tuple<HiveTask, HiveTask>(child, hiveTask))
-          .ContinueWith((x) => log.LogException(x.Exception), TaskContinuationOptions.OnlyOnFaulted));
+          }, new Tuple<HiveTask, HiveTask>(child, hiveTask));
+          task.ContinueWith((x) => log.LogException(x.Exception), TaskContinuationOptions.OnlyOnFaulted);
+          tasks.Add(task);
         }
         taskUploadSemaphore.Release(); semaphoreReleased = true; // the semaphore has to be release before waitall!
-        try {
-          TS.Task.WaitAll(tasks.ToArray());
-        }
-        catch (AggregateException ae) {
-          if (!ae.InnerExceptions.All(e => e is TaskCanceledException)) throw ae; // for some reason the WaitAll throws a AggregateException containg a TaskCanceledException. i don't know where it comes from, however the tasks all finish properly, so for now just ignore it
-        }
+        TS.Task.WaitAll(tasks.ToArray());
       }
       finally {

trunk/sources/HeuristicLab.Clients.Hive/3.3/HiveTasks/HiveTask.cs

@@ -39 +39 @@
     protected static object locker = new object();
     protected ReaderWriterLockSlim childHiveTasksLock = new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);
+    protected ReaderWriterLockSlim itemTaskLock = new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);
 
     public static new Image StaticItemImage {
@@ -83 +84 @@
       set {
         if (itemTask != null && syncTasksWithOptimizers) {
-          this.childHiveTasks.Clear();
+          childHiveTasksLock.EnterWriteLock();
+          try {
+            childHiveTasks.Clear();
+          }
+          finally { childHiveTasksLock.ExitWriteLock(); }
         }
         if (itemTask != value) {
-          DergisterItemTaskEvents();
-          itemTask = value;
-          RegisterItemTaskEvents();
+          itemTaskLock.EnterWriteLock();
+          try {
+            DergisterItemTaskEvents();
+            itemTask = value;
+            RegisterItemTaskEvents();
+          }
+          finally { itemTaskLock.ExitWriteLock(); }
           OnItemTaskChanged();
           IsFinishedTaskDownloaded = true;
         }
+
       }
     }

trunk/sources/HeuristicLab.Clients.Hive/3.3/HiveTasks/OptimizerHiveTask.cs

@@ -31 +31 @@
 namespace HeuristicLab.Clients.Hive {
   public class OptimizerHiveTask : HiveTask<OptimizerTask> {
-
-    Object batchRunLocker = new Object();
-
     #region Constructors and Cloning
     public OptimizerHiveTask() { }
@@ -60 +57 @@
     protected override void UpdateChildHiveTasks() {
       base.UpdateChildHiveTasks();
-      if (Task != null && syncTasksWithOptimizers) {
-        if (!ItemTask.ComputeInParallel) {
-          this.childHiveTasks.Clear();
-        } else {
-          if (ItemTask.Item is Optimization.Experiment) {
-            Optimization.Experiment experiment = (Optimization.Experiment)ItemTask.Item;
-            foreach (IOptimizer childOpt in experiment.Optimizers) {
-              var optimizerHiveTask = new OptimizerHiveTask(childOpt);
-              optimizerHiveTask.Task.Priority = Task.Priority; //inherit priority from parent
-              this.childHiveTasks.Add(optimizerHiveTask);
-            }
-          } else if (ItemTask.Item is Optimization.BatchRun) {
-            Optimization.BatchRun batchRun = ItemTask.OptimizerAsBatchRun;
-            if (batchRun.Optimizer != null) {
-              while (this.childHiveTasks.Count < batchRun.Repetitions) {
-                var optimizerHiveTask = new OptimizerHiveTask(batchRun.Optimizer);
-                optimizerHiveTask.Task.Priority = Task.Priority;
+      childHiveTasksLock.EnterWriteLock();
+      try {
+        if (Task != null && syncTasksWithOptimizers) {
+          if (!ItemTask.ComputeInParallel) {
+            this.childHiveTasks.Clear();
+          } else {
+            if (ItemTask.Item is Optimization.Experiment) {
+              Optimization.Experiment experiment = (Optimization.Experiment)ItemTask.Item;
+              foreach (IOptimizer childOpt in experiment.Optimizers) {
+                var optimizerHiveTask = new OptimizerHiveTask(childOpt);
+                optimizerHiveTask.Task.Priority = Task.Priority; //inherit priority from parent
                 this.childHiveTasks.Add(optimizerHiveTask);
               }
-              while (this.childHiveTasks.Count > batchRun.Repetitions) {
-                this.childHiveTasks.Remove(this.childHiveTasks.Last());
+            } else if (ItemTask.Item is Optimization.BatchRun) {
+              Optimization.BatchRun batchRun = ItemTask.OptimizerAsBatchRun;
+              if (batchRun.Optimizer != null) {
+                while (this.childHiveTasks.Count < batchRun.Repetitions) {
+                  var optimizerHiveTask = new OptimizerHiveTask(batchRun.Optimizer);
+                  optimizerHiveTask.Task.Priority = Task.Priority;
+                  this.childHiveTasks.Add(optimizerHiveTask);
+                }
+                while (this.childHiveTasks.Count > batchRun.Repetitions) {
+                  this.childHiveTasks.Remove(this.childHiveTasks.Last());
+                }
               }
             }
           }
         }
+      }
+      finally {
+        childHiveTasksLock.ExitWriteLock();
       }
     }
@@ -223 +226 @@
     /// </summary>
     private void UpdateOptimizerInBatchRun(BatchRun batchRun, OptimizerTask optimizerTask) {
-      if (batchRun.Optimizer == null) {
-        batchRun.Optimizer = (IOptimizer)optimizerTask.Item; // only set the first optimizer as Optimizer. if every time the Optimizer would be set, the runs would be cleared each time
-      }
-      lock (batchRunLocker) {
+      itemTaskLock.EnterWriteLock();
+      try {
+        if (batchRun.Optimizer == null) {
+          batchRun.Optimizer = (IOptimizer)optimizerTask.Item; // only set the first optimizer as Optimizer. if every time the Optimizer would be set, the runs would be cleared each time
+        }
         foreach (IRun run in optimizerTask.Item.Runs) {
           if (!batchRun.Runs.Contains(run)) {
@@ -234 +238 @@
         }
       }
+      finally {
+        itemTaskLock.ExitWriteLock();
+      }
     }
 
@@ -241 +248 @@
     /// </summary>
     private void UpdateOptimizerInExperiment(Optimization.Experiment experiment, OptimizerTask optimizerTask) {
-      if (optimizerTask.IndexInParentOptimizerList < 0)
-        throw new IndexOutOfRangeException("IndexInParentOptimizerList must be equal or greater than zero! The Task is invalid and the optimizer-tree cannot be reassembled.");
-
-      while (experiment.Optimizers.Count < optimizerTask.IndexInParentOptimizerList) {
-        experiment.Optimizers.Add(new UserDefinedAlgorithm("Placeholder")); // add dummy-entries to Optimizers so that its possible to insert the optimizerTask at the correct position
-      }
-      if (experiment.Optimizers.Count < optimizerTask.IndexInParentOptimizerList + 1) {
-        experiment.Optimizers.Add(optimizerTask.Item);
-      } else {
-        // if ComputeInParallel==true, don't replace the optimizer (except it is still a Placeholder)
-        // this is because Jobs with ComputeInParallel get submitted to hive with their child-optimizers deleted
-        if (!optimizerTask.ComputeInParallel || experiment.Optimizers[optimizerTask.IndexInParentOptimizerList].Name == "Placeholder") {
-          experiment.Optimizers[optimizerTask.IndexInParentOptimizerList] = optimizerTask.Item;
-        }
+      itemTaskLock.EnterWriteLock();
+      try {
+        if (optimizerTask.IndexInParentOptimizerList < 0)
+          throw new IndexOutOfRangeException("IndexInParentOptimizerList must be equal or greater than zero! The Task is invalid and the optimizer-tree cannot be reassembled.");
+
+        while (experiment.Optimizers.Count < optimizerTask.IndexInParentOptimizerList) {
+          experiment.Optimizers.Add(new UserDefinedAlgorithm("Placeholder")); // add dummy-entries to Optimizers so that its possible to insert the optimizerTask at the correct position
+        }
+        if (experiment.Optimizers.Count < optimizerTask.IndexInParentOptimizerList + 1) {
+          experiment.Optimizers.Add(optimizerTask.Item);
+        } else {
+          // if ComputeInParallel==true, don't replace the optimizer (except it is still a Placeholder)
+          // this is because Jobs with ComputeInParallel get submitted to hive with their child-optimizers deleted
+          if (!optimizerTask.ComputeInParallel || experiment.Optimizers[optimizerTask.IndexInParentOptimizerList].Name == "Placeholder") {
+            experiment.Optimizers[optimizerTask.IndexInParentOptimizerList] = optimizerTask.Item;
+          }
+        }
+      }
+      finally {
+        itemTaskLock.ExitWriteLock();
       }
     }
@@ -365 +378 @@
     }
 
+    public void ExecuteReadActionOnItemTask(Action action) {
+      itemTaskLock.EnterReadLock();
+      try {
+        action();
+      }
+      finally {
+        itemTaskLock.ExitReadLock();
+      }
+    }
+
     #region Helpers
     /// <summary>

trunk/sources/HeuristicLab.Clients.Hive/3.3/RefreshableJob.cs

@@ -36 +36 @@
     private object locker = new object();
     private object downloadFinishedLocker = new object();
+    object jobResultReceivedLocker = new object();
 
     public bool IsProgressing { get; set; }
@@ -261 +262 @@
 
     private void jobResultPoller_JobResultReceived(object sender, EventArgs<IEnumerable<LightweightTask>> e) {
-      foreach (LightweightTask lightweightTask in e.Value) {
-        HiveTask hiveTask = GetHiveTaskById(lightweightTask.Id);
-        if (hiveTask != null) {
-          // lastJobDataUpdate equals DateTime.MinValue right after it was uploaded. When the first results are polled, this value is updated
-          if (hiveTask.Task.State == TaskState.Offline && lightweightTask.State != TaskState.Finished && lightweightTask.State != TaskState.Failed && lightweightTask.State != TaskState.Aborted) {
-            hiveTask.Task.LastTaskDataUpdate = lightweightTask.LastTaskDataUpdate;
-          }
-
-          hiveTask.UpdateFromLightweightJob(lightweightTask);
-
-          if (!hiveTask.IsFinishedTaskDownloaded && !hiveTask.IsDownloading && hiveTask.Task.LastTaskDataUpdate < lightweightTask.LastTaskDataUpdate) {
-            log.LogMessage(string.Format("Downloading task {0}", lightweightTask.Id));
-            hiveTask.IsDownloading = true;
-            jobDownloader.DownloadTaskData(hiveTask.Task, (localJob, itemJob) => {
-              lock (downloadFinishedLocker) {
-                log.LogMessage(string.Format("Finished downloading task {0}", localJob.Id));
-                HiveTask localHiveTask = GetHiveTaskById(localJob.Id);
-
-                if (itemJob == null) {
-                  // something bad happened to this task. bad task, BAAAD task!
-                  localHiveTask.IsDownloading = false;
-                } else {
-                  // if the task is paused, download but don't integrate into parent optimizer (to avoid Prepare) 
-                  if (localJob.State == TaskState.Paused) {
-                    localHiveTask.ItemTask = itemJob;
+      lock (jobResultReceivedLocker) {
+        foreach (LightweightTask lightweightTask in e.Value) {
+          HiveTask hiveTask = GetHiveTaskById(lightweightTask.Id);
+          if (hiveTask != null) {
+            // lastJobDataUpdate equals DateTime.MinValue right after it was uploaded. When the first results are polled, this value is updated
+            if (hiveTask.Task.State == TaskState.Offline && lightweightTask.State != TaskState.Finished && lightweightTask.State != TaskState.Failed && lightweightTask.State != TaskState.Aborted) {
+              hiveTask.Task.LastTaskDataUpdate = lightweightTask.LastTaskDataUpdate;
+            }
+
+            hiveTask.UpdateFromLightweightJob(lightweightTask);
+
+            if (!hiveTask.IsFinishedTaskDownloaded && !hiveTask.IsDownloading && hiveTask.Task.LastTaskDataUpdate < lightweightTask.LastTaskDataUpdate) {
+              log.LogMessage(string.Format("Downloading task {0}", lightweightTask.Id));
+              hiveTask.IsDownloading = true;
+              jobDownloader.DownloadTaskData(hiveTask.Task, (localJob, itemJob) => {
+                lock (downloadFinishedLocker) {
+                  log.LogMessage(string.Format("Finished downloading task {0}", localJob.Id));
+                  HiveTask localHiveTask = GetHiveTaskById(localJob.Id);
+
+                  if (itemJob == null) {
+                    // something bad happened to this task. bad task, BAAAD task!
+                    localHiveTask.IsDownloading = false;
                   } else {
-                    if (localJob.ParentTaskId.HasValue) {
-                      HiveTask parentHiveTask = GetHiveTaskById(localJob.ParentTaskId.Value);
-                      parentHiveTask.IntegrateChild(itemJob, localJob.Id);
+                    // if the task is paused, download but don't integrate into parent optimizer (to avoid Prepare) 
+                    if (localJob.State == TaskState.Paused) {
+                      localHiveTask.ItemTask = itemJob;
                     } else {
-                      localHiveTask.ItemTask = itemJob;
+                      if (localJob.ParentTaskId.HasValue) {
+                        HiveTask parentHiveTask = GetHiveTaskById(localJob.ParentTaskId.Value);
+                        parentHiveTask.IntegrateChild(itemJob, localJob.Id);
+                      } else {
+                        localHiveTask.ItemTask = itemJob;
+                      }
                     }
+                    localHiveTask.IsDownloading = false;
+                    localHiveTask.Task.LastTaskDataUpdate = lightweightTask.LastTaskDataUpdate;
                   }
-                  localHiveTask.IsDownloading = false;
-                  localHiveTask.Task.LastTaskDataUpdate = lightweightTask.LastTaskDataUpdate;
                 }
-              }
-            });
-          }
-        }
-      }
-      GC.Collect(); // force GC, because .NET is too lazy here (deserialization takes a lot of memory)
-      if (AllJobsFinished()) {
-        this.ExecutionState = Core.ExecutionState.Stopped;
-        StopResultPolling();
-      }
-      UpdateTotalExecutionTime();
-      UpdateStatistics();
-      OnStateLogListChanged();
-      OnTaskReceived();
+              });
+            }
+          } else
+            throw new Exception("This should not happen");
+        }
+        GC.Collect(); // force GC, because .NET is too lazy here (deserialization takes a lot of memory)
+        if (AllJobsFinished()) {
+          this.ExecutionState = Core.ExecutionState.Stopped;
+          StopResultPolling();
+        }
+        UpdateTotalExecutionTime();
+        UpdateStatistics();
+        OnStateLogListChanged();
+        OnTaskReceived();
+      }
     }