Changeset 5153 for branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine/3.3

Timestamp:

12/21/10 18:16:24 (13 years ago)

Author:

cneumuel

Message:

#1260

• increased timeouts for sent jobs (which are needed if the jobs take long time to deserialize on slave)
• added DeleteJob to ClientService
• made optimizer actually Pause instead of Stop when stop is called explicitly (so they can be resumed later)
• temporarily disabled job-abortion from server because it aborted jobs which took too long to deserialize on slaves (this issue needs to be investigated)
• reduced locking of engines on slave so that the deserialization does not block heartbeats

#1347

• worked on HiveEngine
• added test project for HiveEngine

Location:

branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine/3.3

Files:

• HeuristicLab.HiveEngine-3.3.csproj (modified) (6 diffs)
• HeuristicLab.snk (added)
• HiveEngine.cs (modified) (7 diffs)
• Views (added)
• Views/HiveEngineView.Designer.cs (added)
• Views/HiveEngineView.cs (added)
• Views/HiveEngineView.resx (added)

branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine/3.3/HeuristicLab.HiveEngine-3.3.csproj

@@ -32 +32 @@
     <WarningLevel>4</WarningLevel>
   </PropertyGroup>
+  <PropertyGroup>
+    <SignAssembly>true</SignAssembly>
+  </PropertyGroup>
+  <PropertyGroup>
+    <AssemblyOriginatorKeyFile>HeuristicLab.snk</AssemblyOriginatorKeyFile>
+  </PropertyGroup>
   <ItemGroup>
     <Reference Include="HeuristicLab.Clients.Common-3.3">
@@ -48 +54 @@
       <HintPath>..\..\..\..\..\..\..\..\Program Files\HeuristicLab 3.3\HeuristicLab.Core-3.3.dll</HintPath>
     </Reference>
+    <Reference Include="HeuristicLab.Core.Views-3.3">
+      <HintPath>..\..\..\..\..\..\..\..\Program Files\HeuristicLab 3.3\HeuristicLab.Core.Views-3.3.dll</HintPath>
+    </Reference>
+    <Reference Include="HeuristicLab.MainForm-3.3">
+      <HintPath>..\..\..\..\..\..\..\..\Program Files\HeuristicLab 3.3\HeuristicLab.MainForm-3.3.dll</HintPath>
+    </Reference>
+    <Reference Include="HeuristicLab.MainForm.WindowsForms-3.3">
+      <HintPath>..\..\..\..\..\..\..\..\Program Files\HeuristicLab 3.3\HeuristicLab.MainForm.WindowsForms-3.3.dll</HintPath>
+    </Reference>
+    <Reference Include="HeuristicLab.Optimization.Views-3.3">
+      <HintPath>..\..\..\..\..\..\..\..\Program Files\HeuristicLab 3.3\HeuristicLab.Optimization.Views-3.3.dll</HintPath>
+    </Reference>
     <Reference Include="HeuristicLab.Persistence-3.3">
       <HintPath>..\..\..\..\..\..\..\..\Program Files\HeuristicLab 3.3\HeuristicLab.Persistence-3.3.dll</HintPath>
@@ -53 +71 @@
     <Reference Include="HeuristicLab.PluginInfrastructure-3.3">
       <HintPath>..\..\..\..\..\..\..\..\Program Files\HeuristicLab 3.3\HeuristicLab.PluginInfrastructure-3.3.dll</HintPath>
+    </Reference>
+    <Reference Include="HeuristicLab.Random-3.3">
+      <HintPath>..\..\..\..\..\..\..\..\Program Files\HeuristicLab 3.3\HeuristicLab.Random-3.3.dll</HintPath>
     </Reference>
     <Reference Include="HeuristicLab.SequentialEngine-3.3">
@@ -60 +81 @@
     <Reference Include="System.Core" />
     <Reference Include="System.Drawing" />
+    <Reference Include="System.Windows.Forms" />
     <Reference Include="System.Xml.Linq" />
     <Reference Include="System.Data.DataSetExtensions" />
@@ -67 +89 @@
   <ItemGroup>
     <Compile Include="ScopeMergeException.cs" />
+    <Compile Include="Views\HiveEngineView.cs">
+      <SubType>UserControl</SubType>
+    </Compile>
+    <Compile Include="Views\HiveEngineView.Designer.cs">
+      <DependentUpon>HiveEngineView.cs</DependentUpon>
+    </Compile>
+    <None Include="HeuristicLab.snk" />
     <None Include="Properties\AssemblyInfo.cs.frame" />
     <None Include="HeuristicLabHiveEnginePlugin.cs.frame" />
