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 0 : HcclResult PreSyncQues(const std::vector<InsQuePtr> &syncQueues, const u32 postQueIdx, u32 topicId,
16 : bool enableCounterNotify)
17 : {
18 0 : if (syncQueues.size() <= 1) {
19 0 : HCCL_WARNING("[InsCollAlgFactory] [AlgDataTrans] PreSyncQues: syncQueues size [%zu], do nothing.",
20 : 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 0 : HcclResult PostSyncQues(const std::vector<InsQuePtr> &syncQueues, const u32 waitQueIdx, u32 topicId,
65 : bool enableCounterNotify)
66 : {
67 0 : if (syncQueues.size() <= 1) {
68 0 : HCCL_WARNING("[InsCollAlgFactory] [AlgDataTrans] PreSyncQues: syncQueues size [%zu], do nothing.",
69 : 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 0 : HcclResult TxReduceWithFin(const LinkData &link, InsQuePtr queue, const ReduceSlicesList &slices, u32 topicId,
262 : 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 0 : HcclResult RxReduceWithFin(const LinkData &link, InsQuePtr queue, const ReduceSlicesList &slices, u32 topicId,
280 : 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(const std::vector<LinkData> &links, const std::vector<InsQuePtr> &queues,
295 : const std::vector<SlicesList> &slices, u32 topicId, DmaMode dmaMode)
296 : {
297 0 : CHK_PRT_RET(!DevCapability::GetInstance().IsSupportWriteWithNotify(),
298 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxDataWithFinCounter: inter-rank counterNotify is "
299 : "supported only when the device support WriteWithNotify."),
300 : HcclResult::HCCL_E_INTERNAL);
301 :
302 0 : CHK_PRT_RET(
303 : links.size() != queues.size(),
304 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxDataWithFinCounter: num of links [%zu] given non-equal "
305 : "with num of queues given [%zu].",
306 : links.size(), queues.size()),
307 : HcclResult::HCCL_E_INTERNAL);
308 :
309 0 : CHK_PRT_RET(
310 : links.size() != slices.size(),
311 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxDataWithFinCounter: num of links [%zu] given non-equal "
312 : "with num of slices given [%zu].",
313 : links.size(), slices.size()),
314 : HcclResult::HCCL_E_INTERNAL);
315 :
316 0 : auto linkIter = links.begin();
317 0 : auto queIter = queues.begin();
318 0 : auto sliceListIter = slices.begin();
319 :
320 0 : DmaMode mode;
321 0 : for (; linkIter != links.end(); linkIter++, queIter++, sliceListIter++) {
322 0 : CHK_RET(GetDMAMode(dmaMode, linkIter->GetType(), mode));
323 0 : CHK_PRT_RET(
324 : mode != DmaMode::PUT,
325 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxDataWithFinCounter: inter-rank counterNotify is "
326 : "supported only in PUT MODE."),
327 : HcclResult::HCCL_E_INTERNAL);
328 :
329 0 : CHK_RET(WriteSlicesListsWithFin((*linkIter), (*queIter), TransSlicesInfo((*sliceListIter), true), topicId));
330 : }
331 0 : return HcclResult::HCCL_SUCCESS;
332 : }
333 :
334 0 : HcclResult MultiRxDataWithFinCounter(const std::vector<LinkData> &links, const std::vector<InsQuePtr> &queues,
335 : const std::vector<SlicesList> &slices, u32 topicId, DmaMode dmaMode)
336 : {
337 : (void)slices;
338 0 : CHK_PRT_RET(
339 : !DevCapability::GetInstance().IsSupportWriteWithNotify(),
340 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxDataWithMultiFinCounter: inter-rank counterNotify is "
341 : "supported only when the device support WriteWithNotify."),
342 : HcclResult::HCCL_E_INTERNAL);
343 :
344 0 : std::unique_ptr<InsWaitGroupFin> insWaitGroupFin = std::make_unique<InsWaitGroupFin>(topicId);
345 :
346 0 : DmaMode mode;
347 0 : for (auto linkIter = links.begin(); linkIter != links.end(); linkIter++) {
348 0 : CHK_RET(GetDMAMode(dmaMode, linkIter->GetType(), mode));
349 0 : CHK_PRT_RET(
350 : mode != DmaMode::PUT,
351 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] RxDataWithMultiFinCounter: inter-rank counterNotify is "
352 : "supported only in PUT MODE."),
353 : HcclResult::HCCL_E_INTERNAL);
354 :
355 0 : insWaitGroupFin->Append((*linkIter));
356 : }
357 0 : queues[0]->Append(std::move(insWaitGroupFin));
358 0 : return HcclResult::HCCL_SUCCESS;
359 0 : }
360 :
361 0 : HcclResult MultiTxReduceWithFinCounter(const std::vector<LinkData> &links, const std::vector<InsQuePtr> &queues,
362 : const std::vector<ReduceSlicesList> &slices, u32 topicId, DmaMode dmaMode)
363 : {
364 0 : CHK_PRT_RET(
365 : !DevCapability::GetInstance().IsSupportWriteWithNotify(),
366 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxReduceWithFinCounter: inter-rank counterNotify is "
367 : "supported only when the device support WriteWithNotify."),
368 : HcclResult::HCCL_E_INTERNAL);
369 :
370 0 : CHK_PRT_RET(
371 : links.size() != queues.size(),
372 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxReduceWithFinCounter: num of links [%u] given non-equal "
373 : "with num of queues given [%u].",
374 : links.size(), queues.size()),
375 : HcclResult::HCCL_E_INTERNAL);
376 :
377 0 : CHK_PRT_RET(
378 : links.size() != slices.size(),
379 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxReduceWithFinCounter: num of links [%u] given non-equal "
380 : "with num of slices given [%u].",
381 : links.size(), slices.size()),
382 : HcclResult::HCCL_E_INTERNAL);
383 :
384 0 : auto linkIter = links.begin();
385 0 : auto queIter = queues.begin();
386 0 : auto sliceListIter = slices.begin();
387 :
388 0 : DmaMode mode;
389 0 : for (; linkIter != links.end(); linkIter++, queIter++, sliceListIter++) {
390 0 : CHK_RET(GetDMAMode(dmaMode, linkIter->GetType(), mode));
391 0 : CHK_PRT_RET(
392 : mode != DmaMode::PUT,
393 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxReduceWithFinCounter: inter-rank counterNotify is "
394 : "supported only in PUT MODE."),
395 : HcclResult::HCCL_E_INTERNAL);
396 :
397 0 : CHK_RET(WriteSlicesListsWithFin((*linkIter), (*queIter), TransSlicesInfo((*sliceListIter), true), topicId));
398 : }
399 0 : return HcclResult::HCCL_SUCCESS;
400 : }
401 :
402 0 : HcclResult MultiRxReduceWithFinCounter(const std::vector<LinkData> &links, const std::vector<InsQuePtr> &queues,
403 : const std::vector<ReduceSlicesList> &slices, u32 topicId, DmaMode dmaMode)
404 : {
405 : (void)slices;
406 0 : CHK_PRT_RET(queues.empty(), HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiRxReduceWithFinCounter: queue is empty"), HcclResult::HCCL_E_INTERNAL);
407 0 : CHK_PTR_NULL(queues[0]);
408 0 : CHK_PRT_RET(
409 : !DevCapability::GetInstance().IsSupportWriteWithNotify(),
410 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiRxReduceWithFinCounter: inter-rank counterNotify is "
411 : "supported only when the device support WriteWithNotify."),
412 : HcclResult::HCCL_E_INTERNAL);
413 :
414 0 : std::unique_ptr<InsWaitGroupFin> insWaitGroupFin = std::make_unique<InsWaitGroupFin>(topicId);
415 :
416 0 : DmaMode mode;
417 0 : for (auto linkIter = links.begin(); linkIter != links.end(); linkIter++) {
418 0 : CHK_RET(GetDMAMode(dmaMode, linkIter->GetType(), mode));
419 0 : CHK_PRT_RET(
420 : mode != DmaMode::PUT,
421 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiRxReduceWithFinCounter: inter-rank counterNotify is "
422 : "supported only in PUT MODE."),
423 : HcclResult::HCCL_E_INTERNAL);
424 :
425 0 : insWaitGroupFin->Append((*linkIter));
426 : }
427 0 : queues[0]->Append(std::move(insWaitGroupFin));
428 0 : return HcclResult::HCCL_SUCCESS;
429 0 : }
430 :
431 0 : HcclResult TxRxReady(const TxRxLinks &txRxlinks, InsQuePtr queue, u32 topicId, DmaMode dmaMode)
432 : {
433 : (void)topicId;
434 0 : DmaMode txMode;
435 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), txMode));
436 0 : DmaMode rxMode;
437 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), rxMode));
438 0 : CHK_PRT_RET(txMode != rxMode,
439 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] TxRxReady: DmaMode of txLink inconsistent with rxLink."),
440 : HcclResult::HCCL_E_INTERNAL);
441 :
442 0 : if (txMode == DmaMode::PUT) {
443 0 : queue->Append(std::make_unique<InsPostReady>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_));
444 0 : queue->Append(std::make_unique<InsWaitReady>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_));
445 : } else {
446 0 : queue->Append(std::make_unique<InsPostReady>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_));
447 0 : queue->Append(std::make_unique<InsWaitReady>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_));
448 : }
449 0 : return HcclResult::HCCL_SUCCESS;
450 : }
451 :
452 0 : HcclResult TxRxFin(const TxRxLinks &txRxlinks, InsQuePtr queue, u32 topicId, DmaMode dmaMode)
453 : {
454 : (void)topicId;
455 0 : DmaMode txMode;
456 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), txMode));
457 0 : DmaMode rxMode;
458 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), rxMode));
459 0 : CHK_PRT_RET(txMode != rxMode,
460 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] TxRxFin: DmaMode of txLink inconsistent with rxLink."),
461 : HcclResult::HCCL_E_INTERNAL);
462 :
463 0 : if (txMode == DmaMode::PUT) {
464 0 : queue->Append(std::make_unique<InsPostFin>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_));
465 0 : queue->Append(std::make_unique<InsWaitFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_));
466 : } else {
467 0 : queue->Append(std::make_unique<InsPostFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_));
468 0 : queue->Append(std::make_unique<InsWaitFin>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_));
469 : }
470 :
471 0 : return HcclResult::HCCL_SUCCESS;
472 : }
473 :
474 0 : HcclResult TxRxFinAck(const TxRxLinks &txRxlinks, InsQuePtr queue, u32 topicId, DmaMode dmaMode)
475 : {
476 : (void)topicId;
477 0 : if (!DevCapability::GetInstance().IsSupportStarsPollNetCq()) {
478 0 : bool isTxLinkNet = txRxlinks.txLink_.GetType() == PortDeploymentType::DEV_NET;
479 0 : bool isRxLinkNet = txRxlinks.rxLink_.GetType() == PortDeploymentType::DEV_NET;
480 0 : if (isTxLinkNet && isRxLinkNet) {
481 0 : queue->Append(std::make_unique<InsPostFinAck>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_));
482 0 : queue->Append(std::make_unique<InsWaitFinAck>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_));
483 0 : } else if (isTxLinkNet) {
484 0 : DmaMode mode;
485 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), mode));
486 0 : CHK_PRT_RET(
487 : mode != DmaMode::PUT,
488 : HCCL_ERROR(
489 : "[InsCollAlgFactory] [AlgDataTrans] TxRxFinAck: DmaMode of txLink inconsistent with rxLink."),
490 : HcclResult::HCCL_E_INTERNAL);
491 0 : queue->Append(std::make_unique<InsWaitFinAck>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_));
492 0 : } else if (isRxLinkNet) {
493 0 : DmaMode mode;
494 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), mode));
495 0 : CHK_PRT_RET(
496 : mode != DmaMode::PUT,
497 : HCCL_ERROR(
498 : "[InsCollAlgFactory] [AlgDataTrans] TxRxFinAck: DmaMode of txLink inconsistent with rxLink."),
499 : HcclResult::HCCL_E_INTERNAL);
500 0 : queue->Append(std::make_unique<InsPostFinAck>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_));
501 : }
502 : }
503 :
504 0 : return HcclResult::HCCL_SUCCESS;
505 : }
506 :
507 0 : HcclResult TxRxData(const TxRxLinks &txRxlinks, InsQuePtr queue, const TxRxSlicesList &txRxSlices, DmaMode dmaMode)
508 : {
509 0 : DmaMode txMode;
510 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), txMode));
511 0 : DmaMode rxMode;
512 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), rxMode));
513 0 : CHK_PRT_RET(txMode != rxMode,
514 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] TxRxData: DmaMode of txLink inconsistent with rxLink."),
515 : HcclResult::HCCL_E_INTERNAL);
516 :
517 0 : if (txMode == DmaMode::PUT) {
518 0 : TransSlicesInfo transSlicesInfo = TransSlicesInfo(txRxSlices.txSlicesList_);
519 0 : CHK_RET(TransSlicesLists(txRxlinks.txLink_, queue, transSlicesInfo, DmaMode::PUT));
520 0 : } else {
521 0 : TransSlicesInfo transSlicesInfo = TransSlicesInfo(txRxSlices.rxSlicesList_);
522 0 : CHK_RET(TransSlicesLists(txRxlinks.rxLink_, queue, transSlicesInfo, DmaMode::GET));
523 0 : }
524 :
525 0 : return HcclResult::HCCL_SUCCESS;
526 : }
527 :
528 0 : HcclResult TxRxReduce(const TxRxLinks &txRxlinks, InsQuePtr queue, const TxRxReduceSlicesList &txRxSlices,
529 : DmaMode dmaMode)
530 : {
531 0 : DmaMode txMode;
532 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), txMode));
533 0 : DmaMode rxMode;
534 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), rxMode));
535 0 : CHK_PRT_RET(
536 : txMode != rxMode,
537 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] TxRxReduce: DmaMode of txLink inconsistent with rxLink."),
538 : HcclResult::HCCL_E_INTERNAL);
539 :
540 0 : if (txMode == DmaMode::PUT) {
541 0 : CHK_RET(TransSlicesLists(txRxlinks.txLink_, queue,
542 : TransSlicesInfo(txRxSlices.txSlicesList_, txRxSlices.dataType_, txRxSlices.reduceOp_),
543 : DmaMode::PUT));
544 : } else {
545 0 : CHK_RET(TransSlicesLists(txRxlinks.rxLink_, queue,
546 : TransSlicesInfo(txRxSlices.rxSlicesList_, txRxSlices.dataType_, txRxSlices.reduceOp_),
547 : DmaMode::GET));
548 : }
549 :
550 0 : return HcclResult::HCCL_SUCCESS;
551 : }
552 :
553 0 : HcclResult TxRxDataWithFin(const TxRxLinks &txRxlinks, InsQuePtr queue, const TxRxSlicesList &txRxSlices, u32 topicId,
554 : DmaMode dmaMode)
555 : {
556 0 : CHK_PTR_NULL(queue);
557 0 : DmaMode txMode;
558 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), txMode));
559 0 : DmaMode rxMode;
560 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), rxMode));
561 0 : CHK_PRT_RET(
562 : txMode != rxMode,
563 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] TxRxReduce: DmaMode of txLink inconsistent with rxLink."),
564 : HcclResult::HCCL_E_INTERNAL);
565 0 : if (txMode == DmaMode::PUT) {
566 0 : if (DevCapability::GetInstance().IsSupportWriteWithNotify()) {
567 0 : CHK_RET(WriteSlicesListsWithFin(txRxlinks.txLink_, queue, TransSlicesInfo(txRxSlices.txSlicesList_),
568 : topicId)); // write + postFin
569 :
570 0 : queue->Append(
571 0 : std::make_unique<InsWaitFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_)); // waitFin
572 : } else {
573 0 : CHK_RET(TransSlicesLists(txRxlinks.txLink_, queue, TransSlicesInfo(txRxSlices.txSlicesList_),
574 : DmaMode::PUT)); // write data
575 0 : queue->Append(
576 0 : std::make_unique<InsPostFin>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_)); // postFin
577 0 : queue->Append(
578 0 : std::make_unique<InsWaitFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_)); // waitFin
579 : }
580 : } else {
581 0 : CHK_RET(TransSlicesLists(txRxlinks.rxLink_, queue, TransSlicesInfo(txRxSlices.rxSlicesList_),
582 : DmaMode::GET)); // read data
583 0 : queue->Append(std::make_unique<InsPostFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_)); // postFin
584 0 : queue->Append(std::make_unique<InsWaitFin>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_)); // waitFin
585 : }
586 :
587 0 : return HcclResult::HCCL_SUCCESS;
588 : }
589 :
590 0 : HcclResult TxRxReduceWithFin(const TxRxLinks &txRxlinks, InsQuePtr queue, const TxRxReduceSlicesList &txRxSlices,
591 : u32 topicId, DmaMode dmaMode)
592 : {
593 0 : DmaMode txMode;
594 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.txLink_.GetType(), txMode));
595 0 : DmaMode rxMode;
596 0 : CHK_RET(GetDMAMode(dmaMode, txRxlinks.rxLink_.GetType(), rxMode));
597 0 : CHK_PRT_RET(
598 : txMode != rxMode,
599 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] TxRxReduceWithFin: DmaMode of txLink inconsistent with rxLink."),
600 : HcclResult::HCCL_E_INTERNAL);
601 :
602 0 : if (txMode == DmaMode::PUT) {
603 : TransSlicesInfo transSlicesInfo
604 0 : = TransSlicesInfo(txRxSlices.txSlicesList_, txRxSlices.dataType_, txRxSlices.reduceOp_);
605 :
606 0 : if (DevCapability::GetInstance().IsSupportWriteWithNotify()) {
607 0 : CHK_RET(WriteSlicesListsWithFin(txRxlinks.txLink_, queue, transSlicesInfo, topicId)); // write + postFin
608 :
609 0 : queue->Append(
610 0 : std::make_unique<InsWaitFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_)); // waitFin
611 : } else {
612 0 : CHK_RET(TransSlicesLists(txRxlinks.txLink_, queue, transSlicesInfo, DmaMode::PUT)); // writeReduce data
613 0 : queue->Append(
614 0 : std::make_unique<InsPostFin>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_)); // postFin
615 0 : queue->Append(
616 0 : std::make_unique<InsWaitFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_)); // waitFin
617 : }
618 0 : } else {
619 0 : CHK_RET(TransSlicesLists(txRxlinks.rxLink_, queue,
620 : TransSlicesInfo(txRxSlices.rxSlicesList_, txRxSlices.dataType_, txRxSlices.reduceOp_),
621 : DmaMode::GET)); // readReduce data
622 0 : queue->Append(std::make_unique<InsPostFin>(txRxlinks.rxLink_.GetRemoteRankId(), txRxlinks.rxLink_)); // postFin
623 0 : queue->Append(std::make_unique<InsWaitFin>(txRxlinks.txLink_.GetRemoteRankId(), txRxlinks.txLink_)); // waitFin
624 : }
625 :
626 0 : return HcclResult::HCCL_SUCCESS;
627 : }
628 :
629 0 : HcclResult MultiTxRxDataWithFinCounter(const std::vector<TxRxLinks> &links, const std::vector<InsQuePtr> &queues,
630 : const std::vector<TxRxSlicesList> &slices, u32 topicId, DmaMode dmaMode)
631 : {
632 0 : CHK_PRT_RET(
633 : !DevCapability::GetInstance().IsSupportWriteWithNotify(),
634 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxRxDataWithFinCounter: inter-rank counterNotify is "
635 : "supported only when the device support WriteWithNotify."),
636 : HcclResult::HCCL_E_INTERNAL);
637 :
638 0 : CHK_PRT_RET(
639 : links.size() != queues.size(),
640 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxRxDataWithFinCounter: num of links [%u] given non-equal "
641 : "with num of queues given [%u].",
642 : links.size(), queues.size()),
643 : HcclResult::HCCL_E_INTERNAL);
644 :
645 0 : CHK_PRT_RET(
646 : links.size() != slices.size(),
647 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxRxDataWithFinCounter: num of links [%zu] given non-equal "
648 : "with num of slices given [%zu].",
649 : links.size(), slices.size()),
650 : HcclResult::HCCL_E_INTERNAL);
651 :
652 0 : auto txRxLinkIter = links.begin();
653 0 : auto queIter = queues.begin();
654 0 : auto sliceListIter = slices.begin();
655 :
656 0 : DmaMode txMode;
657 0 : DmaMode rxMode;
658 0 : std::unique_ptr<InsWaitGroupFin> insWaitGroupFin = std::make_unique<InsWaitGroupFin>(topicId);
659 0 : for (; txRxLinkIter != links.end(); txRxLinkIter++, queIter++, sliceListIter++) {
660 0 : CHK_RET(GetDMAMode(dmaMode, (*txRxLinkIter).txLink_.GetType(), txMode));
661 0 : CHK_RET(GetDMAMode(dmaMode, (*txRxLinkIter).rxLink_.GetType(), rxMode));
662 0 : CHK_PRT_RET(
663 : ((txMode != DmaMode::PUT) || (rxMode != DmaMode::PUT)),
664 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxRxDataWithFinCounter: inter-rank counterNotify is "
665 : "supported only in PUT MODE."),
666 : HcclResult::HCCL_E_INTERNAL);
667 :
668 0 : TransSlicesInfo transSlicesInfo = TransSlicesInfo((*sliceListIter).txSlicesList_, true);
669 0 : CHK_RET(WriteSlicesListsWithFin((*txRxLinkIter).txLink_, (*queIter), transSlicesInfo, topicId));
670 :
671 0 : insWaitGroupFin->Append((*txRxLinkIter).rxLink_);
672 0 : }
673 :
674 0 : queues[0]->Append(std::move(insWaitGroupFin));
675 :
676 0 : return HcclResult::HCCL_SUCCESS;
677 0 : }
678 :
679 0 : HcclResult MultiTxRxReduceWithFinCounter(const std::vector<TxRxLinks> &links, const std::vector<InsQuePtr> &queues,
680 : const std::vector<TxRxReduceSlicesList> &slices, u32 topicId, DmaMode dmaMode)
681 : {
682 0 : CHK_PRT_RET(
683 : !DevCapability::GetInstance().IsSupportWriteWithNotify(),
684 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxRxReduceWithFinCounter: inter-rank counterNotify is "
685 : "supported only when the device support WriteReduceWithNotify."),
686 : HcclResult::HCCL_E_INTERNAL);
687 :
688 0 : CHK_PRT_RET(
689 : links.size() != queues.size(),
690 : HCCL_ERROR(
691 : "[InsCollAlgFactory] [AlgDataTrans] MultiTxRxReduceWithFinCounter: num of links [%u] given non-equal "
692 : "with num of queues given [%u].",
693 : links.size(), queues.size()),
694 : HcclResult::HCCL_E_INTERNAL);
695 :
696 0 : CHK_PRT_RET(
697 : links.size() != slices.size(),
698 : HCCL_ERROR(
699 : "[InsCollAlgFactory] [AlgDataTrans] MultiTxRxReduceWithFinCounter: num of links [%u] given non-equal "
700 : "with num of slices given [%u].",
701 : links.size(), slices.size()),
702 : HcclResult::HCCL_E_INTERNAL);
703 :
704 0 : auto txRxLinkIter = links.begin();
705 0 : auto queIter = queues.begin();
706 0 : auto sliceListIter = slices.begin();
707 :
708 0 : DmaMode txMode;
709 0 : DmaMode rxMode;
710 0 : std::unique_ptr<InsWaitGroupFin> insWaitGroupFin = std::make_unique<InsWaitGroupFin>(topicId);
711 0 : for (; txRxLinkIter != links.end(); txRxLinkIter++, queIter++, sliceListIter++) {
712 0 : CHK_RET(GetDMAMode(dmaMode, (*txRxLinkIter).txLink_.GetType(), txMode));
713 0 : CHK_RET(GetDMAMode(dmaMode, (*txRxLinkIter).rxLink_.GetType(), rxMode));
714 0 : CHK_PRT_RET(
715 : ((txMode != DmaMode::PUT) || (rxMode != DmaMode::PUT)),
716 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] MultiTxRxReduceWithFinCounter: inter-rank counterNotify "
717 : "is supported only in PUT MODE."),
718 : HcclResult::HCCL_E_INTERNAL);
719 :
720 : TransSlicesInfo transSlicesInfo
721 0 : = TransSlicesInfo(sliceListIter->txSlicesList_, sliceListIter->dataType_, sliceListIter->reduceOp_, true);
722 0 : CHK_RET(WriteSlicesListsWithFin((*txRxLinkIter).txLink_, (*queIter), transSlicesInfo, topicId));
723 :
724 0 : insWaitGroupFin->Append((*txRxLinkIter).rxLink_);
725 0 : }
726 :
727 0 : queues[0]->Append(std::move(insWaitGroupFin));
728 :
729 0 : return HcclResult::HCCL_SUCCESS;
730 0 : }
731 :
732 0 : HcclResult LocalReduce(InsQuePtr queue, const DataSlice &srcSlice, const DataSlice &dstSlice, const DataType dataType,
733 : const ReduceOp reduceOp)
734 : {
735 0 : CHK_PRT_RET(
736 : srcSlice.GetSize() != dstSlice.GetSize(),
737 : HCCL_ERROR(
738 : "[InsCollAlgFactory] [AlgDataTrans] LocalReduce: src slice size [%zu] is not equal to dst slice size [%zu].",
739 : srcSlice.GetSize(), dstSlice.GetSize()),
740 : HcclResult::HCCL_E_INTERNAL);
741 :
742 : std::unique_ptr<InsLocalReduce> insLocalReduce
743 0 : = std::make_unique<InsLocalReduce>(srcSlice, dstSlice, dataType, reduceOp);
744 0 : queue->Append(std::move(insLocalReduce));
745 :
746 0 : return HcclResult::HCCL_SUCCESS;
747 0 : }
748 :
749 0 : HcclResult LocalReduceSlices(InsQuePtr queue, const std::vector<DataSlice> &srcSlices,
750 : const std::vector<DataSlice> &dstSlices, const DataType dataType, const ReduceOp reduceOp)
751 : {
752 0 : CHK_PRT_RET(srcSlices.size() != dstSlices.size(),
753 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] LocalReduceSlices: num of src slices [%zu], is not equal "
754 : "to num of dst slices [%zu].",
755 : srcSlices.size(), dstSlices.size()),
756 : HcclResult::HCCL_E_INTERNAL);
757 :
758 : // tmpSlices: slices to be transfer in this loop
759 0 : DataSlice tmpSrcSlice = srcSlices[0];
760 0 : DataSlice tmpDstSlice = dstSlices[0];
761 :
762 0 : for (u32 sliceIdx = 0; sliceIdx < srcSlices.size(); sliceIdx++) {
763 0 : CHK_PRT_RET(
764 : srcSlices[sliceIdx].GetSize() != dstSlices[sliceIdx].GetSize(),
765 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] LocalReduceSlices: [%zu]-th slice, src slice size [%zu] "
766 : "is not equal to dst slice size [%zu].",
767 : sliceIdx, srcSlices[sliceIdx].GetSize(), dstSlices[sliceIdx].GetSize()),
768 : HcclResult::HCCL_E_INTERNAL);
769 : try {
770 0 : if (sliceIdx == (srcSlices.size() - 1)) {
771 : // last slice
772 : std::unique_ptr<InsLocalReduce> insLocalReduce
773 0 : = std::make_unique<InsLocalReduce>(tmpSrcSlice, tmpDstSlice, dataType, reduceOp);
774 0 : queue->Append(std::move(insLocalReduce));
775 0 : } else if (IsContinuousSlice(srcSlices[sliceIdx + 1], tmpSrcSlice)
776 0 : && IsContinuousSlice(dstSlices[sliceIdx + 1], tmpDstSlice)) {
777 : // nxtSlice is continuous with tmpSlice, update tmpSlice
778 0 : u64 newTmpSize = tmpSrcSlice.GetSize() + srcSlices[sliceIdx + 1].GetSize();
779 0 : tmpSrcSlice = DataSlice(tmpSrcSlice.GetType(), tmpSrcSlice.GetOffset(), newTmpSize);
780 0 : tmpDstSlice = DataSlice(tmpDstSlice.GetType(), tmpDstSlice.GetOffset(), newTmpSize);
781 : } else {
782 : // nxtSlice is not continuous with tmpSlice, copy tmpSlice, update tmpSlice with nxtSlice
783 : std::unique_ptr<InsLocalReduce> insLocalReduce
784 0 : = std::make_unique<InsLocalReduce>(tmpSrcSlice, tmpDstSlice, dataType, reduceOp);
785 0 : queue->Append(std::move(insLocalReduce));
786 :
787 0 : tmpSrcSlice = srcSlices[sliceIdx + 1];
788 0 : tmpDstSlice = dstSlices[sliceIdx + 1];
789 0 : }
790 0 : } catch (const std::bad_alloc& e) {
791 0 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] LocalReduceSlices: memory allocation failed");
792 0 : return HcclResult::HCCL_E_MEMORY;
793 0 : } catch (const std::exception& e) {
794 0 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] LocalReduceSlices: exception occurred - %s", e.what());
795 0 : return HcclResult::HCCL_E_INTERNAL;
796 0 : }
797 : }
798 :
799 0 : return HcclResult::HCCL_SUCCESS;
800 : }
801 :
802 0 : HcclResult LocalCopy(InsQuePtr queue, const DataSlice &srcSlice, const DataSlice &dstSlice)
803 : {
804 0 : CHK_PRT_RET(
805 : srcSlice.GetSize() != dstSlice.GetSize(),
806 : HCCL_ERROR(
807 : "[InsCollAlgFactory] [AlgDataTrans] LocalCopy: src slice size [%zu] is not equal to dst slice size [%zu].",
808 : srcSlice.GetSize(), dstSlice.GetSize()),
809 : HcclResult::HCCL_E_INTERNAL);
810 :
811 0 : std::unique_ptr<InsLocalCopy> insLocalCopy = std::make_unique<InsLocalCopy>(srcSlice, dstSlice);
812 0 : queue->Append(std::move(insLocalCopy));
813 0 : return HcclResult::HCCL_SUCCESS;
814 0 : }
815 :
816 0 : HcclResult LocalCopySlices(InsQuePtr queue, const std::vector<DataSlice> &srcSlices,
817 : const std::vector<DataSlice> &dstSlices)
818 : {
819 0 : CHK_PRT_RET(srcSlices.size() != dstSlices.size(),
820 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] LocalCopySlices: num of src slices [%u], is not equal "
821 : "to num of dst slices [%u].",
822 : srcSlices.size(), dstSlices.size()),
823 : HcclResult::HCCL_E_INTERNAL);
824 :
825 : // tmpSlices: slices to be transfer in this loop
826 0 : DataSlice tmpSrcSlice = srcSlices[0];
827 0 : DataSlice tmpDstSlice = dstSlices[0];
828 :
829 0 : for (u32 sliceIdx = 0; sliceIdx < srcSlices.size(); sliceIdx++) {
830 0 : CHK_PRT_RET(srcSlices[sliceIdx].GetSize() != dstSlices[sliceIdx].GetSize(),
831 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] LocalCopySlices: [%u]-th slice, src slice size [%zu] "
832 : "is not equal to dst slice size [%zu].",
833 : sliceIdx, srcSlices[sliceIdx].GetSize(), dstSlices[sliceIdx].GetSize()),
834 : HcclResult::HCCL_E_INTERNAL);
835 :
836 0 : if (sliceIdx == (srcSlices.size() - 1)) {
837 : // last slice
838 0 : std::unique_ptr<InsLocalCopy> insLocalCopy = std::make_unique<InsLocalCopy>(tmpSrcSlice, tmpDstSlice);
839 0 : queue->Append(std::move(insLocalCopy));
840 0 : } else if (IsContinuousSlice(srcSlices[sliceIdx + 1], tmpSrcSlice)
841 0 : && IsContinuousSlice(dstSlices[sliceIdx + 1], tmpDstSlice)) {
842 : // nxtSlice is continuous with tmpSlice, update tmpSlice
843 0 : u64 newTmpSize = tmpSrcSlice.GetSize() + srcSlices[sliceIdx + 1].GetSize();
844 0 : tmpSrcSlice = DataSlice(tmpSrcSlice.GetType(), tmpSrcSlice.GetOffset(), newTmpSize);
845 0 : tmpDstSlice = DataSlice(tmpDstSlice.GetType(), tmpDstSlice.GetOffset(), newTmpSize);
846 : } else {
847 : // nxtSlice is not continuous with tmpSlice, copy tmpSlice, update tmpSlice with nxtSlice
848 0 : std::unique_ptr<InsLocalCopy> insLocalCopy = std::make_unique<InsLocalCopy>(tmpSrcSlice, tmpDstSlice);
849 0 : queue->Append(std::move(insLocalCopy));
850 :
851 0 : tmpSrcSlice = srcSlices[sliceIdx + 1];
852 0 : tmpDstSlice = dstSlices[sliceIdx + 1];
853 0 : }
854 : }
855 :
856 0 : return HcclResult::HCCL_SUCCESS;
857 : }
858 :
859 0 : HcclResult StreamSync(std::vector<InsQuePtr> &queues)
860 : {
861 0 : CHK_PRT_RET(queues.empty(), HCCL_ERROR("[alg_data_trans_wrapper_mid][StreamSync] empty queue"),
862 : HcclResult::HCCL_E_INTERNAL);
863 0 : CHK_PTR_NULL(queues[0]);
864 0 : for (auto &queue : queues) {
865 0 : std::unique_ptr<InsPreStreamSync> insPreStreamSync = std::make_unique<InsPreStreamSync>();
866 0 : queue->Append(std::move(insPreStreamSync));
867 0 : }
868 0 : std::unique_ptr<InsStreamSync> insStreamSync = std::make_unique<InsStreamSync>();
869 0 : queues[0]->Append(std::move(insStreamSync));
870 0 : return HcclResult::HCCL_SUCCESS;
871 0 : }
872 :
873 0 : HcclResult AicpuReduce(InsQuePtr queue, const DataSlice &srcSlice, const DataSlice &dstSlice, const DataType dataType,
874 : const ReduceOp reduceOp)
875 : {
876 0 : CHK_PRT_RET(
877 : srcSlice.GetSize() != dstSlice.GetSize(),
878 : HCCL_ERROR(
879 : "[InsCollAlgFactory] [AlgDataTrans] AicpuReduce: src slice size [%zu] is not equal to dst slice size [%zu].",
880 : srcSlice.GetSize(), dstSlice.GetSize()),
881 : HcclResult::HCCL_E_INTERNAL);
882 :
883 : std::unique_ptr<InsAicpuReduce> insAicpuReduce
884 0 : = std::make_unique<InsAicpuReduce>(srcSlice, dstSlice, dataType, reduceOp);
885 0 : queue->Append(std::move(insAicpuReduce));
886 :
887 0 : return HcclResult::HCCL_SUCCESS;
888 0 : }
889 :
890 0 : HcclResult AicpuReduceSlices(InsQuePtr queue, const std::vector<DataSlice> &srcSlices,
891 : const std::vector<DataSlice> &dstSlices, const DataType dataType, const ReduceOp reduceOp)
892 : {
893 0 : CHK_PRT_RET(srcSlices.size() != dstSlices.size(),
894 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] AicpuReduceSlices: num of src slices [%zu], is not equal "
895 : "to num of dst slices [%zu].",
896 : srcSlices.size(), dstSlices.size()),
897 : HcclResult::HCCL_E_INTERNAL);
898 :
899 : // tmpSlices: slices to be transfer in this loop
900 0 : DataSlice tmpSrcSlice = srcSlices[0];
901 0 : DataSlice tmpDstSlice = dstSlices[0];
902 :
903 0 : for (u32 sliceIdx = 0; sliceIdx < srcSlices.size(); sliceIdx++) {
904 0 : CHK_PRT_RET(
905 : srcSlices[sliceIdx].GetSize() != dstSlices[sliceIdx].GetSize(),
906 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] AicpuReduceSlices: [%u]-th slice, src slice size [%zu] "
907 : "is not equal to dst slice size [%zu].",
908 : sliceIdx, srcSlices[sliceIdx].GetSize(), dstSlices[sliceIdx].GetSize()),
909 : HcclResult::HCCL_E_INTERNAL);
910 :
911 0 : if (sliceIdx == (srcSlices.size() - 1)) {
912 : // last slice
913 : std::unique_ptr<InsAicpuReduce> insAicpuReduce
914 0 : = std::make_unique<InsAicpuReduce>(tmpSrcSlice, tmpDstSlice, dataType, reduceOp);
915 0 : queue->Append(std::move(insAicpuReduce));
916 0 : } else if (IsContinuousSlice(srcSlices[sliceIdx + 1], tmpSrcSlice)
917 0 : && IsContinuousSlice(dstSlices[sliceIdx + 1], tmpDstSlice)) {
918 : // nxtSlice is continuous with tmpSlice, update tmpSlice
919 0 : u64 newTmpSize = tmpSrcSlice.GetSize() + srcSlices[sliceIdx + 1].GetSize();
920 0 : tmpSrcSlice = DataSlice(tmpSrcSlice.GetType(), tmpSrcSlice.GetOffset(), newTmpSize);
921 0 : tmpDstSlice = DataSlice(tmpDstSlice.GetType(), tmpDstSlice.GetOffset(), newTmpSize);
922 : } else {
923 : // nxtSlice is not continuous with tmpSlice, copy tmpSlice, update tmpSlice with nxtSlice
924 : std::unique_ptr<InsAicpuReduce> insAicpuReduce
925 0 : = std::make_unique<InsAicpuReduce>(tmpSrcSlice, tmpDstSlice, dataType, reduceOp);
926 0 : queue->Append(std::move(insAicpuReduce));
927 :
928 0 : tmpSrcSlice = srcSlices[sliceIdx + 1];
929 0 : tmpDstSlice = dstSlices[sliceIdx + 1];
930 0 : }
931 : }
932 :
933 0 : return HcclResult::HCCL_SUCCESS;
934 : }
935 :
936 : } // namespace Hccl
|
