source: branches/OptimizationNetworks/HeuristicLab.Networks.IntegratedOptimization/3.3/OrchestratedAlgorithmNode.cs @ 14586

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using HeuristicLab.Common;
using HeuristicLab.Core;
using HeuristicLab.Core.Networks;
using HeuristicLab.Data;
using HeuristicLab.Operators;
using HeuristicLab.Optimization;
using HeuristicLab.Persistence.Default.CompositeSerializers.Storable;

namespace HeuristicLab.Networks.IntegratedOptimization {
  [Item("OrchestratedAlgorithmNode", "A node of a network which contains a HeuristicLab algorithm and can be orchestrated by an orchestrator.")]
  [StorableClass]
  public class OrchestratedAlgorithmNode : AlgorithmNode {
    #region Constants
    private const string OrchestrationPortName = "OrchestrationPort";
    private const string EvaluationPortName = "EvaluationPort";
    #endregion

    private bool sendResultMessage = true;

    #region Ports
    public IConfigurationPort OrchestrationPort {
      get { return (IConfigurationPort)Ports[OrchestrationPortName]; }
    }
    public IConfigurationPort EvaluationPort {
      get { return (IConfigurationPort)Ports[EvaluationPortName]; }
    }
    #endregion

    [Storable]
    private IHookOperator evalHook;
    public IHookOperator EvalHook {
      get { return evalHook; }
      set {
        if (evalHook == value) return;
        evalHook = value;
        evalHook.Port = EvaluationPort;
      }
    }

    [Storable]
    public bool CloneAlgorithm { get; set; }

    [StorableConstructor]
    protected OrchestratedAlgorithmNode(bool deserializing) : base(deserializing) { }
    protected OrchestratedAlgorithmNode(OrchestratedAlgorithmNode original, Cloner cloner) : base(original, cloner) {
      CloneAlgorithm = original.CloneAlgorithm;
      EvalHook = cloner.Clone(original.EvalHook);

      OrchestrationPort.MessageReceived += OrchestrationPort_MessageReceived;
    }
    public OrchestratedAlgorithmNode() : this("OrchestratedAlgorithmNode") { }
    public OrchestratedAlgorithmNode(string name) : base(name) {
      #region Configure Ports
      #region OrchestrationPort
      var orchestrationPort = new ConfigurationPort(OrchestrationPortName);
      orchestrationPort.MessageReceived += OrchestrationPort_MessageReceived;
      Ports.Add(orchestrationPort);
      #endregion

      #region EvaluationPort
      var evaluationPort = new ConfigurationPort(EvaluationPortName);
      Ports.Add(evaluationPort);
      #endregion
      #endregion
    }

    public override IDeepCloneable Clone(Cloner cloner) {
      return new OrchestratedAlgorithmNode(this, cloner);
    }

    [StorableHook(HookType.AfterDeserialization)]
    private void AfterDeserialization() {
      OrchestrationPort.MessageReceived += OrchestrationPort_MessageReceived;
    }

    #region Port Events
    private void OrchestrationPort_MessageReceived(object sender, EventArgs<IMessage, CancellationToken> e) {
      var message = ((EnumValue<OrchestrationMessage>)e.Value["OrchestrationMessage"]).Value;
      IAlgorithm alg = null;

      if (message.HasFlag(OrchestrationMessage.Prepare)) {
        // TODO: what if CloneAlgorithm == true?
        IMessageValue problemMsgVal;
        if (e.Value.Values.TryGetValue("Problem", out problemMsgVal)) {
          alg = CloneAlgorithm ? (IAlgorithm)Algorithm.Clone() : Algorithm;
          var prob = (IHeuristicOptimizationProblem)problemMsgVal.Value.Clone();

          if (message.HasFlag(OrchestrationMessage.QualityAdaption)) {
            var instEval = prob.Evaluator as InstrumentedOperator;
            if (instEval != null && EvalHook != null) instEval.AfterExecutionOperators.Add(EvalHook);
          }

          var parameters = new Dictionary<string, IItem>();
          alg.CollectParameterValues(parameters);
          var entry = parameters.SingleOrDefault(x => x.Value is IMultiAnalyzer);
          var checkedStates = new Dictionary<Type, bool>();
          if (entry.Key != null) {
            var multiAnalyzer = entry.Value as IMultiAnalyzer;
            foreach (var analyzer in multiAnalyzer.Operators)
              checkedStates.Add(analyzer.GetType(), multiAnalyzer.Operators.ItemChecked(analyzer));
          }

          alg.Problem = prob;

          if (entry.Key != null) {
            var multiAnalyzer = entry.Value as IMultiAnalyzer;
            foreach (var state in checkedStates) {
              var analyzer = multiAnalyzer.Operators.Single(x => x.GetType() == state.Key);
              multiAnalyzer.Operators.SetItemCheckedState(analyzer, state.Value);
            }
          }

          alg.Prepare();
        }
      }

      if (message.HasFlag(OrchestrationMessage.Start)) {
        // TODO: what if CloneAlgorithm == true?
        StartAlgorithm(alg ?? (CloneAlgorithm ? (IAlgorithm)Algorithm.Clone() : Algorithm));

        if (sendResultMessage) {
          var msg = OrchestrationPort.PrepareMessage();
          msg["Results"] = (ResultCollection)alg.Results.Clone();
          OrchestrationPort.SendMessage(msg);
        } else sendResultMessage = true;
      }

      if (message.HasFlag(OrchestrationMessage.Pause)) {
        // TODO: what if CloneAlgorithm == true?
        sendResultMessage = false;
        (alg ?? (CloneAlgorithm ? (IAlgorithm)Algorithm.Clone() : Algorithm)).Pause();
      }

      if (message.HasFlag(OrchestrationMessage.Stop)) {
        // TODO: what if CloneAlgorithm == true?
        (alg ?? (CloneAlgorithm ? (IAlgorithm)Algorithm.Clone() : Algorithm)).Stop();
      }
    }

    private void StartAlgorithm(IAlgorithm a) {
      var trigger = new ManualResetEvent(false);
      Exception ex = null;
      a.Stopped += (src, e) => { trigger.Set(); };
      a.ExceptionOccurred += (src, e) => { ex = e.Value; trigger.Set(); };
      a.Prepare();
      a.Start();
      trigger.WaitOne();
      if (ex != null) throw ex;
    }
    #endregion
  }
}