Changeset 17825 for branches/3040_VectorBasedGP/HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs

Timestamp:

01/27/21 14:10:56 (4 years ago)

Author:

pfleck

Message:

#3040 Merged trunk into branch.

Location:

branches/3040_VectorBasedGP

Files:

• . (modified) (1 prop)
• HeuristicLab.Services.Hive (modified) (1 prop)
• HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs (modified) (3 diffs)

branches/3040_VectorBasedGP/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();
-      });
     }
 
@@ -154 +120 @@
       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()
+         where taskIds.Contains(task.TaskId)
+         let lastStateLog = task.StateLogs.OrderByDescending(x => x.DateTime).FirstOrDefault(x => x.State == DA.TaskState.Transferring)
+         select new {
+           TaskId = task.TaskId,
+           JobId = task.JobId,
+           State = task.State,
+           Command = task.Command,
+           SlaveId = lastStateLog != null ? lastStateLog.SlaveId : Guid.Empty
+         }).ToList()
       );