Changeset 15065 for branches/Async/HeuristicLab.Algorithms.DataAnalysis/3.4

Timestamp:

06/26/17 09:45:36 (7 years ago)

Author:

jkarder

Message:

#2258: refactored async methods

• synchronously called IExecutables are now executed in the caller's thread
• removed old synchronization code from unit tests

Location:

branches/Async/HeuristicLab.Algorithms.DataAnalysis/3.4

Files:

• CrossValidation.cs (modified) (9 diffs)
• FixedDataAnalysisAlgorithm.cs (modified) (2 diffs)

branches/Async/HeuristicLab.Algorithms.DataAnalysis/3.4/CrossValidation.cs

@@ -40 +40 @@
   [StorableClass]
   public sealed class CrossValidation : ParameterizedNamedItem, IAlgorithm, IStorableContent {
-    private readonly ManualResetEvent signaler = new ManualResetEvent(true);
-    private CancellationToken cancellationToken;
+    private SemaphoreSlim ticket;
+    private ManualResetEventSlim signal;
 
     public CrossValidation()
@@ -269 +269 @@
       Prepare();
     }
+
     public void Start() {
-      StartAsync().Wait();
-    }
-    public async Task StartAsync() {
-      await StartAsync(CancellationToken.None);
-    }
+      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 (Algorithm == null) return;
+      //create cloned algorithms
+      if (clonedAlgorithms.Count == 0) {
+        int testSamplesCount = (SamplesEnd.Value - SamplesStart.Value) / Folds.Value;
+
+        for (int i = 0; i < Folds.Value; i++) {
+          IAlgorithm clonedAlgorithm = (IAlgorithm)algorithm.Clone();
+          clonedAlgorithm.Name = algorithm.Name + " Fold " + i;
+          IDataAnalysisProblem problem = clonedAlgorithm.Problem as IDataAnalysisProblem;
+          ISymbolicDataAnalysisProblem symbolicProblem = problem as ISymbolicDataAnalysisProblem;
+
+          int testStart = (i * testSamplesCount) + SamplesStart.Value;
+          int testEnd = (i + 1) == Folds.Value ? SamplesEnd.Value : (i + 1) * testSamplesCount + SamplesStart.Value;
+
+          problem.ProblemData.TrainingPartition.Start = SamplesStart.Value;
+          problem.ProblemData.TrainingPartition.End = SamplesEnd.Value;
+          problem.ProblemData.TestPartition.Start = testStart;
+          problem.ProblemData.TestPartition.End = testEnd;
+          DataAnalysisProblemData problemData = problem.ProblemData as DataAnalysisProblemData;
+          if (problemData != null) {
+            problemData.TrainingPartitionParameter.Hidden = false;
+            problemData.TestPartitionParameter.Hidden = false;
+          }
+
+          if (symbolicProblem != null) {
+            symbolicProblem.FitnessCalculationPartition.Start = SamplesStart.Value;
+            symbolicProblem.FitnessCalculationPartition.End = SamplesEnd.Value;
+          }
+          clonedAlgorithm.Prepare();
+          clonedAlgorithms.Add(clonedAlgorithm);
+        }
+      }
+
+      OnStarted();
+      ticket = new SemaphoreSlim(NumberOfWorkers.Value);
+      signal = new ManualResetEventSlim(false);
+
+      //start prepared or paused cloned algorithms
+      foreach (IAlgorithm clonedAlgorithm in clonedAlgorithms) {
+        if (pausePending || stopPending) break;
+        if (clonedAlgorithm.ExecutionState == ExecutionState.Prepared ||
+            clonedAlgorithm.ExecutionState == ExecutionState.Paused) {
+          ticket.Wait();
+          clonedAlgorithm.StartAsync(cancellationToken);
+        }
+      }
+
+      signal.Wait();
+      if (pausePending) OnPaused();
+      else OnStopped();
+    }
+
+    public async Task StartAsync() { await StartAsync(CancellationToken.None); }
     public async Task StartAsync(CancellationToken cancellationToken) {
-      this.cancellationToken = cancellationToken;
-      signaler.Reset();
-      await Task.Run(() => {
-        if ((ExecutionState != ExecutionState.Prepared) && (ExecutionState != ExecutionState.Paused))
-          throw new InvalidOperationException(string.Format("Start not allowed in execution state \"{0}\".", ExecutionState));
-
-        if (Algorithm != null) {
-          //create cloned algorithms
-          if (clonedAlgorithms.Count == 0) {
-            int testSamplesCount = (SamplesEnd.Value - SamplesStart.Value) / Folds.Value;
-
-            for (int i = 0; i < Folds.Value; i++) {
-              IAlgorithm clonedAlgorithm = (IAlgorithm)algorithm.Clone();
-              clonedAlgorithm.Name = algorithm.Name + " Fold " + i;
-              IDataAnalysisProblem problem = clonedAlgorithm.Problem as IDataAnalysisProblem;
-              ISymbolicDataAnalysisProblem symbolicProblem = problem as ISymbolicDataAnalysisProblem;
-
-              int testStart = (i * testSamplesCount) + SamplesStart.Value;
-              int testEnd = (i + 1) == Folds.Value ? SamplesEnd.Value : (i + 1) * testSamplesCount + SamplesStart.Value;
-
-              problem.ProblemData.TrainingPartition.Start = SamplesStart.Value;
-              problem.ProblemData.TrainingPartition.End = SamplesEnd.Value;
-              problem.ProblemData.TestPartition.Start = testStart;
-              problem.ProblemData.TestPartition.End = testEnd;
-              DataAnalysisProblemData problemData = problem.ProblemData as DataAnalysisProblemData;
-              if (problemData != null) {
-                problemData.TrainingPartitionParameter.Hidden = false;
-                problemData.TestPartitionParameter.Hidden = false;
-              }
-
-              if (symbolicProblem != null) {
-                symbolicProblem.FitnessCalculationPartition.Start = SamplesStart.Value;
-                symbolicProblem.FitnessCalculationPartition.End = SamplesEnd.Value;
-              }
-              clonedAlgorithm.Prepare();
-              clonedAlgorithms.Add(clonedAlgorithm);
-            }
-          }
-
-          //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.StartAsync(cancellationToken);
-                  signal.WaitOne();
-                  clonedAlgorithm.Started -= signalSetter;
-
-                  startedAlgorithms++;
-                }
-              }
-            }
-          }
-          OnStarted();
-        }
-      }, cancellationToken);
+      await Task.Factory.StartNew((ct) => Start((CancellationToken)ct), cancellationToken, cancellationToken);
     }
 
