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

#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.IO;
using System.Runtime.CompilerServices;
using System.ServiceModel;
using System.Threading;
using HeuristicLab.Clients.Hive.Slave.ServiceContracts;
using HeuristicLab.Common;
using HeuristicLab.Core;
using HeuristicLab.Services.Hive.Common;
using HeuristicLab.Services.Hive.Common.DataTransfer;


namespace HeuristicLab.Clients.Hive.Slave {
  /// <summary>
  /// The core component of the Hive Client
  /// </summary>
  public class Core : MarshalByRefObject {
    public static bool abortRequested { get; set; }
    private Semaphore waitShutdownSem = new Semaphore(0, 1);
    public static ILog Log { get; set; }

    private Dictionary<Guid, Executor> engines = new Dictionary<Guid, Executor>();
    private Dictionary<Guid, AppDomain> appDomains = new Dictionary<Guid, AppDomain>();
    private Dictionary<Guid, Job> jobs = new Dictionary<Guid, Job>();

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

    private ISlaveCommunication ClientCom;
    private ServiceHost slaveComm;

    public Dictionary<Guid, Executor> ExecutionEngines {
      get { return engines; }
    }

    internal Dictionary<Guid, Job> Jobs {
      get { return jobs; }
    }

    public Core() {
    }

    /// <summary>
    /// Main Method for the client
    /// </summary>
    public void Start() {
      coreThreadId = Thread.CurrentThread.ManagedThreadId;
      abortRequested = false;

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

      ClientCom = SlaveClientCom.Instance.ClientCom;
      ClientCom.LogMessage("Hive Slave started");

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

      wcfService = WcfService.Instance;
      RegisterServiceEvents();

      StartHeartbeats(); // Start heartbeats thread
      DispatchMessageQueue(); // dispatch messages until abortRequested

      DeRegisterServiceEvents();
      waitShutdownSem.Release();
    }

