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))
233, m_parallel_mng_internal(new ParallelMngInternal(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();
343 ITimeMetricCollector* c =
nullptr;
359 m_time_metric_collector = c;
366TimeMetricAction ParallelMngDispatcher::
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);
412 broadcast(bytes, rank);
422 mpAllGather(_messagePassingMng(), send_serializer, recv_serializer);
431 TimeMetricSentry tphase(Timer::phaseAction(
timeStats(), TP_Communication));
435 message_list->addMessage(m);
437 message_list->waitMessages(Parallel::WaitAll);
446 TimeMetricSentry tphase(Timer::phaseAction(
timeStats(), TP_Communication));
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.
Int64 largeSize() const
Number of elements in the vector (in 64 bits).
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.
void resize(Int64 s)
Changes the number of elements in the array to s.
Constant view of an array of type T.
Class implementing a High-Precision real number.
Interface of the parallelism manager for a subdomain.
virtual Ref< Parallel::IRequestList > createRequestListRef()=0
Creates a request list for this manager.
virtual ITimeMetricCollector * metricCollector()=0
Associated collection interface.
Interface of the message passing manager.
Interface for a serialization message list.
Information about the source of a message.
Information for sending/receiving a point-to-point message.
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.
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.
constexpr __host__ __device__ SizeType size() const noexcept
Returns the size of the array.
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.
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.