Changeset 7115 for trunk/sources/HeuristicLab.Clients.Hive

Timestamp:

12/03/11 01:01:47 (12 years ago)

Author:

ascheibe

Message:

#1672

• speed up download of tasks by avoiding unnecessary service calls
• display download progress correctly

Location:

trunk/sources/HeuristicLab.Clients.Hive/3.3

Files:

• ConcurrentTaskDownloader.cs (modified) (2 diffs)
• RefreshableJob.cs (modified) (1 diff)
• TaskDownloader.cs (modified) (6 diffs)

trunk/sources/HeuristicLab.Clients.Hive/3.3/ConcurrentTaskDownloader.cs

@@ -42 +42 @@
     }
 
-    public void DownloadTask(Task job, Action<Task, T> onFinishedAction) {
-      Task<T> task = Task<TaskData>.Factory.StartNew((x) => DownloadTask(x), job.Id)
-                                     .ContinueWith((x) => DeserializeTask(x.Result));
-      task.ContinueWith((x) => OnTaskFinished(job, x, onFinishedAction), TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnRanToCompletion);
-      task.ContinueWith((x) => OnTaskFailed(job, x, onFinishedAction), TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnFaulted);
+    public void DownloadTaskData(Task t, Action<Task, T> onFinishedAction) {
+      Task<Tuple<Task, T>> task = Task<Tuple<Task, TaskData>>.Factory.StartNew(DownloadTaskData, t)
+                                     .ContinueWith((y) => DeserializeTask(y.Result));
+
+      task.ContinueWith((x) => OnTaskFinished(x, onFinishedAction), TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnRanToCompletion);
+      task.ContinueWith((x) => OnTaskFailed(x, onFinishedAction), TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnFaulted);
     }
 
-    private void OnTaskFinished(Task job, Task<T> task, Action<Task, T> onFinishedAction) {
-      onFinishedAction(job, task.Result);
+    public void DownloadTaskDataAndTask(Guid taskId, Action<Task, T> onFinishedAction) {
+      Task<Tuple<Task, T>> task = Task<Task>.Factory.StartNew(DownloadTask, taskId)
+                                     .ContinueWith((x) => DownloadTaskData(x.Result))
+                                     .ContinueWith((y) => DeserializeTask(y.Result));
+
+      task.ContinueWith((x) => OnTaskFinished(x, onFinishedAction), TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnRanToCompletion);
+      task.ContinueWith((x) => OnTaskFailed(x, onFinishedAction), TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnFaulted);
     }
-    private void OnTaskFailed(Task job, Task<T> task, Action<Task, T> onFinishedAction) {
+
+    private void OnTaskFinished(Task<Tuple<Task, T>> task, Action<Task, T> onFinishedAction) {
+      onFinishedAction(task.Result.Item1, task.Result.Item2);
+    }
+    private void OnTaskFailed(Task<Tuple<Task, T>> task, Action<Task, T> onFinishedAction) {
       task.Exception.Flatten().Handle((e) => { return true; });
       OnExceptionOccured(task.Exception.Flatten());
-      onFinishedAction(job, null);
+      onFinishedAction(task.Result.Item1, null);
     }
 
-    protected TaskData DownloadTask(object taskId) {
+    private Task DownloadTask(object taskId) {
+      return ServiceLocator.Instance.CallHiveService(s => s.GetTask((Guid)taskId));
+    }
+
+    protected Tuple<Task, TaskData> DownloadTaskData(object taskId) {
+      return DownloadTaskData((Task)taskId);
+    }
+
+    protected Tuple<Task, TaskData> DownloadTaskData(Task task) {
       downloadSemaphore.WaitOne();
       deserializeSemaphore.WaitOne();
@@ -64 +82 @@
       try {
         if (abort) return null;
-        result = ServiceLocator.Instance.CallHiveService(s => s.GetTaskData((Guid)taskId));
-      }
-      finally {
+        result = ServiceLocator.Instance.CallHiveService(s => s.GetTaskData(task.Id));
+      } finally {
         downloadSemaphore.Release();
       }
-      return result;
+      return new Tuple<Task, TaskData>(task, result);
     }
 
-    protected T DeserializeTask(TaskData taskData) {
+    protected Tuple<Task, T> DeserializeTask(Tuple<Task, TaskData> taskData) {
       try {
-        if (abort || taskData == null) return null;
-        Task task = ServiceLocator.Instance.CallHiveService(s => s.GetTask(taskData.TaskId));
-        if (task == null) return null;
-        var deserializedJob = PersistenceUtil.Deserialize<T>(taskData.Data);
-        taskData.Data = null; // reduce memory consumption.
-        return deserializedJob;
-      }
-      finally {
+        if (abort || taskData.Item2 == null || taskData.Item1 == null) return null;
+        var deserializedJob = PersistenceUtil.Deserialize<T>(taskData.Item2.Data);
+        taskData.Item2.Data = null; // reduce memory consumption.
+        return new Tuple<Task, T>(taskData.Item1, deserializedJob);
+      } finally {
         deserializeSemaphore.Release();
       }

trunk/sources/HeuristicLab.Clients.Hive/3.3/RefreshableJob.cs

@@ -277 +277 @@
             log.LogMessage(string.Format("Downloading task {0}", lightweightTask.Id));
             hiveTask.IsDownloading = true;
-            jobDownloader.DownloadTask(hiveTask.Task, (localJob, itemJob) => {
+            jobDownloader.DownloadTaskData(hiveTask.Task, (localJob, itemJob) => {
               log.LogMessage(string.Format("Finished downloading task {0}", localJob.Id));
               HiveTask localHiveTask = GetHiveJobById(localJob.Id);

trunk/sources/HeuristicLab.Clients.Hive/3.3/TaskDownloader.cs

@@ -25 +25 @@
 using HeuristicLab.Clients.Hive.Jobs;
 using HeuristicLab.Common;
+using System.Threading;
 
 namespace HeuristicLab.Clients.Hive {
@@ -33 +34 @@
     private bool exceptionOccured = false;
     private Exception currentException;
+    private ReaderWriterLockSlim resultsLock = new ReaderWriterLockSlim();
 
     public bool IsFinished {
       get {
-        return results.Count == taskIds.Count();
+          try {        
+              resultsLock.EnterReadLock();
+              return results.Count == taskIds.Count();
+          } finally { resultsLock.ExitReadLock(); }
       }
     }
@@ -54 +59 @@
     public int FinishedCount {
       get {
-        return results.Count;
+            try {
+              resultsLock.EnterReadLock();
+              return results.Count;
+             } finally { resultsLock.ExitReadLock(); }
       }
     }
@@ -60 +68 @@
     public IDictionary<Guid, HiveTask> Results {
       get {
-        return results;
+            try {
+              resultsLock.EnterReadLock();
+              return results;
+            } finally { resultsLock.ExitReadLock(); }
       }
     }
@@ -73 +84 @@
     public void StartAsync() {
       foreach (Guid taskId in taskIds) {
-        Task task = ServiceLocator.Instance.CallHiveService(s => s.GetTask(taskId));
-
-        taskDownloader.DownloadTask(task,
+        taskDownloader.DownloadTaskDataAndTask(taskId,
           (localJob, itemJob) => {
             if (localJob != null && itemJob != null) {
@@ -85 +94 @@
               }
               hiveTask.Task = localJob;
-              this.results.Add(localJob.Id, hiveTask);
+              try {
+                resultsLock.EnterWriteLock();
+                this.results.Add(localJob.Id, hiveTask);
+              } finally { resultsLock.ExitWriteLock(); }
             }
           });