Changeset 219

Timestamp:

05/06/08 12:22:58 (17 years ago)

Author:

gkronber

Message:

• extracted communication code out of the DistributedEngine into class JobManager
• implemented a method to retrieve the JobState for a given job (specified by it's guid) in GridServer
• implemented restarting of jobs in JobManager
• improved exception handling in JobManager and DistributedEngine

(ticket #136)

Location:

trunk/sources

Files:

• HeuristicLab.DistributedEngine/DistributedEngine.cs (modified) (4 diffs)
• HeuristicLab.DistributedEngine/HeuristicLab.DistributedEngine.csproj (modified) (1 diff)
• HeuristicLab.DistributedEngine/JobManager.cs (added)
• HeuristicLab.Grid/ClientForm.cs (modified) (5 diffs)
• HeuristicLab.Grid/EngineStore.cs (modified) (9 diffs)
• HeuristicLab.Grid/GridServer.cs (modified) (1 diff)
• HeuristicLab.Grid/IEngineStore.cs (modified) (1 diff)
• HeuristicLab.Grid/IGridServer.cs (modified) (1 diff)
• HeuristicLab.Grid/ProcessingEngine.cs (modified) (1 diff)

trunk/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs

@@ -30 +30 @@
 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 +44 @@
           serverAddress = value;
         }
-      }
-    }
-    public override bool Terminated {
-      get {
-        return myExecutionStack.Count == 0 && runningEngines.Count == 0 && waitingOperations==null;
       }
     }
@@ -66 +60 @@
 
     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 +69 @@
     }
 
-    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]);
         }
       }

trunk/sources/HeuristicLab.DistributedEngine/HeuristicLab.DistributedEngine.csproj

@@ -50 +50 @@
   <ItemGroup>
     <Compile Include="HeuristicLabDistributedEnginePlugin.cs" />
+    <Compile Include="JobManager.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="DistributedEngine.cs" />

trunk/sources/HeuristicLab.Grid/ClientForm.cs

@@ -46 +46 @@
     private ProcessingEngine currentEngine;
     private string clientUrl;
