communications.cpp

/****************************************************************************
* Copyright (c) 2026, CEA
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
* 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
* 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
* 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
* IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
* OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*****************************************************************************/
 
#include <InOutCommBuffers.h>
#include <communications.h>
#include <Schema_Comm.h>
#include <TRUSTArrays.h>
#include <TRUSTTabs.h>
#include <TRUSTTab.h>
#include <Motcle.h>
 
/*! @brief Synchronous point-to-point message sending between source and target.
 *
 * The envoyer function completes at the latest when the message has been
 *   received by the corresponding "recevoir" instruction, and the send can be blocking
 *   (blocking is forced in debug mode, according to Comm_Group::check_enabled())
 *   These communications use Comm_Group::send() and Comm_Group::recv()
 *   Examples of possible uses:
 *   1) point-to-point communication (processor 2 to processor 5)
 *        if (me() == 2)
 *          envoyer(objet, 5, canal);
 *        else if (me() == 5)
 *          recevoir(objet, 2, canal);
 *      or equivalently:
 *        envoyer(objet, 2, 5, canal); // does nothing on processors other than 2
 *        recevoir(objet, 2, 5, canal); // does nothing on processors other than 5
 *   2) gather: processor 0 retrieves the data of object t from everyone
 *        envoyer(t, me(), 0, canal); // Everyone sends to processor 0, on master it does nothing
 *        if (je_suis_maitre())
 *          for (i = 0; i < nproc(); i++) {
 *            recevoir(t, 0, i; canal); // If i==0, does nothing.
 *            ... do something with t
 *          }
 *   3) broadcast: processor 0 sends the object to all other processors
 *      It is not necessary to do the test "if (je_suis_maitre())"
 *        envoyer(objet, 0, -1, canal); // Processors other than 0 ignore the call
 *        recevoir(objet, 0, -1, canal);// Processor 0 ignores the call
 *      It is better to use envoyer_broadcast() for this purpose.
 *   4) broadcast: equivalent form
 *        if (je_suis_maitre())
 *          envoyer(objet, -1, canal); // Emission by master to all processors, except itself
 *        else
 *          recevoir(objet, 0, canal); // Reception of a message from the master processor
 *      It is better to use envoyer_broadcast() for this purpose.
 *  In the implementation of envoyer_, the received object is encapsulated in a buffer and the buffer is sent.
 *  This therefore works for complex types as long as Entree and Sortie can read and write the object.
 *  Each send requires two messages: first the buffer size, then the content.
 *
 */
template <typename T>
bool envoyer_buffered_(const T& objet, int source, int cible, int canal)
{
  const int moi = Process::me();
  if (moi == cible)
    return false;
 
  if (source != moi && source != -1)
    return false;
 
  OutputCommBuffer buffer;
  buffer << objet;
  const char * data = buffer.get_buffer();
  const int sz   = buffer.get_buffer_size();
  const Comm_Group& grp = PE_Groups::current_group();
  if (cible < 0)
    {
      const int n = grp.nproc();
      for (int i = 0; i < n; i++)
        {
          if (i != moi)
            {
              grp.send(i, &sz, sizeof(int), canal);
              grp.send(i, data, sz, canal);
            }
        }
    }
  else
    {
      grp.send(cible, &sz, sizeof(int), canal);
      grp.send(cible, data, sz, canal);
    }
  return true;
}
 
/*! @brief Reception of a message from the source processor.
 *
 */
template <typename T>
bool recevoir_buffered_(T& objet, int source, int cible, int canal)
{
  const int moi = Process::me();
  if (moi == source)
    return false;
 
  if (cible != moi && cible != -1)
    return false;
 
  int sz;
  const Comm_Group& grp = PE_Groups::current_group();
  grp.recv(source, &sz, sizeof(int), canal);
  InputCommBuffer buffer;
  char * data = buffer.reserve_buffer(sz);
  grp.recv(source, data, sz, canal);
  buffer.create_stream();
  buffer >> objet;
  buffer.clear();
  return true;
}
 
/*! @brief Broadcast of the object by the source processor to all other processors.
 *
 * Works for complex types supported by Entree or Sortie.
 *
 */
template <typename T>
bool envoyer_broadcast_buffered_(T& objet, int source)
{
  const Comm_Group& grp = PE_Groups::current_group();
  if (grp.rank() == source)
    {
      OutputCommBuffer buffer;
      buffer << objet;
      const char * data = buffer.get_buffer();
      int sz   = buffer.get_buffer_size();
      grp.broadcast(&sz, sizeof(int), source);
      grp.broadcast((char*) data, sz, source);
    }
  else
    {
      int sz;
      grp.broadcast(&sz, sizeof(int), source);
      InputCommBuffer buffer;
      char * data = buffer.reserve_buffer(sz);
      grp.broadcast((char*) data, sz, source);
      buffer.create_stream();
      buffer >> objet;
      buffer.clear();
    }
  return true;
}
 
