source: branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave/3.4/Manager/JobManager.cs @ 6381

#region License Information
/* HeuristicLab
 * Copyright (C) 2002-2011 Heuristic and Evolutionary Algorithms Laboratory (HEAL)
 *
 * This file is part of HeuristicLab.
 *
 * HeuristicLab is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * HeuristicLab is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with HeuristicLab. If not, see <http://www.gnu.org/licenses/>.
 */
#endregion

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using HeuristicLab.Common;
using HeuristicLab.Core;

namespace HeuristicLab.Clients.Hive.SlaveCore {

  /// <summary>
  /// Holds a list of slave jobs and manages access to this list. 
  /// Forwards events from SlaveJob and forwards commands to SlaveJob. 
  /// </summary>
  public class JobManager {
    private static ReaderWriterLockSlim slaveJobsLocker = new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);
    private Dictionary<Guid, SlaveJob> slaveJobs;
    private ILog log;
    private PluginManager pluginManager;

    public int JobCount {
      get {
        slaveJobsLocker.EnterReadLock();
        try {
          return slaveJobs.Count;
        }
        finally { slaveJobsLocker.ExitReadLock(); }
      }
    }

    public Guid[] JobIds {
      get {
        slaveJobsLocker.EnterReadLock();
        try {
          return slaveJobs.Keys.ToArray();
        }
        finally { slaveJobsLocker.ExitReadLock(); }
      }
    }

    public JobManager(PluginManager pluginCache, ILog log) {
      this.pluginManager = pluginCache;
      this.log = log;
      this.slaveJobs = new Dictionary<Guid, SlaveJob>();
    }

    #region Job Control methods
    public void StartJobAsync(Job job, JobData jobData) {
      SlaveJob slaveJob = null;
      slaveJobsLocker.EnterUpgradeableReadLock();
      try {
        if (slaveJobs.ContainsKey(job.Id)) {
          SlaveStatusInfo.IncrementExceptionOccured();
          throw new JobAlreadyRunningException(job.Id);
        } else {
          slaveJob = new SlaveJob(pluginManager, job.CoresNeeded, log);
          AddSlaveJob(job, slaveJob);
          SlaveStatusInfo.IncrementJobsFetched();
        }
      }
      finally { slaveJobsLocker.ExitUpgradeableReadLock(); }

      if (slaveJob != null) {
        try {
          slaveJob.StartJobAsync(job, jobData);
        }
        catch (Exception) {
          RemoveSlaveJob(job.Id, slaveJob); // clean up and rethrow
          slaveJob.DisposeAppDomain();
          throw;
        }
      }
    }

    public void PauseJobAsync(Guid jobId) {
      slaveJobsLocker.EnterUpgradeableReadLock();
      try {
        if (!slaveJobs.ContainsKey(jobId)) throw new JobNotRunningException(jobId);
        SlaveJob slaveJob = slaveJobs[jobId];
        slaveJob.PauseJob();
      }
      finally { slaveJobsLocker.ExitUpgradeableReadLock(); }
    }

    public void StopJobAsync(Guid jobId) {
      slaveJobsLocker.EnterUpgradeableReadLock();
      try {
        if (!slaveJobs.ContainsKey(jobId)) throw new JobNotRunningException(jobId);
        SlaveJob slaveJob = slaveJobs[jobId];
        slaveJob.StopJob();
      }
      finally { slaveJobsLocker.ExitUpgradeableReadLock(); }
    }

    public void AbortJob(Guid jobId) {
      SlaveJob slaveJob = null;
      slaveJobsLocker.EnterUpgradeableReadLock();
      try {
        if (!slaveJobs.ContainsKey(jobId)) throw new JobNotRunningException(jobId);
        slaveJob = slaveJobs[jobId];
        if (!slaveJob.IsPrepared) throw new AppDomainNotCreatedException();
        RemoveSlaveJob(jobId, slaveJob);
      }
      finally { slaveJobsLocker.ExitUpgradeableReadLock(); }
      slaveJob.DisposeAppDomain();
      SlaveStatusInfo.IncrementJobsAborted();
      OnJobAborted(slaveJob);
    }

    public void PauseAllJobsAsync() {
      slaveJobsLocker.EnterUpgradeableReadLock();
      try {
        foreach (var slaveJob in slaveJobs.Values) {
          slaveJob.PauseJob();
        }
      }
      finally { slaveJobsLocker.ExitUpgradeableReadLock(); }
    }

