MpiParallelMng.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
//-----------------------------------------------------------------------------
/*---------------------------------------------------------------------------*/
/* MpiParallelMng.cc                                           (C) 2000-2026 */
/*                                                                           */
/* Parallelism manager using MPI.                                            */
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#include "arcane/utils/Collection.h"
#include "arcane/utils/Enumerator.h"
#include "arcane/utils/ScopedPtr.h"
#include "arcane/utils/PlatformUtils.h"
#include "arcane/utils/TimeoutException.h"
#include "arcane/utils/NotImplementedException.h"
#include "arcane/utils/ArgumentException.h"
#include "arcane/utils/ITraceMng.h"
#include "arcane/utils/ValueConvert.h"
#include "arcane/utils/Exception.h"
#include "arcane/utils/HPReal.h"
 
#include "arcane/core/IIOMng.h"
#include "arcane/core/Timer.h"
#include "arcane/core/IItemFamily.h"
#include "arcane/core/IParallelTopology.h"
#include "arcane/core/parallel/IStat.h"
#include "arcane/core/internal/SerializeMessage.h"
#include "arcane/core/internal/ParallelMngInternal.h"
#include "arcane/core/internal/MachineShMemWinMemoryAllocator.h"
 
#include "arcane/parallel/mpi/MpiParallelMng.h"
#include "arcane/parallel/mpi/MpiParallelDispatch.h"
#include "arcane/parallel/mpi/MpiTimerMng.h"
#include "arcane/parallel/mpi/MpiSerializeMessage.h"
#include "arcane/parallel/mpi/MpiParallelNonBlockingCollective.h"
#include "arcane/parallel/mpi/MpiDatatype.h"
#include "arcane/parallel/mpi/IVariableSynchronizerMpiCommunicator.h"
 
#include "arcane/impl/ParallelReplication.h"
#include "arcane/impl/SequentialParallelMng.h"
#include "arcane/impl/internal/ParallelMngUtilsFactoryBase.h"
#include "arcane/impl/internal/VariableSynchronizer.h"
 
#include "arccore/message_passing_mpi/internal/MpiMessagePassingMng.h"
#include "arccore/message_passing_mpi/internal/MpiSerializeDispatcher.h"
#include "arccore/message_passing_mpi/internal/MpiRequestList.h"
#include "arccore/message_passing_mpi/internal/MpiAdapter.h"
#include "arccore/message_passing_mpi/internal/MpiLock.h"
#include "arccore/message_passing_mpi/internal/MpiMachineShMemWinBaseInternalCreator.h"
#include "arccore/message_passing_mpi/internal/MpiContigMachineShMemWinBaseInternal.h"
#include "arccore/message_passing_mpi/internal/MpiMachineShMemWinBaseInternal.h"
#include "arccore/message_passing/Dispatchers.h"
#include "arccore/message_passing/Messages.h"
#include "arccore/message_passing/internal/SerializeMessageList.h"
 
#include "arcane_packages.h"
 
