source: branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Services.Hive/3.4/HeartbeatManager.cs @ 6269

using System;
using System.Collections.Generic;
using System.Linq;
using HeuristicLab.Services.Hive.Common;
using HeuristicLab.Services.Hive.Common.DataTransfer;

namespace HeuristicLab.Services.Hive {
  public class HeartbeatManager {
    private DataAccess.IHiveDao dao {
      get { return ServiceLocator.Instance.HiveDao; }
    }
    private HeuristicLab.Services.Hive.DataAccess.TransactionManager trans {
      get { return ServiceLocator.Instance.TransactionManager; }
    }
    private IAuthorizationManager auth {
      get { return ServiceLocator.Instance.AuthorizationManager; }
    }

    /// <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 {
        // update slave data
        slave.FreeCores = heartbeat.FreeCores;
        slave.FreeMemory = heartbeat.FreeMemory;
        slave.CpuUtilization = heartbeat.CpuUtilization;
        slave.IsAllowedToCalculate = true; // Todo: look into calendar
        slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0) ? SlaveState.Calculating : SlaveState.Idle;
        slave.LastHeartbeat = DateTime.Now;
        dao.UpdateSlave(slave);

        // update job data
        actions.AddRange(UpdateJobs(heartbeat));

        // assign new job
        if (heartbeat.AssignJob && 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);
          }
        }
      }
      return actions;
    }

    private void AssignJob(Slave slave, Job job) {
      job = dao.UpdateJobState(job.Id, JobState.Transferring, slave.Id, null, null);
      dao.UpdateSlave(slave);

      // from now on the job has some time to send the next heartbeat (ApplicationConstants.TransferringJobHeartbeatTimeout)
      job.LastHeartbeat = DateTime.Now;
      dao.UpdateJob(job);
    }

    /// <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));
          LogFactory.GetLogger(this.GetType().Namespace).Log("Job does not exist in DB: " + jobProgress.Key);
        } else {
          if (curJob.CurrentStateLog.SlaveId == Guid.Empty || curJob.CurrentStateLog.SlaveId != heartbeat.SlaveId) {
            // assigned slave does not match heartbeat
            actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
            LogFactory.GetLogger(this.GetType().Namespace).Log("The slave " + heartbeat.SlaveId + " is not supposed to calculate Job: " + curJob);
          } else if (!JobIsAllowedToBeCalculatedBySlave(heartbeat.SlaveId, curJob)) {
            // assigned resources ids of job do not match with slaveId (and parent resourceGroupIds); this might happen when slave is moved to different group
            actions.Add(new MessageContainer(MessageContainer.MessageType.PauseJob, curJob.Id));
          } else {
            // save job execution time
            curJob.ExecutionTime = jobProgress.Value;
            curJob.LastHeartbeat = DateTime.Now;

            switch (curJob.Command) {
              case Command.Stop:
                actions.Add(new MessageContainer(MessageContainer.MessageType.StopJob, curJob.Id));
                break;
              case Command.Pause:
                actions.Add(new MessageContainer(MessageContainer.MessageType.PauseJob, curJob.Id));
                break;
              case Command.Abort:
                actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
                break;
            }
            dao.UpdateJob(curJob);
          }
        }
      }
      return actions;
    }

    private bool JobIsAllowedToBeCalculatedBySlave(Guid slaveId, Job curJob) {
      var assignedResourceIds = dao.GetAssignedResources(curJob.Id).Select(x => x.Id);
      var slaveResourceIds = dao.GetParentResources(slaveId).Select(x => x.Id);
      return assignedResourceIds.Any(x => slaveResourceIds.Contains(x));
    }
  }
}