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 "log.h"
12 :
13 : #include "ins_coll_alg_registry.h"
14 : #include "ins_v2_batchSendRecv_executor.h"
15 : #include "alg_data_trans_wrapper.h"
16 :
17 : #include "hccl_aiv_utils.h"
18 : #include "aiv_ins.h"
19 : #include "executor_utils.h"
20 :
21 : using namespace std;
22 :
23 : namespace Hccl {
24 :
25 : template <typename AlgTopoMatch>
26 0 : InsV2BatchSendRecvExecutor<AlgTopoMatch>::InsV2BatchSendRecvExecutor() : InsCollAlgBase()
27 : {
28 0 : }
29 :
30 : template <typename AlgTopoMatch>
31 0 : InsV2BatchSendRecvExecutor<AlgTopoMatch>::~InsV2BatchSendRecvExecutor()
32 : {
33 0 : }
34 :
35 : template <typename AlgTopoMatch>
36 0 : void InsV2BatchSendRecvExecutor<AlgTopoMatch>::SetRmaDataBufferMgr(const RmtDataBufferMgr* rmaDataBufferMgr)
37 : {
38 0 : rmaDataBufferMgr_ = const_cast<RmtDataBufferMgr*>(rmaDataBufferMgr);
39 0 : return;
40 : }
41 :
42 : template <typename AlgTopoMatch>
43 0 : HcclResult InsV2BatchSendRecvExecutor<AlgTopoMatch>::InitParams(const CollAlgOperator &op, const CollAlgParams ¶ms)
44 : {
45 0 : op_ = op;
46 0 : opMode_ = params.opMode;
47 0 : maxTmpMemSize_ = params.maxTmpMemSize;
48 0 : CHK_PRT_RET((maxTmpMemSize_ == 0),
49 : HCCL_ERROR("[InitParams] maxTmpMemSize equals to zero for OPBASE."), HcclResult::HCCL_E_PARA);
50 0 : HcclSendRecvItem* itemPtr = reinterpret_cast<HcclSendRecvItem *>(op.batchSendRecvDataDes.sendRecvItemsPtr);
51 0 : u32 itemNum = op.batchSendRecvDataDes.itemNum;
52 0 : CHK_PTR_NULL(itemPtr);
53 0 : commTargetUserRankSet_.clear();
54 0 : for (u32 i = 0; i < itemNum; i++) {
55 0 : commTargetUserRankSet_.insert((itemPtr + i)->remoteRank);
56 0 : HCCL_DEBUG("[InsV2BatchSendRecvExecutor][ParseParam] insert remoteUserRank[%u] to Set ",
57 : (itemPtr + i)->remoteRank);
58 : }
59 0 : HCCL_DEBUG("[InitParams]commTargetUserRankSet_ size[%zu]", commTargetUserRankSet_.size());
60 0 : return HcclResult::HCCL_SUCCESS;
61 : }
62 :
63 : template <typename AlgTopoMatch>
64 0 : HcclResult InsV2BatchSendRecvExecutor<AlgTopoMatch>::InitCommInfo(const RankGraph *rankGraph)
65 : {
66 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
67 0 : CHK_RET(topoMatch.SetTargetRanks(commTargetUserRankSet_));
68 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
69 :
70 0 : return HcclResult::HCCL_SUCCESS;
71 0 : }
72 :
73 : template <typename AlgTopoMatch>
74 0 : HcclResult InsV2BatchSendRecvExecutor<AlgTopoMatch>::InitCommInfo(const AlgTopoInfo &topoInfo)
75 : {
76 0 : CHK_PRT_RET(topoInfo.vTopo.size() == 0,
77 : HCCL_ERROR("[InsV2BatchSendRecvExecutor] Rank[%d], vTopo size is zero.", myRank_),
78 : HcclResult::HCCL_E_PARA);
79 :
80 0 : CHK_PRT_RET(topoInfo.virtRankMap.size() == 0,
81 : HCCL_ERROR("[InsV2BatchSendRecvExecutor] Rank[%d], virtRankMap size is zero.", myRank_),
82 : HcclResult::HCCL_E_PARA);
83 :
84 0 : CHK_PRT_RET(topoInfo.virtRanks.size() == 0,
85 : HCCL_ERROR("[InsV2BatchSendRecvExecutor] Rank[%d], virtRanks size is zero.", myRank_),
86 : HcclResult::HCCL_E_PARA);
87 :
88 0 : vTopo_ = topoInfo.vTopo[0]; // 本通信域内的通信平面
89 0 : virtRankMap_ = topoInfo.virtRankMap[0]; // 本通信域内的 rank 映射表
90 0 : virtRanks_ = topoInfo.virtRanks[0]; // 本通信域内的 rank 集合
91 0 : return HcclResult::HCCL_SUCCESS;
92 : }
93 :
94 : template <typename AlgTopoMatch>
95 0 : HcclResult InsV2BatchSendRecvExecutor<AlgTopoMatch>::CalNumBlocks(
96 : u32& blockDim, u64 dataSize, u32 blockDimLimit)
97 : {
98 : (void)dataSize;
99 0 : u32 rankNum = 2;
100 0 : HCCL_INFO("[InsV2BatchSendRecvExecutor] Limit core num[%u]", blockDimLimit);
101 :
102 0 : if (blockDimLimit < rankNum) { // batchSendRecv至少需要两个核，分别去收发
103 0 : HCCL_ERROR("[InsV2BatchSendRecvExecutor] core num[%u] is less than 2", blockDimLimit);
104 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
105 : }
106 :
107 0 : blockDim = blockDimLimit / rankNum * rankNum;
108 0 : HCCL_INFO("[InsV2BatchSendRecvExecutor] Actually use core num[%u]", blockDim);
109 :
110 0 : return HcclResult::HCCL_SUCCESS;
111 : }
112 :
113 : template <typename AlgTopoMatch>
114 0 : bool InsV2BatchSendRecvExecutor<AlgTopoMatch>::SortSendItems(HcclSendRecvItem* a, HcclSendRecvItem* b) const{
115 0 : u32 aFlag = (a->remoteRank <= static_cast<uint32_t>(myRank_)) ?
116 0 : (a->remoteRank + rankSize_) : a->remoteRank;
117 0 : u32 bFlag = (b->remoteRank <= static_cast<uint32_t>(myRank_)) ?
118 0 : (b->remoteRank + rankSize_) : b->remoteRank;
119 0 : if (aFlag > bFlag) {
120 0 : return true;
121 0 : } else if (aFlag < bFlag) {
122 0 : return false;
123 : }
124 0 : return a->count > b->count;
125 : }
126 :
127 : template <typename AlgTopoMatch>
128 0 : bool InsV2BatchSendRecvExecutor<AlgTopoMatch>::SortRecvItems(HcclSendRecvItem* a, HcclSendRecvItem* b) const{
129 0 : u32 aFlag = (a->remoteRank < static_cast<uint32_t>(myRank_)) ?
130 0 : (a->remoteRank + rankSize_) : a->remoteRank;
131 0 : u32 bFlag = (b->remoteRank < static_cast<uint32_t>(myRank_)) ?
132 0 : (b->remoteRank + rankSize_) : b->remoteRank;
133 0 : if (aFlag > bFlag) {
134 0 : return false;
135 0 : } else if (aFlag < bFlag) {
136 0 : return true;
137 : }
138 0 : return a->count > b->count;
139 : }
140 :
141 : template <typename AlgTopoMatch>
142 0 : HcclResult InsV2BatchSendRecvExecutor<AlgTopoMatch>::GetPairWiseList()
143 : {
144 0 : HCCL_INFO("[InsV2BatchSendRecvExecutor][GetPairWiseList] Start sort the batchSendRecv tasklist.");
145 :
146 0 : HcclSendRecvItem *sendRecvInfo = static_cast<HcclSendRecvItem *>(op_.batchSendRecvDataDes.sendRecvItemsPtr);
147 0 : u32 itemNum = op_.batchSendRecvDataDes.itemNum;
148 0 : if (itemNum > BATCH_SEND_RECV_ITEM_SIZE) {
149 0 : HCCL_ERROR("[InsV2BatchSendRecvExecutor][GetPairWiseList] itemNum [%u] is greater than BATCH_SEND_RECV_ITEM_SIZE [%u]",
150 : itemNum, BATCH_SEND_RECV_ITEM_SIZE);
151 0 : return HcclResult::HCCL_E_PARA;
152 : }
153 :
154 0 : CHK_PTR_NULL(sendRecvInfo);
155 0 : std::set<DataType> hcclDataTypeSet;
156 :
157 0 : for (u32 i = 0; i < itemNum; i++) {
158 0 : CHK_PTR_NULL(sendRecvInfo->buf);
159 0 : HCCL_INFO("[InsV2BatchSendRecvExecutor][GetPairWiseList] index is %u, itemNum is %u,"\
160 : "localRankID is %d, sendRecvType is %u, buf is %p, count is %u, dataType is %u, remoteRank is %u, rankSize is %u.",
161 : i, itemNum, myRank_, static_cast<u32>(sendRecvInfo->sendRecvType), sendRecvInfo->buf, sendRecvInfo->count,
162 : static_cast<u32>(sendRecvInfo->dataType), sendRecvInfo->remoteRank, rankSize_);
163 :
164 0 : hcclDataTypeSet.insert(HcclDataTypeToDataType(sendRecvInfo->dataType));
165 :
166 0 : if (sendRecvInfo->sendRecvType == HcclSendRecvType::HCCL_SEND) {
167 0 : sendDeque_.push_back(sendRecvInfo);
168 0 : } else if (sendRecvInfo->sendRecvType == HcclSendRecvType::HCCL_RECV) {
169 0 : recvDeque_.push_back(sendRecvInfo);
170 : } else {
171 0 : HCCL_ERROR("[InsV2BatchSendRecvExecutor][GetPairWiseList] sendRecvType wrong sendrecvType is %d, "\
172 : "rankID is %d, remoteRank is %u.", sendRecvInfo->sendRecvType, myRank_,
173 : sendRecvInfo->remoteRank);
174 0 : return HcclResult::HCCL_E_PARA;
175 : }
176 0 : sendRecvInfo++;
177 : }
178 : // 如果item里面的数据类型都一样，那就用item里面的，如果不一样，就统一用UINT8
179 0 : dataType_ = DataType::UINT8;
180 0 : if (hcclDataTypeSet.size() == 1) {
181 0 : dataType_ = *hcclDataTypeSet.begin();
182 : }
183 0 : HCCL_INFO("[InsV2BatchSendRecvExecutor][GetPairWiseList] dataType num is %u, so the final dataType_ is %u",
184 : hcclDataTypeSet.size(), static_cast<u32>(dataType_));
185 :
186 : /* 此处的排序逻辑(pair-wise算法):
187 : 1.sendDeque元素顺序是:先放remoteRank号小于等于root rank的第一个任务，依次减小(循环索引)直至放完
188 : 2.recvDeque元素顺序是:先放remoteRank号大于等于root rank的第一个任务，依次增大(循环索引)直至放完
189 : 如果有rank间重复send/recv场景，按照收发数据从大到小排序
190 : */
191 0 : auto sendCompare = [this](HcclSendRecvItem* a, HcclSendRecvItem* b) {
192 0 : return this->SortSendItems(a, b);
193 : };
194 :
195 0 : auto recvCompare = [this](HcclSendRecvItem* a, HcclSendRecvItem* b) {
196 0 : return this->SortRecvItems(a, b);
197 : };
198 :
199 0 : std::stable_sort(sendDeque_.begin(), sendDeque_.end(), sendCompare);
200 0 : std::stable_sort(recvDeque_.begin(), recvDeque_.end(), recvCompare);
201 :
202 : // 校验自收发任务，校验数据量和数据类型是否一一对应
203 : // 遍历收发队列
204 0 : for (auto& item : sendDeque_) {
205 0 : if (item->remoteRank == static_cast<uint32_t>(myRank_)) {
206 0 : sendToSelfDeque_.push_back(item);
207 : }
208 : }
209 :
210 0 : for (auto& item : recvDeque_) {
211 0 : if (item->remoteRank == static_cast<uint32_t>(myRank_)) {
212 0 : recvFromSelfDeque_.push_back(item);
213 : }
214 : }
215 :
216 0 : if (sendToSelfDeque_.size() != recvFromSelfDeque_.size()) {
217 0 : HCCL_ERROR("[InsV2BatchSendRecvExecutor][GetPairWiseList] selfSendRecv is not equal,vsendQue size is [%u], recvQue size is [%u]",
218 : sendToSelfDeque_.size(), recvFromSelfDeque_.size());
219 0 : return HcclResult::HCCL_E_PARA;
220 : }
221 :
222 : // 收发队列应该一一对应
223 0 : for (u32 i = 0; i < sendToSelfDeque_.size(); i++) {
224 0 : if ((sendToSelfDeque_[i]->count != recvFromSelfDeque_[i]->count) ||
225 0 : (sendToSelfDeque_[i]->dataType != recvFromSelfDeque_[i]->dataType)) {
226 0 : HCCL_ERROR("[InsV2BatchSendRecvExecutor][GetPairWiseList] selfSendRecv is not equal, "\
227 : "sendQue count is [%u], sendQue dataType is [%u]; recvQue count is [%u], recvQue dataType is [%u]",
228 : sendToSelfDeque_[i]->count, static_cast<u32>(sendToSelfDeque_[i]->dataType),
229 : recvFromSelfDeque_[i]->count, static_cast<u32>(recvFromSelfDeque_[i]->dataType));
230 0 : return HcclResult::HCCL_E_PARA;
231 : }
232 : }
233 :
234 0 : HCCL_INFO("[CollBatchSendRecvExecutor][GetPairWiseList] End sort the batchSendRecv tasklist.");
235 0 : return HcclResult::HCCL_SUCCESS;
236 0 : }
237 :
238 : // 算子执行aiv接口，这个接口需要补齐
239 : template <typename AlgTopoMatch>
240 0 : HcclResult InsV2BatchSendRecvExecutor<AlgTopoMatch>::Orchestrate(
241 : const RankGraph *rankGraph,
242 : const CollAlgOperator &op,
243 : const CollAlgParams ¶ms,
244 : InsQuePtr insQue)
245 : {
246 0 : HCCL_INFO("[InsV2BatchSendRecvExecutor][Orchestrate] Begin to Generate Instruction Queue for BatchSendRecv.");
247 : // init and check params
248 0 : CHK_RET(Init(op, params, insQue));
249 0 : CHK_RET(InitCommInfo(rankGraph));
250 :
251 0 : CHK_PRT_RET(rankSize_ == 1,
252 : HCCL_ERROR("BatchSendRecv Executor orchestrate failed, do not support single rank."),
253 : HcclResult::HCCL_E_PARA);
254 :
255 : // calculate required insQues and prepare queue
256 0 : AlgTempResReq tempResReq;
257 0 : CHK_RET(CalcRes(tempResReq));
258 :
259 0 : CHK_RET(InitQueue(tempResReq.queNum, requiredQue_));
260 0 : HCCL_DEBUG("[InsV2BatchSendRecvExecutor] Rank[%d], requiredQue Num [%u].", myRank_, tempResReq.queNum);
261 :
262 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, tempResLinks_));
263 :
264 0 : CHK_RET(ExecAiv()); // 这里进入算法编排，把参数按照顺序构造好发下去
265 0 : HCCL_INFO("[InsV2BatchSendRecvExecutor][Orchestrate] Orchestrate AIV End");
266 0 : return HcclResult::HCCL_SUCCESS;
267 0 : }
268 :
269 : // 算子执行aicpu接口
270 : template <typename AlgTopoMatch>
271 0 : HcclResult InsV2BatchSendRecvExecutor<AlgTopoMatch>::Orchestrate(const AlgTopoInfo &topoInfo,
272 : const CollAlgOperator &op,
273 : const CollAlgParams ¶ms,
274 : ConnectedLinkMgr *linkMgr,
275 : InsQuePtr insQue)
276 : {
277 0 : HCCL_INFO("[InsV2BatchSendRecvExecutor][Orchestrate] Begin to Generate Instruction Queue for BatchSendRecv.");
278 : // init and check params
279 0 : CHK_RET(Init(op, params, insQue));
280 0 : CHK_RET(InitCommInfo(topoInfo));
281 :
282 0 : CHK_PRT_RET(rankSize_ == 1,
283 : HCCL_ERROR("BatchSendRecv Executor orchestrate failed, do not support single rank."),
284 : HcclResult::HCCL_E_PARA);
285 :
286 : // calculate required insQues and prepare queue
287 0 : AlgTempResReq tempResReq;
288 0 : CHK_RET(CalcRes(tempResReq));
289 :
290 0 : CHK_RET(InitQueue(tempResReq.queNum, requiredQue_));
291 0 : HCCL_DEBUG("[InsV2BatchSendRecvExecutor] Rank[%d], requiredQue Num [%u].", myRank_, tempResReq.queNum);
292 :
293 0 : CHK_RET(PrepResLinks(myRank_, tempResReq.links, linkMgr, tempResLinks_));
294 :
295 0 : CHK_RET(ExecAiv()); // 这里进入算法编排，把参数按照顺序构造好发下去
296 0 : HCCL_INFO("[InsV2BatchSendRecvExecutor][Orchestrate] Orchestrate AICPU End");
297 0 : return HcclResult::HCCL_SUCCESS;
298 0 : }
299 :
300 : template <typename AlgTopoMatch>
301 0 : HcclResult InsV2BatchSendRecvExecutor<AlgTopoMatch>::ExecAiv()
302 : {
303 0 : HCCL_INFO("[InsV2SendExecutor][ExecAiv] start: rank[%d]", myRank_);
304 :
305 0 : CHK_RET(GetPairWiseList());
306 :
307 0 : u64 transportBoundDataSize = UB_MAX_DATA_SIZE;
308 0 : u64 maxScratchDataSize = std::min(transportBoundDataSize, maxTmpMemSize_);
309 0 : std::vector<LinkData> allLinks;
310 0 : for (auto iter = tempResLinks_.begin(); iter != tempResLinks_.end(); ++iter) {
311 0 : allLinks.emplace_back(iter->second.at(0));
312 : }
313 :
314 0 : sliceId_++;
315 :
316 0 : AivOpArgs aivBatchSendRecvArgs;
317 0 : aivBatchSendRecvArgs.cmdType = HcclCMDType::HCCL_CMD_BATCH_SEND_RECV;
318 0 : aivBatchSendRecvArgs.input = 0; // ins_rules.cc里面，这里会和起始地址累加起来作为input
319 0 : aivBatchSendRecvArgs.output = 0;
320 0 : aivBatchSendRecvArgs.rank = u32(myRank_);
321 0 : aivBatchSendRecvArgs.rankSize = rankSize_;
322 0 : aivBatchSendRecvArgs.count = maxScratchDataSize; // 把整个 CCLBuffer的size发过去，因为这里没法确认单次send/recv的dataType
323 0 : aivBatchSendRecvArgs.dataType = dataType_;
324 0 : aivBatchSendRecvArgs.aivTag = sliceId_; // 传入aivTag，Lauch时重新组装为aivTag
325 0 : aivBatchSendRecvArgs.isOpBase = (opMode_ == OpMode::OPBASE);
326 0 : aivBatchSendRecvArgs.xRankSize = rankSize_;
327 0 : aivBatchSendRecvArgs.yRankSize = 0;
328 0 : aivBatchSendRecvArgs.zRankSize = 0;
329 0 : CHK_RET(CalNumBlocks(aivBatchSendRecvArgs.numBlocks, 0, op_.numBlocksLimit)); // 为什么前面计算的值不能用吗，这里要再计算一遍
330 :
331 0 : aivBatchSendRecvArgs.extraArgs.itemNum = op_.batchSendRecvDataDes.itemNum;
332 :
333 : // 遍历收、发队列
334 0 : u32 curQue = 0;
335 0 : for (auto& item : sendDeque_) {
336 0 : aivBatchSendRecvArgs.extraArgs.sendRecvInfo[curQue].sendRecvType = static_cast<uint32_t>(item->sendRecvType);
337 0 : aivBatchSendRecvArgs.extraArgs.sendRecvInfo[curQue].bufAddr = reinterpret_cast<uint64_t>(item->buf);
338 0 : aivBatchSendRecvArgs.extraArgs.sendRecvInfo[curQue].count = item->count;
339 0 : aivBatchSendRecvArgs.extraArgs.sendRecvInfo[curQue].dataTypeSize = DATA_TYPE_SIZE_MAP.at(HcclDataTypeToDataType(item->dataType));
340 0 : aivBatchSendRecvArgs.extraArgs.sendRecvInfo[curQue].remoteRank = item->remoteRank;
341 0 : curQue++;
342 : }
343 :
344 0 : for (auto& item : recvDeque_) {
345 0 : aivBatchSendRecvArgs.extraArgs.sendRecvInfo[curQue].sendRecvType = static_cast<uint32_t>(item->sendRecvType);
346 0 : aivBatchSendRecvArgs.extraArgs.sendRecvInfo[curQue].bufAddr = reinterpret_cast<uint64_t>(item->buf);
347 0 : aivBatchSendRecvArgs.extraArgs.sendRecvInfo[curQue].count = item->count;
348 0 : aivBatchSendRecvArgs.extraArgs.sendRecvInfo[curQue].dataTypeSize = DATA_TYPE_SIZE_MAP.at(HcclDataTypeToDataType(item->dataType));
349 0 : aivBatchSendRecvArgs.extraArgs.sendRecvInfo[curQue].remoteRank = item->remoteRank;
350 0 : curQue++;
351 : }
352 :
353 0 : aivBatchSendRecvArgs.inputSliceStride = 0;
354 0 : aivBatchSendRecvArgs.outputSliceStride = 0;
355 0 : aivBatchSendRecvArgs.repeatNum = 1; // 不重复
356 0 : aivBatchSendRecvArgs.inputRepeatStride = 0;
357 0 : aivBatchSendRecvArgs.outputRepeatStride = 0;
358 :
359 0 : std::unique_ptr<Instruction> aivInsBatchSendRecv = std::make_unique<AivInstruction>(allLinks, aivBatchSendRecvArgs);
360 :
361 0 : requiredQue_[0]->Append(std::move(aivInsBatchSendRecv));
362 :
363 0 : HCCL_INFO("[InsV2BatchSendRecvExecutor][ExecAiv] end: rank[%d]", myRank_);
364 0 : return HcclResult::HCCL_SUCCESS;
365 0 : }
366 :
367 : template <typename AlgTopoMatch>
368 0 : HcclResult InsV2BatchSendRecvExecutor<AlgTopoMatch>::CalcResLinksPartialMesh
369 : (const RankId myRank, const std::vector<std::vector<RankId>> &tempVTopo,
370 : const u32 linkNumBtwPeers, AlgTempResReq &tempResReq)
371 : {
372 : u32 myAlgRank;
373 0 : u32 partialRankSize = commTargetUserRankSet_.size() + 1;
374 :
375 0 : if (tempVTopo.size() < 1) {
376 0 : HCCL_ERROR("[InsV2BatchSendRecvExecutor][CalcResLinksPartialMesh] Rank[%d], tempVTopo size is zero.", myRank);
377 0 : return HCCL_E_PARA;
378 : }
379 0 : for (u32 i = 0; i < tempVTopo.size(); i++) { // 遍历level0的2个平面
380 0 : CHK_RET(GetAlgRank(myRank, tempVTopo[i], myAlgRank));
381 0 : for (u32 queIdx = 0; queIdx < tempResReq.queNum; queIdx++) {
382 : // find neighbors : virtualRank
383 0 : u32 remoteAlgRank = (myAlgRank + 1 + queIdx + partialRankSize) % partialRankSize;
384 0 : if (remoteAlgRank >= tempVTopo[i].size()) {
385 0 : continue;
386 : }
387 0 : RankId neighborRank = tempVTopo[i][remoteAlgRank];
388 0 : HCCL_DEBUG("tempVTopo[%u] index[%u] value[%d]", i, remoteAlgRank, neighborRank);
389 0 : auto rankInRankSet = std::find(commTargetUserRankSet_.begin(), commTargetUserRankSet_.end(),
390 0 : static_cast<u32>(neighborRank));
391 0 : if (rankInRankSet != commTargetUserRankSet_.end() && neighborRank != myRank) {
392 : // LinkNum
393 0 : tempResReq.links[neighborRank] = linkNumBtwPeers;
394 0 : HCCL_DEBUG("myRank[%d] neighborRank[%d] links is [%u]", myRank, neighborRank, linkNumBtwPeers);
395 : }
396 : }
397 : }
398 :
399 0 : return HcclResult::HCCL_SUCCESS;
400 : }
401 :
402 : template <typename AlgTopoMatch>
403 0 : HcclResult InsV2BatchSendRecvExecutor<AlgTopoMatch>::CalcRes(AlgTempResReq &tempResReq)
404 : {
405 0 : tempResReq.queNum = 1; // aiv只需要1条流
406 0 : tempResReq.streamNum = tempResReq.queNum;
407 :
408 0 : CHK_RET(CalcResLinksPartialMesh(myRank_, vTopo_, 1, tempResReq));
409 0 : HCCL_DEBUG("[InsV2BatchSendRecvExecutor][CalcRes] Rank[%d] vTopoSize[%lu] requiredQue Num[%u].",
410 : myRank_, vTopo_[0].size(), tempResReq.queNum);
411 0 : return HcclResult::HCCL_SUCCESS;
412 : }
413 :
414 : template <typename AlgTopoMatch>
415 0 : HcclResult InsV2BatchSendRecvExecutor<AlgTopoMatch>::CalcResOffload(const RankGraph *rankGraph,
416 : const u64 &dataSize,
417 : CollOffloadOpResReq &resReq)
418 : {
419 : (void)rankGraph;
420 : (void)dataSize;
421 : (void)resReq;
422 0 : HCCL_ERROR("[InsCollAlgFactory][InsV2BatchSendRecvExecutor][CalcResOffload] offload is not support");
423 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
424 : }
425 :
426 : template <typename AlgTopoMatch>
427 0 : HcclResult InsV2BatchSendRecvExecutor<AlgTopoMatch>::CalcRes(const RankGraph *rankGraph,
428 : CollAlgResReq &algResReq)
429 : {
430 : // Topo Match
431 0 : CHK_RET(InitCommInfo(rankGraph));
432 :
433 0 : algResReq.topoInfo.UpdateSingleLevelTopo(virtRanks_, virtRankMap_, vTopo_);
434 :
435 0 : for (u32 i = 0; i < vTopo_.size(); i++) { // 遍历level0
436 0 : for (u32 j = 0; j < vTopo_[i].size(); j++) { // 遍历平面内的所有rank
437 0 : HCCL_DEBUG("[InsV2BatchSendRecvExecutor][CalcResLinksPartialMesh] vTopo_[%u][%u] is [%d].",
438 : i, j, vTopo_[i][j]);
439 : }
440 : }
441 0 : HCCL_DEBUG("[InsV2BatchSendRecvExecutor][CalcRes]topoInfo.virtRanks[%u], topoInfo.virtRankMap[%u],"\
442 : "topoInfo.vTopo[%u]", algResReq.topoInfo.virtRanks.size(),
443 : algResReq.topoInfo.virtRankMap.size(), algResReq.topoInfo.vTopo.size());
444 :
445 : // calculate required insQues and prepare queue
446 0 : AlgTempResReq tempResReq;
447 0 : if (enableDetour_) {
448 0 : HCCL_DEBUG("[InsV2BatchSendRecvExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
449 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
450 : } else {
451 0 : HCCL_DEBUG("[InsV2BatchSendRecvExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
452 0 : CHK_RET(CalcRes(tempResReq));
453 : }
454 :
455 0 : algResReq.primQueueNum = tempResReq.streamNum;
456 0 : algResReq.queueNotifys = tempResReq.queNotifys;
457 0 : HCCL_DEBUG("[InsV2BatchSendRecvExecutor] Rank[%d], requiredQueNum [%u].", myRank_, algResReq.primQueueNum);
458 :
459 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, tempResReq.links, algResReq.levelRankPairs));
460 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, algResReq.links));
461 0 : HCCL_DEBUG("[InsV2BatchSendRecvExecutor] Rank[%d], algResReq.links size[%zu].", myRank_, algResReq.links.size());
462 :
463 0 : return HcclResult::HCCL_SUCCESS;
464 0 : }
465 :
466 : // 注册
467 : INS_REGISTER_IMPL_BY_TOPO(OpType::BATCHSENDRECV, AivBatchSendRecv, InsV2BatchSendRecvExecutor, TopoMatchPartialMesh);
468 :
469 : } // namespace Hccl
|
