Changeset 9385

Timestamp:

04/19/13 17:16:49 (12 years ago)

Author:

pfleck

Message:

#2030
Replaced lazy loading with specialized queries.
Compiled queries used for Heardbeat queries.
Changed result types to IQueryable<T> for later query modification.

Location:

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

Files:

• HiveDao.cs (modified) (8 diffs)
• Interfaces/IHiveDao.cs (modified) (3 diffs)
• Manager/HeartbeatManager.cs (modified) (5 diffs)

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

@@ -22 +22 @@
 using System;
 using System.Collections.Generic;
+using System.Data.Linq;
 using System.Linq;
 using System.Linq.Expressions;
@@ -45 +46 @@
     public Task GetTaskDA(Guid id) {
       var db = HiveOperationContext.Current.DataContext;
-      return db.Tasks.SingleOrDefault(x => x.TaskId == id);
-    }
+      return GetTaskByIdQuery(db, id).SingleOrDefault();
+    }
+
+    private static Func<HiveDataContext, Guid, IQueryable<Task>> GetTaskByIdQuery = CompiledQuery.Compile((HiveDataContext db, Guid id) =>
+      from t in db.Tasks
+      where t.TaskId == id
+      select t
+    );
 
     public IEnumerable<DT.Task> GetTasks(Expression<Func<Task, bool>> predicate) {
@@ -205 +212 @@
     }
 
-    public IEnumerable<TaskInfoForScheduler> GetWaitingTasksDA(Slave slave) {
+    public IQueryable<TaskInfoForScheduler> GetWaitingTasksDA(Slave slave) {
       var db = HiveOperationContext.Current.DataContext;
+
       var parentResources = GetParentResourcesDA(slave.ResourceId);
       var resourceIds = parentResources.Select(x => x.ResourceId);
@@ -213 +221 @@
       //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;
-    }
+      return 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 };
+    }
+
+    /*private static Func<HiveDataContext, Guid, Slave, IQueryable<TaskInfoForScheduler>> GetWaitingTasksQuery = CompiledQuery.Compile((HiveDataContext db, Guid id, Slave slave) =>
+      from ar in db.AssignedResources
+      where ar.ResourceId == id
+          && !(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 }
+    );*/
 
     public DT.Task UpdateTaskState(Guid taskId, TaskState taskState, Guid? slaveId, Guid? userId, string exception) {
@@ -265 +281 @@
       });
 
-      var task = db.Tasks.SingleOrDefault(x => x.TaskId == taskId);
+      var task = GetTaskDA(taskId);
       task.State = taskState;
 
@@ -567 +583 @@
     public Slave GetSlaveDA(Guid id) {
       var db = HiveOperationContext.Current.DataContext;
-      return db.Resources.OfType<Slave>().SingleOrDefault(x => x.ResourceId == id);
-    }
+      return GetSlaveByIdQuery(db, id).SingleOrDefault();
+    }
+
+    private static Func<HiveDataContext, Guid, IQueryable<Slave>> GetSlaveByIdQuery = CompiledQuery.Compile((HiveDataContext db, Guid slaveId) =>
+      from s in db.Resources.OfType<Slave>()
+      where s.ResourceId == slaveId
+      select s
+    );
 
     public IEnumerable<DT.Slave> GetSlaves(Expression<Func<Slave, bool>> predicate) {
@@ -714 +736 @@
     }
 
+    public IQueryable<Guid> GetAssignedResourcesIdsDA(Guid taskId) {
+      var db = HiveOperationContext.Current.DataContext;
+      return GetAssignedResourcesIdQuery(db, taskId);
+    }
+
+    private static Func<HiveDataContext, Guid, IQueryable<Guid>> GetAssignedResourcesIdQuery = CompiledQuery.Compile((HiveDataContext db, Guid taskId) =>
+     from ar in db.AssignedResources
+     where ar.TaskId == taskId
+     select ar.ResourceId
+   );
+
     /// <summary>
     /// Returns all parent resources of a resource (the given resource is also added)
@@ -727 +760 @@
     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;
