source: branches/HivePerformance/sources/HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs @ 9381

#region License Information
/* HeuristicLab
 * Copyright (C) 2002-2012 Heuristic and Evolutionary Algorithms Laboratory (HEAL)
 *
 * This file is part of HeuristicLab.
 *
 * HeuristicLab is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * HeuristicLab is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with HeuristicLab. If not, see <http://www.gnu.org/licenses/>.
 */
#endregion

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using HeuristicLab.Services.Hive.DataTransfer;
using DA = HeuristicLab.Services.Hive.DataAccess;

namespace HeuristicLab.Services.Hive {
  public class HeartbeatManager {
    private const string MutexName = "HiveTaskSchedulingMutex";

    private IHiveDao dao {
      get { return ServiceLocator.Instance.HiveDao; }
    }
    private ITaskScheduler taskScheduler {
      get { return ServiceLocator.Instance.TaskScheduler; }
    }
    private DataAccess.ITransactionManager trans {
      get { return ServiceLocator.Instance.TransactionManager; }
    }

    /// <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>();

      DA.Slave slave = null;
      trans.UseTransaction(() => {
        slave = dao.GetSlaveDA(heartbeat.SlaveId);

        if (slave == null) {
          actions.Add(new MessageContainer(MessageContainer.MessageType.SayHello));
        } else {
          if (heartbeat.HbInterval != slave.HbInterval) {
            actions.Add(new MessageContainer(MessageContainer.MessageType.NewHBInterval));
          }
          if (ShutdownSlaveComputer(slave.ResourceId)) {
            actions.Add(new MessageContainer(MessageContainer.MessageType.ShutdownComputer));
          }

          // update slave data
          slave.FreeCores = heartbeat.FreeCores;
          slave.FreeMemory = heartbeat.FreeMemory;
          slave.CpuUtilization = heartbeat.CpuUtilization;
          slave.IsAllowedToCalculate = SlaveIsAllowedToCalculate(slave.ResourceId);
          slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0) ? DA.SlaveState.Calculating : DA.SlaveState.Idle;
          slave.LastHeartbeat = DateTime.Now;

          dao.UpdateSlaveDA(slave);
        }
      });

      // update task data
      actions.AddRange(UpdateTasks(heartbeat, slave.IsAllowedToCalculate));

      // assign new task
      if (heartbeat.AssignJob && slave.IsAllowedToCalculate && heartbeat.FreeCores > 0) {
        bool mutexAquired = false;
        var mutex = new Mutex(false, MutexName);
        try {

          mutexAquired = mutex.WaitOne(Properties.Settings.Default.SchedulingPatience);
          if (!mutexAquired)
            DA.LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling could not be aquired.");
          else {
            trans.UseTransaction(() => {
              IEnumerable<TaskInfoForScheduler> availableTasks = null;
              availableTasks = taskScheduler.Schedule(dao.GetWaitingTasksDA(slave));
              if (availableTasks.Any()) {
                var task = availableTasks.First();
                AssignJob(slave, task.TaskId);
                actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateTask, task.TaskId));
              }
            });
          }
        }
        catch (AbandonedMutexException) {
          DA.LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling has been abandoned.");
        }
        catch (Exception ex) {
          DA.LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager threw an exception in ProcessHeartbeat: " + ex.ToString());
        }
        finally {
          if (mutexAquired) mutex.ReleaseMutex();
        }
      }
      return actions;
    }

    private void AssignJob(DA.Slave slave, Guid taskId) {
      var task = dao.UpdateTaskStateDA(taskId, DataAccess.TaskState.Transferring, slave.ResourceId, null, null);

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

    /// <summary>
    /// Update the progress of each task
    /// Checks if all the task sent by heartbeat are supposed to be calculated by this slave
    /// </summary>
    private IEnumerable<MessageContainer> UpdateTasks(Heartbeat heartbeat, bool IsAllowedToCalculate) {
      List<MessageContainer> actions = new List<MessageContainer>();

      if (heartbeat.JobProgress == null)
        return actions;

      if (!IsAllowedToCalculate && heartbeat.JobProgress.Count != 0) {
        actions.Add(new MessageContainer(MessageContainer.MessageType.PauseAll));
      } else {
        trans.UseTransaction(() => {
          // process the jobProgresses
          foreach (var jobProgress in heartbeat.JobProgress) {
            var curTask = dao.GetTaskDA(jobProgress.Key);
            if (curTask == null) {
              // task does not exist in db
              actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, jobProgress.Key));
              DA.LogFactory.GetLogger(this.GetType().Namespace).Log("Task on slave " + heartbeat.SlaveId + " does not exist in DB: " + jobProgress.Key);
            } else {
              var currentStateLog = curTask.StateLogs.LastOrDefault();
              if (currentStateLog.SlaveId == Guid.Empty || currentStateLog.SlaveId != heartbeat.SlaveId) {
                // assigned slave does not match heartbeat
                actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.TaskId));
                DA.LogFactory.GetLogger(this.GetType().Namespace).Log("The slave " + heartbeat.SlaveId + " is not supposed to calculate task: " + curTask);
              } else if (!TaskIsAllowedToBeCalculatedBySlave(heartbeat.SlaveId, curTask)) {
                // assigned resources ids of task do not match with slaveId (and parent resourceGroupIds); this might happen when slave is moved to different group
                actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.TaskId));
              } else {
                // save task execution time
                curTask.ExecutionTimeMs = jobProgress.Value.TotalMilliseconds;
                curTask.LastHeartbeat = DateTime.Now;

                switch (curTask.Command) {
                  case DA.Command.Stop:
                    actions.Add(new MessageContainer(MessageContainer.MessageType.StopTask, curTask.TaskId));
                    break;
                  case DA.Command.Pause:
                    actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.TaskId));
                    break;
                  case DA.Command.Abort:
                    actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.TaskId));
                    break;
                }
                dao.UpdateTaskDA(curTask);
              }
            }
          }
        });
      }
      return actions;
    }

    private bool TaskIsAllowedToBeCalculatedBySlave(Guid slaveId, DA.Task curTask) {
      var assignedResourceIds = curTask.AssignedResources.Select(ar => ar.Resource.ResourceId);
      var slaveResourceIds = dao.GetParentResourcesDA(slaveId).Select(x => x.ResourceId);
      return assignedResourceIds.Any(r => slaveResourceIds.Contains(r));
    }

    private bool SlaveIsAllowedToCalculate(Guid slaveId) {
      var parentResources = dao.GetParentResourcesDA(slaveId);
      // the slave may only calculate if there is no downtime right now. this needs to be checked for every parent resource also
      return parentResources.All(r => r.Downtimes.Where(d => d.DowntimeType == DA.DowntimeType.Offline && (DateTime.Now >= d.StartDate) && (DateTime.Now <= d.EndDate)).Count() == 0);
    }

    private bool ShutdownSlaveComputer(Guid slaveId) {
      var parentResources = dao.GetParentResourcesDA(slaveId);

      return parentResources.Any(r => r.Downtimes.Where(d => d.DowntimeType == DA.DowntimeType.Shutdown && (DateTime.Now >= d.StartDate) && (DateTime.Now <= d.EndDate)).Count() != 0);
    }
  }
}