communicator.hh

Go to the documentation of this file.

// -*- tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*-
// vi: set et ts=4 sw=2 sts=2:
// SPDX-FileCopyrightInfo: Copyright © DUNE Project contributors, see file LICENSE.md in module root
// SPDX-License-Identifier: LicenseRef-GPL-2.0-only-with-DUNE-exception
#ifndef DUNE_COMMON_PARALLEL_COMMUNICATOR_HH
#define DUNE_COMMON_PARALLEL_COMMUNICATOR_HH
 
#if HAVE_MPI
 
#include <cassert>
#include <cstddef>
#include <iostream>
#include <map>
#include <type_traits>
#include <utility>
 
#include <mpi.h>
 
#include <dune/common/exceptions.hh>
#include <dune/common/parallel/interface.hh>
#include <dune/common/parallel/remoteindices.hh>
#include <dune/common/stdstreams.hh>
 
namespace Dune
{
  struct SizeOne
  {};
 
  struct VariableSize
  {};
 
 
  template<class V>
  struct CommPolicy
  {
    typedef V Type;
 
    typedef typename V::value_type IndexedType;
 
    typedef SizeOne IndexedTypeFlag;
 
    static const void* getAddress(const V& v, int index);
 
    static int getSize(const V&, int index);
  };
 
  template<class K, int n> class FieldVector;
 
  template<class B, class A> class VariableBlockVector;
 
  template<class K, class A, int n>
  struct CommPolicy<VariableBlockVector<FieldVector<K, n>, A> >
  {
    typedef VariableBlockVector<FieldVector<K, n>, A> Type;
 
    typedef typename Type::B IndexedType;
 
    typedef VariableSize IndexedTypeFlag;
 
    static const void* getAddress(const Type& v, int i);
 
    static int getSize(const Type& v, int i);
  };
 
  class CommunicationError : public IOError
  {};
 
  template<class T>
  struct CopyGatherScatter
  {
    typedef typename CommPolicy<T>::IndexedType IndexedType;
 
    static const IndexedType& gather(const T& vec, std::size_t i);
 
    static void scatter(T& vec, const IndexedType& v, std::size_t i);
 
  };
 
  template<typename T>
  class DatatypeCommunicator : public InterfaceBuilder
  {
  public:
 
    typedef T ParallelIndexSet;
 
    typedef Dune::RemoteIndices<ParallelIndexSet> RemoteIndices;
 
    typedef typename RemoteIndices::GlobalIndex GlobalIndex;
 
    typedef typename RemoteIndices::Attribute Attribute;
 
    typedef typename RemoteIndices::LocalIndex LocalIndex;
 
    DatatypeCommunicator();
 
    ~DatatypeCommunicator();
 
    template<class T1, class T2, class V>
    void build(const RemoteIndices& remoteIndices, const T1& sourceFlags, V& sendData, const T2& destFlags, V& receiveData);
 
    void forward();
 
    void backward();
 
    void free();
  private:
    constexpr static int commTag_ = 234;
 
    const RemoteIndices* remoteIndices_;
 
    typedef std::map<int,std::pair<MPI_Datatype,MPI_Datatype> >
    MessageTypeMap;
 
    MessageTypeMap messageTypes;
 
    void* data_;
 
    MPI_Request* requests_[2];
 
    bool created_;
 
    template<class V, bool FORWARD>
    void createRequests(V& sendData, V& receiveData);
 
    template<class T1, class T2, class V, bool send>
    void createDataTypes(const T1& source, const T2& destination, V& data);
 
    void sendRecv(MPI_Request* req);
 
    struct IndexedTypeInformation
    {
      void build(int i)
      {
        length = new int[i];
        displ  = new MPI_Aint[i];
        size = i;
      }
 
      void free()
      {
        delete[] length;
        delete[] displ;
      }
      int* length;
      MPI_Aint* displ;
      int elements;
      int size;
    };
 
    template<class V>
    struct MPIDatatypeInformation
    {
      MPIDatatypeInformation(const V& data) : data_(data)
      {}
 
