Changeset 414

Timestamp:

07/31/08 13:08:41 (16 years ago)

Author:

gkronber

Message:

implemented #216 (ProcessingEngine should terminate on breakpoints)

Location:

trunk/sources

Files:

• HeuristicLab.DistributedEngine/DistributedEngine.cs (modified) (7 diffs)
• HeuristicLab.Grid/ClientForm.cs (modified) (1 diff)
• HeuristicLab.Grid/JobManager.cs (modified) (1 diff)
• HeuristicLab.Grid/ProcessingEngine.cs (modified) (5 diffs)

trunk/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs

@@ -36 +36 @@
 namespace HeuristicLab.DistributedEngine {
   public class DistributedEngine : EngineBase, IEditable {
+    private List<KeyValuePair<ProcessingEngine, AtomicOperation>> suspendedEngines = new List<KeyValuePair<ProcessingEngine, AtomicOperation>>();
     private JobManager jobManager;
     private string serverAddress;
@@ -59 +60 @@
     }
 
+    public override bool Terminated {
+      get {
+        return base.Terminated && suspendedEngines.Count == 0;
+      }
+    }
+
+    public override void Reset() {
+      suspendedEngines.Clear();
+      base.Reset();
+    }
+
     public override void Execute() {
       if(jobManager == null) this.jobManager = new JobManager(serverAddress);
@@ -70 +82 @@
 
     protected override void ProcessNextOperation() {
-      IOperation operation = myExecutionStack.Pop();
+      if(suspendedEngines.Count > 0) {
+        ProcessSuspendedEngines();
+      } else {
+        IOperation operation = myExecutionStack.Pop();
+        ProcessOperation(operation);
+      }
+    }
+
+    private void ProcessSuspendedEngines() {
+      WaitHandle[] waitHandles = new WaitHandle[suspendedEngines.Count];
+      int i = 0;
+      foreach(KeyValuePair<ProcessingEngine, AtomicOperation> suspendedPair in suspendedEngines) {
+        waitHandles[i++] = jobManager.BeginExecuteEngine(suspendedPair.Key);
+      }
+      WaitForAll(waitHandles);
+      // collect results
+      List<KeyValuePair<ProcessingEngine, AtomicOperation>> results = new List<KeyValuePair<ProcessingEngine, AtomicOperation>>();
+      try {
+        foreach(KeyValuePair<ProcessingEngine, AtomicOperation> suspendedPair in suspendedEngines) {
+          KeyValuePair<ProcessingEngine, AtomicOperation> p = new KeyValuePair<ProcessingEngine, AtomicOperation>(
+              jobManager.EndExecuteOperation(suspendedPair.Value),
+              suspendedPair.Value);
+          results.Add(p);
+        }
+      } catch(Exception e) {
+        // this exception means there was a problem with the underlying communication infrastructure
+        // -> show message dialog and abort engine
+        Abort();
+        ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(e); });
+        return;
+      }
+      // got all result engines without an exception -> merge results
+      ProcessResults(results);
+    }
+
+    private void ProcessOperation(IOperation operation) {
       if(operation is AtomicOperation) {
         AtomicOperation atomicOperation = (AtomicOperation)operation;
@@ -89 +136 @@
         CompositeOperation compositeOperation = (CompositeOperation)operation;
         if(compositeOperation.ExecuteInParallel) {
-          try {
-            WaitHandle[] waithandles = new WaitHandle[compositeOperation.Operations.Count];
-            int i = 0;
-            // HACK: assume that all atomicOperations have the same parent scope.
-            // 1) find that parent scope
-            // 2) remove all branches starting from the global scope that don't lead to the parentScope of the parallel operation
-            // 3) keep the branches to 'repair' the scope-tree later
-            // 4) for each parallel job attach only the sub-scope that this operation uses
-            // 5) after starting all parallel jobs restore the whole scope-tree
-            IScope parentScope = FindParentScope(GlobalScope, ((AtomicOperation)compositeOperation.Operations[0]).Scope);
-            List<IList<IScope>> prunedScopes = new List<IList<IScope>>();
-            PruneToParentScope(GlobalScope, parentScope, prunedScopes);
-            List<IScope> subScopes = new List<IScope>(parentScope.SubScopes);
-            foreach(IScope scope in subScopes) {
-              parentScope.RemoveSubScope(scope);
-            }
-            // start all parallel jobs
-            foreach(AtomicOperation parOperation in compositeOperation.Operations) {
-              parentScope.AddSubScope(parOperation.Scope);
-              waithandles[i++] = jobManager.BeginExecuteOperation(GlobalScope, parOperation);
-              parentScope.RemoveSubScope(parOperation.Scope);
-            }
-            foreach(IScope scope in subScopes) {
-              parentScope.AddSubScope(scope);
-            }
-            prunedScopes.Reverse();
-            RestoreFullTree(GlobalScope, prunedScopes);
-
-            // wait until all jobs are finished
-            // 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();
-                waithandles[i].Close();
-              }
-            }
-
-            CompositeOperation canceledOperations = new CompositeOperation();
-            canceledOperations.ExecuteInParallel = true;
-            // retrieve results and merge into scope-tree
-            foreach(AtomicOperation parOperation in compositeOperation.Operations) {
-              ProcessingEngine resultEngine = jobManager.EndExecuteOperation(parOperation);
-              if(resultEngine.Canceled) {
-                ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(new JobExecutionException(resultEngine.ErrorMessage)); });
-                canceledOperations.AddOperation(parOperation);
-              } else {
-                // if everything went fine we can merge the results into our local scope-tree
-                MergeScope(parOperation.Scope, resultEngine.InitialOperation.Scope);
-              }
-            }
-
-            if(canceledOperations.Operations.Count > 0) {
-              myExecutionStack.Push(canceledOperations);
-              Abort();
-            }
-          } catch(Exception e) {
-            myExecutionStack.Push(compositeOperation);
-            Abort();
-            ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(e); });
-          }
+          ProcessParallelOperation(compositeOperation);
           OnOperationExecuted(compositeOperation);
         } else {
@@ -156 +142 @@
             myExecutionStack.Push(compositeOperation.Operations[i]);
         }
