Changeset 228 for trunk/sources/HeuristicLab.DistributedEngine/JobManager.cs

Timestamp:

05/09/08 16:22:33 (16 years ago)

Author:

gkronber

Message:

bug fixing in DistributedEngine and Grid-Infrastructure

File:

• trunk/sources/HeuristicLab.DistributedEngine/JobManager.cs (modified) (4 diffs)

trunk/sources/HeuristicLab.DistributedEngine/JobManager.cs

@@ -21 +21 @@
     private const int MAX_RESTARTS = 5;
     private Exception exception;
+    private ChannelFactory<IGridServer> factory;
     public Exception Exception {
       get { return exception; }
@@ -31 +32 @@
     internal void Reset() {
       lock(locker) {
-        // open a new channel
-        NetTcpBinding binding = new NetTcpBinding();
-        binding.MaxReceivedMessageSize = 100000000; // 100Mbytes
-        binding.ReaderQuotas.MaxStringContentLength = 100000000; // also 100M chars
-        binding.ReaderQuotas.MaxArrayLength = 100000000; // also 100M elements;
-        binding.Security.Mode = SecurityMode.None;
-        ChannelFactory<IGridServer> factory = new ChannelFactory<IGridServer>(binding);
-        server = factory.CreateChannel(new EndpointAddress(address));
-
+        ResetConnection();
         foreach(WaitHandle wh in waithandles.Values) wh.Close();
         waithandles.Clear();
@@ -48 +41 @@
     }
 
+    private void ResetConnection() {
+      // open a new channel
+      NetTcpBinding binding = new NetTcpBinding();
+      binding.MaxReceivedMessageSize = 100000000; // 100Mbytes
+      binding.ReaderQuotas.MaxStringContentLength = 100000000; // also 100M chars
+      binding.ReaderQuotas.MaxArrayLength = 100000000; // also 100M elements;
+      binding.Security.Mode = SecurityMode.None;
+      factory = new ChannelFactory<IGridServer>(binding);
+      server = factory.CreateChannel(new EndpointAddress(address));
+    }
+
     public WaitHandle BeginExecuteOperation(IOperatorGraph operatorGraph, IScope globalScope, AtomicOperation operation) {
       ProcessingEngine engine = new ProcessingEngine(operatorGraph, globalScope, operation); // OperatorGraph not needed?
       MemoryStream memStream = new MemoryStream();
       GZipStream stream = new GZipStream(memStream, CompressionMode.Compress, true);
-      PersistenceManager.Save(engine, stream); // Careful! Make sure that persistence is thread-safe!
+      PersistenceManager.Save(engine, stream);
       stream.Close();
+      if(factory.State != CommunicationState.Opened)
+        ResetConnection();
       Guid currentEngineGuid = server.BeginExecuteEngine(memStream.ToArray());
       lock(locker) {
@@ -69 +75 @@
       do {
         try {
+          if(factory.State != CommunicationState.Opened) ResetConnection();
           byte[] resultXml = server.TryEndExecuteEngine(engineGuid, 100);
           if(resultXml != null) {