Loading...
Searching...
No Matches
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
remoteindices.hh
Classes describing a distributed indexset.
exceptions.hh
A few common exception classes.
stdstreams.hh
Standard Dune debug streams.
Dune::ArrayList::begin
iterator begin()
Get an iterator that is positioned at the first element.
Definition arraylist.hh:515
Dune::ArrayList::size
size_type size() const
Get the number of elements in the list.
Definition arraylist.hh:470
Dune::ArrayList::end
iterator end()
Get a random access iterator positioned after the last element.
Definition arraylist.hh:527
Dune::ArrayList::clear
void clear()
Delete all entries from the list.
Definition arraylist.hh:462
DUNE_THROW
#define DUNE_THROW(E,...)
Definition exceptions.hh:314
Dune::type
MPI_Datatype MPITraits< ParallelLocalIndex< T > >::type
Definition plocalindex.hh:317
Dune::dvverb
DVVerbType dvverb(std::cout)
stream for very verbose output.
Definition stdstreams.hh:96
Dune
Dune namespace
Definition alignedallocator.hh:13
Dune::size
constexpr std::integral_constant< std::size_t, sizeof...(II)> size(std::integer_sequence< T, II... >)
Return the size of the sequence.
Definition integersequence.hh:75
Dune::iterator
Dune::MPITraits
A traits class describing the mapping of types onto MPI_Datatypes.
Definition mpitraits.hh:41
Dune::IOError
Default exception class for I/O errors.
Definition exceptions.hh:325
Dune::SizeOne
Flag for marking indexed data structures where data at each index is of the same size.
Definition communicator.hh:110
Dune::VariableSize
Flag for marking indexed data structures where the data at each index may be a variable multiple of a...
Definition communicator.hh:118
Dune::CommPolicy
Default policy used for communicating an indexed type.
Definition communicator.hh:128
Dune::CommPolicy::IndexedType
V::value_type IndexedType
The type we get at each index with operator[].
Definition communicator.hh:147
Dune::CommPolicy::getSize
static int getSize(const V &, int index)
Get the number of primitive elements at that index.
Dune::CommPolicy::IndexedTypeFlag
SizeOne IndexedTypeFlag
Whether the indexed type has variable size or there is always one value at each index.
Definition communicator.hh:153
Dune::CommPolicy::getAddress
static const void * getAddress(const V &v, int index)
Get the address of entry at an index.
Dune::CommPolicy::Type
V Type
The type the policy is for.
Definition communicator.hh:140
Dune::CommPolicy::FieldVector
Definition communicator.hh:173
Dune::CommPolicy::VariableBlockVector
Definition communicator.hh:175
Dune::CommPolicy::CommPolicy< VariableBlockVector< FieldVector< K, n >, A > >::getSize
static int getSize(const Type &v, int i)
Dune::CommPolicy::CommPolicy< VariableBlockVector< FieldVector< K, n >, A > >::IndexedTypeFlag
VariableSize IndexedTypeFlag
Definition communicator.hh:184
Dune::CommPolicy::CommPolicy< VariableBlockVector< FieldVector< K, n >, A > >::Type
VariableBlockVector< FieldVector< K, n >, A > Type
Definition communicator.hh:180
Dune::CommPolicy::CommPolicy< VariableBlockVector< FieldVector< K, n >, A > >::getAddress
static const void * getAddress(const Type &v, int i)
Dune::CommPolicy::CommPolicy< VariableBlockVector< FieldVector< K, n >, A > >::IndexedType
Type::B IndexedType
Definition communicator.hh:182
Dune::CommPolicy::CommunicationError
Error thrown if there was a problem with the communication.
Definition communicator.hh:195
Dune::CommPolicy::CopyGatherScatter
GatherScatter default implementation that just copies data.
Definition communicator.hh:202
Dune::CommPolicy::CopyGatherScatter::scatter
static void scatter(T &vec, const IndexedType &v, std::size_t i)
Dune::CommPolicy::CopyGatherScatter::IndexedType
CommPolicy< T >::IndexedType IndexedType
Definition communicator.hh:203
Dune::CommPolicy::CopyGatherScatter::gather
static const IndexedType & gather(const T &vec, std::size_t i)
Dune::BufferedCommunicator::backward
void backward(Data &data)
Backward send where target and source are the same.
Dune::BufferedCommunicator::BufferedCommunicator
BufferedCommunicator()
Constructor.
Dune::BufferedCommunicator::~BufferedCommunicator
~BufferedCommunicator()
Destructor.
Dune::BufferedCommunicator::forward
void forward(const Data &source, Data &dest)
Send from source to target.
Dune::BufferedCommunicator::free
void free()
Free the allocated memory (i.e. buffers and message information.
Dune::BufferedCommunicator::build
std::enable_if< std::is_same< SizeOne, typenameCommPolicy< Data >::IndexedTypeFlag >::value, void >::type build(const Interface &interface)
Build the buffers and information for the communication process.
Dune::BufferedCommunicator::backward
void backward(Data &source, const Data &dest)
Communicate in the reverse direction, i.e. send from target to source.
Dune::BufferedCommunicator::build
void build(const Data &source, const Data &target, const Interface &interface)
Build the buffers and information for the communication process.
Dune::BufferedCommunicator::forward
void forward(Data &data)
Forward send where target and source are the same.
Dune::ParallelIndexSet
Manager class for the mapping between local indices and globally unique indices.
Definition indexset.hh:218
Dune::InterfaceBuilder
Base class of all classes representing a communication interface.
Definition parallel/interface.hh:44
Dune::InterfaceInformation
Information describing an interface.
Definition parallel/interface.hh:110
Dune::Interface
Communication interface between remote and local indices.
Definition parallel/interface.hh:218
Dune::LocalIndex
An index present on the local process.
Definition localindex.hh:35
Dune::RemoteIndices
The indices present on remote processes.
Definition remoteindices.hh:190
Dune::RemoteIndices::GlobalIndex
ParallelIndexSet::GlobalIndex GlobalIndex
The type of the global index.
Definition remoteindices.hh:216
Dune::RemoteIndices::Attribute
LocalIndex::Attribute Attribute
The type of the attribute.
Definition remoteindices.hh:227
Dune::RemoteIndices::LocalIndex
ParallelIndexSet::LocalIndex LocalIndex
The type of the local index.
Definition remoteindices.hh:222
Dune::type
interface.hh
Provides classes for building the communication interface between remote indices.