//#define ARCANE_TRACE_MPI
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
namespace Arcane
{
using namespace Arcane::MessagePassing;
using namespace Arcane::MessagePassing::Mpi;
using BasicSerializeMessage = Arcane::MessagePassing::internal::BasicSerializeMessage;
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
extern "C++" IIOMng*
arcaneCreateIOMng(IParallelMng* psm);
 
#if defined(ARCANE_HAS_MPI_NEIGHBOR)
// Defined in MpiNeighborVariableSynchronizeDispatcher
extern "C++" Ref<IDataSynchronizeImplementationFactory>
arcaneCreateMpiNeighborVariableSynchronizerFactory(MpiParallelMng* mpi_pm,
                                                   Ref<IVariableSynchronizerMpiCommunicator> synchronizer_communicator);
#endif
extern "C++" Ref<IDataSynchronizeImplementationFactory>
arcaneCreateMpiBlockVariableSynchronizerFactory(MpiParallelMng* mpi_pm, Int32 block_size, Int32 nb_sequence);
extern "C++" Ref<IDataSynchronizeImplementationFactory>
arcaneCreateMpiVariableSynchronizerFactory(MpiParallelMng* mpi_pm);
extern "C++" Ref<IDataSynchronizeImplementationFactory>
arcaneCreateMpiDirectSendrecvVariableSynchronizerFactory(MpiParallelMng* mpi_pm);
extern "C++" Ref<IDataSynchronizeImplementationFactory>
arcaneCreateMpiLegacyVariableSynchronizerFactory(MpiParallelMng* mpi_pm);
#if defined(ARCANE_HAS_PACKAGE_NCCL)
extern "C++" Ref<IDataSynchronizeImplementationFactory>
arcaneCreateNCCLVariableSynchronizerFactory(IParallelMng* mpi_pm);
#endif
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
MpiParallelMngBuildInfo::
MpiParallelMngBuildInfo(MPI_Comm comm, MPI_Comm machine_comm)
: is_parallel(false)
, comm_rank(MessagePassing::A_NULL_RANK)
, comm_nb_rank(0)
, mpi_comm(comm)
, mpi_machine_comm(machine_comm)
, is_mpi_comm_owned(true)
{
  ::MPI_Comm_rank(comm, &comm_rank);
  ::MPI_Comm_size(comm, &comm_nb_rank);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
class VariableSynchronizerMpiCommunicator
: public IVariableSynchronizerMpiCommunicator
{
 public:
 
  explicit VariableSynchronizerMpiCommunicator(MpiParallelMng* pm)
  : m_mpi_parallel_mng(pm)
  {}
  ~VariableSynchronizerMpiCommunicator() override
  {
    _checkFreeCommunicator();
  }
  MPI_Comm communicator() const override
  {
    return m_topology_communicator;
  }
  void compute(VariableSynchronizer* var_syncer) override
  {
    Int32ConstArrayView comm_ranks = var_syncer->communicatingRanks();
    const Int32 nb_message = comm_ranks.size();
 
    MpiParallelMng* pm = m_mpi_parallel_mng;
 
    MPI_Comm old_comm = pm->communicator();
 
    UniqueArray<int> destinations(nb_message);
    for (Integer i = 0; i < nb_message; ++i) {
      destinations[i] = comm_ranks[i];
    }
 
    _checkFreeCommunicator();
 
    int r = MPI_Dist_graph_create_adjacent(old_comm, nb_message, destinations.data(), MPI_UNWEIGHTED,
                                           nb_message, destinations.data(), MPI_UNWEIGHTED,
                                           MPI_INFO_NULL, 0, &m_topology_communicator);
 
    if (r != MPI_SUCCESS)
      ARCANE_FATAL("Error '{0}' in MPI_Dist_graph_create", r);
 
    // Checks that the rank order for the MPI implementation is the same as the one we have in
    // the VariableSynchronizer.
    {
      int indegree = 0;
      int outdegree = 0;
      int weighted = 0;
      MPI_Dist_graph_neighbors_count(m_topology_communicator, &indegree, &outdegree, &weighted);
 
      if (indegree != nb_message)
        ARCANE_FATAL("Bad value '{0}' for 'indegree' (expected={1})", indegree, nb_message);
      if (outdegree != nb_message)
        ARCANE_FATAL("Bad value '{0}' for 'outdegree' (expected={1})", outdegree, nb_message);
 
      UniqueArray<int> srcs(indegree);
      UniqueArray<int> dsts(outdegree);
 
      MPI_Dist_graph_neighbors(m_topology_communicator, indegree, srcs.data(), MPI_UNWEIGHTED, outdegree, dsts.data(), MPI_UNWEIGHTED);
 
      for (int k = 0; k < outdegree; ++k) {
        int x = dsts[k];
        if (x != comm_ranks[k])
          ARCANE_FATAL("Invalid destination rank order k={0} v={1} expected={2}", k, x, comm_ranks[k]);
      }
 
      for (int k = 0; k < indegree; ++k) {
        int x = srcs[k];
        if (x != comm_ranks[k])
          ARCANE_FATAL("Invalid source rank order k={0} v={1} expected={2}", k, x, comm_ranks[k]);
      }
    }
  }
 
 private:
 
  MpiParallelMng* m_mpi_parallel_mng = nullptr;
  MPI_Comm m_topology_communicator = MPI_COMM_NULL;
 
 private:
 
  void _checkFreeCommunicator()
  {
    if (m_topology_communicator != MPI_COMM_NULL)
      MPI_Comm_free(&m_topology_communicator);
    m_topology_communicator = MPI_COMM_NULL;
  }
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
class MpiVariableSynchronizer
: public VariableSynchronizer
{
 public:
 
  MpiVariableSynchronizer(IParallelMng* pm, const ItemGroup& group,
                          Ref<IDataSynchronizeImplementationFactory> implementation_factory,
                          Ref<IVariableSynchronizerMpiCommunicator> topology_info)
  : VariableSynchronizer(pm, group, implementation_factory)
  , m_topology_info(topology_info)
  {
  }
 
 public:
 
  void compute() override
  {
    VariableSynchronizer::compute();
    // If not null, calculate the topology
    if (m_topology_info.get())
      m_topology_info->compute(this);
  }
 
 private:
 
  Ref<IVariableSynchronizerMpiCommunicator> m_topology_info;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
class MpiParallelMngUtilsFactory
: public ParallelMngUtilsFactoryBase
{
 public:
 
  MpiParallelMngUtilsFactory()
  : m_synchronizer_version(2)
  {
    if (platform::getEnvironmentVariable("ARCANE_SYNCHRONIZE_VERSION") == "1")
      m_synchronizer_version = 1;
    if (platform::getEnvironmentVariable("ARCANE_SYNCHRONIZE_VERSION") == "2")
      m_synchronizer_version = 2;
    if (platform::getEnvironmentVariable("ARCANE_SYNCHRONIZE_VERSION") == "3")
      m_synchronizer_version = 3;
    if (platform::getEnvironmentVariable("ARCANE_SYNCHRONIZE_VERSION") == "4") {
      m_synchronizer_version = 4;
      String v = platform::getEnvironmentVariable("ARCANE_SYNCHRONIZE_BLOCK_SIZE");
      if (!v.null()) {
        Int32 block_size = 0;
        if (!builtInGetValue(block_size, v))
          m_synchronize_block_size = block_size;
        m_synchronize_block_size = std::clamp(m_synchronize_block_size, 0, 1000000000);
      }
      v = platform::getEnvironmentVariable("ARCANE_SYNCHRONIZE_NB_SEQUENCE");
      if (!v.null()) {
        Int32 nb_sequence = 0;
        if (!builtInGetValue(nb_sequence, v))
          m_synchronize_nb_sequence = nb_sequence;
        m_synchronize_nb_sequence = std::clamp(m_synchronize_nb_sequence, 1, 1024 * 1024);
      }
    }
    if (platform::getEnvironmentVariable("ARCANE_SYNCHRONIZE_VERSION") == "5")
      m_synchronizer_version = 5;
    if (platform::getEnvironmentVariable("ARCANE_SYNCHRONIZE_VERSION") == "6")
      m_synchronizer_version = 6;
  }
 
 public:
 
  Ref<IVariableSynchronizer> createSynchronizer(IParallelMng* pm, IItemFamily* family) override
  {
    return _createSynchronizer(pm, family->allItems());
  }
 
  Ref<IVariableSynchronizer> createSynchronizer(IParallelMng* pm, const ItemGroup& group) override
  {
    return _createSynchronizer(pm, group);
  }
 
 private:
 
  Ref<IVariableSynchronizer> _createSynchronizer(IParallelMng* pm, const ItemGroup& group)
  {
    Ref<IVariableSynchronizerMpiCommunicator> topology_info;
    MpiParallelMng* mpi_pm = ARCANE_CHECK_POINTER(dynamic_cast<MpiParallelMng*>(pm));
    ITraceMng* tm = pm->traceMng();
    Ref<IDataSynchronizeImplementationFactory> generic_factory;
    // Only displays information for the group of all cells to avoid displaying
    // the same message multiple times.
    bool do_print = (group.isAllItems() && group.itemKind() == IK_Cell);
    if (m_synchronizer_version == 2) {
      if (do_print)
        tm->info() << "Using MpiSynchronizer V2";
      generic_factory = arcaneCreateMpiVariableSynchronizerFactory(mpi_pm);
    }
    else if (m_synchronizer_version == 3) {
      if (do_print)
        tm->info() << "Using MpiSynchronizer V3";
      generic_factory = arcaneCreateMpiDirectSendrecvVariableSynchronizerFactory(mpi_pm);
    }
    else if (m_synchronizer_version == 4) {
      if (do_print)
        tm->info() << "Using MpiSynchronizer V4 block_size=" << m_synchronize_block_size
                   << " nb_sequence=" << m_synchronize_nb_sequence;
      generic_factory = arcaneCreateMpiBlockVariableSynchronizerFactory(mpi_pm, m_synchronize_block_size, m_synchronize_nb_sequence);
    }
    else if (m_synchronizer_version == 5) {
      if (do_print)
        tm->info() << "Using MpiSynchronizer V5";
      topology_info = createRef<VariableSynchronizerMpiCommunicator>(mpi_pm);
#if defined(ARCANE_HAS_MPI_NEIGHBOR)
      generic_factory = arcaneCreateMpiNeighborVariableSynchronizerFactory(mpi_pm, topology_info);
#else
      throw NotSupportedException(A_FUNCINFO, "Synchronize implementation V5 is not supported with this version of MPI");
#endif
    }
#if defined(ARCANE_HAS_PACKAGE_NCCL)
    else if (m_synchronizer_version == 6) {
      if (do_print)
        tm->info() << "Using NCCLSynchronizer";
      generic_factory = arcaneCreateNCCLVariableSynchronizerFactory(mpi_pm);
    }
#endif
    else {
      if (do_print)
        tm->info() << "Using MpiSynchronizer V1";
      generic_factory = arcaneCreateMpiLegacyVariableSynchronizerFactory(mpi_pm);
    }
    if (!generic_factory.get())
      ARCANE_FATAL("No factory created");
    return createRef<MpiVariableSynchronizer>(pm, group, generic_factory, topology_info);
  }
 
 private:
 
  Integer m_synchronizer_version = 1;
  Int32 m_synchronize_block_size = 32000;
  Int32 m_synchronize_nb_sequence = 1;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
class MpiParallelMng::Impl
: public ParallelMngInternal
{
 public:
 
  explicit Impl(MpiParallelMng* pm)
  : ParallelMngInternal(pm)
  , m_parallel_mng(pm)
  , m_alloc(makeRef(new MachineShMemWinMemoryAllocator(pm)))
  {}
 
  ~Impl() override = default;
 
 public:
 
  Int32 masterParallelIORank() const override { return m_parallel_mng->commRank(); }
  Int32 nbSendersToMasterParallelIO() const override { return 1; }
 
  void initializeWindowCreator() override
  {
    m_parallel_mng->traceMng()->info() << "initializeWindowCreator() MPI";
    m_parallel_mng->adapter()->initializeWindowCreator(m_parallel_mng->machineCommunicator());
  }
 
  bool isMachineShMemWinAvailable() override
  {
    if (m_shmem_available == 1) {
      return true;
    }
 
    if (m_shmem_available == 0) {
      Ref<IParallelTopology> topo = m_parallel_mng->_internalUtilsFactory()->createTopology(m_parallel_mng);
      if (topo->machineRanks().size() == m_parallel_mng->adapter()->windowCreator()->machineRanks().size()) {
        m_shmem_available = 1;
        return true;
      }
      // Problem with MPI. May occur if MPICH is compiled in ch3:sock mode.
      m_shmem_available = 2;
      return false;
    }
 
    return false;
  }
 
  Ref<IContigMachineShMemWinBaseInternal> createContigMachineShMemWinBase(Int64 sizeof_segment, Int32 sizeof_type) override
  {
    return makeRef(m_parallel_mng->adapter()->windowCreator()->createWindow(sizeof_segment, sizeof_type));
  }
 
  Ref<IMachineShMemWinBaseInternal> createMachineShMemWinBase(Int64 sizeof_segment, Int32 sizeof_type) override
  {
    return makeRef(m_parallel_mng->adapter()->windowCreator()->createDynamicWindow(sizeof_segment, sizeof_type));
  }
 
  MemoryAllocationOptions machineShMemWinMemoryAllocator() override
  {
    return MemoryAllocationOptions{ m_alloc.get() };
  }
 
  ConstArrayView<Int32> machineRanks() override
  {
    return m_parallel_mng->adapter()->windowCreator()->machineRanks();
  }
 
  void machineBarrier() override
  {
    m_parallel_mng->adapter()->windowCreator()->machineBarrier();
  }
 
 private:
 
  MpiParallelMng* m_parallel_mng;
  Ref<MachineShMemWinMemoryAllocator> m_alloc;
 
  // 0 = Attribute not initialized
  // 1 = Shared memory available
  // 2 = Shared memory not available
  Int8 m_shmem_available = 0;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
MpiParallelMng::
MpiParallelMng(const MpiParallelMngBuildInfo& bi)
: ParallelMngDispatcher(ParallelMngDispatcherBuildInfo(bi.commRank(),bi.commSize(),MP::Communicator(bi.mpiComm())))
, m_trace(bi.trace_mng)
, m_thread_mng(bi.thread_mng)
, m_world_parallel_mng(bi.world_parallel_mng)
, m_timer_mng(bi.timer_mng)
, m_replication(new ParallelReplication())
, m_is_parallel(bi.is_parallel)
, m_comm_rank(bi.commRank())
, m_comm_size(bi.commSize())
, m_stat(bi.stat)
, m_communicator(bi.mpiComm())
, m_machine_communicator(bi.mpiMachineComm())
, m_is_communicator_owned(bi.is_mpi_comm_owned)
, m_mpi_lock(bi.mpi_lock)
, m_non_blocking_collective(nullptr)
, m_utils_factory(createRef<MpiParallelMngUtilsFactory>())
, m_parallel_mng_internal(new Impl(this))
{
  if (!m_world_parallel_mng) {
    m_trace->debug() << "[MpiParallelMng] No m_world_parallel_mng found, reverting to ourselves!";
    m_world_parallel_mng = this;
  }
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
MpiParallelMng::
~MpiParallelMng()
{
  delete m_parallel_mng_internal;
  delete m_non_blocking_collective;
  m_sequential_parallel_mng.reset();
  if (m_is_communicator_owned) {
    MpiLock::Section ls(m_mpi_lock);
    MPI_Comm_free(&m_communicator);
    MPI_Comm_free(&m_machine_communicator);
  }
  delete m_replication;
  delete m_io_mng;
  if (m_is_timer_owned)
    delete m_timer_mng;
  arcaneCallFunctionAndTerminateIfThrow([&]() { m_adapter->destroy(); });
  delete m_datatype_list;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
namespace
{
 
  /*---------------------------------------------------------------------------*/
  /*---------------------------------------------------------------------------*/
  // Class to create the different dispatchers
  class DispatchCreator
  {
   public:
 
    DispatchCreator(ITraceMng* tm, IMessagePassingMng* mpm, MpiAdapter* adapter, MpiDatatypeList* datatype_list)
    : m_tm(tm)
    , m_mpm(mpm)
    , m_adapter(adapter)
    , m_datatype_list(datatype_list)
    {}
 
   public:
 
    template <typename DataType> MpiParallelDispatchT<DataType>*
    create()
    {
      MpiDatatype* dt = m_datatype_list->datatype(DataType());
      return new MpiParallelDispatchT<DataType>(m_tm, m_mpm, m_adapter, dt);
    }
 
    ITraceMng* m_tm;
    IMessagePassingMng* m_mpm;
    MpiAdapter* m_adapter;
    MpiDatatypeList* m_datatype_list;
  };
 
  /*---------------------------------------------------------------------------*/
  /*---------------------------------------------------------------------------*/
 
  class ControlDispatcherDecorator
  : public ParallelMngDispatcher::DefaultControlDispatcher
  {
   public:
 
    ControlDispatcherDecorator(IParallelMng* pm, MpiAdapter* adapter)
    : ParallelMngDispatcher::DefaultControlDispatcher(pm)
    , m_adapter(adapter)
    {}
 
    IMessagePassingMng* commSplit(bool keep) override
    {
      return m_adapter->commSplit(keep);
    }
    MP::IProfiler* profiler() const override { return m_adapter->profiler(); }
    void setProfiler(MP::IProfiler* p) override { m_adapter->setProfiler(p); }
 
   private:
 
    MpiAdapter* m_adapter;
  };
} // namespace
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void MpiParallelMng::
build()
{
  ITraceMng* tm = traceMng();
  if (!m_timer_mng) {
    m_timer_mng = new MpiTimerMng(tm);
    m_is_timer_owned = true;
  }
 
  // Created the associated sequential manager.
  {
    SequentialParallelMngBuildInfo bi(timerMng(), worldParallelMng());
    bi.setTraceMng(traceMng());
    bi.setCommunicator(communicator());
    bi.setThreadMng(threadMng());
    m_sequential_parallel_mng = arcaneCreateSequentialParallelMngRef(bi);
  }
 
  // Indicates that reduces must be performed in processor order
  // in order to guarantee deterministic execution
  bool is_ordered_reduce = false;
  if (platform::getEnvironmentVariable("ARCANE_ORDERED_REDUCE") == "TRUE")
    is_ordered_reduce = true;
  m_datatype_list = new MpiDatatypeList(is_ordered_reduce);
 
  ARCANE_CHECK_POINTER(m_stat);
 
  MpiAdapter* adapter = new MpiAdapter(m_trace, m_stat->toArccoreStat(), m_communicator, m_mpi_lock);
  m_adapter = adapter;
  auto mpm = _messagePassingMng();
 
  // NOTE: this instance will be destroyed by the ParallelMngDispatcher
  auto* control_dispatcher = new ControlDispatcherDecorator(this, m_adapter);
  _setControlDispatcher(control_dispatcher);
 
  // NOTE: this instance will be destroyed by the ParallelMngDispatcher
  auto* serialize_dispatcher = new MpiSerializeDispatcher(m_adapter, mpm);
  m_mpi_serialize_dispatcher = serialize_dispatcher;
  _setSerializeDispatcher(serialize_dispatcher);
 
  DispatchCreator creator(m_trace, mpm, m_adapter, m_datatype_list);
  this->createDispatchers(creator);
 
  m_io_mng = arcaneCreateIOMng(this);
 
  m_non_blocking_collective = new MpiParallelNonBlockingCollective(tm, this, adapter);
  m_non_blocking_collective->build();
  if (m_mpi_lock)
    m_trace->info() << "Using mpi with locks.";
 
  m_parallel_mng_internal->initializeWindowCreator();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*----------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void MpiParallelMng::
initialize()
{
  Trace::Setter mci(m_trace, "Mpi");
  if (m_is_initialized) {
    m_trace->warning() << "MpiParallelMng already initialized";
    return;
  }
 
  // Initialization of MpiParallelMng
  m_trace->info() << "Initialisation de MpiParallelMng";
  m_sequential_parallel_mng->initialize();
 
  m_adapter->setTimeMetricCollector(timeMetricCollector());
 
  m_is_initialized = true;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void MpiParallelMng::
sendSerializer(ISerializer* s, Int32 rank)
{
  Trace::Setter mci(m_trace, "Mpi");
  Timer::Phase tphase(timeStats(), TP_Communication);
  MessageTag mpi_tag = BasicSerializeMessage::defaultTag();
  m_mpi_serialize_dispatcher->legacySendSerializer(s, { MessageRank(rank), mpi_tag, Blocking });
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
ISerializeMessage* MpiParallelMng::
createSendSerializer(Int32 rank)
{
  return m_utils_factory->createSendSerializeMessage(this, rank)._release();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Request MpiParallelMng::
sendSerializer(ISerializer* s, Int32 rank, [[maybe_unused]] ByteArray& bytes)
{
  Trace::Setter mci(m_trace, "Mpi");
  Timer::Phase tphase(timeStats(), TP_Communication);
  MessageTag mpi_tag = BasicSerializeMessage::defaultTag();
  return m_mpi_serialize_dispatcher->legacySendSerializer(s, { MessageRank(rank), mpi_tag, NonBlocking });
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void MpiParallelMng::
broadcastSerializer(ISerializer* values, Int32 rank)
{
  Timer::Phase tphase(timeStats(), TP_Communication);
  m_mpi_serialize_dispatcher->broadcastSerializer(values, MessageRank(rank));
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void MpiParallelMng::
recvSerializer(ISerializer* values, Int32 rank)
{
  Trace::Setter mci(m_trace, "Mpi");
  Timer::Phase tphase(timeStats(), TP_Communication);
  MessageTag mpi_tag = BasicSerializeMessage::defaultTag();
  m_mpi_serialize_dispatcher->legacyReceiveSerializer(values, MessageRank(rank), mpi_tag);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
ISerializeMessage* MpiParallelMng::
createReceiveSerializer(Int32 rank)
{
  return m_utils_factory->createReceiveSerializeMessage(this, rank)._release();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
auto MpiParallelMng::
probe(const PointToPointMessageInfo& message) -> MessageId
{
  return m_adapter->probeMessage(message);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
auto MpiParallelMng::
legacyProbe(const PointToPointMessageInfo& message) -> MessageSourceInfo
{
  return m_adapter->legacyProbeMessage(message);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Request MpiParallelMng::
sendSerializer(const ISerializer* s, const PointToPointMessageInfo& message)
{
  return m_mpi_serialize_dispatcher->sendSerializer(s, message);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Request MpiParallelMng::
receiveSerializer(ISerializer* s, const PointToPointMessageInfo& message)
{
  return m_mpi_serialize_dispatcher->receiveSerializer(s, message);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void MpiParallelMng::
freeRequests(ArrayView<Parallel::Request> requests)
{
  for (Integer i = 0, is = requests.size(); i < is; ++i)
    m_adapter->freeRequest(requests[i]);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void MpiParallelMng::
_checkFinishedSubRequests()
{
  m_mpi_serialize_dispatcher->checkFinishedSubRequests();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Ref<IParallelMng> MpiParallelMng::
sequentialParallelMngRef()
{
  return m_sequential_parallel_mng;
}
 
IParallelMng* MpiParallelMng::
sequentialParallelMng()
{
  return m_sequential_parallel_mng.get();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void MpiParallelMng::
printStats()
{
  if (m_stat)
    m_stat->print(m_trace);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void MpiParallelMng::
barrier()
{
  traceMng()->flush();
  m_adapter->barrier();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void MpiParallelMng::
waitAllRequests(ArrayView<Request> requests)
{
  m_adapter->waitAllRequests(requests);
  _checkFinishedSubRequests();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
UniqueArray<Integer> MpiParallelMng::
waitSomeRequests(ArrayView<Request> requests)
{
  return _waitSomeRequests(requests, false);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
UniqueArray<Integer> MpiParallelMng::
testSomeRequests(ArrayView<Request> requests)
{
  return _waitSomeRequests(requests, true);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
UniqueArray<Integer> MpiParallelMng::
_waitSomeRequests(ArrayView<Request> requests, bool is_non_blocking)
{
  UniqueArray<Integer> results;
  UniqueArray<bool> done_indexes(requests.size());
 
  m_adapter->waitSomeRequests(requests, done_indexes, is_non_blocking);
  for (int i = 0; i < requests.size(); i++) {
    if (done_indexes[i])
      results.add(i);
  }
  return results;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
ISerializeMessageList* MpiParallelMng::
_createSerializeMessageList()
{
  return new MP::internal::SerializeMessageList(messagePassingMng());
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
IGetVariablesValuesParallelOperation* MpiParallelMng::
createGetVariablesValuesOperation()
{
  return m_utils_factory->createGetVariablesValuesOperation(this)._release();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
ITransferValuesParallelOperation* MpiParallelMng::
createTransferValuesOperation()
{
  return m_utils_factory->createTransferValuesOperation(this)._release();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
IParallelExchanger* MpiParallelMng::
createExchanger()
{
  return m_utils_factory->createExchanger(this)._release();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
IVariableSynchronizer* MpiParallelMng::
createSynchronizer(IItemFamily* family)
{
  return m_utils_factory->createSynchronizer(this, family)._release();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
IVariableSynchronizer* MpiParallelMng::
createSynchronizer(const ItemGroup& group)
{
  return m_utils_factory->createSynchronizer(this, group)._release();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
IParallelTopology* MpiParallelMng::
createTopology()
{
  return m_utils_factory->createTopology(this)._release();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
IParallelReplication* MpiParallelMng::
replication() const
{
  return m_replication;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void MpiParallelMng::
setReplication(IParallelReplication* v)
{
  delete m_replication;
  m_replication = v;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
IParallelMng* MpiParallelMng::
_createSubParallelMng(MPI_Comm sub_communicator)
{
  // If null, this rank is not part of the sub-communicator
  if (sub_communicator == MPI_COMM_NULL)
    return nullptr;
 
  int sub_rank = -1;
  MPI_Comm_rank(sub_communicator, &sub_rank);
 
  MPI_Comm sub_machine_communicator = MPI_COMM_NULL;
  MPI_Comm_split_type(sub_communicator, MPI_COMM_TYPE_SHARED, sub_rank, MPI_INFO_NULL, &sub_machine_communicator);
 
  MpiParallelMngBuildInfo bi(sub_communicator, sub_machine_communicator);
  bi.is_parallel = isParallel();
  bi.stat = m_stat;
  bi.timer_mng = m_timer_mng;
  bi.thread_mng = m_thread_mng;
  bi.trace_mng = m_trace;
  bi.world_parallel_mng = m_world_parallel_mng;
  bi.mpi_lock = m_mpi_lock;
 
  IParallelMng* sub_pm = new MpiParallelMng(bi);
  sub_pm->build();
  return sub_pm;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Ref<IParallelMng> MpiParallelMng::
_createSubParallelMngRef(Int32 color, Int32 key)
{
  if (color < 0)
    color = MPI_UNDEFINED;
  MPI_Comm sub_communicator = MPI_COMM_NULL;
  MPI_Comm_split(m_communicator, color, key, &sub_communicator);
  IParallelMng* sub_pm = _createSubParallelMng(sub_communicator);
  return makeRef(sub_pm);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
IParallelMng* MpiParallelMng::
_createSubParallelMng(Int32ConstArrayView kept_ranks)
{
  MPI_Group mpi_group = MPI_GROUP_NULL;
  MPI_Comm_group(m_communicator, &mpi_group);
  Integer nb_sub_rank = kept_ranks.size();
  UniqueArray<int> mpi_kept_ranks(nb_sub_rank);
  for (Integer i = 0; i < nb_sub_rank; ++i)
    mpi_kept_ranks[i] = (int)kept_ranks[i];
 
  MPI_Group final_group = MPI_GROUP_NULL;
  MPI_Group_incl(mpi_group, nb_sub_rank, mpi_kept_ranks.data(), &final_group);
  MPI_Comm sub_communicator = MPI_COMM_NULL;
 
  MPI_Comm_create(m_communicator, final_group, &sub_communicator);
  MPI_Group_free(&final_group);
  return _createSubParallelMng(sub_communicator);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/

class MpiParallelMng::RequestList
: public MpiRequestList
{
  using Base = MpiRequestList;
 
 public:
 
  explicit RequestList(MpiParallelMng* pm)
  : Base(pm->m_adapter)
  , m_parallel_mng(pm)
  {}
 
 public:
 
  void _wait(Parallel::eWaitType wait_type) override
  {
    Base::_wait(wait_type);
    m_parallel_mng->_checkFinishedSubRequests();
  };
 
 private:
 
  MpiParallelMng* m_parallel_mng;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Ref<IRequestList> MpiParallelMng::
createRequestListRef()
{
  return createRef<RequestList>(this);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Ref<IParallelMngUtilsFactory> MpiParallelMng::
_internalUtilsFactory() const
{
  return m_utils_factory;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
bool MpiParallelMng::
_isAcceleratorAware() const
{
  return arcaneIsAcceleratorAwareMPI();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
} // End namespace Arcane
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/