source: branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine/3.3/HiveEngine.cs @ 5136

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using HeuristicLab.Persistence.Default.CompositeSerializers.Storable;
using HeuristicLab.Core;
using HeuristicLab.Common;
using HeuristicLab.Hive.Contracts.Interfaces;
using HeuristicLab.Clients.Common;
using HeuristicLab.Hive.ExperimentManager;
using HeuristicLab.Hive.Contracts.BusinessObjects;
using HeuristicLab.PluginInfrastructure;
using HeuristicLab.Hive.Contracts.ResponseObjects;
using System.Threading;

namespace HeuristicLab.HiveEngine {
  /// <summary>
  /// Represents an engine that executes operations which can be executed in parallel on the hive
  /// </summary>
  [StorableClass]
  [Item("Hive Parallel Engine", "Engine for parallel execution on the hive.")]
  public class HiveEngine : Engine {
    private IOperator currentOperator;

    #region constructors and cloning
    public HiveEngine() { }
    [StorableConstructor]
    protected HiveEngine(bool deserializing) : base(deserializing) { }
    protected HiveEngine(HiveEngine original, Cloner cloner)
      : base(original, cloner) {
    }
    public override IDeepCloneable Clone(Cloner cloner) {
      return new HiveEngine(this, cloner);
    }
    #endregion

    protected override void ProcessNextOperation() {
      currentOperator = null;
      IOperation next = ExecutionStack.Pop();
      OperationCollection coll = next as OperationCollection;

      while (coll != null) {
        if (coll.Parallel) {
          IScope scope = ((IAtomicOperation)coll.Where(x => x is IAtomicOperation).First()).Scope;
          IEnumerable<IOperation> finishedOperations = ExecuteOnHive(coll.Select(op => new OperationJob(op)));
          foreach (IOperation op in finishedOperations) {
            ReIntegrateScope(scope, op);
          }
        } else {
          for (int i = coll.Count - 1; i >= 0; i--)
            ExecutionStack.Push(coll[i]);
        }
        next = ExecutionStack.Count > 0 ? ExecutionStack.Pop() : null;
        coll = next as OperationCollection;
      }

      IAtomicOperation operation = next as IAtomicOperation;
      if (operation != null) {
        try {
          currentOperator = operation.Operator;
          ExecutionStack.Push(operation.Operator.Execute((IExecutionContext)operation));
        }
        catch (Exception ex) {
          ExecutionStack.Push(operation);
          OnExceptionOccurred(new OperatorExecutionException(operation.Operator, ex));
          Pause();
        }
        if (operation.Operator.Breakpoint) {
          Log.LogMessage(string.Format("Breakpoint: {0}", operation.Operator.Name != string.Empty ? operation.Operator.Name : operation.Operator.ItemName));
          Pause();
        }
      }
    }

    private static void ReIntegrateScope(IScope scope, IOperation operation) {
      if (operation is IAtomicOperation) {
        MergeScopes(scope, ((IAtomicOperation)operation).Scope);
      } else if (operation is OperationCollection) {
        foreach (IOperation op in (OperationCollection)operation) {
          ReIntegrateScope(scope, op);
        }
      }
    }

    private static void MergeScopes(IScope target, IScope source) {
      foreach (IVariable variable in source.Variables) {
        if (target.Variables.ContainsKey(variable.Name)) {
          if (target.Variables[variable.Name] != source.Variables[variable.Name]) {
            // problem: ResultCollection are not equal altough they contain the same values.
            //             - either implement ResultCollection.Equals, or consider ResultCollection as special case here
            //throw new ScopeMergeException(string.Format("Variable {0} already exists in target scope and has a different value ({1}, {2}).", variable.Name, target.Variables[variable.Name].Value, variable.Value));
          }
        } else {
          target.Variables.Add(variable);
        }
      }

      //this is not useful on the parentscope of the initial scope
      //foreach (IScope subScope in source.SubScopes) {
      //  target.SubScopes.Add(subScope);
      //}

      if (target.Parent != null && source.Parent != null)
        MergeScopes(target.Parent, source.Parent);
    }


    private IEnumerable<IOperation> TestExecuteOnHive(IEnumerable<OperationJob> operationJobs) {
      //var clonedJobs = operationJobs.Select(x => (OperationJob)x.Clone()).ToArray();
      var results = new List<IOperation>();

      foreach (var job in operationJobs) {
        job.Start();
        while (job.ExecutionState != Core.ExecutionState.Stopped) {
          Thread.Sleep(100);
        }
        results.Add(job.Operation);
      }
      return results;
    }

    private IEnumerable<IOperation> ExecuteOnHive(IEnumerable<OperationJob> operationJobs) {
      IEnumerable<Guid> jobIds;
      JobResultList results;


      using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
        List<JobDto> jobs = new List<JobDto>();
        foreach (OperationJob operationJob in operationJobs) {
          var groups = new string[] { "HEAL" };
          SerializedJob serializedJob = new SerializedJob();
          serializedJob.SerializedJobData = SerializedJob.Serialize(operationJob);
          serializedJob.JobInfo = new JobDto();
          serializedJob.JobInfo.State = JobState.Offline;
          serializedJob.JobInfo.CoresNeeded = 1;
          serializedJob.JobInfo.PluginsNeeded = ApplicationManager.Manager.Plugins.Select(x => new HivePluginInfoDto { Name = x.Name, Version = x.Version }).ToList();
          ResponseObject<JobDto> response = service.Obj.AddJobWithGroupStrings(serializedJob, groups);
          jobs.Add(response.Obj);
        }
        jobIds = jobs.Select(x => x.Id);
        results = service.Obj.GetJobResults(jobIds).Obj;
      }

      while (!results.All(
          x => x.State == JobState.Finished ||
          x.State == JobState.Failed ||
          x.State == JobState.Aborted)) {
        Thread.Sleep(5000);
        using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
          results = service.Obj.GetJobResults(jobIds).Obj;
        }
      }

      // all finished
      List<OperationJob> finishedJobs = new List<OperationJob>();
      using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
        foreach (Guid jobId in jobIds) {
          SerializedJob serializedJob = service.Obj.GetLastSerializedResult(jobId).Obj;
          OperationJob operationJob = SerializedJob.Deserialize<OperationJob>(serializedJob.SerializedJobData);
          finishedJobs.Add(operationJob);
        }
      }

      // delete jobs
      // TODO

      return finishedJobs.Select(x => x.Operation);
    }

    public override void Pause() {
      base.Pause();
      if (currentOperator != null) currentOperator.Abort();
    }
    public override void Stop() {
      base.Stop();
      if (currentOperator != null) currentOperator.Abort();
    }
  }
}