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_small_count_deterministic_executor.h"
12 :
13 : namespace hccl {
14 : // 准入条件: 确定性&小数据量
15 0 : CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::CollAllReduceMeshOpbaseSmallCountDeterministicExecutor(
16 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
17 0 : : CollAllReduceExecutor(dispatcher, topoMatcher)
18 : {
19 0 : DMAReduceFlag_ = true;
20 0 : if (!IsPowerOfTwo(topoAttr_.deviceNumPerAggregation)) {
21 : // localreduce + broadcast rank0的cclout要用来存放要reduce的数据
22 0 : CCLMemSlice_ = false;
23 : }
24 0 : }
25 :
26 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::CalcStreamNum(u32& streamNum)
27 : {
28 : u32 totalStreamNum;
29 0 : if (!IsPowerOfTwo(topoAttr_.deviceNumPerAggregation)) {
30 : // level0 为localreduce+Bcast时，需要level0 ranksize条
31 0 : totalStreamNum = topoAttr_.deviceNumPerAggregation;
32 : } else {
33 : // Doubling、nhr/ring算法只需要一条主流
34 0 : totalStreamNum = 1U;
35 : }
36 :
37 0 : streamNum = totalStreamNum - 1U;
38 0 : HCCL_INFO(
39 : "[CollAllReduceMeshOpbaseSmallCountDeterministicExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(),
40 : streamNum);
41 0 : return HCCL_SUCCESS;
42 : }
43 :
44 : HcclResult
45 0 : CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
46 : {
47 0 : TransportMemType inputType = TransportMemType::RESERVED;
48 0 : TransportMemType outputType = TransportMemType::RESERVED;
49 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
50 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
51 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
52 0 : return HCCL_SUCCESS;
53 : }
54 :
55 0 : bool CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::IsPowerOfTwo(u32 num) const
56 : {
57 0 : return (num & (num - 1)) == 0;
58 : }
59 :
60 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::CalcTransportMemType(
61 : TransportMemType& inputType, TransportMemType& outputType) const
62 : {
63 0 : inputType = TransportMemType::CCL_INPUT;
64 0 : outputType = TransportMemType::CCL_OUTPUT;
65 0 : HCCL_INFO(
66 : "[CollAllReduceMeshOpbaseSmallCountDeterministicExecutor][CalcTransportMemType]"
67 : "tag[%s] inputType[%d], outputType[%d]",
68 : tag_.c_str(), inputType, outputType);
69 0 : return HCCL_SUCCESS;
70 : }
71 :
72 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::CalcLevel0CommInfo(
73 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
74 : {
75 : CommType commType;
76 0 : if (topoAttr_.deviceNumPerAggregation > 1 && IsPowerOfTwo(topoAttr_.deviceNumPerAggregation)) {
77 : // Doubling
78 0 : commType = CommType::COMM_TAG_HALVING_DOUBLING;
79 : } else {
80 : // reduce + broadcast
81 0 : commType = CommType::COMM_TAG_MESH;
82 : }
83 0 : CommParaInfo commParaInfo(COMM_LEVEL0, commType);
84 0 : commParaInfo.meshSinglePlane = false;
85 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL0], inputType, outputType));
86 0 : return HCCL_SUCCESS;
87 0 : }
88 :
89 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::CalcLevel1CommInfo(
90 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
91 : {
92 0 : HCCL_INFO("[%s][CalcLevel1CommInfo]tag[%s] start", __func__, tag_.c_str());
93 0 : CommParaInfo commParaLevel1(COMM_LEVEL1, CommType::COMM_TAG_MAX);
94 0 : if (IsPowerOfTwo(topoAttr_.moduleNum) || algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_HD) {
95 0 : commParaLevel1.commType = CommType::COMM_TAG_HALVING_DOUBLING;
96 0 : HCCL_INFO("[%s][CalcLevel1CommInfo]tag[%s] Calc HDCommInfo", __func__, tag_.c_str());
97 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
98 0 : commParaLevel1.commType = CommType::COMM_TAG_RING_INNER;
99 0 : HCCL_INFO("[%s][CalcLevel1CommInfo]tag[%s] Calc RingCommInfo", __func__, tag_.c_str());
100 : } else {
101 0 : commParaLevel1.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING;
102 0 : HCCL_INFO("[%s][CalcLevel1CommInfo]tag[%s] Calc NHRCommInfo", __func__, tag_.c_str());
103 : }
104 0 : commParaLevel1.forceRdma = false;
105 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel1, opTransport[commParaLevel1.commPlane], inputType, outputType));
106 0 : HCCL_INFO("[%s][CalcLevel1CommInfo]tag[%s] Calc CommInfo Finish", __func__, tag_.c_str());
107 :
108 0 : return HCCL_SUCCESS;
109 0 : }
110 :
111 0 : u64 CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
112 : {
113 : u64 maxCountPerLoop;
114 0 : if (IsPowerOfTwo(topoAttr_.deviceNumPerAggregation)) {
115 : // local doubling SDMA cclin->cclout 一定是字节对齐的
116 0 : maxCountPerLoop = cclBuffSize / unitSize;
117 : } else {
118 : // template-localreduce_bcast 没有 128B对齐
119 0 : maxCountPerLoop = cclBuffSize / unitSize / (topoAttr_.deviceNumPerAggregation - 1);
120 : }
121 0 : return maxCountPerLoop;
122 : }
123 :
124 0 : bool CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::IsHugeData(const u64 curSize)
125 : {
126 0 : bool hugeData = curSize > RDMA_SEND_MAX_SIZE || curSize > SDMA_SEND_MAX_SIZE;
127 0 : return hugeData;
128 : }
129 :
130 0 : bool CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::IsSmallData(
131 : [[maybe_unused]] const u64 totalSize, [[maybe_unused]] const u64 curSize)
132 : {
133 : // 选到本执行器必为小数据量
134 0 : return true;
135 : }
136 :
137 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
138 : {
139 0 : HCCL_CONFIG_INFO(
140 : HCCL_ALG, "[CollAllReduceMeshOpbaseSmallCountDeterministicExecutor][Run]"
141 : "CollAllReduceMeshOpbaseSmallCountDeterministicExecutor begins.");
142 :
143 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
144 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
145 0 : u32 commIndex = level0CommInfo.localRank;
146 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
147 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
148 :
149 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, param.DataDes.dataType, param.reduceType);
150 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
151 0 : u64 curSize = execMem.count * unitSize; // 单位：字节
152 : // Run level0
153 0 : if (IsPowerOfTwo(level0CommInfo.localRankSize)) {
154 : // userin->cclin
155 0 : DeviceMem userInMem(execMem.inputPtr, curSize);
156 0 : DeviceMem cclInMem = execMem.inputMem.range(0, curSize);
157 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, cclInMem, userInMem, const_cast<Stream&>(param.stream)));
158 0 : CHK_RET(RunDoublingSingleLevel(param, reduceAttr, execMem, level0CommInfo));
159 0 : HCCL_INFO("allreduce small count deterministic: using doubling algo intra-server.");
160 0 : } else {
161 0 : HcomCollOpInfo opInfo = {"",
162 0 : execMem.inputPtr,
163 0 : execMem.inputMem.ptr(),
164 0 : execMem.count,
165 0 : param.DataDes.dataType,
166 0 : param.root,
167 0 : param.reduceType,
168 0 : 0};
169 0 : CHK_RET(RunReduceBcastSingleLevel(param, opInfo, reduceAttr, execMem, level0CommInfo));
170 0 : HCCL_INFO("allreduce small count deterministic: using reduce bcast algo intra-server.");
171 : }
172 : // Run level1
173 0 : if (IsPowerOfTwo(level1CommInfo.localRankSize)) {
174 0 : CHK_RET(RunDoublingSingleLevel(param, reduceAttr, execMem, level1CommInfo));
175 0 : HCCL_INFO("allreduce small count deterministic: using doubling algo inter-server.");
176 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_HD) {
177 0 : CHK_RET(RunTempLevel1(
178 : TemplateType::TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING, param, reduceAttr, execMem, level1CommInfo));
179 0 : HCCL_INFO("allreduce small count deterministic: using rhd algo inter-server.");
180 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
181 0 : CHK_RET(RunTempLevel1(TemplateType::TEMPLATE_ALL_REDUCE_RING, param, reduceAttr, execMem, level1CommInfo));
182 0 : HCCL_INFO("allreduce small count deterministic: using default ring algo inter-server.");
183 : } else {
184 : // 默认nhr
185 0 : CHK_RET(RunTempLevel1(TemplateType::TEMPLATE_ALL_REDUCE_NHR, param, reduceAttr, execMem, level1CommInfo));
186 0 : HCCL_INFO("allreduce small count deterministic: using nhr algo inter-server.");
187 : }
188 0 : DeviceMem dstMem(execMem.outputPtr, curSize);
189 0 : DeviceMem srcMem;
190 0 : if (IsPowerOfTwo(level1CommInfo.localRankSize)) {
191 : // cclin->userout
192 0 : srcMem = execMem.inputMem.range(0, curSize);
193 : } else {
194 : // cclout->userout
195 0 : srcMem = execMem.outputMem.range(0, curSize);
196 : }
197 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
198 :
199 0 : return HCCL_SUCCESS;
200 0 : }
201 :
202 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::RunDoublingSingleLevel(
203 : const OpParam& param, u64 reduceAttr, ExecMem& execMem, SubCommInfo& levelCommInfo) const
204 : {
205 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
206 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
207 0 : TemplateType::TEMPLATE_ALL_REDUCE_DOUBLING_LOCAL_REDUCE, dispatcher_);
208 0 : CHK_SMART_PTR_NULL(tempAlg);
209 0 : CHK_RET(tempAlg->Prepare(reduceAttr));
210 0 : CHK_RET(tempAlg->Prepare(
211 : execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.stream,
212 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), 0));
213 :
214 0 : CHK_RET(tempAlg->RegisterProfiler(
215 : (levelCommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + levelCommInfo.localRank, PROF_STAGE_0,
216 : HCCL_EXEC_STEP_NOT_SET, param.stream));
217 :
218 0 : CHK_RET(RunTemplate(tempAlg, levelCommInfo));
219 0 : return HCCL_SUCCESS;
220 0 : }
221 :
222 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::RunReduceBcastSingleLevel(
223 : const OpParam& param, HcomCollOpInfo& opInfo, u64 reduceAttr, ExecMem& execMem, SubCommInfo& levelCommInfo) const
224 : {
225 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
226 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
227 0 : TemplateType::TEMPLATE_ALL_REDUCE_LOCAL_REDUCE_BCAST, dispatcher_);
228 0 : CHK_SMART_PTR_NULL(tempAlg);
229 0 : CHK_RET(tempAlg->Prepare(
230 : reduceAttr, algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
231 : levelCommInfo.localRank, levelCommInfo.localRankSize, topoAttr_.userRank, &opInfo));
232 0 : CHK_SMART_PTR_NULL(tempAlg);
233 0 : CHK_RET(tempAlg->Prepare(
234 : execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.stream,
235 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), 0));
236 :
237 0 : CHK_RET(tempAlg->RegisterProfiler(
238 : (levelCommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + levelCommInfo.localRank, PROF_STAGE_0,
239 : HCCL_EXEC_STEP_NOT_SET, param.stream));
240 0 : CHK_RET(RunTemplate(tempAlg, levelCommInfo));
241 0 : return HCCL_SUCCESS;
242 0 : }
243 :
244 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::RunTempLevel1(
245 : const TemplateType type, const OpParam& param, u64 reduceAttr, ExecMem& execMem, SubCommInfo& level1CommInfo) const
246 : {
247 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
248 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(type, dispatcher_);
249 0 : CHK_SMART_PTR_NULL(tempAlg);
250 0 : CHK_RET(tempAlg->Prepare(reduceAttr));
251 :
252 0 : CHK_RET(tempAlg->Prepare(
253 : execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.stream,
254 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), 0));
255 0 : CHK_RET(tempAlg->RegisterProfiler(
256 : (level1CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank, PROF_STAGE_0,
257 : HCCL_EXEC_STEP_NOT_SET, param.stream));
258 :
259 0 : CHK_RET(RunTemplate(tempAlg, level1CommInfo));
260 0 : if (type == TemplateType::TEMPLATE_ALL_REDUCE_NHR) {
261 0 : tempAlg->CloseBarrier();
262 : }
263 0 : return HCCL_SUCCESS;
264 0 : }
265 :
266 : REGISTER_EXEC(
267 : "AllReduceMeshOpbaseSmallCountDeterministicExecutor", AllReduceMeshOpbaseSmallCountDeterministic,
268 : CollAllReduceMeshOpbaseSmallCountDeterministicExecutor);
269 :
270 : } // namespace hccl
|
