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

#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 job.Engine.OperatorGraph; }
    //}

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

    public TimeSpan ExecutionTime {
      get { return job.Engine.ExecutionTime; }
    }

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

    //public bool Running {
    //  get { return job.Engine.ExecutionState == Core.ExecutionState.Started; }
    //}

    //public bool Canceled {
    //  get { return job.Engine.Canceled; }
    //}

    //public bool Terminated {
    //  get { return job.Engine.Terminated; }
    //}

    public ExecutionState ExecutionState {
      get { return job.Engine.ExecutionState; }
    }

    protected override void ProcessNextOperation() {
      var jobObj = CreateJobObj();
      IExecutionEngineFacade executionEngineFacade = ServiceLocator.CreateExecutionEngineFacade(HiveServerUrl);

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

      /*if(!String.Empty.Equals(RessourceIds)) {
        String[] ids = RessourceIds.Split(';');
        foreach (string sid in ids) {        
          try {
            System.Guid gid = new Guid(sid);
            jobObj.JobInfo.AssignedResourceIds.Add(gid);
          } catch(Exception ex) {
            
          }   
        }
      } */

      int loops = 1;

      Int32.TryParse(MultiSubmitCount, out loops);
      if (loops == 0)
        loops = 1;

      for (int i = 0; i < loops; i++) {
        ResponseObject<Contracts.BusinessObjects.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;

        //ControlManager.Manager.ShowControl(job.Engine.CreateView());
        //OnChanged();

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

    private HeuristicLab.Hive.Contracts.BusinessObjects.SerializedJob CreateJobObj() {
      HeuristicLab.Hive.Contracts.BusinessObjects.JobDto jobObj = new HeuristicLab.Hive.Contracts.BusinessObjects.JobDto();

      // create xml document with <Root> as root-node and the persisted job as child
      MemoryStream memStream = new MemoryStream();
      XmlGenerator.Serialize(job, memStream);

      // convert memStream to byte-array
      HeuristicLab.Hive.Contracts.BusinessObjects.SerializedJob executableJobObj = new HeuristicLab.Hive.Contracts.BusinessObjects.SerializedJob();
      executableJobObj.JobInfo = jobObj;
      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();

      jobObj.CoresNeeded = 1;
      jobObj.PluginsNeeded = pluginsNeeded;
      jobObj.State = HeuristicLab.Hive.Contracts.BusinessObjects.State.offline;
      return executableJobObj;
    }

    /// <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.");
      IEnumerable<IPluginDescription> plugins = from t in types
                                                select ApplicationManager.Manager.GetDeclaringPlugin(t);
      return plugins;
    }

    /// <summary>
    /// Recursively finds all types used in type which are in a namespace which starts with namespaceStart
    /// </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) {

      // search is not performed on attributes

      if (!types.Contains(type) && type.Namespace.StartsWith(namespaceStart)) {
        types.Add(type);

        // 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);
        }

        // constructors : maybe constructors them out (?)
        foreach (ConstructorInfo info in type.GetConstructors()) {
          foreach (ParameterInfo paramInfo in info.GetParameters()) {
            FindTypes(paramInfo.ParameterType, 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);
        }
      }
    }

    //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 Reset() {
    //  abortRequested = false;

    //  throw new NotImplementedException("[chn] how to reset Engine?");
    //  //job.Engine.Reset();

    //  jobId = Guid.NewGuid();
    //  OnInitialized();
    //}

    //public void RequestSnapshot() {
    //  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 = (Job)PersistenceManager.RestoreFromGZip(jobResult.SerializedJobData);
    //    //ControlManager.Manager.ShowControl(job.Engine.CreateView());
    //    throw new NotImplementedException("TODO[chn]use persistency-3.3");
    //  }
    //  //HiveLogger.Debug("HiveEngine: Results-polling - Exception!");
    //  //Exception ex = new Exception(response.Obj.Exception.Message);
    //  //ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(ex); });
    //}

    //public override System.Xml.XmlNode GetXmlNode(string name, System.Xml.XmlDocument document, IDictionary<Guid, IStorable> persistedObjects) {
    //  XmlNode node = base.GetXmlNode(name, document, persistedObjects);
    //  XmlAttribute attr = document.CreateAttribute("HiveServerUrl");
    //  attr.Value = HiveServerUrl;
    //  node.Attributes.Append(attr);
    //  node.AppendChild(PersistenceManager.Persist("Job", job, document, persistedObjects));
    //  return node;
    //}

    //public override void Populate(System.Xml.XmlNode node, IDictionary<Guid, IStorable> restoredObjects) {
    //  base.Populate(node, restoredObjects);
    //  HiveServerUrl = node.Attributes["HiveServerUrl"].Value;
    //  job = (Job)PersistenceManager.Restore(node.SelectSingleNode("Job"), restoredObjects);
    //}


  }
}