HybridParallelSuperMng.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
//-----------------------------------------------------------------------------
/*---------------------------------------------------------------------------*/
/* HybridParallelSuperMng.cc                                   (C) 2000-2026 */
/*                                                                           */
/* Parallelism manager using MPI and shared memory.                          */
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#include "arcane/utils/ApplicationInfo.h"
#include "arcane/utils/NullThreadMng.h"
#include "arcane/utils/FatalErrorException.h"
#include "arcane/utils/ITraceMng.h"
#include "arcane/utils/PlatformUtils.h"
#include "arcane/utils/ValueConvert.h"
#include "arcane/utils/TraceClassConfig.h"
#include "arcane/utils/ArgumentException.h"
#include "arcane/utils/CommandLineArguments.h"
 
#include "arcane/parallel/IStat.h"
 
#include "arcane/parallel/mpi/MpiParallelMng.h"
#include "arcane/parallel/mpi/MpiParallelDispatch.h"
#include "arcane/parallel/mpi/MpiErrorHandler.h"
 
#include "arccore/message_passing_mpi/internal/MpiAdapter.h"
#include "arccore/message_passing_mpi/internal/MpiLock.h"
 
#include "arcane/parallel/thread/SharedMemoryMessageQueue.h"
#include "arcane/parallel/thread/SharedMemoryParallelMng.h"
#include "arcane/parallel/thread/SharedMemoryParallelSuperMng.h"
#include "arcane/parallel/thread/internal/SharedMemoryThreadMng.h"
 
#include "arcane/parallel/mpithread/HybridParallelMng.h"
#include "arcane/parallel/mpithread/HybridParallelDispatch.h"
#include "arcane/parallel/mpithread/internal/HybridMachineShMemWinBaseInternalCreator.h"
 
#include "arcane/core/FactoryService.h"
#include "arcane/core/AbstractService.h"
#include "arcane/core/IApplication.h"
#include "arcane/core/ParallelSuperMngDispatcher.h"
#include "arcane/core/ApplicationBuildInfo.h"
#include "arcane/core/ServiceBuilder.h"
 
#include "arcane/core/IMainFactory.h"
 
