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