Index: trunk/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs
===================================================================
--- trunk/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs	(revision 91)
+++ trunk/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs	(revision 219)
@@ -30,13 +30,12 @@
 using System.IO;
 using System.IO.Compression;
+using HeuristicLab.PluginInfrastructure;
+using System.Windows.Forms;
 
 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; }
@@ -45,9 +44,4 @@
           serverAddress = value;
         }
-      }
-    }
-    public override bool Terminated {
-      get {
-        return myExecutionStack.Count == 0 && runningEngines.Count == 0 && waitingOperations==null;
       }
     }
@@ -66,12 +60,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,99 +69,38 @@
     }
 
-    public override void Abort() {
-      lock(runningEngines) {
-        cancelRequested = true;
-        foreach(Guid engineGuid in runningEngines) {
-          server.AbortEngine(engineGuid);
+    protected override void ProcessNextOperation() {
+      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); });
         }
-      }
-    }
-    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;
+        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) {
+          WaitHandle[] waithandles = new WaitHandle[compositeOperation.Operations.Count];
+          int i = 0;
+          foreach(AtomicOperation parOperation in compositeOperation.Operations) {
+            waithandles[i++] = jobManager.BeginExecuteOperation(OperatorGraph, GlobalScope, parOperation);
           }
-          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);
+          WaitHandle.WaitAll(waithandles);
+          if(jobManager.Exception != null) {
+            myExecutionStack.Push(compositeOperation);
+            Abort();
+            ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(jobManager.Exception); });
           }
-          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]);
         }
       }
Index: trunk/sources/HeuristicLab.DistributedEngine/HeuristicLab.DistributedEngine.csproj
===================================================================
--- trunk/sources/HeuristicLab.DistributedEngine/HeuristicLab.DistributedEngine.csproj	(revision 91)
+++ trunk/sources/HeuristicLab.DistributedEngine/HeuristicLab.DistributedEngine.csproj	(revision 219)
@@ -50,4 +50,5 @@
   <ItemGroup>
     <Compile Include="HeuristicLabDistributedEnginePlugin.cs" />
+    <Compile Include="JobManager.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="DistributedEngine.cs" />
Index: trunk/sources/HeuristicLab.DistributedEngine/JobManager.cs
===================================================================
--- trunk/sources/HeuristicLab.DistributedEngine/JobManager.cs	(revision 219)
+++ trunk/sources/HeuristicLab.DistributedEngine/JobManager.cs	(revision 219)
@@ -0,0 +1,129 @@
+﻿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, AtomicOperation> engineOperations = new Dictionary<Guid, AtomicOperation>();
+    private Dictionary<Guid, byte[]> runningEngines = new Dictionary<Guid, byte[]>();
+    private Dictionary<Guid, ManualResetEvent> waithandles = new Dictionary<Guid, ManualResetEvent>();
+    private object locker = new object();
+    private const int MAX_RESTARTS = 5;
+    private Exception exception;
+    public Exception Exception {
+      get { return exception; }
+    }
+
+    public JobManager(string address) {
+      this.address = address;
+    }
+
+    internal void Reset() {
+      lock(locker) {
+        // 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;
+        ChannelFactory<IGridServer> factory = new ChannelFactory<IGridServer>(binding);
+        server = factory.CreateChannel(new EndpointAddress(address));
+
+        foreach(WaitHandle wh in waithandles.Values) wh.Close();
+        waithandles.Clear();
+        engineOperations.Clear();
+        runningEngines.Clear();
+        exception = null;
+      }
+    }
+
+    public WaitHandle BeginExecuteOperation(IOperatorGraph operatorGraph, IScope globalScope, AtomicOperation operation) {
+      ProcessingEngine engine = new ProcessingEngine(operatorGraph, globalScope, operation); // OperatorGraph not needed?
+      MemoryStream memStream = new MemoryStream();
+      GZipStream stream = new GZipStream(memStream, CompressionMode.Compress, true);
+      PersistenceManager.Save(engine, stream); // Careful! Make sure that persistence is thread-safe!
+      stream.Close();
+      Guid currentEngineGuid = server.BeginExecuteEngine(memStream.ToArray());
+      lock(locker) {
+        runningEngines[currentEngineGuid] = memStream.ToArray();
+        engineOperations[currentEngineGuid] = operation;
+        waithandles[currentEngineGuid] = new ManualResetEvent(false);
+        ThreadPool.QueueUserWorkItem(new WaitCallback(TryGetResult), currentEngineGuid);
+      }
+      return waithandles[currentEngineGuid];
+    }
+
+    private void TryGetResult(object state) {
+      Guid engineGuid = (Guid)state;
+      int restartCounter = 0;
+      do {
+        try {
+          byte[] resultXml = server.TryEndExecuteEngine(engineGuid, 100);
+          if(resultXml != null) {
+            // restore the engine 
+            GZipStream stream = new GZipStream(new MemoryStream(resultXml), CompressionMode.Decompress);
+            ProcessingEngine resultEngine = (ProcessingEngine)PersistenceManager.Load(stream);
+
+            // merge the results
+            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);
+            }
+            foreach(KeyValuePair<string, string> alias in resultEngine.InitialOperation.Scope.Aliases) {
+              oldScope.AddAlias(alias.Key, alias.Value);
+            }
+
+            lock(locker) {
+              // signal the wait handle and clean up then return
+              waithandles[engineGuid].Set();
+              engineOperations.Remove(engineGuid);
+              waithandles.Remove(engineGuid);
+              runningEngines.Remove(engineGuid);
+            }
+            return;
+          } else {
+            // check if the server is still working on the job
+            JobState jobState = server.JobState(engineGuid);
+            if(jobState == JobState.Unkown) {
+              // restart job
+              byte[] packedEngine;
+              lock(locker) {
+                packedEngine = runningEngines[engineGuid];
+              }
+              server.BeginExecuteEngine(packedEngine);
+              restartCounter++;
+            }
+          }
+        } catch(Exception e) {
+          // catch all exceptions set an exception flag, signal the wait-handle and exit the routine
+          this.exception = new Exception("There was a problem with parallel execution", e);
+          waithandles[engineGuid].Set();
+          return;
+        }
+
+        // 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) {
+          this.exception = new Exception("Maximal number of parallel job restarts reached");
+          waithandles[engineGuid].Set();
+          return;
+        }
+
+        Thread.Sleep(TimeSpan.FromSeconds(10.0));
+      } while(true);
+    }
+  }
+}
