Changeset 12858 for branches/HiveStatistics/sources/HeuristicLab.Services.Hive/3.3/Manager/NewHeartbeatManager.cs

Timestamp:

08/13/15 15:22:51 (9 years ago)

Author:

ascheibe

Message:

#2388

• prevent disposing of hive data context
• more cleanups

File:

• branches/HiveStatistics/sources/HeuristicLab.Services.Hive/3.3/Manager/NewHeartbeatManager.cs (modified) (1 diff)

branches/HiveStatistics/sources/HeuristicLab.Services.Hive/3.3/Manager/NewHeartbeatManager.cs

@@ -47 +47 @@
     public List<MessageContainer> ProcessHeartbeat(Heartbeat heartbeat) {
       List<MessageContainer> actions = new List<MessageContainer>();
-      using (var pm = PersistenceManager) {
-        var slaveDao = pm.SlaveDao;
-        var taskDao = pm.TaskDao;
-        var slave = pm.UseTransaction(() => slaveDao.GetById(heartbeat.SlaveId));
-        if (slave == null) {
-          actions.Add(new MessageContainer(MessageContainer.MessageType.SayHello));
-        } else {
-          if (heartbeat.HbInterval != slave.HbInterval) {
-            actions.Add(new MessageContainer(MessageContainer.MessageType.NewHBInterval));
-          }
-          if (slaveDao.SlaveHasToShutdownComputer(slave.ResourceId)) {
-            actions.Add(new MessageContainer(MessageContainer.MessageType.ShutdownComputer));
-          }
-          // update slave data  
-          slave.FreeCores = heartbeat.FreeCores;
-          slave.FreeMemory = heartbeat.FreeMemory;
-          slave.CpuUtilization = heartbeat.CpuUtilization;
-          slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0)
-            ? DA.SlaveState.Calculating
-            : DA.SlaveState.Idle;
-          slave.LastHeartbeat = DateTime.Now;
-          pm.UseTransaction(() => {
-            slave.IsAllowedToCalculate = slaveDao.SlaveIsAllowedToCalculate(slave.ResourceId);
-            pm.SubmitChanges();
-          });
-
-          // update task data
-          actions.AddRange(UpdateTasks(pm, heartbeat, slave.IsAllowedToCalculate));
-
-          // assign new task
-          if (heartbeat.AssignJob && slave.IsAllowedToCalculate && heartbeat.FreeCores > 0) {
-            bool mutexAquired = false;
-            var mutex = new Mutex(false, MutexName);
-            try {
-              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);
-                if (availableTasks.Any()) {
-                  var task = availableTasks.First();
-                  AssignTask(pm, slave, task.TaskId);
-                  actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateTask, task.TaskId));
-                }
-              } else {
-                LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling could not be aquired.");
+      var pm = PersistenceManager;
+      var slaveDao = pm.SlaveDao;
+      var taskDao = pm.TaskDao;
+      var slave = pm.UseTransaction(() => slaveDao.GetById(heartbeat.SlaveId));
+      if (slave == null) {
+        actions.Add(new MessageContainer(MessageContainer.MessageType.SayHello));
+      } else {
+        if (heartbeat.HbInterval != slave.HbInterval) {
+          actions.Add(new MessageContainer(MessageContainer.MessageType.NewHBInterval));
+        }
+        if (slaveDao.SlaveHasToShutdownComputer(slave.ResourceId)) {
+          actions.Add(new MessageContainer(MessageContainer.MessageType.ShutdownComputer));
+        }
+        // update slave data  
+        slave.FreeCores = heartbeat.FreeCores;
+        slave.FreeMemory = heartbeat.FreeMemory;
+        slave.CpuUtilization = heartbeat.CpuUtilization;
+        slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0)
+          ? DA.SlaveState.Calculating
+          : DA.SlaveState.Idle;
+        slave.LastHeartbeat = DateTime.Now;
+        pm.UseTransaction(() => {
+          slave.IsAllowedToCalculate = slaveDao.SlaveIsAllowedToCalculate(slave.ResourceId);
+          pm.SubmitChanges();
+        });
+
+        // update task data
+        actions.AddRange(UpdateTasks(pm, heartbeat, slave.IsAllowedToCalculate));
+
+        // assign new task
+        if (heartbeat.AssignJob && slave.IsAllowedToCalculate && heartbeat.FreeCores > 0) {
+          bool mutexAquired = false;
+          var mutex = new Mutex(false, MutexName);
+          try {
+            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));
               }
+            } else {
+              LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling could not be aquired.");
             }
-            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 {
-              if (mutexAquired) mutex.ReleaseMutex();
-            }
+          }
+          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 {
+            if (mutexAquired) mutex.ReleaseMutex();
           }
         }