source: branches/2205_OptimizationNetworks/HeuristicLab.Networks.IntegratedOptimization/3.3/OrchestratedAlgorithmNode.cs @ 16539

#region License Information
/* HeuristicLab
 * Copyright (C) 2002-2017 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.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) {
      Ports.Add(CreateOrchestrationPort<IProblem>(OrchestrationPortName));
      Ports.Add(CreateEvaluationPort<IItem>(EvaluationPortName, "Solution", "Quality"));

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

    protected IMessagePort CreateOrchestrationPort<T>(string portName) where T : class, IProblem {
      var orchestrationPort = new MessagePort(portName);
      orchestrationPort.Parameters.Add(new PortParameter<EnumValue<OrchestrationMessage>>("OrchestrationMessage") {
        Type = PortParameterType.Input
      });
      orchestrationPort.Parameters.Add(new PortParameter<T>("Problem") {
        Type = PortParameterType.Input
      });
      orchestrationPort.Parameters.Add(new PortParameter<ResultCollection>("Results") {
        Type = PortParameterType.Output
      });
      return orchestrationPort;
    }

    protected IMessagePort CreateEvaluationPort<T>(string portName, string solutionName, string qualityName) where T : class, IItem {
      var evaluationPort = new MessagePort(portName);
      evaluationPort.Parameters.Add(new PortParameter<T>(solutionName) {
        Type = PortParameterType.Output
      });
      evaluationPort.Parameters.Add(new PortParameter<DoubleValue>(qualityName) {
        Type = PortParameterType.Input | PortParameterType.Output
      });
      return evaluationPort;
    }

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

      #region Prepare
      if (messageFlags.HasFlag(OrchestrationMessage.Prepare)) {
        switch (Algorithm.ExecutionState) {
          case ExecutionState.Prepared:
          case ExecutionState.Paused:
          case ExecutionState.Stopped:
            var problem = (IProblem)requestMessage["Problem"];
            if (problem != null) {
              problem = (IProblem)problem.Clone();

              var heuristicOptimizationProblem = problem as IHeuristicOptimizationProblem;
              if (heuristicOptimizationProblem != null && messageFlags.HasFlag(OrchestrationMessage.SetEvalHook)) {
                var instEval = heuristicOptimizationProblem.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 = problem;

              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;
                      if (!checkedStates.TryGetValue(analyzer.GetType(), out checkedState))
                        checkedState = analyzer.EnabledByDefault;
                      multiAnalyzer.Operators.SetItemCheckedState(analyzer, checkedState);
                    }
                  }
                }
              }
            }


            Algorithm.Prepare(messageFlags.HasFlag(OrchestrationMessage.ClearRuns));
            break;
        }
      }
      #endregion

      #region Start
      if (messageFlags.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) {
              requestMessage["Results"] = (ResultCollection)Algorithm.Results.Clone();
              var responseMessage = OrchestrationPort.PrepareMessage();
              responseMessage["Results"] = (ResultCollection)Algorithm.Results.Clone();
              OrchestrationPort.SendMessage(responseMessage);
            } else sendResultMessage = true;
            break;
        }
      }
      #endregion

      #region Pause
      if (messageFlags.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 (messageFlags.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
  }
}