Changeset 17574 for trunk/HeuristicLab.Services.Hive/3.3/Manager

Timestamp:

05/29/20 13:28:25 (4 years ago)

Author:

jkarder

Message:

#3062: overhauled statistics generation and cleanup

• switched to a single thread for database cleanup and statistics generation (executed sequentially)
• switched to preemptive deletion of items that are in status DeletionPending (for jobs: statelogs, taskdata, tasks)
• added code that aborts tasks whose jobs have already been marked for deletion
• added method UseTransactionAndSubmit in addition to UseTransaction in PersistenceManager
• updated DAO methods and introduced more bare metal sql
• introduced DAO methods for batch deletion
• fixed usage of enum values in DAO sql queries
• deleted unnecessary triggers tr_JobDeleteCascade and tr_TaskDeleteCascade in Prepare Hive Database.sql
• changed scheduling for less interference with janitor and other heartbeats
  • increased scheduling patience from 20 to 70 seconds (to wait longer to get the mutex for scheduling)
  • changed signature of ITaskScheduler.Schedule
  • added base class for TaskSchedulers and moved assignment of tasks to slaves into it
  • changed RoundRobinTaskScheduler to use bare metal sql
• made MessageContainer a storable type (leftover)
• updated HiveJanitorServiceInstaller.nsi

Location:

trunk/HeuristicLab.Services.Hive/3.3/Manager

Files:

• EventManager.cs (modified) (3 diffs)
• HeartbeatManager.cs (modified) (2 diffs)

trunk/HeuristicLab.Services.Hive/3.3/Manager/EventManager.cs

@@ -28 +28 @@
   public class EventManager : IEventManager {
     private const string SlaveTimeout = "Slave timed out.";
+    private static readonly TaskState[] CompletedStates = { TaskState.Finished, TaskState.Aborted, TaskState.Failed };
+
     private IPersistenceManager PersistenceManager {
       get { return ServiceLocator.Instance.PersistenceManager; }
@@ -33 +35 @@
 
     public void Cleanup() {
+      Console.WriteLine("started cleanup");
       var pm = PersistenceManager;
 
-      pm.UseTransaction(() => {
-        FinishJobDeletion(pm);
-        pm.SubmitChanges();
-      });
+      // preemptiv delete obsolete entities
+      // speeds up job deletion
+      BatchDelete((p, s) => p.StateLogDao.DeleteObsolete(s), 100, 100, true, pm, "DeleteObsoleteStateLogs");
+      BatchDelete((p, s) => p.TaskDataDao.DeleteObsolete(s), 100, 20, true, pm, "DeleteObsoleteTaskData");
+      BatchDelete((p, s) => p.TaskDao.DeleteObsolete(s), 100, 20, false, pm, "DeleteObsoleteTasks");
+      BatchDelete((p, s) => p.JobDao.DeleteByState(JobState.DeletionPending, s), 100, 20, true, pm, "DeleteObsoleteJobs");
 
-      pm.UseTransaction(() => {
-        SetTimeoutSlavesOffline(pm);
-        SetTimeoutTasksWaiting(pm);
-        DeleteObsoleteSlaves(pm);
-        pm.SubmitChanges();
-      });
+      LogFactory.GetLogger(typeof(HiveJanitor).Namespace).Log("HiveJanitor: SetTimeoutSlavesOffline");
+      Console.WriteLine("5");
+      pm.UseTransactionAndSubmit(() => { SetTimeoutSlavesOffline(pm); });
+      LogFactory.GetLogger(typeof(HiveJanitor).Namespace).Log("HiveJanitor: SetTimeoutTasksWaiting");
+      Console.WriteLine("6");
+      pm.UseTransactionAndSubmit(() => { SetTimeoutTasksWaiting(pm); });
+      LogFactory.GetLogger(typeof(HiveJanitor).Namespace).Log("HiveJanitor: DeleteObsoleteSlaves");
+      Console.WriteLine("7");
+      pm.UseTransactionAndSubmit(() => { DeleteObsoleteSlaves(pm); });
+      LogFactory.GetLogger(typeof(HiveJanitor).Namespace).Log("HiveJanitor: AbortObsoleteTasks");
+      Console.WriteLine("8");
+      pm.UseTransactionAndSubmit(() => { AbortObsoleteTasks(pm); });
+      LogFactory.GetLogger(typeof(HiveJanitor).Namespace).Log("HiveJanitor: FinishParentTasks");
+      Console.WriteLine("9");
+      pm.UseTransactionAndSubmit(() => { FinishParentTasks(pm); });
+      LogFactory.GetLogger(typeof(HiveJanitor).Namespace).Log("HiveJanitor: DONE");
+      Console.WriteLine("10");
+    }
 
-      pm.UseTransaction(() => {
-        FinishParentTasks(pm);
-        pm.SubmitChanges();
-      });
+    private void BatchDelete(
+      Func<IPersistenceManager, int, int> deletionFunc,
+      int batchSize,
+      int maxCalls,
+      bool limitIsBatchSize,
+      IPersistenceManager pm,
+      string logMessage
+    ) {
+      int totalDeleted = 0;
+      while (maxCalls > 0) {
+        maxCalls--;
+        LogFactory.GetLogger(typeof(HiveJanitor).Namespace).Log($"HiveJanitor: {logMessage}");
+        Console.WriteLine($"HiveJanitor: {logMessage}");
+        var deleted = pm.UseTransactionAndSubmit(() => { return deletionFunc(pm, batchSize); });
+        LogFactory.GetLogger(typeof(HiveJanitor).Namespace).Log($"HiveJanitor: {logMessage} DONE (deleted {deleted}, {maxCalls} calls left)");
+        Console.WriteLine($"HiveJanitor: {logMessage} DONE (deleted {deleted}, {maxCalls} calls left)");
+        totalDeleted += deleted;
+        if (limitIsBatchSize && deleted < batchSize || deleted <= 0) return;
+      }
+      LogFactory.GetLogger(typeof(HiveJanitor).Namespace).Log($"HiveJanitor: Possible rows left to delete (total deleted: {totalDeleted}).");
+      Console.WriteLine($"HiveJanitor: Possible rows left to delete (total deleted: {totalDeleted}).");
     }
 
@@ -136 +170 @@
       }
     }
