Changeset 5263 for branches/HeuristicLab.Hive-3.3/sources

Timestamp:

01/10/11 02:29:22 (14 years ago)

Author:

cneumuel

Message:

#1260

• clearance of unneeded scopes HiveEngine

File:

• branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine/3.3/HiveEngine.cs (modified) (14 diffs)

branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine/3.3/HiveEngine.cs

@@ -24 +24 @@
   public class HiveEngine : Engine {
     private Semaphore maxConcurrentConnections = new Semaphore(4, 4); // avoid too many connections
-    private Semaphore maxConcurrentSerializations = new Semaphore(1, 1); // allow ony ONE concurrent serialization, because the operations share the same ParentScopes and serializing the same objects concurrently causes problems
-    private Semaphore maxSerializedJobsInMemory = new Semaphore(2, 2); // avoid memory problems
+    private Semaphore maxSerializedJobsInMemory = new Semaphore(4, 4); // avoid memory problems
     private CancellationToken cancellationToken;
 
@@ -68 +67 @@
           coll = (OperationCollection)next;
           if (coll.Parallel) {
-            IDictionary<IOperation, OperationJob> jobs = new Dictionary<IOperation, OperationJob>();
-            foreach (IOperation op in coll) {
-              jobs.Add(op, new OperationJob(op));
-            }
-
-            ExecuteOnHive(jobs, cancellationToken);
-
-            foreach (var kvp in jobs) {
-              if (kvp.Key is IAtomicOperation) {
-                ReIntegrateScope(kvp.Value.Operation as IAtomicOperation, kvp.Key as IAtomicOperation);
-              } else if (kvp.Key is OperationCollection) {
+            OperationJob[] jobs = new OperationJob[coll.Count];
+            for (int i = 0; i < coll.Count; i++) {
+              jobs[i] = new OperationJob(coll[i]);
+            }
+
+            IScope[] scopes = ExecuteOnHive(jobs, cancellationToken);
+
+            for (int i = 0; i < coll.Count; i++) {
+              if (coll[i] is IAtomicOperation) {
+                ExchangeScope(scopes[i], ((IAtomicOperation)coll[i]).Scope);
+              } else if (coll[i] is OperationCollection) {
                 // todo ??
               }
@@ -136 +135 @@
     /// TODO: Cancelation needs to be refined; all tasks currently stay in Semaphore.WaitOne after cancelation
     /// </summary>
-    /// <param name="jobDict"></param>
-    private void ExecuteOnHive(IDictionary<IOperation, OperationJob> jobDict, CancellationToken cancellationToken) {
-      LogMessage(string.Format("Executing {0} operations on the hive.", jobDict.Count));
-
+    /// <param name="jobs"></param>
+    private IScope[] ExecuteOnHive(OperationJob[] jobs, CancellationToken cancellationToken) {
+      LogMessage(string.Format("Executing {0} operations on the hive.", jobs.Length));
+      IScope[] scopes = new Scope[jobs.Length];
       object locker = new object();
+
       try {
-        IDictionary<Guid, IOperation> jobIds = new Dictionary<Guid, IOperation>();
+        IDictionary<Guid, int> jobIndices = new Dictionary<Guid, int>();
         List<Guid> remainingJobIds = new List<Guid>();
-        IDictionary<Guid, int> jobNumbers = new Dictionary<Guid, int>(); // for better readability of log
         JobResultList results;
         var pluginsNeeded = ApplicationManager.Manager.Plugins.Select(x => new HivePluginInfoDto { Name = x.Name, Version = x.Version }).ToList();
@@ -152 +151 @@
         // create upload-tasks
         var uploadTasks = new List<Task<JobDto>>();
-        foreach (var kvp in jobDict) {
+        for (int i = 0; i < jobs.Length; i++) {
+          var job = jobs[i];
+
           // shuffle random variable to avoid the same random sequence in each operation; todo: does not yet work (it cannot find the random variable)
-          IRandom random = FindRandomParameter(kvp.Key as IExecutionContext);
+          IRandom random = FindRandomParameter(job.Operation as IExecutionContext);
           if (random != null)
             random.Reset(random.Next());
@@ -160 +161 @@
           uploadTasks.Add(Task.Factory.StartNew<JobDto>((keyValuePairObj) => {
             return UploadJob(pluginsNeeded, keyValuePairObj, cancellationToken);
-          }, kvp, cancellationToken));
+          }, new KeyValuePair<int, OperationJob>(i, job), cancellationToken));
         }
 
@@ -176 +177 @@
             }
 
-            IOperation key = ((KeyValuePair<IOperation, OperationJob>)task.AsyncState).Key;
+            int key = ((KeyValuePair<int, OperationJob>)task.AsyncState).Key;
             JobDto jobDto = task.Result;
             lock (locker) {
               uploadCount++;
-              jobIds.Add(jobDto.Id, key);
+              jobIndices.Add(jobDto.Id, key);
               remainingJobIds.Add(jobDto.Id);
-              jobNumbers.Add(jobDto.Id, uploadCount);
-            }
-            LogMessage(string.Format("Submitted job #{0}", jobNumbers[jobDto.Id], jobDto.Id));
+            }
+            jobs[key] = null; // relax memory
+            LogMessage(string.Format("Submitted job #{0}", key + 1, jobDto.Id));
             uploadTasks.Remove(task);
           }
@@ -203 +204 @@
           finishedCount += jobsFinished.Count();
           var totalExecutionTime = TimeSpan.FromMilliseconds(results.Select(j => j.ExecutionTime).Union(executionTimes).Select(e => e.TotalMilliseconds).Sum());
-          LogMessage(string.Format("Results polled. Jobs finished: {0}/{1}, TotalExecutionTime: {2}", finishedCount, jobDict.Count, totalExecutionTime));
+          LogMessage(string.Format("Results polled. Jobs finished: {0}/{1}, TotalExecutionTime: {2}", finishedCount, jobs.Length, totalExecutionTime));
           foreach (var result in jobsFinished) {
             if (result.State == JobState.Finished) {
               downloadTasks.Add(Task.Factory.StartNew<OperationJob>((jobIdObj) => {
-                return DownloadJob(jobNumbers, jobIdObj, cancellationToken);
+                return DownloadJob(jobIndices, jobIdObj, cancellationToken);
               }, result.Id, cancellationToken));
             } else if (result.State == JobState.Aborted) {
-              LogMessage(string.Format("Job #{0} aborted (id: {1})", jobNumbers[result.Id], result.Id));
+              LogMessage(string.Format("Job #{0} aborted (id: {1})", jobIndices[result.Id] + 1, result.Id));
             } else if (result.State == JobState.Failed) {
-              LogMessage(string.Format("Job {0} failed (id: {1}): {2}", jobNumbers[result.Id], result.Id, result.Exception));
+              LogMessage(string.Format("Job {0} failed (id: {1}): {2}", jobIndices[result.Id] + 1, result.Id, result.Exception));
             }
             remainingJobIds.Remove(result.Id);
@@ -231 +232 @@
             throw task.Exception;
           }
-          jobDict[jobIds[(Guid)task.AsyncState]] = task.Result;
+          scopes[jobIndices[(Guid)task.AsyncState]] = ((IAtomicOperation)task.Result.Operation).Scope;
           downloadTasks.Remove(task);
         }
@@ -238 +239 @@
         // delete jobs
         using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
-          foreach (Guid jobId in jobIds.Keys) {
+          foreach (Guid jobId in jobIndices.Keys) {
             service.Obj.DeleteJob(jobId);
           }
         }
 
-        LogMessage(string.Format("Operations on the hive finished.", jobDict.Count));
+        LogMessage(string.Format("Operations on the hive finished.", jobs.Length));
+        return scopes;
       }
       catch (Exception e) {
@@ -251 +253 @@
     }
 
+    private static object locker = new object();
     private JobDto UploadJob(List<HivePluginInfoDto> pluginsNeeded, object keyValuePairObj, CancellationToken cancellationToken) {
-      var keyValuePair = (KeyValuePair<IOperation, OperationJob>)keyValuePairObj;
+      var keyValuePair = (KeyValuePair<int, OperationJob>)keyValuePairObj;
       var groups = ResourceIds.Split(';');
       maxSerializedJobsInMemory.WaitOne();
@@ -267 +270 @@
         }
       }
