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

#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.IO;
using System.Linq;
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, Executor> executors = new Dictionary<Guid, Executor>();
    private Dictionary<Guid, AppDomain> appDomains = new Dictionary<Guid, AppDomain>();

    // signalizes if the Executor.Start method has properly finished. only then the appdomain may be unloaded
    private Dictionary<Guid, Semaphore> semaphores = new Dictionary<Guid, Semaphore>();

    private WcfService wcfService;
    private HeartbeatManager heartbeatManager;
    private int coreThreadId;

    private ISlaveCommunication clientCom;
    private ServiceHost slaveComm;

    public Dictionary<Guid, Executor> Executors {
      get { return executors; }
    }

    public Core() { }

    /// <summary>
    /// Main Method for the client
    /// </summary>
    public void Start() {
      coreThreadId = Thread.CurrentThread.ManagedThreadId;
      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;
              Job job = wcfService.GetJob(jobId);
              if (job == null) throw new JobNotFoundException(jobId);
              JobData jobData = wcfService.GetJobData(job.Id);
              if (jobData == null) throw new JobDataNotFoundException(jobId);
              SlaveStatusInfo.JobsFetched++;
              job = wcfService.UpdateJobState(job.Id, JobState.Calculating, null);
              if (job == null) throw new JobNotFoundException(jobId);
              StartJobInAppDomain(job, jobData);
            }, container.JobId)
            .ContinueWith((t) => {
              // handle exception of task
              clientCom.LogMessage(t.Exception.ToString());
              wcfService.UpdateJobState(container.JobId, JobState.Failed, t.Exception.ToString());
              SlaveStatusInfo.JobsAborted++;
            }, 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.JobsAborted++;
            KillAppDomain(container.JobId);
            break;
          case MessageContainer.MessageType.StopJob:
            DoStopJob(container.JobId);
            break;
          case MessageContainer.MessageType.PauseJob:
            DoPauseJob(container.JobId);
            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 DoPauseJob(Guid jobId) {
      if (!executors.ContainsKey(jobId)) {
        clientCom.LogMessage("DoPauseJob: Can't find job with id " + jobId);
      } else {
        Job job = wcfService.GetJob(jobId);

        if (job != null && executors.ContainsKey(job.Id)) {
          executors[job.Id].Pause();
          JobData sJob = executors[job.Id].GetPausedJob();
          job.ExecutionTime = executors[job.Id].ExecutionTime;

          try {
            if (executors[job.Id].CurrentException != string.Empty) {
              wcfService.UpdateJobState(job.Id, JobState.Failed, executors[job.Id].CurrentException);
              SlaveStatusInfo.JobsAborted++;
            } else {
              SlaveStatusInfo.JobsProcessed++;
            }
            clientCom.LogMessage("Sending the paused job with id: " + job.Id);
            wcfService.UpdateJobData(job, sJob, ConfigManager.Instance.GetClientInfo().Id, JobState.Paused);
          }
          catch (Exception e) {
            clientCom.LogMessage("Transmitting to server failed. Storing the paused job with id: " + job.Id + " to hdd (" + e.ToString() + ")");
          }
          finally {
            KillAppDomain(job.Id); // kill app-domain in every case          
          }
        }
      }
    }

    private void DoStopJob(Guid jobId) {
      if (!executors.ContainsKey(jobId)) {
        clientCom.LogMessage("DoStopJob: Can't find job with id " + jobId);
      } else {
        Job job = wcfService.GetJob(jobId);

        if (job != null) {
          executors[job.Id].Stop();
          JobData sJob = executors[job.Id].GetFinishedJob();
          job.ExecutionTime = executors[job.Id].ExecutionTime;

          try {
            if (executors[job.Id].CurrentException != string.Empty) {
              wcfService.UpdateJobState(job.Id, JobState.Failed, executors[job.Id].CurrentException);
            }
            SlaveStatusInfo.JobsAborted++;

            clientCom.LogMessage("Sending the stopped job with id: " + job.Id);
            wcfService.UpdateJobData(job, sJob, ConfigManager.Instance.GetClientInfo().Id, JobState.Aborted);
          }
          catch (Exception e) {
            clientCom.LogMessage("Transmitting to server failed. Storing the paused job with id: " + job.Id + " to hdd (" + e.ToString() + ")");
          }
          finally {
            KillAppDomain(job.Id); // kill app-domain in every case          
          }
        }
      }
    }

    /// <summary>
    /// aborts all running jobs, no results are sent back
    /// </summary>
    private void DoAbortAll() {
      List<Guid> jobIds;
      lock (executors) {
        jobIds = new List<Guid>(executors.Keys);
      }
      foreach (Guid jobId in jobIds) {
        KillAppDomain(jobId);
      }
      clientCom.LogMessage("Aborted all jobs!");
    }

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

      //copy guids because there will be removed items from 'Jobs'
      List<Guid> jobIds;
      lock (executors) {
        jobIds = new List<Guid>(Executors.Keys);
      }

      foreach (Guid jobId in jobIds) {
        DoPauseJob(jobId);
      }
    }

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

      //copy guids because there will be removed items from 'Jobs'
      List<Guid> jobIds;
      lock (executors) {
        jobIds = new List<Guid>(executors.Keys);
      }

      foreach (Guid jobId in jobIds) {
        DoStopJob(jobId);
      }
    }

    /// <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");


      lock (executors) {
        clientCom.LogMessage("executors locked");
        foreach (KeyValuePair<Guid, AppDomain> kvp in appDomains) {
          clientCom.LogMessage("Shutting down Appdomain for " + kvp.Key);
          appDomains[kvp.Key].UnhandledException -= new UnhandledExceptionEventHandler(AppDomain_UnhandledException);
          AppDomain.Unload(kvp.Value);
        }
      }
      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>
    /// Pauses a job, which means sending it to the server and killing it locally;
    /// atm only used when executor is waiting for child jobs
    /// </summary>
    public void PauseWaitJob(JobData data) {
      if (!Executors.ContainsKey(data.JobId)) {
        clientCom.LogMessage("Can't find job with id " + data.JobId);
      } else {
        Job job = wcfService.GetJob(data.JobId);
        wcfService.UpdateJobData(job, data, ConfigManager.Instance.GetClientInfo().Id, JobState.Paused);
        wcfService.UpdateJobState(job.Id, JobState.Waiting, null);
      }
      KillAppDomain(data.JobId);
    }

    /// <summary>
    /// serializes the finished job and submits it to the server. If, at the time, a network connection is unavailable, the Job gets stored on the disk. 
    /// once the connection gets reestablished, the job gets submitted
    /// </summary>
    public void SendFinishedJob(Guid jobId) {
      try {
        clientCom.LogMessage("Getting the finished job with id: " + jobId);
        if (!executors.ContainsKey(jobId)) {
          clientCom.LogMessage("Executor doesn't exist");
          return;
        }
        if (!executors.ContainsKey(jobId)) {
          clientCom.LogMessage("Job doesn't exist");
          return;
        }
        Job job = wcfService.GetJob(jobId);
        job.ExecutionTime = executors[jobId].ExecutionTime;

        if (executors[jobId].Aborted) {
          SlaveStatusInfo.JobsAborted++;
        } else {
          SlaveStatusInfo.JobsProcessed++;
        }

        if (executors[jobId].CurrentException != string.Empty) {
          wcfService.UpdateJobState(jobId, JobState.Failed, executors[jobId].CurrentException);
        }

        JobData sJob = executors[jobId].GetFinishedJob();
        try {
          clientCom.LogMessage("Sending the finished job with id: " + jobId);
          wcfService.UpdateJobData(job, sJob, ConfigManager.Instance.GetClientInfo().Id, JobState.Finished);
        }
        catch (Exception e) {
          clientCom.LogMessage("Transmitting to server failed. Storing the finished job with id: " + jobId + " to hdd (" + e.ToString() + ")");
        }
        finally {
          KillAppDomain(jobId);
          heartbeatManager.AwakeHeartBeatThread();
        }
      }
      catch (Exception e) {
        OnExceptionOccured(e);
      }
    }

    private static object startInAppDomainLocker = new object();
    private static Mutex startInAppDomainMutex = new Mutex();
    /// <summary>
    /// A new Job from the wcfService has been received and will be started within a AppDomain.
    /// </summary>    
    private void StartJobInAppDomain(Job job, JobData jobData) {
      clientCom.LogMessage("Received new job with id " + job.Id);
      clientCom.StatusChanged(ConfigManager.Instance.GetStatusForClientConsole());

      startInAppDomainMutex.WaitOne(); // mutex is used instead of lock to be able to release it before executor.Start() is called (which may take long time)
      bool released = false;

        if (executors.ContainsKey(job.Id)) {
          clientCom.LogMessage("Job with key " + job.Id + " already exists. Job will be ignored.");
          return;
        }

        String pluginDir = Path.Combine(PluginCache.Instance.PluginTempBaseDir, job.Id.ToString());
        bool pluginsPrepared = false;
        string configFileName = string.Empty;

        try {
          PluginCache.Instance.PreparePlugins(job, out configFileName);
          clientCom.LogMessage("Plugins fetched for job " + job.Id);
          pluginsPrepared = true;
        }
        catch (Exception exception) {
          clientCom.LogMessage(string.Format("Copying plugins for job {0} failed: {1}", job.Id, exception));
          wcfService.UpdateJobState(job.Id, JobState.Failed, exception.ToString());
          SlaveStatusInfo.JobsAborted++;
        }

        if (pluginsPrepared) {
          try {
            //TODO: switch back to unprivileged sandbox
            AppDomain appDomain = HeuristicLab.PluginInfrastructure.Sandboxing.SandboxManager.CreateAndInitPrivilegedSandbox(job.Id.ToString(), pluginDir, Path.Combine(pluginDir, configFileName));
            appDomain.UnhandledException += new UnhandledExceptionEventHandler(AppDomain_UnhandledException);
            Executor executor;
            appDomains.Add(job.Id, appDomain);
            clientCom.LogMessage("Creating AppDomain");
            executor = (Executor)appDomain.CreateInstanceAndUnwrap(typeof(Executor).Assembly.GetName().Name, typeof(Executor).FullName);
            clientCom.LogMessage("Created AppDomain");
            executor.Core = this;
            executor.JobId = job.Id;
            executor.CoresNeeded = job.CoresNeeded;
            executor.MemoryNeeded = job.MemoryNeeded;
            clientCom.LogMessage("Starting Executor for job " + job.Id);
            lock (executors) {
              if (executors.ContainsKey(job.Id)) {
                throw new JobAlreadyExistsException(job.Id);
              }
              executors.Add(job.Id, executor);
            }
            startInAppDomainMutex.ReleaseMutex();
            released = true;
            semaphores[job.Id] = new Semaphore(0, 1);
            executor.Start(jobData.Data);
            semaphores[job.Id].Release();
          }
          catch (JobAlreadyExistsException e) {
            clientCom.LogMessage(string.Format("Job {0} has already been started. Job will be ignored", e.JobId));
          }
          catch (Exception exception) {
            clientCom.LogMessage("Creating the Appdomain and loading the job failed for job " + job.Id);
            clientCom.LogMessage("Error thrown is: " + exception.ToString());

            if (executors.ContainsKey(job.Id) && executors[job.Id].CurrentException != string.Empty) {
              wcfService.UpdateJobState(job.Id, JobState.Failed, executors[job.Id].CurrentException);
            } else {
              wcfService.UpdateJobState(job.Id, JobState.Failed, exception.ToString());
            }
            SlaveStatusInfo.JobsAborted++;

            KillAppDomain(job.Id);
          }

          if (!released)
            startInAppDomainMutex.ReleaseMutex();
        }
      heartbeatManager.AwakeHeartBeatThread();
    }

    public event EventHandler<EventArgs<Exception>> ExceptionOccured;
    private void OnExceptionOccured(Exception e) {
      clientCom.LogMessage("Error: " + e.ToString());
      var handler = ExceptionOccured;
      if (handler != null) handler(this, new EventArgs<Exception>(e));
    }

    private void AppDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e) {
      clientCom.LogMessage("Exception in AppDomain: " + e.ExceptionObject.ToString());
      KillAppDomain(new Guid(e.ExceptionObject.ToString()));
    }

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

    /// <summary>
    /// Kill a appdomain with a specific id.
    /// </summary>
    /// <param name="jobId">the GUID of the job</param>    
    public void KillAppDomain(Guid jobId) {
      if (Thread.CurrentThread.ManagedThreadId != this.coreThreadId) {
        EnqueueExecutorMessage<Guid>(KillAppDomain, jobId);
        return;
      }

      clientCom.LogMessage("Shutting down Appdomain for Job " + jobId);
      lock (executors) {
        try {
          if (executors.ContainsKey(jobId)) {
            executors[jobId].Dispose();
            executors.Remove(jobId);
          }

          if (appDomains.ContainsKey(jobId)) {
            appDomains[jobId].UnhandledException -= new UnhandledExceptionEventHandler(AppDomain_UnhandledException);
            int repeat = 5;
            while (repeat > 0) {
              try {
                semaphores[jobId].WaitOne();
                AppDomain.Unload(appDomains[jobId]);
                semaphores[jobId].Dispose();
                semaphores.Remove(jobId);
                repeat = 0;
              }
              catch (CannotUnloadAppDomainException) {
                clientCom.LogMessage("Could not unload AppDomain, will try again in 1 sec.");
                Thread.Sleep(1000);
                repeat--;
                if (repeat == 0) {
                  clientCom.LogMessage("Could not unload AppDomain, shutting down core...");
                  throw; // rethrow and let app crash
                }
              }
            }
            appDomains.Remove(jobId);
          }

          PluginCache.Instance.DeletePluginsForJob(jobId);
          GC.Collect();
        }
        catch (Exception ex) {
          clientCom.LogMessage("Exception when unloading the appdomain: " + ex.ToString());
        }
      }
      GC.Collect();
      clientCom.StatusChanged(ConfigManager.Instance.GetStatusForClientConsole());
    }

    public override object InitializeLifetimeService() {
      return null; // avoid destruction of proxy object after 5 minutes
    }

    public int GetCoresNeeded() {
      lock (executors) {
        return executors.Sum(x => x.Value.CoresNeeded);
      }
    }
  }
}