Changeset 1216 for branches/CEDMA-Refactoring-Ticket419/HeuristicLab.CEDMA.Server/Executer.cs

Timestamp:

02/12/09 14:05:33 (16 years ago)

Author:

gkronber

Message:

Improved dispatching mechanism. #419 (Refactor CEDMA plugins).

File:

• branches/CEDMA-Refactoring-Ticket419/HeuristicLab.CEDMA.Server/Executer.cs (modified) (4 diffs)

branches/CEDMA-Refactoring-Ticket419/HeuristicLab.CEDMA.Server/Executer.cs

@@ -41 +41 @@
 namespace HeuristicLab.CEDMA.Server {
   public class Executer {
-    private static int MaxActiveJobs {
-      get { return 100; }
-    }
     private Dispatcher dispatcher;
     private JobManager jobManager;
     private IStore store;
+    private Dictionary<WaitHandle, Execution> activeExecutions;
+
+    private TimeSpan StartJobInterval {
+      get { return TimeSpan.FromMilliseconds(500); }
+    }
+
+    private TimeSpan WaitForFinishedJobsTimeout {
+      get { return TimeSpan.FromMilliseconds(100); }
+    }
+
+    private int maxActiveJobs;
+    public int MaxActiveJobs {
+      get { return maxActiveJobs; }
+      set {
+        if (value < 0) throw new ArgumentException("Only positive values are allowed for MaxActiveJobs");
+        maxActiveJobs = value;
+      }
+    }
 
     public Executer(Dispatcher dispatcher, IStore store, string gridUrl) {
+      activeExecutions = new Dictionary<WaitHandle, Execution>();
+      maxActiveJobs = 10;
       this.dispatcher = dispatcher;
       this.store = store;
@@ -61 +78 @@
     private void StartJobs() {
       List<WaitHandle> wh = new List<WaitHandle>();
-      Dictionary<WaitHandle, AtomicOperation> activeOperations = new Dictionary<WaitHandle,AtomicOperation>();
-      Dictionary<WaitHandle, Execution> activeExecutions = new Dictionary<WaitHandle,Execution>();
+      Dictionary<WaitHandle, AtomicOperation> activeOperations = new Dictionary<WaitHandle, AtomicOperation>();
       while (true) {
         try {
           // start new jobs as long as there are less than MaxActiveJobs 
           while (wh.Count < MaxActiveJobs) {
+            Thread.Sleep(StartJobInterval);
             // get an execution from the dispatcher and execute in grid via job-manager
             Execution execution = dispatcher.GetNextJob();
@@ -74 +91 @@
               wh.Add(opWh);
               activeOperations.Add(opWh, op);
-              activeExecutions.Add(opWh, execution);
+              lock (activeExecutions) {
+                activeExecutions.Add(opWh, execution);
+              }
             }
           }
           // wait until any job is finished
           WaitHandle[] whArr = wh.ToArray();
-          int readyHandleIndex = WaitHandle.WaitAny(whArr);
-          WaitHandle readyHandle = whArr[readyHandleIndex];
-          AtomicOperation finishedOp = activeOperations[readyHandle];
-          Execution finishedExecution = activeExecutions[readyHandle];
-          wh.Remove(readyHandle);
-          activeExecutions.Remove(readyHandle);
-          activeOperations.Remove(readyHandle);
-          ProcessingEngine finishedEngine = null;
-          try {
-            finishedEngine = jobManager.EndExecuteOperation(finishedOp);
-          } catch (Exception badEx) {
-            Trace.WriteLine("CEDMA Executer: Exception in job execution thread. " + badEx.Message);
-          }
-          if (finishedEngine != null) {
-            StoreResults(finishedExecution, finishedEngine);
+          int readyHandleIndex = WaitHandle.WaitAny(whArr, WaitForFinishedJobsTimeout);
+          if (readyHandleIndex != WaitHandle.WaitTimeout) {
+            WaitHandle readyHandle = whArr[readyHandleIndex];
+            AtomicOperation finishedOp = activeOperations[readyHandle];
+            wh.Remove(readyHandle);
+            Execution finishedExecution = null;
+            lock (activeExecutions) {
+              finishedExecution = activeExecutions[readyHandle];
+              activeExecutions.Remove(readyHandle);
+            }
+            activeOperations.Remove(readyHandle);
+            ProcessingEngine finishedEngine = null;
+            try {
+              finishedEngine = jobManager.EndExecuteOperation(finishedOp);
+            }
+            catch (Exception badEx) {
+              Trace.WriteLine("CEDMA Executer: Exception in job execution thread. " + badEx.Message);
+            }
+            if (finishedEngine != null) {
+              StoreResults(finishedExecution, finishedEngine);
+            }
           }
         }
@@ -115 +140 @@
       StoreModelAttribute(model, Ontology.TreeHeight, bestModelScope.GetVariableValue<IntData>("TreeHeight", false).Data);
       StoreModelAttribute(model, Ontology.EvaluatedSolutions, bestModelScope.GetVariableValue<IntData>("EvaluatedSolutions", false).Data);
-       
+
       byte[] serializedModel = PersistenceManager.SaveToGZip(bestModelScope.GetVariableValue("FunctionTree", false));
       store.Add(new Statement(model, Ontology.PredicateSerializedData, new Literal(Convert.ToBase64String(serializedModel))));
-   }
+    }
 
     private void StoreModelAttribute(Entity model, Entity predicate, object value) {
       store.Add(new Statement(model, predicate, new Literal(value)));
     }
+
+    internal string[] GetJobs() {
+      lock (activeExecutions) {
+        string[] retVal = new string[activeExecutions.Count];
+        int i = 0;
+        foreach (Execution e in activeExecutions.Values) {
+          retVal[i++] = "Target-Variable: "+e.TargetVariable+" "+e.Description;
+        }
+        return retVal;
+      }
+    }
   }
 }