source: branches/MPI/HeuristicLab.Operators.MPISupport/3.3/MPIUniformSubscopesProcessor.cs @ 13508

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using HeuristicLab.Core;
using HeuristicLab.Persistence.Default.CompositeSerializers.Storable;
using HeuristicLab.Parameters;
using HeuristicLab.Data;
using HeuristicLab.Common;
using MPI;
using HeuristicLab.Operators.MPISupport.BinaryTransport;

namespace HeuristicLab.Operators.MPISupport {
  [Item("MPIUniformSubscopesProcessor", "An operator which applies a specified operator on all sub-scopes at the given depth of the current scope.")]
  [StorableClass]
  public class MPIUniformSubscopesProcessor : SingleSuccessorOperator {
    private OperatorParameter OperatorParameter {
      get { return (OperatorParameter)Parameters["Operator"]; }
    }
    public ValueLookupParameter<BoolValue> ParallelParameter {
      get { return (ValueLookupParameter<BoolValue>)Parameters["Parallel"]; }
    }

    public IOperator Operator {
      get { return OperatorParameter.Value; }
      set { OperatorParameter.Value = value; }
    }
    public BoolValue Parallel {
      get { return ParallelParameter.Value; }
      set { ParallelParameter.Value = value; }
    }

    [StorableConstructor]
    private MPIUniformSubscopesProcessor(bool deserializing) : base(deserializing) { }
    private MPIUniformSubscopesProcessor(MPIUniformSubscopesProcessor original, Cloner cloner)
      : base(original, cloner) {
    }
    public MPIUniformSubscopesProcessor()
      : base() {
      Parameters.Add(new OperatorParameter("Operator", "The operator which should be applied on all sub-scopes of the current scope."));
      Parameters.Add(new ValueLookupParameter<BoolValue>("Parallel", "True if the operator should be applied in parallel on all sub-scopes, otherwise false.", new BoolValue(false)));
    }

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

    public override IOperation Apply() {
      OperationCollection next = new OperationCollection(base.Apply());
      ScopeList scopes = ExecutionContext.Scope.SubScopes;

      if (Operator != null) {
        if (MPI.Communicator.world != null && MPI.Communicator.world.Size > 2) {
          int rank = MPI.Communicator.world.Rank;
          int size = MPI.Communicator.world.Size -1;

          int count = scopes.Count / size;
          int start = count * (rank - 1);
          int end = (rank == size ? scopes.Count : start + count);

          List<Request> requests = new List<Request>();
          IScope parent = new Scope();

          for (int i = start; i < end; i++) {
            IAtomicOperation op = ExecutionContext.CreateOperation(Operator, scopes[i]);
            MPIHelper.Execute(op);
            
            //SEND results to other clients
            parent.SubScopes.Add(MPIHelper.ShallowCopy(scopes[i]));           
          }

          for (int dest = 1; dest <= size; dest++) {
            if (dest != rank) {
              var result = new MPIBinaryTransportWrapper(parent);
              requests.Add(MPI.Communicator.world.ImmediateSend<MPIBinaryTransportWrapper>(result, dest, 0));
            }
          }

          IExecutionContext globalScope = ExecutionContext;
          while (globalScope.Parent != null) {
            globalScope = globalScope.Parent;
          }

          //RECEIVE results from other clients
          for (int source = 1; source <= size; source++) {
            if (source != rank) {
              var result = MPI.Communicator.world.Receive<MPIBinaryTransportWrapper>(source, 0).GetInnerItem(globalScope) as IScope;

              /*Console.WriteLine(result.SubScopes.Count);
              foreach (IVariable var in result.SubScopes[0].SubScopes[0].Variables) {
                Console.WriteLine("VAR " + var.Name + " " + var.Value);
              }*/ 

              int offset = count * (source - 1);
              for (int scopeIndex = 0; scopeIndex < result.SubScopes.Count; scopeIndex++) {
                scopes[scopeIndex + offset] = result.SubScopes[scopeIndex];
              } 

              /*for (int scopeIndex = 0; scopeIndex < count; scopeIndex++) {
                IScope root = new Scope();
                root.SubScopes.Add(scopes[scopeIndex + offset]);
                scopes[scopeIndex + offset] = root;
              }*/
            }
          }

          foreach (Request request in requests) {
            request.Wait();
          }
        } else {
          OperationCollection inner = new OperationCollection();
          inner.Parallel = Parallel == null ? false : Parallel.Value;
          for (int i = 0; i < scopes.Count; i++) {
            inner.Add(ExecutionContext.CreateOperation(Operator, scopes[i]));
          }
          next.Insert(0, inner);
        }
      }

      return next;
    }
  }
}