Changeset 9363 for branches/OaaS/HeuristicLab.Services.Hive/3.3/HiveDao.cs

Timestamp:

04/16/13 13:13:41 (11 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/HiveDao.cs (modified) (19 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/HiveDao.cs

@@ -22 +22 @@
 using System;
 using System.Collections.Generic;
+using System.Configuration;
+using System.Data.Common;
 using System.Linq;
 using System.Linq.Expressions;
+using HeuristicLab.Services.Hive.Interfaces;
 using DT = HeuristicLab.Services.Hive.DataTransfer;
-using System.ServiceModel;
-using HeuristicLab.Services.Hive.Interfaces;
-using System.Data.Common;
-using System.Configuration;
 
 namespace HeuristicLab.Services.Hive.DataAccess {
   public class HiveDao : IHiveDao {
-    
+
     public HiveDataContext CreateContext(bool longRunning = false) {
       //var context = new HiveDataContext(Settings.Default.HeuristicLab_Hive_LinqConnectionString);
@@ -38 +37 @@
       if (ConfigurationManager.ConnectionStrings[Settings.Default.HiveConnectionStringName] == null) {
         context = new HiveDataContext(Settings.Default.HeuristicLab_Hive_LinqConnectionString);
-      }
-      else {
+      } else {
         context = new HiveDataContext(provider.GetOpenConnection(Settings.Default.HiveConnectionStringName));
       }
@@ -93 +91 @@
     }
 
+    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 IEnumerable<DT.LightweightTask> GetLightweightTasksWithoutStateLog(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.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 = new List<DT.StateLog>();
+          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) {
       return ExecuteWithContext<Guid>((db) => {
@@ -106 +148 @@
     }
 
-    public void UpdateTask(DT.Task dto) {
+    public void UpdateTaskAndPlugins(DT.Task dto) {
       ExecuteWithContext((db) => {
         var entity = db.Tasks.FirstOrDefault(x => x.TaskId == dto.Id);
@@ -118 +160 @@
         db.SubmitChanges();
       });
+    }
+
+    public void UpdateTaskAndStateLogs(DT.Task dto) {
+      using (var db = CreateContext()) {
+        var entity = db.Tasks.FirstOrDefault(x => x.TaskId == dto.Id);
+        if (entity == null) db.Tasks.InsertOnSubmit(DT.Convert.ToEntity(dto));
+        else DT.Convert.ToEntity(dto, entity);
+        db.SubmitChanges();
+      }
+    }
+
+    public void UpdateTask(DT.Task dto) {
+      using (var db = CreateContext()) {
+        db.DeferredLoadingEnabled = false;
+
+        var entity = db.Tasks.FirstOrDefault(x => x.TaskId == dto.Id);
+        if (entity == null) db.Tasks.InsertOnSubmit(DT.Convert.ToEntity(dto));
+        else DT.Convert.ToEntityTaskOnly(dto, entity);
+        db.SubmitChanges();
+      }
     }
 
@@ -156 +218 @@
     }
 
-    public IEnumerable<DT.Task> GetWaitingTasks(DT.Slave slave, int count) {
-      return ExecuteWithContext<IEnumerable<DT.Task>>((db) => {
+    public IEnumerable<TaskInfoForScheduler> GetWaitingTasks(DT.Slave slave) {
+      return ExecuteWithContext<IEnumerable<TaskInfoForScheduler>>((db) => {
         var resourceIds = GetParentResources(slave.Id).Select(r => r.Id);
         //Originally we checked here if there are parent tasks which should be calculated (with GetParentTasks(resourceIds, count, false);).
@@ -169 +231 @@
                        && ar.Task.CoresNeeded <= slave.FreeCores
                        && ar.Task.MemoryNeeded <= slave.FreeMemory
-                    orderby ar.Task.Priority descending, db.Random() // take random task to avoid the race condition that occurs when this method is called concurrently (the same task would be returned)
-                    select DT.Convert.ToDto(ar.Task);
-        var waitingTasks = (count == 0 ? query : query.Take(count)).ToArray();
+                    select new TaskInfoForScheduler() { TaskId = ar.Task.TaskId, JobId = ar.Task.JobId, Priority = ar.Task.Priority };
+        var waitingTasks = query.ToArray();
         return waitingTasks;
       });
@@ -178 +239 @@
     public DT.Task UpdateTaskState(Guid taskId, TaskState taskState, Guid? slaveId, Guid? userId, string exception) {
       return ExecuteWithContext<DT.Task>((db) => {
-        var job = db.Tasks.SingleOrDefault(x => x.TaskId == taskId);
-        job.State = taskState;
+        db.DeferredLoadingEnabled = false;
+        var task = db.Tasks.SingleOrDefault(x => x.TaskId == taskId);
+        task.State = taskState;
         db.StateLogs.InsertOnSubmit(new StateLog {
           TaskId = taskId,
@@ -189 +251 @@
         });
         db.SubmitChanges();
-        job = db.Tasks.SingleOrDefault(x => x.TaskId == taskId);
-        return DT.Convert.ToDto(job);
+        task = db.Tasks.SingleOrDefault(x => x.TaskId == taskId);
+        return DT.Convert.ToDto(task);
       });
     }
@@ -299 +361 @@
     }
 
+    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) {
       return ExecuteWithContext<Guid>((db) => {
         var entity = DT.Convert.ToEntity(dto);
         db.Jobs.InsertOnSubmit(entity);
+        if (!db.UserPriorities.Any(x => x.UserId == dto.OwnerUserId))
+          EnqueueUserPriority(new DT.UserPriority { Id = dto.OwnerUserId, DateEnqueued = dto.DateCreated });
         db.SubmitChanges();
         return entity.JobId;
@@ -374 +444 @@
             // not allowed, delete
             db.JobPermissions.DeleteOnSubmit(jobPermission);
-          }
-          else {
+          } else {
             // update
             jobPermission.Permission = permission;
             jobPermission.GrantedByUserId = grantedByUserId; // update grantedByUserId, always the last "granter" is stored
           }
-        }
-        else {
+        } else {
           // insert
           if (permission != Permission.NotAllowed) {
@@ -441 +509 @@
 
     public IEnumerable<DT.PluginData> GetPluginDatas(Expression<Func<PluginData, bool>> predicate) {
-      return ExecuteWithContext <IEnumerable<DT.PluginData>>((db) => {
+      return ExecuteWithContext<IEnumerable<DT.PluginData>>((db) => {
         return db.PluginDatas.Where(predicate).Select(x => DT.Convert.ToDto(x)).ToArray();
       });
@@ -599 +667 @@
     }
 
-    public void AssignJobToResource(Guid jobId, Guid resourceId) {
-      ExecuteWithContext((db) => {
-        var job = db.Tasks.Where(x => x.TaskId == jobId).Single();
-        job.AssignedResources.Add(new AssignedResource() { TaskId = jobId, ResourceId = resourceId });
+    public void AssignJobToResource(Guid taskId, IEnumerable<Guid> resourceIds) {
+      ExecuteWithContext((db) => {
+        db.DeferredLoadingEnabled = false;
+
+        List<AssignedResource> assignedResources = new List<AssignedResource>();
+        foreach (Guid rId in resourceIds) {
+          assignedResources.Add(new AssignedResource() { TaskId = taskId, ResourceId = rId });
+        }
+        db.AssignedResources.InsertAllOnSubmit(assignedResources);
         db.SubmitChanges();
       });
@@ -608 +681 @@
 
     public IEnumerable<DT.Resource> GetAssignedResources(Guid jobId) {
-      return ExecuteWithContext <IEnumerable<DT.Resource>>((db) => {
+      return ExecuteWithContext<IEnumerable<DT.Resource>>((db) => {
         var job = db.Tasks.Where(x => x.TaskId == jobId).Single();
         return job.AssignedResources.Select(x => DT.Convert.ToDto(x.Resource)).ToArray();
@@ -636 +709 @@
     public IEnumerable<DT.Resource> GetChildResources(Guid resourceId) {
       return ExecuteWithContext<IEnumerable<DT.Resource>>((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(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;
     }
 
@@ -661 +738 @@
     #region ResourcePermission Methods
     public DT.ResourcePermission GetResourcePermission(Guid resourceId, Guid grantedUserId) {
-      return ExecuteWithContext <DT.ResourcePermission> ((db) => {
+      return ExecuteWithContext<DT.ResourcePermission>((db) => {
         return DT.Convert.ToDto(db.ResourcePermissions.SingleOrDefault(x => x.ResourceId == resourceId && x.GrantedUserId == grantedUserId));
       });
@@ -734 +811 @@
         if (entity != null) {
           entity.LastCleanup = datetime;
-        }
-        else {
+        } else {
           entity = new Lifecycle();
           entity.LifecycleId = 0; // always only one entry with ID:0
@@ -825 +901 @@
         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);
+      }
     }
 
@@ -904 +1020 @@
     #endregion
 
+    #region UserPriority Methods
+    public IEnumerable<DT.UserPriority> GetUserPriorities(Expression<Func<UserPriority, bool>> predicate) {
+      using (var db = CreateContext()) {
+        return db.UserPriorities.Where(predicate).Select(x => DT.Convert.ToDto(x)).ToArray();
+      }
+    }
+
+    public void EnqueueUserPriority(DT.UserPriority dto) {
+      using (var db = CreateContext()) {
+        var entity = db.UserPriorities.FirstOrDefault(x => x.UserId == dto.Id);
+        if (entity == null) db.UserPriorities.InsertOnSubmit(DT.Convert.ToEntity(dto));
+        else DT.Convert.ToEntity(dto, entity);
+        db.SubmitChanges();
+      }
+    }
+    #endregion
+
     #region Helpers
     private void CollectChildTasks(HiveDataContext db, Guid parentTaskId, List<Task> collection) {

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