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

#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 static ILog Log { get; set; }
    public static bool AbortRequested { get; set; }

    private static HeartbeatManager heartbeatManager;
    public static HeartbeatManager HeartbeatManager {
      get { return heartbeatManager; }
    }

    private ISlaveCommunication clientCom;
    private ServiceHost slaveComm;
    private Semaphore waitShutdownSem = new Semaphore(0, 1);
    private Dictionary<Guid, SlaveJob> slaveJobs = new Dictionary<Guid, SlaveJob>();
    private WcfService wcfService;

    public EventLog ServiceEventLog { get; set; }
    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(string.Format("Hive Slave threw exception: {0} with stack trace: {1}", ex.ToString(), 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(string.Format("Uncaught exception: {0} {1} Core is going to shutdown.", ex.ToString(), Environment.NewLine));
        }
        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);
        if (!AbortRequested) {
          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;
    }

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

    private 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(string.Format("Message: {0} for job: {1} ", container.Message.ToString(), 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(HandleCalculateJob, container.JobId)
            .ContinueWith((t) => {
              // handle exception of task
              clientCom.LogMessage(t.Exception.ToString());
              if (t.Exception.GetType() != typeof(JobNotFoundException)) {
                wcfService.UpdateJobState(container.JobId, JobState.Waiting, 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(HandleAbortJob, container.JobId)
             .ContinueWith((t) => {
               clientCom.LogMessage(t.Exception.ToString());
             }, TaskContinuationOptions.OnlyOnFaulted);
            break;
          case MessageContainer.MessageType.StopJob:
            Task.Factory.StartNew(HandleStopJob, container.JobId)
             .ContinueWith((t) => {
               clientCom.LogMessage(t.Exception.ToString());
             }, TaskContinuationOptions.OnlyOnFaulted);
            break;
          case MessageContainer.MessageType.PauseJob:
            Task.Factory.StartNew(HandlePauseJob, container.JobId)
             .ContinueWith((t) => {
               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);
      }
    }

    private void HandleCalculateJob(object jobIdObj) {
      Guid jobId = (Guid)jobIdObj;
      SlaveJob newJob = new SlaveJob(this);
      bool start = true;

      lock (slaveJobs) {
        if (ConfigManager.Instance.GetFreeCores() < 1) {
          wcfService.UpdateJobState(jobId, JobState.Waiting, "Slave set status waiting because no cores were available");
          clientCom.LogMessage(string.Format("Setting job with id {0} to waiting, all cores are used", jobId));
          start = false;
        } else {
          if (slaveJobs.ContainsKey(jobId)) {
            start = false;
            clientCom.LogMessage(string.Format("Job with id {0} already exists. Start aborted.", jobId));
          } else {
            slaveJobs.Add(jobId, newJob);
            newJob.PrepareJob(jobId);
          }
        }
      }

      if (start) {
        newJob.CalculateJob();
      }
    }

    private void HandleAbortJob(object 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();
      }
    }

    private void HandleStopJob(object 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();
      }
    }

    private void HandlePauseJob(object 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();
      }
    }

    /// <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");
      ConfigManager.Instance.Asleep = false;
      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 - not accepting any new jobs");
      ConfigManager.Instance.Asleep = true;
      DoPauseAll(); //TODO: or stop? can't decide...      
    }

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