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_mix_executor.h"
12 :
13 : namespace hccl {
14 :
15 0 : CollAllReduceMixExecutor::CollAllReduceMixExecutor(
16 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
17 0 : : CollAllReduceExecutor(dispatcher, topoMatcher)
18 : {
19 0 : DMAReduceFlag_ = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
20 0 : && topoAttr_.deviceType == DevType::DEV_TYPE_910_93;
21 0 : }
22 :
23 0 : void CollAllReduceMixExecutor::ParseParam(const OpParam& param)
24 : {
25 0 : tag_ = param.tag;
26 : bool isInlineReduce
27 0 : = IsSupportSDMAReduce(param.inputPtr, param.outputPtr, param.DataDes.dataType, param.reduceType);
28 0 : meshSinglePlane_ = (topoAttr_.deviceType == DevType::DEV_TYPE_910B)
29 0 : && topoMatcher_->GetExternalInputHcclDeterministic() == DETERMINISTIC_DISABLE && isInlineReduce
30 0 : && (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE);
31 0 : }
32 :
33 0 : HcclResult CollAllReduceMixExecutor::CalcStreamNum(u32& streamNum)
34 : {
35 0 : u32 totalStreamNum = 0;
36 :
37 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
38 0 : totalStreamNum = topoAttr_.deviceNumPerAggregation;
39 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
40 : totalStreamNum
41 0 : = (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING ? LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE :
42 : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE);
43 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
44 0 : totalStreamNum *= STREAM_NUM_FOR_DMAREDUCE_ONE_RING;
45 : }
46 : }
47 :
48 0 : streamNum = totalStreamNum - 1;
49 0 : HCCL_INFO("[CollAllReduceMixExecutor][CalcStreamNum] tag[%s] streamNum[%u].", tag_.c_str(), streamNum);
50 0 : return HCCL_SUCCESS;
51 : }
52 :
53 0 : HcclResult CollAllReduceMixExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
54 : {
55 0 : TransportMemType inputType = TransportMemType::RESERVED;
56 0 : TransportMemType outputType = TransportMemType::RESERVED;
57 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
58 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
59 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
60 :
61 : // mix在server间使用NHR通信域，并在多机A+X场景下当未设置使用RDMA时，默认使用RDMA
62 0 : std::vector<SingleSubCommTransport>& commTransportLevel1 = opTransport[COMM_LEVEL1];
63 0 : for (u32 ringIndex = 0; ringIndex < commTransportLevel1.size(); ringIndex++) {
64 0 : for (auto& transportRequest : commTransportLevel1[ringIndex].transportRequests) {
65 0 : transportRequest.isUsedRdma = true;
66 : }
67 : }
68 0 : return HCCL_SUCCESS;
69 : }
70 :
71 : HcclResult
72 0 : CollAllReduceMixExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType) const
73 : {
74 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
75 0 : inputType = TransportMemType::CCL_INPUT;
76 0 : outputType = TransportMemType::CCL_OUTPUT;
77 : } else {
78 0 : inputType = TransportMemType::PARAM_INPUT;
79 0 : outputType = TransportMemType::PARAM_OUTPUT;
80 : }
81 0 : HCCL_INFO(
82 : "[CollAllReduceMixExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d].", tag_.c_str(),
83 : inputType, outputType);
84 0 : return HCCL_SUCCESS;
85 : }
86 :
87 0 : HcclResult CollAllReduceMixExecutor::CalcLevel0CommInfo(
88 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
89 : {
90 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
91 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
92 0 : commParaLevel0.meshSinglePlane = meshSinglePlane_;
93 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
94 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
95 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
96 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
97 0 : }
98 0 : HCCL_DEBUG("[CollAllReduceMixExecutor][CalcLevel0CommInfo]Calculate for Level0CommInfo success");
99 0 : return HCCL_SUCCESS;
100 : }
101 :
102 0 : bool CollAllReduceMixExecutor::IsSmallData([[maybe_unused]] const u64 totalSize, const u64 curSize)
103 : {
104 0 : bool smallData = IsAllReduceSmallData(curSize);
105 0 : return smallData;
106 : }
107 :
108 0 : bool CollAllReduceMixExecutor::IsHugeData(const u64 curSize)
109 : {
110 0 : bool hugeData = curSize / topoAttr_.deviceNumPerAggregation / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE
111 0 : || curSize > SDMA_SEND_MAX_SIZE;
112 0 : return hugeData;
113 : }
114 :
115 0 : HcclResult CollAllReduceMixExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
116 : {
117 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllReduceMixExecutor][Run]The CollAllReduceMixExecutor starts.");
118 0 : u32 perDataSize = 0;
119 0 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, perDataSize));
120 :
121 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
122 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
123 0 : u32 sliceNum = level0CommInfo.localRankSize;
124 :
125 0 : std::vector<Slice> dataSegsSlice; // 数据分成ranksize份，每份的起始偏移和大小
126 : // 根据数据量计算每个环上数据的偏移和大小
127 0 : CHK_RET(AlgTemplateBase::PrepareSliceData(execMem.count, perDataSize, sliceNum, 0, dataSegsSlice));
128 0 : std::vector<std::vector<Slice>> multRingsSliceZero; // 910_93数据基于该rank上环0的偏移
129 :
130 : /* 三步算法step1：外层 - 节点内 reduce-scatter */
131 0 : CHK_RET(ActiveSlaveStreams(param.stream));
132 :
133 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
134 : // 多环数据切分
135 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
136 0 : multRingsSliceZero = PrepareMultiRingSlice(dataSegsSlice, param.tag, false, topoAttr_.nicList);
137 : } else {
138 0 : multRingsSliceZero.push_back(dataSegsSlice);
139 : }
140 :
141 : // 第一步的reducescatter输出放在CCL buffer上，通过设置nullptr指示不做最后一步的DMA削减动作
142 0 : HcomCollOpInfo reduceScatterOpInfo
143 0 : = {"", execMem.inputPtr, nullptr, execMem.count, param.DataDes.dataType, param.root, param.reduceType, 0};
144 0 : HcomCollOpInfo reduceScatterGraphModeOpInfo
145 0 : = {"", execMem.inputMem.ptr(), nullptr, execMem.count, param.DataDes.dataType,
146 0 : param.root, param.reduceType, 0};
147 0 : HcomCollOpInfo* reduceScatterOpInfoPtr = nullptr;
148 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
149 0 : reduceScatterOpInfoPtr = &reduceScatterGraphModeOpInfo;
150 : }
151 0 : if (DMAReduceFlag_) {
152 0 : reduceScatterOpInfoPtr = &reduceScatterOpInfo;
153 : }
154 0 : const std::vector<std::vector<Slice>> multRingsUserMemSliceDefault = std::vector<std::vector<Slice>>(0);
155 0 : CHK_RET(MultiRingReduceScatter(
156 : param.tag, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.reduceType,
157 : multRingsSliceZero, param.stream, PROF_STAGE_0, 0, reduceScatterOpInfoPtr, multRingsUserMemSliceDefault));
158 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
159 0 : if (topoMatcher_->GetExternalInputHcclDeterministic() == DETERMINISTIC_DISABLE
160 0 : && (param.DataDes.dataType != HCCL_DATA_TYPE_INT64)
161 0 : && (topoAttr_.deviceType == DevType::DEV_TYPE_910B && param.reduceType != HCCL_REDUCE_PROD)) {
162 0 : CHK_RET(MultiStreamReduceScatterMeshAtomic(
163 : param.tag, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.reduceType,
164 : dataSegsSlice, const_cast<Stream&>(param.stream), COMM_LEVEL0));
165 : } else {
166 0 : std::vector<std::vector<Slice>> multiStreamSlice; // 每个stream使用的数据基于用户buffer的偏移
167 0 : CHK_RET(AlgTemplateBase::PrepareSliceMeshStreams(dataSegsSlice, sliceNum - 1, multiStreamSlice));
168 0 : CHK_RET(MultiStreamReduceScatterMesh(
169 : param.tag, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.reduceType,
170 : multiStreamSlice, const_cast<Stream&>(param.stream), COMM_LEVEL0));
171 0 : }
172 : }
173 :
174 0 : HCCL_INFO("[CollAllReduceMixExecutor][KernelRun]AllReduce mix stage0 run success");
175 :
176 : /* 三步算法step2: 内层 - 节点间 allreduce */
177 0 : u32 commIndex = level0CommInfo.localRank;
178 0 : CHK_PRT_RET(
179 : commIndex >= dataSegsSlice.size(),
180 : HCCL_ERROR(
181 : "[CollAllReduceMixExecutor][Run]commIndex[%u] >= dataSegsSlice size[%zu]", commIndex, dataSegsSlice.size()),
182 : HCCL_E_INTERNAL);
183 :
184 0 : DeviceMem allreduceInput = execMem.inputMem.range(dataSegsSlice[commIndex].offset, dataSegsSlice[commIndex].size);
185 0 : CHK_SMART_PTR_NULL(allreduceInput);
186 0 : DeviceMem allreduceOutput = execMem.outputMem.range(dataSegsSlice[commIndex].offset, dataSegsSlice[commIndex].size);
187 0 : CHK_SMART_PTR_NULL(allreduceOutput);
188 :
189 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
190 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
191 :
192 0 : u64 reduceAttr = GetReduceAttr(allreduceInput, allreduceOutput, param.DataDes.dataType, param.reduceType);
193 :
194 0 : std::unique_ptr<AlgTemplateBase> level1Executor;
195 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
196 : level1Executor
197 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RING, dispatcher_);
198 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_RING in COMM_LEVEL1", __func__);
199 0 : CHK_SMART_PTR_NULL(level1Executor);
200 0 : CHK_RET(level1Executor->Prepare(reduceAttr));
201 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
202 0 : u64 curSize = execMem.count * perDataSize; // 单位 byte
203 0 : HCCL_DEBUG(
204 : "[CollAllReduceMixExecutor][KernelRun] curSize[%llu] deviceNumPerAggregation[%u] commLevel0Size[%u]",
205 : curSize, topoAttr_.deviceNumPerAggregation, level0CommInfo.localRankSize);
206 0 : if (curSize / topoAttr_.deviceNumPerAggregation <= NHR_ALLREDUCE_SMALL_SIZE) {
207 0 : level1Executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
208 0 : TemplateType::TEMPLATE_ALL_REDUCE_NHR_ONESHOT, dispatcher_);
209 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_NHR_ONESHOT in COMM_LEVEL1", __func__);
210 : } else {
211 : level1Executor
212 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR, dispatcher_);
213 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NHR in COMM_LEVEL1", __func__);
214 : }
215 0 : CHK_SMART_PTR_NULL(level1Executor);
216 0 : CHK_RET(level1Executor->Prepare(reduceAttr));
217 : } else {
218 0 : HCCL_ERROR(
219 : "[CollAllReduceMixExecutor][KernelRun]AllReduce mix: algType[%u] is not supported.", algType_.algoLevel1);
220 0 : return HCCL_E_NOT_SUPPORT;
221 : }
222 0 : CHK_SMART_PTR_NULL(level1Executor);
223 0 : u32 rankSize = level1CommInfo.localRankSize;
224 :
225 0 : u64 level1Count = dataSegsSlice[commIndex].size / perDataSize;
226 0 : CHK_RET(level1Executor->Prepare(
227 : allreduceInput, allreduceOutput, allreduceOutput, level1Count, param.DataDes.dataType, param.stream,
228 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), dataSegsSlice[commIndex].offset));
229 0 : CHK_RET(level1Executor->RegisterProfiler(
230 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank, PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET,
231 : param.stream));
232 0 : CHK_RET(RunTemplate(level1Executor, level1CommInfo));
233 :
234 0 : HCCL_INFO("[CollAllReduceMixExecutor][KernelRun]AllReduce mix stage1 run success.");
235 :
236 : /* 三步算法step3：外层 - 节点内 allgather */
237 : // 第三步的allgather输入放在CCL buffer上，通过设置nullptr指示要从CCL buffer获取输入
238 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
239 0 : HcomCollOpInfo allgatherOpInfo
240 0 : = {"", nullptr, execMem.outputPtr, execMem.count, param.DataDes.dataType, param.root, param.reduceType, 0};
241 0 : HcomCollOpInfo allgatherOpInfoGraphModeOpInfo = {
242 0 : "", nullptr, execMem.outputMem.ptr(), execMem.count, param.DataDes.dataType, param.root, param.reduceType,
243 0 : 0};
244 0 : HcomCollOpInfo* allgatherOpInfoPtr = nullptr;
245 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
246 0 : allgatherOpInfoPtr = &allgatherOpInfoGraphModeOpInfo;
247 : }
248 0 : if (DMAReduceFlag_) {
249 0 : allgatherOpInfoPtr = &allgatherOpInfo;
250 : }
251 0 : CHK_RET(MultiRingAllGather(
252 : param.tag, execMem.inputMem, execMem.outputMem, level1Count, param.DataDes.dataType, multRingsSliceZero,
253 : param.stream, PROF_STAGE_2, 0, allgatherOpInfoPtr));
254 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
255 0 : std::unique_ptr<AlgTemplateBase> level0Executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
256 0 : TemplateType::TEMPLATE_ALL_GATHER_MESH_ATOMIC, dispatcher_);
257 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_MESH_ATOMIC in COMM_LEVEL0", __func__);
258 0 : CHK_SMART_PTR_NULL(level0Executor);
259 0 : CHK_RET(level0Executor->Prepare(
260 : algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux, topoAttr_.userRank, nullptr,
261 : level0CommInfo.localRank, level0CommInfo.localRankSize));
262 :
263 0 : u32 rankSize = level0CommInfo.localRankSize;
264 0 : CHK_RET(level0Executor->Prepare(
265 : execMem.outputMem, execMem.outputMem, execMem.inputMem, execMem.count, param.DataDes.dataType, param.stream,
266 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, dataSegsSlice, 0));
267 :
268 0 : CHK_RET(level0Executor->RegisterProfiler(
269 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_2,
270 : HCCL_EXEC_STEP_NOT_SET, param.stream));
271 :
272 0 : CHK_RET(RunTemplate(level0Executor, level0CommInfo));
273 0 : }
274 :
275 0 : HCCL_INFO("[CollAllReduceMixExecutor][KernelRun]AllReduce mix stage2 run success.");
276 0 : return HCCL_SUCCESS;
277 0 : }
278 :
279 : REGISTER_EXEC("AllReduceMixExecutor", AllReduceMix, CollAllReduceMixExecutor);
280 :
281 : } // namespace hccl
|