+      }
+    }
+
+    private void ProcessParallelOperation(CompositeOperation compositeOperation) {
+      // send operations to grid
+      WaitHandle[] waithandles = BeginExecuteOperations(compositeOperation);
+      WaitForAll(waithandles);
+      // 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) {
+        // 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
+        myExecutionStack.Push(compositeOperation);
+        Abort();
+        ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(e); });
+        return;
+      }
+      // got all result engines without an exception -> merge results
+      ProcessResults(results);
+    }
+
+    private WaitHandle[] BeginExecuteOperations(CompositeOperation compositeOperation) {
+      WaitHandle[] waithandles = new WaitHandle[compositeOperation.Operations.Count];
+      int i = 0;
+      // HACK: assume that all atomicOperations have the same parent scope.
+      // 1) find that parent scope
+      // 2) remove all branches starting from the global scope that don't lead to the parentScope of the parallel operation
+      // 3) keep the branches to 'repair' the scope-tree later
+      // 4) for each parallel job attach only the sub-scope that this operation uses
+      // 5) after starting all parallel jobs restore the whole scope-tree
+      IScope parentScope = FindParentScope(GlobalScope, ((AtomicOperation)compositeOperation.Operations[0]).Scope);
+      List<IList<IScope>> prunedScopes = new List<IList<IScope>>();
+      PruneToParentScope(GlobalScope, parentScope, prunedScopes);
+      List<IScope> subScopes = new List<IScope>(parentScope.SubScopes);
+      foreach(IScope scope in subScopes) {
+        parentScope.RemoveSubScope(scope);
+      }
+      // start all parallel jobs
+      foreach(AtomicOperation parOperation in compositeOperation.Operations) {
+        parentScope.AddSubScope(parOperation.Scope);
+        waithandles[i++] = jobManager.BeginExecuteOperation(GlobalScope, parOperation);
+        parentScope.RemoveSubScope(parOperation.Scope);
+      }
+      foreach(IScope scope in subScopes) {
+        parentScope.AddSubScope(scope);
+      }
+      prunedScopes.Reverse();
+      RestoreFullTree(GlobalScope, prunedScopes);
+
+      return waithandles;
+    }
+
+    private void WaitForAll(WaitHandle[] waithandles) {
+      // wait until all jobs are finished
+      // WaitAll works only with maximally 64 waithandles
+      if(waithandles.Length <= 64) {
+        WaitHandle.WaitAll(waithandles);
+      } else {
+        int i;
+        for(i = 0; i < waithandles.Length; i++) {
+          waithandles[i].WaitOne();
+          waithandles[i].Close();
+        }
+      }
+    }
+
+    private void ProcessResults(List<KeyValuePair<ProcessingEngine, AtomicOperation>> results) {
+      // create a new compositeOperation to hold canceled operations that should be restarted
+      CompositeOperation canceledOperations = new CompositeOperation();
+      canceledOperations.ExecuteInParallel = true;
+
+      suspendedEngines.Clear();
+      // retrieve results and merge into scope-tree
+      foreach(KeyValuePair<ProcessingEngine, AtomicOperation> p in results) {
+        ProcessingEngine resultEngine = p.Key;
+        AtomicOperation parOperation = p.Value;
+        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)
+          // but don't merge the results of the aborted engine
+          ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(new JobExecutionException(resultEngine.ErrorMessage)); });
+          canceledOperations.AddOperation(parOperation);
+        } 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)
+          MergeScope(parOperation.Scope, resultEngine.InitialOperation.Scope);
+          resultEngine.InitialOperation = parOperation;
+          suspendedEngines.Add(new KeyValuePair<ProcessingEngine, AtomicOperation>(resultEngine, parOperation));
+        } else {
+          // engine is finished ->
+          // simply merge the results into our local scope-tree
+          MergeScope(parOperation.Scope, resultEngine.InitialOperation.Scope);
+        }
+      }
+      // if there were exceptions -> abort 
+      if(canceledOperations.Operations.Count > 0) {
+        // requeue the aborted operations 
+        myExecutionStack.Push(canceledOperations);
+        Abort();
+      }
+      // if there were breakpoints -> abort
+      if(suspendedEngines.Count > 0) {
+        Abort();
       }
     }
