Changeset 2055

Timestamp:

06/17/09 18:07:15 (16 years ago)

Author:

gkronber

Message:

Refactored JobManager and DistributedEngine to fix bugs in the GridExecuter. #644.

Location:

trunk/sources

Files:

• HeuristicLab.CEDMA.Server/3.3/GridExecuter.cs (modified) (4 diffs)
• HeuristicLab.DistributedEngine/3.2/DistributedEngine.cs (modified) (20 diffs)
• HeuristicLab.Grid/3.2/AsyncGridResult.cs (added)
• HeuristicLab.Grid/3.2/HeuristicLab.Grid-3.2.csproj (modified) (1 diff)
• HeuristicLab.Grid/3.2/IGridServer.cs (modified) (1 diff)
• HeuristicLab.Grid/3.2/JobManager.cs (modified) (8 diffs)
• HeuristicLab.SupportVectorMachines/3.2/SupportVectorRegression.cs (modified) (1 diff)

trunk/sources/HeuristicLab.CEDMA.Server/3.3/GridExecuter.cs

@@ -43 +43 @@
   public class GridExecuter : ExecuterBase {
     private JobManager jobManager;
-    private Dictionary<WaitHandle, IAlgorithm> activeAlgorithms;
+    private Dictionary<AsyncGridResult, IAlgorithm> activeAlgorithms;
 
     private TimeSpan StartJobInterval {
@@ -56 +56 @@
       : base(dispatcher, store) {
       this.jobManager = new JobManager(gridUrl);
-      activeAlgorithms = new Dictionary<WaitHandle, IAlgorithm>();
+      activeAlgorithms = new Dictionary<AsyncGridResult, IAlgorithm>();
       jobManager.Reset();
     }
 
     protected override void StartJobs() {
-      List<WaitHandle> wh = new List<WaitHandle>();
-      Dictionary<WaitHandle, AtomicOperation> activeOperations = new Dictionary<WaitHandle, AtomicOperation>();
+      Dictionary<WaitHandle, AsyncGridResult> asyncResults = new Dictionary<WaitHandle,AsyncGridResult>();
       while (true) {
         try {
           // start new jobs as long as there are less than MaxActiveJobs 
-          while (wh.Count < MaxActiveJobs) {
+          while (asyncResults.Count < MaxActiveJobs) {
             Thread.Sleep(StartJobInterval);
             // get an execution from the dispatcher and execute in grid via job-manager
@@ -72 +71 @@
             if (algorithm != null) {
               AtomicOperation op = new AtomicOperation(algorithm.Engine.OperatorGraph.InitialOperator, algorithm.Engine.GlobalScope);
-              WaitHandle opWh = jobManager.BeginExecuteOperation(algorithm.Engine.GlobalScope, op);
-              wh.Add(opWh);
-              activeOperations.Add(opWh, op);
+              AsyncGridResult asyncResult = jobManager.BeginExecuteEngine(new ProcessingEngine(algorithm.Engine.GlobalScope, op));
+              asyncResults.Add(asyncResult.WaitHandle, asyncResult);
               lock (activeAlgorithms) {
-                activeAlgorithms.Add(opWh, algorithm);
+                activeAlgorithms.Add(asyncResult, algorithm);
               }
             }
           }
           // wait until any job is finished
-          WaitHandle[] whArr = wh.ToArray();
+          WaitHandle[] whArr = asyncResults.Keys.ToArray();
           int readyHandleIndex = WaitHandle.WaitAny(whArr, WaitForFinishedJobsTimeout);
           if (readyHandleIndex != WaitHandle.WaitTimeout) {
             WaitHandle readyHandle = whArr[readyHandleIndex];
-            AtomicOperation finishedOp = activeOperations[readyHandle];
-            wh.Remove(readyHandle);
             IAlgorithm finishedAlgorithm = null;
+            AsyncGridResult finishedResult = null;
             lock (activeAlgorithms) {
-              finishedAlgorithm = activeAlgorithms[readyHandle];
-              activeAlgorithms.Remove(readyHandle);
+              finishedResult = asyncResults[readyHandle];
+              finishedAlgorithm = activeAlgorithms[finishedResult];
+              activeAlgorithms.Remove(finishedResult);
+              asyncResults.Remove(readyHandle);
             }
-            activeOperations.Remove(readyHandle);
-            readyHandle.Close();
             try {
-              ProcessingEngine finishedEngine = jobManager.EndExecuteOperation(finishedOp);
+              IEngine finishedEngine = jobManager.EndExecuteEngine(finishedResult);
+              SetResults(finishedEngine.GlobalScope, finishedAlgorithm.Engine.GlobalScope);
               StoreResults(finishedAlgorithm);
             }
@@ -106 +104 @@
           Trace.WriteLine("CEDMA Executer: Exception in job-management thread. " + ex.Message);
         }
+      }
+    }
+
+    private void SetResults(IScope src, IScope target) {
+      foreach (IVariable v in src.Variables) {
+        target.AddVariable(v);
+      }
+      foreach (IScope subScope in src.SubScopes) {
+        target.AddSubScope(subScope);
+      }
+      foreach (KeyValuePair<string, string> alias in src.Aliases) {
+        target.AddAlias(alias.Key, alias.Value);
       }
     }

trunk/sources/HeuristicLab.DistributedEngine/3.2/DistributedEngine.cs

@@ -42 +42 @@
       get { return serverAddress; }
       set {
-        if(value != serverAddress) {
+        if (value != serverAddress) {
           serverAddress = value;
         }
@@ -72 +72 @@
 
     public override void Execute() {
-      if(jobManager == null) this.jobManager = new JobManager(serverAddress);
+      if (jobManager == null) this.jobManager = new JobManager(serverAddress);
       jobManager.Reset();
       base.Execute();
@@ -82 +82 @@
 
     protected override void ProcessNextOperation() {
-      if(suspendedEngines.Count > 0) {
+      if (suspendedEngines.Count > 0) {
         ProcessSuspendedEngines();
       } else {
@@ -91 +91 @@
 
     private void ProcessSuspendedEngines() {
-      WaitHandle[] waitHandles = new WaitHandle[suspendedEngines.Count];
+      AsyncGridResult[] asyncResults = new AsyncGridResult[suspendedEngines.Count];
       int i = 0;
-      foreach(KeyValuePair<ProcessingEngine, AtomicOperation> suspendedPair in suspendedEngines) {
-        waitHandles[i++] = jobManager.BeginExecuteEngine(suspendedPair.Key);
-      }
-      WaitForAll(waitHandles);
+      foreach (KeyValuePair<ProcessingEngine, AtomicOperation> suspendedPair in suspendedEngines) {
+        asyncResults[i++] = jobManager.BeginExecuteEngine(suspendedPair.Key);
+      }
+      WaitForAll(asyncResults);
       // collect results
       List<KeyValuePair<ProcessingEngine, AtomicOperation>> results = new List<KeyValuePair<ProcessingEngine, AtomicOperation>>();
       try {
-        foreach(KeyValuePair<ProcessingEngine, AtomicOperation> suspendedPair in suspendedEngines) {
+        int resultIndex = 0;
+        foreach (KeyValuePair<ProcessingEngine, AtomicOperation> suspendedPair in suspendedEngines) {
           KeyValuePair<ProcessingEngine, AtomicOperation> p = new KeyValuePair<ProcessingEngine, AtomicOperation>(
-              jobManager.EndExecuteOperation(suspendedPair.Value),
+              (ProcessingEngine)jobManager.EndExecuteEngine(asyncResults[resultIndex++]),
               suspendedPair.Value);
           results.Add(p);
         }
-      } catch(Exception e) {
+      }
+      catch (Exception e) {
         // this exception means there was a problem with the underlying communication infrastructure
         // -> show message dialog and abort engine
@@ -118 +120 @@
 
     private void ProcessOperation(IOperation operation) {
-      if(operation is AtomicOperation) {
+      if (operation is AtomicOperation) {
         AtomicOperation atomicOperation = (AtomicOperation)operation;
         IOperation next = null;
         try {
           next = atomicOperation.Operator.Execute(atomicOperation.Scope);
-        } catch(Exception ex) {
+        }
+        catch (Exception ex) {
           // push operation on stack again
           myExecutionStack.Push(atomicOperation);
@@ -129 +132 @@
           ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(ex); });
         }
-        if(next != null)
+        if (next != null)
           myExecutionStack.Push(next);
         OnOperationExecuted(atomicOperation);
-        if(atomicOperation.Operator.Breakpoint) Abort();
-      } else if(operation is CompositeOperation) {
+        if (atomicOperation.Operator.Breakpoint) Abort();
+      } else if (operation is CompositeOperation) {
         CompositeOperation compositeOperation = (CompositeOperation)operation;
-        if(compositeOperation.ExecuteInParallel) {
+        if (compositeOperation.ExecuteInParallel) {
           ProcessParallelOperation(compositeOperation);
           OnOperationExecuted(compositeOperation);
         } else {
-          for(int i = compositeOperation.Operations.Count - 1; i >= 0; i--)
+          for (int i = compositeOperation.Operations.Count - 1; i >= 0; i--)
             myExecutionStack.Push(compositeOperation.Operations[i]);
         }
@@ -147 +150 @@
     private void ProcessParallelOperation(CompositeOperation compositeOperation) {
       // send operations to grid
-      WaitHandle[] waithandles = BeginExecuteOperations(compositeOperation);
-      WaitForAll(waithandles);
+      AsyncGridResult[] asyncResults = BeginExecuteOperations(compositeOperation);
+      WaitForAll(asyncResults);
       // 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) {
+        int i = 0;
+        foreach (AtomicOperation parOperation in compositeOperation.Operations) {
+          results.Add(new KeyValuePair<ProcessingEngine, AtomicOperation>(
+              (ProcessingEngine)jobManager.EndExecuteEngine(asyncResults[i++]), parOperation));
+        }
+      }
+      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
@@ -169 +173 @@
     }
 
-    private WaitHandle[] BeginExecuteOperations(CompositeOperation compositeOperation) {
-      WaitHandle[] waithandles = new WaitHandle[compositeOperation.Operations.Count];
+    private AsyncGridResult[] BeginExecuteOperations(CompositeOperation compositeOperation) {
+      AsyncGridResult[] asyncResults = new AsyncGridResult[compositeOperation.Operations.Count];
       int i = 0;
       // HACK: assume that all atomicOperations have the same parent scope.
@@ -182 +186 @@
       PruneToParentScope(GlobalScope, parentScope, prunedScopes);
       List<IScope> subScopes = new List<IScope>(parentScope.SubScopes);
-      foreach(IScope scope in subScopes) {
+      foreach (IScope scope in subScopes) {
         parentScope.RemoveSubScope(scope);
       }
       // start all parallel jobs
-      foreach(AtomicOperation parOperation in compositeOperation.Operations) {
+      foreach (AtomicOperation parOperation in compositeOperation.Operations) {
         parentScope.AddSubScope(parOperation.Scope);
-        waithandles[i++] = jobManager.BeginExecuteOperation(GlobalScope, parOperation);
+        asyncResults[i++] = jobManager.BeginExecuteEngine(new ProcessingEngine(GlobalScope, parOperation));
         parentScope.RemoveSubScope(parOperation.Scope);
       }
-      foreach(IScope scope in subScopes) {
+      foreach (IScope scope in subScopes) {
         parentScope.AddSubScope(scope);
       }
@@ -197 +201 @@
       RestoreFullTree(GlobalScope, prunedScopes);
 
-      return waithandles;
-    }
-
-    private void WaitForAll(WaitHandle[] waithandles) {
+      return asyncResults;
+    }
+
+    private void WaitForAll(AsyncGridResult[] asyncResults) {
       // wait until all jobs are finished
       // WaitAll works only with maximally 64 waithandles
-      if(waithandles.Length <= 64) {
-        WaitHandle.WaitAll(waithandles);
+      if (asyncResults.Length <= 64) {
+        WaitHandle[] waitHandles = new WaitHandle[asyncResults.Length];
+        for (int i = 0; i < asyncResults.Length; i++) {
+          waitHandles[i] = asyncResults[i].WaitHandle;
+        }
+        WaitHandle.WaitAll(waitHandles);
       } else {
         int i;
-        for(i = 0; i < waithandles.Length; i++) {
-          waithandles[i].WaitOne();
-          waithandles[i].Close();
+        for (i = 0; i < asyncResults.Length; i++) {
+          asyncResults[i].WaitHandle.WaitOne();
         }
       }
@@ -221 +228 @@
       suspendedEngines.Clear();
       // retrieve results and merge into scope-tree
-      foreach(KeyValuePair<ProcessingEngine, AtomicOperation> p in results) {
+      foreach (KeyValuePair<ProcessingEngine, AtomicOperation> p in results) {
         ProcessingEngine resultEngine = p.Key;
         AtomicOperation parOperation = p.Value;
-        if(resultEngine.Canceled && !resultEngine.Suspended) {
+        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)
@@ -230 +237 @@
           ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(new JobExecutionException(resultEngine.ErrorMessage)); });
           canceledOperations.AddOperation(parOperation);
-        } else if(resultEngine.Suspended) {
+        } 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)
@@ -243 +250 @@
       }
       // if there were exceptions -> abort 
-      if(canceledOperations.Operations.Count > 0) {
+      if (canceledOperations.Operations.Count > 0) {
         // requeue the aborted operations 
         myExecutionStack.Push(canceledOperations);
@@ -249 +256 @@
       }
       // if there were breakpoints -> abort
-      if(suspendedEngines.Count > 0) {
+      if (suspendedEngines.Count > 0) {
         Abort();
       }
@@ -255 +262 @@
 
     private void RestoreFullTree(IScope currentScope, IList<IList<IScope>> savedScopes) {
-      if(savedScopes.Count == 0) return;
+      if (savedScopes.Count == 0) return;
       IScope remainingBranch = currentScope.SubScopes[0];
       currentScope.RemoveSubScope(remainingBranch);
       IList<IScope> savedScopesForCurrent = savedScopes[0];
-      foreach(IScope savedScope in savedScopesForCurrent) {
+      foreach (IScope savedScope in savedScopesForCurrent) {
         currentScope.AddSubScope(savedScope);
       }
@@ -267 +274 @@
 
     private IScope PruneToParentScope(IScope currentScope, IScope scope, IList<IList<IScope>> prunedScopes) {
-      if(currentScope == scope) return currentScope;
-      if(currentScope.SubScopes.Count == 0) return null;
+      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) {
+      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
+        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) {
+        foreach (IScope subScope in subScopes) {
           currentScope.RemoveSubScope(subScope);
         }
@@ -291 +298 @@
 
     private IScope FindParentScope(IScope currentScope, IScope childScope) {
-      if(currentScope.SubScopes.Contains(childScope)) return currentScope;
-      foreach(IScope subScope in currentScope.SubScopes) {
+      if (currentScope.SubScopes.Contains(childScope)) return currentScope;
+      foreach (IScope subScope in currentScope.SubScopes) {
         IScope result = FindParentScope(subScope, childScope);
-        if(result != null) return result;
+        if (result != null) return result;
       }
       return null;
@@ -302 +309 @@
       // merge the results
       original.Clear();
-      foreach(IVariable variable in result.Variables) {
+      foreach (IVariable variable in result.Variables) {
         original.AddVariable(variable);
       }
-      foreach(IScope subScope in result.SubScopes) {
+      foreach (IScope subScope in result.SubScopes) {
         original.AddSubScope(subScope);
       }
-      foreach(KeyValuePair<string, string> alias in result.Aliases) {
+      foreach (KeyValuePair<string, string> alias in result.Aliases) {
         original.AddAlias(alias.Key, alias.Value);
       }
@@ -319 +326 @@
       addressAttribute.Value = ServerAddress;
       node.Attributes.Append(addressAttribute);
-      if(suspendedEngines.Count > 0) {
+      if (suspendedEngines.Count > 0) {
         XmlNode suspendedEnginesNode = document.CreateElement("SuspendedEngines");
-        foreach(KeyValuePair<ProcessingEngine, AtomicOperation> p in suspendedEngines) {
+        foreach (KeyValuePair<ProcessingEngine, AtomicOperation> p in suspendedEngines) {
           XmlNode n = document.CreateElement("Entry");
           n.AppendChild(PersistenceManager.Persist(p.Key, document, persistedObjects));
@@ -335 +342 @@
       ServerAddress = node.Attributes["ServerAddress"].Value;
       XmlNode suspendedEnginesNode = node.SelectSingleNode("SuspendedEngines");
-      if(suspendedEnginesNode != null) {
-        foreach(XmlNode n in suspendedEnginesNode.ChildNodes) {
+      if (suspendedEnginesNode != null) {
+        foreach (XmlNode n in suspendedEnginesNode.ChildNodes) {
           KeyValuePair<ProcessingEngine, AtomicOperation> p = new KeyValuePair<ProcessingEngine, AtomicOperation>(
             (ProcessingEngine)PersistenceManager.Restore(n.ChildNodes[0], restoredObjects),

trunk/sources/HeuristicLab.Grid/3.2/HeuristicLab.Grid-3.2.csproj

@@ -98 +98 @@
       <DependentUpon>ClientForm.cs</DependentUpon>
     </Compile>
+    <Compile Include="AsyncGridResult.cs" />
     <Compile Include="Database.cs" />
     <Compile Include="EngineRunner.cs" />

trunk/sources/HeuristicLab.Grid/3.2/IGridServer.cs

@@ -27 +27 @@
 namespace HeuristicLab.Grid {
   public enum JobState {
-    Unknown,
+    Unknown = 0, // default value
     Waiting,
     Busy,

trunk/sources/HeuristicLab.Grid/3.2/JobManager.cs

@@ -43 +43 @@
     private const int RESULT_POLLING_TIMEOUT = 5;
 
-    private class Job {
-      public Guid guid;
-      public ProcessingEngine engine;
-      public ManualResetEvent waitHandle;
-      public int restarts;
-    }
-
     private IGridServer server;
     private string address;
     private object waitingQueueLock = new object();
-    private Queue<Job> waitingJobs = new Queue<Job>();
+    private Queue<AsyncGridResult> waitingJobs = new Queue<AsyncGridResult>();
     private object runningQueueLock = new object();
-    private Queue<Job> runningJobs = new Queue<Job>();
-    private Dictionary<AtomicOperation, byte[]> results = new Dictionary<AtomicOperation, byte[]>();
-
-    private List<IOperation> erroredOperations = new List<IOperation>();
+    private Queue<AsyncGridResult> runningJobs = new Queue<AsyncGridResult>();
     private object connectionLock = new object();
-    private object dictionaryLock = new object();
 
     private AutoResetEvent runningWaitHandle = new AutoResetEvent(false);
@@ -77 +66 @@
     public void Reset() {
       ResetConnection();
-      lock(dictionaryLock) {
-        foreach(Job j in waitingJobs) {
-          j.waitHandle.Close();
+      lock (waitingQueueLock) {
+        foreach (AsyncGridResult r in waitingJobs) {
+          r.WaitHandle.Close();
         }
         waitingJobs.Clear();
-        foreach(Job j in runningJobs) {
-          j.waitHandle.Close();
+      }
+      lock (runningQueueLock) {
+        foreach (AsyncGridResult r in runningJobs) {
+          r.WaitHandle.Close();
         }
         runningJobs.Clear();
-        results.Clear();
-        erroredOperations.Clear();
       }
     }
@@ -93 +82 @@
     private void ResetConnection() {
       Trace.TraceInformation("Reset connection in JobManager");
-      lock(connectionLock) {
+      lock (connectionLock) {
         // open a new channel
         NetTcpBinding binding = new NetTcpBinding();
@@ -107 +96 @@
     public void StartEngines() {
       try {
-        while(true) {
-          Job job = null;
-          lock(waitingQueueLock) {
-            if(waitingJobs.Count > 0) job = waitingJobs.Dequeue();
-          }
-          if(job==null) waitingWaitHandle.WaitOne(); // no jobs waiting
+        while (true) {
+          AsyncGridResult job = null;
+          lock (waitingQueueLock) {
+            if (waitingJobs.Count > 0) job = waitingJobs.Dequeue();
+          }
+          if (job == null) waitingWaitHandle.WaitOne(); // no jobs waiting
           else {
-            Guid currentEngineGuid = TryStartExecuteEngine(job.engine);
-            if(currentEngineGuid == Guid.Empty) {
+            Guid currentEngineGuid = TryStartExecuteEngine(job.Engine);
+            if (currentEngineGuid == Guid.Empty) {
               // couldn't start the job -> requeue
-              if(job.restarts < MAX_RESTARTS) {
-                job.restarts++;
-                lock(waitingQueueLock) waitingJobs.Enqueue(job);
+              if (job.Restarts < MAX_RESTARTS) {
+                job.Restarts++;
+                lock (waitingQueueLock) waitingJobs.Enqueue(job);
                 waitingWaitHandle.Set();
               } else {
                 // max restart count reached -> give up on this job and flag error
-                lock(dictionaryLock) {
-                  erroredOperations.Add(job.engine.InitialOperation);
-                  job.waitHandle.Set();
-                }
+                job.Aborted = true;
+                job.SignalFinished();
               }
             } else {
               // job started successfully
-              job.guid = currentEngineGuid;
-              lock(runningQueueLock) {
+              job.Guid = currentEngineGuid;
+              lock (runningQueueLock) {
                 runningJobs.Enqueue(job);
                 runningWaitHandle.Set();
@@ -138 +125 @@
           }
         }
-      } catch(Exception e) {
-        Trace.TraceError("Exception "+e+" in JobManager.StartEngines() killed the start-engine thread\n"+e.StackTrace);
+      }
+      catch (Exception e) {
+        Trace.TraceError("Exception " + e + " in JobManager.StartEngines() killed the start-engine thread\n" + e.StackTrace);
       }
     }
@@ -146 +134 @@
     public void GetResults() {
       try {
-        while(true) {
-          Job job = null;
-          lock(runningQueueLock) {
-            if(runningJobs.Count > 0) job = runningJobs.Dequeue();
-          }
-          if(job == null) runningWaitHandle.WaitOne(); // no jobs running
+        while (true) {
+          AsyncGridResult job = null;
+          lock (runningQueueLock) {
+            if (runningJobs.Count > 0) job = runningJobs.Dequeue();
+          }
+          if (job == null) runningWaitHandle.WaitOne(); // no jobs running
           else {
-            byte[] zippedResult = TryEndExecuteEngine(server, job.guid);
-            if(zippedResult != null) { // successful
-              lock(dictionaryLock) {
-                // store result
-                results[job.engine.InitialOperation] = zippedResult;
-                // notify consumer that result is ready
-                job.waitHandle.Set();
-              }
+            byte[] zippedResult = TryEndExecuteEngine(server, job.Guid);
+            if (zippedResult != null) { 
+              // successful => store result
+              job.ZippedResult = zippedResult;
+              // notify consumer that result is ready
+              job.SignalFinished();
             } else {
               // there was a problem -> check the state of the job and restart if necessary
-              JobState jobState = TryGetJobState(server, job.guid);
-              if(jobState == JobState.Unknown) {
-                job.restarts++;
-                lock(waitingQueueLock) {
+              JobState jobState = TryGetJobState(server, job.Guid);
+              if (jobState == JobState.Unknown) {
+                job.Restarts++;
+                lock (waitingQueueLock) {
                   waitingJobs.Enqueue(job);
                   waitingWaitHandle.Set();
@@ -172 +158 @@
               } else {
                 // job still active at the server 
-                lock(runningQueueLock) {
+                lock (runningQueueLock) {
                   runningJobs.Enqueue(job);
                   runningWaitHandle.Set();
@@ -181 +167 @@
           }
         }
-      } catch(Exception e) {
-        Trace.TraceError("Exception " + e + " in JobManager.GetResults() killed the results-gathering thread\n"+ e.StackTrace);
-      }
-    }
-
-    public WaitHandle BeginExecuteOperation(IScope globalScope, AtomicOperation operation) {
-      return BeginExecuteEngine(new ProcessingEngine(globalScope, operation));
-    }
-
-    public WaitHandle BeginExecuteEngine(ProcessingEngine engine) {
-      Job job = new Job();
-      job.engine = engine;
-      job.waitHandle = new ManualResetEvent(false);
-      job.restarts = 0;
-      lock(waitingQueueLock) {
-        waitingJobs.Enqueue(job);
+      }
+      catch (Exception e) {
+        Trace.TraceError("Exception " + e + " in JobManager.GetResults() killed the results-gathering thread\n" + e.StackTrace);
+      }
+    }
+
+    public AsyncGridResult BeginExecuteEngine(ProcessingEngine engine) {
+      AsyncGridResult asyncResult = new AsyncGridResult(engine);
+      asyncResult.Engine = engine;
+      lock (waitingQueueLock) {
+        waitingJobs.Enqueue(asyncResult);
       }
       waitingWaitHandle.Set();
-      return job.waitHandle;
-    }
-
-    private byte[] ZipEngine(ProcessingEngine engine) {
+      return asyncResult;
+    }
+
+    private byte[] ZipEngine(IEngine engine) {
       return PersistenceManager.SaveToGZip(engine);
     }
 
-    public ProcessingEngine EndExecuteOperation(AtomicOperation operation) {
-      if(erroredOperations.Contains(operation)) {
-        erroredOperations.Remove(operation);
+    public IEngine EndExecuteEngine(AsyncGridResult asyncResult) {
+      if (asyncResult.Aborted) {
         throw new JobExecutionException("Maximal number of job restarts reached. There is a problem with the connection to the grid-server.");
       } else {
-        byte[] zippedResult = null;
-        lock(dictionaryLock) {
-          zippedResult = results[operation];
-          results.Remove(operation);
-        }
         // restore the engine 
-        return (ProcessingEngine)PersistenceManager.RestoreFromGZip(zippedResult);
-      }
-    }
-
-    private Guid TryStartExecuteEngine(ProcessingEngine engine) {
+        return (IEngine)PersistenceManager.RestoreFromGZip(asyncResult.ZippedResult);
+      }
+    }
+
+    private Guid TryStartExecuteEngine(IEngine engine) {
       byte[] zippedEngine = ZipEngine(engine);
+      return SavelyExecute(() => server.BeginExecuteEngine(zippedEngine));
+    }
+
+    private byte[] TryEndExecuteEngine(IGridServer server, Guid engineGuid) {
+      return SavelyExecute(() => {
+        byte[] zippedResult = server.TryEndExecuteEngine(engineGuid);
+        return zippedResult;
+      });
+    }
+
+    private JobState TryGetJobState(IGridServer server, Guid engineGuid) {
+      return SavelyExecute(() => server.JobState(engineGuid));
+    }
+
+    private TResult SavelyExecute<TResult>(Func<TResult> a) {
       int retries = 0;
-      Guid guid = Guid.Empty;
       do {
         try {
-          lock(connectionLock) {
-            guid = server.BeginExecuteEngine(zippedEngine);
-          }
-          return guid;
-        } catch(TimeoutException) {
+          lock (connectionLock) {
+            return a();
+          }
+        }
+        catch (TimeoutException) {
           retries++;
           Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
-        } catch(CommunicationException) {
+        }
+        catch (CommunicationException) {
           ResetConnection();
           retries++;
           Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
         }
-      } while(retries < MAX_CONNECTION_RETRIES);
-      Trace.TraceWarning("Reached max connection retries in TryStartExecuteEngine");
-      return Guid.Empty;
-    }
-
-    private byte[] TryEndExecuteEngine(IGridServer server, Guid engineGuid) {
-      int retries = 0;
-      do {
-        try {
-          lock(connectionLock) {
-            byte[] zippedResult = server.TryEndExecuteEngine(engineGuid);
-            return zippedResult;
-          }
-        } catch(TimeoutException) {
-          retries++;
-          Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
-        } catch(CommunicationException) {
-          ResetConnection();
-          retries++;
-          Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
-        }
-      } while(retries < MAX_CONNECTION_RETRIES);
-      Trace.TraceWarning("Reached max connection retries in TryEndExecuteEngine");
-      return null;
-    }
-
-    private JobState TryGetJobState(IGridServer server, Guid engineGuid) {
-      // check if the server is still working on the job
-      int retries = 0;
-      do {
-        try {
-          lock(connectionLock) {
-            JobState jobState = server.JobState(engineGuid);
-            return jobState;
-          }
-        } catch(TimeoutException) {
-          retries++;
-          Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
-        } catch(CommunicationException) {
-          ResetConnection();
-          retries++;
-          Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
-        }
-      } while(retries < MAX_CONNECTION_RETRIES);
-      Trace.TraceWarning("Reached max connection retries in TryGetJobState");
-      return JobState.Unknown;
+      } while (retries < MAX_CONNECTION_RETRIES);
+      Trace.TraceWarning("Reached max connection retries");
+      return default(TResult);
     }
   }

trunk/sources/HeuristicLab.SupportVectorMachines/3.2/SupportVectorRegression.cs

@@ -361 +361 @@
       model.TestVarianceAccountedFor = bestModelScope.GetVariableValue<DoubleData>("TestVAF", false).Data;
 
-      model.Data = bestModelScope.GetVariableValue<SVMModel>("BestValidationModel", false);
+      model.Data = bestModelScope.GetVariableValue<SVMModel>("Model", false);
       HeuristicLab.DataAnalysis.Dataset ds = bestModelScope.GetVariableValue<Dataset>("Dataset", true);
       model.Dataset = ds;