@@ -346 +340 @@
       if (!pausePending) {
         pausePending = true;
-        PauseAllClonedAlgorithms();
-      }
-    }
-    private void PauseAllClonedAlgorithms() {
-      foreach (IAlgorithm clonedAlgorithm in clonedAlgorithms) {
-        if (clonedAlgorithm.ExecutionState == ExecutionState.Started)
-          clonedAlgorithm.Pause();
+        var toPause = clonedAlgorithms.Where(x => x.ExecutionState == ExecutionState.Started);
+        if (toPause.Any()) {
+          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) { }
+          }
+        }
+        if (ExecutionState != ExecutionState.Paused) OnPaused();
       }
     }
@@ -363 +358 @@
       if (!stopPending) {
         stopPending = true;
-        StopAllClonedAlgorithms();
-      }
-    }
-    private void StopAllClonedAlgorithms() {
-      foreach (IAlgorithm clonedAlgorithm in clonedAlgorithms) {
-        if (clonedAlgorithm.ExecutionState == ExecutionState.Started ||
-            clonedAlgorithm.ExecutionState == ExecutionState.Paused)
-          clonedAlgorithm.Stop();
+        var toStop = clonedAlgorithms.Where(x => x.ExecutionState == ExecutionState.Started || x.ExecutionState == ExecutionState.Paused);
+        if (toStop.Any()) {
+          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) { }
+          }
+        }
+        if (ExecutionState != ExecutionState.Stopped) OnStopped();
       }
     }
