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