Changeset 9700 for stable/HeuristicLab.Services.Hive

Timestamp:

07/06/13 11:38:38 (11 years ago)

Author:

ascheibe

Message:

#2030 merged r9665,r9666,r9675 into the stable branch

Location:

stable

Files:

• . (modified) (1 prop)
• HeuristicLab.Services.Hive (modified) (1 prop)
• HeuristicLab.Services.Hive/3.3 (modified) (1 prop)
• HeuristicLab.Services.Hive/3.3/Convert.cs (modified) (3 diffs)
• HeuristicLab.Services.Hive/3.3/HeuristicLab.Services.Hive-3.3.csproj (modified) (1 diff)
• HeuristicLab.Services.Hive/3.3/HiveDao.cs (modified) (2 diffs)
• HeuristicLab.Services.Hive/3.3/HiveOperationContext.cs (copied) (copied from trunk/sources/HeuristicLab.Services.Hive/3.3/HiveOperationContext.cs)
• HeuristicLab.Services.Hive/3.3/HiveService.cs (modified) (9 diffs)
• HeuristicLab.Services.Hive/3.3/Interfaces/IOptimizedHiveDao.cs (copied) (copied from trunk/sources/HeuristicLab.Services.Hive/3.3/Interfaces/IOptimizedHiveDao.cs)
• HeuristicLab.Services.Hive/3.3/Interfaces/IServiceLocator.cs (modified) (1 diff)
• HeuristicLab.Services.Hive/3.3/Manager/EventManager.cs (modified) (1 diff)
• HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs (modified) (9 diffs)
• HeuristicLab.Services.Hive/3.3/OptimizedHiveDao.cs (copied) (copied from trunk/sources/HeuristicLab.Services.Hive/3.3/OptimizedHiveDao.cs)
• HeuristicLab.Services.Hive/3.3/ServiceLocator.cs (modified) (1 diff)

stable/HeuristicLab.Services.Hive/3.3

@@ -2 +2 @@
 bin
 obj
+*.user

@@ -2 +2 @@
 bin
 obj
+*.user

stable/HeuristicLab.Services.Hive/3.3/Convert.cs

@@ -102 +102 @@
     public static DT.TaskData ToDto(DB.TaskData source) {
       if (source == null) return null;
-      return new DT.TaskData { TaskId = source.TaskId, Data = source.Data.ToArray(), LastUpdate = source.LastUpdate };
+      return new DT.TaskData { TaskId = source.TaskId, Data = source.Data, LastUpdate = source.LastUpdate };
     }
     public static DB.TaskData ToEntity(DT.TaskData source) {
@@ -111 +111 @@
     public static void ToEntity(DT.TaskData source, DB.TaskData target) {
       if ((source != null) && (target != null)) {
-        target.TaskId = source.TaskId; target.Data = new Binary(source.Data); target.LastUpdate = source.LastUpdate;
+        target.TaskId = source.TaskId;
+        target.Data = source.Data;
+        target.LastUpdate = source.LastUpdate;
       }
     }
@@ -213 +215 @@
     public static void ToEntity(DT.PluginData source, DB.PluginData target) {
       if ((source != null) && (target != null)) {
-        target.PluginDataId = source.Id; target.PluginId = source.PluginId; target.Data = new Binary(source.Data); target.FileName = source.FileName;
+        target.PluginDataId = source.Id;
+        target.PluginId = source.PluginId;
+        target.Data = source.Data;
+        target.FileName = source.FileName;
       }
     }

stable/HeuristicLab.Services.Hive/3.3/HeuristicLab.Services.Hive-3.3.csproj

@@ -137 +137 @@
     <Compile Include="DataTransfer\Statistics.cs" />
     <Compile Include="DataTransfer\UserStatistics.cs" />
+    <Compile Include="OptimizedHiveDao.cs" />
     <Compile Include="HiveDao.cs" />
     <Compile Include="HiveJanitor.cs" />
+    <Compile Include="HiveOperationContext.cs" />
+    <Compile Include="Interfaces\IOptimizedHiveDao.cs" />
     <Compile Include="Interfaces\IHiveDao.cs" />
     <Compile Include="Interfaces\ITaskScheduler.cs" />

stable/HeuristicLab.Services.Hive/3.3/HiveDao.cs

@@ -35 +35 @@
     }
 
-    public HiveDao() { }
-
     #region Task Methods
     public DT.Task GetTask(Guid id) {
@@ -666 +664 @@
     }
 
