variablesizecommunicator.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:
#ifndef DUNE_COMMON_PARALLEL_VARIABLESIZECOMMUNICATOR_HH // Still fits the line!
#define DUNE_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 Dune
{
 
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)
  {
    buffer_[position_++]=data;
  }
 
  void read(T& data)
  {
    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_;
  }
  operator T*()
  {
    return buffer_;
  }
 
private:
  T* buffer_;
  std::size_t size_;
  std::size_t position_;
};
 
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 std::allocator_traits<Allocator>::template rebind_alloc< std::pair<const int,std::pair<InterfaceInformation,InterfaceInformation> > > > 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_);
  }
 
  VariableSizeCommunicator(const VariableSizeCommunicator& other) {
    maxBufferSize_ = other.maxBufferSize_;
    interface_ = other.interface_;
    MPI_Comm_dup(other.communicator_, &communicator_);
  }
 
  VariableSizeCommunicator& operator=(const VariableSizeCommunicator& other) {
    if(this == &other) // don't do anything if objects are the same
      return *this;
 
    maxBufferSize_ = other.maxBufferSize_;
    interface_ = other.interface_;
    MPI_Comm_free(&communicator_);
    MPI_Comm_dup(other.communicator_, &communicator_);
 
    return *this;
  }
 
  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>
  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())))
        {
          handle.gather(buffer, tracker.index());
          packed+=handle.size(tracker.index());
          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);
  }
};
 
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, 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, 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)
      MPI_Issend(buffer, size, 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(), 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()]),
                    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());
}
 
 
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