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

Timestamp:

05/16/11 00:18:48 (13 years ago)

Author:

cneumuel

Message:

#1233

• stability improvements for HiveEngine

File:

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

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

@@ -7 +7 @@
 using HeuristicLab.Common;
 using HeuristicLab.Core;
-using HeuristicLab.Hive;
 using HeuristicLab.Persistence.Default.CompositeSerializers.Storable;
-using HeuristicLab.PluginInfrastructure;
 
 namespace HeuristicLab.HiveEngine {
@@ -18 +16 @@
   [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 Semaphore maxConcurrentConnections = new Semaphore(4, 4); // avoid too many connections
-    private Semaphore maxSerializedJobsInMemory = new Semaphore(4, 4); // avoid memory problems
+    private static object logLocker = new object();
     private CancellationToken cancellationToken;
-
+    
     [Storable]
     private IOperator currentOperator;
@@ -54 +51 @@
     }
 
-    [Storable]
-    private ItemCollection<RefreshableHiveExperiment> hiveExperiments;
+    // [Storable] -> HiveExperiment can't be storable, so RefreshableHiveExperiment can't be stored
+    private ItemCollection<RefreshableHiveExperiment> hiveExperiments = new ItemCollection<RefreshableHiveExperiment>();
     public ItemCollection<RefreshableHiveExperiment> HiveExperiments {
       get { return hiveExperiments; }
@@ -76 +73 @@
     public HiveEngine() {
       ResourceNames = "HEAL";
-      HiveExperiments = new ItemCollection<RefreshableHiveExperiment>();
       Priority = 0;
     }
@@ -89 +85 @@
       this.executionTimeOnHive = original.executionTimeOnHive;
       this.useLocalPlugins = original.useLocalPlugins;
+      this.hiveExperiments = cloner.Clone(original.hiveExperiments);
     }
     public override IDeepCloneable Clone(Cloner cloner) {
@@ -202 +199 @@
       target.SubScopes.AddRange(source.SubScopes);
       // TODO: validate if parent scopes match - otherwise source is invalid
-    }
-
-    // testfunction:
-    private IScope[] ExecuteLocally(EngineJob[] jobs, IScope parentScopeClone, CancellationToken cancellationToken) {
-      IScope[] scopes = new Scope[jobs.Length];
-      for (int i = 0; i < jobs.Length; i++) {
-        var serialized = PersistenceUtil.Serialize(jobs[i]);
-        var deserialized = PersistenceUtil.Deserialize<IJob>(serialized);
-        deserialized.Start();
-        while (deserialized.ExecutionState != ExecutionState.Stopped) {
-          Thread.Sleep(100);
-        }
-        var serialized2 = PersistenceUtil.Serialize(deserialized);
-        var deserialized2 = PersistenceUtil.Deserialize<EngineJob>(serialized2);
-        var newScope = ((IAtomicOperation)deserialized2.InitialOperation).Scope;
-        scopes[i] = newScope;
-      }
-      return scopes;
     }
 
@@ -256 +235 @@
             random.Reset(random.Next());
         }
-        ExperimentManagerClient.StartExperiment((e) => { throw e; }, refreshableHiveExperiment);
+        ExperimentManagerClient.StartExperiment((e) => {
+          LogException(e);
+        }, refreshableHiveExperiment);
+
         // do polling until experiment is finished and all jobs are downloaded
         while (!refreshableHiveExperiment.AllJobsFinished()) {
@@ -273 +255 @@
         refreshableHiveExperiment.RefreshAutomatically = false;
         DeleteHiveExperiment(hiveExperiment.Id);
+        ClearData(refreshableHiveExperiment);
         return scopes;
       }
@@ -290 +273 @@
     }
 
