Changeset 5137 for branches

Timestamp:

12/20/10 19:58:01 (14 years ago)

Author:

ascheibe

Message:

#1233

• more tests for the hive slave
• implemented a better way to write tests for the slave
• get rid of MessageTypes for core and executor
• various improvements in core and executor (better communication, bugfixes,...)

Location:

branches/HeuristicLab.Hive-3.4/sources

Files:

• . (modified) (1 prop)
• HeuristicLab.Clients.Hive.Slave.Tests/Mocks/MockHiveService.cs (modified) (11 diffs)
• HeuristicLab.Clients.Hive.Slave.Tests/Mocks/MockJob.cs (modified) (2 diffs)
• HeuristicLab.Clients.Hive.Slave.Tests/SlaveTest.cs (modified) (4 diffs)
• HeuristicLab.Clients.Hive.Slave/3.4/Core.cs (modified) (16 diffs)
• HeuristicLab.Clients.Hive.Slave/3.4/Executor.cs (modified) (13 diffs)
• HeuristicLab.Clients.Hive.Slave/3.4/ExecutorMessageContainer.cs (added)
• HeuristicLab.Clients.Hive.Slave/3.4/HeartbeatManager.cs (modified) (7 diffs)
• HeuristicLab.Clients.Hive.Slave/3.4/HeuristicLab.Clients.Hive.Slave-3.4.csproj (modified) (1 diff)
• HeuristicLab.Clients.Hive.Slave/3.4/Logger.cs (modified) (3 diffs)
• HeuristicLab.Services.Hive.Common/3.4/MessageContainer.cs (modified) (1 diff)

branches/HeuristicLab.Hive-3.4/sources

@@ +1 @@
+*.suo
 HeuristicLab.Hive-3.4.suo
 TestResults
-HeuristicLab.Hive 3.4.suo

@@ +1 @@
+*.suo
 HeuristicLab.Hive-3.4.suo
 TestResults
-HeuristicLab.Hive 3.4.suo

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave.Tests/Mocks/MockHiveService.cs

@@ -1 +1 @@
 using System;
 using System.Collections.Generic;
+using System.IO;
 using System.Linq;
-using System.Text;
+using HeuristicLab.PluginInfrastructure;
+using HeuristicLab.PluginInfrastructure.Manager;
+using HeuristicLab.Services.Hive.Common;
+using HeuristicLab.Services.Hive.Common.DataTransfer;
 using HeuristicLab.Services.Hive.Common.ServiceContracts;
