Arcane  4.2.3.0
Documentation développeur
Chargement...
Recherche...
Aucune correspondance
SharedMemoryParallelDispatch.cc
1// -*- tab-width: 2; indent-tabs-mode: nil; coding: utf-8-with-signature -*-
2//-----------------------------------------------------------------------------
3// Copyright 2000-2026 CEA (www.cea.fr) IFPEN (www.ifpenergiesnouvelles.com)
4// See the top-level COPYRIGHT file for details.
5// SPDX-License-Identifier: Apache-2.0
6//-----------------------------------------------------------------------------
7/*---------------------------------------------------------------------------*/
8/* SharedMemoryParallelDispatch.cc (C) 2000-2026 */
9/* */
10/* Implémentation des messages en mémoire partagée. */
11/*---------------------------------------------------------------------------*/
12/*---------------------------------------------------------------------------*/
13
14#include "arcane/utils/ArcanePrecomp.h"
15
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"
25
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"
31
32#include "arcane/parallel/thread/SharedMemoryParallelDispatch.h"
33#include "arcane/parallel/thread/SharedMemoryParallelMng.h"
34#include "arcane/parallel/thread/ISharedMemoryMessageQueue.h"
35
36#include "arccore/message_passing/PointToPointMessageInfo.h"
37
38/*---------------------------------------------------------------------------*/
39/*---------------------------------------------------------------------------*/
40
42{
43
44/*---------------------------------------------------------------------------*/
45/*---------------------------------------------------------------------------*/
46
47/*
48 * TODO: pour simplifier le debug lorsqu'il y a un décalage des appels
49 * collectifs entre les threads, il faudrait faire un type de barrière
50 * par type d'appel collectif alors qu'actuellement tous les appels
51 * collectifs utilisent la même barrière (via _collectiveBarrier()).
52 * A cause de cela, des problèmes peuvent survenir qui ne sont pas
53 * facilement détectable. Par exemple:
54 *
55 * Thread1:
56 * allGather();
57 * barrier();
58 * allReduce();
59 * Thread2:
60 * barrier();
61 * allGather();
62 * allReduce();
63 *
64 * Dans ce cas, le code ne plantera pas mais les valeurs des collectives ne
65 * seront pas bonnes.
66 */
67
68/*---------------------------------------------------------------------------*/
69/*---------------------------------------------------------------------------*/
70
71SharedMemoryParallelDispatchBase::
72SharedMemoryParallelDispatchBase(ITraceMng* tm, SharedMemoryParallelMng* parallel_mng,
73 ISharedMemoryMessageQueue* message_queue,
74 ArrayView<SharedMemoryParallelDispatchBase*> all_dispatchs_base)
75: TraceAccessor(tm)
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)
81{
82}
83
84/*---------------------------------------------------------------------------*/
85/*---------------------------------------------------------------------------*/
86
87void SharedMemoryParallelDispatchBase::
88_collectiveBarrier()
89{
90 m_parallel_mng->getThreadBarrier()->wait();
91}
92
93/*---------------------------------------------------------------------------*/
94/*---------------------------------------------------------------------------*/
95
96void SharedMemoryParallelDispatchBase::
97_genericAllToAll(ConstMemoryView send_buf, MutableMemoryView recv_buf, Int32 count)
98{
99 Int32 nb_rank = m_nb_rank;
100
101 //TODO: Faire une version sans allocation
102 Int32UniqueArray send_count(nb_rank, count);
103 Int32UniqueArray recv_count(nb_rank, count);
104
105 Int32UniqueArray send_indexes(nb_rank);
106 Int32UniqueArray recv_indexes(nb_rank);
107 for (Integer i = 0; i < nb_rank; ++i) {
108 send_indexes[i] = count * i;
109 recv_indexes[i] = count * i;
110 }
111 _genericAllToAllVariable(send_buf, send_count, send_indexes, recv_buf, recv_count, recv_indexes);
112}
113
114/*---------------------------------------------------------------------------*/
115/*---------------------------------------------------------------------------*/
116
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)
124{
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();
132 Integer global_index = 0;
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;
142 }
143 _collectiveBarrier();
144}
145
146/*---------------------------------------------------------------------------*/
147/*---------------------------------------------------------------------------*/
148
149void SharedMemoryParallelDispatchBase::
150_genericAllGather(ConstMemoryView send_buf, MutableMemoryView recv_buf)
151{
152 m_const_view = send_buf;
153 _collectiveBarrier();
154 MutableMemoryView recv_mem_view(recv_buf);
155 Int64 index = 0;
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);
160 index += size;
161 }
162 _collectiveBarrier();
163}
164
165/*---------------------------------------------------------------------------*/
166/*---------------------------------------------------------------------------*/
167
168void SharedMemoryParallelDispatchBase::
169_genericAllGatherVariable(ConstMemoryView send_buf, IResizableArray* recv_buf)
170{
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();
176 }
177 recv_buf->resize(total_size);
178 MutableMemoryView recv_mem_view(recv_buf->memoryView());
179 Int64 index = 0;
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);
184 index += size;
185 }
186 _collectiveBarrier();
187}
188
189/*---------------------------------------------------------------------------*/
190/*---------------------------------------------------------------------------*/
191
192void SharedMemoryParallelDispatchBase::
193_genericScatterVariable(ConstMemoryView send_buf, MutableMemoryView recv_buf, Int32 root)
194{
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);
200 Int64 index = 0;
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));
205 index += size;
206 }
207 }
208 _collectiveBarrier();
209}
210
211/*---------------------------------------------------------------------------*/
212/*---------------------------------------------------------------------------*/
213
214Request SharedMemoryParallelDispatchBase::
215_genericSend(ConstMemoryView send_buffer, const PointToPointMessageInfo& message2)
216{
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));
222 if (is_blocking) {
223 m_message_queue->waitAll(m_rank, ArrayView<Request>(1, &r));
224 return Request();
225 }
226 return r;
227 }
228 if (message.isMessageId()) {
229 // Le send avec un MessageId n'existe pas.
230 ARCCORE_THROW(NotSupportedException, "Invalid generic send with MessageId");
231 }
232 ARCCORE_THROW(NotSupportedException, "Invalid message_info");
233}
234
235/*---------------------------------------------------------------------------*/
236/*---------------------------------------------------------------------------*/
237
238Request SharedMemoryParallelDispatchBase::
239_genericReceive(MutableMemoryView recv_buffer, const PointToPointMessageInfo& message2)
240{
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);
246 if (is_blocking) {
247 m_message_queue->waitAll(m_rank, ArrayView<Request>(1, &r));
248 return MP::Request();
249 }
250 return r;
251}
252
253/*---------------------------------------------------------------------------*/
254/*---------------------------------------------------------------------------*/
255
256void SharedMemoryParallelDispatchBase::
257_genericBroadcast(MutableMemoryView send_buf, Int32 rank)
258{
259 m_broadcast_view = send_buf;
260 _collectiveBarrier();
261 MemoryUtils::copyHost(m_broadcast_view, m_all_dispatchs_base[rank]->m_broadcast_view);
262 _collectiveBarrier();
263}
264
265/*---------------------------------------------------------------------------*/
266/*---------------------------------------------------------------------------*/
267
268/*---------------------------------------------------------------------------*/
269/*---------------------------------------------------------------------------*/
270
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)
277{
278 m_reduce_infos.m_index = 0;
279 m_all_dispatchs[m_rank] = this;
280 m_all_dispatchs_base[m_rank] = this;
281}
282
283/*---------------------------------------------------------------------------*/
284/*---------------------------------------------------------------------------*/
285
286template <class Type> SharedMemoryParallelDispatch<Type>::
287~SharedMemoryParallelDispatch()
288{
289 finalize();
290}
291
292/*---------------------------------------------------------------------------*/
293/*---------------------------------------------------------------------------*/
294
295template <class Type> void SharedMemoryParallelDispatch<Type>::
296finalize()
297{
298}
299
300/*---------------------------------------------------------------------------*/
301/*---------------------------------------------------------------------------*/
302
303template <typename T>
304class _ThreadIntegralType
305{
306 public:
307
308 typedef FalseType IsIntegral;
309};
310
311#define ARCANE_DEFINE_INTEGRAL_TYPE(datatype) \
312 template <> \
313 class _ThreadIntegralType<datatype> \
314 { \
315 public: \
316\
317 typedef TrueType IsIntegral; \
318 }
319
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);
331
332/*---------------------------------------------------------------------------*/
333/*---------------------------------------------------------------------------*/
334
335namespace
336{
337
338 template <class Type> void
339 _computeMinMaxSum2(ArrayView<SharedMemoryParallelDispatch<Type>*> all_dispatchs,
340 Type& min_val, Type& max_val, Type& sum_val,
341 Int32& min_rank, Int32& max_rank, Int32 nb_rank, FalseType)
342 {
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);
350
351 throw NotImplementedException(A_FUNCINFO);
352 }
353
354 /*---------------------------------------------------------------------------*/
355 /*---------------------------------------------------------------------------*/
356
357 template <class Type> void
358 _computeMinMaxSum2(ArrayView<SharedMemoryParallelDispatch<Type>*> all_dispatchs,
359 Type& min_val, Type& max_val, Type& sum_val,
360 Int32& min_rank, Int32& max_rank, Int32 nb_rank, TrueType)
361 {
362 Type _min_val = all_dispatchs[0]->m_reduce_infos.reduce_value;
363 Type _max_val = _min_val;
364 Type _sum_val = _min_val;
365 Integer _min_rank = 0;
366 Integer _max_rank = 0;
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) {
370 _min_val = cval;
371 _min_rank = i;
372 }
373 if (cval > _max_val) {
374 _max_val = cval;
375 _max_rank = i;
376 }
377 _sum_val = (Type)(_sum_val + cval);
378 }
379 min_val = _min_val;
380 max_val = _max_val;
381 sum_val = _sum_val;
382 min_rank = _min_rank;
383 max_rank = _max_rank;
384 }
385
386} // namespace
387
388/*---------------------------------------------------------------------------*/
389/*---------------------------------------------------------------------------*/
390
391template <class Type> void SharedMemoryParallelDispatch<Type>::
392computeMinMaxSum(Type val, Type& min_val, Type& max_val, Type& sum_val,
393 Int32& min_rank, Int32& max_rank)
394{
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();
400}
401
402/*---------------------------------------------------------------------------*/
403/*---------------------------------------------------------------------------*/
404
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)
412{
413 // Implémentation sous-optimale qui ne vectorise pas le calcul
414 // (c'est actuellement un copier-coller d'au-dessus mis dans une boucle)
415 typedef typename _ThreadIntegralType<Type>::IsIntegral IntegralType;
416 Integer n = values.size();
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();
423 }
424}
425
426/*---------------------------------------------------------------------------*/
427/*---------------------------------------------------------------------------*/
428
429template <class Type> void SharedMemoryParallelDispatch<Type>::
430broadcast(Span<Type> send_buf, Int32 rank)
431{
432 _genericBroadcast(MutableMemoryView(send_buf), rank);
433}
434
435/*---------------------------------------------------------------------------*/
436/*---------------------------------------------------------------------------*/
437
438template <class Type> void SharedMemoryParallelDispatch<Type>::
439allGather(Span<const Type> send_buf, Span<Type> recv_buf)
440{
441 _genericAllGather(ConstMemoryView{ send_buf }, MutableMemoryView{ recv_buf });
442}
443
444/*---------------------------------------------------------------------------*/
445/*---------------------------------------------------------------------------*/
446
447template <class Type> void SharedMemoryParallelDispatch<Type>::
448gather(Span<const Type> send_buf, Span<Type> recv_buf, Int32 root_rank)
449{
450 UniqueArray<Type> tmp_buf;
451 if (m_rank == root_rank)
452 allGather(send_buf, recv_buf);
453 else {
454 tmp_buf.resize(send_buf.size() * m_nb_rank);
455 allGather(send_buf, tmp_buf);
456 }
457}
458
459/*---------------------------------------------------------------------------*/
460/*---------------------------------------------------------------------------*/
461
462template <class Type> void SharedMemoryParallelDispatch<Type>::
463allGatherVariable(Span<const Type> send_buf, Array<Type>& recv_buf)
464{
465 ResizableArrayRef recv_buf_ref(recv_buf);
466 _genericAllGatherVariable(ConstMemoryView(send_buf), &recv_buf_ref);
467}
468
469/*---------------------------------------------------------------------------*/
470/*---------------------------------------------------------------------------*/
471
472template <class Type> void SharedMemoryParallelDispatch<Type>::
473gatherVariable(Span<const Type> send_buf, Array<Type>& recv_buf, Int32 root_rank)
474{
475 UniqueArray<Type> tmp_buf;
476 if (m_rank == root_rank)
477 allGatherVariable(send_buf, recv_buf);
478 else
479 allGatherVariable(send_buf, tmp_buf);
480}
481
482/*---------------------------------------------------------------------------*/
483/*---------------------------------------------------------------------------*/
484
485template <class Type> void SharedMemoryParallelDispatch<Type>::
486scatterVariable(Span<const Type> send_buf, Span<Type> recv_buf, Int32 root)
487{
488 _genericScatterVariable(ConstMemoryView(send_buf), MutableMemoryView(recv_buf), root);
489}
490
491/*---------------------------------------------------------------------------*/
492/*---------------------------------------------------------------------------*/
493
494template <class Type> void SharedMemoryParallelDispatch<Type>::
495allToAll(Span<const Type> send_buf, Span<Type> recv_buf, Int32 count)
496{
497 _genericAllToAll(ConstMemoryView(send_buf), MutableMemoryView(recv_buf), count);
498}
499
500/*---------------------------------------------------------------------------*/
501/*---------------------------------------------------------------------------*/
502
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)
508{
509 _genericAllToAllVariable(ConstMemoryView(send_buf), send_count, send_index,
510 MutableMemoryView(recv_buf), recv_count, recv_index);
511}
512
513/*---------------------------------------------------------------------------*/
514/*---------------------------------------------------------------------------*/
515
516template <class Type> auto SharedMemoryParallelDispatch<Type>::
517send(Span<const Type> send_buffer, Int32 rank, bool is_blocking) -> Request
518{
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);
522}
523
524/*---------------------------------------------------------------------------*/
525/*---------------------------------------------------------------------------*/
526
527template <class Type> void SharedMemoryParallelDispatch<Type>::
528send(ConstArrayView<Type> send_buf, Int32 rank)
529{
530 send(send_buf, rank, true);
531}
532
533/*---------------------------------------------------------------------------*/
534/*---------------------------------------------------------------------------*/
535
536template <class Type> Parallel::Request SharedMemoryParallelDispatch<Type>::
537receive(Span<Type> recv_buffer, Int32 rank, bool is_blocking)
538{
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);
542}
543
544/*---------------------------------------------------------------------------*/
545/*---------------------------------------------------------------------------*/
546
547template <class Type> Request SharedMemoryParallelDispatch<Type>::
548send(Span<const Type> send_buffer, const PointToPointMessageInfo& message2)
549{
550 return _genericSend(ConstMemoryView(send_buffer), message2);
551}
552
553/*---------------------------------------------------------------------------*/
554/*---------------------------------------------------------------------------*/
555
556template <class Type> Request SharedMemoryParallelDispatch<Type>::
557receive(Span<Type> recv_buffer, const PointToPointMessageInfo& message2)
558{
559 return _genericReceive(MutableMemoryView(recv_buffer), message2);
560}
561
562/*---------------------------------------------------------------------------*/
563/*---------------------------------------------------------------------------*/
564
565template <class Type> void SharedMemoryParallelDispatch<Type>::
566recv(ArrayView<Type> recv_buffer, Integer rank)
567{
568 recv(recv_buffer, rank, true);
569}
570
571/*---------------------------------------------------------------------------*/
572/*---------------------------------------------------------------------------*/
573
574template <class Type> void SharedMemoryParallelDispatch<Type>::
575sendRecv(ConstArrayView<Type> send_buffer, ArrayView<Type> recv_buffer, Integer proc)
576{
577 ARCANE_UNUSED(send_buffer);
578 ARCANE_UNUSED(recv_buffer);
579 ARCANE_UNUSED(proc);
580 throw NotImplementedException(A_FUNCINFO);
581}
582
583/*---------------------------------------------------------------------------*/
584/*---------------------------------------------------------------------------*/
585
586template <class Type> Type SharedMemoryParallelDispatch<Type>::
587allReduce(eReduceType op, Type send_buf)
588{
589 m_reduce_infos.reduce_value = send_buf;
590 //cout << "ALL REDUCE BEGIN RANk=" << m_rank << " TYPE=" << (int)op << " MY=" << send_buf << '\n';
591 std::cout.flush();
592 _collectiveBarrier();
593 Type ret = m_all_dispatchs[0]->m_reduce_infos.reduce_value;
594 switch (op) {
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);
598 break;
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);
602 break;
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);
606 break;
607 default:
608 ARCANE_FATAL("Bad reduce type {0}", (int)op);
609 }
610 //cout << "ALL REDUCE RANK=" << m_rank << " TYPE=" << (int)op << " MY=" << send_buf << " GLOBAL=" << ret << '\n';
611 _collectiveBarrier();
612 return ret;
613}
614
615/*---------------------------------------------------------------------------*/
616/*---------------------------------------------------------------------------*/
617
618template <class Type> void SharedMemoryParallelDispatch<Type>::
619_allReduceOrScan(eReduceType op, Span<const Type> send_buf, Span<Type> receive_buf, bool is_scan)
620{
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;
627 if (use_in_place) {
628 receive_buf_tmp.resize(buf_size);
629 receive_buf = receive_buf_tmp;
630 }
631 //cout << "ALL REDUCE BEGIN RANk=" << m_rank << " TYPE=" << (int)op << " MY=" << send_buf << '\n';
632 //cout.flush();
633 _collectiveBarrier();
634 {
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}",
640 index0, indexi, i);
641 }
642 }
643 }
644 Int32 nb_rank = m_nb_rank;
645 if (is_scan)
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];
649 switch (op) {
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]);
654 break;
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]);
659 break;
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]);
664 break;
665 default:
666 ARCANE_FATAL("Bad reduce type");
667 }
668 //cout << "ALL REDUCE RANK=" << m_rank << " TYPE=" << (int)op << " MY=" << send_buf << " GLOBAL=" << ret << '\n';
669 _collectiveBarrier();
670 for (Integer j = 0; j < buf_size; ++j)
671 final_buf[j] = receive_buf[j];
672}
673
674/*---------------------------------------------------------------------------*/
675/*---------------------------------------------------------------------------*/
676
677template <class Type> void SharedMemoryParallelDispatch<Type>::
678allReduce(eReduceType op, Span<Type> send_buf)
679{
680 _allReduceOrScan(op, send_buf, send_buf, false);
681}
682
683/*---------------------------------------------------------------------------*/
684/*---------------------------------------------------------------------------*/
685
686template <class Type> void SharedMemoryParallelDispatch<Type>::
687allReduce(eReduceType op, Span<const Type> send_buf, Span<Type> receive_buf)
688{
689 _allReduceOrScan(op, send_buf, receive_buf, false);
690}
691
692/*---------------------------------------------------------------------------*/
693/*---------------------------------------------------------------------------*/
694
695template <class Type> Request SharedMemoryParallelDispatch<Type>::
696nonBlockingAllReduce(eReduceType op, Span<const Type> send_buf, Span<Type> recv_buf)
697{
698 ARCANE_UNUSED(op);
699 ARCANE_UNUSED(send_buf);
700 ARCANE_UNUSED(recv_buf);
701 throw NotImplementedException(A_FUNCINFO);
702}
703
704/*---------------------------------------------------------------------------*/
705/*---------------------------------------------------------------------------*/
706
707template <class Type> Request SharedMemoryParallelDispatch<Type>::
708nonBlockingAllGather(Span<const Type> send_buf, Span<Type> recv_buf)
709{
710 ARCANE_UNUSED(send_buf);
711 ARCANE_UNUSED(recv_buf);
712 throw NotImplementedException(A_FUNCINFO);
713}
714
715/*---------------------------------------------------------------------------*/
716/*---------------------------------------------------------------------------*/
717
718template <class Type> Request SharedMemoryParallelDispatch<Type>::
719nonBlockingBroadcast(Span<Type> send_buf, Int32 rank)
720{
721 ARCANE_UNUSED(send_buf);
722 ARCANE_UNUSED(rank);
723 throw NotImplementedException(A_FUNCINFO);
724}
725
726/*---------------------------------------------------------------------------*/
727/*---------------------------------------------------------------------------*/
728
729template <class Type> Request SharedMemoryParallelDispatch<Type>::
730nonBlockingGather(Span<const Type> send_buf, Span<Type> recv_buf, Int32 rank)
731{
732 ARCANE_UNUSED(send_buf);
733 ARCANE_UNUSED(recv_buf);
734 ARCANE_UNUSED(rank);
735 throw NotImplementedException(A_FUNCINFO);
736}
737
738/*---------------------------------------------------------------------------*/
739/*---------------------------------------------------------------------------*/
740
741template <class Type> Request SharedMemoryParallelDispatch<Type>::
742nonBlockingAllToAll(Span<const Type> send_buf, Span<Type> recv_buf, Int32 count)
743{
744 ARCANE_UNUSED(send_buf);
745 ARCANE_UNUSED(recv_buf);
746 ARCANE_UNUSED(count);
747 throw NotImplementedException(A_FUNCINFO);
748}
749
750/*---------------------------------------------------------------------------*/
751/*---------------------------------------------------------------------------*/
752
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)
757{
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);
765}
766
767/*---------------------------------------------------------------------------*/
768/*---------------------------------------------------------------------------*/
769
770template <class Type> Type SharedMemoryParallelDispatch<Type>::
771scan(eReduceType op, Type send_buf)
772{
773 ARCANE_UNUSED(op);
774 ARCANE_UNUSED(send_buf);
775 throw NotImplementedException(A_FUNCINFO);
776}
777
778/*---------------------------------------------------------------------------*/
779/*---------------------------------------------------------------------------*/
780
781template <class Type> void SharedMemoryParallelDispatch<Type>::
782scan(eReduceType op, ArrayView<Type> send_buf)
783{
784 _allReduceOrScan(op, send_buf, send_buf, true);
785}
786
787/*---------------------------------------------------------------------------*/
788/*---------------------------------------------------------------------------*/
789
790template <class Type> void SharedMemoryParallelDispatch<Type>::
791waitAll()
792{
793 // TEMPORAIRE: a priori pas utilisé
794 throw NotImplementedException(A_FUNCINFO);
795}
796
797/*---------------------------------------------------------------------------*/
798/*---------------------------------------------------------------------------*/
799
800template <class Type> Request SharedMemoryParallelDispatch<Type>::
802{
803 throw NotImplementedException(A_FUNCINFO);
804}
805
806/*---------------------------------------------------------------------------*/
807/*---------------------------------------------------------------------------*/
808
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>;
829
830/*---------------------------------------------------------------------------*/
831/*---------------------------------------------------------------------------*/
832
833} // End namespace Arcane::MessagePassing
834
835/*---------------------------------------------------------------------------*/
836/*---------------------------------------------------------------------------*/
#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.
Definition UtilsTypes.h:339
Type
Type of JSON value.
Definition rapidjson.h:730