Changeset 12920 for trunk

Timestamp:

08/27/15 14:52:42 (9 years ago)

Author:

jkarder

Message:

#2468: implemented checkpointing

Location:

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

Files:

• Core.cs (modified) (1 diff)
• Executor.cs (modified) (1 diff)
• Manager/TaskManager.cs (modified) (15 diffs)
• Properties/Settings.Designer.cs (modified) (1 diff)
• Properties/Settings.settings (modified) (1 diff)
• SlaveTask.cs (modified) (1 diff)
• app.config (modified) (1 diff)

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

@@ -481 +481 @@
       SlaveClientCom.Instance.LogMessage("Stopping heartbeat");
       heartbeatManager.StopHeartBeat();
+      SlaveClientCom.Instance.LogMessage("Stopping checkpointing");
+      taskManager.StopCheckpointing();
+
       abortRequested = true;
 

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

@@ -24 +24 @@
 using System.Linq;
 using System.Threading;
+using HeuristicLab.Clients.Hive.SlaveCore.Properties;
 using HeuristicLab.Common;
 using HeuristicLab.Core;
@@ -32 +33 @@
   /// Holds a list of slave tasks and manages access to this list. 
   /// Forwards events from SlaveTask and forwards commands to SlaveTask. 
+  /// Periodically sends task data to the server to avoid loss of progress when the slave crashes.
   /// </summary>
   public class TaskManager {
-    private static ReaderWriterLockSlim slaveTasksLocker = new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);
-    private Dictionary<Guid, SlaveTask> slaveTasks;
-    private ILog log;
-    private PluginManager pluginManager;
+    private static readonly ReaderWriterLockSlim slaveTasksLocker = new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);
+    private readonly Dictionary<Guid, SnapshotInfo> slaveTasks;
+    private readonly ILog log;
+    private readonly PluginManager pluginManager;
+    private readonly CancellationTokenSource cts;
+    private readonly CancellationToken ct;
+    private readonly AutoResetEvent waitHandle;
+    private readonly WcfService wcfService;
+    private readonly TimeSpan checkpointInterval;
+    private readonly TimeSpan checkpointCheckInterval;
 
     public int TaskCount {
@@ -62 +70 @@
       this.pluginManager = pluginCache;
       this.log = log;
-      this.slaveTasks = new Dictionary<Guid, SlaveTask>();
-    }
+      this.slaveTasks = new Dictionary<Guid, SnapshotInfo>();
+
+      cts = new CancellationTokenSource();
+      ct = cts.Token;
+      waitHandle = new AutoResetEvent(true);
+      wcfService = WcfService.Instance;
+      checkpointInterval = Settings.Default.CheckpointInterval;
+      checkpointCheckInterval = Settings.Default.CheckpointCheckInterval;
+
+      System.Threading.Tasks.Task.Factory.StartNew(Checkpointing, ct);
+    }
+
+    #region Checkpointing
+    private void Checkpointing() {
+      while (!ct.IsCancellationRequested) {
+        slaveTasksLocker.EnterWriteLock();
+        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);
+          }
+        } finally { slaveTasksLocker.ExitWriteLock(); }
+        waitHandle.WaitOne(checkpointCheckInterval);
+      }
+    }
+
+    public void StopCheckpointing() {
+      cts.Cancel();
+      waitHandle.Set();
+      waitHandle.Close();
+    }
+    #endregion
 
     #region Task Control methods
@@ -97 +146 @@
       try {
         if (!slaveTasks.ContainsKey(taskId)) throw new TaskNotRunningException(taskId);
-        SlaveTask slaveTask = slaveTasks[taskId];
+        SlaveTask slaveTask = slaveTasks[taskId].Task;
         slaveTask.PauseTask();
       }
@@ -107 +156 @@
       try {
         if (!slaveTasks.ContainsKey(taskId)) throw new TaskNotRunningException(taskId);
-        SlaveTask slaveTask = slaveTasks[taskId];
+        SlaveTask slaveTask = slaveTasks[taskId].Task;
         slaveTask.StopTask();
       }
