source: branches/ParallelEngine/HeuristicLab.ParallelEngine/3.3/ParallelEngine.cs @ 5176

#region License Information
/* HeuristicLab
 * Copyright (C) 2002-2010 Heuristic and Evolutionary Algorithms Laboratory (HEAL)
 *
 * This file is part of HeuristicLab.
 *
 * HeuristicLab is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * HeuristicLab is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with HeuristicLab. If not, see <http://www.gnu.org/licenses/>.
 */
#endregion

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using HeuristicLab.Common;
using HeuristicLab.Core;
using HeuristicLab.Persistence.Default.CompositeSerializers.Storable;

namespace HeuristicLab.ParallelEngine {
  /// <summary>
  /// Represents an engine that executes its steps in parallel (if possible) using multiple threads.
  /// This engine is suitable for parallel processing on shared memory systems which provide multiple cores.
  /// </summary>
  [StorableClass]
  [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;

    [StorableConstructor]
    protected ParallelEngine(bool deserializing) : base(deserializing) { }
    protected ParallelEngine(ParallelEngine original, Cloner cloner) : base(original, cloner) { }
    public ParallelEngine() : base() { }

    public override IDeepCloneable Clone(Cloner cloner) {
      return new ParallelEngine(this, cloner);
    }

    protected override void ProcessNextOperation() {
      using (cancellationTokenSource = new CancellationTokenSource()) {
        cancellationToken = cancellationTokenSource.Token;
        try {
          ProcessNextOperation(ExecutionStack);
        }
        catch (Exception ex) {
          OnExceptionOccurred(ex);
          Pause();
        }
      }
      cancellationTokenSource = null;
    }

    private void ProcessOperations(object state) {
      Stack<IOperation> executionStack = (Stack<IOperation>)state;
      while ((executionStack.Count > 0) && (!cancellationToken.IsCancellationRequested))
        ProcessNextOperation(executionStack);
      cancellationToken.ThrowIfCancellationRequested();
    }

    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);
          }
          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 > 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;
          }

          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);
        }
      }
    }

    public override void Pause() {
      base.Pause();
      if (cancellationTokenSource != null) cancellationTokenSource.Cancel();
    }
    public override void Stop() {
      base.Stop();
      if (cancellationTokenSource != null) cancellationTokenSource.Cancel();
    }
  }
}