CepsContainerCommunication.hpp

Go to the documentation of this file.

/*~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
   This file is part of CEPS.
 
   CEPS 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.
 
   CEPS 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 CEPS (see file LICENSE at root of project).
   If not, see <https://www.gnu.org/licenses/>.
 
 
   Copyright 2019-2024 Inria, Universite de Bordeaux
 
   Authors, in alphabetical order:
 
     Pierre-Elliott BECUE, Florian CARO, Yves COUDIERE(*), Andjela DAVIDOVIC, 
     Charlie DOUANLA-LONTSI, Marc FUENTES, Mehdi JUHOOR, Michael LEGUEBE(*), 
     Pauline MIGERDITICHAN, Valentin PANNETIER(*), Nejib ZEMZEMI.
     * : currently active authors
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~*/
# pragma once
 
#include "common/core/CepsParallelTools.hpp"
#include "common/core/CepsIoMacros.hpp"
#include "common/core/CepsTypeTools.hpp"
#include "common/containers/CepsMap.hpp"
#include "common/containers/CepsVector.hpp"
 
namespace ceps
{
 
  template <typename T>
  int
  allGatherv(const CepsVector<T>& localTab, int localSize, 
             CepsVector<T>& globalTab, CepsBool masterOnly = false);
 
  int
  allGatherv(const std::unordered_map<CepsUInt, CepsUInt>& orig,
                   std::unordered_map<CepsUInt, CepsUInt>& dest);
 
  template<typename _Key, typename _Val, 
           template<typename,typename> typename _Container>
  int
  broadcastMap(const CepsUInt& sender, _Container<_Key,_Val>& table);
 
  template< typename _Key, typename _Val, 
            template<typename,typename> typename _Container>
  int
  sendMap(
    const CepsUInt&        sender,
    const CepsUInt&        receiver,
    _Container<_Key,_Val>& table,
    const CepsUInt&        nComms =1,
    const CepsUInt&        commId =1
  );
 
  template<typename T>
  void
  sendRecvWithMatrix(const Eigen::Matrix<CepsUInt, Eigen::Dynamic,Eigen::Dynamic>& commMat,
                     CepsVector<CepsVector<T>>& sendBuffers,
                     CepsVector<CepsVector<T>>& recvBuffers);
 
} // namespace ceps
 
 
template <typename T>
int
ceps::allGatherv(const CepsVector<T>& localTab, int localSize, CepsVector<T>& globalTab, CepsBool masterOnly)
{
  CepsUInt gridSize = ceps::getGridSize();
  MPI_Comm comm     = ceps::getCommunicator();
 
  CepsVector<int> localSizes(ceps::getGridSize(),0);
  MPI_Allgather(&localSize,1,MPI_INT,localSizes.data(),1,MPI_INT,comm);
 
  // We do not use ceps::cumSum to avoid forward declaration
  CepsVector<int> displacements(gridSize+1,0);
  CepsUInt count(0u);
  for (CepsUInt i=0u; i<gridSize; i++)
  {
    count += localSizes[i];
    displacements[i+1] = count;
  }
 
  T*    localNonConst = const_cast<T*>(localTab.data());   // we need of these two casts
  void* localVoid     = static_cast<void*>(localNonConst); // because first argument of MPI_Allgatherv is void*
 
  MPI_Datatype mpiType = MpiType<T>::getType();
 
  if (masterOnly)
  {
    if (ceps::isMaster())
      globalTab.resize(displacements[gridSize]);
    return MPI_Gatherv(localVoid, localSize, mpiType, globalTab.data(), localSizes.data(),
                       displacements.data(), mpiType, CEPS_MASTER_PROC, comm);
  }
  else
  {
    globalTab.resize(displacements[gridSize]);
    return MPI_Allgatherv(localVoid, localSize, mpiType, globalTab.data(), localSizes.data(),
                          displacements.data(), mpiType, comm);
  }
}
 
template< typename _Key, typename _Val, 
          template<typename,typename> typename _Container>
int
ceps::broadcastMap(const CepsUInt& sender, _Container<_Key,_Val>&table) 
{
 
  MPI_Comm comm     = ceps::getCommunicator();
  CepsBool isSender = ceps::getRank() == sender;
  CepsUInt nItems   = isSender ? table.size() : 0;
 
  if (not isSender)
    table.clear();
 
  MPI_Bcast(&nItems,1,CEPS_MPI_UINT,sender,comm);
  CepsVector<_Key> keys   = ceps::keysOf  (table);
  CepsVector<_Val> values = ceps::valuesOf(table);
  if (not isSender) 
  {
    keys  .resize(nItems);
    values.resize(nItems);
  }
  MPI_Bcast(keys  .data(),nItems,MpiType<_Key>::getType(),sender,comm);
  MPI_Bcast(values.data(),nItems,MpiType<_Val>::getType(),sender,comm);
  if (not isSender)
    for (CepsUInt i = 0U; i < keys.size (); ++i)
      table.emplace (keys[i], values[i]);
 
  return MPI_SUCCESS;
 
}
 