      void reserve(int proc, int size)
      {
        information_[proc].build(size);
      }
      void add(int proc, int local)
      {
        IndexedTypeInformation& info=information_[proc];
        assert((info.elements)<info.size);
        MPI_Get_address( const_cast<void*>(CommPolicy<V>::getAddress(data_, local)),
                         info.displ+info.elements);
        info.length[info.elements]=CommPolicy<V>::getSize(data_, local);
        info.elements++;
      }
 
      std::map<int,IndexedTypeInformation> information_;
      const V& data_;
 
    };
 
  };
 
  class BufferedCommunicator
  {
 
  public:
    BufferedCommunicator();
 
    template<class Data, class Interface>
    typename std::enable_if<std::is_same<SizeOne,typename CommPolicy<Data>::IndexedTypeFlag>::value, void>::type
    build(const Interface& interface);
 
    template<class Data, class Interface>
    void build(const Data& source, const Data& target, const Interface& interface);
 
    template<class GatherScatter, class Data>
    void forward(const Data& source, Data& dest);
 
    template<class GatherScatter, class Data>
    void backward(Data& source, const Data& dest);
 
    template<class GatherScatter, class Data>
    void forward(Data& data);
 
    template<class GatherScatter, class Data>
    void backward(Data& data);
 
    void free();
 
    ~BufferedCommunicator();
 
  private:
 
    typedef std::map<int,std::pair<InterfaceInformation,InterfaceInformation> >
    InterfaceMap;
 
 
    template<class Data, typename IndexedTypeFlag>
    struct MessageSizeCalculator
    {};
 
    template<class Data>
    struct MessageSizeCalculator<Data,SizeOne>
    {
      inline int operator()(const InterfaceInformation& info) const;
      inline int operator()(const Data& data, const InterfaceInformation& info) const;
    };
 
    template<class Data>
    struct MessageSizeCalculator<Data,VariableSize>
    {
      inline int operator()(const Data& data, const InterfaceInformation& info) const;
    };
 
    template<class Data, class GatherScatter, bool send, typename IndexedTypeFlag>
    struct MessageGatherer
    {};
 
    template<class Data, class GatherScatter, bool send>
    struct MessageGatherer<Data,GatherScatter,send,SizeOne>
    {
      typedef typename CommPolicy<Data>::IndexedType Type;
 
      typedef GatherScatter Gatherer;
 
      constexpr static bool forward = send;
 
      inline void operator()(const InterfaceMap& interface, const Data& data, Type* buffer, size_t bufferSize) const;
    };
 
    template<class Data, class GatherScatter, bool send>
    struct MessageGatherer<Data,GatherScatter,send,VariableSize>
    {
      typedef typename CommPolicy<Data>::IndexedType Type;
 
      typedef GatherScatter Gatherer;
 
      constexpr static bool forward = send;
 
      inline void operator()(const InterfaceMap& interface, const Data& data, Type* buffer, size_t bufferSize) const;
    };
 
    template<class Data, class GatherScatter, bool send, typename IndexedTypeFlag>
    struct MessageScatterer
    {};
 
    template<class Data, class GatherScatter, bool send>
    struct MessageScatterer<Data,GatherScatter,send,SizeOne>
    {
      typedef typename CommPolicy<Data>::IndexedType Type;
 
      typedef GatherScatter Scatterer;
 
      constexpr static bool forward = send;
 
      inline void operator()(const InterfaceMap& interface, Data& data, Type* buffer, const int& proc) const;
    };
    template<class Data, class GatherScatter, bool send>
    struct MessageScatterer<Data,GatherScatter,send,VariableSize>
    {
      typedef typename CommPolicy<Data>::IndexedType Type;
 
      typedef GatherScatter Scatterer;
 
      constexpr static bool forward = send;
 
      inline void operator()(const InterfaceMap& interface, Data& data, Type* buffer, const int& proc) const;
    };
 
    struct MessageInformation
    {
      MessageInformation()
        : start_(0), size_(0)
      {}
 
      MessageInformation(size_t start, size_t size)
        : start_(start), size_(size)
      {}
      size_t start_;
      size_t size_;
    };
 
    typedef std::map<int,std::pair<MessageInformation,MessageInformation> >
    InformationMap;
    InformationMap messageInformation_;
    char* buffers_[2];
    size_t bufferSize_[2];
 
    constexpr static int commTag_ = 0;
 
    std::map<int,std::pair<InterfaceInformation,InterfaceInformation> > interfaces_;
 
