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_mesh_opbase_executor.h"
12 : #include <numeric>
13 : namespace hccl {
14 :
15 0 : CollReduceScatterVMeshOpbaseExecutor::CollReduceScatterVMeshOpbaseExecutor(
16 : const HcclDispatcher dispatcher,
17 0 : std::unique_ptr<TopoMatcher> &topoMatcher)
18 0 : : CollReduceScatterVExecutor(dispatcher, topoMatcher)
19 : {
20 0 : DMAReduceFlag_ = true;
21 0 : CCLMemSlice_ = false;
22 0 : }
23 :
24 0 : void CollReduceScatterVMeshOpbaseExecutor::ParseParam(const OpParam& param)
25 : {
26 0 : aicpuUnfoldMode_ = param.aicpuUnfoldMode;
27 0 : DMAReduceFlag_ = topoAttr_.moduleNum > 1 ? false : true ;
28 0 : }
29 :
30 0 : HcclResult CollReduceScatterVMeshOpbaseExecutor::CalcStreamNum(u32& streamNum)
31 : {
32 0 : u32 totalStreamNum = topoAttr_.deviceNumPerAggregation;
33 0 : streamNum = totalStreamNum - 1U;
34 0 : HCCL_INFO("[CollReduceScatterVMeshOpbaseExecutor][CalcStreamNum] tag[%s] streamNum[%u]",
35 : tag_.c_str(), streamNum);
36 0 : return HCCL_SUCCESS;
37 : }
38 :
39 0 : HcclResult CollReduceScatterVMeshOpbaseExecutor::CalcCommInfo(
40 : std::vector<LevelNSubCommTransport>& opTransport)
41 : {
42 0 : TransportMemType inputType = TransportMemType::RESERVED;
43 0 : TransportMemType outputType = TransportMemType::RESERVED;
44 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
45 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
46 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
47 0 : return HCCL_SUCCESS;
48 : }
49 :
50 0 : HcclResult CollReduceScatterVMeshOpbaseExecutor::CalcTransportMemType(TransportMemType &inputType,
51 : TransportMemType &outputType)
52 : {
53 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
54 0 : inputType = TransportMemType::CCL_INPUT;
55 0 : outputType = TransportMemType::CCL_OUTPUT;
56 : } else {
57 0 : inputType = TransportMemType::PARAM_INPUT;
58 0 : outputType = TransportMemType::PARAM_OUTPUT;
59 : }
60 0 : HCCL_INFO("[CollReduceScatterVMeshOpbaseExecutor][CalcTransportMemType] tag[%s] inputType[%d],"
61 : " outputType[%d]", tag_.c_str(), inputType, outputType);
62 0 : return HCCL_SUCCESS;
63 : }
64 :
65 0 : HcclResult CollReduceScatterVMeshOpbaseExecutor::CalcLevel0CommInfo(TransportMemType inputType,
66 : TransportMemType outputType,
67 : std::vector<LevelNSubCommTransport>& opTransport)
68 : {
69 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
70 0 : commParaLevel0.meshSinglePlane = true;
71 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
72 0 : return HCCL_SUCCESS;
73 0 : }
74 :
75 0 : bool CollReduceScatterVMeshOpbaseExecutor::IsHugeData(const u64 curSize, const OpParam ¶m)
76 : {
77 0 : const auto *countsPtr = static_cast<const u64*>(param.VDataDes.counts);
78 0 : u64 totalCounts = std::accumulate(countsPtr, countsPtr + topoAttr_.userRankSize, 0ULL);
79 0 : return (totalCounts * SIZE_TABLE[param.VDataDes.dataType] > RDMA_SEND_MAX_SIZE) || (curSize > SDMA_SEND_MAX_SIZE);
80 : }
81 :
82 0 : HcclResult CollReduceScatterVMeshOpbaseExecutor::CalcCurCountsAndCurDisplsSingleModule(const u64 maxTotalCount,
83 : std::vector<u64> &countsLeft, std::vector<u64> &displs, std::vector<u64> &curCounts, std::vector<u64> &curDispls,
84 : bool &finished)
85 : {
86 0 : finished = true;
87 0 : curCounts.resize(countsLeft.size(), 0);
88 0 : curDispls.resize(displs.size(), 0);
89 :
90 : // 先设置本轮的displacements，等于入参displs
91 0 : std::copy(displs.begin(), displs.end(), curDispls.begin());
92 : // 分配好每个rank的counts
93 0 : for (auto i = 0U; i < countsLeft.size(); ++i) {
94 0 : const auto curCount = countsLeft[i] < maxTotalCount ? countsLeft[i] : maxTotalCount;
95 0 : curCounts[i] = curCount;
96 0 : countsLeft[i] -= curCount;
97 0 : displs[i] += curCount;
98 :
99 0 : if(countsLeft[i] != 0) {
100 0 : finished = false;
101 : }
102 : }
103 0 : HCCL_INFO("[%s] Calc CurCountsAndCurDispls for SingleModule finish.", __func__);
104 0 : return HCCL_SUCCESS;
105 : }
106 :
107 0 : HcclResult CollReduceScatterVMeshOpbaseExecutor::CalcCurCountsAndCurDisplsMultiModule(const u64 maxTotalCount,
108 : std::vector<u64> &countsLeft, std::vector<u64> &displs, std::vector<u64> &curCounts, std::vector<u64> &curDispls,
109 : bool &finished)
110 : {
111 0 : curCounts = std::vector<u64>(countsLeft.size(), 0);
112 0 : curDispls = std::vector<u64>(displs.size(), 0);
113 0 : auto allocatableCount = maxTotalCount;
114 :
115 : // 先设置本轮的displacements，等于入参displs
116 0 : std::copy(displs.begin(), displs.end(), curDispls.begin());
117 :
118 : // 分配本轮的counts，如果CCLbuffer空间还没完全利用，则再进行分配
119 0 : while (allocatableCount > 0) {
120 : // 计算现在还有几个rank还有数据需要去通信(countsLeft不为0)
121 : const auto nonZeroCount =
122 0 : std::count_if(countsLeft.begin(), countsLeft.end(), [](const u64 count) { return count != 0; });
123 0 : if (nonZeroCount == 0) {
124 0 : finished = true;
125 0 : HCCL_INFO("[%s] Calc CurCountsAndCurDispls for multiModule finish.", __func__);
126 0 : return HCCL_SUCCESS;
127 : }
128 : // 计算每个rank可以分到多少count
129 0 : const auto perRankCount = allocatableCount / nonZeroCount;
130 0 : if (perRankCount == 0) {
131 0 : break;
132 : }
133 0 : HCCL_DEBUG("[CollReduceScatterVMeshOpbaseExecutor]Calc for perRankCount start");
134 0 : for (auto i = 0U; i < countsLeft.size(); ++i) {
135 0 : const auto curCount = countsLeft[i] < perRankCount ? countsLeft[i] : perRankCount;
136 0 : allocatableCount -= curCount;
137 0 : curCounts[i] += curCount;
138 0 : countsLeft[i] -= curCount;
139 0 : displs[i] += curCount;
140 : }
141 : }
142 : //特殊情况下，allocatableCount 刚好使用完毕时，不仅如此while循环，导致RunLoop额外循环一次
143 : const auto nonZeroCount =
144 0 : std::count_if(countsLeft.begin(), countsLeft.end(), [](const u64 count) { return count != 0; });
145 0 : if (nonZeroCount == 0) {
146 0 : finished = true;
147 : }
148 0 : HCCL_INFO("[%s] Calc CurCountsAndCurDispls for multiModule finish.", __func__);
149 0 : return HCCL_SUCCESS;
150 : }
151 :
152 0 : HcclResult CollReduceScatterVMeshOpbaseExecutor::CalcCurCountsAndCurDispls(const u64 maxTotalCount,
153 : std::vector<u64> &countsLeft, std::vector<u64> &displs, std::vector<u64> &curCounts, std::vector<u64> &curDispls,
154 : bool &finished)
155 : {
156 0 : if (topoAttr_.moduleNum > 1){
157 0 : CHK_RET(CalcCurCountsAndCurDisplsMultiModule(maxTotalCount, countsLeft, displs, curCounts, curDispls, finished));
158 : } else {
159 0 : CHK_RET(CalcCurCountsAndCurDisplsSingleModule(maxTotalCount, countsLeft, displs, curCounts, curDispls, finished));
160 : }
161 0 : return HCCL_SUCCESS;
162 : }
163 :
164 0 : HcclResult CollReduceScatterVMeshOpbaseExecutor::RunReduceScattervLevel0SingleModule(const OpParam ¶m, ExecMem &execMem,
165 : SubCommInfo &level0CommInfo)
166 : {
167 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollReduceScatterVMeshOpbaseExecutor] Run ReduceScatterV Level0 SingleModule ");
168 0 : HcclDataType dataType = param.VDataDes.dataType;
169 0 : const u32 unitSize = SIZE_TABLE[dataType];
170 0 : u32 level0RankSize = level0CommInfo.localRankSize;
171 :
172 : /* *******************节点内reducescatter ******************************************/
173 : // reduce_scatter_v 计算slice，数据分成ranksize份，每份的起始偏移和大小
174 0 : std::vector<Slice> inputSlices;
175 0 : const auto counts = static_cast<u64*>(param.VDataDes.counts);
176 0 : const auto displs = static_cast<u64*>(param.VDataDes.displs);
177 0 : for (u32 rankId = 0; rankId < level0RankSize; ++rankId) {
178 0 : Slice userslice;
179 0 : userslice.offset = displs[rankId] * unitSize;
180 0 : userslice.size = counts[rankId] * unitSize;
181 0 : inputSlices.emplace_back(std::move(userslice));
182 : }
183 :
184 0 : HcomCollOpInfo *opInfoPtr = nullptr;
185 0 : HcomCollOpInfo opInfo = {"", execMem.inputPtr, execMem.outputPtr, 0, dataType,
186 0 : param.root, param.reduceType};
187 0 : if (DMAReduceFlag_) {
188 0 : opInfoPtr = &opInfo;
189 : }
190 :
191 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, dataType, param.reduceType);
192 0 : std::unique_ptr<AlgTemplateBase> TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
193 0 : TemplateType::TEMPLATE_REDUCESCATTER_MESH_DIRECT, dispatcher_);
194 0 : CHK_SMART_PTR_NULL(TempAlg);
195 :
196 0 : CHK_RET(TempAlg->Prepare(execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count, dataType,
197 : param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, inputSlices, 0, reduceAttr,
198 : algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
199 : topoAttr_.userRank, opInfoPtr));
200 :
201 0 : CHK_RET(TempAlg->RegisterProfiler(
202 : (level0CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank,
203 : PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
204 :
205 0 : CHK_RET(RunTemplate(TempAlg, level0CommInfo));
206 :
207 0 : return HCCL_SUCCESS;
208 0 : }
209 :
210 0 : HcclResult CollReduceScatterVMeshOpbaseExecutor::RunReduceScattervLevel0(const OpParam ¶m, ExecMem &execMem,
211 : SubCommInfo &level0CommInfo)
212 : {
213 0 : HcclDataType dataType = param.VDataDes.dataType;
214 0 : const u32 unitSize = SIZE_TABLE[dataType];
215 0 : u32 level0rankSize = level0CommInfo.localRankSize;
216 0 : u32 commIndex = level0CommInfo.localRank; // 找到rank所在的节点间平面
217 : /* *******************节点内reducescatter ******************************************/
218 :
219 0 : std::vector<Slice> inputSlices;
220 0 : const auto counts = static_cast<u64*>(param.VDataDes.counts);
221 0 : u64 offset = 0;
222 :
223 0 : for (u32 moduleId = 0; moduleId < topoAttr_.moduleNum; moduleId++) {
224 0 : for (u32 rankId = 0; rankId < level0rankSize; ++rankId) {
225 0 : if (topoAttr_.userRank / level0rankSize == moduleId) {
226 0 : Slice userslice;
227 0 : userslice.size = counts[rankId + moduleId * level0rankSize] * unitSize;
228 0 : userslice.offset = offset * unitSize;
229 0 : inputSlices.emplace_back(std::move(userslice));
230 : }
231 0 : offset += counts[rankId + moduleId * level0rankSize];
232 : }
233 : }
234 :
235 0 : HcomCollOpInfo *opInfoPtr = nullptr;
236 0 : HcomCollOpInfo opInfo = {"", execMem.inputPtr, execMem.outputPtr, 0, dataType,
237 0 : param.root, param.reduceType};
238 0 : if (DMAReduceFlag_) {
239 0 : opInfoPtr = &opInfo;
240 : }
241 :
242 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, dataType, param.reduceType);
243 0 : std::unique_ptr<AlgTemplateBase> TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
244 0 : TemplateType::TEMPLATE_REDUCESCATTER_MESH_ATOMIC, dispatcher_);
245 0 : CHK_SMART_PTR_NULL(TempAlg);
246 :
247 0 : CHK_RET(TempAlg->Prepare(execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count, dataType,
248 : param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, inputSlices, 0, reduceAttr,
249 : algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
250 : topoAttr_.userRank, opInfoPtr));
251 :
252 0 : CHK_RET(TempAlg->RegisterProfiler(
253 : (level0CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank,
254 : PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
255 :
256 0 : CHK_RET(RunTemplate(TempAlg, level0CommInfo));
257 :
258 : // 机间reduceScatter 结果 搬运到 cclout
259 0 : DeviceMem srcMem = execMem.inputMem.range(inputSlices[commIndex].offset,
260 0 : inputSlices[commIndex].size);
261 0 : CHK_SMART_PTR_NULL(srcMem);
262 0 : Stream stream = param.stream;
263 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, execMem.outputMem, srcMem, stream));
264 0 : return HCCL_SUCCESS;
265 0 : }
266 :
267 0 : HcclResult CollReduceScatterVMeshOpbaseExecutor::CalReduceScatterVSliceData(const OpParam ¶m, u32 level0RankSize, u32 level1RankSize, std::vector<Slice> &dataSlices)
268 : {
269 0 : HcclDataType dataType = param.VDataDes.dataType;
270 0 : u32 unitSize = SIZE_TABLE[dataType];
271 0 : std::vector<Slice> slices;
272 0 : const auto curCounts = static_cast<u64*>(param.VDataDes.counts);
273 0 : u64 offset = 0;
274 0 : for(u32 moduleId = 0; moduleId < level1RankSize; moduleId++) {
275 0 : u64 size = 0;
276 0 : for( u32 rankid = 0; rankid < level0RankSize; rankid++) {
277 0 : size += curCounts[rankid + moduleId * level0RankSize];
278 : }
279 0 : Slice slice;
280 0 : slice.size = size * unitSize;
281 0 : slice.offset = offset * unitSize;
282 0 : slices.emplace_back(std::move(slice));
283 0 : offset += size;
284 : }
285 0 : dataSlices = std::move(slices);
286 0 : return HCCL_SUCCESS;
287 0 : }
288 :
289 0 : HcclResult CollReduceScatterVMeshOpbaseExecutor::RunReduceScattervLevel1(const OpParam ¶m, ExecMem &execMem,
290 : const SubCommInfo &level0CommInfo)
291 : {
292 0 : u32 commIndex = level0CommInfo.localRank; // 找到rank所在的节点间平面
293 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
294 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
295 :
296 0 : HcclDataType dataType = param.VDataDes.dataType;
297 :
298 0 : u32 level0RankSize = level0CommInfo.localRankSize;
299 0 : u32 level1RankSize = level1CommInfo.localRankSize;
300 : /* ******************第一步: 机间reducescatter *******************************/
301 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, dataType, param.reduceType);
302 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
303 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
304 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
305 0 : TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
306 0 : CHK_SMART_PTR_NULL(level1TempAlg);
307 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
308 0 : HCCL_INFO("reducescatterv mesh: using ring algo inter-server.");
309 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
310 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
311 0 : TemplateType::TEMPLATE_REDUCESCATTER_NB, dispatcher_);
312 0 : HCCL_INFO("reducescatterv mesh: using nonuniform-bruck algo inter-server.");
313 0 : CHK_SMART_PTR_NULL(level1TempAlg);
314 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
315 : } else {
316 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
317 0 : TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
318 0 : HCCL_INFO("reducescatterv mesh: using nhr algo inter-server.");
319 0 : CHK_SMART_PTR_NULL(level1TempAlg);
320 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr, false));
321 0 : level1TempAlg->CloseBarrier();
322 : }
323 :
324 0 : std::vector<Slice> slices;
325 0 : CalReduceScatterVSliceData(param, level0RankSize, level1RankSize, slices);
326 :
327 0 : CHK_RET(level1TempAlg->Prepare(execMem.inputMem, execMem.inputMem, execMem.scratchMem, 0,
328 : dataType, param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, slices));
329 :
330 0 : CHK_RET(level1TempAlg->RegisterProfiler(
331 : (level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank,
332 : PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
333 :
334 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
335 0 : return HCCL_SUCCESS;
336 0 : }
337 :
338 0 : HcclResult CollReduceScatterVMeshOpbaseExecutor::KernelRun(const OpParam ¶m, ExecMem &execMem)
339 : {
340 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollReduceScatterVMeshOpbaseExecutor] reducescatterv mesh run");
341 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
342 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
343 :
344 0 : if (topoAttr_.moduleNum > 1) {
345 0 : CHK_RET(RunReduceScattervLevel1(param, execMem, level0CommInfo));
346 0 : CHK_RET(RunReduceScattervLevel0(param, execMem, level0CommInfo));
347 : } else {
348 0 : CHK_RET(RunReduceScattervLevel0SingleModule(param, execMem, level0CommInfo));
349 : }
350 0 : return HCCL_SUCCESS;
351 0 : }
352 :
353 : REGISTER_EXEC("ReduceScatterVMeshOpbaseExecutor",
354 : ReduceScatterVMeshOpbase, CollReduceScatterVMeshOpbaseExecutor);
355 : }
|
