Changeset 15292 for stable/HeuristicLab.Algorithms.DataAnalysis/3.4

Timestamp:

07/28/17 16:47:42 (7 years ago)

Author:

jkarder

Message:

#2258: merged r15287 into stable

Location:

stable

Files:

• . (modified) (1 prop)
• HeuristicLab.Algorithms.DataAnalysis (modified) (1 prop)
• HeuristicLab.Algorithms.DataAnalysis/3.4/CrossValidation.cs (modified) (11 diffs)
• HeuristicLab.Algorithms.DataAnalysis/3.4/FixedDataAnalysisAlgorithm.cs (modified) (1 diff)

stable/HeuristicLab.Algorithms.DataAnalysis/3.4/CrossValidation.cs

@@ -25 +25 @@
 using System.Linq;
 using System.Threading;
+using System.Threading.Tasks;
 using HeuristicLab.Collections;
 using HeuristicLab.Common;
@@ -42 +43 @@
     [Storable]
     private int seed;
+
+    private SemaphoreSlim availableWorkers; // limits the number of concurrent algorithm executions
+    private ManualResetEventSlim allAlgorithmsFinished; // this indicates that all started algorithms have been paused or stopped
 
     public CrossValidation()
@@ -269 +273 @@
 
     public void Prepare() {
+      if (startPending) return;
       if (ExecutionState == ExecutionState.Started)
         throw new InvalidOperationException(string.Format("Prepare not allowed in execution state \"{0}\".", ExecutionState));
@@ -283 +288 @@
     }
 
