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 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
18 0 : : CollReduceScatterVExecutor(dispatcher, topoMatcher)
19 : {
20 0 : DMAReduceFlag_ = true;
21 0 : CCLMemSlice_ = false;
22 0 : isNeedSpaceBorrow_ = false;
23 0 : }
24 :
25 0 : void CollReduceScatterVDeterExecutor::ParseParam(const OpParam& param)
26 : {
27 : // 是否需要scratch memory（图模式没有cclbuffer，需要额外申请scratchMem）
28 0 : scratchMemFlag_ = (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE);
29 : // 记录图模式总数据量
30 0 : if (scratchMemFlag_) {
31 0 : u64 maxCount = 0;
32 0 : const u64* counts = static_cast<const u64*>(param.VDataDes.counts);
33 0 : for (u32 i = 0; i < topoAttr_.userRankSize; i++) {
34 0 : maxCount = counts[i] > maxCount ? counts[i] : maxCount;
35 : }
36 0 : maxCount_ = maxCount;
37 0 : totalSize_ = maxCount * topoAttr_.userRankSize * SIZE_TABLE[param.VDataDes.dataType];
38 : isMeshTopo_
39 0 : = (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 : maxCountPerLoop
51 0 : = inCCLbufferSize_ / topoAttr_.userRankSize / HCCL_MIN_SLICE_ALIGN * 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(
58 : const u64 maxTotalCount, std::vector<u64>& countsLeft, std::vector<u64>& displs, std::vector<u64>& curCounts,
59 : std::vector<u64>& curDispls, 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 : streamNum
96 0 : = std::min(std::max(level0StreamNum - 1, level1StreamNum), DEVICE_EIGHT + DEVICE_EIGHT / FACTOR_NUM_TWO - 1);
97 0 : HCCL_INFO(
98 : "[%s]tag[%s] level0StreamNum[%u], level1StreamNum[%u], streamNum[%u]", __func__, tag_.c_str(), level0StreamNum,
99 : level1StreamNum, streamNum);
100 0 : return HCCL_SUCCESS;
101 : }
102 :
103 0 : HcclResult CollReduceScatterVDeterExecutor::CalcScratchMemSize(u64& scratchMemSize)
104 : {
105 0 : scratchMemSize = scratchMemFlag_ && isMeshTopo_ ? totalSize_ : 0U;
106 0 : HCCL_INFO("[%s]tag[%s] scratchMemSize[%llu]", __func__, tag_.c_str(), scratchMemSize);
107 0 : return HCCL_SUCCESS;
108 : }
109 :
110 0 : HcclResult CollReduceScatterVDeterExecutor::CalcCommInfo(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 : HcclResult
125 0 : CollReduceScatterVDeterExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
126 : {
127 : // scratchMemFlag_ 对应图模式场景（图模式没有cclbuffer）, PARAM_INPUT -> userInput
128 0 : inputType = scratchMemFlag_ ? TransportMemType::PARAM_INPUT : TransportMemType::CCL_INPUT;
129 0 : outputType = scratchMemFlag_ ? (isMeshTopo_ ? TransportMemType::SCRATCH : TransportMemType::PARAM_OUTPUT) :
130 : TransportMemType::CCL_OUTPUT;
131 0 : HCCL_INFO("[%s]tag[%s] inputType[%d], outputType[%d]", __func__, tag_.c_str(), inputType, outputType);
132 0 : return HCCL_SUCCESS;
133 : }
134 :
135 0 : HcclResult CollReduceScatterVDeterExecutor::CalcLevel0CommInfo(
136 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
137 : {
138 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
139 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
140 0 : return HCCL_SUCCESS;
141 0 : }
142 :
143 0 : HcclResult CollReduceScatterVDeterExecutor::CalcLevel1CommInfoForMeshTopo(
144 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
145 : {
146 0 : if (topoAttr_.moduleNum > 1) {
147 0 : CommParaInfo commParaLevel1(COMM_LEVEL1, CommType::COMM_TAG_MESH);
148 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel1, opTransport[COMM_LEVEL1], inputType, outputType));
149 0 : }
150 0 : return HCCL_SUCCESS;
151 : }
152 :
153 0 : bool CollReduceScatterVDeterExecutor::IsContainZeroSlice(const OpParam& param)
154 : {
155 0 : const auto curCounts = static_cast<u64*>(param.VDataDes.counts);
156 0 : auto it = std::find(curCounts, curCounts + topoAttr_.userRankSize, 0ULL);
157 0 : return (it != curCounts + topoAttr_.userRankSize);
158 : }
159 :
160 0 : bool CollReduceScatterVDeterExecutor::IsHugeData(const u64 curSize, const OpParam& param)
161 : {
162 0 : bool hugeData = (curSize * topoAttr_.userRankSize / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE)
163 0 : || (curSize > SDMA_SEND_MAX_SIZE);
164 0 : return hugeData || IsContainZeroSlice(param);
165 : }
166 :
167 0 : HcclResult CollReduceScatterVDeterExecutor::RunReduceScattervLevel0(
168 : const OpParam& param, ExecMem& execMem, SubCommInfo& level0CommInfo)
169 : {
170 0 : CHK_RET(ActiveSlaveStreams(param.stream));
171 0 : HcclDataType dataType = param.VDataDes.dataType;
172 0 : const u32 unitSize = SIZE_TABLE[dataType];
173 0 : u32 level0RankSize = level0CommInfo.localRankSize;
174 :
175 0 : const auto curCounts = static_cast<u64*>(param.VDataDes.counts);
176 0 : const auto curDispls = static_cast<u64*>(param.VDataDes.displs);
177 0 : GroupSlicesInfo groupSlicesInfoLevel0;
178 0 : for (u32 groupId = 0; groupId < topoAttr_.moduleNum; groupId++) {
179 0 : MemBlockInfo memInfo;
180 0 : u32 groupSlicesOffset = groupId * level0RankSize;
181 0 : for (u32 localRankId = 0; localRankId < level0RankSize; localRankId++) {
182 0 : u64 size = curCounts[localRankId + groupSlicesOffset] * unitSize;
183 0 : u64 userMemInOffset = curDispls[localRankId + groupSlicesOffset] * unitSize;
184 :
185 0 : memInfo.size.push_back(size);
186 0 : memInfo.userInputOffsets.push_back(userMemInOffset);
187 0 : memInfo.inputOffsets.push_back(minBiasOffset_ * unitSize * (localRankId + groupSlicesOffset));
188 0 : memInfo.outputOffsets.push_back(minBiasOffset_ * unitSize * (localRankId + groupSlicesOffset));
189 : }
190 0 : groupSlicesInfoLevel0.push_back(memInfo);
191 0 : }
192 :
193 0 : all2allOffset_ = topoAttr_.moduleNum > 1 ? 1 : 0; // 多机场景需要偏移1（给L1预留计算位，减少拷贝次数）
194 0 : std::unique_ptr<AlgTemplateBase> level0TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
195 0 : TemplateType::TEMPLATE_REDUCESCATTER_PLANT_LOCAL_REDUCE, dispatcher_);
196 0 : CHK_SMART_PTR_NULL(level0TempAlg);
197 :
198 : // execMem.scratchMem在单算子模式下为cclout，图模式为scrach，因此output传入scrach即可
199 0 : CHK_RET(level0TempAlg->Prepare(
200 : execMem.inputPtr, execMem.inputMem, execMem.scratchMem, param.stream, algResResp_->slaveStreams,
201 : algResResp_->notifiesMain, algResResp_->notifiesAux, groupSlicesInfoLevel0, param.reduceType, all2allOffset_,
202 : dataType, isNeedSpaceBorrow_));
203 :
204 0 : CHK_RET(level0TempAlg->RegisterProfiler(
205 : (level0CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_2,
206 : HCCL_EXEC_STEP_NOT_SET, param.stream));
207 0 : CHK_RET(RunTemplate(level0TempAlg, level0CommInfo));
208 0 : return HCCL_SUCCESS;
209 0 : }
210 :
211 0 : HcclResult CollReduceScatterVDeterExecutor::RunReduceScattervLevel1ForMeshTopo(
212 : const OpParam& param, ExecMem& execMem, SubCommInfo& level0CommInfo)
213 : {
214 0 : u32 level0RankId = level0CommInfo.localRank;
215 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, level0RankId + 1));
216 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, level0RankId);
217 0 : u32 level0Ranksize = level0CommInfo.localRankSize;
218 :
219 : // 切分数据，记录每组的起始偏移和大小（仅1组）
220 0 : auto unitSize = SIZE_TABLE[param.VDataDes.dataType];
221 0 : u32 inputBaseIndex
222 0 : = (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(
243 : execMem.inputMem, execMem.scratchMem, param.stream, algResResp_->slaveStreams, algResResp_->notifiesMain,
244 : algResResp_->notifiesAux, memInfo, param.reduceType, param.VDataDes.dataType,
245 : level0RankId == level0LastRank - 1, level0RankId == level0LastRank, isNeedSpaceBorrow_));
246 :
247 0 : CHK_RET(level1TempAlg->RegisterProfiler(
248 : (level0Ranksize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_2,
249 : HCCL_EXEC_STEP_NOT_SET, param.stream));
250 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
251 0 : return HCCL_SUCCESS;
252 0 : }
253 :
254 0 : HcclResult CollReduceScatterVDeterExecutor::CalReduceScatterVSliceData(
255 : const OpParam& param, u32 level0RankSize, u32 level1RankSize, std::vector<Slice>& dataSlices)
256 : {
257 : (void)level0RankSize;
258 0 : u32 unitSize = SIZE_TABLE[param.VDataDes.dataType];
259 0 : std::vector<Slice> slices;
260 0 : const auto curCounts = static_cast<u64*>(param.VDataDes.counts);
261 0 : u64 offset = 0;
262 0 : for (u32 moduleId = 0; moduleId < level1RankSize; moduleId++) {
263 0 : Slice slice;
264 0 : slice.size = curCounts[moduleId] * unitSize;
265 0 : slice.offset = offset * unitSize;
266 0 : slices.emplace_back(std::move(slice));
267 0 : offset += curCounts[moduleId];
268 : }
269 0 : dataSlices = std::move(slices);
270 0 : return HCCL_SUCCESS;
271 0 : }
272 :
273 0 : HcclResult CollReduceScatterVDeterExecutor::RunReduceScattervLevel1(
274 : const OpParam& param, ExecMem& execMem, const SubCommInfo& level0CommInfo)
275 : {
276 0 : u32 commIndex = level0CommInfo.localRank; // 找到rank所在的节点间平面
277 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
278 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
279 :
280 0 : HcclDataType dataType = param.VDataDes.dataType;
281 :
282 0 : u32 level0RankSize = level0CommInfo.localRankSize;
283 0 : u32 level1RankSize = level1CommInfo.localRankSize;
284 0 : HCCL_DEBUG("RunReduceScattervLevel1 begin");
285 : /* ******************第一步: 机间reducescatter *******************************/
286 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, dataType, param.reduceType);
287 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
288 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
289 : level1TempAlg
290 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
291 0 : CHK_SMART_PTR_NULL(level1TempAlg);
292 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
293 0 : HCCL_INFO("reducescatterv mesh: using ring algo inter-server");
294 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
295 : level1TempAlg
296 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NB, dispatcher_);
297 0 : HCCL_INFO("reducescatterv mesh: using nonuniform-bruck algo inter-server");
298 0 : CHK_SMART_PTR_NULL(level1TempAlg);
299 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
300 : } else {
301 : level1TempAlg
302 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
303 0 : HCCL_INFO("reducescatterv mesh: using nhr algo inter-server");
304 0 : CHK_SMART_PTR_NULL(level1TempAlg);
305 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr, false));
306 0 : level1TempAlg->CloseBarrier();
307 : }
308 :
309 0 : std::vector<Slice> slices;
310 0 : CHK_RET(CalReduceScatterVSliceData(param, level0RankSize, level1RankSize, slices));
311 :
312 0 : CHK_RET(level1TempAlg->Prepare(
313 : execMem.inputMem, execMem.inputMem, execMem.inputMem, 0, dataType, param.stream, param.reduceType,
314 : LEVEL0_BRIDGE_RANK_ID, slices));
315 :
316 0 : CHK_RET(level1TempAlg->RegisterProfiler(
317 : (level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank, PROF_STAGE_0,
318 : HCCL_EXEC_STEP_NOT_SET, param.stream));
319 :
320 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
321 0 : return HCCL_SUCCESS;
322 0 : }
323 :
324 0 : HcclResult CollReduceScatterVDeterExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
325 : {
326 0 : HCCL_CONFIG_INFO(
327 : HCCL_ALG, "[%s][CollReduceScatterVDeterExecutor] ReduceScatterV deter run start, tag[%s]", __func__,
328 : tag_.c_str());
329 :
330 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
331 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
332 :
333 0 : auto unitSize = SIZE_TABLE[param.VDataDes.dataType];
334 0 : const auto curCounts = static_cast<u64*>(param.VDataDes.counts);
335 0 : const auto curDispls = static_cast<u64*>(param.VDataDes.displs);
336 0 : u64 dataSize = execMem.count * unitSize;
337 0 : DeviceMem srcMem;
338 :
339 0 : if (isMeshTopo_) {
340 0 : u64 maxCount = *std::max_element(curCounts, curCounts + topoAttr_.userRankSize);
341 0 : u64 maxCountPerloop = CalcLoopMaxCount(unitSize);
342 0 : minBiasOffset_ = maxCount < maxCountPerloop ? maxCount : maxCountPerloop;
343 : // L0 节点内 reduce scatter v
344 0 : CHK_RET(RunReduceScattervLevel0(param, execMem, level0CommInfo));
345 : // L1 节点间 reduce scatter v
346 0 : if (topoAttr_.moduleNum > 1) {
347 0 : CHK_RET(RunReduceScattervLevel1ForMeshTopo(param, execMem, level0CommInfo));
348 : }
349 : srcMem
350 0 : = execMem.scratchMem.range(minBiasOffset_ * topoAttr_.userRank * unitSize, dataSize); // Opbase: CO/Sr->UO
351 : } else { // 处理 Nx1 场景的图模式
352 0 : CHK_RET(RunReduceScattervLevel1(param, execMem, level0CommInfo));
353 0 : srcMem = execMem.inputMem.range(curDispls[topoAttr_.userRank] * unitSize, dataSize); // Offload:UI->UO
354 : }
355 :
356 0 : Stream stream = param.stream;
357 0 : DeviceMem dstMem = DeviceMem::create(execMem.outputPtr, dataSize);
358 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream));
359 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s]ReduceScatterV deter run success, tag[%s]", __func__, tag_.c_str());
360 0 : return HCCL_SUCCESS;
361 0 : }
362 :
363 : REGISTER_EXEC("ReduceScatterVDeterExecutor", ReduceScatterVDeterExecutor, CollReduceScatterVDeterExecutor);
364 : } // namespace hccl
|