-using HeuristicLab.Services.Hive.Common.DataTransfer;
-using HeuristicLab.Services.Hive.Common;
-using HeuristicLab.PluginInfrastructure.Manager;
-using System.IO;
-using System.Reflection;
-using HeuristicLab.PluginInfrastructure;
 
 namespace HeuristicLab.Clients.Hive.Slave.Tests {
@@ -15 +13 @@
     private static List<Job> jobs;
     private static List<JobData> jobDatas;
-    private static List<HeuristicLab.Services.Hive.Common.DataTransfer.HiveExperiment> hiveExperiments;    
+    private static List<HeuristicLab.Services.Hive.Common.DataTransfer.HiveExperiment> hiveExperiments;
     private static List<Plugin> plugins = null;
+    public List<Job> ResultJobs { get; set; }
+    public List<JobData> ResultJobDatas { get; set; }
     private static IEnumerable<PluginDescription> pDescs = null;
     private static int cnt = 0;
@@ -28 +28 @@
       ServerConfigFilePath = Path.Combine(ServerPluginCachePath, "HeuristicLab 3.3.exe.config");
 
-      byte[] data = PersistenceUtil.Serialize(new MockJob(4000, false));
       if (plugins == null)
         plugins = ReadPluginsFromServerCache();
 
-      jobs = new List<Job>();
+
+      //TODO: reuse as fallback?
+      /*jobs = new List<Job>();
       jobs.Add(new Job { Id = Guid.NewGuid(), JobState = JobState.Waiting, DateCreated = DateTime.Now, CoresNeeded = 1, MemoryNeeded = 0 });
       jobs.Add(new Job { Id = Guid.NewGuid(), JobState = JobState.Offline, DateCreated = DateTime.Now, CoresNeeded = 1, MemoryNeeded = 100, ParentJobId = jobs.First().Id });
       jobs.Add(new Job { Id = Guid.NewGuid(), JobState = JobState.Offline, DateCreated = DateTime.Now, CoresNeeded = 1, MemoryNeeded = 1024, ParentJobId = jobs.First().Id });
       jobs.Add(new Job { Id = Guid.NewGuid(), JobState = JobState.Offline, DateCreated = DateTime.Now, CoresNeeded = 1, MemoryNeeded = 4096, ParentJobId = jobs.First().Id });
-
+      byte[] data = PersistenceUtil.Serialize(new MockJob(400, false));
       jobDatas = new List<JobData>();
       foreach (var job in jobs) {
         job.PluginsNeededIds = new List<Guid>();
         job.PluginsNeededIds.AddRange(plugins.Select(a => a.Id));
-        jobDatas.Add(new JobData() { JobId = job.Id, Data = data });        
-      }
-      
+        jobDatas.Add(new JobData() { JobId = job.Id, Data = data });
+      }
+
+      hiveExperiments = new List<HeuristicLab.Services.Hive.Common.DataTransfer.HiveExperiment>();
+      hiveExperiments.Add(new HeuristicLab.Services.Hive.Common.DataTransfer.HiveExperiment() { Id = Guid.NewGuid(), Name = "Hive Exp 1", Description = "", RootJobId = jobs[0].Id });*/
+    }
+
+    public void updateJobs(List<Job> js, MockJob m) {
+      jobs = js;
+      byte[] data = PersistenceUtil.Serialize(m);
+
+      jobDatas = new List<JobData>();
+      foreach (var job in jobs) {
+        job.PluginsNeededIds = new List<Guid>();
+        job.PluginsNeededIds.AddRange(plugins.Select(a => a.Id));
+        jobDatas.Add(new JobData() { JobId = job.Id, Data = data });
+      }
+
       hiveExperiments = new List<HeuristicLab.Services.Hive.Common.DataTransfer.HiveExperiment>();
       hiveExperiments.Add(new HeuristicLab.Services.Hive.Common.DataTransfer.HiveExperiment() { Id = Guid.NewGuid(), Name = "Hive Exp 1", Description = "", RootJobId = jobs[0].Id });
+
+      ResultJobDatas = new List<JobData>();
+      ResultJobs = new List<Job>();
     }
 
@@ -60 +79 @@
         p.Id = Guid.NewGuid();
         plugins.Add(p);
+
       }
       return plugins;
@@ -87 +107 @@
       pd.Data = cf;
       return pd;
+
     }
 
     public Job GetJob(Guid jobId) {
+      //MockHBM.SendHeartbeat();
       return jobs.Where(j => j.Id == jobId).SingleOrDefault();
     }
@@ -106 +128 @@
 
     public JobData GetJobData(Guid jobId) {
-      return jobDatas.Where(jd => jd.JobId == jobId).SingleOrDefault();
+      JobData ret = jobDatas.Where(jd => jd.JobId == jobId).SingleOrDefault();
+      //MockHBM.SendHeartbeat();
+      return ret;
     }
 
     public void UpdateJob(Job jobDto, JobData jobDataDto) {
       Console.WriteLine("Update Job called!");
+      ResultJobDatas.Add(jobDataDto);
+      ResultJobs.Add(jobDto);
     }
 
@@ -123 +149 @@
 
     #region JobControl Methods
+    int curJobIdx = 0;
     public Job AquireJob(Guid slaveId) {
-      var job = jobs.First();
-      job.SlaveId = slaveId;
-      return job;
+      if (curJobIdx < jobs.Count) {
+        var job = jobs[curJobIdx];
+        job.SlaveId = slaveId;
+        curJobIdx++;
+        return job;
+      }
+      throw new Exception("No Jobs left on MockHiveService!");
     }
 
