#region License Information /* HeuristicLab * Copyright (C) 2002-2011 Heuristic and Evolutionary Algorithms Laboratory (HEAL) * * This file is part of HeuristicLab. * * HeuristicLab is free software: you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation, either version 3 of the License, or * (at your option) any later version. * * HeuristicLab is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with HeuristicLab. If not, see . */ #endregion using System; using System.Collections.Generic; using System.Configuration; using System.IO; using System.Linq; using System.Security.Cryptography; using System.Threading; using System.Threading.Tasks; using HeuristicLab.Common; using HeuristicLab.Core; using HeuristicLab.PluginInfrastructure; namespace HeuristicLab.Clients.Hive { [Item("HiveClient", "Hive client.")] public sealed class HiveClient : IContent { private static HiveClient instance; public static HiveClient Instance { get { if (instance == null) instance = new HiveClient(); return instance; } } #region Properties private ItemCollection hiveExperiments; public ItemCollection HiveExperiments { get { return hiveExperiments; } set { if (value != hiveExperiments) { hiveExperiments = value; OnHiveExperimentsChanged(); } } } private List onlinePlugins; public List OnlinePlugins { get { return onlinePlugins; } set { onlinePlugins = value; } } private List alreadyUploadedPlugins; public List AlreadyUploadedPlugins { get { return alreadyUploadedPlugins; } set { alreadyUploadedPlugins = value; } } #endregion public HiveClient() { } #region Refresh public void Refresh() { OnRefreshing(); try { hiveExperiments = new HiveItemCollection(); var he = ServiceLocator.Instance.CallHiveService>(s => s.GetHiveExperiments()); hiveExperiments.AddRange(he.Select(x => new RefreshableHiveExperiment(x)).OrderBy(x => x.HiveExperiment.Name)); } catch { hiveExperiments = null; throw; } finally { OnRefreshed(); } } public void RefreshAsync(Action exceptionCallback) { var call = new Func(delegate() { try { Refresh(); } catch (Exception ex) { return ex; } return null; }); call.BeginInvoke(delegate(IAsyncResult result) { Exception ex = call.EndInvoke(result); if (ex != null) exceptionCallback(ex); }, null); } #endregion #region Store public static void Store(IHiveItem item, CancellationToken cancellationToken) { if (item.Id == Guid.Empty) { if (item is RefreshableHiveExperiment) { HiveClient.Instance.UploadExperiment((RefreshableHiveExperiment)item, cancellationToken); } if (item is HiveExperimentPermission) { var hep = (HiveExperimentPermission)item; hep.GrantedUserId = ServiceLocator.Instance.CallHiveService((s) => s.GetUserIdByUsername(hep.GrantedUserName)); if (hep.GrantedUserId == Guid.Empty) { throw new ArgumentException(string.Format("The user {0} was not found.", hep.GrantedUserName)); } ServiceLocator.Instance.CallHiveService((s) => s.GrantPermission(hep.HiveExperimentId, hep.GrantedUserId, hep.Permission)); } } else { if (item is HiveExperiment) ServiceLocator.Instance.CallHiveService(s => s.UpdateHiveExperiment((HiveExperiment)item)); } } public static void StoreAsync(Action exceptionCallback, IHiveItem item, CancellationToken cancellationToken) { var call = new Func(delegate() { try { Store(item, cancellationToken); } catch (Exception ex) { return ex; } return null; }); call.BeginInvoke(delegate(IAsyncResult result) { Exception ex = call.EndInvoke(result); if (ex != null) exceptionCallback(ex); }, null); } #endregion #region Delete public static void Delete(IHiveItem item) { if (item is HiveExperiment) ServiceLocator.Instance.CallHiveService(s => s.DeleteHiveExperiment(item.Id)); if (item is RefreshableHiveExperiment) ServiceLocator.Instance.CallHiveService(s => s.DeleteHiveExperiment(item.Id)); if (item is HiveExperimentPermission) { var hep = (HiveExperimentPermission)item; ServiceLocator.Instance.CallHiveService(s => s.RevokePermission(hep.HiveExperimentId, hep.GrantedUserId)); } item.Id = Guid.Empty; } #endregion #region Events public event EventHandler Refreshing; private void OnRefreshing() { EventHandler handler = Refreshing; if (handler != null) handler(this, EventArgs.Empty); } public event EventHandler Refreshed; private void OnRefreshed() { var handler = Refreshed; if (handler != null) handler(this, EventArgs.Empty); } public event EventHandler HiveExperimentsChanged; private void OnHiveExperimentsChanged() { var handler = HiveExperimentsChanged; if (handler != null) handler(this, EventArgs.Empty); } #endregion public static void StartExperiment(Action exceptionCallback, RefreshableHiveExperiment refreshableHiveExperiment, CancellationToken cancellationToken) { HiveClient.StoreAsync( new Action((Exception ex) => { refreshableHiveExperiment.HiveExperiment.ExecutionState = ExecutionState.Prepared; exceptionCallback(ex); }), refreshableHiveExperiment, cancellationToken); refreshableHiveExperiment.HiveExperiment.ExecutionState = ExecutionState.Started; } public static void PauseExperiment(HiveExperiment hiveExperiment) { ServiceLocator.Instance.CallHiveService(service => { foreach (HiveJob job in hiveExperiment.GetAllHiveJobs()) { if (job.Job.State != JobState.Finished && job.Job.State != JobState.Aborted && job.Job.State != JobState.Failed) service.PauseJob(job.Job.Id); } }); hiveExperiment.ExecutionState = ExecutionState.Paused; } public static void StopExperiment(HiveExperiment hiveExperiment) { ServiceLocator.Instance.CallHiveService(service => { foreach (HiveJob job in hiveExperiment.GetAllHiveJobs()) { if (job.Job.State != JobState.Finished && job.Job.State != JobState.Aborted && job.Job.State != JobState.Failed) service.StopJob(job.Job.Id); } }); // execution state does not need to be set. it will be set to Stopped, when all jobs have been downloaded } #region Upload Experiment private Semaphore jobUploadSemaphore = new Semaphore(4, 4); // todo: take magic number into config private static object jobCountLocker = new object(); private static object pluginLocker = new object(); private void UploadExperiment(RefreshableHiveExperiment refreshableHiveExperiment, CancellationToken cancellationToken) { try { refreshableHiveExperiment.HiveExperiment.Progress = new Progress("Connecting to server..."); refreshableHiveExperiment.HiveExperiment.IsProgressing = true; IEnumerable resourceNames = ToResourceNameList(refreshableHiveExperiment.HiveExperiment.ResourceNames); var resourceIds = new List(); foreach (var resourceName in resourceNames) { Guid resourceId = ServiceLocator.Instance.CallHiveService((s) => s.GetResourceId(resourceName)); if (resourceId == Guid.Empty) { throw new ResourceNotFoundException(string.Format("Could not find the resource '{0}'", resourceName)); } resourceIds.Add(resourceId); } foreach (OptimizerHiveJob hiveJob in refreshableHiveExperiment.HiveExperiment.HiveJobs.OfType()) { hiveJob.SetIndexInParentOptimizerList(null); } // upload HiveExperiment refreshableHiveExperiment.HiveExperiment.Progress.Status = "Uploading HiveExperiment..."; refreshableHiveExperiment.HiveExperiment.Id = ServiceLocator.Instance.CallHiveService((s) => s.AddHiveExperiment(refreshableHiveExperiment.HiveExperiment)); cancellationToken.ThrowIfCancellationRequested(); int totalJobCount = refreshableHiveExperiment.HiveExperiment.GetAllHiveJobs().Count(); int[] jobCount = new int[1]; // use a reference type (int-array) instead of value type (int) in order to pass the value via a delegate to task-parallel-library cancellationToken.ThrowIfCancellationRequested(); // upload plugins refreshableHiveExperiment.HiveExperiment.Progress.Status = "Uploading plugins..."; this.OnlinePlugins = ServiceLocator.Instance.CallHiveService((s) => s.GetPlugins()); this.AlreadyUploadedPlugins = new List(); Plugin configFilePlugin = ServiceLocator.Instance.CallHiveService((s) => UploadConfigurationFile(s, onlinePlugins)); this.alreadyUploadedPlugins.Add(configFilePlugin); cancellationToken.ThrowIfCancellationRequested(); if (refreshableHiveExperiment.RefreshAutomatically) refreshableHiveExperiment.StartResultPolling(); // upload jobs refreshableHiveExperiment.HiveExperiment.Progress.Status = "Uploading jobs..."; var tasks = new List(); foreach (HiveJob hiveJob in refreshableHiveExperiment.HiveExperiment.HiveJobs) { tasks.Add(Task.Factory.StartNew((hj) => { UploadJobWithChildren(refreshableHiveExperiment.HiveExperiment.Progress, (HiveJob)hj, null, resourceIds, jobCount, totalJobCount, configFilePlugin.Id, refreshableHiveExperiment.HiveExperiment.Id, refreshableHiveExperiment.Log, refreshableHiveExperiment.HiveExperiment.IsPrivileged, cancellationToken); }, hiveJob) .ContinueWith((x) => refreshableHiveExperiment.Log.LogException(x.Exception), TaskContinuationOptions.OnlyOnFaulted)); } try { Task.WaitAll(tasks.ToArray()); } catch (AggregateException ae) { if (!ae.InnerExceptions.All(e => e is TaskCanceledException)) throw ae; // for some reason the WaitAll throws a AggregateException containg a TaskCanceledException. i don't know where it comes from, however the tasks all finish properly, so for now just ignore it } } finally { refreshableHiveExperiment.HiveExperiment.IsProgressing = false; } } /// /// Uploads the local configuration file as plugin /// private static Plugin UploadConfigurationFile(IHiveService service, List onlinePlugins) { string exeFilePath = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "HeuristicLab 3.3.exe"); string configFileName = Path.GetFileName(ConfigurationManager.OpenExeConfiguration(exeFilePath).FilePath); string configFilePath = ConfigurationManager.OpenExeConfiguration(exeFilePath).FilePath; byte[] hash; byte[] data = File.ReadAllBytes(configFilePath); using (SHA1 sha1 = SHA1.Create()) { hash = sha1.ComputeHash(data); } Plugin configPlugin = new Plugin() { Name = "Configuration", Version = new Version(), Hash = hash }; PluginData configFile = new PluginData() { FileName = configFileName, Data = data }; IEnumerable onlineConfig = onlinePlugins.Where(p => p.Hash.SequenceEqual(hash)); if (onlineConfig.Count() > 0) { return onlineConfig.First(); } else { configPlugin.Id = service.AddPlugin(configPlugin, new List { configFile }); return configPlugin; } } /// /// Uploads the given job and all its child-jobs while setting the proper parentJobId values for the childs /// /// shall be null if its the root job private void UploadJobWithChildren(IProgress progress, HiveJob hiveJob, HiveJob parentHiveJob, IEnumerable groups, int[] jobCount, int totalJobCount, Guid configPluginId, Guid hiveExperimentId, ILog log, bool isPrivileged, CancellationToken cancellationToken) { jobUploadSemaphore.WaitOne(); bool semaphoreReleased = false; try { cancellationToken.ThrowIfCancellationRequested(); lock (jobCountLocker) { jobCount[0]++; } JobData jobData; List plugins; if (hiveJob.ItemJob.ComputeInParallel && (hiveJob.ItemJob.Item is Optimization.Experiment || hiveJob.ItemJob.Item is Optimization.BatchRun)) { hiveJob.Job.IsParentJob = true; hiveJob.Job.FinishWhenChildJobsFinished = true; jobData = hiveJob.GetAsJobData(true, out plugins); } else { hiveJob.Job.IsParentJob = false; hiveJob.Job.FinishWhenChildJobsFinished = false; jobData = hiveJob.GetAsJobData(false, out plugins); } cancellationToken.ThrowIfCancellationRequested(); TryAndRepeat(() => { if (!cancellationToken.IsCancellationRequested) { lock (pluginLocker) { ServiceLocator.Instance.CallHiveService((s) => hiveJob.Job.PluginsNeededIds = PluginUtil.GetPluginDependencies(s, this.onlinePlugins, this.alreadyUploadedPlugins, plugins)); } } }, -1, "Failed to upload plugins"); cancellationToken.ThrowIfCancellationRequested(); hiveJob.Job.PluginsNeededIds.Add(configPluginId); hiveJob.Job.HiveExperimentId = hiveExperimentId; hiveJob.Job.IsPrivileged = isPrivileged; log.LogMessage(string.Format("Uploading job ({0} kb, {1} objects)", jobData.Data.Count() / 1024, hiveJob.ItemJob.GetObjectGraphObjects().Count())); TryAndRepeat(() => { if (!cancellationToken.IsCancellationRequested) { if (parentHiveJob != null) { hiveJob.Job.Id = ServiceLocator.Instance.CallHiveService((s) => s.AddChildJob(parentHiveJob.Job.Id, hiveJob.Job, jobData)); } else { hiveJob.Job.Id = ServiceLocator.Instance.CallHiveService((s) => s.AddJob(hiveJob.Job, jobData, groups.ToList())); } } }, 50, "Failed to add job", log); cancellationToken.ThrowIfCancellationRequested(); lock (jobCountLocker) { progress.ProgressValue = (double)jobCount[0] / totalJobCount; progress.Status = string.Format("Uploaded job ({0} of {1})", jobCount[0], totalJobCount); } var tasks = new List(); foreach (HiveJob child in hiveJob.ChildHiveJobs) { tasks.Add(Task.Factory.StartNew((tuple) => { var arguments = (Tuple)tuple; UploadJobWithChildren(progress, arguments.Item1, arguments.Item2, groups, jobCount, totalJobCount, configPluginId, hiveExperimentId, log, isPrivileged, cancellationToken); }, new Tuple(child, hiveJob)) .ContinueWith((x) => log.LogException(x.Exception), TaskContinuationOptions.OnlyOnFaulted)); } jobUploadSemaphore.Release(); semaphoreReleased = true; // the semaphore has to be release before waitall! try { Task.WaitAll(tasks.ToArray()); } catch (AggregateException ae) { if (!ae.InnerExceptions.All(e => e is TaskCanceledException)) throw ae; // for some reason the WaitAll throws a AggregateException containg a TaskCanceledException. i don't know where it comes from, however the tasks all finish properly, so for now just ignore it } } finally { if(!semaphoreReleased) jobUploadSemaphore.Release(); } } #endregion #region Download Experiment public static void LoadExperiment(HiveExperiment hiveExperiment) { hiveExperiment.Progress = new Progress(); try { hiveExperiment.IsProgressing = true; int totalJobCount = 0; IEnumerable allJobs; hiveExperiment.Progress.Status = "Connecting to Server..."; // fetch all Job objects to create the full tree of tree of HiveJob objects hiveExperiment.Progress.Status = "Downloading list of jobs..."; allJobs = ServiceLocator.Instance.CallHiveService(s => s.GetLightweightExperimentJobs(hiveExperiment.Id)); totalJobCount = allJobs.Count(); HiveJobDownloader downloader = new HiveJobDownloader(allJobs.Select(x => x.Id)); downloader.StartAsync(); while (!downloader.IsFinished) { hiveExperiment.Progress.ProgressValue = downloader.FinishedCount / (double)totalJobCount; hiveExperiment.Progress.Status = string.Format("Downloading/deserializing jobs... ({0}/{1} finished)", downloader.FinishedCount, totalJobCount); Thread.Sleep(500); if (downloader.IsFaulted) { throw downloader.Exception; } } IDictionary allHiveJobs = downloader.Results; hiveExperiment.HiveJobs = new ItemCollection(allHiveJobs.Values.Where(x => !x.Job.ParentJobId.HasValue)); if (hiveExperiment.IsFinished()) { hiveExperiment.ExecutionState = Core.ExecutionState.Stopped; } else { hiveExperiment.ExecutionState = Core.ExecutionState.Started; } // build child-job tree foreach (HiveJob hiveJob in hiveExperiment.HiveJobs) { BuildHiveJobTree(hiveJob, allJobs, allHiveJobs); } hiveExperiment.OnLoaded(); } finally { hiveExperiment.IsProgressing = false; } } private static void BuildHiveJobTree(HiveJob parentHiveJob, IEnumerable allJobs, IDictionary allHiveJobs) { IEnumerable childJobs = from job in allJobs where job.ParentJobId.HasValue && job.ParentJobId.Value == parentHiveJob.Job.Id orderby job.DateCreated ascending select job; foreach (LightweightJob job in childJobs) { HiveJob childHiveJob = allHiveJobs[job.Id]; parentHiveJob.AddChildHiveJob(childHiveJob); BuildHiveJobTree(childHiveJob, allJobs, allHiveJobs); } } #endregion /// /// Converts a string which can contain Ids separated by ';' to a enumerable /// private static IEnumerable ToResourceNameList(string resourceNames) { if (!string.IsNullOrEmpty(resourceNames)) { return resourceNames.Split(';'); } else { return new List(); } } public static ItemJob LoadItemJob(Guid jobId) { JobData jobData = ServiceLocator.Instance.CallHiveService(s => s.GetJobData(jobId)); try { return PersistenceUtil.Deserialize(jobData.Data); } catch { return null; } } /// /// Executes the action. If it throws an exception it is repeated until repetition-count is reached. /// If repetitions is -1, it is repeated infinitely. /// public static void TryAndRepeat(Action action, int repetitions, string errorMessage, ILog log = null) { while (true) { try { action(); return; } catch (Exception e) { if (repetitions == 0) throw new HiveException(errorMessage, e); if (log != null) log.LogMessage(string.Format("{0}: {1} - will try again!", errorMessage, e.ToString())); repetitions--; } } } public static HiveItemCollection GetHiveExperimentPermissions(Guid hiveExperimentId) { return ServiceLocator.Instance.CallHiveService((service) => { IEnumerable heps = service.GetHiveExperimentPermissions(hiveExperimentId); foreach (var hep in heps) { hep.GrantedUserName = service.GetUsernameByUserId(hep.GrantedUserId); } return new HiveItemCollection(heps); }); } } }