Changeset 33

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

Location:

trunk/sources

Files:

• HeuristicLab.DistributedEngine/DistributedEngine.cs (modified) (6 diffs)
• HeuristicLab.Grid/EngineStore.cs (modified) (6 diffs)
• HeuristicLab.Grid/GridServer.cs (modified) (1 diff)
• HeuristicLab.Grid/IGridServer.cs (modified) (1 diff)

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]);
         }
       }

trunk/sources/HeuristicLab.Grid/EngineStore.cs

@@ -32 +32 @@
     private Dictionary<Guid, byte[]> waitingEngines;
     private Dictionary<Guid, byte[]> runningEngines;
+    private Dictionary<Guid, ManualResetEvent> waitHandles;
     private Dictionary<Guid, byte[]> results;
     private Dictionary<Guid, string> runningClients;
     private object bigLock;
     private ChannelFactory<IClient> clientChannelFactory;
-
-    private event EventHandler ResultRecieved;
-
     public int WaitingJobs {
       get {
@@ -62 +60 @@
       runningEngines = new Dictionary<Guid, byte[]>();
       runningClients = new Dictionary<Guid, string>();
+      waitHandles = new Dictionary<Guid, ManualResetEvent>();
       results = new Dictionary<Guid, byte[]>();
       bigLock = new object();
@@ -98 +97 @@
         runningClients.Remove(guid);
         results[guid] = result;
-        OnResultRecieved(guid);
+        waitHandles[guid].Set();
       }
     }
@@ -106 +105 @@
         engineQueue.Enqueue(guid);
         waitingEngines.Add(guid, engine);
+        waitHandles.Add(guid, new ManualResetEvent(false));
       }
     }
@@ -118 +118 @@
 
     internal byte[] GetResult(Guid guid) {
-      ManualResetEvent waitHandle = new ManualResetEvent(false);
+      return GetResult(guid, System.Threading.Timeout.Infinite);
+    }
+    internal byte[] GetResult(Guid guid, int timeout) {
       lock(bigLock) {
-        if(results.ContainsKey(guid)) {
-          byte[] result = results[guid];
-          results.Remove(guid);
-          return result;
+        if(waitHandles.ContainsKey(guid)) {
+          ManualResetEvent waitHandle = waitHandles[guid];
+          if(waitHandle.WaitOne(timeout, false)) {
+            waitHandle.Close();
+            waitHandles.Remove(guid);
+            byte[] result = results[guid];
+            results.Remove(guid);
+            return result;
+          } else {
+            return null;
+          }
         } else {
-          ResultRecieved += delegate(object source, EventArgs args) {
-            ResultRecievedEventArgs resultArgs = (ResultRecievedEventArgs)args;
-            if(resultArgs.resultGuid == guid) {
-              waitHandle.Set();
-            }
-          };
+          return null;
         }
-      }
-
-      waitHandle.WaitOne();
-      waitHandle.Close();
-
-      lock(bigLock) {
-        byte[] result = results[guid];
-        results.Remove(guid);
-        return result;
       }
     }
@@ -157 +152 @@
       }
     }
-
-    private void OnResultRecieved(Guid guid) {
-      ResultRecievedEventArgs args = new ResultRecievedEventArgs();
-      args.resultGuid = guid;
-      if(ResultRecieved != null) {
-        ResultRecieved(this, args);
-      }
-    }
-
-    private class ResultRecievedEventArgs : EventArgs {
-      public Guid resultGuid;
-    }
   }
 }

trunk/sources/HeuristicLab.Grid/GridServer.cs

@@ -44 +44 @@
       return engineStore.GetResult(guid);
     }
+    public byte[] TryEndExecuteEngine(Guid guid, int timeout) {
+      return engineStore.GetResult(guid, timeout);
+    }
 
     public void AbortEngine(Guid engine) {

trunk/sources/HeuristicLab.Grid/IGridServer.cs

@@ -30 +30 @@
     [OperationContract]
     Guid BeginExecuteEngine(byte[] engine);
-    
     [OperationContract]
     byte[] EndExecuteEngine(Guid engineGuid);
-
+    [OperationContract]
+    byte[] TryEndExecuteEngine(Guid engineGuid, int timeout);
     [OperationContract]
     void AbortEngine(Guid engineGuid);