#region License Information /* HeuristicLab * Copyright (C) 2002-2015 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.MainForm; using HeuristicLab.PluginInfrastructure; using TS = System.Threading.Tasks; using Microsoft.AspNetCore.Hosting; namespace HeuristicLab.Clients.Hive.WebJobManager.Services.Imports { /// /// Rewritten to use the HiveServiceLocatorWeb /// [Item("HiveClientWeb", "Hive client.")] public sealed class HiveClientWeb : IContent { public IHostingEnvironment CurrentEnv { get; set; } //Current building area #region Properties private HiveItemCollection jobs; public HiveItemCollection Jobs { get { return jobs; } set { if (value != jobs) { jobs = value; OnHiveJobsChanged(); } } } 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 private HiveServiceLocatorWeb serviceLocator; public Guid UserId { get; set; } public HiveClientWeb(HiveServiceLocatorWeb serv, Guid u) { serviceLocator = serv; UserId = u; } /// /// Reaches to the webloginservice to get the current instance of the servicelocator. /// public void updateServiceLocator() {// No idea if needed serviceLocator = WebLoginService.Instance.getServiceLocator(UserId); } public void ClearHiveClientWeb() { Jobs.ClearWithoutHiveDeletion(); foreach (var j in Jobs) { if (j.RefreshAutomatically) { j.RefreshAutomatically = false; // stop result polling } j.Dispose(); } Jobs = null; if (onlinePlugins != null) onlinePlugins.Clear(); if (alreadyUploadedPlugins != null) alreadyUploadedPlugins.Clear(); } #region Refresh public void Refresh() { OnRefreshing(); try { jobs = new HiveItemCollection(); var jobsLoaded = serviceLocator.CallHiveService>(s => s.GetJobs()); foreach (var j in jobsLoaded) { jobs.Add(new RefreshableJob(j)); } } catch { jobs = 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 void Store(IHiveItem item, CancellationToken cancellationToken) { if (item.Id == Guid.Empty) { if (item is RefreshableJob) { this.UploadJob((RefreshableJob)item, cancellationToken); } if (item is JobPermission) { var hep = (JobPermission)item; hep.GrantedUserId = serviceLocator.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.CallHiveService((s) => s.GrantPermission(hep.JobId, hep.GrantedUserId, hep.Permission)); } } else { if (item is Job) serviceLocator.CallHiveService(s => s.UpdateJob((Job)item)); } } public 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 void Delete(IHiveItem item) { if (item.Id == Guid.Empty && item.GetType() != typeof(JobPermission)) return; if (item is Job) serviceLocator.CallHiveService(s => s.DeleteJob(item.Id)); if (item is RefreshableJob) { RefreshableJob job = (RefreshableJob)item; if (job.RefreshAutomatically) { job.StopResultPolling(); } serviceLocator.CallHiveService(s => s.DeleteJob(item.Id)); } if (item is JobPermission) { var hep = (JobPermission)item; serviceLocator.CallHiveService(s => s.RevokePermission(hep.JobId, 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 HiveJobsChanged; private void OnHiveJobsChanged() { var handler = HiveJobsChanged; if (handler != null) handler(this, EventArgs.Empty); } #endregion public void StartJob(Action exceptionCallback, RefreshableJob refreshableJob, CancellationToken cancellationToken) { this.StoreAsync( new Action((Exception ex) => { exceptionCallback(ex); }), refreshableJob, cancellationToken); } public void PauseJob(RefreshableJob refreshableJob) { serviceLocator.CallHiveService(service => { foreach (HiveTask task in refreshableJob.GetAllHiveTasks()) { if (task.Task.State != TaskState.Finished && task.Task.State != TaskState.Aborted && task.Task.State != TaskState.Failed) service.PauseTask(task.Task.Id); } }); } public void StopJob(RefreshableJob refreshableJob) { serviceLocator.CallHiveService(service => { foreach (HiveTask task in refreshableJob.GetAllHiveTasks()) { if (task.Task.State != TaskState.Finished && task.Task.State != TaskState.Aborted && task.Task.State != TaskState.Failed) service.StopTask(task.Task.Id); } }); } public void ResumeJob(RefreshableJob refreshableJob) { serviceLocator.CallHiveService(service => { foreach (HiveTask task in refreshableJob.GetAllHiveTasks()) { if (task.Task.State == TaskState.Paused) { service.RestartTask(task.Task.Id); } } }); } #region Upload Job private Semaphore taskUploadSemaphore = new Semaphore(5, 5); private static object jobCountLocker = new object(); private static object pluginLocker = new object(); private void UploadJob(RefreshableJob refreshableJob, CancellationToken cancellationToken) { try { refreshableJob.IsProgressing = true; refreshableJob.Progress.Start("Connecting to server..."); IEnumerable resourceNames = ToResourceNameList(refreshableJob.Job.ResourceNames); var resourceIds = new List(); foreach (var resourceName in resourceNames) { Guid resourceId = serviceLocator.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 (OptimizerHiveTask hiveJob in refreshableJob.HiveTasks.OfType()) { hiveJob.SetIndexInParentOptimizerList(null); } // upload Job refreshableJob.Progress.Status = "Uploading Job..."; refreshableJob.Job.Id = serviceLocator.CallHiveService((s) => s.AddJob(refreshableJob.Job)); refreshableJob.Job = serviceLocator.CallHiveService((s) => s.GetJob(refreshableJob.Job.Id)); // update owner and permissions cancellationToken.ThrowIfCancellationRequested(); int totalJobCount = refreshableJob.GetAllHiveTasks().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 refreshableJob.Progress.Status = "Uploading plugins..."; this.OnlinePlugins = serviceLocator.CallHiveService((s) => s.GetPlugins()); this.AlreadyUploadedPlugins = new List(); Plugin configFilePlugin = serviceLocator.CallHiveService((s) => UploadConfigurationFile(s, onlinePlugins)); this.alreadyUploadedPlugins.Add(configFilePlugin); cancellationToken.ThrowIfCancellationRequested(); // upload tasks refreshableJob.Progress.Status = "Uploading tasks..."; var tasks = new List(); foreach (HiveTask hiveTask in refreshableJob.HiveTasks) { var task = TS.Task.Factory.StartNew((hj) => { UploadTaskWithChildren(refreshableJob.Progress, (HiveTask)hj, null, resourceIds, jobCount, totalJobCount, configFilePlugin.Id, refreshableJob.Job.Id, refreshableJob.Log, cancellationToken); }, hiveTask); task.ContinueWith((x) => refreshableJob.Log.LogException(x.Exception), TaskContinuationOptions.OnlyOnFaulted); tasks.Add(task); } TS.Task.WaitAll(tasks.ToArray()); } finally { refreshableJob.IsProgressing = false; refreshableJob.Progress.Finish(); } } /// /// Uploads the local configuration file as plugin /// private Plugin UploadConfigurationFile(IHiveService service, List onlinePlugins) { string exeFilePath = Path.Combine( CurrentEnv.WebRootPath, "bin" , HeuristicLab.Clients.Hive.Settings.Default.HLBinaryName); 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 task and all its child-jobs while setting the proper parentJobId values for the childs /// /// shall be null if its the root task private void UploadTaskWithChildren(IProgress progress, HiveTask hiveTask, HiveTask parentHiveTask, IEnumerable groups, int[] taskCount, int totalJobCount, Guid configPluginId, Guid jobId, ILog log, CancellationToken cancellationToken) { taskUploadSemaphore.WaitOne(); bool semaphoreReleased = false; try { cancellationToken.ThrowIfCancellationRequested(); lock (jobCountLocker) { taskCount[0]++; } TaskData taskData; List plugins; if (hiveTask.ItemTask.ComputeInParallel) { hiveTask.Task.IsParentTask = true; hiveTask.Task.FinishWhenChildJobsFinished = true; taskData = hiveTask.GetAsTaskData(true, out plugins); } else { hiveTask.Task.IsParentTask = false; hiveTask.Task.FinishWhenChildJobsFinished = false; taskData = hiveTask.GetAsTaskData(false, out plugins); } cancellationToken.ThrowIfCancellationRequested(); TryAndRepeat(() => { if (!cancellationToken.IsCancellationRequested) { lock (pluginLocker) { serviceLocator.CallHiveService((s) => hiveTask.Task.PluginsNeededIds = PluginUtil.GetPluginDependencies(s, this.onlinePlugins, this.alreadyUploadedPlugins, plugins)); } } }, 5, "Failed to upload plugins"); cancellationToken.ThrowIfCancellationRequested(); hiveTask.Task.PluginsNeededIds.Add(configPluginId); hiveTask.Task.JobId = jobId; log.LogMessage(string.Format("Uploading task ({0} kb, {1} objects)", taskData.Data.Count() / 1024, hiveTask.ItemTask.GetObjectGraphObjects().Count())); TryAndRepeat(() => { if (!cancellationToken.IsCancellationRequested) { if (parentHiveTask != null) { hiveTask.Task.Id = serviceLocator.CallHiveService((s) => s.AddChildTask(parentHiveTask.Task.Id, hiveTask.Task, taskData)); } else { hiveTask.Task.Id = serviceLocator.CallHiveService((s) => s.AddTask(hiveTask.Task, taskData, groups.ToList())); } } }, 5, "Failed to add task", log); cancellationToken.ThrowIfCancellationRequested(); lock (jobCountLocker) { progress.ProgressValue = (double)taskCount[0] / totalJobCount; progress.Status = string.Format("Uploaded task ({0} of {1})", taskCount[0], totalJobCount); } var tasks = new List(); foreach (HiveTask child in hiveTask.ChildHiveTasks) { var task = TS.Task.Factory.StartNew((tuple) => { var arguments = (Tuple)tuple; UploadTaskWithChildren(progress, arguments.Item1, arguments.Item2, groups, taskCount, totalJobCount, configPluginId, jobId, log, cancellationToken); }, new Tuple(child, hiveTask)); task.ContinueWith((x) => log.LogException(x.Exception), TaskContinuationOptions.OnlyOnFaulted); tasks.Add(task); } taskUploadSemaphore.Release(); semaphoreReleased = true; // the semaphore has to be release before waitall! TS.Task.WaitAll(tasks.ToArray()); } finally { if (!semaphoreReleased) taskUploadSemaphore.Release(); } } #endregion #region Download Experiment public void LoadJob(RefreshableJob refreshableJob) { var hiveExperiment = refreshableJob.Job; refreshableJob.IsProgressing = true; TaskDownloaderWeb downloader = null; try { int totalJobCount = 0; IEnumerable allTasks; // fetch all task objects to create the full tree of tree of HiveTask objects refreshableJob.Progress.Start("Downloading list of tasks..."); allTasks = serviceLocator.CallHiveService(s => s.GetLightweightJobTasksWithoutStateLog(hiveExperiment.Id)); totalJobCount = allTasks.Count(); refreshableJob.Progress.Status = "Downloading tasks..."; downloader = new TaskDownloaderWeb(allTasks.Select(x => x.Id), serviceLocator); downloader.StartAsync(); while (!downloader.IsFinished) { refreshableJob.Progress.ProgressValue = downloader.FinishedCount / (double)totalJobCount; refreshableJob.Progress.Status = string.Format("Downloading/deserializing tasks... ({0}/{1} finished)", downloader.FinishedCount, totalJobCount); Thread.Sleep(500); if (downloader.IsFaulted) { throw downloader.Exception; } } IDictionary allHiveTasks = downloader.Results; var parents = allHiveTasks.Values.Where(x => !x.Task.ParentTaskId.HasValue); refreshableJob.Progress.Status = "Downloading/deserializing complete. Displaying tasks..."; // build child-task tree foreach (HiveTask hiveTask in parents) { BuildHiveJobTree(hiveTask, allTasks, allHiveTasks); } refreshableJob.HiveTasks = new ItemCollection(parents); refreshableJob.OnLoaded(); } finally { refreshableJob.IsProgressing = false; refreshableJob.Progress.Finish(); if (downloader != null) { downloader.Dispose(); } } } private static void BuildHiveJobTree(HiveTask parentHiveTask, IEnumerable allTasks, IDictionary allHiveTasks) { IEnumerable childTasks = from job in allTasks where job.ParentTaskId.HasValue && job.ParentTaskId.Value == parentHiveTask.Task.Id orderby job.DateCreated ascending select job; foreach (LightweightTask task in childTasks) { HiveTask childHiveTask = allHiveTasks[task.Id]; BuildHiveJobTree(childHiveTask, allTasks, allHiveTasks); parentHiveTask.AddChildHiveTask(childHiveTask); } } #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(new[] { ';' }, StringSplitOptions.RemoveEmptyEntries); } else { return new List(); } } public ItemTask LoadItemJob(Guid jobId) { TaskData taskData = serviceLocator.CallHiveService(s => s.GetTaskData(jobId)); try { return PersistenceUtil.Deserialize(taskData.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 HiveItemCollection GetJobPermissions(Guid jobId) { return serviceLocator.CallHiveService((service) => { IEnumerable jps = service.GetJobPermissions(jobId); foreach (var hep in jps) { hep.UnmodifiedGrantedUserNameUpdate(service.GetUsernameByUserId(hep.GrantedUserId)); } return new HiveItemCollection(jps); }); } } }