Changeset 4423 for branches/3.3-Hive/sources/HeuristicLab.Hive/HeuristicLab.Hive.Slave.ExecutionEngine

Timestamp:

09/17/10 10:26:55 (14 years ago)

Author:

cneumuel

Message:

• Refactored HL.Hive.Experiment. JobItems are not called HiveJobs and OptimizerJobs do not contain a hierarchy anymore.
• Dynamic generation of jobs on a slave are not reflected on the client user interface.
• Optimizer-Trees are now strictly synchronized with the HiveJob-Trees (also the ComputeInParallel property is taken into account when the Child HiveJobs are created)
• Improved the way a class can report progress and lock the UI (IProgressReporter, IProgress, Progress, ProgressView)
• Changes were made to the config-files, so that server and clients work with blade12.hpc.fh-hagenberg.at
• Lots of small changes and bugfixes

File:

• branches/3.3-Hive/sources/HeuristicLab.Hive/HeuristicLab.Hive.Slave.ExecutionEngine/3.3/Executor.cs (modified) (11 diffs)

branches/3.3-Hive/sources/HeuristicLab.Hive/HeuristicLab.Hive.Slave.ExecutionEngine/3.3/Executor.cs

@@ -39 +39 @@
     public IJob Job { get; set; }
     public MessageContainer.MessageType CurrentMessage { get; set; }
-    public Exception CurrentException { get; set; }
+
+    private Exception currentException;
+    public String CurrentException {
+      get {
+        if (currentException != null) {
+          return currentException.ToString();
+        } else {
+          return string.Empty;
+        }
+      }
+    }
     public MessageQueue Queue { get; set; }
 
@@ -50 +60 @@
     }
 
