Changeset 6203 for branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave/3.4/Core.cs

Timestamp:

05/16/11 21:19:34 (13 years ago)

Author:

ascheibe

Message:

#1233

• dropped dependency of Core from Executor
• enabled sandboxing
• moved most parts of Job handling from Core to SlaveJob to simplify locking
• optimized how UsedCores is handled
• SlaveStatusInfo is now thread-save and counts jobs more correct

File:

• branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave/3.4/Core.cs (modified) (12 diffs)

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave/3.4/Core.cs

@@ -23 +23 @@
 using System.Collections.Generic;
 using System.Diagnostics;
-using System.IO;
-using System.Linq;
 using System.ServiceModel;
 using System.Threading;
@@ -45 +43 @@
     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 Dictionary<Guid, SlaveJob> slaveJobs = new Dictionary<Guid, SlaveJob>();
 
     private WcfService wcfService;
-    private HeartbeatManager heartbeatManager;
-    private int coreThreadId;
+    private static HeartbeatManager heartbeatManager;
+    public static HeartbeatManager HBManager { get { return heartbeatManager; } }
 
     private ISlaveCommunication clientCom;
     private ServiceHost slaveComm;
 
-    public Dictionary<Guid, Executor> Executors {
-      get { return executors; }
+    public Dictionary<Guid, SlaveJob> SlaveJobs {
+      get { return slaveJobs; }
     }
 
@@ -68 +62 @@
     /// </summary>
     public void Start() {
-      coreThreadId = Thread.CurrentThread.ManagedThreadId;
       abortRequested = false;
 
@@ -160 +153 @@
             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);
+              SlaveJob newJob = new SlaveJob(this);
+              bool start = true;
+
+              lock (slaveJobs) {
+                if (slaveJobs.ContainsKey(jobId)) {
+                  start = false;
+                  clientCom.LogMessage(string.Format("Job with id {0} already exists. Start aborted.", jobId));
+                } else {
+                  slaveJobs.Add(jobId, newJob);
+                }
+              }
+
+              if (start) {
+                newJob.CalculateJob(jobId);
+              }
             }, container.JobId)
             .ContinueWith((t) => {
@@ -173 +173 @@
               clientCom.LogMessage(t.Exception.ToString());
               wcfService.UpdateJobState(container.JobId, JobState.Failed, t.Exception.ToString());
-              SlaveStatusInfo.JobsAborted++;
+              SlaveStatusInfo.IncrementJobsFailed();
             }, TaskContinuationOptions.OnlyOnFaulted);
             break;
@@ -189 +189 @@
             break;
           case MessageContainer.MessageType.AbortJob:
-            SlaveStatusInfo.JobsAborted++;
-            KillAppDomain(container.JobId);
+            SlaveStatusInfo.IncrementJobsAborted(); //TODO: move to a sane place
+
+            Task.Factory.StartNew((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();
+              }
+            }, container.JobId)
+             .ContinueWith((t) => {
+               // handle exception of task
+               clientCom.LogMessage(t.Exception.ToString());
+             }, TaskContinuationOptions.OnlyOnFaulted);
             break;
           case MessageContainer.MessageType.StopJob:
-            DoStopJob(container.JobId);
+            Task.Factory.StartNew((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();
+              }
+            }, container.JobId)
+             .ContinueWith((t) => {
+               // handle exception of task
+               clientCom.LogMessage(t.Exception.ToString());
+             }, TaskContinuationOptions.OnlyOnFaulted);
             break;
           case MessageContainer.MessageType.PauseJob:
-            DoPauseJob(container.JobId);
+            Task.Factory.StartNew((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();
+              }
+            }, container.JobId)
+             .ContinueWith((t) => {
+               // handle exception of task
+               clientCom.LogMessage(t.Exception.ToString());
+             }, TaskContinuationOptions.OnlyOnFaulted);
             break;
           case MessageContainer.MessageType.Restart:
@@ -213 +274 @@
     }
 
-    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);
+    /// <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();
           }
-          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!");
@@ -294 +294 @@
       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);
+      lock (slaveJobs) {
+        foreach (SlaveJob sj in slaveJobs.Values) {
+          if (!sj.Finished) {
+            sj.PauseJob();
+          }
+        }
       }
     }
@@ -311 +309 @@
       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);
+      lock (slaveJobs) {
+        foreach (SlaveJob sj in slaveJobs.Values) {
+          if (!sj.Finished) {
+            sj.StopJob();
+          }
+        }
       }
     }
@@ -341 +337 @@
       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);
-        }
-      }
+      DoAbortAll();
+
       WcfService.Instance.Disconnect();
       clientCom.Shutdown();
@@ -382 +371 @@
 
     /// <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.
@@ -549 +383 @@
     }
 
-    /// <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);
+    public void RemoveSlaveJobFromList(Guid jobId) {
+      lock (slaveJobs) {
+        if (slaveJobs.ContainsKey(jobId)) {
+          slaveJobs.Remove(jobId);
+        }
       }
     }