Changeset 5329 for branches

Timestamp:

01/18/11 17:57:14 (14 years ago)

Author:

cneumuel

Message:

#1260

• robustified HiveEngine and HiveJobDownloader (handling of unobserved exceptions from tasks)

Location:

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

Files:

• HeuristicLab.Hive.ExperimentManager/3.3/HiveJobDownloader.cs (modified) (2 diffs)
• HeuristicLab.HiveEngine/3.3/HiveEngine.cs (modified) (4 diffs)

branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.Hive.ExperimentManager/3.3/HiveJobDownloader.cs

@@ -58 +58 @@
       abort = false;
       tasks = new List<Task<HiveJob>>();
+      TaskScheduler.UnobservedTaskException += new EventHandler<UnobservedTaskExceptionEventArgs>(TaskScheduler_UnobservedTaskException);
       foreach (Guid jobId in jobIds) {
         tasks.Add(Task<SerializedJob>.Factory.StartNew(
@@ -64 +65 @@
       }
     }
+
+    private void TaskScheduler_UnobservedTaskException(object sender, UnobservedTaskExceptionEventArgs e) {
+      e.SetObserved(); // evoid crash of process because task crashes. first exception found is handled in Results property
+    }
+
     // use semaphore to ensure only few concurrenct connections and few SerializedJob objects in memory
     private Semaphore downloadSemaphore = new Semaphore(2, 2);

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

@@ -67 +67 @@
       OperationCollection coll;
       IAtomicOperation operation;
+      TaskScheduler.UnobservedTaskException += new EventHandler<UnobservedTaskExceptionEventArgs>(TaskScheduler_UnobservedTaskException);
 
       while (ExecutionStack.Count > 0) {
@@ -116 +117 @@
         }
       }
+    }
+
+    private void TaskScheduler_UnobservedTaskException(object sender, UnobservedTaskExceptionEventArgs e) {
+      e.SetObserved(); // avoid crash of process
     }
 
@@ -275 +280 @@
       var keyValuePair = (KeyValuePair<int, OperationJob>)keyValuePairObj;
       var groups = ResourceIds.Split(';');
-      maxSerializedJobsInMemory.WaitOne();
-      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();
+      ResponseObject<JobDto> response = null;
+      try {
+        maxSerializedJobsInMemory.WaitOne();
+        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);
+          }
+        }
+        // clone operation and remove unnecessary scopes; don't do this earlier to avoid memory problems
+        lock (locker) {
+          ((IAtomicOperation)keyValuePair.Value.Operation).Scope.Parent = parentScopeClone;
+          keyValuePair.Value.Operation = (IOperation)keyValuePair.Value.Operation.Clone();
+          if (keyValuePair.Value.Operation is IAtomicOperation)
+            ((IAtomicOperation)keyValuePair.Value.Operation).Scope.ClearParentScopes();
+          serializedJob.SerializedJobData = SerializedJob.Serialize(keyValuePair.Value);
+        }
+        serializedJob.JobInfo = new JobDto();
+        serializedJob.JobInfo.State = JobState.Offline;
+        serializedJob.JobInfo.CoresNeeded = 1;
+        serializedJob.JobInfo.PluginsNeeded = pluginsNeeded;
+        serializedJob.JobInfo.Priority = priority;
         try {
-          lock (Log) {
-            serializedJob = new SerializedJob();
-          }
-        }
-        catch (Exception e) {
-          LogException(e);
-        }
-      }
-      // clone operation and remove unnecessary scopes; don't do this earlier to avoid memory problems
-      lock (locker) {
-        ((IAtomicOperation)keyValuePair.Value.Operation).Scope.Parent = parentScopeClone;
-        keyValuePair.Value.Operation = (IOperation)keyValuePair.Value.Operation.Clone();
-        if (keyValuePair.Value.Operation is IAtomicOperation)
-          ((IAtomicOperation)keyValuePair.Value.Operation).Scope.ClearParentScopes();
-        serializedJob.SerializedJobData = SerializedJob.Serialize(keyValuePair.Value);
-      }
-      serializedJob.JobInfo = new JobDto();
-      serializedJob.JobInfo.State = JobState.Offline;
-      serializedJob.JobInfo.CoresNeeded = 1;
-      serializedJob.JobInfo.PluginsNeeded = pluginsNeeded;
-      serializedJob.JobInfo.Priority = priority;
-      ResponseObject<JobDto> response = null;
-      maxConcurrentConnections.WaitOne();
-      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();
+          maxConcurrentConnections.WaitOne();
+          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;
+              }
+            }
+            catch (Exception e) {
+              LogException(e);
+            }
+          }
+        }
+        finally {
+          maxSerializedJobsInMemory.Release();
+        }
+      }
+      finally {
+        maxConcurrentConnections.Release();
+      }
       return response.Obj;
     }
@@ -323 +336 @@
       Guid jobId = (Guid)jobIdObj;
       SerializedJob serializedJob = null;
-      maxSerializedJobsInMemory.WaitOne();
-      maxConcurrentConnections.WaitOne();
-      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();
-      OperationJob operationJob = SerializedJob.Deserialize<OperationJob>(serializedJob.SerializedJobData);
-      serializedJob = null;
-      maxSerializedJobsInMemory.Release();
-      LogMessage(string.Format("Downloaded job #{0}", jobIndices[jobId] + 1, jobId));
+      OperationJob operationJob = null;
+      try {
+        maxSerializedJobsInMemory.WaitOne();
+        maxConcurrentConnections.WaitOne();
+        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);
+          }
+        }
+        operationJob = SerializedJob.Deserialize<OperationJob>(serializedJob.SerializedJobData);
+        serializedJob = null;
+        LogMessage(string.Format("Downloaded job #{0}", jobIndices[jobId] + 1, jobId));
+      }
+      finally {
+        maxConcurrentConnections.Release();
+        maxSerializedJobsInMemory.Release();
+      }
       return operationJob;
     }