Free cookie consent management tool by TermsFeed Policy Generator

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

Last change on this file since 6110 was 6110, checked in by cneumuel, 13 years ago

#1233

  • renamed engines to executors
  • changed locking in StartJobInAppDomain
  • avoid destruction of proxy object after 5 minutes for Slave.Core
  • added JobStarted event and fixed ExecutionStateChanged and ExecutionTimeChanged
  • slaves which are moved to another slavegroup will pause their jobs now, if they must not calculate them
File size: 5.1 KB
Line 
1using System;
2using System.Collections.Generic;
3using System.Linq;
4using HeuristicLab.Services.Hive.Common;
5using HeuristicLab.Services.Hive.Common.DataTransfer;
6
7namespace HeuristicLab.Services.Hive {
8  public class HeartbeatManager {
9    private DataAccess.IHiveDao dao {
10      get { return ServiceLocator.Instance.HiveDao; }
11    }
12    private HeuristicLab.Services.Hive.DataAccess.TransactionManager trans {
13      get { return ServiceLocator.Instance.TransactionManager; }
14    }
15    private IAuthorizationManager auth {
16      get { return ServiceLocator.Instance.AuthorizationManager; }
17    }
18
19    /// <summary>
20    /// This method will be called every time a slave sends a heartbeat (-> very often; concurrency is important!)
21    /// </summary>
22    /// <returns>a list of actions the slave should do</returns>
23    public List<MessageContainer> ProcessHeartbeat(Heartbeat heartbeat) {
24      List<MessageContainer> actions = new List<MessageContainer>();
25      Slave slave = dao.GetSlave(heartbeat.SlaveId);
26      if (slave == null) {
27        actions.Add(new MessageContainer(MessageContainer.MessageType.SayHello));
28      } else {
29        // update slave data
30        slave.FreeCores = heartbeat.FreeCores;
31        slave.FreeMemory = heartbeat.FreeMemory;
32        slave.IsAllowedToCalculate = true; // Todo: look into calendar
33        slave.SlaveState = (heartbeat.JobProgress != null && heartbeat.JobProgress.Count > 0) ? SlaveState.Calculating : SlaveState.Idle;
34        slave.LastHeartbeat = DateTime.Now;
35        dao.UpdateSlave(slave);
36
37        // update job data
38        actions.AddRange(UpdateJobs(heartbeat));
39
40        // assign new job
41        if (heartbeat.AssignJob && slave.IsAllowedToCalculate && heartbeat.FreeCores > 0) {
42          var availableJobs = dao.GetWaitingJobs(slave, 1);
43          if (availableJobs.Count() > 0) {
44            var job = availableJobs.First();
45            actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateJob, job.Id));
46            AssignJob(slave, job);
47          }
48        }
49      }
50      return actions;
51    }
52
53    private void AssignJob(Slave slave, Job job) {
54      job = dao.UpdateJobState(job.Id, JobState.Transferring, slave.Id, null, null);
55      dao.UpdateSlave(slave);
56
57      // from now on the job has some time to send the next heartbeat (ApplicationConstants.TransferringJobHeartbeatTimeout)
58      job.LastHeartbeat = DateTime.Now;
59      dao.UpdateJob(job);
60    }
61
62    /// <summary>
63    /// Update the progress of each job
64    /// Checks if all the jobs sent by heartbeat are supposed to be calculated by this slave
65    /// </summary>
66    private IEnumerable<MessageContainer> UpdateJobs(Heartbeat heartbeat) {
67      List<MessageContainer> actions = new List<MessageContainer>();
68
69      if (heartbeat.JobProgress == null)
70        return actions;
71
72      // process the jobProgresses
73      foreach (var jobProgress in heartbeat.JobProgress) {
74        Job curJob = dao.GetJob(jobProgress.Key);
75        if (curJob == null) {
76          // job does not exist in db
77          actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, jobProgress.Key));
78          LogFactory.GetLogger(this.GetType().Namespace).Log("Job does not exist in DB: " + jobProgress.Key);
79        } else {
80          if (curJob.CurrentStateLog.SlaveId == Guid.Empty || curJob.CurrentStateLog.SlaveId != heartbeat.SlaveId) {
81            // assigned slave does not match heartbeat
82            actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
83            LogFactory.GetLogger(this.GetType().Namespace).Log("The slave " + heartbeat.SlaveId + " is not supposed to calculate Job: " + curJob);
84          } else if (!JobIsAllowedToBeCalculatedBySlave(heartbeat.SlaveId, curJob)) {
85            // assigned resources ids of job do not match with slaveId (and parent resourceGroupIds); this might happen when slave is moved to different group
86            actions.Add(new MessageContainer(MessageContainer.MessageType.PauseJob, curJob.Id));
87          } else {
88            // save job execution time
89            curJob.ExecutionTime = jobProgress.Value;
90            curJob.LastHeartbeat = DateTime.Now;
91
92            switch (curJob.Command) {
93              case Command.Stop:
94                actions.Add(new MessageContainer(MessageContainer.MessageType.StopJob, curJob.Id));
95                break;
96              case Command.Pause:
97                actions.Add(new MessageContainer(MessageContainer.MessageType.PauseJob, curJob.Id));
98                break;
99              case Command.Abort:
100                actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
101                break;
102            }
103            dao.UpdateJob(curJob);
104          }
105        }
106      }
107      return actions;
108    }
109
110    private bool JobIsAllowedToBeCalculatedBySlave(Guid slaveId, Job curJob) {
111      var assignedResourceIds = dao.GetAssignedResources(curJob.Id).Select(x => x.Id);
112      var slaveResourceIds = dao.GetParentResources(slaveId).Select(x => x.Id);
113      return assignedResourceIds.Any(x => slaveResourceIds.Contains(x));
114    }
115  }
116}
Note: See TracBrowser for help on using the repository browser.