Changeset 4423 for branches/3.3-Hive/sources/HeuristicLab.Hive/HeuristicLab.Hive.Slave.Core/3.3/Core.cs

Timestamp:

09/17/10 10:26:55 (14 years ago)

Author:

cneumuel

Message:

• Refactored HL.Hive.Experiment. JobItems are not called HiveJobs and OptimizerJobs do not contain a hierarchy anymore.
• Dynamic generation of jobs on a slave are not reflected on the client user interface.
• Optimizer-Trees are now strictly synchronized with the HiveJob-Trees (also the ComputeInParallel property is taken into account when the Child HiveJobs are created)
• Improved the way a class can report progress and lock the UI (IProgressReporter, IProgress, Progress, ProgressView)
• Changes were made to the config-files, so that server and clients work with blade12.hpc.fh-hagenberg.at
• Lots of small changes and bugfixes

File:

• branches/3.3-Hive/sources/HeuristicLab.Hive/HeuristicLab.Hive.Slave.Core/3.3/Core.cs (modified) (25 diffs)

branches/3.3-Hive/sources/HeuristicLab.Hive/HeuristicLab.Hive.Slave.Core/3.3/Core.cs

@@ -36 +36 @@
 using HeuristicLab.Hive.Slave.ExecutionEngine;
 using HeuristicLab.Tracing;
+using HeuristicLab.Common;
 
 namespace HeuristicLab.Hive.Slave.Core {
@@ -44 +45 @@
     public static bool abortRequested { get; set; }
 
+    private Dictionary<Guid, Executor> engines = new Dictionary<Guid, Executor>();
+    private Dictionary<Guid, AppDomain> appDomains = new Dictionary<Guid, AppDomain>();
+    private Dictionary<Guid, JobDto> jobs = new Dictionary<Guid, JobDto>();
+
+    private WcfService wcfService;
+    private HeartbeatManager heartbeatManager;
+
     private bool currentlyFetching;
     private bool CurrentlyFetching {
@@ -55 +63 @@
     }
 
-    private Dictionary<Guid, Executor> engines = new Dictionary<Guid, Executor>();
-    private Dictionary<Guid, AppDomain> appDomains = new Dictionary<Guid, AppDomain>();
-    private Dictionary<Guid, JobDto> jobs = new Dictionary<Guid, JobDto>();
-
-    private WcfService wcfService;
-    private HeartbeatManager beat;
+    public Dictionary<Guid, Executor> ExecutionEngines {
+      get { return engines; }
+    }
+
+    internal Dictionary<Guid, JobDto> Jobs {
+      get { return jobs; }
+    }
 
     /// <summary>
@@ -70 +79 @@
       SlaveConsoleServer server = new SlaveConsoleServer();
       server.Start();
-      
+
       ConfigManager manager = ConfigManager.Instance;
       manager.Core = this;
@@ -95 +104 @@
     private void StartHeartbeats() {
       //Initialize the heartbeat
-      beat = new HeartbeatManager { Interval = new TimeSpan(0, 0, 10) };
-      beat.StartHeartbeat();
+      heartbeatManager = new HeartbeatManager { Interval = new TimeSpan(0, 0, 10) };
+      heartbeatManager.StartHeartbeat();
     }
 
@@ -111 +120 @@
       wcfService.GetFinishedJobResultCompleted += new EventHandler<StoreFinishedJobResultCompletedEventArgs>(wcfService_StoreFinishedJobResultCompleted);
       wcfService.ProcessSnapshotCompleted += new EventHandler<ProcessSnapshotCompletedEventArgs>(wcfService_ProcessSnapshotCompleted);
-      wcfService.ConnectionRestored += new EventHandler(wcfService_ConnectionRestored);
-      //wcfService.ServerChanged += new EventHandler(wcfService_ServerChanged);
       wcfService.Connected += new EventHandler(wcfService_Connected);
     }
@@ -120 +127 @@
       wcfService.GetFinishedJobResultCompleted -= new EventHandler<StoreFinishedJobResultCompletedEventArgs>(wcfService_StoreFinishedJobResultCompleted);
       wcfService.ProcessSnapshotCompleted -= new EventHandler<ProcessSnapshotCompletedEventArgs>(wcfService_ProcessSnapshotCompleted);
-      wcfService.ConnectionRestored -= new EventHandler(wcfService_ConnectionRestored);
-      //wcfService.ServerChanged -= new EventHandler(wcfService_ServerChanged);
       wcfService.Connected -= new EventHandler(wcfService_Connected);
     }
@@ -135 +140 @@
         case MessageContainer.MessageType.AbortJob:
           if (engines.ContainsKey(container.JobId))
