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

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;
using HeuristicLab.Random;
using System.Threading.Tasks;

namespace HeuristicLab.HiveEngine {
  /// <summary>
  /// Represents an engine that executes operations which can be executed in parallel on the hive
  /// </summary>
  [StorableClass]
  [Item("Hive Engine", "Engine for parallel execution on the hive. You need enable `Parallel` for at least one operator in your operator graph to have all childoperations parallelized. Also those childoperations must not have sideeffects on a higher scope.")]
  public class HiveEngine : Engine {
    private Semaphore maxConcurrentConnections = new Semaphore(4, 4); // avoid too many connections
    private Semaphore maxSerializedJobsInMemory = new Semaphore(4, 4); // avoid memory problems
    private CancellationToken cancellationToken;

    [Storable]
    private IOperator currentOperator;

    [Storable]
    public string ResourceIds { get; set; }

    [Storable]
    private int priority;
    public int Priority {
      get { return priority; }
      set { priority = value; }
    }

    [Storable]
    private TimeSpan executionTimeOnHive;
    public TimeSpan ExecutionTimeOnHive {
      get { return executionTimeOnHive; }
      set {
        if (value != executionTimeOnHive) {
          executionTimeOnHive = value;
          OnExecutionTimeOnHiveChanged();
        }
      }
    }
    
    #region constructors and cloning
    public HiveEngine() {
      ResourceIds = "HEAL";
    }

    [StorableConstructor]
    protected HiveEngine(bool deserializing) : base(deserializing) { }
    protected HiveEngine(HiveEngine original, Cloner cloner)
      : base(original, cloner) {
      this.ResourceIds = original.ResourceIds;
      this.currentOperator = cloner.Clone(original.currentOperator);
      this.priority = original.priority;
      this.executionTimeOnHive = original.executionTimeOnHive;
    }
    public override IDeepCloneable Clone(Cloner cloner) {
      return new HiveEngine(this, cloner);
    }
    #endregion

    #region Events
    protected override void OnPrepared() {
      base.OnPrepared();
      this.ExecutionTimeOnHive = TimeSpan.Zero;
    }

    public event EventHandler ExecutionTimeOnHiveChanged;
    protected virtual void OnExecutionTimeOnHiveChanged() {
      var handler = ExecutionTimeOnHiveChanged;
      if (handler != null) handler(this, EventArgs.Empty);
    }
    #endregion

    protected override void Run(CancellationToken cancellationToken) {
      this.cancellationToken = cancellationToken;
      Run(ExecutionStack);
    }

    private void Run(object state) {
      Stack<IOperation> executionStack = (Stack<IOperation>)state;
      IOperation next;
      OperationCollection coll;
      IAtomicOperation operation;
      TaskScheduler.UnobservedTaskException += new EventHandler<UnobservedTaskExceptionEventArgs>(TaskScheduler_UnobservedTaskException);

      while (ExecutionStack.Count > 0) {
        cancellationToken.ThrowIfCancellationRequested();

        next = ExecutionStack.Pop();
        if (next is OperationCollection) {
          coll = (OperationCollection)next;
          if (coll.Parallel) {
            // clone the parent scope here and reuse it for each operation. otherwise for each job the whole scope-tree first needs to be copied and then cleaned, which causes a lot of work for the Garbage Collector
            IScope parentScopeClone = (IScope)((IAtomicOperation)coll.First()).Scope.Parent.Clone();
            parentScopeClone.SubScopes.Clear();
            parentScopeClone.ClearParentScopes();

            OperationJob[] jobs = new OperationJob[coll.Count];
            for (int i = 0; i < coll.Count; i++) {
              jobs[i] = new OperationJob(coll[i]);
            }

            IScope[] scopes = ExecuteOnHive(jobs, parentScopeClone, cancellationToken);

            for (int i = 0; i < coll.Count; i++) {
              if (coll[i] is IAtomicOperation) {
                ExchangeScope(scopes[i], ((IAtomicOperation)coll[i]).Scope);
              } else if (coll[i] is OperationCollection) {
                // todo ??
              }
            }
          } else {
            for (int i = coll.Count - 1; i >= 0; i--)
              if (coll[i] != null) executionStack.Push(coll[i]);
          }
        } else if (next is IAtomicOperation) {
          operation = (IAtomicOperation)next;
          try {
            next = operation.Operator.Execute((IExecutionContext)operation, cancellationToken);
          }
          catch (Exception ex) {
            ExecutionStack.Push(operation);
            if (ex is OperationCanceledException) throw ex;
            else throw new OperatorExecutionException(operation.Operator, ex);
          }
          if (next != null) ExecutionStack.Push(next);

          if (operation.Operator.Breakpoint) {
            LogMessage(string.Format("Breakpoint: {0}", operation.Operator.Name != string.Empty ? operation.Operator.Name : operation.Operator.ItemName));
            Pause();
          }
        }
      }
    }

