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 "ins_broadcast_sole_executor.h"
12 : #include "log.h"
13 : #include "ins_coll_alg_registry.h"
14 : #ifndef CCL_KERNEL_AICPU
15 : #include "ccu_temp_broadcast_mesh_1D.h"
16 : #include "ccu_temp_broadcast_mesh_2D.h"
17 : #endif
18 : #include "topo_match_mesh.h"
19 : #include "topo_match_concurr_mesh.h"
20 : #include "alg_data_trans_wrapper.h"
21 :
22 : namespace Hccl {
23 : template <typename AlgTopoMatch, typename InsAlgTemplate>
24 0 : InsBroadcastSoleExecutor<AlgTopoMatch, InsAlgTemplate>::InsBroadcastSoleExecutor() : InsCollAlgBase()
25 0 : {}
26 :
27 : template <typename AlgTopoMatch, typename InsAlgTemplate>
28 0 : InsBroadcastSoleExecutor<AlgTopoMatch, InsAlgTemplate>::~InsBroadcastSoleExecutor()
29 0 : {}
30 :
31 : template <typename AlgTopoMatch, typename InsAlgTemplate>
32 0 : HcclResult InsBroadcastSoleExecutor<AlgTopoMatch, InsAlgTemplate>::CalcResOffload(
33 : const RankGraph* rankGraph, const u64& dataSize, CollOffloadOpResReq& resReq)
34 : {
35 : (void)dataSize;
36 0 : resReq.requiredScratchMemSize = 0;
37 :
38 : // Topo Match
39 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
40 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
41 :
42 : // instantiate a template
43 0 : InsAlgTemplate tempAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
44 :
45 : // calculate required insQueues and prepare queue
46 0 : AlgTempResReq tempResReq;
47 0 : if (enableDetour_) {
48 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsBroadcastSoleExecutor], CalcRes with detouring enabled.");
49 0 : CHK_RET(tempAlg.CalcResDetour(rankGraph, tempResReq));
50 : } else {
51 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsBroadcastSoleExecutor], CalcRes with detouring disabled.");
52 0 : CHK_RET(tempAlg.CalcRes(tempResReq));
53 : }
54 :
55 0 : resReq.requiredSubQueNum = tempResReq.streamNum - 1;
56 :
57 0 : return HcclResult::HCCL_SUCCESS;
58 0 : }
59 :
60 : template <typename AlgTopoMatch, typename InsAlgTemplate>
61 : HcclResult
62 0 : InsBroadcastSoleExecutor<AlgTopoMatch, InsAlgTemplate>::CalcRes(const RankGraph* rankGraph, CollAlgResReq& algResReq)
63 : {
64 : // Topo Match
65 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
66 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
67 0 : algResReq.topoInfo.UpdateSingleLevelTopo(virtRanks_, virtRankMap_, vTopo_);
68 :
69 : // instantiate a template
70 0 : InsAlgTemplate tempAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
71 0 : tempAlg.SetRoot(op_.root);
72 : // calculate required insQues and prepare queue
73 0 : AlgTempResReq tempResReq;
74 0 : if (enableDetour_) {
75 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsBroadcastSoleExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
76 0 : CHK_RET(tempAlg.CalcResDetour(rankGraph, tempResReq));
77 : } else {
78 0 : HCCL_DEBUG(
79 : "[InsCollAlgFactory] [InsBroadcastSoleExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
80 0 : CHK_RET(tempAlg.CalcRes(tempResReq));
81 : }
82 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, tempResReq.links, algResReq.levelRankPairs));
83 0 : algResReq.primQueueNum = tempResReq.streamNum;
84 0 : algResReq.queueNotifys = tempResReq.queNotifys;
85 0 : algResReq.localWaitGroupCntNotify = tempResReq.localWaitGroupCntNotify;
86 0 : algResReq.localBcastPostCntNotify = tempResReq.localBcastPostCntNotify;
87 0 : HCCL_DEBUG(
88 : "[InsCollAlgFactory] [InsBroadcastSoleExecutor] Rank[%d], requiredQueNum [%u].", myRank_,
89 : algResReq.primQueueNum);
90 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, algResReq.links));
91 :
92 0 : return HcclResult::HCCL_SUCCESS;
93 0 : }
94 :
95 : // dataSize_ as input
96 : template <typename AlgTopoMatch, typename InsAlgTemplate>
97 0 : HcclResult InsBroadcastSoleExecutor<AlgTopoMatch, InsAlgTemplate>::Orchestrate(
98 : const RankGraph* rankGraph, const CollAlgOperator& op, const CollAlgParams& params, InsQuePtr insQue)
99 : {
100 0 : HCCL_INFO("[InsCollAlgFactory] [InsBroadcastSoleExecutor] Host Orchestrate begins.");
101 : // init and check params
102 0 : CHK_RET(Init(op, params, insQue));
103 :
104 : // Topo Match
105 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
106 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
107 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], [%s].", myRank_, topoMatch.Describe().c_str());
108 0 : dataTypeSize_ = DataTypeSizeGet(dataType_);
109 0 : dataSize_ = dataCount_ * dataTypeSize_;
110 0 : CHK_PRT_RET(
111 : dataTypeSize_ == 0,
112 : HCCL_ERROR("Broadcast_[CollAlgFactory] Rank [%d], Invalid dataTypeSize_ [%u].", myRank_, dataTypeSize_),
113 : HcclResult::HCCL_E_INTERNAL);
114 :
115 : // 实例化算法模板类
116 0 : HCCL_DEBUG(
117 : "[InsBroadcastSoleExecutor] Rank[%d], Init insAlgTemplate with rankSize [%u] and dmaMode [%s].", myRank_,
118 : rankSize_, dmaMode_.Describe().c_str());
119 0 : InsAlgTemplate tempAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
120 0 : tempAlg.SetDmaMode(dmaMode_);
121 0 : tempAlg.SetCollOp(op); // CCU template需要传递op信息
122 0 : tempAlg.SetRoot(root_);
123 :
124 : // 计算算法模板所需资源
125 0 : AlgTempResReq tempResReq;
126 0 : if (enableDetour_) {
127 0 : HCCL_DEBUG(
128 : "[InsCollAlgFactory] [InsBroadcastSoleExecutor] Rank[%d], CalcRes with detouring enabled for Orchestrate.",
129 : myRank_);
130 0 : CHK_RET(tempAlg.CalcResDetour(rankGraph, tempResReq));
131 : } else {
132 0 : HCCL_DEBUG(
133 : "[InsCollAlgFactory] [InsBroadcastSoleExecutor] Rank[%d], CalcRes with detouring disabled for Orchestrate.",
134 : myRank_);
135 0 : CHK_RET(tempAlg.CalcRes(tempResReq));
136 : }
137 : // 申请算法模板所需资源
138 0 : CHK_RET(InitQueue(tempResReq.queNum, requiredQue_));
139 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, tempResLinks_));
140 :
141 0 : if (opMode_ == OpMode::OFFLOAD) {
142 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OFFLOAD Mode for HOST.", myRank_);
143 0 : CHK_RET(OrchestrateOffload(tempAlg));
144 : } else { // OPBASE
145 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OPBASE Mode for HOST.", myRank_);
146 0 : CHK_RET(OrchestrateOpbase(tempAlg));
147 : }
148 :
149 0 : return HcclResult::HCCL_SUCCESS;
150 0 : }
151 :
152 : // 算子执行aicpu接口
153 : template <typename AlgTopoMatch, typename InsAlgTemplate>
154 0 : HcclResult InsBroadcastSoleExecutor<AlgTopoMatch, InsAlgTemplate>::Orchestrate(
155 : const AlgTopoInfo& topoInfo, const CollAlgOperator& op, const CollAlgParams& params, ConnectedLinkMgr* linkMgr,
156 : InsQuePtr insQue)
157 : {
158 0 : HCCL_INFO("[InsCollAlgFactory] [InsBroadcastSoleExecutor] [InsReduceSoleExecutor] AiCpu Orchestrate begins.");
159 : // 参数校验和初始化
160 0 : CHK_RET(Init(op, params, insQue));
161 :
162 : // soleEsecutor 只支持单层拓扑, 所以只取第 0 级通信域的信息
163 0 : vTopo_ = topoInfo.vTopo[0]; // 本通信域内的通信平面
164 0 : virtRankMap_ = topoInfo.virtRankMap[0]; // 本通信域内的 rank 映射表
165 0 : virtRanks_ = topoInfo.virtRanks[0]; // 本通信域内的 rank 集合
166 0 : dataTypeSize_ = DataTypeSizeGet(dataType_);
167 0 : dataSize_ = dataCount_ * dataTypeSize_;
168 0 : CHK_PRT_RET(
169 : dataTypeSize_ == 0,
170 : HCCL_ERROR("Broadcast_[CollAlgFactory] Rank [%d], Invalid dataTypeSize_ [%u].", myRank_, dataTypeSize_),
171 : HcclResult::HCCL_E_INTERNAL);
172 :
173 : // 实例化算法模板类
174 0 : HCCL_DEBUG(
175 : "Broadcast_[InsReduceSoleExecutor] Rank[%d], Init insAlgTemplate with rankSize [%u] and dmaMode [%s].", myRank_,
176 : rankSize_, dmaMode_.Describe().c_str());
177 0 : InsAlgTemplate tempAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
178 0 : tempAlg.SetDmaMode(dmaMode_);
179 0 : tempAlg.SetCollOp(op); // CCU template需要传递op信息
180 0 : tempAlg.SetRoot(root_);
181 :
182 : // 计算算法模板所需资源
183 0 : AlgTempResReq tempResReq;
184 0 : if (enableDetour_) {
185 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsBroadcastSoleExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
186 0 : CHK_RET(tempAlg.CalcResDetour(linkMgr, tempResReq));
187 : } else {
188 0 : HCCL_DEBUG(
189 : "[InsCollAlgFactory] [InsBroadcastSoleExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
190 0 : CHK_RET(tempAlg.CalcRes(tempResReq));
191 : }
192 :
193 : // 申请算法模板所需资源
194 0 : CHK_RET(InitQueue(tempResReq.queNum, requiredQue_));
195 0 : CHK_RET(PrepResLinks(myRank_, tempResReq.links, linkMgr, tempResLinks_));
196 :
197 0 : if (opMode_ == OpMode::OFFLOAD) {
198 0 : HCCL_DEBUG(
199 : "[InsReduceSoleExecutor] Rank[%d], Generating Instruction Queues in OFFLOAD Mode for AICPU.", myRank_);
200 0 : CHK_RET(OrchestrateOffload(tempAlg));
201 : } else { // OPBASE
202 0 : HCCL_DEBUG(
203 : "[InsReduceSoleExecutor] Rank[%d], Generating Instruction Queues in OPBASE Mode for AICPU.", myRank_);
204 0 : CHK_RET(OrchestrateOpbase(tempAlg));
205 : }
206 :
207 0 : return HcclResult::HCCL_SUCCESS;
208 0 : }
209 :
210 : template <typename AlgTopoMatch, typename InsAlgTemplate>
211 0 : HcclResult InsBroadcastSoleExecutor<AlgTopoMatch, InsAlgTemplate>::OrchestrateOpbase(InsAlgTemplate& tempAlg)
212 : {
213 0 : BuffInfo buffInfo;
214 0 : buffInfo.inBuffType = BufferType::SCRATCH;
215 0 : buffInfo.outBuffType = BufferType::SCRATCH;
216 0 : buffInfo.inBuffBaseOff = 0;
217 0 : buffInfo.outBuffBaseOff = 0;
218 :
219 : // 基本参数配置
220 0 : AllignInfo allignInfo = {enableAllign_, allignSize_, dataType_};
221 0 : TempFuncs tempFuncs;
222 0 : tempFuncs.opMode = opMode_;
223 0 : tempFuncs.enableCounterNotify = IsEnableCounterNotify();
224 0 : tempFuncs.isForepart = true; // Usr Buff to CCL Buff required
225 0 : tempFuncs.isBottom = true; // CCL Buff to Usr Buff required
226 :
227 : // 根据CCL Buffer 大小，计算出一轮中最多能输出多少数据
228 0 : u64 maxLoopOutputCount
229 0 : = std::min(static_cast<u64>(maxTmpMemSize_), static_cast<u64>(UB_MAX_DATA_SIZE)) / dataTypeSize_;
230 0 : CHK_PRT_RET(
231 : maxLoopOutputCount == 0, HCCL_ERROR("[InsReduceSoleExecutor] maxLoopOutputCount is zero."),
232 : HcclResult::HCCL_E_PARA);
233 0 : u64 dataSize = dataCount_ * dataTypeSize_;
234 :
235 : // offload模式根据UB_MAX_DATA_SIZE，计算出一轮中最多能输出多少数据
236 0 : u64 maxLoopOutputSize = maxLoopOutputCount * dataTypeSize_;
237 :
238 0 : u64 loopTimes = dataSize / maxLoopOutputSize + static_cast<u64>(dataSize % maxLoopOutputSize != 0);
239 :
240 0 : for (u32 loop = 0; loop < loopTimes; loop++) {
241 0 : u64 currloopOffset = loop * maxLoopOutputSize;
242 0 : u64 currSize = (loop == (loopTimes - 1)) ? dataSize - currloopOffset : maxLoopOutputSize;
243 0 : UsrData usrData;
244 0 : usrData.usrInSlices.emplace_back(BufferType::INPUT, currloopOffset, currSize);
245 0 : usrData.scratchInSlices.emplace_back(BufferType::SCRATCH, 0, currSize);
246 0 : usrData.scratchOutSlices.emplace_back(BufferType::SCRATCH, 0, currSize);
247 0 : usrData.usrOutSlices.emplace_back(BufferType::INPUT, currloopOffset, currSize);
248 0 : tempFuncs.usrData = usrData;
249 :
250 0 : RankSliceInfo sliceInfoVec;
251 0 : CHK_RET(tempAlg.CalcSliceInfo(allignInfo, currSize, sliceInfoVec));
252 0 : CHK_RET(tempAlg.Run(tempFuncs, sliceInfoVec, buffInfo, tempResLinks_, requiredQue_));
253 0 : HCCL_DEBUG(
254 : "[InsCollAlgFactory] [InsBroadcastSoleExecutor] Rank[%d], done generating instruction queues, "
255 : "currSize[%llu], currOffset[%llu].",
256 : myRank_, currSize, currloopOffset);
257 : }
258 :
259 0 : return HcclResult::HCCL_SUCCESS;
260 0 : }
261 :
262 : template <typename AlgTopoMatch, typename InsAlgTemplate>
263 0 : HcclResult InsBroadcastSoleExecutor<AlgTopoMatch, InsAlgTemplate>::OrchestrateOffload(InsAlgTemplate& tempAlg)
264 : {
265 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], done calculating slice information.", myRank_);
266 :
267 0 : BuffInfo buffInfo;
268 0 : buffInfo.inBuffType = BufferType::INPUT;
269 0 : buffInfo.outBuffType = BufferType::INPUT;
270 0 : buffInfo.inBuffBaseOff = 0;
271 0 : buffInfo.outBuffBaseOff = 0;
272 0 : HCCL_DEBUG(
273 : "[InsCollAlgFactory] Rank[%d], input buffer type [%s], output buffer type [%s], input buffer base "
274 : "offset [%u], output buffer base offset [%u].",
275 : myRank_, buffInfo.inBuffType.Describe().c_str(), buffInfo.outBuffType.Describe().c_str(),
276 : buffInfo.inBuffBaseOff, buffInfo.outBuffBaseOff);
277 :
278 0 : AllignInfo allignInfo = {enableAllign_, allignSize_, dataType_};
279 0 : TempFuncs tempFuncs;
280 0 : tempFuncs.opMode = opMode_;
281 0 : tempFuncs.enableCounterNotify = IsEnableCounterNotify();
282 0 : tempFuncs.isForepart = true; // Usr Buff to CCL Buff required
283 0 : tempFuncs.isBottom = true; // CCL Buff to Usr Buff required
284 :
285 0 : u64 dataSize = dataCount_ * dataTypeSize_;
286 :
287 : // offload模式根据UB_MAX_DATA_SIZE，计算出一轮中最多能输出多少数据
288 0 : u64 maxLoopOutputSize = (UB_MAX_DATA_SIZE / dataTypeSize_) * dataTypeSize_;
289 :
290 0 : u64 loopTimes = dataSize / maxLoopOutputSize + static_cast<u64>(dataSize % maxLoopOutputSize != 0);
291 :
292 0 : for (u32 loop = 0; loop < loopTimes; loop++) {
293 0 : u64 currloopOffset = loop * maxLoopOutputSize;
294 0 : u64 currSize = (loop == (loopTimes - 1)) ? dataSize - currloopOffset : maxLoopOutputSize;
295 0 : UsrData usrData;
296 0 : usrData.usrInSlices.emplace_back(BufferType::INPUT, currloopOffset, currSize);
297 0 : usrData.scratchInSlices.emplace_back(BufferType::SCRATCH, 0, currSize);
298 0 : usrData.scratchOutSlices.emplace_back(BufferType::SCRATCH, 0, currSize);
299 0 : usrData.usrOutSlices.emplace_back(BufferType::INPUT, currloopOffset, currSize);
300 0 : tempFuncs.usrData = usrData;
301 0 : RankSliceInfo sliceInfoVec;
302 0 : CHK_RET(tempAlg.CalcSliceInfo(allignInfo, currSize, sliceInfoVec));
303 0 : CHK_RET(tempAlg.Run(tempFuncs, sliceInfoVec, buffInfo, tempResLinks_, requiredQue_));
304 0 : HCCL_DEBUG(
305 : "[InsCollAlgFactory] [InsBroadcastSoleExecutor] Rank[%d], done generating instruction queues, "
306 : "currSize[%llu], currOffset[%llu].",
307 : myRank_, currSize, currloopOffset);
308 : }
309 :
310 0 : return HcclResult::HCCL_SUCCESS;
311 0 : }
312 :
313 : #ifndef CCL_KERNEL_AICPU
314 : INS_REGISTER_IMPL_BY_TEMP(
315 : OpType::BROADCAST, CcuBroadcastMesh1D, InsBroadcastSoleExecutor, TopoMatchMesh, CcuTempBroadcastMesh1D);
316 : INS_REGISTER_IMPL_BY_TEMP(
317 : OpType::BROADCAST, CcuBroadcastMesh2D, InsBroadcastSoleExecutor, TopoMatchConcurrMesh, CcuTempBroadcastMesh2D);
318 : #endif
319 : } // namespace Hccl
|
