Changeset 6401 for branches/MPI/HeuristicLab.MPIAlgorithmRunner/3.3/AlgorithmBroker.cs

Timestamp:

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

Author:

svonolfe

Message:

Implemented result and algorithm streaming (#1542)

File:

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

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