Index: branches/3.0/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs
===================================================================
--- branches/3.0/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs	(revision 134)
+++ branches/3.0/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs	(revision 279)
@@ -30,13 +30,13 @@
 using System.IO;
 using System.IO.Compression;
+using HeuristicLab.PluginInfrastructure;
+using System.Windows.Forms;
+using System.Diagnostics;
 
 namespace HeuristicLab.DistributedEngine {
   public class DistributedEngine : EngineBase, IEditable {
-    private IGridServer server;
-    private Dictionary<Guid, AtomicOperation> engineOperations = new Dictionary<Guid, AtomicOperation>();
-    private List<Guid> runningEngines = new List<Guid>();
+    private JobManager jobManager;
+    private CompositeOperation waitingOperations;
     private string serverAddress;
-    private bool cancelRequested;
-    private CompositeOperation waitingOperations;
     public string ServerAddress {
       get { return serverAddress; }
@@ -47,9 +47,4 @@
       }
     }
-    public override bool Terminated {
-      get {
-        return myExecutionStack.Count == 0 && runningEngines.Count == 0 && waitingOperations==null;
-      }
-    }
     public override object Clone(IDictionary<Guid, object> clonedObjects) {
       DistributedEngine clone = (DistributedEngine)base.Clone(clonedObjects);
@@ -66,12 +61,6 @@
 
     public override void Execute() {
-      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;
-      ChannelFactory<IGridServer> factory = new ChannelFactory<IGridServer>(binding);
-      server = factory.CreateChannel(new EndpointAddress(serverAddress));
-
+      if(jobManager == null) this.jobManager = new JobManager(serverAddress);
+      jobManager.Reset();
       base.Execute();
     }
@@ -81,100 +70,136 @@
     }
 
-    public override void Abort() {
-      lock(runningEngines) {
-        cancelRequested = true;
-        foreach(Guid engineGuid in runningEngines) {
-          server.AbortEngine(engineGuid);
-        }
-      }
-    }
-    public override void Reset() {
-      base.Reset();
-      engineOperations.Clear();
-      runningEngines.Clear();
-      cancelRequested = false;
-    }
-
     protected override void ProcessNextOperation() {
-      lock(runningEngines) {
-        if(runningEngines.Count == 0 && cancelRequested) {
-          base.Abort();
-          cancelRequested = false;
-          if(waitingOperations != null && waitingOperations.Operations.Count != 0) {
-            myExecutionStack.Push(waitingOperations);
-            waitingOperations = null;
+      IOperation operation = myExecutionStack.Pop();
+      if(operation is AtomicOperation) {
+        AtomicOperation atomicOperation = (AtomicOperation)operation;
+        IOperation next = null;
+        try {
+          next = atomicOperation.Operator.Execute(atomicOperation.Scope);
+        } catch(Exception ex) {
+          // push operation on stack again
+          myExecutionStack.Push(atomicOperation);
+          Abort();
+          ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(ex); });
+        }
+        if(next != null)
+          myExecutionStack.Push(next);
+        OnOperationExecuted(atomicOperation);
+        if(atomicOperation.Operator.Breakpoint) Abort();
+      } else if(operation is CompositeOperation) {
+        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, compositeOperation);
+            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();
+              }
+            }
+            // 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(e); });
           }