    MPI_Comm communicator_;
 
    template<class GatherScatter, bool FORWARD, class Data>
    void sendRecv(const Data& source, Data& target);
 
  };
 
#ifndef DOXYGEN
 
  template<class V>
  inline const void* CommPolicy<V>::getAddress(const V& v, int index)
  {
    return &(v[index]);
  }
 
  template<class V>
  inline int CommPolicy<V>::getSize([[maybe_unused]] const V& v, [[maybe_unused]] int index)
  {
    return 1;
  }
 
  template<class K, class A, int n>
  inline const void* CommPolicy<VariableBlockVector<FieldVector<K, n>, A> >::getAddress(const Type& v, int index)
  {
    return &(v[index][0]);
  }
 
  template<class K, class A, int n>
  inline int CommPolicy<VariableBlockVector<FieldVector<K, n>, A> >::getSize(const Type& v, int index)
  {
    return v[index].getsize();
  }
 
  template<class T>
  inline const typename CopyGatherScatter<T>::IndexedType& CopyGatherScatter<T>::gather(const T & vec, std::size_t i)
  {
    return vec[i];
  }
 
  template<class T>
  inline void CopyGatherScatter<T>::scatter(T& vec, const IndexedType& v, std::size_t i)
  {
    vec[i]=v;
  }
 
  template<typename T>
  DatatypeCommunicator<T>::DatatypeCommunicator()
    : remoteIndices_(0), created_(false)
  {
    requests_[0]=0;
    requests_[1]=0;
  }
 
 
 
  template<typename T>
  DatatypeCommunicator<T>::~DatatypeCommunicator()
  {
    free();
  }
 
  template<typename T>
  template<class T1, class T2, class V>
  inline void DatatypeCommunicator<T>::build(const RemoteIndices& remoteIndices,
                                             const T1& source, V& sendData,
                                             const T2& destination, V& receiveData)
  {
    remoteIndices_ = &remoteIndices;
    free();
    createDataTypes<T1,T2,V,false>(source,destination, receiveData);
    createDataTypes<T1,T2,V,true>(source,destination, sendData);
    createRequests<V,true>(sendData, receiveData);
    createRequests<V,false>(receiveData, sendData);
    created_=true;
  }
 
  template<typename T>
  void DatatypeCommunicator<T>::free()
  {
    if(created_) {
      delete[] requests_[0];
      delete[] requests_[1];
      typedef MessageTypeMap::iterator iterator;
      typedef MessageTypeMap::const_iterator const_iterator;
 
      const const_iterator end=messageTypes.end();
 
      for(iterator process = messageTypes.begin(); process != end; ++process) {
        MPI_Datatype *type = &(process->second.first);
        int finalized=0;
        MPI_Finalized(&finalized);
        if(*type!=MPI_DATATYPE_NULL && !finalized)
          MPI_Type_free(type);
        type = &(process->second.second);
        if(*type!=MPI_DATATYPE_NULL && !finalized)
          MPI_Type_free(type);
      }
      messageTypes.clear();
      created_=false;
    }
 
  }
 
  template<typename T>
  template<class T1, class T2, class V, bool send>
  void DatatypeCommunicator<T>::createDataTypes(const T1& sourceFlags, const T2& destFlags, V& data)
  {
 
    MPIDatatypeInformation<V>  dataInfo(data);
    this->template buildInterface<RemoteIndices,T1,T2,MPIDatatypeInformation<V>,send>(*remoteIndices_,sourceFlags, destFlags, dataInfo);
 
    typedef typename RemoteIndices::RemoteIndexMap::const_iterator const_iterator;
    const const_iterator end=this->remoteIndices_->end();
 
    // Allocate MPI_Datatypes and deallocate memory for the type construction.
    for(const_iterator process=this->remoteIndices_->begin(); process != end; ++process) {
      IndexedTypeInformation& info=dataInfo.information_[process->first];
      // Shift the displacement
      MPI_Aint base;
      MPI_Get_address(const_cast<void *>(CommPolicy<V>::getAddress(data, 0)), &base);
 
      for(int i=0; i< info.elements; i++) {
        info.displ[i]-=base;
      }
 
      // Create data type
      MPI_Datatype* type = &( send ? messageTypes[process->first].first : messageTypes[process->first].second);
      MPI_Type_create_hindexed(info.elements, info.length, info.displ,
                               MPITraits<typename CommPolicy<V>::IndexedType>::getType(), type);
      MPI_Type_commit(type);
      // Deallocate memory
      info.free();
    }
  }
 
