Changeset 9363 for branches/OaaS/HeuristicLab.Services.Hive/3.3/Manager

Timestamp:

04/16/13 13:13:41 (12 years ago)

Author:

spimming

Message:

#1888:

• Merged revisions from trunk

Location:

branches/OaaS

Files:

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

branches/OaaS

@@ -21 +21 @@
 protoc.exe
 _ReSharper.HeuristicLab 3.3 Tests
+Google.ProtocolBuffers-2.4.1.473.dll
 packages

@@ -21 +21 @@
 protoc.exe
 _ReSharper.HeuristicLab 3.3 Tests
+Google.ProtocolBuffers-2.4.1.473.dll
 packages

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

@@ -23 +23 @@
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using HeuristicLab.Services.Hive.DataTransfer;
 using DA = HeuristicLab.Services.Hive.DataAccess;
@@ -28 +29 @@
 namespace HeuristicLab.Services.Hive {
   public class HeartbeatManager {
+    private const string MutexName = "HiveTaskSchedulingMutex";
+
     private IHiveDao dao {
       get { return ServiceLocator.Instance.HiveDao; }
     }
-    private IAuthorizationManager auth {
-      get { return ServiceLocator.Instance.AuthorizationManager; }
+    private ITaskScheduler taskScheduler {
+      get { return ServiceLocator.Instance.TaskScheduler; }
+    }
+    private DataAccess.ITransactionManager trans {
+      get { return ServiceLocator.Instance.TransactionManager; }
     }
 
@@ -41 +47 @@
     public List<MessageContainer> ProcessHeartbeat(Heartbeat heartbeat) {
       List<MessageContainer> actions = new List<MessageContainer>();
-      Slave slave = dao.GetSlave(heartbeat.SlaveId);
+      Slave slave = null;
+      slave = trans.UseTransaction(() => { return dao.GetSlave(heartbeat.SlaveId); });
+
       if (slave == null) {
         actions.Add(new MessageContainer(MessageContainer.MessageType.SayHello));
@@ -47 +55 @@
         if (heartbeat.HbInterval != slave.HbInterval) {
           actions.Add(new MessageContainer(MessageContainer.MessageType.NewHBInterval));
+        }
+        if (ShutdownSlaveComputer(slave.Id)) {
+          actions.Add(new MessageContainer(MessageContainer.MessageType.ShutdownComputer));
         }
 
@@ -56 +67 @@
         slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0) ? SlaveState.Calculating : SlaveState.Idle;
         slave.LastHeartbeat = DateTime.Now;
-        dao.UpdateSlave(slave);
+
+        trans.UseTransaction(() => { dao.UpdateSlave(slave); });
 
         // update task data
@@ -63 +75 @@
         // assign new task
         if (heartbeat.AssignJob && slave.IsAllowedToCalculate && heartbeat.FreeCores > 0) {
-          var availableJobs = dao.GetWaitingTasks(slave, 1);
-          if (availableJobs.Count() > 0) {
-            var job = availableJobs.First();
-            if (AssignJob(slave, job))
-              actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateTask, job.Id));
+          bool mutexAquired = false;
+          var mutex = new Mutex(false, MutexName);
+          try {
+            mutexAquired = mutex.WaitOne(Properties.Settings.Default.SchedulingPatience);
+            if (!mutexAquired)
+              DA.LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling could not be aquired.");
+            else {
+              IEnumerable<TaskInfoForScheduler> availableTasks = null;
+              availableTasks = trans.UseTransaction(() => { return taskScheduler.Schedule(dao.GetWaitingTasks(slave)); });
+              if (availableTasks.Any()) {
+                var task = availableTasks.First();
+                AssignJob(slave, task.TaskId);
+                actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateTask, task.TaskId));
+              }
+            }
+          }
+          catch (AbandonedMutexException) {
+            DA.LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling has been abandoned.");
+          }
+          catch (Exception ex) {
+            DA.LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager threw an exception in ProcessHeartbeat: " + ex.ToString());
+          }
+          finally {
+            if (mutexAquired) mutex.ReleaseMutex();
           }
         }
@@ -74 +105 @@
     }
 
