Changeset 5404 for branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive/3.4/ExperimentManager/HiveExperimentClient.cs

Timestamp:

02/01/11 15:51:11 (13 years ago)

Author:

cneumuel

Message:

#1233

• changed the workflow of aquireing a new job from server.
  • if a job is available for calculation, the slave receives the jobId already with the heartbeats. The job is then exclusively assigned to this slave.
• extended the metainfo for a slave by OperatingSystem and CpuArchitecture
• enhanced the way plugin-dependencies are discovered by using the types used by XmlGenerator. Now only mimimum amount of plugins are transferred.
• selection of waiting jobs now consideres assigned slave-group
• more unit tests for service
• added unit tests for experiment manager

File:

• branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive/3.4/ExperimentManager/HiveExperimentClient.cs (modified) (11 diffs)

branches/HeuristicLab.Hive-3.4/sources/HeuristicLab.Clients.Hive/3.4/ExperimentManager/HiveExperimentClient.cs

@@ -114 +114 @@
     }
 
-    private List<Guid> pluginsNeededIds;
-    public List<Guid> PluginsNeededIds {
-      get { return pluginsNeededIds; }
-      set { pluginsNeededIds = value; } 
+    private IEnumerable<Plugin> onlinePlugins;
+    public IEnumerable<Plugin> OnlinePlugins {
+      get { return onlinePlugins; }
+      set { onlinePlugins = value; } 
+    }
+
+    private List<Plugin> alreadyUploadedPlugins;
+    public List<Plugin> AlreadyUploadedPlugins {
+      get { return alreadyUploadedPlugins; }
+      set { alreadyUploadedPlugins = value; }
     }
 
@@ -247 +253 @@
 
           this.progress.Status = "Uploading plugins...";
-          this.PluginsNeededIds = GetPluginsNeededIds(this.useLocalPlugins);
+          this.OnlinePlugins = service.Obj.GetPlugins();
+          this.AlreadyUploadedPlugins = new List<Plugin>();
 
           this.progress.Status = "Uploading jobs...";
@@ -269 +276 @@
         IsProgressing = false;
       }
-    }
-
-    /// <summary>
-    /// Gets the Ids of all plugins needed for executing the job.
-    /// All loaded plugins are assumed to be necessary.
-    /// If a plugin with the same name and version is already online, it is used. Otherwise the local plugin is uploaded.
-    /// If useLocalPlugins is true, all local plugins are uploaded regardless of the existence of the same plugin online.
-    /// </summary>
-    public static List<Guid> GetPluginsNeededIds(bool useLocalPlugins) {
-      IEnumerable<IPluginDescription> localPlugins = ApplicationManager.Manager.Plugins;
-      List<Guid> pluginsNeededIds = new List<Guid>();
-
-      using (var service = ServiceLocator.Instance.GetService()) {
-        IEnumerable<Plugin> onlinePlugins = service.Obj.GetPlugins();
-
-        foreach (IPluginDescription localPlugin in localPlugins) {
-          Plugin found = onlinePlugins.Where(onlinePlugin => onlinePlugin.Name == localPlugin.Name && onlinePlugin.Version == localPlugin.Version).SingleOrDefault();
-          if (!useLocalPlugins && found != null) {
-            // plugin is available online; reuse
-            pluginsNeededIds.Add(found.Id);
-          } else {
-            // upload the plugin
-            Plugin p = new Plugin() { Name = localPlugin.Name, Version = localPlugin.Version, IsLocal = useLocalPlugins };
-            List<PluginData> pluginDatas = new List<PluginData>();
-
-            foreach (IPluginFile pf in localPlugin.Files) {
-              PluginData pluginData = new PluginData();
-
-              pluginData.Data = File.ReadAllBytes(pf.Name);
-              pluginDatas.Add(pluginData);
-            }
-            pluginsNeededIds.Add(service.Obj.AddPlugin(p, pluginDatas));
-          }
-        }
-      }
-      return pluginsNeededIds;
     }
 
@@ -319 +290 @@
       this.progress.Status = string.Format("Serializing job {0} of {1}", jobCount, totalJobCount);
       JobData jobData;