    public void StopAllJobsAsync() {
      slaveJobsLocker.EnterUpgradeableReadLock();
      try {
        foreach (var slaveJob in slaveJobs.Values) {
          slaveJob.StopJob();
        }
      }
      finally { slaveJobsLocker.ExitUpgradeableReadLock(); }
    }

    public void AbortAllJobs() {
      slaveJobsLocker.EnterUpgradeableReadLock();
      try {
        foreach (var slaveJob in slaveJobs.Values.ToArray()) {
          AbortJob(slaveJob.JobId);
        }
      }
      finally { slaveJobsLocker.ExitUpgradeableReadLock(); }
    }
    #endregion

    #region Add/Remove SlaveJob
    private void AddSlaveJob(Job job, SlaveJob slaveJob) {
      slaveJobsLocker.EnterWriteLock();
      try {
        slaveJobs.Add(job.Id, slaveJob);
        RegisterSlaveJobEvents(slaveJob);
      }
      finally { slaveJobsLocker.ExitWriteLock(); }
    }

    private void RemoveSlaveJob(Guid jobId, SlaveJob slaveJob) {
      slaveJobsLocker.EnterWriteLock();
      try {
        slaveJobs.Remove(jobId);
        DeregisterSlaveJobEvents(slaveJob);
      }
      finally { slaveJobsLocker.ExitWriteLock(); }
    }
    #endregion

    #region SlaveJob Events
    private void RegisterSlaveJobEvents(SlaveJob slaveJob) {
      slaveJob.JobStarted += new EventHandler<EventArgs<Guid>>(slaveJob_JobStarted);
      slaveJob.JobPaused += new EventHandler<EventArgs<Guid>>(slaveJob_JobPaused);
      slaveJob.JobStopped += new EventHandler<EventArgs<Guid>>(slaveJob_JobStopped);
      slaveJob.JobFailed += new EventHandler<EventArgs<Guid, Exception>>(slaveJob_JobFailed);
      slaveJob.ExceptionOccured += new EventHandler<EventArgs<Guid, Exception>>(slaveJob_ExceptionOccured);
    }

    private void DeregisterSlaveJobEvents(SlaveJob slaveJob) {
      slaveJob.JobStarted -= new EventHandler<EventArgs<Guid>>(slaveJob_JobStarted);
      slaveJob.JobPaused -= new EventHandler<EventArgs<Guid>>(slaveJob_JobPaused);
      slaveJob.JobStopped -= new EventHandler<EventArgs<Guid>>(slaveJob_JobStopped);
      slaveJob.JobFailed -= new EventHandler<EventArgs<Guid, Exception>>(slaveJob_JobFailed);
      slaveJob.ExceptionOccured -= new EventHandler<EventArgs<Guid, Exception>>(slaveJob_ExceptionOccured);
    }

    private void slaveJob_JobStarted(object sender, EventArgs<Guid> e) {
      SlaveJob slaveJob;
      slaveJobsLocker.EnterUpgradeableReadLock();
      try {
        slaveJob = slaveJobs[e.Value];
      }
      finally { slaveJobsLocker.ExitUpgradeableReadLock(); }

      SlaveStatusInfo.IncrementJobsStarted();
      OnJobStarted(slaveJob);
    }

    private void slaveJob_JobPaused(object sender, EventArgs<Guid> e) {
      SlaveJob slaveJob;
      slaveJobsLocker.EnterUpgradeableReadLock();
      try {
        slaveJob = slaveJobs[e.Value];
        RemoveSlaveJob(e.Value, slaveJob);
      }
      finally { slaveJobsLocker.ExitUpgradeableReadLock(); }

      JobData jobData = null;
      try {
        jobData = slaveJob.GetJobData();
        if (jobData == null) throw new SerializationException();
        SlaveStatusInfo.IncrementJobsFinished();
        OnJobPaused(slaveJob, jobData);
      }
      catch (Exception ex) {
        RemoveSlaveJob(e.Value, slaveJob);
        SlaveStatusInfo.IncrementJobsFailed();
        OnJobFailed(slaveJob, jobData, ex);
      }
    }

