Changeset 1287 for trunk/sources/HeuristicLab.CEDMA.Server/Executer.cs

Timestamp:

03/08/09 12:48:18 (15 years ago)

Author:

gkronber

Message:

Merged change sets from CEDMA branch to trunk:

• r957:r990 r992 r1003 r1041:1044 (#419)
• r1051:1053 r1060 (#419, #224)
• r1072:1075 r1104:1106 r1108:1109 r1115:1116 r1129:1130 r1151 r1156 (#419)
• r1201:1202 (#224)
• r1208 (#487)
• r1214:1217 r1230:1231 (#419)
• r1235:1236 r1238:1239 r1245 (#224)
• r1246 (#505)
• r1251:1254 r1262:1263 (#224)
• r1265:1266, r1275:1277 (#419)

File:

• trunk/sources/HeuristicLab.CEDMA.Server/Executer.cs (copied) (copied from branches/CEDMA-Refactoring-Ticket419/HeuristicLab.CEDMA.Server/Executer.cs) (3 diffs)

trunk/sources/HeuristicLab.CEDMA.Server/Executer.cs

@@ -41 +41 @@
 namespace HeuristicLab.CEDMA.Server {
   public class Executer {
-    private static int MaxActiveJobs {
-      get { return 20; }
-    }
-    private Dispatcher dispatcher;
+    private IDispatcher dispatcher;
     private JobManager jobManager;
     private IStore store;
+    private Dictionary<WaitHandle, Execution> activeExecutions;
 
-    public Executer(Dispatcher dispatcher, IStore store, string gridUrl) {
+    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(IDispatcher 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();
-            AtomicOperation op = new AtomicOperation(execution.Engine.OperatorGraph.InitialOperator, execution.Engine.GlobalScope);
-            WaitHandle opWh = jobManager.BeginExecuteOperation(execution.Engine.GlobalScope, op);
-            wh.Add(opWh);
-            activeOperations.Add(opWh, op);
+            if (execution != null) {
+              AtomicOperation op = new AtomicOperation(execution.Engine.OperatorGraph.InitialOperator, execution.Engine.GlobalScope);
+              WaitHandle opWh = jobManager.BeginExecuteOperation(execution.Engine.GlobalScope, op);
+              wh.Add(opWh);
+              activeOperations.Add(opWh, op);
+              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) {
-            Entity modelEntity = new Entity(Ontology.CedmaNameSpace+Guid.NewGuid());
-            store.Add(new Statement(modelEntity, Ontology.PredicateInstanceOf, Ontology.TypeGeneticProgrammingFunctionTree));
-            Entity targetVariableAttr = new Entity(Ontology.CedmaNameSpace+Guid.NewGuid());
-            store.Add(new Statement(modelEntity, Ontology.PredicateModelAttribute, targetVariableAttr));
-            store.Add(new Statement(targetVariableAttr, Ontology.PredicateModelAttributeName, new Literal("TargetVariable")));
-            store.Add(new Statement(targetVariableAttr, Ontology.PredicateModelAttributeValue, new Literal(finishedExecution.TargetVariable)));
-            store.Add(new Statement(targetVariableAttr, Ontology.PredicateModelAttributeType, Ontology.TypeCategoricalAttribute));
-            store.Add(new Statement(finishedExecution.DataSetEntity, Ontology.PredicateHasModel, modelEntity));
+          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);
+            }
           }
         }
@@ -105 +126 @@
       }
     }
+
+    private void StoreResults(Execution finishedExecution, ProcessingEngine finishedEngine) {
+      Entity model = new Entity(Ontology.CedmaNameSpace + Guid.NewGuid());
+      store.Add(new Statement(model, Ontology.PredicateInstanceOf, Ontology.TypeGeneticProgrammingFunctionTree));
+      store.Add(new Statement(finishedExecution.DataSetEntity, Ontology.PredicateHasModel, model));
+      StoreModelAttribute(model, Ontology.TargetVariable, finishedExecution.TargetVariable);
+      Scope bestModelScope = finishedEngine.GlobalScope.GetVariableValue<Scope>("BestValidationSolution", false);
+      StoreModelVariable(model, Ontology.TrainingMeanSquaredError, bestModelScope, "Quality");
+      StoreModelVariable(model, Ontology.ValidationMeanSquaredError, bestModelScope, "ValidationQuality");
+      StoreModelVariable(model, Ontology.TestMeanSquaredError, bestModelScope, "TestQuality");
+      StoreModelVariable(model, Ontology.TrainingMeanAbsolutePercentageError, bestModelScope, "TrainingMAPE");
+      StoreModelVariable(model, Ontology.ValidationMeanAbsolutePercentageError, bestModelScope, "ValidationMAPE");
+      StoreModelVariable(model, Ontology.TestMeanAbsolutePercentageError, bestModelScope, "TestMAPE");
+      StoreModelVariable(model, Ontology.TrainingMeanAbsolutePercentageOfRangeError, bestModelScope, "TrainingMAPRE");
+      StoreModelVariable(model, Ontology.ValidationMeanAbsolutePercentageOfRangeError, bestModelScope, "ValidationMAPRE");
+      StoreModelVariable(model, Ontology.TestMeanAbsolutePercentageOfRangeError, bestModelScope, "TestMAPRE");
+      StoreModelVariable(model, Ontology.TrainingCoefficientOfDetermination, bestModelScope, "TrainingR2");
+      StoreModelVariable(model, Ontology.ValidationCoefficientOfDetermination, bestModelScope, "ValidationR2");
+      StoreModelVariable(model, Ontology.TestCoefficientOfDetermination, bestModelScope, "TestR2");
+      StoreModelVariable(model, Ontology.TrainingTheilsInequalityCoefficient, bestModelScope, "TrainingTheilInequalityCoefficient");
+      StoreModelVariable(model, Ontology.ValidationTheilsInequalityCoefficient, bestModelScope, "ValidationTheilInequalityCoefficient");
+      StoreModelVariable(model, Ontology.TestTheilsInequalityCoefficient, bestModelScope, "TestTheilInequalityCoefficient");
+      StoreModelVariable(model, Ontology.TrainingAccuracy, bestModelScope, "TrainingAccuracy");
+      StoreModelVariable(model, Ontology.ValidationAccuracy, bestModelScope, "ValidationAccuracy");
+      StoreModelVariable(model, Ontology.TestAccuracy, bestModelScope, "TestAccuracy");
+      StoreModelVariable(model, Ontology.TreeSize, bestModelScope, "TreeSize");
+      StoreModelVariable(model, Ontology.TreeHeight, bestModelScope, "TreeHeight");
+      StoreModelVariable(model, Ontology.EvaluatedSolutions, bestModelScope, "EvaluatedSolutions");
+
+      byte[] serializedModel = PersistenceManager.SaveToGZip(bestModelScope.GetVariableValue("FunctionTree", false));
+      store.Add(new Statement(model, Ontology.PredicateSerializedData, new Literal(Convert.ToBase64String(serializedModel))));
+    }
+
+    private void StoreModelVariable(Entity model, Entity entity, Scope scope, string variableName) {
+      if (scope.GetVariable(variableName) != null)
+        StoreModelAttribute(model, entity, scope.GetVariableValue<ObjectData>(variableName, false).Data);
+    }
+
+    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;
+      }
+    }
   }
 }