Changeset 5228 for branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine

Timestamp:

01/06/11 18:34:38 (14 years ago)

Author:

cneumuel

Message:

#1260

• further tweaked HiveEngine to be faster and more robust

Location:

branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine/3.3

Files:

• HeuristicLab.HiveEngine-3.3.csproj (modified) (1 diff)
• HiveEngine.cs (modified) (6 diffs)
• HiveEngineException.cs (added)

branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine/3.3/HeuristicLab.HiveEngine-3.3.csproj

@@ -88 +88 @@
   </ItemGroup>
   <ItemGroup>
+    <Compile Include="HiveEngineException.cs" />
     <Compile Include="ScopeMergeException.cs" />
     <Compile Include="Views\HiveEngineView.cs">

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

@@ -23 +23 @@
   [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 maxConcurrentSerializations = new Semaphore(1, 1); // allow ony ONE concurrent serialization, because the operations share the same ParentScopes and serializing the same objects concurrently causes problems
+    private Semaphore maxSerializedJobsInMemory = new Semaphore(2, 2); // avoid memory problems
     private CancellationToken cancellationToken;
 
@@ -76 +79 @@
                 ReIntegrateScope(kvp.Value.Operation as IAtomicOperation, kvp.Key as IAtomicOperation);
               } else if (kvp.Key is OperationCollection) {
-                OperationCollection ocoll = (OperationCollection)kvp.Key;
-                for (int i = ocoll.Count - 1; i >= 0; i--)
-                  if (ocoll[i] != null) executionStack.Push(ocoll[i]);
+                // todo ??
               }
             }
@@ -137 +138 @@
     private void ExecuteOnHive(IDictionary<IOperation, OperationJob> jobDict) {
       LogMessage(string.Format("Executing {0} operations on the hive.", jobDict.Count));
-      Semaphore concurrentUploads = new Semaphore(6, 6);
+
+      object locker = new object();
       try {
         IDictionary<Guid, IOperation> jobIds = new Dictionary<Guid, IOperation>();
@@ -144 +146 @@
         JobResultList results;
         var pluginsNeeded = ApplicationManager.Manager.Plugins.Select(x => new HivePluginInfoDto { Name = x.Name, Version = x.Version }).ToList();
-
-        List<JobDto> jobs = new List<JobDto>();
+        int finishedCount = 0;
+
+        // create upload-tasks
+        var uploadTasks = new List<Task<JobDto>>();
         foreach (var kvp in jobDict) {
-
           // 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(kvp.Key as IExecutionContext);
@@ -153 +156 @@
             random.Reset(random.Next());
 
-          Task.Factory.StartNew((operationJob) => {
-            var groups = ResourceIds.Split(';');
-            SerializedJob serializedJob = new SerializedJob();
-            serializedJob.SerializedJobData = SerializedJob.Serialize(operationJob);
-            serializedJob.JobInfo = new JobDto();
-            serializedJob.JobInfo.State = JobState.Offline;
-            serializedJob.JobInfo.CoresNeeded = 1;
-            serializedJob.JobInfo.PluginsNeeded = pluginsNeeded;
-            ResponseObject<JobDto> response;
-            concurrentUploads.WaitOne();
-            using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
-              response = service.Obj.AddJobWithGroupStrings(serializedJob, groups);
+          uploadTasks.Add(Task.Factory.StartNew<JobDto>((keyValuePairObj) => {
+            return UploadJob(pluginsNeeded, keyValuePairObj);
+          }, kvp));
+        }
+
+        Task processUploadedJobsTask = Task.Factory.StartNew(() => {
+          // process finished upload-tasks
+          int uploadTasksCount = uploadTasks.Count;
+          for (int i = 0; i < uploadTasksCount; i++) {
+            var uploadTasksArray = uploadTasks.ToArray();
+            var task = uploadTasksArray[Task.WaitAny(uploadTasksArray)];
+            if (task.Status == TaskStatus.Faulted) {
+              LogException(task.Exception);
+              throw task.Exception;
             }
-            concurrentUploads.Release();
-            jobs.Add(response.Obj);
-            jobIds.Add(response.Obj.Id, kvp.Key);
-            lock (remainingJobIds) {
-              remainingJobIds.Add(response.Obj.Id);
-              jobNumbers.Add(response.Obj.Id, remainingJobIds.Count);
-            }
-            LogMessage(string.Format("Submitted job #{0} (id: {1})", jobNumbers[response.Obj.Id], response.Obj.Id));
-          }, kvp.Value);
-        }
-
-        while (remainingJobIds.Count != jobDict.Count) {
-          Thread.Sleep(1000);
-        }
-
-        LogMessage("Waiting for results...");
-        int jobsFinishedCount = 0;
-        Semaphore concurrentDownloads = new Semaphore(4, 4);
-        while (remainingJobIds.Count > 0) {
-          Thread.Sleep(5000);
+
+            IOperation key = ((KeyValuePair<IOperation, OperationJob>)task.AsyncState).Key;
+            JobDto jobDto = task.Result;
+
+            jobIds.Add(jobDto.Id, key);
+            remainingJobIds.Add(jobDto.Id);
+            jobNumbers.Add(jobDto.Id, remainingJobIds.Count);
+
+            LogMessage(string.Format("Submitted job #{0}", jobNumbers[jobDto.Id], jobDto.Id));
+            uploadTasks.Remove(task);
+          }
+        });
+
+        // poll job-statuses and create tasks for those which are finished
+        var downloadTasks = new List<Task<OperationJob>>();
+        var executionTimes = new List<TimeSpan>();
+        while (processUploadedJobsTask.Status != TaskStatus.RanToCompletion || remainingJobIds.Count > 0) {
+          Thread.Sleep(10000);
           using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
             results = service.Obj.GetJobResults(remainingJobIds).Obj;
           }
-          foreach (var result in results) {
+          var jobsFinished = results.Where(j => j.State == JobState.Finished || j.State == JobState.Failed || j.State == JobState.Aborted);
+          finishedCount += jobsFinished.Count();
+          var totalExecutionTime = TimeSpan.FromMilliseconds(results.Select(j => j.ExecutionTime).Union(executionTimes).Select(e => e.TotalMilliseconds).Sum());
+          LogMessage(string.Format("Results polled. Jobs finished: {0}/{1}, TotalExecutionTime: {2}", finishedCount, jobDict.Count, totalExecutionTime));
+          foreach (var result in jobsFinished) {
             if (result.State == JobState.Finished) {
-              Task.Factory.StartNew((jobIdObj) => {
-                Guid jobId = (Guid)jobIdObj;
-                SerializedJob serializedJob;
-                concurrentDownloads.WaitOne();
-                using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
-                  serializedJob = service.Obj.GetLastSerializedResult(jobId).Obj;
-                }
-                concurrentDownloads.Release();
-                OperationJob operationJob = SerializedJob.Deserialize<OperationJob>(serializedJob.SerializedJobData);
-                jobDict[jobIds[jobId]] = operationJob;
-                LogMessage(string.Format("Downloaded job #{0} (id: {1})", jobNumbers[jobId], jobId));
-                jobsFinishedCount++;
-              }, result.Id);
-
-              remainingJobIds.Remove(result.Id);
+              downloadTasks.Add(Task.Factory.StartNew<OperationJob>((jobIdObj) => {
+                return DownloadJob(jobNumbers, jobIdObj);
+              }, result.Id));
             } else if (result.State == JobState.Aborted) {
               LogMessage(string.Format("Job #{0} aborted (id: {1})", jobNumbers[result.Id], result.Id));
-              remainingJobIds.Remove(result.Id);
             } else if (result.State == JobState.Failed) {
               LogMessage(string.Format("Job {0} failed (id: {1}): {2}", jobNumbers[result.Id], result.Id, result.Exception));
-              remainingJobIds.Remove(result.Id);
             }
-          }
-        }
-
-        // wait for all tasks to finish downloading and deserializing
-        while (jobsFinishedCount != jobDict.Count) {
-          Thread.Sleep(1000);
-        }
-
-        LogMessage("All jobs finished. Deleting jobs on hive.");
+            remainingJobIds.Remove(result.Id);
+            executionTimes.Add(result.ExecutionTime);
+          }
+        }
+
+        // process finished download-tasks
+        int downloadTasksCount = downloadTasks.Count;
+        for (int i = 0; i < downloadTasksCount; i++) {
+          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;
+          }
+          jobDict[jobIds[(Guid)task.AsyncState]] = task.Result;
+          downloadTasks.Remove(task);
+        }
+
+        LogMessage(string.Format("All jobs finished (TotalExecutionTime: {0}). Deleting jobs on hive.", TimeSpan.FromMilliseconds(executionTimes.Select(e => e.TotalMilliseconds).Sum())));
         // delete jobs
         using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
