Changeset 11082

Timestamp:

07/03/14 16:03:14 (11 years ago)

Author:

ascheibe

Message:

#2153

• added a new method HandleStartStopPauseError in Executor to handle error conditions in the same way
• added timeouts for semaphores so that failed tasks or tasks with endless loops don't block the slave
• removed ExceptionOccured events from Executor/SlaveTask/TaskManager and use TaskFailed instead
• removed another ExcpetionOccured event in HeartbeatManager that was never used

Location:

trunk/sources/HeuristicLab.Clients.Hive.Slave/3.3

Files:

• Core.cs (modified) (6 diffs)
• Executor.cs (modified) (6 diffs)
• ExecutorMessage.cs (modified) (1 diff)
• Manager/ConfigManager.cs (modified) (1 diff)
• Manager/HeartbeatManager.cs (modified) (4 diffs)
• Manager/TaskManager.cs (modified) (7 diffs)
• SlaveStatusInfo.cs (modified) (3 diffs)
• SlaveTask.cs (modified) (7 diffs)
• StatusCommons.cs (modified) (2 diffs)

trunk/sources/HeuristicLab.Clients.Hive.Slave/3.3/Core.cs

@@ -204 +204 @@
       TS.Task.Factory.StartNew(HandleCalculateTask, jobId)
       .ContinueWith((t) => {
-        SlaveStatusInfo.IncrementExceptionOccured();
+        SlaveStatusInfo.IncrementTasksFailed();
         SlaveClientCom.Instance.LogMessage(t.Exception.ToString());
       }, TaskContinuationOptions.OnlyOnFaulted);
@@ -212 +212 @@
       TS.Task.Factory.StartNew(HandleStopTask, jobId)
        .ContinueWith((t) => {
-         SlaveStatusInfo.IncrementExceptionOccured();
+         SlaveStatusInfo.IncrementTasksFailed();
          SlaveClientCom.Instance.LogMessage(t.Exception.ToString());
        }, TaskContinuationOptions.OnlyOnFaulted);
@@ -220 +220 @@
       TS.Task.Factory.StartNew(HandlePauseTask, jobId)
        .ContinueWith((t) => {
-         SlaveStatusInfo.IncrementExceptionOccured();
+         SlaveStatusInfo.IncrementTasksFailed();
          SlaveClientCom.Instance.LogMessage(t.Exception.ToString());
        }, TaskContinuationOptions.OnlyOnFaulted);
@@ -228 +228 @@
       TS.Task.Factory.StartNew(HandleAbortTask, jobId)
        .ContinueWith((t) => {
-         SlaveStatusInfo.IncrementExceptionOccured();
+         SlaveStatusInfo.IncrementTasksFailed();
          SlaveClientCom.Instance.LogMessage(t.Exception.ToString());
        }, TaskContinuationOptions.OnlyOnFaulted);
@@ -328 +328 @@
       this.taskManager.TaskStopped += new EventHandler<EventArgs<SlaveTask, TaskData>>(taskManager_TaskStopped);
       this.taskManager.TaskFailed += new EventHandler<EventArgs<Tuple<SlaveTask, TaskData, Exception>>>(taskManager_TaskFailed);
-      this.taskManager.ExceptionOccured += new EventHandler<EventArgs<SlaveTask, Exception>>(taskManager_ExceptionOccured);
       this.taskManager.TaskAborted += new EventHandler<EventArgs<SlaveTask>>(taskManager_TaskAborted);
     }
@@ -391 +390 @@
       }
       catch (TaskNotFoundException ex) {
-        SlaveStatusInfo.IncrementExceptionOccured();
+        SlaveStatusInfo.IncrementTasksFailed();
         SlaveClientCom.Instance.LogMessage(ex.ToString());
       }
       catch (Exception ex) {
-        SlaveStatusInfo.IncrementExceptionOccured();
+        SlaveStatusInfo.IncrementTasksFailed();
         SlaveClientCom.Instance.LogMessage(ex.ToString());
       }
-    }
-
-    private void taskManager_ExceptionOccured(object sender, EventArgs<SlaveTask, Exception> e) {
-      SlaveStatusInfo.DecrementUsedCores(e.Value.CoresNeeded);
-      SlaveStatusInfo.IncrementExceptionOccured();
-      heartbeatManager.AwakeHeartBeatThread();
-      SlaveClientCom.Instance.LogMessage(string.Format("Exception occured for task {0}: {1}", e.Value.TaskId, e.Value2.ToString()));
-      wcfService.UpdateJobState(e.Value.TaskId, TaskState.Waiting, e.Value2.ToString());
     }
 