template<typename _Key, typename _Val, 
         template<typename,typename> typename _Container>
int
ceps::sendMap(
  const CepsUInt&        sender,
  const CepsUInt&        receiver,
  _Container<_Key,_Val>& table,
  const CepsUInt&        nComms,
  const CepsUInt&        commId
) 
{
 
  CepsBool isSender   = ceps::getRank() == sender;
  CepsBool isReceiver = ceps::getRank() == receiver;
  if (isReceiver or isSender)
  {
 
    CepsUInt   gSize  = ceps::getGridSize();
    MPI_Comm   comm   = ceps::getCommunicator();
    CepsUInt   nItems = isSender ? table.size() : 0;
    MPI_Status status;
 
    if (isReceiver)
      table.clear();
  
    CepsInt tag = commId + nComms*(receiver + gSize*sender);
  
    if (isSender)
      MPI_Send(&nItems,1,CEPS_MPI_UINT,receiver,3*tag,comm);
    if (isReceiver)
      MPI_Recv(&nItems,1,CEPS_MPI_UINT,sender,3*tag,comm,&status);
 
    CepsVector<_Key> keys   = ceps::keysOf  (table);
    CepsVector<_Val> values = ceps::valuesOf(table);
    if (isSender)
      MPI_Send(keys.data(),nItems,MpiType<_Key>::getType(),receiver,3*tag+1,comm);
    if (isReceiver)
    {
      keys  .resize(nItems);
      MPI_Recv(keys.data(),nItems,MpiType<_Key>::getType(),sender,3*tag+1,comm,&status);
    }
 
    if (isSender)
    {
      MPI_Send(values.data(),nItems,MpiType<_Val>::getType(),receiver,3*tag+2,comm);
    }
    if (isReceiver)
    {
      values.resize(nItems);
      MPI_Recv(values.data(),nItems,MpiType<_Val>::getType(),sender,3*tag+2,comm,&status);
      for (CepsUInt i = 0U; i < keys.size (); ++i)
        table.emplace (keys[i], values[i]);
    }
  }
 
  return MPI_SUCCESS;
 
}
 
 
template <typename T>
void ceps::
sendRecvWithMatrix(const Eigen::Matrix<CepsUInt, Eigen::Dynamic, Eigen::Dynamic>& commMat, 
                   CepsVector<CepsVector<T>>& sendBuffers, 
                   CepsVector<CepsVector<T>>& recvBuffers)
{
  CepsUInt gridSize = ceps::getGridSize();
  MPI_Comm comm     = ceps::getCommunicator();
  CepsUInt rank     = ceps::getRank();
  CepsUInt ncols    = commMat.cols();
  CepsUInt nrows    = commMat.rows();
 
  CepsBool badShape = nrows > gridSize or ncols > gridSize;
  CEPS_ABORT_IF(badShape,
    "given comm Matrix has wrong shape " << nrows << " " << ncols
  );
 
  std::map<CepsUInt, CepsUInt> connectivity;
  for (CepsUInt i = 0U; i < gridSize; ++i)
    connectivity[i] = commMat(rank,i);
 
  CepsUInt     i            = 0;
  MPI_Request *sendRequests = ceps::newArray<MPI_Request>(connectivity.size());
 
  for (auto &[destination, data_size] : connectivity)
    MPI_Isend(sendBuffers[destination].data (), data_size, MpiType<T>::getType(),
              destination,rank*gridSize+destination,comm, &sendRequests[i++]);
 
  // do all the receives (blocking)
  connectivity.clear();
  for (CepsUInt j=0U; j<gridSize; ++j)
    connectivity[j] = commMat(j,rank);
 
  CepsUInt nbReceives          = connectivity.size();
  i                            = 0;
  MPI_Request *receiveRequests = ceps::newArray<MPI_Request>(nbReceives);
  MPI_Status  *receiveStatus   = ceps::newArray<MPI_Status >(nbReceives);
 
  recvBuffers.resize(gridSize);
  for (auto &[sender, data_size] : connectivity)
  {
    recvBuffers[sender].resize(data_size);
    MPI_Recv(recvBuffers[sender].data(), data_size, MpiType<T>::getType(), 
             sender, sender*gridSize+rank, comm, &receiveStatus[i++]);
  }
 
  ceps::destroyTabular(receiveRequests);
  ceps::destroyTabular(receiveStatus  );
  ceps::destroyTabular(sendRequests   );
  return;
}
 
template <>
void ceps::
sendRecvWithMatrix(const Eigen::Matrix<CepsUInt, Eigen::Dynamic, Eigen::Dynamic>& commMat, 
                   CepsVector<CepsVector<CepsHash3>>& sendBuffers, 
                   CepsVector<CepsVector<CepsHash3>>& recvBuffers);