Changeset 7544 for branches/MPI/HeuristicLab.MPIAlgorithmRunner/3.3/Program.cs

Timestamp:

03/05/12 10:18:06 (12 years ago)

Author:

svonolfe

Message:

Improved performance, added MPISolutionsCreator (#1542)

File:

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

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

@@ -27 +27 @@
 using HeuristicLab.Persistence.Default.Xml;
 using System.Threading;
+using Microsoft.Hpc.Scheduler;
+using System.Net;
+using System.ServiceModel;
+using HeuristicLab.Operators.MPISupport;
+using HeuristicLab.Core;
+using System.Diagnostics;
 
 namespace HeuristicLab.MPIAlgorithmRunner {
@@ -33 +39 @@
 
     static void Main(string[] args) {
-      if (args.Length != 2) {
-        Console.WriteLine("Args:" + args.Length);
-
-        throw new ArgumentException("You must provide two arguments - the algorithm file and the result file");
+      if (args.Length == 2) {
+        string fileName = args[0];
+        string resultName = args[1];
+
+        using (new MPI.Environment(ref args)) {
+          if (MPI.Communicator.world.Rank != 0) {
+            Program p = new Program();
+            p.StartAlgorithm(fileName, resultName + MPI.Communicator.world.Rank + ".hl");
+          }
+        }
+      } else {
+        using (new MPI.Environment(ref args)) {
+          int clients = MPI.Communicator.world.Group.Size - 1;
+          Console.WriteLine("Clients: " + clients);
+
+          MPI.Communicator communicator = MPI.Communicator.world.Clone() as MPI.Communicator;
+
+          if (communicator.Rank == 0) {
+            ServiceHost service = AlgorithmBroker.StartService(communicator);
+            AlgorithmBroker broker = (service.SingletonInstance as AlgorithmBroker);
+
+            string address = GetAddress(service);
+            SetJobStatus(address);
+
+            bool[] finished = new bool[clients];
+            int finishedCount = 0;
+
+            while (finishedCount != clients) {
+              ItemList<ResultCollection> results = new ItemList<ResultCollection>();
+
+              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 (broker.resultLocker)
+                broker.Results = results;
+            }
+
+            lock (broker.ExitWaitHandle) {
+              broker.Terminated = true;
+            }
+
+            broker.ExitWaitHandle.WaitOne();
+            Console.WriteLine("Finished.");
+            Thread.Sleep(1000);
+            service.Close();
+          } else {
+              Program p = new Program();
+              p.StartAlgorithm(communicator);
+          }
+        }
+      }       
+    }
+
+    private static void SetJobStatus(string address) {
+      Console.WriteLine("Started service... at " + address);
+
+      // Discover the job's context from the environment
+      String headNodeName = System.Environment.GetEnvironmentVariable("CCP_SCHEDULER");
+      int jobId = System.Convert.ToInt32(System.Environment.GetEnvironmentVariable("CCP_JOBID"));
+
+      Console.WriteLine(jobId + "@" + headNodeName);
+
+      // Connect to the head node and get the job
+      IScheduler scheduler = new Scheduler();
+      scheduler.Connect(headNodeName);
+      ISchedulerJob job = scheduler.OpenJob(jobId);
+
+      job.SetCustomProperty("address", address);
+
+      job.Commit();
+    }
+
+    private static string GetAddress(ServiceHost service) {
+      IPHostEntry IPHost = Dns.GetHostByName(Dns.GetHostName());
+      string address = "net.tcp://" + IPHost.AddressList[0].ToString() + ":" + service.ChannelDispatchers[0].Listener.Uri.Port + "/AlgorithmBroker";
+
+      return address;
+    }
+
+    public void StartAlgorithm(MPI.Communicator communicator) {
+      IAlgorithm alg = communicator.Receive<MPITransportWrapper<IAlgorithm>>(0, 0).InnerItem;
+      int updateInterval = communicator.Receive<int>(0, 1);
+
+      Console.WriteLine("Starting algorithm...");
+
+      alg.Prepare(true);
+      waitHandle = new EventWaitHandle(false, EventResetMode.ManualReset);
+      alg.Started += new EventHandler(alg_Started);
+      alg.Start();
+
+      waitHandle.WaitOne();
+
+      alg.Started -= new EventHandler(alg_Started);
+
+      while (alg.ExecutionState != ExecutionState.Stopped) {
+        Console.WriteLine("Pausing alg...");
+        alg.Pause();
+
+        while (alg.ExecutionState == ExecutionState.Started) {
+          Thread.Sleep(100);
+        }
+
+        communicator.Send<MPITransportWrapper<ResultCollection>>(
+          new MPITransportWrapper<ResultCollection>(alg.Results), 0, 1);
+
+        Console.WriteLine("Sending update...");
+
+        Console.WriteLine("Resuming alg...");
+        if (alg.ExecutionState == ExecutionState.Paused)
+          alg.Start();
+        Thread.Sleep(updateInterval);
       }
 
-      string fileName = args[0];
-      string resultName = args[1];
-
-      using (new MPI.Environment(ref args)) {
-        Program p = new Program();
-        p.StartAlgorithm(fileName, resultName + MPI.Communicator.world.Rank + ".hl");
-      }
+      communicator.Send<int>(communicator.Rank, 0, 2);
+      communicator.Send<MPITransportWrapper<ResultCollection>>(
+          new MPITransportWrapper<ResultCollection>(alg.Results), 0, 1);
+    }
+
+    void alg_Started(object sender, EventArgs e) {
+      waitHandle.Set();
     }
 
@@ -56 +188 @@
       alg.Prepare(true);
       alg.Start();
-
-      Thread.Sleep(10000);      
+      Console.WriteLine("ALG STARTED");
+      Stopwatch sw = new Stopwatch();
+      sw.Start();
+
+     /* Thread.Sleep(10000);      
       Thread t = new Thread(delegate() {
         while (true) {
@@ -82 +217 @@
       });
       t.IsBackground = true;
-      t.Start();
+      t.Start();*/
 
       waitHandle.WaitOne();
 
+      Console.WriteLine("ALG STOPPED");
+      sw.Stop();
+      Console.WriteLine("TIME: " + sw.ElapsedMilliseconds + "ms");
+
+      if (alg.Results.ContainsKey("Best valid Solution Distance") && alg.Results.ContainsKey("Best valid Solution Vehicles")) {
+        Console.WriteLine("BestDistance: " + alg.Results["Best valid Solution Distance"].Value.ToString() + ", " + 
+                          "BestVehicles: " + alg.Results["Best valid Solution Vehicles"].Value.ToString());
+      }
+
       XmlGenerator.Serialize(alg, resultName);
     }
 
     void algorithm_Stopped(object sender, EventArgs e) {
-      waitHandle.Set();
+      waitHandle.Set(); 
     }
   }