14#include "arcane/parallel/thread/SharedMemoryParallelMng.h"
16#include "arcane/utils/NotImplementedException.h"
17#include "arcane/utils/NotSupportedException.h"
18#include "arcane/utils/PlatformUtils.h"
19#include "arcane/utils/TraceInfo.h"
20#include "arcane/utils/NumericTypes.h"
21#include "arcane/utils/ArgumentException.h"
22#include "arcane/utils/FatalErrorException.h"
23#include "arcane/utils/ITraceMng.h"
25#include "arcane/parallel/IStat.h"
27#include "arcane/parallel/thread/SharedMemoryParallelDispatch.h"
28#include "arcane/parallel/thread/ISharedMemoryMessageQueue.h"
30#include "arcane/core/Timer.h"
31#include "arcane/core/IIOMng.h"
32#include "arcane/core/ISerializeMessageList.h"
33#include "arcane/core/IItemFamily.h"
34#include "arcane/core/internal/SerializeMessage.h"
36#include "arcane/impl/TimerMng.h"
37#include "arcane/impl/ParallelReplication.h"
38#include "arcane/impl/internal/ParallelMngUtilsFactoryBase.h"
40#include "arccore/message_passing/RequestListBase.h"
41#include "arccore/message_passing/SerializeMessageList.h"
50extern "C++" ARCANE_IMPL_EXPORT
IIOMng*
68 explicit RequestList(SharedMemoryParallelMng* pm)
69 : m_parallel_mng(pm), m_message_queue(pm->m_message_queue),
70 m_local_rank(m_parallel_mng->commRank()){}
75 case Parallel::WaitAll:
76 return m_message_queue->waitAll(_requests());
78 return m_message_queue->waitSome(m_local_rank,_requests(),_requestsDone(),
false);
80 return m_message_queue->waitSome(m_local_rank,_requests(),_requestsDone(),
true);
84 SharedMemoryParallelMng* m_parallel_mng;
92SharedMemoryParallelMng::
95, m_trace(build_info.trace_mng)
96, m_thread_mng(build_info.thread_mng)
97, m_sequential_parallel_mng(build_info.sequential_parallel_mng)
100, m_world_parallel_mng(build_info.world_parallel_mng)
102, m_message_queue(build_info.message_queue)
103, m_is_parallel(build_info.nb_rank!=1)
104, m_rank(build_info.rank)
105, m_nb_rank(build_info.nb_rank)
106, m_is_initialized(false)
107, m_stat(
Parallel::createDefaultStat())
108, m_thread_barrier(build_info.thread_barrier)
109, m_all_dispatchers(build_info.all_dispatchers)
110, m_sub_builder_factory(build_info.sub_builder_factory)
111, m_parent_container_ref(build_info.container)
112, m_mpi_communicator(build_info.communicator)
115 if (!m_world_parallel_mng)
116 m_world_parallel_mng =
this;
122SharedMemoryParallelMng::
123~SharedMemoryParallelMng()
125 delete m_replication;
126 m_sequential_parallel_mng.reset();
141 DispatchCreator(ITraceMng* tm,SharedMemoryParallelMng* mpm,
142 ISharedMemoryMessageQueue* message_queue,
143 SharedMemoryAllDispatcher* all_dispatchers)
144 : m_tm(tm), m_mpm(mpm), m_message_queue(message_queue),
145 m_all_dispatchers(all_dispatchers){}
147 template<
typename DataType> SharedMemoryParallelDispatch<DataType>*
150 ISharedMemoryMessageQueue* tmq = m_message_queue;
151 SharedMemoryAllDispatcher* ad = m_all_dispatchers;
152 auto& field = ad->instance((DataType*)
nullptr);
153 return new SharedMemoryParallelDispatch<DataType>(m_tm,m_mpm,tmq,field);
157 SharedMemoryParallelMng* m_mpm;
158 ISharedMemoryMessageQueue* m_message_queue;
159 SharedMemoryAllDispatcher* m_all_dispatchers;
172 DispatchCreator creator(m_trace.get(),
this,m_message_queue,m_all_dispatchers);
173 this->createDispatchers(creator);
175 m_io_mng = arcaneCreateIOMng(
this);
186 m_trace->warning() <<
"SharedMemoryParallelMng already initialized";
199 return m_utils_factory->createGetVariablesValuesOperation(
this)._release();
205 return m_utils_factory->createTransferValuesOperation(
this)._release();
211 return m_utils_factory->createExchanger(
this)._release();
220void SharedMemoryParallelMng::
223 auto p2p_message =
buildMessage(dest_rank,Parallel::Blocking);
234 ARCANE_UNUSED(bytes);
244 return m_utils_factory->createSendSerializeMessage(
this, rank)._release();
250void SharedMemoryParallelMng::
262 m_message_queue->waitAll(requests);
265 recvSerializer(values,rank);
272void SharedMemoryParallelMng::
273recvSerializer(ISerializer* values,
Int32 rank)
275 auto p2p_message =
buildMessage(rank,Parallel::Blocking);
276 Request r = m_message_queue->addReceive(p2p_message,ReceiveBufferInfo(values));
277 m_message_queue->waitAll(ArrayView<Request>(1,&r));
286 return m_utils_factory->createReceiveSerializeMessage(
this, rank)._release();
295 ARCANE_UNUSED(requests);
306 m_stat->print(m_trace.get());
315 m_thread_barrier->wait();
325 m_message_queue->waitAll(requests);
327 m_stat->add(
"WaitAll",(end_time-begin_time),0);
334_createSerializeMessageList()
347 return m_message_queue->probe(p2p_message);
358 return m_message_queue->legacyProbe(p2p_message);
364auto SharedMemoryParallelMng::
368 return m_message_queue->addSend(p2p_message,
SendBufferInfo(values));
374auto SharedMemoryParallelMng::
387 return m_utils_factory->createSynchronizer(
this,family)._release();
396 return m_utils_factory->createSynchronizer(
this,group)._release();
405 return m_utils_factory->createTopology(
this)._release();
414 return m_replication;
423 delete m_replication;
433 ARCANE_UNUSED(kept_ranks);
446 if (kept_ranks.
empty())
455 Int32 my_new_rank = (-1);
456 for(
Integer i=0; i<nb_rank; ++i )
457 if (kept_ranks[i]==
m_rank){
465 builder = m_sub_builder_factory->_createParallelMngBuilder(nb_rank,m_mpi_communicator);
467 m_all_dispatchers->m_create_sub_parallel_mng_info.m_builder = builder;
471 builder = m_all_dispatchers->m_create_sub_parallel_mng_info.m_builder;
476 new_parallel_mng = builder->_createParallelMng(my_new_rank,
traceMng());
487 m_all_dispatchers->m_create_sub_parallel_mng_info.m_builder.reset();
491 return new_parallel_mng;
508sequentialParallelMngRef()
510 return m_sequential_parallel_mng;
516 return m_sequential_parallel_mng.get();
545 return m_utils_factory;
#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.
Vue modifiable d'un tableau d'un type T.
void add(ConstReferenceType val)
Ajoute l'élément val à la fin du tableau.
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.
Echange d'informations entre processeurs.
Interface du gestionnaire de parallélisme pour un sous-domaine.
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.
Envoie de valeurs sur différents processeurs.
Interface d'un service de synchronisation de variable.
Groupe d'entités de maillage.
Liste de requête de messages.
Interface d'une liste de messages de sérialisation.
Interface d'un message de sérialisation entre IMessagePassingMng.
Interface d'une file de messages avec les threads.
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.
Implémentation de IRequestList pour SharedMemoryParallelMng.
void _wait(Parallel::eWaitType wait_type) override
Effectue l'attente ou le test.
IParallelTopology * createTopology() override
Créé une instance contenant les infos sur la topologie des rangs de ce gestionnnaire.
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...
Ref< Parallel::IRequestList > createRequestListRef() override
Créé une liste de requêtes pour ce gestionnaire.
void build() override
Construit l'instance.
Int32 m_nb_rank
Nombre de rangs.
bool m_is_initialized
true si déjà initialisé
void printStats() override
Affiche des statistiques liées à ce gestionnaire du parallélisme.
void initialize() override
Initialise le gestionnaire du parallélisme.
void barrier() override
Effectue une barière.
void setReplication(IParallelReplication *v) override
Positionne les Informations sur la réplication.
PointToPointMessageInfo buildMessage(Int32 dest, MP::eBlockingType is_blocking)
Construit un message avec pour destinataire dest.
void waitAllRequests(ArrayView< Request > requests) override
Bloque en attendant que les requêtes rvalues soient terminées.
Ref< IParallelMngUtilsFactory > _internalUtilsFactory() const override
Fabrique des fonctions utilitaires.
IVariableSynchronizer * createSynchronizer(IItemFamily *family) override
Retourne une interface pour synchroniser des variables sur le groupe de la famille family.
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.
ITraceMng * traceMng() const override
Gestionnaire de traces.
IParallelMng * sequentialParallelMng() override
Retourne un gestionnaire de parallélisme séquentiel.
Int32 m_rank
Rang de l'instance.
IParallelExchanger * createExchanger() override
Retourne une interface pour transférer des messages entre processeurs.
void freeRequests(ArrayView< Parallel::Request > requests) override
Libère les requêtes.
ISerializeMessage * createReceiveSerializer(Int32 rank) override
Créé un message non bloquant pour recevoir des données sérialisées du rang rank.
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.
MessageSourceInfo legacyProbe(const PointToPointMessageInfo &message) override
Sonde si des messages sont disponibles.
IParallelReplication * replication() const override
Informations sur la réplication.
Classe de base d'une liste de requêtes.
Exception lorsqu'une fonction n'est pas implémentée.
Exception lorsqu'une opération n'est pas supporté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é
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.
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.
@ 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 -*-
Ref< TrueType > createRef(Args &&... args)
Créé une instance de type TrueType avec les arguments Args et retourne une référence dessus.
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.
double Real
Type représentant un réel.
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 SharedMemoryParallelMng.