Changeset 4333 for branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Experiment/3.3/HiveExperiment.cs

Timestamp:

08/27/10 08:35:43 (14 years ago)

Author:

cneumuel

Message:

added authorizationManager which checks for permission to specific jobs (#1168)

File:

• branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Experiment/3.3/HiveExperiment.cs (modified) (15 diffs)

branches/3.3-HiveMigration/sources/HeuristicLab.Hive/HeuristicLab.Hive.Experiment/3.3/HiveExperiment.cs

@@ -67 +67 @@
     private Semaphore fetchJobSemaphore = new Semaphore(2, 2);
 
+    private static object pendingOptimizerMappingsLocker = new object();
+
     private bool stopResultsPollingPending = false;
 
@@ -197 +199 @@
       clone.pendingOptimizersByJobId = new Dictionary<Guid, IOptimizer>();
 
-      foreach (var pair in this.pendingOptimizersByJobId)
-        clone.pendingOptimizersByJobId[pair.Key] = (IOptimizer)cloner.Clone(pair.Value);
-
-      foreach (var pair in this.parentOptimizersByPendingOptimizer)
-        clone.parentOptimizersByPendingOptimizer[(IOptimizer)cloner.Clone(pair.Key)] = (IOptimizer)cloner.Clone(pair.Value);
-
+      lock (pendingOptimizerMappingsLocker) {
+        foreach (var pair in this.pendingOptimizersByJobId)
+          clone.pendingOptimizersByJobId[pair.Key] = (IOptimizer)cloner.Clone(pair.Value);
+
+        foreach (var pair in this.parentOptimizersByPendingOptimizer)
+          clone.parentOptimizersByPendingOptimizer[(IOptimizer)cloner.Clone(pair.Key)] = (IOptimizer)cloner.Clone(pair.Value);
+      }
       clone.log = (ILog)cloner.Clone(log);
       clone.stopPending = this.stopPending;
@@ -243 +246 @@
       if (experiment != null) {
         StopResultPolling();
-        lock (pendingOptimizersByJobId) {
+        lock (pendingOptimizerMappingsLocker) {
           pendingOptimizersByJobId.Clear();
-        }
-        parentOptimizersByPendingOptimizer.Clear();
+          parentOptimizersByPendingOptimizer.Clear();
+        }
         lock (jobItems) {
           jobItems.Clear();
@@ -275 +278 @@
 
           IEnumerable<string> groups = ResourceGroups;
-
-          foreach (IOptimizer optimizer in parentOptimizersByPendingOptimizer.Keys) {
-            SerializedJob serializedJob = CreateSerializedJob(optimizer);
-            ResponseObject<JobDto> response = clientFacade.AddJobWithGroupStrings(serializedJob, groups);
-            lock (pendingOptimizersByJobId) {
+          lock (pendingOptimizerMappingsLocker) {
+            foreach (IOptimizer optimizer in parentOptimizersByPendingOptimizer.Keys) {
+              SerializedJob serializedJob = CreateSerializedJob(optimizer);
+              ResponseObject<JobDto> response = clientFacade.AddJobWithGroupStrings(serializedJob, groups);
               pendingOptimizersByJobId.Add(response.Obj.Id, optimizer);
+
+              JobItem jobItem = new JobItem() {
+                JobDto = response.Obj,
+                LatestSnapshot = null,
+                Optimizer = optimizer
+              };
+              lock (jobItems) {
+                jobItems.Add(jobItem);
+              }
+              LogMessage(jobItem.JobDto.Id, "Job sent to Hive");
             }
-
-            JobItem jobItem = new JobItem() {
-              JobDto = response.Obj,
-              LatestSnapshot = null,
-              Optimizer = optimizer
-            };
-            lock (jobItems) {
-              jobItems.Add(jobItem);
-            }
-            LogMessage(jobItem.JobDto.Id, "Job sent to Hive");
           }
         }
@@ -396 +398 @@
 
     private bool NoMorePendingOptimizers() {
-      lock (pendingOptimizersByJobId) {
+      lock (pendingOptimizerMappingsLocker) {
         return pendingOptimizersByJobId.Count == 0;
       }
@@ -408 +410 @@
     /// <param name="jobId"></param>
     private void DisposeOptimizerMappings(Guid jobId) {
-      lock (pendingOptimizersByJobId) {
-        LogMessage(jobId, "Disposing Optimizer Mappings");
+      LogMessage(jobId, "Disposing Optimizer Mappings");
+      lock (pendingOptimizerMappingsLocker) {
         parentOptimizersByPendingOptimizer.Remove(pendingOptimizersByJobId[jobId]);
         pendingOptimizersByJobId.Remove(jobId);
@@ -436 +438 @@
     private void JobItem_JobStateChanged(object sender, EventArgs e) {
       JobItem jobItem = (JobItem)sender;
+
       Thread t = new Thread(() => {
-        if (jobItem.State == JobState.Finished) {
-          FetchAndUpdateJob(jobItem.JobDto.Id);
-          DisposeOptimizerMappings(jobItem.JobDto.Id);
-        } else if (jobItem.State == JobState.Failed) {
-          DisposeOptimizerMappings(jobItem.JobDto.Id);
-        }
-
-        if (NoMorePendingOptimizers()) {
-          StopResultPolling();
-          this.ExecutionState = Core.ExecutionState.Stopped;
-          OnStopped();
+        try {
+          if (jobItem.State == JobState.Finished) {
+            FetchAndUpdateJob(jobItem.JobDto.Id);
+            DisposeOptimizerMappings(jobItem.JobDto.Id);
+          } else if (jobItem.State == JobState.Failed) {
+            DisposeOptimizerMappings(jobItem.JobDto.Id);
+          }
+
+          if (NoMorePendingOptimizers()) {
+            StopResultPolling();
+            this.ExecutionState = Core.ExecutionState.Stopped;
+            OnStopped();
+          }
+        }
+        catch (Exception ex) {
+          Logger.Error("JobItem_JobStateChanged failed badly: " + ex.Message);
+          LogMessage("JobItem_JobStateChanged failed badly: " + ex.Message);
         }
       });
@@ -457 +466 @@
     /// </summary>
     private void FetchAndUpdateJob(Guid jobId) {
+      bool tryagain = false;
       LogMessage(jobId, "FetchAndUpdateJob started");
-      IClientFacade clientFacade = CreateStreamedClientFacade();
-      IOptimizer originalOptimizer;
-      lock (pendingOptimizersByJobId) {
-        originalOptimizer = pendingOptimizersByJobId[jobId];
-      }
-
-      fetchJobSemaphore.WaitOne();
-      ResponseObject<SerializedJob> jobResponse = clientFacade.GetLastSerializedResult(jobId, false, false);
-      ServiceLocator.DisposeClientFacade(clientFacade);
-      IJob restoredObject = XmlParser.Deserialize<IJob>(new MemoryStream(jobResponse.Obj.SerializedJobData));
-      IOptimizer restoredOptimizer = ((OptimizerJob)restoredObject).Optimizer;
-
-      ReplaceOptimizer(parentOptimizersByPendingOptimizer[originalOptimizer], originalOptimizer, restoredOptimizer);
-      fetchJobSemaphore.Release();
-      LogMessage(jobId, "FetchAndUpdateJob ended");
+      if (fetchJobSemaphore.WaitOne(new TimeSpan(0, 2, 0))) {
+        IClientFacade clientFacade = null;
+        try {
+          clientFacade = CreateStreamedClientFacade();
+          IOptimizer originalOptimizer;
+          originalOptimizer = pendingOptimizersByJobId[jobId];
+
+          ResponseObject<SerializedJob> jobResponse = clientFacade.GetLastSerializedResult(jobId, false, false);
+          IJob restoredObject = XmlParser.Deserialize<IJob>(new MemoryStream(jobResponse.Obj.SerializedJobData));
+          IOptimizer restoredOptimizer = ((OptimizerJob)restoredObject).Optimizer;
+          ReplaceOptimizer(parentOptimizersByPendingOptimizer[originalOptimizer], originalOptimizer, restoredOptimizer);
+          LogMessage(jobId, "FetchAndUpdateJob ended");
+        }
+        catch (Exception e) {
+          LogMessage(jobId, "FetchAndUpdateJob failed: " + e.Message + ". Will try again!");
+          tryagain = true;
+        }
+        finally {
+          ServiceLocator.DisposeClientFacade(clientFacade);
+          fetchJobSemaphore.Release();
+        }
+      } else {
+        LogMessage(jobId, "FetchAndUpdateJob timed out. Will try again!");
+        tryagain = true;
+      }
+
+      if (tryagain) {
+        FetchAndUpdateJob(jobId);
+      }
     }
 
@@ -509 +533 @@
         PluginsNeeded = pluginsNeeded,
         State = JobState.Offline,
-        MemoryNeeded = 0,
-        UserId = Guid.Empty // [chn] set real userid here!
+        MemoryNeeded = 0
       };
 
@@ -537 +560 @@
     public void StopResultPolling() {
       this.stopResultsPollingPending = true;
-      resultPollingThread.Interrupt();
+      if (resultPollingThread != null && resultPollingThread.ThreadState == System.Threading.ThreadState.WaitSleepJoin) {
+        resultPollingThread.Interrupt();
+      }
       this.stopResultsPollingPending = false;
     }
@@ -581 +606 @@
           // thread has been interuppted
         }
+        catch (Exception e) {
+          LogMessage("Result Polling Thread failed badly: " + e.Message);
+          Logger.Error("Result Polling Thread failed badly: " + e.Message);
+        }
         finally {
           this.IsPollingResults = false;
@@ -598 +627 @@
     public void RequestSnapshot(Guid jobId) {
       Thread t = new Thread(() => {
-        IClientFacade clientFacade = CreateStreamedClientFacade();
+        IClientFacade clientFacade = null;
         try {
+          clientFacade = CreateStreamedClientFacade();
+
           ResponseObject<SerializedJob> response;
           int retryCount = 0;
@@ -633 +664 @@
           }
         }
+        catch (Exception e) {
+          LogMessage("RequestSnapshot Thread failed badly: " + e.Message);
+          Logger.Error("RequestSnapshot Thread failed badly: " + e.Message);
+        }
         finally {
           ServiceLocator.DisposeClientFacade(clientFacade);
@@ -885 +920 @@
           clientFacade = ServiceLocator.CreateClientFacade(Settings.Default.HiveServerIp);
 
-        } catch (EndpointNotFoundException exception) {
+        }
+        catch (EndpointNotFoundException exception) {
           LogMessage("Could not connect to Server: " + exception.Message + ". Will try again in " + (resultPollingIntervalMs / 1000) + " sec.");
           Thread.Sleep(resultPollingIntervalMs);
@@ -899 +935 @@
           //clientFacade = ServiceLocator.CreateStreamedClientFacade(string.Format("http://{0}:{1}/{2}", Settings.Default.HiveServerIp, Settings.Default.HiveServerPort, WcfSettings.ClientStreamedServiceName));
           clientFacade = ServiceLocator.CreateStreamedClientFacade(Settings.Default.HiveServerIp);
-        } catch (EndpointNotFoundException exception) {
+        }
+        catch (EndpointNotFoundException exception) {
           LogMessage("Could not connect to Server: " + exception.Message + ". Will try again in " + (resultPollingIntervalMs / 1000) + " sec.");
           Thread.Sleep(resultPollingIntervalMs);