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_mid_count_aiv_rdma_executor.h"
12 :
13 : namespace hccl {
14 : constexpr s32 INTRA_RS_STEP = 0;
15 : constexpr s32 INTRA_AG_STEP = 2;
16 :
17 0 : CollAllReduceMidCountAivRdmaExecutor::CollAllReduceMidCountAivRdmaExecutor(
18 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
19 0 : : CollAllReduceExecutor(dispatcher, topoMatcher)
20 : {
21 0 : DMAReduceFlag_ = false;
22 0 : desc_.isAivMode = true;
23 0 : desc_.aivTagNum = AIV_A2_ALL_REDUCE_RDMA_KERNEL_NUM;
24 0 : }
25 :
26 0 : HcclResult CollAllReduceMidCountAivRdmaExecutor::CalcStreamNum(u32& streamNum)
27 : {
28 0 : u32 totalStreamNum = topoAttr_.deviceNumPerAggregation > 1U ? topoAttr_.deviceNumPerAggregation - 1U : 1U;
29 0 : streamNum = totalStreamNum - 1U;
30 0 : HCCL_INFO("[CollAllReduceMidCountAivRdmaExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
31 0 : return HCCL_SUCCESS;
32 : }
33 :
34 0 : HcclResult CollAllReduceMidCountAivRdmaExecutor::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 : HcclResult
45 0 : CollAllReduceMidCountAivRdmaExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
46 : {
47 : // 中数据量：使用AIVIN，标记区在AIVIN末尾，单算子模式用CCLOUT，图模式用USEROUT
48 0 : inputType = TransportMemType::AIV_INPUT;
49 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
50 0 : outputType = TransportMemType::CCL_OUTPUT;
51 : } else {
52 0 : outputType = TransportMemType::PARAM_OUTPUT;
53 : }
54 0 : HCCL_INFO(
55 : "[CollAllReduceMidCountAivRdmaExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]",
56 : tag_.c_str(), inputType, outputType);
57 0 : return HCCL_SUCCESS;
58 : }
59 :
60 0 : HcclResult CollAllReduceMidCountAivRdmaExecutor::CalcLevel0CommInfo(
61 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
62 : {
63 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
64 0 : commParaLevel0.meshSinglePlane = true;
65 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
66 0 : return HCCL_SUCCESS;
67 0 : }
68 :
69 0 : HcclResult CollAllReduceMidCountAivRdmaExecutor::CalNumBlocks(
70 : u32& numBlocks, u32 rankSize, [[maybe_unused]] u64 dataSize, [[maybe_unused]] HcclCMDType cmdType)
71 : {
72 0 : numBlocks = rankSize; // 默认情况使用rankSize个AIV
73 0 : u32 bestNumBlocks = numBlocks;
74 :
75 0 : CHK_PRT_RET(
76 : numBlocks_ < numBlocks,
77 : HCCL_WARNING(
78 : "[CollAllReduceMidCountAivRdmaExecutor][CalNumBlocks]aivCore[%u] is invalid, at least need [%u].",
79 : numBlocks_, numBlocks),
80 : HCCL_E_PARA);
81 :
82 0 : HCCL_INFO(
83 : "[CollAllReduceMidCountAivRdmaExecutor][CalNumBlocks] numBlocks is set to [%u], limit[%u], recommanded[%u]",
84 : numBlocks, numBlocks_, bestNumBlocks);
85 0 : return HCCL_SUCCESS;
86 : }
87 :
88 0 : HcclResult CollAllReduceMidCountAivRdmaExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
89 : {
90 0 : HcclUs startut = TIME_NOW();
91 0 : tag_ = param.tag;
92 0 : algResResp_ = &algRes;
93 :
94 : // 中数据量：使用AIVIN，标记区在AIVIN末尾，单算子模式用CCLOUT，图模式用USEROUT
95 0 : ExecMem execMem;
96 0 : execMem.count = param.DataDes.count;
97 0 : execMem.inputPtr = param.inputPtr;
98 0 : execMem.inputMem = algRes.aivInputMem;
99 0 : execMem.outputPtr = param.outputPtr;
100 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
101 0 : execMem.outputMem = algRes.cclOutputMem;
102 : } else {
103 0 : execMem.outputMem = algRes.paramOutputMem;
104 : }
105 0 : HcclResult ret = KernelRun(param, execMem);
106 :
107 0 : CHK_PRT_RET(
108 : ret != HCCL_SUCCESS,
109 : HCCL_ERROR(
110 : "[CollAllReduceMidCountAivRdmaExecutor]errNo[0x%016llx] tag[%s] executor kernel run failed",
111 : HCCL_ERROR_CODE(ret), param.tag.c_str()),
112 : ret);
113 :
114 0 : HCCL_INFO(
115 : "tag[%s], AllReduce executor orchestrate success, take time [%lld]us.", param.tag.c_str(),
116 : DURATION_US(TIME_NOW() - startut));
117 0 : return HCCL_SUCCESS;
118 0 : }
119 :
120 0 : HcclResult CollAllReduceMidCountAivRdmaExecutor::GetAdjInfo(
121 : [[maybe_unused]] AlgResourceResponse& algRes, [[maybe_unused]] AdjInfo& adjInfo)
122 : {
123 0 : return HCCL_SUCCESS;
124 : }
125 :
126 0 : HcclResult CollAllReduceMidCountAivRdmaExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
127 : {
128 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllReduceMidCountAivRdmaExecutor][KernelRun]AllReduce aiv enter");
129 0 : HcclWorkflowMode workflow = workflowMode_;
130 0 : bool isOpbase = (workflow == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE);
131 0 : CHK_RET(ActiveSlaveStreams(param.stream));
132 :
133 : // 获取通信域信息
134 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
135 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
136 0 : u32 commIndex = level0CommInfo.localRank;
137 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
138 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
139 :
140 : // 数据准备，按照server内rankSize切片
141 0 : u32 perDataSize = SIZE_TABLE[param.DataDes.dataType];
142 0 : u64 totalSize = param.DataDes.count * perDataSize;
143 0 : std::vector<Slice> dataSegsSlice; // 数据分成ranksize份，每份的起始偏移和大小
144 0 : u32 sliceNum = level0CommInfo.localRankSize;
145 0 : CHK_RET(PrepareSliceDataWithAlignSize(totalSize, sliceNum, 0, dataSegsSlice, HCCL_ALIGN_COUNT_32_B));
146 0 : CHK_PRT_RET(
147 : commIndex >= dataSegsSlice.size(),
148 : HCCL_ERROR(
149 : "[CollAllReduceMidCountAivRdmaExecutor][Run]commIndex[%u] >= dataSegsSlice size[%zu]", commIndex,
150 : dataSegsSlice.size()),
151 : HCCL_E_INTERNAL);
152 0 : std::vector<hccl::LINK> intraLinks = level0CommInfo.links;
153 0 : std::vector<hccl::LINK> interLinks = level1CommInfo.links;
154 0 : u32 intraRankSize = level0CommInfo.localRankSize;
155 0 : u32 intraRankId = level0CommInfo.localRank;
156 :
157 : // reduce scatter阶段，inputMem0-31m做数据区，32M开始后的1M做标记区
158 : void* dataBuffers[MAX_RANK_SIZE];
159 : void* flagBuffers[MAX_RANK_SIZE]; // 标记区的具体偏移在kernel中决定
160 0 : CHK_RET(PrepareAivBuffers(
161 : intraRankSize, intraRankId, 0, execMem.inputMem, execMem.inputMem, intraLinks, dataBuffers, flagBuffers,
162 : UserMemType::INPUT_MEM, UserMemType::INPUT_MEM, 0, HCCL_MID_COUNT_32_MB));
163 : // 先做本地拷贝到AIVIN再跨片拷贝；output统一为allreduceInput的位置，即buffer中原位
164 :
165 0 : AivOpArgs opArgs{HcclCMDType::HCCL_CMD_ALLREDUCE, execMem.inputPtr, nullptr, execMem.count,
166 0 : param.DataDes.dataType, param.reduceType, 0, isOpbase};
167 0 : AivTopoArgs topoArgs{intraRankId, intraRankSize};
168 : u32 numBlocks;
169 0 : CHK_PRT_RET(
170 : CalNumBlocks(numBlocks, intraRankSize) != HCCL_SUCCESS, HCCL_ERROR("[%s] CalNumBlocks failed", __func__),
171 : HCCL_E_PARA);
172 0 : numBlocks_ = numBlocks;
173 0 : topoArgs.identify = algoAttr_.identifier;
174 0 : AivResourceArgs resourceArgs{param.tag, param.stream.ptr(), dataBuffers, flagBuffers, execMem.inputMem.size(),
175 0 : numBlocks_, param.aivTag};
176 0 : AivAlgArgs algArgs{INTRA_RS_STEP, false};
177 0 : algArgs.execTimeOut = topoMatcher_->GetExecTimeOutConfig();
178 0 : algArgs.execTimeOutSet = true;
179 0 : struct AivProfilingInfo aivProfilingInfo;
180 0 : aivProfilingInfo.counter = opCounter_;
181 :
182 0 : CHK_RET(ExecuteKernelLaunch(opArgs, topoArgs, resourceArgs, algArgs, aivProfilingInfo));
183 :
184 : // allreduce 阶段
185 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
186 0 : DeviceMem allreduceInput = execMem.inputMem.range(dataSegsSlice[commIndex].offset, dataSegsSlice[commIndex].size);
187 0 : CHK_SMART_PTR_NULL(allreduceInput);
188 0 : DeviceMem allreduceOutput = execMem.outputMem.range(dataSegsSlice[commIndex].offset, dataSegsSlice[commIndex].size);
189 0 : CHK_SMART_PTR_NULL(allreduceOutput);
190 :
191 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, param.DataDes.dataType, param.reduceType);
192 0 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
193 0 : auto opMeta = HcclOpMetaInfo::GetOneForAllReduce(
194 0 : autoSelectedAlgTypeLevel1, param.DataDes.dataType, ReduceType::INLINE_REDUCE, IsAllReduceSmallData(totalSize),
195 : 1, false, hccl::CopyPattern::BCOPY, 1, true);
196 0 : CHK_RET(InitTask(dispatcher_, const_cast<Stream&>(param.stream), opMeta.isEnableCache, opMeta.GetCacheKey()));
197 :
198 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
199 : level1TempAlg
200 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RING, dispatcher_);
201 0 : HCCL_INFO("AllReduce mesh: using ring algo inter-server.");
202 0 : CHK_SMART_PTR_NULL(level1TempAlg);
203 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
204 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
205 0 : u64 curSize = execMem.count * perDataSize; // 单位 byte
206 0 : HCCL_DEBUG(
207 : "AllReduce mesh: curSize[%llu] deviceNumPerAggregation[%u] commLevel0Size[%u]", curSize,
208 : topoAttr_.deviceNumPerAggregation, level0CommInfo.localRankSize);
209 0 : if (curSize / topoAttr_.deviceNumPerAggregation <= NHR_ALLREDUCE_SMALL_SIZE) {
210 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
211 0 : TemplateType::TEMPLATE_ALL_REDUCE_NHR_ONESHOT, dispatcher_);
212 : } else {
213 : level1TempAlg
214 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR, dispatcher_);
215 : }
216 0 : HCCL_INFO("AllReduce mesh: using nhr algo inter-server.");
217 0 : CHK_SMART_PTR_NULL(level1TempAlg);
218 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
219 0 : level1TempAlg->CloseBarrier();
220 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
221 : level1TempAlg
222 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR_V1, dispatcher_);
223 0 : HCCL_INFO("AllReduce mesh: using nhr_v1 algo inter-server.");
224 0 : CHK_SMART_PTR_NULL(level1TempAlg);
225 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
226 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
227 : level1TempAlg
228 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NB, dispatcher_);
229 0 : HCCL_INFO("AllReduce mesh: using nb algo inter-server.");
230 0 : CHK_SMART_PTR_NULL(level1TempAlg);
231 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
232 : } else {
233 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
234 0 : TemplateType::TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING, dispatcher_);
235 0 : HCCL_INFO("AllReduce mesh: using Recursive halving-doubling algo inter-server.");
236 0 : CHK_SMART_PTR_NULL(level1TempAlg);
237 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
238 : }
239 0 : CHK_SMART_PTR_NULL(level1TempAlg);
240 :
241 0 : u32 rankSize = level1CommInfo.localRankSize;
242 0 : u64 hdCount = dataSegsSlice[commIndex].size / perDataSize;
243 0 : CHK_RET(level1TempAlg->Prepare(
244 : allreduceInput, allreduceOutput, allreduceOutput, hdCount, param.DataDes.dataType, param.stream,
245 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), dataSegsSlice[commIndex].offset));
246 :
247 0 : CHK_RET(level1TempAlg->RegisterProfiler(
248 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank, PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET,
249 : param.stream));
250 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
251 0 : HCCL_INFO("[CollAllReduceMidCountAivRdmaExecutor] rdma stage run success.");
252 0 : CHK_RET(LaunchTask(dispatcher_, const_cast<Stream&>(param.stream)));
253 :
254 : // allgather阶段，outputMem做数据区，32M开始后的1M做标记区
255 0 : CHK_RET(PrepareAivBuffers(
256 : intraRankSize, intraRankId, 0, execMem.outputMem, execMem.inputMem, intraLinks, dataBuffers, flagBuffers,
257 : UserMemType::OUTPUT_MEM, UserMemType::INPUT_MEM, 0, HCCL_MID_COUNT_32_MB));
258 : // 输入统一为allreduceOutput的位置，各卡不同；单算子模式需要outputAddr，先做本地拷贝再跨片拷贝；图模式结果直接放在CCL
259 : // Out中
260 :
261 0 : opArgs.input = nullptr;
262 0 : opArgs.output = execMem.outputPtr;
263 0 : resourceArgs.buffersIn = dataBuffers;
264 0 : resourceArgs.buffersOut = flagBuffers;
265 0 : resourceArgs.aivTag = GetNextAivTag(resourceArgs.aivTag);
266 0 : algArgs.step = INTRA_AG_STEP;
267 :
268 0 : CHK_RET(ExecuteKernelLaunch(opArgs, topoArgs, resourceArgs, algArgs, aivProfilingInfo));
269 :
270 0 : HCCL_INFO("[CollAllReduceMidCountAivRdmaExecutor][KernelRun]AllReduce aiv run success");
271 0 : return HCCL_SUCCESS;
272 0 : }
273 :
274 : REGISTER_EXEC("AllReduceMidCountAivRdmaExecutor", AllReduceMidCountAivRdma, CollAllReduceMidCountAivRdmaExecutor);
275 :
276 : } // namespace hccl
|
