Changeset 15630 for branches/HiveProjectManagement/HeuristicLab.Services.Hive

Timestamp:

01/18/18 15:08:25 (7 years ago)

Author:

jzenisek

Message:

#2839

• updated Heartbeat processing (regarding: checking against AssignedJobResources and handling of the updated Job deletion routine)
• updated Job deletion routine(still in progress at GenerateStatistics)

Location:

branches/HiveProjectManagement/HeuristicLab.Services.Hive/3.3

Files:

• Converter.cs (modified) (4 diffs)
• DataTransfer/Job.cs (modified) (1 diff)
• DataTransfer/JobState.cs (added)
• HeuristicLab.Services.Hive-3.3.csproj (modified) (1 diff)
• HiveService.cs (modified) (2 diffs)
• Manager/EventManager.cs (modified) (2 diffs)
• Manager/HeartbeatManager.cs (modified) (1 diff)
• ServiceContracts/IHiveService.cs (modified) (1 diff)

branches/HiveProjectManagement/HeuristicLab.Services.Hive/3.3/Converter.cs

@@ -121 +121 @@
         OwnerUserId = source.OwnerUserId,
         DateCreated = source.DateCreated,
-        ProjectId = source.ProjectId
+        ProjectId = source.ProjectId,
+        State = source.State.ToDto()
       };
     }
@@ -140 +141 @@
       target.DateCreated = source.DateCreated;
       target.ProjectId = source.ProjectId;
+      target.State = source.State.ToEntity();
     }
     #endregion
@@ -242 +244 @@
     #endregion
 
-    #region State
+    #region TaskState
     public static DT.TaskState ToDto(this DA.TaskState source) {
       switch (source) {
@@ -268 +270 @@
         case DT.TaskState.Waiting: return DA.TaskState.Waiting;
         default: return DA.TaskState.Failed;
+      }
+    }
+    #endregion
+
+    #region JobState
+    public static DT.JobState ToDto(this DA.JobState source) {
+      switch (source) {
+        case DA.JobState.Online: return DT.JobState.Online;
+        case DA.JobState.StatisticsPending: return DT.JobState.StatisticsPending;
+        case DA.JobState.DeletionPending: return DT.JobState.DeletionPending;
+        default: return DT.JobState.Online;
+      }
+    }
+
+    public static DA.JobState ToEntity(this DT.JobState source) {
+      switch (source) {
+        case DT.JobState.Online: return DA.JobState.Online;
+        case DT.JobState.StatisticsPending: return DA.JobState.StatisticsPending;
+        case DT.JobState.DeletionPending: return DA.JobState.DeletionPending;
+        default: return DA.JobState.Online;
       }
     }

branches/HiveProjectManagement/HeuristicLab.Services.Hive/3.3/DataTransfer/Job.cs

@@ -37 +37 @@
     [DataMember]
     public string OwnerUsername { get; set; }
+    [DataMember]
+    public JobState State { get; set; }
 
     /* ==== some computed statistics ==== */

branches/HiveProjectManagement/HeuristicLab.Services.Hive/3.3/HeuristicLab.Services.Hive-3.3.csproj

@@ -122 +122 @@
     <Compile Include="DataTransfer\AssignedProjectResource.cs" />
     <Compile Include="DataTransfer\Command.cs" />
+    <Compile Include="DataTransfer\JobState.cs" />
     <Compile Include="DataTransfer\Project.cs" />
     <Compile Include="DataTransfer\ProjectPermission.cs" />

branches/HiveProjectManagement/HeuristicLab.Services.Hive/3.3/HiveService.cs