-          return;
-        }
-        if(runningEngines.Count != 0) {
-          Guid engineGuid = runningEngines[0];
-          byte[] resultXml = server.TryEndExecuteEngine(engineGuid, 100);
-          if(resultXml != null) {
-            GZipStream stream = new GZipStream(new MemoryStream(resultXml), CompressionMode.Decompress);
-            ProcessingEngine resultEngine = (ProcessingEngine)PersistenceManager.Load(stream);
-            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);
-            }
-            OnOperationExecuted(engineOperations[engineGuid]);
-
-            if(cancelRequested & resultEngine.ExecutionStack.Count != 0) {
-              if(waitingOperations == null) {
-                waitingOperations = new CompositeOperation();
-                waitingOperations.ExecuteInParallel = false;
-              }
-              CompositeOperation task = new CompositeOperation();
-              while(resultEngine.ExecutionStack.Count > 0) {
-                AtomicOperation oldOperation = (AtomicOperation)resultEngine.ExecutionStack.Pop();
-                if(oldOperation.Scope == resultEngine.InitialOperation.Scope) {
-                  oldOperation = new AtomicOperation(oldOperation.Operator, oldScope);
-                }
-                task.AddOperation(oldOperation);
-              }
-              waitingOperations.AddOperation(task);
-            }
-            runningEngines.Remove(engineGuid);
-            engineOperations.Remove(engineGuid);
-          }
-          return;
-        }
-        IOperation operation = myExecutionStack.Pop();
-        if(operation is AtomicOperation) {
-          AtomicOperation atomicOperation = (AtomicOperation)operation;
-          IOperation next = null;
-          try {
-            next = atomicOperation.Operator.Execute(atomicOperation.Scope);
-          } catch(Exception ex) {
-            // push operation on stack again
-            myExecutionStack.Push(atomicOperation);
-            Abort();
-            ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(ex); });
-          }
-          if(next != null)
-            myExecutionStack.Push(next);
-          OnOperationExecuted(atomicOperation);
-          if(atomicOperation.Operator.Breakpoint) Abort();
-        } else if(operation is CompositeOperation) {
-          CompositeOperation compositeOperation = (CompositeOperation)operation;
-          if(compositeOperation.ExecuteInParallel) {
-            foreach(AtomicOperation parOperation in compositeOperation.Operations) {
-              ProcessingEngine engine = new ProcessingEngine(OperatorGraph, GlobalScope, parOperation); // OperatorGraph not needed?
-              MemoryStream memStream = new MemoryStream();
-              GZipStream stream = new GZipStream(memStream, CompressionMode.Compress, true);
-              PersistenceManager.Save(engine, stream);
-              stream.Close();
-              Guid currentEngineGuid = server.BeginExecuteEngine(memStream.ToArray());
-              runningEngines.Add(currentEngineGuid);
-              engineOperations[currentEngineGuid] = parOperation;
-            }
-          } else {
-            for(int i = compositeOperation.Operations.Count - 1; i >= 0; i--)
-              myExecutionStack.Push(compositeOperation.Operations[i]);
-          }
-        }
+        } else {
+          for(int i = compositeOperation.Operations.Count - 1; i >= 0; i--)
+            myExecutionStack.Push(compositeOperation.Operations[i]);
+        }
+      }
+    }
+
+    private void RestoreFullTree(IScope currentScope, IList<IList<IScope>> savedScopes) {
+      if(savedScopes.Count == 0) return;
+      IScope remainingBranch = currentScope.SubScopes[0];
+      currentScope.RemoveSubScope(remainingBranch);
+      IList<IScope> savedScopesForCurrent = savedScopes[0];
+      foreach(IScope savedScope in savedScopesForCurrent) {
+        currentScope.AddSubScope(savedScope);
+      }
+      savedScopes.RemoveAt(0);
+      RestoreFullTree(remainingBranch, savedScopes);
+    }
+
+    private IScope PruneToParentScope(IScope currentScope, IScope scope, IList<IList<IScope>> prunedScopes) {
+      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) {
+        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
+        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) { 
+          currentScope.RemoveSubScope(subScope);
+        }
+        // add only the branch that leads to the scope that I search for
+        currentScope.AddSubScope(foundScope);
+        return currentScope; // return that this scope contains the branch that leads to the searched scopes
+      } else {
+        return null; // otherwise we didn't find the searched scope and we can return null
+      }
+    }
+
+    private IScope FindParentScope(IScope currentScope, CompositeOperation compositeOperation) {
+      AtomicOperation currentOperation = (AtomicOperation)compositeOperation.Operations[0];
+      if(currentScope.SubScopes.Contains(currentOperation.Scope)) return currentScope;
+      foreach(IScope subScope in currentScope.SubScopes) {
+        IScope result = FindParentScope(subScope, compositeOperation);
+        if(result != null) return result;
+      }
+      return null;
+    }
+
+    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);
       }
     }
Index: branches/3.0/sources/HeuristicLab.DistributedEngine/HeuristicLab.DistributedEngine.csproj
===================================================================
--- branches/3.0/sources/HeuristicLab.DistributedEngine/HeuristicLab.DistributedEngine.csproj	(revision 134)
+++ branches/3.0/sources/HeuristicLab.DistributedEngine/HeuristicLab.DistributedEngine.csproj	(revision 279)
@@ -50,4 +50,5 @@
   <ItemGroup>
     <Compile Include="HeuristicLabDistributedEnginePlugin.cs" />
