14#include "arcane/utils/ArcanePrecomp.h"
16#include "arcane/utils/Array.h"
17#include "arcane/utils/PlatformUtils.h"
18#include "arcane/utils/String.h"
19#include "arcane/utils/ITraceMng.h"
20#include "arcane/utils/NumericTypes.h"
21#include "arcane/utils/APReal.h"
22#include "arcane/utils/NotImplementedException.h"
23#include "arcane/utils/MemoryView.h"
26#include "arcane/core/MeshVariable.h"
27#include "arcane/core/IParallelMng.h"
28#include "arcane/core/ItemGroup.h"
29#include "arcane/core/IMesh.h"
30#include "arcane/core/IBase.h"
32#include "arcane/parallel/thread/SharedMemoryParallelDispatch.h"
33#include "arcane/parallel/thread/SharedMemoryParallelMng.h"
34#include "arcane/parallel/thread/ISharedMemoryMessageQueue.h"
36#include "arccore/message_passing/PointToPointMessageInfo.h"
71SharedMemoryParallelDispatchBase::
74 ArrayView<SharedMemoryParallelDispatchBase*> all_dispatchs_base)
76, m_parallel_mng(parallel_mng)
77, m_rank(parallel_mng->commRank())
78, m_nb_rank(parallel_mng->commSize())
79, m_message_queue(message_queue)
80, m_all_dispatchs_base(all_dispatchs_base)
87void SharedMemoryParallelDispatchBase::
90 m_parallel_mng->getThreadBarrier()->wait();
96void SharedMemoryParallelDispatchBase::
97_genericAllToAll(ConstMemoryView send_buf, MutableMemoryView recv_buf, Int32 count)
99 Int32 nb_rank = m_nb_rank;
107 for (Integer i = 0; i < nb_rank; ++i) {
108 send_indexes[i] = count * i;
109 recv_indexes[i] = count * i;
111 _genericAllToAllVariable(send_buf, send_count, send_indexes, recv_buf, recv_count, recv_indexes);
117void SharedMemoryParallelDispatchBase::
118_genericAllToAllVariable(ConstMemoryView send_buf,
119 Span<const Int32> send_count,
120 Span<const Int32> send_index,
121 MutableMemoryView recv_buf,
122 Span<const Int32> recv_count,
123 Span<const Int32> recv_index)
125 m_alltoallv_infos.send_buf = send_buf;
126 m_alltoallv_infos.send_count = send_count;
127 m_alltoallv_infos.send_index = send_index;
128 m_alltoallv_infos.recv_buf = recv_buf;
129 m_alltoallv_infos.recv_count = recv_count;
130 m_alltoallv_infos.recv_index = recv_index;
131 _collectiveBarrier();
133 Int32 my_rank = m_rank;
134 MutableMemoryView recv_mem_buf(recv_buf);
135 for (Integer i = 0; i < m_nb_rank; ++i) {
136 AllToAllVariableInfo ainfo = m_all_dispatchs_base[i]->m_alltoallv_infos;
137 ConstMemoryView view(ainfo.send_buf);
138 Integer index = ainfo.send_index[my_rank];
139 Integer count = ainfo.send_count[my_rank];
140 MemoryUtils::copyHost(recv_mem_buf.subView(global_index, count), view.subView(index, count));
141 global_index += count;
143 _collectiveBarrier();
149void SharedMemoryParallelDispatchBase::
150_genericAllGather(ConstMemoryView send_buf, MutableMemoryView recv_buf)
152 m_const_view = send_buf;
153 _collectiveBarrier();
154 MutableMemoryView recv_mem_view(recv_buf);
156 for (Int32 i = 0; i < m_nb_rank; ++i) {
157 ConstMemoryView view(m_all_dispatchs_base[i]->m_const_view);
158 Int64 size = view.nbElement();
159 MemoryUtils::copyHost(recv_mem_view.subView(index, size), view);
162 _collectiveBarrier();
168void SharedMemoryParallelDispatchBase::
169_genericAllGatherVariable(ConstMemoryView send_buf, IResizableArray* recv_buf)
171 m_const_view = send_buf;
172 _collectiveBarrier();
173 Int64 total_size = 0;
174 for (Integer i = 0; i < m_nb_rank; ++i) {
175 total_size += m_all_dispatchs_base[i]->m_const_view.nbElement();
177 recv_buf->resize(total_size);
178 MutableMemoryView recv_mem_view(recv_buf->memoryView());
180 for (Integer i = 0; i < m_nb_rank; ++i) {
181 ConstMemoryView view(m_all_dispatchs_base[i]->m_const_view);
182 Int64 size = view.nbElement();
183 MemoryUtils::copyHost(recv_mem_view.subView(index, size), view);
186 _collectiveBarrier();
192void SharedMemoryParallelDispatchBase::
193_genericScatterVariable(ConstMemoryView send_buf, MutableMemoryView recv_buf, Int32 root)
195 m_const_view = send_buf;
196 m_recv_view = recv_buf;
197 _collectiveBarrier();
198 if (m_rank == root) {
199 ConstMemoryView const_view(m_const_view);
201 for (Integer i = 0; i < m_nb_rank; ++i) {
202 MutableMemoryView view(m_all_dispatchs_base[i]->m_recv_view);
203 Int64 size = view.nbElement();
204 MemoryUtils::copyHost(view, const_view.subView(index, size));
208 _collectiveBarrier();
214Request SharedMemoryParallelDispatchBase::
215_genericSend(ConstMemoryView send_buffer,
const PointToPointMessageInfo& message2)
217 PointToPointMessageInfo message(message2);
218 message.setEmiterRank(MessageRank(m_rank));
219 bool is_blocking = message.isBlocking();
220 if (message.isRankTag()) {
221 Request r = m_message_queue->addSend(message, SendBufferInfo(send_buffer));
223 m_message_queue->waitAll(m_rank, ArrayView<Request>(1, &r));
228 if (message.isMessageId()) {
230 ARCCORE_THROW(NotSupportedException,
"Invalid generic send with MessageId");
232 ARCCORE_THROW(NotSupportedException,
"Invalid message_info");
238Request SharedMemoryParallelDispatchBase::
239_genericReceive(MutableMemoryView recv_buffer,
const PointToPointMessageInfo& message2)
241 PointToPointMessageInfo message(message2);
242 bool is_blocking = message.isBlocking();
243 message.setEmiterRank(MessageRank(m_rank));
244 ReceiveBufferInfo buf{ recv_buffer };
245 Request r = m_message_queue->addReceive(message, buf);
247 m_message_queue->waitAll(m_rank, ArrayView<Request>(1, &r));
248 return MP::Request();
256void SharedMemoryParallelDispatchBase::
257_genericBroadcast(MutableMemoryView send_buf, Int32 rank)
259 m_broadcast_view = send_buf;
260 _collectiveBarrier();
261 MemoryUtils::copyHost(m_broadcast_view, m_all_dispatchs_base[rank]->m_broadcast_view);
262 _collectiveBarrier();
271template <
class Type> SharedMemoryParallelDispatch<Type>::
272SharedMemoryParallelDispatch(ITraceMng* tm, SharedMemoryParallelMng* parallel_mng,
273 ISharedMemoryMessageQueue* message_queue,
274 impl::ShareMemoryDispatcherContainer<Type>& containers)
275: BaseClass(tm, parallel_mng, message_queue, containers.all_dispatchs_base)
276, m_all_dispatchs(containers.all_dispatchs)
278 m_reduce_infos.m_index = 0;
279 m_all_dispatchs[m_rank] =
this;
280 m_all_dispatchs_base[m_rank] =
this;
286template <
class Type> SharedMemoryParallelDispatch<Type>::
287~SharedMemoryParallelDispatch()
295template <
class Type>
void SharedMemoryParallelDispatch<Type>::
304class _ThreadIntegralType
308 typedef FalseType IsIntegral;
311#define ARCANE_DEFINE_INTEGRAL_TYPE(datatype) \
313 class _ThreadIntegralType<datatype> \
317 typedef TrueType IsIntegral; \
320ARCANE_DEFINE_INTEGRAL_TYPE(
long long);
321ARCANE_DEFINE_INTEGRAL_TYPE(
long);
322ARCANE_DEFINE_INTEGRAL_TYPE(
int);
323ARCANE_DEFINE_INTEGRAL_TYPE(
short);
324ARCANE_DEFINE_INTEGRAL_TYPE(
unsigned long long);
325ARCANE_DEFINE_INTEGRAL_TYPE(
unsigned long);
326ARCANE_DEFINE_INTEGRAL_TYPE(
unsigned int);
327ARCANE_DEFINE_INTEGRAL_TYPE(
unsigned short);
328ARCANE_DEFINE_INTEGRAL_TYPE(
double);
329ARCANE_DEFINE_INTEGRAL_TYPE(
float);
330ARCANE_DEFINE_INTEGRAL_TYPE(HPReal);
338 template <
class Type>
void
339 _computeMinMaxSum2(ArrayView<SharedMemoryParallelDispatch<Type>*> all_dispatchs,
341 Int32& min_rank, Int32& max_rank, Int32 nb_rank, FalseType)
343 ARCANE_UNUSED(all_dispatchs);
344 ARCANE_UNUSED(min_val);
345 ARCANE_UNUSED(max_val);
346 ARCANE_UNUSED(sum_val);
347 ARCANE_UNUSED(min_rank);
348 ARCANE_UNUSED(max_rank);
349 ARCANE_UNUSED(nb_rank);
351 throw NotImplementedException(A_FUNCINFO);
357 template <
class Type>
void
358 _computeMinMaxSum2(ArrayView<SharedMemoryParallelDispatch<Type>*> all_dispatchs,
360 Int32& min_rank, Int32& max_rank, Int32 nb_rank, TrueType)
362 Type _min_val = all_dispatchs[0]->m_reduce_infos.reduce_value;
363 Type _max_val = _min_val;
364 Type _sum_val = _min_val;
367 for (Integer i = 1; i < nb_rank; ++i) {
368 Type cval = all_dispatchs[i]->m_reduce_infos.reduce_value;
369 if (cval < _min_val) {
373 if (cval > _max_val) {
377 _sum_val = (
Type)(_sum_val + cval);
382 min_rank = _min_rank;
383 max_rank = _max_rank;
391template <
class Type>
void SharedMemoryParallelDispatch<Type>::
393 Int32& min_rank, Int32& max_rank)
395 typedef typename _ThreadIntegralType<Type>::IsIntegral IntegralType;
396 m_reduce_infos.reduce_value = val;
397 _collectiveBarrier();
398 _computeMinMaxSum2(m_all_dispatchs, min_val, max_val, sum_val, min_rank, max_rank, m_nb_rank, IntegralType());
399 _collectiveBarrier();
405template <
class Type>
void SharedMemoryParallelDispatch<Type>::
406computeMinMaxSum(ConstArrayView<Type> values,
407 ArrayView<Type> min_values,
408 ArrayView<Type> max_values,
409 ArrayView<Type> sum_values,
410 ArrayView<Int32> min_ranks,
411 ArrayView<Int32> max_ranks)
415 typedef typename _ThreadIntegralType<Type>::IsIntegral IntegralType;
417 for (Integer i = 0; i < n; ++i) {
418 m_reduce_infos.reduce_value = values[i];
419 _collectiveBarrier();
420 _computeMinMaxSum2(m_all_dispatchs, min_values[i], max_values[i], sum_values[i],
421 min_ranks[i], max_ranks[i], m_nb_rank, IntegralType());
422 _collectiveBarrier();
429template <
class Type>
void SharedMemoryParallelDispatch<Type>::
430broadcast(Span<Type> send_buf, Int32 rank)
432 _genericBroadcast(MutableMemoryView(send_buf), rank);
438template <
class Type>
void SharedMemoryParallelDispatch<Type>::
439allGather(Span<const Type> send_buf, Span<Type> recv_buf)
441 _genericAllGather(ConstMemoryView{ send_buf }, MutableMemoryView{ recv_buf });
447template <
class Type>
void SharedMemoryParallelDispatch<Type>::
448gather(Span<const Type> send_buf, Span<Type> recv_buf, Int32 root_rank)
450 UniqueArray<Type> tmp_buf;
451 if (m_rank == root_rank)
452 allGather(send_buf, recv_buf);
454 tmp_buf.resize(send_buf.size() * m_nb_rank);
455 allGather(send_buf, tmp_buf);
462template <
class Type>
void SharedMemoryParallelDispatch<Type>::
463allGatherVariable(Span<const Type> send_buf, Array<Type>& recv_buf)
465 ResizableArrayRef recv_buf_ref(recv_buf);
466 _genericAllGatherVariable(ConstMemoryView(send_buf), &recv_buf_ref);
472template <
class Type>
void SharedMemoryParallelDispatch<Type>::
473gatherVariable(Span<const Type> send_buf, Array<Type>& recv_buf, Int32 root_rank)
475 UniqueArray<Type> tmp_buf;
476 if (m_rank == root_rank)
477 allGatherVariable(send_buf, recv_buf);
479 allGatherVariable(send_buf, tmp_buf);
485template <
class Type>
void SharedMemoryParallelDispatch<Type>::
486scatterVariable(Span<const Type> send_buf, Span<Type> recv_buf, Int32 root)
488 _genericScatterVariable(ConstMemoryView(send_buf), MutableMemoryView(recv_buf), root);
494template <
class Type>
void SharedMemoryParallelDispatch<Type>::
495allToAll(Span<const Type> send_buf, Span<Type> recv_buf, Int32 count)
497 _genericAllToAll(ConstMemoryView(send_buf), MutableMemoryView(recv_buf), count);
503template <
class Type>
void SharedMemoryParallelDispatch<Type>::
504allToAllVariable(Span<const Type> send_buf, ConstArrayView<Int32> send_count,
505 ConstArrayView<Int32> send_index,
506 Span<Type> recv_buf, ConstArrayView<Int32> recv_count,
507 Int32ConstArrayView recv_index)
509 _genericAllToAllVariable(ConstMemoryView(send_buf), send_count, send_index,
510 MutableMemoryView(recv_buf), recv_count, recv_index);
516template <
class Type>
auto SharedMemoryParallelDispatch<Type>::
517send(Span<const Type> send_buffer, Int32 rank,
bool is_blocking) -> Request
519 auto block_mode = (is_blocking) ? Parallel::Blocking : Parallel::NonBlocking;
520 auto p2p_message = m_parallel_mng->buildMessage(rank, block_mode);
521 return send(send_buffer, p2p_message);
527template <
class Type>
void SharedMemoryParallelDispatch<Type>::
528send(ConstArrayView<Type> send_buf, Int32 rank)
530 send(send_buf, rank,
true);
536template <
class Type> Parallel::Request SharedMemoryParallelDispatch<Type>::
537receive(Span<Type> recv_buffer, Int32 rank,
bool is_blocking)
539 auto block_mode = (is_blocking) ? Parallel::Blocking : Parallel::NonBlocking;
540 auto p2p_message = m_parallel_mng->buildMessage(rank, block_mode);
541 return receive(recv_buffer, p2p_message);
547template <
class Type> Request SharedMemoryParallelDispatch<Type>::
548send(Span<const Type> send_buffer,
const PointToPointMessageInfo& message2)
550 return _genericSend(ConstMemoryView(send_buffer), message2);
556template <
class Type> Request SharedMemoryParallelDispatch<Type>::
557receive(Span<Type> recv_buffer,
const PointToPointMessageInfo& message2)
559 return _genericReceive(MutableMemoryView(recv_buffer), message2);
565template <
class Type>
void SharedMemoryParallelDispatch<Type>::
566recv(ArrayView<Type> recv_buffer, Integer rank)
568 recv(recv_buffer, rank,
true);
574template <
class Type>
void SharedMemoryParallelDispatch<Type>::
575sendRecv(ConstArrayView<Type> send_buffer, ArrayView<Type> recv_buffer, Integer proc)
577 ARCANE_UNUSED(send_buffer);
578 ARCANE_UNUSED(recv_buffer);
580 throw NotImplementedException(A_FUNCINFO);
586template <
class Type>
Type SharedMemoryParallelDispatch<Type>::
587allReduce(eReduceType op,
Type send_buf)
589 m_reduce_infos.reduce_value = send_buf;
592 _collectiveBarrier();
593 Type ret = m_all_dispatchs[0]->m_reduce_infos.reduce_value;
595 case Parallel::ReduceMin:
596 for (Integer i = 1; i < m_nb_rank; ++i)
597 ret = math::min(ret, m_all_dispatchs[i]->m_reduce_infos.reduce_value);
599 case Parallel::ReduceMax:
600 for (Integer i = 1; i < m_nb_rank; ++i)
601 ret = math::max(ret, m_all_dispatchs[i]->m_reduce_infos.reduce_value);
603 case Parallel::ReduceSum:
604 for (Integer i = 1; i < m_nb_rank; ++i)
605 ret = (
Type)(ret + m_all_dispatchs[i]->m_reduce_infos.reduce_value);
611 _collectiveBarrier();
618template <
class Type>
void SharedMemoryParallelDispatch<Type>::
619_allReduceOrScan(eReduceType op, Span<const Type> send_buf, Span<Type> receive_buf,
bool is_scan)
621 m_reduce_infos.reduce_buf = send_buf;
622 ++m_reduce_infos.m_index;
623 Int64 buf_size = send_buf.size();
624 Span<Type> final_buf = receive_buf;
625 bool use_in_place = (send_buf.data() == receive_buf.data());
626 UniqueArray<Type> receive_buf_tmp;
628 receive_buf_tmp.resize(buf_size);
629 receive_buf = receive_buf_tmp;
633 _collectiveBarrier();
635 Integer index0 = m_all_dispatchs[0]->m_reduce_infos.m_index;
636 for (Integer i = 0; i < m_nb_rank; ++i) {
637 Integer indexi = m_all_dispatchs[i]->m_reduce_infos.m_index;
638 if (index0 != m_all_dispatchs[i]->m_reduce_infos.m_index) {
639 ARCANE_FATAL(
"INTERNAL: incoherent all reduce i0={0} in={1} n={2}",
644 Int32 nb_rank = m_nb_rank;
646 nb_rank = m_rank + 1;
647 for (Integer j = 0; j < buf_size; ++j)
648 receive_buf[j] = m_all_dispatchs[0]->m_reduce_infos.reduce_buf[j];
650 case Parallel::ReduceMin:
651 for (Integer i = 1; i < nb_rank; ++i)
652 for (Integer j = 0; j < buf_size; ++j)
653 receive_buf[j] = math::min(receive_buf[j], m_all_dispatchs[i]->m_reduce_infos.reduce_buf[j]);
655 case Parallel::ReduceMax:
656 for (Integer i = 1; i < nb_rank; ++i)
657 for (Integer j = 0; j < buf_size; ++j)
658 receive_buf[j] = math::max(receive_buf[j], m_all_dispatchs[i]->m_reduce_infos.reduce_buf[j]);
660 case Parallel::ReduceSum:
661 for (Integer i = 1; i < nb_rank; ++i)
662 for (Integer j = 0; j < buf_size; ++j)
663 receive_buf[j] = (
Type)(receive_buf[j] + m_all_dispatchs[i]->m_reduce_infos.reduce_buf[j]);
669 _collectiveBarrier();
670 for (Integer j = 0; j < buf_size; ++j)
671 final_buf[j] = receive_buf[j];
677template <
class Type>
void SharedMemoryParallelDispatch<Type>::
678allReduce(eReduceType op, Span<Type> send_buf)
680 _allReduceOrScan(op, send_buf, send_buf,
false);
686template <
class Type>
void SharedMemoryParallelDispatch<Type>::
687allReduce(eReduceType op, Span<const Type> send_buf, Span<Type> receive_buf)
689 _allReduceOrScan(op, send_buf, receive_buf,
false);
695template <
class Type> Request SharedMemoryParallelDispatch<Type>::
696nonBlockingAllReduce(eReduceType op, Span<const Type> send_buf, Span<Type> recv_buf)
699 ARCANE_UNUSED(send_buf);
700 ARCANE_UNUSED(recv_buf);
701 throw NotImplementedException(A_FUNCINFO);
707template <
class Type> Request SharedMemoryParallelDispatch<Type>::
708nonBlockingAllGather(Span<const Type> send_buf, Span<Type> recv_buf)
710 ARCANE_UNUSED(send_buf);
711 ARCANE_UNUSED(recv_buf);
712 throw NotImplementedException(A_FUNCINFO);
718template <
class Type> Request SharedMemoryParallelDispatch<Type>::
719nonBlockingBroadcast(Span<Type> send_buf, Int32 rank)
721 ARCANE_UNUSED(send_buf);
723 throw NotImplementedException(A_FUNCINFO);
729template <
class Type> Request SharedMemoryParallelDispatch<Type>::
730nonBlockingGather(Span<const Type> send_buf, Span<Type> recv_buf, Int32 rank)
732 ARCANE_UNUSED(send_buf);
733 ARCANE_UNUSED(recv_buf);
735 throw NotImplementedException(A_FUNCINFO);
741template <
class Type> Request SharedMemoryParallelDispatch<Type>::
742nonBlockingAllToAll(Span<const Type> send_buf, Span<Type> recv_buf, Int32 count)
744 ARCANE_UNUSED(send_buf);
745 ARCANE_UNUSED(recv_buf);
746 ARCANE_UNUSED(count);
747 throw NotImplementedException(A_FUNCINFO);
753template <
class Type> Request SharedMemoryParallelDispatch<Type>::
754nonBlockingAllToAllVariable(Span<const Type> send_buf, ConstArrayView<Int32> send_count,
755 ConstArrayView<Int32> send_index, Span<Type> recv_buf,
756 ConstArrayView<Int32> recv_count, ConstArrayView<Int32> recv_index)
758 ARCANE_UNUSED(send_buf);
759 ARCANE_UNUSED(recv_buf);
760 ARCANE_UNUSED(send_count);
761 ARCANE_UNUSED(recv_count);
762 ARCANE_UNUSED(send_index);
763 ARCANE_UNUSED(recv_index);
764 throw NotImplementedException(A_FUNCINFO);
770template <
class Type>
Type SharedMemoryParallelDispatch<Type>::
771scan(eReduceType op,
Type send_buf)
774 ARCANE_UNUSED(send_buf);
775 throw NotImplementedException(A_FUNCINFO);
781template <
class Type>
void SharedMemoryParallelDispatch<Type>::
782scan(eReduceType op, ArrayView<Type> send_buf)
784 _allReduceOrScan(op, send_buf, send_buf,
true);
790template <
class Type>
void SharedMemoryParallelDispatch<Type>::
794 throw NotImplementedException(A_FUNCINFO);
800template <
class Type> Request SharedMemoryParallelDispatch<Type>::
803 throw NotImplementedException(A_FUNCINFO);
809template class SharedMemoryParallelDispatch<char>;
810template class SharedMemoryParallelDispatch<signed char>;
811template class SharedMemoryParallelDispatch<unsigned char>;
812template class SharedMemoryParallelDispatch<short>;
813template class SharedMemoryParallelDispatch<unsigned short>;
814template class SharedMemoryParallelDispatch<int>;
815template class SharedMemoryParallelDispatch<unsigned int>;
816template class SharedMemoryParallelDispatch<long>;
817template class SharedMemoryParallelDispatch<unsigned long>;
818template class SharedMemoryParallelDispatch<long long>;
819template class SharedMemoryParallelDispatch<unsigned long long>;
820template class SharedMemoryParallelDispatch<float>;
821template class SharedMemoryParallelDispatch<double>;
822template class SharedMemoryParallelDispatch<long double>;
823template class SharedMemoryParallelDispatch<APReal>;
824template class SharedMemoryParallelDispatch<Real2>;
825template class SharedMemoryParallelDispatch<Real3>;
826template class SharedMemoryParallelDispatch<Real2x2>;
827template class SharedMemoryParallelDispatch<Real3x3>;
828template class SharedMemoryParallelDispatch<HPReal>;
#define ARCANE_FATAL(...)
Macro envoyant une exception FatalErrorException.
#define ARCCORE_THROW(exception_class,...)
Macro pour envoyer une exception avec formattage.
Fonctions de gestion mémoire et des allocateurs.
Informations pour un message 'gather' pour le type de données DataType.
Interface d'une file de messages avec les threads.
Gestionnaire du parallélisme utilisant les threads.
Déclarations des types et méthodes utilisés par les mécanismes d'échange de messages.
Int32 Integer
Type représentant un entier.
UniqueArray< Int32 > Int32UniqueArray
Tableau dynamique à une dimension d'entiers 32 bits.