Changeset 4133 for branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Experiment

Timestamp:

08/02/10 17:27:24 (14 years ago)

Author:

cneumuel

Message:

• Made HiveExperiment storable, so that a running HiveExperiment can be disconnected, stored and later resumed. (#1115)
• Added Log to each JobItem (#1115)

Location:

branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Experiment/3.3

Files:

• HeuristicLab.Hive.Experiment-3.3.csproj (modified) (1 diff)
• HiveExperiment.cs (modified) (15 diffs)
• JobItem.cs (modified) (4 diffs)
• JobItemList.cs (modified) (1 diff)
• Properties/AssemblyInfo.cs (modified) (1 diff)

branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Experiment/3.3/HeuristicLab.Hive.Experiment-3.3.csproj

@@ -85 +85 @@
       <HintPath>..\..\..\..\..\..\..\..\..\Programme\HeuristicLab 3.3\HeuristicLab.PluginInfrastructure-3.3.dll</HintPath>
     </Reference>
-    <Reference Include="HeuristicLab.Tracing-3.3, Version=3.2.0.0, Culture=neutral, PublicKeyToken=ba48961d6f65dcec, processorArchitecture=MSIL" />
+    <Reference Include="HeuristicLab.Tracing-3.3" />
     <Reference Include="System" />
     <Reference Include="System.Core" />

branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Experiment/3.3/HiveExperiment.cs

@@ -53 +53 @@
     private const string itemDescription = "An experiment which contains multiple batch runs of algorithms which are executed in the Hive.";
     private const int resultPollingIntervalMs = 15000;
-
+    private const int maxSnapshotRetries = 20;
     private object locker = new object();
-    private const int maxSnapshotRetries = 20;
+
     private System.Timers.Timer timer;
     private bool pausePending, stopPending;
+
+    [Storable]
     private DateTime lastUpdateTime;
 
+    private bool isPollingResults;
+    public bool IsPollingResults {
+      get { return isPollingResults; }
+      private set {
+        if (isPollingResults != value) {
+          isPollingResults = value;
+          OnIsPollingResultsChanged();
+        }
+      }
+    }
+
+    private bool stopResultsPollingPending = false;
+
+    private IDictionary<Guid, Thread> resultPollingThreads;
+    
     [Storable]
     private IDictionary<Guid, IOptimizer> pendingOptimizers = new Dictionary<Guid, IOptimizer>();
@@ -68 +85 @@
       get { return jobItems; }
     }
-    
+
 
     [Storable]
@@ -76 +93 @@
       set {
         if (serverUrl != value) {
-          serverUrl = value; 
+          serverUrl = value;
           OnServerUrlChanged();
         }
@@ -88 +105 @@
       set {
         if (resourceIds != value) {
-          resourceIds = value; 
+          resourceIds = value;
           OnResourceIdsChanged();
         }
@@ -115 +132 @@
     public HiveExperiment(bool deserializing)
       : base(deserializing) {
+      this.resultPollingThreads = new Dictionary<Guid, Thread>();
+      jobItems = new JobItemList();
     }
 
@@ -124 +143 @@
       pausePending = stopPending = false;
       jobItems = new JobItemList();
+      isPollingResults = false;
+      resultPollingThreads = new Dictionary<Guid, Thread>();
       InitTimer();
     }
@@ -142 +163 @@
       clone.pausePending = this.pausePending;
       clone.jobItems = (JobItemList)cloner.Clone(jobItems);
+      clone.lastUpdateTime = this.lastUpdateTime;
+      clone.isPollingResults = this.isPollingResults;
       return clone;
     }
@@ -148 +171 @@
     private void AfterDeserialization() {
       InitTimer();
+      this.IsPollingResults = false;
+      this.stopResultsPollingPending = false;
       LogMessage("I was deserialized.");
     }
@@ -215 +240 @@
       this.ExecutionState = Core.ExecutionState.Started;
       Thread t = new Thread(() => {
-        IExecutionEngineFacade executionEngineFacade = ServiceLocator.CreateExecutionEngineFacade(ServerUrl);
+        IExecutionEngineFacade executionEngineFacade = GetExecutionEngineFacade();
 
         pendingOptimizers = new Dictionary<Guid, IOptimizer>();
@@ -234 +259 @@
           };
           jobItems.Add(jobItem);
-
-          LogMessage("Sent job to server (jobId: " + response.Obj.Id + ")");
-        }
-
+          jobItem.LogMessage("Job sent to Hive");
+
+          LogMessage("Sent job to Hive (jobId: " + response.Obj.Id + ")");
+        }
+        
         // start results polling after sending sending the jobs to the server (to avoid race conflicts at the optimizers-collection)