-      maxConcurrentSerializations.WaitOne();
-      serializedJob.SerializedJobData = SerializedJob.Serialize(keyValuePair.Value);
-      maxConcurrentSerializations.Release();
+      // clone operation and remove unnecessary scopes; don't do this earlier to avoid memory problems
+      lock (locker) {
+        keyValuePair.Value.Operation = (IOperation)keyValuePair.Value.Operation.Clone();
+        if (keyValuePair.Value.Operation is IAtomicOperation)
+          ((IAtomicOperation)keyValuePair.Value.Operation).Scope.ClearParentScopes();
+        serializedJob.SerializedJobData = SerializedJob.Serialize(keyValuePair.Value);
+      }
       serializedJob.JobInfo = new JobDto();
       serializedJob.JobInfo.State = JobState.Offline;
@@ -293 +300 @@
     }
 
-    private OperationJob DownloadJob(IDictionary<Guid, int> jobNumbers, object jobIdObj, CancellationToken cancellationToken) {
+    private OperationJob DownloadJob(IDictionary<Guid, int> jobIndices, object jobIdObj, CancellationToken cancellationToken) {
       Guid jobId = (Guid)jobIdObj;
       SerializedJob serializedJob = null;
@@ -313 +320 @@
       serializedJob = null;
       maxSerializedJobsInMemory.Release();
-      LogMessage(string.Format("Downloaded job #{0}", jobNumbers[jobId], jobId));
+      LogMessage(string.Format("Downloaded job #{0}", jobIndices[jobId] + 1, jobId));
       return operationJob;
     }
@@ -335 +342 @@
     }
   }
+
+  public static class ScopeExtensions {
+    public static void ClearParentScopes(this IScope scope) {
+      scope.ClearParentScopes(null);
+    }
+
+    public static void ClearParentScopes(this IScope scope, IScope childScope) {
+      if (childScope != null) {
+        scope.SubScopes.Clear();
+        scope.SubScopes.Add(childScope);
+      }
+      if (scope.Parent != null)
+        scope.Parent.ClearParentScopes(scope);
+    }
+  }
 }