Changeset 5405 for branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Services.Hive

Timestamp:

02/01/11 18:12:46 (13 years ago)

Author:

cneumuel

Message:

#1233

• moved heartbeat timestamps of slaves and jobs into database to make server stateless
• made slave use the right authentication ("hiveslave" instead of HL username/password)
• moved heartbeat related methods into HeartbeatManager
• changed signature of Service.Hello method, so all hardware related information is transferred in that method withing the Slave-object

Location:

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Services.Hive/3.4

Files:

• HeartbeatManager.cs (added)
• HeuristicLab.Services.Hive-3.4.csproj (modified) (1 diff)
• HiveService.cs (modified) (3 diffs)
• Interfaces/ILifecycleManager.cs (modified) (1 diff)
• Interfaces/IServiceLocator.cs (modified) (1 diff)
• LifecycleManager.cs (modified) (4 diffs)
• ServiceLocator.cs (modified) (1 diff)

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Services.Hive/3.4/HeuristicLab.Services.Hive-3.4.csproj

@@ -112 +112 @@
     <None Include="HeuristicLabServicesHivePlugin.cs.frame" />
     <None Include="Properties\AssemblyInfo.cs.frame" />
+    <Compile Include="HeartbeatManager.cs" />
     <Compile Include="Interfaces\ILifecycleManager.cs" />
     <Compile Include="Interfaces\IServiceLocator.cs" />

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Services.Hive/3.4/HiveService.cs

@@ -25 +25 @@
     private ILifecycleManager lifecycleManager {
       get { return ServiceLocator.Instance.LifecycleManager; }
+    }
+    private HeartbeatManager heartbeatManager {
+      get { return ServiceLocator.Instance.HeartbeatManager; }
     }
 
@@ -162 +165 @@
 
     #region Login Methods
-    public void Hello(Guid slaveId, string name, int cores, int memory) {
-      using (trans.OpenTransaction()) {
-        var slave = dao.GetSlave(slaveId);
+    public void Hello(Slave slaveInfo) {
+      using (trans.OpenTransaction()) {
+        var slave = dao.GetSlave(slaveInfo.Id);
 
         if (slave == null) {
-          slave = new Slave { Id = slaveId, Name = name, Cores = cores, Memory = memory };
-          slave.IsAllowedToCalculate = true; //a little bit to optimistic?
-          slave.SlaveState = SlaveState.Idle;
-          dao.AddSlave(slave);
+          dao.AddSlave(slaveInfo);
         } else {
-          //TODO: error handling? 
+          dao.UpdateSlave(slaveInfo);
         }
       }
@@ -191 +191 @@
     public List<MessageContainer> Heartbeat(Heartbeat heartbeat) {
       using (trans.OpenTransaction()) {
-        return lifecycleManager.ProcessHeartbeat(heartbeat);
+        return heartbeatManager.ProcessHeartbeat(heartbeat);
       }
     }

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Services.Hive/3.4/Interfaces/ILifecycleManager.cs

@@ -14 +14 @@
 
     void Stop();
-
-    List<MessageContainer> ProcessHeartbeat(Heartbeat heartbeat);
   }
 }

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Services.Hive/3.4/Interfaces/IServiceLocator.cs

@@ -12 +12 @@
     ILifecycleManager LifecycleManager { get; }
     TransactionManager TransactionManager { get; }
+    HeartbeatManager HeartbeatManager { get; }
   }
 }

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Services.Hive/3.4/LifecycleManager.cs

@@ -25 +25 @@
 
     private static object locker = new object();
-    private Dictionary<Guid, DateTime> heartbeats = new Dictionary<Guid, DateTime>();
 
     // Windows-Forms timer is single threaded, so callbacks will be synchron
-    System.Windows.Forms.Timer timer = new System.Windows.Forms.Timer();
+    System.Windows.Forms.Timer timer;
 