-    public double Progress {
+    public TimeSpan ExecutionTime {
       get {
-        return Job.Progress;
+        return Job.ExecutionTime;
       }
     }
@@ -58 +68 @@
     public DateTime CreationTime { get; set; }
 
+    /// <summary>
+    /// 
+    /// </summary>
+    /// <param name="serializedJob"></param>
+    /// <param name="collectChildJobs">if true, all child-jobs are downloaded and the job will be resumed.</param>
     public void Start(byte[] serializedJob) {
-      CreationTime = DateTime.Now;
-      Job = DeserializeJobObject(serializedJob);
-
+      try {
+        CreationTime = DateTime.Now;
+        Job = SerializedJob.Deserialize<IJob>(serializedJob);
+
+        RegisterJobEvents();
+
+        if (Job.CollectChildJobs) {
+          Queue.AddMessage(new MessageContainerWithCallback<SerializedJobList>(MessageContainer.MessageType.GetChildJobs, JobId,
+            (childjobs) => {
+              Job.Resume(childjobs.Select(j => SerializedJob.Deserialize<IJob>(j.SerializedJobData)));
+            }
+          ));
+        } else {
+          Job.Prepare();
+          Job.Start();
+        }
+      }
+      catch (OutOfMemoryException e) {
+        JobIsFinished = true;
+        this.currentException = e;
+        Queue.AddMessage(new MessageContainer(MessageContainer.MessageType.JobFailed, JobId));
+      }
+    }
+
+    public void StartOnlyJob() {
+      try {
+        Job.Start();
+      }
+      catch (OutOfMemoryException e) {
+        JobIsFinished = true;
+        this.currentException = e;
+        Queue.AddMessage(new MessageContainer(MessageContainer.MessageType.JobFailed, JobId));
+      }
+    }
+
+    public void Abort() {
+      CurrentMessage = MessageContainer.MessageType.AbortJob;
+      if ((ExecutionState == ExecutionState.Started) || (ExecutionState == ExecutionState.Paused)) {
+        Job.Stop();
+      }
+    }
+
+    private void RegisterJobEvents() {
       Job.JobStopped += new EventHandler(Job_JobStopped);
       Job.JobFailed += new EventHandler(Job_JobFailed);
       Job.NewChildJob += new EventHandler<EventArgs<IJob>>(Job_NewChildJob);
       Job.WaitForChildJobs += new EventHandler(Job_WaitForChildJobs);
-
-      if (Job.ExecutionState == Core.ExecutionState.Paused) {
-        Queue.AddMessage(new MessageContainerWithCallback<SerializedJobList>(MessageContainer.MessageType.GetChildJobs, JobId,
-          (childjobs) => {
-            Job.Resume(childjobs.Select(j => DeserializeJobObject(j.SerializedJobData)));
-          }
-        ));
-      } else {
-        Job.Prepare();
-        Job.Start();
-      }
-    }
-
-    void Job_NewChildJob(object sender, EventArgs<IJob> e) {
-      byte[] jobByteArray = SerializeJobObject(e.Value);
+      Job.DeleteChildJobs += new EventHandler(Job_DeleteChildJobs);
+    }
+
+    private void DeregisterJobEvents() {
+      Job.JobStopped -= new EventHandler(Job_JobStopped);
+      Job.JobFailed -= new EventHandler(Job_JobFailed);
+      Job.NewChildJob -= new EventHandler<EventArgs<IJob>>(Job_NewChildJob);
+      Job.WaitForChildJobs -= new EventHandler(Job_WaitForChildJobs);
+      Job.DeleteChildJobs -= new EventHandler(Job_DeleteChildJobs);
+    }
+
+    private void Job_NewChildJob(object sender, EventArgs<IJob> e) {
+      byte[] jobByteArray = SerializedJob.Serialize(e.Value);
 
       SerializedJob serializedJob = new SerializedJob() {
         JobInfo = new JobDto() {
           State = JobState.Offline,
-          CoresNeeded = e.Value.CoresNeeded,
-          MemoryNeeded = e.Value.MemoryNeeded,
+          CoresNeeded = 1,
+          MemoryNeeded = 0,
           PluginsNeeded = HivePluginInfoDto.FindPluginsNeeded(e.Value.GetType()),
         },
@@ -95 +148 @@
     }
 
-    void Job_WaitForChildJobs(object sender, EventArgs e) {
+    private void Job_WaitForChildJobs(object sender, EventArgs e) {
       // Pause the job and send it back to the hive. The server will awake it when all child-jobs are finished
-      byte[] jobByteArray = SerializeJobObject(Job);
+      this.Job.CollectChildJobs = true;
+      byte[] jobByteArray = SerializedJob.Serialize(Job);
 
       SerializedJob serializedJob = new SerializedJob() {
@@ -111 +165 @@
     }
 
-    void Job_JobFailed(object sender, EventArgs e) {
+    private void Job_DeleteChildJobs(object sender, EventArgs e) {
+      Queue.AddMessage(new MessageContainer(MessageContainer.MessageType.DeleteChildJobs, this.JobId));
+    }
+
+    private void Job_JobFailed(object sender, EventArgs e) {
       HeuristicLab.Common.EventArgs<Exception> ex = (HeuristicLab.Common.EventArgs<Exception>)e;
-      CurrentException = ex.Value;
+      currentException = ex.Value;
       Queue.AddMessage(new MessageContainer(MessageContainer.MessageType.FinishedJob, JobId));
     }
 
-    public void StartOnlyJob() {
-      Job.Start();
-    }
-
-    public void Abort() {
-      CurrentMessage = MessageContainer.MessageType.AbortJob;
-      if (Job.ExecutionState != Core.ExecutionState.Stopped) {
-        Job.Stop();
-      }
-    }
-
-    void Job_JobStopped(object sender, EventArgs e) {
+    private void Job_JobStopped(object sender, EventArgs e) {
       if (CurrentMessage == MessageContainer.MessageType.NoMessage) {
         Queue.AddMessage(new MessageContainer(MessageContainer.MessageType.FinishedJob, JobId));
@@ -147 +194 @@
         CurrentMessage = MessageContainer.MessageType.NoMessage;
         // Pack the whole job inside an xml document
-        byte[] job = SerializeJobObject(Job);
+        byte[] job = SerializedJob.Serialize(Job);
         // Restart the job
         // Return the Snapshot
@@ -159 +206 @@
         throw new InvalidStateException("Job is still running");
       } else {
-        byte[] jobArr = SerializeJobObject(Job);
+        byte[] jobArr = SerializedJob.Serialize(Job);
         return jobArr;
       }
@@ -169 +216 @@
     }
 
-    private byte[] SerializeJobObject(IJob job) {
-      MemoryStream memStream = new MemoryStream();
-      XmlGenerator.Serialize(job, memStream);
-      byte[] jobByteArray = memStream.ToArray();
-      memStream.Dispose();
-      return jobByteArray;
-    }
-
-    private IJob DeserializeJobObject(byte[] sjob) {
-      MemoryStream memStream = new MemoryStream(sjob);
-      IJob job = XmlParser.Deserialize<IJob>(memStream);
-      memStream.Dispose();
-      return job;
-    }
-
     private void RestoreJobObject(byte[] sjob) {
-      Job = DeserializeJobObject(sjob);
+      Job = SerializedJob.Deserialize<IJob>(sjob);
     }
 
@@ -191 +223 @@
       CurrentMessage = MessageContainer.MessageType.NoMessage;
       JobIsFinished = false;
-      //Job = new TestJob();
     }
 
@@ -197 +228 @@
 
     public void Dispose() {
-      Job.JobFailed -= new EventHandler(Job_JobFailed);
-      Job.JobFailed -= new EventHandler(Job_JobStopped);
+      DeregisterJobEvents();
       Queue = null;
       Job = null;
@@ -204 +234 @@
 
     #endregion
+
   }
 }