-    private void CollectParentResources(List<Resource> resources, Resource resource) {
+    private static void CollectParentResources(ICollection<Resource> resources, Resource resource) {
       if (resource == null) return;
       resources.Add(resource);

stable/HeuristicLab.Services.Hive/3.3/HiveService.cs

@@ -37 +37 @@
   /// </summary>
   [ServiceBehavior(InstanceContextMode = InstanceContextMode.PerCall, IgnoreExtensionDataObject = true)]
+  [HiveOperationContextBehavior]
   public class HiveService : IHiveService {
     private IHiveDao dao {
       get { return ServiceLocator.Instance.HiveDao; }
     }
+    private IOptimizedHiveDao optimizedDao {
+      get { return ServiceLocator.Instance.OptimizedHiveDao; }
+    }
     private Access.IRoleVerifier authen {
       get { return ServiceLocator.Instance.RoleVerifier; }
@@ -64 +68 @@
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client);
       return trans.UseTransaction(() => {
-        task.Id = dao.AddTask(task);
-        taskData.TaskId = task.Id;
-        taskData.LastUpdate = DateTime.Now;
-        dao.AssignJobToResource(task.Id, resourceIds);
-        dao.AddTaskData(taskData);
-        dao.UpdateTaskState(task.Id, DA.TaskState.Waiting, null, userManager.CurrentUserId, null);
-        return taskData.TaskId;
+        var t = DT.Convert.ToEntity(task);
+        t.RequiredPlugins.AddRange(task.PluginsNeededIds.Select(pluginId => new DA.RequiredPlugin { Task = t, PluginId = pluginId }));
+
+        t.JobData = DT.Convert.ToEntity(taskData);
+        t.JobData.LastUpdate = DateTime.Now;
+
+        optimizedDao.AddTask(t);
+
+        dao.AssignJobToResource(t.TaskId, resourceIds);
+
+        optimizedDao.UpdateTaskState(t.TaskId, DA.TaskState.Waiting, null, userManager.CurrentUserId, null);
+
+        return t.TaskId;
       }, false, true);
     }
@@ -76 +86 @@
     public Guid AddChildTask(Guid parentTaskId, Task task, TaskData taskData) {
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client);
-      return trans.UseTransaction(() => {
-        task.ParentTaskId = parentTaskId;
-        return AddTask(task, taskData, dao.GetAssignedResources(parentTaskId).Select(x => x.Id));
-      }, false, true);
+      task.ParentTaskId = parentTaskId;
+      return AddTask(task, taskData, optimizedDao.GetAssignedResourceIds(parentTaskId));
     }
 
@@ -87 +95 @@
 
       return trans.UseTransaction(() => {
-        return dao.GetTask(taskId);
+        return DT.Convert.ToDto(optimizedDao.GetTaskById(taskId));
       }, false, false);
     }
@@ -128 +136 @@
 
       return trans.UseTransaction(() => {
-        return dao.GetLightweightTasks(task => task.JobId == jobId).ToArray();
+        return optimizedDao.GetLightweightTasks(jobId).ToArray();
       }, false, true);
     }
@@ -155 +163 @@
 
       trans.UseTransaction(() => {
-        dao.UpdateTaskAndPlugins(taskDto);
+        var task = optimizedDao.GetTaskByDto(taskDto);
+        optimizedDao.UpdateTask(task);
       });
     }
@@ -164 +173 @@
 
       trans.UseTransaction(() => {
-        dao.UpdateTaskAndPlugins(task);
-      });
-
-      trans.UseTransaction(() => {
-        taskData.LastUpdate = DateTime.Now;
-        dao.UpdateTaskData(taskData);
+        var t = optimizedDao.GetTaskByDto(task);
+        optimizedDao.UpdateTask(t);
+      });
+
+      trans.UseTransaction(() => {
+        var data = optimizedDao.GetTaskDataByDto(taskData);
+        data.LastUpdate = DateTime.Now;
+        optimizedDao.UpdateTaskData(data);
       });
     }
@@ -196 +207 @@
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client, HiveRoles.Slave);
       author.AuthorizeForTask(taskId, Permission.Full);
