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

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 IMessagePort OrchestrationPort {
      get { return (IMessagePort)Ports[OrchestrationPortName]; }
    }
    public IMessagePort EvaluationPort {
      get { return (IMessagePort)Ports[EvaluationPortName]; }
    }
    #endregion

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

    [StorableConstructor]
    protected OrchestratedAlgorithmNode(bool deserializing) : base(deserializing) { }
    protected OrchestratedAlgorithmNode(OrchestratedAlgorithmNode original, Cloner cloner) : base(original, cloner) {
      EvalHook = cloner.Clone(original.EvalHook);
      RegisterEvents();
    }
    public OrchestratedAlgorithmNode() : this("OrchestratedAlgorithmNode") { }
    public OrchestratedAlgorithmNode(string name) : base(name) {
      var orchestrationPort = new MessagePort(OrchestrationPortName);
      Ports.Add(orchestrationPort);

      var evaluationPort = new MessagePort(EvaluationPortName);
      Ports.Add(evaluationPort);

      RegisterEvents();
    }

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

    [StorableHook(HookType.AfterDeserialization)]
    private void AfterDeserialization() {
      RegisterEvents();
    }

    private void RegisterEvents() {
      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;

      #region Prepare
      if (message.HasFlag(OrchestrationMessage.Prepare)) {
        switch (Algorithm.ExecutionState) {
          case ExecutionState.Prepared:
          case ExecutionState.Paused:
          case ExecutionState.Stopped:
            IMessageValue problemMsgVal;
            if (e.Value.Values.TryGetValue("Problem", out problemMsgVal)) {
              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);
              }

              IParameter analyzerParameter = null;
              IMultiAnalyzer multiAnalyzer = null;
              var checkedStates = new Dictionary<Type, bool>();
              if (Algorithm.Parameters.TryGetValue("Analyzer", out analyzerParameter)) {
                multiAnalyzer = analyzerParameter.ActualValue as IMultiAnalyzer;
                foreach (var item in multiAnalyzer.Operators)
                  checkedStates.Add(item.GetType(), multiAnalyzer.Operators.ItemChecked(item));
              }

              Algorithm.Problem = prob;

              if (multiAnalyzer != null) {
                if (Algorithm.Parameters.TryGetValue("Analyzer", out analyzerParameter)) {
                  multiAnalyzer = analyzerParameter.ActualValue as IMultiAnalyzer;
                  if (multiAnalyzer != null && checkedStates.Any()) {
                    foreach (var analyzer in multiAnalyzer.Operators) {
                      bool checkedState;
                      checkedStates.TryGetValue(analyzer.GetType(), out checkedState);
                      multiAnalyzer.Operators.SetItemCheckedState(analyzer, checkedState);
                    }
                  }
                }
              }

              Algorithm.Prepare();
            }
            break;
        }
      }
      #endregion

      #region Start
      if (message.HasFlag(OrchestrationMessage.Start)) {
        switch (Algorithm.ExecutionState) {
          case ExecutionState.Prepared:
          case ExecutionState.Paused:
            var trigger = new ManualResetEvent(false);
            Exception ex = null;

            EventHandler onStopped = null;
            EventHandler<EventArgs<Exception>> onExceptionOccurred = null;
            Algorithm.Stopped += onStopped = (src, args) => {
              Algorithm.Stopped -= onStopped;
              Algorithm.ExceptionOccurred -= onExceptionOccurred;
              trigger.Set();
            };
            Algorithm.ExceptionOccurred += onExceptionOccurred = (src, args) => {
              Algorithm.Stopped -= onStopped;
              Algorithm.ExceptionOccurred -= onExceptionOccurred;
              ex = args.Value;
              trigger.Set();
            };

            Algorithm.Start();
            trigger.WaitOne();

            if (ex != null) throw ex;

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

      #region Pause
      if (message.HasFlag(OrchestrationMessage.Pause)) {
        if (Algorithm.ExecutionState == ExecutionState.Started) {
          sendResultMessage = false;
          try {
            Algorithm.Pause();
          } catch (InvalidOperationException) {
            // ExecutionState might have changed since we accepted the message
          }
        }
      }
      #endregion

      #region Stop
      if (message.HasFlag(OrchestrationMessage.Stop)) {
        switch (Algorithm.ExecutionState) {
          case ExecutionState.Started:
          case ExecutionState.Paused:
            try {
              Algorithm.Stop();
            } catch (InvalidOperationException) {
              // ExecutionState might have changed since we accepted the message
            } catch (NullReferenceException) {
              // BasicAlgorithm might have stopped since we accepted the message
              // CancellationTokenSource is null in this case
            }
            break;
        }
      }
      #endregion
    }
    #endregion
  }
}