@@ -602 +597 @@
     private void Algorithm_ExecutionStateChanged(object sender, EventArgs e) {
       switch (Algorithm.ExecutionState) {
-        case ExecutionState.Prepared: OnPrepared();
+        case ExecutionState.Prepared:
+          OnPrepared();
           break;
         case ExecutionState.Started: throw new InvalidOperationException("Algorithm template can not be started.");
         case ExecutionState.Paused: throw new InvalidOperationException("Algorithm template can not be paused.");
-        case ExecutionState.Stopped: OnStopped();
+        case ExecutionState.Stopped:
+          OnStopped();
           break;
       }
@@ -674 +671 @@
     private void ClonedAlgorithm_Paused(object sender, EventArgs e) {
       lock (locker) {
-        if (pausePending && clonedAlgorithms.All(alg => alg.ExecutionState != ExecutionState.Started))
-          OnPaused();
+        if (ExecutionState != ExecutionState.Paused) {
+          if (clonedAlgorithms.All(alg => alg.ExecutionState != ExecutionState.Started)) {
+            pausePending = true;
+            signal.Set();
+            ticket.Release();
+          }
+        }
       }
     }
@@ -681 +683 @@
     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.StartAsync(cancellationToken);
-        }
         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 (clonedAlgorithms.All(alg => alg.ExecutionState == ExecutionState.Stopped || stopPending && alg.ExecutionState == ExecutionState.Prepared)) {
+            stopPending = true;
+            signal.Set();
+          }
+          ticket.Release();
         }
       }
@@ -727 +723 @@
       pausePending = false;
       ExecutionState = ExecutionState.Paused;
-      signaler.Set();
       EventHandler handler = Paused;
       if (handler != null) handler(this, EventArgs.Empty);
@@ -740 +735 @@
       runs.Add(new Run(string.Format("{0} Run {1}", Name, runsCounter), this));
       ExecutionState = ExecutionState.Stopped;
-      signaler.Set();
       EventHandler handler = Stopped;
       if (handler != null) handler(this, EventArgs.Empty);

branches/Async/HeuristicLab.Algorithms.DataAnalysis/3.4/FixedDataAnalysisAlgorithm.cs

@@ -22 +22 @@
 using System;
 using System.Threading;
-using System.Threading.Tasks;
 using HeuristicLab.Common;
 using HeuristicLab.Optimization;
@@ -70 +69 @@
     }
 
-    public override async Task StartAsync(CancellationToken cancellationToken) {
-      await base.StartAsync(cancellationToken);
-      var cancellationTokenSource = new CancellationTokenSource();
+    public override void Start(CancellationToken cancellationToken) {
+      base.Start(cancellationToken);
+      OnStarted();
 
-      OnStarted();
-      using (var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationTokenSource.Token, cancellationToken)) {
-        Task task = Task.Factory.StartNew(Run, cts.Token, cts.Token);
-        await task.ContinueWith(t => {
-          try {
-            t.Wait();
-          }
-          catch (AggregateException ex) {
-            try {
-              ex.Flatten().Handle(x => x is OperationCanceledException);
-            }
-            catch (AggregateException remaining) {
-              if (remaining.InnerExceptions.Count == 1) OnExceptionOccurred(remaining.InnerExceptions[0]);
-              else OnExceptionOccurred(remaining);
-            }
-          }
-          cancellationTokenSource.Dispose();
-          cancellationTokenSource = null;
-          OnStopped();
-        });
+      try {
+        Run(cancellationToken);
+      } catch (OperationCanceledException) {
+      } catch (AggregateException ae) {
+        if (ae.InnerExceptions.Count == 1) OnExceptionOccurred(ae.InnerExceptions[0]);
+        else OnExceptionOccurred(ae);
+      } catch (Exception e) {
+        OnExceptionOccurred(e);
       }
+
+      OnStopped();
     }
     private void Run(object state) {