HybridParallelDispatch.h

// -*- tab-width: 2; indent-tabs-mode: nil; coding: utf-8-with-signature -*-
//-----------------------------------------------------------------------------
// Copyright 2000-2024 CEA (www.cea.fr) IFPEN (www.ifpenergiesnouvelles.com)
// See the top-level COPYRIGHT file for details.
// SPDX-License-Identifier: Apache-2.0
//-----------------------------------------------------------------------------
/*---------------------------------------------------------------------------*/
/* HybridParallelDispatch.h                                    (C) 2000-2024 */
/*                                                                           */
/* Implémentation des messages en mode hybride MPI/Mémoire partagée..        */
/*---------------------------------------------------------------------------*/
#ifndef ARCANE_PARALLEL_THREAD_HYBRIDPARALLELDISPATCH_H
#define ARCANE_PARALLEL_THREAD_HYBRIDPARALLELDISPATCH_H
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#include "arcane/utils/TraceAccessor.h"
#include "arcane/utils/Array.h"
#include "arcane/utils/HPReal.h"
#include "arcane/utils/DataTypeContainer.h"
 
#include "arcane/core/IParallelDispatch.h"
#include "arcane/core/ISerializer.h"
 
#include "arcane/parallel/thread/ISharedMemoryMessageQueue.h"
 
#include "arccore/base/ReferenceCounterImpl.h"
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
namespace Arcane
{
template<class Type>
class MpiParallelDispatchT;
}
 
