1 | #region License Information
|
---|
2 | /* HeuristicLab
|
---|
3 | * Copyright (C) 2002-2015 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 |
|
---|
22 | using System;
|
---|
23 | using System.Threading;
|
---|
24 | using System.Threading.Tasks;
|
---|
25 | using HeuristicLab.Common;
|
---|
26 | using HeuristicLab.Hive;
|
---|
27 |
|
---|
28 | namespace HeuristicLab.Clients.Hive {
|
---|
29 | /// <summary>
|
---|
30 | /// Downloads and deserializes jobs. It avoids too many jobs beeing downloaded or deserialized at the same time to avoid memory problems
|
---|
31 | /// </summary>
|
---|
32 | public class ConcurrentTaskDownloader<T> : IDisposable where T : class, ITask {
|
---|
33 | private bool abort = false;
|
---|
34 | // use semaphore to ensure only few concurrenct connections and few SerializedJob objects in memory
|
---|
35 | private Semaphore downloadSemaphore;
|
---|
36 | private Semaphore deserializeSemaphore;
|
---|
37 |
|
---|
38 | public ConcurrentTaskDownloader(int concurrentDownloads, int concurrentDeserializations) {
|
---|
39 | downloadSemaphore = new Semaphore(concurrentDownloads, concurrentDownloads);
|
---|
40 | deserializeSemaphore = new Semaphore(concurrentDeserializations, concurrentDeserializations);
|
---|
41 | }
|
---|
42 |
|
---|
43 | public void DownloadTaskData(Task t, Action<Task, T> onFinishedAction) {
|
---|
44 | Task<Tuple<Task, T>> task = Task<Tuple<Task, TaskData>>.Factory.StartNew(DownloadTaskData, t)
|
---|
45 | .ContinueWith((y) => DeserializeTask(y.Result));
|
---|
46 |
|
---|
47 | task.ContinueWith((x) => OnTaskFinished(x, onFinishedAction), TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnRanToCompletion);
|
---|
48 | task.ContinueWith((x) => OnTaskFailed(x), TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnFaulted);
|
---|
49 | }
|
---|
50 |
|
---|
51 | public void DownloadTaskDataAndTask(Guid taskId, Action<Task, T> onFinishedAction) {
|
---|
52 | Task<Tuple<Task, T>> task = Task<Task>.Factory.StartNew(DownloadTask, taskId)
|
---|
53 | .ContinueWith((x) => DownloadTaskData(x.Result))
|
---|
54 | .ContinueWith((y) => DeserializeTask(y.Result));
|
---|
55 |
|
---|
56 | task.ContinueWith((x) => OnTaskFinished(x, onFinishedAction), TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnRanToCompletion);
|
---|
57 | task.ContinueWith((x) => OnTaskFailed(x), TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnFaulted);
|
---|
58 | }
|
---|
59 |
|
---|
60 | private void OnTaskFinished(Task<Tuple<Task, T>> task, Action<Task, T> onFinishedAction) {
|
---|
61 | onFinishedAction(task.Result.Item1, task.Result.Item2);
|
---|
62 | }
|
---|
63 | private void OnTaskFailed(Task<Tuple<Task, T>> task) {
|
---|
64 | task.Exception.Flatten().Handle((e) => { return true; });
|
---|
65 | OnExceptionOccured(task.Exception.Flatten());
|
---|
66 | }
|
---|
67 |
|
---|
68 | private Task DownloadTask(object taskId) {
|
---|
69 | Task t = null;
|
---|
70 | HiveClient.TryAndRepeat(() => {
|
---|
71 | t = HiveServiceLocator.Instance.CallHiveService(s => s.GetTask((Guid)taskId));
|
---|
72 | }, Settings.Default.MaxRepeatServiceCalls, "Failed to download task.");
|
---|
73 | return t;
|
---|
74 | }
|
---|
75 |
|
---|
76 | protected Tuple<Task, TaskData> DownloadTaskData(object taskId) {
|
---|
77 | return DownloadTaskData((Task)taskId);
|
---|
78 | }
|
---|
79 |
|
---|
80 | protected Tuple<Task, TaskData> DownloadTaskData(Task task) {
|
---|
81 | downloadSemaphore.WaitOne();
|
---|
82 | TaskData result = null;
|
---|
83 | try {
|
---|
84 | if (abort) return null;
|
---|
85 | HiveClient.TryAndRepeat(() => {
|
---|
86 | result = HiveServiceLocator.Instance.CallHiveService(s => s.GetTaskData(task.Id));
|
---|
87 | }, Settings.Default.MaxRepeatServiceCalls, "Failed to download task data.");
|
---|
88 | }
|
---|
89 | finally {
|
---|
90 | downloadSemaphore.Release();
|
---|
91 | }
|
---|
92 | return new Tuple<Task, TaskData>(task, result);
|
---|
93 | }
|
---|
94 |
|
---|
95 | protected Tuple<Task, T> DeserializeTask(Tuple<Task, TaskData> taskData) {
|
---|
96 | deserializeSemaphore.WaitOne();
|
---|
97 | try {
|
---|
98 | if (abort || taskData.Item2 == null || taskData.Item1 == null) return null;
|
---|
99 | var deserializedJob = PersistenceUtil.Deserialize<T>(taskData.Item2.Data);
|
---|
100 | taskData.Item2.Data = null; // reduce memory consumption.
|
---|
101 | return new Tuple<Task, T>(taskData.Item1, deserializedJob);
|
---|
102 | }
|
---|
103 | finally {
|
---|
104 | deserializeSemaphore.Release();
|
---|
105 | }
|
---|
106 | }
|
---|
107 |
|
---|
108 | public event EventHandler<EventArgs<Exception>> ExceptionOccured;
|
---|
109 | private void OnExceptionOccured(Exception exception) {
|
---|
110 | var handler = ExceptionOccured;
|
---|
111 | if (handler != null) handler(this, new EventArgs<Exception>(exception));
|
---|
112 | }
|
---|
113 |
|
---|
114 | #region IDisposable Members
|
---|
115 | public void Dispose() {
|
---|
116 | deserializeSemaphore.Dispose();
|
---|
117 | downloadSemaphore.Dispose();
|
---|
118 | }
|
---|
119 | #endregion
|
---|
120 | }
|
---|
121 | }
|
---|