    private void TaskScheduler_UnobservedTaskException(object sender, UnobservedTaskExceptionEventArgs e) {
      e.SetObserved(); // avoid crash of process
    }

    private IRandom FindRandomParameter(IExecutionContext ec) {
      try {
        if (ec == null)
          return null;

        foreach (var p in ec.Parameters) {
          if (p.Name == "Random" && p is IValueParameter)
            return ((IValueParameter)p).Value as IRandom;
        }
        return FindRandomParameter(ec.Parent);
      }
      catch { return null; }
    }

    private static void ReIntegrateScope(IAtomicOperation source, IAtomicOperation target) {
      ExchangeScope(source.Scope, target.Scope);
    }

    private static void ExchangeScope(IScope source, IScope target) {
      target.Variables.Clear();
      target.Variables.AddRange(source.Variables);
      target.SubScopes.Clear();
      target.SubScopes.AddRange(source.SubScopes);
      // TODO: validate if parent scopes match - otherwise source is invalid
    }

    /// <summary>
    /// This method blocks until all jobs are finished
    /// TODO: Cancelation needs to be refined; all tasks currently stay in Semaphore.WaitOne after cancelation
    /// </summary>
    /// <param name="jobs"></param>
    private IScope[] ExecuteOnHive(OperationJob[] jobs, IScope parentScopeClone, CancellationToken cancellationToken) {
      LogMessage(string.Format("Executing {0} operations on the hive.", jobs.Length));
      IScope[] scopes = new Scope[jobs.Length];
      object locker = new object();
      IDictionary<Guid, int> jobIndices = new Dictionary<Guid, int>();

      try {
        List<Guid> remainingJobIds = new List<Guid>();
        JobResultList results;
        var pluginsNeeded = ApplicationManager.Manager.Plugins.Select(x => new HivePluginInfoDto { Name = x.Name, Version = x.Version }).ToList();
        int finishedCount = 0;
        int uploadCount = 0;

        // create upload-tasks
        var uploadTasks = new List<Task<JobDto>>();
        for (int i = 0; i < jobs.Length; i++) {
          var job = jobs[i];

          // shuffle random variable to avoid the same random sequence in each operation; todo: does not yet work (it cannot find the random variable)
          IRandom random = FindRandomParameter(job.Operation as IExecutionContext);
          if (random != null)
            random.Reset(random.Next());

          uploadTasks.Add(Task.Factory.StartNew<JobDto>((keyValuePairObj) => {
            return UploadJob(pluginsNeeded, keyValuePairObj, parentScopeClone, cancellationToken);
          }, new KeyValuePair<int, OperationJob>(i, job), cancellationToken));
        }

        Task processUploadedJobsTask = new Task(() => {
          // process finished upload-tasks
          int uploadTasksCount = uploadTasks.Count;
          for (int i = 0; i < uploadTasksCount; i++) {
            cancellationToken.ThrowIfCancellationRequested();

            var uploadTasksArray = uploadTasks.ToArray();
            var task = uploadTasksArray[Task.WaitAny(uploadTasksArray)];
            if (task.Status == TaskStatus.Faulted) {
              LogException(task.Exception);
              throw task.Exception;
            }

            int key = ((KeyValuePair<int, OperationJob>)task.AsyncState).Key;
            JobDto jobDto = task.Result;
            lock (locker) {
              uploadCount++;
              jobIndices.Add(jobDto.Id, key);
              remainingJobIds.Add(jobDto.Id);
            }
            jobs[key] = null; // relax memory
            LogMessage(string.Format("Uploaded job #{0}", key + 1, jobDto.Id));
            uploadTasks.Remove(task);
          }
        }, cancellationToken, TaskCreationOptions.PreferFairness);
        processUploadedJobsTask.Start();

        // poll job-statuses and create tasks for those which are finished
        var downloadTasks = new List<Task<OperationJob>>();
        var executionTimes = new List<TimeSpan>();
        var executionTimeOnHiveBefore = executionTimeOnHive;
        while (processUploadedJobsTask.Status != TaskStatus.RanToCompletion || remainingJobIds.Count > 0) {
          cancellationToken.ThrowIfCancellationRequested();

          Thread.Sleep(10000);
          try {
            using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
              results = service.Obj.GetJobResults(remainingJobIds).Obj;
            }
            var jobsFinished = results.Where(j => j.State == JobState.Finished || j.State == JobState.Failed || j.State == JobState.Aborted);
            finishedCount += jobsFinished.Count();
            if (jobsFinished.Count() > 0) LogMessage(string.Format("Finished: {0}/{1}", finishedCount, jobs.Length));
            ExecutionTimeOnHive = executionTimeOnHiveBefore + executionTimes.Sum() + results.Select(x => x.ExecutionTime).Sum();

            foreach (var result in jobsFinished) {
              if (result.State == JobState.Finished) {
                downloadTasks.Add(Task.Factory.StartNew<OperationJob>((jobIdObj) => {
                  return DownloadJob(jobIndices, jobIdObj, cancellationToken);
                }, result.Id, cancellationToken));
              } else if (result.State == JobState.Aborted) {
                LogMessage(string.Format("Job #{0} aborted (id: {1})", jobIndices[result.Id] + 1, result.Id));
              } else if (result.State == JobState.Failed) {
                LogMessage(string.Format("Job #{0} failed (id: {1}): {2}", jobIndices[result.Id] + 1, result.Id, result.Exception));
              }
              remainingJobIds.Remove(result.Id);
              executionTimes.Add(result.ExecutionTime);
            }
          }
          catch (Exception e) {
            LogException(e);
          }
        }

        // process finished download-tasks
        int downloadTasksCount = downloadTasks.Count;
        for (int i = 0; i < downloadTasksCount; i++) {
          cancellationToken.ThrowIfCancellationRequested();

          var downloadTasksArray = downloadTasks.ToArray();
          var task = downloadTasksArray[Task.WaitAny(downloadTasksArray)];
          var jobId = (Guid)task.AsyncState;
          if (task.Status == TaskStatus.Faulted) {
            LogException(task.Exception);
            throw task.Exception;
          }
          scopes[jobIndices[(Guid)task.AsyncState]] = ((IAtomicOperation)task.Result.Operation).Scope;
          downloadTasks.Remove(task);
        }

        LogMessage(string.Format("All jobs finished (TotalExecutionTime: {0}).", executionTimes.Sum()));
        DeleteJobs(jobIndices);

        return scopes;
      }
      catch (OperationCanceledException e) {
        lock (locker) {
          if (jobIndices != null) DeleteJobs(jobIndices);
        }
        throw e;
      }
      catch (Exception e) {
        lock (locker) {
          if (jobIndices != null) DeleteJobs(jobIndices);
        }
        LogException(e);
        throw e;
      }
    }