trunk/sources/HeuristicLab.Clients.Hive.Slave/3.3/Executor.cs

@@ -85 +85 @@
         task.Start();
         if (!startTaskSem.WaitOne(Settings.Default.ExecutorSemTimeouts)) {
-          taskDataInvalid = true;
           throw new TimeoutException("Timeout when starting the task. TaskStarted event was not fired.");
         }
       }
       catch (Exception e) {
-        this.CurrentException = e;
-        taskDataInvalid = true;
-        Task_TaskFailed(this, new EventArgs<Exception>(e));
-      } finally {
+        HandleStartStopPauseError(e);
+      }
+      finally {
         taskStartedSem.Set();
       }
@@ -103 +101 @@
       taskStartedSem.WaitOne(Settings.Default.ExecutorSemTimeouts);
       if (task == null) {
-        CurrentException = new Exception("Pausing task " + this.TaskId + ": Task is null");
-        executorQueue.AddMessage(ExecutorMessageType.ExceptionOccured);
+        HandleStartStopPauseError(new Exception("Pausing task " + this.TaskId + ": Task is null"));
         return;
       }
@@ -112 +109 @@
           task.Pause();
           //we need to block the pause...
-          pauseStopSem.WaitOne();
+          if (!pauseStopSem.WaitOne(Settings.Default.ExecutorSemTimeouts)) {
+            throw new Exception("Pausing task " + this.TaskId + " timed out.");
+          }
         }
         catch (Exception ex) {
-          CurrentException = new Exception("Error pausing task " + this.TaskId + ": " + ex.ToString());
-          executorQueue.AddMessage(ExecutorMessageType.ExceptionOccured);
+          HandleStartStopPauseError(ex);
         }
       }
@@ -125 +123 @@
       // wait until task is started. if this does not happen, the Task is null an we give up
       taskStartedSem.WaitOne(Settings.Default.ExecutorSemTimeouts);
+      wasTaskAborted = true;
+
       if (task == null) {
-        CurrentException = new Exception("Stopping task " + this.TaskId + ": Task is null");
-        executorQueue.AddMessage(ExecutorMessageType.ExceptionOccured);
-      }
-      wasTaskAborted = true;
+        HandleStartStopPauseError(new Exception("Stopping task " + this.TaskId + ": Task is null"));
+        return;
+      }
 
       if ((ExecutionState == ExecutionState.Started) || (ExecutionState == ExecutionState.Paused)) {
         try {
           task.Stop();
-          pauseStopSem.WaitOne();
+          if (!pauseStopSem.WaitOne(Settings.Default.ExecutorSemTimeouts)) {
+            throw new Exception("Stopping task " + this.TaskId + " timed out.");
+          }
         }
         catch (Exception ex) {
-          CurrentException = new Exception("Error stopping task " + this.TaskId + ": " + ex.ToString());
-          executorQueue.AddMessage(ExecutorMessageType.ExceptionOccured);
+          HandleStartStopPauseError(ex);
         }
       }
@@ -190 +190 @@
       if (task != null && task.ExecutionState == ExecutionState.Started) {
         throw new InvalidStateException("Task is still running");
+      }
+
+      TaskData taskData = null;
+      if (task == null) {
+        if (CurrentException == null) {
+          CurrentException = new Exception("Task with id " + this.TaskId + " is null, sending empty task");
+        }
       } else {
-        TaskData taskData = new TaskData();
-        if (task == null) {
-          //send empty task and save exception
-          taskData.Data = PersistenceUtil.Serialize(new TaskData());
-          if (CurrentException == null) {
-            CurrentException = new Exception("Task with id " + this.TaskId + " is null, sending empty task");
-          }
-        } else {
-          taskData.Data = PersistenceUtil.Serialize(task);
-        }
+        taskData = new TaskData();
+        taskData.Data = PersistenceUtil.Serialize(task);
         taskData.TaskId = TaskId;
-        return taskData;
-      }
+      }
+      return taskData;
     }
 
@@ -211 +210 @@
       task = null;
     }
+
+    private void HandleStartStopPauseError(Exception e) {
+      taskDataInvalid = true;
+      Task_TaskFailed(this, new EventArgs<Exception>(e));
+    }
   }
 }

trunk/sources/HeuristicLab.Clients.Hive.Slave/3.3/ExecutorMessage.cs

@@ -30 +30 @@
     TaskStopped,
     TaskFailed,
-    StopExecutorMonitoringThread,
-    ExceptionOccured
+    StopExecutorMonitoringThread
   }
 