@@ -169 +200 @@
     public void GoodBye() {
       // do nothing
-    }    
-    
+    }
+
     public List<MessageContainer> Heartbeat(Heartbeat heartbeat) {
       if (Messages != null && cnt < Messages.Count) {
@@ -178 +209 @@
       }
     }
-      
+
     #endregion
 
     #region Plugin Methods
-    public Guid AddPlugin(Plugin plugin, List<PluginData> pluginData) {
-      // todo
-      return Guid.NewGuid();
+    public Guid AddPlugin(Plugin p, List<PluginData> pds) {
+      /*p.Id = Guid.NewGuid();
+      plugins.Add(p);
+      foreach (var pluginData in pds) {
+        pluginData.PluginId = p.Id;
+        pluginDatas.Add(pluginData);
+
+      }
+      return p.Id;*/
+      throw new NotImplementedException();
+
     }
 
@@ -190 +229 @@
       if (plugins == null)
         plugins = ReadPluginsFromServerCache();
-      return plugins;              
+      return plugins;
     }
 
@@ -207 +246 @@
               pluginDatas.Add(data);
             }
-          }          
-        }     
+          }
+        }
       }
       return pluginDatas;

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave.Tests/Mocks/MockJob.cs

@@ -114 +114 @@
         OnJobStopped();
       }
+      catch (ThreadAbortException) {
+        //this happens when the appdomain is killed (e.g. abort from server)
+        Stop();
+      }
       catch (Exception e) {
         this.ExecutionState = Core.ExecutionState.Stopped;
-        OnJobFailed();
+        OnJobFailed(e);
       }
     }
@@ -170 +174 @@
                                                       
     public event EventHandler JobFailed;
