Changeset 6178 for branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.HiveEngine/3.4/HiveEngine.cs

Timestamp:

05/10/11 17:58:59 (13 years ago)

Author:

cneumuel

Message:

#1233

• added semaphores to ensure an appdomain is never unloaded when the start method has not finished
• HiveEngine uploading and downloading of jobs works and is displayed in the view

File:

• branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.HiveEngine/3.4/HiveEngine.cs (modified) (6 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;
@@ -140 +141 @@
 
             IScope[] scopes = ExecuteOnHive(jobs, parentScopeClone, cancellationToken);
-            
+            //IScope[] scopes = ExecuteLocally(jobs, parentScopeClone, cancellationToken);
+
             for (int i = 0; i < coll.Count; i++) {
               if (coll[i] is IAtomicOperation) {
@@ -203 +205 @@
 
     // 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 job = (EngineJob)jobs[i].Clone();
-    //    job.Start();
-    //    while (job.ExecutionState != ExecutionState.Stopped) {
-    //      Thread.Sleep(100);
-    //    }
-    //    scopes[i] = ((IAtomicOperation)job.InitialOperation).Scope;
-    //  }
-    //  return scopes;
-    //}
+    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;
+    }
 
     /// <summary>
@@ -230 +236 @@
       try {
         List<Guid> remainingJobIds = new List<Guid>();
-        List<LightweightJob> lightweightJobs;
-
-        int finishedCount = 0;
-        int uploadCount = 0;
 
         // create hive experiment
         hiveExperiment.Name = "HiveEngine Run " + hiveExperiments.Count;
+        hiveExperiment.DateCreated = DateTime.Now;
         hiveExperiment.UseLocalPlugins = this.UseLocalPlugins;
         hiveExperiment.ResourceNames = this.ResourceNames;
-        hiveExperiment.Id = ServiceLocator.Instance.CallHiveService(s => s.AddHiveExperiment(hiveExperiment));
         var refreshableHiveExperiment = new RefreshableHiveExperiment(hiveExperiment);
+        refreshableHiveExperiment.IsControllable = false;
         hiveExperiments.Add(refreshableHiveExperiment);
 
@@ -246 +249 @@
         var uploadTasks = new List<Task<Job>>();
         for (int i = 0; i < jobs.Length; i++) {
-          var job = jobs[i];
+          hiveExperiment.HiveJobs.Add(new EngineHiveJob(jobs[i], parentScopeClone));
 
           // 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(job.InitialOperation as IExecutionContext);
+          IRandom random = FindRandomParameter(jobs[i].InitialOperation as IExecutionContext);
           if (random != null)
             random.Reset(random.Next());
-
-          uploadTasks.Add(Task.Factory.StartNew<Job>((keyValuePairObj) => {
-            return UploadJob(keyValuePairObj, parentScopeClone, cancellationToken, GetResourceIds(), hiveExperiment.Id);
-          }, new KeyValuePair<int, EngineJob>(i, job), cancellationToken));
-        }
-
-        Task processUploadedJobsTask = new Task(() => {
-          // process finished upload-tasks
-          int uploadTasksCount = uploadTasks.Count;
-          for (int i = 0; i < uploadTasksCount; i++) {
-            cancellationToken.ThrowIfCancellationRequested();
-
-            var uploadTasksArray = uploadTasks.ToArray();
-            var task = uploadTasksArray[Task.WaitAny(uploadTasksArray)];
-            if (task.Status == TaskStatus.Faulted) {
-              LogException(task.Exception);
-              throw task.Exception;
-            }
-
-            int key = ((KeyValuePair<int, EngineJob>)task.AsyncState).Key;
-            Job job = task.Result;
-            lock (locker) {
-              uploadCount++;
-              jobIndices.Add(job.Id, key);
-              remainingJobIds.Add(job.Id);
-            }
-            jobs[key] = null; // relax memory
-            LogMessage(string.Format("Uploaded job #{0}", key + 1, job.Id));
-            uploadTasks.Remove(task);
-          }
-        }, cancellationToken, TaskCreationOptions.PreferFairness);
-        processUploadedJobsTask.Start();
-
-        refreshableHiveExperiment.RefreshAutomatically = true;
-        
+        }
+        ExperimentManagerClient.StartExperiment((e) => { throw e; }, refreshableHiveExperiment);
+        // do polling until experiment is finished and all jobs are downloaded
         while (!refreshableHiveExperiment.AllJobsFinished()) {
-          Thread.Sleep(1000);
-          // update time
-          // handle cancellation
-        }
-
-
-
-        // poll job-statuses and create tasks for those which are finished
-        var downloadTasks = new List<Task<EngineJob>>();
-        var executionTimes = new List<TimeSpan>();
-        var executionTimeOnHiveBefore = executionTimeOnHive;
-        while (processUploadedJobsTask.Status != TaskStatus.RanToCompletion || remainingJobIds.Count > 0) {
-          cancellationToken.ThrowIfCancellationRequested();
-
-          Thread.Sleep(10000);
-          try {
-            lightweightJobs = ServiceLocator.Instance.CallHiveService(s => s.GetLightweightJobs(remainingJobIds));
-
-            var jobsFinished = lightweightJobs.Where(j => j.State == JobState.Finished || j.State == JobState.Failed || j.State == JobState.Aborted);
-            finishedCount += jobsFinished.Count();
-            if (jobsFinished.Count() > 0) LogMessage(string.Format("Finished: {0}/{1}", finishedCount, jobs.Length));
-            ExecutionTimeOnHive = executionTimeOnHiveBefore + executionTimes.Sum() + lightweightJobs.Select(x => x.ExecutionTime.HasValue ? x.ExecutionTime.Value : TimeSpan.Zero).Sum();
-
-            foreach (var result in jobsFinished) {
-              if (result.State == JobState.Finished) {
-                downloadTasks.Add(Task.Factory.StartNew<EngineJob>((jobIdObj) => {
-                  return DownloadJob(jobIndices, jobIdObj, cancellationToken);
-                }, result.Id, cancellationToken));
-              } else if (result.State == JobState.Aborted) {
-                LogMessage(string.Format("Job #{0} aborted (id: {1})", jobIndices[result.Id] + 1, result.Id));
-              } else if (result.State == JobState.Failed) {
-                LogMessage(string.Format("Job #{0} failed (id: {1}): {2}", jobIndices[result.Id] + 1, result.Id, result.CurrentStateLog != null ? result.CurrentStateLog.Exception : string.Empty));
-              }
-              remainingJobIds.Remove(result.Id);
-              executionTimes.Add(result.ExecutionTime.HasValue ? result.ExecutionTime.Value : TimeSpan.Zero);
-            }
-          }
-          catch (Exception e) {
-            LogException(e);
-          }
-        }
-
-        // process finished download-tasks
-        int downloadTasksCount = downloadTasks.Count;
-        for (int i = 0; i < downloadTasksCount; i++) {
-          cancellationToken.ThrowIfCancellationRequested();
-
-          var downloadTasksArray = downloadTasks.ToArray();
-          var task = downloadTasksArray[Task.WaitAny(downloadTasksArray)];
-          var jobId = (Guid)task.AsyncState;
-          if (task.Status == TaskStatus.Faulted) {
-            LogException(task.Exception);
-            throw task.Exception;
-          }
-          scopes[jobIndices[(Guid)task.AsyncState]] = ((IAtomicOperation)task.Result.InitialOperation).Scope;
-          downloadTasks.Remove(task);
-        }
-
-        LogMessage(string.Format("All jobs finished (TotalExecutionTime: {0}).", executionTimes.Sum()));
+          Thread.Sleep(500);
+          this.ExecutionTimeOnHive = TimeSpan.FromMilliseconds(hiveExperiments.Sum(x => x.HiveExperiment.ExecutionTime.TotalMilliseconds));
+        }
+        LogMessage(string.Format("{0} finished (TotalExecutionTime: {1}).", refreshableHiveExperiment.ToString(), refreshableHiveExperiment.HiveExperiment.ExecutionTime));
+
+        // get scopes
+        int j = 0;
+        foreach (var hiveJob in hiveExperiment.HiveJobs) {
+          var scope = ((IAtomicOperation) ((EngineJob)hiveJob.ItemJob).InitialOperation).Scope;
+          scopes[j++] = scope;
+        }
+        refreshableHiveExperiment.RefreshAutomatically = false;
         DeleteHiveExperiment(hiveExperiment.Id);
-
         return scopes;
       }
@@ -505 +428 @@
   }
 
-  public static class ScopeExtensions {
-    public static void ClearParentScopes(this IScope scope) {
-      scope.ClearParentScopes(null);
-    }
-
-    public static void ClearParentScopes(this IScope scope, IScope childScope) {
-      if (childScope != null) {
-        scope.SubScopes.Clear();
-        scope.SubScopes.Add(childScope);
-      }
-      if (scope.Parent != null)
-        scope.Parent.ClearParentScopes(scope);
-    }
-  }
-
   public static class EnumerableExtensions {
     public static TimeSpan Sum(this IEnumerable<TimeSpan> times) {