Loading...
Searching...
No Matches
VariableSizeCommunicator.hpp
// -*- tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*-
// vi: set et ts=4 sw=2 sts=2:
// This file is copied from file dune/common/variablesizecommunicator.hh
// of dune-common as of commit 30e78443884f7730229e6c59b14406051d71c032.
// It is needed as we need an important bugfix (commit 30e78443884)
// that is not included in DUNE
// versions below 2.8. We have moved the classes to the Opm namespace.
// The file's main author is Markus Blatt. Others have only contributed
// minor fixes proposed by compilers, and codespell to prevent unused warnings
// and variables or improved documentation.
/*
Copyright 2013-2018 Dr. Blatt - HPC-Simulation-Software & Services
Copyright 2013-2017 Statoil ASA
*/
#ifndef OPM_COMMON_PARALLEL_VARIABLESIZECOMMUNICATOR_HH // Still fits the line!
#define OPM_COMMON_PARALLEL_VARIABLESIZECOMMUNICATOR_HH
#if HAVE_MPI
#include <algorithm>
#include <cassert>
#include <cstddef>
#include <functional>
#include <map>
#include <memory>
#include <utility>
#include <vector>
#include <algorithm>
#include <mpi.h>
#include <dune/common/parallel/interface.hh>
#include <dune/common/parallel/mpitraits.hh>
#include <dune/common/unused.hh>
namespace Opm
{
namespace
{
template<class T, class Allocator=std::allocator<T> >
class MessageBuffer
{
public:
explicit MessageBuffer(int size)
: buffer_(new T[size]), size_(size), position_(0)
{}
explicit MessageBuffer(const MessageBuffer& o)
: buffer_(new T[o.size_]), size_(o.size_), position_(o.position_)
{
}
~MessageBuffer()
{
delete[] buffer_;
}
void write(const T& data)
{
assert(position_<size_);
buffer_[position_++]=data;
}
void read(T& data)
{
assert(position_<size_);
data=buffer_[position_++];
}
void reset()
{
position_=0;
}
bool finished()
{
return position_==size_;
}
bool hasSpaceForItems(int noItems)
{
return position_+noItems<=size_;
}
std::size_t size() const
{
return size_;
}
std::size_t position() const
{
return position_;
}
operator T*()
{
return buffer_;
}
private:
T* buffer_;
std::size_t size_;
std::size_t position_;
};
using Dune::InterfaceInformation;
using Dune::Interface;
class InterfaceTracker
{
public:
InterfaceTracker(int rank, InterfaceInformation info, std::size_t fixedsize=0,
bool allocateSizes=false)
: fixedSize(fixedsize),rank_(rank), index_(), interface_(info), sizes_()
{
if(allocateSizes)
{
sizes_.resize(info.size());
}
}
void moveToNextIndex()
{
index_++;
assert(index_<=interface_.size());
skipZeroIndices();
}
void increment(std::size_t i)
{
index_+=i;
assert(index_<=interface_.size());
}
bool finished() const
{
return index_==interface_.size();
}
void skipZeroIndices()
{
// skip indices with size zero!
while(sizes_.size() && index_!=interface_.size() &&!size())
++index_;
}
std::size_t index() const
{
return interface_[index_];
}
std::size_t size() const
{
assert(sizes_.size());
return sizes_[index_];
}
std::size_t* getSizesPointer()
{
return &sizes_[0];
}
bool empty() const
{
return !interface_.size();
}
std::size_t indicesLeft() const
{
return interface_.size()-index_;
}
std::size_t fixedSize;
int rank() const
{
return rank_;
}
std::size_t offset() const
{
return index_;
}
private:
int rank_;
std::size_t index_;
InterfaceInformation interface_;
std::vector<std::size_t> sizes_;
};
} // end unnamed namespace
template<class Allocator=std::allocator<std::pair<InterfaceInformation,InterfaceInformation> > >
class VariableSizeCommunicator
{
public:
typedef std::map<int,std::pair<InterfaceInformation,InterfaceInformation>,
std::less<int>,
typename Allocator::template rebind<std::pair<const int,std::pair<InterfaceInformation,InterfaceInformation> > >::other> InterfaceMap;
#ifndef DUNE_PARALLEL_MAX_COMMUNICATION_BUFFER_SIZE
VariableSizeCommunicator(MPI_Comm comm, const InterfaceMap& inf)
: maxBufferSize_(32768), interface_(&inf)
{
MPI_Comm_dup(comm, &communicator_);
}
VariableSizeCommunicator(const Interface& inf)
: maxBufferSize_(32768), interface_(&inf.interfaces())
{
MPI_Comm_dup(inf.communicator(), &communicator_);
}
#else
VariableSizeCommunicator(MPI_Comm comm, InterfaceMap& inf)
: maxBufferSize_(DUNE_PARALLEL_MAX_COMMUNICATION_BUFFER_SIZE),
interface_(&inf)
{
MPI_Comm_dup(comm, &communicator_);
}
VariableSizeCommunicator(const Interface& inf)
: maxBufferSize_(DUNE_PARALLEL_MAX_COMMUNICATION_BUFFER_SIZE),
interface_(&inf.interfaces())
{
MPI_Comm_dup(inf.communicator(), &communicator_);
}
#endif
VariableSizeCommunicator(MPI_Comm comm, const InterfaceMap& inf, std::size_t max_buffer_size)
: maxBufferSize_(max_buffer_size), interface_(&inf)
{
MPI_Comm_dup(comm, &communicator_);
}
VariableSizeCommunicator(const Interface& inf, std::size_t max_buffer_size)
: maxBufferSize_(max_buffer_size), interface_(&inf.interfaces())
{
MPI_Comm_dup(inf.communicator(), &communicator_);
}
~VariableSizeCommunicator()
{
MPI_Comm_free(&communicator_);
}
template<class DataHandle>
void forward(DataHandle& handle)
{
communicate<true>(handle);
}
template<class DataHandle>
void backward(DataHandle& handle)
{
communicate<false>(handle);
}
private:
template<bool FORWARD, class DataHandle>
void communicateSizes(DataHandle& handle,
std::vector<InterfaceTracker>& recv_trackers);
template<bool forward,class DataHandle>
void communicate(DataHandle& handle);
template<bool FORWARD, class DataHandle>
void setupInterfaceTrackers(DataHandle& handle,
std::vector<InterfaceTracker>& send_trackers,
std::vector<InterfaceTracker>& recv_trackers);
template<bool FORWARD, class DataHandle>
[[maybe_unused]] void communicateFixedSize(DataHandle& handle);
template<bool FORWARD, class DataHandle>
void communicateVariableSize(DataHandle& handle);
std::size_t maxBufferSize_;
const InterfaceMap* interface_;
MPI_Comm communicator_;
};
namespace
{
template<class DataHandle>
class SizeDataHandle
{
public:
typedef std::size_t DataType;
SizeDataHandle(DataHandle& data,
std::vector<InterfaceTracker>& trackers)
: data_(data), trackers_(trackers), index_()
{}
bool fixedsize()
{
return true;
}
std::size_t size(std::size_t i)
{
DUNE_UNUSED_PARAMETER(i);
return 1;
}
template<class B>
void gather(B& buf, int i)
{
buf.write(data_.size(i));
}
void setReceivingIndex(std::size_t i)
{
index_=i;
}
std::size_t* getSizesPointer()
{
return trackers_[index_].getSizesPointer();
}
private:
DataHandle& data_;
std::vector<InterfaceTracker>& trackers_;
int index_;
};
template<class T>
void setReceivingIndex(T&, int)
{}
template<class T>
void setReceivingIndex(SizeDataHandle<T>& t, int i)
{
t.setReceivingIndex(i);
}
template<bool FORWARD>
struct InterfaceInformationChooser
{
static const InterfaceInformation&
getSend(const std::pair<InterfaceInformation,InterfaceInformation>& info)
{
return info.first;
}
static const InterfaceInformation&
getReceive(const std::pair<InterfaceInformation,InterfaceInformation>& info)
{
return info.second;
}
};
template<>
struct InterfaceInformationChooser<false>
{
static const InterfaceInformation&
getSend(const std::pair<InterfaceInformation,InterfaceInformation>& info)
{
return info.second;
}
static const InterfaceInformation&
getReceive(const std::pair<InterfaceInformation,InterfaceInformation>& info)
{
return info.first;
}
};
template<class DataHandle>
struct PackEntries
{
int operator()(DataHandle& handle, InterfaceTracker& tracker,
MessageBuffer<typename DataHandle::DataType>& buffer,
int i) const
{
DUNE_UNUSED_PARAMETER(i);
return operator()(handle,tracker,buffer);
}
int operator()(DataHandle& handle, InterfaceTracker& tracker,
MessageBuffer<typename DataHandle::DataType>& buffer) const
{
if(tracker.fixedSize) // fixed size if variable is >0!
{
std::size_t noIndices=std::min(buffer.size()/tracker.fixedSize, tracker.indicesLeft());
for(std::size_t i=0; i< noIndices; ++i)
{
handle.gather(buffer, tracker.index());
tracker.moveToNextIndex();
}
return noIndices*tracker.fixedSize;
}
else
{
int packed=0;
tracker.skipZeroIndices();
while(!tracker.finished())
if(buffer.hasSpaceForItems(handle.size(tracker.index())))
{
assert(std::size_t(packed) == buffer.position());
handle.gather(buffer, tracker.index());
packed+=handle.size(tracker.index());
assert(std::size_t(packed) <= buffer.size());
assert(std::size_t(packed) == buffer.position());
tracker.moveToNextIndex();
}
else
break;
return packed;
}
}
};
template<class DataHandle>
struct UnpackEntries{
bool operator()(DataHandle& handle, InterfaceTracker& tracker,
MessageBuffer<typename DataHandle::DataType>& buffer,
int count=0)
{
if(tracker.fixedSize) // fixed size if variable is >0!
{
std::size_t noIndices=std::min(buffer.size()/tracker.fixedSize, tracker.indicesLeft());
for(std::size_t i=0; i< noIndices; ++i)
{
handle.scatter(buffer, tracker.index(), tracker.fixedSize);
tracker.moveToNextIndex();
}
return tracker.finished();
}
else
{
assert(count);
for(int unpacked=0;unpacked<count;)
{
assert(!tracker.finished());
assert(buffer.hasSpaceForItems(tracker.size()));
handle.scatter(buffer, tracker.index(), tracker.size());
unpacked+=tracker.size();
tracker.moveToNextIndex();
}
return tracker.finished();
}
}
};
template<class DataHandle>
struct UnpackSizeEntries{
bool operator()(SizeDataHandle<DataHandle>& handle, InterfaceTracker& tracker,
MessageBuffer<typename SizeDataHandle<DataHandle>::DataType>& buffer) const
{
std::size_t noIndices=std::min(buffer.size(), tracker.indicesLeft());
std::copy(static_cast<std::size_t*>(buffer), static_cast<std::size_t*>(buffer)+noIndices,
handle.getSizesPointer()+tracker.offset());
tracker.increment(noIndices);
return noIndices;
}
bool operator()(SizeDataHandle<DataHandle>& handle, InterfaceTracker& tracker,
MessageBuffer<typename SizeDataHandle<DataHandle>::DataType>& buffer, int) const
{
return operator()(handle,tracker,buffer);
}
};
[[maybe_unused]] void sendFixedSize(std::vector<InterfaceTracker>& send_trackers,
std::vector<MPI_Request>& send_requests,
std::vector<InterfaceTracker>& recv_trackers,
std::vector<MPI_Request>& recv_requests,
MPI_Comm communicator)
{
typedef std::vector<InterfaceTracker>::iterator TIter;
std::vector<MPI_Request>::iterator mIter=recv_requests.begin();
for(TIter iter=recv_trackers.begin(), end=recv_trackers.end(); iter!=end;
++iter, ++mIter)
{
MPI_Irecv(&(iter->fixedSize), 1, Dune::MPITraits<std::size_t>::getType(),
iter->rank(), 933881, communicator, &(*mIter));
}
// Send our size to all neighbours using non-blocking synchronous communication.
std::vector<MPI_Request>::iterator mIter1=send_requests.begin();
for(TIter iter=send_trackers.begin(), end=send_trackers.end();
iter!=end;
++iter, ++mIter1)
{
MPI_Issend(&(iter->fixedSize), 1, Dune::MPITraits<std::size_t>::getType(),
iter->rank(), 933881, communicator, &(*mIter1));
}
}
template<class DataHandle>
struct SetupSendRequest{
void operator()(DataHandle& handle,
InterfaceTracker& tracker,
MessageBuffer<typename DataHandle::DataType>& buffer,
MPI_Request& request,
MPI_Comm comm) const
{
buffer.reset();
int size=PackEntries<DataHandle>()(handle, tracker, buffer);
// Skip indices of zero size.
while(!tracker.finished() && !handle.size(tracker.index()))
tracker.moveToNextIndex();
if(size)
{
assert(std::size_t(size) <= buffer.size());
MPI_Issend(buffer, size, Dune::MPITraits<typename DataHandle::DataType>::getType(),
tracker.rank(), 933399, comm, &request);
}
}
};
template<class DataHandle>
struct SetupRecvRequest{
void operator()(DataHandle& /*handle*/,
InterfaceTracker& tracker,
MessageBuffer<typename DataHandle::DataType>& buffer,
MPI_Request& request,
MPI_Comm comm) const
{
buffer.reset();
if(tracker.indicesLeft())
MPI_Irecv(buffer, buffer.size(), Dune::MPITraits<typename DataHandle::DataType>::getType(),
tracker.rank(), 933399, comm, &request);
}
};
template<class DataHandle>
struct NullPackUnpackFunctor
{
int operator()(DataHandle&, InterfaceTracker&,
MessageBuffer<typename DataHandle::DataType>&, int)
{
return 0;
}
int operator()(DataHandle&, InterfaceTracker&,
MessageBuffer<typename DataHandle::DataType>&)
{
return 0;
}
};
template<class DataHandle, class BufferFunctor, class CommunicationFunctor>
std::size_t checkAndContinue(DataHandle& handle,
std::vector<InterfaceTracker>& trackers,
std::vector<MPI_Request>& requests,
std::vector<MPI_Request>& requests2,
std::vector<MessageBuffer<typename DataHandle::DataType> >& buffers,
MPI_Comm comm,
BufferFunctor buffer_func,
CommunicationFunctor comm_func,
bool valid=true,
bool getCount=false)
{
std::size_t size=requests.size();
std::vector<MPI_Status> statuses(size);
int no_completed;
std::vector<int> indices(size, -1); // the indices for which the communication finished.
MPI_Testsome(size, &(requests[0]), &no_completed, &(indices[0]), &(statuses[0]));
indices.resize(no_completed);
for(std::vector<int>::iterator index=indices.begin(), end=indices.end();
index!=end; ++index)
{
InterfaceTracker& tracker=trackers[*index];
setReceivingIndex(handle, *index);
if(getCount)
{
// Get the number of entries received
int count;
MPI_Get_count(&(statuses[index-indices.begin()]),
Dune::MPITraits<typename DataHandle::DataType>::getType(),
&count);
// Communication completed, we can reuse the buffers, e.g. unpack or repack
buffer_func(handle, tracker, buffers[*index], count);
}else
buffer_func(handle, tracker, buffers[*index]);
tracker.skipZeroIndices();
if(!tracker.finished()){
// Maybe start another communication.
comm_func(handle, tracker, buffers[*index], requests2[*index], comm);
tracker.skipZeroIndices();
if(valid)
--no_completed; // communication not finished, decrement counter for finished ones.
}
}
return no_completed;
}
template<class DataHandle>
std::size_t receiveSizeAndSetupReceive(DataHandle& handle,
std::vector<InterfaceTracker>& trackers,
std::vector<MPI_Request>& size_requests,
std::vector<MPI_Request>& data_requests,
std::vector<MessageBuffer<typename DataHandle::DataType> >& buffers,
MPI_Comm comm)
{
return checkAndContinue(handle, trackers, size_requests, data_requests, buffers, comm,
NullPackUnpackFunctor<DataHandle>(), SetupRecvRequest<DataHandle>(), false);
}
template<class DataHandle>
std::size_t checkSendAndContinueSending(DataHandle& handle,
std::vector<InterfaceTracker>& trackers,
std::vector<MPI_Request>& requests,
std::vector<MessageBuffer<typename DataHandle::DataType> >& buffers,
MPI_Comm comm)
{
return checkAndContinue(handle, trackers, requests, requests, buffers, comm,
NullPackUnpackFunctor<DataHandle>(), SetupSendRequest<DataHandle>());
}
template<class DataHandle>
std::size_t checkReceiveAndContinueReceiving(DataHandle& handle,
std::vector<InterfaceTracker>& trackers,
std::vector<MPI_Request>& requests,
std::vector<MessageBuffer<typename DataHandle::DataType> >& buffers,
MPI_Comm comm)
{
return checkAndContinue(handle, trackers, requests, requests, buffers, comm,
UnpackEntries<DataHandle>(), SetupRecvRequest<DataHandle>(),
true, !handle.fixedsize());
}
[[maybe_unused]] bool validRecvRequests(const std::vector<MPI_Request> reqs)
{
for(std::vector<MPI_Request>::const_iterator i=reqs.begin(), end=reqs.end();
i!=end; ++i)
if(*i!=MPI_REQUEST_NULL)
return true;
return false;
}
template<class DataHandle, class Functor>
std::size_t setupRequests(DataHandle& handle,
std::vector<InterfaceTracker>& trackers,
std::vector<MessageBuffer<typename DataHandle::DataType> >& buffers,
std::vector<MPI_Request>& requests,
const Functor& setupFunctor,
MPI_Comm communicator)
{
typedef typename std::vector<InterfaceTracker>::iterator TIter;
typename std::vector<MessageBuffer<typename DataHandle::DataType> >::iterator
biter=buffers.begin();
typename std::vector<MPI_Request>::iterator riter=requests.begin();
std::size_t complete=0;
for(TIter titer=trackers.begin(), end=trackers.end(); titer!=end; ++titer, ++biter, ++riter)
{
setupFunctor(handle, *titer, *biter, *riter, communicator);
complete+=titer->finished();
}
return complete;
}
} // end unnamed namespace
template<class Allocator>
template<bool FORWARD, class DataHandle>
void VariableSizeCommunicator<Allocator>::setupInterfaceTrackers(DataHandle& handle,
std::vector<InterfaceTracker>& send_trackers,
std::vector<InterfaceTracker>& recv_trackers)
{
if(interface_->size()==0)
return;
send_trackers.reserve(interface_->size());
recv_trackers.reserve(interface_->size());
int fixedsize=0;
if(handle.fixedsize())
++fixedsize;
typedef typename InterfaceMap::const_iterator IIter;
for(IIter inf=interface_->begin(), end=interface_->end(); inf!=end; ++inf)
{
if(handle.fixedsize() && InterfaceInformationChooser<FORWARD>::getSend(inf->second).size())
fixedsize=handle.size(InterfaceInformationChooser<FORWARD>::getSend(inf->second)[0]);
assert(!handle.fixedsize()||fixedsize>0);
send_trackers.push_back(InterfaceTracker(inf->first,
InterfaceInformationChooser<FORWARD>::getSend(inf->second), fixedsize));
recv_trackers.push_back(InterfaceTracker(inf->first,
InterfaceInformationChooser<FORWARD>::getReceive(inf->second), fixedsize, fixedsize==0));
}
}
template<class Allocator>
template<bool FORWARD, class DataHandle>
void VariableSizeCommunicator<Allocator>::communicateFixedSize(DataHandle& handle)
{
std::vector<MPI_Request> size_send_req(interface_->size());
std::vector<MPI_Request> size_recv_req(interface_->size());
std::vector<InterfaceTracker> send_trackers;
std::vector<InterfaceTracker> recv_trackers;
setupInterfaceTrackers<FORWARD>(handle,send_trackers, recv_trackers);
sendFixedSize(send_trackers, size_send_req, recv_trackers, size_recv_req, communicator_);
std::vector<MPI_Request> data_send_req(interface_->size(), MPI_REQUEST_NULL);
std::vector<MPI_Request> data_recv_req(interface_->size(), MPI_REQUEST_NULL);
typedef typename DataHandle::DataType DataType;
std::vector<MessageBuffer<DataType> > send_buffers(interface_->size(), MessageBuffer<DataType>(maxBufferSize_)),
recv_buffers(interface_->size(), MessageBuffer<DataType>(maxBufferSize_));
setupRequests(handle, send_trackers, send_buffers, data_send_req,
SetupSendRequest<DataHandle>(), communicator_);
std::size_t no_size_to_recv, no_to_send, no_to_recv, old_size;
no_size_to_recv = no_to_send = no_to_recv = old_size = interface_->size();
// Skip empty interfaces.
typedef typename std::vector<InterfaceTracker>::const_iterator Iter;
for(Iter i=recv_trackers.begin(), end=recv_trackers.end(); i!=end; ++i)
if(i->empty())
--no_to_recv;
for(Iter i=send_trackers.begin(), end=send_trackers.end(); i!=end; ++i)
if(i->empty())
--no_to_send;
while(no_size_to_recv+no_to_send+no_to_recv)
{
// Receive the fixedsize and setup receives accordingly
if(no_size_to_recv)
no_size_to_recv -= receiveSizeAndSetupReceive(handle,recv_trackers, size_recv_req,
data_recv_req, recv_buffers,
communicator_);
// Check send completion and initiate other necessary sends
if(no_to_send)
no_to_send -= checkSendAndContinueSending(handle, send_trackers, data_send_req,
send_buffers, communicator_);
if(validRecvRequests(data_recv_req))
// Receive data and setup new unblocking receives if necessary
no_to_recv -= checkReceiveAndContinueReceiving(handle, recv_trackers, data_recv_req,
recv_buffers, communicator_);
}
// Wait for completion of sending the size.
//std::vector<MPI_Status> statuses(interface_->size(), MPI_STATUSES_IGNORE);
MPI_Waitall(size_send_req.size(), &(size_send_req[0]), MPI_STATUSES_IGNORE);
}
template<class Allocator>
template<bool FORWARD, class DataHandle>
void VariableSizeCommunicator<Allocator>::communicateSizes(DataHandle& handle,
std::vector<InterfaceTracker>& data_recv_trackers)
{
std::vector<InterfaceTracker> send_trackers;
std::vector<InterfaceTracker> recv_trackers;
std::size_t size = interface_->size();
std::vector<MPI_Request> send_requests(size, MPI_REQUEST_NULL);
std::vector<MPI_Request> recv_requests(size, MPI_REQUEST_NULL);
std::vector<MessageBuffer<std::size_t> >
send_buffers(size, MessageBuffer<std::size_t>(maxBufferSize_)),
recv_buffers(size, MessageBuffer<std::size_t>(maxBufferSize_));
SizeDataHandle<DataHandle> size_handle(handle,data_recv_trackers);
setupInterfaceTrackers<FORWARD>(size_handle,send_trackers, recv_trackers);
setupRequests(size_handle, send_trackers, send_buffers, send_requests,
SetupSendRequest<SizeDataHandle<DataHandle> >(), communicator_);
setupRequests(size_handle, recv_trackers, recv_buffers, recv_requests,
SetupRecvRequest<SizeDataHandle<DataHandle> >(), communicator_);
// Count valid requests that we have to wait for.
auto valid_req_func =
[](const MPI_Request& req) { return req != MPI_REQUEST_NULL; };
auto size_to_send = std::count_if(send_requests.begin(), send_requests.end(),
valid_req_func);
auto size_to_recv = std::count_if(recv_requests.begin(), recv_requests.end(),
valid_req_func);
while(size_to_send+size_to_recv)
{
if(size_to_send)
size_to_send -=
checkSendAndContinueSending(size_handle, send_trackers, send_requests,
send_buffers, communicator_);
if(size_to_recv)
// Could have done this using checkSendAndContinueSending
// But the call below is more efficient as UnpackSizeEntries
// uses std::copy.
size_to_recv -=
checkAndContinue(size_handle, recv_trackers, recv_requests, recv_requests,
recv_buffers, communicator_, UnpackSizeEntries<DataHandle>(),
SetupRecvRequest<SizeDataHandle<DataHandle> >());
}
}
template<class Allocator>
template<bool FORWARD, class DataHandle>
void VariableSizeCommunicator<Allocator>::communicateVariableSize(DataHandle& handle)
{
std::vector<InterfaceTracker> send_trackers;
std::vector<InterfaceTracker> recv_trackers;
setupInterfaceTrackers<FORWARD>(handle, send_trackers, recv_trackers);
std::vector<MPI_Request> send_requests(interface_->size(), MPI_REQUEST_NULL);
std::vector<MPI_Request> recv_requests(interface_->size(), MPI_REQUEST_NULL);
typedef typename DataHandle::DataType DataType;
std::vector<MessageBuffer<DataType> >
send_buffers(interface_->size(), MessageBuffer<DataType>(maxBufferSize_)),
recv_buffers(interface_->size(), MessageBuffer<DataType>(maxBufferSize_));
communicateSizes<FORWARD>(handle, recv_trackers);
// Setup requests for sending and receiving.
setupRequests(handle, send_trackers, send_buffers, send_requests,
SetupSendRequest<DataHandle>(), communicator_);
setupRequests(handle, recv_trackers, recv_buffers, recv_requests,
SetupRecvRequest<DataHandle>(), communicator_);
// Determine number of valid requests.
auto valid_req_func =
[](const MPI_Request& req) { return req != MPI_REQUEST_NULL;};
auto no_to_send = std::count_if(send_requests.begin(), send_requests.end(),
valid_req_func);
auto no_to_recv = std::count_if(recv_requests.begin(), recv_requests.end(),
valid_req_func);
while(no_to_send+no_to_recv)
{
// Check send completion and initiate other necessary sends
if(no_to_send)
no_to_send -= checkSendAndContinueSending(handle, send_trackers, send_requests,
send_buffers, communicator_);
if(no_to_recv)
// Receive data and setup new unblocking receives if necessary
no_to_recv -= checkReceiveAndContinueReceiving(handle, recv_trackers, recv_requests,
recv_buffers, communicator_);
}
}
template<class Allocator>
template<bool FORWARD, class DataHandle>
void VariableSizeCommunicator<Allocator>::communicate(DataHandle& handle)
{
if( interface_->size() == 0)
// Simply return as otherwise we will index an empty container
// either for MPI_Wait_all or MPI_Test_some.
return;
if(handle.fixedsize())
communicateFixedSize<FORWARD>(handle);
else
communicateVariableSize<FORWARD>(handle);
}
} // end namespace Dune
#endif // HAVE_MPI
#endif
Opm
Holds the implementation of the CpGrid as a pimple.
Definition CellQuadrature.cpp:68