-    protected virtual void OnJobFailed() {
+    protected virtual void OnJobFailed(Exception e) {
       EventHandler handler = JobFailed;
-      if (handler != null) handler(this, EventArgs.Empty);
+      EventArgs<Exception> ev = new EventArgs<Exception>(e);          
+      if (handler != null) handler(this, ev);
     }
 

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave.Tests/SlaveTest.cs

@@ -1 +1 @@
 using System;
-using System.Text;
 using System.Collections.Generic;
 using System.Linq;
+using HeuristicLab.Clients.Common;
+using HeuristicLab.Services.Hive.Common;
+using HeuristicLab.Services.Hive.Common.DataTransfer;
+using HeuristicLab.Services.Hive.Common.ServiceContracts;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
-using HeuristicLab.Clients.Hive.Salve;
-using HeuristicLab.Services.Hive.Common;
-using HeuristicLab.Services.Hive.Common.ServiceContracts;
-using HeuristicLab.Clients.Common;
 
 namespace HeuristicLab.Clients.Hive.Slave.Tests {
@@ -16 +15 @@
     [ClassInitialize]
     public static void MyClassInitialize(TestContext testContext) {
-      PluginLoader.pluginAssemblies.Any();      
+      PluginLoader.pluginAssemblies.Any();
       ServiceLocator.Instance = new MockServiceLocator();
     }
 
-    public List<List<MessageContainer>> CreateMsgs() {
+    private List<List<MessageContainer>> CreateMsgsForSingleJob() {
       List<List<MessageContainer>> allMsgs = new List<List<MessageContainer>>();
       //get slave to fetch job
@@ -28 +27 @@
 
       //do nothing
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
       msg = new List<MessageContainer>();
       allMsgs.Add(msg);
@@ -38 +43 @@
 
     [TestMethod]
-    public void TestMethod1() {
-      using (Disposable<IHiveService> service = ServiceLocator.Instance.GetService()) { 
-        ((MockHiveService)service.Obj).Messages = CreateMsgs();
+    public void TestSingleJob() {
+      Job testJob = new Job { Id = Guid.NewGuid(), JobState = JobState.Waiting, DateCreated = DateTime.Now, CoresNeeded = 1, MemoryNeeded = 0 };
+      List<Job> jobList = new List<Job>();
+      jobList.Add(testJob);
+      MockJob job = new MockJob(400, false);
+
+      using (Disposable<IHiveService> service = ServiceLocator.Instance.GetService()) {
+        MockHiveService ms = (MockHiveService)service.Obj;
+        ((MockHiveService)service.Obj).Messages = CreateMsgsForSingleJob();
+        ((MockHiveService)service.Obj).updateJobs(jobList, job);
+
+
+        HeuristicLab.Clients.Hive.Salve.Core core = new Salve.Core();
+        core.Start();
+
+        Assert.AreEqual<int>(1, ms.ResultJobs.Count);
+        Assert.AreEqual<Guid>(testJob.Id, ms.ResultJobs[0].Id);
       }
-      
-      HeuristicLab.Clients.Hive.Salve.Core core = new Salve.Core();
-      core.Start();
     }
+
+    private List<List<MessageContainer>> CreateMsgsForShutdownSlaveWhileJobRunning() {
+      List<List<MessageContainer>> allMsgs = new List<List<MessageContainer>>();
+      //get slave to fetch job
+      List<MessageContainer> msg = new List<MessageContainer>();
+      msg.Add(new MessageContainer(MessageContainer.MessageType.AquireJob));
+      allMsgs.Add(msg);
+
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+
+      msg = new List<MessageContainer>();
+      msg.Add(new MessageContainer(MessageContainer.MessageType.ShutdownSlave));
+      allMsgs.Add(msg);
+      return allMsgs;
+    }
+
+    [TestMethod]
+    public void TestShutdownSlaveWhileJobRunning() {
+      Job testJob = new Job { Id = Guid.NewGuid(), JobState = JobState.Waiting, DateCreated = DateTime.Now, CoresNeeded = 1, MemoryNeeded = 0 };
+      List<Job> jobList = new List<Job>();
+      jobList.Add(testJob);
+      MockJob job = new MockJob(10000, false);
+
+      using (Disposable<IHiveService> service = ServiceLocator.Instance.GetService()) {
+        MockHiveService ms = (MockHiveService)service.Obj;
+        ((MockHiveService)service.Obj).Messages = CreateMsgsForShutdownSlaveWhileJobRunning();
+        ((MockHiveService)service.Obj).updateJobs(jobList, job);
+
+
+        HeuristicLab.Clients.Hive.Salve.Core core = new Salve.Core();
+        core.Start();
+
+      }
+    }
+
+
+    private List<List<MessageContainer>> CreateMsgsForTwoJobs() {
+      List<List<MessageContainer>> allMsgs = new List<List<MessageContainer>>();
+
+      //get slave to fetch jobs
+      List<MessageContainer> msg = new List<MessageContainer>();
+      msg.Add(new MessageContainer(MessageContainer.MessageType.AquireJob));
+      allMsgs.Add(msg);
+
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+
+      msg = new List<MessageContainer>();
+      msg.Add(new MessageContainer(MessageContainer.MessageType.AquireJob));
+      allMsgs.Add(msg);
+
+
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+      msg = new List<MessageContainer>();
+      allMsgs.Add(msg);
+
+      msg = new List<MessageContainer>();
+      msg.Add(new MessageContainer(MessageContainer.MessageType.ShutdownSlave));
+      allMsgs.Add(msg);
+      return allMsgs;
+    }
+
+    [TestMethod]
+    public void TestTwoJobs() {
+      Job testJob1 = new Job { Id = Guid.NewGuid(), JobState = JobState.Waiting, DateCreated = DateTime.Now, CoresNeeded = 1, MemoryNeeded = 0 };
+      Job testJob2 = new Job { Id = Guid.NewGuid(), JobState = JobState.Waiting, DateCreated = DateTime.Now, CoresNeeded = 1, MemoryNeeded = 0 };
+      List<Job> jobList = new List<Job>();
+      jobList.Add(testJob1);
+      jobList.Add(testJob2);
+      MockJob job = new MockJob(2000, false);
+
+      using (Disposable<IHiveService> service = ServiceLocator.Instance.GetService()) {
+        MockHiveService ms = (MockHiveService)service.Obj;
+        ((MockHiveService)service.Obj).Messages = CreateMsgsForTwoJobs();
+        ((MockHiveService)service.Obj).updateJobs(jobList, job);
+
+
+        HeuristicLab.Clients.Hive.Salve.Core core = new Salve.Core();
+        core.Start();
+        Assert.AreEqual<int>(2, ms.ResultJobs.Count);
+
+      }
+    }
+
+
+
+
+
+
+
+
+
   }
 }

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

@@ -23 +23 @@
 using System.Collections.Generic;
 using System.IO;
+using System.Runtime.CompilerServices;
 using System.Threading;
+using HeuristicLab.Clients.Hive.Slave;
 using HeuristicLab.Common;
 using HeuristicLab.Core;
+using HeuristicLab.Services.Hive.Common;
 using HeuristicLab.Services.Hive.Common.DataTransfer;
-using HeuristicLab.Services.Hive.Common;
-using HeuristicLab.Clients.Hive.Slave;
 
 
@@ -37 +38 @@
   public class Core : MarshalByRefObject {
     public static bool abortRequested { get; set; }
-
     public static ILog Log { get; set; }
 
@@ -45 +45 @@
 
     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 HeartbeatManager heartbeatManager;
+
+    private int coreThreadId;
 
     public Dictionary<Guid, Executor> ExecutionEngines {
@@ -62 +53 @@
     }
 
-    internal Dictionary<Guid, Job> Jobs {    
+    internal Dictionary<Guid, Job> Jobs {
       get { return jobs; }
+    }
+
+    public Core() {
+      coreThreadId = Thread.CurrentThread.ManagedThreadId;
     }
 
@@ -86 +81 @@
 
       DeRegisterServiceEvents();
-      
+
       //server.Close();
       Logger.Info("Program shutdown");
@@ -92 +87 @@
 
     private void StartHeartbeats() {
-      //Initialize the heartbeat
+      //Initialize the heartbeat      
       heartbeatManager = new HeartbeatManager { Interval = new TimeSpan(0, 0, 10) };
       heartbeatManager.StartHeartbeat();
@@ -119 +114 @@
     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: 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
+              Logger.Error("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!!
-            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;
+            StartJobInAppDomain(myJob, jobData);
+            break;
+
+          //Hard shutdown of the client
+          case MessageContainer.MessageType.ShutdownSlave:
+            ShutdownCore();
+            break;
+        }
+      } else {
+        Logger.Warn("Unknown MessageContainer: " + container);
       }
     }
@@ -163 +152 @@
     public void ShutdownCore() {
       Logger.Info("Shutdown Signal received");
-      
       Logger.Debug("Stopping heartbeat");
       heartbeatManager.StopHeartBeat();
-      abortRequested = true;     
+      abortRequested = true;
       Logger.Debug("Logging out");
 
@@ -177 +165 @@
         }
       }
-     
-      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);      
-    }
-            
-    
+      WcfService.Instance.Disconnect();
+    }
+
+    /// <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)) {
+        Logger.Error("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. 
@@ -201 +190 @@
     /// </summary>
     /// <param name="jobId"></param>
+    [MethodImpl(MethodImplOptions.Synchronized)]
     public void SendFinishedJob(object jobId) {
       try {
@@ -208 +198 @@
           Logger.Info("Engine doesn't exist");
           return;
-        }  
+        }
         if (!jobs.ContainsKey(jId)) {
           Logger.Info("Job doesn't exist");
@@ -218 +208 @@
         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++;          
+          SlaveStatusInfo.JobsProcessed++;
         }
         catch (Exception e) {
-          Logger.Info("Transmitting to server failed. Storing the finished job with id: " + jId + " to hdd (" + e.ToString() + ")");          
+          Logger.Info("Transmitting to server failed. Storing the finished job with id: " + jId + " to hdd (" + e.ToString() + ")");
         }
         finally {
@@ -236 +226 @@
       }
     }
-  
+
     /// <summary>
     /// A new Job from the wcfService has been received and will be started within a AppDomain.
@@ -242 +232 @@
     /// <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;
+    private void StartJobInAppDomain(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(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;
-    }
-  
+          Logger.Error("Creating the Appdomain and loading the job failed for job " + myJob.Id);
+          Logger.Error("Error thrown is: ", exception);
+          KillAppDomain(myJob.Id);
+        }
+      }
+    }
+
     public event EventHandler<EventArgs<Exception>> ExceptionOccured;
     private void OnExceptionOccured(Exception e) {
@@ -295 +283 @@
     }
 
