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