+
+    /// <summary>
+    /// Aborts tasks whose jobs have already been marked for deletion
+    /// </summary>
+    /// <param name="pm"></param>
+    private void AbortObsoleteTasks(IPersistenceManager pm) {
+      var jobDao = pm.JobDao;
+      var taskDao = pm.TaskDao;
+
+      var obsoleteTasks = (from jobId in jobDao.GetJobIdsByState(JobState.StatisticsPending)
+                           join task in taskDao.GetAll() on jobId equals task.JobId
+                           where !CompletedStates.Contains(task.State) && task.Command == null
+                           select task).ToList();
+
+      foreach (var t in obsoleteTasks) {
+        t.State = TaskState.Aborted;
+      }
+    }
   }
 }

trunk/HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs

@@ -83 +83 @@
             mutexAquired = mutex.WaitOne(Properties.Settings.Default.SchedulingPatience);
             if (mutexAquired) {
-              var waitingTasks = pm.UseTransaction(() => taskDao.GetWaitingTasks(slave)
-                  .Select(x => new TaskInfoForScheduler {
-                    TaskId = x.TaskId,
-                    JobId = x.JobId,
-                    Priority = x.Priority
-                  })
-                  .ToList()
-              );
-              var availableTasks = TaskScheduler.Schedule(waitingTasks).ToArray();
-              if (availableTasks.Any()) {
-                var task = availableTasks.First();
-                AssignTask(pm, slave, task.TaskId);
-                actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateTask, task.TaskId));
+              var scheduledTaskIds = TaskScheduler.Schedule(slave, 1).ToArray();
+              foreach (var id in scheduledTaskIds) {
+                actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateTask, id));
               }
             } else {
-              LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling could not be aquired.");
+              LogFactory.GetLogger(this.GetType().Namespace).Log($"HeartbeatManager: The mutex used for scheduling could not be aquired. (HB from Slave {slave.ResourceId})");
             }
-          }
-          catch (AbandonedMutexException) {
-            LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling has been abandoned.");
-          }
-          catch (Exception ex) {
-            LogFactory.GetLogger(this.GetType().Namespace).Log(string.Format("HeartbeatManager threw an exception in ProcessHeartbeat: {0}", ex));
-          }
-          finally {
+          } catch (AbandonedMutexException) {
+            LogFactory.GetLogger(this.GetType().Namespace).Log($"HeartbeatManager: The mutex used for scheduling has been abandoned. (HB from Slave {slave.ResourceId})");
+          } catch (Exception ex) {
+            LogFactory.GetLogger(this.GetType().Namespace).Log($"HeartbeatManager threw an exception in ProcessHeartbeat (HB from Slave {slave.ResourceId}): {ex}");
+          } finally {
             if (mutexAquired) mutex.ReleaseMutex();
           }
@@ -113 +100 @@
       }
       return actions;
-    }
-
-    private void AssignTask(IPersistenceManager pm, DA.Slave slave, Guid taskId) {
-      const DA.TaskState transferring = DA.TaskState.Transferring;
-      DateTime now = DateTime.Now;
-      var taskDao = pm.TaskDao;
-      var stateLogDao = pm.StateLogDao;
-      pm.UseTransaction(() => {
-        var task = taskDao.GetById(taskId);
-        stateLogDao.Save(new DA.StateLog {
-          State = transferring,
-          DateTime = now,
-          TaskId = taskId,
-          SlaveId = slave.ResourceId,
-          UserId = null,
-          Exception = null
-        });
-        task.State = transferring;
-        task.LastHeartbeat = now;
-        pm.SubmitChanges();
-      });
     }