@@ -223 +318 @@
       XmlAttribute addressAttribute = document.CreateAttribute("ServerAddress");
       addressAttribute.Value = ServerAddress;
-      node.Attributes.Append(addressAttribute);
+      if(suspendedEngines.Count > 0) {
+        node.Attributes.Append(addressAttribute);
+        XmlNode suspendedEnginesNode = document.CreateElement("SuspendedEngines");
+        foreach(KeyValuePair<ProcessingEngine, AtomicOperation> p in suspendedEngines) {
+          XmlNode n = document.CreateElement("Entry");
+          n.AppendChild(PersistenceManager.Persist(p.Key, document, persistedObjects));
+          n.AppendChild(PersistenceManager.Persist(p.Value, document, persistedObjects));
+          suspendedEnginesNode.AppendChild(n);
+        }
+        node.AppendChild(suspendedEnginesNode);
+      }
       return node;
     }
@@ -229 +334 @@
       base.Populate(node, restoredObjects);
       ServerAddress = node.Attributes["ServerAddress"].Value;
+      XmlNode suspendedEnginesNode = node.SelectSingleNode("SuspendedEngines");
+      if(suspendedEnginesNode != null) {
+        foreach(XmlNode n in suspendedEnginesNode.ChildNodes) {
+          KeyValuePair<ProcessingEngine, AtomicOperation> p = new KeyValuePair<ProcessingEngine, AtomicOperation>(
+            (ProcessingEngine)PersistenceManager.Restore(n.ChildNodes[0], restoredObjects),
+            (AtomicOperation)PersistenceManager.Restore(n.ChildNodes[1], restoredObjects));
+          suspendedEngines.Add(p);
+        }
+      }
     }
     #endregion

trunk/sources/HeuristicLab.Grid/ClientForm.cs

@@ -153 +153 @@
       ProcessingEngine engine = (ProcessingEngine)sender;
 
-      // if the engine was stopped because of an error it's not necessary to return the whole engine
+      // if the engine was stopped because of an error (not suspended because of a breakpoint) 
+      // it's not necessary to return the whole engine
       // instead just return an empty engine that has the aborted flag set
