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