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

Timestamp:

06/08/11 11:07:33 (13 years ago)

Author:

svonolfe

Message:

Worked on the MPIEngine (#1542)

File:

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

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

@@ -31 +31 @@
 using HeuristicLab.Optimization;
 using System.Linq;
+using Microsoft.Hpc.Scheduler;
+using System.ServiceModel;
+using HeuristicLab.MPIAlgorithmRunner;
+using HeuristicLab.Operators.MPISupport;
 
 namespace HeuristicLab.MPIEngine {
@@ -49 +53 @@
     }
 
-    private string algFile;
+    private IAlgorithm algorithm;
 
     public override void Start() {
@@ -64 +68 @@
         alg.Engine = new SequentialEngine.SequentialEngine();
 
-        algFile = Path.GetTempFileName();
-        XmlGenerator.Serialize(alg, algFile);
+        algorithm = alg;
       }
 
@@ -77 +80 @@
         IScope globalScope = context.Scope;
 
-        string resultFile = Path.GetTempFileName();
-        ItemList<ResultCollection> empty = new ItemList<ResultCollection>();
-        XmlGenerator.Serialize(empty, resultFile);
+        string exec = @"mpiexec";
+        string args = @"-c 3 C:\public\MPISupport\HeuristicLab.MPIAlgorithmRunner-3.3.exe ";
 
-        string exec = @"C:\Program Files\Microsoft Compute Cluster Pack\Bin\mpiexec.exe";
-        string args = @"-n 3 HeuristicLab.MPIAlgorithmRunner-3.3.exe " + algFile + " " + resultFile + " 5000";
+        IScheduler scheduler = new Scheduler();
+        scheduler.Connect("blade00.hpc.fh-hagenberg.at");
 
-        System.Threading.Thread pollThread = new System.Threading.Thread(delegate(object state) {
-          while (true) {
-            System.Threading.Thread.Sleep(5000);
+        ISchedulerJob job = scheduler.CreateJob();
+        job.Name = "HeuristicLab.MPIEngine";
+        job.RequestedNodes.Add("BLADE03");
+        ISchedulerTask task = job.CreateTask();
+        task.Name = "HeuristicLab.MPIAlgorithmRunner";
+        task.CommandLine = exec + " " + args;
+        task.StdOutFilePath = "stdout.txt";
+        task.StdErrFilePath = "stderr.txt";
+        task.MinimumNumberOfCores = task.MaximumNumberOfCores = 3;
+        job.AddTask(task);
 
-            lock (resultFile) {
-              object results = XmlParser.Deserialize(resultFile);
-              ResultCollection resultCollection = (globalScope.Variables["Results"].Value as ResultCollection);
+        scheduler.SubmitJob(job, @"HPC\svonolfe", @"Vunkopf!3");
 
-              if (resultCollection != null) {
-                if (!resultCollection.ContainsKey("MPIResults"))
-                  resultCollection.Add(new Result("MPIResults", results as IItem));
+        try {
+          string address = null;
+          int timeout = 10;
 
-                resultCollection["MPIResults"].Value = results as IItem;
+          while (address == null && timeout > 0) {
+            ISchedulerJob schedulerJob = scheduler.OpenJob(job.Id);
+            if (schedulerJob != null) {
+              INameValueCollection properties = schedulerJob.GetCustomProperties();
+              NameValue item = properties.FirstOrDefault(i => i.Name == "address");
+              if (item != null) {
+                address = item.Value;
+              } else {
+                System.Threading.Thread.Sleep(1000);
+                timeout--;
               }
-
             }
           }
-        });
-        pollThread.Start();
 
-        Process p = Process.Start(exec, args);
-        p.WaitForExit();
+          if (address == null) {
+            throw new Exception("A timeout occurred when starting the MPIAlgorithmRunner");
+          }
 
-        pollThread.Abort();
-        File.Delete(algFile);
-        File.Delete(resultFile);
+          NetTcpBinding netTCPBinding = new NetTcpBinding(SecurityMode.None);
+          ChannelFactory<IAlgorithmBroker> factory = new ChannelFactory<IAlgorithmBroker>(netTCPBinding, address);
+          IAlgorithmBroker proxy = factory.CreateChannel();
+
+          proxy.TransmitAlgorithm(new MPITransportWrapper<IAlgorithm>(algorithm));
+
+          while (!proxy.IsAlgorithmTerminated()) {
+            ItemList<ResultCollection> results = proxy.GetResults().InnerItem;
+
+            ResultCollection resultCollection = (globalScope.Variables["Results"].Value as ResultCollection);
+
+            if (resultCollection != null && results != null) {
+              if (!resultCollection.ContainsKey("MPIResults"))
+                resultCollection.Add(new Result("MPIResults", results));
+
+              resultCollection["MPIResults"].Value = results;
+            }
+
+            System.Threading.Thread.Sleep(5000);
+          }
+        }
+        catch (Exception e) {
+          scheduler.CancelJob(job.Id, "Exception: " + e.Message);
+          throw e;
+        }
       }
     }