    private void StartHeartbeats() {
      //Initialize the heartbeat      
      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);
      }
    }

    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);
      //TODO: find a better solution
      if (container is ExecutorMessageContainer<Guid>) {
        ExecutorMessageContainer<Guid> c = (ExecutorMessageContainer<Guid>)container;
        c.execute();
      } else if (container is MessageContainer) {
        switch (container.Message) {
          //Server requests to abort a job
          case MessageContainer.MessageType.AbortJob:
            if (engines.ContainsKey(container.JobId))
              try {
                engines[container.JobId].Abort();
              }
              catch (AppDomainUnloadedException) {
                // appdomain already unloaded. Finishing job probably ongoing
              } else
              ClientCom.LogMessage("AbortJob: Engine doesn't exist");
            break;

          //Pull a Job from the Server
          case MessageContainer.MessageType.AquireJob:
            Job myJob = wcfService.AquireJob();
            //TODO: handle in own thread!!
            JobData jobData = wcfService.GetJobData(myJob.Id);
            StartJobInAppDomain(myJob, jobData);
            break;

          //Hard shutdown of the client
          case MessageContainer.MessageType.ShutdownSlave:
            ShutdownCore();
            break;
          case MessageContainer.MessageType.HardPause:
            doHardPause();
            break;
          case MessageContainer.MessageType.SoftPause:
            doSoftPause();
            break;
          case MessageContainer.MessageType.Restart:
            doRestart();
            break;
        }
      } else {
        ClientCom.LogMessage("Unknown MessageContainer: " + container);
      }
    }

    /// <summary>
    /// reinitializes everything and continues operation, 
    /// can be called after SoftPause() or HardPause()
    /// </summary>
    public void Restart() {
      MessageContainer mc = new MessageContainer(MessageContainer.MessageType.Restart);
      MessageQueue.GetInstance().AddMessage(mc);
    }

    private void doRestart() {
      ClientCom.LogMessage("Restart received");
      StartHeartbeats();
      ClientCom.LogMessage("Restart done");
    }

    /// <summary>
    /// wait for jobs to finish, then pause client
    /// </summary>
    public void SoftPause() {
      MessageContainer mc = new MessageContainer(MessageContainer.MessageType.SoftPause);
      MessageQueue.GetInstance().AddMessage(mc);
    }

    private void doSoftPause() {
      ClientCom.LogMessage("Soft pause received");

      //TODO: jobs get removed from Jobs map, is this a problem?
      foreach (Job job in Jobs.Values) {
        engines[job.Id].Pause();
        JobData sJob = engines[job.Id].GetFinishedJob();
        job.Exception = engines[job.Id].CurrentException;
        job.ExecutionTime = engines[job.Id].ExecutionTime;

        try {
          ClientCom.LogMessage("Sending the paused job with id: " + job.Id);
          wcfService.UpdateJob(job, sJob);
          SlaveStatusInfo.JobsProcessed++;    //TODO: count or not count, thats the question
        }
        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          
        }
      }

      heartbeatManager.StopHeartBeat();
      WcfService.Instance.Disconnect();
      ClientCom.LogMessage("Soft pause done");
    }

    /// <summary>
    /// pause slave immediately
    /// </summary>
    public void HardPause() {
      MessageContainer mc = new MessageContainer(MessageContainer.MessageType.HardPause);
      MessageQueue.GetInstance().AddMessage(mc);
    }

    private void doHardPause() {
      ClientCom.LogMessage("Hard pause received");
      heartbeatManager.StopHeartBeat();

      lock (engines) {
        ClientCom.LogMessage("engines 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.LogMessage("Hard pause done");
    }

    public void Shutdown() {
      MessageContainer mc = new MessageContainer(MessageContainer.MessageType.ShutdownSlave);
      MessageQueue.GetInstance().AddMessage(mc);
      waitShutdownSem.WaitOne();
    }

    /// <summary>
    /// hard 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 (engines) {
        ClientCom.LogMessage("engines 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>
    /// 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>
    /// <param name="data"></param>
    [MethodImpl(MethodImplOptions.Synchronized)]
    public void PauseJob(JobData data) {
      if (!Jobs.ContainsKey(data.JobId)) {
        ClientCom.LogMessage("Can't find job with id " + data.JobId);
      } else {
        Job job = Jobs[data.JobId];
        job.JobState = JobState.WaitingForChildJobs;
        wcfService.UpdateJob(job, data);
      }
      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>
    /// <param name="jobId"></param>
    [MethodImpl(MethodImplOptions.Synchronized)]
    public void SendFinishedJob(object jobId) {
      try {
        Guid jId = (Guid)jobId;
        ClientCom.LogMessage("Getting the finished job with id: " + jId);
        if (!engines.ContainsKey(jId)) {
          ClientCom.LogMessage("Engine doesn't exist");
          return;
        }
        if (!jobs.ContainsKey(jId)) {
          ClientCom.LogMessage("Job doesn't exist");
          return;
        }
        Job cJob = jobs[jId];

        JobData sJob = engines[jId].GetFinishedJob();
        cJob.Exception = engines[jId].CurrentException;
        cJob.ExecutionTime = engines[jId].ExecutionTime;

        try {
          ClientCom.LogMessage("Sending the finished job with id: " + jId);
          wcfService.UpdateJob(cJob, sJob);
          SlaveStatusInfo.JobsProcessed++;
        }
        catch (Exception e) {
          ClientCom.LogMessage("Transmitting to server failed. Storing the finished job with id: " + jId + " to hdd (" + e.ToString() + ")");
        }
        finally {
          KillAppDomain(jId); // kill app-domain in every case
          heartbeatManager.AwakeHeartBeatThread();
        }
      }
      catch (Exception e) {
        OnExceptionOccured(e);
      }
    }

    /// <summary>
    /// A new Job from the wcfService has been received and will be started within a AppDomain.
    /// </summary>
    /// <param name="sender"></param>
    /// <param name="e"></param>
    private void StartJobInAppDomain(Job myJob, JobData jobData) {
      ClientCom.LogMessage("Received new job with id " + myJob.Id);
      String pluginDir = Path.Combine(PluginCache.Instance.PluginTempBaseDir, myJob.Id.ToString());
      bool pluginsPrepared = false;

      try {
        PluginCache.Instance.PreparePlugins(myJob, jobData);
        ClientCom.LogMessage("Plugins fetched for job " + myJob.Id);
        pluginsPrepared = true;
      }
      catch (Exception exception) {
        ClientCom.LogMessage(string.Format("Copying plugins for job {0} failed: {1}", myJob.Id, exception));
      }

      if (pluginsPrepared) {
        try {
          AppDomain appDomain = HeuristicLab.PluginInfrastructure.Sandboxing.SandboxManager.CreateAndInitSandbox(myJob.Id.ToString(), pluginDir, Path.Combine(pluginDir, PluginCache.ConfigFileName));
          appDomain.UnhandledException += new UnhandledExceptionEventHandler(appDomain_UnhandledException);
          lock (engines) {
            if (!jobs.ContainsKey(myJob.Id)) {
              jobs.Add(myJob.Id, myJob);
              appDomains.Add(myJob.Id, appDomain);
              ClientCom.LogMessage("Creating AppDomain");
              Executor engine = (Executor)appDomain.CreateInstanceAndUnwrap(typeof(Executor).Assembly.GetName().Name, typeof(Executor).FullName);
              ClientCom.LogMessage("Created AppDomain");
              engine.JobId = myJob.Id;
              engine.core = this;
              ClientCom.LogMessage("Starting Engine for job " + myJob.Id);
              engines.Add(myJob.Id, engine);
              engine.Start(jobData.Data);
              SlaveStatusInfo.JobsFetched++;
              ClientCom.LogMessage("Increment FetchedJobs to:" + SlaveStatusInfo.JobsFetched);
            }
          }
          heartbeatManager.AwakeHeartBeatThread();
        }
        catch (Exception exception) {
          ClientCom.LogMessage("Creating the Appdomain and loading the job failed for job " + myJob.Id);
          ClientCom.LogMessage("Error thrown is: " + exception.ToString());
          KillAppDomain(myJob.Id);
        }
      }
    }

    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>
    /// <typeparam name="T"></typeparam>
    /// <param name="action"></param>
    /// <param name="parameter"></param>
    /// <returns>true if the calling method can continue execution, else false</returns>
    private bool EnqueueExecutorMessage<T>(Action<T> action, T parameter) {
      if (Thread.CurrentThread.ManagedThreadId != this.coreThreadId) {
        ExecutorMessageContainer<T> container = new ExecutorMessageContainer<T>();
        container.Callback = action;
        container.CallbackParameter = parameter;
        MessageQueue.GetInstance().AddMessage(container);
        return false;
      } else {
        return true;
      }
    }

    /// <summary>
    /// Kill a appdomain with a specific id.
    /// </summary>
    /// <param name="id">the GUID of the job</param>
    [MethodImpl(MethodImplOptions.Synchronized)]
    public void KillAppDomain(Guid id) {
      if (EnqueueExecutorMessage<Guid>(KillAppDomain, id)) {
        ClientCom.LogMessage("Shutting down Appdomain for Job " + id);
        lock (engines) {
          try {
            if (engines.ContainsKey(id)) {
              engines[id].Dispose();
              engines.Remove(id);
            }

            if (appDomains.ContainsKey(id)) {
              appDomains[id].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);

              int repeat = 5;
              while (repeat > 0) {
                try {
                  AppDomain.Unload(appDomains[id]);
                  repeat = 0;
                }
                catch (CannotUnloadAppDomainException) {
                  ClientCom.LogMessage("Could not unload AppDomain, will try again in 1 sec.");
                  Thread.Sleep(1000);
                  repeat--;
                  if (repeat == 0) {
                    throw; // rethrow and let app crash
                  }
                }
              }
              appDomains.Remove(id);
            }

            jobs.Remove(id);
            PluginCache.Instance.DeletePluginsForJob(id);
            GC.Collect();
          }
          catch (Exception ex) {
            ClientCom.LogMessage("Exception when unloading the appdomain: " + ex.ToString());
          }
        }
        GC.Collect();
      }
    }
  }
}