[6983] | 1 | #region License Information
|
---|
| 2 | /* HeuristicLab
|
---|
[7259] | 3 | * Copyright (C) 2002-2012 Heuristic and Evolutionary Algorithms Laboratory (HEAL)
|
---|
[6983] | 4 | *
|
---|
| 5 | * This file is part of HeuristicLab.
|
---|
| 6 | *
|
---|
| 7 | * HeuristicLab is free software: you can redistribute it and/or modify
|
---|
| 8 | * it under the terms of the GNU General Public License as published by
|
---|
| 9 | * the Free Software Foundation, either version 3 of the License, or
|
---|
| 10 | * (at your option) any later version.
|
---|
| 11 | *
|
---|
| 12 | * HeuristicLab is distributed in the hope that it will be useful,
|
---|
| 13 | * but WITHOUT ANY WARRANTY; without even the implied warranty of
|
---|
| 14 | * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
---|
| 15 | * GNU General Public License for more details.
|
---|
| 16 | *
|
---|
| 17 | * You should have received a copy of the GNU General Public License
|
---|
| 18 | * along with HeuristicLab. If not, see <http://www.gnu.org/licenses/>.
|
---|
| 19 | */
|
---|
| 20 | #endregion
|
---|
| 21 |
|
---|
| 22 | using System;
|
---|
| 23 | using System.Collections.Generic;
|
---|
| 24 | using System.Linq;
|
---|
[8707] | 25 | using System.Threading;
|
---|
[6983] | 26 | using HeuristicLab.Services.Hive.DataTransfer;
|
---|
| 27 | using DA = HeuristicLab.Services.Hive.DataAccess;
|
---|
| 28 |
|
---|
| 29 | namespace HeuristicLab.Services.Hive {
|
---|
| 30 | public class HeartbeatManager {
|
---|
[8707] | 31 | private const string MutexName = "HiveTaskSchedulingMutex";
|
---|
| 32 |
|
---|
[6983] | 33 | private IHiveDao dao {
|
---|
| 34 | get { return ServiceLocator.Instance.HiveDao; }
|
---|
| 35 | }
|
---|
[8687] | 36 | private ITaskScheduler taskScheduler {
|
---|
| 37 | get { return ServiceLocator.Instance.TaskScheduler; }
|
---|
[6983] | 38 | }
|
---|
| 39 |
|
---|
| 40 | /// <summary>
|
---|
| 41 | /// This method will be called every time a slave sends a heartbeat (-> very often; concurrency is important!)
|
---|
| 42 | /// </summary>
|
---|
| 43 | /// <returns>a list of actions the slave should do</returns>
|
---|
| 44 | public List<MessageContainer> ProcessHeartbeat(Heartbeat heartbeat) {
|
---|
| 45 | List<MessageContainer> actions = new List<MessageContainer>();
|
---|
| 46 | Slave slave = dao.GetSlave(heartbeat.SlaveId);
|
---|
| 47 | if (slave == null) {
|
---|
| 48 | actions.Add(new MessageContainer(MessageContainer.MessageType.SayHello));
|
---|
[7723] | 49 | } else {
|
---|
[7187] | 50 | if (heartbeat.HbInterval != slave.HbInterval) {
|
---|
| 51 | actions.Add(new MessageContainer(MessageContainer.MessageType.NewHBInterval));
|
---|
[6983] | 52 | }
|
---|
[7723] | 53 |
|
---|
[6983] | 54 | // update slave data
|
---|
| 55 | slave.FreeCores = heartbeat.FreeCores;
|
---|
| 56 | slave.FreeMemory = heartbeat.FreeMemory;
|
---|
| 57 | slave.CpuUtilization = heartbeat.CpuUtilization;
|
---|
| 58 | slave.IsAllowedToCalculate = SlaveIsAllowedToCalculate(slave.Id);
|
---|
| 59 | slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0) ? SlaveState.Calculating : SlaveState.Idle;
|
---|
| 60 | slave.LastHeartbeat = DateTime.Now;
|
---|
| 61 | dao.UpdateSlave(slave);
|
---|
| 62 |
|
---|
| 63 | // update task data
|
---|
| 64 | actions.AddRange(UpdateTasks(heartbeat, slave.IsAllowedToCalculate));
|
---|
| 65 |
|
---|
| 66 | // assign new task
|
---|
| 67 | if (heartbeat.AssignJob && slave.IsAllowedToCalculate && heartbeat.FreeCores > 0) {
|
---|
[8707] | 68 | bool mutexAquired = false;
|
---|
| 69 | var mutex = new Mutex(false, MutexName);
|
---|
| 70 | try {
|
---|
| 71 | mutexAquired = mutex.WaitOne(Properties.Settings.Default.SchedulingPatience);
|
---|
| 72 | if (!mutexAquired)
|
---|
| 73 | DA.LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling could not be aquired.");
|
---|
| 74 | else {
|
---|
| 75 | var availableJobs = taskScheduler.Schedule(dao.GetWaitingTasks(slave));
|
---|
| 76 | if (availableJobs.Any()) {
|
---|
| 77 | var job = availableJobs.First();
|
---|
| 78 | if (AssignJob(slave, job))
|
---|
| 79 | actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateTask, job.Id));
|
---|
| 80 | }
|
---|
| 81 | }
|
---|
[6983] | 82 | }
|
---|
[8707] | 83 | catch (AbandonedMutexException) {
|
---|
| 84 | DA.LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling has been abandoned.");
|
---|
| 85 | }
|
---|
| 86 | finally {
|
---|
| 87 | if (mutexAquired) mutex.ReleaseMutex();
|
---|
| 88 | }
|
---|
[6983] | 89 | }
|
---|
| 90 | }
|
---|
| 91 | return actions;
|
---|
| 92 | }
|
---|
| 93 |
|
---|
| 94 | // returns true if assignment was successful
|
---|
| 95 | private bool AssignJob(Slave slave, Task task) {
|
---|
| 96 | // load task again and check if it is still available (this is an attempt to reduce the race condition which causes multiple heartbeats to get the same task assigned)
|
---|
| 97 | if (dao.GetTask(task.Id).State != TaskState.Waiting) return false;
|
---|
| 98 |
|
---|
| 99 | task = dao.UpdateTaskState(task.Id, DataAccess.TaskState.Transferring, slave.Id, null, null);
|
---|
| 100 |
|
---|
| 101 | // from now on the task has some time to send the next heartbeat (ApplicationConstants.TransferringJobHeartbeatTimeout)
|
---|
| 102 | task.LastHeartbeat = DateTime.Now;
|
---|
| 103 | dao.UpdateTask(task);
|
---|
| 104 | return true;
|
---|
| 105 | }
|
---|
| 106 |
|
---|
| 107 | /// <summary>
|
---|
| 108 | /// Update the progress of each task
|
---|
| 109 | /// Checks if all the task sent by heartbeat are supposed to be calculated by this slave
|
---|
| 110 | /// </summary>
|
---|
| 111 | private IEnumerable<MessageContainer> UpdateTasks(Heartbeat heartbeat, bool IsAllowedToCalculate) {
|
---|
| 112 | List<MessageContainer> actions = new List<MessageContainer>();
|
---|
| 113 |
|
---|
| 114 | if (heartbeat.JobProgress == null)
|
---|
| 115 | return actions;
|
---|
| 116 |
|
---|
| 117 | if (!IsAllowedToCalculate && heartbeat.JobProgress.Count != 0) {
|
---|
| 118 | actions.Add(new MessageContainer(MessageContainer.MessageType.PauseAll));
|
---|
| 119 | } else {
|
---|
| 120 | // process the jobProgresses
|
---|
| 121 | foreach (var jobProgress in heartbeat.JobProgress) {
|
---|
[7723] | 122 | Task curTask = dao.GetTask(jobProgress.Key);
|
---|
| 123 | if (curTask == null) {
|
---|
[6983] | 124 | // task does not exist in db
|
---|
| 125 | actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, jobProgress.Key));
|
---|
[7723] | 126 | DA.LogFactory.GetLogger(this.GetType().Namespace).Log("Task on slave " + heartbeat.SlaveId + " does not exist in DB: " + jobProgress.Key);
|
---|
[6983] | 127 | } else {
|
---|
[7723] | 128 | if (curTask.CurrentStateLog.SlaveId == Guid.Empty || curTask.CurrentStateLog.SlaveId != heartbeat.SlaveId) {
|
---|
[6983] | 129 | // assigned slave does not match heartbeat
|
---|
[7723] | 130 | actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.Id));
|
---|
| 131 | DA.LogFactory.GetLogger(this.GetType().Namespace).Log("The slave " + heartbeat.SlaveId + " is not supposed to calculate task: " + curTask);
|
---|
| 132 | } else if (!TaskIsAllowedToBeCalculatedBySlave(heartbeat.SlaveId, curTask)) {
|
---|
[6983] | 133 | // assigned resources ids of task do not match with slaveId (and parent resourceGroupIds); this might happen when slave is moved to different group
|
---|
[7723] | 134 | actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.Id));
|
---|
[6983] | 135 | } else {
|
---|
| 136 | // save task execution time
|
---|
[7723] | 137 | curTask.ExecutionTime = jobProgress.Value;
|
---|
| 138 | curTask.LastHeartbeat = DateTime.Now;
|
---|
[6983] | 139 |
|
---|
[7723] | 140 | switch (curTask.Command) {
|
---|
[6983] | 141 | case Command.Stop:
|
---|
[7723] | 142 | actions.Add(new MessageContainer(MessageContainer.MessageType.StopTask, curTask.Id));
|
---|
[6983] | 143 | break;
|
---|
| 144 | case Command.Pause:
|
---|
[7723] | 145 | actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.Id));
|
---|
[6983] | 146 | break;
|
---|
| 147 | case Command.Abort:
|
---|
[7723] | 148 | actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.Id));
|
---|
[6983] | 149 | break;
|
---|
| 150 | }
|
---|
[7723] | 151 | dao.UpdateTask(curTask);
|
---|
[6983] | 152 | }
|
---|
| 153 | }
|
---|
| 154 | }
|
---|
| 155 | }
|
---|
| 156 | return actions;
|
---|
| 157 | }
|
---|
| 158 |
|
---|
| 159 | private bool TaskIsAllowedToBeCalculatedBySlave(Guid slaveId, Task curTask) {
|
---|
| 160 | var assignedResourceIds = dao.GetAssignedResources(curTask.Id).Select(x => x.Id);
|
---|
| 161 | var slaveResourceIds = dao.GetParentResources(slaveId).Select(x => x.Id);
|
---|
| 162 | return assignedResourceIds.Any(x => slaveResourceIds.Contains(x));
|
---|
| 163 | }
|
---|
| 164 |
|
---|
| 165 | private bool SlaveIsAllowedToCalculate(Guid slaveId) {
|
---|
| 166 | // the slave may only calculate if there is no downtime right now. this needs to be checked for every parent resource also
|
---|
| 167 | return dao.GetParentResources(slaveId).All(r => dao.GetDowntimes(x => x.ResourceId == r.Id && (DateTime.Now >= x.StartDate) && (DateTime.Now <= x.EndDate)).Count() == 0);
|
---|
| 168 | }
|
---|
| 169 | }
|
---|
| 170 | }
|
---|