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

#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.Threading;
using HeuristicLab.Common;
using HeuristicLab.Core;
using HeuristicLab.Services.Hive.Common.DataTransfer;
using HeuristicLab.Services.Hive.Common;
using HeuristicLab.Clients.Hive.Slave;


namespace HeuristicLab.Clients.Hive.Salve {
  /// <summary>
  /// The core component of the Hive Client
  /// </summary>
  public class Core : MarshalByRefObject {
    public static bool abortRequested { get; set; }

    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 bool currentlyFetching;
    private bool CurrentlyFetching {
      get {
        return currentlyFetching;
      }
      set {
        currentlyFetching = value;
        Logger.Debug("Set CurrentlyFetching to " + currentlyFetching);
      }
    }

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

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

    /// <summary>
    /// Main Method for the client
    /// </summary>
    public void Start() {
      abortRequested = false;
      Logger.Info("Hive Slave started");
      //TODO: implement slave console server
      //SlaveConsoleServer server = new SlaveConsoleServer();
      //server.Start();

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

      wcfService = WcfService.Instance;
      RegisterServiceEvents();

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

      DeRegisterServiceEvents();
      
      //server.Close();
      Logger.Info("Program shutdown");
    }

    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() {
      //TODO
    }

    private void DeRegisterServiceEvents() {
      //TODO
    }

    /// <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) {
      Logger.Info("Message: " + container.Message.ToString() + " for job: " + container.JobId);
      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
            Logger.Error("AbortJob: Engine doesn't exist");
          break;
                        
        //Pull a Job from the Server
        case MessageContainer.MessageType.AquireJob:          
          if (!CurrentlyFetching) {
            //TODO: handle in own thread!!
            Job myJob = wcfService.AquireJob();
            JobData jobData = wcfService.GetJobData(myJob.Id);
            wcfService_GetJobCompleted(myJob, jobData);            
            CurrentlyFetching = true;
          } else
            Logger.Info("Currently fetching, won't fetch this time!");
          break;

        //A Job has finished and can be sent back to the server
        case MessageContainer.MessageType.JobStopped:
          SendFinishedJob(container.JobId);
          break;

        case MessageContainer.MessageType.JobFailed:
          SendFinishedJob(container.JobId);
          break;        
              
        //Hard shutdown of the client
        case MessageContainer.MessageType.ShutdownSlave:
          ShutdownCore();
          break;
      }
    }

    public void ShutdownCore() {
      Logger.Info("Shutdown Signal received");
      
      Logger.Debug("Stopping heartbeat");
      heartbeatManager.StopHeartBeat();
      abortRequested = true;     
      Logger.Debug("Logging out");

      lock (engines) {
        Logger.Debug("engines locked");
        foreach (KeyValuePair<Guid, AppDomain> kvp in appDomains) {
          Logger.Debug("Shutting down Appdomain for " + kvp.Key);
          appDomains[kvp.Key].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
          AppDomain.Unload(kvp.Value);
        }
      }
     
      WcfService.Instance.Disconnect();    
    }

    //TODO: make synchronized
    //TODO: start own thread?
    //only called when waiting for child jobs?
    public void PauseJob(JobData data ) {
      if(!Jobs.ContainsKey(data.JobId)) {
        //TODO: error handling 
      }
      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>
    public void SendFinishedJob(object jobId) {
      try {
        Guid jId = (Guid)jobId;
        Logger.Info("Getting the finished job with id: " + jId);
        if (!engines.ContainsKey(jId)) {
          Logger.Info("Engine doesn't exist");
          return;
        }  
        if (!jobs.ContainsKey(jId)) {
          Logger.Info("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 {
          Logger.Info("Sending the finished job with id: " + jId);
          wcfService.UpdateJob(cJob, sJob);
          SlaveStatusInfo.JobsProcessed++;          
        }
        catch (Exception e) {
          Logger.Info("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>
    void wcfService_GetJobCompleted(Job myJob, JobData jobData) {      
        Logger.Info("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);
          Logger.Debug("Plugins fetched for job " + myJob.Id);
          pluginsPrepared = true;
        }
        catch (Exception exception) {
          Logger.Error(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);
                Logger.Debug("Creating AppDomain");
                Executor engine = (Executor)appDomain.CreateInstanceAndUnwrap(typeof(Executor).Assembly.GetName().Name, typeof(Executor).FullName);
                Logger.Debug("Created AppDomain");
                engine.JobId = myJob.Id;
                engine.core = this;
                Logger.Debug("Starting Engine for job " + myJob.Id);
                engines.Add(myJob.Id, engine);
                engine.Start(jobData.Data);
                SlaveStatusInfo.JobsFetched++;
                Logger.Info("Increment FetchedJobs to:" + SlaveStatusInfo.JobsFetched);
              }
            }
            heartbeatManager.AwakeHeartBeatThread();
          }
          catch (Exception exception) {
            Logger.Error("Creating the Appdomain and loading the job failed for job " + myJob.Id);
            Logger.Error("Error thrown is: ", exception);
            CurrentlyFetching = false;
            KillAppDomain(myJob.Id);
          }
        }      
      CurrentlyFetching = false;
    }
  
    public event EventHandler<EventArgs<Exception>> ExceptionOccured;
    private void OnExceptionOccured(Exception e) {
      Logger.Error("Error: " + e.ToString());
      var handler = ExceptionOccured;
      if (handler != null) handler(this, new EventArgs<Exception>(e));
    }

    void appDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e) {
      Logger.Error("Exception in AppDomain: " + e.ExceptionObject.ToString());
      KillAppDomain(new Guid(e.ExceptionObject.ToString()));
    }

    /// <summary>
    /// Kill a appdomain with a specific id.
    /// </summary>
    /// <param name="id">the GUID of the job</param>
    public void KillAppDomain(Guid id) {
      Logger.Debug("Shutting down Appdomain for Job " + id);
      lock (engines) {
        try {
          if (engines.ContainsKey(id))
            engines[id].Dispose();
          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) {
                Logger.Error("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);
          }

          engines.Remove(id);
          jobs.Remove(id);
          PluginCache.Instance.DeletePluginsForJob(id);
          GC.Collect();
        }
        catch (Exception ex) {
          Logger.Error("Exception when unloading the appdomain: ", ex);
        }
      }
      GC.Collect();
    } 
  }   


}