Changeset 6401

Timestamp:

06/09/11 12:52:48 (14 years ago)

Author:

svonolfe

Message:

Implemented result and algorithm streaming (#1542)

Location:

branches/MPI

Files:

• HeuristicLab.MPIAlgorithmRunner/3.3/AlgorithmBroker.cs (modified) (3 diffs)
• HeuristicLab.MPIAlgorithmRunner/3.3/IAlgorithmBroker.cs (modified) (2 diffs)
• HeuristicLab.MPIAlgorithmRunner/3.3/Program.cs (modified) (3 diffs)
• HeuristicLab.MPIEngine/3.3/MPIEngine.cs (modified) (3 diffs)
• HeuristicLab.Operators.MPISupport/3.3/MPITransportWrapper.cs (modified) (1 diff)

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

@@ -33 +33 @@
 using System.Threading;
 using System.Xml;
+using System.IO;
+using HeuristicLab.Persistence.Default.Xml;
                      
 namespace HeuristicLab.MPIAlgorithmRunner {
+  public class StreamingHelper {
+    public static IItem StreamItem(Stream stream) {
+      using (MemoryStream memoryStream = new MemoryStream()) {
+        byte[] buffer = new byte[256];
+        int read;
+        while ((read = stream.Read(buffer, 0, buffer.Length)) != 0)
+          memoryStream.Write(buffer, 0, read);
+
+        using (MemoryStream itemStream = new MemoryStream(memoryStream.GetBuffer())) {
+          IItem result = XmlParser.Deserialize(itemStream) as IItem;
+          return result;
+        }
+      }
+    }
+  }
+
   [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single)]
   class AlgorithmBroker: IAlgorithmBroker {
@@ -54 +72 @@
       endpoint.Name = "AlgorithmBrokerEndpoint";
       NetTcpBinding netTCPBinding = new NetTcpBinding(SecurityMode.None);
+      netTCPBinding.TransferMode = TransferMode.Streamed;
       netTCPBinding.MaxReceivedMessageSize = int.MaxValue;
-      XmlDictionaryReaderQuotas quotas = new XmlDictionaryReaderQuotas();
-      quotas.MaxArrayLength = int.MaxValue;
-      netTCPBinding.ReaderQuotas = quotas;
-      netTCPBinding.SendTimeout = new TimeSpan(600000000);
-      netTCPBinding.ReceiveTimeout = new TimeSpan(600000000);
       endpoint.Binding = netTCPBinding;
       int port = FindFreeTcpPort();
@@ -92 +106 @@
     #region IAlgorithmBroker Members
 
-    public void TransmitAlgorithm(MPITransportWrapper<IAlgorithm> algorithm, int updateInterval) {
-      Console.WriteLine("Transmitting Algorithm...");
+    public void SetUpdateInterval(int updateInterval) {
+      Console.WriteLine("Set update interval...");
+
       int clients = MPI.Communicator.world.Group.Size - 1;
-
-      for(int i = 0; i < clients; i++) {
+      for (int i = 0; i < clients; i++) {
         int client = i + 1;
-        communicator.Send<MPITransportWrapper<IAlgorithm>>(algorithm, client, 0);
         communicator.Send<int>(updateInterval, client, 1);
       }
     }
 
+    public void TransmitAlgorithm(Stream stream) {
+      Console.WriteLine("Transmitting Algorithm...");
+
+      IAlgorithm alg = StreamingHelper.StreamItem(stream) as IAlgorithm;
+      MPITransportWrapper<IAlgorithm> algorithm = new MPITransportWrapper<IAlgorithm>(alg);
+
+      int clients = MPI.Communicator.world.Group.Size - 1;
+      for(int i = 0; i < clients; i++) {
+        int client = i + 1;
+        communicator.Send<MPITransportWrapper<IAlgorithm>>(algorithm, client, 0);
+      }
+    }
+
+    public object resultLocker = new object();
     public ItemList<ResultCollection> Results { get; set; }
 
-    public MPITransportWrapper<ItemList<ResultCollection>> GetResults() {
+    public Stream GetResults() {
       Console.WriteLine("Transmitting Results...");
-      return new MPITransportWrapper<ItemList<ResultCollection>>(Results);
+      MemoryStream stream = new MemoryStream();
+      lock (resultLocker) {
+        XmlGenerator.Serialize(Results, stream);
+      }
+      stream = new MemoryStream((stream as MemoryStream).GetBuffer());
+
+      return stream;
     }
 

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

@@ -28 +28 @@
 using HeuristicLab.Operators.MPISupport;
 using HeuristicLab.Core;
+using System.IO;
 
 namespace HeuristicLab.MPIAlgorithmRunner {
@@ -33 +34 @@
   public interface IAlgorithmBroker {
     [OperationContract]
-    void TransmitAlgorithm(MPITransportWrapper<IAlgorithm> algorithm, int updateInterval);
+    void SetUpdateInterval(int updateInterval);
+    
+    [OperationContract]
+    void TransmitAlgorithm(Stream stream);
 
     [OperationContract]
-    MPITransportWrapper<ItemList<ResultCollection>> GetResults();
+    Stream GetResults();
 
     [OperationContract]

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

@@ -77 +77 @@
           SetJobStatus(address);
 
-          ItemList<ResultCollection> results = new ItemList<ResultCollection>();
-          
           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]) {
@@ -105 +105 @@
 
             Console.WriteLine("Update results");
-            broker.Results = results;
+            lock (broker.resultLocker)
+              broker.Results = results;
           }
 
@@ -114 +115 @@
           broker.ExitWaitHandle.WaitOne();
           Console.WriteLine("Finished.");
+          Thread.Sleep(1000);
           service.Close();
         } else {

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

@@ -229 +229 @@
         task.StdOutFilePath = "stdout.txt";
         task.StdErrFilePath = "stderr.txt";
-        task.WorkDirectory = path; 
-        task.MinimumNumberOfCores = task.MaximumNumberOfCores = 3;
+        task.WorkDirectory = path;
+        task.MinimumNumberOfCores = task.MaximumNumberOfCores = cpuPerNode * requestedNodes.Count;
         job.AddTask(task);
 
@@ -260 +260 @@
 
           NetTcpBinding netTCPBinding = new NetTcpBinding(SecurityMode.None);
-          XmlDictionaryReaderQuotas quotas = new XmlDictionaryReaderQuotas();
-          quotas.MaxArrayLength = int.MaxValue;
-          netTCPBinding.ReaderQuotas = quotas;
+          netTCPBinding.TransferMode = TransferMode.Streamed;
           netTCPBinding.MaxReceivedMessageSize = int.MaxValue;
-          netTCPBinding.SendTimeout = new TimeSpan(600000000);
-          netTCPBinding.ReceiveTimeout = new TimeSpan(600000000);
           ChannelFactory<IAlgorithmBroker> factory = new ChannelFactory<IAlgorithmBroker>(netTCPBinding, address);
           IAlgorithmBroker proxy = factory.CreateChannel();
 
-          proxy.TransmitAlgorithm(new MPITransportWrapper<IAlgorithm>(algorithm), updateInterval);
+          Stream stream = new MemoryStream();
+          XmlGenerator.Serialize(algorithm, stream);
+          stream = new MemoryStream((stream as MemoryStream).GetBuffer());
+
+          proxy.TransmitAlgorithm(stream);
+          proxy.SetUpdateInterval(updateInterval);
 
           while (!proxy.IsAlgorithmTerminated()) {
             cancellationToken.ThrowIfCancellationRequested();
 
-            ItemList<ResultCollection> results = proxy.GetResults().InnerItem;
+            ItemList<ResultCollection> results = StreamingHelper.StreamItem(proxy.GetResults()) as ItemList<ResultCollection>;
 
             ResultCollection resultCollection = (globalScope.Variables["Results"].Value as ResultCollection);
@@ -292 +293 @@
           throw e;
         }
-        finally {
-           /*ISchedulerJob schedulerJob = scheduler.OpenJob(job.Id);
-           if (schedulerJob != null &&
-             (schedulerJob.State == JobState.Running || schedulerJob.State == JobState.Queued)) {
-               scheduler.CancelJob(job.Id, "Cancelled");            
-           } */
-        }
       }
     }

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

@@ -45 +45 @@
 
     public MPITransportWrapper(T item) {
-      innerItem = item;
+      innerItem = item.Clone() as T;
     }