[5405] | 1 | using System;
|
---|
| 2 | using System.Collections.Generic;
|
---|
| 3 | using System.Linq;
|
---|
| 4 | using HeuristicLab.Services.Hive.Common;
|
---|
| 5 | using HeuristicLab.Services.Hive.Common.DataTransfer;
|
---|
| 6 |
|
---|
| 7 | namespace HeuristicLab.Services.Hive {
|
---|
| 8 | public class HeartbeatManager {
|
---|
| 9 | private DataAccess.IHiveDao dao {
|
---|
| 10 | get { return ServiceLocator.Instance.HiveDao; }
|
---|
| 11 | }
|
---|
| 12 | private HeuristicLab.Services.Hive.DataAccess.TransactionManager trans {
|
---|
| 13 | get { return ServiceLocator.Instance.TransactionManager; }
|
---|
| 14 | }
|
---|
| 15 | private IAuthorizationManager auth {
|
---|
| 16 | get { return ServiceLocator.Instance.AuthorizationManager; }
|
---|
| 17 | }
|
---|
| 18 |
|
---|
| 19 | /// <summary>
|
---|
| 20 | /// This method will be called every time a slave sends a heartbeat (-> very often; concurrency is important!)
|
---|
| 21 | /// </summary>
|
---|
| 22 | /// <returns>a list of actions the slave should do</returns>
|
---|
| 23 | public List<MessageContainer> ProcessHeartbeat(Heartbeat heartbeat) {
|
---|
| 24 | List<MessageContainer> actions = new List<MessageContainer>();
|
---|
| 25 | Slave slave = dao.GetSlave(heartbeat.SlaveId);
|
---|
| 26 | if (slave == null) {
|
---|
| 27 | actions.Add(new MessageContainer(MessageContainer.MessageType.SayHello));
|
---|
| 28 | } else {
|
---|
| 29 | // update slave data
|
---|
| 30 | slave.FreeCores = heartbeat.FreeCores;
|
---|
| 31 | slave.FreeMemory = heartbeat.FreeMemory;
|
---|
| 32 | slave.IsAllowedToCalculate = true; // Todo: look into calendar
|
---|
| 33 | slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0) ? SlaveState.Calculating : SlaveState.Idle;
|
---|
| 34 | slave.LastHeartbeat = DateTime.Now;
|
---|
| 35 | dao.UpdateSlave(slave);
|
---|
| 36 |
|
---|
| 37 | // update job data
|
---|
| 38 | actions.AddRange(UpdateJobs(heartbeat));
|
---|
| 39 |
|
---|
| 40 | // assign new job
|
---|
| 41 | if (heartbeat.AssignJob && this.IsAllowedToSendJobs() && slave.IsAllowedToCalculate && heartbeat.FreeCores > 0) {
|
---|
| 42 | var availableJobs = dao.GetWaitingJobs(slave, 1);
|
---|
| 43 | if (availableJobs.Count() > 0) {
|
---|
| 44 | var job = availableJobs.First();
|
---|
| 45 | actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateJob, job.Id));
|
---|
| 46 | AssignJob(slave, job);
|
---|
| 47 | }
|
---|
| 48 | }
|
---|
| 49 | }
|
---|
| 50 | return actions;
|
---|
| 51 | }
|
---|
| 52 |
|
---|
| 53 | private void AssignJob(Slave slave, Job job) {
|
---|
[5636] | 54 | dao.UpdateJobState(job.Id, JobState.Transferring, slave.Id, null, null);
|
---|
[5405] | 55 | dao.UpdateSlave(slave);
|
---|
| 56 | }
|
---|
| 57 |
|
---|
| 58 | /// <summary>
|
---|
| 59 | /// Update the progress of each job
|
---|
| 60 | /// Checks if all the jobs sent by heartbeat are supposed to be calculated by this slave
|
---|
| 61 | /// </summary>
|
---|
| 62 | private IEnumerable<MessageContainer> UpdateJobs(Heartbeat heartbeat) {
|
---|
| 63 | List<MessageContainer> actions = new List<MessageContainer>();
|
---|
| 64 |
|
---|
| 65 | if (heartbeat.JobProgress == null)
|
---|
| 66 | return actions;
|
---|
| 67 |
|
---|
| 68 | // process the jobProgresses
|
---|
| 69 | foreach (var jobProgress in heartbeat.JobProgress) {
|
---|
| 70 | Job curJob = dao.GetJob(jobProgress.Key);
|
---|
| 71 | if (curJob == null) {
|
---|
| 72 | // job does not exist in db
|
---|
| 73 | actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, jobProgress.Key));
|
---|
[5718] | 74 | LogFactory.GetLogger(this.GetType().Namespace).Log("Job does not exist in DB: " + jobProgress.Key);
|
---|
[5405] | 75 | } else {
|
---|
[5511] | 76 | if (curJob.CurrentStateLog.SlaveId == Guid.Empty || curJob.CurrentStateLog.SlaveId != heartbeat.SlaveId) {
|
---|
[5405] | 77 | // assigned slave does not match heartbeat
|
---|
| 78 | actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
|
---|
[5718] | 79 | LogFactory.GetLogger(this.GetType().Namespace).Log("The slave " + heartbeat.SlaveId + " is not supposed to calculate Job: " + curJob);
|
---|
[5405] | 80 | } else {
|
---|
| 81 | // save job execution time
|
---|
| 82 | curJob.ExecutionTime = jobProgress.Value;
|
---|
| 83 | curJob.LastHeartbeat = DateTime.Now;
|
---|
| 84 |
|
---|
[5511] | 85 | if (curJob.State == JobState.Aborted) {
|
---|
[5405] | 86 | // a request to abort the job has been set
|
---|
| 87 | actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
|
---|
| 88 | }
|
---|
| 89 | dao.UpdateJob(curJob);
|
---|
| 90 | }
|
---|
| 91 | }
|
---|
| 92 | }
|
---|
| 93 | return actions;
|
---|
| 94 | }
|
---|
| 95 |
|
---|
| 96 | /// <summary>
|
---|
| 97 | /// Returns true if there are enough resources to send a job
|
---|
| 98 | /// There should not be too many jobs sent simultaniously
|
---|
| 99 | /// </summary>
|
---|
| 100 | private bool IsAllowedToSendJobs() {
|
---|
| 101 | return true; // JobsCurrentlyTransferring < ApplicationConstants.MaxJobTransferCount;
|
---|
| 102 | // Todo: see if unlimited job transfer count works. if not, look into db and count jobs in state Transferring
|
---|
| 103 | }
|
---|
| 104 | }
|
---|
| 105 | }
|
---|