Changeset 2055 for trunk/sources/HeuristicLab.DistributedEngine

Timestamp:

06/17/09 18:07:15 (16 years ago)

Author:

gkronber

Message:

Refactored JobManager and DistributedEngine to fix bugs in the GridExecuter. #644.

File:

• trunk/sources/HeuristicLab.DistributedEngine/3.2/DistributedEngine.cs (modified) (20 diffs)

trunk/sources/HeuristicLab.DistributedEngine/3.2/DistributedEngine.cs ¶

@@ -42 +42 @@
       get { return serverAddress; }
       set {
-        if(value != serverAddress) {
+        if (value != serverAddress) {
           serverAddress = value;
         }
@@ -72 +72 @@
 
     public override void Execute() {
-      if(jobManager == null) this.jobManager = new JobManager(serverAddress);
+      if (jobManager == null) this.jobManager = new JobManager(serverAddress);
       jobManager.Reset();
       base.Execute();
@@ -82 +82 @@
 
     protected override void ProcessNextOperation() {
-      if(suspendedEngines.Count > 0) {
+      if (suspendedEngines.Count > 0) {
         ProcessSuspendedEngines();
       } else {
@@ -91 +91 @@
 
     private void ProcessSuspendedEngines() {
-      WaitHandle[] waitHandles = new WaitHandle[suspendedEngines.Count];
+      AsyncGridResult[] asyncResults = new AsyncGridResult[suspendedEngines.Count];
       int i = 0;
-      foreach(KeyValuePair<ProcessingEngine, AtomicOperation> suspendedPair in suspendedEngines) {
-        waitHandles[i++] = jobManager.BeginExecuteEngine(suspendedPair.Key);
-      }
-      WaitForAll(waitHandles);
+      foreach (KeyValuePair<ProcessingEngine, AtomicOperation> suspendedPair in suspendedEngines) {
+        asyncResults[i++] = jobManager.BeginExecuteEngine(suspendedPair.Key);
+      }
+      WaitForAll(asyncResults);
       // collect results
       List<KeyValuePair<ProcessingEngine, AtomicOperation>> results = new List<KeyValuePair<ProcessingEngine, AtomicOperation>>();
       try {
-        foreach(KeyValuePair<ProcessingEngine, AtomicOperation> suspendedPair in suspendedEngines) {
+        int resultIndex = 0;
+        foreach (KeyValuePair<ProcessingEngine, AtomicOperation> suspendedPair in suspendedEngines) {
           KeyValuePair<ProcessingEngine, AtomicOperation> p = new KeyValuePair<ProcessingEngine, AtomicOperation>(
-              jobManager.EndExecuteOperation(suspendedPair.Value),
+              (ProcessingEngine)jobManager.EndExecuteEngine(asyncResults[resultIndex++]),
               suspendedPair.Value);
           results.Add(p);
         }
-      } catch(Exception e) {
+      }
+      catch (Exception e) {
         // this exception means there was a problem with the underlying communication infrastructure
         // -> show message dialog and abort engine
@@ -118 +120 @@
 
     private void ProcessOperation(IOperation operation) {
-      if(operation is AtomicOperation) {
+      if (operation is AtomicOperation) {
         AtomicOperation atomicOperation = (AtomicOperation)operation;
         IOperation next = null;
         try {
           next = atomicOperation.Operator.Execute(atomicOperation.Scope);
-        } catch(Exception ex) {
+        }
+        catch (Exception ex) {
           // push operation on stack again
           myExecutionStack.Push(atomicOperation);
@@ -129 +132 @@
           ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(ex); });
         }
-        if(next != null)
+        if (next != null)
           myExecutionStack.Push(next);
         OnOperationExecuted(atomicOperation);
-        if(atomicOperation.Operator.Breakpoint) Abort();
-      } else if(operation is CompositeOperation) {
+        if (atomicOperation.Operator.Breakpoint) Abort();
+      } else if (operation is CompositeOperation) {
         CompositeOperation compositeOperation = (CompositeOperation)operation;
-        if(compositeOperation.ExecuteInParallel) {
+        if (compositeOperation.ExecuteInParallel) {
           ProcessParallelOperation(compositeOperation);
           OnOperationExecuted(compositeOperation);
         } else {
-          for(int i = compositeOperation.Operations.Count - 1; i >= 0; i--)
+          for (int i = compositeOperation.Operations.Count - 1; i >= 0; i--)
             myExecutionStack.Push(compositeOperation.Operations[i]);
         }
@@ -147 +150 @@
     private void ProcessParallelOperation(CompositeOperation compositeOperation) {
       // send operations to grid
-      WaitHandle[] waithandles = BeginExecuteOperations(compositeOperation);
-      WaitForAll(waithandles);
+      AsyncGridResult[] asyncResults = BeginExecuteOperations(compositeOperation);
+      WaitForAll(asyncResults);
       // collect results
       List<KeyValuePair<ProcessingEngine, AtomicOperation>> results = new List<KeyValuePair<ProcessingEngine, AtomicOperation>>();
       try {
-        foreach(AtomicOperation parOperation in compositeOperation.Operations) {
-          KeyValuePair<ProcessingEngine, AtomicOperation> p = new KeyValuePair<ProcessingEngine, AtomicOperation>(
-            jobManager.EndExecuteOperation(parOperation), parOperation);
-          results.Add(p);
-        }
-      } catch(Exception e) {
+        int i = 0;
+        foreach (AtomicOperation parOperation in compositeOperation.Operations) {
+          results.Add(new KeyValuePair<ProcessingEngine, AtomicOperation>(
+              (ProcessingEngine)jobManager.EndExecuteEngine(asyncResults[i++]), parOperation));
+        }
+      }
+      catch (Exception e) {
         // this exception means there was a problem with the underlying communication infrastructure
         // -> show message dialog, abort engine, requeue the whole composite operation again and return
@@ -169 +173 @@
     }
 
-    private WaitHandle[] BeginExecuteOperations(CompositeOperation compositeOperation) {
-      WaitHandle[] waithandles = new WaitHandle[compositeOperation.Operations.Count];
+    private AsyncGridResult[] BeginExecuteOperations(CompositeOperation compositeOperation) {
+      AsyncGridResult[] asyncResults = new AsyncGridResult[compositeOperation.Operations.Count];
       int i = 0;
       // HACK: assume that all atomicOperations have the same parent scope.
@@ -182 +186 @@
       PruneToParentScope(GlobalScope, parentScope, prunedScopes);
       List<IScope> subScopes = new List<IScope>(parentScope.SubScopes);
-      foreach(IScope scope in subScopes) {
+      foreach (IScope scope in subScopes) {
         parentScope.RemoveSubScope(scope);
       }
       // start all parallel jobs
-      foreach(AtomicOperation parOperation in compositeOperation.Operations) {
+      foreach (AtomicOperation parOperation in compositeOperation.Operations) {
         parentScope.AddSubScope(parOperation.Scope);
-        waithandles[i++] = jobManager.BeginExecuteOperation(GlobalScope, parOperation);
+        asyncResults[i++] = jobManager.BeginExecuteEngine(new ProcessingEngine(GlobalScope, parOperation));
         parentScope.RemoveSubScope(parOperation.Scope);
       }
-      foreach(IScope scope in subScopes) {
+      foreach (IScope scope in subScopes) {
         parentScope.AddSubScope(scope);
       }
@@ -197 +201 @@
       RestoreFullTree(GlobalScope, prunedScopes);
 
-      return waithandles;
-    }
-
-    private void WaitForAll(WaitHandle[] waithandles) {
+      return asyncResults;
+    }
+
+    private void WaitForAll(AsyncGridResult[] asyncResults) {
       // wait until all jobs are finished
       // WaitAll works only with maximally 64 waithandles
-      if(waithandles.Length <= 64) {
-        WaitHandle.WaitAll(waithandles);
+      if (asyncResults.Length <= 64) {
+        WaitHandle[] waitHandles = new WaitHandle[asyncResults.Length];
+        for (int i = 0; i < asyncResults.Length; i++) {
+          waitHandles[i] = asyncResults[i].WaitHandle;
+        }
+        WaitHandle.WaitAll(waitHandles);
       } else {
         int i;
-        for(i = 0; i < waithandles.Length; i++) {
-          waithandles[i].WaitOne();
-          waithandles[i].Close();
+        for (i = 0; i < asyncResults.Length; i++) {
+          asyncResults[i].WaitHandle.WaitOne();
         }
       }
@@ -221 +228 @@
       suspendedEngines.Clear();
       // retrieve results and merge into scope-tree
-      foreach(KeyValuePair<ProcessingEngine, AtomicOperation> p in results) {
+      foreach (KeyValuePair<ProcessingEngine, AtomicOperation> p in results) {
         ProcessingEngine resultEngine = p.Key;
         AtomicOperation parOperation = p.Value;
-        if(resultEngine.Canceled && !resultEngine.Suspended) {
+        if (resultEngine.Canceled && !resultEngine.Suspended) {
           // when an engine was canceled but not suspended this means there was a problem
           // show error message and queue the operation for restart (again parallel)
@@ -230 +237 @@
           ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(new JobExecutionException(resultEngine.ErrorMessage)); });
           canceledOperations.AddOperation(parOperation);
-        } else if(resultEngine.Suspended) {
+        } else if (resultEngine.Suspended) {
           // when the engine was suspended it means it was stopped because of a breakpoint 
           // -> merge the partial results and queue the engine (which has operations remaining in the execution stack) to be resumed (parallel)
@@ -243 +250 @@
       }
       // if there were exceptions -> abort 
-      if(canceledOperations.Operations.Count > 0) {
+      if (canceledOperations.Operations.Count > 0) {
         // requeue the aborted operations 
         myExecutionStack.Push(canceledOperations);
@@ -249 +256 @@
       }
       // if there were breakpoints -> abort
-      if(suspendedEngines.Count > 0) {
+      if (suspendedEngines.Count > 0) {
         Abort();
       }
@@ -255 +262 @@
 
     private void RestoreFullTree(IScope currentScope, IList<IList<IScope>> savedScopes) {
-      if(savedScopes.Count == 0) return;
+      if (savedScopes.Count == 0) return;
       IScope remainingBranch = currentScope.SubScopes[0];
       currentScope.RemoveSubScope(remainingBranch);
       IList<IScope> savedScopesForCurrent = savedScopes[0];
-      foreach(IScope savedScope in savedScopesForCurrent) {
+      foreach (IScope savedScope in savedScopesForCurrent) {
         currentScope.AddSubScope(savedScope);
       }
@@ -267 +274 @@
 
     private IScope PruneToParentScope(IScope currentScope, IScope scope, IList<IList<IScope>> prunedScopes) {
-      if(currentScope == scope) return currentScope;
-      if(currentScope.SubScopes.Count == 0) return null;
+      if (currentScope == scope) return currentScope;
+      if (currentScope.SubScopes.Count == 0) return null;
       IScope foundScope = null;
       // try to find the searched scope in all my sub-scopes
-      foreach(IScope subScope in currentScope.SubScopes) {
+      foreach (IScope subScope in currentScope.SubScopes) {
         foundScope = PruneToParentScope(subScope, scope, prunedScopes);
-        if(foundScope != null) break; // we can stop as soon as we find the scope in a branch
-      }
-      if(foundScope != null) { // when we found the scopes in my sub-scopes
+        if (foundScope != null) break; // we can stop as soon as we find the scope in a branch
+      }
+      if (foundScope != null) { // when we found the scopes in my sub-scopes
         List<IScope> subScopes = new List<IScope>(currentScope.SubScopes); // store the list of sub-scopes
         prunedScopes.Add(subScopes);
         // remove all my sub-scopes
-        foreach(IScope subScope in subScopes) {
+        foreach (IScope subScope in subScopes) {
           currentScope.RemoveSubScope(subScope);
         }
@@ -291 +298 @@
 
     private IScope FindParentScope(IScope currentScope, IScope childScope) {
-      if(currentScope.SubScopes.Contains(childScope)) return currentScope;
-      foreach(IScope subScope in currentScope.SubScopes) {
+      if (currentScope.SubScopes.Contains(childScope)) return currentScope;
+      foreach (IScope subScope in currentScope.SubScopes) {
         IScope result = FindParentScope(subScope, childScope);
-        if(result != null) return result;
+        if (result != null) return result;
       }
       return null;
@@ -302 +309 @@
       // merge the results
       original.Clear();
-      foreach(IVariable variable in result.Variables) {
+      foreach (IVariable variable in result.Variables) {
         original.AddVariable(variable);
       }
-      foreach(IScope subScope in result.SubScopes) {
+      foreach (IScope subScope in result.SubScopes) {
         original.AddSubScope(subScope);
       }
-      foreach(KeyValuePair<string, string> alias in result.Aliases) {
+      foreach (KeyValuePair<string, string> alias in result.Aliases) {
         original.AddAlias(alias.Key, alias.Value);
       }
@@ -319 +326 @@
       addressAttribute.Value = ServerAddress;
       node.Attributes.Append(addressAttribute);
-      if(suspendedEngines.Count > 0) {
+      if (suspendedEngines.Count > 0) {
         XmlNode suspendedEnginesNode = document.CreateElement("SuspendedEngines");
-        foreach(KeyValuePair<ProcessingEngine, AtomicOperation> p in suspendedEngines) {
+        foreach (KeyValuePair<ProcessingEngine, AtomicOperation> p in suspendedEngines) {
           XmlNode n = document.CreateElement("Entry");
           n.AppendChild(PersistenceManager.Persist(p.Key, document, persistedObjects));
@@ -335 +342 @@
       ServerAddress = node.Attributes["ServerAddress"].Value;
       XmlNode suspendedEnginesNode = node.SelectSingleNode("SuspendedEngines");
-      if(suspendedEnginesNode != null) {
-        foreach(XmlNode n in suspendedEnginesNode.ChildNodes) {
+      if (suspendedEnginesNode != null) {
+        foreach (XmlNode n in suspendedEnginesNode.ChildNodes) {
           KeyValuePair<ProcessingEngine, AtomicOperation> p = new KeyValuePair<ProcessingEngine, AtomicOperation>(
             (ProcessingEngine)PersistenceManager.Restore(n.ChildNodes[0], restoredObjects),