source: branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave/3.4/Core.cs @ 6203

#region License Information
/* HeuristicLab
 * Copyright (C) 2002-2010 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.Diagnostics;
using System.ServiceModel;
using System.Threading;
using System.Threading.Tasks;
using HeuristicLab.Clients.Hive.SlaveCore.ServiceContracts;
using HeuristicLab.Common;
using HeuristicLab.Core;


namespace HeuristicLab.Clients.Hive.SlaveCore {
  /// <summary>
  /// The core component of the Hive Slave. 
  /// Handles commands sent from the Hive Server. 
  /// </summary>
  public class Core : MarshalByRefObject {
    public EventLog ServiceEventLog { get; set; }

    public static bool abortRequested { get; set; }
    private Semaphore waitShutdownSem = new Semaphore(0, 1);
    public static ILog Log { get; set; }

    private Dictionary<Guid, SlaveJob> slaveJobs = new Dictionary<Guid, SlaveJob>();

    private WcfService wcfService;
    private static HeartbeatManager heartbeatManager;
    public static HeartbeatManager HBManager { get { return heartbeatManager; } }

    private ISlaveCommunication clientCom;
    private ServiceHost slaveComm;

    public Dictionary<Guid, SlaveJob> SlaveJobs {
      get { return slaveJobs; }
    }

    public Core() { }

    /// <summary>
    /// Main Method for the client
    /// </summary>
    public void Start() {
      abortRequested = false;

      try {
        ConfigManager manager = ConfigManager.Instance;
        manager.Core = this;

        //start the client communication service (pipe between slave and slave gui)
        slaveComm = new ServiceHost(typeof(SlaveCommunicationService));
        slaveComm.Open();
        clientCom = SlaveClientCom.Instance.ClientCom;

        // delete all left over job directories
        PluginCache.Instance.CleanPluginTemp();
        clientCom.LogMessage("Hive Slave started");

        wcfService = WcfService.Instance;
        RegisterServiceEvents();

        StartHeartbeats(); // Start heartbeats thread        
        DispatchMessageQueue(); // dispatch messages until abortRequested
      }
      catch (Exception ex) {
        if (ServiceEventLog != null) {
          try {
            ServiceEventLog.WriteEntry("Hive Slave threw exception: " + ex.ToString() + " with stack trace: " + ex.StackTrace);
          }
          catch (Exception) { }
        } else {
          //try to log with clientCom. if this works the user sees at least a message, 
          //else an exception will be thrown anyways.
          clientCom.LogMessage("Uncaught exception: " + ex.ToString() +
            Environment.NewLine + "Core is going to shutdown.");
        }
        ShutdownCore();
      }
      finally {
        DeRegisterServiceEvents();
        waitShutdownSem.Release();
      }
    }

    private void StartHeartbeats() {
      //Initialize the heartbeat     
      if (heartbeatManager == null) {
        heartbeatManager = new HeartbeatManager { Interval = new TimeSpan(0, 0, 10) };
        heartbeatManager.StartHeartbeat();
      }
    }

    private void DispatchMessageQueue() {
      MessageQueue queue = MessageQueue.GetInstance();
      while (!abortRequested) {
        MessageContainer container = queue.GetMessage();
        DetermineAction(container);
        clientCom.StatusChanged(ConfigManager.Instance.GetStatusForClientConsole());
      }
    }

    private void RegisterServiceEvents() {
      WcfService.Instance.Connected += new EventHandler(WcfService_Connected);
      WcfService.Instance.ExceptionOccured += new EventHandler<EventArgs<Exception>>(WcfService_ExceptionOccured);
    }

    private void DeRegisterServiceEvents() {
      WcfService.Instance.Connected -= WcfService_Connected;
      WcfService.Instance.ExceptionOccured -= WcfService_ExceptionOccured;
    }

    void WcfService_ExceptionOccured(object sender, EventArgs<Exception> e) {
      clientCom.LogMessage("Connection to server interruped with exception: " + e.Value.Message);
    }

    void WcfService_Connected(object sender, EventArgs e) {
      clientCom.LogMessage("Connected successfully to Hive server");
    }

    /// <summary>
    /// Reads and analyzes the Messages from the MessageQueue and starts corresponding actions
    /// </summary>
    /// <param name="container">The container, containing the message</param>
    private void DetermineAction(MessageContainer container) {
      clientCom.LogMessage("Message: " + container.Message.ToString() + " for job: " + container.JobId);

      if (container is ExecutorMessageContainer<Guid>) {
        ExecutorMessageContainer<Guid> c = (ExecutorMessageContainer<Guid>)container;
        c.execute();
      } else if (container is MessageContainer) {
        switch (container.Message) {
          case MessageContainer.MessageType.CalculateJob:
            Task.Factory.StartNew((jobIdObj) => {
              Guid jobId = (Guid)jobIdObj;
              SlaveJob newJob = new SlaveJob(this);
              bool start = true;

              lock (slaveJobs) {
                if (slaveJobs.ContainsKey(jobId)) {
                  start = false;
                  clientCom.LogMessage(string.Format("Job with id {0} already exists. Start aborted.", jobId));
                } else {
                  slaveJobs.Add(jobId, newJob);
                }
              }

              if (start) {
                newJob.CalculateJob(jobId);
              }
            }, container.JobId)
            .ContinueWith((t) => {
              // handle exception of task
              clientCom.LogMessage(t.Exception.ToString());
              wcfService.UpdateJobState(container.JobId, JobState.Failed, t.Exception.ToString());
              SlaveStatusInfo.IncrementJobsFailed();
            }, TaskContinuationOptions.OnlyOnFaulted);
            break;
          case MessageContainer.MessageType.ShutdownSlave:
            ShutdownCore();
            break;
          case MessageContainer.MessageType.StopAll:
            DoStopAll();
            break;
          case MessageContainer.MessageType.PauseAll:
            DoPauseAll();
            break;
          case MessageContainer.MessageType.AbortAll:
            DoAbortAll();
            break;
          case MessageContainer.MessageType.AbortJob:
            SlaveStatusInfo.IncrementJobsAborted(); //TODO: move to a sane place

            Task.Factory.StartNew((jobIdObj) => {
              Guid jobId = (Guid)jobIdObj;
              bool abort = true;
              SlaveJob sj = null;

              lock (slaveJobs) {
                if (!slaveJobs.ContainsKey(jobId)) {
                  clientCom.LogMessage(string.Format("Job with id {0} doesn't exist. Abort aborted.", jobId));
                  abort = false;
                } else {
                  sj = slaveJobs[jobId];
                }
              }
              if (abort && !sj.Finished) {
                sj.KillAppDomain();
              }
            }, container.JobId)
             .ContinueWith((t) => {
               // handle exception of task
               clientCom.LogMessage(t.Exception.ToString());
             }, TaskContinuationOptions.OnlyOnFaulted);
            break;
          case MessageContainer.MessageType.StopJob:
            Task.Factory.StartNew((jobIdObj) => {
              Guid jobId = (Guid)jobIdObj;
              bool stop = true;
              SlaveJob sj = null;

              lock (slaveJobs) {
                if (!slaveJobs.ContainsKey(jobId)) {
                  clientCom.LogMessage(string.Format("Job with id {0} doesn't exist. Stop aborted.", jobId));
                  stop = false;
                } else {
                  sj = slaveJobs[jobId];
                }
              }
              if (stop && !sj.Finished) {
                sj.StopJob();
              }
            }, container.JobId)
             .ContinueWith((t) => {
               // handle exception of task
               clientCom.LogMessage(t.Exception.ToString());
             }, TaskContinuationOptions.OnlyOnFaulted);
            break;
          case MessageContainer.MessageType.PauseJob:
            Task.Factory.StartNew((jobIdObj) => {
              Guid jobId = (Guid)jobIdObj;
              bool pause = true;
              SlaveJob sj = null;

              lock (slaveJobs) {
                if (!slaveJobs.ContainsKey(jobId)) {
                  clientCom.LogMessage(string.Format("Job with id {0} doesn't exist. Pause aborted.", jobId));
                  pause = false;
                } else {
                  sj = slaveJobs[jobId];
                }
              }
              if (pause && !sj.Finished) {
                sj.PauseJob();
              }
            }, container.JobId)
             .ContinueWith((t) => {
               // handle exception of task
               clientCom.LogMessage(t.Exception.ToString());
             }, TaskContinuationOptions.OnlyOnFaulted);
            break;
          case MessageContainer.MessageType.Restart:
            DoStartSlave();
            break;
          case MessageContainer.MessageType.Sleep:
            Sleep();
            break;
          case MessageContainer.MessageType.SayHello:
            wcfService.Connect(ConfigManager.Instance.GetClientInfo());
            break;
        }
      } else {
        clientCom.LogMessage("Unknown MessageContainer: " + container);
      }
    }

    /// <summary>
    /// aborts all running jobs, no results are sent back
    /// </summary>
    private void DoAbortAll() {
      lock (slaveJobs) {
        foreach (SlaveJob sj in slaveJobs.Values) {
          if (!sj.Finished) {
            sj.KillAppDomain();
          }
        }
      }
      clientCom.LogMessage("Aborted all jobs!");
    }

    /// <summary>
    /// wait for jobs to finish, then pause client
    /// </summary>
    private void DoPauseAll() {
      clientCom.LogMessage("Pause all received");

      lock (slaveJobs) {
        foreach (SlaveJob sj in slaveJobs.Values) {
          if (!sj.Finished) {
            sj.PauseJob();
          }
        }
      }
    }

    /// <summary>
    /// pause slave immediately
    /// </summary>
    private void DoStopAll() {
      clientCom.LogMessage("Stop all received");

      lock (slaveJobs) {
        foreach (SlaveJob sj in slaveJobs.Values) {
          if (!sj.Finished) {
            sj.StopJob();
          }
        }
      }
    }

    /// <summary>
    /// completly shudown slave
    /// </summary>
    public void Shutdown() {
      MessageContainer mc = new MessageContainer(MessageContainer.MessageType.ShutdownSlave);
      MessageQueue.GetInstance().AddMessage(mc);
      waitShutdownSem.WaitOne();
    }

    /// <summary>
    /// complete shutdown, should be called before the the application is exited
    /// </summary>
    private void ShutdownCore() {
      clientCom.LogMessage("Shutdown Signal received");
      clientCom.LogMessage("Stopping heartbeat");
      heartbeatManager.StopHeartBeat();
      abortRequested = true;
      clientCom.LogMessage("Logging out");

      DoAbortAll();

      WcfService.Instance.Disconnect();
      clientCom.Shutdown();
      SlaveClientCom.Close();

      if (slaveComm.State != CommunicationState.Closed)
        slaveComm.Close();
    }

    /// <summary>
    /// reinitializes everything and continues operation, 
    /// can be called after Sleep()
    /// </summary>  
    private void DoStartSlave() {
      clientCom.LogMessage("Restart received");
      StartHeartbeats();
      clientCom.LogMessage("Restart done");
    }

    /// <summary>
    /// stop slave, except for client gui communication, 
    /// primarily used by gui if core is running as windows service
    /// </summary>    
    private void Sleep() {
      clientCom.LogMessage("Sleep received");
      heartbeatManager.StopHeartBeat();
      heartbeatManager = null;
      DoStopAll();
      WcfService.Instance.Disconnect();
      clientCom.LogMessage("Sleep done");
    }

    /// <summary>
    /// Enqueues messages from the executor to the message queue. 
    /// This is necessary if the core thread has to execute certain actions, e.g.
    /// killing of an app domain. 
    /// </summary>    
    /// <returns>true if the calling method can continue execution, else false</returns>
    public void EnqueueExecutorMessage<T>(Action<T> action, T parameter) {
      ExecutorMessageContainer<T> container = new ExecutorMessageContainer<T>();
      container.Callback = action;
      container.CallbackParameter = parameter;
      MessageQueue.GetInstance().AddMessage(container);
    }

    public void RemoveSlaveJobFromList(Guid jobId) {
      lock (slaveJobs) {
        if (slaveJobs.ContainsKey(jobId)) {
          slaveJobs.Remove(jobId);
        }
      }
    }
  }
}