Changeset 15373 for branches/ParallelExperiment/HeuristicLab.Optimization

Timestamp:

09/19/17 15:58:21 (7 years ago)

Author:

pfleck

Message:

#2822:

• Merged recent trunk-changes into branch.
• Using task-continuations for semaphore-releasing to avoid manually started optimizers to mess with the semaphore.

Location:

branches/ParallelExperiment/HeuristicLab.Optimization

Files:

• . (modified) (1 prop)
• 3.3/Algorithms/BasicAlgorithm.cs (modified) (1 diff)
• 3.3/MetaOptimizers/Experiment.cs (modified) (5 diffs)

branches/ParallelExperiment/HeuristicLab.Optimization/3.3/Algorithms/BasicAlgorithm.cs

@@ -82 +82 @@
       } catch (OperationCanceledException) {
       } catch (AggregateException ae) {
-        OnExceptionOccurred(ae.InnerExceptions.SingleOrDefault() ?? ae);
+        ae.FlattenAndHandle(new[] { typeof(OperationCanceledException) }, e => OnExceptionOccurred(e));
       } catch (Exception e) {
         OnExceptionOccurred(e);

branches/ParallelExperiment/HeuristicLab.Optimization/3.3/MetaOptimizers/Experiment.cs

@@ -122 +122 @@
     private bool experimentStopped = false;
 
-    private ManualResetEventSlim allOptimizerFinished; // this indicates that all started optimizers have been paused or stopped
-    private SemaphoreSlim availableWorkers; // limits the number of concurrent optimizer executions
+    private readonly ManualResetEventSlim allOptimizerFinished = new ManualResetEventSlim(false); // this indicates that all started optimizers have been paused or stopped
 
     public Experiment()
@@ -202 +201 @@
       Start(CancellationToken.None);
     }
-
     public void Start(CancellationToken cancellationToken) {
       if ((ExecutionState != ExecutionState.Prepared) && (ExecutionState != ExecutionState.Paused))
         throw new InvalidOperationException(string.Format("Start not allowed in execution state \"{0}\".", ExecutionState));
-      if (Optimizers.Count == 0) return;
+      if (!Optimizers.Any(x => x.ExecutionState == ExecutionState.Prepared || x.ExecutionState == ExecutionState.Paused)) return;
 
       experimentStarted = true;
       experimentStopped = false;
-      allOptimizerFinished = new ManualResetEventSlim(false);
-      availableWorkers = new SemaphoreSlim(NumberOfWorkers, NumberOfWorkers);
-
-      var runnableOptimizers = Optimizers.Where(o => o.ExecutionState == ExecutionState.Prepared || o.ExecutionState == ExecutionState.Paused).ToList();
-      if (!runnableOptimizers.Any()) return;
-
-      while (runnableOptimizers.Any()) {
-        try {
-          availableWorkers.Wait(cancellationToken); // an optimizer was pause/stopped previously
-          if (experimentStopped || !experimentStarted) break;
-          // some optimizers might be started manually somewhere else
-          runnableOptimizers.RemoveAll(o => !(o.ExecutionState == ExecutionState.Prepared || o.ExecutionState == ExecutionState.Paused));
-          var optimizer = runnableOptimizers.FirstOrDefault();
-          runnableOptimizers.Remove(optimizer);
-          optimizer?.StartAsync(cancellationToken);
-        } catch (InvalidOperationException) { } catch (OperationCanceledException) { }
-      }
-
-      allOptimizerFinished.Wait();
+
+      var startedTasks = new List<Task>();
+      using (var availableWorkers = new SemaphoreSlim(NumberOfWorkers, NumberOfWorkers)) {
+        var runnableOptimizers = Optimizers.Where(o => o.ExecutionState == ExecutionState.Prepared || o.ExecutionState == ExecutionState.Paused).ToList();
+        if (!runnableOptimizers.Any()) return;
+
+        while (runnableOptimizers.Any()) {
+          try {
+            availableWorkers.Wait(cancellationToken);
+            if (experimentStopped || !experimentStarted) break;
+            // some optimizers might be started manually somewhere else
+            runnableOptimizers.RemoveAll(o => !(o.ExecutionState == ExecutionState.Prepared || o.ExecutionState == ExecutionState.Paused));
+            if (runnableOptimizers.Any()) {
+              var optimizer = runnableOptimizers.First();
+              runnableOptimizers.Remove(optimizer);
+              var startedTask = optimizer.StartAsync(cancellationToken).ContinueWith(x => {
+                availableWorkers.Release(); // is not disposed yet because Task.WaitAll at the end of the using is blocking
+              });
+              startedTasks.Add(startedTask);
+            }
+          } catch (InvalidOperationException) { } catch (OperationCanceledException) { }
+        }
+
+        allOptimizerFinished.Wait();
+        Task.WaitAll(startedTasks.ToArray()); // retreive exeptions of the asyncrounously started optimizer
+      }
     }
     public async Task StartAsync() { await StartAsync(CancellationToken.None); }
@@ -294 +299 @@
     private void OnPaused() {
       ExecutionState = ExecutionState.Paused;
+      allOptimizerFinished.Set();
       EventHandler handler = Paused;
       if (handler != null) handler(this, EventArgs.Empty);
-      allOptimizerFinished.Set();
     }
     public event EventHandler Stopped;
     private void OnStopped() {
       ExecutionState = ExecutionState.Stopped;
+      allOptimizerFinished.Set();
       EventHandler handler = Stopped;
       if (handler != null) handler(this, EventArgs.Empty);
-      allOptimizerFinished.Set();
     }
     public event EventHandler<EventArgs<Exception>> ExceptionOccurred;
@@ -398 +403 @@
     private void optimizer_Paused(object sender, EventArgs e) {
       lock (locker) {
-        if (availableWorkers.CurrentCount < NumberOfWorkers)
-          availableWorkers.Release();
         if (Optimizers.All(x => x.ExecutionState != ExecutionState.Started)) {
           OnPaused();
@@ -415 +418 @@
     private void optimizer_Stopped(object sender, EventArgs e) {
       lock (locker) {
-        if (availableWorkers.CurrentCount < NumberOfWorkers)
-          availableWorkers.Release();
         if (experimentStopped) {
-          if (Optimizers.All(x => (x.ExecutionState == ExecutionState.Stopped) || (x.ExecutionState == ExecutionState.Prepared))) {
-            OnStopped();
-          }
+          if (Optimizers.All(x => (x.ExecutionState == ExecutionState.Stopped) || (x.ExecutionState == ExecutionState.Prepared))) OnStopped();
         } else {
           if (experimentStarted && Optimizers.Any(x => (x.ExecutionState == ExecutionState.Prepared) || (x.ExecutionState == ExecutionState.Paused))) return;