Changeset 6354 for branches/MPI

Timestamp:

06/01/11 17:37:05 (13 years ago)

Author:

svonolfe

Message:

Added first version of the MPIEngine (#1542)

Location:

branches/MPI

Files:

• HeuristicLab.ExtLibs/HeuristicLab.MPInet/MPIAlgorithmRunner/3.3/HeuristicLab.MPIAlgorithmRunner-3.3.csproj (modified) (2 diffs)
• HeuristicLab.ExtLibs/HeuristicLab.MPInet/MPIAlgorithmRunner/3.3/HeuristicLabMPIAlgorithmRunnerPlugin.cs.frame (modified) (1 diff)
• HeuristicLab.ExtLibs/HeuristicLab.MPInet/MPIAlgorithmRunner/3.3/Program.cs (modified) (3 diffs)
• HeuristicLab.MPIEngine/3.3/HeuristicLab.MPIEngine-3.3.csproj (modified) (3 diffs)
• HeuristicLab.MPIEngine/3.3/MPIEngine.cs (modified) (3 diffs)
• HeuristicLab.Operators.MPISupport/3.3/MPITransportWrapper.cs (modified) (1 diff)
• HeuristicLab.Operators.MPISupport/3.3/MPIUnidirectionalRingMigrator.cs (modified) (2 diffs)

branches/MPI/HeuristicLab.ExtLibs/HeuristicLab.MPInet/MPIAlgorithmRunner/3.3/HeuristicLab.MPIAlgorithmRunner-3.3.csproj

@@ -120 +120 @@
   </ItemGroup>
   <ItemGroup>
+    <ProjectReference Include="..\..\..\..\HeuristicLab.Collections\3.3\HeuristicLab.Collections-3.3.csproj">
+      <Project>{958B43BC-CC5C-4FA2-8628-2B3B01D890B6}</Project>
+      <Name>HeuristicLab.Collections-3.3</Name>
+    </ProjectReference>
     <ProjectReference Include="..\..\..\..\HeuristicLab.Common\3.3\HeuristicLab.Common-3.3.csproj">
       <Project>{A9AD58B9-3EF9-4CC1-97E5-8D909039FF5C}</Project>
@@ -127 +131 @@
       <Project>{C36BD924-A541-4A00-AFA8-41701378DDC5}</Project>
       <Name>HeuristicLab.Core-3.3</Name>
+    </ProjectReference>
+    <ProjectReference Include="..\..\..\..\HeuristicLab.Operators.MPISupport\3.3\HeuristicLab.Operators.MPISupport-3.3.csproj">
+      <Project>{6BD69CDA-4875-4045-8B35-6FD4602854F5}</Project>
+      <Name>HeuristicLab.Operators.MPISupport-3.3</Name>
     </ProjectReference>
     <ProjectReference Include="..\..\..\..\HeuristicLab.PluginInfrastructure\3.3\HeuristicLab.PluginInfrastructure-3.3.csproj">

branches/MPI/HeuristicLab.ExtLibs/HeuristicLab.MPInet/MPIAlgorithmRunner/3.3/HeuristicLabMPIAlgorithmRunnerPlugin.cs.frame

@@ -30 +30 @@
   [PluginDependency("HeuristicLab.Common", "3.3")]
   [PluginDependency("HeuristicLab.Core", "3.3")]
+  [PluginDependency("HeuristicLab.Collections", "3.3")]
   [PluginDependency("HeuristicLab.MPInet", "1.0.0")]
   [PluginDependency("HeuristicLab.Optimization", "3.3")]
   [PluginDependency("HeuristicLab.Persistence", "3.3")]
+  [PluginDependency("HeuristicLab.Operators.MPISupport", "3.3")]
   public class HeuristicLabProblemsVehicleRoutingPlugin : PluginBase {
   }

branches/MPI/HeuristicLab.ExtLibs/HeuristicLab.MPInet/MPIAlgorithmRunner/3.3/Program.cs

@@ -27 +27 @@
 using HeuristicLab.Persistence.Default.Xml;
 using System.Threading;
+using HeuristicLab.Operators.MPISupport;
+using HeuristicLab.Core;
 
 namespace HeuristicLab.MPIAlgorithmRunner {
@@ -33 +35 @@
 
     static void Main(string[] args) {
-      if (args.Length != 2) {
+      if (args.Length != 3) {
         Console.WriteLine("Args:" + args.Length);
 
-        throw new ArgumentException("You must provide two arguments - the algorithm file and the result file");
+        throw new ArgumentException("You must provide 3 arguments - the algorithm file, the result file and the update interval (in ms)");
       }
 
       string fileName = args[0];
       string resultName = args[1];
+      int updateInterval = int.Parse(args[2]);
 
       using (new MPI.Environment(ref args)) {
-        Program p = new Program();
-        p.StartAlgorithm(fileName, resultName + MPI.Communicator.world.Rank + ".hl");
+        MPI.Communicator communicator = MPI.Communicator.world.Clone() as MPI.Communicator;
+
+        if (communicator.Rank == 0) {
+          IAlgorithm alg = XmlParser.Deserialize<IAlgorithm>(fileName);
+          MPITransportWrapper<IAlgorithm> transport = new MPITransportWrapper<IAlgorithm>(alg);
+
+          int clients = communicator.Size - 1;
+          Console.WriteLine("Sending alg to " + clients + " clients");
+          for (int i = 0; i < clients; i++) {
+            int client = i + 1;
+            communicator.Send(transport, client, 0);
+          }
+
+          ItemList<ResultCollection> results = new ItemList<ResultCollection>();
+          
+          bool[] finished = new bool[clients];
+          int finishedCount = 0;
+
+          while (finishedCount != clients) {
+            for (int i = 0; i < clients; i++) {
+              if (!finished[i]) {
+                int client = i + 1;
+                ResultCollection result = communicator.Receive<MPITransportWrapper<ResultCollection>>(client, 1).InnerItem;
+
+                Console.WriteLine("Received result " + result);
+
+                if (results.Count != clients) {
+                  results.Add(result);
+                } else {
+                  results[i] = result;
+                }
+
+                Console.WriteLine("Probing...");
+                if (communicator.ImmediateProbe(client, 2) != null) {
+                  finished[i] = true;
+                  finishedCount++;
+                }
+              }
+            }
+
+            Console.WriteLine("Update results");
+            lock (resultName) {
+              XmlGenerator.Serialize(results, resultName);
+            }
+          }
+
+          Console.WriteLine("Finished.");
+        } else {
+          Program p = new Program();
+          p.StartAlgorithm(updateInterval, communicator);
+        }
       }
     }
 
-    public void StartAlgorithm(string fileName, string resultName) {
-      IAlgorithm alg = XmlParser.Deserialize<HeuristicLab.Optimization.IAlgorithm>(fileName);
+    public void StartAlgorithm(int updateInterval, MPI.Communicator communicator) {
+      IAlgorithm alg = communicator.Receive<MPITransportWrapper<IAlgorithm>>(0, 0).InnerItem;
+
+      Console.WriteLine("Starting algorithm...");
 
       alg.Stopped += new EventHandler(algorithm_Stopped);
@@ -57 +111 @@
       alg.Start();
 
+      Timer t = new Timer(delegate(object state) {
+        if (alg.ExecutionState == ExecutionState.Started) {
+          Console.WriteLine("Pausing alg...");
+          alg.Pause();
+
+          while (alg.ExecutionState != ExecutionState.Paused) {
+            Thread.Sleep(100);
+          }
+
+          communicator.Send<MPITransportWrapper<ResultCollection>>(
+            new MPITransportWrapper<ResultCollection>(alg.Results), 0, 1);
+
+          Console.WriteLine("Sending update...");
+
+          Console.WriteLine("Resuming alg...");
+          alg.Start();
+        }
+      }, null, updateInterval, updateInterval);
+
       waitHandle.WaitOne();
 
-      XmlGenerator.Serialize(alg, resultName);
+      communicator.Send<int>(communicator.Rank, 0, 2);
+      communicator.Send<MPITransportWrapper<ResultCollection>>(
+          new MPITransportWrapper<ResultCollection>(alg.Results), 0, 1);
     }
 

branches/MPI/HeuristicLab.MPIEngine/3.3/HeuristicLab.MPIEngine-3.3.csproj

@@ -122 +122 @@
   </ItemGroup>
   <ItemGroup>
+    <ProjectReference Include="..\..\HeuristicLab.Collections\3.3\HeuristicLab.Collections-3.3.csproj">
+      <Project>{958B43BC-CC5C-4FA2-8628-2B3B01D890B6}</Project>
+      <Name>HeuristicLab.Collections-3.3</Name>
+    </ProjectReference>
     <ProjectReference Include="..\..\HeuristicLab.Common\3.3\HeuristicLab.Common-3.3.csproj">
       <Project>{A9AD58B9-3EF9-4CC1-97E5-8D909039FF5C}</Project>
@@ -130 +134 @@
       <Name>HeuristicLab.Core-3.3</Name>
     </ProjectReference>
+    <ProjectReference Include="..\..\HeuristicLab.Optimization\3.3\HeuristicLab.Optimization-3.3.csproj">
+      <Project>{14AB8D24-25BC-400C-A846-4627AA945192}</Project>
+      <Name>HeuristicLab.Optimization-3.3</Name>
+    </ProjectReference>
     <ProjectReference Include="..\..\HeuristicLab.Persistence\3.3\HeuristicLab.Persistence-3.3.csproj">
       <Project>{102BC7D3-0EF9-439C-8F6D-96FF0FDB8E1B}</Project>
@@ -137 +145 @@
       <Project>{94186A6A-5176-4402-AE83-886557B53CCA}</Project>
       <Name>HeuristicLab.PluginInfrastructure-3.3</Name>
+    </ProjectReference>
+    <ProjectReference Include="..\..\HeuristicLab.SequentialEngine\3.3\HeuristicLab.SequentialEngine-3.3.csproj">
+      <Project>{DC3D7072-7999-4719-B65D-3997744D5DC1}</Project>
+      <Name>HeuristicLab.SequentialEngine-3.3</Name>
     </ProjectReference>
   </ItemGroup>

branches/MPI/HeuristicLab.MPIEngine/3.3/MPIEngine.cs

@@ -22 +22 @@
 using System;
 using System.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
 using HeuristicLab.Common;
 using HeuristicLab.Core;
 using HeuristicLab.Persistence.Default.CompositeSerializers.Storable;
+using System.Reflection;
+using System.IO;
+using HeuristicLab.Persistence.Default.Xml;
+using System.Diagnostics;
+using HeuristicLab.Optimization;
+using System.Linq;
 
 namespace HeuristicLab.MPIEngine {
@@ -36 +40 @@
   [Item("MPI Engine", "Engine for parallel execution of algorithms using multiple processes (suitable for distributed memory systems with multiple cores).")]
   public class MPIEngine : Engine {
-    private CancellationToken cancellationToken;
-
     [StorableConstructor]
     protected MPIEngine(bool deserializing) : base(deserializing) { }
@@ -47 +49 @@
     }
 
-    protected override void Run(CancellationToken cancellationToken) {
-      this.cancellationToken = cancellationToken;
-      Run(ExecutionStack);
+    private string algFile;
+
+    public override void Start() {
+      if (ExecutionStack.Count == 1) {
+        ExecutionContext context = ExecutionStack.First() as ExecutionContext;
+
+        ExecutionContext algorithmContext = context.Parent as ExecutionContext;
+
+        EngineAlgorithm alg = typeof(ExecutionContext).InvokeMember("parameterizedItem",
+          BindingFlags.GetField | BindingFlags.NonPublic |
+          BindingFlags.Instance, null, algorithmContext, null) as EngineAlgorithm;
+
+        alg = alg.Clone() as EngineAlgorithm;
+        alg.Engine = new SequentialEngine.SequentialEngine();
+
+        algFile = Path.GetTempFileName();
+        XmlGenerator.Serialize(alg, algFile);
+      }
+
+      base.Start();
     }
 
-    private void Run(object state) {
-      Stack<IOperation> executionStack = (Stack<IOperation>)state;
-      IOperation next;
-      OperationCollection coll;
-      IAtomicOperation operation;
+    protected override void Run(System.Threading.CancellationToken cancellationToken) {
+      if (ExecutionStack.Count == 1) {
+        ExecutionContext context = ExecutionStack.Pop() as ExecutionContext;
 
-      while (executionStack.Count > 0) {
-        cancellationToken.ThrowIfCancellationRequested();
+        IScope globalScope = context.Scope;
 
-        next = executionStack.Pop();
-        if (next is OperationCollection) {
-          coll = (OperationCollection)next;
-          if (coll.Parallel) {
-            Task[] tasks = new Task[coll.Count];
-            Stack<IOperation>[] stacks = new Stack<IOperation>[coll.Count];
-            for (int i = 0; i < coll.Count; i++) {
-              stacks[i] = new Stack<IOperation>();
-              stacks[i].Push(coll[i]);
-              tasks[i] = Task.Factory.StartNew(Run, stacks[i], cancellationToken);
+        string resultFile = Path.GetTempFileName();
+        ItemList<ResultCollection> empty = new ItemList<ResultCollection>();
+        XmlGenerator.Serialize(empty, resultFile);
+
+        string exec = @"C:\Program Files\Microsoft Compute Cluster Pack\Bin\mpiexec.exe";
+        string args = @"-n 3 HeuristicLab.MPIAlgorithmRunner-3.3.exe " + algFile + " " + resultFile + " 5000";
+
+        System.Threading.Thread pollThread = new System.Threading.Thread(delegate(object state) {
+          while (true) {
+            System.Threading.Thread.Sleep(5000);
+
+            lock (resultFile) {
+              object results = XmlParser.Deserialize(resultFile);
+              ResultCollection resultCollection = (globalScope.Variables["Results"].Value as ResultCollection);
+
+              if (resultCollection != null) {
+                if (!resultCollection.ContainsKey("MPIResults"))
+                  resultCollection.Add(new Result("MPIResults", results as IItem));
+
+                resultCollection["MPIResults"].Value = results as IItem;
+              }
+
             }
-            try {
-              Task.WaitAll(tasks);
-            }
-            catch (AggregateException ex) {
-              OperationCollection remaining = new OperationCollection() { Parallel = true };
-              for (int i = 0; i < stacks.Length; i++) {
-                if (stacks[i].Count == 1)
-                  remaining.Add(stacks[i].Pop());
-                if (stacks[i].Count > 1) {
-                  OperationCollection ops = new OperationCollection();
-                  while (stacks[i].Count > 0)
-                    ops.Add(stacks[i].Pop());
-                  remaining.Add(ops);
-                }
-              }
-              if (remaining.Count > 0) executionStack.Push(remaining);
-              throw ex;
-            }
-          } else {
-            for (int i = coll.Count - 1; i >= 0; i--)
-              if (coll[i] != null) executionStack.Push(coll[i]);
           }
-        } else if (next is IAtomicOperation) {
-          operation = (IAtomicOperation)next;
-          try {
-            next = operation.Operator.Execute((IExecutionContext)operation, cancellationToken);
-          }
-          catch (Exception ex) {
-            executionStack.Push(operation);
-            if (ex is OperationCanceledException) throw ex;
-            else throw new OperatorExecutionException(operation.Operator, ex);
-          }
-          if (next != null) executionStack.Push(next);
+        });
+        pollThread.Start();
 
-          if (operation.Operator.Breakpoint) {
-            string message = string.Format("Breakpoint: {0}", operation.Operator.Name != string.Empty ? operation.Operator.Name : operation.Operator.ItemName);
-            Log.LogMessage(message);
-            throw new OperationCanceledException(message);
-          }
-        }
+        Process p = Process.Start(exec, args);
+        p.WaitForExit();
+
+        pollThread.Abort();
+        File.Delete(algFile);
+        File.Delete(resultFile);
       }
     }

branches/MPI/HeuristicLab.Operators.MPISupport/3.3/MPITransportWrapper.cs

@@ -34 +34 @@
   /// </summary>
   [Serializable]
-  class MPITransportWrapper<T>: ISerializable where T: class, IItem {
+  public class MPITransportWrapper<T>: ISerializable where T: class, IItem {
     [NonSerialized]
     private T innerItem;

branches/MPI/HeuristicLab.Operators.MPISupport/3.3/MPIUnidirectionalRingMigrator.cs

@@ -43 +43 @@
     }
 
+    public static int Channel = 100;
+
     public override IOperation Apply() {
       if (MPI.Communicator.world != null) {
-        int size = MPI.Communicator.world.Group.Size;
-        if (size > 1) {
+        int size = MPI.Communicator.world.Size;
+        if (size > 2) {
           int i = MPI.Communicator.world.Rank;
           IScope scope = ExecutionContext.Scope;
@@ -52 +54 @@
           IScope emigrants = scope.SubScopes[1];
           scope.SubScopes.Remove(emigrants);
-          int recipent = (i + 1) % size;
+          int recipent = i + 1;
+          if (recipent == size)
+            recipent = 1;
 
           Console.WriteLine("MIGRATE " + i + " TO " + recipent);
           MPI.Communicator.world.Send<MPITransportWrapper<IScope>>(
-            new MPITransportWrapper<IScope>(emigrants), recipent, 0);
+            new MPITransportWrapper<IScope>(emigrants), recipent, Channel);
 
           IScope immigrants = null;
           int sender = i - 1;
-          if (sender < 0)
+          if (sender < 1)
             sender = size - 1;
 
           Console.WriteLine("MIGRATE " + i + " FROM " + sender);
-          immigrants = MPI.Communicator.world.Receive<MPITransportWrapper<IScope>>(sender, 0).InnerItem;
+          immigrants = MPI.Communicator.world.Receive<MPITransportWrapper<IScope>>(sender, Channel).InnerItem;
 
           scope.SubScopes.Add(immigrants);