-    // returns true if assignment was successful
-    private bool AssignJob(Slave slave, Task task) {
-      // load task again and check if it is still available (this is an attempt to reduce the race condition which causes multiple heartbeats to get the same task assigned)
-      if (dao.GetTask(task.Id).State != TaskState.Waiting) return false;
+    private void AssignJob(Slave slave, Guid taskId) {
+      trans.UseTransaction(() => {
+        var task = dao.UpdateTaskState(taskId, DataAccess.TaskState.Transferring, slave.Id, null, null);
 
-      task = dao.UpdateTaskState(task.Id, DataAccess.TaskState.Transferring, slave.Id, null, null);
-
-      // from now on the task has some time to send the next heartbeat (ApplicationConstants.TransferringJobHeartbeatTimeout)
-      task.LastHeartbeat = DateTime.Now;
-      dao.UpdateTask(task);
-      return true;
+        // from now on the task has some time to send the next heartbeat (ApplicationConstants.TransferringJobHeartbeatTimeout)
+        task.LastHeartbeat = DateTime.Now;
+        dao.UpdateTask(task);
+      });
     }
 
@@ -102 +130 @@
         // process the jobProgresses
         foreach (var jobProgress in heartbeat.JobProgress) {
-          Task curTask = dao.GetTask(jobProgress.Key);
+          Task curTask = null;
+          curTask = trans.UseTransaction(() => { return dao.GetTask(jobProgress.Key); });
           if (curTask == null) {
             // task does not exist in db
@@ -131 +160 @@
                   break;
               }
-              dao.UpdateTask(curTask);
+              trans.UseTransaction(() => { dao.UpdateTask(curTask); });
             }
           }
@@ -140 +169 @@
 
     private bool TaskIsAllowedToBeCalculatedBySlave(Guid slaveId, Task curTask) {
-      var assignedResourceIds = dao.GetAssignedResources(curTask.Id).Select(x => x.Id);
-      var slaveResourceIds = dao.GetParentResources(slaveId).Select(x => x.Id);
-      return assignedResourceIds.Any(x => slaveResourceIds.Contains(x));
+      return trans.UseTransaction(() => {
+        var assignedResourceIds = dao.GetAssignedResources(curTask.Id).Select(x => x.Id);
+        var slaveResourceIds = dao.GetParentResources(slaveId).Select(x => x.Id);
+        return assignedResourceIds.Any(x => slaveResourceIds.Contains(x));
+      });
     }
 
     private bool SlaveIsAllowedToCalculate(Guid slaveId) {
       // the slave may only calculate if there is no downtime right now. this needs to be checked for every parent resource also
-      return dao.GetParentResources(slaveId).All(r => dao.GetDowntimes(x => x.ResourceId == r.Id && (DateTime.Now >= x.StartDate) && (DateTime.Now <= x.EndDate)).Count() == 0);
+      return trans.UseTransaction(() => { return dao.GetParentResources(slaveId).All(r => dao.GetDowntimes(x => x.ResourceId == r.Id && x.DowntimeType == DA.DowntimeType.Offline && (DateTime.Now >= x.StartDate) && (DateTime.Now <= x.EndDate)).Count() == 0); });
+    }
+
+    private bool ShutdownSlaveComputer(Guid slaveId) {
+      return trans.UseTransaction(() => { return dao.GetParentResources(slaveId).Any(r => dao.GetDowntimes(x => x.ResourceId == r.Id && x.DowntimeType == DA.DowntimeType.Shutdown && (DateTime.Now >= x.StartDate) && (DateTime.Now <= x.EndDate)).Count() != 0); });
     }
   }

@@ -21 +21 @@
 protoc.exe
 _ReSharper.HeuristicLab 3.3 Tests
+Google.ProtocolBuffers-2.4.1.473.dll
 packages