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 "coll_reduce_scatter_v_deter_executor.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 :
16 0 : CollReduceScatterVDeterExecutor::CollReduceScatterVDeterExecutor(
17 : const HcclDispatcher dispatcher,
18 0 : std::unique_ptr<TopoMatcher> &topoMatcher)
19 0 : : CollReduceScatterVExecutor(dispatcher, topoMatcher)
20 : {
21 0 : DMAReduceFlag_ = true;
22 0 : CCLMemSlice_ = false;
23 0 : isNeedSpaceBorrow_ = false;
24 0 : }
25 :
26 0 : void CollReduceScatterVDeterExecutor::ParseParam(const OpParam& param)
27 : {
28 : // 是否需要scratch memory（图模式没有cclbuffer，需要额外申请scratchMem）
29 0 : scratchMemFlag_ = (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE);
30 : // 记录图模式总数据量
31 0 : if( scratchMemFlag_ ) {
32 0 : u64 maxCount = 0;
33 0 : const u64* counts = static_cast<const u64*>(param.VDataDes.counts);
34 0 : for( u32 i = 0; i < topoAttr_.userRankSize; i++ ){
35 0 : maxCount = counts[i] > maxCount ? counts[i] : maxCount;
36 : }
37 0 : maxCount_ = maxCount;
38 0 : totalSize_ = maxCount * topoAttr_.userRankSize * SIZE_TABLE[param.VDataDes.dataType];
39 0 : isMeshTopo_ = (topoType_ == TopoType::TOPO_TYPE_NP_MESH || topoType_ == TopoType::TOPO_TYPE_4P_MESH ||
40 0 : topoType_ == TopoType::TOPO_TYPE_2P_MESH || topoType_ == TopoType::TOPO_TYPE_1P_MESH);
41 : }
42 0 : }
43 :
44 0 : u64 CollReduceScatterVDeterExecutor::CalcLoopMaxCount(const u32 unitSize)
45 : {
46 : u64 maxCountPerLoop ;
47 0 : if(scratchMemFlag_) {
48 0 : maxCountPerLoop = maxCount_;
49 : } else {
50 0 : maxCountPerLoop = inCCLbufferSize_ / topoAttr_.userRankSize / HCCL_MIN_SLICE_ALIGN
51 0 : * HCCL_MIN_SLICE_ALIGN / unitSize;
52 : }
53 0 : HCCL_INFO("[CollReduceScatterVDeterExecutor][CalcLoopMaxCount] maxCountPerLoop = [%llu] .", maxCountPerLoop);
54 0 : return maxCountPerLoop;
55 : }
56 :
57 0 : HcclResult CollReduceScatterVDeterExecutor::CalcCurCountsAndCurDispls(const u64 maxTotalCount,
58 : std::vector<u64> &countsLeft, std::vector<u64> &displs, std::vector<u64> &curCounts, std::vector<u64> &curDispls,
59 : bool &finished)
60 : {
61 0 : finished = true;
62 0 : curCounts.resize(countsLeft.size(), 0);
63 0 : curDispls.resize(displs.size(), 0);
64 :
65 : // 先设置本轮的displacements，等于入参displs
66 0 : std::copy(displs.begin(), displs.end(), curDispls.begin());
67 :
68 : // 分配好每个rank的counts
69 0 : for (auto i = 0U; i < countsLeft.size(); ++i) {
70 0 : const auto curCount = countsLeft[i] < maxTotalCount ? countsLeft[i] : maxTotalCount;
71 :
72 0 : curCounts[i] = curCount;
73 0 : countsLeft[i] -= curCount;
74 0 : displs[i] += curCount;
75 :
76 0 : if(countsLeft[i] != 0) {
77 0 : finished = false;
78 : }
79 : }
80 0 : return HCCL_SUCCESS;
81 : }
82 :
83 0 : u32 CollReduceScatterVDeterExecutor::CalReduceStreamNum(const u32& localRankSize)
84 : {
85 0 : return (1 << static_cast<int>(std::floor(log2(localRankSize))));
86 : }
87 :
88 0 : HcclResult CollReduceScatterVDeterExecutor::CalcStreamNum(u32& streamNum)
89 : {
90 : // Level0RankSize条流给alltoall，剩下的流给LocalReduce使用
91 0 : u32 level0StreamNum = topoAttr_.deviceNumPerAggregation - 1 + CalReduceStreamNum(topoAttr_.deviceNumPerAggregation);
92 : // level1主流分给alltoall，从流给LocalReduce使用
93 0 : u32 level1StreamNum = CalReduceStreamNum(topoAttr_.moduleNum);
94 : // 总流数上限：7（alltoall使用，提前的本地拷贝任务不需要并行）+ 4（LocalReduce使用）
95 0 : streamNum = std::min(std::max(level0StreamNum - 1, level1StreamNum),
96 0 : DEVICE_EIGHT + DEVICE_EIGHT / FACTOR_NUM_TWO - 1);
97 0 : HCCL_INFO("[%s]tag[%s] level0StreamNum[%u], level1StreamNum[%u], streamNum[%u]", __func__, tag_.c_str(),
98 : level0StreamNum, level1StreamNum, streamNum);
99 0 : return HCCL_SUCCESS;
100 : }
101 :
102 0 : HcclResult CollReduceScatterVDeterExecutor::CalcScratchMemSize(u64& scratchMemSize)
103 : {
104 0 : scratchMemSize = scratchMemFlag_ && isMeshTopo_ ? totalSize_ : 0U;
105 0 : HCCL_INFO("[%s]tag[%s] scratchMemSize[%llu]", __func__, tag_.c_str(), scratchMemSize);
106 0 : return HCCL_SUCCESS;
107 : }
108 :
109 0 : HcclResult CollReduceScatterVDeterExecutor::CalcCommInfo(
110 : std::vector<LevelNSubCommTransport>& opTransport)
111 : {
112 0 : TransportMemType inputType = TransportMemType::RESERVED;
113 0 : TransportMemType outputType = TransportMemType::RESERVED;
114 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
115 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
116 0 : if (isMeshTopo_) {
117 0 : CHK_RET(CalcLevel1CommInfoForMeshTopo(inputType, outputType, opTransport));
118 : } else {
119 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
120 : }
121 0 : return HCCL_SUCCESS;
122 : }
123 :
124 0 : HcclResult CollReduceScatterVDeterExecutor::CalcTransportMemType(TransportMemType &inputType,
125 : TransportMemType &outputType)
126 : {
127 : // scratchMemFlag_ 对应图模式场景（图模式没有cclbuffer）, PARAM_INPUT -> userInput
128 0 : inputType = scratchMemFlag_ ? TransportMemType::PARAM_INPUT : TransportMemType::CCL_INPUT;
129 0 : outputType = scratchMemFlag_ ?
130 0 : ( isMeshTopo_ ? TransportMemType::SCRATCH : TransportMemType::PARAM_OUTPUT )
131 : : TransportMemType::CCL_OUTPUT;
132 0 : HCCL_INFO("[%s]tag[%s] inputType[%d], outputType[%d]", __func__, tag_.c_str(), inputType, outputType);
133 0 : return HCCL_SUCCESS;
134 : }
135 :
136 0 : HcclResult CollReduceScatterVDeterExecutor::CalcLevel0CommInfo(TransportMemType inputType,
137 : TransportMemType outputType,
138 : std::vector<LevelNSubCommTransport>& opTransport)
139 : {
140 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
141 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
142 0 : return HCCL_SUCCESS;
143 0 : }
144 :
145 0 : HcclResult CollReduceScatterVDeterExecutor::CalcLevel1CommInfoForMeshTopo(TransportMemType inputType,
146 : TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
147 : {
148 0 : if (topoAttr_.moduleNum > 1) {
149 0 : CommParaInfo commParaLevel1(COMM_LEVEL1, CommType::COMM_TAG_MESH);
150 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel1, opTransport[COMM_LEVEL1], inputType, outputType));
151 0 : }
152 0 : return HCCL_SUCCESS;
153 : }
154 :
155 0 : bool CollReduceScatterVDeterExecutor::IsContainZeroSlice(const OpParam ¶m)
156 : {
157 0 : const auto curCounts = static_cast<u64*>(param.VDataDes.counts);
158 0 : auto it = std::find(curCounts, curCounts + topoAttr_.userRankSize, 0ULL);
159 0 : return (it != curCounts + topoAttr_.userRankSize);
160 : }
161 :
162 0 : bool CollReduceScatterVDeterExecutor::IsHugeData(const u64 curSize, const OpParam ¶m)
163 : {
164 0 : bool hugeData = (curSize * topoAttr_.userRankSize / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE) ||
165 : (curSize > SDMA_SEND_MAX_SIZE);
166 0 : return hugeData || IsContainZeroSlice(param);
167 : }
168 :
169 0 : HcclResult CollReduceScatterVDeterExecutor::RunReduceScattervLevel0(const OpParam ¶m, ExecMem &execMem,
170 : SubCommInfo &level0CommInfo)
171 : {
172 0 : CHK_RET(ActiveSlaveStreams(param.stream));
173 0 : HcclDataType dataType = param.VDataDes.dataType;
174 0 : const u32 unitSize = SIZE_TABLE[dataType];
175 0 : u32 level0RankSize = level0CommInfo.localRankSize;
176 :
177 0 : const auto curCounts = static_cast<u64*>(param.VDataDes.counts);
178 0 : const auto curDispls = static_cast<u64*>(param.VDataDes.displs);
179 0 : GroupSlicesInfo groupSlicesInfoLevel0;
180 0 : for (u32 groupId = 0; groupId < topoAttr_.moduleNum; groupId++) {
181 0 : MemBlockInfo memInfo;
182 0 : u32 groupSlicesOffset = groupId * level0RankSize ;
183 0 : for (u32 localRankId = 0; localRankId < level0RankSize; localRankId++) {
184 0 : u64 size = curCounts[localRankId + groupSlicesOffset] * unitSize;
185 0 : u64 userMemInOffset = curDispls[localRankId + groupSlicesOffset] * unitSize;
186 :
187 0 : memInfo.size.push_back(size);
188 0 : memInfo.userInputOffsets.push_back(userMemInOffset);
189 0 : memInfo.inputOffsets.push_back(minBiasOffset_ * unitSize * (localRankId + groupSlicesOffset));
190 0 : memInfo.outputOffsets.push_back(minBiasOffset_ * unitSize * (localRankId + groupSlicesOffset));
191 : }
192 0 : groupSlicesInfoLevel0.push_back(memInfo);
193 0 : }
194 :
195 0 : all2allOffset_ = topoAttr_.moduleNum > 1 ? 1 : 0; // 多机场景需要偏移1（给L1预留计算位，减少拷贝次数）
196 0 : std::unique_ptr<AlgTemplateBase> level0TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
197 0 : TemplateType::TEMPLATE_REDUCESCATTER_PLANT_LOCAL_REDUCE, dispatcher_);
198 0 : CHK_SMART_PTR_NULL(level0TempAlg);
199 :
200 : // execMem.scratchMem在单算子模式下为cclout，图模式为scrach，因此output传入scrach即可
201 0 : CHK_RET(level0TempAlg->Prepare(execMem.inputPtr, execMem.inputMem, execMem.scratchMem, param.stream,
202 : algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
203 : groupSlicesInfoLevel0, param.reduceType, all2allOffset_, dataType, isNeedSpaceBorrow_));
204 :
205 0 : CHK_RET(level0TempAlg->RegisterProfiler(
206 : (level0CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank,
207 : PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET, param.stream));
208 0 : CHK_RET(RunTemplate(level0TempAlg, level0CommInfo));
209 0 : return HCCL_SUCCESS;
210 0 : }
211 :
212 0 : HcclResult CollReduceScatterVDeterExecutor::RunReduceScattervLevel1ForMeshTopo(const OpParam ¶m, ExecMem &execMem,
213 : SubCommInfo &level0CommInfo)
214 : {
215 0 : u32 level0RankId = level0CommInfo.localRank;
216 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, level0RankId + 1));
217 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, level0RankId);
218 0 : u32 level0Ranksize = level0CommInfo.localRankSize;
219 :
220 : // 切分数据，记录每组的起始偏移和大小（仅1组）
221 0 : auto unitSize = SIZE_TABLE[param.VDataDes.dataType];
222 0 : u32 inputBaseIndex = (all2allOffset_ + level0RankId) % level0Ranksize; // 多机场景需要偏移1（给L1预留计算位，减少拷贝次数）
223 0 : u32 level1Ranksize = level1CommInfo.localRankSize;
224 0 : const auto curCounts = static_cast<u64*>(param.VDataDes.counts);
225 0 : MemBlockInfo memInfo;
226 0 : for (u32 localRank = 0; localRank < level1Ranksize; localRank++) {
227 0 : u64 inputIndex = inputBaseIndex + localRank * level0Ranksize;
228 0 : u64 outputIndex = level0RankId + localRank * level0Ranksize;
229 0 : u64 size = curCounts[level0RankId + localRank * level0Ranksize] * unitSize;
230 :
231 0 : memInfo.userInputOffsets.push_back(minBiasOffset_ * unitSize * outputIndex);
232 0 : memInfo.inputOffsets.push_back(minBiasOffset_ * unitSize* inputIndex);
233 0 : memInfo.outputOffsets.push_back(minBiasOffset_ * unitSize * outputIndex);
234 0 : memInfo.size.push_back(size);
235 : }
236 :
237 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
238 0 : TemplateType::TEMPLATE_REDUCESCATTER_PLANT_LOCAL_REDUCE_COMBINE, dispatcher_);
239 0 : CHK_SMART_PTR_NULL(level1TempAlg);
240 :
241 0 : u32 level0LastRank = level0Ranksize - 1;
242 0 : CHK_RET(level1TempAlg->Prepare(execMem.inputMem, execMem.scratchMem,
243 : param.stream, algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
244 : memInfo, param.reduceType, param.VDataDes.dataType, level0RankId == level0LastRank - 1,
245 : level0RankId == level0LastRank, isNeedSpaceBorrow_));
246 :
247 0 : CHK_RET(level1TempAlg->RegisterProfiler((level0Ranksize << PROF_RANKSIZE_OFFSET_OF_PLANEID) +
248 : level0CommInfo.localRank, PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET, param.stream));
249 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
250 0 : return HCCL_SUCCESS;
251 0 : }
252 :
253 0 : HcclResult CollReduceScatterVDeterExecutor::CalReduceScatterVSliceData(const OpParam ¶m, u32 level0RankSize, u32 level1RankSize, std::vector<Slice> &dataSlices)
254 : {
255 : (void) level0RankSize;
256 0 : u32 unitSize = SIZE_TABLE[param.VDataDes.dataType];
257 0 : std::vector<Slice> slices;
258 0 : const auto curCounts = static_cast<u64*>(param.VDataDes.counts);
259 0 : u64 offset = 0;
260 0 : for(u32 moduleId = 0; moduleId < level1RankSize; moduleId++) {
261 0 : Slice slice;
262 0 : slice.size = curCounts[moduleId] * unitSize;
263 0 : slice.offset = offset * unitSize;
264 0 : slices.emplace_back(std::move(slice));
265 0 : offset += curCounts[moduleId];
266 : }
267 0 : dataSlices = std::move(slices);
268 0 : return HCCL_SUCCESS;
269 0 : }
270 :
271 0 : HcclResult CollReduceScatterVDeterExecutor::RunReduceScattervLevel1(const OpParam ¶m, ExecMem &execMem,
272 : const SubCommInfo &level0CommInfo)
273 : {
274 0 : u32 commIndex = level0CommInfo.localRank; // 找到rank所在的节点间平面
275 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
276 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
277 :
278 0 : HcclDataType dataType = param.VDataDes.dataType;
279 :
280 0 : u32 level0RankSize = level0CommInfo.localRankSize;
281 0 : u32 level1RankSize = level1CommInfo.localRankSize;
282 0 : HCCL_DEBUG("RunReduceScattervLevel1 begin");
283 : /* ******************第一步: 机间reducescatter *******************************/
284 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, dataType, param.reduceType);
285 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
286 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
287 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
288 0 : TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
289 0 : CHK_SMART_PTR_NULL(level1TempAlg);
290 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
291 0 : HCCL_INFO("reducescatterv mesh: using ring algo inter-server");
292 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
293 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
294 0 : TemplateType::TEMPLATE_REDUCESCATTER_NB, dispatcher_);
295 0 : HCCL_INFO("reducescatterv mesh: using nonuniform-bruck algo inter-server");
296 0 : CHK_SMART_PTR_NULL(level1TempAlg);
297 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
298 : } else {
299 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
300 0 : TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
301 0 : HCCL_INFO("reducescatterv mesh: using nhr algo inter-server");
302 0 : CHK_SMART_PTR_NULL(level1TempAlg);
303 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr, false));
304 0 : level1TempAlg->CloseBarrier();
305 : }
306 :
307 0 : std::vector<Slice> slices;
308 0 : CHK_RET(CalReduceScatterVSliceData(param, level0RankSize, level1RankSize, slices));
309 :
310 0 : CHK_RET(level1TempAlg->Prepare(execMem.inputMem, execMem.inputMem, execMem.inputMem, 0,
311 : dataType, param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, slices));
312 :
313 0 : CHK_RET(level1TempAlg->RegisterProfiler(
314 : (level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank,
315 : PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
316 :
317 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
318 0 : return HCCL_SUCCESS;
319 0 : }
320 :
321 0 : HcclResult CollReduceScatterVDeterExecutor::KernelRun(const OpParam ¶m, ExecMem &execMem)
322 : {
323 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s][CollReduceScatterVDeterExecutor] ReduceScatterV deter run start, tag[%s]", __func__, tag_.c_str());
324 :
325 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
326 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
327 :
328 0 : auto unitSize = SIZE_TABLE[param.VDataDes.dataType];
329 0 : const auto curCounts = static_cast<u64*>(param.VDataDes.counts);
330 0 : const auto curDispls = static_cast<u64*>(param.VDataDes.displs);
331 0 : u64 dataSize = execMem.count * unitSize;
332 0 : DeviceMem srcMem;
333 :
334 0 : if (isMeshTopo_) {
335 0 : u64 maxCount = *std::max_element(curCounts, curCounts + topoAttr_.userRankSize);
336 0 : u64 maxCountPerloop = CalcLoopMaxCount(unitSize);
337 0 : minBiasOffset_ = maxCount < maxCountPerloop ? maxCount : maxCountPerloop;
338 : // L0 节点内 reduce scatter v
339 0 : CHK_RET(RunReduceScattervLevel0(param, execMem, level0CommInfo));
340 : // L1 节点间 reduce scatter v
341 0 : if (topoAttr_.moduleNum > 1) {
342 0 : CHK_RET(RunReduceScattervLevel1ForMeshTopo(param, execMem, level0CommInfo));
343 : }
344 0 : srcMem = execMem.scratchMem.range(minBiasOffset_ * topoAttr_.userRank * unitSize, dataSize);// Opbase: CO/Sr->UO
345 : } else { // 处理 Nx1 场景的图模式
346 0 : CHK_RET(RunReduceScattervLevel1(param, execMem, level0CommInfo));
347 0 : srcMem = execMem.inputMem.range(curDispls[topoAttr_.userRank] * unitSize, dataSize);// Offload:UI->UO
348 : }
349 :
350 0 : Stream stream = param.stream;
351 0 : DeviceMem dstMem = DeviceMem::create(execMem.outputPtr, dataSize);
352 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream));
353 0 : HCCL_CONFIG_INFO(HCCL_ALG,"[%s]ReduceScatterV deter run success, tag[%s]", __func__, tag_.c_str());
354 0 : return HCCL_SUCCESS;
355 0 : }
356 :
357 : REGISTER_EXEC("ReduceScatterVDeterExecutor", ReduceScatterVDeterExecutor, CollReduceScatterVDeterExecutor);
358 : }
|