+      List<IPluginDescription> plugins;
       if (hiveJob.OptimizerJob.ComputeInParallel &&
         (hiveJob.OptimizerJob.Optimizer is Optimization.Experiment || hiveJob.OptimizerJob.Optimizer is Optimization.BatchRun)) {
         hiveJob.Job.JobState = JobState.WaitingForChildJobs;
         hiveJob.OptimizerJob.CollectChildJobs = false; // don't collect child-jobs on slaves
-        jobData = hiveJob.GetAsJobData(true);
+        jobData = hiveJob.GetAsJobData(true, out plugins);
       } else {
-        jobData = hiveJob.GetAsJobData(false);
-      }
-
-      hiveJob.Job.PluginsNeededIds = this.PluginsNeededIds;
+        jobData = hiveJob.GetAsJobData(false, out plugins);
+      }
+
+      hiveJob.Job.PluginsNeededIds = GetPluginDependencies(service, onlinePlugins, alreadyUploadedPlugins, plugins, useLocalPlugins);
 
       this.progress.Status = string.Format("Uploading job {0} of {1} ({2} kb)", jobCount, totalJobCount, jobData.Data.Count() / 1024);
@@ -347 +319 @@
       }
     }
-
+    
     /// <summary>
     /// Converts a string which can contain Ids separated by ';' to a enumerable
@@ -369 +341 @@
     #endregion
 
-    public void StartResultPolling() {
-      if (!jobResultPoller.IsPolling) {
-        jobResultPoller.Start();
-      } else {
-        throw new JobResultPollingException("Result polling already running");
-      }
-    }
-
-    public void StopResultPolling() {
-      if (jobResultPoller.IsPolling) {
-        jobResultPoller.Stop();
-      } else {
-        throw new JobResultPollingException("Result polling not running");
-      }
-    }
 
     #region HiveJob Events
@@ -483 +440 @@
 
     #region JobResultPoller Events
+
+    public void StartResultPolling() {
+      if (!jobResultPoller.IsPolling) {
+        jobResultPoller.Start();
+      } else {
+        throw new JobResultPollingException("Result polling already running");
+      }
+    }
+
+    public void StopResultPolling() {
+      if (jobResultPoller.IsPolling) {
+        jobResultPoller.Stop();
+      } else {
+        throw new JobResultPollingException("Result polling not running");
+      }
+    }
+
     private void RegisterResultPollingEvents() {
       jobResultPoller.ExceptionOccured += new EventHandler<EventArgs<Exception>>(jobResultPoller_ExceptionOccured);
@@ -497 +471 @@
       jobResultPoller.IsPollingChanged -= new EventHandler(jobResultPoller_IsPollingChanged);
     }
