Changeset 279 for branches/3.0/sources/HeuristicLab.Grid/EngineStore.cs

Timestamp:

05/30/08 13:41:27 (16 years ago)

Author:

gkronber

Message:

merged changesets r219:r228, r240, r241:258, r263:265, r267,r268, r269 from trunk into the HL3 stable branch

File:

• branches/3.0/sources/HeuristicLab.Grid/EngineStore.cs (modified) (6 diffs)

branches/3.0/sources/HeuristicLab.Grid/EngineStore.cs

@@ -34 +34 @@
     private Dictionary<Guid, ManualResetEvent> waitHandles;
     private Dictionary<Guid, byte[]> results;
-    private Dictionary<Guid, string> runningClients;
+    private Dictionary<Guid, DateTime> resultDate;
+    private Dictionary<Guid, DateTime> runningEngineDate;
+    private const int RESULT_EXPIRY_TIME_MIN = 10;
+    private const int RUNNING_JOB_EXPIRY_TIME_MIN = 10;
     private object bigLock;
-    private ChannelFactory<IClient> clientChannelFactory;
     public int WaitingJobs {
       get {
@@ -59 +61 @@
       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>();
+      runningEngineDate = new Dictionary<Guid, DateTime>();
       bigLock = new object();
-
-      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;
-
-      clientChannelFactory = new ChannelFactory<IClient>(binding);
     }
 
-    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) {
@@ -85 +80 @@
           waitingEngines.Remove(guid);
           runningEngines[guid] = engine;
-          runningClients[guid] = clientUrl;
+          runningEngineDate[guid] = DateTime.Now;
           return true;
         }
@@ -93 +88 @@
     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
-
+        // clear old results
+        List<Guid> expiredResults = FindExpiredResults(DateTime.Now.AddMinutes(-RESULT_EXPIRY_TIME_MIN));
+        foreach(Guid expiredGuid in expiredResults) {
+          results.Remove(expiredGuid);
+          waitHandles.Remove(expiredGuid);
+          resultDate.Remove(expiredGuid);
+        }
+        // add the new result
         runningEngines.Remove(guid);
-        runningClients.Remove(guid);
+        runningEngineDate.Remove(guid);
         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;
+    }
+    private List<Guid> FindExpiredJobs(DateTime expirationDate) {
+      List<Guid> expiredJobs = new List<Guid>();
+      foreach(Guid guid in runningEngines.Keys) {
+        if(runningEngineDate[guid] < expirationDate) {
+          expiredJobs.Add(guid);
+        }
+      }
+      return expiredJobs;
     }
 
@@ -113 +134 @@
       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);
+          }
+          byte[] result = results[guid];
+          results.Remove(guid);
+          return result;
+        } 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, check for which jobs we waited too long and requeue those jobs
+            List<Guid> expiredJobs = FindExpiredJobs(DateTime.Now.AddMinutes(-RUNNING_JOB_EXPIRY_TIME_MIN));
+            foreach(Guid expiredGuid in expiredJobs) {
+              engineList.Insert(0, expiredGuid);
+              waitingEngines[expiredGuid] = runningEngines[expiredGuid];
+              runningEngines.Remove(expiredGuid);
+              runningEngineDate.Remove(expiredGuid);
+            }
             return null;
           }
-        } else {
-          return null;
         }
       }
@@ -133 +176 @@
 
     internal void AbortEngine(Guid guid) {
-      string clientUrl = "";
+      throw new NotImplementedException();
+    }
+
+    internal JobState JobState(Guid guid) {
       lock(bigLock) {
-        if(runningClients.ContainsKey(guid)) {
-          clientUrl = runningClients[guid];
-          IClient client = clientChannelFactory.CreateChannel(new EndpointAddress(clientUrl));
-          client.Abort(guid);
-        } else if(waitingEngines.ContainsKey(guid)) {
-          byte[] engine = waitingEngines[guid];
-          waitingEngines.Remove(guid);
-          engineList.Remove(guid);
-          waitHandles[guid].Set();
-          results.Add(guid, engine);
-        }
+        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;
       }
     }