-      return trans.UseTransaction(() => {
-        Task task = dao.UpdateTaskState(taskId, DataTransfer.Convert.ToEntity(taskState), slaveId, userId, exception);
-
-        if (task.Command.HasValue && task.Command.Value == Command.Pause && task.State == TaskState.Paused) {
+
+      return trans.UseTransaction(() => {
+        var task = optimizedDao.UpdateTaskState(taskId, DT.Convert.ToEntity(taskState), slaveId, userId, exception);
+
+        if (task.Command.HasValue && task.Command.Value == DA.Command.Pause && task.State == DA.TaskState.Paused) {
           task.Command = null;
-        } else if (task.Command.HasValue && task.Command.Value == Command.Abort && task.State == TaskState.Aborted) {
+        } else if (task.Command.HasValue && task.Command.Value == DA.Command.Abort && task.State == DA.TaskState.Aborted) {
           task.Command = null;
-        } else if (task.Command.HasValue && task.Command.Value == Command.Stop && task.State == TaskState.Aborted) {
+        } else if (task.Command.HasValue && task.Command.Value == DA.Command.Stop && task.State == DA.TaskState.Aborted) {
           task.Command = null;
         } else if (taskState == TaskState.Paused && !task.Command.HasValue) {
           // slave paused and uploaded the task (no user-command) -> set waiting.
-          task = dao.UpdateTaskState(taskId, DataTransfer.Convert.ToEntity(TaskState.Waiting), slaveId, userId, exception);
-        }
-
-        dao.UpdateTaskAndPlugins(task);
-        return task;
+          task = optimizedDao.UpdateTaskState(taskId, DA.TaskState.Waiting, slaveId, userId, exception);
+        }
+
+        return DT.Convert.ToDto(task);
       });
     }
@@ -463 +474 @@
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client, HiveRoles.Slave);
       return trans.UseTransaction(() => {
-        return dao.GetPlugin(pluginId);
+        return DT.Convert.ToDto(optimizedDao.GetPluginById(pluginId));
       });
     }

stable/HeuristicLab.Services.Hive/3.3/Interfaces/IServiceLocator.cs

@@ -27 +27 @@
     IAuthorizationManager AuthorizationManager { get; }
     IHiveDao HiveDao { get; }
+    IOptimizedHiveDao OptimizedHiveDao { get; }
     IEventManager EventManager { get; }
     ITransactionManager TransactionManager { get; }

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

@@ -78 +78 @@
       //we have to find another way to deal with this.  
       //until then the next line is commented out...
-      //stats.UserStatistics = dao.GetUserStatistics();
+      //stats.UserStatistics = dtoDao.GetUserStatistics();
       dao.AddStatistics(stats);
     }

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

@@ -24 +24 @@
 using System.Linq;
 using System.Threading;
