Changeset 5093 for branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.Hive.Server.Core/3.3/SlaveCommunicator.cs

Timestamp:

12/13/10 14:13:15 (13 years ago)

Author:

cneumuel

Message:

#1260

• moved all state-information into lifecycleManager
• changed isolation level for transactions to ReadCommited
• made currentlyFetching-status on slave more rubust
• made LogServiceReader more rubust

File:

• branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.Hive.Server.Core/3.3/SlaveCommunicator.cs (modified) (21 diffs)

branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.Hive.Server.Core/3.3/SlaveCommunicator.cs

@@ -36 +36 @@
 using HeuristicLab.Hive.Tracing;
 using HeuristicLab.PluginInfrastructure.Manager;
+using System.Web;
+using System.Web.SessionState;
 
 namespace HeuristicLab.Hive.Server.Core {
@@ -42 +44 @@
   /// </summary>
   public class SlaveCommunicator : ISlaveCommunicator, IInternalSlaveCommunicator {
-    private static Dictionary<Guid, DateTime> lastHeartbeats = new Dictionary<Guid, DateTime>();
-
-    /// <summary>
-    /// Contains a list job JobIds which have been sent to a slave, but the slave has not yet sent 
-    /// a jobProgress of the job with a heartbeat, because he has not finished downloading/deserializing it.
-    /// The number value indicates a TimeToLive count that is decremented with each server-heartbeat.
-    /// When the number reaches zero, the jobs is assumed to be lost and is set Offline again.
-    /// </summary>
-    private static Dictionary<Guid, int> newAssignedJobs = new Dictionary<Guid, int>();
-
-    /// <summary>
-    /// When a slave reconnects and he has finished results waiting it calls IsJobStillNeeded. If the finished 
-    /// result has not yet been collected from anywhere else, the job will be sent by the slave and the job state is set to Pending.
-    /// Now the job be in pending state until it is received from the reconnected slave or the TimeToLive value of this dictionary has reached zero.
-    /// </summary>
-    private static Dictionary<Guid, int> pendingJobs = new Dictionary<Guid, int>();
     private static int PENDING_TIMEOUT = 100;
 
@@ -74 +60 @@
     /// </summary>
     public SlaveCommunicator() {
+      Logger.Debug("ServiceCommunicator instantiated");
       lifecycleManager = ServiceLocator.GetLifecycleManager();
       jobManager = ServiceLocator.GetJobManager() as IInternalJobManager;
@@ -97 +84 @@
             SlaveDto slave = DaoLocator.SlaveDao.FindById(slaveId);
 
-            if (!lastHeartbeats.ContainsKey(slave.Id)) {
-              Logger.Info("No previous hearbeats are available for " + slave.Id + " although it is in state " + slave.State);
+            if (!lifecycleManager.LastHeartbeats.ContainsKey(slave.Id)) {
+              Logger.Info("No previous hearbeats are available for " + slave.Name + "(" + slave.Id + "), although it is in state " + slave.State);
 
               // add a heartbeat NOW and give the slave time to say something for HEARTBEAT_MAX_DIF
               // otherwise alls the slaves jobs would be aborted, which is not desirable if the server has just been restarted
               heartbeatLock.EnterWriteLock();
-              lastHeartbeats.Add(slave.Id, DateTime.Now);
+              lifecycleManager.LastHeartbeats.Add(slave.Id, DateTime.Now);
               heartbeatLock.ExitWriteLock();
             } else {
-              DateTime lastHeartbeatOfSlave = lastHeartbeats[slave.Id];
+              DateTime lastHeartbeatOfSlave = lifecycleManager.LastHeartbeats[slave.Id];
 
               TimeSpan diff = DateTime.Now.Subtract(lastHeartbeatOfSlave);
@@ -112 +99 @@
               if (diff.TotalSeconds > ApplicationConstants.HEARTBEAT_MAX_DIF) {
                 // if slave calculated jobs, the job must be reset
-                Logger.Info("Slave timed out and is on RESET");
+                Logger.Info("Slave timed out and is on RESET (no message for " + diff.TotalSeconds + " seconds.)");
                 foreach (JobDto job in DaoLocator.JobDao.FindActiveJobsOfSlave(slave)) {
                   DaoLocator.JobDao.SetJobOffline(job);
-                  lock (newAssignedJobs) {
-                    if (newAssignedJobs.ContainsKey(job.Id))
-                      newAssignedJobs.Remove(job.Id);
+                  lock (lifecycleManager.NewAssignedJobs) {
+                    if (lifecycleManager.NewAssignedJobs.ContainsKey(job.Id))
+                      lifecycleManager.NewAssignedJobs.Remove(job.Id);
                   }
                 }
@@ -127 +114 @@
                 Logger.Debug("removing it from the heartbeats list");
                 heartbeatLock.EnterWriteLock();
-                lastHeartbeats.Remove(slave.Id);
+                lifecycleManager.LastHeartbeats.Remove(slave.Id);
                 heartbeatLock.ExitWriteLock();
               }
@@ -148 +135 @@
 
       foreach (JobDto currJob in pendingJobsInDb) {
-        lock (pendingJobs) {
-          if (pendingJobs.ContainsKey(currJob.Id)) {
-            if (pendingJobs[currJob.Id] <= 0) {
+        lock (lifecycleManager.PendingJobs) {
+          if (lifecycleManager.PendingJobs.ContainsKey(currJob.Id)) {
+            if (lifecycleManager.PendingJobs[currJob.Id] <= 0) {
               currJob.State = JobState.Offline;
               DaoLocator.JobDao.Update(currJob);
             } else {
-              pendingJobs[currJob.Id]--;
+              lifecycleManager.PendingJobs[currJob.Id]--;
             }
           }
@@ -173 +160 @@
 
       heartbeatLock.EnterWriteLock();
-      if (lastHeartbeats.ContainsKey(slave.Id)) {
-        lastHeartbeats[slave.Id] = DateTime.Now;
+      if (lifecycleManager.LastHeartbeats.ContainsKey(slave.Id)) {
+        lifecycleManager.LastHeartbeats[slave.Id] = DateTime.Now;
       } else {
-        lastHeartbeats.Add(slave.Id, DateTime.Now);
+        lifecycleManager.LastHeartbeats.Add(slave.Id, DateTime.Now);
       }
       heartbeatLock.ExitWriteLock();
@@ -247 +234 @@
         using (TransactionScope scope = new TransactionScope(TransactionScopeOption.Required, new TransactionOptions { IsolationLevel = ApplicationConstants.ISOLATION_LEVEL_SCOPE })) {
           SlaveDto slave = UpdateSlaveData(heartbeatData);
+          DaoLocator.SlaveDao.Update(slave);
+
           SaveTimestamp(heartbeatData);
 
           //ProcessJobProgress(heartbeatData, response);
           response.ActionRequest = ProcessJobProgress(heartbeatData);
-
+          
           //check if new Cal must be loaded
           if (slave.CalendarSyncStatus == CalendarState.Fetch || slave.CalendarSyncStatus == CalendarState.ForceFetch) {
@@ -270 +259 @@
           }
 
-          DaoLocator.SlaveDao.Update(slave);
           scope.Complete();
         }
@@ -286 +274 @@
     }
 
-    private static void SaveTimestamp(HeartBeatData heartbeatData) {
+    private void SaveTimestamp(HeartBeatData heartbeatData) {
       heartbeatLock.EnterWriteLock();
-      if (lastHeartbeats.ContainsKey(heartbeatData.SlaveId)) {
-        lastHeartbeats[heartbeatData.SlaveId] = DateTime.Now;
+      if (lifecycleManager.LastHeartbeats.ContainsKey(heartbeatData.SlaveId)) {
+        lifecycleManager.LastHeartbeats[heartbeatData.SlaveId] = DateTime.Now;
       } else {
-        lastHeartbeats.Add(heartbeatData.SlaveId, DateTime.Now);
+        lifecycleManager.LastHeartbeats.Add(heartbeatData.SlaveId, DateTime.Now);
       }
       heartbeatLock.ExitWriteLock();
@@ -316 +304 @@
 
       // find all the jobs in jobProgress which are not in the database -> they are not supposed to be calculated by this slave
-      IEnumerable<Guid> jobsToAbort = GetJobsNotInDatabase(heartbeatData.SlaveId, heartbeatData.JobProgress.Keys);
-      foreach (Guid jobId in jobsToAbort) {
-        actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, jobId));
-        heartbeatData.JobProgress.Remove(jobId);
-      }
+      //IEnumerable<Guid> jobsToAbort = GetJobsNotCalculatedByThisSlave(heartbeatData.SlaveId, heartbeatData.JobProgress.Keys);
+      //foreach (Guid jobId in jobsToAbort) {
+      //  Logger.Error("Job shall not be caculated by this slave or does not exist in DB: " + jobId);
+      //  actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, jobId));
+      //  heartbeatData.JobProgress.Remove(jobId);
+      //}
 
       // process all the remaining jobProgresses
@@ -327 +316 @@
         if (curJob == null) {
           // job does not exist in db
+          Logger.Error("Job does not exist in DB: " + jobProgress.Key);
           actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, jobProgress.Key));
-          Logger.Error("Job does not exist in DB: " + jobProgress.Key);
         } else {
           curJob.Slave = DaoLocator.SlaveDao.GetSlaveForJob(curJob.Id);
+          Guid id = curJob.Slave != null ? curJob.Slave.Id : Guid.Empty;
           if (curJob.Slave == null || curJob.Slave.Id != heartbeatData.SlaveId) {
             // assigned slave does not match heartbeat
+            Logger.Error("The slave " + heartbeatData.SlaveId + " is not supposed to calculate Job: " + curJob.ToString());
             actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
-            Logger.Error("The slave " + heartbeatData.SlaveId + " is not supposed to calculate Job: " + curJob);
           } else {
             // save job execution time
@@ -341 +331 @@
             if (curJob.State == JobState.Aborted) {
               // a request to abort the job has been set
+              Logger.Error("Job is in state aborted, send AbortJob: " + curJob.Id);
               actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
             } else if (curJob.State == JobState.SnapshotRequested) {
@@ -355 +346 @@
       foreach (JobDto currJob in jobsOfSlave) {
         if (heartbeatData.JobProgress.ContainsKey(currJob.Id)) {
-          lock (newAssignedJobs) {
-            if (newAssignedJobs.ContainsKey(currJob.Id)) {
+          lock (lifecycleManager.NewAssignedJobs) {
+            if (lifecycleManager.NewAssignedJobs.ContainsKey(currJob.Id)) {
               Logger.Info("Job is sending a heart beat, removing it from the newAssignedJobList: " + currJob);
-              newAssignedJobs.Remove(currJob.Id);
+              lifecycleManager.NewAssignedJobs.Remove(currJob.Id);
             }
           }
         } else {
-          lock (newAssignedJobs) {
-            if (newAssignedJobs.ContainsKey(currJob.Id)) {
-              newAssignedJobs[currJob.Id]--;
-              Logger.Error("Job TTL Reduced by one for job: " + currJob + "and is now: " + newAssignedJobs[currJob.Id] + ". User that sucks: " + currJob.Slave);
-              if (newAssignedJobs[currJob.Id] <= 0) {
+          lock (lifecycleManager.NewAssignedJobs) {
+            if (lifecycleManager.NewAssignedJobs.ContainsKey(currJob.Id)) {
+              lifecycleManager.NewAssignedJobs[currJob.Id]--;
+              Logger.Error("Job TTL Reduced by one for job: " + currJob + "and is now: " + lifecycleManager.NewAssignedJobs[currJob.Id] + ". User that sucks: " + currJob.Slave);
+              if (lifecycleManager.NewAssignedJobs[currJob.Id] <= 0) {
                 Logger.Error("Job TTL reached Zero, Job gets removed: " + currJob + " and set back to offline. User that sucks: " + currJob.Slave);
 
@@ -374 +365 @@
                 actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, currJob.Id));
 
-                newAssignedJobs.Remove(currJob.Id);
+                lifecycleManager.NewAssignedJobs.Remove(currJob.Id);
               }
             } else {
@@ -388 +379 @@
 
     /// <summary>
-    /// Returns the jobIds of the jobs which are not assigned to this slave in the database
-    /// </summary>
-    private IEnumerable<Guid> GetJobsNotInDatabase(Guid slaveId, IEnumerable<Guid> jobIds) {
+    /// Returns the jobIds of the jobs which are not assigned to this slave in the database (they either are not stored in DB or they are assigned
+    /// </summary>
+    private IEnumerable<Guid> GetJobsNotCalculatedByThisSlave(Guid slaveId, IEnumerable<Guid> jobIds) {
       IEnumerable<Guid> activeJobsOfSlave = DaoLocator.JobDao.FindActiveJobsOfSlave(DaoLocator.SlaveDao.FindById(slaveId)).Select(j => j.Id);
       return jobIds.Except(activeJobsOfSlave).ToList();
@@ -520 +511 @@
 
         Logger.Info("Job pulled: " + job2Calculate + " for user " + slaveId);
-        lock (newAssignedJobs) {
-          if (!newAssignedJobs.ContainsKey(job2Calculate.Id))
-            newAssignedJobs.Add(job2Calculate.Id, ApplicationConstants.JOB_TIME_TO_LIVE);
+        lock (lifecycleManager.NewAssignedJobs) {
+          if (!lifecycleManager.NewAssignedJobs.ContainsKey(job2Calculate.Id))
+            lifecycleManager.NewAssignedJobs.Add(job2Calculate.Id, ApplicationConstants.JOB_TIME_TO_LIVE);
         }
       } else {
@@ -701 +692 @@
 
       heartbeatLock.EnterWriteLock();
-      if (lastHeartbeats.ContainsKey(slaveId))
-        lastHeartbeats.Remove(slaveId);
+      if (lifecycleManager.LastHeartbeats.ContainsKey(slaveId))
+        lifecycleManager.LastHeartbeats.Remove(slaveId);
       heartbeatLock.ExitWriteLock();
 
@@ -745 +736 @@
       }
       job.State = JobState.Pending;
-      lock (pendingJobs) {
-        pendingJobs.Add(job.Id, PENDING_TIMEOUT);
+      lock (lifecycleManager.PendingJobs) {
+        lifecycleManager.PendingJobs.Add(job.Id, PENDING_TIMEOUT);
       }
 
@@ -765 +756 @@
         if (ipd != null) {
           response.List.Add(ConvertPluginDescriptorToDto(ipd));
-        } else {
-          response.StatusMessage = ResponseStatus.GetPlugins_PluginsNotAvailable;
-          return response;
         }
       }