-        foreach (JobItem jobItem in jobItems) {
-          StartResultPollingThread(jobItem.JobDto);
-        }
+        StartResultPolling();
       });
       t.Start();
+    }
+
+    private void CreateResultPollingThreads() {
+      foreach(JobItem jobItem in JobItems) {
+        resultPollingThreads.Add(jobItem.JobDto.Id, CreateResultPollingThread(jobItem.JobDto));
+      }
+    }
+
+    public void StartResultPolling() {
+      this.stopResultsPollingPending = false;
+      CreateResultPollingThreads();
+      foreach (Thread pollingThread in resultPollingThreads.Values) {
+        pollingThread.Start();
+      }
+      this.IsPollingResults = true;
+    }
+
+    public void StopResultPolling() {
+      this.stopResultsPollingPending = true;
+      foreach (Thread pollingThread in resultPollingThreads.Values) {
+        pollingThread.Interrupt();
+      }
+      this.stopResultsPollingPending = false;
+    }
+
+    private JobItem GetJobItemById(Guid jobId) {
+      return jobItems.Single(x => x.JobDto.Id == jobId);
     }
 
@@ -267 +318 @@
 
     public void Stop() {
-      // todo
-    }
-    
+      foreach(JobItem jobItem in jobItems) {
+        AbortJob(jobItem.JobDto.Id);
+      }
+    }
+
+    public void AbortJob(Guid jobId) {
+      IExecutionEngineFacade executionEngineFacade = GetExecutionEngineFacade();
+      executionEngineFacade.AbortJob(jobId);
+      resultPollingThreads[jobId].Interrupt();
+      GetJobItemById(jobId).LogMessage("Aborting Job");
+    }
     #endregion
+
+    private IExecutionEngineFacade GetExecutionEngineFacade() {
+      return ServiceLocator.CreateExecutionEngineFacade(ServerUrl);
+    }
 
     private SerializedJob CreateSerializedJob(IOptimizer optimizer) {
@@ -307 +370 @@
     }
 
-    private void StartResultPollingThread(JobDto job) {
-      Thread t = new Thread(() => {
-        IExecutionEngineFacade executionEngineFacade = ServiceLocator.CreateExecutionEngineFacade(ServerUrl);
-        IJob restoredObject = null;
-
-        do {
-          Thread.Sleep(resultPollingIntervalMs);
-          //lock (locker) { [chn] try without locking for better performance
-            if (stopPending) return;
+    private Thread CreateResultPollingThread(JobDto job) {
+      return new Thread(() => {
+        try {
+          GetJobItemById(job.Id).LogMessage("Starting job results polling");
+          IExecutionEngineFacade executionEngineFacade = GetExecutionEngineFacade();
+          IJob restoredObject = null;
+
+          do {
+            Thread.Sleep(resultPollingIntervalMs);
+            if (stopPending || !this.IsPollingResults) {
+              return;
+            }
 
             ResponseObject<JobDto> response = executionEngineFacade.GetJobById(job.Id);
             LogMessage("Response: " + response.StatusMessage + " (jobId: " + job.Id + ")");
+            GetJobItemById(job.Id).LogMessage("Response: " + response.StatusMessage);
 
             if (response.Obj != null) {
               UpdateJobItem(response.Obj);
             }
-            
+
             // loop while 
             // 1. the user doesn't request an abort 
@@ -334 +401 @@
               UpdateSnapshot(jobResponse);
             }
-          //}
-        } while (restoredObject == null || restoredObject.ExecutionState != Core.ExecutionState.Stopped);
-
-        LogMessage("Job finished (jobId: " + job.Id + ")");
-        // job retrieved... replace the existing optimizers with the finished one
-        IOptimizer originalOptimizer = pendingOptimizers[job.Id];
-        IOptimizer restoredOptimizer = ((OptimizerJob)restoredObject).Optimizer;
-
-        ReplaceOptimizer(originalOptimizer, restoredOptimizer);
-        pendingOptimizers.Remove(job.Id);
-
-        if (pendingOptimizers.Count == 0) {
-          // finished
-          this.ExecutionState = Core.ExecutionState.Stopped;
-          OnStopped();
+          } while (restoredObject == null || restoredObject.ExecutionState != Core.ExecutionState.Stopped);
+
+          LogMessage("Job finished (jobId: " + job.Id + ")");
+          GetJobItemById(job.Id).LogMessage("Job finished");
+          // job retrieved... replace the existing optimizers with the finished one
+          IOptimizer originalOptimizer = pendingOptimizers[job.Id];
+          IOptimizer restoredOptimizer = ((OptimizerJob)restoredObject).Optimizer;
+
+          ReplaceOptimizer(originalOptimizer, restoredOptimizer);
+          pendingOptimizers.Remove(job.Id);
+
+          if (pendingOptimizers.Count == 0) {
+            // finished
+            this.ExecutionState = Core.ExecutionState.Stopped;
+            OnStopped();
+          }
+        } catch (ThreadInterruptedException exception) {
+
+        } finally {
+          GetJobItemById(job.Id).LogMessage("ResultsPolling Thread stopped");
+          resultPollingThreads.Remove(job.Id);
+          if (resultPollingThreads.Count == 0) {
+            IsPollingResults = false;
+          }
         }
       });
-      t.Start();
     }
 
@@ -370 +445 @@
       }
     }
-    
+
     #region Required Plugin Search
     /// <summary>
@@ -523 +598 @@
     }
 
