Loading...
Searching...
No Matches
p2pcommunicator_impl.hh
/*
Copyright 2015 IRIS AS
This file is part of the Open Porous Media project (OPM).
OPM 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.
OPM 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 OPM. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef DUNE_POINT2POINTCOMMUNICATOR_IMPL_HEADER_INCLUDED
#define DUNE_POINT2POINTCOMMUNICATOR_IMPL_HEADER_INCLUDED
#include <iostream>
namespace Dune
{
template <class MsgBuffer>
inline void
Point2PointCommunicator< MsgBuffer >::
removeLinkage()
{
sendLinkage_.clear();
recvLinkage_.clear();
sendDest_.clear();
recvSource_.clear();
// clear previously stored buffer sizes
_recvBufferSizes.clear();
_recvBufferSizesComputed = false ;
}
template <class MsgBuffer>
inline void
Point2PointCommunicator< MsgBuffer >::
computeDestinations( const linkage_t& linkage, vector_t& dest )
{
typedef linkage_t::const_iterator const_iterator ;
dest.resize( linkage.size() );
const const_iterator linkageEnd = linkage.end ();
for( const_iterator i = linkage.begin (); i != linkageEnd; ++i )
{
dest[ (*i).second ] = (*i).first;
}
}
template <class MsgBuffer>
inline void
Point2PointCommunicator< MsgBuffer >::
insertRequest( const std::set< int >& sendLinks, const std::set< int >& recvLinks )
{
// remove old linkage
removeLinkage();
const int me_rank = rank ();
{
typedef std::map< int, int >::iterator iterator ;
typedef std::set< int >::const_iterator const_iterator;
const iterator sendEnd = sendLinkage_.end ();
const iterator recvEnd = recvLinkage_.end ();
const const_iterator sendLinksEnd = sendLinks.end ();
int sendLink = 0 ;
int recvLink = 0 ;
for (const_iterator i = sendLinks.begin (); i != sendLinksEnd; ++i )
{
const int rank = (*i);
// if rank was not inserted, insert with current link number
if( rank != me_rank && (sendLinkage_.find ( rank ) == sendEnd ) )
{
sendLinkage_.insert( std::make_pair( rank, sendLink++) );
}
}
const const_iterator recvLinksEnd = recvLinks.end ();
for (const_iterator i = recvLinks.begin (); i != recvLinksEnd; ++i )
{
const int rank = (*i);
// if rank was not inserted, insert with current link number
if( rank != me_rank && (recvLinkage_.find ( rank ) == recvEnd ) )
{
recvLinkage_.insert( std::make_pair( rank, recvLink++) );
}
}
}
// compute send destinations
computeDestinations( sendLinkage_, sendDest_ );
// compute send destinations
computeDestinations( recvLinkage_, recvSource_ );
}
// non-blocking communication object
// this class is defined here since it contains MPI information
#if HAVE_MPI
#ifndef NDEBUG
// this is simply to avoid warning of unused variables
#define MY_INT_TEST int test =
#else
#define MY_INT_TEST
#endif
template < class P2PCommunicator >
class NonBlockingExchangeImplementation
{
typedef P2PCommunicator P2PCommunicatorType ;
const P2PCommunicatorType& _p2pCommunicator;
const int _sendLinks;
const int _recvLinks;
const int _tag;
MPI_Request* _sendRequest;
MPI_Request* _recvRequest;
const bool _recvBufferSizesKnown;
bool _needToSend ;
// no copying
NonBlockingExchangeImplementation( const NonBlockingExchangeImplementation& );
// return vector of send requests for number of send links is positive
MPI_Request* createSendRequest() const
{
return ( _sendLinks > 0 ) ? new MPI_Request [ _sendLinks ] : 0;
}
// return vector of recv requests when
// number of recv links is positive and symmetric is true
MPI_Request* createRecvRequest( const bool recvBufferSizesKnown ) const
{
return ( _recvLinks > 0 && recvBufferSizesKnown ) ? new MPI_Request [ _recvLinks ] : 0;
}
// call cast operator on CollectiveCommunication to retreive MPI_Comm
MPI_Comm mpiCommunicator() const { return static_cast< MPI_Comm > (_p2pCommunicator); }
public:
typedef typename P2PCommunicatorType :: DataHandleInterface DataHandleInterface;
typedef typename P2PCommunicatorType :: MessageBufferType MessageBufferType;
NonBlockingExchangeImplementation( const P2PCommunicatorType& p2pComm,
const int tag,
const bool recvBufferSizesKnown = false )
: _p2pCommunicator( p2pComm ),
_sendLinks( _p2pCommunicator.sendLinks() ),
_recvLinks( _p2pCommunicator.recvLinks() ),
_tag( tag ),
_sendRequest( createSendRequest() ),
_recvRequest( createRecvRequest( recvBufferSizesKnown ) ),
_recvBufferSizesKnown( recvBufferSizesKnown ),
_needToSend( true )
{
// make sure every process has the same tag
#ifndef NDEBUG
int mytag = tag ;
assert ( mytag == _p2pCommunicator.max( mytag ) );
#endif
}
NonBlockingExchangeImplementation( const P2PCommunicatorType& p2pComm,
const int tag,
const std::vector< MessageBufferType > & sendBuffers )
: _p2pCommunicator( p2pComm ),
_sendLinks( _p2pCommunicator.sendLinks() ),
_recvLinks( _p2pCommunicator.recvLinks() ),
_tag( tag ),
_sendRequest( createSendRequest() ),
_recvRequest( createRecvRequest( false ) ),
_recvBufferSizesKnown( false ),
_needToSend( false )
{
// make sure every process has the same tag
#ifndef NDEBUG
int mytag = tag ;
assert ( mytag == _p2pCommunicator.max( mytag ) );
#endif
assert ( _sendLinks == int( sendBuffers.size() ) );
sendImpl( sendBuffers );
}
// interface methods
~NonBlockingExchangeImplementation()
{
if( _sendRequest )
{
delete [] _sendRequest;
_sendRequest = 0;
}
if( _recvRequest )
{
delete [] _recvRequest;
_recvRequest = 0;
}
}
// virtual methods
void send( const std::vector< MessageBufferType > & sendBuffers ) { sendImpl( sendBuffers ); }
std::vector< MessageBufferType > receive() { return receiveImpl(); }
// implementation
// send data implementation
void sendImpl( const std::vector< MessageBufferType > & sendBuffers )
{
// get mpi communicator
MPI_Comm comm = mpiCommunicator();
// get vector with destinations
const std::vector< int >& sendDest = _p2pCommunicator.sendDest();
// send data
for (int link = 0; link < _sendLinks; ++link)
{
sendLink( sendDest[ link ], _tag, sendBuffers[ link ], _sendRequest[ link ], comm );
}
// set send info
_needToSend = false ;
}
// receive data without buffer given
std::vector< MessageBufferType > receiveImpl ()
{
// create vector of empty streams
std::vector< MessageBufferType > recvBuffer( _recvLinks );
receiveImpl( recvBuffer );
return recvBuffer;
}
// receive data implementation with given buffers
void receiveImpl ( std::vector< MessageBufferType >& recvBuffers, DataHandleInterface* dataHandle = 0)
{
// do nothing if number of links is zero
if( (_recvLinks + _sendLinks) == 0 ) return;
// get mpi communicator
MPI_Comm comm = mpiCommunicator();
// get vector with destinations
const std::vector< int >& recvSource = _p2pCommunicator.recvSource();
// check whether out vector has more than one stream
const bool useFirstStreamOnly = (recvBuffers.size() == 1) ;
// flag vector holding information about received links
std::vector< bool > linkNotReceived( _recvLinks, true );
// count noumber of received messages
int numReceived = 0;
while( numReceived < _recvLinks )
{
// check for all links messages
for (int link = 0; link < _recvLinks; ++link )
{
// if message was not received yet, check again
if( linkNotReceived[ link ] )
{
// get appropriate object stream
MessageBufferType& recvBuffer = useFirstStreamOnly ? recvBuffers[ 0 ] : recvBuffers[ link ];
// check whether a message was completely received
// if message was received the unpack data
if( probeAndReceive( comm, recvSource[ link ], _tag, recvBuffer ) )
{
// if data handle was given do unpack
if( dataHandle ) dataHandle->unpack( link, recvBuffer );
// mark link as received
linkNotReceived[ link ] = false ;
// increase number of received messages
++ numReceived;
}
}
}
}
// if send request exists, i.e. some messages have been sent
if( _sendRequest )
{
// wait until all processes are done with receiving
MY_INT_TEST MPI_Waitall ( _sendLinks, _sendRequest, MPI_STATUSES_IGNORE);
assert (test == MPI_SUCCESS);
}
}
// receive data implementation with given buffers
void unpackRecvBufferSizeKnown( std::vector< MessageBufferType >& recvBuffers, DataHandleInterface& dataHandle )
{
// do nothing if number of links is zero
if( _recvLinks == 0 ) return;
// flag vector holding information about received links
std::vector< bool > linkNotReceived( _recvLinks, true );
// count noumber of received messages
int numReceived = 0;
while( numReceived < _recvLinks )
{
// check for all links messages
for (int link = 0; link < _recvLinks; ++link )
{
// if message was not received yet, check again
if( linkNotReceived[ link ] )
{
assert( _recvRequest );
// check whether message was received, and if unpack data
if( receivedMessage( _recvRequest[ link ], recvBuffers[ link ] ) )
{
// if data handle was given do unpack
dataHandle.unpack( link, recvBuffers[ link ] );
// mark link as received
linkNotReceived[ link ] = false ;
// increase number of received messages
++ numReceived;
}
}
}
}
// if send request exists, i.e. some messages have been sent
if( _sendRequest )
{
// wait until all processes are done with receiving
MY_INT_TEST MPI_Waitall ( _sendLinks, _sendRequest, MPI_STATUSES_IGNORE);
assert (test == MPI_SUCCESS);
}
}
// receive data implementation with given buffers
void send( std::vector< MessageBufferType >& sendBuffers,
DataHandleInterface& dataHandle )
{
std::vector< MessageBufferType > recvBuffers;
send( sendBuffers, recvBuffers, dataHandle );
}
// receive data implementation with given buffers
void send( std::vector< MessageBufferType >& sendBuffer,
std::vector< MessageBufferType >& recvBuffer,
DataHandleInterface& dataHandle )
{
if( _needToSend )
{
// get mpi communicator
MPI_Comm comm = mpiCommunicator();
// get vector with destinations
const std::vector< int >& sendDest = _p2pCommunicator.sendDest();
// send data
for (int link = 0; link < _sendLinks; ++link)
{
// pack data
dataHandle.pack( link, sendBuffer[ link ] );
// send data
sendLink( sendDest[ link ], _tag, sendBuffer[ link ], _sendRequest[ link ], comm );
}
// set send info
_needToSend = false ;
}
// resize receive buffer if in symmetric mode
if( _recvBufferSizesKnown )
{
// get mpi communicator
MPI_Comm comm = mpiCommunicator();
recvBuffer.resize( _recvLinks );
// get vector with destinations
const std::vector< int >& recvSource = _p2pCommunicator.recvSource();
const std::vector< int >& recvBufferSizes = _p2pCommunicator.recvBufferSizes();
// send data
for (int link = 0; link < _recvLinks; ++link)
{
// send data
const int bufferSize = recvBufferSizes[ link ];
// post receive if in symmetric mode
assert( _recvRequest );
assert( &_recvRequest[ link ] );
postReceive( recvSource[ link ], _tag, bufferSize, recvBuffer[ link ], _recvRequest[ link ], comm );
}
}
}
// receive data implementation with given buffers
void receive( DataHandleInterface& dataHandle )
{
// do work that can be done between send and receive
dataHandle.localComputation() ;
// create receive message buffers
std::vector< MessageBufferType > recvBuffer( 1 );
// receive data
receiveImpl( recvBuffer, &dataHandle );
}
// receive data implementation with given buffers
void exchange( DataHandleInterface& dataHandle )
{
const int recvLinks = _p2pCommunicator.recvLinks();
// do nothing if number of links is zero
if( (recvLinks + _sendLinks) == 0 ) return;
// send message buffers, we need several because of the
// non-blocking send routines, send might not be finished
// when we start recieving
std::vector< MessageBufferType > recvBuffers ;
// note: due to non-blocking we cannot limit scope
std::vector< MessageBufferType > sendBuffers;
// if data was noy send yet, do it now
if( _needToSend )
{
// resize message buffer vector
sendBuffers.resize(_sendLinks);
// send data
send( sendBuffers, recvBuffers, dataHandle );
}
// now receive data
if( _recvBufferSizesKnown )
unpackRecvBufferSizeKnown( recvBuffers, dataHandle );
else
receive( dataHandle );
}
protected:
int sendLink( const int dest, const int tag,
const MessageBufferType& msgBuffer, MPI_Request& request, MPI_Comm& comm )
{
// buffer = point to mem and size
std::pair< char*, int > buffer = msgBuffer.buffer();
MY_INT_TEST MPI_Isend ( buffer.first, buffer.second, MPI_BYTE, dest, tag, comm, &request );
assert (test == MPI_SUCCESS);
return buffer.second;
}
void postReceive( const int source, const int tag, const int bufferSize,
MessageBufferType& msgBuffer, MPI_Request& request, MPI_Comm& comm )
{
// reserve memory for receive buffer
msgBuffer.resize( bufferSize );
// reset read position
msgBuffer.resetReadPosition();
// get buffer and size
std::pair< char*, int > buffer = msgBuffer.buffer();
// MPI receive (non-blocking)
{
MY_INT_TEST MPI_Irecv ( buffer.first, buffer.second, MPI_BYTE, source, tag, comm, & request);
assert (test == MPI_SUCCESS);
}
}
// does receive operation for one link
bool receivedMessage( MPI_Request& request, MessageBufferType&
#ifndef NDEBUG
buffer
#endif
)
{
#ifndef NDEBUG
// for checking whether the buffer size is correct
MPI_Status status ;
#endif
// msg received, 0 or 1
int received = 0;
// if receive of message is finished, unpack
MPI_Test( &request, &received,
#ifndef NDEBUG
&status
#else
MPI_STATUS_IGNORE // ignore status in non-debug mode for performance reasons
#endif
);
#ifndef NDEBUG
if( received )
{
int checkBufferSize = -1;
MPI_Get_count ( & status, MPI_BYTE, &checkBufferSize );
if( checkBufferSize != int(buffer.size()) )
std::cout << "Buffer sizes don't match: " << checkBufferSize << " " << buffer.size() << std::endl;
assert( checkBufferSize == int(buffer.size()) );
}
#endif
return bool(received);
}
// does receive operation for one link
bool probeAndReceive( MPI_Comm& comm,
const int source,
const int tag,
MessageBufferType& recvBuffer )
{
// corresponding MPI status
MPI_Status status;
// msg available, 0 or 1
// available does not mean already received
int available = 0;
// check for any message with tag (nonblocking)
MPI_Iprobe( source, tag, comm, &available, &status );
// receive message if available flag is true
if( available )
{
// this should be the same, otherwise we got an error
assert ( source == status.MPI_SOURCE );
// length of message
int bufferSize = -1;
// get length of message
{
MY_INT_TEST MPI_Get_count ( &status, MPI_BYTE, &bufferSize );
assert (test == MPI_SUCCESS);
}
// reserve memory
recvBuffer.resize( bufferSize );
// reset read position for unpack
recvBuffer.resetReadPosition();
// get buffer
std::pair< char*, int > buffer = recvBuffer.buffer();
// MPI receive (blocking)
{
MY_INT_TEST MPI_Recv ( buffer.first, buffer.second, MPI_BYTE, status.MPI_SOURCE, tag, comm, & status);
assert (test == MPI_SUCCESS);
}
return true ; // received
}
return false ; // not yet received
}
}; // end NonBlockingExchangeImplementation
#undef MY_INT_TEST
#endif // #if HAVE_MPI
// --exchange
template <class MsgBuffer>
inline void
Point2PointCommunicator< MsgBuffer >::
#if HAVE_MPI
exchange( DataHandleInterface& handle) const
#else
exchange( DataHandleInterface&) const
#endif
{
assert( _recvBufferSizes.empty () );
#if HAVE_MPI
NonBlockingExchangeImplementation< ThisType > nonBlockingExchange( *this, getMessageTag() );
nonBlockingExchange.exchange( handle );
#endif
}
// --exchange
template <class MsgBuffer>
inline std::vector< MsgBuffer >
Point2PointCommunicator< MsgBuffer >::
exchange( const std::vector< MessageBufferType > & in ) const
{
#if HAVE_MPI
// note: for the non-blocking exchange the message tag
// should be different each time to avoid MPI problems
NonBlockingExchangeImplementation< ThisType > nonBlockingExchange( *this, getMessageTag(), in );
return nonBlockingExchange.receiveImpl();
#else
// don nothing when MPI is not found
return in;
#endif
}
// --exchange
template <class MsgBuffer>
inline void
Point2PointCommunicator< MsgBuffer >::
#if HAVE_MPI
exchangeCached( DataHandleInterface& handle ) const
#else
exchangeCached( DataHandleInterface&) const
#endif
{
#if HAVE_MPI
if( ! _recvBufferSizesComputed )
{
const int nSendLinks = sendLinks();
std::vector< MsgBuffer > buffers( nSendLinks );
// pack all data
for( int link=0; link<nSendLinks; ++link )
{
handle.pack( link, buffers[ link ] );
}
// exchange data
buffers = exchange( buffers );
const int nRecvLinks = recvLinks();
// unpack all data
for( int link=0; link<nRecvLinks; ++link )
{
handle.unpack( link, buffers[ link ] );
}
// store receive buffer sizes
_recvBufferSizes.resize( nRecvLinks );
for( int link=0; link<nRecvLinks; ++link )
{
_recvBufferSizes[ link ] = buffers[ link ].size();
}
_recvBufferSizesComputed = true ;
}
else
{
NonBlockingExchangeImplementation< ThisType > nonBlockingExchange( *this, getMessageTag(), _recvBufferSizesComputed );
nonBlockingExchange.exchange( handle );
}
#endif
}
} // namespace Dune
#endif // #ifndef DUNE_POINT2POINTCOMMUNICATOR_IMPL_HEADER_INCLUDED
Dune::Point2PointCommunicator::DataHandleInterface
Definition p2pcommunicator.hh:166
Dune::Point2PointCommunicator
Point-2-Point communicator for exchange messages between processes.
Definition p2pcommunicator.hh:132
Dune::Point2PointCommunicator::recvLinks
int recvLinks() const
return number of processes we will receive data from
Definition p2pcommunicator.hh:193
Dune::Point2PointCommunicator::recvLink
int recvLink(const int rank) const
return recv link number for a given recv rank number
Definition p2pcommunicator.hh:206
Dune::Point2PointCommunicator::sendLink
int sendLink(const int rank) const
return send link number for a given send rank number
Definition p2pcommunicator.hh:199
Dune::Point2PointCommunicator::exchangeCached
virtual void exchangeCached(DataHandleInterface &) const
exchange data with peers, handle defines pack and unpack of data, if receive buffers are known from p...
Definition p2pcommunicator_impl.hh:620
Dune::Point2PointCommunicator::sendLinks
int sendLinks() const
return number of processes we will send data to
Definition p2pcommunicator.hh:190
Dune::Point2PointCommunicator::exchange
virtual std::vector< MessageBufferType > exchange(const std::vector< MessageBufferType > &) const
exchange message buffers with peers defined by inserted linkage
Definition p2pcommunicator_impl.hh:600
Dune::Point2PointCommunicator::Point2PointCommunicator
Point2PointCommunicator(const MPICommunicator &mpiComm=MPIHelper::getCommunicator())
constructor taking mpi communicator
Definition p2pcommunicator.hh:179
Dune::Point2PointCommunicator::insertRequest
void insertRequest(const std::set< int > &sendLinks, const std::set< int > &recvLinks)
insert communication request with a set os ranks to send to and a set of ranks to receive from
Definition p2pcommunicator_impl.hh:59
Dune::Point2PointCommunicator::removeLinkage
void removeLinkage()
remove stored linkage
Definition p2pcommunicator_impl.hh:30
Dune
The namespace Dune is the main namespace for all Dune code.
Definition CartesianIndexMapper.hpp:10