-    void appDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e) {
+    private void appDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e) {
       Logger.Error("Exception in AppDomain: " + e.ExceptionObject.ToString());
       KillAppDomain(new Guid(e.ExceptionObject.ToString()));
@@ -301 +289 @@
 
     /// <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) {
-      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
+      if (EnqueueExecutorMessage<Guid>(KillAppDomain, id)) {
+        Logger.Debug("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) {
+                  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);
             }
-            appDomains.Remove(id);
+
+            jobs.Remove(id);
+            PluginCache.Instance.DeletePluginsForJob(id);
+            GC.Collect();
           }
-
-          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();
-    } 
-  }   
-
-
+          catch (Exception ex) {
+            Logger.Error("Exception when unloading the appdomain: ", ex);
+          }
+        }
+        GC.Collect();
+      }
+    }
+  }
 }

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

@@ -21 +21 @@
 
 using System;
+using System.Collections.Generic;
 using System.Linq;
+using HeuristicLab.Clients.Hive.Slave;
 using HeuristicLab.Common;
 using HeuristicLab.Core;
-using HeuristicLab.Services.Hive.Common;
 using HeuristicLab.Hive;
+using HeuristicLab.PluginInfrastructure;
 using HeuristicLab.Services.Hive.Common.DataTransfer;
