source: branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave/3.4/SlaveJob.cs @ 6248

using System;
using System.IO;
using System.Threading;
using HeuristicLab.Clients.Hive.SlaveCore.ServiceContracts;

namespace HeuristicLab.Clients.Hive.SlaveCore {

  public class SlaveJob : MarshalByRefObject {
    private Executor executor;
    private AppDomain appDomain;
    private Semaphore waitForStartBeforeKillSem;
    private bool executorMonitoringRun;
    private Thread executorMonitoringThread;
    private Core core;
    private bool finished;
    private int coresNeeded;

    private ISlaveCommunication clientCom;
    private WcfService wcfService;

    public Guid JobId;
    public Executor JobExecutor { get { return executor; } }
    public bool Finished { get { return finished; } }


    public SlaveJob(Core core) {
      clientCom = SlaveClientCom.Instance.ClientCom;
      wcfService = WcfService.Instance;
      waitForStartBeforeKillSem = new Semaphore(0, 1);
      executorMonitoringRun = true;
      this.core = core;
      finished = false;
    }

    public void PrepareJob(Guid jobId) {
      JobId = jobId;
      Job job = wcfService.GetJob(jobId);
      if (job == null) throw new JobNotFoundException(jobId);
      coresNeeded = job.CoresNeeded;
      SlaveStatusInfo.IncrementUsedCores(coresNeeded);
    }

    public void CalculateJob() {
      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.IncrementJobsFetched();
      job = wcfService.UpdateJobState(job.Id, JobState.Calculating, null);
      if (job == null) throw new JobNotFoundException(JobId);
      StartJobInAppDomain(job, jobData);
    }

    public void StopJob() {
      if (executor == null) {
        clientCom.LogMessage(string.Format("StopJob: job with id {0} is missing the executor", JobId));
      } else {
        Job job = wcfService.GetJob(JobId);

        if (job != null) {
          executor.Stop();

          try {
            JobData sJob = executor.GetFinishedJob();
            job.ExecutionTime = executor.ExecutionTime;

            if (executor.CurrentException != string.Empty) {
              wcfService.UpdateJobState(job.Id, JobState.Failed, executor.CurrentException);
            }
            SlaveStatusInfo.IncrementJobsAborted();

            clientCom.LogMessage(string.Format("Sending the stopped job with id: {0}", job.Id));
            wcfService.UpdateJobData(job, sJob, ConfigManager.Instance.GetClientInfo().Id, JobState.Aborted);
          }
          catch (Exception e) {
            clientCom.LogMessage(string.Format("Transmitting the stopped job with id {0} to server failed. Exception is: {1}", job.Id, e.ToString()));
          }
          finally {
            KillAppDomain(); // kill appdomain in every case          
          }
        }
      }
    }

    public void PauseJob() {
      if (executor == null) {
        clientCom.LogMessage("PauseJob: Can't pause job with uninitialized executor");
      } else {
        Job job = wcfService.GetJob(JobId);

        if (job != null) {
          executor.Pause();
          JobData sJob = executor.GetPausedJob();
          job.ExecutionTime = executor.ExecutionTime;

          try {
            if (executor.CurrentException != string.Empty) {
              wcfService.UpdateJobState(job.Id, JobState.Failed, executor.CurrentException);
              SlaveStatusInfo.IncrementJobsFailed();
            } else {
              SlaveStatusInfo.IncrementJobsFinished();
            }
            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(string.Format("Transmitting the paused job with id {0} to server failed. Exception is: {1}", job.Id, e.ToString()));
          }
          finally {
            KillAppDomain(); // kill appdomain in every case          
          }
        }
      }
    }

    /// <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) {
      try {
        if (executor == null) {
          clientCom.LogMessage(string.Format("PauseWaitJob: Can't pause job with id {0} with uninitialized executor", 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);
        }
      }
      catch (Exception ex) {
        clientCom.LogMessage(string.Format("Pausing job with id {0} failed. Exception: {1}", JobId, ex.ToString()));
      }
      finally {
        KillAppDomain();
      }
    }

    /// <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() {
      try {
        clientCom.LogMessage(string.Format("Getting the finished job with id: {0} ", JobId));
        if (executor == null) {
          clientCom.LogMessage(string.Format("SendFinishedJob: Can't pause job with id {0} with uninitialized executor", JobId));
          return;
        }

        Job job = wcfService.GetJob(JobId);
        job.ExecutionTime = executor.ExecutionTime;

        if (executor.Aborted) {
          SlaveStatusInfo.IncrementJobsAborted();
        } else {
          SlaveStatusInfo.IncrementJobsFinished();
        }

        if (executor.CurrentException != string.Empty) {
          wcfService.UpdateJobState(JobId, JobState.Failed, executor.CurrentException);
        }

        try {
          JobData sJob = executor.GetFinishedJob();
          if (sJob != null) {
            clientCom.LogMessage(string.Format("Sending the finished job with id: {0}", JobId));
            wcfService.UpdateJobData(job, sJob, ConfigManager.Instance.GetClientInfo().Id, JobState.Finished);
          }
        }
        catch (Exception e) {
          clientCom.LogMessage(string.Format("Transmitting the job with id {0} to server failed. Exception is: {1}", job.Id, e.ToString()));
        }
        finally {
          KillAppDomain();
          Core.HBManager.AwakeHeartBeatThread();
        }
      }
      catch (Exception e) {
        clientCom.LogMessage(string.Format("SendFinishedJob: The following exception has been thrown: {0}", e.ToString()));
      }
    }

