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 "alg_common_interface.h"
13 : #include "log.h"
14 :
15 : namespace Hccl {
16 :
17 0 : HcclResult GetDMAMode(const DmaMode setMode, const PortDeploymentType linkPortType, DmaMode& mode)
18 : {
19 0 : if (setMode == DmaMode::GET) {
20 0 : mode = DmaMode::GET;
21 0 : } else if (setMode == DmaMode::PUT) {
22 0 : mode = DmaMode::PUT;
23 : } else {
24 0 : if (linkPortType == PortDeploymentType::P2P) {
25 0 : mode = DmaMode::GET;
26 : } else {
27 0 : mode = DmaMode::PUT;
28 : }
29 : }
30 0 : return HcclResult::HCCL_SUCCESS;
31 : }
32 :
33 0 : bool IsContinuousSlice(const DataSlice& nxtSlice, const DataSlice& currSlice)
34 : {
35 0 : if (nxtSlice.GetType() != currSlice.GetType()) {
36 0 : return false;
37 : }
38 0 : if (nxtSlice.GetOffset() != currSlice.GetOffset() + currSlice.GetSize()) {
39 0 : return false;
40 : }
41 0 : return true;
42 : }
43 :
44 0 : bool isSupportBatchTransfer()
45 : {
46 0 : if (IsAicpuMode()) {
47 0 : return true;
48 : } else {
49 0 : return false;
50 : }
51 : }
52 :
53 0 : void TransSlice(const LinkData& link, InsQuePtr queue, const SlicePair& txRxSlice, DmaMode dmaMode, bool reduceFlag)
54 : {
55 0 : if ((dmaMode == DmaMode::PUT) && (!reduceFlag)) {
56 0 : queue->Append(std::make_unique<InsWrite>(
57 0 : link.GetRemoteRankId(), link, txRxSlice.srcSlice_,
58 0 : txRxSlice.dstSlice_)); // src as local
59 0 : } else if ((dmaMode == DmaMode::PUT) && (reduceFlag)) {
60 0 : queue->Append(std::make_unique<InsWriteReduce>(
61 0 : link.GetRemoteRankId(), link, txRxSlice.srcSlice_, txRxSlice.dstSlice_, txRxSlice.dataType_,
62 0 : txRxSlice.reduceOp_));
63 0 : } else if ((dmaMode == DmaMode::GET) && (!reduceFlag)) {
64 0 : queue->Append(std::make_unique<InsRead>(
65 0 : link.GetRemoteRankId(), link, txRxSlice.dstSlice_,
66 0 : txRxSlice.srcSlice_)); // dst as local
67 : } else {
68 0 : queue->Append(std::make_unique<InsReadReduce>(
69 0 : link.GetRemoteRankId(), link, txRxSlice.dstSlice_, txRxSlice.srcSlice_, txRxSlice.dataType_,
70 0 : txRxSlice.reduceOp_));
71 : }
72 :
73 0 : return;
74 : }
75 :
76 : HcclResult
77 0 : IndividualTransSlicesLists(const LinkData& link, InsQuePtr queue, const TransSlicesInfo& slices, DmaMode dmaMode)
78 : {
79 0 : CHK_PRT_RET(
80 : slices.dstSlices.size() != slices.srcSlices.size(),
81 : HCCL_ERROR(
82 : "[InsCollAlgFactory] [AlgDataTrans] IndividualTransSlicesLists: recv slice num [%zu] is not equal to "
83 : "send slice num [%zu].",
84 : slices.dstSlices.size(), slices.srcSlices.size()),
85 : HcclResult::HCCL_E_INTERNAL);
86 :
87 0 : if (slices.srcSlices.size() == 0) {
88 0 : HCCL_DEBUG("[InsCollAlgFactory] [AlgDataTrans] IndividualTransSlicesLists: empty slices do nothing.");
89 0 : return HcclResult::HCCL_SUCCESS;
90 : }
91 :
92 : // tmpSlices: slices to be transfer in this loop
93 0 : DataSlice tmpSrcSlice = slices.srcSlices[0];
94 0 : DataSlice tmpDstSlice = slices.dstSlices[0];
95 :
96 0 : for (u32 sliceIdx = 0; sliceIdx < slices.srcSlices.size(); sliceIdx++) {
97 0 : CHK_PRT_RET(
98 : slices.srcSlices[sliceIdx].GetSize() != slices.dstSlices[sliceIdx].GetSize(),
99 : HCCL_ERROR(
100 : "[InsCollAlgFactory] [AlgDataTrans] TransSlicesLists: [%u]-th slice, recv slice size [%zu] "
101 : "is not equal to send slice size [%zu].",
102 : sliceIdx, slices.dstSlices[sliceIdx].GetSize(), slices.srcSlices[sliceIdx].GetSize()),
103 : HcclResult::HCCL_E_INTERNAL);
104 :
105 0 : if (sliceIdx == (slices.srcSlices.size() - 1)) {
106 : // last slice, transfer immediately
107 0 : SlicePair txRxSlice = SlicePair(tmpSrcSlice, tmpDstSlice);
108 0 : if (slices.reduceFlag) {
109 0 : txRxSlice.dataType_ = slices.dataType_;
110 0 : txRxSlice.reduceOp_ = slices.reduceOp_;
111 : }
112 0 : TransSlice(link, queue, txRxSlice, dmaMode, slices.reduceFlag);
113 0 : } else if (
114 0 : IsContinuousSlice(slices.srcSlices[sliceIdx + 1], tmpSrcSlice)
115 0 : && IsContinuousSlice(slices.dstSlices[sliceIdx + 1], tmpDstSlice)) {
116 : // nxtSlice is continuous with tmpSlice, updata tmpSlice
117 0 : u64 newTmpSize = tmpSrcSlice.GetSize() + slices.srcSlices[sliceIdx + 1].GetSize();
118 0 : tmpSrcSlice = DataSlice(tmpSrcSlice.GetType(), tmpSrcSlice.GetOffset(), newTmpSize);
119 0 : tmpDstSlice = DataSlice(tmpDstSlice.GetType(), tmpDstSlice.GetOffset(), newTmpSize);
120 : } else {
121 : // nxtSlice is not continuous with tmpSlice, transfer tmpSlice, update tmpSlice with nxtSlice
122 0 : SlicePair txRxSlice = SlicePair(tmpSrcSlice, tmpDstSlice);
123 0 : if (slices.reduceFlag) {
124 0 : txRxSlice.reduceOp_ = slices.reduceOp_;
125 0 : txRxSlice.dataType_ = slices.dataType_;
126 : }
127 0 : TransSlice(link, queue, txRxSlice, dmaMode, slices.reduceFlag);
128 :
129 0 : tmpSrcSlice = slices.srcSlices[sliceIdx + 1];
130 0 : tmpDstSlice = slices.dstSlices[sliceIdx + 1];
131 : }
132 : }
133 0 : return HcclResult::HCCL_SUCCESS;
134 : }
135 :
136 : HcclResult
137 0 : IndividualWriteSlicesListsWithFin(const LinkData& link, InsQuePtr queue, const TransSlicesInfo& slices, u32 topicId)
138 : {
139 0 : CHK_PRT_RET(
140 : slices.srcSlices.size() == 0,
141 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] WriteSlicesListsWithFin: invalid input with empty slices."),
142 : HcclResult::HCCL_E_INTERNAL);
143 :
144 0 : CHK_PRT_RET(
145 : slices.dstSlices.size() != slices.srcSlices.size(),
146 : HCCL_ERROR(
147 : "[InsCollAlgFactory] [AlgDataTrans] WriteSlicesListsWithFin: dst slice num [%zu] is not equal to "
148 : "src slice num [%zu].",
149 : slices.dstSlices.size(), slices.srcSlices.size()),
150 : HcclResult::HCCL_E_INTERNAL);
151 :
152 0 : DataSlice tmpSrcSlice = slices.srcSlices[0];
153 0 : DataSlice tmpDstSlice = slices.dstSlices[0];
154 :
155 0 : for (u32 sliceIdx = 0; sliceIdx < slices.srcSlices.size(); sliceIdx++) {
156 0 : CHK_PRT_RET(
157 : slices.srcSlices[sliceIdx].GetSize() != slices.dstSlices[sliceIdx].GetSize(),
158 : HCCL_ERROR(
159 : "[InsCollAlgFactory] [AlgDataTrans] WriteSlicesListsWithFin: [%u]-th slice, recv slice "
160 : "size [%zu] is not equal to send slice size [%zu].",
161 : sliceIdx, slices.dstSlices[sliceIdx].GetSize(), slices.srcSlices[sliceIdx].GetSize()),
162 : HcclResult::HCCL_E_INTERNAL);
163 :
164 0 : if (sliceIdx == (slices.srcSlices.size() - 1)) {
165 : // last slice, transfer immediately, src as local
166 0 : NotifyType notifyType = slices.enableCounterNotify_ ? NotifyType::COUNTER : NotifyType::NORMAL;
167 0 : if (!slices.reduceFlag) {
168 0 : queue->Append(std::make_unique<InsWriteWithFin>(
169 0 : link.GetRemoteRankId(), link, tmpSrcSlice, tmpDstSlice, notifyType, topicId)); // src as local
170 : } else {
171 0 : queue->Append(std::make_unique<InsWriteReduceWithFin>(
172 0 : link.GetRemoteRankId(), link, tmpSrcSlice, tmpDstSlice, slices.dataType_, slices.reduceOp_,
173 : notifyType, topicId));
174 : }
175 0 : } else if (
176 0 : IsContinuousSlice(slices.srcSlices[sliceIdx + 1], tmpSrcSlice)
177 0 : && IsContinuousSlice(slices.dstSlices[sliceIdx + 1], tmpDstSlice)) {
178 : // nxtSlice is continuous with tmpSlice, updata tmpSlice
179 0 : u64 newTmpSize = tmpSrcSlice.GetSize() + slices.srcSlices[sliceIdx + 1].GetSize();
180 0 : tmpSrcSlice = DataSlice(tmpSrcSlice.GetType(), tmpSrcSlice.GetOffset(), newTmpSize);
181 0 : tmpDstSlice = DataSlice(tmpDstSlice.GetType(), tmpDstSlice.GetOffset(), newTmpSize);
182 : } else {
183 : // nxtSlice is not continuous with tmpSlice, transfer tmpSlice, update tmpSlice with nxtSlice
184 0 : SlicePair txRxSlice = SlicePair(tmpSrcSlice, tmpDstSlice);
185 0 : if (slices.reduceFlag) {
186 0 : txRxSlice.dataType_ = slices.dataType_;
187 0 : txRxSlice.reduceOp_ = slices.reduceOp_;
188 : }
189 :
190 : // not last slice, therefore no Fin sync
191 0 : TransSlice(link, queue, txRxSlice, DmaMode::PUT, slices.reduceFlag);
192 :
193 0 : tmpSrcSlice = slices.srcSlices[sliceIdx + 1];
194 0 : tmpDstSlice = slices.dstSlices[sliceIdx + 1];
195 : }
196 : }
197 :
198 0 : return HcclResult::HCCL_SUCCESS;
199 : }
200 :
201 0 : HcclResult BatchTransSlicesLists(const LinkData& link, InsQuePtr queue, const TransSlicesInfo& slices, DmaMode dmaMode)
202 : {
203 0 : CHK_PRT_RET(
204 : slices.dstSlices.size() != slices.srcSlices.size(),
205 : HCCL_ERROR(
206 : "[InsCollAlgFactory] [AlgDataTrans] BatchTransSlicesLists: recv slice num [%zu] is not equal to "
207 : "send slice num [%zu].",
208 : slices.dstSlices.size(), slices.srcSlices.size()),
209 : HcclResult::HCCL_E_INTERNAL);
210 :
211 0 : if (slices.srcSlices.size() == 0) {
212 0 : HCCL_DEBUG("[InsCollAlgFactory] [AlgDataTrans] BatchTransSlicesLists: empty slices do nothing.");
213 0 : return HcclResult::HCCL_SUCCESS;
214 : }
215 :
216 0 : if (dmaMode == DmaMode::PUT) {
217 0 : unique_ptr<InsBatchWrite> batchInstruction = std::make_unique<InsBatchWrite>(link.GetRemoteRankId(), link);
218 0 : for (u32 sliceIdx = 0; sliceIdx < slices.srcSlices.size(); sliceIdx++) {
219 0 : CHK_PRT_RET(
220 : slices.srcSlices[sliceIdx].GetSize() != slices.dstSlices[sliceIdx].GetSize(),
221 : HCCL_ERROR(
222 : "[InsCollAlgFactory] [AlgDataTrans] BatchTransSlicesLists: [%u]-th slice, recv slice size [%zu] "
223 : "is not equal to send slice size [%zu].",
224 : sliceIdx, slices.dstSlices[sliceIdx].GetSize(), slices.srcSlices[sliceIdx].GetSize()),
225 : HcclResult::HCCL_E_INTERNAL);
226 0 : if (!slices.reduceFlag) {
227 0 : batchInstruction->PushWriteIns(std::make_unique<InsWrite>(
228 0 : link.GetRemoteRankId(), link, slices.srcSlices[sliceIdx], slices.dstSlices[sliceIdx]));
229 : } else {
230 0 : batchInstruction->PushWriteIns(std::make_unique<InsWriteReduce>(
231 0 : link.GetRemoteRankId(), link, slices.srcSlices[sliceIdx], slices.dstSlices[sliceIdx],
232 0 : slices.dataType_, slices.reduceOp_));
233 : }
234 : }
235 0 : queue->Append(std::move(batchInstruction));
236 0 : } else if (dmaMode == DmaMode::GET) {
237 0 : unique_ptr<InsBatchRead> batchInstruction = std::make_unique<InsBatchRead>(link.GetRemoteRankId(), link);
238 0 : for (u32 sliceIdx = 0; sliceIdx < slices.srcSlices.size(); sliceIdx++) {
239 0 : CHK_PRT_RET(
240 : slices.srcSlices[sliceIdx].GetSize() != slices.dstSlices[sliceIdx].GetSize(),
241 : HCCL_ERROR(
242 : "[InsCollAlgFactory] [AlgDataTrans] BatchTransSlicesLists: [%u]-th slice, recv slice size [%zu] "
243 : "is not equal to send slice size [%zu].",
244 : sliceIdx, slices.dstSlices[sliceIdx].GetSize(), slices.srcSlices[sliceIdx].GetSize()),
245 : HcclResult::HCCL_E_INTERNAL);
246 0 : if (!slices.reduceFlag) {
247 0 : batchInstruction->PushReadIns(std::make_unique<InsRead>(
248 0 : link.GetRemoteRankId(), link, slices.dstSlices[sliceIdx], slices.srcSlices[sliceIdx]));
249 : } else {
250 0 : batchInstruction->PushReadIns(std::make_unique<InsReadReduce>(
251 0 : link.GetRemoteRankId(), link, slices.dstSlices[sliceIdx], slices.srcSlices[sliceIdx],
252 0 : slices.dataType_, slices.reduceOp_));
253 : }
254 : }
255 0 : queue->Append(std::move(batchInstruction));
256 0 : } else {
257 0 : HCCL_ERROR(
258 : "[InsCollAlgFactory] [AlgDataTrans] BatchTransSlicesLists: dmaMode [%s] is not supported.",
259 : dmaMode.Describe().c_str());
260 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
261 : }
262 0 : return HcclResult::HCCL_SUCCESS;
263 : }
264 :
265 0 : HcclResult TransSlicesLists(const LinkData& link, InsQuePtr queue, const TransSlicesInfo& slices, DmaMode dmaMode)
266 : {
267 0 : if (isSupportBatchTransfer()) {
268 0 : CHK_RET(BatchTransSlicesLists(link, queue, slices, dmaMode));
269 : } else {
270 0 : CHK_RET(IndividualTransSlicesLists(link, queue, slices, dmaMode));
271 : }
272 0 : return HcclResult::HCCL_SUCCESS;
273 : }
274 :
275 0 : HcclResult WriteSlicesListsWithFin(const LinkData& link, InsQuePtr queue, const TransSlicesInfo& slices, u32 topicId)
276 : {
277 0 : if (isSupportBatchTransfer() && slices.srcSlices.size() > 1) {
278 0 : CHK_RET(BatchTransSlicesLists(link, queue, slices, DmaMode::PUT));
279 0 : queue->Append(std::make_unique<InsPostFin>(link.GetRemoteRankId(), link));
280 : } else {
281 0 : CHK_RET(IndividualWriteSlicesListsWithFin(link, queue, slices, topicId));
282 : }
283 0 : return HcclResult::HCCL_SUCCESS;
284 : }
285 :
286 0 : HcclResult ProceedMultiLinks(
287 : const std::vector<DataInfo>& dataInfo, const std::vector<InsQuePtr>& queues,
288 : const MultiDataLinksDmaModeInfo& dmaModeInfo, std::vector<InsQuePtr>& syncQues, bool& hasDiffDmaMode)
289 : {
290 0 : auto dataInfoIter = dataInfo.begin();
291 0 : auto queIter = queues.begin();
292 :
293 0 : RankId remoteRank = dataInfoIter->link_.GetRemoteRankId();
294 0 : PortDeploymentType linkType = dataInfoIter->link_.GetType();
295 0 : DmaMode mode;
296 0 : CHK_RET(GetDMAMode(dmaModeInfo.modeSet_, linkType, mode));
297 0 : dataInfoIter++;
298 0 : queIter++;
299 :
300 0 : DmaMode tmpMode;
301 0 : u32 netLinkNum = 0;
302 0 : for (; dataInfoIter != dataInfo.end(); dataInfoIter++, queIter++) {
303 0 : CHK_PRT_RET(
304 : dataInfoIter->link_.GetRemoteRankId() != remoteRank,
305 : HCCL_ERROR(
306 : "[InsCollAlgFactory] [AlgDataTrans] Send/RecvThruMultiLinks: only support identical "
307 : "remote rank, now we have got rank [%d] and rank [%d].",
308 : remoteRank, dataInfoIter->link_.GetRemoteRankId()),
309 : HcclResult::HCCL_E_INTERNAL);
310 :
311 0 : PortDeploymentType tmpLinkType = (dataInfoIter->link_).GetType();
312 0 : if (tmpLinkType == PortDeploymentType::DEV_NET) {
313 0 : CHK_PRT_RET(
314 : linkType == PortDeploymentType::P2P,
315 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] Send/RecvThruMultiLinks: portType of first link "
316 : "should be DEV_NET when there exists NET links."),
317 : HcclResult::HCCL_E_INTERNAL);
318 0 : netLinkNum++;
319 : }
320 :
321 0 : CHK_RET(GetDMAMode(dmaModeInfo.modeSet_, (dataInfoIter->link_).GetType(), tmpMode));
322 :
323 0 : if (tmpMode != mode) {
324 0 : hasDiffDmaMode = true;
325 : }
326 :
327 0 : if (tmpMode == dmaModeInfo.modeNeedSync_) {
328 0 : syncQues.push_back(*queIter); // que sync is required only when mode is PUT for send and GET for recv
329 : }
330 : }
331 :
332 0 : CHK_PRT_RET(
333 : netLinkNum > 1,
334 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] Send/RecvThruMultiLinks: more than one net links, use "
335 : "mid-level wrapper instead as DEV_NET is async."),
336 : HcclResult::HCCL_E_INTERNAL);
337 :
338 0 : return HcclResult::HCCL_SUCCESS;
339 : }
340 :
341 0 : HcclResult ProceedMultiLinks(
342 : const std::vector<DataReduceInfo>& dataInfo, const std::vector<InsQuePtr>& queues,
343 : const MultiDataLinksDmaModeInfo& dmaModeInfo, std::vector<InsQuePtr>& syncQues, bool& hasDiffDmaMode)
344 : {
345 0 : RankId remoteRank = dataInfo[0].link_.GetRemoteRankId();
346 0 : PortDeploymentType linkType = dataInfo[0].link_.GetType();
347 0 : DmaMode mode;
348 0 : CHK_RET(GetDMAMode(dmaModeInfo.modeSet_, linkType, mode));
349 :
350 0 : auto dataInfoIter = dataInfo.begin();
351 0 : auto queIter = queues.begin();
352 0 : dataInfoIter++;
353 0 : queIter++;
354 :
355 0 : DmaMode tmpMode;
356 0 : u32 netLinkNum = 0;
357 0 : for (; dataInfoIter != dataInfo.end(); dataInfoIter++, queIter++) {
358 0 : CHK_PRT_RET(
359 : dataInfoIter->link_.GetRemoteRankId() != remoteRank,
360 : HCCL_ERROR(
361 : "[InsCollAlgFactory] [AlgDataTrans] Send/RecvReduceThruMultiLinks: only support "
362 : "identical remote rank, now we have got rank [%d] and rank [%d].",
363 : remoteRank, dataInfoIter->link_.GetRemoteRankId()),
364 : HcclResult::HCCL_E_INTERNAL);
365 :
366 0 : PortDeploymentType tmpLinkType = (dataInfoIter->link_).GetType();
367 0 : if (tmpLinkType == PortDeploymentType::DEV_NET) {
368 0 : CHK_PRT_RET(
369 : linkType == PortDeploymentType::P2P,
370 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] Send/RecvReduceThruMultiLinks: portType of first link "
371 : "should be DEV_NET when there exists NET links."),
372 : HcclResult::HCCL_E_INTERNAL);
373 0 : netLinkNum++;
374 : }
375 :
376 0 : CHK_RET(GetDMAMode(dmaModeInfo.modeSet_, (dataInfoIter->link_).GetType(), tmpMode));
377 :
378 0 : if (tmpMode == dmaModeInfo.modeNeedSync_) {
379 0 : syncQues.push_back(*queIter); // que sync is required only when mode is PUT for send and GET for recv
380 : }
381 :
382 0 : if (tmpMode != mode) {
383 0 : hasDiffDmaMode = true;
384 : }
385 : }
386 :
387 0 : CHK_PRT_RET(
388 : netLinkNum > 1,
389 : HCCL_ERROR("[InsCollAlgFactory] [AlgDataTrans] Send/RecvReduceThruMultiLinks: more than one net links, "
390 : "use mid-level wrapper instead as DEV_NET is async."),
391 : HcclResult::HCCL_E_INTERNAL);
392 :
393 0 : return HcclResult::HCCL_SUCCESS;
394 : }
395 :
396 : } // namespace Hccl
|