-    void jobResultPoller_IsPollingChanged(object sender, EventArgs e) {
+    private void jobResultPoller_IsPollingChanged(object sender, EventArgs e) {
       this.IsPollingResults = jobResultPoller.IsPolling;
     }
-    void jobResultPoller_PollingFinished(object sender, EventArgs e) {
+    private void jobResultPoller_PollingFinished(object sender, EventArgs e) {
       LogMessage("Polling results finished");
     }
-    void jobResultPoller_PollingStarted(object sender, EventArgs e) {
+    private void jobResultPoller_PollingStarted(object sender, EventArgs e) {
       LogMessage("Polling results started");
     }
-    void jobResultPoller_JobResultReceived(object sender, EventArgs<IEnumerable<LightweightJob>> e) {
+    private void jobResultPoller_JobResultReceived(object sender, EventArgs<IEnumerable<LightweightJob>> e) {
       foreach (LightweightJob lightweightJob in e.Value) {
         HiveJob hj = hiveJob.GetHiveJobByJobId(lightweightJob.Id);
@@ -543 +517 @@
     }
 
-    void jobResultPoller_ExceptionOccured(object sender, EventArgs<Exception> e) {
+    private void jobResultPoller_ExceptionOccured(object sender, EventArgs<Exception> e) {
       OnExceptionOccured(e.Value);
     }
@@ -578 +552 @@
     #endregion
 
+    #region Job Loading
     /// <summary>
     /// Downloads the root job from hive and sets the experiment, rootJob and rootJobItem
@@ -674 +649 @@
       }
     }
-    
+    #endregion 
+
+    #region Plugin Management
+    /// <summary>
+    /// Checks if plugins are available on Hive Server. If not they are uploaded. Ids are returned.
+    /// </summary>
+    /// <param name="service">An active service-proxy</param>
+    /// <param name="onlinePlugins">List of plugins which are available online</param>
+    /// <param name="alreadyUploadedPlugins">List of plugins which have been uploaded from this HiveExperiment</param>
+    /// <param name="neededPlugins">List of plugins which need to be uploaded</param>
+    /// <param name="useLocalPlugins">If true, the plugins which are already online are ignored. All local plugins are uploaded, but only once.</param>
+    /// <returns></returns>
+    private static List<Guid> GetPluginDependencies(IHiveService service, IEnumerable<Plugin> onlinePlugins, List<Plugin> alreadyUploadedPlugins, IEnumerable<IPluginDescription> neededPlugins, bool useLocalPlugins) {
+      var pluginIds = new List<Guid>();
+      foreach (var neededPlugin in neededPlugins) {
+        Plugin foundPlugin = alreadyUploadedPlugins.SingleOrDefault(p => p.Name == neededPlugin.Name && p.Version == neededPlugin.Version);
+        if (foundPlugin == null) {
+          foundPlugin = onlinePlugins.SingleOrDefault(p => p.Name == neededPlugin.Name && p.Version == neededPlugin.Version);
+          if (useLocalPlugins || foundPlugin == null) {
+            Plugin p = CreatePlugin(neededPlugin, useLocalPlugins);
+            List<PluginData> pd = CreatePluginDatas(neededPlugin);
+            p.Id = service.AddPlugin(p, pd);
+            alreadyUploadedPlugins.Add(p);
+          } else {
+            pluginIds.Add(foundPlugin.Id);
+          }
+        } else {
+          pluginIds.Add(foundPlugin.Id);
+        }
+      }
+      return pluginIds;
+    }
+
+    private static Plugin CreatePlugin(IPluginDescription plugin, bool useLocalPlugins) {
+      return new Plugin() { Name = plugin.Name, Version = plugin.Version, IsLocal = useLocalPlugins };
+    }
+
+    private static List<PluginData> CreatePluginDatas(IPluginDescription plugin) {
+      List<PluginData> pluginDatas = new List<PluginData>();
+
+      foreach (IPluginFile pf in plugin.Files) {
+        PluginData pluginData = new PluginData();
+
+        pluginData.Data = File.ReadAllBytes(pf.Name);
+        pluginData.FileName = Path.GetFileName(pf.Name);
+        pluginDatas.Add(pluginData);
+      }
+      return pluginDatas;
+    }
+
+    /// <summary>
+    /// Gets the Ids of all plugins needed for executing the job.
+    /// All loaded plugins are assumed to be necessary.
+    /// If a plugin with the same name and version is already online, it is used. Otherwise the local plugin is uploaded.
+    /// If useLocalPlugins is true, all local plugins are uploaded regardless of the existence of the same plugin online.
+    /// </summary>
+    //public static List<Guid> GetPluginsNeededIds(bool useLocalPlugins) {
+    //  IEnumerable<IPluginDescription> localPlugins = ApplicationManager.Manager.Plugins;
+    //  List<Guid> pluginsNeededIds = new List<Guid>();
+
+    //  using (var service = ServiceLocator.Instance.GetService()) {
+    //    IEnumerable<Plugin> onlinePlugins = service.Obj.GetPlugins();
+
+    //    foreach (IPluginDescription localPlugin in localPlugins) {
+    //      Plugin found = onlinePlugins.Where(onlinePlugin => onlinePlugin.Name == localPlugin.Name && onlinePlugin.Version == localPlugin.Version).SingleOrDefault();
+    //      if (!useLocalPlugins && found != null) {
+    //        // plugin is available online; reuse
+    //        pluginsNeededIds.Add(found.Id);
+    //      } else {
+    //        // upload the plugin
+    //        Plugin p = new Plugin() { Name = localPlugin.Name, Version = localPlugin.Version, IsLocal = useLocalPlugins };
+    //        List<PluginData> pluginDatas = new List<PluginData>();
+
+    //        foreach (IPluginFile pf in localPlugin.Files) {
+    //          PluginData pluginData = new PluginData();
+
+    //          pluginData.Data = File.ReadAllBytes(pf.Name);
+    //          pluginDatas.Add(pluginData);
+    //        }
+    //        pluginsNeededIds.Add(service.Obj.AddPlugin(p, pluginDatas));
+    //      }
+    //    }
+    //  }
+    //  return pluginsNeededIds;
+    //}
+    #endregion
   }
 }