Free cookie consent management tool by TermsFeed Policy Generator

source: trunk/sources/HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs @ 12448

Last change on this file since 12448 was 12012, checked in by ascheibe, 10 years ago

#2212 merged r12008, r12009, r12010 back into trunk

File size: 8.1 KB
RevLine 
[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
22using System;
23using System.Collections.Generic;
24using System.Linq;
[9123]25using System.Threading;
[9665]26using HeuristicLab.Services.Hive.DataAccess;
27using Heartbeat = HeuristicLab.Services.Hive.DataTransfer.Heartbeat;
[6983]28
29namespace 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}
Note: See TracBrowser for help on using the repository browser.