Changeset 9381 for branches/HivePerformance

Timestamp:

04/18/13 16:35:14 (12 years ago)

Author:

pfleck

Message:

#2030
Activated Delayed Loading for binary data.
Added HiveOperationContext to store HiveDataContext for whole ServiceOperation duration.
Added HiveDao methods to query database objects, not DTOs.
Changed HartbeatManager to use only database objects from new queries.

Location:

branches/HivePerformance/sources

Files:

• HeuristicLab.Services.Hive.DataAccess/3.3/HiveDataContext.dbml (modified) (2 diffs)
• HeuristicLab.Services.Hive.DataAccess/3.3/HiveDataContext.designer.cs (modified) (5 diffs)
• HeuristicLab.Services.Hive/3.3/HeuristicLab.Services.Hive-3.3.csproj (modified) (1 diff)
• HeuristicLab.Services.Hive/3.3/HiveDao.cs (modified) (7 diffs)
• HeuristicLab.Services.Hive/3.3/HiveOperationContext.cs (added)
• HeuristicLab.Services.Hive/3.3/Interfaces/IHiveDao.cs (modified) (4 diffs)
• HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs (modified) (4 diffs)

branches/HivePerformance/sources/HeuristicLab.Services.Hive.DataAccess/3.3/HiveDataContext.dbml

@@ -114 +114 @@
     <Type Name="TaskData">
       <Column Name="TaskId" Storage="_JobId" Type="System.Guid" DbType="UniqueIdentifier NOT NULL" IsPrimaryKey="true" CanBeNull="false" />
-      <Column Name="Data" Type="System.Data.Linq.Binary" DbType="VarBinary(MAX)" CanBeNull="false" UpdateCheck="Never" />
+      <Column Name="Data" Type="System.Data.Linq.Binary" DbType="VarBinary(MAX)" CanBeNull="false" UpdateCheck="Never" IsDelayLoaded="true" />
       <Column Name="LastUpdate" Type="System.DateTime" DbType="DateTime" CanBeNull="false" />
       <Association Name="Task_TaskData" Member="Task" Storage="_Job" ThisKey="TaskId" OtherKey="TaskId" Type="Task" IsForeignKey="true" />
@@ -123 +123 @@
       <Column Name="PluginDataId" Type="System.Guid" DbType="UniqueIdentifier NOT NULL" IsPrimaryKey="true" IsDbGenerated="true" CanBeNull="false" />
       <Column Name="PluginId" Type="System.Guid" DbType="UniqueIdentifier NOT NULL" CanBeNull="false" />
-      <Column Name="Data" Type="System.Data.Linq.Binary" DbType="VarBinary(MAX) NOT NULL" CanBeNull="false" UpdateCheck="Never" />
+      <Column Name="Data" Type="System.Data.Linq.Binary" DbType="VarBinary(MAX) NOT NULL" CanBeNull="false" UpdateCheck="Never" IsDelayLoaded="true" />
       <Column Name="FileName" Type="System.String" DbType="VarChar(MAX)" CanBeNull="false" />
       <Association Name="Plugin_PluginData" Member="Plugin" ThisKey="PluginId" OtherKey="PluginId" Type="Plugin" IsForeignKey="true" />

branches/HivePerformance/sources/HeuristicLab.Services.Hive.DataAccess/3.3/HiveDataContext.designer.cs

@@ -3 +3 @@
 // <auto-generated>
 //     This code was generated by a tool.
-//     Runtime Version:4.0.30319.17929
+//     Runtime Version:4.0.30319.18034
 //
 //     Changes to this file may cause incorrect behavior and will be lost if
@@ -2661 +2661 @@
     private System.Guid _JobId;
     
-    private System.Data.Linq.Binary _Data;
+    private System.Data.Linq.Link<System.Data.Linq.Binary> _Data;
     
     private System.DateTime _LastUpdate;
