Changeset 9109

Timestamp:

01/04/13 16:29:03 (11 years ago)

Author:

ascheibe

Message:

#1712 merged changes from trunk into branch and improved scheduler performance

Location:

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3

Files:

• HeuristicLab.Services.Hive-3.3.csproj (modified) (1 diff)
• HiveDao.cs (modified) (4 diffs)
• HiveService.cs (modified) (1 diff)
• Interfaces/IHiveDao.cs (modified) (3 diffs)
• Manager/HeartbeatManager.cs (modified) (1 diff)
• Scheduler/JobInfoForScheduler.cs (added)
• Scheduler/RoundRobinTaskScheduler.cs (modified) (3 diffs)

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/HeuristicLab.Services.Hive-3.3.csproj

@@ -141 +141 @@
     <Compile Include="Interfaces\IHiveDao.cs" />
     <Compile Include="Interfaces\ITaskScheduler.cs" />
+    <Compile Include="Scheduler\JobInfoForScheduler.cs" />
     <Compile Include="Scheduler\RoundRobinTaskScheduler.cs" />
     <Compile Include="Settings.cs" />

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/HiveDao.cs

@@ -49 +49 @@
     }
 
+    public IEnumerable<DT.LightweightTask> GetLightweightTasks(Expression<Func<Task, bool>> predicate) {
+      List<DT.LightweightTask> tasks = new List<DT.LightweightTask>();
+
+      using (var db = CreateContext()) {
+        var tasksQuery = db.Tasks.Where(predicate).Select(task => new { task.TaskId, task.ExecutionTimeMs, task.ParentTaskId, task.StateLogs, task.State, task.Command });
+        var taskDatasQuery = db.Tasks.Where(predicate).Where(task => task.JobData != null).Select(task => new { task.TaskId, task.JobData.LastUpdate });
+
+        foreach (var task in tasksQuery) {
+          DT.LightweightTask t = new DT.LightweightTask();
+          t.Id = task.TaskId;
+          t.ExecutionTime = TimeSpan.FromMilliseconds(task.ExecutionTimeMs);
+          t.ParentTaskId = task.ParentTaskId;
+          t.StateLog = task.StateLogs == null ? new List<DT.StateLog>() : task.StateLogs.Select(x => DataTransfer.Convert.ToDto(x)).OrderBy(x => x.DateTime).ToList();
+          t.State = DataTransfer.Convert.ToDto(task.State);
+          t.Command = DataTransfer.Convert.ToDto(task.Command);
+          t.LastTaskDataUpdate = taskDatasQuery.Where(x => x.TaskId == task.TaskId).Count() > 0 ? taskDatasQuery.Select(x => x.LastUpdate).First() : DateTime.MinValue;
+          tasks.Add(t);
+        }
+      }
+      return tasks;
+    }
+
     public Guid AddTask(DT.Task dto) {
       using (var db = CreateContext()) {
@@ -255 +277 @@
     }
 
+    public IEnumerable<JobInfoForScheduler> GetJobInfoForScheduler(Expression<Func<Job, bool>> predicate) {
+      using (var db = CreateContext()) {
+        return db.Jobs.Where(predicate).Select(x => new JobInfoForScheduler() { Id = x.JobId, DateCreated = x.DateCreated, OwnerUserId = x.OwnerUserId }).ToArray();
+      }
+    }
+
     public Guid AddJob(DT.Job dto) {
       using (var db = CreateContext()) {
         var entity = DT.Convert.ToEntity(dto);
         db.Jobs.InsertOnSubmit(entity);
-        if (!db.Jobs.Any(x => x.OwnerUserId == dto.OwnerUserId))
+        if (!db.UserPriorities.Any(x => x.UserId == dto.OwnerUserId))
           EnqueueUserPriority(new DT.UserPriority { Id = dto.OwnerUserId, DateEnqueued = dto.DateCreated });
         db.SubmitChanges();
@@ -592 +620 @@
     public IEnumerable<DT.Resource> GetChildResources(Guid resourceId) {
       using (var db = CreateContext()) {
-        var childs = new List<DT.Resource>();
-        foreach (var child in db.Resources.Where(x => x.ParentResourceId == resourceId)) {
-          childs.Add(DT.Convert.ToDto(child));
-          childs.AddRange(GetChildResources(child.ResourceId));
-        }
-        return childs;
-      }
+        return CollectChildResources(resourceId, db);
+      }
+    }
+
+    public IEnumerable<DT.Resource> CollectChildResources(Guid resourceId, HiveDataContext db) {
+      var childs = new List<DT.Resource>();
+      foreach (var child in db.Resources.Where(x => x.ParentResourceId == resourceId)) {
+        childs.Add(DT.Convert.ToDto(child));
+        childs.AddRange(CollectChildResources(child.ResourceId, db));
+      }
+      return childs;
     }
 
@@ -779 +811 @@
         if (entity != null) db.Statistics.DeleteOnSubmit(entity);
         db.SubmitChanges();
+      }
+    }
+
+    public Dictionary<Guid, int> GetWaitingTasksByUser() {
+      using (var db = CreateContext()) {
+        var waitingTasksByUser = from task in db.Tasks
+                                 where task.State == TaskState.Waiting
+                                 group task by task.Job.OwnerUserId into g
+                                 select new { UserId = g.Key, UsedCores = g.Count() };
+        return waitingTasksByUser.ToDictionary(x => x.UserId, x => x.UsedCores);
+      }
+    }
+
+    public Dictionary<Guid, int> GetWaitingTasksByUserForResources(List<Guid> resourceIds) {
+      using (var db = CreateContext()) {
+        var waitingTasksByUser = from task in db.Tasks
+                                 where task.State == TaskState.Waiting && task.AssignedResources.Any(x => resourceIds.Contains(x.ResourceId))
+                                 group task by task.Job.OwnerUserId into g
+                                 select new { UserId = g.Key, UsedCores = g.Count() };
+        return waitingTasksByUser.ToDictionary(x => x.UserId, x => x.UsedCores);
+      }
+    }
+
+    public Dictionary<Guid, int> GetCalculatingTasksByUser() {
+      using (var db = CreateContext()) {
+        var calculatingTasksByUser = from task in db.Tasks
+                                     where task.State == TaskState.Calculating
+                                     group task by task.Job.OwnerUserId into g
+                                     select new { UserId = g.Key, UsedCores = g.Count() };
+        return calculatingTasksByUser.ToDictionary(x => x.UserId, x => x.UsedCores);
+      }
+    }
+
+    public Dictionary<Guid, int> GetCalculatingTasksByUserForResources(List<Guid> resourceIds) {
+      using (var db = CreateContext()) {
+        var calculatingTasksByUser = from task in db.Tasks
+                                     where task.State == TaskState.Calculating && task.AssignedResources.Any(x => resourceIds.Contains(x.ResourceId))
+                                     group task by task.Job.OwnerUserId into g
+                                     select new { UserId = g.Key, UsedCores = g.Count() };
+        return calculatingTasksByUser.ToDictionary(x => x.UserId, x => x.UsedCores);
       }
     }

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/HiveService.cs

