Changeset 9257

Timestamp:

02/28/13 13:57:49 (11 years ago)

Author:

ascheibe

Message:

#2019

• added missing transactions in the Hive service
• split scheduling transaction into smaller transactions
• improved speed of job uploading (AddTask)
• changed highest isolation level from Serializable to RepeatableRead as phantom reads shouldn't be a problem

Location:

trunk/sources

Files:

• HeuristicLab.Services.Hive.DataAccess/3.3/TransactionManager.cs (modified) (4 diffs)
• HeuristicLab.Services.Hive/3.3/HiveDao.cs (modified) (1 diff)
• HeuristicLab.Services.Hive/3.3/HiveService.cs (modified) (13 diffs)
• HeuristicLab.Services.Hive/3.3/Interfaces/IHiveDao.cs (modified) (1 diff)
• HeuristicLab.Services.Hive/3.3/Manager/EventManager.cs (modified) (1 diff)
• HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs (modified) (8 diffs)

trunk/sources/HeuristicLab.Services.Hive.DataAccess/3.3/TransactionManager.cs

@@ -22 +22 @@
 using System;
 using System.Transactions;
-using HeuristicLab.Services.Hive.DataAccess;
 
-namespace HeuristicLab.Services.Hive.DataAccess {  
-  public class TransactionManager : ITransactionManager {    
-    public void UseTransaction(Action call, bool serializable = false, bool longRunning = false) {
+namespace HeuristicLab.Services.Hive.DataAccess {
+  public class TransactionManager : ITransactionManager {
+    public void UseTransaction(Action call, bool repeatableRead = false, bool longRunning = false) {
       int n = 10;
       while (n > 0) {
-        TransactionScope transaction = CreateTransaction(serializable, longRunning);
+        TransactionScope transaction = CreateTransaction(repeatableRead, longRunning);
         try {
           call();
@@ -37 +36 @@
         catch (System.Data.SqlClient.SqlException e) {
           n--; // probably deadlock situation, let it roll back and repeat the transaction n times
-          LogFactory.GetLogger(typeof(TransactionManager).Namespace).Log(string.Format("Exception occured, repeating transaction {0} more times. Details: {1}", n, e.ToString()));          
+          LogFactory.GetLogger(typeof(TransactionManager).Namespace).Log(string.Format("Exception occured, repeating transaction {0} more times. Details: {1}", n, e.ToString()));
           if (n <= 0) throw;
         }
@@ -46 +45 @@
     }
 
-    public T UseTransaction<T>(Func<T> call, bool serializable = false, bool longRunning = false) {
+    public T UseTransaction<T>(Func<T> call, bool repeatableRead = false, bool longRunning = false) {
       int n = 10;
       while (n > 0) {
-        TransactionScope transaction = CreateTransaction(serializable, longRunning);
+        TransactionScope transaction = CreateTransaction(repeatableRead, longRunning);
         try {
           T result = call();
@@ -68 +67 @@
     }
 
-    private TransactionScope CreateTransaction(bool serializable, bool longRunning) {
+    private TransactionScope CreateTransaction(bool repeatableRead, bool longRunning) {
       var options = new TransactionOptions();
-      if (serializable)
-        options.IsolationLevel = IsolationLevel.Serializable;
+      if (repeatableRead)
+        options.IsolationLevel = IsolationLevel.RepeatableRead;
       else
         options.IsolationLevel = IsolationLevel.ReadUncommitted;

trunk/sources/HeuristicLab.Services.Hive/3.3/HiveDao.cs

@@ -613 +613 @@
     }
 
-    public void AssignJobToResource(Guid jobId, Guid resourceId) {
-      using (var db = CreateContext()) {
-        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) {
+      using (var db = CreateContext()) {
+        var task = db.Tasks.Where(x => x.TaskId == taskId).Single();
+        foreach (Guid rId in resourceIds) {
+          task.AssignedResources.Add(new AssignedResource() { TaskId = taskId, ResourceId = rId });
+        }
         db.SubmitChanges();
       }

trunk/sources/HeuristicLab.Services.Hive/3.3/HiveService.cs

@@ -67 +67 @@
         taskData.TaskId = task.Id;
         taskData.LastUpdate = DateTime.Now;
-        foreach (Guid slaveGroupId in resourceIds) {
-          dao.AssignJobToResource(task.Id, slaveGroupId);
-        }
+        dao.AssignJobToResource(task.Id, resourceIds);
         dao.AddTaskData(taskData);
         dao.UpdateTaskState(task.Id, DA.TaskState.Waiting, null, userManager.CurrentUserId, null);
@@ -95 +93 @@
     public IEnumerable<Task> GetTasks() {
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client);
-      var tasks = dao.GetTasks(x => true);
-      foreach (var task in tasks)
-        author.AuthorizeForTask(task.Id, Permission.Read);
-      return tasks;
+      return trans.UseTransaction(() => {
+        var tasks = dao.GetTasks(x => true);
+        foreach (var task in tasks)
+          author.AuthorizeForTask(task.Id, Permission.Read);
+        return tasks;
+      });
     }
 
@@ -144 +144 @@
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client, HiveRoles.Slave);
       author.AuthorizeForTask(taskId, Permission.Read);
-      return dao.GetTaskData(taskId);
+
+      return trans.UseTransaction(() => {
+        return dao.GetTaskData(taskId);
+      });
     }
 
@@ -150 +153 @@
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client, HiveRoles.Slave);
       author.AuthorizeForTask(taskDto.Id, Permission.Full);
+
       trans.UseTransaction(() => {
         dao.UpdateTaskAndPlugins(taskDto);
@@ -158 +162 @@
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client, HiveRoles.Slave);
       author.AuthorizeForTask(task.Id, Permission.Full);
-      author.AuthorizeForTask(taskData.TaskId, Permission.Full);
-      //trans.UseTransaction(() => { // cneumuel: try without transaction
-      taskData.LastUpdate = DateTime.Now;
-      dao.UpdateTaskAndPlugins(task);
-      dao.UpdateTaskData(taskData);
-      //}, false, true);
+
+      trans.UseTransaction(() => {
+        dao.UpdateTaskAndPlugins(task);
+      });
+
+      trans.UseTransaction(() => {
+        taskData.LastUpdate = DateTime.Now;
+        dao.UpdateTaskData(taskData);
+      });
     }
 
@@ -415 +422 @@
       authen.AuthenticateForAnyRole(HiveRoles.Slave);
 
-      List<MessageContainer> result = trans.UseTransaction(() => heartbeatManager.ProcessHeartbeat(heartbeat));
+      List<MessageContainer> result = new List<MessageContainer>();
+      try {
+        result = heartbeatManager.ProcessHeartbeat(heartbeat);
+      }
+      catch (Exception ex) {
+        DA.LogFactory.GetLogger(this.GetType().Namespace).Log("Exception processing Heartbeat: " + ex.ToString());
+      }
 
       if (HeuristicLab.Services.Hive.Properties.Settings.Default.TriggerEventManagerInHeartbeat) {
@@ -449 +462 @@
     public Plugin GetPlugin(Guid pluginId) {
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client, HiveRoles.Slave);
-      return dao.GetPlugin(pluginId);
+      return trans.UseTransaction(() => {
+        return dao.GetPlugin(pluginId);
+      });
     }
 
     public Plugin GetPluginByHash(byte[] hash) {
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client, HiveRoles.Slave);
-      return dao.GetPlugins(x => x.Hash == hash).FirstOrDefault();
+      return trans.UseTransaction(() => {
+        return dao.GetPlugins(x => x.Hash == hash).FirstOrDefault();
+      });
     }
 
@@ -461 +478 @@
     public IEnumerable<Plugin> GetPlugins() {
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client, HiveRoles.Slave);
-      return dao.GetPlugins(x => x.Hash != null);
+      return trans.UseTransaction(() => {
+        return dao.GetPlugins(x => x.Hash != null);
+      });
     }
 
@@ -477 +496 @@
     public void DeletePlugin(Guid pluginId) {
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client, HiveRoles.Slave);
-      dao.DeletePlugin(pluginId);
+      trans.UseTransaction(() => {
+        dao.DeletePlugin(pluginId);
+      });
     }
     #endregion
@@ -516 +537 @@
     #region Resource Methods
     public IEnumerable<Resource> GetChildResources(Guid resourceId) {
-      return dao.GetChildResources(resourceId);
+      return trans.UseTransaction(() => { return dao.GetChildResources(resourceId); });
     }
     #endregion
@@ -523 +544 @@
     public int GetNewHeartbeatInterval(Guid slaveId) {
       authen.AuthenticateForAnyRole(HiveRoles.Slave);
-      Slave s = dao.GetSlave(slaveId);
+
+      Slave s = trans.UseTransaction(() => { return dao.GetSlave(slaveId); });
       if (s != null) {
         return s.HbInterval;
@@ -543 +565 @@
     public Slave GetSlave(Guid slaveId) {
       authen.AuthenticateForAnyRole(HiveRoles.Administrator);
-      return dao.GetSlave(slaveId);
+      return trans.UseTransaction(() => { return dao.GetSlave(slaveId); });
     }
 
     public SlaveGroup GetSlaveGroup(Guid slaveGroupId) {
       authen.AuthenticateForAnyRole(HiveRoles.Administrator);
-      return dao.GetSlaveGroup(slaveGroupId);
+      return trans.UseTransaction(() => { return dao.GetSlaveGroup(slaveGroupId); });
     }
 
     public IEnumerable<Slave> GetSlaves() {
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client);
-      return dao.GetSlaves(x => true).Where(x => x.OwnerUserId == null
-                                         || x.OwnerUserId == userManager.CurrentUserId
-                                         || userManager.VerifyUser(userManager.CurrentUserId, GetResourcePermissions(x.Id).Select(y => y.GrantedUserId).ToList())
-                                         || authen.IsInRole(HiveRoles.Administrator)).ToArray();
+      return trans.UseTransaction(() => {
+        return dao.GetSlaves(x => true).Where(x => x.OwnerUserId == null
+                                           || x.OwnerUserId == userManager.CurrentUserId
+                                           || userManager.VerifyUser(userManager.CurrentUserId, GetResourcePermissions(x.Id).Select(y => y.GrantedUserId).ToList())
+                                           || authen.IsInRole(HiveRoles.Administrator)).ToArray();
+      });
     }
 
     public IEnumerable<SlaveGroup> GetSlaveGroups() {
       authen.AuthenticateForAnyRole(HiveRoles.Administrator, HiveRoles.Client);
-      return dao.GetSlaveGroups(x => true).Where(x => x.OwnerUserId == null
-                                              || x.OwnerUserId == userManager.CurrentUserId
-                                              || userManager.VerifyUser(userManager.CurrentUserId, GetResourcePermissions(x.Id).Select(y => y.GrantedUserId).ToList())
-                                              || authen.IsInRole(HiveRoles.Administrator)).ToArray();
+      return trans.UseTransaction(() => {
+        return dao.GetSlaveGroups(x => true).Where(x => x.OwnerUserId == null
+                                                || x.OwnerUserId == userManager.CurrentUserId
+                                                || userManager.VerifyUser(userManager.CurrentUserId, GetResourcePermissions(x.Id).Select(y => y.GrantedUserId).ToList())
+                                                || authen.IsInRole(HiveRoles.Administrator)).ToArray();
+      });
     }
 
@@ -717 +743 @@
     #region Statistics Methods
     public IEnumerable<Statistics> GetStatistics() {
-      return dao.GetStatistics(x => true);
+      return trans.UseTransaction(() => { return dao.GetStatistics(x => true); });
     }
     public IEnumerable<Statistics> GetStatisticsForTimePeriod(DateTime from, DateTime to) {
-      return dao.GetStatistics(x => x.Timestamp >= from && x.Timestamp <= to);
+      return trans.UseTransaction(() => { return dao.GetStatistics(x => x.Timestamp >= from && x.Timestamp <= to); });
     }
     #endregion

trunk/sources/HeuristicLab.Services.Hive/3.3/Interfaces/IHiveDao.cs

@@ -114 +114 @@
     void UpdateResource(DT.Resource dto);
     void DeleteResource(Guid id);
-    void AssignJobToResource(Guid jobId, Guid resourceId);
+    void AssignJobToResource(Guid taskId, IEnumerable<Guid> resourceIds);
     IEnumerable<DT.Resource> GetAssignedResources(Guid jobId);
     IEnumerable<DT.Resource> GetParentResources(Guid resourceId);

trunk/sources/HeuristicLab.Services.Hive/3.3/Manager/EventManager.cs

@@ -50 +50 @@
         SetTimeoutTasksWaiting();
         DeleteObsoleteSlaves();
-      }, true);
+      });
 
       trans.UseTransaction(() => {
         FinishParentTasks();
         UpdateStatistics();
-      }, false);
+      });
     }
 