@@ -92 +121 @@
     </ProjectReference>
   </ItemGroup>
+  <ItemGroup>
+    <EmbeddedResource Include="Views\HiveEngineView.resx">
+      <DependentUpon>HiveEngineView.cs</DependentUpon>
+    </EmbeddedResource>
+  </ItemGroup>
   <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
   <PropertyGroup>

branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine/3.3/HiveEngine.cs

@@ -13 +13 @@
 using HeuristicLab.Hive.Contracts.ResponseObjects;
 using System.Threading;
+using HeuristicLab.Random;
 
 namespace HeuristicLab.HiveEngine {
@@ -19 +20 @@
   /// </summary>
   [StorableClass]
-  [Item("Hive Parallel Engine", "Engine for parallel execution on the hive.")]
+  [Item("Hive Engine", "Engine for parallel execution on the hive. You need enable `Parallel` for at least one operator in your operator graph to have all childoperations parallelized. Also those childoperations must not have sideeffects on a higher scope.")]
   public class HiveEngine : Engine {
+    [Storable]
     private IOperator currentOperator;
 
+    [Storable]
+    public string ResourceIds { get; set; }
+
     #region constructors and cloning
-    public HiveEngine() { }
+    public HiveEngine() {
+      ResourceIds = "HEAL";
+    }
     [StorableConstructor]
     protected HiveEngine(bool deserializing) : base(deserializing) { }
     protected HiveEngine(HiveEngine original, Cloner cloner)
       : base(original, cloner) {
+      this.ResourceIds = original.ResourceIds;
+      this.currentOperator = cloner.Clone(original.currentOperator);
     }
     public override IDeepCloneable Clone(Cloner cloner) {
@@ -42 +51 @@
       while (coll != null) {
         if (coll.Parallel) {
-          IScope scope = ((IAtomicOperation)coll.Where(x => x is IAtomicOperation).First()).Scope;
-          IEnumerable<IOperation> finishedOperations = ExecuteOnHive(coll.Select(op => new OperationJob(op)));
-          foreach (IOperation op in finishedOperations) {
-            ReIntegrateScope(scope, op);
+          IDictionary<IOperation, OperationJob> jobs = new Dictionary<IOperation, OperationJob>();
+          foreach (IOperation op in coll) {
+            jobs.Add(op, new OperationJob(op));
+          }
+
+          ExecuteOnHive(jobs);
+
+          foreach (var kvp in jobs) {
+            if (kvp.Key is IAtomicOperation) {
+              ReIntegrateScope(kvp.Value.Operation as IAtomicOperation, kvp.Key as IAtomicOperation);
+            } else if (kvp.Key is OperationCollection) {
+              // todo
+            }
           }
         } else {
@@ -73 +91 @@
     }
 
-    private static void ReIntegrateScope(IScope scope, IOperation operation) {
-      if (operation is IAtomicOperation) {
-        MergeScopes(scope, ((IAtomicOperation)operation).Scope);
-      } else if (operation is OperationCollection) {
-        foreach (IOperation op in (OperationCollection)operation) {
-          ReIntegrateScope(scope, op);
+    private IRandom FindRandomParameter(IExecutionContext ec) {
+      try {
+        if (ec == null)
+          return null;
+
+        foreach (var p in ec.Parameters) {
+          if (p.Name == "Random" && p is IValueParameter)
+            return ((IValueParameter)p).Value as IRandom;
         }
+        return FindRandomParameter(ec.Parent);
       }
+      catch { return null; }
     }
 
-    private static void MergeScopes(IScope target, IScope source) {
-      foreach (IVariable variable in source.Variables) {
-        if (target.Variables.ContainsKey(variable.Name)) {
-          if (target.Variables[variable.Name] != source.Variables[variable.Name]) {
-            // problem: ResultCollection are not equal altough they contain the same values.
-            //             - either implement ResultCollection.Equals, or consider ResultCollection as special case here
-            //throw new ScopeMergeException(string.Format("Variable {0} already exists in target scope and has a different value ({1}, {2}).", variable.Name, target.Variables[variable.Name].Value, variable.Value));
-          }
-        } else {
-          target.Variables.Add(variable);
-        }
-      }
-
-      //this is not useful on the parentscope of the initial scope
-      //foreach (IScope subScope in source.SubScopes) {
-      //  target.SubScopes.Add(subScope);
-      //}
-
-      if (target.Parent != null && source.Parent != null)
-        MergeScopes(target.Parent, source.Parent);
+    private static void ReIntegrateScope(IAtomicOperation source, IAtomicOperation target) {
+      ExchangeScope(source.Scope, target.Scope);
     }
 
-
-    private IEnumerable<IOperation> TestExecuteOnHive(IEnumerable<OperationJob> operationJobs) {
-      //var clonedJobs = operationJobs.Select(x => (OperationJob)x.Clone()).ToArray();
-      var results = new List<IOperation>();
-
-      foreach (var job in operationJobs) {
-        job.Start();
-        while (job.ExecutionState != Core.ExecutionState.Stopped) {
-          Thread.Sleep(100);
-        }
-        results.Add(job.Operation);
-      }
-      return results;
+    private static void ExchangeScope(IScope source, IScope target) {
+      target.Variables.Clear();
+      target.Variables.AddRange(source.Variables);
+      target.SubScopes.Clear();
+      target.SubScopes.AddRange(source.SubScopes);
+      // TODO: validate if parent scopes match - otherwise source is invalid
     }
 
-    private IEnumerable<IOperation> ExecuteOnHive(IEnumerable<OperationJob> operationJobs) {
-      IEnumerable<Guid> jobIds;
+    /// <summary>
+    /// This method blocks until all jobs are finished
+    /// </summary>
+    /// <param name="jobDict"></param>
+    private void ExecuteOnHive(IDictionary<IOperation, OperationJob> jobDict) {
+      Log.LogMessage(string.Format("Executing {0} operations on the hive.", jobDict.Count));
+      IDictionary<Guid, IOperation> jobIds = new Dictionary<Guid, IOperation>();
       JobResultList results;
-
 
       using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
         List<JobDto> jobs = new List<JobDto>();
-        foreach (OperationJob operationJob in operationJobs) {
-          var groups = new string[] { "HEAL" };
+        foreach (var kvp in jobDict) {
+          // shuffle random variable to avoid the same random sequence in each operation
+          IRandom random = FindRandomParameter(kvp.Key as IExecutionContext);
+          if (random != null)
+            random.Reset(random.Next());
+
+          var groups = ResourceIds.Split(';');
           SerializedJob serializedJob = new SerializedJob();
-          serializedJob.SerializedJobData = SerializedJob.Serialize(operationJob);
+          serializedJob.SerializedJobData = SerializedJob.Serialize(kvp.Value);
           serializedJob.JobInfo = new JobDto();
           serializedJob.JobInfo.State = JobState.Offline;
@@ -137 +143 @@
           ResponseObject<JobDto> response = service.Obj.AddJobWithGroupStrings(serializedJob, groups);
           jobs.Add(response.Obj);
+          jobIds.Add(response.Obj.Id, kvp.Key);
         }
-        jobIds = jobs.Select(x => x.Id);
-        results = service.Obj.GetJobResults(jobIds).Obj;
+        results = service.Obj.GetJobResults(jobIds.Keys).Obj;
       }
 
@@ -148 +154 @@
         Thread.Sleep(5000);
         using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
-          results = service.Obj.GetJobResults(jobIds).Obj;
+          results = service.Obj.GetJobResults(jobIds.Keys).Obj;
         }
       }
 
       // all finished
-      List<OperationJob> finishedJobs = new List<OperationJob>();
       using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
-        foreach (Guid jobId in jobIds) {
+        foreach (Guid jobId in jobIds.Keys) {
           SerializedJob serializedJob = service.Obj.GetLastSerializedResult(jobId).Obj;
           OperationJob operationJob = SerializedJob.Deserialize<OperationJob>(serializedJob.SerializedJobData);
-          finishedJobs.Add(operationJob);
+          jobDict[jobIds[jobId]] = operationJob;
         }
       }
 
       // delete jobs
-      // TODO
+      using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
+        foreach (Guid jobId in jobIds.Keys) {
+          service.Obj.DeleteJob(jobId);
+        }
+      }
 
-      return finishedJobs.Select(x => x.Operation);
+      Log.LogMessage(string.Format("Operations on the hive finished.", jobDict.Count));
     }
 
@@ -176 +185 @@
       if (currentOperator != null) currentOperator.Abort();
     }
+
   }
 }