namespace Arcane::MessagePassing
{
 
class HybridParallelMng;
class HybridMessageQueue;
template<class Type>
class HybridParallelDispatch;
 
template<typename DataType>
class MpiThreadDispatcherContainerTraits
{
 public:
  typedef UniqueArray<HybridParallelDispatch<DataType>*> InstanceType;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
class MpiThreadAllDispatcher
: public ArcaneDataTypeContainer<MpiThreadDispatcherContainerTraits>
{
 public:
  class CreateSubParallelMngInfo
  {
   public:
    Ref<IParallelMngContainer> m_builder;
  };
 public:
  void resize(Integer n)
  {
    m_char.resize(n);
    m_signed_char.resize(n);
    m_unsigned_char.resize(n);
    m_short.resize(n);
    m_unsigned_short.resize(n);
    m_int.resize(n);
    m_unsigned_int.resize(n);
    m_long.resize(n);
    m_unsigned_long.resize(n);
    m_long_long.resize(n);
    m_unsigned_long_long.resize(n);
    m_float.resize(n);
    m_double.resize(n);
    m_long_double.resize(n);
    m_apreal.resize(n);
    m_real2.resize(n);
    m_real3.resize(n);
    m_real2x2.resize(n);
    m_real3x3.resize(n);
    m_hpreal.resize(n);
  }
 public:
  CreateSubParallelMngInfo m_create_sub_parallel_mng_info;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
template<class Type>
class HybridParallelDispatch
: public TraceAccessor
, public IParallelDispatchT<Type>
, public ReferenceCounterImpl
, public ITypeDispatcher<Type>
{
  ARCCORE_INTERNAL_DEFINE_REFERENCE_COUNTED_INCLASS_METHODS();
 
  class AllToAllVariableInfo
  {
   public:
    Span<const Type> send_buf;
    Int32ConstArrayView send_count;
    Int32ConstArrayView send_index;
    Span<Type> recv_buf;
    Int32ConstArrayView recv_count;
    Int32ConstArrayView recv_index;
  };
 
  class ReduceInfo
  {
   public:
 
    Span<Type> reduce_buf_span;
    Type reduce_value;
    int m_index;
  };
 
 private:
  class MinMaxSumInfo
  {
   public:
    Int32 m_min_rank = A_NULL_RANK;
    Int32 m_max_rank = A_NULL_RANK;
    Type m_min_value = Type();
    Type m_max_value = Type();
    Type m_sum_value = Type();
  };
 public:
 
  HybridParallelDispatch(ITraceMng* tm,HybridParallelMng* parallel_mng,HybridMessageQueue* message_queue,
                         ArrayView<HybridParallelDispatch<Type>*> all_dispatchs);
  ~HybridParallelDispatch() override;
  void finalize() override;
 
 public:
 
  using ITypeDispatcher<Type>::gather;
 
 public:
 
  void broadcast(Span<Type> send_buf,Int32 sub_domain) override;
  void allGather(Span<const Type> send_buf,Span<Type> recv_buf) override;
  void allGatherVariable(Span<const Type> send_buf,Array<Type>& recv_buf) override;
  void gather(Span<const Type> send_buf,Span<Type> recv_buf,Int32 rank) override;
  void gatherVariable(Span<const Type> send_buf,Array<Type>& recv_buf,Int32 rank) override;
  void scatterVariable(Span<const Type> send_buf,Span<Type> recv_buf,Int32 root) override;
  void allReduce(eReduceType op,Span<Type> send_buf) override;
  void allToAll(Span<const Type> send_buf,Span<Type> recv_buf,Int32 count) override;
  void allToAllVariable(Span<const Type> send_buf,ConstArrayView<Int32> send_count,
                        ConstArrayView<Int32> send_index,Span<Type> recv_buf,
                        ConstArrayView<Int32> recv_count,ConstArrayView<Int32> recv_index) override;
  Request send(Span<const Type> send_buffer,Int32 proc,bool is_blocked) override;
  Request send(Span<const Type> send_buffer,const PointToPointMessageInfo& message) override;
  Request receive(Span<Type> recv_buffer,Int32 rank,bool is_blocked) override;
  Request receive(Span<Type> recv_buffer,const PointToPointMessageInfo& message) override;
  Request nonBlockingAllReduce(eReduceType op,Span<const Type> send_buf,Span<Type> recv_buf) override;
  Request nonBlockingAllGather(Span<const Type> send_buf, Span<Type> recv_buf) override;
  Request nonBlockingBroadcast(Span<Type> send_buf, Int32 rank) override;
  Request nonBlockingGather(Span<const Type> send_buf, Span<Type> recv_buf, Int32 rank) override;
  Request nonBlockingAllToAll(Span<const Type> send_buf, Span<Type> recv_buf, Int32 count) override;
  Request nonBlockingAllToAllVariable(Span<const Type> send_buf, ConstArrayView<Int32> send_count,
                                      ConstArrayView<Int32> send_index, Span<Type> recv_buf,
                                      ConstArrayView<Int32> recv_count, ConstArrayView<Int32> recv_index) override;
  Request gather(Arccore::MessagePassing::GatherMessageInfo<Type>&) override;
 
  void broadcast(ArrayView<Type> send_buf,Integer sub_domain) override
  { this->broadcast(Span<Type>(send_buf),sub_domain); }
  void allGather(ConstArrayView<Type> send_buf,ArrayView<Type> recv_buf) override
  { this->allGather(Span<const Type>(send_buf),Span<Type>(recv_buf)); }
  void allGatherVariable(ConstArrayView<Type> send_buf,Array<Type>& recv_buf) override
  { this->allGatherVariable(Span<const Type>(send_buf),recv_buf); }
  void gather(ConstArrayView<Type> send_buf,ArrayView<Type> recv_buf,Integer rank) override
  { this->gather(Span<const Type>(send_buf),Span<Type>(recv_buf),rank); }
  void gatherVariable(ConstArrayView<Type> send_buf,Array<Type>& recv_buf,Integer rank) override
  { this->gatherVariable(Span<const Type>(send_buf),recv_buf,rank); }
  void scatterVariable(ConstArrayView<Type> send_buf,ArrayView<Type> recv_buf,Integer root) override
  { this->scatterVariable(Span<const Type>(send_buf),Span<Type>(recv_buf),root); }
  void allToAll(ConstArrayView<Type> send_buf,ArrayView<Type> recv_buf,Integer count) override
  { this->allToAll(Span<const Type>(send_buf),Span<Type>(recv_buf),count); }
  void allToAllVariable(ConstArrayView<Type> send_buf,Int32ConstArrayView send_count,
                        Int32ConstArrayView send_index,ArrayView<Type> recv_buf,
                        Int32ConstArrayView recv_count,Int32ConstArrayView recv_index) override
  { this->allToAllVariable(Span<const Type>(send_buf),send_count,send_index,
                           Span<Type>(recv_buf),recv_count,recv_index); }
  Request send(ConstArrayView<Type> send_buffer,Integer proc,bool is_blocked) override
  { return this->send(Span<const Type>(send_buffer),proc,is_blocked); }
  Request recv(ArrayView<Type> recv_buffer,Integer proc,bool is_blocked) override
  { return this->receive(Span<Type>(recv_buffer),proc,is_blocked); }
  void allReduce(eReduceType op,ArrayView<Type> send_buf) override
  { return this->allReduce(op,Span<Type>(send_buf)); }
 
  void send(ConstArrayView<Type> send_buffer,Integer proc) override;
  void recv(ArrayView<Type> recv_buffer,Integer proc) override;
  void sendRecv(ConstArrayView<Type> send_buffer,ArrayView<Type> recv_buffer,Integer proc) override;
  Type allReduce(eReduceType op,Type send_buf) override;
  Type scan(eReduceType op,Type send_buf) override;
  void scan(eReduceType op,ArrayView<Type> send_buf) override;
  void computeMinMaxSum(Type val,Type& min_val,Type& max_val,Type& sum_val,
                                Int32& min_rank,
                                Int32& max_rank) override;
  void computeMinMaxSum(ConstArrayView<Type> values,
                        ArrayView<Type> min_values,
                        ArrayView<Type> max_values,
                        ArrayView<Type> sum_values,
                        ArrayView<Int32> min_ranks,
                        ArrayView<Int32> max_ranks) override;
  ITypeDispatcher<Type>* toArccoreDispatcher() override { return this; }
 
 private:
 
  HybridParallelMng* m_parallel_mng = nullptr;
  Int32 m_local_rank = A_NULL_RANK;
  Int32 m_local_nb_rank = 0;
  Int32 m_global_rank = A_NULL_RANK;
  Int32 m_global_nb_rank = 0;
  Int32 m_mpi_rank = A_NULL_RANK;
  Int32 m_mpi_nb_rank = 0;
 
 public:
 
  using AllDispatchView = ArrayView<HybridParallelDispatch<Type>*>;
  Int32 globalRank() const { return m_global_rank; }
  AllDispatchView m_all_dispatchs;
 
 private:
 
  Span<Type> m_broadcast_view;
  Span<const Type> m_const_view;
  Span<Type> m_recv_view;
  Span<const Type> m_send_view;
  AllToAllVariableInfo m_alltoallv_infos;
 
 public:
 
  ReduceInfo m_reduce_infos;
  MinMaxSumInfo m_min_max_sum_infos;
 
 private:
 
  HybridMessageQueue* m_message_queue = nullptr;
  MpiParallelDispatchT<Type>* m_mpi_dispatcher = nullptr;
 
 private:
 
  void _collectiveBarrier();
  void _allReduceOrScan(eReduceType op, Span<Type> send_buf, bool is_scan);
  void _applyReduceOperator(eReduceType op, Span<Type> result, AllDispatchView dispatch_view,
                            Int32 first_rank, Int32 last_rank);
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
} // End namespace Arcane::MessagePassing
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#endif