MessagePassingUtils.h

// -*- 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
//-----------------------------------------------------------------------------
/*---------------------------------------------------------------------------*/
/* MessagePassingUtils.h                                       (C) 2000-2026 */
/*                                                                           */
/* Various utilities to handle message passing.                              */
/*---------------------------------------------------------------------------*/
#ifndef ARCCORE_ALINA_MESSAGEPASSINGUTILS_H
#define ARCCORE_ALINA_MESSAGEPASSINGUTILS_H
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
/*
 * This file is based on the work on AMGCL library (version march 2026)
 * which can be found at https://github.com/ddemidov/amgcl.
 *
 * Copyright (c) 2012-2022 Denis Demidov <dennis.demidov@gmail.com>
 * SPDX-License-Identifier: MIT
 */
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#include "arccore/base/FixedArray.h"
#include "arccore/common/Array.h"
 
#include "arccore/message_passing_mpi/StandaloneMpiMessagePassingMng.h"
#include "arccore/message_passing/Messages.h"
#include "arccore/message_passing/PointToPointMessageInfo.h"
 
#include "arccore/alina/ValueTypeInterface.h"
#include "arccore/alina/AlinaUtils.h"
 
#include <vector>
#include <numeric>
#include <complex>
#include <type_traits>
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
namespace Arcane::Alina
{
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/

struct mpi_init
{
  mpi_init(int* argc, char*** argv)
  {
    MPI_Init(argc, argv);
  }
 
  ~mpi_init()
  {
    MPI_Finalize();
  }
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/

struct mpi_init_thread
{
  mpi_init_thread(int* argc, char*** argv)
  {
    int _;
    MPI_Init_thread(argc, argv, MPI_THREAD_MULTIPLE, &_);
  }
 
  ~mpi_init_thread()
  {
    MPI_Finalize();
  }
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
struct mpi_communicator
{
  MPI_Comm comm = MPI_COMM_NULL;
  int rank = 0;
  int size = 0;
  Ref<IMessagePassingMng> m_message_passing_mng;
 
  mpi_communicator() = default;
 
  explicit mpi_communicator(MPI_Comm comm)
  : comm(comm)
  {
    MPI_Comm_rank(comm, &rank);
    MPI_Comm_size(comm, &size);
    m_message_passing_mng = MessagePassing::Mpi::StandaloneMpiMessagePassingMng::createRef(comm);
  };
 
  operator MPI_Comm() const
  {
    return comm;
  }

  template <typename T>
  std::vector<T> exclusive_sum(T n) const
  {
    // TODO: Use scan.
    std::vector<T> v(size + 1);
    v[0] = 0;
    T v0 = n;
    ConstArrayView<T> v0_view(1, &v0);
    ArrayView<T> out_view(static_cast<Int32>(v.size()), &v[1]);
    mpAllGather(m_message_passing_mng.get(), v0_view, out_view);
    std::partial_sum(v.begin(), v.end(), v.begin());
    return v;
  }
 
  std::complex<long double> reduceSum(const std::complex<long double>& lval) const
  {
    return _reduceSumForComplex(lval);
  }
  std::complex<double> reduceSum(const std::complex<double>& lval) const
  {
    return _reduceSumForComplex(lval);
  }
  std::complex<float> reduceSum(const std::complex<float>& lval) const
  {
    return _reduceSumForComplex(lval);
  }
 
  template <typename T> T reduceSum(const T& lval) const
  {
    return mpAllReduce(m_message_passing_mng.get(), MessagePassing::eReduceType::ReduceSum, lval);
  }
 
  void waitAll(ArrayView<MessagePassing::Request> requests) const
  {
    mpWaitAll(m_message_passing_mng.get(), requests);
  }
  void wait(MessagePassing::Request request) const
  {
    ArrayView<MessagePassing::Request> requests(1, &request);
    mpWaitAll(m_message_passing_mng.get(), requests);
  }

  template <class Condition, class Message>
  void check(const Condition& cond, const Message& message)
  {
    int lc = static_cast<int>(cond);
    int gc = _reduce(MPI_PROD, lc);
 
    if (gc==0) {
      IMessagePassingMng* pm = m_message_passing_mng.get();
      UniqueArray<int> c(size);
      if (rank == 0)
        c.resize(size);
      ConstArrayView<int> in_view(1, &lc);
      mpGather(pm, in_view, c, 0);
      if (rank == 0) {
        std::cerr << "Failed assumption: " << message << std::endl;
        std::cerr << "Offending processes:";
        for (int i = 0; i < size; ++i)
          if (!c[i])
            std::cerr << " " << i;
        std::cerr << std::endl;
      }
      mpBarrier(pm);
      ARCCORE_FATAL("CheckError in MessagePassingUtils: {0}", message);
    }
  }
 
  template <typename T> MessagePassing::Request
  doIReceive(T* buf, int count, int source, int tag) const
  {
    using namespace Arcane::MessagePassing;
    Span<T> s(buf, count);
    Span<unsigned char> schar(reinterpret_cast<unsigned char*>(s.data()), s.sizeBytes());
    PointToPointMessageInfo msg_info(MessageRank{ source }, MessageTag{ tag }, eBlockingType::NonBlocking);
    return mpReceive(m_message_passing_mng.get(), schar, msg_info);
  }
 
  template <typename T> void
  doReceive(T* buf, int count, int source, int tag) const
  {
    using namespace Arcane::MessagePassing;
    Span<T> s(buf, count);
    Span<unsigned char> schar(reinterpret_cast<unsigned char*>(s.data()), s.sizeBytes());
    PointToPointMessageInfo msg_info(MessageRank{ source }, MessageTag{ tag }, eBlockingType::Blocking);
    mpReceive(m_message_passing_mng.get(), schar, msg_info);
  }
 
  template <typename T> MessagePassing::Request
  doISend(const T* buf, int count, int dest, int tag) const
  {
    using namespace Arcane::MessagePassing;
    Span<const T> s(buf, count);
    Span<const unsigned char> schar(reinterpret_cast<const unsigned char*>(s.data()), s.sizeBytes());
    PointToPointMessageInfo msg_info(MessageRank{ dest }, MessageTag{ tag }, eBlockingType::NonBlocking);
    return mpSend(m_message_passing_mng.get(), schar, msg_info);
  }
 
  template <typename T> void
  doSend(const T* buf, int count, int dest, int tag) const
  {
    using namespace Arcane::MessagePassing;
    Span<const T> s(buf, count);
    Span<const unsigned char> schar(reinterpret_cast<const unsigned char*>(s.data()), s.sizeBytes());
    PointToPointMessageInfo msg_info(MessageRank{ dest }, MessageTag{ tag }, eBlockingType::Blocking);
    mpSend(m_message_passing_mng.get(), schar, msg_info);
  }
 
 private:
 
  int _reduce(MPI_Op op, int lval) const
  {
    int gval = 0;
 
    MPI_Allreduce((void*)&lval, &gval, 1, MPI_INT, op, comm);
    return gval;
  }
 
  template <typename T> std::complex<T>
  _reduceSumForComplex(const std::complex<T>& lval) const
  {
    // Specialisation for 'std::complex<float>' as 2 float.
    FixedArray<T, 2> values = { { lval.real(), lval.imag() } };
    mpAllReduce(m_message_passing_mng.get(), MessagePassing::eReduceType::ReduceSum, values.view());
    return std::complex<T>(values[0], values[1]);
  }
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
} // namespace Arcane::Alina
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#endif