  template<typename T>
  template<class V, bool createForward>
  void DatatypeCommunicator<T>::createRequests(V& sendData, V& receiveData)
  {
    typedef std::map<int,std::pair<MPI_Datatype,MPI_Datatype> >::const_iterator MapIterator;
    int rank;
    static int index = createForward ? 1 : 0;
    int noMessages = messageTypes.size();
    // allocate request handles
    requests_[index] = new MPI_Request[2*noMessages];
    const MapIterator end = messageTypes.end();
    int request=0;
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
 
    // Set up the requests for receiving first
    for(MapIterator process = messageTypes.begin(); process != end;
        ++process, ++request) {
      MPI_Datatype type = createForward ? process->second.second : process->second.first;
      void* address = const_cast<void*>(CommPolicy<V>::getAddress(receiveData,0));
      MPI_Recv_init(address, 1, type, process->first, commTag_, this->remoteIndices_->communicator(), requests_[index]+request);
    }
 
    // And now the send requests
 
    for(MapIterator process = messageTypes.begin(); process != end;
        ++process, ++request) {
      MPI_Datatype type = createForward ? process->second.first : process->second.second;
      void* address =  const_cast<void*>(CommPolicy<V>::getAddress(sendData, 0));
      MPI_Ssend_init(address, 1, type, process->first, commTag_, this->remoteIndices_->communicator(), requests_[index]+request);
    }
  }
 
  template<typename T>
  void DatatypeCommunicator<T>::forward()
  {
    sendRecv(requests_[1]);
  }
 
  template<typename T>
  void DatatypeCommunicator<T>::backward()
  {
    sendRecv(requests_[0]);
  }
 
  template<typename T>
  void DatatypeCommunicator<T>::sendRecv(MPI_Request* requests)
  {
    int noMessages = messageTypes.size();
    // Start the receive calls first
    MPI_Startall(noMessages, requests);
    // Now the send calls
    MPI_Startall(noMessages, requests+noMessages);
 
    // Wait for completion of the communication send first then receive
    MPI_Status* status=new MPI_Status[2*noMessages];
    for(int i=0; i<2*noMessages; i++)
      status[i].MPI_ERROR=MPI_SUCCESS;
 
    int send = MPI_Waitall(noMessages, requests+noMessages, status+noMessages);
    int receive = MPI_Waitall(noMessages, requests, status);
 
    // Error checks
    int success=1, globalSuccess=0;
    if(send==MPI_ERR_IN_STATUS) {
      int rank;
      MPI_Comm_rank(this->remoteIndices_->communicator(), &rank);
      std::cerr<<rank<<": Error in sending :"<<std::endl;
      // Search for the error
      for(int i=noMessages; i< 2*noMessages; i++)
        if(status[i].MPI_ERROR!=MPI_SUCCESS) {
          char message[300];
          int messageLength;
          MPI_Error_string(status[i].MPI_ERROR, message, &messageLength);
          std::cerr<<" source="<<status[i].MPI_SOURCE<<" message: ";
          for(int j = 0; j < messageLength; j++)
            std::cout << message[j];
        }
      std::cerr<<std::endl;
      success=0;
    }
 
    if(receive==MPI_ERR_IN_STATUS) {
      int rank;
      MPI_Comm_rank(this->remoteIndices_->communicator(), &rank);
      std::cerr<<rank<<": Error in receiving!"<<std::endl;
      // Search for the error
      for(int i=0; i< noMessages; i++)
        if(status[i].MPI_ERROR!=MPI_SUCCESS) {
          char message[300];
          int messageLength;
          MPI_Error_string(status[i].MPI_ERROR, message, &messageLength);
          std::cerr<<" source="<<status[i].MPI_SOURCE<<" message: ";
          for(int j = 0; j < messageLength; j++)
            std::cerr << message[j];
        }
      std::cerr<<std::endl;
      success=0;
    }
 
    MPI_Allreduce(&success, &globalSuccess, 1, MPI_INT, MPI_MIN, this->remoteIndices_->communicator());
 
    delete[] status;
 
    if(!globalSuccess)
      DUNE_THROW(CommunicationError, "A communication error occurred!");
 
  }
 
