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