Changeset 15004 for trunk

Timestamp:

05/30/17 18:55:38 (7 years ago)

Author:

jkarder

Message:

#2791: improved checkpointing (task is paused and sent back to the server, new one is assigned via next heartbeat)

Location:

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

Files:

• Executor.cs (modified) (1 diff)
• Manager/TaskManager.cs (modified) (16 diffs)
• SlaveTask.cs (modified) (1 diff)

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

@@ -185 +185 @@
     #endregion
 
-    public Tuple<TaskData, DateTime> GetTaskDataSnapshot() {
-      if (taskDataInvalid) return null;
-
-      Tuple<TaskData, DateTime> snapshot = null;
-      if (task == null) {
-        if (CurrentException == null) {
-          CurrentException = new Exception("Task with id " + this.TaskId + " is null, sending empty task");
-        }
-      } else {
-        var taskData = new TaskData();
-
-        var pausedTrigger = new EventWaitHandle(false, EventResetMode.ManualReset);
-        EventHandler pausedHandler = null;
-        pausedHandler = (s, e) => {
-          task.TaskPaused -= pausedHandler;
-          task.TaskPaused += Task_TaskPaused;
-          pausedTrigger.Set();
-        };
-
-        task.TaskPaused -= Task_TaskPaused;
-        task.TaskPaused += pausedHandler;
-        task.Pause();
-        pausedTrigger.WaitOne();
-
-        taskData.Data = PersistenceUtil.Serialize(task);
-        var timestamp = DateTime.Now;
-
-        EventHandler startedHandler = null;
-        startedHandler = (s, e) => {
-          task.TaskStarted -= startedHandler;
-          task.TaskStarted += Task_TaskStarted;
-        };
-
-        task.TaskStarted -= Task_TaskStarted;
-        task.TaskStarted += startedHandler;
-        task.Start();
-
-        taskData.TaskId = TaskId;
-        snapshot = Tuple.Create(taskData, timestamp);
-      }
-
-      return snapshot;
-    }
-
     public TaskData GetTaskData() {
       if (taskDataInvalid) return null;

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

@@ -37 +37 @@
   public class TaskManager {
     private static readonly ReaderWriterLockSlim slaveTasksLocker = new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);
-    private readonly Dictionary<Guid, SnapshotInfo> slaveTasks;
+    private readonly Dictionary<Guid, Tuple<SlaveTask, DateTime>> slaveTasks;
     private readonly ILog log;
     private readonly PluginManager pluginManager;
@@ -43 +43 @@
     private readonly CancellationToken ct;
     private readonly AutoResetEvent waitHandle;
-    private readonly WcfService wcfService;
     private readonly TimeSpan checkpointInterval;
     private readonly TimeSpan checkpointCheckInterval;
@@ -70 +69 @@
       this.pluginManager = pluginCache;
       this.log = log;
-      this.slaveTasks = new Dictionary<Guid, SnapshotInfo>();
+      this.slaveTasks = new Dictionary<Guid, Tuple<SlaveTask, DateTime>>();
 
       cts = new CancellationTokenSource();
       ct = cts.Token;
       waitHandle = new AutoResetEvent(true);
-      wcfService = WcfService.Instance;
       checkpointInterval = Settings.Default.CheckpointInterval;
       checkpointCheckInterval = Settings.Default.CheckpointCheckInterval;
@@ -85 +83 @@
     private void Checkpointing() {
       while (!ct.IsCancellationRequested) {
-        slaveTasksLocker.EnterWriteLock();
+        slaveTasksLocker.EnterUpgradeableReadLock();
         try {
           foreach (var entry in slaveTasks) {
-            var taskId = entry.Key;
-            var snapshotInfo = entry.Value;
-
-            if (DateTime.Now - snapshotInfo.LastSnapshot <= checkpointInterval) continue;
-
-            var task = wcfService.GetTask(taskId);
-            var snapshot = snapshotInfo.Task.GetTaskDataSnapshot();
-
-            if (snapshot == null) continue;
-
-            slaveTasks[taskId].LastSnapshot = snapshot.Item2;
-            var slaveId = ConfigManager.Instance.GetClientInfo().Id;
-            wcfService.UpdateTaskData(task, snapshot.Item1, slaveId, TaskState.Calculating);
+            if (DateTime.Now - entry.Value.Item2 > checkpointInterval)
+              PauseTaskAsync(entry.Key);
           }
-        } finally { slaveTasksLocker.ExitWriteLock(); }
+        } finally { slaveTasksLocker.ExitUpgradeableReadLock(); }
         waitHandle.WaitOne(checkpointCheckInterval);
       }
@@ -146 +133 @@
       try {
         if (!slaveTasks.ContainsKey(taskId)) throw new TaskNotRunningException(taskId);
-        SlaveTask slaveTask = slaveTasks[taskId].Task;
+        SlaveTask slaveTask = slaveTasks[taskId].Item1;
         slaveTask.PauseTask();
       }
@@ -156 +143 @@
       try {
         if (!slaveTasks.ContainsKey(taskId)) throw new TaskNotRunningException(taskId);
-        SlaveTask slaveTask = slaveTasks[taskId].Task;
+        SlaveTask slaveTask = slaveTasks[taskId].Item1;
         slaveTask.StopTask();
       }
@@ -167 +154 @@
       try {
         if (!slaveTasks.ContainsKey(taskId)) throw new TaskNotRunningException(taskId);
-        slaveTask = slaveTasks[taskId].Task;
+        slaveTask = slaveTasks[taskId].Item1;
         if (!slaveTask.IsPrepared) throw new AppDomainNotCreatedException();
         RemoveSlaveTask(taskId, slaveTask);
@@ -181 +168 @@
       try {
         foreach (var slaveTask in slaveTasks.Values) {
-          slaveTask.Task.PauseTask();
+          slaveTask.Item1.PauseTask();
         }
       }
@@ -191 +178 @@
       try {
         foreach (var slaveTask in slaveTasks.Values) {
-          slaveTask.Task.StopTask();
+          slaveTask.Item1.StopTask();
         }
       }
@@ -201 +188 @@
       try {
         foreach (var slaveTask in slaveTasks.Values.ToArray()) {
-          AbortTask(slaveTask.Task.TaskId);
+          AbortTask(slaveTask.Item1.TaskId);
         }
       }
@@ -212 +199 @@
       slaveTasksLocker.EnterWriteLock();
       try {
-        slaveTasks.Add(task.Id, new SnapshotInfo { Task = slaveTask, LastSnapshot = task.DateCreated.GetValueOrDefault() });
+        slaveTasks.Add(task.Id, Tuple.Create(slaveTask, DateTime.Now));
         RegisterSlaveTaskEvents(slaveTask);
       }
@@ -247 +234 @@
       slaveTasksLocker.EnterUpgradeableReadLock();
       try {
-        slaveTask = slaveTasks[e.Value].Task;
+        slaveTask = slaveTasks[e.Value].Item1;
       }
       finally { slaveTasksLocker.ExitUpgradeableReadLock(); }
@@ -259 +246 @@
       slaveTasksLocker.EnterUpgradeableReadLock();
       try {
-        slaveTask = slaveTasks[e.Value].Task;
+        slaveTask = slaveTasks[e.Value].Item1;
         RemoveSlaveTask(e.Value, slaveTask);
       }
@@ -281 +268 @@
       slaveTasksLocker.EnterUpgradeableReadLock();
       try {
-        slaveTask = slaveTasks[e.Value].Task;
+        slaveTask = slaveTasks[e.Value].Item1;
         RemoveSlaveTask(e.Value, slaveTask);
       }
@@ -303 +290 @@
       slaveTasksLocker.EnterUpgradeableReadLock();
       try {
-        slaveTask = slaveTasks[e.Value].Task;
+        slaveTask = slaveTasks[e.Value].Item1;
         RemoveSlaveTask(e.Value, slaveTask);
       }
@@ -353 +340 @@
       slaveTasksLocker.EnterReadLock();
       try {
-        return slaveTasks.ToDictionary(x => x.Key, x => x.Value.Task.ExecutionTime);
+        return slaveTasks.ToDictionary(x => x.Key, x => x.Value.Item1.ExecutionTime);
       }
       finally { slaveTasksLocker.ExitReadLock(); }
-    }
-
-    private sealed class SnapshotInfo {
-      public SlaveTask Task { get; set; }
-      public DateTime LastSnapshot { get; set; }
     }
   }

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

@@ -165 +165 @@
     }
 
-    public Tuple<TaskData, DateTime> GetTaskDataSnapshot() {
-      Tuple<TaskData, DateTime> snapshot = null;
-      try {
-        snapshot = executor.GetTaskDataSnapshot();
-        if (snapshot == null) return Tuple.Create(originalTaskData, DateTime.Now);
-      }
-      catch (Exception ex) {
-        EventLogManager.LogException(ex);
-      }
-      return snapshot;
-    }
-
     public TaskData GetTaskData() {
       TaskData data = null;