-using System.Collections.Generic;
-using HeuristicLab.Clients.Hive.Slave;
-using HeuristicLab.PluginInfrastructure;
-
-                                             
+
+
 namespace HeuristicLab.Clients.Hive.Salve {
   public class Executor : MarshalByRefObject, IDisposable {
     public Guid JobId { get; set; }
     public IJob Job { get; set; }
-    public MessageContainer.MessageType CurrentMessage { get; set; }
-    public Core core { get; set; }    
-        
+    private bool wasJobAborted = false;
+    public Core core { get; set; }
+
     private Exception currentException;
     public String CurrentException {
@@ -49 +48 @@
       }
     }
-    
+
     public ExecutionState ExecutionState {
       get {
@@ -73 +72 @@
         RegisterJobEvents();
 
-        if (Job.CollectChildJobs) {          
+        if (Job.CollectChildJobs) {
           IEnumerable<JobData> childjobs = WcfService.Instance.GetChildJobs(JobId);
           Job.Resume(childjobs.Select(j => PersistenceUtil.Deserialize<IJob>(j.Data)));
@@ -82 +81 @@
       }
       catch (Exception e) {
-        this.currentException = e;        
+        this.currentException = e;
       }
     }
@@ -91 +90 @@
       }
       catch (Exception e) {
-        this.currentException = e;        
+        this.currentException = e;
       }
     }
 
     public void Abort() {
-      CurrentMessage = MessageContainer.MessageType.AbortJob;
+      wasJobAborted = true;
       if ((ExecutionState == ExecutionState.Started) || (ExecutionState == ExecutionState.Paused)) {
         Job.Stop();
@@ -105 +104 @@
 
     private void RegisterJobEvents() {
-      //TODO: warum gibt es jobStopped und jobfailed nicht mehr? 
       Job.JobStopped += new EventHandler(Job_JobStopped);
       Job.JobFailed += new EventHandler(Job_JobFailed);
@@ -123 +121 @@
     private List<Guid> FindPluginsNeeded(IJob obj) {
       List<Guid> guids = new List<Guid>();
-      foreach(IPluginDescription desc in PluginUtil.GetDeclaringPlugins(obj)) {        
+      foreach (IPluginDescription desc in PluginUtil.GetDeclaringPlugins(obj)) {
       }
       throw new NotImplementedException("FindPluginsNeeded for Job_NewChildJob");
@@ -147 +145 @@
       // Pause the job and send it back to the hive. The server will awake it when all child-jobs are finished
       this.Job.CollectChildJobs = true;
-      
+
       JobData jdata = new JobData();
       jdata.Data = PersistenceUtil.Serialize(Job);
       jdata.JobId = this.JobId;
-      
+
       core.PauseJob(jdata);
     }
@@ -167 +165 @@
 
     private void Job_JobStopped(object sender, EventArgs e) {
-     if (CurrentMessage == MessageContainer.MessageType.AbortJob) {
-       core.KillAppDomain(JobId);
+      if (wasJobAborted) {
+        core.KillAppDomain(JobId);
       } else {
         core.SendFinishedJob(JobId);
@@ -176 +174 @@
     public JobData GetFinishedJob() {
       if (Job == null) {
-        throw new InvalidStateException("Job is null"); 
-      }              
+        throw new InvalidStateException("Job is null");
+      }
 
       if (Job.ExecutionState == HeuristicLab.Core.ExecutionState.Started) {
@@ -185 +183 @@
 
       if (Job.ExecutionState == HeuristicLab.Core.ExecutionState.Started) {
-        throw new InvalidStateException("Job is still running"); 
+        throw new InvalidStateException("Job is still running");
       } else {
         JobData jdata = new JobData();
@@ -192 +190 @@
         return jdata;
       }
-    }   
-   
+    }
+
     public Executor() {
-      // CurrentMessage = MessageContainer.MessageType.NoMessage; [chn] check usage of 'CurrentMessage'
     }
 
@@ -201 +198 @@
       if (Job != null)
         DeregisterJobEvents();
-      //Queue = null;
       Job = null;
     }

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

@@ -21 +21 @@
 
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Threading;
 using HeuristicLab.Common;
-using HeuristicLab.Clients.Hive.Salve;
+using HeuristicLab.Services.Hive.Common;
 using HeuristicLab.Services.Hive.Common.DataTransfer;
-using HeuristicLab.Services.Hive.Common;
-using System.Collections.Generic;
 
 namespace HeuristicLab.Clients.Hive.Salve {
@@ -49 +48 @@
     private WcfService wcfService;
 
-    private bool abortThreadPending;
+    private bool threadStopped;
 
     ReaderWriterLockSlim heartBeatThreadIsSleepingLock = new ReaderWriterLockSlim();
-    
+
     /// <summary>
     /// Starts the Heartbeat signal.
@@ -58 +57 @@
     public void StartHeartbeat() {
       this.waitHandle = new AutoResetEvent(true);
-      wcfService = WcfService.Instance;     
-      abortThreadPending = false;
+      wcfService = WcfService.Instance;
+      threadStopped = false;
       heartBeatThread = new Thread(RunHeartBeatThread);
       heartBeatThread.Start();
@@ -68 +67 @@
     /// </summary>
     public void StopHeartBeat() {
-      abortThreadPending = true;
+      threadStopped = true;
       waitHandle.Set();
       heartBeatThread.Join();
     }
-    
+
     /// <summary>
     /// use this method to singalize there is work to do (to avoid the waiting period if its clear that actions are required)
     /// </summary>
     public void AwakeHeartBeatThread() {
-      waitHandle.Set();
+      if (!threadStopped)
+        waitHandle.Set();
     }
 
     private void RunHeartBeatThread() {
-      while (!abortThreadPending) {
+      while (!threadStopped) {
         try {
           lock (locker) {
@@ -87 +87 @@
               wcfService.Connect(ConfigManager.Instance.GetClientInfo()); // Login happens automatically upon successfull connection
             }
-            if(wcfService.ConnState == NetworkEnum.WcfConnState.Connected) {
+            if (wcfService.ConnState == NetworkEnum.WcfConnState.Connected) {
               HeuristicLab.Services.Hive.Common.DataTransfer.Slave info = ConfigManager.Instance.GetClientInfo();
 
-              Heartbeat heartBeatData = new Heartbeat
-              {
+              Heartbeat heartBeatData = new Heartbeat {
                 SlaveId = info.Id,
                 FreeCores = info.Cores.HasValue ? info.Cores.Value - ConfigManager.Instance.GetUsedCores() : 0,
                 FreeMemory = GetFreeMemory(),
-                JobProgress = ConfigManager.Instance.GetExecutionTimeOfAllJobs()                
+                JobProgress = ConfigManager.Instance.GetExecutionTimeOfAllJobs()
               };
-                              
+
               Logger.Debug("Sending Heartbeat: " + heartBeatData);
-              List<MessageContainer> msgs =  wcfService.SendHeartbeat(heartBeatData);
-              
+              List<MessageContainer> msgs = wcfService.SendHeartbeat(heartBeatData);
+
               if (msgs == null) {
                 Logger.Debug("Error getting response from Heartbeat");
@@ -108 +107 @@
               Logger.Debug("Heartbeat Response received: ");
               msgs.ForEach(mc => Logger.Debug(mc.Message.ToString()));
-              msgs.ForEach(mc => MessageQueue.GetInstance().AddMessage(mc));      
+              msgs.ForEach(mc => MessageQueue.GetInstance().AddMessage(mc));
             }
-          } // lock
+          }
         }
         catch (Exception e) {
@@ -117 +116 @@
         }
         waitHandle.WaitOne(this.Interval);
-      } // while
+      }
       waitHandle.Close();
-      abortThreadPending = false;
       Logger.Debug("Heartbeat thread stopped");
     }
-         
+
 
     #region Eventhandler

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive.Slave/3.4/HeuristicLab.Clients.Hive.Slave-3.4.csproj

@@ -72 +72 @@
       <SubType>Code</SubType>
     </Compile>
+    <Compile Include="ExecutorMessageContainer.cs" />
     <Compile Include="MessageQueue.cs" />
     <Compile Include="NetworkEnum.cs" />

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

@@ -1 +1 @@
 using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using HeuristicLab.Core;
 
 namespace HeuristicLab.Clients.Hive.Salve {
@@ -11 +7 @@
   /// TODO: send messages to gui
   /// </summary>
-  internal static class Logger {   
-    private static object locker = new object(); 
+  public static class Logger {
+    private static object locker = new object();
 
     public static void Debug(object message) {
@@ -42 +38 @@
         Console.WriteLine(message.ToString() + " \n Exception is :" + e.ToString());
       }
-    }   
+    }
   }
 }

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Services.Hive.Common/3.4/MessageContainer.cs

@@ -43 +43 @@
       SayHello,  // Slave should say hello, because he is unknown to the server
 
-      // *** events from execution engine ***
-      JobStopped,      // job finished, submit results, unload appdomain
-      JobPaused,       // job paused, submit results, unload appdomain
-      JobFailed,       // job failed with an exception. submit the results, unload appdomain
-
       // *** commands from execution engine ***
       AddChildJob,     // adds a new child job for the provided jobId

@@ +1 @@
+*.suo
 HeuristicLab.Hive-3.4.suo
 TestResults
-HeuristicLab.Hive 3.4.suo