CepsContainerCommunication.cpp

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
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~*/
#include "common/containers/CepsContainerCommunication.hpp"
#include "common/core/CepsTypeTools.hpp"
 
int
ceps::allGatherv(
  const std::unordered_map<CepsUInt,CepsUInt>& orig,
  std::unordered_map<CepsUInt,CepsUInt>&       dest
)
{
  auto gridSize = ceps::getGridSize();
 
  CepsVector<int> localSizes(gridSize  ,0);
  CepsVector<int> displs    (gridSize+1,0);
 
  CepsVector<CepsUInt> keysOrig;
  CepsVector<CepsUInt> valuesOrig;
 
  keysOrig  .reserve(orig.size());
  valuesOrig.reserve(orig.size());
  for (auto &[key, value] : orig)
  {
    keysOrig  .push_back(key);
    valuesOrig.push_back(value);
  }
 
  int origSize = (int)orig.size();
  int res      = MPI_Allgather(&origSize,1,MPI_INT,localSizes.data(),1,MPI_INT,ceps::getCommunicator());
  if (res != 0)
    return res;
 
  for (CepsUInt i = 1U; i < gridSize + 1; i++)
    displs[i] = displs[i-1] + localSizes[i-1];
 
  CepsVector<CepsUInt> keysDest;
  CepsVector<CepsUInt> valuesDest;
 
  keysDest  .resize(displs[gridSize]);
  valuesDest.resize(displs[gridSize]);
 
  int res1 = MPI_Allgatherv(keysOrig.data(),orig.size(),CEPS_MPI_UINT, 
                            keysDest.data(),localSizes.data(),displs.data(), 
                            CEPS_MPI_UINT,ceps::getCommunicator());
  if (res1 != 0)
    return res1;
 
  int res2 = MPI_Allgatherv(valuesOrig.data(),orig.size(),CEPS_MPI_UINT, 
                            valuesDest.data(),localSizes.data(),displs.data(),
                            CEPS_MPI_UINT,ceps::getCommunicator());
 
  if (res2 != 0)
    return res2;
 
  for (CepsUInt i=0U; i<keysDest.size(); ++i)
    dest[keysDest[i]] = valuesDest[i];
 
  return MPI_SUCCESS;
}
 
// Specialization for CepsHash3
template <>
void ceps::
sendRecvWithMatrix(const Eigen::Matrix<CepsUInt, Eigen::Dynamic, Eigen::Dynamic> &commMat, 
                   CepsVector<CepsVector<CepsHash3>> &sendBuffers, 
                   CepsVector<CepsVector<CepsHash3>> &recvBuffers)
{
  auto gridSize = ceps::getGridSize    ();
  auto comm     = ceps::getCommunicator();
  auto rank     = ceps::getRank        ();
 
  CepsBool badShape = (CepsUInt)commMat.rows() > gridSize or (CepsUInt)commMat.cols() > gridSize;
  CEPS_ABORT_IF(badShape,
    "given comm Matrix has wrong shape " << commMat.rows() << " " << commMat.cols()
  );
 
  std::map<CepsUInt, CepsUInt> connectivity;
  CepsArray3<CepsVector<CepsVector<CepsSHash>>> sBuffers;
  CepsArray3<CepsVector<CepsVector<CepsSHash>>> rBuffers;
 
  CepsUInt ns = sendBuffers.size();
  for (CepsUInt i=0; i<3; i++)
  {
    sBuffers[i].resize(ns);
    for (CepsUInt j=0; j<ns; j++)
      sBuffers[i][j].resize(sendBuffers[j].size());
  }
  for (CepsUInt j=0; j<ns; j++)
    for (CepsUInt k=0;k<sendBuffers[j].size();k++)
    {
      sBuffers[0][j][k] = sendBuffers[j][k].h1;
      sBuffers[1][j][k] = sendBuffers[j][k].h2;
      sBuffers[2][j][k] = sendBuffers[j][k].h3;
    }
 
  for (CepsUInt dim=0; dim<3; dim++)
  {
 
    rBuffers[dim].resize(gridSize);
 
    MPI_Status  status;
    MPI_Request request;
    // Try with slow blocking communications
    for (CepsUInt sndr=0u; sndr<gridSize; sndr++)
    for (CepsUInt rcvr=0u; rcvr<gridSize; rcvr++)
    {
      CepsInt tag      = dim+3*(sndr*gridSize+rcvr);
      CepsInt dataSize = commMat(sndr,rcvr);
      if (rank==sndr)
        MPI_Isend(sBuffers[dim][rcvr].data(),dataSize,CEPS_MPI_SHASH,rcvr,tag,comm,&request);
      if (rank==rcvr)
      {
        rBuffers[dim][sndr].resize(dataSize);
        MPI_Recv(rBuffers[dim][sndr].data(),dataSize,CEPS_MPI_SHASH,sndr,tag,comm,&status);
      }
 
    }
 
//     connectivity.clear();
//     for (CepsUInt i=0U; i<gridSize; ++i)
//       connectivity[i] = commMat(rank,i);
 
//     CepsUInt     i            = 0;
//     MPI_Request* sendRequests = new MPI_Request[connectivity.size()];
 
//     for (auto &[destination,dataSize] : connectivity)
//     {
//       MPI_Isend(sBuffers[dim][destination].data(),dataSize,CEPS_MPI_SHASH,
//                 destination,dim+3*(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);
 
//     i = 0;
//     MPI_Status* receiveStatus = new MPI_Status[connectivity.size()];
 
//     rBuffers[dim].resize(gridSize);
//     for (auto &[sender,dataSize] : connectivity)
//     {
//       rBuffers[dim][sender].resize(dataSize);
//       MPI_Recv(rBuffers[dim][sender].data(),dataSize,CEPS_MPI_SHASH, 
//                sender,dim+3*(sender*gridSize+rank),comm,&receiveStatus[i++]);
//     }
 
    // ceps::destroyTabular(receiveRequests);
    // ceps::destroyTabular(receiveStatus);
    // ceps::destroyTabular(sendRequests);
  }
 
  CepsUInt rs = rBuffers[0].size();
  recvBuffers.resize(rs);
  for (CepsUInt j=0; j<rs; j++)
  {
    CepsUInt rss = rBuffers[0][j].size();
    recvBuffers[j].resize(rss);
    for (CepsUInt k=0;k<rss;k++)
    {
      recvBuffers[j][k].h1 = rBuffers[0][j][k];
      recvBuffers[j][k].h2 = rBuffers[1][j][k];
      recvBuffers[j][k].h3 = rBuffers[2][j][k];
    }
  }
 
 
  return;
}