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_small_count_executor.h"
12 :
13 : namespace hccl {
14 :
15 0 : CollAllReduceMeshSmallCountExecutor::CollAllReduceMeshSmallCountExecutor(
16 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
17 0 : : CollAllReduceExecutor(dispatcher, topoMatcher)
18 : {
19 0 : DMAReduceFlag_ = true;
20 0 : desc_.deterministic = 1;
21 0 : }
22 :
23 0 : void CollAllReduceMeshSmallCountExecutor::ParseParam(const OpParam& param)
24 : {
25 0 : tag_ = param.tag;
26 0 : totalSize_ = param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
27 0 : aicpuUnfoldMode_ = param.aicpuUnfoldMode;
28 0 : }
29 :
30 0 : bool CollAllReduceMeshSmallCountExecutor::CalcScratchMemFlag(const u64 totalSize)
31 : {
32 0 : bool isDeter910B = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB
33 0 : && topoAttr_.deviceType == DevType::DEV_TYPE_910B
34 0 : && topoMatcher_->GetExternalInputHcclDeterministic() != DETERMINISTIC_DISABLE
35 0 : && topoAttr_.deviceNumPerAggregation > DEVICE_TWO
36 0 : && topoAttr_.deviceNumPerAggregation < DEVICE_EIGHT && totalSize <= HCCL_SMALL_COUNT_GRAPH_64_KB;
37 0 : return workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB
38 0 : && (isDeter910B || topoAttr_.deviceType == DevType::DEV_TYPE_910_93);
39 : }
40 :
41 0 : HcclResult CollAllReduceMeshSmallCountExecutor::CalcScratchMemSize(u64& scratchMemSize)
42 : {
43 0 : const u32 base = 2;
44 0 : if (CalcScratchMemFlag(totalSize_) == true) {
45 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
46 0 : scratchMemSize = totalSize_ * (topoAttr_.userRankSize - 1);
47 : } else {
48 0 : u64 factor = static_cast<u64>(log2(base * topoAttr_.userRankSize - 1));
49 0 : scratchMemSize = totalSize_ * factor;
50 : }
51 : } else {
52 0 : scratchMemSize = 0U;
53 : }
54 0 : HCCL_INFO(
55 : "[CollAllReduceMeshSmallCountExecutor][CalcScratchMemSize] tag[%s] scratchMemSize[%llu]", tag_.c_str(),
56 : scratchMemSize);
57 0 : return HCCL_SUCCESS;
58 : }
59 :
60 0 : HcclResult CollAllReduceMeshSmallCountExecutor::CalcStreamNum(u32& streamNum)
61 : {
62 0 : u32 totalStreamNum = 0U;
63 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) {
64 0 : totalStreamNum = topoAttr_.deviceNumPerAggregation - 1U;
65 : } else {
66 0 : totalStreamNum = topoAttr_.deviceNumPerAggregation;
67 : }
68 0 : streamNum = totalStreamNum - 1U;
69 0 : HCCL_INFO("[CollAllReduceMeshSmallCountExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
70 0 : return HCCL_SUCCESS;
71 : }
72 :
73 0 : HcclResult CollAllReduceMeshSmallCountExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
74 : {
75 0 : TransportMemType inputType = TransportMemType::RESERVED;
76 0 : TransportMemType outputType = TransportMemType::RESERVED;
77 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
78 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
79 0 : return HCCL_SUCCESS;
80 : }
81 :
82 : HcclResult
83 0 : CollAllReduceMeshSmallCountExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
84 : {
85 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
86 0 : inputType = TransportMemType::CCL_INPUT;
87 0 : outputType = TransportMemType::CCL_OUTPUT;
88 : } else {
89 0 : inputType = TransportMemType::PARAM_INPUT;
90 0 : if (CalcScratchMemFlag(totalSize_) == true) {
91 0 : outputType = TransportMemType::SCRATCH;
92 : } else {
93 0 : outputType = TransportMemType::PARAM_OUTPUT;
94 : }
95 : }
96 0 : HCCL_INFO(
97 : "[CollAllReduceMeshSmallCountExecutor][CalcTransportMemType]"
98 : "tag[%s] inputType[%d], outputType[%d]",
99 : tag_.c_str(), inputType, outputType);
100 0 : return HCCL_SUCCESS;
101 : }
102 :
103 0 : HcclResult CollAllReduceMeshSmallCountExecutor::CalcLevel0CommInfo(
104 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
105 : {
106 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
107 0 : commParaLevel0.meshSinglePlane = true;
108 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
109 0 : return HCCL_SUCCESS;
110 0 : }
111 :
112 0 : HcclResult CollAllReduceMeshSmallCountExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
113 : {
114 0 : HcclUs startut = TIME_NOW();
115 0 : ParseParam(param);
116 0 : algResResp_ = &algRes;
117 0 : ExecMem execMem;
118 0 : execMem.count = param.DataDes.count;
119 0 : execMem.inputPtr = param.inputPtr;
120 0 : execMem.outputPtr = param.outputPtr;
121 0 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
122 0 : execMem.inputMem = algRes.paramInputMem;
123 0 : execMem.outputMem = algRes.paramOutputMem;
124 0 : execMem.scratchMem = algRes.scratchMem;
125 : } else {
126 0 : execMem.inputMem = algRes.cclInputMem;
127 0 : execMem.outputMem = algRes.cclOutputMem;
128 0 : execMem.scratchMem = algRes.scratchMem;
129 : }
130 0 : HcclResult ret = KernelRun(param, execMem);
131 0 : CHK_PRT_RET(
132 : ret != HCCL_SUCCESS,
133 : HCCL_ERROR(
134 : "[CollAllReduceMeshSmallCountExecutor][Orchestrate]errNo[0x%016llx]executor kernel run failed",
135 : HCCL_ERROR_CODE(ret)),
136 : ret);
137 :
138 : // Enforce task launch at the end of Orchestrate
139 : // 注意: 不要删除这里的强制launch, 否则会导致aicpu cache功能问题
140 0 : if (!is310P3Common_) {
141 0 : HCCL_INFO("%s: enforce task launch at the end of Orchestrate", __func__);
142 0 : CHK_RET(LaunchTaskExtend(
143 : dispatcher_, const_cast<Stream&>(param.stream),
144 : const_cast<std::vector<Stream>&>(algResResp_->slaveStreams)));
145 : }
146 :
147 0 : HCCL_INFO(
148 : "[CollAllReduceMeshSmallCountExecutor]tag[%s], AllReduce executor orchestrate success, take time [%lld]us",
149 : param.tag.c_str(), DURATION_US(TIME_NOW() - startut));
150 0 : return HCCL_SUCCESS;
151 0 : }
152 :
153 0 : HcclResult CollAllReduceMeshSmallCountExecutor::GetAdjInfo(AlgResourceResponse& algRes, AdjInfo& adjInfo)
154 : {
155 : (void)algRes;
156 : (void)adjInfo;
157 0 : return HCCL_SUCCESS;
158 : }
159 :
160 0 : HcclResult CollAllReduceMeshSmallCountExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
161 : {
162 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] userRank[%u] starts.", __func__, topoAttr_.userRank);
163 0 : std::vector<Slice> dataSegsSlice; // 数据分成ranksize份，每份的起始偏移和大小
164 0 : if (!CalcScratchMemFlag(totalSize_)) {
165 0 : execMem.scratchMem = execMem.outputMem;
166 : }
167 :
168 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
169 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
170 :
171 0 : ReduceType reduceType
172 0 : = ((param.reduceType != HCCL_REDUCE_PROD) && (param.DataDes.dataType != HCCL_DATA_TYPE_INT64)) ?
173 : ReduceType::INLINE_REDUCE :
174 : ReduceType::TBE_REDUCE;
175 0 : auto originalAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
176 0 : u8 deterministic = topoMatcher_->GetExternalInputHcclDeterministic();
177 0 : auto opMeta = HcclOpMetaInfo::GetOneForAllReduce(
178 0 : originalAlgTypeLevel1, param.DataDes.dataType, reduceType, true, 1, false, CopyPattern::BCOPY, 1, false, true,
179 : false, deterministic);
180 0 : CHK_RET(InitTask(dispatcher_, const_cast<Stream&>(param.stream), opMeta.isEnableCache, opMeta.GetCacheKey()));
181 :
182 0 : CHK_RET(ActiveSlaveStreams(param.stream));
183 :
184 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, param.DataDes.dataType, param.reduceType);
185 0 : HcomCollOpInfo opInfo
186 0 : = {"", execMem.inputPtr, execMem.outputPtr, execMem.count, param.DataDes.dataType, param.root, param.reduceType,
187 0 : 0};
188 :
189 0 : bool isUsedRegister = false;
190 0 : std::unique_ptr<AlgTemplateBase> level0TempAlg;
191 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
192 0 : bool aicpu = true;
193 : level0TempAlg
194 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_HD_OPTIM, dispatcher_);
195 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_HD_OPTIM in COMM_LEVEL0", __func__);
196 0 : CHK_SMART_PTR_NULL(level0TempAlg);
197 0 : CHK_RET(level0TempAlg->Prepare(
198 : reduceAttr, algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
199 : level0CommInfo.localRank, &opInfo, aicpu));
200 0 : } else if (topoMatcher_->GetExternalInputHcclDeterministic() == DETERMINISTIC_DISABLE) {
201 0 : isUsedRegister = true;
202 0 : level0TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
203 0 : TemplateType::TEMPLATE_ALL_REDUCE_REDUCE_BCAST, dispatcher_);
204 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_REDUCE_BCAST in COMM_LEVEL0", __func__);
205 0 : } else if (topoAttr_.deviceNumPerAggregation == DEVICE_EIGHT) {
206 0 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE || aicpuUnfoldMode_) {
207 0 : level0TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
208 0 : TemplateType::TEMPLATE_ALL_REDUCE_DOUBLING, dispatcher_);
209 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_DOUBLING in COMM_LEVEL0", __func__);
210 0 : CHK_SMART_PTR_NULL(level0TempAlg);
211 0 : CHK_RET(level0TempAlg->Prepare(reduceAttr));
212 0 : } else {
213 0 : level0TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
214 0 : TemplateType::TEMPLATE_ALL_REDUCE_DOUBLING_DIRECT, dispatcher_);
215 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_DOUBLING_DIRECT in COMM_LEVEL0", __func__);
216 0 : CHK_SMART_PTR_NULL(level0TempAlg);
217 0 : CHK_RET(level0TempAlg->Prepare(reduceAttr, &opInfo));
218 : }
219 : } else {
220 0 : level0TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
221 0 : TemplateType::TEMPLATE_ALL_REDUCE_LOCAL_REDUCE_BCAST, dispatcher_);
222 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_LOCAL_REDUCE_BCAST in COMM_LEVEL0", __func__);
223 0 : CHK_SMART_PTR_NULL(level0TempAlg);
224 0 : CHK_RET(level0TempAlg->Prepare(
225 : reduceAttr, algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
226 : level0CommInfo.localRank, level0CommInfo.localRankSize, topoAttr_.userRank, &opInfo));
227 : }
228 0 : CHK_SMART_PTR_NULL(level0TempAlg);
229 :
230 0 : if (isUsedRegister) {
231 0 : PrepareData prepareData;
232 0 : prepareData.reduceAttr = reduceAttr;
233 0 : prepareData.subStreamsPtr = &algResResp_->slaveStreams;
234 0 : prepareData.signalPtr = &algResResp_->notifiesMain;
235 0 : prepareData.signalAuxPtr = &algResResp_->notifiesAux;
236 0 : prepareData.interRank = level0CommInfo.localRank;
237 0 : prepareData.interRankSize = level0CommInfo.localRankSize;
238 0 : prepareData.userRank = topoAttr_.userRank;
239 0 : prepareData.opInfo = &opInfo;
240 :
241 0 : prepareData.inputMem = execMem.inputMem;
242 0 : prepareData.outputMem = execMem.outputMem;
243 0 : prepareData.scratchMem = execMem.scratchMem;
244 0 : prepareData.count = execMem.count;
245 0 : prepareData.dataType = param.DataDes.dataType;
246 0 : prepareData.stream = param.stream;
247 0 : prepareData.reductionOp = param.reduceType;
248 0 : prepareData.slicesPtr = &dataSegsSlice;
249 :
250 0 : CHK_RET(level0TempAlg->Prepare(prepareData));
251 0 : } else {
252 0 : CHK_RET(level0TempAlg->Prepare(
253 : execMem.inputMem, execMem.scratchMem, execMem.outputMem, execMem.count, param.DataDes.dataType,
254 : param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, dataSegsSlice, 0));
255 : }
256 :
257 0 : CHK_RET(level0TempAlg->RegisterProfiler(
258 : (level0CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_2,
259 : HCCL_EXEC_STEP_NOT_SET, param.stream));
260 0 : CHK_RET(RunTemplate(level0TempAlg, level0CommInfo));
261 0 : HCCL_INFO("AllReduce small count executor run success.");
262 0 : return HCCL_SUCCESS;
263 0 : }
264 :
265 : REGISTER_EXEC("AllReduceMeshSmallCountExecutor", AllReduceMeshSmallCount, CollAllReduceMeshSmallCountExecutor);
266 :
267 : } // namespace hccl
|
