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(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(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<Type> send_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 UniqueArray<Type> ret(buf_size);
627 _collectiveBarrier();
629 Integer index0 = m_all_dispatchs[0]->m_reduce_infos.m_index;
630 for (Integer i = 0; i < m_nb_rank; ++i) {
631 Integer indexi = m_all_dispatchs[i]->m_reduce_infos.m_index;
632 if (index0 != m_all_dispatchs[i]->m_reduce_infos.m_index) {
633 ARCANE_FATAL(
"INTERNAL: incoherent all reduce i0={0} in={1} n={2}",
638 Int32 nb_rank = m_nb_rank;
640 nb_rank = m_rank + 1;
641 for (Integer j = 0; j < buf_size; ++j)
642 ret[j] = m_all_dispatchs[0]->m_reduce_infos.reduce_buf[j];
644 case Parallel::ReduceMin:
645 for (Integer i = 1; i < nb_rank; ++i)
646 for (Integer j = 0; j < buf_size; ++j)
647 ret[j] = math::min(ret[j], m_all_dispatchs[i]->m_reduce_infos.reduce_buf[j]);
649 case Parallel::ReduceMax:
650 for (Integer i = 1; i < nb_rank; ++i)
651 for (Integer j = 0; j < buf_size; ++j)
652 ret[j] = math::max(ret[j], m_all_dispatchs[i]->m_reduce_infos.reduce_buf[j]);
654 case Parallel::ReduceSum:
655 for (Integer i = 1; i < nb_rank; ++i)
656 for (Integer j = 0; j < buf_size; ++j)
657 ret[j] = (
Type)(ret[j] + m_all_dispatchs[i]->m_reduce_infos.reduce_buf[j]);
663 _collectiveBarrier();
664 for (Integer j = 0; j < buf_size; ++j)
665 send_buf[j] = ret[j];
671template <
class Type>
void SharedMemoryParallelDispatch<Type>::
672allReduce(eReduceType op, Span<Type> send_buf)
674 _allReduceOrScan(op, send_buf,
false);
680template <
class Type> Request SharedMemoryParallelDispatch<Type>::
681nonBlockingAllReduce(eReduceType op, Span<const Type> send_buf, Span<Type> recv_buf)
684 ARCANE_UNUSED(send_buf);
685 ARCANE_UNUSED(recv_buf);
686 throw NotImplementedException(A_FUNCINFO);
692template <
class Type> Request SharedMemoryParallelDispatch<Type>::
693nonBlockingAllGather(Span<const Type> send_buf, Span<Type> recv_buf)
695 ARCANE_UNUSED(send_buf);
696 ARCANE_UNUSED(recv_buf);
697 throw NotImplementedException(A_FUNCINFO);
703template <
class Type> Request SharedMemoryParallelDispatch<Type>::
704nonBlockingBroadcast(Span<Type> send_buf, Int32 rank)
706 ARCANE_UNUSED(send_buf);
708 throw NotImplementedException(A_FUNCINFO);
714template <
class Type> Request SharedMemoryParallelDispatch<Type>::
715nonBlockingGather(Span<const Type> send_buf, Span<Type> recv_buf, Int32 rank)
717 ARCANE_UNUSED(send_buf);
718 ARCANE_UNUSED(recv_buf);
720 throw NotImplementedException(A_FUNCINFO);
726template <
class Type> Request SharedMemoryParallelDispatch<Type>::
727nonBlockingAllToAll(Span<const Type> send_buf, Span<Type> recv_buf, Int32 count)
729 ARCANE_UNUSED(send_buf);
730 ARCANE_UNUSED(recv_buf);
731 ARCANE_UNUSED(count);
732 throw NotImplementedException(A_FUNCINFO);
738template <
class Type> Request SharedMemoryParallelDispatch<Type>::
739nonBlockingAllToAllVariable(Span<const Type> send_buf, ConstArrayView<Int32> send_count,
740 ConstArrayView<Int32> send_index, Span<Type> recv_buf,
741 ConstArrayView<Int32> recv_count, ConstArrayView<Int32> recv_index)
743 ARCANE_UNUSED(send_buf);
744 ARCANE_UNUSED(recv_buf);
745 ARCANE_UNUSED(send_count);
746 ARCANE_UNUSED(recv_count);
747 ARCANE_UNUSED(send_index);
748 ARCANE_UNUSED(recv_index);
749 throw NotImplementedException(A_FUNCINFO);
755template <
class Type>
Type SharedMemoryParallelDispatch<Type>::
756scan(eReduceType op,
Type send_buf)
759 ARCANE_UNUSED(send_buf);
760 throw NotImplementedException(A_FUNCINFO);
766template <
class Type>
void SharedMemoryParallelDispatch<Type>::
767scan(eReduceType op, ArrayView<Type> send_buf)
769 _allReduceOrScan(op, send_buf,
true);
775template <
class Type>
void SharedMemoryParallelDispatch<Type>::
779 throw NotImplementedException(A_FUNCINFO);
785template <
class Type> Request SharedMemoryParallelDispatch<Type>::
788 throw NotImplementedException(A_FUNCINFO);
794template class SharedMemoryParallelDispatch<char>;
795template class SharedMemoryParallelDispatch<signed char>;
796template class SharedMemoryParallelDispatch<unsigned char>;
797template class SharedMemoryParallelDispatch<short>;
798template class SharedMemoryParallelDispatch<unsigned short>;
799template class SharedMemoryParallelDispatch<int>;
800template class SharedMemoryParallelDispatch<unsigned int>;
801template class SharedMemoryParallelDispatch<long>;
802template class SharedMemoryParallelDispatch<unsigned long>;
803template class SharedMemoryParallelDispatch<long long>;
804template class SharedMemoryParallelDispatch<unsigned long long>;
805template class SharedMemoryParallelDispatch<float>;
806template class SharedMemoryParallelDispatch<double>;
807template class SharedMemoryParallelDispatch<long double>;
808template class SharedMemoryParallelDispatch<APReal>;
809template class SharedMemoryParallelDispatch<Real2>;
810template class SharedMemoryParallelDispatch<Real3>;
811template class SharedMemoryParallelDispatch<Real2x2>;
812template class SharedMemoryParallelDispatch<Real3x3>;
813template 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.