bool envoyer(const Objet_U& t, int source, int cible, int canal)
{
  return envoyer_buffered_(t, source, cible, canal);
}
 
bool envoyer(const Objet_U& t, int cible, int canal)
{
  return envoyer_buffered_(t, Process::me(), cible, canal);
}
 
bool recevoir(Objet_U& t, int source, int cible, int canal)
{
  return recevoir_buffered_(t, source, cible, canal);
}
 
bool recevoir(Objet_U& t, int source, int canal)
{
  return recevoir_buffered_(t, source, Process::me(), canal);
}
 
bool envoyer_broadcast(Objet_U& t, int source)
{
  return envoyer_broadcast_buffered_(t, source);
}
 
// Template that works for any object T that looks like a vector (has an operator[]) of size nproc()
template <typename T>
bool envoyer_all_to_all_(const T& src, T& dest)
{
  const Comm_Group& grp = PE_Groups::current_group();
  assert(src.size() == Process::nproc());
  assert(dest.size() == Process::nproc());
  Schema_Comm sch;
  const int nproc = grp.nproc();
  const int moi = grp.rank();
  dest[moi] = src[moi];
  ArrOfInt send_pe_list(1);
  ArrOfInt recv_pe_list(1);
  for (int p = 1; p < nproc; p++)
    {
      const int send_to = (moi + p) % nproc;
      const int recv_from = (moi + nproc - p) % nproc;
      send_pe_list[0] = send_to;
      recv_pe_list[0] = recv_from;
      sch.set_send_recv_pe_list(send_pe_list, recv_pe_list);
      sch.begin_comm();
      sch.send_buffer(send_to) << src[send_to];
      sch.echange_taille_et_messages();
      sch.recv_buffer(recv_from) >> dest[recv_from];
      sch.end_comm();
    }
  return true;
}
 
bool envoyer_all_to_all(const TRUST_Vector<TRUSTTab<double>>& src, TRUST_Vector<TRUSTTab<double>>& dest)
{
  return envoyer_all_to_all_(src, dest);
}
 
bool envoyer_all_to_all(const TRUST_Vector<TRUSTArray<int>>& src, TRUST_Vector<TRUSTArray<int>>& dest)
{
  return envoyer_all_to_all_(src, dest);
}
 
#if INT_is_64_ == 2
bool envoyer_all_to_all(const TRUST_Vector<TRUSTArray<trustIdType>>& src, TRUST_Vector<TRUSTArray<trustIdType>>& dest)
{
  return envoyer_all_to_all_(src, dest);
}
#endif
 
/*! @brief Assumes that the input and output arrays are of size nproc().
 *
 * src[0] is sent to proc 0,
 *   src[1] to proc 1, etc... the value received from processor 0 is placed in
 *   dest[0], processor 1 in dest[1], etc...
 *   It is allowed to call this function with the same array for src and dest.
 *
 */
bool envoyer_all_to_all(std::vector<long long>& src, std::vector<long long>& dest)
{
  const Comm_Group& grp = PE_Groups::current_group();
  assert(static_cast<int>(src.size()) == grp.nproc());
  assert(static_cast<int>(dest.size()) == grp.nproc());
  if (src.data() == dest.data())
    {
      std::vector<long long> tmp(grp.nproc(),0);
      grp.all_to_all(src.data(), tmp.data(), sizeof(long long));
      dest = tmp;
    }
  else
    {
      grp.all_to_all(src.data(), dest.data(), sizeof(long long));
    }
  return true;
}
 
/*! @brief Computes the transpose of a processor list: builds the array dest_list such that:
 *
 *         x is in the src_list on processor y
 *       if and only if
 *         y is in the dest_list on processor x
 *   In practice, this function computes the list of
 *   processors that receive messages when the list of
 *   processors that send messages is known, and vice versa.
 *
 * @param (src_list) A list of PE numbers present in this comm group. These are the indices of the PEs in the current group (0 <= pe < nproc_)
 * @param (dest_list) The array in which to store the result. It is resized and filled appropriately. PE numbers are stored in ascending order. These are the indices of the PEs in the current group (0 <= pe < nproc_)
 */