-using HeuristicLab.Services.Hive.DataTransfer;
-using DA = HeuristicLab.Services.Hive.DataAccess;
+using HeuristicLab.Services.Hive.DataAccess;
+using Heartbeat = HeuristicLab.Services.Hive.DataTransfer.Heartbeat;
 
 namespace HeuristicLab.Services.Hive {
@@ -31 +31 @@
     private const string MutexName = "HiveTaskSchedulingMutex";
 
-    private IHiveDao dao {
-      get { return ServiceLocator.Instance.HiveDao; }
+    private IOptimizedHiveDao dao {
+      get { return ServiceLocator.Instance.OptimizedHiveDao; }
     }
     private ITaskScheduler taskScheduler {
@@ -47 +47 @@
     public List<MessageContainer> ProcessHeartbeat(Heartbeat heartbeat) {
       List<MessageContainer> actions = new List<MessageContainer>();
+
       Slave slave = null;
-      slave = trans.UseTransaction(() => { return dao.GetSlave(heartbeat.SlaveId); });
-
+      trans.UseTransaction(() => {
+        slave = dao.GetSlaveById(heartbeat.SlaveId);
+      });
       if (slave == null) {
         actions.Add(new MessageContainer(MessageContainer.MessageType.SayHello));
@@ -56 +58 @@
           actions.Add(new MessageContainer(MessageContainer.MessageType.NewHBInterval));
         }
-        if (ShutdownSlaveComputer(slave.Id)) {
+        if (dao.SlaveHasToShutdownComputer(slave.ResourceId)) {
           actions.Add(new MessageContainer(MessageContainer.MessageType.ShutdownComputer));
         }
@@ -64 +66 @@
         slave.FreeMemory = heartbeat.FreeMemory;
         slave.CpuUtilization = heartbeat.CpuUtilization;
-        slave.IsAllowedToCalculate = SlaveIsAllowedToCalculate(slave.Id);
+        slave.IsAllowedToCalculate = dao.SlaveIsAllowedToCalculate(slave.ResourceId);
         slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0) ? SlaveState.Calculating : SlaveState.Idle;
         slave.LastHeartbeat = DateTime.Now;
 
-        trans.UseTransaction(() => { dao.UpdateSlave(slave); });
+        trans.UseTransaction(() => {
+          dao.UpdateSlave(slave);
+        });
 
         // update task data
@@ -78 +82 @@
           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.");
+              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));
-              }
+              trans.UseTransaction(() => {
+                IEnumerable<TaskInfoForScheduler> availableTasks = null;
+                availableTasks = taskScheduler.Schedule(dao.GetWaitingTasks(slave).ToArray());
+                if (availableTasks.Any()) {
+                  var task = availableTasks.First();
+                  AssignTask(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.");
+            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());
+            LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager threw an exception in ProcessHeartbeat: " + ex.ToString());
           }
           finally {
@@ -105 +112 @@
     }
 
-    private void AssignJob(Slave slave, Guid taskId) {
-      trans.UseTransaction(() => {
-        var task = dao.UpdateTaskState(taskId, DataAccess.TaskState.Transferring, slave.Id, null, null);
+    private void AssignTask(Slave slave, Guid taskId) {
+      var task = dao.UpdateTaskState(taskId, TaskState.Transferring, slave.ResourceId, null, null);
 
-        // from now on the task has some time to send the next heartbeat (ApplicationConstants.TransferringJobHeartbeatTimeout)
-        task.LastHeartbeat = DateTime.Now;
-        dao.UpdateTask(task);
-      });
+      // from now on the task has some time to send the next heartbeat (ApplicationConstants.TransferringJobHeartbeatTimeout)
+      task.LastHeartbeat = DateTime.Now;
+      dao.UpdateTask(task);
     }
 
@@ -130 +135 @@
         // process the jobProgresses
         foreach (var jobProgress in heartbeat.JobProgress) {
-          Task curTask = null;
-          curTask = trans.UseTransaction(() => { return dao.GetTask(jobProgress.Key); });
+          Tuple<Task, Guid?> taskWithLastStateLogSlaveId = null;
+          trans.UseTransaction(() => {
+            taskWithLastStateLogSlaveId = dao.GetTaskByIdAndLastStateLogSlaveId(jobProgress.Key);
+          });
+          var curTask = taskWithLastStateLogSlaveId != null ? taskWithLastStateLogSlaveId.Item1 : null;
           if (curTask == null) {
             // task does not exist in db
             actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, jobProgress.Key));
-            DA.LogFactory.GetLogger(this.GetType().Namespace).Log("Task on slave " + heartbeat.SlaveId + " does not exist in DB: " + jobProgress.Key);
+            LogFactory.GetLogger(this.GetType().Namespace).Log("Task on slave " + heartbeat.SlaveId + " does not exist in DB: " + jobProgress.Key);
           } else {
-            if (curTask.CurrentStateLog.SlaveId == Guid.Empty || curTask.CurrentStateLog.SlaveId != heartbeat.SlaveId) {
+            var slaveId = taskWithLastStateLogSlaveId.Item2;
+            if (slaveId == Guid.Empty || slaveId != heartbeat.SlaveId) {
               // assigned slave does not match heartbeat
-              actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.Id));
-              DA.LogFactory.GetLogger(this.GetType().Namespace).Log("The slave " + heartbeat.SlaveId + " is not supposed to calculate task: " + curTask);
-            } else if (!TaskIsAllowedToBeCalculatedBySlave(heartbeat.SlaveId, curTask)) {
+              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);
+            } else if (!dao.TaskIsAllowedToBeCalculatedBySlave(curTask.TaskId, 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.Id));
+              actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.TaskId));
             } else {
               // save task execution time
-              curTask.ExecutionTime = jobProgress.Value;
+              curTask.ExecutionTimeMs = jobProgress.Value.TotalMilliseconds;
               curTask.LastHeartbeat = DateTime.Now;
 
               switch (curTask.Command) {
                 case Command.Stop:
-                  actions.Add(new MessageContainer(MessageContainer.MessageType.StopTask, curTask.Id));
+                  actions.Add(new MessageContainer(MessageContainer.MessageType.StopTask, curTask.TaskId));
                   break;
                 case Command.Pause:
-                  actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.Id));
+                  actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.TaskId));
                   break;
                 case Command.Abort:
-                  actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.Id));
+                  actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.TaskId));
                   break;
               }
-              trans.UseTransaction(() => { dao.UpdateTask(curTask); });
+              trans.UseTransaction(() => {
+                dao.UpdateTask(curTask);
+              });
             }
           }
@@ -167 +178 @@
       return actions;
     }
-
-    private bool TaskIsAllowedToBeCalculatedBySlave(Guid slaveId, Task curTask) {
-      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 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); });
-    }
   }
 }

stable/HeuristicLab.Services.Hive/3.3/ServiceLocator.cs

@@ -39 +39 @@
         if (hiveDao == null) hiveDao = new HiveDao();
         return hiveDao;
+      }
+    }
+
+    public IOptimizedHiveDao OptimizedHiveDao {
+      get {
+        var dataContext = HiveOperationContext.Current != null
+                            ? HiveOperationContext.Current.DataContext
+                            : new HiveDataContext(Settings.Default.HeuristicLab_Hive_LinqConnectionString);
+        return new OptimizedHiveDao(dataContext);
       }
     }

@@ -2 +2 @@
 bin
 obj
+*.user