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_recv_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 0 : InsV2RecvExecutor::InsV2RecvExecutor() : InsCollAlgBase() {}
25 :
26 0 : InsV2RecvExecutor::~InsV2RecvExecutor() {}
27 :
28 0 : HcclResult InsV2RecvExecutor::CalNumBlocks(u32& numBlocks, u64 dataSize, u32 numBlocksLimit)
29 : {
30 : (void)dataSize;
31 :
32 0 : if (numBlocksLimit < 1) {
33 0 : HCCL_ERROR("[InsV2RecvExecutor] core num[%u] is less than 1", numBlocksLimit);
34 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
35 : }
36 :
37 0 : numBlocks = numBlocksLimit;
38 0 : HCCL_INFO("[InsV2RecvExecutor] Actually use core num[%u]", numBlocks);
39 :
40 0 : return HcclResult::HCCL_SUCCESS;
41 : }
42 :
43 : HcclResult
44 0 : InsV2RecvExecutor::CalcResOffload(const RankGraph* rankGraph, const u64& dataSize, CollOffloadOpResReq& resReq)
45 : {
46 : (void)rankGraph;
47 : (void)dataSize;
48 : (void)resReq;
49 0 : HCCL_ERROR("[InsCollAlgFactory][InsV2RecvExecutor][CalcResOffload] offload is not support");
50 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
51 : }
52 :
53 0 : HcclResult InsV2RecvExecutor::CalcRes(const RankGraph* rankGraph, CollAlgResReq& algResReq)
54 : {
55 0 : u32 linkNumBtwPeers = 1;
56 0 : algResReq.primQueueNum = 1;
57 0 : AlgTempResReq tempResReq;
58 0 : tempResReq.queNum = 1;
59 0 : tempResReq.streamNum = tempResReq.queNum;
60 0 : if (static_cast<u32>(sendRecvRemoteRank_) > rankSize_ - 1) {
61 0 : HCCL_ERROR(
62 : "[InsCollAlgFactory][InsV2RecvExecutor][CalcRes] Rank[%d] get dest[%d] is invalid", myRank_,
63 : sendRecvRemoteRank_);
64 0 : return HcclResult::HCCL_E_PARA;
65 : }
66 0 : tempResReq.links[sendRecvRemoteRank_] = linkNumBtwPeers;
67 0 : uint32_t linkNum = GetPathsFromRankGraph(rankGraph, myRank_, sendRecvRemoteRank_).size();
68 0 : if (linkNum == 0) {
69 0 : HCCL_ERROR(
70 : "[InsCollAlgFactory][InsV2RecvExecutor][CalcRes] Rank[%d] get path num to dest[%d] is zero", myRank_,
71 : sendRecvRemoteRank_);
72 : }
73 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, algResReq.links));
74 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, tempResReq.links, algResReq.levelRankPairs));
75 0 : return HcclResult::HCCL_SUCCESS;
76 0 : }
77 :
78 : // host
79 0 : HcclResult InsV2RecvExecutor::Orchestrate(
80 : const RankGraph* rankGraph, const CollAlgOperator& op, const CollAlgParams& params, InsQuePtr insQue)
81 : {
82 0 : HCCL_DEBUG("[InsCollAlgFactory][InsV2RecvExecutor][Orchestrate] Begin to Generate Instruction Queue for RECV.");
83 0 : CHK_RET(Init(op, params, insQue));
84 : // 从集合通信算子op中获得remote rank和data type/count相关信息
85 0 : RankId remoteRank = op.sendRecvRemoteRank;
86 0 : u32 dataElemSize = DATA_TYPE_SIZE_MAP.at(op.dataType);
87 0 : u64 totalDataSize = static_cast<u64>(dataElemSize) * op.dataCount;
88 :
89 : // 判断是不是自发自收这种情况，若是，则什么都不做，直接返回
90 0 : if (myRank_ == remoteRank) {
91 0 : HCCL_WARNING("[InsCollAlgFactory][InsV2RecvExecutor][Orchestrate] Self send, Self recv, Do nothing");
92 0 : return HcclResult::HCCL_SUCCESS;
93 : }
94 0 : HCCL_DEBUG("[InsCollAlgFactory][InsV2RecvExecutor][Orchestrate] Other send, Self recv");
95 :
96 : // 从virtualTopo里面拿到 link，转成LinkData格式
97 0 : const std::vector<NetInstance::Path> recvPath = GetPathsFromRankGraph(rankGraph, myRank_, remoteRank);
98 0 : CHK_PRT_RET(
99 : recvPath.size() == 0,
100 : HCCL_ERROR("[InsCollAlgFactory] Unable to obtain valid link, srcRank [%d], dstRank [%d].", myRank_, remoteRank),
101 : HcclResult::HCCL_E_INTERNAL);
102 0 : LinkData recvLinkData(recvPath[0]);
103 0 : HCCL_DEBUG(
104 : "[InsCollAlgFactory][InsV2RecvExecutor][Orchestrate] Total transfer data size [%llu], Max scratch buffer size "
105 : "[%u].",
106 : totalDataSize, params.maxTmpMemSize);
107 :
108 : // 模式判断
109 0 : if (opMode_ == OpMode::OFFLOAD) {
110 0 : HCCL_ERROR("[InsCollAlgFactory][InsV2RecvExecutor][Orchestrate] offload is not support");
111 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
112 : } else {
113 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OPBASE Mode for HOST.", myRank_);
114 :
115 0 : CHK_RET(ExecAiv(op, params, recvLinkData, insQue));
116 0 : return HcclResult::HCCL_SUCCESS;
117 : }
118 : return HcclResult::HCCL_SUCCESS;
119 0 : }
120 :
121 : // aicpu
122 0 : HcclResult InsV2RecvExecutor::Orchestrate(
123 : const AlgTopoInfo& topoInfo, const CollAlgOperator& op, const CollAlgParams& params, ConnectedLinkMgr* linkMgr,
124 : InsQuePtr insQue)
125 : {
126 : (void)topoInfo;
127 0 : HCCL_DEBUG(
128 : "[InsCollAlgFactory][InsV2RecvExecutor][Orchestrate] Begin to Generate Instruction Queue for RECV AICPU mode.");
129 0 : CHK_RET(Init(op, params, insQue));
130 : // 从集合通信算子op中获得remote rank和data type/count相关信息
131 0 : RankId remoteRank = op.sendRecvRemoteRank;
132 0 : u32 dataElemSize = DATA_TYPE_SIZE_MAP.at(op.dataType);
133 0 : u64 totalDataSize = static_cast<u64>(dataElemSize) * op.dataCount;
134 :
135 : // 判断是不是自发自收这种情况，若是，则什么都不做，直接返回
136 0 : if (myRank_ == remoteRank) {
137 0 : HCCL_WARNING("[InsCollAlgFactory][InsV2RecvExecutor][Orchestrate] Self send, Self recv, Do nothing");
138 0 : return HcclResult::HCCL_SUCCESS;
139 : }
140 0 : HCCL_DEBUG("[InsCollAlgFactory][InsV2RecvExecutor][Orchestrate] Other send, Self recv.");
141 :
142 : // 从linkMgr里面拿到 linkData
143 0 : const vector<LinkData> recvPath = linkMgr->GetLinks(remoteRank);
144 0 : CHK_PRT_RET(
145 : recvPath.size() == 0,
146 : HCCL_ERROR("[InsCollAlgFactory] Unable to obtain valid link, srcRank [%d], dstRank [%d].", myRank_, remoteRank),
147 : HcclResult::HCCL_E_INTERNAL);
148 0 : LinkData recvLinkData(recvPath[0]);
149 0 : HCCL_DEBUG(
150 : "[InsCollAlgFactory][InsV2RecvExecutor][Orchestrate] Total transfer data size [%llu], Max scratch buffer size "
151 : "[%u].",
152 : totalDataSize, params.maxTmpMemSize);
153 :
154 : // 模式判断
155 0 : if (opMode_ == OpMode::OFFLOAD) {
156 0 : HCCL_ERROR("[InsCollAlgFactory][InsV2RecvExecutor][Orchestrate] offload is not support");
157 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
158 : } else {
159 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OPBASE Mode for HOST.", myRank_);
160 :
161 0 : CHK_RET(ExecAiv(op, params, recvLinkData, insQue));
162 0 : return HcclResult::HCCL_SUCCESS;
163 : }
164 : return HcclResult::HCCL_SUCCESS;
165 0 : }
166 :
167 0 : HcclResult InsV2RecvExecutor::ExecAiv(
168 : const CollAlgOperator& op, const CollAlgParams& params, LinkData& recvLinkData, InsQuePtr insQue)
169 : {
170 0 : HCCL_INFO(
171 : "[InsV2RecvExecutor][ExecAiv] start: rank is %d, count is %u, dataType is %u, srcRank is %d", myRank_,
172 : op.dataCount, static_cast<u32>(op.dataType), op.sendRecvRemoteRank);
173 :
174 0 : u64 transportBoundDataSize = UB_MAX_DATA_SIZE;
175 0 : u64 maxScratchDataSize = std::min(transportBoundDataSize, params.maxTmpMemSize);
176 0 : u32 dataElemSize = DATA_TYPE_SIZE_MAP.at(op.dataType);
177 0 : u64 maxScratchDataCount = maxScratchDataSize / dataElemSize;
178 0 : CHK_PRT_RET(
179 : maxScratchDataCount == 0, HCCL_ERROR("[InsV2RecvExecutor][Orchestrate] maxScratchDataCount is 0"),
180 : HCCL_E_INTERNAL);
181 :
182 0 : std::vector<LinkData> allLinks;
183 0 : allLinks.emplace_back(recvLinkData);
184 :
185 0 : u64 loopTimes = op.dataCount / maxScratchDataCount + static_cast<u64>(op.dataCount % maxScratchDataCount != 0);
186 0 : u64 processedDataCount = 0;
187 0 : for (u64 loop = 0; loop < loopTimes; loop++) {
188 0 : sliceId_++; // 自动增长sliceId，传入aivTag
189 0 : u64 currDataCount = (loop == loopTimes - 1) ? op.dataCount - processedDataCount : maxScratchDataCount;
190 0 : HCCL_INFO(
191 : "[InsV2RecvExecutor][ExecAiv] myRank[%d], loop[%llu] sliceId_[%llu] currDataCount[%llu], "
192 : "processedDataCount[%llu]",
193 : myRank_, loop, sliceId_, currDataCount, processedDataCount);
194 :
195 0 : AivOpArgs aivRecvArgs;
196 0 : aivRecvArgs.cmdType = HcclCMDType::HCCL_CMD_RECEIVE;
197 0 : aivRecvArgs.input = 0;
198 0 : aivRecvArgs.output = processedDataCount * dataElemSize;
199 0 : aivRecvArgs.rank = u32(myRank_);
200 0 : aivRecvArgs.sendRecvRemoteRank = op.sendRecvRemoteRank;
201 0 : aivRecvArgs.rankSize = rankSize_;
202 0 : aivRecvArgs.count = currDataCount; // 需要传输的数据量
203 0 : aivRecvArgs.dataType = op.dataType;
204 0 : aivRecvArgs.aivTag = sliceId_; // 传入aivTag，Lauch时重新组装为aivTag
205 0 : aivRecvArgs.isOpBase = (opMode_ == OpMode::OPBASE);
206 0 : aivRecvArgs.xRankSize = rankSize_;
207 0 : aivRecvArgs.yRankSize = 0;
208 0 : aivRecvArgs.zRankSize = 0;
209 0 : CHK_RET(CalNumBlocks(aivRecvArgs.numBlocks, 0, op.numBlocksLimit));
210 :
211 0 : aivRecvArgs.inputSliceStride = 0;
212 0 : aivRecvArgs.outputSliceStride = 0;
213 0 : aivRecvArgs.repeatNum = 1; // 不重复
214 0 : aivRecvArgs.inputRepeatStride = 0;
215 0 : aivRecvArgs.outputRepeatStride = 0;
216 :
217 0 : std::unique_ptr<Instruction> aivInsRecvMesh1D = std::make_unique<AivInstruction>(allLinks, aivRecvArgs);
218 :
219 0 : insQue->Append(std::move(aivInsRecvMesh1D));
220 0 : processedDataCount += currDataCount;
221 0 : }
222 :
223 0 : HCCL_INFO("[InsV2RecvExecutor][ExecAiv] end: rank[%d]", myRank_);
224 0 : return HcclResult::HCCL_SUCCESS;
225 0 : }
226 :
227 : // 注册
228 : INS_REGISTER_IMPL(OpType::RECV, AivRecv, InsV2RecvExecutor);
229 :
230 : } // namespace Hccl
|
