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