Changeset 5185 for branches/ParallelEngine/HeuristicLab.ParallelEngine/3.3/ParallelEngine.cs

Timestamp:

12/31/10 04:29:19 (13 years ago)

Author:

swagner

Message:

Worked on cancellation and refactored code (#1333)

File:

• branches/ParallelEngine/HeuristicLab.ParallelEngine/3.3/ParallelEngine.cs (modified) (2 diffs)

branches/ParallelEngine/HeuristicLab.ParallelEngine/3.3/ParallelEngine.cs

@@ -36 +36 @@
   [Item("Parallel Engine", "Engine for parallel execution of algorithms using multiple threads (suitable for shared memory systems with multiple cores).")]
   public class ParallelEngine : Engine {
-    private CancellationTokenSource cancellationTokenSource;
     private CancellationToken cancellationToken;
 
@@ -48 +47 @@
     }
 
-    protected override void ProcessNextOperation() {
-      using (cancellationTokenSource = new CancellationTokenSource()) {
-        cancellationToken = cancellationTokenSource.Token;
-        try {
-          ProcessNextOperation(ExecutionStack);
-        }
-        catch (Exception ex) {
-          OnExceptionOccurred(ex);
-          Pause();
-        }
-      }
-      cancellationTokenSource = null;
+    protected override void Run(CancellationToken cancellationToken) {
+      this.cancellationToken = cancellationToken;
+      Run(ExecutionStack);
     }
 
-    private void ProcessOperations(object state) {
+    private void Run(object state) {
       Stack<IOperation> executionStack = (Stack<IOperation>)state;
-      while ((executionStack.Count > 0) && (!cancellationToken.IsCancellationRequested))
-        ProcessNextOperation(executionStack);
-      cancellationToken.ThrowIfCancellationRequested();
-    }
+      IOperation next;
+      OperationCollection coll;
+      IAtomicOperation operation;
 
-    private void ProcessNextOperation(Stack<IOperation> executionStack) {
-      IOperation next = executionStack.Pop();
-      OperationCollection coll = next as OperationCollection;
-      while (coll != null) {
-        if (coll.Parallel) {
-          Task[] tasks = new Task[coll.Count];
-          Stack<IOperation>[] stacks = new Stack<IOperation>[coll.Count];
-          for (int i = 0; i < coll.Count; i++) {
-            stacks[i] = new Stack<IOperation>();
-            stacks[i].Push(coll[i]);
-            tasks[i] = Task.Factory.StartNew(new Action<object>(ProcessOperations), stacks[i], cancellationToken);
+      while (executionStack.Count > 0) {
+        cancellationToken.ThrowIfCancellationRequested();
+
+        next = executionStack.Pop();
+        if (next is OperationCollection) {
+          coll = (OperationCollection)next;
+          if (coll.Parallel) {
+            Task[] tasks = new Task[coll.Count];
+            Stack<IOperation>[] stacks = new Stack<IOperation>[coll.Count];
+            for (int i = 0; i < coll.Count; i++) {
+              stacks[i] = new Stack<IOperation>();
+              stacks[i].Push(coll[i]);
+              tasks[i] = Task.Factory.StartNew(Run, stacks[i], cancellationToken);
+            }
+            try {
+              Task.WaitAll(tasks);
+            }
+            catch (AggregateException ex) {
+              OperationCollection remaining = new OperationCollection() { Parallel = true };
+              for (int i = 0; i < stacks.Length; i++) {
+                if (stacks[i].Count == 1)
+                  remaining.Add(stacks[i].Pop());
+                if (stacks[i].Count > 1) {
+                  OperationCollection ops = new OperationCollection();
+                  while (stacks[i].Count > 0)
+                    ops.Add(stacks[i].Pop());
+                  remaining.Add(ops);
+                }
+              }
+              if (remaining.Count > 0) executionStack.Push(remaining);
+              throw ex;
+            }
+          } else {
+            for (int i = coll.Count - 1; i >= 0; i--)
+              if (coll[i] != null) ExecutionStack.Push(coll[i]);
           }
+        } else if (next is IAtomicOperation) {
+          operation = (IAtomicOperation)next;
           try {
-            Task.WaitAll(tasks);
+            next = operation.Operator.Execute((IExecutionContext)operation, cancellationToken);
           }
-          catch (AggregateException ex) {
-            OperationCollection remaining = new OperationCollection() { Parallel = true };
-            for (int i = 0; i < stacks.Length; i++) {
-              if (stacks[i].Count > 0) {
-                OperationCollection ops = new OperationCollection();
-                while (stacks[i].Count > 0)
-                  ops.Add(stacks[i].Pop());
-                remaining.Add(ops);
-              }
-            }
-            executionStack.Push(remaining);
-            ex.Flatten().Handle(x => x is OperationCanceledException);
-            return;
+          catch (Exception ex) {
+            executionStack.Push(operation);
+            throw new OperatorExecutionException(operation.Operator, ex);
           }
-
-          next = executionStack.Count > 0 ? executionStack.Pop() : null;
-          coll = next as OperationCollection;
-        } else {
-          for (int i = coll.Count - 1; i >= 0; i--)
-            executionStack.Push(coll[i]);
-          next = executionStack.Count > 0 ? executionStack.Pop() : null;
-          coll = next as OperationCollection;
-        }
-      }
-      IAtomicOperation operation = next as IAtomicOperation;
-      if (operation != null) {
-        try {
-          executionStack.Push(operation.Operator.Execute((IExecutionContext)operation));
-        }
-        catch (Exception ex) {
-          executionStack.Push(operation);
-          throw new OperatorExecutionException(operation.Operator, ex);
+          if (next != null) executionStack.Push(next);
         }
       }
     }
-
-    public override void Pause() {
-      base.Pause();
-      if (cancellationTokenSource != null) cancellationTokenSource.Cancel();
-    }
-    public override void Stop() {
-      base.Stop();
-      if (cancellationTokenSource != null) cancellationTokenSource.Cancel();
-    }
   }
 }