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_order_preserved_for_910_93_executor.h"
12 :
13 : namespace hccl {
14 :
15 0 : CollAllReduceOrderPreservedFor91093Executor::CollAllReduceOrderPreservedFor91093Executor(const HcclDispatcher dispatcher,
16 0 : std::unique_ptr<TopoMatcher> &topoMatcher)
17 0 : : CollAllReduceExecutor(dispatcher, topoMatcher)
18 : {
19 0 : DMAReduceFlag_ = true;
20 0 : CCLMemSlice_ = false;
21 0 : desc_.deterministic = DETERMINISTIC_STRICT;
22 0 : }
23 :
24 0 : void CollAllReduceOrderPreservedFor91093Executor::ParseParam(const OpParam& param)
25 : {
26 0 : tag_ = param.tag;
27 0 : aicpuUnfoldMode_ = param.aicpuUnfoldMode;
28 :
29 0 : u64 sizePerBlock = (param.DataDes.count + topoAttr_.userRankSize - 1) / topoAttr_.userRankSize
30 0 : * SIZE_TABLE[param.DataDes.dataType];
31 0 : sizePerBlock = AlgTemplateBase::RoundUpWithDivisor(sizePerBlock, HCCL_MIN_SLICE_ALIGN);
32 :
33 : // 是否需要scratch memory（图模式没有cclbuffer，需要额外申请scratchMem）
34 0 : u64 inputSize = param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
35 0 : scratchMemFlag_ = (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) &&
36 0 : (inputSize < (topoAttr_.userRankSize - 1) * sizePerBlock);
37 :
38 0 : totalSize_ = std::max(sizePerBlock * topoAttr_.userRankSize, inputSize);
39 0 : }
40 :
41 0 : HcclResult CollAllReduceOrderPreservedFor91093Executor::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 CollAllReduceOrderPreservedFor91093Executor::CalReduceStreamNum(const u32& localRankSize) const
49 : {
50 0 : return (1 << static_cast<int>(std::floor(log2(localRankSize))));
51 : }
52 :
53 0 : HcclResult CollAllReduceOrderPreservedFor91093Executor::CalcStreamNum(u32& streamNum)
54 : {
55 : // 获取超节点内rank数
56 0 : u32 devNumInlocalPod = 0;
57 0 : u32 rankIdxInPod = 0;
58 0 : CHK_RET(topoMatcher_->GetLocalSuperPodRankSize(topoAttr_.userRank, devNumInlocalPod, rankIdxInPod));
59 :
60 0 : if (devNumInlocalPod == 1) {
61 0 : u32 level2StreamNum = std::min(CalReduceStreamNum(topoAttr_.superPodNum) - 1, DEVICE_FOUR);
62 0 : streamNum = level2StreamNum;
63 0 : HCCL_INFO("[%s]tag[%s] single rank per module, level2StreamNum[%u], streamNum[%u]",
64 : __func__, tag_.c_str(), level2StreamNum, streamNum);
65 0 : return HCCL_SUCCESS;
66 : }
67 :
68 : // all2allStreamNum条流给alltoall
69 0 : u32 all2allStreamNum = std::min(devNumInlocalPod, DEVICE_EIGHT);
70 : // reduceStreamNum主流分给alltoall，从流给LocalReduce使用
71 0 : u32 reduceStreamNum = std::min(CalReduceStreamNum(devNumInlocalPod) - 1, DEVICE_FOUR);
72 : // level2StreamNum超节点间reducescatter
73 0 : u32 level2StreamNum = std::min(CalReduceStreamNum(topoAttr_.superPodNum) - 1, DEVICE_FOUR);
74 : // 总流数上限：7（alltoall使用，提前的本地拷贝任务不需要并行）+ 4（LocalReduce使用）
75 0 : streamNum = std::max(all2allStreamNum + reduceStreamNum - 1, level2StreamNum);
76 :
77 0 : HCCL_INFO("[%s]tag[%s] all2allStreamNum[%u], reduceStreamNum[%u], level2StreamNum[%u], streamNum[%u]", __func__, tag_.c_str(),
78 : all2allStreamNum, reduceStreamNum, level2StreamNum, streamNum);
79 0 : return HCCL_SUCCESS;
80 : }
81 :
82 0 : HcclResult CollAllReduceOrderPreservedFor91093Executor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
83 : {
84 0 : TransportMemType inputType = TransportMemType::RESERVED;
85 0 : TransportMemType outputType = TransportMemType::RESERVED;
86 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
87 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
88 0 : CHK_RET(CalcLevel2CommInfo(inputType, outputType, opTransport));
89 0 : return HCCL_SUCCESS;
90 : }
91 :
92 0 : HcclResult CollAllReduceOrderPreservedFor91093Executor::CalcTransportMemType(TransportMemType &inputType,
93 : TransportMemType &outputType) const
94 : {
95 : // 图模式场景使用PARAM_INPUT/OUTPUT -> userInput/userOutPut，不需要scrachMem
96 0 : inputType = workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE ?
97 : TransportMemType::PARAM_INPUT : TransportMemType::CCL_INPUT;
98 0 : outputType = workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE ?
99 : TransportMemType::PARAM_OUTPUT : TransportMemType::CCL_OUTPUT;
100 :
101 0 : if (scratchMemFlag_) {
102 0 : outputType = TransportMemType::SCRATCH;
103 : }
104 :
105 0 : HCCL_INFO("[%s]tag[%s] inputType[%d], outputType[%d]", __func__, tag_.c_str(), inputType, outputType);
106 0 : return HCCL_SUCCESS;
107 : }
108 :
109 0 : HcclResult CollAllReduceOrderPreservedFor91093Executor::CalcLevel1CommInfo(TransportMemType inputType,
110 : TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
111 : {
112 0 : CommParaInfo commParaLevel1(COMM_COMBINE_L1, CommType::COMM_TAG_MESH);
113 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel1, opTransport[COMM_COMBINE_L1], inputType, outputType));
114 0 : return HCCL_SUCCESS;
115 0 : }
116 :
117 0 : HcclResult CollAllReduceOrderPreservedFor91093Executor::CalcLevel2CommInfo(TransportMemType inputType,
118 : TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
119 : {
120 0 : if (topoAttr_.superPodNum > 1) {
121 0 : CommParaInfo commParaLevel2(COMM_LEVEL2, CommType::COMM_TAG_MESH);
122 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel2, opTransport[COMM_LEVEL2], inputType, outputType));
123 0 : }
124 0 : return HCCL_SUCCESS;
125 : }
126 :
127 0 : bool CollAllReduceOrderPreservedFor91093Executor::IsHugeData(const u64 curSize)
128 : {
129 0 : bool hugeData = curSize / topoAttr_.deviceNumPerAggregation / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE ||
130 : curSize > SDMA_SEND_MAX_SIZE;
131 0 : HCCL_DEBUG("[%s]isHugeData[%d], curSize[%llu], topoAttr_.deviceNumPerAggregation[%u]",
132 : __func__, hugeData, curSize, topoAttr_.deviceNumPerAggregation);
133 0 : return hugeData;
134 : }
135 :
136 0 : void CollAllReduceOrderPreservedFor91093Executor::CalcSizePerBlock(const OpParam ¶m, ExecMem &execMem)
137 : {
138 0 : sizePerBlock_ = (execMem.count + topoAttr_.userRankSize - 1) / topoAttr_.userRankSize
139 0 : * SIZE_TABLE[param.DataDes.dataType];
140 0 : sizePerBlock_ = AlgTemplateBase::RoundUpWithDivisor(sizePerBlock_, HCCL_MIN_SLICE_ALIGN);
141 0 : }
142 :
143 0 : void CollAllReduceOrderPreservedFor91093Executor::CalGroupSlices(const OpParam ¶m, const ExecMem &execMem)
144 : {
145 0 : groupSize_.clear();
146 0 : u64 sizeRemain = execMem.count * SIZE_TABLE[param.DataDes.dataType];
147 0 : for (u32 rankId = 0; rankId < topoAttr_.userRankSize; rankId++) {
148 0 : u64 size = (sizeRemain > sizePerBlock_) ? sizePerBlock_ : sizeRemain;
149 0 : groupSize_.push_back(size);
150 0 : sizeRemain -= size;
151 : }
152 0 : }
153 :
154 0 : HcclResult CollAllReduceOrderPreservedFor91093Executor::RunReduceScatterLevel1(const OpParam ¶m, ExecMem &execMem,
155 : SubCommInfo &level1CommInfo)
156 : {
157 0 : if (level1CommInfo.localRankSize == 1) {
158 0 : all2allOffset_ = topoAttr_.superPodNum > 1 ? 1 : 0;
159 0 : HCCL_INFO("[%s] single rank per module, skip L1 AllToAll and LocalReduce, tag[%s]",
160 : __func__, tag_.c_str());
161 0 : CHK_RET(RunReduceScatterLevel1SingleRank(param, execMem, level1CommInfo));
162 0 : return HCCL_SUCCESS;
163 : }
164 :
165 : // 切分数据(ReduceScatter分组，记录每组的起始偏移和大小)
166 0 : GroupSlicesInfo groupSlicesInfoLevel1;
167 0 : for (u32 groupId = 0; groupId < topoAttr_.superPodNum; groupId++) {
168 0 : MemBlockInfo memInfo;
169 0 : for (u32 dataId = 0; dataId < level1CommInfo.localRankSize; dataId ++) {
170 0 : u64 globalDataId = groupId * level1CommInfo.localRankSize + dataId;
171 0 : u64 size = groupSize_[globalDataId];
172 0 : u64 offset = globalDataId * sizePerBlock_;
173 0 : memInfo.size.push_back(size);
174 0 : memInfo.userInputOffsets.push_back(offset);
175 0 : memInfo.inputOffsets.push_back(offset);
176 0 : memInfo.outputOffsets.push_back(offset);
177 : }
178 0 : groupSlicesInfoLevel1.push_back(memInfo);
179 0 : }
180 :
181 0 : CHK_RET(ActiveSlaveStreams(param.stream));
182 0 : all2allOffset_ = topoAttr_.superPodNum > 1 ? 1 : 0; // 多机场景需要偏移1（给L1预留计算位，减少拷贝次数）
183 :
184 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
185 0 : TemplateType::TEMPLATE_REDUCESCATTER_PLANT_LOCAL_REDUCE, dispatcher_);
186 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_PLANT_LOCAL_REDUCE in COMM_COMBINE_L1", __func__);
187 0 : CHK_SMART_PTR_NULL(level1TempAlg);
188 :
189 0 : DeviceMem outputMem = scratchMemFlag_ ? execMem.scratchMem : execMem.outputMem;
190 0 : CHK_RET(level1TempAlg->Prepare(execMem.inputPtr, execMem.inputMem, outputMem, param.stream,
191 : algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux, groupSlicesInfoLevel1,
192 : param.reduceType, all2allOffset_, param.DataDes.dataType, true, false, true));
193 :
194 0 : CHK_RET(level1TempAlg->RegisterProfiler(
195 : (level1CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank,
196 : PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET, param.stream));
197 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
198 0 : return HCCL_SUCCESS;
199 0 : }
200 :
201 0 : HcclResult CollAllReduceOrderPreservedFor91093Executor::RunReduceScatterLevel1SingleRank(const OpParam ¶m,
202 : ExecMem &execMem, const SubCommInfo &level1CommInfo) const
203 : {
204 : (void) level1CommInfo;
205 0 : u64 size = execMem.count * SIZE_TABLE[param.DataDes.dataType];
206 :
207 0 : DeviceMem srcMem = DeviceMem::create(execMem.inputPtr, size);
208 0 : DeviceMem dstMem = scratchMemFlag_ ? execMem.scratchMem.range(0, size) : execMem.inputMem.range(0, size);
209 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
210 :
211 0 : return HCCL_SUCCESS;
212 0 : }
213 :
214 0 : HcclResult CollAllReduceOrderPreservedFor91093Executor::RunReduceScatterLevel2(const OpParam ¶m, ExecMem &execMem,
215 : SubCommInfo &level1CommInfo)
216 : {
217 0 : u32 commIndex = level1CommInfo.localRank;
218 0 : u32 level1Ranksize = level1CommInfo.localRankSize;
219 0 : CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1));
220 0 : SubCommInfo level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
221 :
222 : // 切分数据，记录每组的起始偏移和大小（仅1组）
223 0 : MemBlockInfo memInfo;
224 0 : u32 inputBaseIndex = (all2allOffset_ + commIndex) % level1Ranksize; // 多机场景需要偏移1（给L1预留计算位，减少拷贝次数）
225 0 : for (u32 dataId = 0; dataId < level2CommInfo.localRankSize; dataId ++) {
226 0 : u64 inputIndex = inputBaseIndex + dataId * level1Ranksize;
227 0 : memInfo.inputOffsets.push_back(inputIndex * sizePerBlock_);
228 0 : memInfo.size.push_back(groupSize_[commIndex + dataId * level1Ranksize]);
229 0 : u64 outputIndex = commIndex + dataId * level1Ranksize;
230 0 : memInfo.outputOffsets.push_back(outputIndex * sizePerBlock_);
231 0 : memInfo.userInputOffsets.push_back(outputIndex * sizePerBlock_);
232 : }
233 :
234 0 : DeviceMem outputMem = scratchMemFlag_ ? execMem.scratchMem : execMem.outputMem;
235 0 : std::unique_ptr<AlgTemplateBase> level2TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
236 0 : TemplateType::TEMPLATE_REDUCESCATTER_PLANT_LOCAL_REDUCE_COMBINE, dispatcher_);
237 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_PLANT_LOCAL_REDUCE_COMBINE in COMM_LEVEL2", __func__);
238 0 : CHK_SMART_PTR_NULL(level2TempAlg);
239 :
240 0 : u32 level1LastRank = level1Ranksize - 1;
241 0 : bool isUseCclIn = level1Ranksize == 1 || commIndex == level1LastRank - 1;
242 0 : CHK_RET(level2TempAlg->Prepare(execMem.inputMem, outputMem, param.stream, algResResp_->slaveStreams,
243 : algResResp_->notifiesMain, algResResp_->notifiesAux, memInfo, param.reduceType,
244 : param.DataDes.dataType, isUseCclIn, commIndex == level1LastRank, true));
245 :
246 0 : CHK_RET(level2TempAlg->RegisterProfiler((level1Ranksize << PROF_RANKSIZE_OFFSET_OF_PLANEID) +
247 : level1CommInfo.localRank, PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET, param.stream));
248 0 : CHK_RET(RunTemplate(level2TempAlg, level2CommInfo));
249 0 : return HCCL_SUCCESS;
250 0 : }
251 :
252 0 : HcclResult CollAllReduceOrderPreservedFor91093Executor::RunAllGatherLevel1(const OpParam ¶m, ExecMem &execMem,
253 : SubCommInfo &level1CommInfo)
254 : {
255 0 : u32 level1RankSize = level1CommInfo.localRankSize;
256 0 : u32 commIndex = level1CommInfo.localRank;
257 0 : u64 count = execMem.count / topoAttr_.userRankSize;
258 0 : u64 serverOffsetConut = topoAttr_.userRank / level1RankSize * level1RankSize;
259 :
260 : // allgather 计算slice，数据分成ranksize份，每份的起始偏移和大小
261 0 : std::vector<Slice> dataSegsSlice;
262 0 : for (u32 rank = 0; rank < level1RankSize; rank++) {
263 0 : Slice userslice;
264 0 : userslice.size = groupSize_[rank + serverOffsetConut];
265 0 : userslice.offset = userslice.size == 0 ? 0 : (rank + serverOffsetConut) * sizePerBlock_;
266 0 : dataSegsSlice.emplace_back(std::move(userslice));
267 : }
268 :
269 0 : DeviceMem outputMem = scratchMemFlag_ ? execMem.scratchMem : execMem.outputMem;
270 :
271 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
272 0 : TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
273 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NHR in COMM_COMBINE_L1", __func__);
274 :
275 0 : CHK_SMART_PTR_NULL(level1TempAlg);
276 0 : CHK_RET(level1TempAlg->Prepare(outputMem, outputMem, outputMem, count, param.DataDes.dataType, param.stream,
277 : HCCL_REDUCE_RESERVED, LEVEL0_BRIDGE_RANK_ID, dataSegsSlice, 0));
278 :
279 0 : CHK_RET(level1TempAlg->RegisterProfiler((level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + commIndex,
280 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
281 :
282 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
283 0 : return HCCL_SUCCESS;
284 0 : }
285 :
286 0 : HcclResult CollAllReduceOrderPreservedFor91093Executor::RunAllGatherLevel2(const OpParam ¶m, const ExecMem &execMem,
287 : const SubCommInfo &level1CommInfo)
288 : {
289 0 : CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1));
290 0 : SubCommInfo level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
291 :
292 0 : std::unique_ptr<AlgTemplateBase> level2TempAlg; // Level1Allgather(根据算法选择)
293 0 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_RING) {
294 0 : level2TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
295 0 : TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
296 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_RING in COMM_LEVEL2", __func__);
297 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
298 0 : level2TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
299 0 : TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
300 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NB in COMM_LEVEL2", __func__);
301 : } else {
302 0 : level2TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
303 0 : TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
304 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NHR in COMM_LEVEL2", __func__);
305 : }
306 :
307 0 : DeviceMem outputMem = scratchMemFlag_ ? execMem.scratchMem : execMem.outputMem;
308 :
309 0 : u32 level1RankSize = level1CommInfo.localRankSize;
310 0 : u64 count = execMem.count / level2CommInfo.localRankSize;
311 :
312 0 : std::vector<u64> level2GroupSize;
313 0 : for (u32 rank = 0; rank < level2CommInfo.localRankSize; rank++) {
314 0 : u64 size = 0;
315 0 : for (u32 level1RankId = 0; level1RankId < level1RankSize; level1RankId++) {
316 0 : size += groupSize_[rank * level1RankSize + level1RankId];
317 : }
318 0 : level2GroupSize.push_back(size);
319 : }
320 :
321 : // allgather 计算slice，数据分成ranksize份，每份的起始偏移和大小
322 0 : std::vector<Slice> dataSegsSlice;
323 0 : for (u32 rank = 0; rank < level2CommInfo.localRankSize; rank++) {
324 0 : Slice userslice;
325 0 : userslice.size = level2GroupSize[rank];
326 0 : userslice.offset = rank * level1RankSize * sizePerBlock_;
327 0 : dataSegsSlice.emplace_back(std::move(userslice));
328 : }
329 :
330 0 : CHK_SMART_PTR_NULL(level2TempAlg);
331 0 : CHK_RET(level2TempAlg->Prepare(outputMem, outputMem, outputMem, count, param.DataDes.dataType, param.stream,
332 : HcclReduceOp::HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID, dataSegsSlice));
333 :
334 0 : CHK_RET(level2TempAlg->RegisterProfiler((level2CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID)
335 : + level2CommInfo.localRank, PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET, param.stream));
336 0 : CHK_RET(RunTemplate(level2TempAlg, level2CommInfo));
337 0 : return HCCL_SUCCESS;
338 0 : }
339 :
340 0 : HcclResult CollAllReduceOrderPreservedFor91093Executor::KernelRun(const OpParam ¶m, ExecMem &execMem)
341 : {
342 0 : HCCL_CONFIG_INFO(HCCL_ALG,
343 : "[%s]The CollAllReduceOrderPreservedFor91093Executor starts, tag[%s]", __func__, tag_.c_str());
344 0 : CHK_RET(CheckCommSize(COMM_COMBINE_L1, COMM_INDEX_0 + 1));
345 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_COMBINE_L1, COMM_INDEX_0);
346 :
347 0 : CalcSizePerBlock(param, execMem);
348 0 : CalGroupSlices(param, execMem);
349 :
350 0 : u64 inputSize = param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
351 0 : scratchMemFlag_ = (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) &&
352 0 : (inputSize < (topoAttr_.userRankSize - 1) * sizePerBlock_);
353 :
354 : // L1 节点内 reduce scatter
355 0 : CHK_RET(RunReduceScatterLevel1(param, execMem, level1CommInfo));
356 : // L2 节点间 reduce scatter
357 0 : if (topoAttr_.superPodNum > 1) {
358 0 : CHK_RET(RunReduceScatterLevel2(param, execMem, level1CommInfo));
359 : }
360 :
361 : // Level1 节点内 AllGatherMeshAtomic
362 0 : CHK_RET(RunAllGatherLevel1(param, execMem, level1CommInfo));
363 0 : if (topoAttr_.superPodNum > 1) {
364 : // L2 节点间 allgather
365 0 : CHK_RET(RunAllGatherLevel2(param, execMem, level1CommInfo));
366 : }
367 :
368 : // 单算子需要 execMem.outputMem最后拷贝至UserOut
369 0 : if (scratchMemFlag_ || workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
370 0 : u64 dataSize = execMem.count * SIZE_TABLE[param.DataDes.dataType];
371 0 : void *srcPtr = scratchMemFlag_ ? execMem.scratchMem.ptr() : execMem.outputMem.ptr();
372 0 : DeviceMem srcMem = DeviceMem::create(srcPtr, dataSize);
373 0 : DeviceMem dstMem = DeviceMem::create(execMem.outputPtr, dataSize);
374 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
375 0 : }
376 :
377 0 : HCCL_INFO("[%s]order preserved AllReduce run success, tag[%s]", __func__, tag_.c_str());
378 0 : return HCCL_SUCCESS;
379 0 : }
380 :
381 : REGISTER_EXEC("AllReduceOrderPreservedFor91093Executor", AllReduceOrderPreservedFor91093, CollAllReduceOrderPreservedFor91093Executor);
382 : }
|