-    }
-
-    private void CollectParentResources(List<Resource> resources, Resource resource) {
+      var child = db.Resources.Single(r => r.ResourceId == resourceId);
+
+      yield return child;
+      while (child.ParentResource != null) {
+        child = child.ParentResource;
+        yield return child;
+      }
+    }
+
+    public IEnumerable<Guid> GetParentResourcesIDsDA(Guid resourceId) {
+      var db = HiveOperationContext.Current.DataContext;
+      var child = db.Resources.Single(r => r.ResourceId == resourceId);
+
+      yield return resourceId;
+      while (child.ParentResource != null) {
+        child = child.ParentResource;
+        yield return child.ResourceId;
+      }
+    }
+
+    public IQueryable<int> GetNumberOfDowntimesFromParentResourcesAtCurrentTimeDA(Guid resourceId, DowntimeType type) {
+      var db = HiveOperationContext.Current.DataContext;
+
+      var ids = GetParentResourcesIDsDA(resourceId).ToArray();
+
+      return from r in db.Resources
+             where ids.Contains(r.ResourceId)
+             select (from d in db.Downtimes
+                     where d.ResourceId == r.ResourceId && d.DowntimeType == type && (DateTime.Now >= d.StartDate) && (DateTime.Now <= d.EndDate)
+                     select d).Count();
+    }
+
+    /*private static Func<HiveDataContext, Guid, DowntimeType, int> GetNumberOfDowntimesAtCurrentTimeQuery =
+      CompiledQuery.Compile((HiveDataContext db, Guid ids, DowntimeType type) =>
+        (from d in db.Downtimes
+         where d.ResourceId == ids && d.DowntimeType == type && (DateTime.Now >= d.StartDate) && (DateTime.Now <= d.EndDate)
+         select d).Count()
+    );*/
+
+    private static void CollectParentResources(ICollection<Resource> resources, Resource resource) {
       if (resource == null) return;
       resources.Add(resource);

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

@@ -22 +22 @@
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Linq.Expressions;
 using HeuristicLab.Services.Hive.DataAccess;
@@ -41 +42 @@
     void DeleteTask(Guid id);
     IEnumerable<TaskInfoForScheduler> GetWaitingTasks(DT.Slave slave);
-    IEnumerable<TaskInfoForScheduler> GetWaitingTasksDA(Slave slave);
+    IQueryable<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);
@@ -123 +124 @@
     void AssignJobToResource(Guid taskId, IEnumerable<Guid> resourceIds);
     IEnumerable<DT.Resource> GetAssignedResources(Guid jobId);
+    IQueryable<Guid> GetAssignedResourcesIdsDA(Guid taskId);
     IEnumerable<DT.Resource> GetParentResources(Guid resourceId);
     IEnumerable<Resource> GetParentResourcesDA(Guid resourceId);
+    IEnumerable<Guid> GetParentResourcesIDsDA(Guid resourceId);
+    IQueryable<int> GetNumberOfDowntimesFromParentResourcesAtCurrentTimeDA(Guid resourceId, DowntimeType type);
     IEnumerable<DT.Resource> GetChildResources(Guid resourceId);
     IEnumerable<DT.Task> GetJobsByResourceId(Guid resourceId);

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

@@ -74 +74 @@
       });
 
-      // update task data
-      actions.AddRange(UpdateTasks(heartbeat, slave.IsAllowedToCalculate));
+      if (slave != null) {
+        // 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 {
+        // 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 {
-            trans.UseTransaction(() => {
-              IEnumerable<TaskInfoForScheduler> availableTasks = null;
-              availableTasks = taskScheduler.Schedule(dao.GetWaitingTasksDA(slave));
-              if (availableTasks.Any()) {
-                var task = availableTasks.First();
-                AssignJob(slave, task.TaskId);
-                actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateTask, task.TaskId));
-              }
-            });
+            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 = taskScheduler.Schedule(dao.GetWaitingTasksDA(slave).ToArray());
+                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();
+          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();
+          }
         }
       }
@@ -132 +134 @@
         actions.Add(new MessageContainer(MessageContainer.MessageType.PauseAll));
       } else {
-        trans.UseTransaction(() => {
-          // process the jobProgresses
-          foreach (var jobProgress in heartbeat.JobProgress) {
+        // process the jobProgresses
+        foreach (var jobProgress in heartbeat.JobProgress) {
+          trans.UseTransaction(() => {
             var curTask = dao.GetTaskDA(jobProgress.Key);
             if (curTask == null) {
@@ -141 +143 @@
               DA.LogFactory.GetLogger(this.GetType().Namespace).Log("Task on slave " + heartbeat.SlaveId + " does not exist in DB: " + jobProgress.Key);
             } else {
-              var currentStateLog = curTask.StateLogs.LastOrDefault();
+              var currentStateLog = curTask.StateLogs.Last();
               if (currentStateLog.SlaveId == Guid.Empty || currentStateLog.SlaveId != heartbeat.SlaveId) {
                 // assigned slave does not match heartbeat
@@ -168 +170 @@
               }
             }
-          }
-        });
+          });
+        }
       }
       return actions;
@@ -175 +177 @@
 
     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);
+      var assignedResourceIds = dao.GetAssignedResourcesIdsDA(curTask.TaskId);
+      var slaveResourceIds = dao.GetParentResourcesIDsDA(slaveId).ToArray();
       return assignedResourceIds.Any(r => slaveResourceIds.Contains(r));
     }
 
     private bool SlaveIsAllowedToCalculate(Guid slaveId) {
-      var parentResources = dao.GetParentResourcesDA(slaveId);
+      var downtimes = dao.GetNumberOfDowntimesFromParentResourcesAtCurrentTimeDA(slaveId, DA.DowntimeType.Offline);
       // the slave may only calculate if there is no downtime right now. this needs to be checked for every parent resource also
-      return parentResources.All(r => r.Downtimes.Where(d => d.DowntimeType == DA.DowntimeType.Offline && (DateTime.Now >= d.StartDate) && (DateTime.Now <= d.EndDate)).Count() == 0);
+      return downtimes.All(x => x == 0);
     }
 
     private bool ShutdownSlaveComputer(Guid slaveId) {
-      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);
+      var downtimes = dao.GetNumberOfDowntimesFromParentResourcesAtCurrentTimeDA(slaveId, DA.DowntimeType.Shutdown);
+      return downtimes.Any(x => x != 0);
     }
   }