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