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(const HcclDispatcher dispatcher,
16 0 : std::unique_ptr<TopoMatcher> &topoMatcher): CollAllReduceExecutor(dispatcher, topoMatcher)
17 : {
18 0 : DMAReduceFlag_ = true;
19 0 : if (!IsPowerOfTwo(topoAttr_.deviceNumPerAggregation)) {
20 : // localreduce + broadcast rank0的cclout要用来存放要reduce的数据
21 0 : CCLMemSlice_ = false;
22 : }
23 0 : }
24 :
25 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::CalcStreamNum(u32& streamNum)
26 : {
27 : u32 totalStreamNum;
28 0 : if (!IsPowerOfTwo(topoAttr_.deviceNumPerAggregation)) {
29 : // level0 为localreduce+Bcast时，需要level0 ranksize条
30 0 : totalStreamNum = topoAttr_.deviceNumPerAggregation;
31 : } else {
32 : // Doubling、nhr/ring算法只需要一条主流
33 0 : totalStreamNum = 1U;
34 : }
35 :
36 0 : streamNum = totalStreamNum - 1U;
37 0 : HCCL_INFO("[CollAllReduceMeshOpbaseSmallCountDeterministicExecutor][CalcStreamNum] tag[%s] streamNum[%u]",
38 : tag_.c_str(), streamNum);
39 0 : return HCCL_SUCCESS;
40 : }
41 :
42 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
43 : {
44 0 : TransportMemType inputType = TransportMemType::RESERVED;
45 0 : TransportMemType outputType = TransportMemType::RESERVED;
46 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
47 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
48 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
49 0 : return HCCL_SUCCESS;
50 : }
51 :
52 0 : bool CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::IsPowerOfTwo(u32 num)
53 : {
54 0 : return (num & (num - 1)) == 0;
55 : }
56 :
57 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::CalcTransportMemType(TransportMemType &inputType,
58 : TransportMemType &outputType)
59 : {
60 0 : inputType = TransportMemType::CCL_INPUT;
61 0 : outputType = TransportMemType::CCL_OUTPUT;
62 0 : HCCL_INFO("[CollAllReduceMeshOpbaseSmallCountDeterministicExecutor][CalcTransportMemType]" \
63 : "tag[%s] inputType[%d], outputType[%d]",
64 : tag_.c_str(), inputType, outputType);
65 0 : return HCCL_SUCCESS;
66 : }
67 :
68 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::CalcLevel0CommInfo(TransportMemType inputType,
69 : TransportMemType outputType,
70 : 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(TransportMemType inputType,
87 : 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 :
128 0 : bool CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::IsSmallData(const u64 totalSize, const u64 curSize)
129 : {
130 : // 选到本执行器必为小数据量
131 0 : return true;
132 : }
133 :
134 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::KernelRun(const OpParam ¶m, ExecMem &execMem)
135 : {
136 0 : HCCL_CONFIG_INFO(HCCL_ALG,
137 : "[CollAllReduceMeshOpbaseSmallCountDeterministicExecutor][Run]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, param.root, param.reduceType};
159 0 : CHK_RET(RunReduceBcastSingleLevel(param, opInfo, reduceAttr, execMem, level0CommInfo));
160 0 : HCCL_INFO("allreduce small count deterministic: using reduce bcast algo intra-server.");
161 : }
162 : // Run level1
163 0 : if (IsPowerOfTwo(level1CommInfo.localRankSize)) {
164 0 : CHK_RET(RunDoublingSingleLevel(param, reduceAttr, execMem, level1CommInfo));
165 0 : HCCL_INFO("allreduce small count deterministic: using doubling algo inter-server.");
166 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_HD) {
167 0 : CHK_RET(RunTempLevel1(TemplateType::TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING, param, reduceAttr, execMem,
168 : level1CommInfo));
169 0 : HCCL_INFO("allreduce small count deterministic: using rhd algo inter-server.");
170 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
171 0 : CHK_RET(RunTempLevel1(TemplateType::TEMPLATE_ALL_REDUCE_RING, param, reduceAttr, execMem, level1CommInfo));
172 0 : HCCL_INFO("allreduce small count deterministic: using default ring algo inter-server.");
173 : } else {
174 : // 默认nhr
175 0 : CHK_RET(RunTempLevel1(TemplateType::TEMPLATE_ALL_REDUCE_NHR, param, reduceAttr, execMem, level1CommInfo));
176 0 : HCCL_INFO("allreduce small count deterministic: using nhr algo inter-server.");
177 : }
178 0 : DeviceMem dstMem(execMem.outputPtr, curSize);
179 0 : DeviceMem srcMem;
180 0 : if (IsPowerOfTwo(level1CommInfo.localRankSize)) {
181 : // cclin->userout
182 0 : srcMem = execMem.inputMem.range(0, curSize);
183 : } else {
184 : // cclout->userout
185 0 : srcMem = execMem.outputMem.range(0, curSize);
186 : }
187 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
188 :
189 0 : return HCCL_SUCCESS;
190 0 : }
191 :
192 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::RunDoublingSingleLevel(const OpParam ¶m, u64 reduceAttr, ExecMem &execMem,
193 : SubCommInfo &levelCommInfo)
194 : {
195 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
196 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_DOUBLING_LOCAL_REDUCE,
197 0 : dispatcher_);
198 0 : CHK_SMART_PTR_NULL(tempAlg);
199 0 : CHK_RET(tempAlg->Prepare(reduceAttr));
200 0 : CHK_RET(tempAlg->Prepare(execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count,
201 : param.DataDes.dataType, param.stream, param.reduceType,
202 : LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), 0));
203 :
204 0 : CHK_RET(tempAlg->RegisterProfiler(
205 : (levelCommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + levelCommInfo.localRank,
206 : PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
207 :
208 0 : CHK_RET(RunTemplate(tempAlg, levelCommInfo));
209 0 : return HCCL_SUCCESS;
210 0 : }
211 :
212 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::RunReduceBcastSingleLevel(const OpParam ¶m, HcomCollOpInfo& opInfo, u64 reduceAttr, ExecMem &execMem,
213 : SubCommInfo &levelCommInfo)
214 : {
215 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
216 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_LOCAL_REDUCE_BCAST, dispatcher_);
217 0 : CHK_SMART_PTR_NULL(tempAlg);
218 0 : CHK_RET(tempAlg->Prepare(reduceAttr, algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
219 : levelCommInfo.localRank, levelCommInfo.localRankSize, topoAttr_.userRank, &opInfo));
220 0 : CHK_SMART_PTR_NULL(tempAlg);
221 0 : CHK_RET(tempAlg->Prepare(execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count,
222 : param.DataDes.dataType, param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), 0));
223 :
224 0 : CHK_RET(tempAlg->RegisterProfiler(
225 : (levelCommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + levelCommInfo.localRank,
226 : PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
227 0 : CHK_RET(RunTemplate(tempAlg, levelCommInfo));
228 0 : return HCCL_SUCCESS;
229 0 : }
230 :
231 0 : HcclResult CollAllReduceMeshOpbaseSmallCountDeterministicExecutor::RunTempLevel1(const TemplateType type,
232 : const OpParam ¶m, u64 reduceAttr, ExecMem &execMem, SubCommInfo &level1CommInfo)
233 : {
234 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
235 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(type, dispatcher_);
236 0 : CHK_SMART_PTR_NULL(tempAlg);
237 0 : CHK_RET(tempAlg->Prepare(reduceAttr));
238 :
239 0 : CHK_RET(tempAlg->Prepare(execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count,
240 : param.DataDes.dataType, param.stream, param.reduceType,
241 : LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), 0));
242 0 : CHK_RET(tempAlg->RegisterProfiler((level1CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) +
243 : level1CommInfo.localRank, PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
244 :
245 0 : CHK_RET(RunTemplate(tempAlg, level1CommInfo));
246 0 : if (type == TemplateType::TEMPLATE_ALL_REDUCE_NHR) {
247 0 : tempAlg->CloseBarrier();
248 : }
249 0 : return HCCL_SUCCESS;
250 0 : }
251 :
252 : REGISTER_EXEC("AllReduceMeshOpbaseSmallCountDeterministicExecutor",
253 : AllReduceMeshOpbaseSmallCountDeterministic, CollAllReduceMeshOpbaseSmallCountDeterministicExecutor);
254 :
255 : } // namespace hccl
|