#include "arccore/base/ReferenceCounterImpl.h"
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
namespace Arcane::MessagePassing
{
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/

class HybridParallelMngContainer
: public ParallelMngContainerBase
{
 public:
 
  HybridParallelMngContainer(IApplication* app, Int32 nb_local_rank,
                             MP::Communicator mpi_comm, MP::Communicator mpi_machine_comm,
                             IParallelMngContainerFactory* factory,
                             Parallel::IStat* stat, MpiLock* mpi_lock);
  ~HybridParallelMngContainer() override;
 
 public:
 
  void build();
  Ref<IParallelMng> _createParallelMng(Int32 local_rank, ITraceMng* tm) override;
 
 public:
 
  IApplication* m_application; 
  Parallel::IStat* m_stat = nullptr; 
  IThreadMng* m_thread_mng = nullptr;
  MpiLock* m_mpi_lock = nullptr;
  ISharedMemoryMessageQueue* m_message_queue = nullptr;
  IThreadBarrier* m_thread_barrier = nullptr;
  Int32 m_local_nb_rank = -1;
  MpiThreadAllDispatcher* m_all_dispatchers = nullptr;
  // This object is shared by all HybridParallelMngs.
  UniqueArray<HybridParallelMng*>* m_parallel_mng_list = nullptr;
  Mutex* m_internal_create_mutex = nullptr;
  IParallelMngContainerFactory* m_sub_builder_factory = nullptr;
  HybridMachineShMemWinBaseInternalCreator* m_window_creator = nullptr;
 
 private:
 
  MPI_Comm m_mpi_communicator; 
  Int32 m_mpi_comm_rank = -1; 
  Int32 m_mpi_comm_size = -1; 
  MPI_Comm m_mpi_machine_communicator; 
 
 private:
 
  void _setMPICommunicator();
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
HybridParallelMngContainer::
HybridParallelMngContainer(IApplication* app, Int32 nb_local_rank,
                           MP::Communicator mpi_comm, MP::Communicator mpi_machine_comm,
                           IParallelMngContainerFactory* factory,
                           Parallel::IStat* stat, MpiLock* mpi_lock)
: m_application(app)
, m_stat(stat)
, m_thread_mng(new SharedMemoryThreadMng())
, m_mpi_lock(mpi_lock)
, m_local_nb_rank(nb_local_rank)
, m_parallel_mng_list(new UniqueArray<HybridParallelMng*>())
, m_sub_builder_factory(factory)
, m_mpi_communicator(mpi_comm)
, m_mpi_machine_communicator(mpi_machine_comm)
{
  _setMPICommunicator();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
HybridParallelMngContainer::
~HybridParallelMngContainer()
{
  // TODO: check if the communicator needs to be destroyed
  m_thread_barrier->destroy();
  delete m_message_queue;
  delete m_thread_mng;
  delete m_all_dispatchers;
  delete m_parallel_mng_list;
  delete m_internal_create_mutex;
  delete m_window_creator;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void HybridParallelMngContainer::
build()
{
  m_internal_create_mutex = new Mutex();
 
  m_all_dispatchers = new MpiThreadAllDispatcher();
  m_all_dispatchers->resize(m_local_nb_rank);
 
  m_parallel_mng_list->resize(m_local_nb_rank);
  m_parallel_mng_list->fill(nullptr);
 
  m_message_queue = new SharedMemoryMessageQueue();
  m_message_queue->init(m_local_nb_rank);
 
  m_thread_barrier = platform::getThreadImplementationService()->createBarrier();
  m_thread_barrier->init(m_local_nb_rank);
 
  m_window_creator = new HybridMachineShMemWinBaseInternalCreator(m_local_nb_rank, m_thread_barrier);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void HybridParallelMngContainer::
_setMPICommunicator()
{
  MPI_Comm comm = static_cast<MPI_Comm>(m_mpi_communicator);
 
  if (comm == MPI_COMM_NULL)
    ARCANE_THROW(ArgumentException, "Null MPI Communicator");
  m_mpi_communicator = comm;
 
  int rank = 0;
  MPI_Comm_rank(m_mpi_communicator, &rank);
  int size = 0;
  MPI_Comm_size(m_mpi_communicator, &size);
 
  m_mpi_comm_rank = rank;
  m_mpi_comm_size = size;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Ref<IParallelMng> HybridParallelMngContainer::
_createParallelMng(Int32 local_rank, ITraceMng* tm)
{
  if (local_rank < 0 || local_rank >= m_local_nb_rank)
    ARCANE_THROW(ArgumentException, "Bad value '{0}' for local_rank (max={1})",
                 local_rank, m_local_nb_rank);
 
  // This method is not reentrant.
  Mutex::ScopedLock sl(m_internal_create_mutex);
 
  Int32 nb_process = m_mpi_comm_size;
  bool is_parallel = nb_process > 1;
 
  // The communicator passed as an argument remains our property.
  MpiParallelMngBuildInfo bi(m_mpi_communicator, m_mpi_machine_communicator);
  bi.is_parallel = is_parallel;
  bi.stat = m_stat;
  bi.trace_mng = tm;
  bi.thread_mng = m_thread_mng;
  bi.is_mpi_comm_owned = false;
  bi.mpi_lock = m_mpi_lock;
 
  if (m_mpi_lock)
    tm->info() << "MPI implementation need serialized threads : using locks";
 
  MpiParallelMng* mpi_pm = new MpiParallelMng(bi);
 
  mpi_pm->build();
  mpi_pm->initialize();
  mpi_pm->adapter()->enableDebugRequest(false);
 
  HybridParallelMngBuildInfo build_info;
  build_info.local_rank = local_rank;
  build_info.local_nb_rank = m_local_nb_rank;
  build_info.mpi_parallel_mng = mpi_pm;
  build_info.trace_mng = tm;
  build_info.thread_mng = m_thread_mng;
  build_info.message_queue = m_message_queue;
  build_info.thread_barrier = m_thread_barrier;
  build_info.parallel_mng_list = m_parallel_mng_list;
  build_info.all_dispatchers = m_all_dispatchers;
  build_info.sub_builder_factory = m_sub_builder_factory;
  build_info.container = makeRef<IParallelMngContainer>(this);
  build_info.window_creator = m_window_creator;
 
  // NOTE: This instance will be destroyed by the caller of this method
  HybridParallelMng* pm = new HybridParallelMng(build_info);
  pm->build();
  (*m_parallel_mng_list)[local_rank] = pm;
 
  return makeRef<IParallelMng>(pm);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
class HybridParallelMngContainerFactory
: public AbstractService
, public IParallelMngContainerFactory
{
 public:
 
  HybridParallelMngContainerFactory(const ServiceBuildInfo& sbi)
  : AbstractService(sbi)
  , m_application(sbi.application())
  {}
 
 public:
 
  Ref<IParallelMngContainer>
  _createParallelMngBuilder(Int32 nb_rank, MP::Communicator mpi_communicator,
                            MP::Communicator mpi_machine_communicator) override
  {
    auto x = new HybridParallelMngContainer(m_application, nb_rank, mpi_communicator,
                                            mpi_machine_communicator,
                                            this, m_stat, m_mpi_lock);
    x->build();
    return makeRef<IParallelMngContainer>(x);
  }
 
 private:
 
  IApplication* m_application;
 
 public:
 
  MpiLock* m_mpi_lock = nullptr;
  Parallel::IStat* m_stat = nullptr;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
ARCANE_REGISTER_SERVICE(HybridParallelMngContainerFactory,
                        ServiceProperty("HybridParallelMngContainerFactory", ST_Application),
                        ARCANE_SERVICE_INTERFACE(IParallelMngContainerFactory));
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/

class HybridParallelSuperMng
: public ParallelSuperMngDispatcher
{
 public:
 
  explicit HybridParallelSuperMng(const ServiceBuildInfo& sbi);
  ~HybridParallelSuperMng() override;
 
  void initialize() override;
  void build() override;
 
  IApplication* application() const override { return m_application; }
  IThreadMng* threadMng() const override { return m_container->m_thread_mng; }
  bool isParallel() const override { return true; }
  Int32 commRank() const override { return m_mpi_comm_rank; }
  Int32 commSize() const override { return m_mpi_comm_size; }
  Int32 traceRank() const override { return m_mpi_comm_rank * m_container->m_local_nb_rank; }
  void* getMPICommunicator() override { return &m_mpi_communicator; }
  MP::Communicator communicator() const override { return m_communicator; }
  Ref<IParallelMng> internalCreateWorldParallelMng(Int32 local_rank) override;
  void tryAbort() override;
  bool isMasterIO() const override { return commRank() == 0; }
  Integer masterIORank() const override { return 0; }
  Integer nbLocalSubDomain() override { return m_container->m_local_nb_rank; }
  void barrier() override;
 
 public:
 
  HybridParallelMngContainer* m_container = nullptr;
  Ref<IParallelMngContainerFactory> m_builder_factory;
  Ref<IParallelMngContainer> m_main_builder;
 
  IApplication* m_application; 
  Parallel::IStat* m_stat; 
  Int32 m_mpi_comm_rank = -1; 
  Int32 m_mpi_comm_size = -1; 
  MPI_Comm m_mpi_communicator;
  MP::Communicator m_communicator;
  MP::Communicator m_machine_communicator;
  Int32 m_local_nb_rank = A_NULL_RANK;
  MpiLock* m_mpi_lock;
  MpiAdapter* m_mpi_adapter;
  MpiDatatypeList* m_datatype_list;
  MpiErrorHandler m_error_handler;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
HybridParallelSuperMng::
HybridParallelSuperMng(const ServiceBuildInfo& sbi)
: m_application(sbi.application())
, m_stat(nullptr)
, m_mpi_comm_size(0)
, m_mpi_communicator(MPI_COMM_NULL)
, m_local_nb_rank(A_NULL_RANK)
, m_mpi_lock(nullptr)
, m_mpi_adapter(nullptr)
, m_datatype_list(nullptr)
{
  m_stat = Parallel::createDefaultStat();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
HybridParallelSuperMng::
~HybridParallelSuperMng()
{
  m_error_handler.removeHandler();
  _finalize();
  delete m_datatype_list;
  if (m_mpi_adapter)
    arcaneCallFunctionAndTerminateIfThrow([&]() { m_mpi_adapter->destroy(); });
  delete m_mpi_lock;
  delete m_stat;
  MPI_Barrier(m_mpi_communicator);
  MPI_Comm_free(&m_mpi_communicator);
  m_mpi_communicator = MPI_COMM_NULL;
  MPI_Comm_free(static_cast<MPI_Comm*>(m_machine_communicator.communicatorAddress()));
  arcaneFinalizeMPI();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void HybridParallelSuperMng::
initialize()
{
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void HybridParallelSuperMng::
build()
{
  if (!arcaneHasThread())
    ARCANE_FATAL("Can not create HybridParallelSuperMng because threads are disabled");
 
  Request::setNullRequest(Request(0, nullptr, MPI_REQUEST_NULL));
  Communicator::setNullCommunicator(Communicator(MPI_COMM_NULL));
 
  int rank, size;
  int* argc = nullptr;
  char*** argv = nullptr;
 
  const ApplicationInfo& app_info = m_application->applicationInfo();
  const CommandLineArguments& cmd_line_args = app_info.commandLineArguments();
  argc = cmd_line_args.commandLineArgc();
  argv = cmd_line_args.commandLineArgv();
 
  bool need_serialize = false;
 
  // Try with MPI_THREAD_MULTIPLE
  // which allows calling MPI from multiple threads simultaneously
  // If this level is not available, at least the
  // MPI_THREAD_SERIALIZED level is required, which allows calling MPI from multiple
  // threads but only one at a time. Therefore, in this case, locks
  // must be used around MPI calls. In our case, locks
  // are only used for point-to-point communications because
  // for collective operations, only one thread performs them
  // at a time.
  int thread_wanted = MPI_THREAD_MULTIPLE;
  int thread_provided = 0;
 
  Real start_time = platform::getRealTime();
  arcaneInitializeMPI(argc, argv, thread_wanted);
  Real end_time = platform::getRealTime();
  MPI_Query_thread(&thread_provided);
 
  if (thread_provided < MPI_THREAD_MULTIPLE) {
    if (thread_provided >= MPI_THREAD_SERIALIZED)
      need_serialize = true;
    else {
      // The required thread level is not available.
      // Launch a fatal error but only on a single processor.
      int my_rank = 0;
      MPI_Comm_rank(MPI_COMM_WORLD, &my_rank);
      if (my_rank != 0)
        ARCANE_FATAL("MPI thread level provided!=wanted ({0}!={1})",
                     thread_provided, thread_wanted);
    }
  }
 
  // Uncomment to force the use of locks for testing.
  // need_serialize = true;
 
  if (need_serialize)
    m_mpi_lock = new MpiLock();
 
  MPI_Comm_dup(MPI_COMM_WORLD, &m_mpi_communicator);
  m_communicator = Communicator(m_mpi_communicator);
  MPI_Comm_rank(m_mpi_communicator, &rank);
  MPI_Comm_size(m_mpi_communicator, &size);
 
  m_mpi_comm_rank = rank;
  m_mpi_comm_size = size;
 
  MPI_Comm mpi_machine_communicator = MPI_COMM_NULL;
  MPI_Comm_split_type(MPI_COMM_WORLD, MPI_COMM_TYPE_SHARED, m_mpi_comm_rank, MPI_INFO_NULL, &mpi_machine_communicator);
  m_machine_communicator = Communicator(mpi_machine_communicator);
 
  IApplication* app = m_application;
 
  m_error_handler.registerHandler(m_mpi_communicator);
 
  Integer n = app->applicationBuildInfo().nbSharedMemorySubDomain();
  if (n == 0)
    ARCANE_FATAL("Number of shared memory sub-domains is not defined");
  m_local_nb_rank = n;
 
  if (rank == 0) {
    ITraceMng* tm = app->traceMng();
    tm->info() << "MPI has non blocking collective";
    tm->info() << "MPI: sizeof(MPI_Count)=" << sizeof(MPI_Count);
    tm->info() << "MPI: is Cuda Aware?=" << arcaneIsCudaAwareMPI();
    tm->info() << "MPI: init_time (seconds)=" << (end_time - start_time);
  }
 
  /*cout << "ThreadSuperParallelMng: nb_sub_domain=" << m_local_nb_rank
       << " env=" << s
       << " mpi_rank=" << m_mpi_comm_rank << "/" << m_mpi_comm_size
       << " first_rank=" << m_current_rank << '\n';*/
 
  ITraceMng* tm = app->traceMng();
  m_datatype_list = new MpiDatatypeList(false);
  auto* adapter = new MpiAdapter(tm, m_stat->toArccoreStat(), m_communicator, nullptr);
  m_mpi_adapter = adapter;
  auto c = createBuiltInDispatcher<Byte>(tm, nullptr, m_mpi_adapter, m_datatype_list);
  auto i32 = createBuiltInDispatcher<Int32>(tm, nullptr, m_mpi_adapter, m_datatype_list);
  auto i64 = createBuiltInDispatcher<Int64>(tm, nullptr, m_mpi_adapter, m_datatype_list);
  auto r = createBuiltInDispatcher<Real>(tm, nullptr, m_mpi_adapter, m_datatype_list);
  _setDispatchers(c, i32, i64, r);
 
  ServiceBuilder<IParallelMngContainerFactory> sb(m_application);
  String service_name = "HybridParallelMngContainerFactory";
  m_builder_factory = sb.createReference(service_name);
  auto* true_builder = dynamic_cast<HybridParallelMngContainerFactory*>(m_builder_factory.get());
  ARCANE_CHECK_POINTER(true_builder);
  true_builder->m_stat = m_stat;
  true_builder->m_mpi_lock = m_mpi_lock;
 
  Ref<IParallelMngContainer> x = m_builder_factory->_createParallelMngBuilder(n, m_communicator, m_machine_communicator);
  m_main_builder = x;
  m_container = dynamic_cast<HybridParallelMngContainer*>(x.get());
  ARCANE_CHECK_POINTER(m_container);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Ref<IParallelMng> HybridParallelSuperMng::
internalCreateWorldParallelMng(Int32 local_rank)
{
  if (local_rank < 0 || local_rank >= m_local_nb_rank)
    throw ArgumentException(A_FUNCINFO, "Bad value for local_rank");
 
  Int32 current_global_rank = local_rank + m_local_nb_rank * commRank();
  ITraceMng* app_tm = m_application->traceMng();
  app_tm->info() << "Create SharedMemoryParallelMng rank=" << current_global_rank;
 
  ITraceMng* tm = nullptr;
  if (local_rank == 0) {
    // The first created subdomain uses the default traceMng().
    tm = app_tm;
  }
  else {
    tm = m_application->createAndInitializeTraceMng(app_tm, String::fromNumber(current_global_rank));
  }
 
  Ref<IParallelMng> pm = m_container->_createParallelMng(local_rank, tm);
  return pm;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void HybridParallelSuperMng::
tryAbort()
{
  m_application->traceMng()->flush();
  MPI_Abort(m_communicator, 2);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void HybridParallelSuperMng::
barrier()
{
  MPI_Barrier(m_communicator);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
ARCANE_REGISTER_SERVICE(HybridParallelSuperMng,
                        ServiceProperty("HybridParallelSuperMng", ST_Application),
                        ARCANE_SERVICE_INTERFACE(IParallelSuperMng));
 
// Old name
ARCANE_REGISTER_SERVICE(HybridParallelSuperMng,
                        ServiceProperty("MpiThreadParallelSuperMng", ST_Application),
                        ARCANE_SERVICE_INTERFACE(IParallelSuperMng));
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/

class MpiSharedMemoryParallelSuperMng
: public SharedMemoryParallelSuperMng
{
 public:
 
  explicit MpiSharedMemoryParallelSuperMng(const ServiceBuildInfo& sbi)
  : SharedMemoryParallelSuperMng(sbi, Parallel::Communicator(MPI_COMM_WORLD), true)
  {
    Request::setNullRequest(Request(0, nullptr, MPI_REQUEST_NULL));
    Communicator::setNullCommunicator(Communicator(MPI_COMM_NULL));
  }
 
  ~MpiSharedMemoryParallelSuperMng() override
  {
    //arcaneFinalizeMPI();
  }
 
  void build() override
  {
    SharedMemoryParallelSuperMng::build();
  }
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
ARCANE_REGISTER_SERVICE(MpiSharedMemoryParallelSuperMng,
                        ServiceProperty("MpiSharedMemoryParallelSuperMng", ST_Application),
                        ARCANE_SERVICE_INTERFACE(IParallelSuperMng));
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
} // End namespace Arcane::MessagePassing
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/