Changeset 5227 for branches/HeuristicLab.Hive-3.3/sources/HeuristicLab.Hive/HeuristicLab.HiveEngine/3.3/HiveEngine.cs

Timestamp:

01/06/11 13:17:56 (14 years ago)

Author:

cneumuel

Message:

#1260

• improved efficiency of HiveEngine

File:

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

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

@@ -98 +98 @@
 
           if (operation.Operator.Breakpoint) {
-            Log.LogMessage(string.Format("Breakpoint: {0}", operation.Operator.Name != string.Empty ? operation.Operator.Name : operation.Operator.ItemName));
+            LogMessage(string.Format("Breakpoint: {0}", operation.Operator.Name != string.Empty ? operation.Operator.Name : operation.Operator.ItemName));
             Pause();
           }
@@ -136 +136 @@
     /// <param name="jobDict"></param>
     private void ExecuteOnHive(IDictionary<IOperation, OperationJob> jobDict) {
-      Log.LogMessage(string.Format("Executing {0} operations on the hive.", jobDict.Count));
+      LogMessage(string.Format("Executing {0} operations on the hive.", jobDict.Count));
+      Semaphore concurrentUploads = new Semaphore(6, 6);
       try {
         IDictionary<Guid, IOperation> jobIds = new Dictionary<Guid, IOperation>();
+        List<Guid> remainingJobIds = new List<Guid>();
+        IDictionary<Guid, int> jobNumbers = new Dictionary<Guid, int>(); // for better readability of log
         JobResultList results;
-
-        using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
-          List<JobDto> jobs = new List<JobDto>();
-          foreach (var kvp in jobDict) {
-            // shuffle random variable to avoid the same random sequence in each operation
-            IRandom random = FindRandomParameter(kvp.Key as IExecutionContext);
-            if (random != null)
-              random.Reset(random.Next());
-
+        var pluginsNeeded = ApplicationManager.Manager.Plugins.Select(x => new HivePluginInfoDto { Name = x.Name, Version = x.Version }).ToList();
+
+        List<JobDto> jobs = new List<JobDto>();
+        foreach (var kvp in jobDict) {
+
+          // 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);
+          if (random != null)
+            random.Reset(random.Next());
+
+          Task.Factory.StartNew((operationJob) => {
             var groups = ResourceIds.Split(';');
             SerializedJob serializedJob = new SerializedJob();
-            serializedJob.SerializedJobData = SerializedJob.Serialize(kvp.Value);
+            serializedJob.SerializedJobData = SerializedJob.Serialize(operationJob);
             serializedJob.JobInfo = new JobDto();
             serializedJob.JobInfo.State = JobState.Offline;
             serializedJob.JobInfo.CoresNeeded = 1;
-            serializedJob.JobInfo.PluginsNeeded = ApplicationManager.Manager.Plugins.Select(x => new HivePluginInfoDto { Name = x.Name, Version = x.Version }).ToList();
-            ResponseObject<JobDto> response = service.Obj.AddJobWithGroupStrings(serializedJob, groups);
+            serializedJob.JobInfo.PluginsNeeded = pluginsNeeded;
+            ResponseObject<JobDto> response;
+            concurrentUploads.WaitOne();
+            using (Disposable<IClientFacade> service = ServiceLocator.Instance.StreamedClientFacadePool.GetService()) {
+              response = service.Obj.AddJobWithGroupStrings(serializedJob, groups);
+            }
+            concurrentUploads.Release();
             jobs.Add(response.Obj);
             jobIds.Add(response.Obj.Id, kvp.Key);
-          }
-          results = service.Obj.GetJobResults(jobIds.Keys).Obj;
-        }
-
-        while (!results.All(
-            x => x.State == JobState.Finished ||
-            x.State == JobState.Failed ||
-            x.State == JobState.Aborted)) {
+            lock (remainingJobIds) {
+              remainingJobIds.Add(response.Obj.Id);
+              jobNumbers.Add(response.Obj.Id, remainingJobIds.Count);
+            }
+            LogMessage(string.Format("Submitted job #{0} (id: {1})", jobNumbers[response.Obj.Id], response.Obj.Id));
+          }, kvp.Value);
+        }
+
+        while (remainingJobIds.Count != jobDict.Count) {
+          Thread.Sleep(1000);
+        }
+
+        LogMessage("Waiting for results...");
+        int jobsFinishedCount = 0;
+        Semaphore concurrentDownloads = new Semaphore(4, 4);
+        while (remainingJobIds.Count > 0) {
           Thread.Sleep(5000);
           using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
-            results = service.Obj.GetJobResults(jobIds.Keys).Obj;
-          }
-        }
-
-        // all finished
-        using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
-          foreach (Guid jobId in jobIds.Keys) {
-            SerializedJob serializedJob = service.Obj.GetLastSerializedResult(jobId).Obj;
-            OperationJob operationJob = SerializedJob.Deserialize<OperationJob>(serializedJob.SerializedJobData);
-            jobDict[jobIds[jobId]] = operationJob;
-          }
-        }
-
+            results = service.Obj.GetJobResults(remainingJobIds).Obj;
+          }
+          foreach (var result in results) {
+            if (result.State == JobState.Finished) {
+              Task.Factory.StartNew((jobIdObj) => {
+                Guid jobId = (Guid)jobIdObj;
+                SerializedJob serializedJob;
+                concurrentDownloads.WaitOne();
+                using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
+                  serializedJob = service.Obj.GetLastSerializedResult(jobId).Obj;
+                }
+                concurrentDownloads.Release();
+                OperationJob operationJob = SerializedJob.Deserialize<OperationJob>(serializedJob.SerializedJobData);
+                jobDict[jobIds[jobId]] = operationJob;
+                LogMessage(string.Format("Downloaded job #{0} (id: {1})", jobNumbers[jobId], jobId));
+                jobsFinishedCount++;
+              }, result.Id);
+
+              remainingJobIds.Remove(result.Id);
+            } else if (result.State == JobState.Aborted) {
+              LogMessage(string.Format("Job #{0} aborted (id: {1})", jobNumbers[result.Id], result.Id));
+              remainingJobIds.Remove(result.Id);
+            } else if (result.State == JobState.Failed) {
+              LogMessage(string.Format("Job {0} failed (id: {1}): {2}", jobNumbers[result.Id], result.Id, result.Exception));
+              remainingJobIds.Remove(result.Id);
+            }
+          }
+        }
+
+        // wait for all tasks to finish downloading and deserializing
+        while (jobsFinishedCount != jobDict.Count) {
+          Thread.Sleep(1000);
+        }
+
+        LogMessage("All jobs finished. Deleting jobs on hive.");
         // delete jobs
         using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