-            engines[container.JobId].Abort();
+            try {
+              engines[container.JobId].Abort();
+            }
+            catch (AppDomainUnloadedException) {
+              // appdomain already unloaded. Finishing job probably ongoing
+            }
           else
             Logger.Error("AbortJob: Engine doesn't exist");
@@ -156 +166 @@
         //Snapshot is ready and can be sent back to the Server
         case MessageContainer.MessageType.SnapshotReady:
-          ThreadPool.QueueUserWorkItem(new WaitCallback(GetSnapshot), container.JobId);
+          GetSnapshot(container.JobId);
           break;
 
@@ -168 +178 @@
           break;
 
-
         //A Job has finished and can be sent back to the server
         case MessageContainer.MessageType.FinishedJob:
-          ThreadPool.QueueUserWorkItem(new WaitCallback(GetFinishedJob), container.JobId);
-          break;
-
+          SendFinishedJob(container.JobId);
+          break;
 
         //When the timeslice is up
         case MessageContainer.MessageType.UptimeLimitDisconnect:
           Logger.Info("Uptime Limit reached, storing jobs and sending them back");
-
           ShutdownRunningJobsAndSubmitSnapshots();
           break;
@@ -201 +208 @@
           Logger.Debug("Stopping heartbeat");
           abortRequested = true;
-          beat.StopHeartBeat();
+          heartbeatManager.StopHeartBeat();
           Logger.Debug("Logging out");
           WcfService.Instance.Logout(ConfigManager.Instance.GetClientInfo().Id);
@@ -216 +223 @@
 
         case MessageContainer.MessageType.GetChildJobs:
-          // send the job back to hive
           GetChildJobs((MessageContainerWithCallback<SerializedJobList>)container);
+          break;
+
+        case MessageContainer.MessageType.DeleteChildJobs:
+          wcfService.DeleteChildJobs(container.JobId);
           break;
       }