@@ -2714 +2714 @@
       get
       {
-        return this._Data;
-      }
-      set
-      {
-        if ((this._Data != value))
+        return this._Data.Value;
+      }
+      set
+      {
+        if ((this._Data.Value != value))
         {
           this.OnDataChanging(value);
           this.SendPropertyChanging();
-          this._Data = value;
+          this._Data.Value = value;
           this.SendPropertyChanged("Data");
           this.OnDataChanged();
@@ -2814 +2814 @@
     private System.Guid _PluginId;
     
-    private System.Data.Linq.Binary _Data;
+    private System.Data.Linq.Link<System.Data.Linq.Binary> _Data;
     
     private string _FileName;
@@ -2889 +2889 @@
       get
       {
-        return this._Data;
-      }
-      set
-      {
-        if ((this._Data != value))
+        return this._Data.Value;
+      }
+      set
+      {
+        if ((this._Data.Value != value))
         {
           this.OnDataChanging(value);
           this.SendPropertyChanging();
-          this._Data = value;
+          this._Data.Value = value;
           this.SendPropertyChanged("Data");
           this.OnDataChanged();

branches/HivePerformance/sources/HeuristicLab.Services.Hive/3.3/HeuristicLab.Services.Hive-3.3.csproj

@@ -139 +139 @@
     <Compile Include="HiveDao.cs" />
     <Compile Include="HiveJanitor.cs" />
+    <Compile Include="HiveOperationContext.cs" />
     <Compile Include="Interfaces\IHiveDao.cs" />
     <Compile Include="Interfaces\ITaskScheduler.cs" />

branches/HivePerformance/sources/HeuristicLab.Services.Hive/3.3/HiveDao.cs

@@ -43 +43 @@
     }
 
+    public Task GetTaskDA(Guid id) {
+      var db = HiveOperationContext.Current.DataContext;
+      return db.Tasks.SingleOrDefault(x => x.TaskId == id);
+    }
+
     public IEnumerable<DT.Task> GetTasks(Expression<Func<Task, bool>> predicate) {
       using (var db = CreateContext()) {
@@ -138 +143 @@
         db.SubmitChanges();
       }
+    }
+
+    public void UpdateTaskDA(Task task) {
+      var db = HiveOperationContext.Current.DataContext;
+      db.SubmitChanges();
     }
 
@@ -195 +205 @@
     }
 
+    public IEnumerable<TaskInfoForScheduler> GetWaitingTasksDA(Slave slave) {
+      var db = HiveOperationContext.Current.DataContext;
+      var parentResources = GetParentResourcesDA(slave.ResourceId);
+      var resourceIds = parentResources.Select(x => x.ResourceId);
+      //Originally we checked here if there are parent tasks which should be calculated (with GetParentTasks(resourceIds, count, false);).
+      //Because there is at the moment no case where this makes sense (there don't exist parent tasks which need to be calculated), 
+      //we skip this step because it's wasted runtime
+
+      var query = from ar in db.AssignedResources
+                  where resourceIds.Contains(ar.ResourceId)
+                     && !(ar.Task.IsParentTask && ar.Task.FinishWhenChildJobsFinished)
+                     && ar.Task.State == TaskState.Waiting
+                     && ar.Task.CoresNeeded <= slave.FreeCores
+                     && ar.Task.MemoryNeeded <= slave.FreeMemory
+                  select new TaskInfoForScheduler() { TaskId = ar.Task.TaskId, JobId = ar.Task.JobId, Priority = ar.Task.Priority };
+      var waitingTasks = query.ToArray();
+      return waitingTasks;
+    }
+
     public DT.Task UpdateTaskState(Guid taskId, TaskState taskState, Guid? slaveId, Guid? userId, string exception) {
       using (var db = CreateContext()) {
@@ -222 +251 @@
         return DT.Convert.ToDto(task);
       }
+    }
+
+    public Task UpdateTaskStateDA(Guid taskId, TaskState taskState, Guid? slaveId, Guid? userId, string exception) {
+      var db = HiveOperationContext.Current.DataContext;
+
+      db.StateLogs.InsertOnSubmit(new StateLog {
+        TaskId = taskId,
+        State = taskState,
+        SlaveId = slaveId,
+        UserId = userId,
+        Exception = exception,
+        DateTime = DateTime.Now
+      });
+
+      var task = db.Tasks.SingleOrDefault(x => x.TaskId == taskId);
+      task.State = taskState;
+
+      db.SubmitChanges();
+
+      return task;
     }
     #endregion
@@ -516 +565 @@
     }
 
+    public Slave GetSlaveDA(Guid id) {
+      var db = HiveOperationContext.Current.DataContext;
+      return db.Resources.OfType<Slave>().SingleOrDefault(x => x.ResourceId == id);
+    }
+
     public IEnumerable<DT.Slave> GetSlaves(Expression<Func<Slave, bool>> predicate) {
       using (var db = CreateContext()) {
@@ -538 +592 @@
         db.SubmitChanges();
       }
+    }
+
+    public void UpdateSlaveDA(Slave slave) {
+      var db = HiveOperationContext.Current.DataContext;
+      db.SubmitChanges();
     }
 
@@ -664 +723 @@
         return resources.Select(r => DT.Convert.ToDto(r)).ToArray();
       }
