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

Timestamp:

01/07/11 17:08:36 (14 years ago)

Author:

cneumuel

Message:

#1260

• added cancellation to HiveEngine
• made HiveEngine more failproof for flaky connection to hive server
• increased WCF message quotas to 300MB

File:

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

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

@@ -73 +73 @@
             }
 
-            ExecuteOnHive(jobs);
+            ExecuteOnHive(jobs, cancellationToken);
 
             foreach (var kvp in jobs) {
@@ -134 +134 @@
     /// <summary>
     /// This method blocks until all jobs are finished
+    /// TODO: Cancelation needs to be refined; all tasks currently stay in Semaphore.WaitOne after cancelation
     /// </summary>
     /// <param name="jobDict"></param>
-    private void ExecuteOnHive(IDictionary<IOperation, OperationJob> jobDict) {
+    private void ExecuteOnHive(IDictionary<IOperation, OperationJob> jobDict, CancellationToken cancellationToken) {
       LogMessage(string.Format("Executing {0} operations on the hive.", jobDict.Count));
 
@@ -147 +148 @@
         var pluginsNeeded = ApplicationManager.Manager.Plugins.Select(x => new HivePluginInfoDto { Name = x.Name, Version = x.Version }).ToList();
         int finishedCount = 0;
+        int uploadCount = 0;
 
         // create upload-tasks
@@ -157 +159 @@
 
           uploadTasks.Add(Task.Factory.StartNew<JobDto>((keyValuePairObj) => {
-            return UploadJob(pluginsNeeded, keyValuePairObj);
-          }, kvp));
-        }
-
-        Task processUploadedJobsTask = Task.Factory.StartNew(() => {
+            return UploadJob(pluginsNeeded, keyValuePairObj, cancellationToken);
+          }, kvp, 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)];
@@ -174 +178 @@
             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);
-
+            lock (locker) {
+              uploadCount++;
+              jobIds.Add(jobDto.Id, key);
+              remainingJobIds.Add(jobDto.Id);
+              jobNumbers.Add(jobDto.Id, uploadCount);
+            }
             LogMessage(string.Format("Submitted job #{0}", jobNumbers[jobDto.Id], jobDto.Id));
             uploadTasks.Remove(task);
           }
-        });
+        }, cancellationToken, TaskCreationOptions.PreferFairness);
+        processUploadedJobsTask.Start();
 
         // poll job-statuses and create tasks for those which are finished
@@ -188 +194 @@
         var executionTimes = new List<TimeSpan>();
         while (processUploadedJobsTask.Status != TaskStatus.RanToCompletion || remainingJobIds.Count > 0) {
+          cancellationToken.ThrowIfCancellationRequested();
+
           Thread.Sleep(10000);
           using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
@@ -199 +207 @@
             if (result.State == JobState.Finished) {
               downloadTasks.Add(Task.Factory.StartNew<OperationJob>((jobIdObj) => {
-                return DownloadJob(jobNumbers, jobIdObj);
-              }, result.Id));
+                return DownloadJob(jobNumbers, jobIdObj, cancellationToken);
+              }, result.Id, cancellationToken));
             } else if (result.State == JobState.Aborted) {
               LogMessage(string.Format("Job #{0} aborted (id: {1})", jobNumbers[result.Id], result.Id));
@@ -214 +222 @@
         int downloadTasksCount = downloadTasks.Count;
         for (int i = 0; i < downloadTasksCount; i++) {
+          cancellationToken.ThrowIfCancellationRequested();
+
           var downloadTasksArray = downloadTasks.ToArray();
           var task = downloadTasksArray[Task.WaitAny(downloadTasksArray)];
@@ -241 +251 @@
     }
 
-    private JobDto UploadJob(List<HivePluginInfoDto> pluginsNeeded, object keyValuePairObj) {
+    private JobDto UploadJob(List<HivePluginInfoDto> pluginsNeeded, object keyValuePairObj, CancellationToken cancellationToken) {
       var keyValuePair = (KeyValuePair<IOperation, OperationJob>)keyValuePairObj;
       var groups = ResourceIds.Split(';');
       maxSerializedJobsInMemory.WaitOne();
-      SerializedJob serializedJob = new SerializedJob();
+      SerializedJob serializedJob = null;
+      while (serializedJob == null) { // repeat until success; rare race-conditions occur at serializations (enumeration was changed-exceptions); maybe this is because all the parent-scopes and execution-contexts at some point contain the hiveengine and the Log in here
+        cancellationToken.ThrowIfCancellationRequested();
+        try {
+          lock (Log) {
+            serializedJob = new SerializedJob();
+          }
+        }
+        catch (Exception e) {
+          LogException(e);
+        }
+      }
       maxConcurrentSerializations.WaitOne();
       serializedJob.SerializedJobData = SerializedJob.Serialize(keyValuePair.Value);
@@ -253 +274 @@
       serializedJob.JobInfo.CoresNeeded = 1;
       serializedJob.JobInfo.PluginsNeeded = pluginsNeeded;
-      ResponseObject<JobDto> response;
+      ResponseObject<JobDto> response = null;
       maxConcurrentConnections.WaitOne();
-      using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
-        response = service.Obj.AddJobWithGroupStrings(serializedJob, groups);
-        serializedJob = null;
-        maxSerializedJobsInMemory.Release();
+      while (response == null) { // repeat until success
+        cancellationToken.ThrowIfCancellationRequested();
+        try {
+          using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
+            response = service.Obj.AddJobWithGroupStrings(serializedJob, groups);
+            serializedJob = null;
+            maxSerializedJobsInMemory.Release();
+          }
+        }
+        catch (Exception e) {
+          LogException(e);
+        }
       }
       maxConcurrentConnections.Release();
@@ -264 +293 @@
     }
 
-    private OperationJob DownloadJob(IDictionary<Guid, int> jobNumbers, object jobIdObj) {
+    private OperationJob DownloadJob(IDictionary<Guid, int> jobNumbers, object jobIdObj, CancellationToken cancellationToken) {
       Guid jobId = (Guid)jobIdObj;
-      SerializedJob serializedJob;
+      SerializedJob serializedJob = null;
+      maxSerializedJobsInMemory.WaitOne();
       maxConcurrentConnections.WaitOne();
-      maxSerializedJobsInMemory.WaitOne();
-      using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
-        serializedJob = service.Obj.GetLastSerializedResult(jobId).Obj;
+      while (serializedJob == null) { // repeat until success
+        cancellationToken.ThrowIfCancellationRequested();
+        try {
+          using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
+            serializedJob = service.Obj.GetLastSerializedResult(jobId).Obj;
+          }
+        }
+        catch (Exception e) {
+          LogException(e);
+        }
       }
       maxConcurrentConnections.Release();