@@ -224 +234 @@
     private void GetChildJobs(MessageContainerWithCallback<SerializedJobList> mc) {
       ResponseObject<SerializedJobList> response = wcfService.GetChildJobs(mc.JobId);
-      if (response.StatusMessage != ResponseStatus.Ok) {
-        Logger.Error("GetChildJobs failed: " + response.StatusMessage);
+      if (response != null && response.StatusMessage == ResponseStatus.Ok) {
+        mc.Callback(response.Obj);
       } else {
-        mc.Callback(response.Obj);
+        if (response != null) {
+          Logger.Error(string.Format("GetChildJobs failed: {0}", response.StatusMessage));
+        } else {
+          Logger.Error("GetChildJobs failed.");
+        }
       }
     }
@@ -234 +248 @@
       ResponseObject<JobDto> response = wcfService.PauseJob(mc.SerializedJob);
       KillAppDomain(mc.JobId);
-      if (response.StatusMessage != ResponseStatus.Ok) {
+      if (response == null || response.StatusMessage != ResponseStatus.Ok) {
         Logger.Error("PauseJob failed: " + response.StatusMessage);
       }
@@ -241 +255 @@
     private ResponseObject<JobDto> AddChildJob(MessageContainerWithJob mc) {
       ResponseObject<JobDto> response = wcfService.AddChildJob(mc.JobId, mc.SerializedJob);
-      if (response.StatusMessage != ResponseStatus.Ok) {
+      if (response == null || response.StatusMessage != ResponseStatus.Ok) {
         Logger.Error("AddChildJob failed: " + response.StatusMessage);
       }
@@ -267 +281 @@
     /// </summary>
     /// <param name="jobId"></param>
-    private void GetFinishedJob(object jobId) {
-      Guid jId = (Guid)jobId;
-      Logger.Info("Getting the finished job with id: " + jId);
+    private void SendFinishedJob(object jobId) {
       try {
+        Guid jId = (Guid)jobId;
+        Logger.Info("Getting the finished job with id: " + jId);
         if (!engines.ContainsKey(jId)) {
           Logger.Info("Engine doesn't exist");
@@ -278 +292 @@
         byte[] sJob = engines[jId].GetFinishedJob();
 
-        if (WcfService.Instance.LoggedIn) {
+        try {
           Logger.Info("Sending the finished job with id: " + jId);
-          wcfService.GetFinishedJobResultAsync(ConfigManager.Instance.GetClientInfo().Id, jId, sJob, 1.0, engines[jId].CurrentException, true);
+          wcfService.GetFinishedJobResultAsync(ConfigManager.Instance.GetClientInfo().Id, jId, sJob, engines[jId].ExecutionTime, engines[jId].CurrentException, true);
+        }
+        catch (Exception e) {
+          Logger.Info("Transmitting to server failed. Storing the finished job with id: " + jId + " to hdd (" + e.ToString() + ")");
+          JobStorageManager.PersistObjectToDisc(wcfService.ServerIp, 0, jId, sJob); // [chn] Port is not unique anymore (since we need two ports for http and net.tcp-streaming). also the port is now specified only in app.config. use port 0 for the moment
+        }
+        finally {
+          KillAppDomain(jId); // kill app-domain in every case
+        }
+      }
+      catch (Exception e) {
+        OnExceptionOccured(e);
+      }
+    }
+
+    private void GetSnapshot(object jobId) {
+      try {
+        Logger.Info("Fetching a snapshot for job " + jobId);
+        Guid jId = (Guid)jobId;
+        byte[] obj = engines[jId].GetSnapshot();
+        wcfService.ProcessSnapshotSync(ConfigManager.Instance.GetClientInfo().Id, jId, obj, engines[jId].ExecutionTime, null);
+
+        //Uptime Limit reached, now is a good time to destroy this jobs.
+        Logger.Debug("Checking if uptime limit is reached");
+        if (!UptimeManager.Instance.IsAllowedToCalculate()) {
+          Logger.Debug("Uptime limit reached");
+          Logger.Debug("Killing Appdomain");
+          KillAppDomain(jId);
+          //Still anything running?  
+          if (engines.Count == 0) {
+            Logger.Info("All jobs snapshotted and sent back, disconnecting");
+            WcfService.Instance.Disconnect();
+          } else {
+            Logger.Debug("There are still active Jobs in the Field, not disconnecting");
+          }
         } else {
-          Logger.Info("Storing the finished job with id: " + jId + " to hdd");
-          JobStorageManager.PersistObjectToDisc(wcfService.ServerIp, 0, jId, sJob); // [chn] Port is not unique anymore (since we need two ports for http and net.tcp-streaming). also the port is now specified only in app.config. use port 0 for the moment
-          KillAppDomain(jId);
-        }
-      }
-      catch (InvalidStateException ise) {
-        Logger.Error("Invalid State while Snapshoting:", ise);
-      }
-    }
-
-    private void GetSnapshot(object jobId) {
-      Logger.Info("Fetching a snapshot for job " + jobId);
-      Guid jId = (Guid)jobId;
-      byte[] obj = engines[jId].GetSnapshot();
-      Logger.Debug("BEGIN: Sending snapshot sync");
-      wcfService.ProcessSnapshotSync(ConfigManager.Instance.GetClientInfo().Id,
-        jId,
-        obj,
-        engines[jId].Progress,
-        null);
-      Logger.Debug("END: Sended snapshot sync");
-      //Uptime Limit reached, now is a good time to destroy this jobs.
-      Logger.Debug("Checking if uptime limit is reached");
-      if (!UptimeManager.Instance.IsAllowedToCalculate()) {
-        Logger.Debug("Uptime limit reached");
-        Logger.Debug("Killing Appdomain");
-        KillAppDomain(jId);
-        //Still anything running?  
-        if (engines.Count == 0) {
-          Logger.Info("All jobs snapshotted and sent back, disconnecting");
-          WcfService.Instance.Disconnect();
-        } else {
-          Logger.Debug("There are still active Jobs in the Field, not disconnecting");
-        }
-
-      } else {
-        Logger.Debug("Restarting the job" + jobId);
-        engines[jId].StartOnlyJob();
-        Logger.Info("Restarted the job" + jobId);
+          Logger.Debug("Restarting the job" + jobId);
+          engines[jId].StartOnlyJob();
+          Logger.Info("Restarted the job" + jobId);
+        }
+      }
+      catch (Exception e) {
+        OnExceptionOccured(e);
       }
     }
@@ -349 +365 @@
       if (e.Result.StatusMessage != ResponseStatus.GetJob_NoJobsAvailable) {
         Logger.Info("Received new job with id " + e.Result.Obj.Id);
-        bool sandboxed = false;
         Logger.Debug("Fetching plugins for job " + e.Result.Obj.Id);
         try {
-
           PluginCache.Instance.PreparePlugins(e.Result.Obj.PluginsNeeded);
           PluginCache.Instance.CopyPluginsForJob(e.Result.Obj.PluginsNeeded, e.Result.Obj.Id);
 
-          //        foreach (CachedHivePluginInfoDto plugininfo in PluginCache.Instance.GetPlugins(e.Result.Job.PluginsNeeded))
-          //        files.AddRange(plugininfo.PluginFiles);
           Logger.Debug("Plugins fetched for job " + e.Result.Obj.Id);
           String pluginDir = Path.Combine(PluginCache.Instance.PluginRepositoryDir, e.Result.Obj.Id.ToString());
@@ -375 +387 @@
               engine.Start(e.Data);
               engines.Add(e.Result.Obj.Id, engine);
-
               SlaveStatusInfo.JobsFetched++;
               Logger.Info("Increment FetchedJobs to:" + SlaveStatusInfo.JobsFetched);
             }
           }
-          beat.InterruptHeartBeatThread();
+          heartbeatManager.AwakeHeartBeatThread();
         }
         catch (Exception exception) {
@@ -387 +398 @@
           CurrentlyFetching = false;
           KillAppDomain(e.Result.Obj.Id);
-          wcfService.StoreFinishedJobResultsSync(ConfigManager.Instance.GetClientInfo().Id, e.Result.Obj.Id, new byte[] { }, 1, exception, true);
+          wcfService.StoreFinishedJobResultsSync(ConfigManager.Instance.GetClientInfo().Id, e.Result.Obj.Id, new byte[] { }, e.Result.Obj.ExecutionTime, exception.ToString(), true);
         }
       } else {
@@ -406 +417 @@
         SlaveStatusInfo.JobsProcessed++;
         Logger.Info("Increased ProcessedJobs to:" + SlaveStatusInfo.JobsProcessed);
-        beat.InterruptHeartBeatThread();
+        heartbeatManager.AwakeHeartBeatThread();
       } else {
         Logger.Error("Sending of job " + e.Result.JobId + " failed, job has been wasted. Message: " + e.Result.StatusMessage);
@@ -433 +444 @@
         FetchCalendarFromServer();
       }
-      //if the fetching from the server failed - still set the client online... maybe we get 
-      //a result within the next few heartbeats      
-      //if (!UptimeManager.Instance.CalendarAvailable || UptimeManager.Instance.IsOnline()) {
       Logger.Info("CalendarAvailable is " + UptimeManager.Instance.CalendarAvailable + " and IsOnline is: " + UptimeManager.Instance.IsAllowedToCalculate());
-      Logger.Info("Setting client online");
-      wcfService.Login(ConfigManager.Instance.GetClientInfo());
+      CurrentlyFetching = false;
+      CheckRunningAppDomains();
       JobStorageManager.CheckAndSubmitJobsFromDisc();
-      CurrentlyFetching = false;
     }
 
@@ -459 +466 @@
     }
 
-    //this is a little bit tricky - 
-    void wcfService_ConnectionRestored(object sender, EventArgs e) {
-      Logger.Info("Reconnected to old server - checking currently running appdomains");
-
+    private void CheckRunningAppDomains() {
       foreach (KeyValuePair<Guid, Executor> execKVP in engines) {
         if (execKVP.Value.ExecutionState != ExecutionState.Started && execKVP.Value.CurrentMessage == MessageContainer.MessageType.NoMessage) {
           Logger.Info("Checking for JobId: " + execKVP.Value.JobId);
-          Thread finThread = new Thread(new ParameterizedThreadStart(GetFinishedJob));
+          Thread finThread = new Thread(new ParameterizedThreadStart(SendFinishedJob));
           finThread.Start(execKVP.Value.JobId);
         }
@@ -474 +478 @@
     #endregion
 
-    public Dictionary<Guid, Executor> ExecutionEngines {
-      get { return engines; }
-    }
-
-    internal Dictionary<Guid, JobDto> Jobs {
-      get { return jobs; }
+    public event EventHandler<EventArgs<Exception>> ExceptionOccured;
+    private void OnExceptionOccured(Exception e) {
+      Logger.Error("Error: " + e.ToString());
+      var handler = ExceptionOccured;
+      if (handler != null) handler(this, new EventArgs<Exception>(e));
     }
 
@@ -498 +501 @@
           if (appDomains.ContainsKey(id)) {
             appDomains[id].UnhandledException -= new UnhandledExceptionEventHandler(appDomain_UnhandledException);
-            AppDomain.Unload(appDomains[id]);
+
+            int repeat = 5;
+            while (repeat > 0) {
+              try {
+                AppDomain.Unload(appDomains[id]);
+                repeat = 0;
+              }
+              catch (CannotUnloadAppDomainException) {
+                Logger.Error("Could not unload AppDomain, will try again in 1 sec.");
+                Thread.Sleep(1000);
+                repeat--;
+                if (repeat == 0) {
+                  throw; // rethrow and let app crash
+                }
+              }
+            }
             appDomains.Remove(id);
           }