@@ -128 +128 @@
 
       return trans.UseTransaction(() => {
-        return dao.GetTasks(x => x.JobId == jobId).Select(x => new LightweightTask(x)).ToArray();
+        return dao.GetLightweightTasks(task => task.JobId == jobId).ToArray();
       }, false, false);
     }

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/Interfaces/IHiveDao.cs

@@ -31 +31 @@
     DT.Task GetTask(Guid id);
     IEnumerable<DT.Task> GetTasks(Expression<Func<Task, bool>> predicate);
+    IEnumerable<DT.LightweightTask> GetLightweightTasks(Expression<Func<Task, bool>> predicate);
     Guid AddTask(DT.Task dto);
     void UpdateTask(DT.Task dto);
@@ -58 +59 @@
     DT.Job GetJob(Guid id);
     IEnumerable<DT.Job> GetJobs(Expression<Func<Job, bool>> predicate);
+    IEnumerable<JobInfoForScheduler> GetJobInfoForScheduler(Expression<Func<Job, bool>> predicate);
     Guid AddJob(DT.Job dto);
     void UpdateJob(DT.Job dto);
@@ -145 +147 @@
 
     #region Statistics Methods
+    Dictionary<Guid, int> GetWaitingTasksByUser();
+    Dictionary<Guid, int> GetWaitingTasksByUserForResources(List<Guid> resourceIds);
+    Dictionary<Guid, int> GetCalculatingTasksByUser();
+    Dictionary<Guid, int> GetCalculatingTasksByUserForResources(List<Guid> resourceIds);
     DT.Statistics GetStatistic(Guid id);
     IEnumerable<DT.Statistics> GetStatistics(Expression<Func<Statistics, bool>> predicate);

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

@@ -87 +87 @@
             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();

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/Scheduler/RoundRobinTaskScheduler.cs

@@ -36 +36 @@
       var userPriorities = dao.GetUserPriorities(x => true).OrderBy(x => x.DateEnqueued).ToArray();
 
-      var jobs = new List<Job>();
+      var jobs = new List<JobInfoForScheduler>();
       foreach (var userPriority in userPriorities) {
-        jobs.AddRange(dao.GetJobs(x => x.OwnerUserId == userPriority.Id));
+        jobs.AddRange(dao.GetJobInfoForScheduler(x => x.OwnerUserId == userPriority.Id));
       }
 
@@ -44 +44 @@
         task => task.JobId,
         job => job.Id,
-        (task, job) => new { Task = task, Job = job })
+        (task, job) => new { Task = task, JobInfo = job })
         .OrderByDescending(x => x.Task.Priority)
         .ToList();
@@ -55 +55 @@
       for (int i = 0; i < count; i++) {
         var defaultEntry = taskJobRelations.First(); // search first task which is not included yet
-        var priorityEntries = taskJobRelations.Where(x => x.Job.OwnerUserId == userPriorities[priorityIndex].Id).ToArray(); // search for tasks with desired user priority
+        var priorityEntries = taskJobRelations.Where(x => x.JobInfo.OwnerUserId == userPriorities[priorityIndex].Id).ToArray(); // search for tasks with desired user priority
         while (!priorityEntries.Any() && ++priorityIndex < userPriorities.Length)
-          priorityEntries = taskJobRelations.Where(x => x.Job.OwnerUserId == userPriorities[priorityIndex].Id).ToArray();
+          priorityEntries = taskJobRelations.Where(x => x.JobInfo.OwnerUserId == userPriorities[priorityIndex].Id).ToArray();
         if (priorityEntries.Any()) { // tasks with desired user priority found
-          var priorityEntry = priorityEntries.OrderByDescending(x => x.Task.Priority).ThenBy(x => x.Job.DateCreated).First();
+          var priorityEntry = priorityEntries.OrderByDescending(x => x.Task.Priority).ThenBy(x => x.JobInfo.DateCreated).First();
           if (defaultEntry.Task.Priority <= priorityEntry.Task.Priority) {
             taskJobRelations.Remove(priorityEntry);