Changeset 33 for trunk/sources/HeuristicLab.DistributedEngine

Timestamp:

03/01/08 15:05:37 (17 years ago)

Author:

gkronber

Message:

• worked on #2
• fixed some problems in communication between Grid and DistributedEngine

File:

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

trunk/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs

@@ -34 +34 @@
   public class DistributedEngine : EngineBase, IEditable {
     private IGridServer server;
-    private Dictionary<Guid, AtomicOperation> runningEngines = new Dictionary<Guid, AtomicOperation>();
-
+    private Dictionary<Guid, AtomicOperation> engineOperations = new Dictionary<Guid, AtomicOperation>();
+    private List<Guid> runningEngines = new List<Guid>();
     private string serverAddress;
     public string ServerAddress {
@@ -43 +43 @@
           serverAddress = value;
         }
+      }
+    }
+    public override bool Terminated {
+      get {
+        return myExecutionStack.Count == 0 && runningEngines.Count == 0;
       }
     }
@@ -77 +82 @@
     public override void Abort() {
       base.Abort();
-      foreach(Guid engineGuid in runningEngines.Keys) {
+      foreach(Guid engineGuid in runningEngines) {
         server.AbortEngine(engineGuid);
       }
@@ -83 +88 @@
 
     protected override void ProcessNextOperation() {
-      ProcessNextOperation(myExecutionStack, 0);
-    }
-    private void ProcessNextOperation(Stack<IOperation> stack, int currentOperatorIndex) {
-      IOperation operation = stack.Pop();
+      if(runningEngines.Count != 0) {
+        Guid engineGuid = runningEngines[0];
+        byte[] scopeXml = server.TryEndExecuteEngine(engineGuid,100);
+        if(scopeXml != null) {
+          GZipStream stream = new GZipStream(new MemoryStream(scopeXml), CompressionMode.Decompress);
+          IScope newScope = (IScope)PersistenceManager.Load(stream);
+          IScope oldScope = engineOperations[engineGuid].Scope;
+          oldScope.Clear();
+          foreach(IVariable variable in newScope.Variables) {
+            oldScope.AddVariable(variable);
+          }
+          foreach(IScope subScope in newScope.SubScopes) {
+            oldScope.AddSubScope(subScope);
+          }
+          runningEngines.Remove(engineGuid);
+          engineOperations.Remove(engineGuid);
+        }
+
+        if(Canceled) {
+          // write back not finished tasks
+          //CompositeOperation remaining = new CompositeOperation();
+          //remaining.ExecuteInParallel = true;
+          //for(int i = 0; i < list.tasks.Length; i++) {
+          //  if(list.tasks[i].Count > 0) {
+          //    CompositeOperation task = new CompositeOperation();
+          //    while(list.tasks[i].Count > 0)
+          //      task.AddOperation(list.tasks[i].Pop());
+          //    remaining.AddOperation(task);
+          //  }
+          //}
+          //if(remaining.Operations.Count > 0)
+          //  stack.Push(remaining);
+        }
+        return;
+      }
+      IOperation operation = myExecutionStack.Pop();
       if(operation is AtomicOperation) {
         AtomicOperation atomicOperation = (AtomicOperation)operation;
@@ -94 +131 @@
         } catch(Exception ex) {
           // push operation on stack again
-          stack.Push(atomicOperation);
+          myExecutionStack.Push(atomicOperation);
           Abort();
           ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(ex); });
         }
         if(next != null)
-          stack.Push(next);
+          myExecutionStack.Push(next);
         OnOperationExecuted(atomicOperation);
         if(atomicOperation.Operator.Breakpoint) Abort();
@@ -112 +149 @@
             stream.Close();
             Guid currentEngineGuid = server.BeginExecuteEngine(memStream.ToArray());
-            runningEngines[currentEngineGuid] = parOperation;
+            runningEngines.Add(currentEngineGuid);
+            engineOperations[currentEngineGuid] = parOperation;
           }
-          foreach(Guid engineGuid in runningEngines.Keys) {
-            byte[] scopeXml = server.EndExecuteEngine(engineGuid);
-            GZipStream stream = new GZipStream(new MemoryStream(scopeXml), CompressionMode.Decompress);
-            IScope newScope = (IScope)PersistenceManager.Load(stream);
-            IScope oldScope = runningEngines[engineGuid].Scope;
-            oldScope.Clear();
-            foreach(IVariable variable in newScope.Variables) {
-              oldScope.AddVariable(variable);
-            }
-            foreach(IScope subScope in newScope.SubScopes) {
-              oldScope.AddSubScope(subScope);
-            }
-          }
-
-          // TASK (gkronber 12.2.08)
-          //if (Canceled) {
-          //  // write back not finished tasks
-          //  CompositeOperation remaining = new CompositeOperation();
-          //  remaining.ExecuteInParallel = true;
-          //  for (int i = 0; i < list.tasks.Length; i++) {
-          //    if (list.tasks[i].Count > 0) {
-          //      CompositeOperation task = new CompositeOperation();
-          //      while (list.tasks[i].Count > 0)
-          //        task.AddOperation(list.tasks[i].Pop());
-          //      remaining.AddOperation(task);
-          //    }
-          //  }
-          //  if (remaining.Operations.Count > 0)
-          //    stack.Push(remaining);
-          //}
         } else {
           for(int i = compositeOperation.Operations.Count - 1; i >= 0; i--)
-            stack.Push(compositeOperation.Operations[i]);
+            myExecutionStack.Push(compositeOperation.Operations[i]);
         }
       }