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