Changeset 15337 for branches/ParallelExperiment/HeuristicLab.Optimization

Timestamp:

08/22/17 14:08:52 (7 years ago)

Author:

pfleck

Message:

#2822 Added parallel execution support for Experiment (similar to CrossValidation).

• Fixed race-condition in BasicAlgorithm (potential NullReference of CancellationTokenSource).
• Fixed race-condition in BatchRun (calling .Stop potentially fires both OnStopped and OnPaused).

Location:

branches/ParallelExperiment/HeuristicLab.Optimization/3.3

Files:

• Algorithms/BasicAlgorithm.cs (modified) (4 diffs)
• MetaOptimizers/BatchRun.cs (modified) (3 diffs)
• MetaOptimizers/Experiment.cs (modified) (11 diffs)

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

@@ -80 +80 @@
       try {
         Run((object)cancellationTokenSource.Token);
-      }
-      catch (OperationCanceledException) {
-      }
-      catch (AggregateException ae) {
+      } catch (OperationCanceledException) {
+      } catch (AggregateException ae) {
         OnExceptionOccurred(ae.InnerExceptions.SingleOrDefault() ?? ae);
-      }
-      catch (Exception e) {
+      } catch (Exception e) {
         OnExceptionOccurred(e);
       }
@@ -104 +101 @@
       base.Pause();
       pausePending = true;
-      CancellationTokenSource.Cancel();
+      CancellationTokenSource?.Cancel();
     }
 
@@ -112 +109 @@
       base.Stop();
       if (ExecutionState == ExecutionState.Paused) OnStopped();
-      else CancellationTokenSource.Cancel();
+      else CancellationTokenSource?.Cancel();
     }
 
@@ -127 +124 @@
         initialized = true;
         Run(cancellationToken);
-      }
-      finally {
+      } finally {
         timer.Elapsed -= new System.Timers.ElapsedEventHandler(timer_Elapsed);
         timer.Stop();

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

@@ -258 +258 @@
         if (ExecutionState == ExecutionState.Paused || ExecutionState == ExecutionState.Stopped) break;
         Optimizer.Prepare();
+        if (ExecutionState == ExecutionState.Paused || ExecutionState == ExecutionState.Stopped) break;
       }
     }
@@ -271 +272 @@
       if (Optimizer.ExecutionState != ExecutionState.Started) return;
       // a race-condition may occur when the optimizer has changed the state by itself in the meantime
-      try { Optimizer.Pause(); } catch (InvalidOperationException) { }
+      try { Optimizer.Pause(); } catch (InvalidOperationException) { } catch (NotSupportedException) { }
     }
     public void Stop() {
@@ -385 +386 @@
     }
     private void Optimizer_Prepared(object sender, EventArgs e) {
-      if (batchRunAction == BatchRunAction.Prepare || ExecutionState == ExecutionState.Stopped) {
+      if (batchRunAction == BatchRunAction.Prepare/* || ExecutionState == ExecutionState.Stopped*/) {
         ExecutionTime = TimeSpan.Zero;
         runsExecutionTime = TimeSpan.Zero;

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

@@ -97 +97 @@
     }
 
+    [Storable]
+    private int numberOfWorkers = 1;
+    public int NumberOfWorkers {
+      get { return numberOfWorkers; }
+      set {
+        if (value < 1) throw new ArgumentException("Number of Workers must not be lower than one.");
+        numberOfWorkers = value;
+      }
+    }
+
     public IEnumerable<IOptimizer> NestedOptimizers {
       get {
@@ -111 +121 @@
     private bool experimentStarted = false;
     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
 
     public Experiment()
@@ -155 +167 @@
       experimentStarted = original.experimentStarted;
       experimentStopped = original.experimentStopped;
+      numberOfWorkers = original.numberOfWorkers;
       Initialize();
     }
@@ -189 +202 @@
       Start(CancellationToken.None);
     }
+
     public void Start(CancellationToken cancellationToken) {
       if ((ExecutionState != ExecutionState.Prepared) && (ExecutionState != ExecutionState.Paused))
@@ -197 +211 @@
       experimentStopped = false;
       allOptimizerFinished = new ManualResetEventSlim(false);
-
-      IOptimizer optimizer;
-      while ((optimizer = Optimizers.FirstOrDefault(x => (x.ExecutionState == ExecutionState.Prepared) || (x.ExecutionState == ExecutionState.Paused))) != null) {
-        // a race-condition may occur when the optimizer has changed the state by itself in the meantime
-        try { optimizer.Start(cancellationToken); } catch (InvalidOperationException) { }
-        if (ExecutionState == ExecutionState.Paused || ExecutionState == ExecutionState.Stopped) break;
+      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) { }
       }
 
@@ -220 +243 @@
       foreach (IOptimizer optimizer in Optimizers.Where(x => x.ExecutionState == ExecutionState.Started)) {
         // a race-condition may occur when the optimizer has changed the state by itself in the meantime
-        try { optimizer.Pause(); } catch (InvalidOperationException) { }
+        try { optimizer.Pause(); } catch (InvalidOperationException) { } catch (NotSupportedException) { }
       }
     }
@@ -273 +296 @@
       EventHandler handler = Paused;
       if (handler != null) handler(this, EventArgs.Empty);
+      allOptimizerFinished.Set();
     }
     public event EventHandler Stopped;
@@ -279 +303 @@
       EventHandler handler = Stopped;
       if (handler != null) handler(this, EventArgs.Empty);
+      allOptimizerFinished.Set();
     }
     public event EventHandler<EventArgs<Exception>> ExceptionOccurred;
@@ -372 +397 @@
     }
     private void optimizer_Paused(object sender, EventArgs e) {
-      lock (locker)
+      lock (locker) {
+        if (availableWorkers.CurrentCount < NumberOfWorkers)
+          availableWorkers.Release();
         if (Optimizers.All(x => x.ExecutionState != ExecutionState.Started)) {
           OnPaused();
-          allOptimizerFinished.Set();
-        }
+        }
+      }
     }
     private void optimizer_Prepared(object sender, EventArgs e) {
@@ -388 +415 @@
     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();
-            allOptimizerFinished.Set();
           }
         } else {
@@ -397 +425 @@
           else if (Optimizers.All(x => x.ExecutionState == ExecutionState.Stopped)) {
             OnStopped();
-            allOptimizerFinished.Set();
           } else if (Optimizers.Any(x => (x.ExecutionState == ExecutionState.Prepared) || (x.ExecutionState == ExecutionState.Paused)) && Optimizers.All(o => o.ExecutionState != ExecutionState.Started)) {
             OnPaused();
-            allOptimizerFinished.Set();
           }
         }