  inline BufferedCommunicator::BufferedCommunicator()
  {
    buffers_[0]=0;
    buffers_[1]=0;
    bufferSize_[0]=0;
    bufferSize_[1]=0;
  }
 
  template<class Data, class Interface>
  typename std::enable_if<std::is_same<SizeOne, typename CommPolicy<Data>::IndexedTypeFlag>::value, void>::type
  BufferedCommunicator::build(const Interface& interface)
  {
    interfaces_=interface.interfaces();
    communicator_=interface.communicator();
    typedef typename std::map<int,std::pair<InterfaceInformation,InterfaceInformation> >
    ::const_iterator const_iterator;
    typedef typename CommPolicy<Data>::IndexedTypeFlag Flag;
    const const_iterator end = interfaces_.end();
    int lrank;
    MPI_Comm_rank(communicator_, &lrank);
 
    bufferSize_[0]=0;
    bufferSize_[1]=0;
 
    for(const_iterator interfacePair = interfaces_.begin();
        interfacePair != end; ++interfacePair) {
      int noSend = MessageSizeCalculator<Data,Flag>() (interfacePair->second.first);
      int noRecv = MessageSizeCalculator<Data,Flag>() (interfacePair->second.second);
      if (noSend + noRecv > 0)
        messageInformation_.insert(std::make_pair(interfacePair->first,
                                                std::make_pair(MessageInformation(bufferSize_[0],
                                                                                  noSend*sizeof(typename CommPolicy<Data>::IndexedType)),
                                                               MessageInformation(bufferSize_[1],
                                                                                  noRecv*sizeof(typename CommPolicy<Data>::IndexedType)))));
      bufferSize_[0] += noSend;
      bufferSize_[1] += noRecv;
    }
 
    // allocate the buffers
    bufferSize_[0] *= sizeof(typename CommPolicy<Data>::IndexedType);
    bufferSize_[1] *= sizeof(typename CommPolicy<Data>::IndexedType);
 
    buffers_[0] = new char[bufferSize_[0]];
    buffers_[1] = new char[bufferSize_[1]];
  }
 
  template<class Data, class Interface>
  void BufferedCommunicator::build(const Data& source, const Data& dest, const Interface& interface)
  {
 
    interfaces_=interface.interfaces();
    communicator_=interface.communicator();
    typedef typename std::map<int,std::pair<InterfaceInformation,InterfaceInformation> >
    ::const_iterator const_iterator;
    typedef typename CommPolicy<Data>::IndexedTypeFlag Flag;
    const const_iterator end = interfaces_.end();
 
    bufferSize_[0]=0;
    bufferSize_[1]=0;
 
    for(const_iterator interfacePair = interfaces_.begin();
        interfacePair != end; ++interfacePair) {
      int noSend = MessageSizeCalculator<Data,Flag>() (source, interfacePair->second.first);
      int noRecv = MessageSizeCalculator<Data,Flag>() (dest, interfacePair->second.second);
      if (noSend + noRecv > 0)
        messageInformation_.insert(std::make_pair(interfacePair->first,
                                                std::make_pair(MessageInformation(bufferSize_[0],
                                                                                  noSend*sizeof(typename CommPolicy<Data>::IndexedType)),
                                                               MessageInformation(bufferSize_[1],
                                                                                  noRecv*sizeof(typename CommPolicy<Data>::IndexedType)))));
      bufferSize_[0] += noSend;
      bufferSize_[1] += noRecv;
    }
 
    bufferSize_[0] *= sizeof(typename CommPolicy<Data>::IndexedType);
    bufferSize_[1] *= sizeof(typename CommPolicy<Data>::IndexedType);
    // allocate the buffers
    buffers_[0] = new char[bufferSize_[0]];
    buffers_[1] = new char[bufferSize_[1]];
  }
 
  inline void BufferedCommunicator::free()
  {
    messageInformation_.clear();
    if(buffers_[0])
      delete[] buffers_[0];
 
    if(buffers_[1])
      delete[] buffers_[1];
    buffers_[0]=buffers_[1]=0;
  }
 