trunk/sources/HeuristicLab.Clients.Hive.Slave/3.3/Manager/ConfigManager.cs

@@ -117 +117 @@
       st.JobsFetched = SlaveStatusInfo.TasksFetched;
       st.JobsFailed = SlaveStatusInfo.TasksFailed;
-      st.ExceptionsOccured = SlaveStatusInfo.ExceptionsOccured;
 
       st.Jobs = jobManager.GetExecutionTimes().Select(x => new TaskStatus { TaskId = x.Key, ExecutionTime = x.Value }).ToList();

trunk/sources/HeuristicLab.Clients.Hive.Slave/3.3/Manager/HeartbeatManager.cs

@@ -24 +24 @@
 using System.Threading;
 using HeuristicLab.Clients.Hive.SlaveCore.Properties;
-using HeuristicLab.Common;
 
 namespace HeuristicLab.Clients.Hive.SlaveCore {
@@ -115 +114 @@
               if (msgs == null) {
                 SlaveClientCom.Instance.LogMessage("Error getting response from HB");
-                OnExceptionOccured(new Exception("Error getting response from HB"));
               } else {
                 SlaveClientCom.Instance.LogMessage("HB Response received (" + msgs.Count + "): ");
@@ -126 +124 @@
         catch (Exception e) {
           SlaveClientCom.Instance.LogMessage("Heartbeat thread failed: " + e.ToString());
-          OnExceptionOccured(e);
         }
         waitHandle.WaitOne(this.interval);
@@ -132 +129 @@
       SlaveClientCom.Instance.LogMessage("Heartbeat thread stopped");
     }
-
-    #region Eventhandler
-    public event EventHandler<EventArgs<Exception>> ExceptionOccured;
-    private void OnExceptionOccured(Exception e) {
-      var handler = ExceptionOccured;
-      if (handler != null) handler(this, new EventArgs<Exception>(e));
-    }
-    #endregion
   }
 }

trunk/sources/HeuristicLab.Clients.Hive.Slave/3.3/Manager/TaskManager.cs

@@ -71 +71 @@
       try {
         if (slaveTasks.ContainsKey(task.Id)) {
-          SlaveStatusInfo.IncrementExceptionOccured();
+          SlaveStatusInfo.IncrementTasksFailed();
           throw new TaskAlreadyRunningException(task.Id);
         } else {
@@ -185 +185 @@
       slaveTask.TaskStopped += new EventHandler<EventArgs<Guid>>(slaveTask_TaskStopped);
       slaveTask.TaskFailed += new EventHandler<EventArgs<Guid, Exception>>(slaveTask_TaskFailed);
-      slaveTask.ExceptionOccured += new EventHandler<EventArgs<Guid, Exception>>(slaveTask_ExceptionOccured);
     }
 
@@ -193 +192 @@
       slaveTask.TaskStopped -= new EventHandler<EventArgs<Guid>>(slaveTask_TaskStopped);
       slaveTask.TaskFailed -= new EventHandler<EventArgs<Guid, Exception>>(slaveTask_TaskFailed);
-      slaveTask.ExceptionOccured -= new EventHandler<EventArgs<Guid, Exception>>(slaveTask_ExceptionOccured);
     }
 
@@ -220 +218 @@
       try {
         taskData = slaveTask.GetTaskData();
-        if (taskData == null) throw new SerializationException();
         SlaveStatusInfo.IncrementTasksFinished();
         OnTaskPaused(slaveTask, taskData);
@@ -243 +240 @@
       try {
         taskData = slaveTask.GetTaskData();
-        if (taskData == null) throw new SerializationException();
         SlaveStatusInfo.IncrementTasksFinished();
         OnTaskStopped(slaveTask, taskData);
@@ -266 +262 @@
       try {
         taskData = slaveTask.GetTaskData();
-        if (taskData == null) throw new SerializationException();
       }
       catch { /* taskData will be null */ }
       SlaveStatusInfo.IncrementTasksFailed();
       OnTaskFailed(slaveTask, taskData, e.Value2);
-    }
-
-    private void slaveTask_ExceptionOccured(object sender, EventArgs<Guid, Exception> e) {
-      SlaveTask slaveTask;
-      slaveTasksLocker.EnterUpgradeableReadLock();
-      try {
-        slaveTask = slaveTasks[e.Value];
-        RemoveSlaveTask(e.Value, slaveTask);
-      }
-      finally { slaveTasksLocker.ExitUpgradeableReadLock(); }
-
-      SlaveStatusInfo.IncrementExceptionOccured();
-      OnExceptionOccured(slaveTask, e.Value2);
     }
     #endregion
@@ -312 +294 @@
     }
 
