HybridMessageQueue.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
//-----------------------------------------------------------------------------
/*---------------------------------------------------------------------------*/
/* HybridMessageQueue.cc                                       (C) 2000-2025 */
/*                                                                           */
/* Message file for an MPI/Thread implementation.                            */
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#include "arcane/utils/ArcanePrecomp.h"
 
#include "arcane/utils/Array.h"
#include "arcane/utils/FatalErrorException.h"
#include "arcane/utils/NotImplementedException.h"
#include "arcane/utils/NotSupportedException.h"
#include "arcane/utils/ArgumentException.h"
#include "arcane/utils/TraceInfo.h"
#include "arcane/utils/ITraceMng.h"
#include "arcane/utils/ValueConvert.h"
 
#include "arcane/parallel/mpithread/HybridMessageQueue.h"
#include "arcane/parallel/mpi/MpiParallelMng.h"
 
#include "arccore/message_passing_mpi/internal/MpiAdapter.h"
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
// Macro to display messages for debug
#define TRACE_DEBUG(needed_debug_level, format_str, ...) \
  if (m_debug_level >= needed_debug_level) { \
    info() << String::format("Hybrid " format_str, __VA_ARGS__); \
    traceMng()->flush(); \
  }
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
namespace Arcane::MessagePassing
{
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
HybridMessageQueue::
HybridMessageQueue(ISharedMemoryMessageQueue* thread_queue, MpiParallelMng* mpi_pm,
                   Int32 local_nb_rank)
: TraceAccessor(mpi_pm->traceMng())
, m_thread_queue(thread_queue)
, m_mpi_parallel_mng(mpi_pm)
, m_mpi_adapter(mpi_pm->adapter())
, m_local_nb_rank(local_nb_rank)
, m_rank_tag_builder(local_nb_rank)
, m_debug_level(0)
{
  if (auto v = Convert::Type<Int32>::tryParseFromEnvironment("ARCCORE_ALLOW_NULL_RANK_FOR_MPI_ANY_SOURCE", true))
    m_is_allow_null_rank_for_any_source = v.value() != 0;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void HybridMessageQueue::
_checkValidRank(MessageRank rank)
{
  if (rank.isNull())
    ARCANE_THROW(ArgumentException, "null rank");
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void HybridMessageQueue::
_checkValidSource(const PointToPointMessageInfo& message)
{
  MessageRank source = message.emiterRank();
  if (source.isNull())
    ARCANE_THROW(ArgumentException, "null source");
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
PointToPointMessageInfo HybridMessageQueue::
_buildSharedMemoryMessage(const PointToPointMessageInfo& message,
                          const SourceDestinationFullRankInfo& fri)
{
  PointToPointMessageInfo p2p_message(message);
  p2p_message.setEmiterRank(fri.source().localRank());
  p2p_message.setDestinationRank(fri.destination().localRank());
  return p2p_message;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
PointToPointMessageInfo HybridMessageQueue::
_buildMPIMessage(const PointToPointMessageInfo& message,
                 const SourceDestinationFullRankInfo& fri)
{
  PointToPointMessageInfo p2p_message(message);
  p2p_message.setEmiterRank(fri.source().mpiRank());
  p2p_message.setDestinationRank(fri.destination().mpiRank());
  return p2p_message;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void HybridMessageQueue::
waitAll(ArrayView<Request> requests)
{
  // TODO: merge what is possible with waitSome.
  Integer nb_request = requests.size();
  UniqueArray<Request> mpi_requests;
  UniqueArray<Request> thread_requests;
  for (Integer i = 0; i < nb_request; ++i) {
    Request r = requests[i];
    if (!r.isValid())
      continue;
    IRequestCreator* creator = r.creator();
    if (creator == m_mpi_adapter) {
      mpi_requests.add(r);
    }
    else if (creator == m_thread_queue)
      thread_requests.add(r);
    else
      ARCANE_FATAL("Invalid IRequestCreator");
  }
 
  if (mpi_requests.size() != 0)
    m_mpi_adapter->waitAllRequests(mpi_requests);
  if (thread_requests.size() != 0)
    m_thread_queue->waitAll(thread_requests);
 
  // We reset all requests to be able to call the Wait functions!
  for (Request r : requests)
    r.reset();
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
void HybridMessageQueue::
waitSome(Int32 rank, ArrayView<Request> requests, ArrayView<bool> requests_done,
         bool is_non_blocking)
{
  Integer nb_done = 0;
  do {
    TRACE_DEBUG(2, "Hybrid: wait some rank={0} requests n={1} nb_done={2} is_non_blocking={3}",
                rank, requests.size(), nb_done, is_non_blocking);
    nb_done = _testOrWaitSome(rank, requests, requests_done);
    if (is_non_blocking || nb_done == (-1))
      break;
  } while (nb_done == 0);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Integer HybridMessageQueue::
_testOrWaitSome(Int32 rank, ArrayView<Request> requests, ArrayView<bool> requests_done)
{
  Integer nb_request = requests.size();
  TRACE_DEBUG(2, "Hybrid: wait some rank={0} requests n={1}", rank, nb_request);
 
  // We must separate MPI requests from shared memory requests.
  // TODO: with the notion of generalized MPI requests, it might be
  // possible to merge this.
  UniqueArray<Request> mpi_requests;
  UniqueArray<Request> shm_requests;
  // Index of requests in the global list \a requests
  UniqueArray<Integer> mpi_requests_index;
  UniqueArray<Integer> shm_requests_index;
 
  Integer nb_done = 0;
  for (Integer i = 0; i < nb_request; ++i) {
    Request r = requests[i];
    if (!r.isValid())
      continue;
    IRequestCreator* creator = r.creator();
    if (creator == m_mpi_adapter) {
      mpi_requests.add(r);
      mpi_requests_index.add(i);
    }
    else if (creator == m_thread_queue) {
      shm_requests.add(r);
      shm_requests_index.add(i);
    }
    else
      ARCANE_FATAL("Invalid IRequestCreator");
  }
 
  TRACE_DEBUG(2, "Hybrid: wait some rank={0} nb_mpi={1} nb_shm={2}",
              rank, mpi_requests.size(), shm_requests.size());
 
  // If there are no valid requests, there is no need to proceed.
  // However, we must not return '0' because we must make
  // the difference between no requests available for the 'is_non_blocking'
  // mode
  // and no valid requests.
  if (mpi_requests.size() == 0 && shm_requests.size() == 0)
    return (-1);
 
  // Even in waitSome mode, we must use non-blocking mode because
  // we do not know between threads and MPI which requests will be
  // available
 
  // The requests may have been modified if they are not finished.
  // Therefore, they must be put back into the \a requests list. In our
  // case, it is enough to only copy the new value into
  // the corresponding instance of HybridMessageRequest.
  UniqueArray<bool> mpi_done_indexes;
  Integer nb_mpi_request = mpi_requests.size();
 
  if (nb_mpi_request != 0) {
    mpi_done_indexes.resize(nb_mpi_request);
    mpi_done_indexes.fill(false);
    m_mpi_adapter->waitSomeRequests(mpi_requests, mpi_done_indexes, true);
    TRACE_DEBUG(2, "Hybrid: MPI wait some requests n={0} after=", nb_mpi_request, mpi_done_indexes);
    for (Integer i = 0; i < nb_mpi_request; ++i) {
      Integer index_in_global = mpi_requests_index[i];
      if (mpi_done_indexes[i]) {
        requests_done[index_in_global] = true;
        requests[index_in_global].reset();
        ++nb_done;
        TRACE_DEBUG(1, "MPI rank={0} set done i={1} in_global={2}",
                    rank, i, index_in_global);
      }
      else
        requests[index_in_global] = mpi_requests[i];
    }
  }
 
  UniqueArray<bool> shm_done_indexes;
  Integer nb_shm_request = shm_requests.size();
  TRACE_DEBUG(2, "SHM wait some requests n={0}", nb_shm_request);
  if (shm_requests.size() != 0) {
    shm_done_indexes.resize(nb_shm_request);
    shm_done_indexes.fill(false);
    m_thread_queue->waitSome(rank, shm_requests, shm_done_indexes, true);
    for (Integer i = 0; i < nb_shm_request; ++i) {
      Integer index_in_global = shm_requests_index[i];
      if (shm_done_indexes[i]) {
        requests_done[index_in_global] = true;
        requests[index_in_global].reset();
        ++nb_done;
        TRACE_DEBUG(1, "SHM rank={0} set done i={1} in_global={2}",
                    rank, i, index_in_global);
      }
      else
        requests[index_in_global] = shm_requests[i];
    }
  }
  return nb_done;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Request HybridMessageQueue::
_addReceiveRankTag(const PointToPointMessageInfo& message, ReceiveBufferInfo buf_info)
{
  // We do not support receives with ANY_RANK because we do not know
  // whether we need to perform a 'receive' with MPI or in shared memory.
  // In this case, the user must instead use probe()
  // to know what is available and send an addReceive()
  // with a MessageId.
  if (message.destinationRank().isNull())
    ARCANE_THROW(NotSupportedException, "Receive with any rank. Use probe() and MessageId instead");
 
  SourceDestinationFullRankInfo fri = _getFullRankInfo(message);
  bool is_same_mpi_rank = fri.isSameMpiRank();
 
  if (is_same_mpi_rank) {
    TRACE_DEBUG(1, "** MPITMQ SHM ADD RECV S queue={0} message={1}", this, message);
    PointToPointMessageInfo p2p_message(_buildSharedMemoryMessage(message, fri));
    return m_thread_queue->addReceive(p2p_message, buf_info);
  }
 
  ISerializer* serializer = buf_info.serializer();
  if (serializer) {
    TRACE_DEBUG(1, "** MPITMQ MPI ADD RECV S queue={0} message={1}", this, message);
    PointToPointMessageInfo p2p_message(_buildMPIMessage(message, fri));
    p2p_message.setTag(m_rank_tag_builder.tagForReceive(MessageTag(message.tag()), fri));
    return m_mpi_parallel_mng->receiveSerializer(serializer, p2p_message);
  }
  else {
    ByteSpan buf = buf_info.memoryBuffer();
    Int64 size = buf.size();
 
    TRACE_DEBUG(1, "** MPITMQ THREAD ADD RECV B queue={0} message={1} size={2} same_mpi?={3}",
                this, message, size, fri.isSameMpiRank());
 
    //TODO: use the real MPI_Datatype
    MPI_Datatype char_data_type = MpiBuiltIn::datatype(char());
    MessageTag mpi_tag = m_rank_tag_builder.tagForReceive(message.tag(), fri);
    Request r = m_mpi_adapter->directRecv(buf.data(), size, fri.destination().mpiRankValue(), sizeof(char),
                                          char_data_type, mpi_tag.value(), false);
    return r;
  }
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Request HybridMessageQueue::
_addReceiveMessageId(const PointToPointMessageInfo& message, ReceiveBufferInfo buf_info)
{
  MessageId message_id = message.messageId();
  MessageId::SourceInfo si(message_id.sourceInfo());
 
  if (si.rank() != message.destinationRank())
    ARCANE_FATAL("Incohence between messsage_id rank and destination rank x1={0} x2={1}",
                 si.rank(), message.destinationRank());
 
  TRACE_DEBUG(1, "** MPITMQ ADD_RECV (message_id) queue={0} message={1}",
              this, message);
 
  SourceDestinationFullRankInfo fri = _getFullRankInfo(message);
  if (fri.isSameMpiRank()) {
    PointToPointMessageInfo p2p_message(_buildSharedMemoryMessage(message, fri));
    return m_thread_queue->addReceive(p2p_message, buf_info);
  }
 
  TRACE_DEBUG(1, "** MPITMQ MPI ADD RECV (message_id) queue={0} message={1}", this, message);
 
  ISerializer* serializer = buf_info.serializer();
  if (serializer) {
    PointToPointMessageInfo p2p_message(_buildMPIMessage(message, fri));
    //p2p_message.setTag(m_rank_tag_builder.tagForReceive(message.tag(),fri));
    TRACE_DEBUG(1, "** MPI ADD RECV Serializer (message_id) message={0} p2p_message={1}",
                message, p2p_message);
    return m_mpi_parallel_mng->receiveSerializer(serializer, p2p_message);
  }
  else {
    ByteSpan buf = buf_info.memoryBuffer();
    Int64 size = buf.size();
 
    // TODO: use the real MPI_Datatype
    MPI_Datatype char_data_type = MpiBuiltIn::datatype(char());
    MessageId mpi_message(message_id);
    MessageId::SourceInfo mpi_si(si);
    mpi_si.setRank(fri.destination().mpiRank());
    mpi_message.setSourceInfo(mpi_si);
    return m_mpi_adapter->directRecv(buf.data(), size, mpi_message, sizeof(char),
                                     char_data_type, false);
  }
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Request HybridMessageQueue::
addReceive(const PointToPointMessageInfo& message, ReceiveBufferInfo buf)
{
  _checkValidSource(message);
 
  if (!message.isValid())
    return Request();
 
  if (message.isRankTag())
    return _addReceiveRankTag(message, buf);
 
  if (message.isMessageId())
    return _addReceiveMessageId(message, buf);
 
  ARCANE_THROW(NotSupportedException, "Invalid message_info");
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
Request HybridMessageQueue::
addSend(const PointToPointMessageInfo& message, SendBufferInfo buf_info)
{
  if (!message.isValid())
    return Request();
  if (message.destinationRank().isNull())
    ARCCORE_FATAL("Null destination");
  if (!message.isRankTag())
    ARCCORE_FATAL("Invalid message_info for sending: message.isRankTag() is false");
 
  SourceDestinationFullRankInfo fri = _getFullRankInfo(message);
 
  // Same rank, so send via the shared memory queue.
  if (fri.isSameMpiRank()) {
    TRACE_DEBUG(1, "** MPITMQ SHM ADD SEND S queue={0} message={1}", this, message);
    PointToPointMessageInfo p2p_message(_buildSharedMemoryMessage(message, fri));
    return m_thread_queue->addSend(p2p_message, buf_info);
  }
 
  // Send via MPI
  MessageTag mpi_tag = m_rank_tag_builder.tagForSend(message.tag(), fri);
  const ISerializer* serializer = buf_info.serializer();
  if (serializer) {
    PointToPointMessageInfo p2p_message(_buildMPIMessage(message, fri));
    p2p_message.setTag(mpi_tag);
    TRACE_DEBUG(1, "** MPITMQ MPI ADD SEND Serializer queue={0} message={1} p2p_message={2}",
                this, message, p2p_message);
    return m_mpi_parallel_mng->sendSerializer(serializer, p2p_message);
  }
  else {
    ByteConstSpan buf = buf_info.memoryBuffer();
    Int64 size = buf.size();
 
    // TODO: use m_mpi_parallel_mng but must be careful
    // to use blocking mode
    // TODO: use the real MPI_Datatype
    MPI_Datatype char_data_type = MpiBuiltIn::datatype(char());
 
    TRACE_DEBUG(1, "** MPITMQ MPI ADD SEND B queue={0} message={1} size={2} mpi_tag={3} mpi_rank={4}",
                this, message, size, mpi_tag, fri.destination().mpiRank());
 
    return m_mpi_adapter->directSend(buf.data(), size, fri.destination().mpiRankValue(),
                                     sizeof(char), char_data_type, mpi_tag.value(), false);
  }
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
MP::MessageId HybridMessageQueue::
probe(const MP::PointToPointMessageInfo& message)
{
  TRACE_DEBUG(1, "Probe msg='{0}' queue={1} is_valid={2}",
              message, this, message.isValid());
 
  MessageRank orig = message.emiterRank();
  if (orig.isNull())
    ARCANE_THROW(ArgumentException, "null sender");
 
  if (!message.isValid())
    return MessageId();
 
  // The message must be initialized with a (rank/tag) pair.
  if (!message.isRankTag())
    ARCCORE_FATAL("Invalid message_info: message.isRankTag() is false");
 
  MessageRank dest = message.destinationRank();
  MessageTag user_tag = message.tag();
  bool is_blocking = message.isBlocking();
  if (is_blocking)
    ARCANE_THROW(NotImplementedException, "blocking probe");
  if (user_tag.isNull())
    ARCANE_THROW(NotImplementedException, "probe with ANY_TAG");
  FullRankInfo orig_fri = m_rank_tag_builder.rank(orig);
  FullRankInfo dest_fri = m_rank_tag_builder.rank(dest);
  MessageId message_id;
  Int32 found_dest = dest.value();
  const bool is_any_source = dest.isNull() || dest.isAnySource();
  if (dest.isNull() && !m_is_allow_null_rank_for_any_source)
    ARCANE_FATAL("Can not use probe() with null rank. Use MessageRank::anySourceRank() instead");
  if (is_any_source) {
    // Since we don't know who we will receive from, we must
    // test both the thread queue and via MPI.
    MP::PointToPointMessageInfo p2p_message(message);
    p2p_message.setEmiterRank(orig_fri.localRank());
    message_id = m_thread_queue->probe(p2p_message);
    if (message_id.isValid()) {
      // We found a message in the thread list.
      // Since we are in our thread list, the global rank is our
      // MPI rank + the found local rank.
      found_dest = orig_fri.mpiRankValue() * m_local_nb_rank + message_id.sourceInfo().rank().value();
      TRACE_DEBUG(2, "Probe with null_rank (thread) orig={0} found_dest={1} tag={2}",
                  orig, found_dest, user_tag);
    }
    else {
      // Search via MPI.
      // The difficulty is that the local rank of the originating PE of the message
      // is encoded in the tag and we do not know the originating PE.
      // Therefore, we must test all potential tags. Their number is
      // equal to 'm_nb_local_rank'.
      for (Integer z = 0, zn = m_local_nb_rank; z < zn; ++z) {
        MP::PointToPointMessageInfo mpi_message(message);
        MessageTag mpi_tag = m_rank_tag_builder.tagForReceive(user_tag, orig_fri.localRank(), MessageRank(z));
        mpi_message.setTag(mpi_tag);
        TRACE_DEBUG(2, "Probe with null_rank orig={0} dest={1} tag={2}", orig, dest, mpi_tag);
        message_id = m_mpi_adapter->probeMessage(mpi_message);
        if (message_id.isValid()) {
          // We found an MPI message. We must extract the local rank
          // from the tag. The MPI rank is the one in the message.
          MessageRank mpi_rank = message_id.sourceInfo().rank();
          MessageTag ret_tag = message_id.sourceInfo().tag();
          Int32 local_rank = m_rank_tag_builder.getReceiveRankFromTag(ret_tag);
          found_dest = mpi_rank.value() * m_local_nb_rank + local_rank;
          TRACE_DEBUG(2, "Probe null rank found mpi_rank={0} local_rank={1} tag={2}",
                      ret_tag, mpi_rank, local_rank, ret_tag);
          break;
        }
      }
    }
  }
  else {
    // The rank `dest` must be converted to the rank expected by the thread queue
    // or by MPI.
    if (orig_fri.mpiRank() == dest_fri.mpiRank()) {
      MP::PointToPointMessageInfo p2p_message(message);
      p2p_message.setDestinationRank(MP::MessageRank(dest_fri.localRank()));
      p2p_message.setEmiterRank(MessageRank(orig_fri.localRank()));
      message_id = m_thread_queue->probe(p2p_message);
    }
    else {
      MP::PointToPointMessageInfo mpi_message(message);
      MessageTag mpi_tag = m_rank_tag_builder.tagForReceive(user_tag, orig_fri, dest_fri);
      mpi_message.setTag(mpi_tag);
      mpi_message.setDestinationRank(MP::MessageRank(dest_fri.mpiRank()));
      TRACE_DEBUG(2, "Probe orig={0} dest={1} mpi_tag={2} user_tag={3}", orig, dest, mpi_tag, user_tag);
      message_id = m_mpi_adapter->probeMessage(mpi_message);
    }
  }
  if (message_id.isValid()) {
    // The local rank returned by the previous methods must be transformed
    // into a global rank
    MessageId::SourceInfo si = message_id.sourceInfo();
    si.setRank(MessageRank(found_dest));
    message_id.setSourceInfo(si);
  }
  return message_id;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
MP::MessageSourceInfo HybridMessageQueue::
legacyProbe(const MP::PointToPointMessageInfo& message)
{
  TRACE_DEBUG(1, "LegacyProbe msg='{0}' queue={1} is_valid={2}",
              message, this, message.isValid());
 
  MessageRank orig = message.emiterRank();
  if (orig.isNull())
    ARCANE_THROW(ArgumentException, "null sender");
 
  if (!message.isValid())
    return {};
 
  // The message must be initialized with a (rank/tag) pair.
  if (!message.isRankTag())
    ARCCORE_FATAL("Invalid message_info: message.isRankTag() is false");
 
  const MessageRank dest = message.destinationRank();
  const MessageTag user_tag = message.tag();
  const bool is_blocking = message.isBlocking();
  if (is_blocking)
    ARCANE_THROW(NotImplementedException, "blocking probe");
  if (user_tag.isNull())
    ARCANE_THROW(NotImplementedException, "legacyProbe with ANY_TAG");
  FullRankInfo orig_fri = m_rank_tag_builder.rank(orig);
  FullRankInfo dest_fri = m_rank_tag_builder.rank(dest);
  MP::MessageSourceInfo message_source_info;
  Int32 found_dest = dest.value();
  const bool is_any_source = dest.isNull() || dest.isAnySource();
  if (dest.isNull() && !m_is_allow_null_rank_for_any_source)
    ARCANE_FATAL("Can not use legacyProbe() with null rank. Use MessageRank::anySourceRank() instead");
  if (is_any_source) {
    // Since we don't know who we will receive from, we must test
    // both the thread queue and via MPI.
    MP::PointToPointMessageInfo p2p_message(message);
    p2p_message.setEmiterRank(orig_fri.localRank());
    message_source_info = m_thread_queue->legacyProbe(p2p_message);
    if (message_source_info.isValid()) {
      // We found a message in the thread list.
      // Since we are in our thread list, the global rank is our
      // MPI rank + the found local rank.
      found_dest = orig_fri.mpiRankValue() * m_local_nb_rank + message_source_info.rank().value();
      TRACE_DEBUG(2, "LegacyProbe with null_rank (thread) orig={0} found_dest={1} tag={2}",
                  orig, found_dest, user_tag);
    }
    else {
      // Search via MPI.
      // The difficulty is that the local rank of the originating PE of the message
      // is encoded in the tag and we do not know the originating PE.
      // Therefore, we must test all potential tags. Their number is
      // equal to 'm_nb_local_rank'.
      for (Integer z = 0, zn = m_local_nb_rank; z < zn; ++z) {
        MP::PointToPointMessageInfo mpi_message(message);
        MessageTag mpi_tag = m_rank_tag_builder.tagForReceive(user_tag, orig_fri.localRank(), MessageRank(z));
        mpi_message.setTag(mpi_tag);
        TRACE_DEBUG(2, "LegacyProbe with null_rank orig={0} dest={1} tag={2}", orig, dest, mpi_tag);
        message_source_info = m_mpi_adapter->legacyProbeMessage(mpi_message);
        if (message_source_info.isValid()) {
          // We found an MPI message. We must extract the local rank
          // from the tag. The MPI rank is the one in the message.
          MessageRank mpi_rank = message_source_info.rank();
          MessageTag ret_tag = message_source_info.tag();
          Int32 local_rank = m_rank_tag_builder.getReceiveRankFromTag(ret_tag);
          found_dest = mpi_rank.value() * m_local_nb_rank + local_rank;
          TRACE_DEBUG(2, "LegacyProbe null rank found mpi_rank={0} local_rank={1} tag={2}",
                      ret_tag, mpi_rank, local_rank, ret_tag);
          // Restore the original tag to be able to perform a receive with it.
          message_source_info.setTag(user_tag);
          break;
        }
      }
    }
  }
  else {
    // The rank `dest` must be converted to the rank expected by the thread queue
    // or by MPI.
    if (orig_fri.mpiRank() == dest_fri.mpiRank()) {
      MP::PointToPointMessageInfo p2p_message(message);
      p2p_message.setDestinationRank(MP::MessageRank(dest_fri.localRank()));
      p2p_message.setEmiterRank(MessageRank(orig_fri.localRank()));
      TRACE_DEBUG(2, "LegacyProbe SHM orig={0} dest={1} tag={2}", orig, dest, user_tag);
      message_source_info = m_thread_queue->legacyProbe(p2p_message);
    }
    else {
      MP::PointToPointMessageInfo mpi_message(message);
      MessageTag mpi_tag = m_rank_tag_builder.tagForReceive(user_tag, orig_fri, dest_fri);
      mpi_message.setTag(mpi_tag);
      mpi_message.setDestinationRank(MP::MessageRank(dest_fri.mpiRank()));
      TRACE_DEBUG(2, "LegacyProbe MPI orig={0} dest={1} mpi_tag={2} user_tag={3}", orig, dest, mpi_tag, user_tag);
      message_source_info = m_mpi_adapter->legacyProbeMessage(mpi_message);
      if (message_source_info.isValid()) {
        // Restore the original tag to be able to perform a receive with it.
        message_source_info.setTag(user_tag);
      }
    }
  }
  if (message_source_info.isValid()) {
    // The local rank returned by the previous methods must be transformed into a global rank
    message_source_info.setRank(MessageRank(found_dest));
  }
  TRACE_DEBUG(2, "LegacyProbe has matched message? = {0}", message_source_info.isValid());
  return message_source_info;
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
std::ostream& operator<<(std::ostream& o, const FullRankInfo& fri)
{
  return o << "(local=" << fri.m_local_rank << ",global="
           << fri.m_global_rank << ",mpi=" << fri.m_mpi_rank << ")";
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
} // End namespace Arcane::MessagePassing
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/