source: branches/2839_HiveProjectManagement/HeuristicLab.Clients.Hive/3.3/HiveClient.cs @ 15913

Last change on this file since 15913 was 15913, checked in by jzenisek, 4 years ago

#2839: updated genealogy computation for hive job administrator

File size: 26.0 KB
Line 
1#region License Information
2/* HeuristicLab
3 * Copyright (C) 2002-2016 Heuristic and Evolutionary Algorithms Laboratory (HEAL)
4 *
5 * This file is part of HeuristicLab.
6 *
7 * HeuristicLab is free software: you can redistribute it and/or modify
8 * it under the terms of the GNU General Public License as published by
9 * the Free Software Foundation, either version 3 of the License, or
10 * (at your option) any later version.
11 *
12 * HeuristicLab is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15 * GNU General Public License for more details.
16 *
17 * You should have received a copy of the GNU General Public License
18 * along with HeuristicLab. If not, see <http://www.gnu.org/licenses/>.
19 */
20#endregion
21
22using System;
23using System.Collections.Generic;
24using System.Configuration;
25using System.IO;
26using System.Linq;
27using System.Security.Cryptography;
28using System.Threading;
29using System.Threading.Tasks;
30using HeuristicLab.Common;
31using HeuristicLab.Core;
32using HeuristicLab.MainForm;
33using HeuristicLab.PluginInfrastructure;
34using TS = System.Threading.Tasks;
35
36namespace HeuristicLab.Clients.Hive {
37  [Item("HiveClient", "Hive client.")]
38  public sealed class HiveClient : IContent {
39    private static HiveClient instance;
40    public static HiveClient Instance {
41      get {
42        if (instance == null) instance = new HiveClient();
43        return instance;
44      }
45    }
46
47    #region Properties
48    private HiveItemCollection<RefreshableJob> jobs;
49    public HiveItemCollection<RefreshableJob> Jobs {
50      get { return jobs; }
51      set {
52        if (value != jobs) {
53          jobs = value;
54          OnHiveJobsChanged();
55        }
56      }
57    }
58
59    private IItemList<Project> projects;
60    public IItemList<Project> Projects {
61      get { return projects; }
62    }
63
64    private IItemList<Resource> resources;
65    public IItemList<Resource> Resources {
66      get { return resources; }
67    }
68
69    private Dictionary<Guid, HashSet<Guid>> projectAncestors;
70    public Dictionary<Guid, HashSet<Guid>> ProjectAncestors {
71      get { return projectAncestors; }
72    }
73
74    private Dictionary<Guid, HashSet<Guid>> projectDescendants;
75    public Dictionary<Guid, HashSet<Guid>> ProjectDescendants {
76      get { return projectDescendants; }
77    }
78
79    private Dictionary<Guid, HashSet<Guid>> resourceAncestors;
80    public Dictionary<Guid, HashSet<Guid>> ResourceAncestors {
81      get { return resourceAncestors; }
82    }
83
84    private Dictionary<Guid, HashSet<Guid>> resourceDescendants;
85    public Dictionary<Guid, HashSet<Guid>> ResourceDescendants {
86      get { return resourceDescendants; }
87    }
88
89
90    private List<Plugin> onlinePlugins;
91    public List<Plugin> OnlinePlugins {
92      get { return onlinePlugins; }
93      set { onlinePlugins = value; }
94    }
95
96    private List<Plugin> alreadyUploadedPlugins;
97    public List<Plugin> AlreadyUploadedPlugins {
98      get { return alreadyUploadedPlugins; }
99      set { alreadyUploadedPlugins = value; }
100    }
101    #endregion
102
103    private HiveClient() { }
104
105    public void ClearHiveClient() {
106      Jobs.ClearWithoutHiveDeletion();
107      foreach (var j in Jobs) {
108        if (j.RefreshAutomatically) {
109          j.RefreshAutomatically = false; // stop result polling
110        }
111        j.Dispose();
112      }
113      Jobs = null;
114
115      if (onlinePlugins != null)
116        onlinePlugins.Clear();
117      if (alreadyUploadedPlugins != null)
118        alreadyUploadedPlugins.Clear();
119    }
120
121    #region Refresh
122    public void Refresh() {
123      OnRefreshing();
124
125      try {
126        projects = new ItemList<Project>();
127        resources = new ItemList<Resource>();
128        jobs = new HiveItemCollection<RefreshableJob>();
129
130        projectAncestors = new Dictionary<Guid, HashSet<Guid>>();
131        projectDescendants = new Dictionary<Guid, HashSet<Guid>>();
132        resourceAncestors = new Dictionary<Guid, HashSet<Guid>>();
133        resourceDescendants = new Dictionary<Guid, HashSet<Guid>>();
134
135        HiveServiceLocator.Instance.CallHiveService(service => {
136          service.GetProjects().ForEach(p => projects.Add(p));
137          service.GetSlaveGroups().ForEach(g => resources.Add(g));
138          service.GetSlaves().ForEach(s => resources.Add(s));
139          service.GetJobs().ForEach(p => jobs.Add(new RefreshableJob(p)));
140        });
141
142        UpdateResourceGenealogy();
143        UpdateProjectGenealogy();
144      }
145      catch {
146        jobs = null;
147        projects = null;
148        resources = null;
149        throw;
150      }
151      finally {
152        OnRefreshed();
153      }
154    }
155
156    public void RefreshProjectsAndResources() {
157      OnRefreshing();
158
159      try {
160        projects = new ItemList<Project>();
161        resources = new ItemList<Resource>();
162
163        projectAncestors = new Dictionary<Guid, HashSet<Guid>>();
164        projectDescendants = new Dictionary<Guid, HashSet<Guid>>();
165        resourceAncestors = new Dictionary<Guid, HashSet<Guid>>();
166        resourceDescendants = new Dictionary<Guid, HashSet<Guid>>();
167
168        HiveServiceLocator.Instance.CallHiveService(service => {
169          service.GetProjects().ForEach(p => projects.Add(p));
170          service.GetSlaveGroups().ForEach(g => resources.Add(g));
171          service.GetSlaves().ForEach(s => resources.Add(s));
172        });
173
174        UpdateResourceGenealogy();
175        UpdateProjectGenealogy();
176      } catch {
177        projects = null;
178        resources = null;
179        throw;
180      } finally {
181        OnRefreshed();
182      }
183    }
184
185    public void RefreshAsync(Action<Exception> exceptionCallback) {
186      var call = new Func<Exception>(delegate() {
187        try {
188          Refresh();
189        }
190        catch (Exception ex) {
191          return ex;
192        }
193        return null;
194      });
195      call.BeginInvoke(delegate(IAsyncResult result) {
196        Exception ex = call.EndInvoke(result);
197        if (ex != null) exceptionCallback(ex);
198      }, null);
199    }
200
201    private void UpdateResourceGenealogy() {
202      resourceAncestors.Clear();
203      resourceDescendants.Clear();
204
205      // fetch resource ancestor set
206      HiveServiceLocator.Instance.CallHiveService(service => {
207        var ra = service.GetResourceGenealogy();
208        ra.Keys.ToList().ForEach(k => resourceAncestors.Add(k, new HashSet<Guid>()));
209        resourceAncestors.Keys.ToList().ForEach(k => resourceAncestors[k].UnionWith(ra[k]));
210      });
211
212      // build resource descendant set
213      resourceAncestors.Keys.ToList().ForEach(k => resourceDescendants.Add(k, new HashSet<Guid>()));
214      foreach (var ra in resourceAncestors) {
215        foreach(var ancestor in ra.Value) {
216          resourceDescendants[ancestor].Add(ra.Key);
217        }
218      }
219    }
220
221    private void UpdateProjectGenealogy() {
222      projectAncestors.Clear();
223      projectDescendants.Clear();
224
225      // fetch project ancestor list
226      HiveServiceLocator.Instance.CallHiveService(service => {
227        var pa = service.GetProjectGenealogy();
228        pa.Keys.ToList().ForEach(k => projectAncestors.Add(k, new HashSet<Guid>()));
229        projectAncestors.Keys.ToList().ForEach(k => projectAncestors[k].UnionWith(pa[k]));
230      });
231
232      // build project descendant list
233      projectAncestors.Keys.ToList().ForEach(k => projectDescendants.Add(k, new HashSet<Guid>()));
234      foreach(var pa in projectAncestors) {
235        foreach(var ancestor in pa.Value) {
236          projectDescendants[ancestor].Add(pa.Key);
237        }
238      }
239    }
240
241    public IEnumerable<Project> GetAvailableParentProjects(Guid id) {
242      return projects.Where(x => projectAncestors[id].Contains(x.Id));
243    }
244
245    public IEnumerable<Project> GetAvailableChildProjects(Guid id) {
246      return projects.Where(x => projectDescendants[id].Contains(x.Id));
247    }
248
249    public IEnumerable<Resource> GetAvailableParentResources(Guid id) {
250      return resources.Where(x => resourceAncestors[id].Contains(x.Id));
251    }
252
253    public IEnumerable<Resource> GetAvailableChildResources(Guid id) {
254      return resources.Where(x => resourceDescendants[id].Contains(x.Id));
255    }
256
257    #endregion
258
259    #region Store
260    public static void Store(IHiveItem item, CancellationToken cancellationToken) {
261      if (item.Id == Guid.Empty) {
262        if (item is RefreshableJob) {
263          item.Id = HiveClient.Instance.UploadJob((RefreshableJob)item, cancellationToken);
264        }
265        if (item is JobPermission) {
266          var hep = (JobPermission)item;
267          hep.GrantedUserId = HiveServiceLocator.Instance.CallHiveService((s) => s.GetUserIdByUsername(hep.GrantedUserName));
268          if (hep.GrantedUserId == Guid.Empty) {
269            throw new ArgumentException(string.Format("The user {0} was not found.", hep.GrantedUserName));
270          }
271          HiveServiceLocator.Instance.CallHiveService((s) => s.GrantPermission(hep.JobId, hep.GrantedUserId, hep.Permission));
272        }
273        if (item is Project) {
274          item.Id = HiveServiceLocator.Instance.CallHiveService(s => s.AddProject((Project)item));
275        }
276      } else {
277        if (item is Job) {
278          var job = (Job)item;
279          HiveServiceLocator.Instance.CallHiveService(s => s.UpdateJob(job, job.ResourceIds));
280        }
281        if (item is Project)
282          HiveServiceLocator.Instance.CallHiveService(s => s.UpdateProject((Project)item));
283      }
284    }
285    public static void StoreAsync(Action<Exception> exceptionCallback, IHiveItem item, CancellationToken cancellationToken) {
286      var call = new Func<Exception>(delegate() {
287        try {
288          Store(item, cancellationToken);
289        }
290        catch (Exception ex) {
291          return ex;
292        }
293        return null;
294      });
295      call.BeginInvoke(delegate(IAsyncResult result) {
296        Exception ex = call.EndInvoke(result);
297        if (ex != null) exceptionCallback(ex);
298      }, null);
299    }
300    #endregion
301
302    #region Delete
303    public static void Delete(IHiveItem item) {
304      if (item.Id == Guid.Empty && item.GetType() != typeof(JobPermission))
305        return;
306
307      if (item is Job)
308        HiveServiceLocator.Instance.CallHiveService(s => s.UpdateJobState(item.Id, JobState.StatisticsPending));
309      if (item is RefreshableJob) {
310        RefreshableJob job = (RefreshableJob)item;
311        if (job.RefreshAutomatically) {
312          job.StopResultPolling();
313        }
314        HiveServiceLocator.Instance.CallHiveService(s => s.UpdateJobState(item.Id, JobState.StatisticsPending));
315      }
316      if (item is JobPermission) {
317        var hep = (JobPermission)item;
318        HiveServiceLocator.Instance.CallHiveService(s => s.RevokePermission(hep.JobId, hep.GrantedUserId));
319      }
320      item.Id = Guid.Empty;
321    }
322    #endregion
323
324    #region Events
325    public event EventHandler Refreshing;
326    private void OnRefreshing() {
327      EventHandler handler = Refreshing;
328      if (handler != null) handler(this, EventArgs.Empty);
329    }
330    public event EventHandler Refreshed;
331    private void OnRefreshed() {
332      var handler = Refreshed;
333      if (handler != null) handler(this, EventArgs.Empty);
334    }
335    public event EventHandler HiveJobsChanged;
336    private void OnHiveJobsChanged() {
337      var handler = HiveJobsChanged;
338      if (handler != null) handler(this, EventArgs.Empty);
339    }
340    #endregion
341
342    public static void StartJob(Action<Exception> exceptionCallback, RefreshableJob refreshableJob, CancellationToken cancellationToken) {
343      HiveClient.StoreAsync(
344        new Action<Exception>((Exception ex) => {
345          refreshableJob.ExecutionState = ExecutionState.Prepared;
346          exceptionCallback(ex);
347        }), refreshableJob, cancellationToken);
348      refreshableJob.ExecutionState = ExecutionState.Started;
349    }
350
351    public static void PauseJob(RefreshableJob refreshableJob) {
352      HiveServiceLocator.Instance.CallHiveService(service => {
353        foreach (HiveTask task in refreshableJob.GetAllHiveTasks()) {
354          if (task.Task.State != TaskState.Finished && task.Task.State != TaskState.Aborted && task.Task.State != TaskState.Failed)
355            service.PauseTask(task.Task.Id);
356        }
357      });
358      refreshableJob.ExecutionState = ExecutionState.Paused;
359    }
360
361    public static void StopJob(RefreshableJob refreshableJob) {
362      HiveServiceLocator.Instance.CallHiveService(service => {
363        foreach (HiveTask task in refreshableJob.GetAllHiveTasks()) {
364          if (task.Task.State != TaskState.Finished && task.Task.State != TaskState.Aborted && task.Task.State != TaskState.Failed)
365            service.StopTask(task.Task.Id);
366        }
367      });
368      refreshableJob.ExecutionState = ExecutionState.Stopped;
369    }
370
371    public static void ResumeJob(RefreshableJob refreshableJob) {
372      HiveServiceLocator.Instance.CallHiveService(service => {
373        foreach (HiveTask task in refreshableJob.GetAllHiveTasks()) {
374          if (task.Task.State == TaskState.Paused) {
375            service.RestartTask(task.Task.Id);
376          }
377        }
378      });
379      refreshableJob.ExecutionState = ExecutionState.Started;
380    }
381
382    public static void UpdateJob(Action<Exception> exceptionCallback, RefreshableJob refreshableJob, CancellationToken cancellationToken) {
383      HiveServiceLocator.Instance.CallHiveService(service => {
384        HiveClient.StoreAsync(
385          new Action<Exception>((Exception ex) => {
386            //refreshableJob.ExecutionState = ExecutionState.Prepared;
387            exceptionCallback(ex);
388          }), refreshableJob.Job, cancellationToken);
389      });
390    }
391
392    #region Upload Job
393    private Semaphore taskUploadSemaphore = new Semaphore(Settings.Default.MaxParallelUploads, Settings.Default.MaxParallelUploads);
394    private static object jobCountLocker = new object();
395    private static object pluginLocker = new object();
396    private Guid UploadJob(RefreshableJob refreshableJob, CancellationToken cancellationToken) {
397      try {
398        refreshableJob.IsProgressing = true;
399        refreshableJob.Progress.Start("Connecting to server...");
400
401        foreach (OptimizerHiveTask hiveJob in refreshableJob.HiveTasks.OfType<OptimizerHiveTask>()) {
402          hiveJob.SetIndexInParentOptimizerList(null);
403        }
404
405        // upload Job
406        refreshableJob.Progress.Status = "Uploading Job...";
407        refreshableJob.Job.Id = HiveServiceLocator.Instance.CallHiveService((s) => s.AddJob(refreshableJob.Job, refreshableJob.Job.ResourceIds));
408        refreshableJob.Job = HiveServiceLocator.Instance.CallHiveService((s) => s.GetJob(refreshableJob.Job.Id)); // update owner and permissions
409        cancellationToken.ThrowIfCancellationRequested();
410
411        int totalJobCount = refreshableJob.GetAllHiveTasks().Count();
412        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
413        cancellationToken.ThrowIfCancellationRequested();
414
415        // upload plugins
416        refreshableJob.Progress.Status = "Uploading plugins...";
417        this.OnlinePlugins = HiveServiceLocator.Instance.CallHiveService((s) => s.GetPlugins());
418        this.AlreadyUploadedPlugins = new List<Plugin>();
419        Plugin configFilePlugin = HiveServiceLocator.Instance.CallHiveService((s) => UploadConfigurationFile(s, onlinePlugins));
420        this.alreadyUploadedPlugins.Add(configFilePlugin);
421        cancellationToken.ThrowIfCancellationRequested();
422
423        // upload tasks
424        refreshableJob.Progress.Status = "Uploading tasks...";
425
426        var tasks = new List<TS.Task>();
427        foreach (HiveTask hiveTask in refreshableJob.HiveTasks) {
428          var task = TS.Task.Factory.StartNew((hj) => {
429            UploadTaskWithChildren(refreshableJob.Progress, (HiveTask)hj, null, jobCount, totalJobCount, configFilePlugin.Id, refreshableJob.Job.Id, refreshableJob.Log, cancellationToken);
430          }, hiveTask);
431          task.ContinueWith((x) => refreshableJob.Log.LogException(x.Exception), TaskContinuationOptions.OnlyOnFaulted);
432          tasks.Add(task);
433        }
434        TS.Task.WaitAll(tasks.ToArray());
435      }
436      finally {
437        refreshableJob.Job.Modified = false;
438        refreshableJob.IsProgressing = false;
439        refreshableJob.Progress.Finish();
440      }
441      return (refreshableJob.Job != null) ? refreshableJob.Job.Id : Guid.Empty;
442    }
443
444    /// <summary>
445    /// Uploads the local configuration file as plugin
446    /// </summary>
447    private static Plugin UploadConfigurationFile(IHiveService service, List<Plugin> onlinePlugins) {
448      string exeFilePath = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, Settings.Default.HLBinaryName);
449      string configFileName = Path.GetFileName(ConfigurationManager.OpenExeConfiguration(exeFilePath).FilePath);
450      string configFilePath = ConfigurationManager.OpenExeConfiguration(exeFilePath).FilePath;
451      byte[] hash;
452
453      byte[] data = File.ReadAllBytes(configFilePath);
454      using (SHA1 sha1 = SHA1.Create()) {
455        hash = sha1.ComputeHash(data);
456      }
457
458      Plugin configPlugin = new Plugin() { Name = "Configuration", Version = new Version(), Hash = hash };
459      PluginData configFile = new PluginData() { FileName = configFileName, Data = data };
460
461      IEnumerable<Plugin> onlineConfig = onlinePlugins.Where(p => p.Hash.SequenceEqual(hash));
462
463      if (onlineConfig.Count() > 0) {
464        return onlineConfig.First();
465      } else {
466        configPlugin.Id = service.AddPlugin(configPlugin, new List<PluginData> { configFile });
467        return configPlugin;
468      }
469    }
470
471    /// <summary>
472    /// Uploads the given task and all its child-jobs while setting the proper parentJobId values for the childs
473    /// </summary>
474    /// <param name="parentHiveTask">shall be null if its the root task</param>
475    private void UploadTaskWithChildren(IProgress progress, HiveTask hiveTask, HiveTask parentHiveTask, int[] taskCount, int totalJobCount, Guid configPluginId, Guid jobId, ILog log, CancellationToken cancellationToken) {
476      taskUploadSemaphore.WaitOne();
477      bool semaphoreReleased = false;
478      try {
479        cancellationToken.ThrowIfCancellationRequested();
480        lock (jobCountLocker) {
481          taskCount[0]++;
482        }
483        TaskData taskData;
484        List<IPluginDescription> plugins;
485
486        if (hiveTask.ItemTask.ComputeInParallel) {
487          hiveTask.Task.IsParentTask = true;
488          hiveTask.Task.FinishWhenChildJobsFinished = true;
489          taskData = hiveTask.GetAsTaskData(true, out plugins);
490        } else {
491          hiveTask.Task.IsParentTask = false;
492          hiveTask.Task.FinishWhenChildJobsFinished = false;
493          taskData = hiveTask.GetAsTaskData(false, out plugins);
494        }
495        cancellationToken.ThrowIfCancellationRequested();
496
497        TryAndRepeat(() => {
498          if (!cancellationToken.IsCancellationRequested) {
499            lock (pluginLocker) {
500              HiveServiceLocator.Instance.CallHiveService((s) => hiveTask.Task.PluginsNeededIds = PluginUtil.GetPluginDependencies(s, this.onlinePlugins, this.alreadyUploadedPlugins, plugins));
501            }
502          }
503        }, Settings.Default.MaxRepeatServiceCalls, "Failed to upload plugins");
504        cancellationToken.ThrowIfCancellationRequested();
505        hiveTask.Task.PluginsNeededIds.Add(configPluginId);
506        hiveTask.Task.JobId = jobId;
507
508        log.LogMessage(string.Format("Uploading task ({0} kb, {1} objects)", taskData.Data.Count() / 1024, hiveTask.ItemTask.GetObjectGraphObjects().Count()));
509        TryAndRepeat(() => {
510          if (!cancellationToken.IsCancellationRequested) {
511            if (parentHiveTask != null) {
512              hiveTask.Task.Id = HiveServiceLocator.Instance.CallHiveService((s) => s.AddChildTask(parentHiveTask.Task.Id, hiveTask.Task, taskData));
513            } else {
514              hiveTask.Task.Id = HiveServiceLocator.Instance.CallHiveService((s) => s.AddTask(hiveTask.Task, taskData));
515            }
516          }
517        }, Settings.Default.MaxRepeatServiceCalls, "Failed to add task", log);
518        cancellationToken.ThrowIfCancellationRequested();
519
520        lock (jobCountLocker) {
521          progress.ProgressValue = (double)taskCount[0] / totalJobCount;
522          progress.Status = string.Format("Uploaded task ({0} of {1})", taskCount[0], totalJobCount);
523        }
524
525        var tasks = new List<TS.Task>();
526        foreach (HiveTask child in hiveTask.ChildHiveTasks) {
527          var task = TS.Task.Factory.StartNew((tuple) => {
528            var arguments = (Tuple<HiveTask, HiveTask>)tuple;
529            UploadTaskWithChildren(progress, arguments.Item1, arguments.Item2, taskCount, totalJobCount, configPluginId, jobId, log, cancellationToken);
530          }, new Tuple<HiveTask, HiveTask>(child, hiveTask));
531          task.ContinueWith((x) => log.LogException(x.Exception), TaskContinuationOptions.OnlyOnFaulted);
532          tasks.Add(task);
533        }
534        taskUploadSemaphore.Release(); semaphoreReleased = true; // the semaphore has to be release before waitall!
535        TS.Task.WaitAll(tasks.ToArray());
536      } finally {
537        if (!semaphoreReleased) taskUploadSemaphore.Release();
538      }
539    }
540    #endregion
541
542    #region Download Experiment
543    public static void LoadJob(RefreshableJob refreshableJob) {
544      var hiveExperiment = refreshableJob.Job;
545      refreshableJob.IsProgressing = true;
546      TaskDownloader downloader = null;
547
548      try {
549        int totalJobCount = 0;
550        IEnumerable<LightweightTask> allTasks;
551
552        // fetch all task objects to create the full tree of tree of HiveTask objects
553        refreshableJob.Progress.Start("Downloading list of tasks...");
554        allTasks = HiveServiceLocator.Instance.CallHiveService(s => s.GetLightweightJobTasksWithoutStateLog(hiveExperiment.Id));
555        totalJobCount = allTasks.Count();
556
557        refreshableJob.Progress.Status = "Downloading tasks...";
558        downloader = new TaskDownloader(allTasks.Select(x => x.Id));
559        downloader.StartAsync();
560
561        while (!downloader.IsFinished) {
562          refreshableJob.Progress.ProgressValue = downloader.FinishedCount / (double)totalJobCount;
563          refreshableJob.Progress.Status = string.Format("Downloading/deserializing tasks... ({0}/{1} finished)", downloader.FinishedCount, totalJobCount);
564          Thread.Sleep(500);
565
566          if (downloader.IsFaulted) {
567            throw downloader.Exception;
568          }
569        }
570        IDictionary<Guid, HiveTask> allHiveTasks = downloader.Results;
571        var parents = allHiveTasks.Values.Where(x => !x.Task.ParentTaskId.HasValue);
572
573        refreshableJob.Progress.Status = "Downloading/deserializing complete. Displaying tasks...";
574        // build child-task tree
575        foreach (HiveTask hiveTask in parents) {
576          BuildHiveJobTree(hiveTask, allTasks, allHiveTasks);
577        }
578
579        refreshableJob.HiveTasks = new ItemCollection<HiveTask>(parents);
580        if (refreshableJob.IsFinished()) {
581          refreshableJob.ExecutionState = Core.ExecutionState.Stopped;
582        } else if (refreshableJob.IsPaused()) {
583          refreshableJob.ExecutionState = Core.ExecutionState.Paused;
584        } else {
585          refreshableJob.ExecutionState = Core.ExecutionState.Started;
586        }
587        refreshableJob.OnLoaded();
588      }
589      finally {
590        refreshableJob.IsProgressing = false;
591        refreshableJob.Progress.Finish();
592        if (downloader != null) {
593          downloader.Dispose();
594        }
595      }
596    }
597
598    private static void BuildHiveJobTree(HiveTask parentHiveTask, IEnumerable<LightweightTask> allTasks, IDictionary<Guid, HiveTask> allHiveTasks) {
599      IEnumerable<LightweightTask> childTasks = from job in allTasks
600                                                where job.ParentTaskId.HasValue && job.ParentTaskId.Value == parentHiveTask.Task.Id
601                                                orderby job.DateCreated ascending
602                                                select job;
603      foreach (LightweightTask task in childTasks) {
604        HiveTask childHiveTask = allHiveTasks[task.Id];
605        BuildHiveJobTree(childHiveTask, allTasks, allHiveTasks);
606        parentHiveTask.AddChildHiveTask(childHiveTask);
607      }
608    }
609    #endregion
610
611    /// <summary>
612    /// Converts a string which can contain Ids separated by ';' to a enumerable
613    /// </summary>
614    private static IEnumerable<string> ToResourceNameList(string resourceNames) {
615      if (!string.IsNullOrEmpty(resourceNames)) {
616        return resourceNames.Split(new[] { ';' }, StringSplitOptions.RemoveEmptyEntries);
617      } else {
618        return new List<string>();
619      }
620    }
621
622    public static ItemTask LoadItemJob(Guid jobId) {
623      TaskData taskData = HiveServiceLocator.Instance.CallHiveService(s => s.GetTaskData(jobId));
624      try {
625        return PersistenceUtil.Deserialize<ItemTask>(taskData.Data);
626      }
627      catch {
628        return null;
629      }
630    }
631
632    /// <summary>
633    /// Executes the action. If it throws an exception it is repeated until repetition-count is reached.
634    /// If repetitions is -1, it is repeated infinitely.
635    /// </summary>
636    public static void TryAndRepeat(Action action, int repetitions, string errorMessage, ILog log = null) {
637      while (true) {
638        try { action(); return; }
639        catch (Exception e) {
640          if (repetitions == 0) throw new HiveException(errorMessage, e);
641          if (log != null) log.LogMessage(string.Format("{0}: {1} - will try again!", errorMessage, e.ToString()));
642          repetitions--;
643        }
644      }
645    }
646
647    public static HiveItemCollection<JobPermission> GetJobPermissions(Guid jobId) {
648      return HiveServiceLocator.Instance.CallHiveService((service) => {
649        IEnumerable<JobPermission> jps = service.GetJobPermissions(jobId);
650        foreach (var hep in jps) {
651          hep.UnmodifiedGrantedUserNameUpdate(service.GetUsernameByUserId(hep.GrantedUserId));
652        }
653        return new HiveItemCollection<JobPermission>(jps);
654      });
655    }
656  }
657}
Note: See TracBrowser for help on using the repository browser.