Changeset 6354 for branches/MPI/HeuristicLab.MPIEngine/3.3/MPIEngine.cs

Timestamp:

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

Author:

svonolfe

Message:

Added first version of the MPIEngine (#1542)

File:

• branches/MPI/HeuristicLab.MPIEngine/3.3/MPIEngine.cs (modified) (3 diffs)

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);
       }
     }