Changeset 248 for trunk/sources

Timestamp:

05/13/08 23:06:12 (17 years ago)

Author:

gkronber

Message:

improved stability of distributed engine (#149)

Location:

trunk/sources/HeuristicLab.DistributedEngine

Files:

• DistributedEngine.cs (modified) (3 diffs)
• JobManager.cs (modified) (4 diffs)

trunk/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs

@@ -92 +92 @@
             WaitHandle[] waithandles = new WaitHandle[compositeOperation.Operations.Count];
             int i = 0;
+            // start all parallel jobs
             foreach(AtomicOperation parOperation in compositeOperation.Operations) {
               waithandles[i++] = jobManager.BeginExecuteOperation(OperatorGraph, GlobalScope, parOperation);
             }
+
+            // wait until all jobs are finished
             // WaitAll works only with maximally 64 waithandles
             if(waithandles.Length <= 64) {
@@ -103 +106 @@
               }
             }
-            if(jobManager.Exception != null) {
-              myExecutionStack.Push(compositeOperation);
-              Abort();
-              ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(jobManager.Exception); });
+            // retrieve results and merge into scope-tree
+            foreach(AtomicOperation parOperation in compositeOperation.Operations) {
+              IScope result = jobManager.EndExecuteOperation(parOperation);
+              MergeScope(parOperation.Scope, result);
             }
           } catch(Exception e) {
             myExecutionStack.Push(compositeOperation);
             Abort();
-            ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(jobManager.Exception); });
+            ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(e); });
           }
         } else {
@@ -117 +120 @@
             myExecutionStack.Push(compositeOperation.Operations[i]);
         }
+      }
+    }
+
+    private void MergeScope(IScope original, IScope result) {
+      // merge the results
+      original.Clear();
+      foreach(IVariable variable in result.Variables) {
+        original.AddVariable(variable);
+      }
+      foreach(IScope subScope in result.SubScopes) {
+        original.AddSubScope(subScope);
+      }
+      foreach(KeyValuePair<string, string> alias in result.Aliases) {
+        original.AddAlias(alias.Key, alias.Value);
       }
     }

trunk/sources/HeuristicLab.DistributedEngine/JobManager.cs

@@ -18 +18 @@
     private Dictionary<Guid, byte[]> runningEngines = new Dictionary<Guid, byte[]>();
     private Dictionary<Guid, ManualResetEvent> waithandles = new Dictionary<Guid, ManualResetEvent>();
-    private object locker = new object();
+    private Dictionary<AtomicOperation, byte[]> results = new Dictionary<AtomicOperation, byte[]>();
+    private object connectionLock = new object();
+    private object dictionaryLock = new object();
+
     private const int MAX_RESTARTS = 5;
-    private Exception exception;
+    private const int MAX_CONNECTION_RETRIES = 10;
+    private const int RETRY_TIMEOUT_SEC = 10;
+    private const int CHECK_RESULTS_TIMEOUT = 10;
+
     private ChannelFactory<IGridServer> factory;
-    public Exception Exception {
-      get { return exception; }
-    }
 
     public JobManager(string address) {
@@ -31 +34 @@
 
     internal void Reset() {
-      lock(locker) {
-        ResetConnection();
+      ResetConnection();
+      lock(dictionaryLock) {
         foreach(WaitHandle wh in waithandles.Values) wh.Close();
         waithandles.Clear();
         engineOperations.Clear();
         runningEngines.Clear();
-        exception = null;
+        results.Clear();
       }
     }
 
     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));
+      lock(connectionLock) {
+        // 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));
+      }
     }
 
@@ -58 +64 @@
       PersistenceManager.Save(engine, stream);
       stream.Close();
-      if(factory.State != CommunicationState.Opened)
-        ResetConnection();
-      Guid currentEngineGuid = server.BeginExecuteEngine(memStream.ToArray());
-      lock(locker) {
+      byte[] zippedEngine = memStream.ToArray();
+      Guid currentEngineGuid = Guid.Empty;
+      bool success = false;
+      int retryCount = 0;
+      do {
+        lock(connectionLock) {
+          if(factory.State != CommunicationState.Opened)
+            ResetConnection();
+          try {
+            currentEngineGuid = server.BeginExecuteEngine(zippedEngine);
+            success = true;
+          } catch(TimeoutException timeoutException) {
+            if(retryCount < MAX_CONNECTION_RETRIES) {
+              Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+              retryCount++;
+            } else {
+              throw new ApplicationException("Max retries reached.", timeoutException);
+            }
+          } catch(CommunicationException communicationException) {
+            // wait some time and try again (limit with maximal retries if retry count reached throw exception -> engine can decide to stop execution)
+            if(retryCount < MAX_CONNECTION_RETRIES) {
+              Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+              retryCount++;
+            } else {
+              throw new ApplicationException("Max retries reached.", communicationException);
+            }
+          }
+        }
+      } while(!success);
+      lock(dictionaryLock) {
         runningEngines[currentEngineGuid] = memStream.ToArray();
         engineOperations[currentEngineGuid] = operation;
         waithandles[currentEngineGuid] = new ManualResetEvent(false);
-        ThreadPool.QueueUserWorkItem(new WaitCallback(TryGetResult), currentEngineGuid);
       }
+      ThreadPool.QueueUserWorkItem(new WaitCallback(TryGetResult), currentEngineGuid);
       return waithandles[currentEngineGuid];
+    }
+
+    public IScope EndExecuteOperation(AtomicOperation operation) {
+      byte[] zippedResult = results[operation];
+      // restore the engine 
+      GZipStream stream = new GZipStream(new MemoryStream(zippedResult), CompressionMode.Decompress);
+      ProcessingEngine resultEngine = (ProcessingEngine)PersistenceManager.Load(stream);
+
+      return resultEngine.InitialOperation.Scope;
     }
 
