[6983] | 1 | #region License Information
|
---|
| 2 | /* HeuristicLab
|
---|
[12012] | 3 | * Copyright (C) 2002-2015 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;
|
---|
[9123] | 25 | using System.Threading;
|
---|
[9665] | 26 | using HeuristicLab.Services.Hive.DataAccess;
|
---|
| 27 | using Heartbeat = HeuristicLab.Services.Hive.DataTransfer.Heartbeat;
|
---|
[6983] | 28 |
|
---|
| 29 | namespace HeuristicLab.Services.Hive {
|
---|
| 30 | public class HeartbeatManager {
|
---|
[9123] | 31 | private const string MutexName = "HiveTaskSchedulingMutex";
|
---|
| 32 |
|
---|
[9665] | 33 | private IOptimizedHiveDao dao {
|
---|
| 34 | get { return ServiceLocator.Instance.OptimizedHiveDao; }
|
---|
[6983] | 35 | }
|
---|
[9123] | 36 | private ITaskScheduler taskScheduler {
|
---|
| 37 | get { return ServiceLocator.Instance.TaskScheduler; }
|
---|
[6983] | 38 | }
|
---|
[9257] | 39 | private DataAccess.ITransactionManager trans {
|
---|
| 40 | get { return ServiceLocator.Instance.TransactionManager; }
|
---|
| 41 | }
|
---|
[6983] | 42 |
|
---|
| 43 | /// <summary>
|
---|
| 44 | /// This method will be called every time a slave sends a heartbeat (-> very often; concurrency is important!)
|
---|
| 45 | /// </summary>
|
---|
| 46 | /// <returns>a list of actions the slave should do</returns>
|
---|
| 47 | public List<MessageContainer> ProcessHeartbeat(Heartbeat heartbeat) {
|
---|
| 48 | List<MessageContainer> actions = new List<MessageContainer>();
|
---|
[9665] | 49 |
|
---|
[9257] | 50 | Slave slave = null;
|
---|
[9665] | 51 | trans.UseTransaction(() => {
|
---|
| 52 | slave = dao.GetSlaveById(heartbeat.SlaveId);
|
---|
| 53 | });
|
---|
[6983] | 54 | if (slave == null) {
|
---|
| 55 | actions.Add(new MessageContainer(MessageContainer.MessageType.SayHello));
|
---|
[7723] | 56 | } else {
|
---|
[7187] | 57 | if (heartbeat.HbInterval != slave.HbInterval) {
|
---|
| 58 | actions.Add(new MessageContainer(MessageContainer.MessageType.NewHBInterval));
|
---|
[6983] | 59 | }
|
---|
[9665] | 60 | if (dao.SlaveHasToShutdownComputer(slave.ResourceId)) {
|
---|
[8957] | 61 | actions.Add(new MessageContainer(MessageContainer.MessageType.ShutdownComputer));
|
---|
| 62 | }
|
---|
[7723] | 63 |
|
---|
[6983] | 64 | // update slave data
|
---|
| 65 | slave.FreeCores = heartbeat.FreeCores;
|
---|
| 66 | slave.FreeMemory = heartbeat.FreeMemory;
|
---|
| 67 | slave.CpuUtilization = heartbeat.CpuUtilization;
|
---|
[9665] | 68 | slave.IsAllowedToCalculate = dao.SlaveIsAllowedToCalculate(slave.ResourceId);
|
---|
[6983] | 69 | slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0) ? SlaveState.Calculating : SlaveState.Idle;
|
---|
| 70 | slave.LastHeartbeat = DateTime.Now;
|
---|
| 71 |
|
---|
[9665] | 72 | trans.UseTransaction(() => {
|
---|
| 73 | dao.UpdateSlave(slave);
|
---|
| 74 | });
|
---|
[9257] | 75 |
|
---|
[6983] | 76 | // update task data
|
---|
| 77 | actions.AddRange(UpdateTasks(heartbeat, slave.IsAllowedToCalculate));
|
---|
| 78 |
|
---|
| 79 | // assign new task
|
---|
| 80 | if (heartbeat.AssignJob && slave.IsAllowedToCalculate && heartbeat.FreeCores > 0) {
|
---|
[9123] | 81 | bool mutexAquired = false;
|
---|
| 82 | var mutex = new Mutex(false, MutexName);
|
---|
| 83 | try {
|
---|
[9665] | 84 |
|
---|
[9123] | 85 | mutexAquired = mutex.WaitOne(Properties.Settings.Default.SchedulingPatience);
|
---|
| 86 | if (!mutexAquired)
|
---|
[9665] | 87 | LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling could not be aquired.");
|
---|
[9123] | 88 | else {
|
---|
[9665] | 89 | trans.UseTransaction(() => {
|
---|
| 90 | IEnumerable<TaskInfoForScheduler> availableTasks = null;
|
---|
| 91 | availableTasks = taskScheduler.Schedule(dao.GetWaitingTasks(slave).ToArray());
|
---|
| 92 | if (availableTasks.Any()) {
|
---|
| 93 | var task = availableTasks.First();
|
---|
| 94 | AssignTask(slave, task.TaskId);
|
---|
| 95 | actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateTask, task.TaskId));
|
---|
| 96 | }
|
---|
| 97 | });
|
---|
[9123] | 98 | }
|
---|
[6983] | 99 | }
|
---|
[9123] | 100 | catch (AbandonedMutexException) {
|
---|
[9665] | 101 | LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling has been abandoned.");
|
---|
[9123] | 102 | }
|
---|
| 103 | catch (Exception ex) {
|
---|
[9665] | 104 | LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager threw an exception in ProcessHeartbeat: " + ex.ToString());
|
---|
[9123] | 105 | }
|
---|
| 106 | finally {
|
---|
| 107 | if (mutexAquired) mutex.ReleaseMutex();
|
---|
| 108 | }
|
---|
[6983] | 109 | }
|
---|
| 110 | }
|
---|
| 111 | return actions;
|
---|
| 112 | }
|
---|
| 113 |
|
---|
[9665] | 114 | private void AssignTask(Slave slave, Guid taskId) {
|
---|
| 115 | var task = dao.UpdateTaskState(taskId, TaskState.Transferring, slave.ResourceId, null, null);
|
---|
[6983] | 116 |
|
---|
[9665] | 117 | // from now on the task has some time to send the next heartbeat (ApplicationConstants.TransferringJobHeartbeatTimeout)
|
---|
| 118 | task.LastHeartbeat = DateTime.Now;
|
---|
| 119 | dao.UpdateTask(task);
|
---|
[6983] | 120 | }
|
---|
| 121 |
|
---|
| 122 | /// <summary>
|
---|
| 123 | /// Update the progress of each task
|
---|
| 124 | /// Checks if all the task sent by heartbeat are supposed to be calculated by this slave
|
---|
| 125 | /// </summary>
|
---|
| 126 | private IEnumerable<MessageContainer> UpdateTasks(Heartbeat heartbeat, bool IsAllowedToCalculate) {
|
---|
| 127 | List<MessageContainer> actions = new List<MessageContainer>();
|
---|
| 128 |
|
---|
| 129 | if (heartbeat.JobProgress == null)
|
---|
| 130 | return actions;
|
---|
| 131 |
|
---|
| 132 | if (!IsAllowedToCalculate && heartbeat.JobProgress.Count != 0) {
|
---|
| 133 | actions.Add(new MessageContainer(MessageContainer.MessageType.PauseAll));
|
---|
| 134 | } else {
|
---|
| 135 | // process the jobProgresses
|
---|
| 136 | foreach (var jobProgress in heartbeat.JobProgress) {
|
---|
[9665] | 137 | Tuple<Task, Guid?> taskWithLastStateLogSlaveId = null;
|
---|
| 138 | trans.UseTransaction(() => {
|
---|
| 139 | taskWithLastStateLogSlaveId = dao.GetTaskByIdAndLastStateLogSlaveId(jobProgress.Key);
|
---|
| 140 | });
|
---|
| 141 | var curTask = taskWithLastStateLogSlaveId != null ? taskWithLastStateLogSlaveId.Item1 : null;
|
---|
[7723] | 142 | if (curTask == null) {
|
---|
[6983] | 143 | // task does not exist in db
|
---|
| 144 | actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, jobProgress.Key));
|
---|
[9665] | 145 | LogFactory.GetLogger(this.GetType().Namespace).Log("Task on slave " + heartbeat.SlaveId + " does not exist in DB: " + jobProgress.Key);
|
---|
[6983] | 146 | } else {
|
---|
[9665] | 147 | var slaveId = taskWithLastStateLogSlaveId.Item2;
|
---|
| 148 | if (slaveId == Guid.Empty || slaveId != heartbeat.SlaveId) {
|
---|
[6983] | 149 | // assigned slave does not match heartbeat
|
---|
[9665] | 150 | actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.TaskId));
|
---|
| 151 | LogFactory.GetLogger(this.GetType().Namespace).Log("The slave " + heartbeat.SlaveId + " is not supposed to calculate task: " + curTask);
|
---|
| 152 | } else if (!dao.TaskIsAllowedToBeCalculatedBySlave(curTask.TaskId, heartbeat.SlaveId)) {
|
---|
[6983] | 153 | // assigned resources ids of task do not match with slaveId (and parent resourceGroupIds); this might happen when slave is moved to different group
|
---|
[9665] | 154 | actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.TaskId));
|
---|
[6983] | 155 | } else {
|
---|
| 156 | // save task execution time
|
---|
[9665] | 157 | curTask.ExecutionTimeMs = jobProgress.Value.TotalMilliseconds;
|
---|
[7723] | 158 | curTask.LastHeartbeat = DateTime.Now;
|
---|
[6983] | 159 |
|
---|
[7723] | 160 | switch (curTask.Command) {
|
---|
[6983] | 161 | case Command.Stop:
|
---|
[9665] | 162 | actions.Add(new MessageContainer(MessageContainer.MessageType.StopTask, curTask.TaskId));
|
---|
[6983] | 163 | break;
|
---|
| 164 | case Command.Pause:
|
---|
[9665] | 165 | actions.Add(new MessageContainer(MessageContainer.MessageType.PauseTask, curTask.TaskId));
|
---|
[6983] | 166 | break;
|
---|
| 167 | case Command.Abort:
|
---|
[9665] | 168 | actions.Add(new MessageContainer(MessageContainer.MessageType.AbortTask, curTask.TaskId));
|
---|
[6983] | 169 | break;
|
---|
| 170 | }
|
---|
[9665] | 171 | trans.UseTransaction(() => {
|
---|
| 172 | dao.UpdateTask(curTask);
|
---|
| 173 | });
|
---|
[6983] | 174 | }
|
---|
| 175 | }
|
---|
| 176 | }
|
---|
| 177 | }
|
---|
| 178 | return actions;
|
---|
| 179 | }
|
---|
| 180 | }
|
---|
| 181 | }
|
---|