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

Timestamp:

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

Author:

svonolfe

Message:

Implemented result and algorithm streaming (#1542)

File:

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

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");            
-           } */
-        }
       }
     }