Arcane  4.2.1.0
Documentation développeur
Chargement...
Recherche...
Aucune correspondance
HybridParallelDispatch.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/* MpiParallelDispatch.cc (C) 2000-2024 */
9/* */
10/* Gestionnaire de parallélisme utilisant les threads et MPI. */
11/*---------------------------------------------------------------------------*/
12/*---------------------------------------------------------------------------*/
13
14#include "arcane/utils/Array.h"
15#include "arcane/utils/PlatformUtils.h"
16#include "arcane/utils/String.h"
17#include "arcane/utils/ITraceMng.h"
18#include "arcane/utils/Real2.h"
19#include "arcane/utils/Real3.h"
20#include "arcane/utils/Real2x2.h"
21#include "arcane/utils/Real3x3.h"
22#include "arcane/utils/APReal.h"
23#include "arcane/utils/FatalErrorException.h"
24#include "arcane/utils/NotImplementedException.h"
25#include "arcane/utils/NotSupportedException.h"
26#include "arcane/utils/IThreadBarrier.h"
27#include "arcane/utils/CheckedConvert.h"
28
29#include "arcane/core/MeshVariableRef.h"
30#include "arcane/core/IParallelMng.h"
31#include "arcane/core/ItemGroup.h"
32#include "arcane/core/IMesh.h"
33#include "arcane/core/IBase.h"
34
35#include "arcane/parallel/mpithread/HybridParallelDispatch.h"
36#include "arcane/parallel/mpithread/HybridParallelMng.h"
37#include "arcane/parallel/mpithread/HybridMessageQueue.h"
38#include "arcane/parallel/mpi/MpiParallelMng.h"
39#include "arcane/parallel/mpi/MpiParallelDispatch.h"
40
41/*---------------------------------------------------------------------------*/
42/*---------------------------------------------------------------------------*/
43
45{
46
47/*---------------------------------------------------------------------------*/
48/*---------------------------------------------------------------------------*/
49
50//TODO: Fusionner avec ce qui est possible dans SharedMemoryParallelDispatch
51
52/*---------------------------------------------------------------------------*/
53/*---------------------------------------------------------------------------*/
54
55template <class Type> HybridParallelDispatch<Type>::
56HybridParallelDispatch(ITraceMng* tm, HybridParallelMng* pm, HybridMessageQueue* message_queue,
57 ArrayView<HybridParallelDispatch<Type>*> all_dispatchs)
58: TraceAccessor(tm)
59, m_parallel_mng(pm)
60, m_local_rank(pm->localRank())
61, m_local_nb_rank(pm->localNbRank())
62, m_global_rank(pm->commRank())
63, m_global_nb_rank(pm->commSize())
64, m_mpi_rank(pm->mpiParallelMng()->commRank())
65, m_mpi_nb_rank(pm->mpiParallelMng()->commSize())
66, m_all_dispatchs(all_dispatchs)
67, m_message_queue(message_queue)
68, m_mpi_dispatcher(0)
69{
70 m_reduce_infos.m_index = 0;
71
72 // Ce tableau a été dimensionné par le créateur de cette instance.
73 // Il faut juste mettre à jour la valeur correspondant à son rang
74 m_all_dispatchs[m_local_rank] = this;
75
76 // Récupère le dispatcher MPI pour ce type.
77 MpiParallelMng* mpi_pm = pm->mpiParallelMng();
78 IParallelDispatchT<Type>* pd = mpi_pm->dispatcher((Type*)nullptr);
79 if (!pd)
80 ARCANE_FATAL("null dispatcher");
81
82 m_mpi_dispatcher = dynamic_cast<MpiParallelDispatchT<Type>*>(pd);
83 if (!m_mpi_dispatcher)
84 ARCANE_FATAL("null mpi dispatcher");
85}
86
87/*---------------------------------------------------------------------------*/
88/*---------------------------------------------------------------------------*/
89
90template <class Type> HybridParallelDispatch<Type>::
91~HybridParallelDispatch()
92{
93 finalize();
94}
95
96/*---------------------------------------------------------------------------*/
97/*---------------------------------------------------------------------------*/
98
99template <class Type> void HybridParallelDispatch<Type>::
100finalize()
101{
102}
103
104/*---------------------------------------------------------------------------*/
105/*---------------------------------------------------------------------------*/
106
107template <typename T>
109{
110 public:
111
112 typedef FalseType IsIntegral;
113};
114
115#define ARCANE_DEFINE_INTEGRAL_TYPE(datatype) \
116 template <> \
117 class _ThreadIntegralType<datatype> \
118 { \
119 public: \
120\
121 typedef TrueType IsIntegral; \
122 }
123
124ARCANE_DEFINE_INTEGRAL_TYPE(long long);
125ARCANE_DEFINE_INTEGRAL_TYPE(long);
126ARCANE_DEFINE_INTEGRAL_TYPE(int);
127ARCANE_DEFINE_INTEGRAL_TYPE(short);
128ARCANE_DEFINE_INTEGRAL_TYPE(unsigned long long);
129ARCANE_DEFINE_INTEGRAL_TYPE(unsigned long);
130ARCANE_DEFINE_INTEGRAL_TYPE(unsigned int);
131ARCANE_DEFINE_INTEGRAL_TYPE(unsigned short);
132ARCANE_DEFINE_INTEGRAL_TYPE(double);
133ARCANE_DEFINE_INTEGRAL_TYPE(float);
134ARCANE_DEFINE_INTEGRAL_TYPE(HPReal);
135
136/*---------------------------------------------------------------------------*/
137/*---------------------------------------------------------------------------*/
138
139namespace
140{
141
142 template <class Type> void
143 _computeMinMaxSum2(ArrayView<HybridParallelDispatch<Type>*> all_dispatchs,
144 Int32 my_rank, Type& min_val, Type& max_val, Type& sum_val,
145 Int32& min_rank, Int32& max_rank, Int32 nb_rank, FalseType)
146 {
147 ARCANE_UNUSED(all_dispatchs);
148 ARCANE_UNUSED(my_rank);
149 ARCANE_UNUSED(min_val);
150 ARCANE_UNUSED(max_val);
151 ARCANE_UNUSED(sum_val);
152 ARCANE_UNUSED(min_rank);
153 ARCANE_UNUSED(max_rank);
154 ARCANE_UNUSED(nb_rank);
155
156 throw NotImplementedException(A_FUNCINFO);
157 }
158
159 /*---------------------------------------------------------------------------*/
160 /*---------------------------------------------------------------------------*/
161
162 template <class Type> void
163 _computeMinMaxSum2(ArrayView<HybridParallelDispatch<Type>*> all_dispatchs,
164 Int32 my_rank, Type& min_val, Type& max_val, Type& sum_val,
165 Int32& min_rank, Int32& max_rank, Int32 nb_rank, TrueType)
166 {
167 ARCANE_UNUSED(my_rank);
168
169 HybridParallelDispatch<Type>* mtpd0 = all_dispatchs[0];
170 Type cval0 = mtpd0->m_reduce_infos.reduce_value;
171 Type _min_val = cval0;
172 Type _max_val = cval0;
173 Type _sum_val = cval0;
174 Integer _min_rank = 0;
175 Integer _max_rank = 0;
176 for (Integer i = 1; i < nb_rank; ++i) {
177 HybridParallelDispatch<Type>* mtpd = all_dispatchs[i];
178 Type cval = mtpd->m_reduce_infos.reduce_value;
179 Int32 grank = mtpd->globalRank();
180 if (cval < _min_val) {
181 _min_val = cval;
182 _min_rank = grank;
183 }
184 if (_max_val < cval) {
185 _max_val = cval;
186 _max_rank = grank;
187 }
188 _sum_val = (Type)(_sum_val + cval);
189 }
190 min_val = _min_val;
191 max_val = _max_val;
192 sum_val = _sum_val;
193 min_rank = _min_rank;
194 max_rank = _max_rank;
195 }
196
197} // namespace
198
199/*---------------------------------------------------------------------------*/
200/*---------------------------------------------------------------------------*/
201
202template <class Type> void HybridParallelDispatch<Type>::
203computeMinMaxSum(Type val, Type& min_val, Type& max_val, Type& sum_val,
204 Int32& min_rank, Int32& max_rank)
205{
206 typedef typename _ThreadIntegralType<Type>::IsIntegral IntegralType;
207 m_reduce_infos.reduce_value = val;
208 _collectiveBarrier();
209 _computeMinMaxSum2(m_all_dispatchs, m_global_rank, min_val, max_val, sum_val, min_rank, max_rank, m_local_nb_rank, IntegralType());
210 if (m_local_rank == 0) {
211 /*pinfo() << "COMPUTE_MIN_MAX_SUM_B rank=" << m_global_rank
212 << " min_rank=" << min_rank
213 << " max_rank=" << max_rank
214 << " min_val=" << min_val
215 << " max_val=" << max_val
216 << " sum_val=" << sum_val;*/
217 m_mpi_dispatcher->computeMinMaxSumNoInit(min_val, max_val, sum_val, min_rank, max_rank);
218 /*pinfo() << "COMPUTE_MIN_MAX_SUM_A rank=" << m_global_rank
219 << " min_rank=" << min_rank
220 << " max_rank=" << max_rank;*/
221
222 m_min_max_sum_infos.m_min_value = min_val;
223 m_min_max_sum_infos.m_max_value = max_val;
224 m_min_max_sum_infos.m_sum_value = sum_val;
225 m_min_max_sum_infos.m_min_rank = min_rank;
226 m_min_max_sum_infos.m_max_rank = max_rank;
227 }
228 _collectiveBarrier();
229 m_min_max_sum_infos = m_all_dispatchs[0]->m_min_max_sum_infos;
230 min_val = m_min_max_sum_infos.m_min_value;
231 max_val = m_min_max_sum_infos.m_max_value;
232 sum_val = m_min_max_sum_infos.m_sum_value;
233 min_rank = m_min_max_sum_infos.m_min_rank;
234 max_rank = m_min_max_sum_infos.m_max_rank;
235 _collectiveBarrier();
236}
237
238/*---------------------------------------------------------------------------*/
239/*---------------------------------------------------------------------------*/
240
241template <class Type> void HybridParallelDispatch<Type>::
242computeMinMaxSum(ConstArrayView<Type> values,
243 ArrayView<Type> min_values,
244 ArrayView<Type> max_values,
245 ArrayView<Type> sum_values,
246 ArrayView<Int32> min_ranks,
247 ArrayView<Int32> max_ranks)
248{
249 // Implémentation sous-optimale qui ne vectorise pas le calcul
250 // (c'est actuellement un copier-coller d'au-dessus mis dans une boucle)
251 typedef typename _ThreadIntegralType<Type>::IsIntegral IntegralType;
252 Integer n = values.size();
253 for (Integer i = 0; i < n; ++i) {
254 m_reduce_infos.reduce_value = values[i];
255 _collectiveBarrier();
256 _computeMinMaxSum2(m_all_dispatchs, m_global_rank, min_values[i], max_values[i], sum_values[i], min_ranks[i], max_ranks[i], m_local_nb_rank, IntegralType());
257 if (m_local_rank == 0) {
258 /*pinfo() << "COMPUTE_MIN_MAX_SUM_B rank=" << m_global_rank
259 << " min_rank=" << min_rank
260 << " max_rank=" << max_rank
261 << " min_val=" << min_val
262 << " max_val=" << max_val
263 << " sum_val=" << sum_val;*/
264 m_mpi_dispatcher->computeMinMaxSumNoInit(min_values[i], max_values[i], sum_values[i], min_ranks[i], max_ranks[i]);
265 /*pinfo() << "COMPUTE_MIN_MAX_SUM_A rank=" << m_global_rank
266 << " min_rank=" << min_rank
267 << " max_rank=" << max_rank;*/
268
269 m_min_max_sum_infos.m_min_value = min_values[i];
270 m_min_max_sum_infos.m_max_value = max_values[i];
271 m_min_max_sum_infos.m_sum_value = sum_values[i];
272 m_min_max_sum_infos.m_min_rank = min_ranks[i];
273 m_min_max_sum_infos.m_max_rank = max_ranks[i];
274 }
275 _collectiveBarrier();
276 m_min_max_sum_infos = m_all_dispatchs[0]->m_min_max_sum_infos;
277 min_values[i] = m_min_max_sum_infos.m_min_value;
278 max_values[i] = m_min_max_sum_infos.m_max_value;
279 sum_values[i] = m_min_max_sum_infos.m_sum_value;
280 min_ranks[i] = m_min_max_sum_infos.m_min_rank;
281 max_ranks[i] = m_min_max_sum_infos.m_max_rank;
282 _collectiveBarrier();
283 }
284}
285
286/*---------------------------------------------------------------------------*/
287/*---------------------------------------------------------------------------*/
288
289template <class Type> void HybridParallelDispatch<Type>::
290broadcast(Span<Type> send_buf, Int32 rank)
291{
292 m_broadcast_view = send_buf;
293 _collectiveBarrier();
294 FullRankInfo fri = FullRankInfo::compute(MP::MessageRank(rank), m_local_nb_rank);
295 int mpi_rank = fri.mpiRankValue();
296 if (m_mpi_rank == mpi_rank) {
297 // J'ai le même rang MPI que celui qui effectue le broadcast
298 if (m_global_rank == rank) {
299 //TODO: Passage 64 bits.
300 m_parallel_mng->mpiParallelMng()->broadcast(send_buf.smallView(), mpi_rank);
301 }
302 else {
303 m_all_dispatchs[m_local_rank]->m_broadcast_view.copy(m_all_dispatchs[fri.localRankValue()]->m_broadcast_view);
304 }
305 }
306 else {
307 if (m_local_rank == 0) {
308 //TODO: Passage 64 bits.
309 m_parallel_mng->mpiParallelMng()->broadcast(send_buf.smallView(), mpi_rank);
310 }
311 }
312 _collectiveBarrier();
313 if (m_mpi_rank != mpi_rank) {
314 if (m_local_rank != 0)
315 m_all_dispatchs[m_local_rank]->m_broadcast_view.copy(m_all_dispatchs[0]->m_broadcast_view);
316 }
317 _collectiveBarrier();
318}
319
320/*---------------------------------------------------------------------------*/
321/*---------------------------------------------------------------------------*/
322
323template <class Type> void HybridParallelDispatch<Type>::
324allGather(Span<const Type> send_buf, Span<Type> recv_buf)
325{
326 //TODO: fusionner avec allGatherVariable()
327 m_const_view = send_buf;
328 _collectiveBarrier();
329 Int64 total_size = 0;
330 for (Int32 i = 0; i < m_local_nb_rank; ++i) {
331 total_size += m_all_dispatchs[i]->m_const_view.size();
332 }
333 if (m_local_rank == 0) {
334 Int64 index = 0;
335 UniqueArray<Type> local_buf(total_size);
336 for (Integer i = 0; i < m_local_nb_rank; ++i) {
337 Span<const Type> view = m_all_dispatchs[i]->m_const_view;
338 Int64 size = view.size();
339 for (Int64 j = 0; j < size; ++j)
340 local_buf[j + index] = view[j];
341 index += size;
342 }
343 IParallelMng* pm = m_parallel_mng->mpiParallelMng();
344 //TODO: 64bit
345 pm->allGather(local_buf, recv_buf.smallView());
346 m_const_view = recv_buf;
347 }
348 _collectiveBarrier();
349 if (m_local_rank != 0) {
350 Span<const Type> view = m_all_dispatchs[0]->m_const_view;
351 recv_buf.copy(view);
352 }
353 _collectiveBarrier();
354}
355
356/*---------------------------------------------------------------------------*/
357/*---------------------------------------------------------------------------*/
358
359template <class Type> void HybridParallelDispatch<Type>::
360gather(Span<const Type> send_buf, Span<Type> recv_buf, Int32 root_rank)
361{
362 UniqueArray<Type> tmp_buf;
363 if (m_global_rank == root_rank)
364 allGather(send_buf, recv_buf);
365 else {
366 tmp_buf.resize(send_buf.size() * m_global_nb_rank);
367 allGather(send_buf, tmp_buf);
368 }
369}
370
371/*---------------------------------------------------------------------------*/
372/*---------------------------------------------------------------------------*/
373
374template <class Type> void HybridParallelDispatch<Type>::
375allGatherVariable(Span<const Type> send_buf, Array<Type>& recv_buf)
376{
377 m_const_view = send_buf;
378 _collectiveBarrier();
379 Int64 total_size = 0;
380 for (Integer i = 0; i < m_local_nb_rank; ++i) {
381 total_size += m_all_dispatchs[i]->m_const_view.size();
382 }
383 if (m_local_rank == 0) {
384 Int64 index = 0;
385 UniqueArray<Type> local_buf(total_size);
386 for (Integer i = 0; i < m_local_nb_rank; ++i) {
387 Span<const Type> view = m_all_dispatchs[i]->m_const_view;
388 Int64 size = view.size();
389 for (Int64 j = 0; j < size; ++j)
390 local_buf[j + index] = view[j];
391 index += size;
392 }
393 m_parallel_mng->mpiParallelMng()->allGatherVariable(local_buf, recv_buf);
394 m_const_view = recv_buf.constView();
395 }
396 _collectiveBarrier();
397 if (m_local_rank != 0) {
398 Span<const Type> view = m_all_dispatchs[0]->m_const_view;
399 recv_buf.resize(view.size());
400 recv_buf.copy(view);
401 }
402 _collectiveBarrier();
403}
404
405/*---------------------------------------------------------------------------*/
406/*---------------------------------------------------------------------------*/
407
408template <class Type> void HybridParallelDispatch<Type>::
409gatherVariable(Span<const Type> send_buf, Array<Type>& recv_buf, Int32 root_rank)
410{
411 UniqueArray<Type> tmp_buf;
412 if (m_global_rank == root_rank)
413 allGatherVariable(send_buf, recv_buf);
414 else
415 allGatherVariable(send_buf, tmp_buf);
416}
417
418/*---------------------------------------------------------------------------*/
419/*---------------------------------------------------------------------------*/
420
421template <class Type>
422void HybridParallelDispatch<Type>::
423scatterVariable(Span<const Type> send_buf, Span<Type> recv_buf, Int32 root)
424{
425 m_const_view = send_buf;
426 m_recv_view = recv_buf;
427
428 _collectiveBarrier();
429
430 // On calcule le nombre d'élément que veut tous les threads de notre processus.
431 Int64 total_size = 0;
432 for (Integer i = 0; i < m_local_nb_rank; ++i) {
433 total_size += m_all_dispatchs[i]->m_recv_view.size();
434 }
435
436 _collectiveBarrier();
437
438 // Les échanges MPI s'effectuent uniquement par les threads leaders des processus.
439 if (m_local_rank == 0) {
440 FullRankInfo fri(FullRankInfo::compute(MessageRank(root), m_local_nb_rank));
441
442 UniqueArray<Type> local_recv_buf(total_size);
443
444 // Si le thread "root" est dans notre processus.
445 if (m_mpi_rank == fri.mpiRankValue()) {
446 // Le thread leader s'occupe de l'échange.
447 m_parallel_mng->mpiParallelMng()->scatterVariable(m_all_dispatchs[fri.localRankValue()]->m_const_view.smallView(),
448 local_recv_buf, fri.mpiRankValue());
449 }
450 // Les autres threads leaders mettent leurs buffers d'envoi (qu'importe ce
451 // qu'ils contiennent, c'est un scatter).
452 else {
453 m_parallel_mng->mpiParallelMng()->scatterVariable(m_const_view.smallView(), local_recv_buf, fri.mpiRankValue());
454 }
455
456 // On a plus qu'à répartir les données reçues entre les threads.
457 Integer compt = 0;
458 for (Integer i = 0; i < m_local_nb_rank; ++i) {
459 Int64 size = m_all_dispatchs[i]->m_recv_view.size();
460 for (Integer j = 0; j < size; ++j) {
461 m_all_dispatchs[i]->m_recv_view[j] = local_recv_buf[compt++];
462 }
463 }
464 }
465 _collectiveBarrier();
466 recv_buf.copy(m_recv_view);
467 _collectiveBarrier();
468}
469
470/*---------------------------------------------------------------------------*/
471/*---------------------------------------------------------------------------*/
472
473template <class Type> void HybridParallelDispatch<Type>::
474allToAll(Span<const Type> send_buf, Span<Type> recv_buf, Int32 count)
475{
476 Int32 global_nb_rank = m_global_nb_rank;
477 //TODO: Faire une version sans allocation
478 Int32UniqueArray send_count(global_nb_rank, count);
479 Int32UniqueArray recv_count(global_nb_rank, count);
480
481 Int32UniqueArray send_indexes(global_nb_rank);
482 Int32UniqueArray recv_indexes(global_nb_rank);
483 for (Integer i = 0; i < global_nb_rank; ++i) {
484 send_indexes[i] = count * i;
485 recv_indexes[i] = count * i;
486 }
487 this->allToAllVariable(send_buf, send_count, send_indexes, recv_buf, recv_count, recv_indexes);
488}
489
490/*---------------------------------------------------------------------------*/
491/*---------------------------------------------------------------------------*/
492
493template <class Type> void HybridParallelDispatch<Type>::
494allToAllVariable(Span<const Type> g_send_buf,
495 Int32ConstArrayView g_send_count,
496 Int32ConstArrayView g_send_index,
497 Span<Type> g_recv_buf,
498 Int32ConstArrayView g_recv_count,
499 Int32ConstArrayView g_recv_index)
500{
501 m_alltoallv_infos.send_buf = g_send_buf;
502 m_alltoallv_infos.send_count = g_send_count;
503 m_alltoallv_infos.send_index = g_send_index;
504 m_alltoallv_infos.recv_buf = g_recv_buf;
505 m_alltoallv_infos.recv_count = g_recv_count;
506 m_alltoallv_infos.recv_index = g_recv_index;
507
508 _collectiveBarrier();
509
510 UniqueArray<Type> tmp_recv_buf;
511
512 // PREMIERE IMPLEMENTATION
513 // Le proc de rang local 0 fait tout le travail.
514
515 if (m_local_rank == 0) {
516
517 Int32UniqueArray tmp_send_count(m_mpi_nb_rank);
518 tmp_send_count.fill(0);
519 Int32UniqueArray tmp_recv_count(m_mpi_nb_rank);
520 tmp_recv_count.fill(0);
521
522 Int64 total_send_size = 0;
523 Int64 total_recv_size = 0;
524
525 for (Integer i = 0; i < m_local_nb_rank; ++i) {
526 const AllToAllVariableInfo& vinfo = m_all_dispatchs[i]->m_alltoallv_infos;
527 total_send_size += vinfo.send_buf.size();
528 total_recv_size += vinfo.recv_buf.size();
529 }
530
531 UniqueArray<Type> tmp_send_buf(total_send_size);
532 tmp_recv_buf.resize(total_recv_size);
533
534 // Calcule le nombre d'éléments à envoyer et recevoir pour chaque proc.
535 for (Integer i = 0; i < m_local_nb_rank; ++i) {
536 const AllToAllVariableInfo& vinfo = m_all_dispatchs[i]->m_alltoallv_infos;
537
538 for (Integer z = 0; z < m_global_nb_rank; ++z) {
539 FullRankInfo fri(FullRankInfo::compute(MP::MessageRank(z), m_local_nb_rank));
540 Int32 fri_mpi_rank = fri.mpiRankValue();
541
542 Int32 nb_send = vinfo.send_count[z];
543
544 tmp_send_count[fri_mpi_rank] += nb_send;
545 tmp_recv_count[fri_mpi_rank] += vinfo.recv_count[z];
546
547#if 0
548 info() << "my_local=" << i << " dest=" << z
549 << " send_count=" << vinfo.send_count[z] << " send_index=" << vinfo.send_index[z]
550 << " recv_count=" << vinfo.recv_count[z] << " recv_index=" << vinfo.recv_index[z];
551 {
552 Integer vindex = vinfo.send_index[z];
553 for( Integer w=0, wn=vinfo.send_count[z]; w<wn; ++w ){
554 info() << "V=" << vinfo.send_buf[ vindex + w ];
555 }
556 }
557#endif
558 }
559 }
560
561 Int32UniqueArray tmp_send_index(m_mpi_nb_rank);
562 Int32UniqueArray tmp_recv_index(m_mpi_nb_rank);
563 tmp_send_index[0] = 0;
564 tmp_recv_index[0] = 0;
565 for (Integer k = 1, nmpi = m_mpi_nb_rank; k < nmpi; ++k) {
566 tmp_send_index[k] = tmp_send_index[k - 1] + tmp_send_count[k - 1];
567 tmp_recv_index[k] = tmp_recv_index[k - 1] + tmp_recv_count[k - 1];
568 }
569
570 for (Integer i = 0; i < m_local_nb_rank; ++i) {
571 const AllToAllVariableInfo& vinfo = m_all_dispatchs[i]->m_alltoallv_infos;
572
573 for (Integer z = 0; z < m_global_nb_rank; ++z) {
574
575 FullRankInfo fri(FullRankInfo::compute(MP::MessageRank(z), m_local_nb_rank));
576 Int32 fri_mpi_rank = fri.mpiRankValue();
577
578 Integer nb_send = vinfo.send_count[z];
579 {
580
581 Integer tmp_current_index = tmp_send_index[fri_mpi_rank];
582 Integer local_current_index = vinfo.send_index[z];
583 for (Integer j = 0; j < nb_send; ++j)
584 tmp_send_buf[j + tmp_current_index] = vinfo.send_buf[j + local_current_index];
585 tmp_send_index[fri_mpi_rank] += nb_send;
586 }
587 }
588 }
589
590 tmp_send_index[0] = 0;
591 tmp_recv_index[0] = 0;
592 for (Integer k = 1, nmpi = m_mpi_nb_rank; k < nmpi; ++k) {
593 tmp_send_index[k] = tmp_send_index[k - 1] + tmp_send_count[k - 1];
594 tmp_recv_index[k] = tmp_recv_index[k - 1] + tmp_recv_count[k - 1];
595 }
596
597 /* Integer send_index = 0;
598 for( Integer i=0; i<m_local_nb_rank; ++i ){
599 ConstArrayView<Type> send_view = m_all_dispatchs[i]->m_alltoallv_infos.send_buf;
600 Integer send_size = send_view.size();
601 info() << "ADD_TMP_SEND_BUF send_index=" << send_index << " size=" << send_size;
602 for( Integer j=0; j<send_size; ++j )
603 tmp_send_buf[j+send_index] = send_view[j];
604 send_index += send_size;
605 }
606 */
607
608#if 0
609 info() << "AllToAllV nb_send=" << total_send_size << " nb_recv=" << total_recv_size;
610 for( Integer k=0; k<m_mpi_nb_rank; ++k ){
611 info() << "INFOS Rank=" << k << " send_count=" << tmp_send_count[k] << " recv_count=" << tmp_recv_count[k]
612 << " send_index=" << tmp_send_index[k] << " recv_index=" << tmp_recv_index[k];
613 }
614
615 for( Integer i=0; i<tmp_send_buf.size(); ++i )
616 info() << "SEND_BUF[" << i << "] = " << tmp_send_buf[i];
617
618 for( Integer k=0; k<m_mpi_nb_rank; ++k ){
619 info() << "SEND Rank=" << k << " send_count=" << tmp_send_count[k] << " recv_count=" << tmp_recv_count[k]
620 << " send_index=" << tmp_send_index[k] << " recv_index=" << tmp_recv_index[k];
621 Integer vindex = tmp_send_index[k];
622 for( Integer w=0, wn=tmp_send_count[k]; w<wn; ++w ){
623 info() << "V=" << tmp_send_buf[ vindex + w ];
624 }
625 }
626#endif
627
628 m_parallel_mng->mpiParallelMng()->allToAllVariable(tmp_send_buf, tmp_send_count,
629 tmp_send_index, tmp_recv_buf,
630 tmp_recv_count, tmp_recv_index);
631
632#if 0
633 for( Integer i=0; i<tmp_recv_buf.size(); ++i )
634 info() << "RECV_BUF[" << i << "] = " << tmp_recv_buf[i];
635
636 for( Integer k=0; k<m_mpi_nb_rank; ++k ){
637 info() << "RECV Rank=" << k << " send_count=" << tmp_send_count[k] << " recv_count=" << tmp_recv_count[k]
638 << " send_index=" << tmp_send_index[k] << " recv_index=" << tmp_recv_index[k];
639 Integer vindex = tmp_recv_index[k];
640 for( Integer w=0, wn=tmp_recv_count[k]; w<wn; ++w ){
641 info() << "V=" << tmp_recv_buf[ vindex + w ];
642 }
643 }
644#endif
645
646 m_const_view = tmp_recv_buf.constView();
647
648 for (Integer z = 0; z < m_global_nb_rank; ++z) {
649 FullRankInfo fri(FullRankInfo::compute(MP::MessageRank(z), m_local_nb_rank));
650 Int32 fri_mpi_rank = fri.mpiRankValue();
651
652 for (Integer i = 0; i < m_local_nb_rank; ++i) {
653 AllToAllVariableInfo& vinfo = m_all_dispatchs[i]->m_alltoallv_infos;
654 Span<Type> my_buf = vinfo.recv_buf;
655 Int64 recv_size = vinfo.recv_count[z];
656 Int64 recv_index = tmp_recv_index[fri_mpi_rank];
657
658 Span<const Type> recv_view = tmp_recv_buf.span().subSpan(recv_index, recv_size);
659
660 Int64 my_recv_index = vinfo.recv_index[z];
661
662 //info() << "GET i=" << i << " z=" << z << " size=" << recv_size << " index=" << recv_index
663 // << " mpi_rank=" << fri_mpi_rank << " my_index=" << my_recv_index;
664
665 tmp_recv_index[fri_mpi_rank] = CheckedConvert::toInt32(tmp_recv_index[fri_mpi_rank] + recv_size);
666
667 for (Int64 j = 0; j < recv_size; ++j)
668 my_buf[j + my_recv_index] = recv_view[j];
669
670 //for( Integer j=0; j<recv_size; ++j )
671 //info() << "V=" << recv_view[j];
672
673 my_recv_index += recv_size;
674 }
675 }
676 }
677 _collectiveBarrier();
678
679 //info() << "END_PHASE_1_ALL_TO_ALL_V my_rank=" << m_global_rank << " (local=" << m_local_rank << ")";
680
681 //_collectiveBarrier();
682}
683
684/*---------------------------------------------------------------------------*/
685/*---------------------------------------------------------------------------*/
686
687template <class Type> auto HybridParallelDispatch<Type>::
688send(Span<const Type> send_buffer, Int32 rank, bool is_blocked) -> Request
689{
690 eBlockingType block_mode = (is_blocked) ? MP::Blocking : MP::NonBlocking;
691 PointToPointMessageInfo p2p_message(MessageRank(rank), block_mode);
692 return send(send_buffer, p2p_message);
693}
694
695/*---------------------------------------------------------------------------*/
696/*---------------------------------------------------------------------------*/
697
698template <class Type> void HybridParallelDispatch<Type>::
699send(ConstArrayView<Type> send_buf, Int32 rank)
700{
701 send(send_buf, rank, true);
702}
703
704/*---------------------------------------------------------------------------*/
705/*---------------------------------------------------------------------------*/
706
707template <class Type> Parallel::Request HybridParallelDispatch<Type>::
708receive(Span<Type> recv_buffer, Int32 rank, bool is_blocked)
709{
710 eBlockingType block_mode = (is_blocked) ? MP::Blocking : MP::NonBlocking;
711 PointToPointMessageInfo p2p_message(MessageRank(rank), block_mode);
712 return receive(recv_buffer, p2p_message);
713}
714
715/*---------------------------------------------------------------------------*/
716/*---------------------------------------------------------------------------*/
717
718template <class Type> Request HybridParallelDispatch<Type>::
719send(Span<const Type> send_buffer, const PointToPointMessageInfo& message2)
720{
721 PointToPointMessageInfo message(message2);
722 bool is_blocking = message.isBlocking();
723 message.setEmiterRank(MessageRank(m_global_rank));
724 Request r = m_message_queue->addSend(message, ConstMemoryView(send_buffer));
725 if (is_blocking) {
726 m_message_queue->waitAll(ArrayView<MP::Request>(1, &r));
727 return Request();
728 }
729 return r;
730}
731
732/*---------------------------------------------------------------------------*/
733/*---------------------------------------------------------------------------*/
734
735template <class Type> Request HybridParallelDispatch<Type>::
736receive(Span<Type> recv_buffer, const PointToPointMessageInfo& message2)
737{
738 PointToPointMessageInfo message(message2);
739 message.setEmiterRank(MessageRank(m_global_rank));
740 bool is_blocking = message.isBlocking();
741 Request r = m_message_queue->addReceive(message, ReceiveBufferInfo(MutableMemoryView(recv_buffer)));
742 if (is_blocking) {
743 m_message_queue->waitAll(ArrayView<Request>(1, &r));
744 return Request();
745 }
746 return r;
747}
748
749/*---------------------------------------------------------------------------*/
750/*---------------------------------------------------------------------------*/
751
752template <class Type> void HybridParallelDispatch<Type>::
753recv(ArrayView<Type> recv_buffer, Integer rank)
754{
755 recv(recv_buffer, rank, true);
756}
757
758/*---------------------------------------------------------------------------*/
759/*---------------------------------------------------------------------------*/
760
761template <class Type> void HybridParallelDispatch<Type>::
762sendRecv(ConstArrayView<Type> send_buffer, ArrayView<Type> recv_buffer, Integer proc)
763{
764 ARCANE_UNUSED(send_buffer);
765 ARCANE_UNUSED(recv_buffer);
766 ARCANE_UNUSED(proc);
767 throw NotImplementedException(A_FUNCINFO);
768}
769
770/*---------------------------------------------------------------------------*/
771/*---------------------------------------------------------------------------*/
772
773template <class Type> Type HybridParallelDispatch<Type>::
774allReduce(eReduceType op, Type send_buf)
775{
776 m_reduce_infos.reduce_value = send_buf;
777 //pinfo() << "ALL REDUCE BEGIN RANK=" << m_global_rank << " TYPE=" << (int)op << " MY=" << send_buf;
778 std::cout.flush();
779 _collectiveBarrier();
780 if (m_local_rank == 0) {
781 Type ret = m_all_dispatchs[0]->m_reduce_infos.reduce_value;
782 switch (op) {
783 case Parallel::ReduceMin:
784 for (Integer i = 1; i < m_local_nb_rank; ++i)
785 ret = math::min(ret, m_all_dispatchs[i]->m_reduce_infos.reduce_value);
786 break;
787 case Parallel::ReduceMax:
788 for (Integer i = 1; i < m_local_nb_rank; ++i)
789 ret = math::max(ret, m_all_dispatchs[i]->m_reduce_infos.reduce_value);
790 break;
791 case Parallel::ReduceSum:
792 for (Integer i = 1; i < m_local_nb_rank; ++i)
793 ret = (Type)(ret + m_all_dispatchs[i]->m_reduce_infos.reduce_value);
794 break;
795 default:
796 ARCANE_FATAL("Bad reduce type");
797 }
798 ret = m_parallel_mng->mpiParallelMng()->reduce(op, ret);
799 m_all_dispatchs[0]->m_reduce_infos.reduce_value = ret;
800 //pinfo() << "ALL REDUCE RANK=" << m_local_rank << " TYPE=" << (int)op << " MY=" << send_buf << " GLOBAL=" << ret << '\n';
801 }
802 _collectiveBarrier();
803 Type ret = m_all_dispatchs[0]->m_reduce_infos.reduce_value;
804 _collectiveBarrier();
805 return ret;
806}
807
808/*---------------------------------------------------------------------------*/
809/*---------------------------------------------------------------------------*/
810
811template <class Type> void HybridParallelDispatch<Type>::
812_applyReduceOperator(eReduceType op, Span<Type> result, AllDispatchView dispatch_view,
813 Int32 first_rank, Int32 last_rank)
814{
815 Int64 buf_size = result.size();
816 switch (op) {
817 case Parallel::ReduceMin:
818 for (Integer i = first_rank; i <= last_rank; ++i)
819 for (Int64 j = 0; j < buf_size; ++j)
820 result[j] = math::min(result[j], dispatch_view[i]->m_reduce_infos.reduce_buf_span[j]);
821 break;
822 case Parallel::ReduceMax:
823 for (Integer i = first_rank; i <= last_rank; ++i)
824 for (Int64 j = 0; j < buf_size; ++j)
825 result[j] = math::max(result[j], dispatch_view[i]->m_reduce_infos.reduce_buf_span[j]);
826 break;
827 case Parallel::ReduceSum:
828 for (Integer i = first_rank; i <= last_rank; ++i)
829 for (Integer j = 0; j < buf_size; ++j) {
830 result[j] = static_cast<Type>(result[j] + dispatch_view[i]->m_reduce_infos.reduce_buf_span[j]);
831 }
832 break;
833 default:
834 ARCANE_FATAL("Bad reduce type");
835 }
836}
837
838/*---------------------------------------------------------------------------*/
839/*---------------------------------------------------------------------------*/
840
841template <class Type> void HybridParallelDispatch<Type>::
842_allReduceOrScan(eReduceType op, Span<Type> send_buf, bool is_scan)
843{
844 m_reduce_infos.reduce_buf_span = send_buf;
845 ++m_reduce_infos.m_index;
846 Int64 buf_size = send_buf.size();
847 UniqueArray<Type> ret(buf_size);
848 // Valeurs du rang MPI précédent (utilisé uniquement en mode Scan)
849 UniqueArray<Type> previous_rank_ret;
850 MpiParallelMng* mpi_pm = m_parallel_mng->mpiParallelMng();
851 Int32 my_mpi_rank = mpi_pm->commRank();
852 Int32 mpi_nb_rank = mpi_pm->commSize();
853
854 //cout << "ALL REDUCE BEGIN RANk=" << m_local_rank << " TYPE=" << (int)op << " MY=" << send_buf << '\n';
855 //cout.flush();
856 _collectiveBarrier();
857 {
858 Integer index0 = m_all_dispatchs[0]->m_reduce_infos.m_index;
859 for (Integer i = 0; i < m_local_nb_rank; ++i) {
860 Integer indexi = m_all_dispatchs[i]->m_reduce_infos.m_index;
861 if (index0 != m_all_dispatchs[i]->m_reduce_infos.m_index) {
862 ARCANE_FATAL("INTERNAL: incoherent all reduce i0={0} in={1} n={2}",
863 index0, indexi, i);
864 }
865 }
866 }
867
868 if (m_local_rank == 0) {
869 const Int32 nb_local_rank = m_local_nb_rank;
870 for (Integer j = 0; j < buf_size; ++j)
871 ret[j] = m_all_dispatchs[0]->m_reduce_infos.reduce_buf_span[j];
872 _applyReduceOperator(op, ret, m_all_dispatchs, 1, nb_local_rank - 1);
873 if (is_scan) {
874 // Pour le scan, on a besoin de savoir la valeur du scan du rang qui nous précéde.
875 // On utilise ensuite cette valeur et on applique notre opérateur.
876 mpi_pm->scan(op, ret);
877 previous_rank_ret.resize(buf_size);
878 UniqueArray<Request> requests;
879 if (my_mpi_rank != 0)
880 requests.add(mpi_pm->recv(previous_rank_ret, my_mpi_rank - 1, false));
881 if (my_mpi_rank != (mpi_nb_rank - 1))
882 requests.add(mpi_pm->send(ret, my_mpi_rank + 1, false));
883 mpi_pm->waitAllRequests(requests);
884 if (my_mpi_rank != 0) {
885 // Applique le scan à mes valeurs.
886 _applyReduceOperator(op, previous_rank_ret, m_all_dispatchs, 0, 0);
887 send_buf.copy(previous_rank_ret);
888 }
889 else {
890 // Je suis le premier rang local et MPI. J'ai déja les bonnes valeurs
891 // dans \a send_buf.
892 }
893 }
894 else {
895 mpi_pm->reduce(op, ret);
896 send_buf.copy(ret);
897 }
898 }
899
900 _collectiveBarrier();
901
902 if (is_scan) {
903 if (m_local_rank != 0) {
904 Span<const Type> global_buf = m_all_dispatchs[0]->m_reduce_infos.reduce_buf_span;
905 ret.copy(global_buf);
906 // Le scan pour le rank local 0 a déjà été appliqué
907 _applyReduceOperator(op, ret, m_all_dispatchs, 1, m_local_rank);
908 }
909 // TODO: On pourrait éviter cette barrière si on copiait les valeurs de 'send_buf'
910 // avant de les modifier.
911 _collectiveBarrier();
912
913 if (m_local_rank != 0) {
914 send_buf.copy(ret);
915 }
916 }
917 else {
918 if (m_local_rank != 0) {
919 Span<const Type> global_buf = m_all_dispatchs[0]->m_reduce_infos.reduce_buf_span;
920 send_buf.copy(global_buf);
921 }
922 }
923
924 _collectiveBarrier();
925}
926
927/*---------------------------------------------------------------------------*/
928/*---------------------------------------------------------------------------*/
929
930template <class Type> void HybridParallelDispatch<Type>::
931allReduce(eReduceType op, Span<Type> send_buf)
932{
933 _allReduceOrScan(op, send_buf, false);
934}
935
936/*---------------------------------------------------------------------------*/
937/*---------------------------------------------------------------------------*/
938
939template <class Type> Request HybridParallelDispatch<Type>::
940nonBlockingAllReduce(eReduceType op, Span<const Type> send_buf, Span<Type> recv_buf)
941{
942 ARCANE_UNUSED(op);
943 ARCANE_UNUSED(send_buf);
944 ARCANE_UNUSED(recv_buf);
945 throw NotImplementedException(A_FUNCINFO);
946}
947
948/*---------------------------------------------------------------------------*/
949/*---------------------------------------------------------------------------*/
950template <class Type> Request HybridParallelDispatch<Type>::
951nonBlockingAllGather(Span<const Type> send_buf, Span<Type> recv_buf)
952{
953 ARCANE_UNUSED(send_buf);
954 ARCANE_UNUSED(recv_buf);
955 throw NotImplementedException(A_FUNCINFO);
956}
957
958/*---------------------------------------------------------------------------*/
959/*---------------------------------------------------------------------------*/
960
961template <class Type> Request HybridParallelDispatch<Type>::
962nonBlockingBroadcast(Span<Type> send_buf, Int32 rank)
963{
964 ARCANE_UNUSED(send_buf);
965 ARCANE_UNUSED(rank);
966 throw NotImplementedException(A_FUNCINFO);
967}
968
969/*---------------------------------------------------------------------------*/
970/*---------------------------------------------------------------------------*/
971
972template <class Type> Request HybridParallelDispatch<Type>::
973nonBlockingGather(Span<const Type> send_buf, Span<Type> recv_buf, Int32 rank)
974{
975 ARCANE_UNUSED(send_buf);
976 ARCANE_UNUSED(recv_buf);
977 ARCANE_UNUSED(rank);
978 throw NotImplementedException(A_FUNCINFO);
979}
980
981/*---------------------------------------------------------------------------*/
982/*---------------------------------------------------------------------------*/
983
984template <class Type> Request HybridParallelDispatch<Type>::
985nonBlockingAllToAll(Span<const Type> send_buf, Span<Type> recv_buf, Int32 count)
986{
987 ARCANE_UNUSED(send_buf);
988 ARCANE_UNUSED(recv_buf);
989 ARCANE_UNUSED(count);
990 throw NotImplementedException(A_FUNCINFO);
991}
992
993/*---------------------------------------------------------------------------*/
994/*---------------------------------------------------------------------------*/
995
996template <class Type> Request HybridParallelDispatch<Type>::
997nonBlockingAllToAllVariable(Span<const Type> send_buf, ConstArrayView<Int32> send_count,
998 ConstArrayView<Int32> send_index, Span<Type> recv_buf,
999 ConstArrayView<Int32> recv_count, ConstArrayView<Int32> recv_index)
1000{
1001 ARCANE_UNUSED(send_buf);
1002 ARCANE_UNUSED(recv_buf);
1003 ARCANE_UNUSED(send_count);
1004 ARCANE_UNUSED(recv_count);
1005 ARCANE_UNUSED(send_index);
1006 ARCANE_UNUSED(recv_index);
1007 throw NotImplementedException(A_FUNCINFO);
1008}
1009
1010/*---------------------------------------------------------------------------*/
1011/*---------------------------------------------------------------------------*/
1012
1013template <class Type> Type HybridParallelDispatch<Type>::
1014scan(eReduceType op, Type send_buf)
1015{
1016 ARCANE_UNUSED(op);
1017 ARCANE_UNUSED(send_buf);
1018 throw NotImplementedException(A_FUNCINFO);
1019}
1020
1021/*---------------------------------------------------------------------------*/
1022/*---------------------------------------------------------------------------*/
1023
1024template <class Type> void HybridParallelDispatch<Type>::
1025scan(eReduceType op, ArrayView<Type> send_buf)
1026{
1027 _allReduceOrScan(op, send_buf, true);
1028}
1029
1030/*---------------------------------------------------------------------------*/
1031/*---------------------------------------------------------------------------*/
1032
1033template <class Type> Request HybridParallelDispatch<Type>::
1035{
1036 throw NotImplementedException(A_FUNCINFO);
1037}
1038
1039/*---------------------------------------------------------------------------*/
1040/*---------------------------------------------------------------------------*/
1041
1042template <class Type> void HybridParallelDispatch<Type>::
1043_collectiveBarrier()
1044{
1045 m_parallel_mng->getThreadBarrier()->wait();
1046}
1047
1048/*---------------------------------------------------------------------------*/
1049/*---------------------------------------------------------------------------*/
1050
1051template class HybridParallelDispatch<char>;
1052template class HybridParallelDispatch<signed char>;
1053template class HybridParallelDispatch<unsigned char>;
1054template class HybridParallelDispatch<short>;
1055template class HybridParallelDispatch<unsigned short>;
1056template class HybridParallelDispatch<int>;
1057template class HybridParallelDispatch<unsigned int>;
1058template class HybridParallelDispatch<long>;
1059template class HybridParallelDispatch<unsigned long>;
1060template class HybridParallelDispatch<long long>;
1061template class HybridParallelDispatch<unsigned long long>;
1062template class HybridParallelDispatch<float>;
1063template class HybridParallelDispatch<double>;
1064template class HybridParallelDispatch<long double>;
1065template class HybridParallelDispatch<Real2>;
1066template class HybridParallelDispatch<Real3>;
1067template class HybridParallelDispatch<Real2x2>;
1068template class HybridParallelDispatch<Real3x3>;
1069template class HybridParallelDispatch<HPReal>;
1070template class HybridParallelDispatch<APReal>;
1071
1072/*---------------------------------------------------------------------------*/
1073/*---------------------------------------------------------------------------*/
1074
1075} // End namespace Arcane::MessagePassing
1076
1077/*---------------------------------------------------------------------------*/
1078/*---------------------------------------------------------------------------*/
#define ARCANE_FATAL(...)
Macro envoyant une exception FatalErrorException.
Vue modifiable d'un tableau d'un type T.
Classe implémentant un nombre réel de haute précision.
Informations pour un message 'gather' pour le type de données DataType.
Interface d'une file de messages avec les threads.
Interface des messages pour le type Type.
Gestionnaire du parallélisme utilisant les threads.
Exception lorsqu'une fonction n'est pas implémentée.
Déclarations des types et méthodes utilisés par les mécanismes d'échange de messages.
eBlockingType
Type indiquant si un message est bloquant ou non.
Int32 Integer
Type représentant un entier.
UniqueArray< Int32 > Int32UniqueArray
Tableau dynamique à une dimension d'entiers 32 bits.
Definition UtilsTypes.h:339
std::int32_t Int32
Type entier signé sur 32 bits.
Type
Type of JSON value.
Definition rapidjson.h:730
Structure équivalente à la valeur booléenne vrai.
Structure équivalente à la valeur booléenne vrai.