  inline BufferedCommunicator::~BufferedCommunicator()
  {
    free();
  }
 
  template<class Data>
  inline int BufferedCommunicator::MessageSizeCalculator<Data,SizeOne>::operator()
    (const InterfaceInformation& info) const
  {
    return info.size();
  }
 
 
  template<class Data>
  inline int BufferedCommunicator::MessageSizeCalculator<Data,SizeOne>::operator()
    (const Data&, const InterfaceInformation& info) const
  {
    return operator()(info);
  }
 
 
  template<class Data>
  inline int BufferedCommunicator::MessageSizeCalculator<Data, VariableSize>::operator()
    (const Data& data, const InterfaceInformation& info) const
  {
    int entries=0;
 
    for(size_t i=0; i < info.size(); i++)
      entries += CommPolicy<Data>::getSize(data,info[i]);
 
    return entries;
  }
 
 
  template<class Data, class GatherScatter, bool FORWARD>
  inline void BufferedCommunicator::MessageGatherer<Data,GatherScatter,FORWARD,VariableSize>::operator()(const InterfaceMap& interfaces,const Data& data, Type* buffer, [[maybe_unused]] size_t bufferSize) const
  {
    typedef typename InterfaceMap::const_iterator
    const_iterator;
 
    int rank;
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    const const_iterator end = interfaces.end();
    size_t index=0;
 
    for(const_iterator interfacePair = interfaces.begin();
        interfacePair != end; ++interfacePair) {
      int size = forward ? interfacePair->second.first.size() :
                 interfacePair->second.second.size();
 
      for(int i=0; i < size; i++) {
        int local = forward ? interfacePair->second.first[i] :
                    interfacePair->second.second[i];
        for(std::size_t j=0; j < CommPolicy<Data>::getSize(data, local); j++, index++) {
 
#ifdef DUNE_ISTL_WITH_CHECKING
          assert(bufferSize>=(index+1)*sizeof(typename CommPolicy<Data>::IndexedType));
#endif
          buffer[index]=GatherScatter::gather(data, local, j);
        }
 
      }
    }
 
  }
 
 
  template<class Data, class GatherScatter, bool FORWARD>
  inline void BufferedCommunicator::MessageGatherer<Data,GatherScatter,FORWARD,SizeOne>::operator()(
    const InterfaceMap& interfaces, const Data& data, Type* buffer, [[maybe_unused]] size_t bufferSize) const
  {
    typedef typename InterfaceMap::const_iterator
    const_iterator;
    const const_iterator end = interfaces.end();
    size_t index = 0;
 
    int rank;
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
 
    for(const_iterator interfacePair = interfaces.begin();
        interfacePair != end; ++interfacePair) {
      size_t size = FORWARD ? interfacePair->second.first.size() :
                    interfacePair->second.second.size();
 
      for(size_t i=0; i < size; i++) {
 
#ifdef DUNE_ISTL_WITH_CHECKING
        assert(bufferSize>=(index+1)*sizeof(typename CommPolicy<Data>::IndexedType));
#endif
 
        buffer[index++] = GatherScatter::gather(data, FORWARD ? interfacePair->second.first[i] :
                                                interfacePair->second.second[i]);
      }
    }
 
  }
 
 
  template<class Data, class GatherScatter, bool FORWARD>
  inline void BufferedCommunicator::MessageScatterer<Data,GatherScatter,FORWARD,VariableSize>::operator()(const InterfaceMap& interfaces, Data& data, Type* buffer, const int& proc) const
  {
    typedef typename InterfaceMap::value_type::second_type::first_type Information;
    const typename InterfaceMap::const_iterator infoPair = interfaces.find(proc);
 
    assert(infoPair!=interfaces.end());
 
    const Information& info = FORWARD ? infoPair->second.second :
                              infoPair->second.first;
 
    for(size_t i=0, index=0; i < info.size(); i++) {
      for(size_t j=0; j < CommPolicy<Data>::getSize(data, info[i]); j++)
        GatherScatter::scatter(data, buffer[index++], info[i], j);
    }
  }
 
 
  template<class Data, class GatherScatter, bool FORWARD>
  inline void BufferedCommunicator::MessageScatterer<Data,GatherScatter,FORWARD,SizeOne>::operator()(const InterfaceMap& interfaces, Data& data, Type* buffer, const int& proc) const
  {
    typedef typename InterfaceMap::value_type::second_type::first_type Information;
    const typename InterfaceMap::const_iterator infoPair = interfaces.find(proc);
 
    assert(infoPair!=interfaces.end());
 
    const Information& info = FORWARD ? infoPair->second.second :
                              infoPair->second.first;
 
    for(size_t i=0; i < info.size(); i++) {
      GatherScatter::scatter(data, buffer[i], info[i]);
    }
  }
 
 
  template<class GatherScatter,class Data>
  void BufferedCommunicator::forward(Data& data)
  {
    this->template sendRecv<GatherScatter,true>(data, data);
  }
 
 
  template<class GatherScatter, class Data>
  void BufferedCommunicator::backward(Data& data)
  {
    this->template sendRecv<GatherScatter,false>(data, data);
  }
 
 
  template<class GatherScatter, class Data>
  void BufferedCommunicator::forward(const Data& source, Data& dest)
  {
    this->template sendRecv<GatherScatter,true>(source, dest);
  }
 
 
  template<class GatherScatter, class Data>
  void BufferedCommunicator::backward(Data& source, const Data& dest)
  {
    this->template sendRecv<GatherScatter,false>(dest, source);
  }
 
 
  template<class GatherScatter, bool FORWARD, class Data>
  void BufferedCommunicator::sendRecv(const Data& source, Data& dest)
  {
    int rank, lrank;
 
    MPI_Comm_rank(MPI_COMM_WORLD,&rank);
    MPI_Comm_rank(MPI_COMM_WORLD,&lrank);
 
    typedef typename CommPolicy<Data>::IndexedType Type;
    Type *sendBuffer, *recvBuffer;
    size_t sendBufferSize;
#ifndef NDEBUG
    size_t recvBufferSize;
#endif
 
    if(FORWARD) {
      sendBuffer = reinterpret_cast<Type*>(buffers_[0]);
      sendBufferSize = bufferSize_[0];
      recvBuffer = reinterpret_cast<Type*>(buffers_[1]);
#ifndef NDEBUG
      recvBufferSize = bufferSize_[1];
#endif
    }else{
      sendBuffer = reinterpret_cast<Type*>(buffers_[1]);
      sendBufferSize = bufferSize_[1];
      recvBuffer = reinterpret_cast<Type*>(buffers_[0]);
#ifndef NDEBUG
      recvBufferSize = bufferSize_[0];
#endif
    }
    typedef typename CommPolicy<Data>::IndexedTypeFlag Flag;
 
    MessageGatherer<Data,GatherScatter,FORWARD,Flag>() (interfaces_, source, sendBuffer, sendBufferSize);
 
    MPI_Request* sendRequests = new MPI_Request[messageInformation_.size()];
    MPI_Request* recvRequests = new MPI_Request[messageInformation_.size()];
    /* Number of recvRequests that are not MPI_REQUEST_NULL */
    size_t numberOfRealRecvRequests = 0;
 
    // Setup receive first
    typedef typename InformationMap::const_iterator const_iterator;
 
    const const_iterator end = messageInformation_.end();
    size_t i=0;
    int* processMap = new int[messageInformation_.size()];
 
    for(const_iterator info = messageInformation_.begin(); info != end; ++info, ++i) {
      processMap[i]=info->first;
      if(FORWARD) {
        assert(info->second.second.start_*sizeof(typename CommPolicy<Data>::IndexedType)+info->second.second.size_ <= recvBufferSize );
        Dune::dvverb<<rank<<": receiving "<<info->second.second.size_<<" from "<<info->first<<std::endl;
        if(info->second.second.size_) {
          MPI_Irecv(recvBuffer+info->second.second.start_, info->second.second.size_,
                    MPI_BYTE, info->first, commTag_, communicator_,
                    recvRequests+i);
          numberOfRealRecvRequests += 1;
        } else {
          // Nothing to receive -> set request to inactive
          recvRequests[i]=MPI_REQUEST_NULL;
        }
      }else{
        assert(info->second.first.start_*sizeof(typename CommPolicy<Data>::IndexedType)+info->second.first.size_ <= recvBufferSize );
        Dune::dvverb<<rank<<": receiving "<<info->second.first.size_<<" to "<<info->first<<std::endl;
        if(info->second.first.size_) {
          MPI_Irecv(recvBuffer+info->second.first.start_, info->second.first.size_,
                    MPI_BYTE, info->first, commTag_, communicator_,
                    recvRequests+i);
          numberOfRealRecvRequests += 1;
        } else {
          // Nothing to receive -> set request to inactive
          recvRequests[i]=MPI_REQUEST_NULL;
        }
      }
    }
 
    // now the send requests
    i=0;
    for(const_iterator info = messageInformation_.begin(); info != end; ++info, ++i)
      if(FORWARD) {
        assert(info->second.second.start_*sizeof(typename CommPolicy<Data>::IndexedType)+info->second.second.size_ <= recvBufferSize );
        Dune::dvverb<<rank<<": sending "<<info->second.first.size_<<" to "<<info->first<<std::endl;
        assert(info->second.first.start_*sizeof(typename CommPolicy<Data>::IndexedType)+info->second.first.size_ <= sendBufferSize );
        if(info->second.first.size_)
          MPI_Issend(sendBuffer+info->second.first.start_, info->second.first.size_,
                     MPI_BYTE, info->first, commTag_, communicator_,
                     sendRequests+i);
        else
          // Nothing to send -> set request to inactive
          sendRequests[i]=MPI_REQUEST_NULL;
      }else{
        assert(info->second.second.start_*sizeof(typename CommPolicy<Data>::IndexedType)+info->second.second.size_ <= sendBufferSize );
        Dune::dvverb<<rank<<": sending "<<info->second.second.size_<<" to "<<info->first<<std::endl;
        if(info->second.second.size_)
          MPI_Issend(sendBuffer+info->second.second.start_, info->second.second.size_,
                     MPI_BYTE, info->first, commTag_, communicator_,
                     sendRequests+i);
        else
          // Nothing to send -> set request to inactive
          sendRequests[i]=MPI_REQUEST_NULL;
      }
 
    // Wait for completion of receive and immediately start scatter
    i=0;
    //int success = 1;
    int finished = MPI_UNDEFINED;
    MPI_Status status; //[messageInformation_.size()];
    //MPI_Waitall(messageInformation_.size(), recvRequests, status);
 
    for(i=0; i< numberOfRealRecvRequests; i++) {
      status.MPI_ERROR=MPI_SUCCESS;
      MPI_Waitany(messageInformation_.size(), recvRequests, &finished, &status);
      assert(finished != MPI_UNDEFINED);
 
      if(status.MPI_ERROR==MPI_SUCCESS) {
        int& proc = processMap[finished];
        typename InformationMap::const_iterator infoIter = messageInformation_.find(proc);
        assert(infoIter != messageInformation_.end());
 
        MessageInformation info = (FORWARD) ? infoIter->second.second : infoIter->second.first;
        assert(info.start_+info.size_ <= recvBufferSize);
 
        MessageScatterer<Data,GatherScatter,FORWARD,Flag>() (interfaces_, dest, recvBuffer+info.start_, proc);
      }else{
        std::cerr<<rank<<": MPI_Error occurred while receiving message from "<<processMap[finished]<<std::endl;
        //success=0;
      }
    }
 
    MPI_Status recvStatus;
 
    // Wait for completion of sends
    for(i=0; i< messageInformation_.size(); i++)
      if(MPI_SUCCESS!=MPI_Wait(sendRequests+i, &recvStatus)) {
        std::cerr<<rank<<": MPI_Error occurred while sending message to "<<processMap[finished]<<std::endl;
        //success=0;
      }
    /*
       int globalSuccess;
       MPI_Allreduce(&success, &globalSuccess, 1, MPI_INT, MPI_MIN, interface_->communicator());
 
       if(!globalSuccess)
       DUNE_THROW(CommunicationError, "A communication error occurred!");
     */
    delete[] processMap;
    delete[] sendRequests;
    delete[] recvRequests;
 
  }
 
#endif  // DOXYGEN
 
}
 
#endif // HAVE_MPI
#endif // DUNE_COMMON_PARALLEL_COMMUNICATOR_HH