+    }
+
+    public IEnumerable<Resource> GetParentResourcesDA(Guid resourceId) {
+      var db = HiveOperationContext.Current.DataContext;
+      var resources = new List<Resource>();
+      CollectParentResources(resources, db.Resources.Where(r => r.ResourceId == resourceId).Single());
+      return resources;
     }
 

branches/HivePerformance/sources/HeuristicLab.Services.Hive/3.3/Interfaces/IHiveDao.cs

@@ -30 +30 @@
     #region Task Methods
     DT.Task GetTask(Guid id);
+    Task GetTaskDA(Guid id);
     IEnumerable<DT.Task> GetTasks(Expression<Func<Task, bool>> predicate);
     IEnumerable<DT.LightweightTask> GetLightweightTasks(Expression<Func<Task, bool>> predicate);
@@ -37 +38 @@
     void UpdateTaskAndStateLogs(DT.Task dto);
     void UpdateTask(DT.Task dto);
+    void UpdateTaskDA(Task dto);
     void DeleteTask(Guid id);
     IEnumerable<TaskInfoForScheduler> GetWaitingTasks(DT.Slave slave);
+    IEnumerable<TaskInfoForScheduler> GetWaitingTasksDA(Slave slave);
     IEnumerable<DT.Task> GetParentTasks(IEnumerable<Guid> resourceIds, int count, bool finished);
     DT.Task UpdateTaskState(Guid taskId, TaskState taskState, Guid? slaveId, Guid? userId, string exception);
+    Task UpdateTaskStateDA(Guid taskId, TaskState taskState, Guid? slaveId, Guid? userId, string exception);
     #endregion
 
@@ -95 +99 @@
     #region Slave Methods
     DT.Slave GetSlave(Guid id);
+    Slave GetSlaveDA(Guid id);
     IEnumerable<DT.Slave> GetSlaves(Expression<Func<Slave, bool>> predicate);
     Guid AddSlave(DT.Slave dto);
     void UpdateSlave(DT.Slave dto);
+    void UpdateSlaveDA(Slave dto);
     void DeleteSlave(Guid id);
     #endregion
@@ -118 +124 @@
     IEnumerable<DT.Resource> GetAssignedResources(Guid jobId);
     IEnumerable<DT.Resource> GetParentResources(Guid resourceId);
+    IEnumerable<Resource> GetParentResourcesDA(Guid resourceId);
     IEnumerable<DT.Resource> GetChildResources(Guid resourceId);
     IEnumerable<DT.Task> GetJobsByResourceId(Guid resourceId);

branches/HivePerformance/sources/HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs

@@ -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); });
 
