Changeset 228 for trunk/sources/HeuristicLab.DistributedEngine

Timestamp:

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

Author:

gkronber

Message:

bug fixing in DistributedEngine and Grid-Infrastructure

Location:

trunk/sources/HeuristicLab.DistributedEngine

Files:

• DistributedEngine.cs (modified) (1 diff)
• JobManager.cs (modified) (4 diffs)

trunk/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs

@@ -89 +89 @@
         CompositeOperation compositeOperation = (CompositeOperation)operation;
         if(compositeOperation.ExecuteInParallel) {
-          WaitHandle[] waithandles = new WaitHandle[compositeOperation.Operations.Count];
-          int i = 0;
-          foreach(AtomicOperation parOperation in compositeOperation.Operations) {
-            waithandles[i++] = jobManager.BeginExecuteOperation(OperatorGraph, GlobalScope, parOperation);
-          }
-          // WaitAll works only with maximally 64 waithandles
-          if(waithandles.Length <= 64) {
-            WaitHandle.WaitAll(waithandles);
-          } else {
-            for(i = 0; i < waithandles.Length; i++) {
-              waithandles[i].WaitOne();
+          try {
+            WaitHandle[] waithandles = new WaitHandle[compositeOperation.Operations.Count];
+            int i = 0;
+            foreach(AtomicOperation parOperation in compositeOperation.Operations) {
+              waithandles[i++] = jobManager.BeginExecuteOperation(OperatorGraph, GlobalScope, parOperation);
             }
-          }
-          if(jobManager.Exception != null) {
+            // WaitAll works only with maximally 64 waithandles
+            if(waithandles.Length <= 64) {
+              WaitHandle.WaitAll(waithandles);
+            } else {
+              for(i = 0; i < waithandles.Length; i++) {
+                waithandles[i].WaitOne();
+              }
+            }
+            if(jobManager.Exception != null) {
+              myExecutionStack.Push(compositeOperation);
+              Abort();
+              ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(jobManager.Exception); });
+            }
+          } catch(Exception e) {
             myExecutionStack.Push(compositeOperation);
             Abort();

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) {