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_reduce_scatter_order_preserved_executor.h"
12 :
13 : namespace hccl {
14 :
15 0 : CollReduceScatterOrderPreservedExecutor::CollReduceScatterOrderPreservedExecutor(
16 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
17 0 : : CollReduceScatterExecutor(dispatcher, topoMatcher)
18 : {
19 0 : DMAReduceFlag_ = true;
20 0 : }
21 :
22 0 : void CollReduceScatterOrderPreservedExecutor::ParseParam(const OpParam& param)
23 : {
24 0 : tag_ = param.tag;
25 :
26 : // 是否需要scratch memory（图模式没有cclbuffer，需要额外申请scratchMem）
27 0 : scratchMemFlag_ = (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE);
28 :
29 0 : u64 sizePerRank = param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
30 0 : totalSize_ = topoAttr_.userRankSize * sizePerRank;
31 :
32 : // 单算子场景 单机2次幂场景小数据量使用HD性能更优
33 0 : const bool isSmallData = sizePerRank <= HCCL_SMALL_COUNT_32_KB;
34 0 : const bool isSingleModule = topoAttr_.moduleNum == 1;
35 0 : const bool isOpBase = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE;
36 0 : const bool isPowerOfTwoDevices
37 0 : = (topoAttr_.deviceNumPerAggregation == DEVICE_EIGHT) || (topoAttr_.deviceNumPerAggregation == DEVICE_FOUR);
38 0 : isUseHDAlg_ = isSmallData && isSingleModule && isOpBase && isPowerOfTwoDevices;
39 0 : }
40 :
41 0 : HcclResult CollReduceScatterOrderPreservedExecutor::CalcScratchMemSize(u64& scratchMemSize)
42 : {
43 0 : scratchMemSize = scratchMemFlag_ ? totalSize_ : 0U;
44 0 : HCCL_INFO("[%s]tag[%s] scratchMemSize[%llu]", __func__, tag_.c_str(), scratchMemSize);
45 0 : return HCCL_SUCCESS;
46 : }
47 :
48 0 : u32 CollReduceScatterOrderPreservedExecutor::CalReduceStreamNum(const u32& localRankSize)
49 : {
50 0 : return (1 << static_cast<int>(std::floor(log2(localRankSize))));
51 : }
52 :
53 0 : HcclResult CollReduceScatterOrderPreservedExecutor::CalcStreamNum(u32& streamNum)
54 : {
55 0 : if (topoAttr_.deviceNumPerAggregation == 1) {
56 0 : u32 level1StreamNum = CalReduceStreamNum(topoAttr_.moduleNum);
57 0 : streamNum = std::min(level1StreamNum, DEVICE_EIGHT + DEVICE_EIGHT / FACTOR_NUM_TWO - 1);
58 0 : HCCL_INFO(
59 : "[%s]tag[%s] single rank per module, level1StreamNum[%u], streamNum[%u]", __func__, tag_.c_str(),
60 : level1StreamNum, streamNum);
61 0 : return HCCL_SUCCESS;
62 : }
63 :
64 : // Level0RankSize条流给alltoall，剩下的流给LocalReduce使用
65 0 : u32 level0StreamNum = topoAttr_.deviceNumPerAggregation - 1 + CalReduceStreamNum(topoAttr_.deviceNumPerAggregation);
66 : // level1主流分给alltoall，从流给LocalReduce使用
67 0 : u32 level1StreamNum = CalReduceStreamNum(topoAttr_.moduleNum);
68 : // 总流数上限：7（alltoall使用，提前的本地拷贝任务不需要并行）+ 4（LocalReduce使用）
69 : streamNum
70 0 : = std::min(std::max(level0StreamNum - 1, level1StreamNum), DEVICE_EIGHT + DEVICE_EIGHT / FACTOR_NUM_TWO - 1);
71 :
72 0 : HCCL_INFO(
73 : "[%s]tag[%s] level0StreamNum[%u], level1StreamNum[%u], streamNum[%u]", __func__, tag_.c_str(), level0StreamNum,
74 : level1StreamNum, streamNum);
75 0 : return HCCL_SUCCESS;
76 : }
77 :
78 0 : HcclResult CollReduceScatterOrderPreservedExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
79 : {
80 0 : TransportMemType inputType = TransportMemType::RESERVED;
81 0 : TransportMemType outputType = TransportMemType::RESERVED;
82 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
83 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
84 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
85 0 : return HCCL_SUCCESS;
86 : }
87 :
88 : HcclResult
89 0 : CollReduceScatterOrderPreservedExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
90 : {
91 : // scratchMemFlag_ 对应图模式场景（图模式没有cclbuffer）, PARAM_INPUT -> userInput
92 0 : inputType = scratchMemFlag_ ? TransportMemType::PARAM_INPUT : TransportMemType::CCL_INPUT;
93 0 : outputType = scratchMemFlag_ ? TransportMemType::SCRATCH : TransportMemType::CCL_OUTPUT;
94 0 : HCCL_INFO("[%s]tag[%s] inputType[%d], outputType[%d]", __func__, tag_.c_str(), inputType, outputType);
95 0 : return HCCL_SUCCESS;
96 : }
97 :
98 0 : HcclResult CollReduceScatterOrderPreservedExecutor::CalcLevel0CommInfo(
99 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
100 : {
101 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
102 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
103 0 : return HCCL_SUCCESS;
104 0 : }
105 :
106 0 : HcclResult CollReduceScatterOrderPreservedExecutor::CalcLevel1CommInfo(
107 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
108 : {
109 0 : if (topoAttr_.moduleNum > 1) {
110 0 : CommParaInfo commParaLevel1(COMM_LEVEL1, CommType::COMM_TAG_MESH);
111 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel1, opTransport[COMM_LEVEL1], inputType, outputType));
112 0 : }
113 0 : return HCCL_SUCCESS;
114 : }
115 :
116 0 : bool CollReduceScatterOrderPreservedExecutor::IsSmallData(const u64 totalSize, const u64 curSize)
117 : {
118 : (void)curSize;
119 : // 子图复用的阈值（opmeta全一致时，ffts子图复用）
120 0 : return totalSize <= HCCL_SMALL_COUNT_32_KB;
121 : }
122 :
123 0 : HcclResult CollReduceScatterOrderPreservedExecutor::RunReduceScatterLevel0SingleRank(
124 : const OpParam& param, ExecMem& execMem, const SubCommInfo& level0CommInfo) const
125 : {
126 : (void)level0CommInfo;
127 0 : u64 unitSize = SIZE_TABLE[param.DataDes.dataType];
128 0 : u64 curSize = execMem.count * unitSize;
129 0 : DeviceMem bufferMem = scratchMemFlag_ ? execMem.scratchMem : execMem.inputMem;
130 0 : DeviceMem dstMem;
131 0 : DeviceMem srcMem;
132 0 : for (u32 i = 0; i < topoAttr_.userRankSize; i++) {
133 : // 拷贝input上每个slice的数据到中转内存，源端每个slice的size固定为output的size
134 0 : dstMem = bufferMem.range(curSize * i, curSize);
135 0 : srcMem = DeviceMem::create(static_cast<u8*>(execMem.inputPtr) + param.DataDes.count * unitSize * i, curSize);
136 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
137 : }
138 0 : return HCCL_SUCCESS;
139 0 : }
140 :
141 0 : HcclResult CollReduceScatterOrderPreservedExecutor::RunReduceScatterLevel0HD(
142 : const OpParam& param, ExecMem& execMem, SubCommInfo& level0CommInfo)
143 : {
144 : std::unique_ptr<AlgTemplateBase> level0TempAlg
145 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_HDSTAGE, dispatcher_);
146 :
147 0 : std::vector<Slice> dataSegsSlice; // 数据分成ranksize份，每份的起始偏移和大小
148 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, param.DataDes.dataType, param.reduceType);
149 0 : HcomCollOpInfo opInfo = {"",
150 0 : execMem.inputPtr,
151 0 : execMem.outputPtr,
152 0 : param.DataDes.count,
153 0 : param.DataDes.dataType,
154 0 : param.root,
155 0 : param.reduceType,
156 0 : 0};
157 :
158 0 : CHK_SMART_PTR_NULL(level0TempAlg);
159 0 : CHK_RET(level0TempAlg->Prepare(
160 : execMem.inputMem, execMem.scratchMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.stream,
161 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, dataSegsSlice, 0, reduceAttr, algResResp_->slaveStreams,
162 : algResResp_->notifiesMain, algResResp_->notifiesAux, topoAttr_.userRank, &opInfo));
163 :
164 0 : CHK_RET(level0TempAlg->RegisterProfiler(
165 : (level0CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_2,
166 : HCCL_EXEC_STEP_NOT_SET, param.stream));
167 0 : CHK_RET(RunTemplate(level0TempAlg, level0CommInfo));
168 0 : return HCCL_SUCCESS;
169 0 : }
170 :
171 0 : HcclResult CollReduceScatterOrderPreservedExecutor::RunReduceScatterLevel0(
172 : const OpParam& param, ExecMem& execMem, SubCommInfo& level0CommInfo)
173 : {
174 0 : if (level0CommInfo.localRankSize == 1) {
175 0 : all2allOffset_ = topoAttr_.moduleNum > 1 ? 1 : 0;
176 0 : HCCL_INFO("[%s] single rank per module, skip L0 AllToAll and LocalReduce, tag[%s]", __func__, tag_.c_str());
177 0 : CHK_RET(RunReduceScatterLevel0SingleRank(param, execMem, level0CommInfo));
178 0 : return HCCL_SUCCESS;
179 : }
180 :
181 0 : CHK_RET(ActiveSlaveStreams(param.stream));
182 0 : if (isUseHDAlg_) {
183 0 : CHK_RET(RunReduceScatterLevel0HD(param, execMem, level0CommInfo));
184 : } else {
185 : // 切分数据(ReduceScatter分组，记录每组的起始偏移和大小)
186 0 : GroupSlicesInfo groupSlicesInfoLevel0;
187 0 : u64 size = execMem.count * SIZE_TABLE[param.DataDes.dataType];
188 0 : for (u32 groupId = 0; groupId < topoAttr_.moduleNum; groupId++) {
189 0 : MemBlockInfo memInfo;
190 0 : for (u32 dataId = 0; dataId < level0CommInfo.localRankSize; dataId++) {
191 0 : u64 offset = (dataId + groupId * level0CommInfo.localRankSize) * size;
192 0 : u64 userMemInOffset = param.DataDes.count * SIZE_TABLE[param.DataDes.dataType]
193 0 : * (dataId + groupId * level0CommInfo.localRankSize);
194 0 : memInfo.size.push_back(size);
195 0 : memInfo.userInputOffsets.push_back(userMemInOffset);
196 0 : memInfo.inputOffsets.push_back(offset);
197 0 : memInfo.outputOffsets.push_back(offset);
198 : }
199 0 : groupSlicesInfoLevel0.push_back(memInfo);
200 0 : }
201 :
202 0 : all2allOffset_ = topoAttr_.moduleNum > 1 ? 1 : 0; // 多机场景需要偏移1（给L1预留计算位，减少拷贝次数）
203 0 : std::unique_ptr<AlgTemplateBase> level0TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
204 0 : TemplateType::TEMPLATE_REDUCESCATTER_PLANT_LOCAL_REDUCE, dispatcher_);
205 0 : CHK_SMART_PTR_NULL(level0TempAlg);
206 :
207 : // execMem.scratchMem在单算子模式下为cclout，图模式为scrach，因此output传入scrach即可
208 0 : CHK_RET(level0TempAlg->Prepare(
209 : execMem.inputPtr, execMem.inputMem, execMem.scratchMem, param.stream, algResResp_->slaveStreams,
210 : algResResp_->notifiesMain, algResResp_->notifiesAux, groupSlicesInfoLevel0, param.reduceType,
211 : all2allOffset_, param.DataDes.dataType, false));
212 0 : CHK_RET(level0TempAlg->RegisterProfiler(
213 : (level0CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_2,
214 : HCCL_EXEC_STEP_NOT_SET, param.stream));
215 0 : CHK_RET(RunTemplate(level0TempAlg, level0CommInfo));
216 0 : }
217 0 : return HCCL_SUCCESS;
218 : }
219 :
220 0 : HcclResult CollReduceScatterOrderPreservedExecutor::RunReduceScatterLevel1(
221 : const OpParam& param, ExecMem& execMem, SubCommInfo& level0CommInfo)
222 : {
223 0 : u32 commIndex = level0CommInfo.localRank;
224 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
225 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
226 :
227 : // 切分数据，记录每组的起始偏移和大小（仅1组）
228 0 : u64 size = execMem.count * SIZE_TABLE[param.DataDes.dataType];
229 0 : MemBlockInfo memInfo;
230 0 : u32 level0Ranksize = level0CommInfo.localRankSize;
231 0 : u32 inputBaseIndex
232 0 : = (all2allOffset_ + commIndex) % level0Ranksize; // 多机场景需要偏移1（给L1预留计算位，减少拷贝次数）
233 0 : for (u32 dataId = 0; dataId < level1CommInfo.localRankSize; dataId++) {
234 0 : u64 inputIndex = inputBaseIndex + dataId * level0Ranksize;
235 0 : memInfo.inputOffsets.push_back(inputIndex * size);
236 0 : u64 outputIndex = commIndex + dataId * level0Ranksize;
237 0 : memInfo.outputOffsets.push_back(outputIndex * size);
238 0 : memInfo.userInputOffsets.push_back(outputIndex * size);
239 0 : memInfo.size.push_back(size);
240 : }
241 :
242 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
243 0 : TemplateType::TEMPLATE_REDUCESCATTER_PLANT_LOCAL_REDUCE_COMBINE, dispatcher_);
244 0 : CHK_SMART_PTR_NULL(level1TempAlg);
245 :
246 0 : u32 level0LastRank = level0Ranksize - 1;
247 0 : bool isUseCclIn = (level0Ranksize == 1) || (commIndex == level0LastRank - 1);
248 0 : bool borrowSpace = level0Ranksize == 1;
249 0 : CHK_RET(level1TempAlg->Prepare(
250 : execMem.inputMem, execMem.scratchMem, param.stream, algResResp_->slaveStreams, algResResp_->notifiesMain,
251 : algResResp_->notifiesAux, memInfo, param.reduceType, param.DataDes.dataType, isUseCclIn,
252 : commIndex == level0LastRank, borrowSpace));
253 0 : CHK_RET(level1TempAlg->RegisterProfiler(
254 : (level0Ranksize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_2,
255 : HCCL_EXEC_STEP_NOT_SET, param.stream));
256 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
257 0 : return HCCL_SUCCESS;
258 0 : }
259 :
260 0 : HcclResult CollReduceScatterOrderPreservedExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
261 : {
262 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s]CollReduceScatterOrderPreservedExecutor starts, tag[%s]", __func__, tag_.c_str());
263 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
264 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
265 :
266 : // L0 节点内 reduce scatter
267 0 : CHK_RET(RunReduceScatterLevel0(param, execMem, level0CommInfo));
268 : // L1 节点间 reduce scatter
269 0 : if (topoAttr_.moduleNum > 1) {
270 0 : CHK_RET(RunReduceScatterLevel1(param, execMem, level0CommInfo));
271 : }
272 :
273 0 : if (!isUseHDAlg_) {
274 : // 非HD算法 execMem.scratchMem最后拷贝至UserOut
275 0 : u64 dataSize = execMem.count * SIZE_TABLE[param.DataDes.dataType];
276 0 : DeviceMem srcMem = execMem.scratchMem.range(dataSize * topoAttr_.userRank, dataSize);
277 0 : DeviceMem dstMem = DeviceMem::create(execMem.outputPtr, dataSize);
278 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
279 0 : }
280 :
281 0 : HCCL_INFO("[%s]order preserved ReduceScatter run success, tag[%s]", __func__, tag_.c_str());
282 0 : return HCCL_SUCCESS;
283 0 : }
284 :
285 : REGISTER_EXEC(
286 : "ReduceScatterOrderPreservedExecutor", ReduceScatterOrderPreserved, CollReduceScatterOrderPreservedExecutor);
287 : } // namespace hccl
|
