14#include "arcane/parallel/mpithread/HybridParallelMng.h"
16#include "arcane/utils/NotImplementedException.h"
17#include "arcane/utils/NotSupportedException.h"
18#include "arcane/utils/FatalErrorException.h"
19#include "arcane/utils/NumericTypes.h"
20#include "arcane/utils/ArgumentException.h"
21#include "arcane/utils/IThreadBarrier.h"
22#include "arcane/utils/ITraceMng.h"
24#include "arcane/core/IIOMng.h"
25#include "arcane/core/Timer.h"
26#include "arcane/core/ISerializeMessageList.h"
27#include "arcane/core/IItemFamily.h"
28#include "arcane/core/internal/ParallelMngInternal.h"
29#include "arcane/core/internal/SerializeMessage.h"
30#include "arcane/core/internal/DynamicMachineMemoryWindowMemoryAllocator.h"
31#include "arcane/core/parallel/IStat.h"
33#include "arcane/parallel/mpithread/HybridParallelDispatch.h"
34#include "arcane/parallel/mpithread/HybridMessageQueue.h"
35#include "arcane/parallel/mpithread/internal/HybridMachineMemoryWindowBaseInternalCreator.h"
36#include "arcane/parallel/mpithread/internal/HybridMachineMemoryWindowBaseInternal.h"
37#include "arcane/parallel/mpithread/internal/HybridDynamicMachineMemoryWindowBaseInternal.h"
39#include "arcane/parallel/mpi/MpiParallelMng.h"
41#include "arcane/impl/TimerMng.h"
42#include "arcane/impl/ParallelReplication.h"
43#include "arcane/impl/SequentialParallelMng.h"
44#include "arcane/impl/internal/ParallelMngUtilsFactoryBase.h"
47#include "arccore/message_passing/RequestListBase.h"
48#include "arccore/message_passing/internal/SerializeMessageList.h"
67class HybridSerializeMessageList
71 class HybridSerializeMessageRequest
75 : m_message(message), m_request(request){}
84 : m_parallel_mng(mpm), m_trace(mpm->traceMng())
92 m_messages_to_process.add(msg);
102 case Parallel::WaitAll:
105 _wait(Parallel::WaitAll);
106 m_messages_to_process.clear();
129void HybridSerializeMessageList::
132 m_trace->
info() <<
"BEGIN PROCESS MESSAGES";
152 if (wait_mode==Parallel::WaitAll)
153 message_queue->waitAll(all_requests);
155 for( ISerializeMessage* sm : messages )
156 sm->setFinished(
true);
166:
public ParallelMngInternal
171 : ParallelMngInternal(pm)
173 , m_window_creator(window_creator)
177 ~Impl()
override =
default;
183 return makeRef(m_window_creator->createWindow(m_parallel_mng->commRank(), sizeof_segment, sizeof_type, m_parallel_mng->mpiParallelMng()));
188 return makeRef(m_window_creator->createDynamicWindow(m_parallel_mng->commRank(), sizeof_segment, sizeof_type, m_parallel_mng->mpiParallelMng()));
193 return m_alloc.get();
212, m_trace(bi.trace_mng)
213, m_thread_mng(bi.thread_mng)
214, m_world_parallel_mng(bi.world_parallel_mng)
216, m_timer_mng(nullptr)
218, m_message_queue(new
HybridMessageQueue(bi.message_queue,bi.mpi_parallel_mng,bi.local_nb_rank))
219, m_is_initialized(false)
220, m_stat(
Parallel::createDefaultStat())
221, m_thread_barrier(bi.thread_barrier)
222, m_mpi_parallel_mng(bi.mpi_parallel_mng)
223, m_all_dispatchers(bi.all_dispatchers)
224, m_sub_builder_factory(bi.sub_builder_factory)
225, m_parent_container_ref(bi.container)
227, m_parallel_mng_internal(new Impl(this, bi.window_creator))
229 if (!m_world_parallel_mng)
230 m_world_parallel_mng =
this;
237 Int32 mpi_rank = m_mpi_parallel_mng->commRank();
238 Int32 mpi_size = m_mpi_parallel_mng->commSize();
252 m_sequential_parallel_mng.reset();
253 delete m_replication;
255 delete m_message_queue;
258 delete m_mpi_parallel_mng;
259 delete m_parallel_mng_internal;
271 DispatchCreator(ITraceMng* tm,HybridParallelMng* mpm,HybridMessageQueue* message_queue,MpiThreadAllDispatcher* all_dispatchers)
272 : m_tm(tm), m_mpm(mpm), m_message_queue(message_queue), m_all_dispatchers(all_dispatchers){}
274 template<
typename DataType> HybridParallelDispatch<DataType>*
277 HybridMessageQueue* tmq = m_message_queue;
278 MpiThreadAllDispatcher* ad = m_all_dispatchers;
279 auto field = ad->instance((DataType*)
nullptr).view();
280 return new HybridParallelDispatch<DataType>(m_tm,m_mpm,tmq,field);
284 HybridParallelMng* m_mpm;
285 HybridMessageQueue* m_message_queue;
286 MpiThreadAllDispatcher* m_all_dispatchers;
297 tm->
info() <<
"Initialise HybridParallelMng"
300 <<
" mpi_rank=" << m_mpi_parallel_mng->commRank();
310 m_sequential_parallel_mng = arcaneCreateSequentialParallelMngRef(bi);
313 DispatchCreator creator(m_trace,
this,m_message_queue,m_all_dispatchers);
314 this->createDispatchers(creator);
315 m_io_mng = arcaneCreateIOMng(
this);
329 m_trace->warning() <<
"HybridParallelMng already initialized";
354 return m_utils_factory->createGetVariablesValuesOperation(
this)._release();
360 return m_utils_factory->createTransferValuesOperation(
this)._release();
366 return m_utils_factory->createExchanger(
this)._release();
375void HybridParallelMng::
378 auto p2p_message =
buildMessage(rank,Parallel::NonBlocking);
379 Request r = m_message_queue->addSend(p2p_message,s);
386auto HybridParallelMng::
389 ARCANE_UNUSED(bytes);
390 auto p2p_message =
buildMessage(rank,Parallel::NonBlocking);
391 return m_message_queue->addSend(p2p_message,s);
400 return m_utils_factory->createSendSerializeMessage(
this, rank)._release();
406void HybridParallelMng::
412 bool is_broadcaster = (rank==
commRank());
421 Int64 total_size = sbuf->totalSize();
427 Int64 total_size = 0;
429 sbuf->preallocate(total_size);
432 sbuf->setFromSizes();
439void HybridParallelMng::
440recvSerializer(ISerializer* s,
Int32 rank)
442 auto p2p_message =
buildMessage(rank,Parallel::NonBlocking);
443 Request r = m_message_queue->addReceive(p2p_message,ReceiveBufferInfo(s));
444 m_message_queue->waitAll(ArrayView<Request>(1,&r));
453 return m_utils_factory->createReceiveSerializeMessage(
this, rank)._release();
462 ARCANE_UNUSED(requests);
474 return m_message_queue->probe(p2p_message);
485 return m_message_queue->legacyProbe(p2p_message);
495 return m_message_queue->addSend(p2p_message,s);
515 m_stat->print(m_trace);
524 m_thread_barrier->wait();
526 m_mpi_parallel_mng->barrier();
527 m_thread_barrier->wait();
534_createSerializeMessageList()
537 x->setAllowAnyRankReceive(
false);
547 return m_utils_factory->createSynchronizer(
this,family)._release();
556 return m_utils_factory->createSynchronizer(
this,group)._release();
565 return m_utils_factory->createTopology(
this)._release();
574 return m_replication;
583 delete m_replication;
593 return m_sequential_parallel_mng.get();
600sequentialParallelMngRef()
602 return m_sequential_parallel_mng;
615 RequestList(HybridParallelMng* pm)
616 : m_parallel_mng(pm), m_message_queue(pm->m_message_queue),
617 m_local_rank(m_parallel_mng->localRank()) {}
622 case Parallel::WaitAll:
623 m_parallel_mng->m_message_queue->waitAll(_requests());
626 m_message_queue->waitSome(m_local_rank,_requests(),_requestsDone(),
false);
629 m_message_queue->waitSome(m_local_rank,_requests(),_requestsDone(),
true);
633 HybridParallelMng* m_parallel_mng;
654 m_message_queue->waitAll(requests);
663 return m_mpi_parallel_mng->getMPICommunicator();
672 return m_mpi_parallel_mng->communicator();
681 return m_mpi_parallel_mng->machineCommunicator();
707IParallelMng* HybridParallelMng::
710 ARCANE_UNUSED(kept_ranks);
711 ARCANE_THROW(NotSupportedException,
"Use createSubParallelMngRef() instead");
723 if (kept_ranks.
empty())
727 m_trace->info() <<
"CREATE SUB_PARALLEL_MNG_REF";
747 Int32 my_new_global_rank = (-1);
748 Int32 new_local_nb_rank = 0;
749 Int32 my_new_local_rank = (-1);
750 for(
Integer i=0; i<nb_kept_rank; ++i ){
751 Int32 kept_rank = kept_ranks[i];
752 if (kept_rank>=first_global_rank_in_this_mpi && kept_rank<last_global_rank_in_this_mpi)
755 my_new_global_rank = i;
756 my_new_local_rank = new_local_nb_rank - 1;
759 bool has_new_rank = (my_new_global_rank != (-1));
770 Int32 min_new_local_nb_rank = -1;
771 Int32 max_new_local_nb_rank = -1;
772 Int32 sum_new_local_nb_rank = -1;
773 Int32 min_rank = A_NULL_RANK;
774 Int32 max_rank = A_NULL_RANK;
775 computeMinMaxSum(new_local_nb_rank,min_new_local_nb_rank,max_new_local_nb_rank,
776 sum_new_local_nb_rank,min_rank,max_rank);
778 m_trace->info() <<
"CREATE SUB_PARALLEL_MNG_REF new_local_nb_rank=" << new_local_nb_rank
779 <<
" min=" << min_new_local_nb_rank
780 <<
" max=" << max_new_local_nb_rank
781 <<
" sum=" << sum_new_local_nb_rank
782 <<
" new_global_rank=" << my_new_global_rank;
786 if (max_new_local_nb_rank==1){
787 Integer nb_mpi_rank = m_mpi_parallel_mng->commSize();
797 bool do_mpi_call =
false;
798 if (min_new_local_nb_rank==1){
801 kept_mpi_ranks.
resize(nb_mpi_rank);
802 for(
Int32 x=0; x<nb_mpi_rank; ++x )
803 kept_mpi_ranks[x] = x;
812 Int16 v = (has_new_rank) ? 1 : 0;
815 for(
Int32 x=0; x<nb_mpi_rank; ++x )
816 if (gathered_ranks[x]==1)
817 kept_mpi_ranks.
add(x);
820 return m_mpi_parallel_mng->createSubParallelMngRef(kept_mpi_ranks);
825 if (max_new_local_nb_rank!=new_local_nb_rank)
826 ARCANE_FATAL(
"Not same number of new local ranks on every MPI processus: current={0} max={1}",
827 new_local_nb_rank,max_new_local_nb_rank);
829 if (max_new_local_nb_rank<2)
830 ARCANE_FATAL(
"number of local ranks is too low current={0} minimum=2",new_local_nb_rank);
833 m_thread_barrier->wait();
846 builder = m_sub_builder_factory->_createParallelMngBuilder(new_local_nb_rank, c, mc);
848 m_all_dispatchers->m_create_sub_parallel_mng_info.m_builder = builder;
851 m_thread_barrier->wait();
853 builder = m_all_dispatchers->m_create_sub_parallel_mng_info.m_builder;
857 if (my_new_local_rank>=0){
858 new_parallel_mng = builder->_createParallelMng(my_new_local_rank,
traceMng());
860 m_thread_barrier->wait();
866 m_all_dispatchers->m_create_sub_parallel_mng_info.m_builder.reset();
868 m_thread_barrier->wait();
870 return new_parallel_mng;
879 return m_utils_factory;
885bool HybridParallelMng::
886_isAcceleratorAware()
const
#define ARCANE_CHECK_POINTER(ptr)
Macro retournant le pointeur ptr s'il est non nul ou lancant une exception s'il est nul.
#define ARCANE_THROW(exception_class,...)
Macro pour envoyer une exception avec formattage.
#define ARCANE_FATAL(...)
Macro envoyant une exception FatalErrorException.
Liste des fonctions d'échange de message.
Exception lorsqu'un argument est invalide.
Vue modifiable d'un tableau d'un type T.
void resize(Int64 s)
Change le nombre d'éléments du tableau à s.
void add(ConstReferenceType val)
Ajoute l'élément val à la fin du tableau.
Vue constante d'un tableau de type T.
constexpr Integer size() const noexcept
Nombre d'éléments du tableau.
constexpr bool empty() const noexcept
true si le tableau est vide (size()==0)
Opérations pour accéder aux valeurs de variables d'un autre sous-domaine.
Interface du gestionnaire des entrées sorties.
Interface d'une famille d'entités.
Interface d'un allocateur pour la mémoire.
Échange d'informations entre processeurs.
virtual bool isAcceleratorAware() const =0
Indique si l'implémentation gère les accélérateurs.
Interface du gestionnaire de parallélisme pour un sous-domaine.
virtual void computeMinMaxSum(char val, char &min_val, char &max_val, char &sum_val, Int32 &min_rank, Int32 &max_rank)=0
Calcule en une opération la somme, le min, le max d'une valeur.
Informations sur la réplication des sous-domaines en parallèle.
Informations sur la topologie d'allocation des coeurs de calcul.
Interface d'un sérialiseur.
Interface du gestionnaire de traces.
virtual TraceMessage info()=0
Flot pour un message d'information.
Envoie de valeurs sur différents processeurs.
Interface d'un service de synchronisation de variable.
Groupe d'entités de maillage.
Interface d'une file de messages avec les threads.
Ref< IMachineMemoryWindowBaseInternal > createMachineMemoryWindowBase(Int64 sizeof_segment, Int32 sizeof_type) override
Méthode permettant de créer une fenêtre mémoire sur le noeud.
Ref< IDynamicMachineMemoryWindowBaseInternal > createDynamicMachineMemoryWindowBase(Int64 sizeof_segment, Int32 sizeof_type) override
Méthode permettant de créer une fenêtre mémoire dynamique sur le noeud.
Implémentation de IRequestList pour HybridParallelMng.
void _wait(Parallel::eWaitType wait_type) override
Effectue l'attente ou le test.
Gestionnaire du parallélisme utilisant les threads.
IParallelMng * worldParallelMng() const override
Gestionnaire de parallélisme sur l'ensemble des ressources allouées.
void freeRequests(ArrayView< Request > requests) override
Libère les requêtes.
void printStats() override
Affiche des statistiques liées à ce gestionnaire du parallélisme.
IParallelExchanger * createExchanger() override
Retourne une interface pour transférer des messages entre processeurs.
void barrier() override
Effectue une barière.
MP::Communicator communicator() const override
Communicateur MPI associé à ce gestionnaire.
bool m_is_initialized
true si déjà initialisé
ISerializeMessage * createReceiveSerializer(Int32 rank) override
Créé un message non bloquant pour recevoir des données sérialisées du rang rank.
Ref< Parallel::IRequestList > createRequestListRef() override
Créé une liste de requêtes pour ce gestionnaire.
ITimerMng * timerMng() const override
Gestionnaire de timers.
void build() override
Construit l'instance.
void waitAllRequests(ArrayView< Request > requests) override
Bloque en attendant que les requêtes rvalues soient terminées.
void initialize() override
Initialise le gestionnaire du parallélisme.
IThreadMng * threadMng() const override
Gestionnaire de threads.
Int32 m_global_nb_rank
Nombre de rangs globaux.
MessageSourceInfo legacyProbe(const PointToPointMessageInfo &message) override
Sonde si des messages sont disponibles.
Int32 m_local_rank
Rang local du processeur actuel.
ITraceMng * traceMng() const override
Gestionnaire de traces.
IGetVariablesValuesParallelOperation * createGetVariablesValuesOperation() override
Retourne une opération pour récupérer les valeurs d'une variable sur les entités d'un autre sous-doma...
void * getMPICommunicator() override
Adresse du communicateur MPI associé à ce gestionnaire.
IParallelReplication * replication() const override
Informations sur la réplication.
Ref< IParallelMngUtilsFactory > _internalUtilsFactory() const override
Fabrique des fonctions utilitaires.
Ref< IParallelMng > createSubParallelMngRef(Int32ConstArrayView kept_ranks) override
Créé un nouveau gestionnaire de parallélisme pour un sous-ensemble des rangs.
MP::Communicator machineCommunicator() const override
Communicateur MPI issus du communicateur communicator() réunissant tous les processus du noeud de cal...
ITransferValuesParallelOperation * createTransferValuesOperation() override
Retourne une opération pour transférer des valeurs entre sous-domaine.
IParallelTopology * createTopology() override
Créé une instance contenant les infos sur la topologie des rangs de ce gestionnnaire.
void setReplication(IParallelReplication *v) override
Positionne les Informations sur la réplication.
IParallelMng * sequentialParallelMng() override
Retourne un gestionnaire de parallélisme séquentiel.
IVariableSynchronizer * createSynchronizer(IItemFamily *family) override
Retourne une interface pour synchroniser des variables sur le groupe de la famille family.
Int32 m_local_nb_rank
Nombre de rang locaux.
MessageId probe(const PointToPointMessageInfo &message) override
Sonde si des messages sont disponibles.
ISerializeMessage * createSendSerializer(Int32 rank) override
Créé un message non bloquant pour envoyer des données sérialisées au rang rank.
Int32 commRank() const override
Rang de cette instance dans le communicateur.
Int32 m_global_rank
Numéro du processeur actuel.
PointToPointMessageInfo buildMessage(Int32 dest, MP::eBlockingType is_blocking)
Construit un message avec pour destinataire dest.
void processPendingMessages() override
Envoie les messages de la liste qui ne l'ont pas encore été.
Integer waitMessages(Parallel::eWaitType wait_type) override
Attend que les messages aient terminé leur exécution.
void addMessage(ISerializeMessage *msg) override
Ajoute un message à la liste.
Interface du gestionnaire des échanges de messages.
Liste de requête de messages.
Interface d'une liste de messages de sérialisation.
Interface d'un message de sérialisation entre IMessagePassingMng.
Informations sur la source d'un message.
Informations pour envoyer/recevoir un message point à point.
void setEmiterRank(MessageRank rank)
Positionne le rang de l'émetteur du message.
Informations des buffers de réception.
Informations des buffers d'envoie.
Classe de base d'une liste de requêtes.
IParallelMngInternal * _internalApi() override
API interne à Arcane.
Exception lorsqu'une fonction n'est pas implémentée.
Redirige la gestion des messages des sous-domaines suivant le type de l'argument.
IMessagePassingMng * messagePassingMng() const override
Gestionnaire de message de Arccore associé
ITimeStats * timeStats() const override
Gestionnaire de statistiques associé (peut être nul)
Classe de base d'une fabrique pour les fonctions utilitaires de IParallelMng.
Informations sur la réplication des sous-domaines en parallèle.
Référence à une instance.
InstanceType * get() const
Instance associée ou nullptr si aucune.
Implémentation d'un tampon pour la sérialisation.
Vue d'un tableau d'éléments de type T.
Positionne la phase de l'action en cours d'exécution.
Positionne une classe de message.
Vecteur 1D de données avec sémantique par valeur (style STL).
Déclarations des types et méthodes utilisés par les mécanismes d'échange de messages.
C void mpBroadcast(IMessagePassingMng *pm, Span< char > send_buf, Int32 rank)
@ WaitSome
Attend que tous les messages de la liste soient traités.
eBlockingType
Type indiquant si un message est bloquant ou non.
Implémentation de la concurrence.
-*- tab-width: 2; indent-tabs-mode: nil; coding: utf-8-with-signature -*-
ArrayView< Int64 > Int64ArrayView
Equivalent C d'un tableau à une dimension d'entiers 64 bits.
Ref< TrueType > createRef(Args &&... args)
Créé une instance de type TrueType avec les arguments Args et retourne une référence dessus.
std::int64_t Int64
Type entier signé sur 64 bits.
Int32 Integer
Type représentant un entier.
Array< Byte > ByteArray
Tableau dynamique à une dimension de caractères.
ConstArrayView< Int32 > Int32ConstArrayView
Equivalent C d'un tableau à une dimension d'entiers 32 bits.
std::int16_t Int16
Type entier signé sur 16 bits.
auto makeRef(InstanceType *t) -> Ref< InstanceType >
Créé une référence sur un pointeur.
std::int32_t Int32
Type entier signé sur 32 bits.
Infos pour construire un HybridParallelMng.
Infos pour construire un SequentialParallelMng.