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/parallel/IStat.h"
32#include "arcane/parallel/mpithread/HybridParallelDispatch.h"
33#include "arcane/parallel/mpithread/HybridMessageQueue.h"
34#include "arcane/parallel/mpithread/internal/HybridMachineMemoryWindowBaseInternalCreator.h"
35#include "arcane/parallel/mpithread/internal/HybridMachineMemoryWindowBaseInternal.h"
36#include "arcane/parallel/mpithread/internal/HybridDynamicMachineMemoryWindowBaseInternal.h"
38#include "arcane/parallel/mpi/MpiParallelMng.h"
40#include "arcane/impl/TimerMng.h"
41#include "arcane/impl/ParallelReplication.h"
42#include "arcane/impl/SequentialParallelMng.h"
43#include "arcane/impl/internal/ParallelMngUtilsFactoryBase.h"
46#include "arccore/message_passing/RequestListBase.h"
47#include "arccore/message_passing/SerializeMessageList.h"
66class HybridSerializeMessageList
70 class HybridSerializeMessageRequest
74 : m_message(message), m_request(request){}
83 : m_parallel_mng(mpm), m_trace(mpm->traceMng())
91 m_messages_to_process.add(msg);
101 case Parallel::WaitAll:
104 _wait(Parallel::WaitAll);
105 m_messages_to_process.clear();
128void HybridSerializeMessageList::
131 m_trace->
info() <<
"BEGIN PROCESS MESSAGES";
151 if (wait_mode==Parallel::WaitAll)
152 message_queue->waitAll(all_requests);
154 for( ISerializeMessage* sm : messages )
155 sm->setFinished(
true);
165:
public ParallelMngInternal
170 : ParallelMngInternal(pm)
172 , m_window_creator(window_creator)
175 ~Impl()
override =
default;
181 return makeRef(m_window_creator->createWindow(m_parallel_mng->commRank(), sizeof_segment, sizeof_type, m_parallel_mng->mpiParallelMng()));
186 return makeRef(m_window_creator->createDynamicWindow(m_parallel_mng->commRank(), sizeof_segment, sizeof_type, m_parallel_mng->mpiParallelMng()));
191 HybridParallelMng* m_parallel_mng;
204, m_trace(bi.trace_mng)
205, m_thread_mng(bi.thread_mng)
206, m_world_parallel_mng(bi.world_parallel_mng)
208, m_timer_mng(nullptr)
210, m_message_queue(new
HybridMessageQueue(bi.message_queue,bi.mpi_parallel_mng,bi.local_nb_rank))
211, m_is_initialized(false)
212, m_stat(
Parallel::createDefaultStat())
213, m_thread_barrier(bi.thread_barrier)
214, m_mpi_parallel_mng(bi.mpi_parallel_mng)
215, m_all_dispatchers(bi.all_dispatchers)
216, m_sub_builder_factory(bi.sub_builder_factory)
217, m_parent_container_ref(bi.container)
219, m_parallel_mng_internal(new Impl(this, bi.window_creator))
221 if (!m_world_parallel_mng)
222 m_world_parallel_mng =
this;
229 Int32 mpi_rank = m_mpi_parallel_mng->commRank();
230 Int32 mpi_size = m_mpi_parallel_mng->commSize();
244 m_sequential_parallel_mng.reset();
245 delete m_replication;
247 delete m_message_queue;
250 delete m_mpi_parallel_mng;
251 delete m_parallel_mng_internal;
263 DispatchCreator(ITraceMng* tm,HybridParallelMng* mpm,HybridMessageQueue* message_queue,MpiThreadAllDispatcher* all_dispatchers)
264 : m_tm(tm), m_mpm(mpm), m_message_queue(message_queue), m_all_dispatchers(all_dispatchers){}
266 template<
typename DataType> HybridParallelDispatch<DataType>*
269 HybridMessageQueue* tmq = m_message_queue;
270 MpiThreadAllDispatcher* ad = m_all_dispatchers;
271 auto field = ad->instance((DataType*)
nullptr).view();
272 return new HybridParallelDispatch<DataType>(m_tm,m_mpm,tmq,field);
276 HybridParallelMng* m_mpm;
277 HybridMessageQueue* m_message_queue;
278 MpiThreadAllDispatcher* m_all_dispatchers;
289 tm->
info() <<
"Initialise HybridParallelMng"
292 <<
" mpi_rank=" << m_mpi_parallel_mng->commRank();
302 m_sequential_parallel_mng = arcaneCreateSequentialParallelMngRef(bi);
305 DispatchCreator creator(m_trace,
this,m_message_queue,m_all_dispatchers);
306 this->createDispatchers(creator);
307 m_io_mng = arcaneCreateIOMng(
this);
321 m_trace->warning() <<
"HybridParallelMng already initialized";
346 return m_utils_factory->createGetVariablesValuesOperation(
this)._release();
352 return m_utils_factory->createTransferValuesOperation(
this)._release();
358 return m_utils_factory->createExchanger(
this)._release();
367void HybridParallelMng::
370 auto p2p_message =
buildMessage(rank,Parallel::NonBlocking);
371 Request r = m_message_queue->addSend(p2p_message,s);
378auto HybridParallelMng::
381 ARCANE_UNUSED(bytes);
382 auto p2p_message =
buildMessage(rank,Parallel::NonBlocking);
383 return m_message_queue->addSend(p2p_message,s);
392 return m_utils_factory->createSendSerializeMessage(
this, rank)._release();
398void HybridParallelMng::
404 bool is_broadcaster = (rank==
commRank());
413 Int64 total_size = sbuf->totalSize();
419 Int64 total_size = 0;
421 sbuf->preallocate(total_size);
424 sbuf->setFromSizes();
431void HybridParallelMng::
432recvSerializer(ISerializer* s,
Int32 rank)
434 auto p2p_message =
buildMessage(rank,Parallel::NonBlocking);
435 Request r = m_message_queue->addReceive(p2p_message,ReceiveBufferInfo(s));
436 m_message_queue->waitAll(ArrayView<Request>(1,&r));
445 return m_utils_factory->createReceiveSerializeMessage(
this, rank)._release();
454 ARCANE_UNUSED(requests);
466 return m_message_queue->probe(p2p_message);
477 return m_message_queue->legacyProbe(p2p_message);
487 return m_message_queue->addSend(p2p_message,s);
507 m_stat->print(m_trace);
516 m_thread_barrier->wait();
518 m_mpi_parallel_mng->barrier();
519 m_thread_barrier->wait();
526_createSerializeMessageList()
529 x->setAllowAnyRankReceive(
false);
539 return m_utils_factory->createSynchronizer(
this,family)._release();
548 return m_utils_factory->createSynchronizer(
this,group)._release();
557 return m_utils_factory->createTopology(
this)._release();
566 return m_replication;
575 delete m_replication;
585 return m_sequential_parallel_mng.get();
592sequentialParallelMngRef()
594 return m_sequential_parallel_mng;
607 RequestList(HybridParallelMng* pm)
608 : m_parallel_mng(pm), m_message_queue(pm->m_message_queue),
609 m_local_rank(m_parallel_mng->localRank()) {}
614 case Parallel::WaitAll:
615 m_parallel_mng->m_message_queue->waitAll(_requests());
618 m_message_queue->waitSome(m_local_rank,_requests(),_requestsDone(),
false);
621 m_message_queue->waitSome(m_local_rank,_requests(),_requestsDone(),
true);
625 HybridParallelMng* m_parallel_mng;
646 m_message_queue->waitAll(requests);
655 return m_mpi_parallel_mng->getMPICommunicator();
664 return m_mpi_parallel_mng->communicator();
690IParallelMng* HybridParallelMng::
693 ARCANE_UNUSED(kept_ranks);
694 ARCANE_THROW(NotSupportedException,
"Use createSubParallelMngRef() instead");
706 if (kept_ranks.
empty())
710 m_trace->info() <<
"CREATE SUB_PARALLEL_MNG_REF";
730 Int32 my_new_global_rank = (-1);
731 Int32 new_local_nb_rank = 0;
732 Int32 my_new_local_rank = (-1);
733 for(
Integer i=0; i<nb_kept_rank; ++i ){
734 Int32 kept_rank = kept_ranks[i];
735 if (kept_rank>=first_global_rank_in_this_mpi && kept_rank<last_global_rank_in_this_mpi)
738 my_new_global_rank = i;
739 my_new_local_rank = new_local_nb_rank - 1;
742 bool has_new_rank = (my_new_global_rank != (-1));
753 Int32 min_new_local_nb_rank = -1;
754 Int32 max_new_local_nb_rank = -1;
755 Int32 sum_new_local_nb_rank = -1;
756 Int32 min_rank = A_NULL_RANK;
757 Int32 max_rank = A_NULL_RANK;
758 computeMinMaxSum(new_local_nb_rank,min_new_local_nb_rank,max_new_local_nb_rank,
759 sum_new_local_nb_rank,min_rank,max_rank);
761 m_trace->info() <<
"CREATE SUB_PARALLEL_MNG_REF new_local_nb_rank=" << new_local_nb_rank
762 <<
" min=" << min_new_local_nb_rank
763 <<
" max=" << max_new_local_nb_rank
764 <<
" sum=" << sum_new_local_nb_rank
765 <<
" new_global_rank=" << my_new_global_rank;
769 if (max_new_local_nb_rank==1){
770 Integer nb_mpi_rank = m_mpi_parallel_mng->commSize();
780 bool do_mpi_call =
false;
781 if (min_new_local_nb_rank==1){
784 kept_mpi_ranks.
resize(nb_mpi_rank);
785 for(
Int32 x=0; x<nb_mpi_rank; ++x )
786 kept_mpi_ranks[x] = x;
795 Int16 v = (has_new_rank) ? 1 : 0;
798 for(
Int32 x=0; x<nb_mpi_rank; ++x )
799 if (gathered_ranks[x]==1)
800 kept_mpi_ranks.
add(x);
803 return m_mpi_parallel_mng->createSubParallelMngRef(kept_mpi_ranks);
808 if (max_new_local_nb_rank!=new_local_nb_rank)
809 ARCANE_FATAL(
"Not same number of new local ranks on every MPI processus: current={0} max={1}",
810 new_local_nb_rank,max_new_local_nb_rank);
812 if (max_new_local_nb_rank<2)
813 ARCANE_FATAL(
"number of local ranks is too low current={0} minimum=2",new_local_nb_rank);
816 m_thread_barrier->wait();
828 builder = m_sub_builder_factory->_createParallelMngBuilder(new_local_nb_rank,c);
830 m_all_dispatchers->m_create_sub_parallel_mng_info.m_builder = builder;
833 m_thread_barrier->wait();
835 builder = m_all_dispatchers->m_create_sub_parallel_mng_info.m_builder;
839 if (my_new_local_rank>=0){
840 new_parallel_mng = builder->_createParallelMng(my_new_local_rank,
traceMng());
842 m_thread_barrier->wait();
848 m_all_dispatchers->m_create_sub_parallel_mng_info.m_builder.reset();
850 m_thread_barrier->wait();
852 return new_parallel_mng;
861 return m_utils_factory;
867bool HybridParallelMng::
868_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.
É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.
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.