    private void slaveJob_JobStopped(object sender, EventArgs<Guid> e) {
      SlaveJob slaveJob;
      slaveJobsLocker.EnterUpgradeableReadLock();
      try {
        slaveJob = slaveJobs[e.Value];
        RemoveSlaveJob(e.Value, slaveJob);
      }
      finally { slaveJobsLocker.ExitUpgradeableReadLock(); }

      JobData jobData = null;
      try {
        jobData = slaveJob.GetJobData();
        if (jobData == null) throw new SerializationException();
        SlaveStatusInfo.IncrementJobsFinished();
        OnJobStopped(slaveJob, jobData);
      }
      catch (Exception ex) {
        RemoveSlaveJob(e.Value, slaveJob);
        SlaveStatusInfo.IncrementJobsFailed();
        OnJobFailed(slaveJob, jobData, ex);
      }
    }

    private void slaveJob_JobFailed(object sender, EventArgs<Guid, Exception> e) {
      SlaveJob slaveJob;
      slaveJobsLocker.EnterUpgradeableReadLock();
      try {
        slaveJob = slaveJobs[e.Value];
        RemoveSlaveJob(e.Value, slaveJob);
      }
      finally { slaveJobsLocker.ExitUpgradeableReadLock(); }

      JobData jobData = null;
      try {
        jobData = slaveJob.GetJobData();
        if (jobData == null) throw new SerializationException();
      }
      catch { /* jobData will be null */ }
      SlaveStatusInfo.IncrementJobsFailed();
      OnJobFailed(slaveJob, jobData, e.Value2);
    }

    private void slaveJob_ExceptionOccured(object sender, EventArgs<Guid, Exception> e) {
      SlaveJob slaveJob;
      slaveJobsLocker.EnterUpgradeableReadLock();
      try {
        slaveJob = slaveJobs[e.Value];
        RemoveSlaveJob(e.Value, slaveJob);
      }
      finally { slaveJobsLocker.ExitUpgradeableReadLock(); }

      SlaveStatusInfo.IncrementExceptionOccured();
      OnExceptionOccured(slaveJob, e.Value2);
    }
    #endregion

    #region EventHandler
    public event EventHandler<EventArgs<SlaveJob>> JobStarted;
    private void OnJobStarted(SlaveJob slaveJob) {
      var handler = JobStarted;
      if (handler != null) handler(this, new EventArgs<SlaveJob>(slaveJob));
    }

    public event EventHandler<EventArgs<SlaveJob, JobData>> JobStopped;
    private void OnJobStopped(SlaveJob slaveJob, JobData jobData) {
      var handler = JobStopped;
      if (handler != null) handler(this, new EventArgs<SlaveJob, JobData>(slaveJob, jobData));
    }

    public event EventHandler<EventArgs<SlaveJob, JobData>> JobPaused;
    private void OnJobPaused(SlaveJob slaveJob, JobData jobData) {
      var handler = JobPaused;
      if (handler != null) handler(this, new EventArgs<SlaveJob, JobData>(slaveJob, jobData));
    }

    public event EventHandler<EventArgs<Tuple<SlaveJob, JobData, Exception>>> JobFailed;
    private void OnJobFailed(SlaveJob slaveJob, JobData jobData, Exception exception) {
      var handler = JobFailed;
      if (handler != null) handler(this, new EventArgs<Tuple<SlaveJob, JobData, Exception>>(new Tuple<SlaveJob, JobData, Exception>(slaveJob, jobData, exception)));
    }

    public event EventHandler<EventArgs<SlaveJob, Exception>> ExceptionOccured;
    private void OnExceptionOccured(SlaveJob slaveJob, Exception exception) {
      var handler = ExceptionOccured;
      if (handler != null) handler(this, new EventArgs<SlaveJob, Exception>(slaveJob, exception));
    }

    public event EventHandler<EventArgs<SlaveJob>> JobAborted;
    private void OnJobAborted(SlaveJob slaveJob) {
      var handler = JobAborted;
      if (handler != null) handler(this, new EventArgs<SlaveJob>(slaveJob));
    }
    #endregion

    public Dictionary<Guid, TimeSpan> GetExecutionTimes() {
      slaveJobsLocker.EnterReadLock();
      try {
        return slaveJobs.ToDictionary(x => x.Key, x => x.Value.ExecutionTime);
      }
      finally { slaveJobsLocker.ExitReadLock(); }
    }
  }
}