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_all_reduce_mesh_opbase_mid_count_deterministic_executor.h"
12 :
13 : namespace hccl {
14 0 : CollAllReduceMeshOpbaseMidCountDeterministicExecutor::CollAllReduceMeshOpbaseMidCountDeterministicExecutor(
15 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
16 0 : : CollAllReduceExecutor(dispatcher, topoMatcher)
17 : {
18 0 : DMAReduceFlag_ = true;
19 0 : }
20 :
21 0 : HcclResult CollAllReduceMeshOpbaseMidCountDeterministicExecutor::CalcStreamNum(u32& streamNum)
22 : {
23 0 : const u32 level0AlltoallStreamNum = topoAttr_.deviceNumPerAggregation - 1;
24 0 : const u32 level0LocalReduceStreamNum = 1 << static_cast<int>(std::floor(log2(topoAttr_.deviceNumPerAggregation)));
25 0 : streamNum = level0AlltoallStreamNum + level0LocalReduceStreamNum;
26 :
27 0 : HCCL_INFO(
28 : "[%s]tag[%s] level0AlltoallStreamNum[%u], level0LocalReduceStreamNum[%u], streamNum[%u]", __func__,
29 : tag_.c_str(), level0AlltoallStreamNum, level0LocalReduceStreamNum, streamNum);
30 0 : return HCCL_SUCCESS;
31 : }
32 :
33 : HcclResult
34 0 : CollAllReduceMeshOpbaseMidCountDeterministicExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
35 : {
36 0 : TransportMemType inputType = TransportMemType::RESERVED;
37 0 : TransportMemType outputType = TransportMemType::RESERVED;
38 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
39 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
40 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
41 0 : return HCCL_SUCCESS;
42 : }
43 :
44 0 : HcclResult CollAllReduceMeshOpbaseMidCountDeterministicExecutor::CalcTransportMemType(
45 : TransportMemType& inputType, TransportMemType& outputType)
46 : {
47 0 : inputType = TransportMemType::CCL_INPUT;
48 0 : outputType = TransportMemType::CCL_OUTPUT;
49 0 : HCCL_INFO(
50 : "[CollAllReduceMeshOpbaseMidCountDeterministicExecutor][CalcTransportMemType]"
51 : "tag[%s] inputType[%d], outputType[%d]",
52 : tag_.c_str(), inputType, outputType);
53 0 : return HCCL_SUCCESS;
54 : }
55 :
56 0 : HcclResult CollAllReduceMeshOpbaseMidCountDeterministicExecutor::CalcLevel0CommInfo(
57 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
58 : {
59 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
60 0 : commParaLevel0.meshSinglePlane = true;
61 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
62 0 : return HCCL_SUCCESS;
63 0 : }
64 :
65 0 : bool CollAllReduceMeshOpbaseMidCountDeterministicExecutor::IsHugeData(const u64 curSize)
66 : {
67 0 : bool hugeData = curSize / topoAttr_.deviceNumPerAggregation / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE
68 0 : || curSize > SDMA_SEND_MAX_SIZE;
69 0 : HCCL_DEBUG(
70 : "[%s]isHugeData[%d], curSize[%llu], topoAttr_.deviceNumPerAggregation[%u]", __func__, hugeData, curSize,
71 : topoAttr_.deviceNumPerAggregation);
72 0 : return hugeData;
73 : }
74 :
75 0 : bool CollAllReduceMeshOpbaseMidCountDeterministicExecutor::IsSmallData(
76 : [[maybe_unused]] const u64 totalSize, const u64 curSize)
77 : {
78 0 : bool smallData = IsAllReduceSmallData(curSize);
79 0 : return smallData;
80 : }
81 :
82 0 : HcclResult CollAllReduceMeshOpbaseMidCountDeterministicExecutor::PrepareSlicesInfo(
83 : const OpParam& param, const ExecMem& execMem, std::vector<Slice>& dataSegsSlice, GroupSlicesInfo& groupSlicesInfo,
84 : const u32 sliceSize)
85 : {
86 0 : const u32 perDataSize = SIZE_TABLE[param.DataDes.dataType];
87 0 : MemBlockInfo memInfo;
88 0 : u64 sizePerBlock = (execMem.count + sliceSize - 1) / sliceSize * perDataSize;
89 0 : sizePerBlock = AlgTemplateBase::RoundUpWithDivisor(sizePerBlock, HCCL_MIN_SLICE_ALIGN);
90 0 : const u64 totalSize = execMem.count * perDataSize;
91 0 : u64 sizeRemain = totalSize;
92 0 : for (u32 dataId = 0; dataId < sliceSize; dataId++) {
93 0 : u64 size = (sizeRemain > sizePerBlock) ? sizePerBlock : sizeRemain;
94 0 : u64 offset = totalSize - sizeRemain;
95 0 : memInfo.size.push_back(size);
96 0 : memInfo.userInputOffsets.push_back(offset);
97 0 : memInfo.inputOffsets.push_back(offset);
98 0 : memInfo.outputOffsets.push_back(offset);
99 0 : Slice slice{offset, size};
100 0 : dataSegsSlice.emplace_back(std::move(slice));
101 0 : sizeRemain -= size;
102 : }
103 0 : groupSlicesInfo.push_back(memInfo);
104 0 : return HCCL_SUCCESS;
105 0 : }
106 :
107 0 : HcclResult CollAllReduceMeshOpbaseMidCountDeterministicExecutor::RunReduceScatterLevel0(
108 : const OpParam& param, ExecMem& execMem, GroupSlicesInfo& groupSlicesInfo)
109 : {
110 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
111 0 : std::unique_ptr<AlgTemplateBase> level0TempAlg;
112 0 : const u32 all2allOffset = 0;
113 0 : level0TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
114 0 : TemplateType::TEMPLATE_REDUCESCATTER_PLANT_LOCAL_REDUCE, dispatcher_);
115 0 : CHK_SMART_PTR_NULL(level0TempAlg);
116 :
117 0 : CHK_RET(level0TempAlg->Prepare(
118 : execMem.inputPtr, execMem.inputMem, execMem.outputMem, param.stream, algResResp_->slaveStreams,
119 : algResResp_->notifiesMain, algResResp_->notifiesAux, groupSlicesInfo, param.reduceType, all2allOffset,
120 : param.DataDes.dataType, true, true));
121 :
122 0 : CHK_RET(level0TempAlg->RegisterProfiler(
123 : (level0CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_2,
124 : HCCL_EXEC_STEP_NOT_SET, param.stream));
125 0 : CHK_RET(RunTemplate(level0TempAlg, level0CommInfo));
126 0 : return HCCL_SUCCESS;
127 0 : }
128 :
129 0 : HcclResult CollAllReduceMeshOpbaseMidCountDeterministicExecutor::RunAllReduceLevel1(
130 : const OpParam& param, ExecMem& execMem, const std::vector<Slice>& dataSegsSlice)
131 : {
132 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
133 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
134 0 : const u32 perDataSize = SIZE_TABLE[param.DataDes.dataType];
135 0 : const u32 commIndex = level0CommInfo.localRank;
136 :
137 0 : CHK_PRT_RET(
138 : commIndex >= dataSegsSlice.size(),
139 : HCCL_ERROR("[%s]commIndex[%u] >= dataSegsSlice size[%zu]", __func__, commIndex, dataSegsSlice.size()),
140 : HCCL_E_INTERNAL);
141 :
142 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
143 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
144 :
145 0 : DeviceMem allreduceInput = execMem.inputMem.range(dataSegsSlice[commIndex].offset, dataSegsSlice[commIndex].size);
146 0 : CHK_SMART_PTR_NULL(allreduceInput);
147 0 : DeviceMem allreduceOutput = execMem.outputMem.range(dataSegsSlice[commIndex].offset, dataSegsSlice[commIndex].size);
148 0 : CHK_SMART_PTR_NULL(allreduceOutput);
149 :
150 0 : const u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, param.DataDes.dataType, param.reduceType);
151 :
152 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
153 : level1TempAlg
154 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RING, dispatcher_);
155 0 : HCCL_INFO("[%s]: using ring algo inter-server.", __func__);
156 0 : CHK_SMART_PTR_NULL(level1TempAlg);
157 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
158 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
159 : level1TempAlg
160 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR, dispatcher_);
161 0 : HCCL_INFO("[%s]: using nhr algo inter-server.", __func__);
162 0 : CHK_SMART_PTR_NULL(level1TempAlg);
163 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
164 0 : level1TempAlg->CloseBarrier();
165 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
166 : level1TempAlg
167 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR_V1, dispatcher_);
168 0 : HCCL_INFO("[%s]: using nhr_v1 algo inter-server.", __func__);
169 0 : CHK_SMART_PTR_NULL(level1TempAlg);
170 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
171 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
172 : level1TempAlg
173 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NB, dispatcher_);
174 0 : HCCL_INFO("[%s]: using nb algo inter-server.", __func__);
175 0 : CHK_SMART_PTR_NULL(level1TempAlg);
176 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
177 : } else {
178 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
179 0 : TemplateType::TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING, dispatcher_);
180 0 : HCCL_INFO("[%s]: using Recursive halving-doubling algo inter-server.", __func__);
181 0 : CHK_SMART_PTR_NULL(level1TempAlg);
182 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
183 : }
184 0 : CHK_SMART_PTR_NULL(level1TempAlg);
185 :
186 0 : const u32 rankSize = level1CommInfo.localRankSize;
187 :
188 0 : const u64 hdCount = dataSegsSlice[commIndex].size / perDataSize;
189 0 : CHK_RET(level1TempAlg->Prepare(
190 : allreduceInput, allreduceOutput, allreduceOutput, hdCount, param.DataDes.dataType, param.stream,
191 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), dataSegsSlice[commIndex].offset));
192 :
193 0 : CHK_RET(level1TempAlg->RegisterProfiler(
194 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank, PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET,
195 : param.stream));
196 :
197 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
198 0 : return HCCL_SUCCESS;
199 0 : }
200 :
201 0 : HcclResult CollAllReduceMeshOpbaseMidCountDeterministicExecutor::RunAllGatherLevel0(
202 : const OpParam& param, ExecMem& execMem, const std::vector<Slice>& dataSegsSlice)
203 : {
204 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
205 0 : std::unique_ptr<AlgTemplateBase> level0TempAlg;
206 : level0TempAlg
207 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_MESH_ATOMIC, dispatcher_);
208 :
209 0 : CHK_SMART_PTR_NULL(level0TempAlg);
210 0 : CHK_RET(level0TempAlg->Prepare(
211 : algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux, topoAttr_.userRank, nullptr,
212 : level0CommInfo.localRank, level0CommInfo.localRankSize));
213 :
214 0 : u32 rankSize = level0CommInfo.localRankSize;
215 0 : CHK_RET(level0TempAlg->Prepare(
216 : execMem.outputMem, execMem.outputMem, execMem.inputMem, execMem.count, param.DataDes.dataType, param.stream,
217 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, dataSegsSlice, 0));
218 :
219 0 : CHK_RET(level0TempAlg->RegisterProfiler(
220 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET,
221 : param.stream));
222 :
223 0 : CHK_RET(RunTemplate(level0TempAlg, level0CommInfo));
224 0 : return HCCL_SUCCESS;
225 0 : }
226 :
227 0 : HcclResult CollAllReduceMeshOpbaseMidCountDeterministicExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
228 : {
229 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s]userRank[%u] starts.", __func__, topoAttr_.userRank);
230 :
231 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, 1));
232 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
233 0 : const u32 sliceNum = level0CommInfo.localRankSize;
234 :
235 0 : std::vector<Slice> dataSegsSlice;
236 0 : GroupSlicesInfo groupSlicesInfoLevel0;
237 0 : CHK_RET(PrepareSlicesInfo(param, execMem, dataSegsSlice, groupSlicesInfoLevel0, sliceNum));
238 0 : CHK_RET(ActiveSlaveStreams(param.stream));
239 :
240 : /* STAGE 0: level 0 reduce scatter - plant local reduce */
241 0 : CHK_RET(RunReduceScatterLevel0(param, execMem, groupSlicesInfoLevel0));
242 0 : HCCL_INFO("[%s]AllReduce stage0 run success.", __func__);
243 :
244 : /* STAGE 1: level1 all_reduce - auto selected */
245 0 : CHK_RET(RunAllReduceLevel1(param, execMem, dataSegsSlice));
246 0 : HCCL_INFO("[%s]AllReduce stage1 run success.", __func__);
247 :
248 : /* STAGE 2: level0 all_gather - mesh atomic */
249 0 : CHK_RET(RunAllGatherLevel0(param, execMem, dataSegsSlice));
250 0 : HCCL_INFO("[%s]AllReduce stage2 run success", __func__);
251 :
252 0 : const u64 curSize = execMem.count * SIZE_TABLE[param.DataDes.dataType];
253 0 : DeviceMem outCommMem = execMem.outputMem.range(0, curSize);
254 0 : DeviceMem outMem(execMem.outputPtr, curSize);
255 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, outMem, outCommMem, const_cast<Stream&>(param.stream)));
256 :
257 0 : return HCCL_SUCCESS;
258 0 : }
259 :
260 0 : HcclResult CollAllReduceMeshOpbaseMidCountDeterministicExecutor::RunLoopInner(
261 : OpParam& param, const ReduceType& reduceType, ExecMem& execMem)
262 : {
263 0 : const u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
264 0 : const u64 curSize = execMem.count * unitSize;
265 0 : HCCL_DEBUG(
266 : "[%s]inputMem[%p][%llu], outputMem[%p][%llu], "
267 : "intputPtr[%p], outputPtr[%p], curCount[%llu], curSize[%llu]",
268 : __func__, execMem.inputMem.ptr(), execMem.inputMem.size(), execMem.outputMem.ptr(), execMem.outputMem.size(),
269 : execMem.inputPtr, execMem.outputPtr, execMem.count, curSize);
270 0 : CHK_PRT_RET((execMem.count == 0), HCCL_ERROR("[%s]In OP_BASE curCount is zero.", __func__), HCCL_E_PARA);
271 :
272 : /* init task */
273 0 : const auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
274 0 : const bool hugeData = IsHugeData(curSize);
275 0 : const bool smallData = IsSmallData(param.DataDes.count * unitSize, curSize);
276 0 : u64 sliceNum = 0;
277 0 : CHK_RET(GetSliceNum(execMem.count * unitSize, smallData, sliceNum, unitSize));
278 0 : const bool dataSplit = true;
279 0 : const u8 deterministic = topoMatcher_->GetExternalInputHcclDeterministic();
280 0 : const CopyPattern copy = CopyPattern::ZCOPY;
281 0 : const auto opMeta = HcclOpMetaInfo::GetOneForAllReduce(
282 : autoSelectedAlgTypeLevel1, param.DataDes.dataType, reduceType, smallData, 1, hugeData, copy, sliceNum, false,
283 : true, dataSplit, deterministic);
284 0 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
285 :
286 : /* kernel run */
287 0 : CHK_RET(KernelRun(param, execMem));
288 0 : CHK_RET(LaunchTaskExtend(
289 : dispatcher_, const_cast<Stream&>(param.stream), const_cast<std::vector<Stream>&>(algResResp_->slaveStreams)));
290 :
291 0 : return HCCL_SUCCESS;
292 : }
293 :
294 : REGISTER_EXEC(
295 : "AllReduceMeshOpbaseMidCountDeterministicExecutor", AllReduceMeshOpbaseMidCountDeterministic,
296 : CollAllReduceMeshOpbaseMidCountDeterministicExecutor);
297 :
298 : } // namespace hccl
|
