Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "alg_data_trans_wrapper.h"
12 : #include "log.h"
13 :
14 : namespace Hccl {
15 : HcclResult
16 0 : PreSyncQues(const std::vector<InsQuePtr>& syncQueues, const u32 postQueIdx, u32 topicId, bool enableCounterNotify)
17 : {
18 0 : if (syncQueues.size() <= 1) {
19 0 : HCCL_WARNING(
20 : "[InsCollAlgFactory] [AlgDataTrans] PreSyncQues: syncQueues size [%zu], do nothing.", syncQueues.size());
21 0 : return HcclResult::HCCL_SUCCESS;
22 : }
23 :
24 0 : CHK_PRT_RET(
25 : postQueIdx >= syncQueues.size(),
26 : HCCL_ERROR(
27 : "[InsCollAlgFactory] [AlgDataTrans] PreSyncQues: postQueIdx [%u] out of idx range for syncQueues [%zu].",
28 : postQueIdx, syncQueues.size()),
29 : HcclResult::HCCL_E_INTERNAL);
30 :
31 0 : if (enableCounterNotify) {
32 0 : std::unique_ptr<InsLocalBcastPost> insLocalBcastPost = std::make_unique<InsLocalBcastPost>(topicId);
33 0 : CHK_PTR_NULL(insLocalBcastPost);
34 0 : for (size_t queIdx = 0; queIdx < syncQueues.size(); queIdx++) {
35 0 : if (queIdx != postQueIdx) {
36 0 : insLocalBcastPost->Append(syncQueues[queIdx]->GetId()); // add queIdx to semaphore post
37 : std::unique_ptr<Instruction> insLocalWaitFrom
38 0 : = std::make_unique<InsLocalWaitFrom>(syncQueues[postQueIdx]->GetId(), NotifyType::COUNTER);
39 0 : CHK_PTR_NULL(insLocalWaitFrom);
40 0 : syncQueues[queIdx]->Append(std::move(insLocalWaitFrom)); // semaphore wait
41 0 : }
42 : }
43 0 : syncQueues[postQueIdx]->Append(std::move(insLocalBcastPost)); // semaphore post
44 0 : } else {
45 0 : for (size_t queIdx = 0; queIdx < syncQueues.size(); queIdx++) {
46 0 : if (queIdx != postQueIdx) {
47 : // semaphore post
48 : std::unique_ptr<Instruction> insLocalPostTo
49 0 : = std::make_unique<InsLocalPostTo>(syncQueues[queIdx]->GetId());
50 0 : CHK_PTR_NULL(insLocalPostTo);
51 0 : syncQueues[postQueIdx]->Append(std::move(insLocalPostTo));
52 : // semaphore wait
53 : std::unique_ptr<Instruction> insLocalWaitFrom
54 0 : = std::make_unique<InsLocalWaitFrom>(syncQueues[postQueIdx]->GetId());
55 0 : CHK_PTR_NULL(insLocalWaitFrom);
56 0 : syncQueues[queIdx]->Append(std::move(insLocalWaitFrom));
57 0 : }
58 : }
59 : }
60 :
61 0 : return HcclResult::HCCL_SUCCESS;
62 : }
63 :
64 : HcclResult
65 0 : PostSyncQues(const std::vector<InsQuePtr>& syncQueues, const u32 waitQueIdx, u32 topicId, bool enableCounterNotify)
66 : {
67 0 : if (syncQueues.size() <= 1) {
68 0 : HCCL_WARNING(
69 : "[InsCollAlgFactory] [AlgDataTrans] PreSyncQues: syncQueues size [%zu], do nothing.", syncQueues.size());
70 0 : return HcclResult::HCCL_SUCCESS;
71 : }
72 :
73 0 : CHK_PRT_RET(
74 : waitQueIdx >= syncQueues.size(),
75 : HCCL_ERROR(
76 : "[InsCollAlgFactory] [AlgDataTrans] PostSyncQues: waitQueIdx [%u] out of idx range for syncQueues [%zu].",
77 : waitQueIdx, syncQueues.size()),
78 : HcclResult::HCCL_E_INTERNAL);
79 :
80 0 : if (enableCounterNotify) {
81 0 : std::unique_ptr<InsLocalWaitGroup> insLocalWaitGroup = std::make_unique<InsLocalWaitGroup>(topicId);
82 0 : CHK_PTR_NULL(insLocalWaitGroup);
83 0 : for (size_t queIdx = 0; queIdx < syncQueues.size(); queIdx++) {
84 0 : if (queIdx != waitQueIdx) {
85 0 : insLocalWaitGroup->Append(syncQueues[queIdx]->GetId()); // add queIdx to semaphore wait
86 :
87 : std::unique_ptr<Instruction> insLocalPostTo
88 0 : = std::make_unique<InsLocalPostTo>(syncQueues[waitQueIdx]->GetId(), NotifyType::COUNTER);
89 0 : CHK_PTR_NULL(insLocalPostTo);
90 0 : syncQueues[queIdx]->Append(std::move(insLocalPostTo)); // semaphore post
91 0 : }
92 : }
93 0 : syncQueues[waitQueIdx]->Append(std::move(insLocalWaitGroup)); // semaphore wait
94 0 : } else {
95 0 : for (size_t queIdx = 0; queIdx < syncQueues.size(); queIdx++) {
96 0 : if (queIdx != waitQueIdx) {
97 : // semaphore post
98 : std::unique_ptr<Instruction> insLocalPostTo
99 0 : = std::make_unique<InsLocalPostTo>(syncQueues[waitQueIdx]->GetId());
100 0 : CHK_PTR_NULL(insLocalPostTo);
101 0 : syncQueues[queIdx]->Append(std::move(insLocalPostTo));
102 : // semaphore wait
103 : std::unique_ptr<Instruction> insLocalWaitFrom
104 0 : = std::make_unique<InsLocalWaitFrom>(syncQueues[queIdx]->GetId());
105 0 : CHK_PTR_NULL(insLocalWaitFrom);
106 0 : syncQueues[waitQueIdx]->Append(std::move(insLocalWaitFrom));
107 0 : }
108 : }
109 : }
110 :
111 0 : return HcclResult::HCCL_SUCCESS;
112 : }
113 :
114 0 : HcclResult TxReady(const LinkData& link, InsQuePtr queue, u32 topicId, DmaMode dmaMode)
115 : {
116 : (void)topicId;
117 0 : DmaMode mode;
118 0 : CHK_RET(GetDMAMode(dmaMode, link.GetType(), mode));
119 0 : if (mode == DmaMode::PUT) {
120 0 : queue->Append(std::make_unique<InsWaitReady>(link.GetRemoteRankId(), link));
121 : } else {
122 0 : queue->Append(std::make_unique<InsPostReady>(link.GetRemoteRankId(), link));
123 : }
124 0 : return HcclResult::HCCL_SUCCESS;
125 : }
126 :
127 0 : HcclResult RxReady(const LinkData& link, InsQuePtr queue, u32 topicId, DmaMode dmaMode)
128 : {
129 : (void)topicId;
130 0 : DmaMode mode;
131 0 : CHK_RET(GetDMAMode(dmaMode, link.GetType(), mode));
132 0 : if (mode == DmaMode::PUT) {
133 0 : queue->Append(std::make_unique<InsPostReady>(link.GetRemoteRankId(), link));
134 : } else {
135 0 : queue->Append(std::make_unique<InsWaitReady>(link.GetRemoteRankId(), link));
136 : }
137 0 : return HcclResult::HCCL_SUCCESS;
138 : }
139 :
140 0 : HcclResult TxFin(const LinkData& link, InsQuePtr queue, u32 topicId, DmaMode dmaMode)
141 : {
142 : (void)topicId;
143 0 : DmaMode mode;
144 0 : CHK_RET(GetDMAMode(dmaMode, link.GetType(), mode));
145 0 : if (mode == DmaMode::PUT) {
146 0 : queue->Append(std::make_unique<InsPostFin>(link.GetRemoteRankId(), link));
147 : } else {
148 0 : queue->Append(std::make_unique<InsWaitFin>(link.GetRemoteRankId(), link));
149 : }
150 0 : return HcclResult::HCCL_SUCCESS;
151 : }
152 :
153 0 : HcclResult RxFin(const LinkData& link, InsQuePtr queue, u32 topicId, DmaMode dmaMode)
154 : {
155 : (void)topicId;
156 0 : DmaMode mode;
157 0 : CHK_RET(GetDMAMode(dmaMode, link.GetType(), mode));
158 0 : CHK_PTR_NULL(queue);
159 0 : if (mode == DmaMode::PUT) {
160 0 : queue->Append(std::make_unique<InsWaitFin>(link.GetRemoteRankId(), link));
161 : } else {
162 0 : queue->Append(std::make_unique<InsPostFin>(link.GetRemoteRankId(), link));
163 : }
164 0 : return HcclResult::HCCL_SUCCESS;
165 : }
166 :
167 0 : HcclResult TxFinAck(const LinkData& link, InsQuePtr queue, u32 topicId, DmaMode dmaMode)
168 : {
169 : (void)topicId;
170 : (void)dmaMode;
171 0 : if ((link.GetType() == PortDeploymentType::DEV_NET) && (!DevCapability::GetInstance().IsSupportStarsPollNetCq())) {
172 : // DmaMode of DEV_NET can only be PUT
173 0 : queue->Append(std::make_unique<InsWaitFinAck>(link.GetRemoteRankId(), link));
174 : }
175 0 : return HcclResult::HCCL_SUCCESS;
176 : }
177 :
178 0 : HcclResult RxFinAck(const LinkData& link, InsQuePtr queue, u32 topicId, DmaMode dmaMode)
179 : {
180 : (void)topicId;
181 : (void)dmaMode;
182 0 : if ((link.GetType() == PortDeploymentType::DEV_NET) && (!DevCapability::GetInstance().IsSupportStarsPollNetCq())) {
183 : // DmaMode of DEV_NET can only be PUT
184 0 : queue->Append(std::make_unique<InsPostFinAck>(link.GetRemoteRankId(), link));
185 : }
186 0 : return HcclResult::HCCL_SUCCESS;
187 : }
188 :
189 0 : HcclResult TxData(const LinkData& link, InsQuePtr queue, const SlicesList& slices, DmaMode dmaMode)
190 : {
191 0 : DmaMode mode;
192 0 : CHK_RET(GetDMAMode(dmaMode, link.GetType(), mode));
193 0 : if (mode == DmaMode::PUT) {
194 0 : CHK_RET(TransSlicesLists(link, queue, TransSlicesInfo(slices), DmaMode::PUT));
195 : }
196 0 : return HcclResult::HCCL_SUCCESS;
197 : }
198 :
199 0 : HcclResult RxData(const LinkData& link, InsQuePtr queue, const SlicesList& slices, DmaMode dmaMode)
200 : {
201 0 : DmaMode mode;
202 0 : CHK_RET(GetDMAMode(dmaMode, link.GetType(), mode));
203 0 : if (mode == DmaMode::GET) {
204 0 : CHK_RET(TransSlicesLists(link, queue, TransSlicesInfo(slices), DmaMode::GET));
205 : }
206 0 : return HcclResult::HCCL_SUCCESS;
207 : }
208 :
209 0 : HcclResult TxReduce(const LinkData& link, InsQuePtr queue, const ReduceSlicesList& slices, DmaMode dmaMode)
210 : {
211 0 : DmaMode mode;
212 0 : CHK_RET(GetDMAMode(dmaMode, link.GetType(), mode));
213 0 : if (mode == DmaMode::PUT) {
214 0 : CHK_RET(TransSlicesLists(link, queue, TransSlicesInfo(slices), DmaMode::PUT));
215 : }
216 0 : return HcclResult::HCCL_SUCCESS;
217 : }
218 :
219 0 : HcclResult RxReduce(const LinkData& link, InsQuePtr queue, const ReduceSlicesList& slices, DmaMode dmaMode)
220 : {
221 0 : DmaMode mode;
222 0 : CHK_RET(GetDMAMode(dmaMode, link.GetType(), mode));
223 0 : if (mode == DmaMode::GET) {
224 0 : CHK_RET(TransSlicesLists(link, queue, TransSlicesInfo(slices), DmaMode::GET));
225 : }
226 0 : return HcclResult::HCCL_SUCCESS;
227 : }
228 :
229 0 : HcclResult TxDataWithFin(const LinkData& link, InsQuePtr queue, const SlicesList& slices, u32 topicId, DmaMode dmaMode)
230 : {
231 0 : DmaMode mode;
232 0 : CHK_RET(GetDMAMode(dmaMode, link.GetType(), mode));
233 0 : if (mode == DmaMode::PUT) {
234 0 : if (DevCapability::GetInstance().IsSupportWriteWithNotify()) {
235 0 : CHK_RET(WriteSlicesListsWithFin(link, queue, TransSlicesInfo(slices), topicId));
236 : } else {
237 0 : CHK_RET(TransSlicesLists(link, queue, TransSlicesInfo(slices),
238 : DmaMode::PUT)); // Write Data
239 0 : queue->Append(std::make_unique<InsPostFin>(link.GetRemoteRankId(), link));
240 : }
241 : } else {
242 0 : queue->Append(std::make_unique<InsWaitFin>(link.GetRemoteRankId(), link));
243 : }
244 0 : return HcclResult::HCCL_SUCCESS;
245 : }
246 :
247 0 : HcclResult RxDataWithFin(const LinkData& link, InsQuePtr queue, const SlicesList& slices, u32 topicId, DmaMode dmaMode)
248 : {
249 : (void)topicId;
250 0 : DmaMode mode;
251 0 : CHK_RET(GetDMAMode(dmaMode, link.GetType(), mode));
252 0 : if (mode == DmaMode::PUT) {
253 0 : queue->Append(std::make_unique<InsWaitFin>(link.GetRemoteRankId(), link));
254 : } else {
255 0 : CHK_RET(TransSlicesLists(link, queue, TransSlicesInfo(slices), DmaMode::GET)); // Read Data
256 0 : queue->Append(std::make_unique<InsPostFin>(link.GetRemoteRankId(), link));
257 : }
258 0 : return HcclResult::HCCL_SUCCESS;
259 : }
260 :
261 : HcclResult
262 0 : TxReduceWithFin(const LinkData& link, InsQuePtr queue, const ReduceSlicesList& slices, u32 topicId, DmaMode dmaMode)
263 : {
264 0 : DmaMode mode;
265 0 : CHK_RET(GetDMAMode(dmaMode, link.GetType(), mode));
266 0 : if (mode == DmaMode::PUT) {
267 0 : if (DevCapability::GetInstance().IsSupportWriteWithNotify()) {
268 0 : CHK_RET(WriteSlicesListsWithFin(link, queue, TransSlicesInfo(slices), topicId));
269 : } else {
270 0 : CHK_RET(TransSlicesLists(link, queue, TransSlicesInfo(slices), DmaMode::PUT)); // WriteReduce Data
271 0 : queue->Append(std::make_unique<InsPostFin>(link.GetRemoteRankId(), link));
272 : }
273 : } else {
274 0 : queue->Append(std::make_unique<InsWaitFin>(link.GetRemoteRankId(), link));
275 : }
276 0 : return HcclResult::HCCL_SUCCESS;
277 : }
278 :
279 : HcclResult
280 0 : RxReduceWithFin(const LinkData& link, InsQuePtr queue, const ReduceSlicesList& slices, u32 topicId, DmaMode dmaMode)
281 : {
282 : (void)topicId;
283 0 : DmaMode mode;
284 0 : CHK_RET(GetDMAMode(dmaMode, link.GetType(), mode));
285 0 : if (mode == DmaMode::PUT) {
286 0 : queue->Append(std::make_unique<InsWaitFin>(link.GetRemoteRankId(), link));
287 : } else {
288 0 : CHK_RET(TransSlicesLists(link, queue, TransSlicesInfo(slices), DmaMode::GET)); // ReadReduce Data
289 0 : queue->Append(std::make_unique<InsPostFin>(link.GetRemoteRankId(), link));
290 : }
291 0 : return HcclResult::HCCL_SUCCESS;
292 : }
293 :
294 0 : HcclResult MultiTxDataWithFinCounter(
295 : const std::vector<LinkData>& links, const std::vector<InsQuePtr>& queues, const std::vector<SlicesList>& slices,
296 : u32 topicId, DmaMode dmaMode)
297 : {
298 0 : CHK_PRT_RET(
299 : !DevCapability::GetInstance().IsSupportWriteWithNotify(),
300 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxDataWithFinCounter: inter-rank counterNotify is "
301 : "supported only when the device support WriteWithNotify."),
302 : HcclResult::HCCL_E_INTERNAL);
303 :
304 0 : CHK_PRT_RET(
305 : links.size() != queues.size(),
306 : HCCL_ERROR(
307 : "[InsCollAlgFactory] [AlgDataTrans] MultiTxDataWithFinCounter: num of links [%zu] given non-equal "
308 : "with num of queues given [%zu].",
309 : links.size(), queues.size()),
310 : HcclResult::HCCL_E_INTERNAL);
311 :
312 0 : CHK_PRT_RET(
313 : links.size() != slices.size(),
314 : HCCL_ERROR(
315 : "[InsCollAlgFactory] [AlgDataTrans] MultiTxDataWithFinCounter: num of links [%zu] given non-equal "
316 : "with num of slices given [%zu].",
317 : links.size(), slices.size()),
318 : HcclResult::HCCL_E_INTERNAL);
319 :
320 0 : auto linkIter = links.begin();
321 0 : auto queIter = queues.begin();
322 0 : auto sliceListIter = slices.begin();
323 :
324 0 : DmaMode mode;
325 0 : for (; linkIter != links.end(); linkIter++, queIter++, sliceListIter++) {
326 0 : CHK_RET(GetDMAMode(dmaMode, linkIter->GetType(), mode));
327 0 : CHK_PRT_RET(
328 : mode != DmaMode::PUT,
329 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxDataWithFinCounter: inter-rank counterNotify is "
330 : "supported only in PUT MODE."),
331 : HcclResult::HCCL_E_INTERNAL);
332 :
333 0 : CHK_RET(WriteSlicesListsWithFin((*linkIter), (*queIter), TransSlicesInfo((*sliceListIter), true), topicId));
334 : }
335 0 : return HcclResult::HCCL_SUCCESS;
336 : }
337 :
338 0 : HcclResult MultiRxDataWithFinCounter(
339 : const std::vector<LinkData>& links, const std::vector<InsQuePtr>& queues, const std::vector<SlicesList>& slices,
340 : u32 topicId, DmaMode dmaMode)
341 : {
342 : (void)slices;
343 0 : CHK_PRT_RET(
344 : !DevCapability::GetInstance().IsSupportWriteWithNotify(),
345 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxDataWithMultiFinCounter: inter-rank counterNotify is "
346 : "supported only when the device support WriteWithNotify."),
347 : HcclResult::HCCL_E_INTERNAL);
348 :
349 0 : std::unique_ptr<InsWaitGroupFin> insWaitGroupFin = std::make_unique<InsWaitGroupFin>(topicId);
350 :
351 0 : DmaMode mode;
352 0 : for (auto linkIter = links.begin(); linkIter != links.end(); linkIter++) {
353 0 : CHK_RET(GetDMAMode(dmaMode, linkIter->GetType(), mode));
354 0 : CHK_PRT_RET(
355 : mode != DmaMode::PUT,
356 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] RxDataWithMultiFinCounter: inter-rank counterNotify is "
357 : "supported only in PUT MODE."),
358 : HcclResult::HCCL_E_INTERNAL);
359 :
360 0 : insWaitGroupFin->Append((*linkIter));
361 : }
362 0 : queues[0]->Append(std::move(insWaitGroupFin));
363 0 : return HcclResult::HCCL_SUCCESS;
364 0 : }
365 :
366 0 : HcclResult MultiTxReduceWithFinCounter(
367 : const std::vector<LinkData>& links, const std::vector<InsQuePtr>& queues,
368 : const std::vector<ReduceSlicesList>& slices, u32 topicId, DmaMode dmaMode)
369 : {
370 0 : CHK_PRT_RET(
371 : !DevCapability::GetInstance().IsSupportWriteWithNotify(),
372 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxReduceWithFinCounter: inter-rank counterNotify is "
373 : "supported only when the device support WriteWithNotify."),
374 : HcclResult::HCCL_E_INTERNAL);
375 :
376 0 : CHK_PRT_RET(
377 : links.size() != queues.size(),
378 : HCCL_ERROR(
379 : "[InsCollAlgFactory] [AlgDataTrans] MultiTxReduceWithFinCounter: num of links [%u] given non-equal "
380 : "with num of queues given [%u].",
381 : links.size(), queues.size()),
382 : HcclResult::HCCL_E_INTERNAL);
383 :
384 0 : CHK_PRT_RET(
385 : links.size() != slices.size(),
386 : HCCL_ERROR(
387 : "[InsCollAlgFactory] [AlgDataTrans] MultiTxReduceWithFinCounter: num of links [%u] given non-equal "
388 : "with num of slices given [%u].",
389 : links.size(), slices.size()),
390 : HcclResult::HCCL_E_INTERNAL);
391 :
392 0 : auto linkIter = links.begin();
393 0 : auto queIter = queues.begin();
394 0 : auto sliceListIter = slices.begin();
395 :
396 0 : DmaMode mode;
397 0 : for (; linkIter != links.end(); linkIter++, queIter++, sliceListIter++) {
398 0 : CHK_RET(GetDMAMode(dmaMode, linkIter->GetType(), mode));
399 0 : CHK_PRT_RET(
400 : mode != DmaMode::PUT,
401 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxReduceWithFinCounter: inter-rank counterNotify is "
402 : "supported only in PUT MODE."),
403 : HcclResult::HCCL_E_INTERNAL);
404 :
405 0 : CHK_RET(WriteSlicesListsWithFin((*linkIter), (*queIter), TransSlicesInfo((*sliceListIter), true), topicId));
406 : }
407 0 : return HcclResult::HCCL_SUCCESS;
408 : }
409 :
410 0 : HcclResult MultiRxReduceWithFinCounter(
411 : const std::vector<LinkData>& links, const std::vector<InsQuePtr>& queues,
412 : const std::vector<ReduceSlicesList>& slices, u32 topicId, DmaMode dmaMode)
413 : {
414 : (void)slices;
415 0 : CHK_PRT_RET(
416 : queues.empty(), HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiRxReduceWithFinCounter: queue is empty"),
417 : HcclResult::HCCL_E_INTERNAL);
418 0 : CHK_PTR_NULL(queues[0]);
419 0 : CHK_PRT_RET(
420 : !DevCapability::GetInstance().IsSupportWriteWithNotify(),
421 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiRxReduceWithFinCounter: inter-rank counterNotify is "
422 : "supported only when the device support WriteWithNotify."),
423 : HcclResult::HCCL_E_INTERNAL);
424 :
425 0 : std::unique_ptr<InsWaitGroupFin> insWaitGroupFin = std::make_unique<InsWaitGroupFin>(topicId);
426 :
427 0 : DmaMode mode;
428 0 : for (auto linkIter = links.begin(); linkIter != links.end(); linkIter++) {
429 0 : CHK_RET(GetDMAMode(dmaMode, linkIter->GetType(), mode));
430 0 : CHK_PRT_RET(
431 : mode != DmaMode::PUT,
432 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiRxReduceWithFinCounter: inter-rank counterNotify is "
433 : "supported only in PUT MODE."),
434 : HcclResult::HCCL_E_INTERNAL);
435 :
436 0 : insWaitGroupFin->Append((*linkIter));
437 : }
438 0 : queues[0]->Append(std::move(insWaitGroupFin));
439 0 : return HcclResult::HCCL_SUCCESS;
440 0 : }
441 :
442 0 : HcclResult TxRxReady(const TxRxLinks& txRxlinks, InsQuePtr queue, u32 topicId, DmaMode dmaMode)
443 : {
444 : (void)topicId;
445 0 : DmaMode txMode;
446 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), txMode));
447 0 : DmaMode rxMode;
448 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), rxMode));
449 0 : CHK_PRT_RET(
450 : txMode != rxMode,
451 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] TxRxReady: DmaMode of txLink inconsistent with rxLink."),
452 : HcclResult::HCCL_E_INTERNAL);
453 :
454 0 : if (txMode == DmaMode::PUT) {
455 0 : queue->Append(std::make_unique<InsPostReady>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_));
456 0 : queue->Append(std::make_unique<InsWaitReady>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_));
457 : } else {
458 0 : queue->Append(std::make_unique<InsPostReady>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_));
459 0 : queue->Append(std::make_unique<InsWaitReady>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_));
460 : }
461 0 : return HcclResult::HCCL_SUCCESS;
462 : }
463 :
464 0 : HcclResult TxRxFin(const TxRxLinks& txRxlinks, InsQuePtr queue, u32 topicId, DmaMode dmaMode)
465 : {
466 : (void)topicId;
467 0 : DmaMode txMode;
468 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), txMode));
469 0 : DmaMode rxMode;
470 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), rxMode));
471 0 : CHK_PRT_RET(
472 : txMode != rxMode,
473 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] TxRxFin: DmaMode of txLink inconsistent with rxLink."),
474 : HcclResult::HCCL_E_INTERNAL);
475 :
476 0 : if (txMode == DmaMode::PUT) {
477 0 : queue->Append(std::make_unique<InsPostFin>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_));
478 0 : queue->Append(std::make_unique<InsWaitFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_));
479 : } else {
480 0 : queue->Append(std::make_unique<InsPostFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_));
481 0 : queue->Append(std::make_unique<InsWaitFin>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_));
482 : }
483 :
484 0 : return HcclResult::HCCL_SUCCESS;
485 : }
486 :
487 0 : HcclResult TxRxFinAck(const TxRxLinks& txRxlinks, InsQuePtr queue, u32 topicId, DmaMode dmaMode)
488 : {
489 : (void)topicId;
490 0 : if (!DevCapability::GetInstance().IsSupportStarsPollNetCq()) {
491 0 : bool isTxLinkNet = txRxlinks.txLink_.GetType() == PortDeploymentType::DEV_NET;
492 0 : bool isRxLinkNet = txRxlinks.rxLink_.GetType() == PortDeploymentType::DEV_NET;
493 0 : if (isTxLinkNet && isRxLinkNet) {
494 0 : queue->Append(std::make_unique<InsPostFinAck>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_));
495 0 : queue->Append(std::make_unique<InsWaitFinAck>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_));
496 0 : } else if (isTxLinkNet) {
497 0 : DmaMode mode;
498 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), mode));
499 0 : CHK_PRT_RET(
500 : mode != DmaMode::PUT,
501 : HCCL_ERROR(
502 : "[InsCollAlgFactory] [AlgDataTrans] TxRxFinAck: DmaMode of txLink inconsistent with rxLink."),
503 : HcclResult::HCCL_E_INTERNAL);
504 0 : queue->Append(std::make_unique<InsWaitFinAck>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_));
505 0 : } else if (isRxLinkNet) {
506 0 : DmaMode mode;
507 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), mode));
508 0 : CHK_PRT_RET(
509 : mode != DmaMode::PUT,
510 : HCCL_ERROR(
511 : "[InsCollAlgFactory] [AlgDataTrans] TxRxFinAck: DmaMode of txLink inconsistent with rxLink."),
512 : HcclResult::HCCL_E_INTERNAL);
513 0 : queue->Append(std::make_unique<InsPostFinAck>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_));
514 : }
515 : }
516 :
517 0 : return HcclResult::HCCL_SUCCESS;
518 : }
519 :
520 0 : HcclResult TxRxData(const TxRxLinks& txRxlinks, InsQuePtr queue, const TxRxSlicesList& txRxSlices, DmaMode dmaMode)
521 : {
522 0 : DmaMode txMode;
523 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), txMode));
524 0 : DmaMode rxMode;
525 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), rxMode));
526 0 : CHK_PRT_RET(
527 : txMode != rxMode,
528 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] TxRxData: DmaMode of txLink inconsistent with rxLink."),
529 : HcclResult::HCCL_E_INTERNAL);
530 :
531 0 : if (txMode == DmaMode::PUT) {
532 0 : TransSlicesInfo transSlicesInfo = TransSlicesInfo(txRxSlices.txSlicesList_);
533 0 : CHK_RET(TransSlicesLists(txRxlinks.txLink_, queue, transSlicesInfo, DmaMode::PUT));
534 0 : } else {
535 0 : TransSlicesInfo transSlicesInfo = TransSlicesInfo(txRxSlices.rxSlicesList_);
536 0 : CHK_RET(TransSlicesLists(txRxlinks.rxLink_, queue, transSlicesInfo, DmaMode::GET));
537 0 : }
538 :
539 0 : return HcclResult::HCCL_SUCCESS;
540 : }
541 :
542 : HcclResult
543 0 : TxRxReduce(const TxRxLinks& txRxlinks, InsQuePtr queue, const TxRxReduceSlicesList& txRxSlices, DmaMode dmaMode)
544 : {
545 0 : DmaMode txMode;
546 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), txMode));
547 0 : DmaMode rxMode;
548 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), rxMode));
549 0 : CHK_PRT_RET(
550 : txMode != rxMode,
551 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] TxRxReduce: DmaMode of txLink inconsistent with rxLink."),
552 : HcclResult::HCCL_E_INTERNAL);
553 :
554 0 : if (txMode == DmaMode::PUT) {
555 0 : CHK_RET(TransSlicesLists(
556 : txRxlinks.txLink_, queue,
557 : TransSlicesInfo(txRxSlices.txSlicesList_, txRxSlices.dataType_, txRxSlices.reduceOp_), DmaMode::PUT));
558 : } else {
559 0 : CHK_RET(TransSlicesLists(
560 : txRxlinks.rxLink_, queue,
561 : TransSlicesInfo(txRxSlices.rxSlicesList_, txRxSlices.dataType_, txRxSlices.reduceOp_), DmaMode::GET));
562 : }
563 :
564 0 : return HcclResult::HCCL_SUCCESS;
565 : }
566 :
567 0 : HcclResult TxRxDataWithFin(
568 : const TxRxLinks& txRxlinks, InsQuePtr queue, const TxRxSlicesList& txRxSlices, u32 topicId, DmaMode dmaMode)
569 : {
570 0 : CHK_PTR_NULL(queue);
571 0 : DmaMode txMode;
572 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), txMode));
573 0 : DmaMode rxMode;
574 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), rxMode));
575 0 : CHK_PRT_RET(
576 : txMode != rxMode,
577 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] TxRxReduce: DmaMode of txLink inconsistent with rxLink."),
578 : HcclResult::HCCL_E_INTERNAL);
579 0 : if (txMode == DmaMode::PUT) {
580 0 : if (DevCapability::GetInstance().IsSupportWriteWithNotify()) {
581 0 : CHK_RET(WriteSlicesListsWithFin(
582 : txRxlinks.txLink_, queue, TransSlicesInfo(txRxSlices.txSlicesList_),
583 : topicId)); // write + postFin
584 :
585 0 : queue->Append(
586 0 : std::make_unique<InsWaitFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_)); // waitFin
587 : } else {
588 0 : CHK_RET(TransSlicesLists(
589 : txRxlinks.txLink_, queue, TransSlicesInfo(txRxSlices.txSlicesList_),
590 : DmaMode::PUT)); // write data
591 0 : queue->Append(
592 0 : std::make_unique<InsPostFin>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_)); // postFin
593 0 : queue->Append(
594 0 : std::make_unique<InsWaitFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_)); // waitFin
595 : }
596 : } else {
597 0 : CHK_RET(TransSlicesLists(
598 : txRxlinks.rxLink_, queue, TransSlicesInfo(txRxSlices.rxSlicesList_),
599 : DmaMode::GET)); // read data
600 0 : queue->Append(std::make_unique<InsPostFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_)); // postFin
601 0 : queue->Append(std::make_unique<InsWaitFin>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_)); // waitFin
602 : }
603 :
604 0 : return HcclResult::HCCL_SUCCESS;
605 : }
606 :
607 0 : HcclResult TxRxReduceWithFin(
608 : const TxRxLinks& txRxlinks, InsQuePtr queue, const TxRxReduceSlicesList& txRxSlices, u32 topicId, DmaMode dmaMode)
609 : {
610 0 : DmaMode txMode;
611 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), txMode));
612 0 : DmaMode rxMode;
613 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), rxMode));
614 0 : CHK_PRT_RET(
615 : txMode != rxMode,
616 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] TxRxReduceWithFin: DmaMode of txLink inconsistent with rxLink."),
617 : HcclResult::HCCL_E_INTERNAL);
618 :
619 0 : if (txMode == DmaMode::PUT) {
620 : TransSlicesInfo transSlicesInfo
621 0 : = TransSlicesInfo(txRxSlices.txSlicesList_, txRxSlices.dataType_, txRxSlices.reduceOp_);
622 :
623 0 : if (DevCapability::GetInstance().IsSupportWriteWithNotify()) {
624 0 : CHK_RET(WriteSlicesListsWithFin(txRxlinks.txLink_, queue, transSlicesInfo, topicId)); // write + postFin
625 :
626 0 : queue->Append(
627 0 : std::make_unique<InsWaitFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_)); // waitFin
628 : } else {
629 0 : CHK_RET(TransSlicesLists(txRxlinks.txLink_, queue, transSlicesInfo, DmaMode::PUT)); // writeReduce data
630 0 : queue->Append(
631 0 : std::make_unique<InsPostFin>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_)); // postFin
632 0 : queue->Append(
633 0 : std::make_unique<InsWaitFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_)); // waitFin
634 : }
635 0 : } else {
636 0 : CHK_RET(TransSlicesLists(
637 : txRxlinks.rxLink_, queue,
638 : TransSlicesInfo(txRxSlices.rxSlicesList_, txRxSlices.dataType_, txRxSlices.reduceOp_),
639 : DmaMode::GET)); // readReduce data
640 0 : queue->Append(std::make_unique<InsPostFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_)); // postFin
641 0 : queue->Append(std::make_unique<InsWaitFin>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_)); // waitFin
642 : }
643 :
644 0 : return HcclResult::HCCL_SUCCESS;
645 : }
646 :
647 0 : HcclResult MultiTxRxDataWithFinCounter(
648 : const std::vector<TxRxLinks>& links, const std::vector<InsQuePtr>& queues,
649 : const std::vector<TxRxSlicesList>& slices, u32 topicId, DmaMode dmaMode)
650 : {
651 0 : CHK_PRT_RET(
652 : !DevCapability::GetInstance().IsSupportWriteWithNotify(),
653 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxRxDataWithFinCounter: inter-rank counterNotify is "
654 : "supported only when the device support WriteWithNotify."),
655 : HcclResult::HCCL_E_INTERNAL);
656 :
657 0 : CHK_PRT_RET(
658 : links.size() != queues.size(),
659 : HCCL_ERROR(
660 : "[InsCollAlgFactory] [AlgDataTrans] MultiTxRxDataWithFinCounter: num of links [%u] given non-equal "
661 : "with num of queues given [%u].",
662 : links.size(), queues.size()),
663 : HcclResult::HCCL_E_INTERNAL);
664 :
665 0 : CHK_PRT_RET(
666 : links.size() != slices.size(),
667 : HCCL_ERROR(
668 : "[InsCollAlgFactory] [AlgDataTrans] MultiTxRxDataWithFinCounter: num of links [%zu] given non-equal "
669 : "with num of slices given [%zu].",
670 : links.size(), slices.size()),
671 : HcclResult::HCCL_E_INTERNAL);
672 :
673 0 : auto txRxLinkIter = links.begin();
674 0 : auto queIter = queues.begin();
675 0 : auto sliceListIter = slices.begin();
676 :
677 0 : DmaMode txMode;
678 0 : DmaMode rxMode;
679 0 : std::unique_ptr<InsWaitGroupFin> insWaitGroupFin = std::make_unique<InsWaitGroupFin>(topicId);
680 0 : for (; txRxLinkIter != links.end(); txRxLinkIter++, queIter++, sliceListIter++) {
681 0 : CHK_RET(GetDMAMode(dmaMode, (*txRxLinkIter).txLink_.GetType(), txMode));
682 0 : CHK_RET(GetDMAMode(dmaMode, (*txRxLinkIter).rxLink_.GetType(), rxMode));
683 0 : CHK_PRT_RET(
684 : ((txMode != DmaMode::PUT) || (rxMode != DmaMode::PUT)),
685 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxRxDataWithFinCounter: inter-rank counterNotify is "
686 : "supported only in PUT MODE."),
687 : HcclResult::HCCL_E_INTERNAL);
688 :
689 0 : TransSlicesInfo transSlicesInfo = TransSlicesInfo((*sliceListIter).txSlicesList_, true);
690 0 : CHK_RET(WriteSlicesListsWithFin((*txRxLinkIter).txLink_, (*queIter), transSlicesInfo, topicId));
691 :
692 0 : insWaitGroupFin->Append((*txRxLinkIter).rxLink_);
693 0 : }
694 :
695 0 : queues[0]->Append(std::move(insWaitGroupFin));
696 :
697 0 : return HcclResult::HCCL_SUCCESS;
698 0 : }
699 :
700 0 : HcclResult MultiTxRxReduceWithFinCounter(
701 : const std::vector<TxRxLinks>& links, const std::vector<InsQuePtr>& queues,
702 : const std::vector<TxRxReduceSlicesList>& slices, u32 topicId, DmaMode dmaMode)
703 : {
704 0 : CHK_PRT_RET(
705 : !DevCapability::GetInstance().IsSupportWriteWithNotify(),
706 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxRxReduceWithFinCounter: inter-rank counterNotify is "
707 : "supported only when the device support WriteReduceWithNotify."),
708 : HcclResult::HCCL_E_INTERNAL);
709 :
710 0 : CHK_PRT_RET(
711 : links.size() != queues.size(),
712 : HCCL_ERROR(
713 : "[InsCollAlgFactory] [AlgDataTrans] MultiTxRxReduceWithFinCounter: num of links [%u] given non-equal "
714 : "with num of queues given [%u].",
715 : links.size(), queues.size()),
716 : HcclResult::HCCL_E_INTERNAL);
717 :
718 0 : CHK_PRT_RET(
719 : links.size() != slices.size(),
720 : HCCL_ERROR(
721 : "[InsCollAlgFactory] [AlgDataTrans] MultiTxRxReduceWithFinCounter: num of links [%u] given non-equal "
722 : "with num of slices given [%u].",
723 : links.size(), slices.size()),
724 : HcclResult::HCCL_E_INTERNAL);
725 :
726 0 : auto txRxLinkIter = links.begin();
727 0 : auto queIter = queues.begin();
728 0 : auto sliceListIter = slices.begin();
729 :
730 0 : DmaMode txMode;
731 0 : DmaMode rxMode;
732 0 : std::unique_ptr<InsWaitGroupFin> insWaitGroupFin = std::make_unique<InsWaitGroupFin>(topicId);
733 0 : for (; txRxLinkIter != links.end(); txRxLinkIter++, queIter++, sliceListIter++) {
734 0 : CHK_RET(GetDMAMode(dmaMode, (*txRxLinkIter).txLink_.GetType(), txMode));
735 0 : CHK_RET(GetDMAMode(dmaMode, (*txRxLinkIter).rxLink_.GetType(), rxMode));
736 0 : CHK_PRT_RET(
737 : ((txMode != DmaMode::PUT) || (rxMode != DmaMode::PUT)),
738 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxRxReduceWithFinCounter: inter-rank counterNotify "
739 : "is supported only in PUT MODE."),
740 : HcclResult::HCCL_E_INTERNAL);
741 :
742 : TransSlicesInfo transSlicesInfo
743 0 : = TransSlicesInfo(sliceListIter->txSlicesList_, sliceListIter->dataType_, sliceListIter->reduceOp_, true);
744 0 : CHK_RET(WriteSlicesListsWithFin((*txRxLinkIter).txLink_, (*queIter), transSlicesInfo, topicId));
745 :
746 0 : insWaitGroupFin->Append((*txRxLinkIter).rxLink_);
747 0 : }
748 :
749 0 : queues[0]->Append(std::move(insWaitGroupFin));
750 :
751 0 : return HcclResult::HCCL_SUCCESS;
752 0 : }
753 :
754 0 : HcclResult LocalReduce(
755 : InsQuePtr queue, const DataSlice& srcSlice, const DataSlice& dstSlice, const DataType dataType,
756 : const ReduceOp reduceOp)
757 : {
758 0 : CHK_PRT_RET(
759 : srcSlice.GetSize() != dstSlice.GetSize(),
760 : HCCL_ERROR(
761 : "[InsCollAlgFactory] [AlgDataTrans] LocalReduce: src slice size [%zu] is not equal to dst slice size "
762 : "[%zu].",
763 : srcSlice.GetSize(), dstSlice.GetSize()),
764 : HcclResult::HCCL_E_INTERNAL);
765 :
766 : std::unique_ptr<InsLocalReduce> insLocalReduce
767 0 : = std::make_unique<InsLocalReduce>(srcSlice, dstSlice, dataType, reduceOp);
768 0 : queue->Append(std::move(insLocalReduce));
769 :
770 0 : return HcclResult::HCCL_SUCCESS;
771 0 : }
772 :
773 0 : HcclResult LocalReduceSlices(
774 : InsQuePtr queue, const std::vector<DataSlice>& srcSlices, const std::vector<DataSlice>& dstSlices,
775 : const DataType dataType, const ReduceOp reduceOp)
776 : {
777 0 : CHK_PRT_RET(
778 : srcSlices.size() != dstSlices.size(),
779 : HCCL_ERROR(
780 : "[InsCollAlgFactory] [AlgDataTrans] LocalReduceSlices: num of src slices [%zu], is not equal "
781 : "to num of dst slices [%zu].",
782 : srcSlices.size(), dstSlices.size()),
783 : HcclResult::HCCL_E_INTERNAL);
784 :
785 : // tmpSlices: slices to be transfer in this loop
786 0 : DataSlice tmpSrcSlice = srcSlices[0];
787 0 : DataSlice tmpDstSlice = dstSlices[0];
788 :
789 0 : for (u32 sliceIdx = 0; sliceIdx < srcSlices.size(); sliceIdx++) {
790 0 : CHK_PRT_RET(
791 : srcSlices[sliceIdx].GetSize() != dstSlices[sliceIdx].GetSize(),
792 : HCCL_ERROR(
793 : "[InsCollAlgFactory] [AlgDataTrans] LocalReduceSlices: [%zu]-th slice, src slice size [%zu] "
794 : "is not equal to dst slice size [%zu].",
795 : sliceIdx, srcSlices[sliceIdx].GetSize(), dstSlices[sliceIdx].GetSize()),
796 : HcclResult::HCCL_E_INTERNAL);
797 : try {
798 0 : if (sliceIdx == (srcSlices.size() - 1)) {
799 : // last slice
800 : std::unique_ptr<InsLocalReduce> insLocalReduce
801 0 : = std::make_unique<InsLocalReduce>(tmpSrcSlice, tmpDstSlice, dataType, reduceOp);
802 0 : queue->Append(std::move(insLocalReduce));
803 0 : } else if (
804 0 : IsContinuousSlice(srcSlices[sliceIdx + 1], tmpSrcSlice)
805 0 : && IsContinuousSlice(dstSlices[sliceIdx + 1], tmpDstSlice)) {
806 : // nxtSlice is continuous with tmpSlice, update tmpSlice
807 0 : u64 newTmpSize = tmpSrcSlice.GetSize() + srcSlices[sliceIdx + 1].GetSize();
808 0 : tmpSrcSlice = DataSlice(tmpSrcSlice.GetType(), tmpSrcSlice.GetOffset(), newTmpSize);
809 0 : tmpDstSlice = DataSlice(tmpDstSlice.GetType(), tmpDstSlice.GetOffset(), newTmpSize);
810 : } else {
811 : // nxtSlice is not continuous with tmpSlice, copy tmpSlice, update tmpSlice with nxtSlice
812 : std::unique_ptr<InsLocalReduce> insLocalReduce
813 0 : = std::make_unique<InsLocalReduce>(tmpSrcSlice, tmpDstSlice, dataType, reduceOp);
814 0 : queue->Append(std::move(insLocalReduce));
815 :
816 0 : tmpSrcSlice = srcSlices[sliceIdx + 1];
817 0 : tmpDstSlice = dstSlices[sliceIdx + 1];
818 0 : }
819 0 : } catch (const std::bad_alloc& e) {
820 0 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] LocalReduceSlices: memory allocation failed");
821 0 : return HcclResult::HCCL_E_MEMORY;
822 0 : } catch (const std::exception& e) {
823 0 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] LocalReduceSlices: exception occurred - %s", e.what());
824 0 : return HcclResult::HCCL_E_INTERNAL;
825 0 : }
826 : }
827 :
828 0 : return HcclResult::HCCL_SUCCESS;
829 : }
830 :
831 0 : HcclResult LocalCopy(InsQuePtr queue, const DataSlice& srcSlice, const DataSlice& dstSlice)
832 : {
833 0 : CHK_PRT_RET(
834 : srcSlice.GetSize() != dstSlice.GetSize(),
835 : HCCL_ERROR(
836 : "[InsCollAlgFactory] [AlgDataTrans] LocalCopy: src slice size [%zu] is not equal to dst slice size [%zu].",
837 : srcSlice.GetSize(), dstSlice.GetSize()),
838 : HcclResult::HCCL_E_INTERNAL);
839 :
840 0 : std::unique_ptr<InsLocalCopy> insLocalCopy = std::make_unique<InsLocalCopy>(srcSlice, dstSlice);
841 0 : queue->Append(std::move(insLocalCopy));
842 0 : return HcclResult::HCCL_SUCCESS;
843 0 : }
844 :
845 : HcclResult
846 0 : LocalCopySlices(InsQuePtr queue, const std::vector<DataSlice>& srcSlices, const std::vector<DataSlice>& dstSlices)
847 : {
848 0 : CHK_PRT_RET(
849 : srcSlices.size() != dstSlices.size(),
850 : HCCL_ERROR(
851 : "[InsCollAlgFactory] [AlgDataTrans] LocalCopySlices: num of src slices [%u], is not equal "
852 : "to num of dst slices [%u].",
853 : srcSlices.size(), dstSlices.size()),
854 : HcclResult::HCCL_E_INTERNAL);
855 :
856 : // tmpSlices: slices to be transfer in this loop
857 0 : DataSlice tmpSrcSlice = srcSlices[0];
858 0 : DataSlice tmpDstSlice = dstSlices[0];
859 :
860 0 : for (u32 sliceIdx = 0; sliceIdx < srcSlices.size(); sliceIdx++) {
861 0 : CHK_PRT_RET(
862 : srcSlices[sliceIdx].GetSize() != dstSlices[sliceIdx].GetSize(),
863 : HCCL_ERROR(
864 : "[InsCollAlgFactory] [AlgDataTrans] LocalCopySlices: [%u]-th slice, src slice size [%zu] "
865 : "is not equal to dst slice size [%zu].",
866 : sliceIdx, srcSlices[sliceIdx].GetSize(), dstSlices[sliceIdx].GetSize()),
867 : HcclResult::HCCL_E_INTERNAL);
868 :
869 0 : if (sliceIdx == (srcSlices.size() - 1)) {
870 : // last slice
871 0 : std::unique_ptr<InsLocalCopy> insLocalCopy = std::make_unique<InsLocalCopy>(tmpSrcSlice, tmpDstSlice);
872 0 : queue->Append(std::move(insLocalCopy));
873 0 : } else if (
874 0 : IsContinuousSlice(srcSlices[sliceIdx + 1], tmpSrcSlice)
875 0 : && IsContinuousSlice(dstSlices[sliceIdx + 1], tmpDstSlice)) {
876 : // nxtSlice is continuous with tmpSlice, update tmpSlice
877 0 : u64 newTmpSize = tmpSrcSlice.GetSize() + srcSlices[sliceIdx + 1].GetSize();
878 0 : tmpSrcSlice = DataSlice(tmpSrcSlice.GetType(), tmpSrcSlice.GetOffset(), newTmpSize);
879 0 : tmpDstSlice = DataSlice(tmpDstSlice.GetType(), tmpDstSlice.GetOffset(), newTmpSize);
880 : } else {
881 : // nxtSlice is not continuous with tmpSlice, copy tmpSlice, update tmpSlice with nxtSlice
882 0 : std::unique_ptr<InsLocalCopy> insLocalCopy = std::make_unique<InsLocalCopy>(tmpSrcSlice, tmpDstSlice);
883 0 : queue->Append(std::move(insLocalCopy));
884 :
885 0 : tmpSrcSlice = srcSlices[sliceIdx + 1];
886 0 : tmpDstSlice = dstSlices[sliceIdx + 1];
887 0 : }
888 : }
889 :
890 0 : return HcclResult::HCCL_SUCCESS;
891 : }
892 :
893 0 : HcclResult StreamSync(std::vector<InsQuePtr>& queues)
894 : {
895 0 : CHK_PRT_RET(
896 : queues.empty(), HCCL_ERROR("[alg_data_trans_wrapper_mid][StreamSync] empty queue"),
897 : HcclResult::HCCL_E_INTERNAL);
898 0 : CHK_PTR_NULL(queues[0]);
899 0 : for (auto& queue : queues) {
900 0 : std::unique_ptr<InsPreStreamSync> insPreStreamSync = std::make_unique<InsPreStreamSync>();
901 0 : queue->Append(std::move(insPreStreamSync));
902 0 : }
903 0 : std::unique_ptr<InsStreamSync> insStreamSync = std::make_unique<InsStreamSync>();
904 0 : queues[0]->Append(std::move(insStreamSync));
905 0 : return HcclResult::HCCL_SUCCESS;
906 0 : }
907 :
908 0 : HcclResult AicpuReduce(
909 : InsQuePtr queue, const DataSlice& srcSlice, const DataSlice& dstSlice, const DataType dataType,
910 : const ReduceOp reduceOp)
911 : {
912 0 : CHK_PRT_RET(
913 : srcSlice.GetSize() != dstSlice.GetSize(),
914 : HCCL_ERROR(
915 : "[InsCollAlgFactory] [AlgDataTrans] AicpuReduce: src slice size [%zu] is not equal to dst slice size "
916 : "[%zu].",
917 : srcSlice.GetSize(), dstSlice.GetSize()),
918 : HcclResult::HCCL_E_INTERNAL);
919 :
920 : std::unique_ptr<InsAicpuReduce> insAicpuReduce
921 0 : = std::make_unique<InsAicpuReduce>(srcSlice, dstSlice, dataType, reduceOp);
922 0 : queue->Append(std::move(insAicpuReduce));
923 :
924 0 : return HcclResult::HCCL_SUCCESS;
925 0 : }
926 :
927 0 : HcclResult AicpuReduceSlices(
928 : InsQuePtr queue, const std::vector<DataSlice>& srcSlices, const std::vector<DataSlice>& dstSlices,
929 : const DataType dataType, const ReduceOp reduceOp)
930 : {
931 0 : CHK_PRT_RET(
932 : srcSlices.size() != dstSlices.size(),
933 : HCCL_ERROR(
934 : "[InsCollAlgFactory] [AlgDataTrans] AicpuReduceSlices: num of src slices [%zu], is not equal "
935 : "to num of dst slices [%zu].",
936 : srcSlices.size(), dstSlices.size()),
937 : HcclResult::HCCL_E_INTERNAL);
938 :
939 : // tmpSlices: slices to be transfer in this loop
940 0 : DataSlice tmpSrcSlice = srcSlices[0];
941 0 : DataSlice tmpDstSlice = dstSlices[0];
942 :
943 0 : for (u32 sliceIdx = 0; sliceIdx < srcSlices.size(); sliceIdx++) {
944 0 : CHK_PRT_RET(
945 : srcSlices[sliceIdx].GetSize() != dstSlices[sliceIdx].GetSize(),
946 : HCCL_ERROR(
947 : "[InsCollAlgFactory] [AlgDataTrans] AicpuReduceSlices: [%u]-th slice, src slice size [%zu] "
948 : "is not equal to dst slice size [%zu].",
949 : sliceIdx, srcSlices[sliceIdx].GetSize(), dstSlices[sliceIdx].GetSize()),
950 : HcclResult::HCCL_E_INTERNAL);
951 :
952 0 : if (sliceIdx == (srcSlices.size() - 1)) {
953 : // last slice
954 : std::unique_ptr<InsAicpuReduce> insAicpuReduce
955 0 : = std::make_unique<InsAicpuReduce>(tmpSrcSlice, tmpDstSlice, dataType, reduceOp);
956 0 : queue->Append(std::move(insAicpuReduce));
957 0 : } else if (
958 0 : IsContinuousSlice(srcSlices[sliceIdx + 1], tmpSrcSlice)
959 0 : && IsContinuousSlice(dstSlices[sliceIdx + 1], tmpDstSlice)) {
960 : // nxtSlice is continuous with tmpSlice, update tmpSlice
961 0 : u64 newTmpSize = tmpSrcSlice.GetSize() + srcSlices[sliceIdx + 1].GetSize();
962 0 : tmpSrcSlice = DataSlice(tmpSrcSlice.GetType(), tmpSrcSlice.GetOffset(), newTmpSize);
963 0 : tmpDstSlice = DataSlice(tmpDstSlice.GetType(), tmpDstSlice.GetOffset(), newTmpSize);
964 : } else {
965 : // nxtSlice is not continuous with tmpSlice, copy tmpSlice, update tmpSlice with nxtSlice
966 : std::unique_ptr<InsAicpuReduce> insAicpuReduce
967 0 : = std::make_unique<InsAicpuReduce>(tmpSrcSlice, tmpDstSlice, dataType, reduceOp);
968 0 : queue->Append(std::move(insAicpuReduce));
969 :
970 0 : tmpSrcSlice = srcSlices[sliceIdx + 1];
971 0 : tmpDstSlice = dstSlices[sliceIdx + 1];
972 0 : }
973 : }
974 :
975 0 : return HcclResult::HCCL_SUCCESS;
976 : }
977 :
978 : } // namespace Hccl
|
