SharedMemoryMessageQueue.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
//-----------------------------------------------------------------------------
/*---------------------------------------------------------------------------*/
/* SharedMemoryMessageQueue.h                                  (C) 2000-2024 */
/*                                                                           */
/* Implémentation d'une file de messages en mémoire partagée.                */
/*---------------------------------------------------------------------------*/
#ifndef ARCANE_PARALLEL_THREAD_SHAREDMEMORYMESSAGEQUEUE_H
#define ARCANE_PARALLEL_THREAD_SHAREDMEMORYMESSAGEQUEUE_H
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#include "arcane/utils/TraceAccessor.h"
#include "arcane/utils/Array.h"
 
#include "arcane/parallel/thread/ISharedMemoryMessageQueue.h"
 
#include "arcane/ISerializer.h"
#include "arcane/Parallel.h"
 
#include <algorithm>
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
namespace Arcane::MessagePassing
{
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
class SharedMemoryParallelMng;
class SharedMemoryMessageRequest;
class SharedMemoryMessageQueue;
using MessageRank = MP::MessageRank;
using MessageTag = MP::MessageTag;
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
class ARCANE_THREAD_EXPORT SharedMemoryMessageQueue
: public ISharedMemoryMessageQueue
{
 public:
 
  class SubQueue;
 
 public:
 
  SharedMemoryMessageQueue()
  : m_atomic_request_id(0)
  {}
  ~SharedMemoryMessageQueue() override;
 
 public:
 
  void init(Int32 nb_thread) override;
  void waitAll(Int32 caller_rank, ArrayView<Request> requests) override;
  void waitSome(Int32 rank, ArrayView<Request> requests, ArrayView<bool> requests_done,
                bool is_non_blocking) override;
  void setTraceMng(Int32 rank, ITraceMng* tm) override;
 
 public:
 
  Request addReceive(const PointToPointMessageInfo& message,ReceiveBufferInfo buf) override;
  Request addSend(const PointToPointMessageInfo& message,SendBufferInfo buf) override;
 
 public:
 
  MessageId probe(const PointToPointMessageInfo& message) override;
  MessageSourceInfo legacyProbe(const PointToPointMessageInfo& message) override;
 
 private:
 
  Int32 m_nb_thread = 0;
  UniqueArray<SubQueue*> m_sub_queues;
  std::atomic<Int64> m_atomic_request_id;
 
 private:
 
  SubQueue* _getSubQueue(MessageRank rank)
  {
    return m_sub_queues[rank.value()];
  }
  Int64 _getNextRequestId()
  {
    return m_atomic_request_id.fetch_add(1);
  }
  Request _request(SharedMemoryMessageRequest* tmr);
  SubQueue* _getSourceSubQueue(const MP::PointToPointMessageInfo& message);
  SubQueue* _getDestinationSubQueue(const MP::PointToPointMessageInfo& message);
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
class ARCANE_THREAD_EXPORT SharedMemoryMessageRequest
{
 public:
  class SortFunctor
  {
   public:
    explicit SortFunctor(Int32 nb_thread) : m_nb_thread(nb_thread){}
    bool operator()(SharedMemoryMessageRequest* r1,SharedMemoryMessageRequest* r2) const
    {
      if (!r1){
        if (!r2)
          return true;
        else
          return false;
      }
      if (!r2)
        return true;
      if (r1->isRecv() && !r2->isRecv())
        return true;
      if (!r1->isRecv() && r2->isRecv())
        return false;
      Int32 i1 = _getQueueIndex(r1->orig(),r1->dest());
      Int32 i2 = _getQueueIndex(r2->orig(),r2->dest());
      if (i1==i2)
        return r1->id() < r2->id();
      return i1 < i2;
    }
    Int32 _getQueueIndex(MessageRank thread1,MessageRank thread2) const
    {
      if (thread1.isNull())
        ARCANE_FATAL("Null rank for thread1");
      if (thread2.isNull())
        ARCANE_FATAL("Null rank for thread2");
      // TODO: gérer dest()==A_NULL_RANK.
      return thread1.value() + (thread2.value() * m_nb_thread);
    }
    Int32 m_nb_thread;
  };
 public:
  using SubQueue = SharedMemoryMessageQueue::SubQueue;
 public:
  SharedMemoryMessageRequest(SubQueue* queue,Int64 request_id,MessageRank orig,
                             MessageRank dest,MessageTag tag,ReceiveBufferInfo buf)
  : m_queue(queue), m_request_id(request_id), m_is_recv(true)
  , m_orig(orig), m_dest(dest), m_tag(tag), m_receive_buffer_info(buf)
  {
  }

  SharedMemoryMessageRequest(SubQueue* queue,Int64 request_id,MessageRank orig,
                             MessageRank dest,MessageTag tag,SendBufferInfo buf)
  : m_queue(queue), m_request_id(request_id), m_is_recv(false)
  , m_orig(orig), m_dest(dest), m_tag(tag), m_send_buffer_info(buf)
  {
  }
 public:
 
  MessageRank orig() const { return m_orig; }
  MessageRank dest() const { return m_dest; }
  MessageTag tag() const { return m_tag; }
  bool isRecv() const { return m_is_recv; }
  bool isDone() const { return m_is_done; }
  void setDone(bool v) { m_is_done = v; }
  SendBufferInfo sendBufferInfo() { return m_send_buffer_info; }
  ReceiveBufferInfo receiveBufferInfo() { return m_receive_buffer_info; }
  SharedMemoryMessageQueue::SubQueue* queue() { return m_queue; }
  const String& queueName() const;
  void copyFromSender(SharedMemoryMessageRequest* sender);
  Int64 id() const { return m_request_id; }
  void destroy();
  ISerializer* recvSerializer() { return m_receive_buffer_info.serializer(); }
  const ISerializer* sendSerializer() { return m_send_buffer_info.serializer(); }
  // Dans le cas ou dest()==A_NULL_RANK, positionne une fois le message recu le rang d'origine.
  void setSource(MessageRank s)
  {
    if (isRecv())
      m_dest = s;
  }
  SharedMemoryMessageRequest* matchingSendRequest() { return m_matching_send_request; }
  void setMatchingSendRequest(SharedMemoryMessageRequest* r) { m_matching_send_request = r; }
  friend std::ostream& operator<<(std::ostream& o, const SharedMemoryMessageRequest& v);
 
 private:
 
  SubQueue* m_queue = nullptr;
  Int64 m_request_id = -1;
  bool m_is_recv = false;
  MessageRank m_orig;
  MessageRank m_dest;
  MessageTag m_tag;
  std::atomic<bool> m_is_done = false;
  SharedMemoryMessageRequest* m_matching_send_request = nullptr;
  bool m_is_destroyed = false;
  SendBufferInfo m_send_buffer_info;
  ReceiveBufferInfo m_receive_buffer_info;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
} // End namespace Arcane::MessagePassing
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#endif