-      if (slave == null) {
-        actions.Add(new MessageContainer(MessageContainer.MessageType.SayHello));
-      } else {
-        if (heartbeat.HbInterval != slave.HbInterval) {
-          actions.Add(new MessageContainer(MessageContainer.MessageType.NewHBInterval));
+      DA.Slave slave = null;
+      trans.UseTransaction(() => {
+        slave = dao.GetSlaveDA(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 (ShutdownSlaveComputer(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.IsAllowedToCalculate = SlaveIsAllowedToCalculate(slave.ResourceId);
+          slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0) ? DA.SlaveState.Calculating : DA.SlaveState.Idle;
+          slave.LastHeartbeat = DateTime.Now;
+
+          dao.UpdateSlaveDA(slave);
         }
-        if (ShutdownSlaveComputer(slave.Id)) {
-          actions.Add(new MessageContainer(MessageContainer.MessageType.ShutdownComputer));
-        }
+      });
 
-        // update slave data
-        slave.FreeCores = heartbeat.FreeCores;
-        slave.FreeMemory = heartbeat.FreeMemory;
-        slave.CpuUtilization = heartbeat.CpuUtilization;
-        slave.IsAllowedToCalculate = SlaveIsAllowedToCalculate(slave.Id);
-        slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0) ? SlaveState.Calculating : SlaveState.Idle;
-        slave.LastHeartbeat = DateTime.Now;
+      // update task data
+      actions.AddRange(UpdateTasks(heartbeat, slave.IsAllowedToCalculate));
 
-        trans.UseTransaction(() => { dao.UpdateSlave(slave); });
+      // assign new task
+      if (heartbeat.AssignJob && slave.IsAllowedToCalculate && heartbeat.FreeCores > 0) {
+        bool mutexAquired = false;
+        var mutex = new Mutex(false, MutexName);
+        try {
 
-        // update task data
-        actions.AddRange(UpdateTasks(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)
-              DA.LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling could not be aquired.");
-            else {
+          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 {
+            trans.UseTransaction(() => {
               IEnumerable<TaskInfoForScheduler> availableTasks = null;
-              availableTasks = trans.UseTransaction(() => { return taskScheduler.Schedule(dao.GetWaitingTasks(slave)); });
+              availableTasks = taskScheduler.Schedule(dao.GetWaitingTasksDA(slave));
               if (availableTasks.Any()) {
                 var task = availableTasks.First();
@@ -89 +95 @@
                 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();
-          }
+        }
+        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();
         }
       }
@@ -105 +111 @@
     }
 
-    private void AssignJob(Slave slave, Guid taskId) {
-      trans.UseTransaction(() => {
-        var task = dao.UpdateTaskState(taskId, DataAccess.TaskState.Transferring, slave.Id, null, null);
+    private void AssignJob(DA.Slave slave, Guid taskId) {
+      var task = dao.UpdateTaskStateDA(taskId, DataAccess.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.UpdateTaskDA(task);
     }
 
@@ -128 +132 @@
         actions.Add(new MessageContainer(MessageContainer.MessageType.PauseAll));
       } else {
-        // process the jobProgresses
-        foreach (var jobProgress in heartbeat.JobProgress) {
-          Task curTask = null;
-          curTask = trans.UseTransaction(() => { return dao.GetTask(jobProgress.Key); });
-          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);
-          } else {
-            if (curTask.CurrentStateLog.SlaveId == Guid.Empty || curTask.CurrentStateLog.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)) {
-              // 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));
+        trans.UseTransaction(() => {
+          // process the jobProgresses
+          foreach (var jobProgress in heartbeat.JobProgress) {
+            var curTask = dao.GetTaskDA(jobProgress.Key);
+            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);
             } else {
-              // save task execution time
-              curTask.ExecutionTime = jobProgress.Value;
-              curTask.LastHeartbeat = DateTime.Now;
+              var currentStateLog = curTask.StateLogs.LastOrDefault();
+              if (currentStateLog.SlaveId == Guid.Empty || currentStateLog.SlaveId != heartbeat.SlaveId) {
+                // assigned slave does not match heartbeat
+                actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.TaskId));
+                DA.LogFactory.GetLogger(this.GetType().Namespace).Log("The slave " + heartbeat.SlaveId + " is not supposed to calculate task: " + curTask);
+              } else if (!TaskIsAllowedToBeCalculatedBySlave(heartbeat.SlaveId, curTask)) {
+                // 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));
+              } else {
+                // save task execution time
+                curTask.ExecutionTimeMs = jobProgress.Value.TotalMilliseconds;
+                curTask.LastHeartbeat = DateTime.Now;
 
-              switch (curTask.Command) {
-                case Command.Stop:
-                  actions.Add(new MessageContainer(MessageContainer.MessageType.StopTask, curTask.Id));
-                  break;
-                case Command.Pause:
-                  actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.Id));
-                  break;
-                case Command.Abort:
-                  actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.Id));
-                  break;
+                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;
+                }
+                dao.UpdateTaskDA(curTask);
               }
-              trans.UseTransaction(() => { dao.UpdateTask(curTask); });
             }
           }
-        }
+        });
       }
       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 TaskIsAllowedToBeCalculatedBySlave(Guid slaveId, DA.Task curTask) {
+      var assignedResourceIds = curTask.AssignedResources.Select(ar => ar.Resource.ResourceId);
+      var slaveResourceIds = dao.GetParentResourcesDA(slaveId).Select(x => x.ResourceId);
+      return assignedResourceIds.Any(r => slaveResourceIds.Contains(r));
     }
 
     private bool SlaveIsAllowedToCalculate(Guid slaveId) {
+      var parentResources = dao.GetParentResourcesDA(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); });
+      return parentResources.All(r => r.Downtimes.Where(d => d.DowntimeType == DA.DowntimeType.Offline && (DateTime.Now >= d.StartDate) && (DateTime.Now <= d.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); });
+      var parentResources = dao.GetParentResourcesDA(slaveId);
+
+      return parentResources.Any(r => r.Downtimes.Where(d => d.DowntimeType == DA.DowntimeType.Shutdown && (DateTime.Now >= d.StartDate) && (DateTime.Now <= d.EndDate)).Count() != 0);
     }
   }