+    private bool startPending;
     public void Start() {
-      if ((ExecutionState != ExecutionState.Prepared) && (ExecutionState != ExecutionState.Paused))
-        throw new InvalidOperationException(string.Format("Start not allowed in execution state \"{0}\".", ExecutionState));
-
-      seed = new FastRandom().NextInt();
-
-      if (Algorithm != null) {
+      Start(CancellationToken.None);
+    }
+    public void Start(CancellationToken cancellationToken) {
+      lock (locker) {
+        if (startPending) return;
+        startPending = true;
+      }
+
+      try {
+        if ((ExecutionState != ExecutionState.Prepared) && (ExecutionState != ExecutionState.Paused))
+          throw new InvalidOperationException(string.Format("Start not allowed in execution state \"{0}\".", ExecutionState));
+        seed = new FastRandom().NextInt();
+
+        if (Algorithm == null) return;
         //create cloned algorithms
         if (clonedAlgorithms.Count == 0) {
@@ -330 +344 @@
         }
 
-        //start prepared or paused cloned algorithms
-        int startedAlgorithms = 0;
-        foreach (IAlgorithm clonedAlgorithm in clonedAlgorithms) {
-          if (startedAlgorithms < NumberOfWorkers.Value) {
-            if (clonedAlgorithm.ExecutionState == ExecutionState.Prepared ||
-                clonedAlgorithm.ExecutionState == ExecutionState.Paused) {
-
-              // start and wait until the alg is started 
-              using (var signal = new ManualResetEvent(false)) {
-                EventHandler signalSetter = (sender, args) => { signal.Set(); };
-                clonedAlgorithm.Started += signalSetter;
-                clonedAlgorithm.Start();
-                signal.WaitOne();
-                clonedAlgorithm.Started -= signalSetter;
-
-                startedAlgorithms++;
-              }
-            }
+        OnStarted();
+      } finally {
+        if (startPending) startPending = false;
+      }
+
+      availableWorkers = new SemaphoreSlim(NumberOfWorkers.Value, NumberOfWorkers.Value);
+      allAlgorithmsFinished = new ManualResetEventSlim(false);
+
+      //start prepared or paused cloned algorithms
+      foreach (IAlgorithm clonedAlgorithm in clonedAlgorithms) {
+        if (pausePending || stopPending || ExecutionState != ExecutionState.Started) break;
+        if (clonedAlgorithm.ExecutionState == ExecutionState.Prepared ||
+            clonedAlgorithm.ExecutionState == ExecutionState.Paused) {
+          availableWorkers.Wait();
+          lock (locker) {
+            if (pausePending || stopPending || ExecutionState != ExecutionState.Started) break;
+            clonedAlgorithm.StartAsync(cancellationToken);
           }
         }
-        OnStarted();
-      }
+      }
+
+      allAlgorithmsFinished.Wait();
+    }
+
+    public async Task StartAsync() { await StartAsync(CancellationToken.None); }
+    public async Task StartAsync(CancellationToken cancellationToken) {
+      await AsyncHelper.DoAsync(Start, cancellationToken);
     }
 
     private bool pausePending;
     public void Pause() {
+      if (startPending) return;
       if (ExecutionState != ExecutionState.Started)
         throw new InvalidOperationException(string.Format("Pause not allowed in execution state \"{0}\".", ExecutionState));
       if (!pausePending) {
         pausePending = true;
-        PauseAllClonedAlgorithms();
-      }
-    }
-    private void PauseAllClonedAlgorithms() {
-      foreach (IAlgorithm clonedAlgorithm in clonedAlgorithms) {
-        if (clonedAlgorithm.ExecutionState == ExecutionState.Started)
-          clonedAlgorithm.Pause();
+        lock (locker) {
+          var toPause = clonedAlgorithms.Where(x => x.ExecutionState == ExecutionState.Started).ToList();
+          foreach (var optimizer in toPause) {
+            // a race-condition may occur when the optimizer has changed the state by itself in the meantime
+            try { optimizer.Pause(); } catch (InvalidOperationException) { }
+          }
+        }
       }
     }
@@ -372 +392 @@
     private bool stopPending;
     public void Stop() {
+      if (startPending) return;
       if ((ExecutionState != ExecutionState.Started) && (ExecutionState != ExecutionState.Paused))
         throw new InvalidOperationException(string.Format("Stop not allowed in execution state \"{0}\".",
@@ -377 +398 @@
       if (!stopPending) {
         stopPending = true;
-        StopAllClonedAlgorithms();
-      }
-    }
-    private void StopAllClonedAlgorithms() {
-      foreach (IAlgorithm clonedAlgorithm in clonedAlgorithms) {
-        if (clonedAlgorithm.ExecutionState == ExecutionState.Started ||
-            clonedAlgorithm.ExecutionState == ExecutionState.Paused)
-          clonedAlgorithm.Stop();
+        lock (locker) {
+          var toStop = clonedAlgorithms.Where(x => x.ExecutionState == ExecutionState.Started || x.ExecutionState == ExecutionState.Paused).ToList();
+          foreach (var optimizer in toStop) {
+            // a race-condition may occur when the optimizer has changed the state by itself in the meantime
+            try { optimizer.Stop(); } catch (InvalidOperationException) { }
+          }
+        }
       }
     }
@@ -685 +705 @@
     }
     private void ClonedAlgorithm_ExceptionOccurred(object sender, EventArgs<Exception> e) {
+      Pause();
       OnExceptionOccurred(e.Value);
     }
@@ -703 +724 @@
     private void ClonedAlgorithm_Paused(object sender, EventArgs e) {
       lock (locker) {
-        if (pausePending && clonedAlgorithms.All(alg => alg.ExecutionState != ExecutionState.Started))
+        availableWorkers.Release();
+        if (clonedAlgorithms.All(alg => alg.ExecutionState != ExecutionState.Started)) {
           OnPaused();
+          allAlgorithmsFinished.Set();
+        }
       }
     }
@@ -710 +734 @@
     private void ClonedAlgorithm_Stopped(object sender, EventArgs e) {
       lock (locker) {
-        if (!stopPending && ExecutionState == ExecutionState.Started) {
-          IAlgorithm preparedAlgorithm = clonedAlgorithms.FirstOrDefault(alg => alg.ExecutionState == ExecutionState.Prepared ||
-                                                                                alg.ExecutionState == ExecutionState.Paused);
-          if (preparedAlgorithm != null) preparedAlgorithm.Start();
-        }
-        if (ExecutionState != ExecutionState.Stopped) {
-          if (clonedAlgorithms.All(alg => alg.ExecutionState == ExecutionState.Stopped))
-            OnStopped();
-          else if (stopPending &&
-                   clonedAlgorithms.All(
-                     alg => alg.ExecutionState == ExecutionState.Prepared || alg.ExecutionState == ExecutionState.Stopped))
-            OnStopped();
+        // if the algorithm was in paused state, its worker has already been released
+        if (availableWorkers.CurrentCount < NumberOfWorkers.Value)
+          availableWorkers.Release();
+        if (clonedAlgorithms.All(alg => alg.ExecutionState == ExecutionState.Stopped)) {
+          OnStopped();
+          allAlgorithmsFinished.Set();
+        } else if (stopPending && clonedAlgorithms.All(alg => alg.ExecutionState == ExecutionState.Prepared || alg.ExecutionState == ExecutionState.Stopped)) {
+          OnStopped();
+          allAlgorithmsFinished.Set();
         }
       }
@@ -748 +769 @@
     public event EventHandler Started;
     private void OnStarted() {
+      startPending = false;
       ExecutionState = ExecutionState.Started;
       EventHandler handler = Started;

stable/HeuristicLab.Algorithms.DataAnalysis/3.4/FixedDataAnalysisAlgorithm.cs

@@ -21 +21 @@
 
 using System;
+using System.Linq;
+using System.Threading;
 using HeuristicLab.Common;
 using HeuristicLab.Optimization;