Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 : #include "log.h"
11 : #include "ins_coll_alg_registry.h"
12 : #include "topo_match_mesh_nhr.h"
13 : #include "alg_data_trans_wrapper.h"
14 :
15 : #include "ins_temp_scatter_mesh_1d.h"
16 : #include "ins_temp_scatter_nhr.h"
17 : #include "ins_temp_all_gather_nhr.h"
18 : #include "ins_temp_all_gather_mesh.h"
19 : #include "ins_broadcast_parallel_aicpu_executor.h"
20 :
21 : namespace Hccl {
22 :
23 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1, typename InsAlgTemplate2,
24 : typename InsAlgTemplate3>
25 0 : HcclResult InsBroadcastParallelAiCpuExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1, InsAlgTemplate2,
26 : InsAlgTemplate3>::PreCalcRes(const RankGraph *rankGraph, AlgTempResReq &resReqIntraScatter,
27 : AlgTempResReq &resReqInterScatter, AlgTempResReq &resReqIntraAllGather, AlgTempResReq &resReqInterAllGather)
28 : {
29 : // Topo Match
30 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
31 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
32 :
33 : // 计算localRankSize
34 0 : CHK_RET(CalcLocalRankSize());
35 :
36 : // 实例化算法模板类
37 0 : InsAlgTemplate0 intraScatterTempAlg(myRank_, intraLocalRankSize_, vTopo_[0], virtRankMap_[0]);
38 0 : InsAlgTemplate1 interScatterTempAlg(myRank_, interLocalRankSize_, vTopo_[1], virtRankMap_[1]);
39 0 : InsAlgTemplate2 intraAllGatherTempAlg(myRank_, intraLocalRankSize_, vTopo_[0], virtRankMap_[0]);
40 0 : InsAlgTemplate3 interAllGatherTempAlg(myRank_, interLocalRankSize_, vTopo_[1], virtRankMap_[1]);
41 0 : std::vector<map<u32, u32>> rank2PathNumMap;
42 0 : HCCL_INFO("[InsBroadcastParallelAiCpuExecutor] CalcRes SetPathNumMap");
43 0 : CHK_RET(SetPathNumMapByRankGraphMultiLevel(rankGraph, virtRanks_, myRank_, rank2PathNumMap));
44 0 : intraAllGatherTempAlg.setPathNumMap(rank2PathNumMap[0]);
45 0 : interAllGatherTempAlg.setPathNumMap(rank2PathNumMap[1]);
46 0 : intraScatterTempAlg.setPathNumMap(rank2PathNumMap[0]);
47 0 : interScatterTempAlg.setPathNumMap(rank2PathNumMap[1]);
48 0 : CHK_RET(CalcSingleAlgRes(intraScatterTempAlg, interScatterTempAlg, intraAllGatherTempAlg, interAllGatherTempAlg,
49 : rankGraph, resReqIntraScatter, resReqInterScatter, resReqIntraAllGather, resReqInterAllGather));
50 0 : return HcclResult::HCCL_SUCCESS;
51 0 : }
52 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1, typename InsAlgTemplate2,
53 : typename InsAlgTemplate3>
54 : template <typename T>
55 0 : HcclResult InsBroadcastParallelAiCpuExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1, InsAlgTemplate2,
56 : InsAlgTemplate3>::CalcSingleAlgRes(InsAlgTemplate0 &intraScatter, InsAlgTemplate1 &interScatter,
57 : InsAlgTemplate2 &intraAllGather, InsAlgTemplate3 &interAllGather, T *type, AlgTempResReq &resReqIntraScatter,
58 : AlgTempResReq &resReqInterScatter, AlgTempResReq &resReqIntraAllGather, AlgTempResReq &resReqInterAllGather) const
59 : {
60 0 : if (enableDetour_) {
61 0 : HCCL_DEBUG("[%s] Rank[%d], CalcRes with detouring enabled.", __func__, myRank_);
62 0 : CHK_RET(intraScatter.CalcResDetour(type, resReqIntraScatter));
63 0 : CHK_RET(intraAllGather.CalcResDetour(type, resReqIntraAllGather));
64 : } else {
65 0 : HCCL_DEBUG("[%s] Rank[%d], CalcRes with detouring disabled.", __func__, myRank_);
66 0 : CHK_RET(intraScatter.CalcRes(resReqIntraScatter));
67 0 : CHK_RET(intraAllGather.CalcRes(resReqIntraAllGather));
68 : }
69 0 : CHK_RET(interScatter.CalcRes(resReqInterScatter));
70 0 : CHK_RET(interAllGather.CalcRes(resReqInterAllGather));
71 0 : return HcclResult::HCCL_SUCCESS;
72 : }
73 :
74 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1, typename InsAlgTemplate2,
75 : typename InsAlgTemplate3>
76 0 : HcclResult InsBroadcastParallelAiCpuExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1, InsAlgTemplate2,
77 : InsAlgTemplate3>::CalcRes(const RankGraph *rankGraph, CollAlgResReq &algResReq)
78 : {
79 0 : HCCL_INFO("[InsBroadcastParallelAiCpuExecutor] CalcRes start, rank[%d]", myRank_);
80 :
81 : // 计算和准备Queue资源
82 0 : AlgTempResReq resReqIntraScatter;
83 0 : AlgTempResReq resReqInterScatter;
84 0 : AlgTempResReq resReqIntraAllGather;
85 0 : AlgTempResReq resReqInterAllGather;
86 :
87 0 : CHK_RET(PreCalcRes(rankGraph, resReqIntraScatter, resReqInterScatter, resReqIntraAllGather, resReqInterAllGather));
88 :
89 0 : algResReq.topoInfo.UpdateMultiLevelTopo(virtRanks_, virtRankMap_, vTopo_);
90 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, resReqIntraScatter.links, algResReq.levelRankPairs));
91 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, resReqInterScatter.links, algResReq.levelRankPairs));
92 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, resReqIntraAllGather.links, algResReq.levelRankPairs));
93 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, resReqInterAllGather.links, algResReq.levelRankPairs));
94 0 : u32 intraQueNum = max(resReqIntraScatter.queNum, resReqIntraAllGather.queNum);
95 0 : u32 interQueNum = max(resReqInterScatter.queNum, resReqInterAllGather.queNum);
96 :
97 0 : algResReq.primQueueNum = intraQueNum + interQueNum;
98 :
99 0 : std::vector<std::tuple<QId, QId, u32>> notifyRequests;
100 :
101 0 : CHK_RET(CalcParallelNotifyReq(algResReq.primQueueNum, resReqIntraAllGather.queNum, algResReq.queueNotifys));
102 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqIntraScatter.links, algResReq.links));
103 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqInterScatter.links, algResReq.links));
104 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqIntraAllGather.links, algResReq.links));
105 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqInterAllGather.links, algResReq.links));
106 :
107 0 : HCCL_INFO(
108 : "[InsBroadcastParallelAiCpuExecutor] CalcRes end, rank[%d], required total que num [%u], que notify num [%u]",
109 : myRank_, algResReq.primQueueNum, algResReq.queueNotifys.size());
110 :
111 0 : return HcclResult::HCCL_SUCCESS;
112 0 : }
113 :
114 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1, typename InsAlgTemplate2,
115 : typename InsAlgTemplate3>
116 0 : HcclResult InsBroadcastParallelAiCpuExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1, InsAlgTemplate2,
117 : InsAlgTemplate3>::CalcResOffload(const RankGraph *rankGraph, const u64 &dataSize, CollOffloadOpResReq &resReq)
118 : {
119 0 : HCCL_INFO("[InsBroadcastParallelAiCpuExecutor] CalcResOffload start, rank[%d]", myRank_);
120 :
121 : (void)dataSize;
122 0 : u64 scratchMemSize = 200 * 1024 * 1024;
123 0 : resReq.requiredScratchMemSize = scratchMemSize; // 200MB
124 :
125 : // 计算和准备Queue资源
126 0 : AlgTempResReq resReqIntraScatter;
127 0 : AlgTempResReq resReqInterScatter;
128 0 : AlgTempResReq resReqIntraAllGather;
129 0 : AlgTempResReq resReqInterAllGather;
130 :
131 0 : CHK_RET(PreCalcRes(rankGraph, resReqIntraScatter, resReqInterScatter, resReqIntraAllGather, resReqInterAllGather));
132 0 : resReq.requiredSubQueNum = resReqIntraScatter.streamNum + resReqInterScatter.streamNum
133 0 : + resReqIntraAllGather.streamNum + resReqInterAllGather.streamNum - 1;
134 :
135 0 : HCCL_INFO("[InsBroadcastParallelAiCpuExecutor] CalcResOffload end, rank[%d], required sub que num is [%u]", myRank_,
136 : resReq.requiredSubQueNum);
137 :
138 0 : return HcclResult::HCCL_SUCCESS;
139 0 : }
140 :
141 : // Host展开
142 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1, typename InsAlgTemplate2,
143 : typename InsAlgTemplate3>
144 0 : HcclResult InsBroadcastParallelAiCpuExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1, InsAlgTemplate2,
145 : InsAlgTemplate3>::Orchestrate(const RankGraph *rankGraph, const CollAlgOperator &op, const CollAlgParams ¶ms,
146 : InsQuePtr insQue)
147 : {
148 0 : HCCL_INFO("[InsBroadcastParallelAiCpuExecutor] Host orchestrate begins.");
149 :
150 : // 初始化参数
151 0 : CHK_RET(Init(op, params, insQue));
152 :
153 : // 获取算法Topo信息
154 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
155 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
156 :
157 : // 计算localRankSize和localRoot
158 0 : CHK_RET(CalcLocalRankSize());
159 0 : CHK_RET(CalcLocalRoot());
160 :
161 : // 实例化算法模板类
162 0 : InsAlgTemplate0 intraScatterTempAlg(myRank_, intraLocalRankSize_, vTopo_[0], virtRankMap_[0]);
163 0 : InsAlgTemplate1 interScatterTempAlg(myRank_, interLocalRankSize_, vTopo_[1], virtRankMap_[1]);
164 0 : InsAlgTemplate2 intraAllGatherTempAlg(myRank_, intraLocalRankSize_, vTopo_[0], virtRankMap_[0]);
165 0 : InsAlgTemplate3 interAllGatherTempAlg(myRank_, interLocalRankSize_, vTopo_[1], virtRankMap_[1]);
166 0 : std::vector<map<u32, u32>> rank2PathNumMap;
167 0 : HCCL_INFO("[InsBroadcastParallelAiCpuExecutor] Orchestrate SetPathNumMap");
168 0 : CHK_RET(SetPathNumMapByRankGraphMultiLevel(rankGraph, virtRanks_, myRank_, rank2PathNumMap));
169 0 : intraAllGatherTempAlg.setPathNumMap(rank2PathNumMap[0]);
170 0 : interAllGatherTempAlg.setPathNumMap(rank2PathNumMap[1]);
171 0 : intraScatterTempAlg.setPathNumMap(rank2PathNumMap[0]);
172 0 : interScatterTempAlg.setPathNumMap(rank2PathNumMap[1]);
173 : // 传入Template参数
174 0 : AlgTemplateInitPara(op, intraScatterTempAlg, interScatterTempAlg, intraAllGatherTempAlg, interAllGatherTempAlg);
175 : // 计算算法模板所需资源
176 0 : CHK_RET(PrepareResForTemplate(
177 : rankGraph, intraScatterTempAlg, interScatterTempAlg, intraAllGatherTempAlg, interAllGatherTempAlg));
178 :
179 : // 算法展开
180 0 : CHK_RET(GenInsQues(intraScatterTempAlg, interScatterTempAlg, intraAllGatherTempAlg, interAllGatherTempAlg));
181 :
182 0 : HCCL_INFO("[InsBroadcastParallelAiCpuExecutor] Host orchestrate success.");
183 0 : return HcclResult::HCCL_SUCCESS;
184 0 : }
185 :
186 : // Aicpu展开
187 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1, typename InsAlgTemplate2,
188 : typename InsAlgTemplate3>
189 0 : HcclResult InsBroadcastParallelAiCpuExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1, InsAlgTemplate2,
190 : InsAlgTemplate3>::Orchestrate(const AlgTopoInfo &topoInfo, const CollAlgOperator &op, const CollAlgParams ¶ms,
191 : ConnectedLinkMgr *linkMgr, InsQuePtr insQue)
192 : {
193 0 : HCCL_INFO("[InsBroadcastParallelAiCpuExecutor] Aicpu orchestrate begins.");
194 :
195 : // 初始化参数
196 0 : CHK_RET(Init(op, params, insQue));
197 :
198 : // 获取算法Topo信息
199 0 : vTopo_ = topoInfo.vTopo; // 本通信域内的通信平面
200 0 : virtRanks_ = topoInfo.virtRanks; // 本通信域内的 rank 集合
201 0 : virtRankMap_ = topoInfo.virtRankMap; // 本通信域内的 rank 映射表
202 :
203 : // 计算localRankSize和localRoot
204 0 : CHK_RET(CalcLocalRankSize());
205 0 : CHK_RET(CalcLocalRoot());
206 :
207 : // 实例化算法模板类
208 0 : InsAlgTemplate0 intraScatterTempAlg(myRank_, intraLocalRankSize_, vTopo_[0], virtRankMap_[0]);
209 0 : InsAlgTemplate1 interScatterTempAlg(myRank_, interLocalRankSize_, vTopo_[1], virtRankMap_[1]);
210 0 : InsAlgTemplate2 intraAllGatherTempAlg(myRank_, intraLocalRankSize_, vTopo_[0], virtRankMap_[0]);
211 0 : InsAlgTemplate3 interAllGatherTempAlg(myRank_, interLocalRankSize_, vTopo_[1], virtRankMap_[1]);
212 :
213 : // 传入Template参数
214 0 : AlgTemplateInitPara(op, intraScatterTempAlg, interScatterTempAlg, intraAllGatherTempAlg, interAllGatherTempAlg);
215 0 : std::vector<std::map<u32, u32>> rank2PathNumMap;
216 0 : CHK_RET(SetPathNumMapByLinkMgrMultiLevel(linkMgr, virtRanks_, myRank_, rank2PathNumMap));
217 0 : intraAllGatherTempAlg.setPathNumMap(rank2PathNumMap[0]);
218 0 : interAllGatherTempAlg.setPathNumMap(rank2PathNumMap[1]);
219 0 : intraScatterTempAlg.setPathNumMap(rank2PathNumMap[0]);
220 0 : interScatterTempAlg.setPathNumMap(rank2PathNumMap[1]);
221 : // 计算算法模板所需资源
222 0 : CHK_RET(PrepareResForTemplate(
223 : linkMgr, intraScatterTempAlg, interScatterTempAlg, intraAllGatherTempAlg, interAllGatherTempAlg));
224 :
225 : // 算法展开
226 0 : CHK_RET(GenInsQues(intraScatterTempAlg, interScatterTempAlg, intraAllGatherTempAlg, interAllGatherTempAlg));
227 :
228 0 : HCCL_INFO("[InsBroadcastParallelAiCpuExecutor] Aicpu orchestrate success.");
229 0 : return HcclResult::HCCL_SUCCESS;
230 0 : }
231 :
232 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1, typename InsAlgTemplate2,
233 : typename InsAlgTemplate3>
234 : template <typename T>
235 0 : HcclResult InsBroadcastParallelAiCpuExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1, InsAlgTemplate2,
236 : InsAlgTemplate3>::PrepareRes(T *type, AlgTempResReq &resReqIntraScatter, AlgTempResReq &resReqInterScatter,
237 : AlgTempResReq &resReqIntraAllGather, AlgTempResReq &resReqInterAllGather)
238 : {
239 : // 申请算法模板所需资源
240 0 : if (resReqIntraScatter.queNum == 0 || resReqInterScatter.queNum == 0 || resReqIntraAllGather.queNum == 0
241 0 : || resReqInterAllGather.queNum == 0) {
242 0 : HCCL_ERROR("queNum must larger than 0.");
243 0 : return HcclResult::HCCL_E_INTERNAL;
244 : }
245 0 : u32 intraQueNum = max(resReqIntraScatter.queNum, resReqIntraAllGather.queNum);
246 0 : u32 interQueNum = max(resReqInterScatter.queNum, resReqInterAllGather.queNum);
247 :
248 0 : u32 totalQueueNum = intraQueNum + interQueNum;
249 0 : CHK_RET(InitQueue(totalQueueNum, requiredQue_));
250 0 : for (u32 i = 0; i < requiredQue_.size(); i++) {
251 0 : if (i < intraQueNum) {
252 0 : intraQue_.push_back(requiredQue_.at(i));
253 : } else {
254 0 : interQue_.push_back(requiredQue_.at(i));
255 : }
256 : }
257 :
258 : // 每个算法的第0条流用于同步
259 0 : syncQueues_.emplace_back(intraQue_.at(0));
260 0 : syncQueues_.emplace_back(interQue_.at(0));
261 :
262 0 : CHK_RET(WrapPrepResLinks(type, resReqIntraScatter.links, scatterIntraLinks_));
263 0 : CHK_RET(WrapPrepResLinks(type, resReqInterScatter.links, scatterInterLinks_));
264 0 : CHK_RET(WrapPrepResLinks(type, resReqIntraAllGather.links, allGatherIntraLinks_));
265 0 : CHK_RET(WrapPrepResLinks(type, resReqInterAllGather.links, allGatherInterLinks_));
266 0 : HCCL_INFO("[InsBroadcastParallelAiCpuExecutor] scatterIntraLinks size[%zu], scatterInterLinks size[%zu], "
267 : "allGatherIntraLinks size[%zu], allGatherInterLinks size[%zu]",
268 : scatterIntraLinks_.size(), scatterInterLinks_.size(), allGatherIntraLinks_.size(), allGatherInterLinks_.size());
269 0 : return HcclResult::HCCL_SUCCESS;
270 : }
271 :
272 : // Host
273 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1, typename InsAlgTemplate2,
274 : typename InsAlgTemplate3>
275 0 : HcclResult InsBroadcastParallelAiCpuExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1, InsAlgTemplate2,
276 : InsAlgTemplate3>::PrepareResForTemplate(const RankGraph *rankGraph, InsAlgTemplate0 &intraScatterTempAlg,
277 : InsAlgTemplate1 &interScatterTempAlg, InsAlgTemplate2 &intraAllGatherTempAlg,
278 : InsAlgTemplate3 &interAllGatherTempAlg)
279 : {
280 0 : AlgTempResReq resReqIntraScatter;
281 0 : AlgTempResReq resReqInterScatter;
282 0 : AlgTempResReq resReqIntraAllGather;
283 0 : AlgTempResReq resReqInterAllGather;
284 0 : CHK_RET(CalcSingleAlgRes(intraScatterTempAlg, interScatterTempAlg, intraAllGatherTempAlg, interAllGatherTempAlg,
285 : rankGraph, resReqIntraScatter, resReqInterScatter, resReqIntraAllGather, resReqInterAllGather));
286 0 : CHK_RET(PrepareRes(rankGraph, resReqIntraScatter, resReqInterScatter, resReqIntraAllGather, resReqInterAllGather));
287 :
288 0 : return HcclResult::HCCL_SUCCESS;
289 0 : }
290 :
291 : // Aicpu
292 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1, typename InsAlgTemplate2,
293 : typename InsAlgTemplate3>
294 0 : HcclResult InsBroadcastParallelAiCpuExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1, InsAlgTemplate2,
295 : InsAlgTemplate3>::PrepareResForTemplate(ConnectedLinkMgr *linkMgr, InsAlgTemplate0 &intraScatterTempAlg,
296 : InsAlgTemplate1 &interScatterTempAlg, InsAlgTemplate2 &intraAllGatherTempAlg,
297 : InsAlgTemplate3 &interAllGatherTempAlg)
298 : {
299 0 : AlgTempResReq resReqIntraScatter;
300 0 : AlgTempResReq resReqInterScatter;
301 0 : AlgTempResReq resReqIntraAllGather;
302 0 : AlgTempResReq resReqInterAllGather;
303 0 : CHK_RET(CalcSingleAlgRes(intraScatterTempAlg, interScatterTempAlg, intraAllGatherTempAlg, interAllGatherTempAlg,
304 : linkMgr, resReqIntraScatter, resReqInterScatter, resReqIntraAllGather, resReqInterAllGather));
305 0 : CHK_RET(PrepareRes(linkMgr, resReqIntraScatter, resReqInterScatter, resReqIntraAllGather, resReqInterAllGather));
306 :
307 0 : return HcclResult::HCCL_SUCCESS;
308 0 : }
309 :
310 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1, typename InsAlgTemplate2,
311 : typename InsAlgTemplate3>
312 0 : void InsBroadcastParallelAiCpuExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1, InsAlgTemplate2,
313 : InsAlgTemplate3>::CalcSlice(std::vector<double> &splitDataSize, float scratchMaxMultiple, SliceConfig &slice)
314 : {
315 : // 数据切分
316 0 : u64 sliceCount = std::min(static_cast<u64>(UB_MAX_DATA_SIZE) / dataTypeSize_, dataCount_);
317 0 : if (scratchMaxMultiple > 0 && maxTmpMemSize_ > 0) {
318 0 : u64 scratchCount = maxTmpMemSize_ / dataTypeSize_; // 按照count来切分
319 : sliceCount
320 0 : = min(sliceCount, static_cast<u64>(static_cast<double>(scratchCount) / scratchMaxMultiple)); // 向下取整，防止Scratch溢出
321 : }
322 : /* 刷新slicecout0 和slicecout1确保是interLocalRankSize_ * intraLocalRankSize_整倍数 */
323 0 : u64 sliceCountPart0 = static_cast<u64>(sliceCount * splitDataSize.at(0));
324 : sliceCountPart0
325 0 : = (sliceCountPart0 / interLocalRankSize_ / intraLocalRankSize_) * interLocalRankSize_ * intraLocalRankSize_;
326 0 : u64 sliceCountPart1 = static_cast<u64>(sliceCount * splitDataSize.at(1));
327 : sliceCountPart1
328 0 : = (sliceCountPart1 / interLocalRankSize_ / intraLocalRankSize_) * interLocalRankSize_ * intraLocalRankSize_;
329 0 : sliceCount = sliceCountPart0 + sliceCountPart1;
330 : // 计算循环次数, 如果sliceCountPart0和liceCountPart1为0说明只有一块数据都是尾块
331 0 : u32 loopTimes = sliceCount == 0 ? 1 : (dataCount_ + sliceCount - 1) / sliceCount;
332 : // 计算尾块
333 0 : u64 finalSliceCount = dataCount_ - (loopTimes - 1) * sliceCount;
334 0 : u64 finalTailCount = finalSliceCount % (interLocalRankSize_ * intraLocalRankSize_);
335 0 : u64 finalSliceCountPart1 = static_cast<u64>(finalSliceCount * splitDataSize.at(1));
336 : // 刷新slicecout0 和slicecout1确保是interLocalRankSize_ * intraLocalRankSize_整倍数
337 0 : finalSliceCountPart1 = (finalSliceCountPart1 / interLocalRankSize_ / intraLocalRankSize_) * interLocalRankSize_
338 0 : * intraLocalRankSize_;
339 0 : u64 finalSliceCountPart0 = finalSliceCount - finalSliceCountPart1 - finalTailCount;
340 0 : slice.loopTimes = loopTimes;
341 0 : slice.sliceCount = sliceCount;
342 0 : slice.sliceCountPart0 = sliceCountPart0;
343 0 : slice.sliceCountPart1 = sliceCountPart1;
344 0 : slice.finalSliceCount = finalSliceCount;
345 0 : slice.finalSliceCountPart0 = finalSliceCountPart0;
346 0 : slice.finalSliceCountPart1 = finalSliceCountPart1;
347 : // 结构体定义中必须确保finalTailCountPart0和finalTailCountPart1初始化为0
348 0 : (finalSliceCountPart0 < finalSliceCountPart1 ? slice.finalTailCountPart0 : slice.finalTailCountPart1) = finalTailCount;
349 0 : return;
350 : }
351 :
352 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1, typename InsAlgTemplate2,
353 : typename InsAlgTemplate3>
354 0 : HcclResult InsBroadcastParallelAiCpuExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1, InsAlgTemplate2,
355 : InsAlgTemplate3>::StageProcess(DataParameters &dataParameters, std::vector<StageProcAlgPara> &algParaVec)
356 : {
357 0 : TemplateDataParams tempAlgParams;
358 0 : TempFuncs tempFuncs;
359 0 : tempFuncs.opMode = opMode_;
360 0 : tempFuncs.enableCounterNotify = false;
361 0 : for (u32 step = 0; step < algParaVec.size(); step++) {
362 0 : bool isFirst = (step == 0);
363 : // 先处理part0数据
364 0 : CHK_RET(PreSyncQues(syncQueues_, 0));
365 : // 第一步的时候server间topo包含root_的rank进行展开，其它rank不展开
366 0 : u64 sliceSizePart0 = max(dataParameters.sliceSize.at(0).at(step), dataParameters.tailSize.at(0).at(step));
367 0 : if ((!isFirst || interLocalRoot_ == root_) && (sliceSizePart0 > 0)) {
368 0 : GenDataParamsStage(0, step, dataParameters, tempAlgParams);
369 0 : CHK_RET(algParaVec.at(step).part0FuncPtr(
370 : tempFuncs, tempAlgParams, algParaVec.at(step).part0links, algParaVec.at(step).part0Que));
371 : }
372 : // 再处理part1数据, 第一步的时候server内topo包含root_的rank进行展开，其它rank不展开
373 0 : u64 sliceSizePart1 = max(dataParameters.sliceSize.at(1).at(step), dataParameters.tailSize.at(1).at(step));
374 0 : if ((!isFirst || intraLocalRoot_ == root_) && sliceSizePart1 > 0) {
375 : // 数据1的server内的scatter算法
376 0 : GenDataParamsStage(1, step, dataParameters, tempAlgParams);
377 0 : CHK_RET(algParaVec.at(step).part1FuncPtr(
378 : tempFuncs, tempAlgParams, algParaVec.at(step).part1links, algParaVec.at(step).part1Que));
379 : }
380 0 : CHK_RET(PostSyncQues(syncQueues_, 0));
381 : }
382 0 : return HcclResult::HCCL_SUCCESS;
383 0 : }
384 :
385 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1, typename InsAlgTemplate2,
386 : typename InsAlgTemplate3>
387 0 : HcclResult InsBroadcastParallelAiCpuExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1, InsAlgTemplate2,
388 : InsAlgTemplate3>::GenInsQues(InsAlgTemplate0 &intraScatterTempAlg, InsAlgTemplate1 &interScatterTempAlg,
389 : InsAlgTemplate2 &intraAllGatherTempAlg, InsAlgTemplate3 &interAllGatherTempAlg)
390 : {
391 0 : LogAlgInfo(intraScatterTempAlg, interScatterTempAlg, intraAllGatherTempAlg, interAllGatherTempAlg);
392 :
393 0 : std::vector<double> dataSplitSize;
394 0 : GetParallelDataSplit(dataSplitSize);
395 : ScratchMultiple scratchMultiple;
396 0 : CalcScratchMultiple(dataSplitSize, scratchMultiple, intraScatterTempAlg, interScatterTempAlg, intraAllGatherTempAlg,
397 : interAllGatherTempAlg);
398 0 : SliceConfig slice;
399 0 : CalcSlice(dataSplitSize, scratchMultiple.maxMultiple, slice);
400 :
401 0 : std::vector<StageProcAlgPara> stageProcAlgParaVec;
402 0 : InitStageProcAlgParaVec(
403 : stageProcAlgParaVec, intraScatterTempAlg, interScatterTempAlg, intraAllGatherTempAlg, interAllGatherTempAlg);
404 0 : DataParameters dataParameters;
405 0 : InitDataParameters(slice, scratchMultiple, dataParameters);
406 0 : for (u32 loopIndex = 0; loopIndex < slice.loopTimes - 1; loopIndex++) {
407 0 : dataParameters.dataOffset[0] = loopIndex * slice.sliceCount * dataTypeSize_;
408 0 : dataParameters.dataOffset[1] = dataParameters.dataOffset[0] + slice.sliceCountPart0 * dataTypeSize_;
409 0 : CHK_RET(StageProcess(dataParameters, stageProcAlgParaVec));
410 : }
411 0 : InitFinalSliceDataParameters(slice, scratchMultiple, dataParameters);
412 0 : dataParameters.dataOffset[0] = (slice.loopTimes - 1) * slice.sliceCount * dataTypeSize_;
413 : dataParameters.dataOffset[1]
414 0 : = dataParameters.dataOffset[0] + (slice.finalSliceCountPart0 + slice.finalTailCountPart0) * dataTypeSize_;
415 0 : CHK_RET(StageProcess(dataParameters, stageProcAlgParaVec));
416 0 : return HcclResult::HCCL_SUCCESS;
417 0 : }
418 :
419 : // 算法注册
420 : INS_REGISTER_IMPL_BY_FOUR_TEMPS(OpType::BROADCAST, AiCpuInsBroadcastParallelMesh1DNHR,
421 : InsBroadcastParallelAiCpuExecutor, TopoMatchMeshNHR, InsTempScatterMesh1D, InsTempScatterNHR,
422 : InsTempAllGatherMesh1D, InsTempAllGatherNHR);
423 : } // namespace Hccl
|
