14#include "arcane/utils/Array.h"
15#include "arcane/utils/HPReal.h"
16#include "arcane/utils/NumericTypes.h"
17#include "arcane/utils/NotImplementedException.h"
18#include "arcane/utils/ScopedPtr.h"
19#include "arcane/utils/ITraceMng.h"
20#include "arcane/utils/ValueConvert.h"
21#include "arcane/utils/FatalErrorException.h"
23#include "arcane/core/ParallelMngDispatcher.h"
24#include "arcane/core/IParallelDispatch.h"
25#include "arcane/core/Timer.h"
26#include "arcane/core/ITimeStats.h"
27#include "arcane/core/IParallelNonBlockingCollective.h"
28#include "arcane/core/internal/ParallelMngInternal.h"
30#include "arcane/accelerator/core/Runner.h"
31#include "arcane/accelerator/core/RunQueueBuildInfo.h"
33#include "arccore/message_passing/Dispatchers.h"
35#include "arccore/message_passing/MessagePassingMng.h"
36#include "arccore/message_passing/IControlDispatcher.h"
37#include "arccore/message_passing/ISerializeDispatcher.h"
38#include "arccore/message_passing/PointToPointMessageInfo.h"
39#include "arccore/message_passing/IRequestList.h"
40#include "arccore/message_passing/ISerializeMessageList.h"
41#include "arccore/trace/internal/TimeMetric.h"
48using namespace Arccore::MessagePassing;
53ParallelMngDispatcherBuildInfo::
54ParallelMngDispatcherBuildInfo(
Int32 comm_rank,
Int32 comm_size)
55: m_comm_rank(comm_rank)
56, m_comm_size(comm_size)
65ParallelMngDispatcherBuildInfo::
66ParallelMngDispatcherBuildInfo(Int32 comm_rank, Int32 comm_size, MP::Communicator communicator)
67: m_comm_rank(comm_rank)
68, m_comm_size(comm_size)
69, m_communicator(communicator)
77void ParallelMngDispatcherBuildInfo::
86ParallelMngDispatcher::DefaultControlDispatcher::
87DefaultControlDispatcher(IParallelMng* pm)
92void ParallelMngDispatcher::DefaultControlDispatcher::
95 m_parallel_mng->waitAllRequests(requests);
98void ParallelMngDispatcher::DefaultControlDispatcher::
100 bool is_non_blocking)
104 done_requests = m_parallel_mng->testSomeRequests(requests);
106 done_requests = m_parallel_mng->waitSomeRequests(requests);
108 for (
int x : done_requests)
120splitCommunicator(
bool keep)
126void ParallelMngDispatcher::DefaultControlDispatcher::
129 m_parallel_mng->barrier();
132Request ParallelMngDispatcher::DefaultControlDispatcher::
135 return m_parallel_mng->nonBlockingCollective()->barrier();
138MessageId ParallelMngDispatcher::DefaultControlDispatcher::
141 return m_parallel_mng->probe(message);
147 return m_parallel_mng->legacyProbe(message);
151createRequestListRef()
153 return m_parallel_mng->createRequestListRef();
159void ParallelMngDispatcher::DefaultControlDispatcher::
160setProfiler(MP::IProfiler* p)
170:
public MP::ISerializeDispatcher
182 return m_parallel_mng->createSerializeMessageListRef();
186 return m_parallel_mng->sendSerializer(s, message);
190 return m_parallel_mng->receiveSerializer(s, message);
204ParallelMngDispatcher::
206: m_comm_rank(bi.commRank())
207, m_comm_size(bi.commSize())
208, m_communicator(bi.communicator())
210, m_unsigned_char(nullptr)
211, m_signed_char(nullptr)
213, m_unsigned_short(nullptr)
215, m_unsigned_int(nullptr)
217, m_unsigned_long(nullptr)
218, m_long_long(nullptr)
219, m_unsigned_long_long(nullptr)
222, m_long_double(nullptr)
229, m_time_stats(nullptr)
230, m_mp_dispatchers_ref(bi.dispatchersRef())
231, m_control_dispatcher(new DefaultControlDispatcher(this))
232, m_serialize_dispatcher(new SerializeDispatcher(this))
235 m_message_passing_mng =
this;
241ParallelMngDispatcher::
242~ParallelMngDispatcher()
244 m_mp_dispatchers_ref.
reset();
246 delete m_parallel_mng_internal;
248 delete m_serialize_dispatcher;
249 delete m_control_dispatcher;
251 delete m_signed_char;
252 delete m_unsigned_char;
254 delete m_unsigned_short;
256 delete m_unsigned_int;
258 delete m_unsigned_long;
260 delete m_unsigned_long_long;
264 delete m_long_double;
275void ParallelMngDispatcher::
276_setControlDispatcher(MP::IControlDispatcher* d)
278 delete m_control_dispatcher;
279 m_control_dispatcher = d;
285void ParallelMngDispatcher::
286_setSerializeDispatcher(MP::ISerializeDispatcher* d)
288 delete m_serialize_dispatcher;
289 m_serialize_dispatcher = d;
295void ParallelMngDispatcher::
296_setArccoreDispatchers()
298 m_mp_dispatchers_ref->setDispatcher(m_char->toArccoreDispatcher());
299 m_mp_dispatchers_ref->setDispatcher(m_signed_char->toArccoreDispatcher());
300 m_mp_dispatchers_ref->setDispatcher(m_unsigned_char->toArccoreDispatcher());
301 m_mp_dispatchers_ref->setDispatcher(m_short->toArccoreDispatcher());
302 m_mp_dispatchers_ref->setDispatcher(m_unsigned_short->toArccoreDispatcher());
303 m_mp_dispatchers_ref->setDispatcher(m_int->toArccoreDispatcher());
304 m_mp_dispatchers_ref->setDispatcher(m_unsigned_int->toArccoreDispatcher());
305 m_mp_dispatchers_ref->setDispatcher(m_long->toArccoreDispatcher());
306 m_mp_dispatchers_ref->setDispatcher(m_unsigned_long->toArccoreDispatcher());
307 m_mp_dispatchers_ref->setDispatcher(m_long_long->toArccoreDispatcher());
308 m_mp_dispatchers_ref->setDispatcher(m_unsigned_long_long->toArccoreDispatcher());
309 m_mp_dispatchers_ref->setDispatcher(m_float->toArccoreDispatcher());
310 m_mp_dispatchers_ref->setDispatcher(m_double->toArccoreDispatcher());
311 m_mp_dispatchers_ref->setDispatcher(m_long_double->toArccoreDispatcher());
314 m_mp_dispatchers_ref->setDispatcher(m_control_dispatcher);
316 m_mp_dispatchers_ref->setDispatcher(m_serialize_dispatcher);
325 return m_message_passing_mng;
331MP::IDispatchers* ParallelMngDispatcher::
334 return m_mp_dispatchers_ref.
get();
359 m_time_metric_collector = c;
367_communicationTimeMetricAction()
const
369 return Timer::phaseAction(
timeStats(), TP_Communication);
375void ParallelMngDispatcher::
376broadcastString(
String& str, Int32 rank)
381 if (rank == my_rank) {
382 len_info[0] = bytes.size();
385 broadcast(utf8_array, rank);
390 broadcast(utf8_array, rank);
391 str = String::fromUtf8(utf8_array);
403 size = bytes.largeSize();
412 broadcast(bytes, rank);
422 mpAllGather(_messagePassingMng(), send_serializer, recv_serializer);
435 message_list->addMessage(m);
437 message_list->waitMessages(Parallel::WaitAll);
450 message_list->addMessage(v.get());
452 message_list->waitMessages(Parallel::WaitAll);
461 return makeRef(_createSerializeMessageList());
470 return _createSerializeMessageList();
479 return _createSubParallelMng(kept_ranks);
488 return makeRef(_createSubParallelMng(kept_ranks));
495_createSubParallelMngRef([[maybe_unused]]
Int32 color, [[maybe_unused]]
Int32 key)
506 Timer::Phase tphase(
timeStats(), TP_Communication);
509 request_list->add(requests);
510 request_list->wait(wait_type);
512 for (
Integer i = 0; i < nb_request; ++i)
513 requests[i] = request_list->request(i);
514 return request_list->doneRequestIndexes();
538#define ARCANE_PARALLEL_MANAGER_DISPATCH(field, type) \
539 void ParallelMngDispatcher:: \
540 allGather(ConstArrayView<type> send_buf, ArrayView<type> recv_buf) \
542 Timer::Phase tphase(timeStats(), TP_Communication); \
543 (field)->allGather(send_buf, recv_buf); \
545 void ParallelMngDispatcher:: \
546 gather(ConstArrayView<type> send_buf, ArrayView<type> recv_buf, Integer rank) \
548 Timer::Phase tphase(timeStats(), TP_Communication); \
549 (field)->gather(send_buf, recv_buf, rank); \
551 void ParallelMngDispatcher:: \
552 allGatherVariable(ConstArrayView<type> send_buf, Array<type>& recv_buf) \
554 Timer::Phase tphase(timeStats(), TP_Communication); \
555 (field)->allGatherVariable(send_buf, recv_buf); \
557 void ParallelMngDispatcher:: \
558 gatherVariable(ConstArrayView<type> send_buf, Array<type>& recv_buf, Integer rank) \
560 Timer::Phase tphase(timeStats(), TP_Communication); \
561 (field)->gatherVariable(send_buf, recv_buf, rank); \
563 void ParallelMngDispatcher:: \
564 scatterVariable(ConstArrayView<type> send_buf, ArrayView<type> recv_buf, Integer root) \
566 Timer::Phase tphase(timeStats(), TP_Communication); \
567 (field)->scatterVariable(send_buf, recv_buf, root); \
569 type ParallelMngDispatcher:: \
570 reduce(eReduceType rt, type v) \
572 Timer::Phase tphase(timeStats(), TP_Communication); \
573 return (field)->allReduce(rt, v); \
575 void ParallelMngDispatcher:: \
576 reduce(eReduceType rt, ArrayView<type> v) \
578 Timer::Phase tphase(timeStats(), TP_Communication); \
579 (field)->allReduce(rt, v); \
581 void ParallelMngDispatcher:: \
582 broadcast(ArrayView<type> send_buf, Integer id) \
584 Timer::Phase tphase(timeStats(), TP_Communication); \
585 (field)->broadcast(send_buf, id); \
587 void ParallelMngDispatcher:: \
588 send(ConstArrayView<type> values, Integer id) \
590 Timer::Phase tphase(timeStats(), TP_Communication); \
591 (field)->send(values, id); \
593 void ParallelMngDispatcher:: \
594 recv(ArrayView<type> values, Integer id) \
596 Timer::Phase tphase(timeStats(), TP_Communication); \
597 (field)->recv(values, id); \
599 Parallel::Request ParallelMngDispatcher:: \
600 send(ConstArrayView<type> values, Integer id, bool is_blocked) \
602 Timer::Phase tphase(timeStats(), TP_Communication); \
603 return (field)->send(values, id, is_blocked); \
605 Request ParallelMngDispatcher:: \
606 send(Span<const type> values, const PointToPointMessageInfo& message) \
608 Timer::Phase tphase(timeStats(), TP_Communication); \
609 return (field)->send(values, message); \
611 Parallel::Request ParallelMngDispatcher:: \
612 recv(ArrayView<type> values, Integer id, bool is_blocked) \
614 Timer::Phase tphase(timeStats(), TP_Communication); \
615 return (field)->recv(values, id, is_blocked); \
617 Request ParallelMngDispatcher:: \
618 receive(Span<type> values, const PointToPointMessageInfo& message) \
620 Timer::Phase tphase(timeStats(), TP_Communication); \
621 return (field)->receive(values, message); \
623 void ParallelMngDispatcher:: \
624 sendRecv(ConstArrayView<type> send_buf, ArrayView<type> recv_buf, Integer id) \
626 Timer::Phase tphase(timeStats(), TP_Communication); \
627 (field)->sendRecv(send_buf, recv_buf, id); \
629 void ParallelMngDispatcher:: \
630 allToAll(ConstArrayView<type> send_buf, ArrayView<type> recv_buf, Integer count) \
632 Timer::Phase tphase(timeStats(), TP_Communication); \
633 (field)->allToAll(send_buf, recv_buf, count); \
635 void ParallelMngDispatcher:: \
636 allToAllVariable(ConstArrayView<type> send_buf, Int32ConstArrayView send_count, \
637 Int32ConstArrayView send_index, ArrayView<type> recv_buf, \
638 Int32ConstArrayView recv_count, Int32ConstArrayView recv_index) \
640 Timer::Phase tphase(timeStats(), TP_Communication); \
641 (field)->allToAllVariable(send_buf, send_count, send_index, recv_buf, recv_count, recv_index); \
643 type ParallelMngDispatcher:: \
644 scan(eReduceType rt, type v) \
646 Timer::Phase tphase(timeStats(), TP_Communication); \
647 return (field)->scan(rt, v); \
649 void ParallelMngDispatcher:: \
650 computeMinMaxSum(type val, type& min_val, type& max_val, type& sum_val, Int32& min_proc, Int32& max_proc) \
652 Timer::Phase tphase(timeStats(), TP_Communication); \
653 (field)->computeMinMaxSum(val, min_val, max_val, sum_val, min_proc, max_proc); \
655 IParallelDispatchT<type>* ParallelMngDispatcher:: \
660 void ParallelMngDispatcher:: \
661 computeMinMaxSum(ConstArrayView<type> values, \
662 ArrayView<type> min_values, \
663 ArrayView<type> max_values, \
664 ArrayView<type> sum_values, \
665 ArrayView<Int32> min_ranks, \
666 ArrayView<Int32> max_ranks) \
668 Timer::Phase tphase(timeStats(), TP_Communication); \
669 (field)->computeMinMaxSum(values, min_values, max_values, sum_values, min_ranks, max_ranks); \
671 void ParallelMngDispatcher:: \
672 scan(eReduceType rt, ArrayView<type> v) \
674 Timer::Phase tphase(timeStats(), TP_Communication); \
675 (field)->scan(rt, v); \
678ARCANE_PARALLEL_MANAGER_DISPATCH(m_char,
char)
679ARCANE_PARALLEL_MANAGER_DISPATCH(m_unsigned_char,
unsigned char)
680ARCANE_PARALLEL_MANAGER_DISPATCH(m_signed_char,
signed char)
681ARCANE_PARALLEL_MANAGER_DISPATCH(m_short,
short)
682ARCANE_PARALLEL_MANAGER_DISPATCH(m_unsigned_short,
unsigned short)
683ARCANE_PARALLEL_MANAGER_DISPATCH(m_int,
int)
684ARCANE_PARALLEL_MANAGER_DISPATCH(m_unsigned_int,
unsigned int)
685ARCANE_PARALLEL_MANAGER_DISPATCH(m_long,
long)
686ARCANE_PARALLEL_MANAGER_DISPATCH(m_unsigned_long,
unsigned long)
687ARCANE_PARALLEL_MANAGER_DISPATCH(m_long_long,
long long)
688ARCANE_PARALLEL_MANAGER_DISPATCH(m_unsigned_long_long,
unsigned long long)
689ARCANE_PARALLEL_MANAGER_DISPATCH(m_float,
float)
690ARCANE_PARALLEL_MANAGER_DISPATCH(m_double,
double)
691ARCANE_PARALLEL_MANAGER_DISPATCH(m_long_double,
long double)
692ARCANE_PARALLEL_MANAGER_DISPATCH(m_apreal,
APReal)
693ARCANE_PARALLEL_MANAGER_DISPATCH(m_real2,
Real2)
694ARCANE_PARALLEL_MANAGER_DISPATCH(m_real3,
Real3)
695ARCANE_PARALLEL_MANAGER_DISPATCH(m_real2x2,
Real2x2)
696ARCANE_PARALLEL_MANAGER_DISPATCH(m_real3x3,
Real3x3)
697ARCANE_PARALLEL_MANAGER_DISPATCH(m_hpreal,
HPReal)
#define ARCANE_CHECK_POINTER(ptr)
Macro returning the pointer ptr if it is not null or throwing an exception if it is null.
#define ARCANE_THROW(exception_class,...)
Macro for throwing an exception with formatting.
Brief list of message exchange functions.
#define ARCCORE_DEFINE_REFERENCE_COUNTED_CLASS(class_name)
Macro to define methods and types for a class that uses a reference counter.
Emulation of real number in arbitrary precision.
Modifiable view of an array of type T.
void fill(const T &o) noexcept
Fills the array with the value o.
constexpr Integer size() const noexcept
Returns the size of the array.
Constant view of an array of type T.
Class implementing a High-Precision real number.
Interface for an 'IParallelMng' container.
Interface of the parallelism manager for a subdomain.
virtual Ref< Parallel::IRequestList > createRequestListRef()=0
Creates a request list for this manager.
Interface managing statistics on execution.
Interface managing execution time statistics.
virtual ITimeMetricCollector * metricCollector()=0
Associated collection interface.
Interface of the message passing manager.
Interface for a serialization message list.
Interface for a serialization message between IMessagePassingMng.
Information about the source of a message.
Information for sending/receiving a point-to-point message.
Exception when a function is not implemented.
Request sendSerializer(const ISerializer *s, const PointToPointMessageInfo &message) override
Sending message.
Request receiveSerializer(ISerializer *s, const PointToPointMessageInfo &message) override
Receiving message.
Ref< ISerializeMessageList > createSerializeMessageListRef() override
Create a list of serialization messages.
ISerializeMessageList * createSerializeMessageList() final
Creates a list to manage 'ISerializeMessage'.
void setTimeStats(ITimeStats *ts) override
Sets the statistics manager.
void allGather(ISerializer *send_serializer, ISerializer *recv_serializer) override
Redefines allGather here to avoid hiding the symbol in derived classes.
Ref< IParallelMng > createSubParallelMngRef(Int32ConstArrayView kept_ranks) override
Creates a new parallelism manager for a subset of ranks.
void processMessages(ConstArrayView< ISerializeMessage * > messages) override
Executes the operations of messages messages.
IMessagePassingMng * messagePassingMng() const override
Associated Arccore message passing manager.
Int32 commRank() const override
Rank of this instance in the communicator.
ITimeMetricCollector * timeMetricCollector() const override
Interface for collecting execution times (can be null).
IParallelMng * createSubParallelMng(Int32ConstArrayView kept_ranks) final
Creates a new parallelism manager for a subset of ranks.
UniqueArray< Integer > testSomeRequests(ArrayView< Request > requests) override
Tests if one of the rvalues requests is complete.
ITimeStats * timeStats() const override
Associated statistics manager (can be null).
Ref< ISerializeMessageList > createSerializeMessageListRef() final
Creates a list to manage 'ISerializeMessage'.
UniqueArray< Integer > waitSomeRequests(ArrayView< Request > requests) override
Blocks while waiting for one of the rvalues requests to complete.
void broadcastMemoryBuffer(ByteArray &bytes, Int32 rank) override
Performs a broadcast of a memory region.
Internal part of IParallelMng.
Class managing a 2-dimensional real vector.
Class managing a 2x2 matrix of reals.
Class managing a 3-dimensional real vector.
Class managing a 3x3 real matrix.
void reset()
Positions the instance to the null pointer.
InstanceType * get() const
Associated instance or nullptr if none.
Reference to an instance.
View of an array of elements of type T.
Unicode character string.
Span< const Byte > bytes() const
Returns the conversion of the instance into UTF-8 encoding.
Sentinel for collecting temporal information.
Positions the phase of the currently executing action.
1D data vector with value semantics (STL style).
@ WaitSome
Wait until all messages in the list are processed.
void mpAllGather(IMessagePassingMng *pm, const ISerializer *send_serializer, ISerializer *receive_serialize)
allGather() message for serialization
-- tab-width: 2; indent-tabs-mode: nil; coding: utf-8-with-signature --
ArrayView< Int64 > Int64ArrayView
C equivalent of a 1D array of 64-bit integers.
std::int64_t Int64
Signed integer type of 64 bits.
Int32 Integer
Type representing an integer.
Array< Byte > ByteArray
Dynamic one-dimensional array of characters.
ConstArrayView< Int32 > Int32ConstArrayView
C equivalent of a 1D array of 32-bit integers.
UniqueArray< Byte > ByteUniqueArray
Dynamic 1D array of characters.
auto makeRef(InstanceType *t) -> Ref< InstanceType >
Creates a reference on a pointer.
std::int32_t Int32
Signed integer type of 32 bits.
Ref< TrueType > createRef(Args &&... args)
Creates an instance of type TrueType with arguments Args and returns a reference to it.