Changeset 9323 for trunk/sources/HeuristicLab.ParallelEngine

Timestamp:

03/20/13 16:51:58 (11 years ago)

Author:

mkommend

Message:

#2024: Changed parallel engine to use parallel.foreach instead of spawning task manually to enable more options for parallelization.

File:

• trunk/sources/HeuristicLab.ParallelEngine/3.3/ParallelEngine.cs (modified) (3 diffs)

trunk/sources/HeuristicLab.ParallelEngine/3.3/ParallelEngine.cs

@@ -37 +37 @@
   public class ParallelEngine : Engine {
     private CancellationToken cancellationToken;
+    private ParallelOptions parallelOptions;
+
+    [Storable(DefaultValue = -1)]
+    private int degreeOfParallelism;
+    public int DegreeOfParallelism {
+      get { return degreeOfParallelism; }
+      set {
+        if (degreeOfParallelism != value) {
+          degreeOfParallelism = value;
+          OnDegreeOfParallelismChanged();
+        }
+      }
+    }
 
     [StorableConstructor]
     protected ParallelEngine(bool deserializing) : base(deserializing) { }
-    protected ParallelEngine(ParallelEngine original, Cloner cloner) : base(original, cloner) { }
+    protected ParallelEngine(ParallelEngine original, Cloner cloner)
+      : base(original, cloner) {
+      this.DegreeOfParallelism = original.DegreeOfParallelism;
+    }
     public ParallelEngine() : base() { }
 
@@ -47 +63 @@
     }
 
+    public event EventHandler DegreeOfParallelismChanged;
+    protected void OnDegreeOfParallelismChanged() {
+      var handler = DegreeOfParallelismChanged;
+      if (handler != null) handler(this, EventArgs.Empty);
+    }
+
+
     protected override void Run(CancellationToken cancellationToken) {
       this.cancellationToken = cancellationToken;
+      parallelOptions = new ParallelOptions();
+      parallelOptions.MaxDegreeOfParallelism = DegreeOfParallelism;
       Run(ExecutionStack);
     }
@@ -65 +90 @@
           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);
+              Parallel.ForEach(stacks, parallelOptions, Run);
             }
             catch (AggregateException ex) {