    private void DeleteJobs(IDictionary<Guid, int> jobIndices) {
      if (jobIndices.Count > 0) {
        using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
          foreach (Guid jobId in jobIndices.Keys) {
            service.Obj.DeleteJob(jobId);
          }
          LogMessage(string.Format("Deleted {0} jobs on hive.", jobIndices.Count));
          jobIndices.Clear();
        }
      }
    }

    private static object locker = new object();
    private JobDto UploadJob(List<HivePluginInfoDto> pluginsNeeded, object keyValuePairObj, IScope parentScopeClone, CancellationToken cancellationToken) {
      var keyValuePair = (KeyValuePair<int, OperationJob>)keyValuePairObj;
      var groups = ResourceIds.Split(';');
      ResponseObject<JobDto> response = null;
      try {
        maxSerializedJobsInMemory.WaitOne();
        SerializedJob serializedJob = null;
        while (serializedJob == null) { // repeat until success; rare race-conditions occur at serializations (enumeration was changed-exceptions); maybe this is because all the parent-scopes and execution-contexts at some point contain the hiveengine and the Log in here
          cancellationToken.ThrowIfCancellationRequested();
          try {
            lock (Log) {
              serializedJob = new SerializedJob();
            }
          }
          catch (Exception e) {
            LogException(e);
          }
        }
        // clone operation and remove unnecessary scopes; don't do this earlier to avoid memory problems
        lock (locker) {
          ((IAtomicOperation)keyValuePair.Value.Operation).Scope.Parent = parentScopeClone;
          keyValuePair.Value.Operation = (IOperation)keyValuePair.Value.Operation.Clone();
          if (keyValuePair.Value.Operation is IAtomicOperation)
            ((IAtomicOperation)keyValuePair.Value.Operation).Scope.ClearParentScopes();
          serializedJob.SerializedJobData = SerializedJob.Serialize(keyValuePair.Value);
        }
        serializedJob.JobInfo = new JobDto();
        serializedJob.JobInfo.State = JobState.Offline;
        serializedJob.JobInfo.CoresNeeded = 1;
        serializedJob.JobInfo.PluginsNeeded = pluginsNeeded;
        serializedJob.JobInfo.Priority = priority;
        try {
          maxConcurrentConnections.WaitOne();
          while (response == null) { // repeat until success
            cancellationToken.ThrowIfCancellationRequested();
            try {
              using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
                response = service.Obj.AddJobWithGroupStrings(serializedJob, groups);
                serializedJob = null;
              }
            }
            catch (Exception e) {
              LogException(e);
            }
          }
        }
        finally {
          maxConcurrentConnections.Release();
        }
      }
      finally {
        maxSerializedJobsInMemory.Release();
      }
      return response.Obj;
    }

    private OperationJob DownloadJob(IDictionary<Guid, int> jobIndices, object jobIdObj, CancellationToken cancellationToken) {
      Guid jobId = (Guid)jobIdObj;
      SerializedJob serializedJob = null;
      OperationJob operationJob = null;
      try {
        maxSerializedJobsInMemory.WaitOne();
        maxConcurrentConnections.WaitOne();
        while (serializedJob == null) { // repeat until success
          cancellationToken.ThrowIfCancellationRequested();
          try {
            using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
              serializedJob = service.Obj.GetLastSerializedResult(jobId).Obj;
            }
          }
          catch (Exception e) {
            LogException(e);
          }
        }
        operationJob = SerializedJob.Deserialize<OperationJob>(serializedJob.SerializedJobData);
        serializedJob = null;
        LogMessage(string.Format("Downloaded job #{0}", jobIndices[jobId] + 1, jobId));
      }
      finally {
        maxConcurrentConnections.Release();
        maxSerializedJobsInMemory.Release();
      }
      return operationJob;
    }

    /// <summary>
    /// Threadsafe message logging
    /// </summary>
    private void LogMessage(string message) {
      lock (Log) {
        Log.LogMessage(message);
      }
    }

    /// <summary>
    /// Threadsafe exception logging
    /// </summary>
    private void LogException(Exception exception) {
      lock (Log) {
        Log.LogException(exception);
      }
    }
  }

  public static class ScopeExtensions {
    public static void ClearParentScopes(this IScope scope) {
      scope.ClearParentScopes(null);
    }

    public static void ClearParentScopes(this IScope scope, IScope childScope) {
      if (childScope != null) {
        scope.SubScopes.Clear();
        scope.SubScopes.Add(childScope);
      }
      if (scope.Parent != null)
        scope.Parent.ClearParentScopes(scope);
    }
  }

  public static class EnumerableExtensions {
    public static TimeSpan Sum(this IEnumerable<TimeSpan> times) {
      return TimeSpan.FromMilliseconds(times.Select(e => e.TotalMilliseconds).Sum());
    }
  }
}