-    private object locker = new object();
 
     public ClientForm() {
@@ -62 +61 @@
 
       // windows XP returns the external ip on index 0 while windows vista returns the external ip on index 2
-      if (System.Environment.OSVersion.Version.Major >= 6) {
-        clientUrl = "net.tcp://" + Dns.GetHostAddresses(Dns.GetHostName())[2] + ":" + clientPort.Text +"/Grid/Client";
+      if(System.Environment.OSVersion.Version.Major >= 6) {
+        clientUrl = "net.tcp://" + Dns.GetHostAddresses(Dns.GetHostName())[2] + ":" + clientPort.Text + "/Grid/Client";
       } else {
-        clientUrl = "net.tcp://" + Dns.GetHostAddresses(Dns.GetHostName())[0] + ":" + clientPort.Text +"/Grid/Client";
+        clientUrl = "net.tcp://" + Dns.GetHostAddresses(Dns.GetHostName())[0] + ":" + clientPort.Text + "/Grid/Client";
       }
 
@@ -87 +86 @@
         statusTextBox.Text = "Waiting for engine";
 
-      } catch (CommunicationException ex) {
+      } catch(CommunicationException ex) {
         MessageBox.Show("Exception while connecting to the server: " + ex.Message);
         clientHost.Abort();
@@ -106 +105 @@
 
     private void fetchOperationTimer_Elapsed(object sender, System.Timers.ElapsedEventArgs e) {
-      lock(locker) {
-        byte[] engineXml;
-        fetchOperationTimer.Stop();
-        if(engineStore.TryTakeEngine(clientUrl, out currentGuid, out engineXml)) {
+      byte[] engineXml;
+      fetchOperationTimer.Stop();
+      try {
+        if(engineStore.TryTakeEngine(out currentGuid, out engineXml)) {
           currentEngine = RestoreEngine(engineXml);
           if(InvokeRequired) { Invoke((MethodInvoker)delegate() { statusTextBox.Text = "Executing engine"; }); } else statusTextBox.Text = "Executing engine";
@@ -126 +125 @@
           fetchOperationTimer.Start();
         }
+      } catch(Exception ex) {
+        currentEngine = null;
+        currentGuid = Guid.Empty;
+        fetchOperationTimer.Interval = 5000;
+        fetchOperationTimer.Start();
       }
     }
     public void Abort(Guid guid) {
-      lock(locker) {
-        if(!IsRunningEngine(guid)) return;
-        currentEngine.Abort();
-      }
+      throw new NotSupportedException();
     }
     public bool IsRunningEngine(Guid guid) {
-      return currentGuid == guid;
+      throw new NotSupportedException();
     }
     private ProcessingEngine RestoreEngine(byte[] engine) {

trunk/sources/HeuristicLab.Grid/EngineStore.cs

@@ -31 +31 @@
     private List<Guid> engineList;
     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 Dictionary<Guid, DateTime> resultDate;
     private object bigLock;
     private ChannelFactory<IClient> clientChannelFactory;
@@ -45 +44 @@
     public int RunningJobs {
       get {
-        return runningEngines.Count;
+        return waitHandles.Count;
       }
     }
@@ -58 +57 @@
       engineList = new List<Guid>();
       waitingEngines = new Dictionary<Guid, byte[]>();
-      runningEngines = new Dictionary<Guid, byte[]>();
-      runningClients = new Dictionary<Guid, string>();
       waitHandles = new Dictionary<Guid, ManualResetEvent>();
       results = new Dictionary<Guid, byte[]>();
+      resultDate = new Dictionary<Guid, DateTime>();
       bigLock = new object();
 
@@ -73 +71 @@
     }
 
-    public bool TryTakeEngine(string clientUrl, out Guid guid, out byte[] engine) {
+    public bool TryTakeEngine(out Guid guid, out byte[] engine) {
       lock(bigLock) {
         if(engineList.Count == 0) {
@@ -84 +82 @@
           engine = waitingEngines[guid];
           waitingEngines.Remove(guid);
-          runningEngines[guid] = engine;
-          runningClients[guid] = clientUrl;
           return true;
         }
@@ -93 +89 @@
     public void StoreResult(Guid guid, byte[] result) {
       lock(bigLock) {
-        if(!runningEngines.ContainsKey(guid)) return; // ignore result when the engine is not known to be running
-
-        runningEngines.Remove(guid);
-        runningClients.Remove(guid);
+        // clear old results
+        List<Guid> expiredResults = FindExpiredResults(DateTime.Now.AddHours(-1.0));
+        foreach(Guid expiredGuid in expiredResults) {
+          results.Remove(expiredGuid);
+          waitHandles.Remove(expiredGuid);
+          resultDate.Remove(expiredGuid);
+        }
+        // add the new result
         results[guid] = result;
+        resultDate[guid] = DateTime.Now;
         waitHandles[guid].Set();
       }
+    }
+
+    private List<Guid> FindExpiredResults(DateTime expirationDate) {
+      List<Guid> expiredResults = new List<Guid>();
+      foreach(Guid guid in results.Keys) {
+        if(resultDate[guid] < expirationDate) {
+          expiredResults.Add(guid);
+        }
+      }
+      return expiredResults;
     }
 
@@ -113 +124 @@
       return GetResult(guid, System.Threading.Timeout.Infinite);
     }
+
     internal byte[] GetResult(Guid guid, int timeout) {
       lock(bigLock) {
-        if(waitHandles.ContainsKey(guid)) {
+        // result already available
+        if(results.ContainsKey(guid)) {
+          // if the wait-handle for this result is still alive then close and remove it
+          if(waitHandles.ContainsKey(guid)) {
+            ManualResetEvent waitHandle = waitHandles[guid];
+            waitHandle.Close();
+            waitHandles.Remove(guid);
+          }
+          return results[guid];
+        } else {
+          // result not yet available, if there is also no wait-handle for that result then we will never have a result and can return null
+          if(!waitHandles.ContainsKey(guid)) return null;
+
+          // otherwise we have a wait-handle and can wait for the result
           ManualResetEvent waitHandle = waitHandles[guid];
+          // wait
           if(waitHandle.WaitOne(timeout, true)) {
+            // ok got the result in within the wait time => close and remove the wait-hande and return the result
             waitHandle.Close();
             waitHandles.Remove(guid);
             byte[] result = results[guid];
-            results.Remove(guid);
             return result;
           } else {
+            // no result yet return without result
             return null;
           }
-        } else {
-          return null;
         }
       }
@@ -133 +158 @@
 
     internal void AbortEngine(Guid guid) {
-      string clientUrl = "";
       lock(bigLock) {
-        if(runningClients.ContainsKey(guid)) {
-          clientUrl = runningClients[guid];
-          IClient client = clientChannelFactory.CreateChannel(new EndpointAddress(clientUrl));
-          client.Abort(guid);
-        } else if(waitingEngines.ContainsKey(guid)) {
+        if(waitingEngines.ContainsKey(guid)) {
           byte[] engine = waitingEngines[guid];
           waitingEngines.Remove(guid);
@@ -148 +168 @@
       }
     }
+
+    internal JobState JobState(Guid guid) {
+      lock(bigLock) {
+        if(waitingEngines.ContainsKey(guid)) return HeuristicLab.Grid.JobState.Waiting;
+        else if(waitHandles.ContainsKey(guid)) return HeuristicLab.Grid.JobState.Busy;
+        else if(results.ContainsKey(guid)) return HeuristicLab.Grid.JobState.Finished;
+        else return HeuristicLab.Grid.JobState.Unkown;
+      }
+    }
   }
 }

trunk/sources/HeuristicLab.Grid/GridServer.cs

@@ -35 +35 @@
     }
 
+    public JobState JobState(Guid guid) {
+      return engineStore.JobState(guid);
+    }
+
     public Guid BeginExecuteEngine(byte[] engine) {
       Guid guid = Guid.NewGuid();

trunk/sources/HeuristicLab.Grid/IEngineStore.cs

@@ -30 +30 @@
   interface IEngineStore {
     [OperationContract]
-    bool TryTakeEngine(string clientUrl, out Guid guid, out byte[] engine);
+    bool TryTakeEngine(out Guid guid, out byte[] engine);
 
     [OperationContract]

trunk/sources/HeuristicLab.Grid/IGridServer.cs

@@ -26 +26 @@
 
 namespace HeuristicLab.Grid {
+  public enum JobState {
+    Unkown,
+    Waiting,
+    Busy,
+    Finished
+  }
+
+
   [ServiceContract(Namespace = "http://HeuristicLab.Grid")]
   public interface IGridServer {
+    [OperationContract]
+    JobState JobState(Guid guid);
     [OperationContract]
     Guid BeginExecuteEngine(byte[] engine);

trunk/sources/HeuristicLab.Grid/ProcessingEngine.cs

@@ -68 +68 @@
           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) {