@@ -189 +229 @@
         }
 
-        Log.LogMessage(string.Format("Operations on the hive finished.", jobDict.Count));
+        LogMessage(string.Format("Operations on the hive finished.", jobDict.Count));
       }
       catch (Exception e) {
-        Log.LogException(e);
+        LogException(e);
         throw e;
+      }
+    }
+
+    private void DownloadJob(Guid jobId, IDictionary<IOperation, OperationJob> jobDict, IDictionary<Guid, IOperation> jobIds) {
+      SerializedJob serializedJob;
+      using (Disposable<IClientFacade> service = ServiceLocator.Instance.ClientFacadePool.GetService()) {
+        serializedJob = service.Obj.GetLastSerializedResult(jobId).Obj;
+      }
+      OperationJob operationJob = SerializedJob.Deserialize<OperationJob>(serializedJob.SerializedJobData);
+      jobDict[jobIds[jobId]] = operationJob;
+      LogMessage(string.Format("Downloaded job (id: {0})", jobId));
+    }
+
+    /// <summary>
+    /// Threadsafe message logging
+    /// </summary>
+    private void LogMessage(string message) {
+      lock (Log) {
+        Log.LogMessage(message);
+      }
+    }
+
+    /// <summary>
+    /// Threadsafe exception logging
+    /// </summary>
+    private void LogException(Exception exception) {
+      lock (Log) {
+        Log.LogException(exception);
       }
     }