+    public event EventHandler IsResultsPollingChanged;
+    private void OnIsPollingResultsChanged() {
+      if (this.IsPollingResults) {
+        LogMessage("Results Polling Started");
+        timer.Start();
+      } else {
+        LogMessage("Results Polling Stopped");
+        timer.Stop();
+      }
+      EventHandler handler = IsResultsPollingChanged;
+      if (handler != null) handler(this, EventArgs.Empty);
+    }
     #endregion
   }

branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Experiment/3.3/JobItem.cs

@@ -8 +8 @@
 using HeuristicLab.Hive.Contracts;
 using System.Drawing;
+using HeuristicLab.Common;
+using System.Diagnostics;
 
 namespace HeuristicLab.Hive.Experiment {
+  [StorableClass]
   public class JobItem : Item {
+    private static object locker = new object();
+
     public override Image ItemImage {
       get {
@@ -59 +64 @@
       get { return log; }
     }
-
+    
     public JobItem() {
       log = new Log();
@@ -73 +78 @@
 
     public event EventHandler LatestSnapshotChanged;
-    public void OnLatestSnapshotChanged() {
+    private void OnLatestSnapshotChanged() {
+      LogMessage("LatestSnapshotChanged");
       EventHandler handler = LatestSnapshotChanged;
       if (handler != null) handler(this, EventArgs.Empty);
@@ -80 +86 @@
     public event EventHandler JobDtoChanged;
     private void OnJobDtoChanged() {
+      LogMessage("JobDtoChanged");
       EventHandler handler = JobDtoChanged;
       if (handler != null) handler(this, EventArgs.Empty);
     }
+
+    public void LogMessage(string message) {
+      lock (locker) {
+        log.LogMessage(message);
+      }
+    }
+
+    public override IDeepCloneable Clone(Cloner cloner) {
+      LogMessage("I am beeing cloned");
+      JobItem clone = (JobItem)base.Clone(cloner);
+      clone.latestSnapshotTime = this.latestSnapshotTime;
+      clone.jobDto = (JobDto)cloner.Clone(this.jobDto);
+      clone.latestSnapshot = (ResponseObject<SerializedJob>)cloner.Clone(this.latestSnapshot);
+      clone.log = (ILog)cloner.Clone(this.log);
+      return clone;
+    }
   }
 }

branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Experiment/3.3/JobItemList.cs

@@ -14 +14 @@
     public JobItemList(int capacity) : base(capacity) { }
     public JobItemList(IEnumerable<JobItem> collection) : base(collection) { }
+
+    public override Common.IDeepCloneable Clone(Common.Cloner cloner) {
+      JobItemList clone = (JobItemList)base.Clone(cloner);
+      return clone;
+    }
   }
 }

branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Experiment/3.3/Properties/AssemblyInfo.cs

@@ -55 +55 @@
 // You can specify all the values or you can default the Revision and Build Numbers 
 // by using the '*' as shown below:
-[assembly: AssemblyVersion("3.3.0.4120")]
-[assembly: AssemblyFileVersion("3.3.0.4120")]
-[assembly: AssemblyBuildDate("2010/07/30 13:39:10")]
+[assembly: AssemblyVersion("3.3.0.4121")]
+[assembly: AssemblyFileVersion("3.3.0.4121")]
+[assembly: AssemblyBuildDate("2010/08/02 17:23:46")]