-      if(engine.Canceled) {
+      if(engine.Canceled && !engine.Suspended) {
         engine.Reset();
         engine.OperatorGraph.Clear();

trunk/sources/HeuristicLab.Grid/JobManager.cs

@@ -189 +189 @@
 
     public WaitHandle BeginExecuteOperation(IScope globalScope, AtomicOperation operation) {
+      return BeginExecuteEngine(new ProcessingEngine(globalScope, operation));
+    }
+
+    public WaitHandle BeginExecuteEngine(ProcessingEngine engine) {
       Job job = new Job();
-      job.engine = new ProcessingEngine(globalScope, operation);
+      job.engine = engine;
       job.waitHandle = new ManualResetEvent(false);
       job.restarts = 0;

trunk/sources/HeuristicLab.Grid/ProcessingEngine.cs

@@ -34 +34 @@
     public AtomicOperation InitialOperation {
       get { return initialOperation; }
+      set { initialOperation = value; }
     }
 
@@ -39 +40 @@
     public string ErrorMessage {
       get { return errorMessage; }
+    }
+
+    private bool suspended;
+    public bool Suspended {
+      get { return suspended; }
     }
 
@@ -51 +57 @@
       myExecutionStack.Push(initialOperation);
     }
-    public override XmlNode GetXmlNode(string name, XmlDocument document, IDictionary<Guid, IStorable> persistedObjects) {
-      XmlNode node = base.GetXmlNode(name, document, persistedObjects);
-      XmlAttribute canceledAttr = document.CreateAttribute("Canceled");
-      canceledAttr.Value = Canceled.ToString();
-      node.Attributes.Append(canceledAttr);
-      if(errorMessage != null) {
-        XmlAttribute errorMessageAttr = document.CreateAttribute("ErrorMessage");
-        errorMessageAttr.Value = ErrorMessage;
-        node.Attributes.Append(errorMessageAttr);
-      }
-      node.AppendChild(PersistenceManager.Persist("InitialOperation", initialOperation, document, persistedObjects));
-      return node;
+
+    private void Suspend() {
+      Abort();
+      suspended = true;
     }
 
-    public override void Populate(XmlNode node, IDictionary<Guid, IStorable> restoredObjects) {
-      base.Populate(node, restoredObjects);
-      myCanceled = bool.Parse(node.Attributes["Canceled"].Value);
-      if(node.Attributes["ErrorMessage"] != null) errorMessage = node.Attributes["ErrorMessage"].Value;
-      initialOperation = (AtomicOperation)PersistenceManager.Restore(node.SelectSingleNode("InitialOperation"), restoredObjects);
+    public override void Execute() {
+      suspended = false;
+      base.Execute();
+    }
+
+    public override void ExecuteSteps(int steps) {
+      suspended = false;
+      base.ExecuteSteps(steps);
     }
 
@@ -88 +89 @@
         if(next != null)
           myExecutionStack.Push(next);
-        if(atomicOperation.Operator.Breakpoint) Abort();
+        if(atomicOperation.Operator.Breakpoint) Suspend();
       } else if(operation is CompositeOperation) {
         CompositeOperation compositeOperation = (CompositeOperation)operation;
@@ -95 +96 @@
       }
     }
+
+    #region persistence
+    public override XmlNode GetXmlNode(string name, XmlDocument document, IDictionary<Guid, IStorable> persistedObjects) {
+      XmlNode node = base.GetXmlNode(name, document, persistedObjects);
+      XmlAttribute canceledAttr = document.CreateAttribute("Canceled");
+      canceledAttr.Value = Canceled.ToString();
+      node.Attributes.Append(canceledAttr);
+      XmlAttribute suspendedAttr = document.CreateAttribute("Suspended");
+      suspendedAttr.Value = Suspended.ToString();
+      node.Attributes.Append(suspendedAttr);
+      if(errorMessage != null) {
+        XmlAttribute errorMessageAttr = document.CreateAttribute("ErrorMessage");
+        errorMessageAttr.Value = ErrorMessage;
+        node.Attributes.Append(errorMessageAttr);
+      }
+      node.AppendChild(PersistenceManager.Persist("InitialOperation", initialOperation, document, persistedObjects));
+      return node;
+    }
+
+    public override void Populate(XmlNode node, IDictionary<Guid, IStorable> restoredObjects) {
+      base.Populate(node, restoredObjects);
+      myCanceled = bool.Parse(node.Attributes["Canceled"].Value);
+      suspended = bool.Parse(node.Attributes["Suspended"].Value);
+      if(node.Attributes["ErrorMessage"] != null) errorMessage = node.Attributes["ErrorMessage"].Value;
+      initialOperation = (AtomicOperation)PersistenceManager.Restore(node.SelectSingleNode("InitialOperation"), restoredObjects);
+    }
+    #endregion
 
     private string CreateErrorMessage(Exception ex) {