GatherGroup.cc

// -*- tab-width: 2; indent-tabs-mode: nil; coding: utf-8-with-signature -*-
//-----------------------------------------------------------------------------
// Copyright 2000-2026 CEA (www.cea.fr) IFPEN (www.ifpenergiesnouvelles.com)
// See the top-level COPYRIGHT file for details.
// SPDX-License-Identifier: Apache-2.0
//-----------------------------------------------------------------------------
/*---------------------------------------------------------------------------*/
/* GatherGroup.cc                                              (C) 2000-2026 */
/*                                                                           */
/* Class allowing the management of data groupings on the one or more        */
/* writing subdomains.                                                       */
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#include "arcane/core/internal/GatherGroup.h"
 
#include "arcane/utils/FatalErrorException.h"
#include "arcane/utils/Array2.h"
 
#include "arcane/core/IParallelMng.h"
#include "arcane/core/internal/IParallelMngInternal.h"
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
namespace Arcane
{
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
GatherGroup::
GatherGroup(GatherGroupInfo* gather_group_info)
: m_gather_group_info(ARCANE_CHECK_POINTER(gather_group_info))
{
  ARCANE_FATAL_IF(!gather_group_info->isComputed(), "GatherGroupInfo is not computed");
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
GatherGroup::
GatherGroup() = default;
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
GatherGroup::
~GatherGroup() = default;
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
bool GatherGroup::
isNeedGather()
{
  ARCANE_CHECK_POINTER(m_gather_group_info);
  IParallelMng* pm = m_gather_group_info->m_parallel_mng;
  // True if Thread or Hybrid or MPI without MPI-IO
  // False if Sequential or MPI + MPI-IO.
  return pm->commSize() != 1 && (pm->isThreadImplementation() || !m_gather_group_info->m_use_collective_io);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void GatherGroup::
gatherToMasterIO(Int64 sizeof_elem, Span<const Byte> in, Span<Byte> out)
{
  ARCANE_CHECK_POINTER(m_gather_group_info);
  IParallelMng* pm = m_gather_group_info->m_parallel_mng;
 
  // If sequential or MPI + MPI-IO.
  if ((pm->commSize() == 1) || (!pm->isThreadImplementation() && m_gather_group_info->m_use_collective_io)) {
    out.copy(in);
    return;
  }
 
  const Int32 writer = m_gather_group_info->m_writer;
 
  if (writer != pm->commRank()) {
    pm->send(in.constSmallView(), writer);
    return;
  }
 
  out.copy(in);
 
  const Int32 nb_sender = m_gather_group_info->m_nb_sender_to_writer;
 
  SmallSpan<const Int32> nb_elem_recved = m_gather_group_info->m_nb_elem_recv.smallSpan();
 
  UniqueArray<Parallel::Request> requests(nb_sender - 1);
 
  Int64 old_size = in.size();
  for (Int32 i = 0; i < nb_sender - 1; ++i) {
    const Int32 rank = i + writer + 1;
    const Int64 sizeof_recved = nb_elem_recved[i] * sizeof_elem;
 
    ArrayView<Byte> recv_elem = out.subSpan(old_size, sizeof_recved).smallView();
    requests[i] = pm->recv(recv_elem, rank, false);
 
    old_size += sizeof_recved;
  }
  pm->waitAllRequests(requests);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void GatherGroup::
setGatherGroupInfo(GatherGroupInfo* gather_group_info)
{
  ARCANE_CHECK_POINTER(gather_group_info);
  ARCANE_FATAL_IF(!gather_group_info->isComputed(), "GatherGroupInfo is not computed");
  m_gather_group_info = gather_group_info;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
GatherGroupInfo::
GatherGroupInfo(IParallelMng* parallel_mng, bool use_collective_io)
: m_parallel_mng(parallel_mng)
, m_use_collective_io(use_collective_io)
{}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
GatherGroupInfo::
~GatherGroupInfo() = default;
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void GatherGroupInfo::
computeSize(Int32 nb_elem_in)
{
  if (m_is_computed)
    return;
  m_is_computed = true;
 
  if (m_use_collective_io) {
    m_writer = m_parallel_mng->_internalApi()->masterParallelIORank();
    m_nb_sender_to_writer = m_parallel_mng->_internalApi()->nbSendersToMasterParallelIO();
  }
  else {
    m_writer = m_parallel_mng->masterIORank();
    m_nb_sender_to_writer = m_parallel_mng->commSize();
  }
 
  // If sequential or MPI + MPI-IO.
  if ((m_parallel_mng->commSize() == 1) || (!m_parallel_mng->isThreadImplementation() && m_use_collective_io)) {
    m_nb_elem_output = nb_elem_in;
    m_nb_writer_global = m_parallel_mng->commSize();
    return;
  }
 
  if (m_writer != m_parallel_mng->commRank()) {
    m_parallel_mng->send({ 1, &nb_elem_in }, m_writer);
    m_nb_elem_output = 0;
  }
  else {
    m_nb_elem_recv.resizeNoInit(m_nb_sender_to_writer - 1);
 
    {
      UniqueArray<Parallel::Request> requests(m_nb_sender_to_writer - 1);
      for (Int32 i = 0; i < m_nb_sender_to_writer - 1; ++i) {
        const Int32 rank = i + m_writer + 1;
        requests[i] = m_parallel_mng->recv({ 1, &m_nb_elem_recv[i] }, rank, false);
      }
      m_parallel_mng->waitAllRequests(requests);
    }
 
    m_nb_elem_output = nb_elem_in;
    for (const Int32 size : m_nb_elem_recv) {
      m_nb_elem_output += size;
    }
  }
 
  m_nb_writer_global = m_parallel_mng->reduce(MessagePassing::ReduceSum, (m_writer == m_parallel_mng->commRank()));
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
SmallSpan<Int32> GatherGroupInfo::
nbElemRecvGatherToMasterIO()
{
  return m_nb_elem_recv.smallSpan();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
} // End namespace Arcane
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/