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