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

Timestamp:

03/22/11 11:36:53 (13 years ago)

Author:

cneumuel

Message:

#1233

• implemented correct numbering of BatchRuns
• improvements in ExperimentManager
• fixed bug in server (jobs were scheduled multiple times)
• added exception handling for task in slave
• improved timeout handling of jobs (LifecycleManager)

File:

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

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

@@ -57 +57 @@
     private int coreThreadId;
 
-    private ISlaveCommunication ClientCom;
+    private ISlaveCommunication clientCom;
     private ServiceHost slaveComm;
 
@@ -84 +84 @@
         slaveComm.Open();
 
-        ClientCom = SlaveClientCom.Instance.ClientCom;
-        ClientCom.LogMessage("Hive Slave started");
+        clientCom = SlaveClientCom.Instance.ClientCom;
+        clientCom.LogMessage("Hive Slave started");
 
         ConfigManager manager = ConfigManager.Instance;
@@ -139 +139 @@
 
     void WcfService_ExceptionOccured(object sender, EventArgs<Exception> e) {
-      ClientCom.LogMessage("Connection to server interruped with exception: " + e.Value.Message);
+      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");
+      clientCom.LogMessage("Connected successfully to Hive server");
     }
 
@@ -151 +151 @@
     /// <param name="container">The Container, containing the message</param>
     private void DetermineAction(MessageContainer container) {
-      ClientCom.LogMessage("Message: " + container.Message.ToString() + " for job: " + container.JobId);
+      clientCom.LogMessage("Message: " + container.Message.ToString() + " for job: " + container.JobId);
 
       if (container is ExecutorMessageContainer<Guid>) {
@@ -159 +159 @@
         switch (container.Message) {
           case MessageContainer.MessageType.CalculateJob:
-            Task.Factory.StartNew(() => {
-              Job job = wcfService.GetJob(container.JobId);
+            clientCom.LogMessage("Task.StartNew[0]: jobId: " + container.JobId);
+            Task.Factory.StartNew((jobIdObj) => {
+              Guid jobId = (Guid)jobIdObj;
+              clientCom.LogMessage("Task.StartNew[1]: jobId: " + jobId);
+              Job job = wcfService.GetJob(jobId);
+              if (job == null) throw new JobNotFoundException(jobId);
               lock (engines) {
                 if (!jobs.ContainsKey(job.Id)) {
@@ -167 +171 @@
               }
               JobData jobData = wcfService.GetJobData(job.Id);
+              if (job == null) throw new JobDataNotFoundException(jobId);
               job = wcfService.UpdateJobState(job.Id, JobState.Calculating, null);
               StartJobInAppDomain(job, jobData);
-            });
+            }, container.JobId)
+            .ContinueWith((t) => {
+              // handle exception of task
+              clientCom.LogMessage(t.Exception.ToString());
+            }, TaskContinuationOptions.OnlyOnFaulted);
             break;
           case MessageContainer.MessageType.ShutdownSlave:
@@ -203 +212 @@
         }
       } else {
-        ClientCom.LogMessage("Unknown MessageContainer: " + container);
+        clientCom.LogMessage("Unknown MessageContainer: " + container);
       }
     }
@@ -217 +226 @@
 
         try {
-          ClientCom.LogMessage("Sending the paused job with id: " + job.Id);
+          clientCom.LogMessage("Sending the paused job with id: " + job.Id);
           wcfService.UpdateJobData(job, sJob, ConfigManager.Instance.GetClientInfo().Id, JobState.Paused);
           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() + ")");
+          clientCom.LogMessage("Transmitting to server failed. Storing the paused job with id: " + job.Id + " to hdd (" + e.ToString() + ")");
         }
         finally {
@@ -240 +249 @@
 
         try {
-          ClientCom.LogMessage("Sending the stoppped job with id: " + job.Id);
+          clientCom.LogMessage("Sending the stoppped job with id: " + job.Id);
           wcfService.UpdateJobData(job, sJob, ConfigManager.Instance.GetClientInfo().Id, JobState.Paused);
           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() + ")");
+          clientCom.LogMessage("Transmitting to server failed. Storing the paused job with id: " + job.Id + " to hdd (" + e.ToString() + ")");
         }
         finally {
@@ -266 +275 @@
       }
 
