Changeset 219 for trunk/sources/HeuristicLab.Grid/EngineStore.cs

Timestamp:

05/06/08 12:22:58 (16 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)

File:

• trunk/sources/HeuristicLab.Grid/EngineStore.cs (modified) (9 diffs)

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;
+      }
+    }
   }
 }