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