@@ -361 +361 @@
         return pm.UseTransaction(() => {
           var jobs = jobDao.GetAll()
-            .Where(x => x.OwnerUserId == currentUserId
-                     || x.JobPermissions.Count(y => y.Permission != DA.Permission.NotAllowed
-                                                 && y.GrantedUserId == currentUserId) > 0)
+            .Where(x => x.State == DA.JobState.Online
+                          && (x.OwnerUserId == currentUserId
+                            || x.JobPermissions.Count(y => y.Permission != DA.Permission.NotAllowed 
+                              && y.GrantedUserId == currentUserId) > 0)
+                          )
             .Select(x => x.ToDto())
             .ToList();
@@ -524 +526 @@
     }
 
-    public void DeleteJob(Guid jobId) {
+    public void UpdateJobState(Guid jobId, DT.JobState jobState) {
       RoleVerifier.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client);
       AuthorizationManager.AuthorizeForJob(jobId, DT.Permission.Full);
       var pm = PersistenceManager;
-      using (new PerformanceLogger("DeleteJob")) {
+      using (new PerformanceLogger("UpdateJobState")) {
         var jobDao = pm.JobDao;
         pm.UseTransaction(() => {
-          // not necessary anymore: child tasks will be deleted by db-trigger
-          // now: tasks are deleted by jobId
-          // entries in AssignedJobResource will be deleted by foreign key clause CASCADE ON DELETE
-          jobDao.Delete(jobId);
-          pm.SubmitChanges();
+          var job = jobDao.GetById(jobId);
+          if(job != null) {
+            var jobStateEntity = jobState.ToEntity();
+            // note: allow solely state changes from "Online" to "StatisticsPending"
+            // and from "StatisticsPending" to "DeletionPending"
+            if (job.State == DA.JobState.Online && jobStateEntity == DA.JobState.StatisticsPending) {
+              job.State = jobStateEntity;
+              foreach(var task in job.Tasks
+              .Where(x => x.State == DA.TaskState.Waiting 
+                || x.State == DA.TaskState.Paused 
+                || x.State == DA.TaskState.Offline)) {
+                task.State = DA.TaskState.Aborted;
+              }
+              pm.SubmitChanges();
+            } else if(job.State == DA.JobState.StatisticsPending && jobStateEntity == DA.JobState.DeletionPending) {
+              job.State = jobStateEntity;
+              pm.SubmitChanges();
+            } 
+          }
         });
       }

branches/HiveProjectManagement/HeuristicLab.Services.Hive/3.3/Manager/EventManager.cs

@@ -34 +34 @@
     public void Cleanup() {
       var pm = PersistenceManager;
+
+      pm.UseTransaction(() => {
+        FinishJobDeletion(pm);
+        pm.SubmitChanges();
+      });
+
       pm.UseTransaction(() => {
         SetTimeoutSlavesOffline(pm);
@@ -45 +51 @@
         pm.SubmitChanges();
       });
+    }
+
+    /// <summary>
+    /// Deletes all jobs which are in state "DeletionPending" (this will include all corresponding tasks).
+    /// The state "DeletionPending" is set by HiveJanitor > StatisticsGenerator
+    /// </summary>
+    private void FinishJobDeletion(IPersistenceManager pm) {
+      var jobDao = pm.JobDao;
+      jobDao.DeleteByState(JobState.DeletionPending);
     }
 

branches/HiveProjectManagement/HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs

@@ -141 +141 @@
     /// </summary>
     private IEnumerable<MessageContainer> UpdateTasks(IPersistenceManager pm, Heartbeat heartbeat, bool isAllowedToCalculate) {
+      var taskDao = pm.TaskDao;
+      var jobDao = pm.JobDao;
+      var assignedJobResourceDao = pm.AssignedJobResourceDao;
+      var actions = new List<MessageContainer>();
+      if (heartbeat.JobProgress == null || !heartbeat.JobProgress.Any())
+        return actions;
+
+      var jobIdsWithStatisticsPending = jobDao.GetJobIdsByState(DA.JobState.StatisticsPending).ToList();
+
+      // select all tasks and statelogs with one query
+      var taskIds = heartbeat.JobProgress.Select(x => x.Key).ToList();
+      var taskInfos = pm.UseTransaction(() =>
+        (from task in taskDao.GetAll()
+          where taskIds.Contains(task.TaskId)
+          let lastStateLog = task.StateLogs.OrderByDescending(x => x.DateTime).FirstOrDefault()
+          select new {
+            TaskId = task.TaskId,
+            JobId = task.JobId,
+            State = task.State,
+            Command = task.Command,
+            SlaveId = lastStateLog != null ? lastStateLog.SlaveId : default(Guid)
+          }).ToList()
+      );
+
+      // process the jobProgresses
+      foreach (var jobProgress in heartbeat.JobProgress) {
+        var progress = jobProgress;
+        var curTask = taskInfos.SingleOrDefault(x => x.TaskId == progress.Key);
+        if (curTask == null) {
+          actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, progress.Key));
+          LogFactory.GetLogger(this.GetType().Namespace).Log("Task on slave " + heartbeat.SlaveId + " does not exist in DB: " + jobProgress.Key);
+        } else if (jobIdsWithStatisticsPending.Contains(curTask.JobId)) {
+          // parenting job of current task has been requested for deletion (indicated by job state "Statistics Pending")
+          // update task execution time
+          pm.UseTransaction(() => {
+            taskDao.UpdateExecutionTime(curTask.TaskId, progress.Value.TotalMilliseconds);
+          });
+          actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.TaskId));
+          LogFactory.GetLogger(this.GetType().Namespace).Log("Abort task " + curTask.TaskId + " on slave " + heartbeat.SlaveId + ". The parenting job " + curTask.JobId + " was requested to be deleted.");
+        } else if (curTask.SlaveId == Guid.Empty || curTask.SlaveId != heartbeat.SlaveId) {
+          // assigned slave does not match heartbeat
+          actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.TaskId));
+          LogFactory.GetLogger(this.GetType().Namespace).Log("The slave " + heartbeat.SlaveId + " is not supposed to calculate task: " + curTask.TaskId);
+        } else if (!isAllowedToCalculate) {
+          actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.TaskId));
+          LogFactory.GetLogger(this.GetType().Namespace).Log("The slave " + heartbeat.SlaveId + " is not allowed to calculate any tasks tue to a downtime. The task is paused.");
+        } else if (assignedJobResourceDao.CheckJobGrantedForResource(curTask.JobId, heartbeat.SlaveId)) { 
+          // assigned resources ids of task do not match with slaveId (and parent resourceGroupIds); this might happen when slave is moved to different group
+          actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.TaskId));
+          LogFactory.GetLogger(this.GetType().Namespace).Log("The slave " + heartbeat.SlaveId + " is not granted to calculate task: " + curTask.TaskId);
+        } else {
+          // update task execution time
+          pm.UseTransaction(() => {
+            taskDao.UpdateExecutionTime(curTask.TaskId, progress.Value.TotalMilliseconds);
+          });
+          switch (curTask.Command) {
+            case DA.Command.Stop:
+              actions.Add(new MessageContainer(MessageContainer.MessageType.StopTask, curTask.TaskId));
+              break;
+            case DA.Command.Pause:
+              actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.TaskId));
+              break;
+            case DA.Command.Abort:
+              actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.TaskId));
+              break;
+          }
+        }
+        
+      } 
+      return actions;
+    }
+
+    /// <summary>
+    /// Update the progress of each task
+    /// Checks if all the task sent by heartbeat are supposed to be calculated by this slave
+    /// </summary>
+    private IEnumerable<MessageContainer> UpdateTasks_Old(IPersistenceManager pm, Heartbeat heartbeat, bool isAllowedToCalculate) {
       var taskDao = pm.TaskDao;
       var assignedResourceDao = pm.AssignedTaskResourceDao;

branches/HiveProjectManagement/HeuristicLab.Services.Hive/3.3/ServiceContracts/IHiveService.cs

@@ -93 +93 @@
 
     [OperationContract]
-    void DeleteJob(Guid JobId);
+    void UpdateJobState(Guid JobId, JobState jobState);
 
     [OperationContract]