    /// <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) {
      JobId = job.Id;

      clientCom.LogMessage(string.Format("Received new job with id {0}", job.Id));
      clientCom.StatusChanged(ConfigManager.Instance.GetStatusForClientConsole());

      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(string.Format("Plugins fetched for job {0}", 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.IncrementJobsFailed();
        core.RemoveSlaveJobFromList(JobId);
        finished = true;
        SlaveStatusInfo.DecrementUsedCores(coresNeeded);
      }

      if (pluginsPrepared) {
        try {
          appDomain = HeuristicLab.PluginInfrastructure.Sandboxing.SandboxManager.CreateAndInitPrivilegedSandbox(job.Id.ToString(), pluginDir, Path.Combine(pluginDir, configFileName));
          appDomain.UnhandledException += new UnhandledExceptionEventHandler(AppDomain_UnhandledException);

          clientCom.LogMessage("Creating AppDomain");
          executor = (Executor)appDomain.CreateInstanceAndUnwrap(typeof(Executor).Assembly.GetName().Name, typeof(Executor).FullName);
          clientCom.LogMessage("Created AppDomain");

          executor.JobId = job.Id;
          executor.CoresNeeded = job.CoresNeeded;
          executor.MemoryNeeded = job.MemoryNeeded;
          clientCom.LogMessage(string.Format("Starting Executor for job {0}", job.Id));

          executor.Start(jobData.Data);
          waitForStartBeforeKillSem.Release();

          StartExecutorMonitoringThread();
        }
        catch (Exception exception) {
          clientCom.LogMessage(string.Format("Creating the Appdomain and loading the job failed for job {0}", job.Id));
          clientCom.LogMessage(string.Format("Error thrown is: {0}", exception.ToString()));

          if (executor != null && executor.CurrentException != string.Empty) {
            wcfService.UpdateJobState(job.Id, JobState.Failed, executor.CurrentException);
          } else {
            wcfService.UpdateJobState(job.Id, JobState.Failed, exception.ToString());
          }
          SlaveStatusInfo.IncrementJobsFailed();

          KillAppDomain();
        }
      }
      Core.HBManager.AwakeHeartBeatThread();
    }

    /// <summary>
    /// Kill a appdomain with a specific id.
    /// </summary>
    /// <param name="JobId">the GUID of the job</param>    
    public void KillAppDomain() {
      clientCom.LogMessage(string.Format("Shutting down Appdomain for Job {0}", JobId));

      try {
        StopExecutorMonitoringThread();
        finished = true;
        SlaveStatusInfo.DecrementUsedCores(coresNeeded);

        if (executor != null) {
          executor.Dispose();
        }

        if (appDomain != null) {
          appDomain.UnhandledException -= new UnhandledExceptionEventHandler(AppDomain_UnhandledException);
          int repeat = 5;
          while (repeat > 0) {
            try {
              waitForStartBeforeKillSem.WaitOne();
              AppDomain.Unload(appDomain);
              waitForStartBeforeKillSem.Dispose();
              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
              }
            }
          }
        }

        PluginCache.Instance.DeletePluginsForJob(JobId);
        GC.Collect();
      }
      catch (Exception ex) {
        clientCom.LogMessage(string.Format("Exception when unloading the appdomain: {0}", ex.ToString()));
      }
      finally {
        core.RemoveSlaveJobFromList(JobId);
      }

      GC.Collect();
      clientCom.StatusChanged(ConfigManager.Instance.GetStatusForClientConsole());
    }

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


    #region ExecutorMonitorThread

    private void StartExecutorMonitoringThread() {
      executorMonitoringThread = new Thread(MonitorExecutor);
      executorMonitoringThread.Start();
    }

    private void StopExecutorMonitoringThread() {
      if (executorMonitoringThread != null) {
        if (executorMonitoringRun) {
          executorMonitoringRun = false;
          executor.executorQueue.AddMessage(ExecutorMessageType.StopExecutorMonitoringThread);
        }
      }
    }

    /// <summary>
    /// Because the executor is in an appdomain and is not able to call back 
    /// (because of security -> lease time for marshall-by-ref object is 5 min),
    /// we have to poll the executor for events we have to react to (e.g. job finished...)    
    /// </summary>
    private void MonitorExecutor() {
      while (executorMonitoringRun) {
        //this blocks through the appdomain border, that's why the lease gets renewed
        ExecutorMessage message = executor.executorQueue.GetMessage();

        switch (message.MessageType) {
          case ExecutorMessageType.JobStopped:
            executorMonitoringRun = false;
            SendFinishedJob();
            break;

          case ExecutorMessageType.JobFailed:
            executorMonitoringRun = false;
            SendFinishedJob();
            break;

          case ExecutorMessageType.NewChildJob:
            WcfService.Instance.AddChildJob(JobId, message.MsgJob, message.MsgData);
            break;

          case ExecutorMessageType.WaitForChildJobs:
            executorMonitoringRun = false;
            PauseWaitJob(message.MsgData);
            break;

          case ExecutorMessageType.DeleteChildJobs:
            WcfService.Instance.DeleteChildJobs(JobId);
            break;

          case ExecutorMessageType.StopExecutorMonitoringThread:
            executorMonitoringRun = false;
            return;
        }
      }
    }
    #endregion
  }
}