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/NotSupportedException.h"
19#include "arcane/utils/ScopedPtr.h"
20#include "arcane/utils/ITraceMng.h"
21#include "arcane/utils/ValueConvert.h"
22#include "arcane/utils/FatalErrorException.h"
24#include "arcane/core/ParallelMngDispatcher.h"
25#include "arcane/core/IParallelDispatch.h"
26#include "arcane/core/Timer.h"
27#include "arcane/core/ITimeStats.h"
28#include "arcane/core/IParallelNonBlockingCollective.h"
29#include "arcane/core/internal/ParallelMngInternal.h"
31#include "arcane/accelerator/core/Runner.h"
32#include "arcane/accelerator/core/RunQueueBuildInfo.h"
34#include "arccore/message_passing/Dispatchers.h"
36#include "arccore/message_passing/MessagePassingMng.h"
37#include "arccore/message_passing/IControlDispatcher.h"
38#include "arccore/message_passing/ISerializeDispatcher.h"
39#include "arccore/message_passing/PointToPointMessageInfo.h"
40#include "arccore/message_passing/IRequestList.h"
41#include "arccore/message_passing/ISerializeMessageList.h"
42#include "arccore/trace/internal/TimeMetric.h"
49using namespace Arccore::MessagePassing;
54ParallelMngDispatcherBuildInfo::
55ParallelMngDispatcherBuildInfo(
Int32 comm_rank,
Int32 comm_size)
56: m_comm_rank(comm_rank)
57, m_comm_size(comm_size)
66ParallelMngDispatcherBuildInfo::
67ParallelMngDispatcherBuildInfo(Int32 comm_rank, Int32 comm_size, MP::Communicator communicator)
68: m_comm_rank(comm_rank)
69, m_comm_size(comm_size)
70, m_communicator(communicator)
78void ParallelMngDispatcherBuildInfo::
87ParallelMngDispatcher::DefaultControlDispatcher::
88DefaultControlDispatcher(ParallelMngDispatcher* pm)
93void ParallelMngDispatcher::DefaultControlDispatcher::
96 m_parallel_mng->waitAllRequests(requests);
99void ParallelMngDispatcher::DefaultControlDispatcher::
101 bool is_non_blocking)
105 done_requests = m_parallel_mng->testSomeRequests(requests);
107 done_requests = m_parallel_mng->waitSomeRequests(requests);
109 for (
int x : done_requests)
121splitCommunicator(
bool keep)
130void ParallelMngDispatcher::DefaultControlDispatcher::
133 m_parallel_mng->barrier();
136Request ParallelMngDispatcher::DefaultControlDispatcher::
139 return m_parallel_mng->nonBlockingCollective()->barrier();
142MessageId ParallelMngDispatcher::DefaultControlDispatcher::
145 return m_parallel_mng->probe(message);
151 return m_parallel_mng->legacyProbe(message);
155createRequestListRef()
157 return m_parallel_mng->createRequestListRef();
163void ParallelMngDispatcher::DefaultControlDispatcher::
164setProfiler(MP::IProfiler* p)
174:
public MP::ISerializeDispatcher
186 return m_parallel_mng->createSerializeMessageListRef();
190 return m_parallel_mng->sendSerializer(s, message);
194 return m_parallel_mng->receiveSerializer(s, message);
208ParallelMngDispatcher::
210: m_comm_rank(bi.commRank())
211, m_comm_size(bi.commSize())
212, m_communicator(bi.communicator())
214, m_unsigned_char(nullptr)
215, m_signed_char(nullptr)
217, m_unsigned_short(nullptr)
219, m_unsigned_int(nullptr)
221, m_unsigned_long(nullptr)
222, m_long_long(nullptr)
223, m_unsigned_long_long(nullptr)
226, m_long_double(nullptr)
233, m_time_stats(nullptr)
234, m_mp_dispatchers_ref(bi.dispatchersRef())
235, m_control_dispatcher(new DefaultControlDispatcher(this))
236, m_serialize_dispatcher(new SerializeDispatcher(this))
239 m_message_passing_mng =
this;
245ParallelMngDispatcher::
246~ParallelMngDispatcher()
248 m_mp_dispatchers_ref.
reset();
250 delete m_parallel_mng_internal;
252 delete m_serialize_dispatcher;
253 delete m_control_dispatcher;
255 delete m_signed_char;
256 delete m_unsigned_char;
258 delete m_unsigned_short;
260 delete m_unsigned_int;
262 delete m_unsigned_long;
264 delete m_unsigned_long_long;
268 delete m_long_double;
279void ParallelMngDispatcher::
280_setControlDispatcher(MP::IControlDispatcher* d)
282 delete m_control_dispatcher;
283 m_control_dispatcher = d;
289void ParallelMngDispatcher::
290_setSerializeDispatcher(MP::ISerializeDispatcher* d)
292 delete m_serialize_dispatcher;
293 m_serialize_dispatcher = d;
299void ParallelMngDispatcher::
300_setArccoreDispatchers()
302 m_mp_dispatchers_ref->setDispatcher(m_char->toArccoreDispatcher());
303 m_mp_dispatchers_ref->setDispatcher(m_signed_char->toArccoreDispatcher());
304 m_mp_dispatchers_ref->setDispatcher(m_unsigned_char->toArccoreDispatcher());
305 m_mp_dispatchers_ref->setDispatcher(m_short->toArccoreDispatcher());
306 m_mp_dispatchers_ref->setDispatcher(m_unsigned_short->toArccoreDispatcher());
307 m_mp_dispatchers_ref->setDispatcher(m_int->toArccoreDispatcher());
308 m_mp_dispatchers_ref->setDispatcher(m_unsigned_int->toArccoreDispatcher());
309 m_mp_dispatchers_ref->setDispatcher(m_long->toArccoreDispatcher());
310 m_mp_dispatchers_ref->setDispatcher(m_unsigned_long->toArccoreDispatcher());
311 m_mp_dispatchers_ref->setDispatcher(m_long_long->toArccoreDispatcher());
312 m_mp_dispatchers_ref->setDispatcher(m_unsigned_long_long->toArccoreDispatcher());
313 m_mp_dispatchers_ref->setDispatcher(m_float->toArccoreDispatcher());
314 m_mp_dispatchers_ref->setDispatcher(m_double->toArccoreDispatcher());
315 m_mp_dispatchers_ref->setDispatcher(m_long_double->toArccoreDispatcher());
318 m_mp_dispatchers_ref->setDispatcher(m_control_dispatcher);
320 m_mp_dispatchers_ref->setDispatcher(m_serialize_dispatcher);
329 return m_message_passing_mng;
335MP::IDispatchers* ParallelMngDispatcher::
338 return m_mp_dispatchers_ref.
get();
363 m_time_metric_collector = c;
371_communicationTimeMetricAction()
const
373 return Timer::phaseAction(
timeStats(), TP_Communication);
379void ParallelMngDispatcher::
380broadcastString(
String& str, Int32 rank)
385 if (rank == my_rank) {
386 len_info[0] = bytes.size();
389 broadcast(utf8_array, rank);
394 broadcast(utf8_array, rank);
395 str = String::fromUtf8(utf8_array);
407 size = bytes.largeSize();
416 broadcast(bytes, rank);
426 mpAllGather(_messagePassingMng(), send_serializer, recv_serializer);
439 message_list->addMessage(m);
441 message_list->waitMessages(Parallel::WaitAll);
454 message_list->addMessage(v.get());
456 message_list->waitMessages(Parallel::WaitAll);
465 return makeRef(_createSerializeMessageList());
474 return _createSerializeMessageList();
483 return _createSubParallelMng(kept_ranks);
492 return makeRef(_createSubParallelMng(kept_ranks));
499_createSubParallelMngRef([[maybe_unused]]
Int32 color, [[maybe_unused]]
Int32 key)
508_createSubParallelMngRef(
bool is_kept)
514 Int8 kept = (is_kept) ? 1 : 0;
519 for (Int32 i = 0; i < nb_rank; ++i)
520 if (all_kept[i] != 0)
528UniqueArray<Integer> ParallelMngDispatcher::
529_doWaitRequests(ArrayView<Request> requests,
eWaitType wait_type)
531 Timer::Phase tphase(
timeStats(), TP_Communication);
534 request_list->add(requests);
535 request_list->wait(wait_type);
537 for (
Integer i = 0; i < nb_request; ++i)
538 requests[i] = request_list->request(i);
539 return request_list->doneRequestIndexes();
563#define ARCANE_PARALLEL_MANAGER_DISPATCH(field, type) \
564 void ParallelMngDispatcher:: \
565 allGather(ConstArrayView<type> send_buf, ArrayView<type> recv_buf) \
567 Timer::Phase tphase(timeStats(), TP_Communication); \
568 (field)->allGather(send_buf, recv_buf); \
570 void ParallelMngDispatcher:: \
571 gather(ConstArrayView<type> send_buf, ArrayView<type> recv_buf, Integer rank) \
573 Timer::Phase tphase(timeStats(), TP_Communication); \
574 (field)->gather(send_buf, recv_buf, rank); \
576 void ParallelMngDispatcher:: \
577 allGatherVariable(ConstArrayView<type> send_buf, Array<type>& recv_buf) \
579 Timer::Phase tphase(timeStats(), TP_Communication); \
580 (field)->allGatherVariable(send_buf, recv_buf); \
582 void ParallelMngDispatcher:: \
583 gatherVariable(ConstArrayView<type> send_buf, Array<type>& recv_buf, Integer rank) \
585 Timer::Phase tphase(timeStats(), TP_Communication); \
586 (field)->gatherVariable(send_buf, recv_buf, rank); \
588 void ParallelMngDispatcher:: \
589 scatterVariable(ConstArrayView<type> send_buf, ArrayView<type> recv_buf, Integer root) \
591 Timer::Phase tphase(timeStats(), TP_Communication); \
592 (field)->scatterVariable(send_buf, recv_buf, root); \
594 type ParallelMngDispatcher:: \
595 reduce(eReduceType rt, type v) \
597 Timer::Phase tphase(timeStats(), TP_Communication); \
598 return (field)->allReduce(rt, v); \
600 void ParallelMngDispatcher:: \
601 reduce(eReduceType rt, ArrayView<type> v) \
603 Timer::Phase tphase(timeStats(), TP_Communication); \
604 (field)->allReduce(rt, v); \
606 void ParallelMngDispatcher:: \
607 broadcast(ArrayView<type> send_buf, Integer id) \
609 Timer::Phase tphase(timeStats(), TP_Communication); \
610 (field)->broadcast(send_buf, id); \
612 void ParallelMngDispatcher:: \
613 send(ConstArrayView<type> values, Integer id) \
615 Timer::Phase tphase(timeStats(), TP_Communication); \
616 (field)->send(values, id); \
618 void ParallelMngDispatcher:: \
619 recv(ArrayView<type> values, Integer id) \
621 Timer::Phase tphase(timeStats(), TP_Communication); \
622 (field)->recv(values, id); \
624 Parallel::Request ParallelMngDispatcher:: \
625 send(ConstArrayView<type> values, Integer id, bool is_blocked) \
627 Timer::Phase tphase(timeStats(), TP_Communication); \
628 return (field)->send(values, id, is_blocked); \
630 Request ParallelMngDispatcher:: \
631 send(Span<const type> values, const PointToPointMessageInfo& message) \
633 Timer::Phase tphase(timeStats(), TP_Communication); \
634 return (field)->send(values, message); \
636 Parallel::Request ParallelMngDispatcher:: \
637 recv(ArrayView<type> values, Integer id, bool is_blocked) \
639 Timer::Phase tphase(timeStats(), TP_Communication); \
640 return (field)->recv(values, id, is_blocked); \
642 Request ParallelMngDispatcher:: \
643 receive(Span<type> values, const PointToPointMessageInfo& message) \
645 Timer::Phase tphase(timeStats(), TP_Communication); \
646 return (field)->receive(values, message); \
648 void ParallelMngDispatcher:: \
649 sendRecv(ConstArrayView<type> send_buf, ArrayView<type> recv_buf, Integer id) \
651 Timer::Phase tphase(timeStats(), TP_Communication); \
652 (field)->sendRecv(send_buf, recv_buf, id); \
654 void ParallelMngDispatcher:: \
655 allToAll(ConstArrayView<type> send_buf, ArrayView<type> recv_buf, Integer count) \
657 Timer::Phase tphase(timeStats(), TP_Communication); \
658 (field)->allToAll(send_buf, recv_buf, count); \
660 void ParallelMngDispatcher:: \
661 allToAllVariable(ConstArrayView<type> send_buf, Int32ConstArrayView send_count, \
662 Int32ConstArrayView send_index, ArrayView<type> recv_buf, \
663 Int32ConstArrayView recv_count, Int32ConstArrayView recv_index) \
665 Timer::Phase tphase(timeStats(), TP_Communication); \
666 (field)->allToAllVariable(send_buf, send_count, send_index, recv_buf, recv_count, recv_index); \
668 type ParallelMngDispatcher:: \
669 scan(eReduceType rt, type v) \
671 Timer::Phase tphase(timeStats(), TP_Communication); \
672 return (field)->scan(rt, v); \
674 void ParallelMngDispatcher:: \
675 computeMinMaxSum(type val, type& min_val, type& max_val, type& sum_val, Int32& min_proc, Int32& max_proc) \
677 Timer::Phase tphase(timeStats(), TP_Communication); \
678 (field)->computeMinMaxSum(val, min_val, max_val, sum_val, min_proc, max_proc); \
680 IParallelDispatchT<type>* ParallelMngDispatcher:: \
685 void ParallelMngDispatcher:: \
686 computeMinMaxSum(ConstArrayView<type> values, \
687 ArrayView<type> min_values, \
688 ArrayView<type> max_values, \
689 ArrayView<type> sum_values, \
690 ArrayView<Int32> min_ranks, \
691 ArrayView<Int32> max_ranks) \
693 Timer::Phase tphase(timeStats(), TP_Communication); \
694 (field)->computeMinMaxSum(values, min_values, max_values, sum_values, min_ranks, max_ranks); \
696 void ParallelMngDispatcher:: \
697 scan(eReduceType rt, ArrayView<type> v) \
699 Timer::Phase tphase(timeStats(), TP_Communication); \
700 (field)->scan(rt, v); \
703ARCANE_PARALLEL_MANAGER_DISPATCH(m_char,
char)
704ARCANE_PARALLEL_MANAGER_DISPATCH(m_unsigned_char,
unsigned char)
705ARCANE_PARALLEL_MANAGER_DISPATCH(m_signed_char,
signed char)
706ARCANE_PARALLEL_MANAGER_DISPATCH(m_short,
short)
707ARCANE_PARALLEL_MANAGER_DISPATCH(m_unsigned_short,
unsigned short)
708ARCANE_PARALLEL_MANAGER_DISPATCH(m_int,
int)
709ARCANE_PARALLEL_MANAGER_DISPATCH(m_unsigned_int,
unsigned int)
710ARCANE_PARALLEL_MANAGER_DISPATCH(m_long,
long)
711ARCANE_PARALLEL_MANAGER_DISPATCH(m_unsigned_long,
unsigned long)
712ARCANE_PARALLEL_MANAGER_DISPATCH(m_long_long,
long long)
713ARCANE_PARALLEL_MANAGER_DISPATCH(m_unsigned_long_long,
unsigned long long)
714ARCANE_PARALLEL_MANAGER_DISPATCH(m_float,
float)
715ARCANE_PARALLEL_MANAGER_DISPATCH(m_double,
double)
716ARCANE_PARALLEL_MANAGER_DISPATCH(m_long_double,
long double)
717ARCANE_PARALLEL_MANAGER_DISPATCH(m_apreal,
APReal)
718ARCANE_PARALLEL_MANAGER_DISPATCH(m_real2,
Real2)
719ARCANE_PARALLEL_MANAGER_DISPATCH(m_real3,
Real3)
720ARCANE_PARALLEL_MANAGER_DISPATCH(m_real2x2,
Real2x2)
721ARCANE_PARALLEL_MANAGER_DISPATCH(m_real3x3,
Real3x3)
722ARCANE_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.
void add(ConstReferenceType val)
Adds element val to the end of the array.
void reserve(Int64 new_capacity)
Reserves memory for new_capacity elements.
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.
Exception when an operation is not supported.
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.
Int32 commSize() const override
Number of instances in the communicator.
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.
auto makeRef(InstanceType *t) -> Ref< InstanceType >
Creates a reference on a pointer.
Arcane::Int8 Int8
Type representing an 8-bit integer.