Changeset 6444 for branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive/3.4/HiveClient.cs

Timestamp:

06/19/11 23:21:21 (13 years ago)

Author:

cneumuel

Message:

#1233

• stability improvements for HiveExperiment and HiveEngine
• parallelized upload of jobs
• enabled cancellation of job upload
• reduced the amount of double-assignment of jobs by an additional check in HeartbeatManager
• tried to tackle the amount of deadlocks by automatically rerunning transactions
• some fixes

File:

• branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive/3.4/HiveClient.cs (modified) (6 diffs)

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive/3.4/HiveClient.cs

@@ -27 +27 @@
 using System.Security.Cryptography;
 using System.Threading;
+using System.Threading.Tasks;
 using HeuristicLab.Common;
 using HeuristicLab.Core;
@@ -104 +105 @@
 
     #region Store
-    public static void Store(IHiveItem item) {
+    public static void Store(IHiveItem item, CancellationToken cancellationToken) {
       if (item.Id == Guid.Empty) {
         if (item is RefreshableHiveExperiment) {
-          HiveClient.Instance.UploadExperiment((RefreshableHiveExperiment)item);
+          HiveClient.Instance.UploadExperiment((RefreshableHiveExperiment)item, cancellationToken);
         }
       } else {
@@ -114 +115 @@
       }
     }
-    public static void StoreAsync(Action<Exception> exceptionCallback, IHiveItem item) {
+    public static void StoreAsync(Action<Exception> exceptionCallback, IHiveItem item, CancellationToken cancellationToken) {
       var call = new Func<Exception>(delegate() {
         try {
-          Store(item);
+          Store(item, cancellationToken);
         }
         catch (Exception ex) {
@@ -159 +160 @@
     #endregion
 
-    public static void StartExperiment(Action<Exception> exceptionCallback, RefreshableHiveExperiment refreshableHiveExperiment) {
+    public static void StartExperiment(Action<Exception> exceptionCallback, RefreshableHiveExperiment refreshableHiveExperiment, CancellationToken cancellationToken) {
       HiveClient.StoreAsync(
         new Action<Exception>((Exception ex) => {
           refreshableHiveExperiment.HiveExperiment.ExecutionState = ExecutionState.Prepared;
           exceptionCallback(ex);
-        }), refreshableHiveExperiment);
+        }), refreshableHiveExperiment, cancellationToken);
       refreshableHiveExperiment.HiveExperiment.ExecutionState = ExecutionState.Started;
     }
@@ -189 +190 @@
 
     #region Upload Experiment
-    private void UploadExperiment(RefreshableHiveExperiment refreshableHiveExperiment) {
+    private Semaphore jobUploadSemaphore = new Semaphore(4, 4); // todo: take magic number into config
+    private static object jobCountLocker = new object();
+    private void UploadExperiment(RefreshableHiveExperiment refreshableHiveExperiment, CancellationToken cancellationToken) {
       try {
         refreshableHiveExperiment.HiveExperiment.Progress = new Progress("Connecting to server...");
         refreshableHiveExperiment.HiveExperiment.IsProgressing = true;
-        ServiceLocator.Instance.CallHiveService(service => {
-          IEnumerable<string> resourceNames = ToResourceNameList(refreshableHiveExperiment.HiveExperiment.ResourceNames);
-          var resourceIds = new List<Guid>();
-          foreach (var resourceName in resourceNames) {
-            Guid resourceId = service.GetResourceId(resourceName);
-            if (resourceId == Guid.Empty) {
-              throw new ResourceNotFoundException(string.Format("Could not find the resource '{0}'", resourceName));
-            }
-            resourceIds.Add(resourceId);
+
+        IEnumerable<string> resourceNames = ToResourceNameList(refreshableHiveExperiment.HiveExperiment.ResourceNames);
+        var resourceIds = new List<Guid>();
+        foreach (var resourceName in resourceNames) {
+          Guid resourceId = ServiceLocator.Instance.CallHiveService((s) => s.GetResourceId(resourceName));
+          if (resourceId == Guid.Empty) {
+            throw new ResourceNotFoundException(string.Format("Could not find the resource '{0}'", resourceName));
           }
-
-          foreach (OptimizerHiveJob hiveJob in refreshableHiveExperiment.HiveExperiment.HiveJobs.OfType<OptimizerHiveJob>()) {
-            hiveJob.SetIndexInParentOptimizerList(null);
-          }
-
-          // upload HiveExperiment
-          refreshableHiveExperiment.HiveExperiment.Progress.Status = "Uploading HiveExperiment...";
-          refreshableHiveExperiment.HiveExperiment.Id = service.AddHiveExperiment(refreshableHiveExperiment.HiveExperiment);
-
-          int totalJobCount = refreshableHiveExperiment.HiveExperiment.GetAllHiveJobs().Count();
-          int jobCount = 0;
-
-          // upload plugins
-          refreshableHiveExperiment.HiveExperiment.Progress.Status = "Uploading plugins...";
-          this.OnlinePlugins = service.GetPlugins();
-          this.AlreadyUploadedPlugins = new List<Plugin>();
-          Plugin configFilePlugin = UploadConfigurationFile(service, onlinePlugins);
-          this.alreadyUploadedPlugins.Add(configFilePlugin);
-
-          // upload jobs
-          refreshableHiveExperiment.HiveExperiment.Progress.Status = "Uploading jobs...";
-
-          foreach (HiveJob hiveJob in refreshableHiveExperiment.HiveExperiment.HiveJobs) {
-            UploadJobWithChildren(refreshableHiveExperiment.HiveExperiment.Progress, service, hiveJob, null, resourceIds, ref jobCount, totalJobCount, configFilePlugin.Id, refreshableHiveExperiment.HiveExperiment.Id, refreshableHiveExperiment.Log, refreshableHiveExperiment.HiveExperiment.IsPrivileged);
-          }
-
-          if (refreshableHiveExperiment.RefreshAutomatically) refreshableHiveExperiment.StartResultPolling();
-        });
+          resourceIds.Add(resourceId);
+        }
+
+        foreach (OptimizerHiveJob hiveJob in refreshableHiveExperiment.HiveExperiment.HiveJobs.OfType<OptimizerHiveJob>()) {
+          hiveJob.SetIndexInParentOptimizerList(null);
+        }
+
+        // upload HiveExperiment
+        refreshableHiveExperiment.HiveExperiment.Progress.Status = "Uploading HiveExperiment...";
+        refreshableHiveExperiment.HiveExperiment.Id = ServiceLocator.Instance.CallHiveService((s) => s.AddHiveExperiment(refreshableHiveExperiment.HiveExperiment));
+        cancellationToken.ThrowIfCancellationRequested();
+
+        int totalJobCount = refreshableHiveExperiment.HiveExperiment.GetAllHiveJobs().Count();
+        int[] jobCount = new int[1]; // use a reference type (int-array) instead of value type (int) in order to pass the value via a delegate to task-parallel-library
+        cancellationToken.ThrowIfCancellationRequested();
+
+        // upload plugins
+        refreshableHiveExperiment.HiveExperiment.Progress.Status = "Uploading plugins...";
+        this.OnlinePlugins = ServiceLocator.Instance.CallHiveService((s) => s.GetPlugins());
+        this.AlreadyUploadedPlugins = new List<Plugin>();
+        Plugin configFilePlugin = ServiceLocator.Instance.CallHiveService((s) => UploadConfigurationFile(s, onlinePlugins));
+        this.alreadyUploadedPlugins.Add(configFilePlugin);
+        cancellationToken.ThrowIfCancellationRequested();
+
+        if (refreshableHiveExperiment.RefreshAutomatically) refreshableHiveExperiment.StartResultPolling();
+
+        // upload jobs
+        refreshableHiveExperiment.HiveExperiment.Progress.Status = "Uploading jobs...";
+
+        var tasks = new List<Task>();
+        foreach (HiveJob hiveJob in refreshableHiveExperiment.HiveExperiment.HiveJobs) {
+          tasks.Add(Task.Factory.StartNew((hj) => {
+            UploadJobWithChildren(refreshableHiveExperiment.HiveExperiment.Progress, (HiveJob)hj, null, resourceIds, jobCount, totalJobCount, configFilePlugin.Id, refreshableHiveExperiment.HiveExperiment.Id, refreshableHiveExperiment.Log, refreshableHiveExperiment.HiveExperiment.IsPrivileged, cancellationToken);
+          }, hiveJob)
+          .ContinueWith((x) => refreshableHiveExperiment.Log.LogException(x.Exception), TaskContinuationOptions.OnlyOnFaulted));
+        }
+        Task.WaitAll(tasks.ToArray());
       }
       finally {
@@ -267 +277 @@
     /// Uploads the given job and all its child-jobs while setting the proper parentJobId values for the childs
     /// </summary>
-    /// <param name="service"></param>
-    /// <param name="hiveJob"></param>
     /// <param name="parentHiveJob">shall be null if its the root job</param>
-    /// <param name="groups"></param>
-    private void UploadJobWithChildren(IProgress progress, IHiveService service, HiveJob hiveJob, HiveJob parentHiveJob, IEnumerable<Guid> groups, ref int jobCount, int totalJobCount, Guid configPluginId, Guid hiveExperimentId, ILog log, bool isPrivileged) {
-      jobCount++;
-      progress.Status = string.Format("Serializing job {0} of {1}", jobCount, totalJobCount);
-      JobData jobData;
-      List<IPluginDescription> plugins;
-
-      if (hiveJob.ItemJob.ComputeInParallel && (hiveJob.ItemJob.Item is Optimization.Experiment || hiveJob.ItemJob.Item is Optimization.BatchRun)) {
-        hiveJob.Job.IsParentJob = true;
-        hiveJob.Job.FinishWhenChildJobsFinished = true;
-        jobData = hiveJob.GetAsJobData(true, out plugins);
-      } else {
-        hiveJob.Job.IsParentJob = false;
-        hiveJob.Job.FinishWhenChildJobsFinished = false;
-        jobData = hiveJob.GetAsJobData(false, out plugins);
-      }
-
-      TryAndRepeat(() => {
-        hiveJob.Job.PluginsNeededIds = PluginUtil.GetPluginDependencies(service, this.onlinePlugins, this.alreadyUploadedPlugins, plugins);
-      }, -1, "Failed to upload plugins");
-      hiveJob.Job.PluginsNeededIds.Add(configPluginId);
-      hiveJob.Job.HiveExperimentId = hiveExperimentId;
-      hiveJob.Job.IsPrivileged = isPrivileged;
-
-      progress.Status = string.Format("Uploading job {0} of {1} ({2} kb, {3} objects)", jobCount, totalJobCount, jobData.Data.Count() / 1024, hiveJob.ItemJob.GetObjectGraphObjects().Count());
-      progress.ProgressValue = (double)jobCount / totalJobCount;
-
-      log.LogMessage(progress.Status);
-      TryAndRepeat(() => {
-        if (parentHiveJob != null) {
-          hiveJob.Job.Id = service.AddChildJob(parentHiveJob.Job.Id, hiveJob.Job, jobData);
-        } else {
-          hiveJob.Job.Id = service.AddJob(hiveJob.Job, jobData, groups.ToList());
-        }
-      }, -1, "Failed to add job", log);
-
-      foreach (HiveJob child in hiveJob.ChildHiveJobs) {
-        UploadJobWithChildren(progress, service, child, hiveJob, groups, ref jobCount, totalJobCount, configPluginId, hiveExperimentId, log, isPrivileged);
+    private void UploadJobWithChildren(IProgress progress, HiveJob hiveJob, HiveJob parentHiveJob, IEnumerable<Guid> groups, int[] jobCount, int totalJobCount, Guid configPluginId, Guid hiveExperimentId, ILog log, bool isPrivileged, CancellationToken cancellationToken) {
+      jobUploadSemaphore.WaitOne();
+      try {
+        cancellationToken.ThrowIfCancellationRequested();
+        lock (jobCountLocker) {
+          jobCount[0]++;
+        }
+        JobData jobData = null;
+        List<IPluginDescription> plugins = null;
+
+        TryAndRepeat(() => { // workaround for persistence bug (thread-safe access to bitmaps) - remove later
+          if (hiveJob.ItemJob.ComputeInParallel && (hiveJob.ItemJob.Item is Optimization.Experiment || hiveJob.ItemJob.Item is Optimization.BatchRun)) {
+            hiveJob.Job.IsParentJob = true;
+            hiveJob.Job.FinishWhenChildJobsFinished = true;
+            jobData = hiveJob.GetAsJobData(true, out plugins);
+          } else {
+            hiveJob.Job.IsParentJob = false;
+            hiveJob.Job.FinishWhenChildJobsFinished = false;
+            jobData = hiveJob.GetAsJobData(false, out plugins);
+          }
+        }, 30, "Could not serialize job");
+        cancellationToken.ThrowIfCancellationRequested();
+
+        TryAndRepeat(() => {
+          if (!cancellationToken.IsCancellationRequested) {
+            ServiceLocator.Instance.CallHiveService((s) => hiveJob.Job.PluginsNeededIds = PluginUtil.GetPluginDependencies(s, this.onlinePlugins, this.alreadyUploadedPlugins, plugins));
+          }
+        }, -1, "Failed to upload plugins");
+        cancellationToken.ThrowIfCancellationRequested();
+        hiveJob.Job.PluginsNeededIds.Add(configPluginId);
+        hiveJob.Job.HiveExperimentId = hiveExperimentId;
+        hiveJob.Job.IsPrivileged = isPrivileged;
+
+        log.LogMessage(string.Format("Uploading job ({0} kb, {1} objects)", jobData.Data.Count() / 1024, hiveJob.ItemJob.GetObjectGraphObjects().Count()));
+        TryAndRepeat(() => {
+          if (!cancellationToken.IsCancellationRequested) {
+            if (parentHiveJob != null) {
+              hiveJob.Job.Id = ServiceLocator.Instance.CallHiveService((s) => s.AddChildJob(parentHiveJob.Job.Id, hiveJob.Job, jobData));
+            } else {
+              hiveJob.Job.Id = ServiceLocator.Instance.CallHiveService((s) => s.AddJob(hiveJob.Job, jobData, groups.ToList()));
+            }
+          }
+        }, 50, "Failed to add job", log);
+        cancellationToken.ThrowIfCancellationRequested();
+
+        lock (jobCountLocker) {
+          progress.ProgressValue = (double)jobCount[0] / totalJobCount;
+          progress.Status = string.Format("Uploaded job ({0} of {1})", jobCount[0], totalJobCount);
+        }
+
+        var tasks = new List<Task>();
+        foreach (HiveJob child in hiveJob.ChildHiveJobs) {
+          tasks.Add(Task.Factory.StartNew((tuple) => {
+            var arguments = (Tuple<HiveJob, HiveJob>)tuple;
+            UploadJobWithChildren(progress, arguments.Item1, arguments.Item2, groups, jobCount, totalJobCount, configPluginId, hiveExperimentId, log, isPrivileged, cancellationToken);
+          }, new Tuple<HiveJob, HiveJob>(child, hiveJob ))
+          .ContinueWith((x) => log.LogException(x.Exception), TaskContinuationOptions.OnlyOnFaulted));
+        }
+        Task.WaitAll(tasks.ToArray());
+      }
+      finally {
+        jobUploadSemaphore.Release();
       }
     }