source: branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Services.Hive/3.4/LifecycleManager.cs @ 6229

using System;
using System.Collections.Generic;
using System.Linq;
using HeuristicLab.Services.Hive.Common;
using HeuristicLab.Services.Hive.Common.DataTransfer;

namespace HeuristicLab.Services.Hive {
  /// <summary>
  /// This class offers methods for cleaning up offline slaves and jobs
  /// </summary>
  public class LifecycleManager : ILifecycleManager {
    private DataAccess.IHiveDao dao {
      get { return ServiceLocator.Instance.HiveDao; }
    }
    private HeuristicLab.Services.Hive.DataAccess.TransactionManager trans {
      get { return ServiceLocator.Instance.TransactionManager; }
    }
    private IAuthorizationManager auth {
      get { return ServiceLocator.Instance.AuthorizationManager; }
    }
    private ILogger log {
      get { return LogFactory.GetLogger(this.GetType().Namespace); }
    }

    public void Cleanup() {
      log.Log("LifecycleManager.Cleanup()");
      SetTimeoutSlavesOffline();
      SetTimeoutJobsWaiting();
      FinishParentJobs();
      UpdateStatistics();
    }

    private void UpdateStatistics() {
      var slaves = dao.GetSlaves(x => x.SlaveState == SlaveState.Calculating || x.SlaveState == SlaveState.Idle);

      var stats = new Statistics();
      stats.TimeStamp = DateTime.Now;
      var slaveStats = new List<SlaveStatistics>();
      foreach (var slave in slaves) {
        slaveStats.Add(new SlaveStatistics() {
          SlaveId = slave.Id,
          Cores = slave.Cores.HasValue ? slave.Cores.Value : 0,
          FreeCores = slave.FreeCores.HasValue ? slave.FreeCores.Value : 0,
          Memory = slave.Memory.HasValue ? slave.Memory.Value : 0,
          FreeMemory = slave.FreeMemory.HasValue ? slave.FreeMemory.Value : 0,
          CpuUtilization = slave.CpuUtilization
        });
      }
      stats.SlaveStatistics = slaveStats;

      stats.UserStatistics = dao.GetUserStatistics();

      dao.AddStatistics(stats);

      //int slaveCount = slaves.Count();
      //int cores = slaves.Select(x => x.Cores.HasValue ? x.Cores.Value : 0).Sum();
      //int usedCores = slaves.Select(x => x.FreeCores.HasValue ? x.FreeCores.Value : 0).Sum();

      //long memoryMb = slaves.Select(x => x.Memory.HasValue ? x.Memory.Value : 0).Sum(); ;
      //long memoryUsedMb = slaves.Select(x => x.FreeMemory.HasValue ? x.FreeMemory.Value : 0).Sum();

      //double averageCpuUtilization = slaves.Select(x => x.CpuUtilization.HasValue ? x.CpuUtilization.Value : 0).Average();

      //int executionTimeDelta = 0; // since last update

      //{ // per user
      //  int activeJobs = 0; // calculating, transferring
      //  int waitingJobs = 0; // waiting
      //  int stoppedJobs = 0; // finished, failed, aborted, paused
      //}

      //{ // per slave
      //  int activeJobs = 0; // calculating, transferring
      //  double cpuUtilization = 0.0;
      //}

    }

    /// <summary>
    /// Searches for slaves which are timed out, puts them and their jobs offline
    /// </summary>
    private void SetTimeoutSlavesOffline() {
      var slaves = dao.GetSlaves(x => x.SlaveState != SlaveState.Offline);
      foreach (Slave slave in slaves) {
        if (!slave.LastHeartbeat.HasValue || (DateTime.Now - slave.LastHeartbeat.Value) > ApplicationConstants.SlaveHeartbeatTimeout) {
          slave.SlaveState = SlaveState.Offline;
          SetJobsWaiting(slave.Id);
          dao.UpdateSlave(slave);
        }
      }
    }

    /// <summary>
    /// Looks for parent jobs which have FinishWhenChildJobsFinished and set their state to finished
    /// </summary>
    private void FinishParentJobs() {
      var parentJobsToFinish = dao.GetParentJobs(dao.GetResources(x => true).Select(x => x.Id), 0, true);
      foreach (var job in parentJobsToFinish) {
        dao.UpdateJobState(job.Id, JobState.Finished, null, null, string.Empty);
      }
    }

    private void SetJobsWaiting(Guid slaveId) {
      var jobs = dao.GetJobs(x => x.State == JobState.Calculating).Where(x => x.StateLog.Last().SlaveId == slaveId);
      foreach (var j in jobs) {
        Job job = dao.UpdateJobState(j.Id, JobState.Waiting, slaveId, null, "Slave timed out.");
        job.Command = null;
        dao.UpdateJob(job);
      }
    }

    /// <summary>
    /// Looks for jobs which have not sent heartbeats for some time and reschedules them for calculation
    /// </summary>
    private void SetTimeoutJobsWaiting() {
      var jobs = dao.GetJobs(x => (x.State == JobState.Calculating && (DateTime.Now - x.LastHeartbeat) > ApplicationConstants.CalculatingJobHeartbeatTimeout) 
                               || (x.State == JobState.Transferring && (DateTime.Now - x.LastHeartbeat) > ApplicationConstants.TransferringJobHeartbeatTimeout));
      foreach (var j in jobs) {
        Job job = dao.UpdateJobState(j.Id, JobState.Waiting, null, null, "Slave timed out.");
        job.Command = null;
        dao.UpdateJob(job);
      }
    }
  }
}