Alien  1.3.0
User documentation
Loading...
Searching...
No Matches
SYCLSendRecvOp.h
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#pragma once
8
9#include <vector>
10
11#include <alien/kernels/sycl/SYCLPrecomp.h>
12#include <alien/utils/Precomp.h>
13#include <alien/utils/Trace.h>
14
15#include <arccore/message_passing/Messages.h>
16#include <arccore/message_passing/Request.h>
17
18#include <alien/kernels/simple_csr/SendRecvOp.h>
19
20#include <alien/kernels/sycl/data/SYCLEnv.h>
21#include <alien/kernels/sycl/data/SYCLEnvInternal.h>
22
23namespace Alien::SYCLInternal
24{
25
26#ifndef USE_SYCL2020
27 using namespace cl ;
28#endif
29
30template <typename ValueT>
31class SYCLSendRecvOp : public Alien::SimpleCSRInternal::IASynchOp
32{
33 public:
34 // clang-format off
35 typedef ValueT ValueType ;
36 typedef sycl::buffer<ValueType, 1> ValueBufferType ;
37
38 typedef sycl::buffer<int> IndexBufferType ;
39 typedef std::unique_ptr<IndexBufferType> IndexBufferPtrType ;
40 // clang-format on
41
42 SYCLSendRecvOp(ValueBufferType& send_buffer,
43 const Alien::SimpleCSRInternal::CommInfo& send_info,
44 IndexBufferType& send_ids,
45 Alien::SimpleCSRInternal::CommProperty::ePolicyType send_policy,
46 ValueBufferType& recv_buffer,
47 const Alien::SimpleCSRInternal::CommInfo& recv_info,
48 IndexBufferType& recv_ids,
49 Alien::SimpleCSRInternal::CommProperty::ePolicyType recv_policy,
50 IMessagePassingMng* mng,
51 Arccore::ITraceMng* trace_mng,
52 int block_size = 1)
53 : m_send_buffer(send_buffer)
54 , m_send_info(send_info)
55 , m_send_ids(send_ids)
56 , m_send_policy(send_policy)
57 , m_recv_buffer(recv_buffer)
58 , m_recv_info(recv_info)
59 , m_recv_ids(recv_ids)
60 , m_recv_policy(recv_policy)
61 , m_parallel_mng(mng)
62 , m_trace(trace_mng)
63 , m_block_size(block_size)
64 {}
65
66 virtual ~SYCLSendRecvOp()
67 {
68#ifdef USE_SYCL_USM
69 auto& queue = SYCLEnv::instance()->internal()->queue();
70 sycl::free(m_rbuffer, queue);
71 sycl::free(m_sbuffer, queue);
72#endif
73 }
74
75 void start([[maybe_unused]] bool insitu = false)
76 {
77 //alien_debug([&] {cout() << "SYCLSendRecvOP START "<<m_send_policy<<" insitu="<<insitu<<" "<<m_block_size;});
78 //Universe().traceMng()->flush() ;
79
80 // clang-format off
81 auto env = SYCLEnv::instance();
82 auto& queue = env->internal()->queue();
83 auto total_threads = env->maxNumThreads() ;
84 // clang-format on
85 if (m_recv_policy == Alien::SimpleCSRInternal::CommProperty::ASynch) {
86 m_recv_request.resize(m_recv_info.m_num_neighbours);
87 Integer total_nb_recv_ids = m_recv_info.m_ids_offset[m_recv_info.m_num_neighbours] - m_recv_info.m_ids_offset[0];
88#ifdef USE_SYCL_USM
89 m_rbuffer = sycl::malloc_shared<ValueT>(total_nb_recv_ids * m_block_size, queue);
90#else
91 m_rbuffer.resize(total_nb_recv_ids * m_block_size);
92#endif
93 for (Integer i = 0; i < m_recv_info.m_num_neighbours; ++i) {
94 Integer off = m_recv_info.m_ids_offset[i] - m_recv_info.m_ids_offset[0];
95 Integer size = (m_recv_info.m_ids_offset[i + 1] - off) ;
96#ifdef USE_SYCL_USM
97 ValueT* ptr = m_rbuffer + off*m_block_size;
98#else
99 ValueT* ptr = m_rbuffer.data() + off * m_block_size;
100#endif
101 Integer rank = m_recv_info.m_ranks[i];
102
103 m_recv_request[i] =
104 Arccore::MessagePassing::mpReceive(m_parallel_mng,
105 ArrayView<ValueT>(size*m_block_size, ptr),
106 rank,
107 false);
108 }
109 }
110 if (m_send_policy == Alien::SimpleCSRInternal::CommProperty::ASynch)
111 m_send_request.resize(m_send_info.m_num_neighbours);
112 if (m_send_info.m_ids.size()) {
113 std::size_t total_nb_send_ids = (m_send_info.m_ids_offset[m_send_info.m_num_neighbours] - m_send_info.m_ids_offset[0]);
114#ifdef USE_SYCL_USM
115 m_sbuffer = sycl::malloc_shared<ValueT>(total_nb_send_ids * m_block_size, queue);
116#else
117 m_sbuffer.resize(total_nb_send_ids * m_block_size);
118#endif
119 {
120#ifdef USE_SYCL_USM
121 sycl::buffer<ValueType> sbuffer{ { sycl::buffer_allocation::empty_view(m_sbuffer, queue.get_device()) }, total_nb_send_ids * m_block_size };
122#else
123 sycl::buffer<ValueType> sbuffer(m_sbuffer.data(), total_nb_send_ids * m_block_size);
124#endif
125 const int block_size = m_block_size;
126 // clang-format off
127 if (block_size == 1)
128 {
129 queue.submit(
130 [&](
131 sycl::handler &cgh)
132 {
133 auto access_send_buffer = m_send_buffer.template get_access<sycl::access::mode::read>(cgh);
134 auto access_ids = m_send_ids.template get_access<sycl::access::mode::read>(cgh);
135 auto access_sbuffer = sycl::accessor(sbuffer, cgh, sycl::write_only, sycl::property::no_init
136 {});
137
138 cgh.parallel_for<class vector_send>(sycl::range<1>
139 { total_threads},
140 [=] (sycl::item<1> itemId)
141 {
142 auto id = itemId.get_id(0);
143 for( auto i=id; i<total_nb_send_ids; i+=total_threads)
144 access_sbuffer[i] = access_send_buffer[access_ids[i]];
145 });
146 });
147 }
148 else
149 {
150 queue.submit(
151 [&](
152 sycl::handler &cgh)
153 {
154 auto access_send_buffer = m_send_buffer.template get_access<sycl::access::mode::read>(cgh);
155 auto access_ids = m_send_ids.template get_access<sycl::access::mode::read>(cgh);
156 auto access_sbuffer = sycl::accessor(sbuffer, cgh, sycl::write_only, sycl::property::no_init
157 {});
158
159 cgh.parallel_for<class vectorN_send>(sycl::range<1>
160 { total_threads},
161 [=] (sycl::item<1> itemId)
162 {
163 auto id = itemId.get_id(0);
164 for( auto i=id; i<total_nb_send_ids; i+=total_threads)
165 #pragma unroll
166 for( auto j=0; j<block_size; ++j)
167 access_sbuffer[i*block_size+j] = access_send_buffer[access_ids[i]*block_size+j];
168 });
169 });
170 }
171 // clang-format on
172 }
173
174 }
175 for (Integer i = 0; i < m_send_info.m_num_neighbours; ++i) {
176 Integer off = m_send_info.m_ids_offset[i];
177 Integer nb_send_ids = m_send_info.m_ids_offset[i + 1] - off;
178#ifdef USE_SYCL_USM
179 ValueT const* ptr = m_sbuffer + off * m_block_size;
180#else
181 ValueT const* ptr = m_sbuffer.data() + off * m_block_size;
182#endif
183 Integer rank = m_send_info.m_ranks[i];
184 if (m_send_policy == Alien::SimpleCSRInternal::CommProperty::ASynch)
185 m_send_request[i] = Arccore::MessagePassing::mpSend(m_parallel_mng,
186 ConstArrayView<ValueT>(nb_send_ids*m_block_size, ptr), rank, false);
187 else
188 Arccore::MessagePassing::mpSend(m_parallel_mng,
189 ConstArrayView<ValueT>(nb_send_ids*m_block_size, ptr), rank);
190 }
191
192 //alien_debug([&] {cout()<<"END SYCLSendRecvOP START"<<m_block_size ; });
193 //Universe().traceMng()->flush() ;
194 }
195
196 void end([[maybe_unused]] bool insitu = false)
197 {
198 //alien_debug([&] {cout() << "SYCLSendRecvOP END : "<<m_recv_policy<<" BS="<<m_block_size<<" insitu="<<insitu;});
199 //Universe().traceMng()->flush() ;
200
201 // clang-format off
202 auto env = SYCLEnv::instance();
203 auto& queue = env->internal()->queue();
204 auto total_threads = env->maxNumThreads() ;
205 // clang-format on
206 if (m_recv_policy == Alien::SimpleCSRInternal::CommProperty::ASynch) {
207 Arccore::MessagePassing::mpWaitAll(m_parallel_mng, m_recv_request);
208
209 //Arccore::Integer total_recv_ids = m_recv_info.m_ids_offset[m_recv_info.m_num_neighbours] - m_recv_info.m_ids_offset[0];
210 //for (Integer i = 0; i < total_recv_ids; ++i) {
211 // alien_debug([&] {cout() << "MPI RECV["<<i<<"]="<<m_rbuffer[i];});
212 //}
213 }
214 else {
215 if (m_recv_info.m_ids.size()) {
216 Arccore::Integer total_recv_ids = m_recv_info.m_ids_offset[m_recv_info.m_num_neighbours] - m_recv_info.m_ids_offset[0];
217#ifdef USE_SYCL_USM
218 m_rbuffer = sycl::malloc_shared<ValueT>(total_recv_ids*m_block_size, queue);
219#else
220 m_rbuffer.resize(total_recv_ids*m_block_size);
221#endif
222 }
223 for (Integer i = 0; i < m_recv_info.m_num_neighbours; ++i) {
224 Integer off = m_recv_info.m_ids_offset[i];
225 Integer size = m_recv_info.m_ids_offset[i + 1] - off;
226#ifdef USE_SYCL_USM
227 ValueT* ptr = m_rbuffer + off*m_block_size;
228#else
229 ValueT* ptr = m_rbuffer.data() + off*m_block_size;
230#endif
231 Integer rank = m_recv_info.m_ranks[i];
232 Arccore::MessagePassing::mpReceive(m_parallel_mng, ArrayView<ValueT>(size*m_block_size, ptr), rank);
233 }
234 }
235 if (m_recv_info.m_ids.size()) {
236 std::size_t total_nb_recv_ids = m_recv_info.m_ids_offset[m_recv_info.m_num_neighbours] - m_recv_info.m_ids_offset[0];
237
238 {
239#ifdef USE_SYCL_USM
240 sycl::buffer<ValueType> rbuffer{ { sycl::buffer_allocation::view(m_rbuffer, queue.get_device()) }, total_nb_recv_ids*m_block_size };
241#else
242 sycl::buffer<ValueType> rbuffer(m_rbuffer.data(), total_nb_recv_ids*m_block_size);
243#endif
244 const int block_size = m_block_size; // copie locale : évite la capture de this
245 // clang-format off
246 if(block_size==1)
247 {
248 queue.submit([&](sycl::handler& cgh)
249 {
250 //auto access_recv_buffer = m_recv_buffer.template get_access<sycl::access::mode::read_write>(cgh);
251 //auto access_ids = m_recv_ids.template get_access<sycl::access::mode::read>(cgh);
252 #ifdef USE_HIPSYCL
253 sycl::accessor<ValueType> access_recv_buffer{m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{}};
254 #endif
255 #ifdef USE_ONEAPI
256 auto access_recv_buffer = sycl::accessor(m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{});
257 #endif
258 #ifdef USE_ACPPSYCL
259 auto access_recv_buffer = sycl::accessor(m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{});
260 #endif
261
262
263 sycl::accessor<ValueType> access_rbuffer{rbuffer, cgh};
264
265 cgh.parallel_for<class vector_mult_recv>(sycl::range<1>{total_threads},
266 [=] (sycl::item<1> itemId)
267 {
268 auto id = itemId.get_id(0);
269 for(auto i=id;i<total_nb_recv_ids;i += total_threads)
270 access_recv_buffer[i] = access_rbuffer[i];
271 });
272 });
273 }
274 else
275 {
276 queue.submit([&](sycl::handler& cgh)
277 {
278 //auto access_recv_buffer = m_recv_buffer.template get_access<sycl::access::mode::read_write>(cgh);
279 //auto access_ids = m_recv_ids.template get_access<sycl::access::mode::read>(cgh);
280 #ifdef USE_HIPSYCL
281 sycl::accessor<ValueType> access_recv_buffer{m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{}};
282 #endif
283 #ifdef USE_ONEAPI
284 auto access_recv_buffer = sycl::accessor(m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{});
285 #endif
286 #ifdef USE_ACPPSYCL
287 auto access_recv_buffer = sycl::accessor(m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{});
288 #endif
289
290 sycl::accessor<ValueType> access_rbuffer{rbuffer, cgh};
291
292 cgh.parallel_for<class vectorN_mult_recv>(sycl::range<1>{total_threads},
293 [=] (sycl::item<1> itemId)
294 {
295 auto id = itemId.get_id(0);
296 for(auto i=id;i<total_nb_recv_ids;i += total_threads)
297 #pragma unroll
298 for(auto j=0;j<block_size;++j)
299 access_recv_buffer[i*block_size+j] = access_rbuffer[i*block_size+j];
300 });
301 });
302
303 }
304 // clang-format on
305 }
306 }
307 if (m_send_policy == Alien::SimpleCSRInternal::CommProperty::ASynch) {
308 Arccore::MessagePassing::mpWaitAll(m_parallel_mng, m_send_request);
309 }
310
311 //alien_debug([&] {cout() << "AFTER SYCLSendRecvOP END : "<<m_recv_policy;});
312 //Universe().traceMng()->flush() ;
313 }
314
315 void lowerRecv([[maybe_unused]] bool insitu = true)
316 {
317 // clang-format off
318 auto env = SYCLEnv::instance();
319 auto& queue = env->internal()->queue();
320 auto total_threads = env->maxNumThreads() ;
321 // clang-format on
322
323 if (m_recv_info.m_ids.size())
324 {
325 Arccore::Integer total_recv_ids = m_recv_info.m_ids_offset[m_recv_info.m_num_neighbours] - m_recv_info.m_ids_offset[0];
326#ifdef USE_SYCL_USM
327 m_rbuffer = sycl::malloc_shared<ValueT>(total_recv_ids*m_block_size, queue);
328#else
329 m_rbuffer.resize(total_recv_ids*m_block_size);
330#endif
331 }
332 for (Integer i = 0; i < m_recv_info.m_first_upper_neighb; ++i)
333 {
334 Integer off = m_recv_info.m_ids_offset[i];
335 Integer size = m_recv_info.m_ids_offset[i + 1] - off;
336#ifdef USE_SYCL_USM
337 ValueT* ptr = m_rbuffer + off*m_block_size;
338#else
339 ValueT* ptr = m_rbuffer.data() + off*m_block_size;
340#endif
341 Integer rank = m_recv_info.m_ranks[i];
342 Arccore::MessagePassing::mpReceive(m_parallel_mng, ArrayView<ValueT>(size*m_block_size, ptr), rank);
343 }
344
345 if (m_recv_info.m_ids.size())
346 {
347 std::size_t total_nb_recv_ids = m_recv_info.m_ids_offset[m_recv_info.m_first_upper_neighb] - m_recv_info.m_ids_offset[0];
348
349 {
350#ifdef USE_SYCL_USM
351 sycl::buffer<ValueType> rbuffer{ { sycl::buffer_allocation::view(m_rbuffer, queue.get_device()) }, total_nb_recv_ids*m_block_size };
352#else
353 sycl::buffer<ValueType> rbuffer(m_rbuffer.data(), total_nb_recv_ids*m_block_size);
354#endif
355 const int block_size = m_block_size; // copie locale : évite la capture de this
356 // clang-format off
357 if(block_size==1)
358 {
359 queue.submit([&](sycl::handler& cgh)
360 {
361 //auto access_recv_buffer = m_recv_buffer.template get_access<sycl::access::mode::read_write>(cgh);
362 //auto access_ids = m_recv_ids.template get_access<sycl::access::mode::read>(cgh);
363 #ifdef USE_HIPSYCL
364 sycl::accessor<ValueType> access_recv_buffer{m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{}};
365 #endif
366 #ifdef USE_ONEAPI
367 auto access_recv_buffer = sycl::accessor(m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{});
368 #endif
369 #ifdef USE_ACPPSYCL
370 auto access_recv_buffer = sycl::accessor(m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{});
371 #endif
372
373
374 sycl::accessor<ValueType> access_rbuffer{rbuffer, cgh};
375
376 cgh.parallel_for<class vector_mult_recv>(sycl::range<1>{total_threads},
377 [=] (sycl::item<1> itemId)
378 {
379 auto id = itemId.get_id(0);
380 for(auto i=id;i<total_nb_recv_ids;i += total_threads)
381 access_recv_buffer[i] = access_rbuffer[i];
382 });
383 });
384 }
385 else
386 {
387 queue.submit([&](sycl::handler& cgh)
388 {
389 //auto access_recv_buffer = m_recv_buffer.template get_access<sycl::access::mode::read_write>(cgh);
390 //auto access_ids = m_recv_ids.template get_access<sycl::access::mode::read>(cgh);
391 #ifdef USE_HIPSYCL
392 sycl::accessor<ValueType> access_recv_buffer{m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{}};
393 #endif
394 #ifdef USE_ONEAPI
395 auto access_recv_buffer = sycl::accessor(m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{});
396 #endif
397 #ifdef USE_ACPPSYCL
398 auto access_recv_buffer = sycl::accessor(m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{});
399 #endif
400
401 sycl::accessor<ValueType> access_rbuffer{rbuffer, cgh};
402
403 cgh.parallel_for<class vectorN_mult_recv>(sycl::range<1>{total_threads},
404 [=] (sycl::item<1> itemId)
405 {
406 auto id = itemId.get_id(0);
407 for(auto i=id;i<total_nb_recv_ids;i += total_threads)
408 #pragma unroll
409 for(auto j=0;j<block_size;++j)
410 access_recv_buffer[i*block_size+j] = access_rbuffer[i*block_size+j];
411 });
412 });
413
414 }
415 // clang-format on
416 }
417 }
418 }
419
420 void lowerSend()
421 {
422 // clang-format off
423 auto env = SYCLEnv::instance();
424 auto& queue = env->internal()->queue();
425 auto total_threads = env->maxNumThreads() ;
426 // clang-format on
427 if (m_send_info.m_ids.size())
428 {
429 std::size_t total_nb_send_ids = (m_send_info.m_ids_offset[m_send_info.m_first_upper_neighb] - m_send_info.m_ids_offset[0]);
430#ifdef USE_SYCL_USM
431 m_sbuffer = sycl::malloc_shared<ValueT>(total_nb_send_ids * m_block_size, queue);
432#else
433 m_sbuffer.resize(total_nb_send_ids * m_block_size);
434#endif
435 {
436#ifdef USE_SYCL_USM
437 sycl::buffer<ValueType> sbuffer{ { sycl::buffer_allocation::empty_view(m_sbuffer, queue.get_device()) }, total_nb_send_ids * m_block_size };
438#else
439 sycl::buffer<ValueType> sbuffer(m_sbuffer.data(), total_nb_send_ids * m_block_size);
440#endif
441 const int block_size = m_block_size;
442 // clang-format off
443 if (block_size == 1)
444 {
445 queue.submit(
446 [&](
447 sycl::handler &cgh)
448 {
449 auto access_send_buffer = m_send_buffer.template get_access<sycl::access::mode::read>(cgh);
450 auto access_ids = m_send_ids.template get_access<sycl::access::mode::read>(cgh);
451 auto access_sbuffer = sycl::accessor(sbuffer, cgh, sycl::write_only, sycl::property::no_init
452 {});
453
454 cgh.parallel_for<class vector_send>(sycl::range<1>
455 { total_threads},
456 [=] (sycl::item<1> itemId)
457 {
458 auto id = itemId.get_id(0);
459 for( auto i=id; i<total_nb_send_ids; i+=total_threads)
460 access_sbuffer[i] = access_send_buffer[access_ids[i]];
461 });
462 });
463 }
464 else
465 {
466 queue.submit(
467 [&](
468 sycl::handler &cgh)
469 {
470 auto access_send_buffer = m_send_buffer.template get_access<sycl::access::mode::read>(cgh);
471 auto access_ids = m_send_ids.template get_access<sycl::access::mode::read>(cgh);
472 auto access_sbuffer = sycl::accessor(sbuffer, cgh, sycl::write_only, sycl::property::no_init
473 {});
474
475 cgh.parallel_for<class vectorN_send>(sycl::range<1>
476 { total_threads},
477 [=] (sycl::item<1> itemId)
478 {
479 auto id = itemId.get_id(0);
480 for( auto i=id; i<total_nb_send_ids; i+=total_threads)
481 #pragma unroll
482 for( auto j=0; j<block_size; ++j)
483 access_sbuffer[i*block_size+j] = access_send_buffer[access_ids[i]*block_size+j];
484 });
485 });
486 }
487 // clang-format on
488 }
489
490 }
491 //for (Integer i = 0; i < m_send_info.m_num_neighbours; ++i) {
492 for (int i = m_send_info.m_first_upper_neighb - 1; i > -1; --i) {
493 Integer off = m_send_info.m_ids_offset[i];
494 Integer nb_send_ids = m_send_info.m_ids_offset[i + 1] - off;
495#ifdef USE_SYCL_USM
496 ValueT const* ptr = m_sbuffer + off * m_block_size;
497#else
498 ValueT const* ptr = m_sbuffer.data() + off * m_block_size;
499#endif
500 Integer rank = m_send_info.m_ranks[i];
501 Arccore::MessagePassing::mpSend(m_parallel_mng,
502 ConstArrayView<ValueT>(nb_send_ids*m_block_size, ptr), rank);
503 }
504 }
505
506 void upperRecv([[maybe_unused]] bool insitu = true)
507 {
508 // clang-format off
509 auto env = SYCLEnv::instance();
510 auto& queue = env->internal()->queue();
511 auto total_threads = env->maxNumThreads() ;
512 // clang-format on
513
514 if (m_recv_info.m_ids.size()) {
515 Arccore::Integer total_recv_ids = m_recv_info.m_ids_offset[m_recv_info.m_num_neighbours] - m_recv_info.m_ids_offset[m_recv_info.m_first_upper_neighb];
516#ifdef USE_SYCL_USM
517 m_rbuffer = sycl::malloc_shared<ValueT>(total_recv_ids*m_block_size, queue);
518#else
519 m_rbuffer.resize(total_recv_ids*m_block_size);
520#endif
521 }
522 //for (Integer i = 0; i < m_recv_info.m_num_neighbours; ++i) {
523 for (int i = m_recv_info.m_num_neighbours - 1; i > m_recv_info.m_first_upper_neighb - 1; --i) {
524 Integer off = m_recv_info.m_ids_offset[i];
525 Integer size = m_recv_info.m_ids_offset[i + 1] - off;
526#ifdef USE_SYCL_USM
527 ValueT* ptr = m_rbuffer + off*m_block_size;
528#else
529 ValueT* ptr = m_rbuffer.data() + off*m_block_size;
530#endif
531 Integer rank = m_recv_info.m_ranks[i];
532 Arccore::MessagePassing::mpReceive(m_parallel_mng, ArrayView<ValueT>(size*m_block_size, ptr), rank);
533 }
534 if (m_recv_info.m_ids.size()) {
535 //std::size_t total_nb_recv_ids = m_recv_info.m_ids_offset[m_recv_info.m_num_neighbours] - m_recv_info.m_ids_offset[0];
536 std::size_t total_nb_recv_ids = m_recv_info.m_ids_offset[m_recv_info.m_num_neighbours] -
537 m_recv_info.m_ids_offset[m_recv_info.m_first_upper_neighb];
538
539 {
540#ifdef USE_SYCL_USM
541 sycl::buffer<ValueType> rbuffer{ { sycl::buffer_allocation::view(m_rbuffer, queue.get_device()) }, total_nb_recv_ids*m_block_size };
542#else
543 sycl::buffer<ValueType> rbuffer(m_rbuffer.data(), total_nb_recv_ids*m_block_size);
544#endif
545 const int block_size = m_block_size; // copie locale : évite la capture de this
546 // clang-format off
547 if(block_size==1)
548 {
549 queue.submit([&](sycl::handler& cgh)
550 {
551 //auto access_recv_buffer = m_recv_buffer.template get_access<sycl::access::mode::read_write>(cgh);
552 //auto access_ids = m_recv_ids.template get_access<sycl::access::mode::read>(cgh);
553 #ifdef USE_HIPSYCL
554 sycl::accessor<ValueType> access_recv_buffer{m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{}};
555 #endif
556 #ifdef USE_ONEAPI
557 auto access_recv_buffer = sycl::accessor(m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{});
558 #endif
559 #ifdef USE_ACPPSYCL
560 auto access_recv_buffer = sycl::accessor(m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{});
561 #endif
562
563
564 sycl::accessor<ValueType> access_rbuffer{rbuffer, cgh};
565
566 cgh.parallel_for<class vector_mult_recv>(sycl::range<1>{total_threads},
567 [=] (sycl::item<1> itemId)
568 {
569 auto id = itemId.get_id(0);
570 for(auto i=id;i<total_nb_recv_ids;i += total_threads)
571 access_recv_buffer[i] = access_rbuffer[i];
572 });
573 });
574 }
575 else
576 {
577 queue.submit([&](sycl::handler& cgh)
578 {
579 //auto access_recv_buffer = m_recv_buffer.template get_access<sycl::access::mode::read_write>(cgh);
580 //auto access_ids = m_recv_ids.template get_access<sycl::access::mode::read>(cgh);
581 #ifdef USE_HIPSYCL
582 sycl::accessor<ValueType> access_recv_buffer{m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{}};
583 #endif
584 #ifdef USE_ONEAPI
585 auto access_recv_buffer = sycl::accessor(m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{});
586 #endif
587 #ifdef USE_ACPPSYCL
588 auto access_recv_buffer = sycl::accessor(m_recv_buffer, cgh, sycl::write_only, sycl::property::no_init{});
589 #endif
590
591 sycl::accessor<ValueType> access_rbuffer{rbuffer, cgh};
592
593 cgh.parallel_for<class vectorN_mult_recv>(sycl::range<1>{total_threads},
594 [=] (sycl::item<1> itemId)
595 {
596 auto id = itemId.get_id(0);
597 for(auto i=id;i<total_nb_recv_ids;i += total_threads)
598 #pragma unroll
599 for(auto j=0;j<block_size;++j)
600 access_recv_buffer[i*block_size+j] = access_rbuffer[i*block_size+j];
601 });
602 });
603
604 }
605 // clang-format on
606 }
607 }
608 }
609
610 void upperSend()
611 {
612 auto env = SYCLEnv::instance();
613 auto& queue = env->internal()->queue();
614 auto total_threads = env->maxNumThreads() ;
615 // clang-format on
616 if (m_send_info.m_ids.size())
617 {
618 int offset = m_send_info.m_ids_offset[m_send_info.m_first_upper_neighb] - m_send_info.m_ids_offset[0];
619 std::size_t total_nb_send_ids = m_send_info.m_ids_offset[m_send_info.m_num_neighbours] - m_send_info.m_ids_offset[m_send_info.m_first_upper_neighb];
620#ifdef USE_SYCL_USM
621 m_sbuffer = sycl::malloc_shared<ValueT>(total_nb_send_ids * m_block_size, queue);
622#else
623 m_sbuffer.resize(total_nb_send_ids * m_block_size);
624#endif
625 {
626#ifdef USE_SYCL_USM
627 sycl::buffer<ValueType> sbuffer{ { sycl::buffer_allocation::empty_view(m_sbuffer, queue.get_device()) }, total_nb_send_ids * m_block_size };
628#else
629 sycl::buffer<ValueType> sbuffer(m_sbuffer.data(), total_nb_send_ids * m_block_size);
630#endif
631 const int block_size = m_block_size;
632 // clang-format off
633 if (block_size == 1)
634 {
635 queue.submit(
636 [&](
637 sycl::handler &cgh)
638 {
639 auto access_send_buffer = m_send_buffer.template get_access<sycl::access::mode::read>(cgh);
640 auto access_ids = m_send_ids.template get_access<sycl::access::mode::read>(cgh);
641 auto access_sbuffer = sycl::accessor(sbuffer, cgh, sycl::write_only, sycl::property::no_init
642 {});
643
644 cgh.parallel_for<class vector_send>(sycl::range<1>
645 { total_threads},
646 [=] (sycl::item<1> itemId)
647 {
648 auto id = itemId.get_id(0);
649 for( auto i=id; i<total_nb_send_ids; i+=total_threads)
650 access_sbuffer[i] = access_send_buffer[access_ids[offset+i]];
651 });
652 });
653 }
654 else
655 {
656 queue.submit(
657 [&](
658 sycl::handler &cgh)
659 {
660 auto access_send_buffer = m_send_buffer.template get_access<sycl::access::mode::read>(cgh);
661 auto access_ids = m_send_ids.template get_access<sycl::access::mode::read>(cgh);
662 auto access_sbuffer = sycl::accessor(sbuffer, cgh, sycl::write_only, sycl::property::no_init
663 {});
664
665 cgh.parallel_for<class vectorN_send>(sycl::range<1>
666 { total_threads},
667 [=] (sycl::item<1> itemId)
668 {
669 auto id = itemId.get_id(0);
670 for( auto i=id; i<total_nb_send_ids; i+=total_threads)
671 #pragma unroll
672 for( auto j=0; j<block_size; ++j)
673 access_sbuffer[i*block_size+j] = access_send_buffer[access_ids[offset+i]*block_size+j];
674 });
675 });
676 }
677 // clang-format on
678 }
679 }
680 //for (Integer i = 0; i < m_send_info.m_num_neighbours; ++i)
681 for (int i = m_send_info.m_first_upper_neighb; i < m_send_info.m_num_neighbours; ++i)
682 {
683 Integer off = m_send_info.m_ids_offset[i];
684 Integer nb_send_ids = m_send_info.m_ids_offset[i + 1] - off;
685#ifdef USE_SYCL_USM
686 ValueT const* ptr = m_sbuffer + off * m_block_size;
687#else
688 ValueT const* ptr = m_sbuffer.data() + off * m_block_size;
689#endif
690 Integer rank = m_send_info.m_ranks[i];
691 Arccore::MessagePassing::mpSend(m_parallel_mng,
692 ConstArrayView<ValueT>(nb_send_ids*m_block_size, ptr), rank);
693 }
694 }
695
696 private:
697 // clang-format off
698 ValueBufferType& m_send_buffer;
699 const Alien::SimpleCSRInternal::CommInfo& m_send_info;
700 IndexBufferType& m_send_ids;
701 Alien::SimpleCSRInternal::CommProperty::ePolicyType m_send_policy;
702
703 ValueBufferType& m_recv_buffer;
704 const Alien::SimpleCSRInternal::CommInfo& m_recv_info;
705 IndexBufferType& m_recv_ids;
706 Alien::SimpleCSRInternal::CommProperty::ePolicyType m_recv_policy;
707#ifdef USE_SYCL_USM
708 ValueT* m_rbuffer = nullptr;
709 ValueT* m_sbuffer = nullptr;
710#else
711 std::vector<ValueT> m_rbuffer ;
712 std::vector<ValueT> m_sbuffer ;
713#endif
714 Arccore::Integer m_block_size = 0;
715 Arccore::MessagePassing::IMessagePassingMng* m_parallel_mng = nullptr;
716 Arccore::ITraceMng* m_trace = nullptr;
717 Arccore::UniqueArray<Arccore::MessagePassing::Request> m_recv_request;
718 Arccore::UniqueArray<Arccore::MessagePassing::Request> m_send_request;
719 // clang-format on
720};
721
722} // namespace Alien::SYCLInternal