+    <Compile Include="JobManager.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="DistributedEngine.cs" />
Index: branches/3.0/sources/HeuristicLab.DistributedEngine/JobManager.cs
===================================================================
--- branches/3.0/sources/HeuristicLab.DistributedEngine/JobManager.cs	(revision 279)
+++ branches/3.0/sources/HeuristicLab.DistributedEngine/JobManager.cs	(revision 279)
@@ -0,0 +1,236 @@
+﻿#region License Information
+/* HeuristicLab
+ * Copyright (C) 2002-2008 Heuristic and Evolutionary Algorithms Laboratory (HEAL)
+ *
+ * This file is part of HeuristicLab.
+ *
+ * HeuristicLab is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * HeuristicLab is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with HeuristicLab. If not, see <http://www.gnu.org/licenses/>.
+ */
+#endregion
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.ServiceModel;
+using HeuristicLab.Grid;
+using System.Threading;
+using HeuristicLab.Core;
+using System.IO;
+using System.IO.Compression;
+using System.Windows.Forms;
+
+namespace HeuristicLab.DistributedEngine {
+  class JobManager {
+    private IGridServer server;
+    private string address;
+    private Dictionary<Guid, ProcessingEngine> engines = new Dictionary<Guid, ProcessingEngine>();
+    private Dictionary<Guid, ManualResetEvent> waithandles = new Dictionary<Guid, ManualResetEvent>();
+    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 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 JobManager(string address) {
+      this.address = address;
+    }
+
+    internal void Reset() {
+      ResetConnection();
+      lock(dictionaryLock) {
+        foreach(WaitHandle wh in waithandles.Values) wh.Close();
+        waithandles.Clear();
+        engines.Clear();
+        results.Clear();
+      }
+    }
+
+    private void ResetConnection() {
+      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));
+      }
+    }
+
+    public WaitHandle BeginExecuteOperation(IScope globalScope, AtomicOperation operation) {
+      ProcessingEngine engine = new ProcessingEngine(globalScope, operation);
+      byte[] zippedEngine = ZipEngine(engine);
+      Guid currentEngineGuid = Guid.Empty;
+      bool success = false;
+      int retryCount = 0;
+      do {
+        lock(connectionLock) {
+          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) {
+            ResetConnection();
+            // 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) {
+        engines[currentEngineGuid] = engine;
+        waithandles[currentEngineGuid] = new ManualResetEvent(false);
+      }
+      ThreadPool.QueueUserWorkItem(new WaitCallback(TryGetResult), currentEngineGuid);
+      return waithandles[currentEngineGuid];
+    }
+
+    private byte[] ZipEngine(ProcessingEngine engine) {
+      MemoryStream memStream = new MemoryStream();
+      GZipStream stream = new GZipStream(memStream, CompressionMode.Compress, true);
+      PersistenceManager.Save(engine, stream);
+      stream.Close();
+      byte[] zippedEngine = memStream.ToArray();
+      memStream.Close();
+      return zippedEngine;
+    }
+
+    public IScope EndExecuteOperation(AtomicOperation operation) {
+      byte[] zippedResult = null;
+      lock(dictionaryLock) {
+        zippedResult = results[operation];
+        results.Remove(operation);
+      }
+      // restore the engine 
+      using(GZipStream stream = new GZipStream(new MemoryStream(zippedResult), CompressionMode.Decompress)) {
+        ProcessingEngine resultEngine = (ProcessingEngine)PersistenceManager.Load(stream);
+        return resultEngine.InitialOperation.Scope;
+      }      
+    }
+
+    private void TryGetResult(object state) {
+      Guid engineGuid = (Guid)state;
+      int restartCounter = 0;
+      do {
+        Thread.Sleep(TimeSpan.FromSeconds(CHECK_RESULTS_TIMEOUT));
+        byte[] zippedResult = null;
+        lock(connectionLock) {
+          bool success = false;
+          int retries = 0;
+          do {
+            try {
+              zippedResult = server.TryEndExecuteEngine(engineGuid, 100);
+              success = true;
+            } catch(TimeoutException timeoutException) {
+              success = false;
+              retries++;
+              Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+            } catch(CommunicationException communicationException) {
+              ResetConnection();
+              success = false;
+              retries++;
+              Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+            }
+
+          } while(!success && retries < MAX_CONNECTION_RETRIES);
+        }
+        if(zippedResult != null) {
+          lock(dictionaryLock) {
+            // store result
+            results[engines[engineGuid].InitialOperation] = zippedResult;
+
+            // signal the wait handle and clean up then return
+            engines.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) {
+                jobState = server.JobState(engineGuid);
+              }
+              success = true;
+            } catch(TimeoutException timeoutException) {
+              retries++;
+              success = false;
+              Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+            } catch(CommunicationException communicationException) {
+              ResetConnection();
+              retries++;
+              success = false;
+              Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+            }
+          } while(!success && retries < MAX_CONNECTION_RETRIES);
+          if(jobState == JobState.Unkown) {
+            // restart job
+            ProcessingEngine engine;
+            lock(dictionaryLock) {
+              engine = engines[engineGuid];
+            }
+            byte[] zippedEngine = ZipEngine(engine);
+            success = false;
+            retries = 0;
+            do {
+              try {
+                lock(connectionLock) {
+                  server.BeginExecuteEngine(zippedEngine);
+                }
+                success = true;
+              } catch(TimeoutException timeoutException) {
+                success = false;
+                retries++;
+                Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+              } catch(CommunicationException communicationException) {
+                ResetConnection();
+                success = false;
+                retries++;
+                Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+              }
+            } while(!success && retries < MAX_CONNECTION_RETRIES);
+            restartCounter++;
+          }
+        }
+
+        // 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) {
+          throw new ApplicationException("Maximum number of job restarts reached.");
+        }
+      } while(true);
+    }
+  }
+}
