source: branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive/3.4/ExperimentManager/HiveExperimentClient.cs @ 5106

#region License Information
/* HeuristicLab
 * Copyright (C) 2002-2010 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;
using HeuristicLab.Tracing;
using HeuristicLab.Services.Hive.Common;
using HeuristicLab.Services.Hive.Common.ServiceContracts;

namespace HeuristicLab.Clients.Hive {
  using DT = HeuristicLab.Services.Hive.Common.DataTransfer;
  using HeuristicLab.Services.Hive.Common.DataTransfer;
  using HeuristicLab.Clients.Hive.Jobs;
  using HeuristicLab.Clients.Common;

  /// <summary>
  /// An experiment which contains multiple batch runs of algorithms.
  /// </summary>
  [Item(itemName, itemDescription)]
  public class HiveExperimentClient : NamedItem, IExecutable, IProgressReporter {
    private object locker = new object();
    private const string itemName = "Hive Experiment";
    private const string itemDescription = "A runner for a single experiment, which's algorithms are executed in the Hive.";
    private System.Timers.Timer timer;
    private DateTime lastUpdateTime;
    private Guid rootJobId;
    private JobResultPoller jobResultPoller;

    private Guid hiveExperimentId;
    public Guid HiveExperimentId {
      get { return hiveExperimentId; }
      set { hiveExperimentId = value; }
    }

    private HiveJob hiveJob;
    public HiveJob HiveJob {
      get { return hiveJob; }
      set {
        DeregisterHiveJobEvents();
        if (hiveJob != value) {
          hiveJob = value;
          RegisterHiveJobEvents();
          OnHiveJobChanged();
        }
      }
    }

    private ILog log;
    public ILog Log {
      get { return log; }
    }

    private string resourceIds;
    public string ResourceIds {
      get { return resourceIds; }
      set {
        if (resourceIds != value) {
          resourceIds = value;
          OnResourceIdsChanged();
        }
      }
    }

    private bool isPollingResults;
    public bool IsPollingResults {
      get { return isPollingResults; }
      private set {
        if (isPollingResults != value) {
          isPollingResults = value;
          OnIsPollingResultsChanged();
        }
      }
    }

    private bool isProgressing;
    public bool IsProgressing {
      get { return isProgressing; }
      set {
        if (isProgressing != value) {
          isProgressing = value;
          OnIsProgressingChanged();
        }
      }
    }

    private IProgress progress;
    public IProgress Progress {
      get { return progress; }
    }

    public HiveExperimentClient() : base(itemName, itemDescription) {
      // TODO //this.ResourceIds = HeuristicLab.Hive.Experiment.Properties.Settings.Default.ResourceIds;
      this.ResourceIds = "MyGroup";
      this.log = new Log();
      InitTimer();
    }
    public HiveExperimentClient(DT.HiveExperiment hiveExperimentDto) : this() {
      UpdateFromDto(hiveExperimentDto);
    }
    protected HiveExperimentClient(HiveExperimentClient original, Cloner cloner)
      : base(original, cloner) {
      this.ResourceIds = original.resourceIds;
      this.ExecutionState = original.executionState;
      this.ExecutionTime = original.executionTime;
      this.log = cloner.Clone(original.log);
      this.lastUpdateTime = original.lastUpdateTime;
      this.rootJobId = original.rootJobId;
    }
    public override IDeepCloneable Clone(Cloner cloner) {
      return new HiveExperimentClient(this, cloner);
    }

    public void UpdateFromDto(DT.HiveExperiment hiveExperimentDto) {
      this.HiveExperimentId = hiveExperimentDto.Id;
      this.Name = hiveExperimentDto.Name;
      this.Description = hiveExperimentDto.Description;
      // TODO: this.ResourceIds = hiveExperimentDto.ResourceIds;
      this.rootJobId = hiveExperimentDto.RootJobId;
    }

    public DT.HiveExperiment ToHiveExperimentDto() {
      return new DT.HiveExperiment() {
        Id = this.HiveExperimentId,
        Name = this.Name,
        Description = this.Description,
        //ResourceIds = this.ResourceIds,
        RootJobId = this.rootJobId
      };
    }

    public void SetExperiment(Optimization.Experiment experiment) {
      this.HiveJob = new HiveJob(experiment);
      Prepare();
    }

    private void RegisterHiveJobEvents() {
      if (HiveJob != null) {
        HiveJob.JobStateChanged += new EventHandler(HiveJob_JobStateChanged);
      }
    }

    private void DeregisterHiveJobEvents() {
      if (HiveJob != null) {
        HiveJob.JobStateChanged -= new EventHandler(HiveJob_JobStateChanged);
      }
    }

    /// <summary>
    /// Returns the experiment from the root HiveJob
    /// </summary>
    public Optimization.Experiment GetExperiment() {
      if (this.HiveJob != null) {
        return HiveJob.OptimizerJob.OptimizerAsExperiment;
      }
      return null;
    }

    #region IExecutable Members
    private Core.ExecutionState executionState;
    public ExecutionState ExecutionState {
      get { return executionState; }
      private set {
        if (executionState != value) {
          executionState = value;
          OnExecutionStateChanged();
        }
      }
    }

    private TimeSpan executionTime;
    public TimeSpan ExecutionTime {
      get { return executionTime; }
      private set {
        if (executionTime != value) {
          executionTime = value;
          OnExecutionTimeChanged();
        }
      }
    }

    public void Pause() {
      throw new NotSupportedException();
    }

    public void Prepare() {
      // do nothing
    }

    public void Start() {
      OnStarted();
      ExecutionTime = new TimeSpan();
      lastUpdateTime = DateTime.Now;
      this.ExecutionState = Core.ExecutionState.Started;

      Thread t = new Thread(RunUploadExperiment);
      t.Name = "RunUploadExperimentThread";
      t.Start();
    }

    private void RunUploadExperiment() {
      try {
        this.progress = new Progress("Connecting to server...");
        IsProgressing = true;
        using (Disposable<IHiveService> service = ServiceLocator.Instance.GetService()) {
          IEnumerable<string> groups = ToResourceIdList(this.ResourceIds);
          this.HiveJob.SetIndexInParentOptimizerList(null);

          int totalJobCount = this.HiveJob.GetAllHiveJobs().Count();
          int jobCount = 0;

          this.progress.Status = "Uploading jobs...";
          UploadJobWithChildren(service.Obj, this.HiveJob, null, groups, ref jobCount, totalJobCount);
          this.rootJobId = this.HiveJob.Job.Id;
          LogMessage("Finished sending jobs to hive");

          // insert or update HiveExperiment
          this.progress.Status = "Uploading HiveExperiment...";

          DT.HiveExperiment he = service.Obj.GetHiveExperiment(service.Obj.AddHiveExperiment(this.ToHiveExperimentDto()));
          this.UpdateFromDto(he);

          StartResultPolling();
        }
      }
      catch (Exception e) {
        OnExceptionOccured(e);
      }
      finally {
        IsProgressing = false;
      }
    }

    /// <summary>
    /// Uploads the given job and all its child-jobs while setting the proper parentJobId values for the childs
    /// 
    /// </summary>
    /// <param name="service"></param>
    /// <param name="hiveJob"></param>
    /// <param name="parentHiveJob">shall be null if its the root job</param>
    /// <param name="groups"></param>
    private void UploadJobWithChildren(IHiveService service, HiveJob hiveJob, HiveJob parentHiveJob, IEnumerable<string> groups, ref int jobCount, int totalJobCount) {
      jobCount++;
      this.progress.Status = string.Format("Serializing job {0} of {1}", jobCount, totalJobCount);
      JobData jobData;
      if (hiveJob.OptimizerJob.ComputeInParallel &&
        (hiveJob.OptimizerJob.Optimizer is Optimization.Experiment || hiveJob.OptimizerJob.Optimizer is Optimization.BatchRun)) {
        hiveJob.Job.JobState = JobState.WaitingForChildJobs;
        hiveJob.OptimizerJob.CollectChildJobs = false; // don't collect child-jobs on slaves
        jobData = hiveJob.GetAsJobData(true);
      } else {
        jobData = hiveJob.GetAsJobData(false);
      }

      this.progress.Status = string.Format("Uploading job {0} of {1} ({2} kb)", jobCount, totalJobCount, jobData.Data.Count() / 1024);
      this.progress.ProgressValue = (double)jobCount / totalJobCount;

      if (parentHiveJob != null) {
        //response = service.AddChildJob(parentHiveJob.Job.Id, serializedJob);
        hiveJob.Job.Id = service.AddChildJob(parentHiveJob.Job.Id, hiveJob.Job, jobData);
      } else {
        // response = service.AddJobWithGroupStrings(serializedJob, groups);
        hiveJob.Job.Id = service.AddJob(hiveJob.Job, jobData);
      }

      LogMessage(hiveJob.Job.Id, "Job sent to Hive");

      foreach (HiveJob child in hiveJob.ChildHiveJobs) {
        UploadJobWithChildren(service, child, hiveJob, groups, ref jobCount, totalJobCount);
      }
    }

    /// <summary>
    /// Converts a string which can contain Ids separated by ';' to a enumerable
    /// </summary>
    private IEnumerable<string> ToResourceIdList(string resourceGroups) {
      if (!string.IsNullOrEmpty(resourceGroups)) {
        return resourceIds.Split(';');
      } else {
        return new List<string>();
      }
    }

    public void Stop() {
      using (Disposable<IHiveService> service = ServiceLocator.Instance.GetService()) {
        foreach (HiveJob hj in HiveJob.GetAllHiveJobs()) {
          service.Obj.StopJob(hj.Job.Id);
        }
      }
    }

    #endregion

    public void StartResultPolling() {
      if (!jobResultPoller.IsPolling) {
        jobResultPoller.Start();
      } else {
        throw new JobResultPollingException("Result polling already running");
      }
    }

    public void StopResultPolling() {
      if (jobResultPoller.IsPolling) {
        jobResultPoller.Stop();
      } else {
        throw new JobResultPollingException("Result polling not running");
      }
    }

    #region HiveJob Events
    void HiveJob_JobStateChanged(object sender, EventArgs e) {
      if (HiveJob != null) {
        rootJobId = HiveJob.Job.Id;
      }
    }
    #endregion

    #region Eventhandler

    public event EventHandler ExecutionTimeChanged;
    private void OnExecutionTimeChanged() {
      EventHandler handler = ExecutionTimeChanged;
      if (handler != null) handler(this, EventArgs.Empty);
    }

    public event EventHandler ExecutionStateChanged;
    private void OnExecutionStateChanged() {
      LogMessage("ExecutionState changed to " + executionState.ToString());
      EventHandler handler = ExecutionStateChanged;
      if (handler != null) handler(this, EventArgs.Empty);
    }

    public event EventHandler Started;
    private void OnStarted() {
      LogMessage("Started");
      timer.Start();
      EventHandler handler = Started;
      if (handler != null) handler(this, EventArgs.Empty);
    }

    public event EventHandler Stopped;
    private void OnStopped() {
      LogMessage("Stopped");
      timer.Stop();
      EventHandler handler = Stopped;
      if (handler != null) handler(this, EventArgs.Empty);
    }

    public event EventHandler Paused;
    private void OnPaused() {
      LogMessage("Paused");
      EventHandler handler = Paused;
      if (handler != null) handler(this, EventArgs.Empty);
    }

    public event EventHandler Prepared;
    protected virtual void OnPrepared() {
      LogMessage("Prepared");
      EventHandler handler = Prepared;
      if (handler != null) handler(this, EventArgs.Empty);
    }

    public event EventHandler ResourceIdsChanged;
    protected virtual void OnResourceIdsChanged() {
      EventHandler handler = ResourceIdsChanged;
      if (handler != null) handler(this, EventArgs.Empty);
    }

    public event EventHandler IsResultsPollingChanged;
    private void OnIsPollingResultsChanged() {
      if (this.IsPollingResults) {
        LogMessage("Results Polling Started");
      } else {
        LogMessage("Results Polling Stopped");
      }
      EventHandler handler = IsResultsPollingChanged;
      if (handler != null) handler(this, EventArgs.Empty);
    }

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

    public event EventHandler HiveJobChanged;
    private void OnHiveJobChanged() {
      if (jobResultPoller != null && jobResultPoller.IsPolling) {
        jobResultPoller.Stop();
        DeregisterResultPollingEvents();
      }
      if (HiveJob != null) {
        jobResultPoller = new JobResultPoller(HiveJob, ApplicationConstants.ResultPollingInterval);
        RegisterResultPollingEvents();
      }
      EventHandler handler = HiveJobChanged;
      if (handler != null) handler(this, EventArgs.Empty);
    }

    public event EventHandler IsProgressingChanged;
    private void OnIsProgressingChanged() {
      var handler = IsProgressingChanged;
      if (handler != null) handler(this, EventArgs.Empty);
    }
    #endregion

    #region JobResultPoller Events
    private void RegisterResultPollingEvents() {
      jobResultPoller.ExceptionOccured += new EventHandler<EventArgs<Exception>>(jobResultPoller_ExceptionOccured);
      jobResultPoller.JobResultsReceived += new EventHandler<EventArgs<IEnumerable<LightweightJob>>>(jobResultPoller_JobResultReceived);
      jobResultPoller.PollingStarted += new EventHandler(jobResultPoller_PollingStarted);
      jobResultPoller.PollingFinished += new EventHandler(jobResultPoller_PollingFinished);
      jobResultPoller.IsPollingChanged += new EventHandler(jobResultPoller_IsPollingChanged);
    }
    private void DeregisterResultPollingEvents() {
      jobResultPoller.ExceptionOccured -= new EventHandler<EventArgs<Exception>>(jobResultPoller_ExceptionOccured);
      jobResultPoller.JobResultsReceived -= new EventHandler<EventArgs<IEnumerable<LightweightJob>>>(jobResultPoller_JobResultReceived);
      jobResultPoller.PollingStarted -= new EventHandler(jobResultPoller_PollingStarted);
      jobResultPoller.PollingFinished -= new EventHandler(jobResultPoller_PollingFinished);
      jobResultPoller.IsPollingChanged -= new EventHandler(jobResultPoller_IsPollingChanged);
    }
    void jobResultPoller_IsPollingChanged(object sender, EventArgs e) {
      this.IsPollingResults = jobResultPoller.IsPolling;
    }
    void jobResultPoller_PollingFinished(object sender, EventArgs e) {
      LogMessage("Polling results finished");
    }
    void jobResultPoller_PollingStarted(object sender, EventArgs e) {
      LogMessage("Polling results started");
    }
    void jobResultPoller_JobResultReceived(object sender, EventArgs<IEnumerable<LightweightJob>> e) {
      foreach (LightweightJob lightweightJob in e.Value) {
        HiveJob hj = hiveJob.GetHiveJobByJobId(lightweightJob.Id);
        if (hj != null) {
          hj.UpdateFromLightweightJob(lightweightJob);
          if ((hj.Job.JobState == JobState.Aborted ||
               hj.Job.JobState == JobState.Failed ||
               hj.Job.JobState == JobState.Finished) &&
              !hj.IsFinishedOptimizerDownloaded) {
            LogMessage(hj.Job.Id, "Downloading optimizer for job");
            OptimizerJob optimizerJob = LoadOptimizerJob(hj.Job.Id);
            if (optimizerJob == null) {
              // something bad happened to this job. set to finished to allow the rest beeing downloaded
              hj.IsFinishedOptimizerDownloaded = true;
            } else {
              if (lightweightJob.ParentJobId.HasValue) {
                HiveJob parentHiveJob = HiveJob.GetHiveJobByJobId(lightweightJob.ParentJobId.Value);
                parentHiveJob.UpdateChildOptimizer(optimizerJob, hj.Job.Id);
              } else {
                this.HiveJob.IsFinishedOptimizerDownloaded = true;
              }
            }
          }
        }
      }
      GC.Collect(); // force GC, because .NET is too lazy here (deserialization takes a lot of memory)
      if (AllJobsFinished()) {
        this.ExecutionState = Core.ExecutionState.Stopped;
        StopResultPolling();
        OnStopped();
      }
    }

    private bool AllJobsFinished() {
      return HiveJob.GetAllHiveJobs().All(hj => hj.IsFinishedOptimizerDownloaded);
    }

    void jobResultPoller_ExceptionOccured(object sender, EventArgs<Exception> e) {
      OnExceptionOccured(e.Value);
    }
    #endregion

    #region Execution Time Timer
    private void InitTimer() {
      timer = new System.Timers.Timer(100);
      timer.AutoReset = true;
      timer.Elapsed += new System.Timers.ElapsedEventHandler(timer_Elapsed);
    }

    private void timer_Elapsed(object sender, System.Timers.ElapsedEventArgs e) {
      DateTime now = DateTime.Now;
      ExecutionTime += now - lastUpdateTime;
      lastUpdateTime = now;
    }
    #endregion

    #region Logging
    private void LogMessage(string message) {
      // HeuristicLab.Log is not Thread-Safe, so lock on every call
      lock (locker) {
        log.LogMessage(message);
        Logger.Debug(message);
      }
    }

    private void LogMessage(Guid jobId, string message) {
      //GetJobItemById(jobId).LogMessage(message);
      LogMessage(message + " (jobId: " + jobId + ")");
    }

    #endregion

    /// <summary>
    /// Downloads the root job from hive and sets the experiment, rootJob and rootJobItem
    /// </summary>
    public void LoadHiveJob() {
      progress = new Progress();
      try {
        IsProgressing = true;
        int totalJobCount = 0;
        int jobCount = 0;
        progress.Status = "Connecting to Server...";
        using (var service = ServiceLocator.Instance.GetService()) {
          // fetch all Job objects to create the full tree of tree of HiveJob objects
          progress.Status = "Downloading list of jobs...";
          IEnumerable<LightweightJob> allResults = service.Obj.GetLightweightChildJobs(rootJobId, true, true);
          totalJobCount = allResults.Count();

          // download them first
          IDictionary<Guid, Job> allJobs = new Dictionary<Guid, Job>();
          IDictionary<Guid, JobData> allJobDatas = new Dictionary<Guid, JobData>();
          foreach (LightweightJob lightweightJob in allResults) {
            jobCount++;
            progress.Status = string.Format("Downloading {0} of {1} jobs...", jobCount, totalJobCount);
            allJobs.Add(lightweightJob.Id, service.Obj.GetJob(lightweightJob.Id));
            allJobDatas.Add(lightweightJob.Id, service.Obj.GetJobData(lightweightJob.Id));
            progress.ProgressValue = (double)jobCount / totalJobCount;
          }

          jobCount = 1;
          progress.Status = string.Format("Deserializing {0} of {1} jobs... ({2} kb)", jobCount, totalJobCount, allJobDatas[this.rootJobId].Data.Count() / 1024);
          this.HiveJob = new HiveJob(allJobs[this.rootJobId], allJobDatas[this.rootJobId], false);
          allJobDatas.Remove(this.rootJobId); // reduce memory footprint
          allJobs.Remove(this.rootJobId);
          progress.ProgressValue = (double)jobCount / totalJobCount;

          if (this.HiveJob.Job.DateFinished.HasValue) {
            this.ExecutionTime = this.HiveJob.Job.DateFinished.Value - this.HiveJob.Job.DateCreated;
            this.lastUpdateTime = this.HiveJob.Job.DateFinished.Value;
            this.ExecutionState = Core.ExecutionState.Stopped;
            OnStopped();
          } else {
            this.ExecutionTime = DateTime.Now - this.HiveJob.Job.DateCreated;
            this.lastUpdateTime = DateTime.Now;
            this.ExecutionState = Core.ExecutionState.Started;
            OnStarted();
          }

          // build child-job tree
          LoadChildResults(service.Obj, this.HiveJob, allResults, allJobs, allJobDatas, progress, totalJobCount, ref jobCount);
          StartResultPolling();
        }
      }
      catch (Exception e) {
        OnExceptionOccured(e);
      }
      finally {
        IsProgressing = false;
      }
    }

    private void LoadChildResults(IHiveService service, HiveJob parentHiveJob, IEnumerable<LightweightJob> allResults, IDictionary<Guid, Job> allJobs, IDictionary<Guid, JobData> allJobDatas, IProgress progress, int totalJobCount, ref int jobCount) {
      IEnumerable<LightweightJob> childResults = from result in allResults
                                                 where result.ParentJobId.HasValue && result.ParentJobId.Value == parentHiveJob.Job.Id
                                                 orderby result.DateCreated ascending
                                                 select result;
      foreach (LightweightJob lightweightJob in childResults) {
        jobCount++;
        progress.Status = string.Format("Deserializing {0} of {1} jobs ({2} kb)...", jobCount, totalJobCount, allJobDatas[lightweightJob.Id].Data.Count() / 1024);
        OptimizerJob optimizerJob = null;
        try {
          optimizerJob = PersistenceUtil.Deserialize<OptimizerJob>(allJobDatas[lightweightJob.Id].Data);
        }
        catch {
          optimizerJob = null;
        }
        progress.ProgressValue = (double)jobCount / totalJobCount;
        HiveJob childHiveJob = new HiveJob(optimizerJob, false);
        parentHiveJob.AddChildHiveJob(childHiveJob);
        childHiveJob.Job = allJobs[lightweightJob.Id];
        allJobDatas.Remove(lightweightJob.Id); // reduce memory footprint
        allJobs.Remove(lightweightJob.Id);
        if (jobCount % 10 == 0) GC.Collect(); // this is needed or otherwise HL takes over the system when the number of jobs is high
        LoadChildResults(service, childHiveJob, allResults, allJobs, allJobDatas, progress, totalJobCount, ref jobCount);
      }
    }

    private OptimizerJob LoadOptimizerJob(Guid jobId) {
      using (var service = ServiceLocator.Instance.GetService()) {
        JobData jobData = service.Obj.GetJobData(jobId);
        try {
          return PersistenceUtil.Deserialize<OptimizerJob>(jobData.Data);
        }
        catch {
          return null;
        }
      }
    }
  }
}