source: branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Engine/3.3/HiveEngine.cs @ 4111

#region License Information
/* HeuristicLab
 * Copyright (C) 2002-2008 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.Text;
using HeuristicLab.Core;
using System.Threading;
using System.IO;
using System.Xml;
using System.IO.Compression;
using System.Linq;
using HeuristicLab.Common;
using HeuristicLab.Hive.JobBase;
using HeuristicLab.Hive.Contracts.Interfaces;
using HeuristicLab.Hive.Contracts;
using HeuristicLab.PluginInfrastructure;
using HeuristicLab.Hive.Contracts.BusinessObjects;
using HeuristicLab.Tracing;
using HeuristicLab.Persistence.Default.Xml;
using System.Reflection;

namespace HeuristicLab.Hive.Engine {
  /// <summary>
  /// Represents an engine that executes its operator-graph on the hive.
  /// in parallel.
  /// </summary>
  public class HiveEngine : HeuristicLab.Core.Engine {
    private const int SNAPSHOT_POLLING_INTERVAL_MS = 1000;
    private const int RESULT_POLLING_INTERVAL_MS = 10000;
    private const int MAX_SNAPSHOT_RETRIES = 20;
    private Guid jobId;
    private Job job;
    private object locker = new object();
    private volatile bool abortRequested;
    private ThreadPriority priority;

    public string HiveServerUrl { get; set; }
    public string MultiSubmitCount { get; set; }
    public string RessourceIds { get; set; }

    public HiveEngine() {
      // <debug>
      HiveServerUrl = "net.tcp://10.42.1.153:9000/ExecutionEngine/ExecutionEngineFacade";
      RessourceIds = "MyGroup";
      // </debug>

      job = new Job();
      abortRequested = false;
      Priority = ThreadPriority.Lowest;
    }

    //public OperatorGraph OperatorGraph {
    //  get { return currentJob.Engine.; }
    //}

    //public IScope GlobalScope {
    //  get { return job.Engine.GlobalScope; }
    //}

    public ThreadPriority Priority {
      get { return this.priority; }
      set { this.priority = value; }
    }

    public new void Prepare(IOperation initialOperation) {
      this.job.Prepare(initialOperation);
      OnPrepared();
    }

    public override void Start() {
      SerializedJob jobObj = CreateSerializedJob();
      IExecutionEngineFacade executionEngineFacade = ServiceLocator.CreateExecutionEngineFacade(HiveServerUrl);

      List<string> groups = new List<string>();
      if (!String.IsNullOrEmpty(RessourceIds)) {
        groups.AddRange(RessourceIds.Split(';'));
      }

      ResponseObject<JobDto> res = executionEngineFacade.AddJobWithGroupStrings(jobObj, groups);
      jobId = res.Obj.Id;

      StartResultPollingThread();
    }

    private void StartResultPollingThread() {
      // start a backgroud thread to poll the final result of the job
      Thread t = new Thread(() => {
        IExecutionEngineFacade executionEngineFacade = ServiceLocator.CreateExecutionEngineFacade(HiveServerUrl);
        ResponseObject<SerializedJob> response = null;
        Job restoredJob = null;
        do {
          Thread.Sleep(RESULT_POLLING_INTERVAL_MS);
          lock (locker) {
            Logger.Debug("HiveEngine: Results-polling - GetLastResult");
            response = executionEngineFacade.GetLastSerializedResult(jobId, false, false);
            Logger.Debug("HiveEngine: Results-polling - Server: " + response.StatusMessage + " success: " + response.Success);
            // loop while 
            // 1. the user doesn't request an abort 
            // 2. there is a problem with server communication (success==false)
            // 3. no result for the job is available yet (response.Obj==null)
            // 4. the result that we get from the server is a snapshot and not the final result
            if (abortRequested) return;
            if (response.Success && response.Obj != null && response.StatusMessage != ApplicationConstants.RESPONSE_JOB_RESULT_NOT_YET_HERE) {
              Logger.Debug("HiveEngine: Results-polling - Got result!");
              restoredJob = XmlParser.Deserialize<Job>(new MemoryStream(response.Obj.SerializedJobData));
              Logger.Debug("HiveEngine: Results-polling - IsSnapshotResult: " + (restoredJob.Progress < 1.0));
            }
          }
        } while (restoredJob == null || (restoredJob.Progress < 1.0));

        job = restoredJob;

        // [chn] how to show the View in 3.3?
        //ControlManager.Manager.ShowControl(job.Engine.CreateView());
        //OnChanged();

        //OnFinished();
      });
      Logger.Debug("HiveEngine: Starting results-polling thread");
      t.Start();
    }

    public override void Pause() {
      base.Pause();
      if (job != null) {
        IExecutionEngineFacade executionEngineFacade = ServiceLocator.CreateExecutionEngineFacade(HiveServerUrl);
        executionEngineFacade.AbortJob(jobId);
      }
    }

    public override void Stop() {
      base.Stop();
      if (job != null) {
        IExecutionEngineFacade executionEngineFacade = ServiceLocator.CreateExecutionEngineFacade(HiveServerUrl);
        executionEngineFacade.AbortJob(jobId);
      }
    }

    /// <summary>
    /// Creates a SerializedJob Object which contains the currentJob as binary array
    /// and some metainfo about that job.
    /// </summary>
    private SerializedJob CreateSerializedJob() {
      JobDto jobDto = new JobDto();

      // serialize current job
      MemoryStream memStream = new MemoryStream();
      XmlGenerator.Serialize(job, memStream);

      // convert memStream to byte-array
      SerializedJob executableJobObj = new SerializedJob();
      executableJobObj.JobInfo = jobDto;
      executableJobObj.SerializedJobData = memStream.ToArray();

      // find out which which plugins are needed to recreate the type
      List<HivePluginInfoDto> pluginsNeeded = (
        from p in GetDeclaringPlugins(job.GetType())
        select new HivePluginInfoDto() {
          Name = p.Name,
          Version = p.Version,
          BuildDate = p.BuildDate,
        }).ToList();

      jobDto.CoresNeeded = 1;
      jobDto.PluginsNeeded = pluginsNeeded;
      jobDto.State = State.offline;
      return executableJobObj;
    }


    //public event EventHandler Initialized;
    ///// <summary>
    ///// Fires a new <c>Initialized</c> event.
    ///// </summary>
    //protected virtual void OnInitialized() {
    //  if (Initialized != null)
    //    Initialized(this, new EventArgs());
    //}

    //public event EventHandler<EventArgs<IOperation>> OperationExecuted;
    ///// <summary>
    ///// Fires a new <c>OperationExecuted</c> event.
    ///// </summary>
    ///// <param name="operation">The operation that has been executed.</param>
    //protected virtual void OnOperationExecuted(IOperation operation) {
    //  if (OperationExecuted != null)
    //    OperationExecuted(this, new EventArgs<IOperation>(operation));
    //}

    //public event EventHandler<EventArgs<Exception>> ExceptionOccurred;
    ///// <summary>
    ///// Aborts the execution and fires a new <c>ExceptionOccurred</c> event.
    ///// </summary>
    ///// <param name="exception">The exception that was thrown.</param>
    //protected virtual void OnExceptionOccurred(Exception exception) {
    //  Abort();
    //  if (ExceptionOccurred != null)
    //    ExceptionOccurred(this, new EventArgs<Exception>(exception));
    //}

    //public event EventHandler ExecutionTimeChanged;
    ///// <summary>
    ///// Fires a new <c>ExecutionTimeChanged</c> event.
    ///// </summary>
    //protected virtual void OnExecutionTimeChanged() {
    //  if (ExecutionTimeChanged != null)
    //    ExecutionTimeChanged(this, new EventArgs());
    //}

    //public event EventHandler Finished;
    ///// <summary>
    ///// Fires a new <c>Finished</c> event.
    ///// </summary>
    //protected virtual void OnFinished() {
    //  if (Finished != null)
    //    Finished(this, new EventArgs());
    //}

    //public void Abort() {
    //  abortRequested = true;
    //  // RequestSnapshot();
    //  IExecutionEngineFacade executionEngineFacade = ServiceLocator.CreateExecutionEngineFacade(HiveServerUrl);
    //  executionEngineFacade.AbortJob(jobId);
    //  //OnChanged();
    //  OnFinished();
    //}

    public void RequestSnapshot() {
      if (job != null) {
        IExecutionEngineFacade executionEngineFacade = ServiceLocator.CreateExecutionEngineFacade(HiveServerUrl);
        int retryCount = 0;
        ResponseObject<SerializedJob> response;
        lock (locker) {
          Logger.Debug("HiveEngine: Abort - RequestSnapshot");
          Response snapShotResponse = executionEngineFacade.RequestSnapshot(jobId);
          if (snapShotResponse.StatusMessage == ApplicationConstants.RESPONSE_JOB_IS_NOT_BEEING_CALCULATED) {
            // job is finished already
            Logger.Debug("HiveEngine: Abort - GetLastResult(false)");
            response = executionEngineFacade.GetLastSerializedResult(jobId, false, false);
            Logger.Debug("HiveEngine: Abort - Server: " + response.StatusMessage + " success: " + response.Success);
          } else {
            // server sent snapshot request to client
            // poll until snapshot is ready
            do {
              Thread.Sleep(SNAPSHOT_POLLING_INTERVAL_MS);
              Logger.Debug("HiveEngine: Abort - GetLastResult(true)");
              response = executionEngineFacade.GetLastSerializedResult(jobId, false, true);
              Logger.Debug("HiveEngine: Abort - Server: " + response.StatusMessage + " success: " + response.Success);
              retryCount++;
              // loop while
              // 1. problem with communication with server
              // 2. job result not yet ready
            } while (
              (retryCount < MAX_SNAPSHOT_RETRIES) && (
              !response.Success ||
              response.StatusMessage == ApplicationConstants.RESPONSE_JOB_RESULT_NOT_YET_HERE)
              );
          }

        }
        SerializedJob jobResult = response.Obj;
        if (jobResult != null) {
          Logger.Debug("HiveEngine: Results-polling - Got result!");

          job = XmlParser.Deserialize<Job>(new MemoryStream(jobResult.SerializedJobData));

          throw new NotImplementedException("[chn] how to create a view in 3.3 and why should i do this here? shouldnt the caller of this method receive a result and decide what to do?");
          //ControlManager.Manager.ShowControl(job.Engine.CreateView());
        }
      }
      //HiveLogger.Debug("HiveEngine: Results-polling - Exception!");
      //Exception ex = new Exception(response.Obj.Exception.Message);
      //ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(ex); });
    }

    protected override void ProcessNextOperation() {
      throw new NotSupportedException();
    }

    #region Required Plugin Search
    /// <summary>
    /// Returns a list of plugins in which the type itself and all members 
    /// of the type are declared. Objectgraph is searched recursively.
    /// </summary>
    private IEnumerable<IPluginDescription> GetDeclaringPlugins(Type type) {
      HashSet<Type> types = new HashSet<Type>();
      FindTypes(type, types, "HeuristicLab.");
      return GetDeclaringPlugins(types);
    }

    /// <summary>
    /// Returns the plugins (including dependencies) in which the given types are declared
    /// </summary>
    private IEnumerable<IPluginDescription> GetDeclaringPlugins(IEnumerable<Type> types) {
      HashSet<IPluginDescription> plugins = new HashSet<IPluginDescription>();
      foreach (Type t in types) {
        FindDeclaringPlugins(ApplicationManager.Manager.GetDeclaringPlugin(t), plugins);
      }
      return plugins;
    }

    /// <summary>
    /// Finds the dependencies of the given plugin and adds it to the plugins hashset.
    /// Also searches the dependencies recursively.
    /// </summary>
    private void FindDeclaringPlugins(IPluginDescription plugin, HashSet<IPluginDescription> plugins) {
      if (!plugins.Contains(plugin)) {
        plugins.Add(plugin);
        foreach (IPluginDescription dependency in plugin.Dependencies) {
          FindDeclaringPlugins(dependency, plugins);
        }
      }
    }

    /// <summary>
    /// Recursively finds all types used in type which are in a namespace which starts with namespaceStart
    /// Be aware that search is not performed on attributes
    /// </summary>
    /// <param name="type">the type to be searched</param>
    /// <param name="types">found types will be stored there, needed in order to avoid duplicates</param>
    /// <param name="namespaceStart">only types from namespaces which start with this will be searched and added</param>
    private void FindTypes(Type type, HashSet<Type> types, string namespaceStart) {
      if (!types.Contains(type) && type.Namespace.StartsWith(namespaceStart)) {
        types.Add(type);

        // constructors
        foreach (ConstructorInfo info in type.GetConstructors()) {
          foreach (ParameterInfo paramInfo in info.GetParameters()) {
            FindTypes(paramInfo.ParameterType, types, namespaceStart);
          }
        }

        // interfaces
        foreach (Type t in type.GetInterfaces()) {
          FindTypes(t, types, namespaceStart);
        }

        // events
        foreach (EventInfo info in type.GetEvents()) {
          FindTypes(info.EventHandlerType, types, namespaceStart);
          FindTypes(info.DeclaringType, types, namespaceStart);
        }

        // properties
        foreach (PropertyInfo info in type.GetProperties()) {
          FindTypes(info.PropertyType, types, namespaceStart);
        }

        // fields
        foreach (FieldInfo info in type.GetFields()) {
          FindTypes(info.FieldType, types, namespaceStart);
        }

        // methods
        foreach (MethodInfo info in type.GetMethods()) {
          foreach (ParameterInfo paramInfo in info.GetParameters()) {
            FindTypes(paramInfo.ParameterType, types, namespaceStart);
          }
          FindTypes(info.ReturnType, types, namespaceStart);
        }
      }
    }
    #endregion
  }
}