@@ -237 +241 @@
     }
 
-    private void DownloadJob(Guid jobId, IDictionary<IOperation, OperationJob> jobDict, IDictionary<Guid, IOperation> jobIds) {
+    private JobDto UploadJob(List<HivePluginInfoDto> pluginsNeeded, object keyValuePairObj) {
+      var keyValuePair = (KeyValuePair<IOperation, OperationJob>)keyValuePairObj;
+      var groups = ResourceIds.Split(';');
+      maxSerializedJobsInMemory.WaitOne();
+      SerializedJob serializedJob = new SerializedJob();
+      maxConcurrentSerializations.WaitOne();
+      serializedJob.SerializedJobData = SerializedJob.Serialize(keyValuePair.Value);
+      maxConcurrentSerializations.Release();
+      serializedJob.JobInfo = new JobDto();
+      serializedJob.JobInfo.State = JobState.Offline;
+      serializedJob.JobInfo.CoresNeeded = 1;
+      serializedJob.JobInfo.PluginsNeeded = pluginsNeeded;
+      ResponseObject<JobDto> response;
+      maxConcurrentConnections.WaitOne();
+      using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
+        response = service.Obj.AddJobWithGroupStrings(serializedJob, groups);
+        serializedJob = null;
+        maxSerializedJobsInMemory.Release();
+      }
+      maxConcurrentConnections.Release();
+      return response.Obj;
+    }
+
+    private OperationJob DownloadJob(IDictionary<Guid, int> jobNumbers, object jobIdObj) {
+      Guid jobId = (Guid)jobIdObj;
       SerializedJob serializedJob;
-      using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
+      maxConcurrentConnections.WaitOne();
+      maxSerializedJobsInMemory.WaitOne();
+      using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
         serializedJob = service.Obj.GetLastSerializedResult(jobId).Obj;
       }
+      maxConcurrentConnections.Release();
       OperationJob operationJob = SerializedJob.Deserialize<OperationJob>(serializedJob.SerializedJobData);
-      jobDict[jobIds[jobId]] = operationJob;
-      LogMessage(string.Format("Downloaded job (id: {0})", jobId));
+      serializedJob = null;
+      maxSerializedJobsInMemory.Release();
+      LogMessage(string.Format("Downloaded job #{0}", jobNumbers[jobId], jobId));
+      return operationJob;
     }