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(const u64 totalSize, const u64 curSize)
128 : {
129 : // 选到本执行器必为小数据量
130 0 : return true;
131 : }
132 :
133 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
134 : {
135 0 : HCCL_CONFIG_INFO(
136 : HCCL_ALG, "[CollAllReduceMeshOpbaseSmallCountDeterministicExecutor][Run]"
137 : "CollAllReduceMeshOpbaseSmallCountDeterministicExecutor begins.");
138 :
139 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
140 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
141 0 : u32 commIndex = level0CommInfo.localRank;
142 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
143 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
144 :
145 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, param.DataDes.dataType, param.reduceType);
146 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
147 0 : u64 curSize = execMem.count * unitSize; // 单位：字节
148 : // Run level0
149 0 : if (IsPowerOfTwo(level0CommInfo.localRankSize)) {
150 : // userin->cclin
151 0 : DeviceMem userInMem(execMem.inputPtr, curSize);
152 0 : DeviceMem cclInMem = execMem.inputMem.range(0, curSize);
153 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, cclInMem, userInMem, const_cast<Stream&>(param.stream)));
154 0 : CHK_RET(RunDoublingSingleLevel(param, reduceAttr, execMem, level0CommInfo));
155 0 : HCCL_INFO("allreduce small count deterministic: using doubling algo intra-server.");
156 0 : } else {
157 0 : HcomCollOpInfo opInfo
158 0 : = {"", execMem.inputPtr, execMem.inputMem.ptr(), execMem.count, param.DataDes.dataType,
159 0 : param.root, param.reduceType};
160 0 : CHK_RET(RunReduceBcastSingleLevel(param, opInfo, reduceAttr, execMem, level0CommInfo));
161 0 : HCCL_INFO("allreduce small count deterministic: using reduce bcast algo intra-server.");
162 : }
163 : // Run level1
164 0 : if (IsPowerOfTwo(level1CommInfo.localRankSize)) {
165 0 : CHK_RET(RunDoublingSingleLevel(param, reduceAttr, execMem, level1CommInfo));
166 0 : HCCL_INFO("allreduce small count deterministic: using doubling algo inter-server.");
167 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_HD) {
168 0 : CHK_RET(RunTempLevel1(
169 : TemplateType::TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING, param, reduceAttr, execMem, level1CommInfo));
170 0 : HCCL_INFO("allreduce small count deterministic: using rhd algo inter-server.");
171 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
172 0 : CHK_RET(RunTempLevel1(TemplateType::TEMPLATE_ALL_REDUCE_RING, param, reduceAttr, execMem, level1CommInfo));
173 0 : HCCL_INFO("allreduce small count deterministic: using default ring algo inter-server.");
174 : } else {
175 : // 默认nhr
176 0 : CHK_RET(RunTempLevel1(TemplateType::TEMPLATE_ALL_REDUCE_NHR, param, reduceAttr, execMem, level1CommInfo));
177 0 : HCCL_INFO("allreduce small count deterministic: using nhr algo inter-server.");
178 : }
179 0 : DeviceMem dstMem(execMem.outputPtr, curSize);
180 0 : DeviceMem srcMem;
181 0 : if (IsPowerOfTwo(level1CommInfo.localRankSize)) {
182 : // cclin->userout
183 0 : srcMem = execMem.inputMem.range(0, curSize);
184 : } else {
185 : // cclout->userout
186 0 : srcMem = execMem.outputMem.range(0, curSize);
187 : }
188 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
189 :
190 0 : return HCCL_SUCCESS;
191 0 : }
192 :
193 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::RunDoublingSingleLevel(
194 : const OpParam& param, u64 reduceAttr, ExecMem& execMem, SubCommInfo& levelCommInfo)
195 : {
196 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
197 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
198 0 : TemplateType::TEMPLATE_ALL_REDUCE_DOUBLING_LOCAL_REDUCE, dispatcher_);
199 0 : CHK_SMART_PTR_NULL(tempAlg);
200 0 : CHK_RET(tempAlg->Prepare(reduceAttr));
201 0 : CHK_RET(tempAlg->Prepare(
202 : execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.stream,
203 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), 0));
204 :
205 0 : CHK_RET(tempAlg->RegisterProfiler(
206 : (levelCommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + levelCommInfo.localRank, PROF_STAGE_0,
207 : HCCL_EXEC_STEP_NOT_SET, param.stream));
208 :
209 0 : CHK_RET(RunTemplate(tempAlg, levelCommInfo));
210 0 : return HCCL_SUCCESS;
211 0 : }
212 :
213 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::RunReduceBcastSingleLevel(
214 : const OpParam& param, HcomCollOpInfo& opInfo, u64 reduceAttr, ExecMem& execMem, SubCommInfo& levelCommInfo)
215 : {
216 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
217 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
218 0 : TemplateType::TEMPLATE_ALL_REDUCE_LOCAL_REDUCE_BCAST, dispatcher_);
219 0 : CHK_SMART_PTR_NULL(tempAlg);
220 0 : CHK_RET(tempAlg->Prepare(
221 : reduceAttr, algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
222 : levelCommInfo.localRank, levelCommInfo.localRankSize, topoAttr_.userRank, &opInfo));
223 0 : CHK_SMART_PTR_NULL(tempAlg);
224 0 : CHK_RET(tempAlg->Prepare(
225 : execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.stream,
226 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), 0));
227 :
228 0 : CHK_RET(tempAlg->RegisterProfiler(
229 : (levelCommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + levelCommInfo.localRank, PROF_STAGE_0,
230 : HCCL_EXEC_STEP_NOT_SET, param.stream));
231 0 : CHK_RET(RunTemplate(tempAlg, levelCommInfo));
232 0 : return HCCL_SUCCESS;
233 0 : }
234 :
235 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::RunTempLevel1(
236 : const TemplateType type, const OpParam& param, u64 reduceAttr, ExecMem& execMem, SubCommInfo& level1CommInfo)
237 : {
238 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
239 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(type, dispatcher_);
240 0 : CHK_SMART_PTR_NULL(tempAlg);
241 0 : CHK_RET(tempAlg->Prepare(reduceAttr));
242 :
243 0 : CHK_RET(tempAlg->Prepare(
244 : execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.stream,
245 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), 0));
246 0 : CHK_RET(tempAlg->RegisterProfiler(
247 : (level1CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank, PROF_STAGE_0,
248 : HCCL_EXEC_STEP_NOT_SET, param.stream));
249 :
250 0 : CHK_RET(RunTemplate(tempAlg, level1CommInfo));
251 0 : if (type == TemplateType::TEMPLATE_ALL_REDUCE_NHR) {
252 0 : tempAlg->CloseBarrier();
253 : }
254 0 : return HCCL_SUCCESS;
255 0 : }
256 :
257 : REGISTER_EXEC(
258 : "AllReduceMeshOpbaseSmallCountDeterministicExecutor", AllReduceMeshOpbaseSmallCountDeterministic,
259 : CollAllReduceMeshOpbaseSmallCountDeterministicExecutor);
260 :
261 : } // namespace hccl
|