bool reverse_send_recv_pe_list(const ArrOfInt& src_list, ArrOfInt& dest_list)
{
  const Comm_Group& grp = PE_Groups::current_group();
  const int np = grp.nproc();
  assert(src_list.size_array() == 0 || max_array(src_list) < np);
  assert(src_list.size_array() == 0 || min_array(src_list) >= 0);
  ArrOfInt flag(np);
  int i;
  const int src_size = src_list.size_array();
  for (i = 0; i < src_size; i++)
    {
      const int pe = src_list[i];
      flag[pe] = 1;
    }
  ArrOfInt flag_recv(np);
  envoyer_all_to_all(flag, flag_recv);
  int n = 0;
  dest_list.resize_array(np);
  for (i = 0; i < np; i++)
    if (flag_recv[i])
      dest_list[n++] = i;
  dest_list.resize_array(n);
  return true;
}
 
/*! @brief Assumes that the input and output arrays have dimension(0)==nproc() and that the other dimensions
 *
 *   are identical. src(0, i, j, ...) is sent to proc 0,
 *   src(0, i, j, ...) to proc 1, etc... the value received from processor 0 is placed in
 *   dest(0, i, j, ...) , processor 1 in dest(1, i, j, ...) , etc...
 *   It is allowed to call this function with the same array for src and dest.
 *
 */
void envoyer_all_to_all(const DoubleTab& src, DoubleTab& dest)
{
  const Comm_Group& grp = PE_Groups::current_group();
  assert(src.dimension(0) == grp.nproc());
  assert(dest.dimension(0) == grp.nproc());
  assert(src.size_array() == dest.size_array());
 
  if (dest.data() == src.data())
    {
      DoubleTab tmp(src);
      envoyer_all_to_all(src, tmp);
      dest = tmp;
    }
  else
    {
      const int sz = src.size() / grp.nproc() * (int)sizeof(double);
      grp.all_to_all(src.data(), dest.data(), sz);
    }
}
template <typename T>
void envoyer_all_gather_(const TRUSTTab<T>& src, TRUSTTab<T>& dest)
{
  const Comm_Group& grp = PE_Groups::current_group();
 
  if (dest.addr() == src.addr())
    {
      assert(src.dimension(0) == grp.nproc());
      assert(dest.dimension(0) == grp.nproc());
      const int sz = src.size() / grp.nproc() * (int)sizeof(T);
      TRUSTTab<T> tmp(src);
      grp.all_gather(src.addr(), tmp.addr(), sz);
      dest = tmp;
    }
  else
    {
      const int sz = src.size() * (int)sizeof(T);
      assert( dest.size_array() == grp.nproc()*src.size_array());
      grp.all_gather(src.addr(), dest.addr(), sz);
    }
}
 
template <typename T>
void envoyer_gather_(const TRUSTTab<T>& src, TRUSTTab<T>& dest, int root)
{
  const Comm_Group& grp = PE_Groups::current_group();
 
  if (dest.addr() == src.addr())
    {
      assert(src.dimension(0) == grp.nproc());
      assert(dest.dimension(0) == grp.nproc());
      const int sz = src.size() / grp.nproc() * (int)sizeof(T);
      TRUSTTab<T> tmp(src);
      grp.gather(src.addr(), tmp.addr(), sz, root);
      dest = tmp;
    }
  else
    {
      const int sz = src.size() * (int)sizeof(T);
      assert( dest.size_array() == grp.nproc()*src.size_array());
      grp.gather(src.addr(), dest.addr(), sz, root);
    }
}
 
void envoyer_gather(const DoubleTab& src, DoubleTab& dest, int root)
{
  envoyer_gather_(src,dest,root);
}
 
void envoyer_gather(const IntTab& src, IntTab& dest, int root)
{
  envoyer_gather_(src,dest,root);
}
 
void envoyer_all_gather(const DoubleTab& src, DoubleTab& dest)
{
  envoyer_all_gather_(src,dest);
}
 
void envoyer_all_gather(const IntTab& src, IntTab& dest)
{
  envoyer_all_gather_(src,dest);
}
 
void envoyer_all_gatherv(const DoubleTab& src, DoubleTab& dest, const IntTab& recv_size)
{
  const Comm_Group& grp = PE_Groups::current_group();
 
  assert(dest.size_array()==local_somme_vect(recv_size));
  assert(src.size_array()==recv_size[grp.me()]);
 
  const int nbprocs = (int)grp.nproc();
 
  std::vector<int> sz(nbprocs);
  for (int p=0; p<nbprocs; p++)
    sz[p] = recv_size[p] * (int)sizeof(double);
 
  int sz0 =  src.size() * (int)sizeof(double);
 
  std::vector<int> displs(nbprocs);
  displs[0] = 0;
  for (int p=1; p<nbprocs; p++)
    displs[p] = displs[p-1]+sz[p-1];
 
  grp.all_gatherv(src.addr(), dest.addr(), sz0 , sz.data(),displs.data());
}
 
 
/*! @brief Returns the Comm_Group::check_enabled() flag.
 *
 * This flag indicates whether the code is running in parallel DEBUG mode. In that case, additional communications take place to verify parallelism consistency and the communication mode
 *   is forced to synchronous. This method allows testing the flag, and if it is set, performing additional checks.
 *
 */
