Free cookie consent management tool by TermsFeed Policy Generator

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

Last change on this file since 5779 was 5779, checked in by cneumuel, 14 years ago

#1233

  • implemented pause, stop for single jobs
  • introduced Command property for jobs (to distinguish between state and command (abort vs. aborted))
  • improved behaviour of ItemTreeView (double click opens new window, selected item stays marked)
  • fixed bugs in StateLogGanttChartListView and HiveJobView
  • fixed cloning of client-side dtos
File size: 4.3 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      dao.UpdateJobState(job.Id, JobState.Transferring, slave.Id, null, null);
55      dao.UpdateSlave(slave);
56    }
57
58    /// <summary>
59    /// Update the progress of each job
60    /// Checks if all the jobs sent by heartbeat are supposed to be calculated by this slave
61    /// </summary>
62    private IEnumerable<MessageContainer> UpdateJobs(Heartbeat heartbeat) {
63      List<MessageContainer> actions = new List<MessageContainer>();
64
65      if (heartbeat.JobProgress == null)
66        return actions;
67
68      // process the jobProgresses
69      foreach (var jobProgress in heartbeat.JobProgress) {
70        Job curJob = dao.GetJob(jobProgress.Key);
71        if (curJob == null) {
72          // job does not exist in db
73          actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, jobProgress.Key));
74          LogFactory.GetLogger(this.GetType().Namespace).Log("Job does not exist in DB: " + jobProgress.Key);
75        } else {
76          if (curJob.CurrentStateLog.SlaveId == Guid.Empty || curJob.CurrentStateLog.SlaveId != heartbeat.SlaveId) {
77            // assigned slave does not match heartbeat
78            actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
79            LogFactory.GetLogger(this.GetType().Namespace).Log("The slave " + heartbeat.SlaveId + " is not supposed to calculate Job: " + curJob);
80          } else {
81            // save job execution time
82            curJob.ExecutionTime = jobProgress.Value;
83            curJob.LastHeartbeat = DateTime.Now;
84
85            switch (curJob.Command) {
86              case Command.Stop:
87                actions.Add(new MessageContainer(MessageContainer.MessageType.StopJob, curJob.Id));
88                break;
89              case Command.Pause:
90                actions.Add(new MessageContainer(MessageContainer.MessageType.PauseJob, curJob.Id));
91                break;
92              case Command.Abort:
93                actions.Add(new MessageContainer(MessageContainer.MessageType.AbortJob, curJob.Id));
94                break;
95            }
96            dao.UpdateJob(curJob);
97          }
98        }
99      }
100      return actions;
101    }
102  }
103}
Note: See TracBrowser for help on using the repository browser.