trunk/sources/HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs

@@ -37 +37 @@
       get { return ServiceLocator.Instance.TaskScheduler; }
     }
+    private DataAccess.ITransactionManager trans {
+      get { return ServiceLocator.Instance.TransactionManager; }
+    }
 
     /// <summary>
@@ -44 +47 @@
     public List<MessageContainer> ProcessHeartbeat(Heartbeat heartbeat) {
       List<MessageContainer> actions = new List<MessageContainer>();
-      Slave slave = dao.GetSlave(heartbeat.SlaveId);
+      Slave slave = null;
+      slave = trans.UseTransaction(() => { return dao.GetSlave(heartbeat.SlaveId); });
+
       if (slave == null) {
         actions.Add(new MessageContainer(MessageContainer.MessageType.SayHello));
@@ -62 +67 @@
         slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0) ? SlaveState.Calculating : SlaveState.Idle;
         slave.LastHeartbeat = DateTime.Now;
-        dao.UpdateSlave(slave);
+
+        trans.UseTransaction(() => { dao.UpdateSlave(slave); });
 
         // update task data
@@ -76 +82 @@
               DA.LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling could not be aquired.");
             else {
-              var availableTasks = taskScheduler.Schedule(dao.GetWaitingTasks(slave));
+              IEnumerable<TaskInfoForScheduler> availableTasks = null;
+              availableTasks = trans.UseTransaction(() => { return taskScheduler.Schedule(dao.GetWaitingTasks(slave)); });
               if (availableTasks.Any()) {
                 var task = availableTasks.First();
@@ -99 +106 @@
 
     private void AssignJob(Slave slave, Guid taskId) {
-      var task = dao.UpdateTaskState(taskId, DataAccess.TaskState.Transferring, slave.Id, null, null);
+      trans.UseTransaction(() => {
+        var task = dao.UpdateTaskState(taskId, DataAccess.TaskState.Transferring, slave.Id, 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);
+      });
     }
 
@@ -121 +130 @@
         // process the jobProgresses
         foreach (var jobProgress in heartbeat.JobProgress) {
-          Task curTask = dao.GetTask(jobProgress.Key);
+          Task curTask = null;
+          curTask = trans.UseTransaction(() => { return dao.GetTask(jobProgress.Key); });
           if (curTask == null) {
             // task does not exist in db
@@ -150 +160 @@
                   break;
               }
-              dao.UpdateTask(curTask);
+              trans.UseTransaction(() => { dao.UpdateTask(curTask); });
             }
           }
@@ -159 +169 @@
 
     private bool TaskIsAllowedToBeCalculatedBySlave(Guid slaveId, Task curTask) {
-      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));
+      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 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 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 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);
+      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); });
     }
   }