+    private void ClearData(RefreshableHiveExperiment refreshableHiveExperiment) {
+      var jobs = refreshableHiveExperiment.HiveExperiment.GetAllHiveJobs();
+      foreach (var job in jobs) {
+        job.ClearData();
+      }
+    }
+
     private void DeleteHiveExperiment(Guid hiveExperimentId) {
-      TryAndRepeat(() => {
+      ExperimentManagerClient.TryAndRepeat(() => {
         ServiceLocator.Instance.CallHiveService(s => s.DeleteHiveExperiment(hiveExperimentId));
       }, 5, string.Format("Could not delete jobs"));
     }
-
-    private static object locker = new object();
-    private Job UploadJob(object keyValuePairObj, IScope parentScopeClone, CancellationToken cancellationToken, List<Guid> resourceIds, Guid hiveExperimentId) {
-      var keyValuePair = (KeyValuePair<int, EngineJob>)keyValuePairObj;
-      Job job = new Job();
-
-      try {
-        maxSerializedJobsInMemory.WaitOne();
-        JobData jobData = new JobData();
-        IEnumerable<Type> usedTypes;
-
-        // clone operation and remove unnecessary scopes; don't do this earlier to avoid memory problems
-        lock (locker) {
-          ((IAtomicOperation)keyValuePair.Value.InitialOperation).Scope.Parent = parentScopeClone;
-          keyValuePair.Value.InitialOperation = (IOperation)keyValuePair.Value.InitialOperation.Clone();
-          if (keyValuePair.Value.InitialOperation is IAtomicOperation)
-            ((IAtomicOperation)keyValuePair.Value.InitialOperation).Scope.ClearParentScopes();
-          jobData.Data = PersistenceUtil.Serialize(keyValuePair.Value, out usedTypes);
-        }
-        var neededPlugins = new List<IPluginDescription>();
-
-        bool useAllLocalPlugins = true;
-        if (useAllLocalPlugins) {
-          // use all plugins
-          neededPlugins.AddRange(ApplicationManager.Manager.Plugins);
-        } else {
-          // use only 
-          PluginUtil.CollectDeclaringPlugins(neededPlugins, usedTypes);
-        }
-
-        job.CoresNeeded = 1;
-        job.PluginsNeededIds = ServiceLocator.Instance.CallHiveService(s => PluginUtil.GetPluginDependencies(s, this.OnlinePlugins, this.AlreadyUploadedPlugins, neededPlugins, useLocalPlugins));
-        job.Priority = priority;
-        job.HiveExperimentId = hiveExperimentId;
-
-        try {
-          maxConcurrentConnections.WaitOne();
-          while (job.Id == Guid.Empty) { // repeat until success
-            cancellationToken.ThrowIfCancellationRequested();
-            try {
-              job.Id = ServiceLocator.Instance.CallHiveService(s => s.AddJob(job, jobData, resourceIds));
-            }
-            catch (Exception e) {
-              LogException(e);
-              LogMessage("Repeating upload");
-            }
-          }
-        }
-        finally {
-          maxConcurrentConnections.Release();
-        }
-      }
-      finally {
-        maxSerializedJobsInMemory.Release();
-      }
-      return job;
-    }
-
+    
     private List<Guid> GetResourceIds() {
       return ServiceLocator.Instance.CallHiveService(service => {
@@ -368 +301 @@
     }
 
-    private EngineJob DownloadJob(IDictionary<Guid, int> jobIndices, object jobIdObj, CancellationToken cancellationToken) {
-      Guid jobId = (Guid)jobIdObj;
-      JobData jobData = null;
-      EngineJob engineJob = null;
-      try {
-        maxSerializedJobsInMemory.WaitOne();
-        maxConcurrentConnections.WaitOne();
-        while (jobData == null) { // repeat until success
-          cancellationToken.ThrowIfCancellationRequested();
-          try {
-            jobData = ServiceLocator.Instance.CallHiveService(s => s.GetJobData(jobId));
-          }
-          catch (Exception e) {
-            LogException(e);
-            LogMessage("Repeating download");
-          }
-        }
-        engineJob = PersistenceUtil.Deserialize<EngineJob>(jobData.Data);
-        jobData = null;
-        LogMessage(string.Format("Downloaded job #{0}", jobIndices[jobId] + 1, jobId));
-      }
-      finally {
-        maxConcurrentConnections.Release();
-        maxSerializedJobsInMemory.Release();
-      }
-      return engineJob;
-    }
-
     /// <summary>
     /// Threadsafe message logging
     /// </summary>
     private void LogMessage(string message) {
-      lock (Log) {
+      lock (logLocker) {
         Log.LogMessage(message);
       }
@@ -409 +314 @@
     /// </summary>
     private void LogException(Exception exception) {
-      lock (Log) {
+      lock (logLocker) {
         Log.LogException(exception);
       }
     }
 
-    /// <summary>
-    /// Executes the action. If it throws an exception it is repeated until repetition-count is reached.
-    /// If repetitions is -1, it is repeated infinitely.
-    /// </summary>
-    private static void TryAndRepeat(Action action, int repetitions, string errorMessage) {
-      try { action(); }
-      catch (Exception e) {
-        repetitions--;
-        if (repetitions <= 0)
-          throw new HiveEngineException(errorMessage, e);
-        TryAndRepeat(action, repetitions, errorMessage);
-      }
-    }
-  }
-
-  public static class EnumerableExtensions {
-    public static TimeSpan Sum(this IEnumerable<TimeSpan> times) {
-      return TimeSpan.FromMilliseconds(times.Select(e => e.TotalMilliseconds).Sum());
-    }
+    // testfunction:
+    //private IScope[] ExecuteLocally(EngineJob[] jobs, IScope parentScopeClone, CancellationToken cancellationToken) {
+    //  IScope[] scopes = new Scope[jobs.Length];
+    //  for (int i = 0; i < jobs.Length; i++) {
+    //    var serialized = PersistenceUtil.Serialize(jobs[i]);
+    //    var deserialized = PersistenceUtil.Deserialize<IJob>(serialized);
+    //    deserialized.Start();
+    //    while (deserialized.ExecutionState != ExecutionState.Stopped) {
+    //      Thread.Sleep(100);
+    //    }
+    //    var serialized2 = PersistenceUtil.Serialize(deserialized);
+    //    var deserialized2 = PersistenceUtil.Deserialize<EngineJob>(serialized2);
+    //    var newScope = ((IAtomicOperation)deserialized2.InitialOperation).Scope;
+    //    scopes[i] = newScope;
+    //  }
+    //  return scopes;
+    //}
   }
 }