-    public event EventHandler<EventArgs<SlaveTask, Exception>> ExceptionOccured;
-    private void OnExceptionOccured(SlaveTask slaveTask, Exception exception) {
-      var handler = ExceptionOccured;
-      if (handler != null) handler(this, new EventArgs<SlaveTask, Exception>(slaveTask, exception));
-    }
-
     public event EventHandler<EventArgs<SlaveTask>> TaskAborted;
     private void OnTaskAborted(SlaveTask slaveTask) {

trunk/sources/HeuristicLab.Clients.Hive.Slave/3.3/SlaveStatusInfo.cs

@@ -31 +31 @@
     private static int tasksAborted;  // server sent abort
     private static int tasksFailed;   // tasks that failed in the sandbox
-    private static int exceptionsOccured; // number jobs failed caused by the business logic, not a faulted task
     private static int usedCores;    // number of cores currently used
 
@@ -60 +59 @@
     }
 
-    public static int ExceptionsOccured {
-      get { return exceptionsOccured; }
-    }
-
     public static void IncrementTasksStarted() {
       Interlocked.Increment(ref tasksStarted);
@@ -84 +79 @@
     }
 
-    public static void IncrementExceptionOccured() {
-      Interlocked.Increment(ref exceptionsOccured);
-    }
-
     public static void IncrementUsedCores(int val) {
       Interlocked.Add(ref usedCores, val);

trunk/sources/HeuristicLab.Clients.Hive.Slave/3.3/SlaveTask.cs

@@ -43 +43 @@
     public Guid TaskId { get; private set; }
     public bool IsPrepared { get; private set; }
+    private TaskData originalTaskData;
 
     private int coresNeeded;
@@ -74 +75 @@
       try {
         this.TaskId = task.Id;
+        originalTaskData = taskData;
         Prepare(task);
         StartTaskInAppDomain(taskData);
@@ -143 +145 @@
         while (repeat > 0) {
           try {
-            waitForStartBeforeKillSem.WaitOne();
+            waitForStartBeforeKillSem.WaitOne(Settings.Default.ExecutorSemTimeouts);
             AppDomain.Unload(appDomain);
             waitForStartBeforeKillSem.Dispose();
@@ -164 +166 @@
     private void AppDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e) {
       DisposeAppDomain();
-      OnExceptionOccured(new Exception("Unhandled exception: " + e.ExceptionObject.ToString()));
+      OnTaskFailed(new Exception("Unhandled exception: " + e.ExceptionObject.ToString()));
     }
 
@@ -171 +173 @@
       try {
         data = executor.GetTaskData();
+        //this means that there was a problem executing the task
+        if (data == null) return originalTaskData;
       }
       catch (Exception ex) {
@@ -233 +237 @@
             executorMonitoringRun = false;
             break;
-
-          case ExecutorMessageType.ExceptionOccured:
-            executorMonitoringRun = false;
-            DisposeAppDomain();
-            if (executor.CurrentException != null) {
-              OnExceptionOccured(executor.CurrentException);
-            } else {
-              OnExceptionOccured(new Exception(string.Format("Unknow exception occured in Executor for task {0}", TaskId)));
-            }
-            break;
         }
       }
@@ -277 +271 @@
       if (handler != null) handler(this, new EventArgs<Guid, Exception>(this.TaskId, exception));
     }
-
-    public event EventHandler<EventArgs<Guid, Exception>> ExceptionOccured;
-    private void OnExceptionOccured(Exception exception) {
-      var handler = ExceptionOccured;
-      if (handler != null) handler(this, new EventArgs<Guid, Exception>(this.TaskId, exception));
-    }
   }
 }

trunk/sources/HeuristicLab.Clients.Hive.Slave/3.3/StatusCommons.cs

@@ -51 +51 @@
     public int JobsFailed { get; set; }
     [DataMember]
-    public int ExceptionsOccured { get; set; }
-    [DataMember]
     public List<TaskStatus> Jobs { get; set; }
     [DataMember]
@@ -58 +56 @@
 
     public override string ToString() {
-      return string.Format("Status: {0}, Fetc,Start,Fin,Abor,Fail,Exc: {1},{2},{3},{4},{5},{6}",
-        Status, JobsFetched, JobsStarted, JobsFinished, JobsAborted, JobsFailed, ExceptionsOccured);
+      return string.Format("Status: {0}, Fetc,Start,Fin,Abor,Fail,Exc: {1},{2},{3},{4},{5}",
+        Status, JobsFetched, JobsStarted, JobsFinished, JobsAborted, JobsFailed);
     }
   }