-    /// <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 Dictionary<Guid, int> newlyAssignedJobs = new Dictionary<Guid, int>();
-
-    /// <summary>
-    /// Counts how many jobs are currently beeing transferred.
-    /// </summary>
-    private int jobsCurrentlyTransfering = 0;
-    public int JobsCurrentlyTransferring {
-      get { return jobsCurrentlyTransfering; }
-      set {
-        if (jobsCurrentlyTransfering != value) {
-          jobsCurrentlyTransfering = value;
-          Logger.Info("JobsCurrentlyTransfering: " + jobsCurrentlyTransfering);
-        }
-      }
+    public ExecutionState ExecutionState {
+      get { return timer.Enabled ? ExecutionState.Started : Core.ExecutionState.Stopped; }
     }
 
-    public ExecutionState ExecutionState {
-      get { return timer.Enabled ? ExecutionState.Started : Core.ExecutionState.Stopped;  }
+    public LifecycleManager() {
+      this.timer = new System.Windows.Forms.Timer();
+      this.timer.Tick += new EventHandler(timer_Tick);
     }
-
-    public LifecycleManager() { }
 
     public void Start() {
       if (ExecutionState == Core.ExecutionState.Stopped) {
         this.timer.Interval = (int)new TimeSpan(0, 0, 10).TotalMilliseconds;
-        this.timer.Tick += new EventHandler(timer_Tick);
         this.timer.Start();
-        AddAllSlavesToHeartbeats();
       }
     }
@@ -74 +52 @@
 
     /// <summary>
-    // add all slaves to hearbeats-collection and give them some time to say hello (HEARTBEAT_TIMEOUT)
-    // otherwise alls the slaves jobs would be aborted immediately, which is not desirable if the server has just been restarted
-    /// </summary>
-    private void AddAllSlavesToHeartbeats() {
-      lock (locker) {
-        using (trans.OpenTransaction()) {
-          Guid[] slaveIds = dao.GetSlaves(x => true).Select(s => s.Id).ToArray();
-          foreach (Guid slaveId in slaveIds) {
-            if (!heartbeats.ContainsKey(slaveId)) {
-              heartbeats.Add(slaveId, DateTime.Now);
-            }
-          }
-        }
-      }
-    }
-
-    /// <summary>
     /// This method is supposed to check if slaves are online
     /// if not -> set them offline and check if they where calculating a job
@@ -97 +58 @@
       lock (locker) {
         using (trans.OpenTransaction()) {
-          Guid[] slaveIds = dao.GetSlaves(x => true).Select(s => s.Id).ToArray();
-          foreach (Guid slaveId in slaveIds) {
-            if (SlaveTimedOut(slaveId)) {
-              var slave = dao.GetSlave(slaveId);
-              if (slave.SlaveState != SlaveState.Offline) {
-                AbortJobs(slaveId);
-                slave.SlaveState = SlaveState.Offline;
-                dao.UpdateSlave(slave);
-              }
-              heartbeats.Remove(slaveId);
+          var slaves = dao.GetSlaves(x => x.SlaveState != SlaveState.Offline);
+          foreach (Slave slave in slaves) {
+            if (!slave.LastHeartbeat.HasValue || (DateTime.Now - slave.LastHeartbeat.Value).TotalSeconds > ApplicationConstants.HeartbeatTimeout) {
+              slave.SlaveState = SlaveState.Offline;
+              AbortJobs(slave.Id);
+              dao.UpdateSlave(slave);
             }
           }
         }
       }
-    }
-
-    private bool SlaveTimedOut(Guid slaveId) {
-      if (!heartbeats.ContainsKey(slaveId))
-        return true;
-
-      if ((DateTime.Now - heartbeats[slaveId]).TotalSeconds > ApplicationConstants.HeartbeatTimeout) {
-        return true;
-      }
-
-      return false;
     }
 
@@ -131 +77 @@
       }
     }
-
-    /// <summary>
-    /// This method will be called every time a slave sends a heartbeat (-> very often; concurrency is important!)
-    /// </summary>
-    /// <returns>a list of actions the slave should do</returns>
-    public List<MessageContainer> ProcessHeartbeat(Heartbeat heartbeat) {
-      List<MessageContainer> actions = new List<MessageContainer>();
-      Slave slave = dao.GetSlave(heartbeat.SlaveId);
-      if (slave == null) {
-        actions.Add(new MessageContainer(MessageContainer.MessageType.SayHello));
-      } else {
-        heartbeats[heartbeat.SlaveId] = DateTime.Now;
-        actions.AddRange(UpdateJobs(heartbeat));
-
-        if (this.IsAllowedToSendJobs() && slave.IsAllowedToCalculate && heartbeat.FreeCores > 0) {
-          var availableJobs = dao.GetWaitingJobs(slave, 1);
-          if (availableJobs.Count() > 0) {
-            var job = availableJobs.First();
-            actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateJob, job.Id));
-            AssignJob(slave, job);
-          }
-        }
-
-        if (slave.FreeCores != heartbeat.FreeCores ||
-            slave.FreeMemory != heartbeat.FreeMemory ||
-            slave.IsAllowedToCalculate != heartbeat.IsAllowedToCalculate ||
-            slave.SlaveState != (heartbeat.JobProgress.Count > 0 ? SlaveState.Calculating : SlaveState.Idle)) { // only update slave when something changed, to avoid unnecessary updates
-          slave.FreeCores = heartbeat.FreeCores;
-          slave.FreeMemory = heartbeat.FreeMemory;
-          slave.IsAllowedToCalculate = heartbeat.IsAllowedToCalculate;
-          slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0) ? SlaveState.Calculating : SlaveState.Idle;
-          dao.UpdateSlave(slave);
-        }
-      }
-      return actions;
-    }
-
-    private void AssignJob(Slave slave, Job job) {
-      job.SlaveId = slave.Id;
-      job.JobState = JobState.Calculating; // Todo: Maybe use State = Transferring (?)
-      job.DateCalculated = DateTime.Now; // Todo: use statelog instead
-      dao.UpdateJob(job);
-      dao.UpdateSlave(slave);
-    }
-
-    /// <summary>
-    /// Update the progress of each job
-    /// Checks if all the jobs sent by heartbeat are supposed to be calculated by this slave
-    /// </summary>
-    private IEnumerable<MessageContainer> UpdateJobs(Heartbeat heartbeat) {
-      List<MessageContainer> actions = new List<MessageContainer>();
-
-      if (heartbeat.JobProgress == null)
-        return actions;
-
-      // process the jobProgresses
-      foreach (var jobProgress in heartbeat.JobProgress) {
-        Job curJob = dao.GetJob(jobProgress.Key);
-        if (curJob == null) {
-          // job does not exist in db
-          actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, jobProgress.Key));
-          Logger.Error("Job does not exist in DB: " + jobProgress.Key);
-        } else {
-          if (curJob.SlaveId == Guid.Empty || curJob.SlaveId != heartbeat.SlaveId) {
-            // assigned slave does not match heartbeat
-            actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
-            Logger.Error("The slave " + heartbeat.SlaveId + " is not supposed to calculate Job: " + curJob);
-          } else {
-            // save job execution time
-            curJob.ExecutionTime = jobProgress.Value;
-
-            if (curJob.JobState == JobState.Aborted) {
-              // a request to abort the job has been set
-              actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
-            }
-            dao.UpdateJob(curJob);
-          }
-        }
-      }
-      return actions;
-    }
-
-    /// <summary>
-    /// Returns true if there are enough resources to send a job
-    /// There should not be too many jobs sent simultaniously
-    /// </summary>
-    private bool IsAllowedToSendJobs() {
-      return JobsCurrentlyTransferring < ApplicationConstants.MaxJobTransferCount;
-    }
   }
 }

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Services.Hive/3.4/ServiceLocator.cs

@@ -48 +48 @@
       }
     }
+
+    private HeartbeatManager heartbeatManager;
+    public HeartbeatManager HeartbeatManager {
+      get {
+        if(heartbeatManager == null) heartbeatManager = new HeartbeatManager();
+        return heartbeatManager;
+      }
+    }
   }
 }