bool comm_check_enabled() { return Comm_Group::check_enabled(); }
 
/*! @brief Returns 1 (on all procs) if the parameter is identical on all procs, otherwise returns 0 (on all procs).
 *
 */
bool is_parallel_object(const int x)
{
  const int x1 = Process::mp_min(x);
  const int x2 = Process::mp_max(x);
  return x1 == x2;
}
 
#if INT_is_64_ == 2
bool is_parallel_object(const trustIdType x)
{
  const trustIdType x1 = Process::mp_min(x);
  const trustIdType x2 = Process::mp_max(x);
  return x1 == x2;
}
#endif
 
/*! @brief Returns 1 (on all procs) if the parameter is identical on all procs, otherwise returns 0 (on all procs).
 *
 */
bool is_parallel_object(const double x)
{
  const Comm_Group& grp = PE_Groups::current_group();
  double tab_x[2];
  tab_x[0] = x;
  tab_x[1] = x;
  double tab_y[2];
  Comm_Group::Collective_Op tab_op[] = { Comm_Group::COLL_MIN, Comm_Group::COLL_MAX };
  grp.mp_collective_op(tab_x, tab_y, tab_op, 2);
  return tab_y[0] == tab_y[1];
}
 
template <class T> int compare(const Objet_U& a, const Objet_U& b, int& erreur)
{
  if (sub_type(T, a))
    {
      const T& a_ = static_cast<const T&> (a);
      const T& b_ = static_cast<const T&> (b);
      if (! (a_ == b_))
        erreur = 1;
      return 1;
    }
  else
    {
      return 0;
    }
}
 
/*! @brief Returns 1 (on all procs) if the parameter is identical on all procs, otherwise returns 0 (on all procs).
 *
 */
bool is_parallel_object(const Objet_U& obj)
{
  int erreur = 0;
  if (Process::je_suis_maitre())
    {
      DerObjU copie;
      const int n = Process::nproc();
      for (int i = 1; i < n; i++)
        {
          recevoir(copie, i, 67 /* tag */);
          if (erreur)
            continue; // We already found an error, we receive the other messages but that's all...
          const Objet_U& obj2 = copie.valeur();
          if (obj2.que_suis_je() != obj.que_suis_je())
            {
              erreur = 1;
            }
          else
            {
              // The == operator is not virtual, so we cannot simply compare with ==.
              // We must call == for the specific sub-type of the object.
              // The order is important: first the most derived type, then its base class.
              if (compare<IntTab> (obj, obj2, erreur)) { /* Do nothing */ }
              else if (compare<DoubleTab> (obj, obj2, erreur)) { /* Do nothing */ }
              else if (compare<IntVect> (obj, obj2, erreur)) { /* Do nothing */ }
              else if (compare<DoubleVect> (obj, obj2, erreur)) { /* Do nothing */ }
              else if (compare<ArrOfInt> (obj, obj2, erreur)) { /* Do nothing */ }
              else if (compare<ArrOfDouble> (obj, obj2, erreur)) { /* Do nothing */ }
              else if (compare<Motcle> (obj, obj2, erreur)) { /* Do nothing */ }
              else if (compare<Nom> (obj, obj2, erreur)) { /* Do nothing */ }
              else
                {
                  Cerr << "Internal error in is_parallel_objec() : it is not coded for " << obj.que_suis_je() << finl;
                  Process::exit();
                }
            }
        }
    }
  else
    {
      // We send an object of type OWN_PTR..
      DerObjU copie(obj);
      envoyer(copie, 0, 67 /* tag */);
    }
 
  envoyer_broadcast(erreur, 0 /* pe source */);
  return !erreur;
}
 
/*! @brief In comm_check_enabled() mode, verifies that the parameter has the same value on all processors.
 *
 * Does not work for all types! Currently ok for Noms, Motcles and arrays.
 *
 */
void assert_parallel(const Objet_U& obj)
{
  if (!Comm_Group::check_enabled())
    return;
 
  if (!is_parallel_object(obj))
    {
      Cerr << "Fatal error in assert_parallel(const Objet_U & obj): objects do not match, see log files" << finl;
      Process::Journal() << "Error in assert_parallel(const Objet_U & obj)"
                         << "This processor has this object : " << obj.que_suis_je() << finl
                         << obj << finl;
      Process::barrier();
      Process::exit();
    }
}