@@ -74 +115 @@
       int restartCounter = 0;
       do {
-        try {
-          if(factory.State != CommunicationState.Opened) ResetConnection();
-          byte[] resultXml = server.TryEndExecuteEngine(engineGuid, 100);
-          if(resultXml != null) {
-            // restore the engine 
-            GZipStream stream = new GZipStream(new MemoryStream(resultXml), CompressionMode.Decompress);
-            ProcessingEngine resultEngine = (ProcessingEngine)PersistenceManager.Load(stream);
-
-            // merge the results
-            IScope oldScope = engineOperations[engineGuid].Scope;
-            oldScope.Clear();
-            foreach(IVariable variable in resultEngine.InitialOperation.Scope.Variables) {
-              oldScope.AddVariable(variable);
-            }
-            foreach(IScope subScope in resultEngine.InitialOperation.Scope.SubScopes) {
-              oldScope.AddSubScope(subScope);
-            }
-            foreach(KeyValuePair<string, string> alias in resultEngine.InitialOperation.Scope.Aliases) {
-              oldScope.AddAlias(alias.Key, alias.Value);
-            }
-
-            lock(locker) {
-              // signal the wait handle and clean up then return
-              waithandles[engineGuid].Set();
-              engineOperations.Remove(engineGuid);
-              waithandles.Remove(engineGuid);
-              runningEngines.Remove(engineGuid);
-            }
-            return;
-          } else {
-            // check if the server is still working on the job
-            JobState jobState = server.JobState(engineGuid);
-            if(jobState == JobState.Unkown) {
-              // restart job
-              byte[] packedEngine;
-              lock(locker) {
-                packedEngine = runningEngines[engineGuid];
+        byte[] zippedResult = null;
+        lock(connectionLock) {
+          bool success = false;
+          int retries = 0;
+          do {
+            if(factory.State != CommunicationState.Opened) ResetConnection();
+            try {
+              zippedResult = server.TryEndExecuteEngine(engineGuid, 100);
+              success = true;
+            } catch(TimeoutException timeoutException) {
+              success = false;
+              retries++;
+              Thread.Sleep(RETRY_TIMEOUT_SEC);
+            } catch(CommunicationException communicationException) {
+              success = false;
+              retries++;
+              Thread.Sleep(RETRY_TIMEOUT_SEC);
+            }
+
+          } while(!success && retries < MAX_CONNECTION_RETRIES);
+        }
+        if(zippedResult != null) {
+          lock(dictionaryLock) {
+            // store result
+            results[engineOperations[engineGuid]] = zippedResult;
+
+            // signal the wait handle and clean up then return
+            engineOperations.Remove(engineGuid);
+            runningEngines.Remove(engineGuid);
+            waithandles[engineGuid].Set();
+            waithandles.Remove(engineGuid);
+          }
+          return;
+        } else {
+          // check if the server is still working on the job
+          bool success = false;
+          int retries = 0;
+          JobState jobState = JobState.Unkown;
+          do {
+            try {
+              lock(connectionLock) {
+                if(factory.State != CommunicationState.Opened) ResetConnection();
+                jobState = server.JobState(engineGuid);
               }
-              server.BeginExecuteEngine(packedEngine);
-              restartCounter++;
-            }
+              success = true;
+            } catch(TimeoutException timeoutException) {
+              retries++;
+              success = false;
+              Thread.Sleep(RETRY_TIMEOUT_SEC);
+            } catch(CommunicationException communicationException) {
+              retries++;
+              success = false;
+              Thread.Sleep(RETRY_TIMEOUT_SEC);
+            }
+          } while(!success && retries < MAX_CONNECTION_RETRIES);
+          if(jobState == JobState.Unkown) {
+            // restart job
+            byte[] packedEngine;
+            lock(dictionaryLock) {
+              packedEngine = runningEngines[engineGuid];
+            }
+            success = false;
+            retries = 0;
+            do {
+              try {
+                lock(connectionLock) {
+                  if(factory.State != CommunicationState.Opened) ResetConnection();
+                  server.BeginExecuteEngine(packedEngine);
+                }
+                success = true;
+              } catch(TimeoutException timeoutException) {
+                success = false;
+                retries++;
+                Thread.Sleep(RETRY_TIMEOUT_SEC);
+              } catch(CommunicationException communicationException) {
+                success = false;
+                retries++;
+                Thread.Sleep(RETRY_TIMEOUT_SEC);
+              }
+            } while(!success && retries < MAX_CONNECTION_RETRIES);
+            restartCounter++;
           }
-        } catch(Exception e) {
-          // catch all exceptions set an exception flag, signal the wait-handle and exit the routine
-          this.exception = new Exception("There was a problem with parallel execution", e);
-          waithandles[engineGuid].Set();
-          return;
         }
 
         // when we reach a maximum amount of restarts => signal the wait-handle and set a flag that there was a problem
         if(restartCounter > MAX_RESTARTS) {
-          this.exception = new Exception("Maximal number of parallel job restarts reached");
-          waithandles[engineGuid].Set();
-          return;
-        }
-
-        Thread.Sleep(TimeSpan.FromSeconds(10.0));
+          throw new ApplicationException("Maximum number of job restarts reached.");
+        }
+
+        Thread.Sleep(TimeSpan.FromSeconds(CHECK_RESULTS_TIMEOUT));
       } while(true);
     }