@@ -118 +167 @@
       try {
         if (!slaveTasks.ContainsKey(taskId)) throw new TaskNotRunningException(taskId);
-        slaveTask = slaveTasks[taskId];
+        slaveTask = slaveTasks[taskId].Task;
         if (!slaveTask.IsPrepared) throw new AppDomainNotCreatedException();
         RemoveSlaveTask(taskId, slaveTask);
@@ -132 +181 @@
       try {
         foreach (var slaveTask in slaveTasks.Values) {
-          slaveTask.PauseTask();
+          slaveTask.Task.PauseTask();
         }
       }
@@ -142 +191 @@
       try {
         foreach (var slaveTask in slaveTasks.Values) {
-          slaveTask.StopTask();
+          slaveTask.Task.StopTask();
         }
       }
@@ -152 +201 @@
       try {
         foreach (var slaveTask in slaveTasks.Values.ToArray()) {
-          AbortTask(slaveTask.TaskId);
+          AbortTask(slaveTask.Task.TaskId);
         }
       }
@@ -163 +212 @@
       slaveTasksLocker.EnterWriteLock();
       try {
-        slaveTasks.Add(task.Id, slaveTask);
+        slaveTasks.Add(task.Id, new SnapshotInfo { Task = slaveTask, LastSnapshot = task.DateCreated.GetValueOrDefault() });
         RegisterSlaveTaskEvents(slaveTask);
       }
@@ -198 +247 @@
       slaveTasksLocker.EnterUpgradeableReadLock();
       try {
-        slaveTask = slaveTasks[e.Value];
+        slaveTask = slaveTasks[e.Value].Task;
       }
       finally { slaveTasksLocker.ExitUpgradeableReadLock(); }
@@ -210 +259 @@
       slaveTasksLocker.EnterUpgradeableReadLock();
       try {
-        slaveTask = slaveTasks[e.Value];
+        slaveTask = slaveTasks[e.Value].Task;
         RemoveSlaveTask(e.Value, slaveTask);
       }
@@ -232 +281 @@
       slaveTasksLocker.EnterUpgradeableReadLock();
       try {
-        slaveTask = slaveTasks[e.Value];
+        slaveTask = slaveTasks[e.Value].Task;
         RemoveSlaveTask(e.Value, slaveTask);
       }
@@ -254 +303 @@
       slaveTasksLocker.EnterUpgradeableReadLock();
       try {
-        slaveTask = slaveTasks[e.Value];
+        slaveTask = slaveTasks[e.Value].Task;
         RemoveSlaveTask(e.Value, slaveTask);
       }
@@ -304 +353 @@
       slaveTasksLocker.EnterReadLock();
       try {
-        return slaveTasks.ToDictionary(x => x.Key, x => x.Value.ExecutionTime);
+        return slaveTasks.ToDictionary(x => x.Key, x => x.Value.Task.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/Properties/Settings.Designer.cs

@@ -299 +299 @@
             }
         }
+        
+        [global::System.Configuration.UserScopedSettingAttribute()]
+        [global::System.Diagnostics.DebuggerNonUserCodeAttribute()]
+        [global::System.Configuration.DefaultSettingValueAttribute("10:00:00")]
+        public global::System.TimeSpan CheckpointInterval {
+            get {
+                return ((global::System.TimeSpan)(this["CheckpointInterval"]));
+            }
+            set {
+                this["CheckpointInterval"] = value;
+            }
+        }
+        
+        [global::System.Configuration.UserScopedSettingAttribute()]
+        [global::System.Diagnostics.DebuggerNonUserCodeAttribute()]
+        [global::System.Configuration.DefaultSettingValueAttribute("00:05:00")]
+        public global::System.TimeSpan CheckpointCheckInterval {
+            get {
+                return ((global::System.TimeSpan)(this["CheckpointCheckInterval"]));
+            }
+            set {
+                this["CheckpointCheckInterval"] = value;
+            }
+        }
     }
 }

trunk/sources/HeuristicLab.Clients.Hive.Slave/3.3/Properties/Settings.settings

@@ -72 +72 @@
       <Value Profile="(Default)">00:00:40</Value>
     </Setting>
+    <Setting Name="CheckpointInterval" Type="System.TimeSpan" Scope="User">
+      <Value Profile="(Default)">10:00:00</Value>
+    </Setting>
+    <Setting Name="CheckpointCheckInterval" Type="System.TimeSpan" Scope="User">
+      <Value Profile="(Default)">00:05:00</Value>
+    </Setting>
   </Settings>
 </SettingsFile>

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

@@ -169 +169 @@
     }
 
+    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;

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

@@ -135 +135 @@
         <value>00:00:40</value>
       </setting>
+      <setting name="CheckpointInterval" serializeAs="String">
+        <value>10:00:00</value>
+      </setting>
+      <setting name="CheckpointCheckInterval" serializeAs="String">
+        <value>00:05:00</value>
+      </setting>
     </HeuristicLab.Clients.Hive.SlaveCore.Properties.Settings>
     <HeuristicLab.Clients.Hive.Settings>