-      ClientCom.LogMessage("Aborted all jobs!");
+      clientCom.LogMessage("Aborted all jobs!");
     }
 
@@ -273 +282 @@
     /// </summary>
     private void DoPauseAll() {
-      ClientCom.LogMessage("Pause all received");
+      clientCom.LogMessage("Pause all received");
 
       //copy guids because there will be removed items from 'Jobs'
@@ -290 +299 @@
     /// </summary>
     private void DoStopAll() {
-      ClientCom.LogMessage("Stop all received");
+      clientCom.LogMessage("Stop all received");
 
       //copy guids because there will be removed items from 'Jobs'
@@ -316 +325 @@
     /// </summary>
     private void ShutdownCore() {
-      ClientCom.LogMessage("Shutdown Signal received");
-      ClientCom.LogMessage("Stopping heartbeat");
+      clientCom.LogMessage("Shutdown Signal received");
+      clientCom.LogMessage("Stopping heartbeat");
       heartbeatManager.StopHeartBeat();
       abortRequested = true;
-      ClientCom.LogMessage("Logging out");
+      clientCom.LogMessage("Logging out");
 
 
       lock (engines) {
-        ClientCom.LogMessage("engines locked");
+        clientCom.LogMessage("engines locked");
         foreach (KeyValuePair<Guid, AppDomain> kvp in appDomains) {
-          ClientCom.LogMessage("Shutting down Appdomain for " + kvp.Key);
+          clientCom.LogMessage("Shutting down Appdomain for " + kvp.Key);
           appDomains[kvp.Key].UnhandledException -= new UnhandledExceptionEventHandler(AppDomain_UnhandledException);
           AppDomain.Unload(kvp.Value);
@@ -332 +341 @@
       }
       WcfService.Instance.Disconnect();
-      ClientCom.Shutdown();
+      clientCom.Shutdown();
       SlaveClientCom.Close();
 
@@ -344 +353 @@
     /// </summary>  
     private void DoStartSlave() {
-      ClientCom.LogMessage("Restart received");
+      clientCom.LogMessage("Restart received");
       StartHeartbeats();
-      ClientCom.LogMessage("Restart done");
+      clientCom.LogMessage("Restart done");
     }
 
@@ -355 +364 @@
     //TODO: do we need an AbortSleep?
     private void Sleep() {
-      ClientCom.LogMessage("Sleep received");
+      clientCom.LogMessage("Sleep received");
       heartbeatManager.StopHeartBeat();
       heartbeatManager = null;
       DoStopAll();
       WcfService.Instance.Disconnect();
-      ClientCom.LogMessage("Sleep done");
+      clientCom.LogMessage("Sleep done");
     }
 
@@ -371 +380 @@
     public void PauseWaitJob(JobData data) {
       if (!Jobs.ContainsKey(data.JobId)) {
-        ClientCom.LogMessage("Can't find job with id " + data.JobId);
+        clientCom.LogMessage("Can't find job with id " + data.JobId);
       } else {
         Job job = Jobs[data.JobId];
@@ -388 +397 @@
     public void SendFinishedJob(Guid jobId) {
       try {
-        ClientCom.LogMessage("Getting the finished job with id: " + jobId);
+        clientCom.LogMessage("Getting the finished job with id: " + jobId);
         if (!engines.ContainsKey(jobId)) {
-          ClientCom.LogMessage("Engine doesn't exist");
+          clientCom.LogMessage("Engine doesn't exist");
           return;
         }
         if (!jobs.ContainsKey(jobId)) {
-          ClientCom.LogMessage("Job doesn't exist");
+          clientCom.LogMessage("Job doesn't exist");
           return;
         }
@@ -405 +414 @@
 
         try {
-          ClientCom.LogMessage("Sending the finished job with id: " + jobId);
+          clientCom.LogMessage("Sending the finished job with id: " + jobId);
           wcfService.UpdateJobData(cJob, sJob, ConfigManager.Instance.GetClientInfo().Id, JobState.Finished);
           SlaveStatusInfo.JobsProcessed++;
         }
         catch (Exception e) {
-          ClientCom.LogMessage("Transmitting to server failed. Storing the finished job with id: " + jobId + " to hdd (" + e.ToString() + ")");
+          clientCom.LogMessage("Transmitting to server failed. Storing the finished job with id: " + jobId + " to hdd (" + e.ToString() + ")");
         }
         finally {
@@ -428 +437 @@
     /// <param name="e"></param>
     private void StartJobInAppDomain(Job myJob, JobData jobData) {
-      ClientCom.LogMessage("Received new job with id " + myJob.Id);
+      clientCom.LogMessage("Received new job with id " + myJob.Id);
       String pluginDir = Path.Combine(PluginCache.Instance.PluginTempBaseDir, myJob.Id.ToString());
       bool pluginsPrepared = false;
@@ -435 +444 @@
       try {
         PluginCache.Instance.PreparePlugins(myJob, out configFileName);
-        ClientCom.LogMessage("Plugins fetched for job " + myJob.Id);
+        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));
+        clientCom.LogMessage(string.Format("Copying plugins for job {0} failed: {1}", myJob.Id, exception));
       }
 
@@ -448 +457 @@
           lock (engines) {
             appDomains.Add(myJob.Id, appDomain);
-            ClientCom.LogMessage("Creating AppDomain");
+            clientCom.LogMessage("Creating AppDomain");
             Executor engine = (Executor)appDomain.CreateInstanceAndUnwrap(typeof(Executor).Assembly.GetName().Name, typeof(Executor).FullName);
-            ClientCom.LogMessage("Created AppDomain");
+            clientCom.LogMessage("Created AppDomain");
             engine.JobId = myJob.Id;
             engine.Core = this;
-            ClientCom.LogMessage("Starting Engine for job " + myJob.Id);
+            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);
+            clientCom.LogMessage("Increment FetchedJobs to:" + SlaveStatusInfo.JobsFetched);
           }
         }
         catch (Exception exception) {
-          ClientCom.LogMessage("Creating the Appdomain and loading the job failed for job " + myJob.Id);
-          ClientCom.LogMessage("Error thrown is: " + exception.ToString());
+          clientCom.LogMessage("Creating the Appdomain and loading the job failed for job " + myJob.Id);
+          clientCom.LogMessage("Error thrown is: " + exception.ToString());
           KillAppDomain(myJob.Id);
         }
@@ -471 +480 @@
     public event EventHandler<EventArgs<Exception>> ExceptionOccured;
     private void OnExceptionOccured(Exception e) {
-      ClientCom.LogMessage("Error: " + e.ToString());
+      clientCom.LogMessage("Error: " + e.ToString());
       var handler = ExceptionOccured;
       if (handler != null) handler(this, new EventArgs<Exception>(e));
@@ -477 +486 @@
 
     private void AppDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e) {
-      ClientCom.LogMessage("Exception in AppDomain: " + e.ExceptionObject.ToString());
+      clientCom.LogMessage("Exception in AppDomain: " + e.ExceptionObject.ToString());
       KillAppDomain(new Guid(e.ExceptionObject.ToString()));
     }
@@ -508 +517 @@
       }
 
-      ClientCom.LogMessage("Shutting down Appdomain for Job " + id);
+      clientCom.LogMessage("Shutting down Appdomain for Job " + id);
       lock (engines) {
         try {
@@ -526 +535 @@
               }
               catch (CannotUnloadAppDomainException) {
-                ClientCom.LogMessage("Could not unload AppDomain, will try again in 1 sec.");
+                clientCom.LogMessage("Could not unload AppDomain, will try again in 1 sec.");
                 Thread.Sleep(1000);
                 repeat--;
@@ -542 +551 @@
         }
         catch (Exception ex) {
-          ClientCom.LogMessage("Exception when unloading the appdomain: " + ex.ToString());
+          clientCom.LogMessage("Exception when unloading the appdomain: " + ex.ToString());
         }
       }