Star507
1017
代码
代码
Issues88
Pull Requests378
Actions
项目讨论
Wiki
分析
Star507
1017
已合并
AG #1931
千弘创建于 4月25日
AG #1931
已合并
千弘创建于 4月25日
从zwlStateflow/hcommmaster合入到cann/hcommmaster
共 7 个文件变更+870-3
Asrc/algorithm/impl/coll_executor/coll_all_gather/coll_all_gather_pipeline_for_910_93_executor.cc+567-0
@@ -0,0 +1,567 @@
1+/**
2+ * Copyright (c) 2026 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_gather_pipeline_for_910_93_executor.h"
12+#include "hccl_types.h"
13+#include "alg_template_register.h"
14+#include "all_gather_nhr_pub.h"
15+#include "aligned_all_gather_double_ring_pub.h"
16+#include "alg_template_base_pub.h"
17+ 
18+namespace hccl {
19+constexpr u32 PIPELINE_NUM = 2;
20+ 
21+CollAllGatherPipelineFor91093Executor::CollAllGatherPipelineFor91093Executor(
22+ const HcclDispatcher dispatcher,
23+ std::unique_ptr<TopoMatcher> &topoMatcher)
24+ : CollAllGatherExecutor(dispatcher, topoMatcher)
25+{
26+ DMAReduceFlag_ = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE;
27+ desc_.level1SupportedAlgos = {
28+ AlgTypeLevel1::ALG_LEVEL1_NHR,
29+ AlgTypeLevel1::ALG_LEVEL1_NB,
30+ AlgTypeLevel1::ALG_LEVEL1_RING
31+ };
32+ desc_.level2SupportedAlgos = {
33+ AlgTypeLevel2::ALG_LEVEL2_NHR,
34+ AlgTypeLevel2::ALG_LEVEL2_NB,
35+ AlgTypeLevel2::ALG_LEVEL2_RING
36+ };
37+}
38+ 
39+HcclResult CollAllGatherPipelineFor91093Executor::CalcStreamNum(u32& streamNum)
40+{
41+ // 计算三级流水线所需的流数量
42+ HCCL_INFO("[CollAllGatherPipelineFor91093Executor][CalcStreamNum] topoType_[%u], workflowMode_[%u]",
43+ topoType_, workflowMode_);
44+ // 基本流数量计算
45+ u32 totalStreamNum = (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING ? LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE :
46+ LEVEL0_PLANE_NUM_IN_NPRING_SINGLE);
47+ if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) { // 工作流模式，双倍的流，用于并行操作
48+ totalStreamNum *= STREAM_NUM_FOR_DMAREDUCE_ONE_RING; // *2
49+ }
50+ 
51+ // 为三级流水线增加额外的流
52+ // 主流用于L2 NHR，从流用于L1 NHR + L0 DoubleRing
53+ totalStreamNum += 2; // 增加一个主流用于L2流水线，2个notify用于两块内存的主从流之间的同步
54+ 
55+ streamNum = totalStreamNum - 1;
56+ HCCL_INFO("[CollAllGatherPipelineFor91093Executor][CalcStreamNum] tag[%s] streamNum[%u]",
57+ tag_.c_str(), streamNum);
58+ return HCCL_SUCCESS;
59+}
60+ 
61+HcclResult CollAllGatherPipelineFor91093Executor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
62+{
63+ TransportMemType inputType = TransportMemType::RESERVED;
64+ TransportMemType outputType = TransportMemType::RESERVED;
65+ CHK_RET(CalcTransportMemType(inputType, outputType));
66+ CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
67+ CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
68+ CHK_RET(CalcLevel2CommInfo(inputType, outputType, opTransport));
69+ return HCCL_SUCCESS;
70+}
71+ 
72+// level0 ring
73+HcclResult CollAllGatherPipelineFor91093Executor::CalcLevel0CommInfo(TransportMemType inputType,
74+ TransportMemType outputType,
75+ std::vector<LevelNSubCommTransport>& opTransport)
76+{
77+ CommParaInfo commParaInfo(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
78+ CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL0], inputType, outputType));
79+ return HCCL_SUCCESS;
80+}
81+ 
82+// level2 NHR
83+HcclResult CollAllGatherPipelineFor91093Executor::CalcLevel2CommInfo(TransportMemType inputType,
84+ TransportMemType outputType,
85+ std::vector<LevelNSubCommTransport>& opTransport)
86+{
87+ CommParaInfo commParaInfo(COMM_LEVEL2, CommType::COMM_TAG_MAX);
88+ if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
89+ commParaInfo.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING;
90+ } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
91+ commParaInfo.commType = CommType::COMM_TAG_NONUNIFORM_BRUCK;
92+ } else {
93+ commParaInfo.commType = CommType::COMM_TAG_RING_INNER;
94+ }
95+ CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL2], inputType, outputType));
96+ return HCCL_SUCCESS;
97+}
98+ 
99+HcclResult CollAllGatherPipelineFor91093Executor::CalcTransportMemType(TransportMemType &inputType,
100+ TransportMemType &outputType)
101+{
102+ inputType = TransportMemType::CCL_INPUT;
103+ outputType = TransportMemType::CCL_OUTPUT;
104+ HCCL_INFO("[CollAllGatherPipelineFor91093Executor][CalcTransportMemType]" \
105+ "tag[%s] inputType[%d], outputType[%d]",
106+ tag_.c_str(), inputType, outputType);
107+ return HCCL_SUCCESS;
108+}
109+ 
110+// 每次循环处理的数据量，双流水的情况下需要满足每个流水线都能满载
111+u64 CollAllGatherPipelineFor91093Executor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
112+{
113+ // 分成两片，做流水ping-pong
114+ u64 maxCountPerLoop = cclBuffSize / PIPELINE_NUM / topoAttr_.userRankSize / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN / unitSize;
千弘
千弘千弘5月12日

maxCountPerLoop除后为0的情况：

likedislike
115+ HCCL_INFO("[%s] tag[%s] maxCountPerLoop[%llu]", __func__, tag_.c_str(), maxCountPerLoop);
116+ 
117+ return maxCountPerLoop;
118+}
119+ 
120+// 编排
121+HcclResult CollAllGatherPipelineFor91093Executor::Orchestrate(
122+ OpParam &param, AlgResourceResponse &algRes)
123+{
124+ HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllGatherPipelineFor91093Executor][Orchestrate] begins.");
125+ 
126+ HcclUs startut = TIME_NOW();
127+ tag_ = param.tag;
128+ algResResp_ = &algRes;
129+ 
130+ // 设置L0和L1通信域信息
131+ CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
132+ level0CommInfo_ = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
133+ u32 commIndex = level0CommInfo_.localRank;
134+ CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
135+ level1CommInfo_ = GetSubCommInfo(COMM_LEVEL1, commIndex);
136+ // 获取L2通信域信息
137+ CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1));
138+ level2CommInfo_ = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
139+ 
140+ // 准备通信资源和流，资源待商榷
141+ mainStreamL2_ = param.stream;
142+ subStreams_ = algResResp_->slaveStreams;
143+ mainStreamL1L0_ = subStreams_.back();
144+ const u32 baseStreamNum = algResResp_->slaveStreams.size() - PIPELINE_NUM;
千弘
千弘千弘5月12日

aclgraph场景？确认一下有没有影响

likedislike
145+ notifyL1L0ToL2A_ = algResResp_->notifiesMain[baseStreamNum];
146+ notifyL1L0ToL2B_ = algResResp_->notifiesMain[baseStreamNum + 1];
147+ notifyL2ToL1L0A_ = algResResp_->notifiesAux[baseStreamNum];
148+ notifyL2ToL1L0B_ = algResResp_->notifiesAux[baseStreamNum + 1];
149+ notifyRingMain_.assign(algResResp_->notifiesMain.begin(), algResResp_->notifiesMain.end() - PIPELINE_NUM);
150+ notifyRingSub_.assign(algResResp_->notifiesAux.begin(), algResResp_->notifiesAux.end() - PIPELINE_NUM);
151+ ringSubStreams_.assign(subStreams_.begin(), subStreams_.end() - PIPELINE_NUM);
152+ 
153+ // 计算通信域信息和内存类型
154+ unitSize_ = SIZE_TABLE[param.DataDes.dataType];
155+ cclInputSizeHalved_ = algResResp_->cclInputMem.size() / 2;
156+ cclInputAMem_ = algResResp_->cclInputMem.range(0, cclInputSizeHalved_);
157+ cclInputBMem_ = algResResp_->cclInputMem.range(cclInputSizeHalved_, cclInputSizeHalved_);
158+ cclOutputSizeHalved_ = algResResp_->cclOutputMem.size() / 2;
159+ cclOutputAMem_ = algResResp_->cclOutputMem.range(0, cclOutputSizeHalved_);
160+ cclOutputBMem_ = algResResp_->cclOutputMem.range(cclOutputSizeHalved_, cclOutputSizeHalved_);
161+ 
162+ CHK_RET(RunLoop(param)); // 运行循环，循环内执行三级流水线
163+ 
164+ HCCL_INFO("tag[%s], Allgather executor orchestrate success, take time [%lld]us.", tag_.c_str(),
165+ DURATION_US(TIME_NOW() - startut));
166+ 
167+ return HCCL_SUCCESS;
168+}
169+ 
170+HcclResult CollAllGatherPipelineFor91093Executor::RunL2Stage(
171+ const OpParam &param, ExecMem &execMem, u64 loopIdx, u64 memIdx, u64 bufferSliceNum)
172+{
173+ // superpod数量不超过1不需要跨超节点；最后一轮循环处理L1L0的最后一片数据，L2不需要参与通信，跳过L2阶段
174+ if (loopIdx >= bufferSliceNum) {
175+ return HCCL_SUCCESS;
176+ }
177+ // 同步：等待 loopIdx-2 的 L1+L0 通信完成，释放 DMA buffer 后 L2 再复用
178+ if (loopIdx >= 2) {
179+ auto notifyL1L0ToL2 = (memIdx == 0) ? notifyL1L0ToL2A_ : notifyL1L0ToL2B_;
180+ CHK_RET(LocalNotify::Wait(mainStreamL2_, dispatcher_, notifyL1L0ToL2, INVALID_VALUE_STAGE));
181+ }
182+ 
183+ // Local Copy: UserIn -> Ccl
184+ u64 curSize = execMem.count * unitSize_;
185+ DeviceMem srcMem = DeviceMem::create(static_cast<u8 *>(execMem.inputPtr), curSize);
186+ DeviceMem dstMem = execMem.inputMem.range(0, curSize);
187+ CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, mainStreamL2_));
188+ 
189+ // OPBASE模式
190+ u64 dstMemOffset = topoAttr_.userRank * curSize;
191+ DeviceMem dmaDst = execMem.outputMem.range(dstMemOffset, curSize);
192+ CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dmaDst, srcMem, mainStreamL2_));
193+ 
194+ // L2 Rx -> DMA[0/1]
195+ u64 baseOffset = memIdx == 0 ? 0 : cclInputSizeHalved_;
196+ CHK_RET(KernelRunInterSuperPod(param, execMem, baseOffset));
197+ auto notifyL2ToL1L0 = (memIdx == 0) ? notifyL2ToL1L0A_ : notifyL2ToL1L0B_;
198+ CHK_RET(LocalNotify::Post(mainStreamL2_, dispatcher_, notifyL2ToL1L0, INVALID_VALUE_STAGE));
199+ return HCCL_SUCCESS;
200+}
201+ 
202+HcclResult CollAllGatherPipelineFor91093Executor::RunL1L0Stage(
203+ const OpParam &param, ExecMem &lastExecMem, u64 loopIdx, u64 memIdx, u64 bufferSliceNum)
204+{
205+ // 第一轮等待L2处理完
206+ if (loopIdx < 1) {
207+ return HCCL_SUCCESS;
208+ }
209+ // 同步：等待上一轮的 L2 通信完成
210+ auto notifyL2ToL1L0 = (memIdx == 0) ? notifyL2ToL1L0A_ : notifyL2ToL1L0B_;
211+ CHK_RET(LocalNotify::Wait(mainStreamL1L0_, dispatcher_, notifyL2ToL1L0, INVALID_VALUE_STAGE));
212+ 
213+ u64 baseOffset = memIdx == 0 ? 0 : cclInputSizeHalved_;
214+ if (level1CommInfo_.localRankSize > 1) {
215+ CHK_RET(KernelRunInterServer(param, lastExecMem, baseOffset));
216+ }
217+ CHK_RET(KernelRunIntraServer(param, lastExecMem, baseOffset));
218+ 
219+ auto notifyL1L0ToL2 = (memIdx == 0) ? notifyL1L0ToL2A_ : notifyL1L0ToL2B_;
220+ CHK_RET(LocalNotify::Post(mainStreamL1L0_, dispatcher_, notifyL1L0ToL2, INVALID_VALUE_STAGE));
221+ return HCCL_SUCCESS;
222+}
223+ 
224+HcclResult CollAllGatherPipelineFor91093Executor::RunLoop(OpParam &param)
225+{
226+ u8* userInputPtr = static_cast<u8 *>(param.inputPtr);
227+ u8* userOutputPtr = static_cast<u8 *>(param.outputPtr);
228+ CHK_PTR_NULL(userInputPtr);
229+ CHK_PTR_NULL(userOutputPtr);
230+ 
231+ u64 maxCountPerLoop = CalcLoopMaxCount(algResResp_->cclInputMem.size(), unitSize_);
232+ CHK_PRT_RET(maxCountPerLoop == 0,
233+ HCCL_ERROR("[CollAllGatherPipelineFor91093Executor][RunLoop]tag[%s] userRankSize[%u] maxCountPerLoop[%llu]",
234+ tag_.c_str(), topoAttr_.userRankSize, maxCountPerLoop), HCCL_E_PARA);
235+ u64 bufferSliceNum = (param.DataDes.count + maxCountPerLoop - 1) / maxCountPerLoop;
236+ if (bufferSliceNum == 0) {
237+ return HCCL_SUCCESS;
238+ }
239+ u64 loopNum = bufferSliceNum + 1;
240+ u64 countLeft = param.DataDes.count; // 剩余的数据量
241+ 
242+ u32 memIdx = 0;
243+ ExecMem lastExecMem;
244+ for (u64 loopIdx = 0; loopIdx < loopNum; loopIdx++) {
245+ u64 curCount = countLeft > maxCountPerLoop ? maxCountPerLoop : countLeft; // 当前循环处理的数据量
246+ countLeft -= curCount;
247+ 
248+ ExecMem execMem;
249+ execMem.count = curCount;
250+ execMem.inputMem = memIdx == 0 ? cclInputAMem_ : cclInputBMem_;
251+ execMem.outputMem = memIdx == 0 ? cclOutputAMem_ : cclOutputBMem_;
252+ execMem.inputPtr = userInputPtr;
253+ execMem.outputPtr = userOutputPtr;
254+ 
255+ CHK_RET(RunL2Stage(param, execMem, loopIdx, memIdx, bufferSliceNum));
256+ CHK_RET(RunL1L0Stage(param, lastExecMem, loopIdx, 1 - memIdx, bufferSliceNum));
257+ 
258+ CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
259+ 
260+ u64 curSize = curCount * unitSize_;
261+ userInputPtr += curSize;
262+ userOutputPtr += curSize;
263+ memIdx = 1 - memIdx; // 双缓冲交替使用
264+ lastExecMem = execMem;
265+ }
266+ 
267+ // 只有 L1L0 阶段实际运行的 buffer 才需要等待
268+ // 当 bufferSliceNum >= 2 时两路都用到，否则只用了一路
269+ CHK_RET(LocalNotify::Wait(mainStreamL2_, dispatcher_, notifyL1L0ToL2A_, INVALID_VALUE_STAGE));
270+ if (bufferSliceNum >= 2) {
271+ CHK_RET(LocalNotify::Wait(mainStreamL2_, dispatcher_, notifyL1L0ToL2B_, INVALID_VALUE_STAGE));
272+ }
273+ 
274+ CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
275+ 
276+ return HCCL_SUCCESS;
277+}
278+ 
279+// 跨超节点
280+HcclResult CollAllGatherPipelineFor91093Executor::KernelRunInterSuperPod(const OpParam &param, ExecMem &execMem, u64 baseOffset)
281+{
282+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] begins, topoType_[%u], DMAReduceFlag_[%u]", __func__, topoType_, DMAReduceFlag_);
283+ std::unique_ptr<AlgTemplateBase> level2AGExecutor;
284+ if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
285+ level2AGExecutor
286+ = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
287+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NB in COMM_LEVEL2", __func__);
288+ } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
289+ level2AGExecutor
290+ = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
291+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NHR in COMM_LEVEL2", __func__);
292+ } else {
293+ level2AGExecutor
294+ = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
295+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_RING in COMM_LEVEL2", __func__);
296+ }
297+ CHK_SMART_PTR_NULL(level2AGExecutor);
298+ 
299+ u64 curDataSegsSliceSize = execMem.count * unitSize_;
300+ std::vector<Slice> level2DataSegsSlice = PrepareSlicesL2(
301+ param, level2CommInfo_, level1CommInfo_, level0CommInfo_, unitSize_, curDataSegsSliceSize);
302+ CHK_RET(level2AGExecutor->Prepare(execMem.outputMem, execMem.outputMem, execMem.inputMem, execMem.count, param.DataDes.dataType,
303+ mainStreamL2_, HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID, level2DataSegsSlice, baseOffset));
304+ 
305+ CHK_RET(RunTemplate(level2AGExecutor, level2CommInfo_));
306+ HCCL_INFO("[%s] AllGather level2 AllGather run success, topoType_[%u]", __func__, topoType_);
307+ return HCCL_SUCCESS;
308+}
309+ 
310+// 超节点内的节点间通信 L1nhr
311+HcclResult CollAllGatherPipelineFor91093Executor::KernelRunInterServer(
312+ const OpParam &param, ExecMem &execMem, u64 baseOffset)
313+{
314+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] begins, topoType_[%u], DMAReduceFlag_[%u]", __func__, topoType_, DMAReduceFlag_);
315+ u64 curDataSegsSliceSize = execMem.count * unitSize_;
316+ std::vector<Slice> level1DataSegsSlice = PrepareSlicesL1(
317+ param, level2CommInfo_, level1CommInfo_, level0CommInfo_, unitSize_, curDataSegsSliceSize);
318+ 
319+ std::unique_ptr<AlgTemplateBase> level1AGExecutor;
320+ if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
321+ level1AGExecutor
322+ = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
323+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_RING in COMM_LEVEL1", __func__);
324+ } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
325+ level1AGExecutor
326+ = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
327+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NB in COMM_LEVEL1", __func__);
328+ } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
329+ level1AGExecutor
330+ = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
331+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NHR in COMM_LEVEL1", __func__);
332+ } else {
333+ HCCL_ERROR("AllGather ring: unsupported algtype [%s].", AlgTypeToStr(algType_).c_str());
334+ return HCCL_E_NOT_SUPPORT;
335+ }
336+ CHK_SMART_PTR_NULL(level1AGExecutor);
337+ CHK_RET(level1AGExecutor->Prepare(execMem.outputMem, execMem.outputMem, execMem.inputMem, execMem.count, param.DataDes.dataType,
338+ mainStreamL1L0_, HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID, level1DataSegsSlice, baseOffset));
339+ 
340+ CHK_RET(RunTemplate(level1AGExecutor, level1CommInfo_));
341+ HCCL_INFO("[%s] AllGather level1 AllGather run success, topoType_[%u]", __func__, topoType_);
342+ return HCCL_SUCCESS;
343+}
344+ 
345+// Server内的通信
346+HcclResult CollAllGatherPipelineFor91093Executor::KernelRunIntraServer(
347+ const OpParam &param, ExecMem &execMem, u64 baseOffset)
348+{
349+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] begins, topoType_[%u], DMAReduceFlag_[%u]", __func__, topoType_, DMAReduceFlag_);
350+ // 节点内做AllGather ring
351+ u64 curDataSegsSliceSize = execMem.count * unitSize_;
352+ std::vector<std::vector<Slice>> multRingsSlice;
353+ CHK_RET(PrepareSlicesL0(multRingsSlice, param, level2CommInfo_, level1CommInfo_, level0CommInfo_,
354+ unitSize_, curDataSegsSliceSize));
355+ 
356+ std::vector<std::vector<Slice>> multRingsUserMemSlice;
357+ CHK_RET(PrepareUserMemSlices(multRingsUserMemSlice, multRingsSlice, param, level2CommInfo_, level1CommInfo_,
358+ level0CommInfo_, unitSize_, curDataSegsSliceSize));
359+ 
360+ // allgather输入放在CCL buffer上，通过设置nullptr指示要从CCL buffer获取输入
361+ l0OpInfo_.inputAddr = nullptr;
362+ l0OpInfo_.outputAddr = execMem.outputPtr;
363+ l0OpInfo_.dataType = param.GetDataType();
364+ l0OpInfo_.count = execMem.count;
365+ l0OpInfo_.root = 0;
366+ l0OpInfo_.reduceOp = HCCL_REDUCE_RESERVED;
367+ l0OpInfo_.strideCount = param.DataDes.strideCount;
368+ 
369+ if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
370+ CHK_RET(DoubleRingAllGather(param.tag, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType,
371+ multRingsSlice, mainStreamL1L0_, PROF_STAGE_2, baseOffset, &l0OpInfo_, multRingsUserMemSlice));
372+ } else if (topoType_ == TopoType::TOPO_TYPE_NP_SINGLE_RING) {
373+ CHK_RET(MultiRingAllGather(param.tag, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType,
374+ multRingsSlice, mainStreamL1L0_, PROF_STAGE_2, baseOffset, &l0OpInfo_, multRingsUserMemSlice, COMM_LEVEL0));
375+ } else {
376+ return HCCL_E_NOT_SUPPORT;
377+ }
378+ HCCL_INFO("[%s] AllGather level0 Ring run success, topoType_[%u]", __func__, topoType_);
379+ return HCCL_SUCCESS;
380+}
381+ 
382+std::vector<Slice> CollAllGatherPipelineFor91093Executor::PrepareSlicesL1(const OpParam &param,
383+ const SubCommInfo &level2CommInfo, const SubCommInfo &level1CommInfo, const SubCommInfo &level0CommInfo,
384+ u32 perDataSize, u64 inputMemSize) const
385+{
386+ const u32 level0RankSize = level0CommInfo.localRankSize;
387+ const u32 level0ServerIndex = level0CommInfo.localRank;
388+ const u32 level1RankSize = level1CommInfo.localRankSize;
389+ const u32 level2RankSize = level2CommInfo.localRankSize;
390+ std::vector<Slice> level1DataSegsSlice;
391+ for (u32 j = 0; j < level1RankSize; j++) {
392+ for (u32 i = 0; i < level2RankSize; i++) {
393+ Slice level1Slice;
394+ level1Slice.size = inputMemSize;
395+ level1Slice.offset = inputMemSize *
396+ (i * level1RankSize * level0RankSize + j * level0RankSize + level0ServerIndex);
397+ 
398+ HCCL_DEBUG("[CollAllGatherPipelineFor91093Executor][PrepareSlicesL1] rank[%u], level1index[%u], level2index[%u], slices.offset=%llu, slices.size=%llu",
399+ level0CommInfo.localRank, j, i, level1Slice.offset, level1Slice.size);
400+ 
401+ level1DataSegsSlice.push_back(level1Slice);
402+ }
403+ }
404+ return level1DataSegsSlice;
405+}
406+ 
407+std::vector<Slice> CollAllGatherPipelineFor91093Executor::PrepareSlicesL2(const OpParam &param,
408+ const SubCommInfo &level2CommInfo, const SubCommInfo &level1CommInfo, const SubCommInfo &level0CommInfo,
409+ u32 perDataSize, u64 inputMemSize) const
410+{
411+ const u32 level0RankSize = level0CommInfo.localRankSize;
412+ const u32 level0ServerIndex = level0CommInfo.localRank;
413+ const u32 level1RankSize = level1CommInfo.localRankSize;
414+ const u32 level1ServerIndex = level1CommInfo.localRank;
415+ const u32 level2RankSize = level2CommInfo.localRankSize;
416+ std::vector<Slice> level2DataSegsSlice;
417+ for (u32 i = 0; i < level2RankSize; i++) {
418+ Slice sliceTemp;
419+ sliceTemp.size = inputMemSize;
420+ sliceTemp.offset = inputMemSize *
421+ (i * level1RankSize * level0RankSize + level1ServerIndex * level0RankSize + level0ServerIndex);
422+ level2DataSegsSlice.push_back(sliceTemp);
423+ }
424+ return level2DataSegsSlice;
425+}
426+ 
427+HcclResult CollAllGatherPipelineFor91093Executor::PrepareSlicesL0(std::vector<std::vector<Slice>> &multRingsSlice,
428+ const OpParam &param, const SubCommInfo &level2CommInfo, const SubCommInfo &level1CommInfo,
429+ const SubCommInfo &level0CommInfo, u32 perDataSize, u64 inputMemSize)
430+{
431+ const u32 level0RankSize = level0CommInfo.localRankSize;
432+ const u32 level1RankSize = level1CommInfo.localRankSize;
433+ const u32 level2RankSize = level2CommInfo.localRankSize;
434+ 
435+ std::vector<Slice> dataSegsSlice;
436+ CHK_RET(PrepareAllgatherSlice(level0RankSize, inputMemSize, dataSegsSlice));
437+ 
438+ // 多环数据切分
439+ std::vector<std::vector<Slice>> multRingsSliceZero; // 数据基于该rank上环0的偏移
440+
441+ if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING &&
442+ !IsSupportUnifiedMarch(param, topoType_, topoAttr_.serverNum, topoAttr_.superPodNum)) {
443+ multRingsSliceZero = PrepareMultiRingSlice(dataSegsSlice, param.tag, false, topoAttr_.nicList);
444+ } else {
445+ multRingsSliceZero.push_back(dataSegsSlice);
446+ }
447+ for (u32 ringIndex = 0; ringIndex < multRingsSliceZero.size(); ringIndex++) {
448+ std::vector<Slice> level2DataSlice;
449+ CHK_RET(CalculateLevel2AllgatherSlice(inputMemSize, level0RankSize, level1RankSize, level2RankSize,
450+ multRingsSliceZero, level2DataSlice, ringIndex));
451+ multRingsSlice.push_back(level2DataSlice);
452+ }
453+ 
454+ return HCCL_SUCCESS;
455+}
456+ 
457+HcclResult CollAllGatherPipelineFor91093Executor::PrepareUserMemSlices(std::vector<std::vector<Slice>> &userMemSlices,
458+ const std::vector<std::vector<Slice>> &multRingsSlice, const OpParam &param, const SubCommInfo &level2CommInfo,
459+ const SubCommInfo &level1CommInfo, const SubCommInfo &level0CommInfo, u32 perDataSize, u64 inputMemSize)
460+{
461+ CHK_PRT_RET(0 < param.DataDes.strideCount && param.DataDes.strideCount < param.DataDes.count,
462+ HCCL_ERROR("[CollAllGatherPipelineFor91093Executor][KernelRun]strideCount[%llu] is smaller than opCount[%llu]",
463+ param.DataDes.strideCount, param.DataDes.count),
464+ HCCL_E_PARA);
465+ HCCL_DEBUG("[CollAllGatherPipelineFor91093Executor][KernelRun]strideCount[%llu], opCount[%llu]",
466+ param.DataDes.strideCount, param.DataDes.count);
467+ 
468+ for (u32 ringIndex = 0; ringIndex < multRingsSlice.size(); ringIndex++) {
469+ std::vector<Slice> userMemSlice;
470+ for (const auto &cclSlice : multRingsSlice[ringIndex]) {
471+ Slice tmpSlice;
472+ u64 count = (param.DataDes.strideCount == 0) ? param.DataDes.count : param.DataDes.strideCount;
473+ tmpSlice.size = cclSlice.size;
474+ tmpSlice.offset
475+ = (cclSlice.offset / inputMemSize) * count * perDataSize + multRingsSlice[ringIndex][0].offset;
476+ userMemSlice.push_back(tmpSlice);
477+ HCCL_DEBUG("rank[%u], ringIndex[%u], tmpSlice.offset=[%llu], size=[%llu]", topoAttr_.userRank, ringIndex,
478+ tmpSlice.offset, tmpSlice.size);
479+ }
480+ userMemSlices.push_back(userMemSlice);
481+ }
482+ return HCCL_SUCCESS;
483+}
484+ 
485+HcclResult CollAllGatherPipelineFor91093Executor::DoubleRingAllGather(
486+ const std::string &tag, DeviceMem inputMem, DeviceMem outputMem,
487+ const u64 count, const HcclDataType dataType, const std::vector<std::vector<Slice> > multRingsSliceZero,
488+ Stream stream, s32 profStage, const u64 baseOffset, const HcomCollOpInfo *opInfo,
489+ const std::vector<std::vector<Slice>> multRingsUserMemSlice)
490+{
491+ HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllGatherPipelineFor91093Executor]userRank[%u], count[%llu]",
492+ topoAttr_.userRank, count);
493+ 
494+ (void)tag;
495+ HCCL_INFO("[CollAllGatherPipelineFor91093Executor][DoubleRingAllGather] DoubleRingAllGather starts");
496+ HcclResult ret = HCCL_SUCCESS;
497+ u32 ringNum = multRingsSliceZero.size();
498+ CHK_RET(CheckCommSize(COMM_LEVEL0, ringNum));
499+ // 拿到ring环映射关系
500+ SubCommInfo level0ZeroCommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
501+ auto nicList = topoAttr_.nicList;
502+ std::vector<std::vector<u32>> multiRingsOrder =
503+ GetRingsOrderByTopoType(level0ZeroCommInfo.localRankSize, topoType_, nicList);
504+ // 生成两个ring上的userMemOut_上对应的slices
505+ std::vector<std::vector<Slice>> userMemOutputSlicesOfDoubleRing;
506+ CHK_RET(CollectMultiRingsUserMemSlices(ringNum, dataType, opInfo, multRingsSliceZero,
507+ multiRingsOrder, multRingsUserMemSlice, userMemOutputSlicesOfDoubleRing));
508+ // 生成两个ring上的rankOrder
509+ std::vector<std::vector<u32>> rankOrders;
510+ CHK_RET(CollectMultiRingsRankOrder(ringNum, multiRingsOrder, rankOrders));
511+ // 初始化executor
512+ std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
513+ TemplateType::TEMPLATE_ALIGNED_ALL_GATHER_DOUBLE_RING, dispatcher_);
514+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALIGNED_ALL_GATHER_DOUBLE_RING in COMM_LEVEL0", __func__);
515+ CHK_SMART_PTR_NULL(tempAlg);
516+ CHK_RET(tempAlg->Prepare(const_cast<HcomCollOpInfo *>(opInfo), topoAttr_.userRank, ringSubStreams_,
517+ notifyRingMain_, notifyRingSub_, rankOrders, userMemOutputSlicesOfDoubleRing));
518+ 
519+ ret = tempAlg->Prepare(outputMem, outputMem, inputMem, count, dataType, stream, multRingsSliceZero,
520+ HCCL_REDUCE_RESERVED, LEVEL0_BRIDGE_RANK_ID, baseOffset);
521+ CHK_PRT_RET(ret != HCCL_SUCCESS,
522+ HCCL_ERROR("[CollAllGatherPipelineFor91093Executor][DoubleRingAllGather]Double ring "
523+ "AllGather failed, return[%d]", ret), ret);
524+ u32 ringIndexOp = COMM_INDEX_0;
525+ u32 rankSize = level0ZeroCommInfo.localRankSize;
526+ ret = tempAlg->RegisterProfiler(
527+ ((ringIndexOp + 1) << PROF_RINGINDEX_OFFSET_OF_PLANEID) +
528+ (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0ZeroCommInfo.localRank,
529+ profStage, HCCL_EXEC_STEP_NOT_SET, stream);
530+ CHK_PRT_RET(ret != HCCL_SUCCESS,
531+ HCCL_ERROR("[CollAllGatherPipelineFor91093Executor][DoubleRingAllGather]Double ring "
532+ "AllGather failed, return[%d]", ret), ret);
533+ 
534+ // 空拷贝用于后续操作附着
535+ CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem, outputMem, stream, dispatcher_));
536+ ret = RunTemplate(tempAlg, level0ZeroCommInfo);
537+ CHK_PRT_RET(ret != HCCL_SUCCESS,
538+ HCCL_ERROR("[CollAllGatherPipelineFor91093Executor][DoubleRingAllGather] Double ring "
539+ "AllGather failed, return[%d]", ret), ret);
540+ // 添加空task,保证执行时不乱序
541+ CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem, outputMem, stream, dispatcher_));
542+ return HCCL_SUCCESS;
543+}
544+ 
545+HcclResult CollAllGatherPipelineFor91093Executor::GetSubStreamInfoOnOneRing(const u32 ringIndex,
546+ std::vector<Stream> &subStreamsInOneRing,
547+ std::vector<std::shared_ptr<LocalNotify>> &mainSignalsInOneRing,
548+ std::vector<std::shared_ptr<LocalNotify>> &subSignalsInOneRing)
549+{
550+ u32 ringNum = algResResp_->slaveStreams.size() - 1;
551+ if (ringNum == LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE * STREAM_NUM_FOR_DMAREDUCE_ONE_RING) {
552+ subStreamsInOneRing.push_back(algResResp_->slaveStreams[ringIndex + 1]);
553+ mainSignalsInOneRing.push_back(algResResp_->notifiesMain[ringIndex + 1]);
554+ subSignalsInOneRing.push_back(algResResp_->notifiesAux[ringIndex + 1]);
555+ } else if (ringNum == LEVEL0_PLANE_NUM_IN_NPRING_SINGLE * STREAM_NUM_FOR_DMAREDUCE_ONE_RING) {
556+ subStreamsInOneRing.push_back(algResResp_->slaveStreams[ringIndex]);
557+ mainSignalsInOneRing.push_back(algResResp_->notifiesMain[ringIndex]);
558+ subSignalsInOneRing.push_back(algResResp_->notifiesAux[ringIndex]);
559+ }
560+ return HCCL_SUCCESS;
561+}
562+ 
563+REGISTER_EXEC("AllGatherPipelineFor91093Executor",
564+ AllGatherPipelineFor91093,
565+ CollAllGatherPipelineFor91093Executor);
566+ 
567+} // namespace hccl
Asrc/algorithm/impl/coll_executor/coll_all_gather/coll_all_gather_pipeline_for_910_93_executor.h+98-0
@@ -0,0 +1,98 @@
1+/**
2+ * Copyright (c) 2026 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+#ifndef COLL_ALLGATHER_PIPELINE_FOR_910_93_EXECUTOR_H
12+#define COLL_ALLGATHER_PIPELINE_FOR_910_93_EXECUTOR_H
13+ 
14+#include "coll_all_gather_executor.h"
15+ 
16+namespace hccl {
17+class CollAllGatherPipelineFor91093Executor
18+ : public CollAllGatherExecutor {
19+public:
20+ explicit CollAllGatherPipelineFor91093Executor(
21+ const HcclDispatcher dispatcher,
22+ std::unique_ptr<TopoMatcher> &topoMatcher);
23+ ~CollAllGatherPipelineFor91093Executor() override = default;
24+ 
25+ // 流水线编排
26+ HcclResult Orchestrate(OpParam& param, AlgResourceResponse& algRes) override;
27+ 
28+private:
29+ // 资源计算
30+ HcclResult CalcStreamNum(u32& streamNum) override;
31+ HcclResult CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport) override;
32+ HcclResult CalcLevel0CommInfo(TransportMemType inputType, TransportMemType outputType,
33+ std::vector<LevelNSubCommTransport>& opTransport) override;
34+ HcclResult CalcLevel2CommInfo(TransportMemType inputType, TransportMemType outputType,
35+ std::vector<LevelNSubCommTransport>& opTransport) override;
36+ HcclResult CalcTransportMemType(TransportMemType &inputType, TransportMemType &outputType);
37+ 
38+ // 算法编排
39+ u64 CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize) override;
40+ HcclResult RunLoop(OpParam &param);
41+ HcclResult RunL2Stage(const OpParam &param, ExecMem &execMem, u64 loopIdx, u64 memIdx, u64 bufferSliceNum);
42+ HcclResult RunL1L0Stage(const OpParam &param, ExecMem &lastExecMem, u64 loopIdx, u64 memIdx, u64 bufferSliceNum);
43+ 
44+ // 层级算法调用方法（从模板移动到执行器）
45+ HcclResult KernelRunInterSuperPod(const OpParam &param, ExecMem &execMem, u64 baseOffset); // 跨超
46+ HcclResult KernelRunIntraServer(const OpParam &param, ExecMem &execMem, u64 baseOffset); // server内
47+ HcclResult KernelRunInterServer(const OpParam &param, ExecMem &execMem, u64 baseOffset); // 跨server
48+ 
49+ virtual std::vector<Slice> PrepareSlicesL2(const OpParam &param, const SubCommInfo &level2CommInfo,
50+ const SubCommInfo &level1CommInfo, const SubCommInfo &level0CommInfo, u32 perDataSize, u64 inputMemSize) const;
51+ virtual std::vector<Slice> PrepareSlicesL1(const OpParam &param, const SubCommInfo &level2CommInfo,
52+ const SubCommInfo &level1CommInfo, const SubCommInfo &level0CommInfo, u32 perDataSize, u64 inputMemSize) const;
53+ virtual HcclResult PrepareSlicesL0(std::vector<std::vector<Slice>> &multRingsSlice, const OpParam &param,
54+ const SubCommInfo &level2CommInfo, const SubCommInfo &level1CommInfo, const SubCommInfo &level0CommInfo,
55+ u32 perDataSize, u64 inputMemSize);
56+ virtual HcclResult PrepareUserMemSlices(std::vector<std::vector<Slice>> &userMemSlices,
57+ const std::vector<std::vector<Slice>> &multRingsSlice, const OpParam &param, const SubCommInfo &level2CommInfo,
58+ const SubCommInfo &level1CommInfo, const SubCommInfo &level0CommInfo, u32 perDataSize, u64 inputMemSize);
59+ 
60+ HcclResult DoubleRingAllGather(const std::string &tag, DeviceMem inputMem, DeviceMem outputMem, const u64 count,
61+ const HcclDataType dataType, const std::vector<std::vector<Slice>> multRingsSliceZero, Stream stream,
62+ s32 profStage, const u64 baseOffset, const HcomCollOpInfo *opInfo,
63+ const std::vector<std::vector<Slice>> multRingsUserMemSlice);
64+ 
65+ HcclResult GetSubStreamInfoOnOneRing(const u32 ringIndex,
66+ std::vector<Stream> &subStreamsInOneRing,
67+ std::vector<std::shared_ptr<LocalNotify>> &mainSignalsInOneRing,
68+ std::vector<std::shared_ptr<LocalNotify>> &subSignalsInOneRing) override;
69+ // 成员变量
70+ u32 unitSize_ = 0;
71+ 
72+ Stream mainStreamL2_; // SDMA+localcopy
73+ std::vector<Stream> subStreams_;
74+ Stream mainStreamL1L0_;
75+ std::vector<Stream> ringSubStreams_;
76+ std::shared_ptr<LocalNotify> notifyL2ToL1L0A_{nullptr};
77+ std::shared_ptr<LocalNotify> notifyL1L0ToL2A_{nullptr};
78+ std::shared_ptr<LocalNotify> notifyL2ToL1L0B_{nullptr};
79+ std::shared_ptr<LocalNotify> notifyL1L0ToL2B_{nullptr};
80+ std::vector<std::shared_ptr<LocalNotify>> notifyRingMain_;
81+ std::vector<std::shared_ptr<LocalNotify>> notifyRingSub_;
82+ SubCommInfo level0CommInfo_{}; // L0层通信域信息
83+ SubCommInfo level1CommInfo_{}; // L1层通信域信息
84+ SubCommInfo level2CommInfo_{}; // L2层通信域信息
85+ 
86+ DeviceMem cclInputAMem_;
87+ DeviceMem cclOutputAMem_;
88+ DeviceMem cclInputBMem_;
89+ DeviceMem cclOutputBMem_;
90+ u64 cclInputSizeHalved_{0};
91+ u64 cclOutputSizeHalved_{0};
92+ 
93+ HcomCollOpInfo l0OpInfo_{};
94+};
95+ 
96+} // namespace hccl
97+ 
98+#endif
Msrc/algorithm/impl/coll_executor/coll_comm_executor.h+1-1
@@ -136,7 +136,7 @@ public:
136 HcclResult GetAdjInfo(AlgResourceResponse& algRes, AdjInfo& adjInfo) override;136 HcclResult GetAdjInfo(AlgResourceResponse& algRes, AdjInfo& adjInfo) override;
137 137 
138protected:138protected:
139- HcclResult GetSubStreamInfoOnOneRing(const u32 ringIndex,139+ virtual HcclResult GetSubStreamInfoOnOneRing(const u32 ringIndex,
140 std::vector<Stream> &subStreamsInOneRing,140 std::vector<Stream> &subStreamsInOneRing,
141 std::vector<std::shared_ptr<LocalNotify>> &mainSignalsInOneRing,141 std::vector<std::shared_ptr<LocalNotify>> &mainSignalsInOneRing,
142 std::vector<std::shared_ptr<LocalNotify>> &subSignalsInOneRing);142 std::vector<std::shared_ptr<LocalNotify>> &subSignalsInOneRing);
Msrc/algorithm/impl/operator/all_gather_operator.cc+14-0
@@ -340,6 +340,15 @@ HcclResult AllGatherOperator::SelectAlgfor91093(const OpParam& param, std::strin
340 CHK_RET(SelectAlgforAHC(dataSize, AHCOpType::AHC_OP_TYPE_ALLGATHER));340 CHK_RET(SelectAlgforAHC(dataSize, AHCOpType::AHC_OP_TYPE_ALLGATHER));
341 }341 }
342 342 
343+ constexpr u64 ALLGATHER_PIPELINE_THRESHOLD = 4 * 1024 * 1024; // 4MB
344+ constexpr u64 ALLGATHER_PIPELINE_SLICE_THRESHOLD = 2; // 2
345+ u64 maxCountPerLoop = cclBufferManager_.GetInCCLbufferSize() / HCCL_DEVICE_NUM_TWO / userRankSize_
346+ / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN / unitSize;
347+ u64 bufferSliceNum = 0;
348+ if (maxCountPerLoop != 0) {
349+ bufferSliceNum = (param.DataDes.count + maxCountPerLoop - 1) / maxCountPerLoop;
350+ }
351+ 
343 bool isHccsPlusSio = userRankSize_ == 2 && pairLinkCounter_[static_cast<u32>(LinkTypeInServer::SIO_TYPE)] == 2 &&352 bool isHccsPlusSio = userRankSize_ == 2 && pairLinkCounter_[static_cast<u32>(LinkTypeInServer::SIO_TYPE)] == 2 &&
344 pairLinkCounter_[static_cast<u32>(LinkTypeInServer::HCCS_TYPE)] == 0;353 pairLinkCounter_[static_cast<u32>(LinkTypeInServer::HCCS_TYPE)] == 0;
345 isHccsPlusSio = false;354 isHccsPlusSio = false;
@@ -377,6 +386,11 @@ HcclResult AllGatherOperator::SelectAlgfor91093(const OpParam& param, std::strin
377 } else {386 } else {
378 algName = "AllGatherRingZerocopyExchangeExecutor"; // 连续数据通信+额外的数据交换（AHC不支持）387 algName = "AllGatherRingZerocopyExchangeExecutor"; // 连续数据通信+额外的数据交换（AHC不支持）
379 }388 }
389+ } else if (superPodNum_ > 1 && maxCountPerLoop * unitSize > ALLGATHER_PIPELINE_THRESHOLD && bufferSliceNum > ALLGATHER_PIPELINE_SLICE_THRESHOLD
390+ && isOpbase && !isAHCAlgo && !multiModuleDiffDeviceNumMode_
391+ && (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING
392+ || topoType_ == TopoType::TOPO_TYPE_NP_SINGLE_RING)) {
393+ algName = "AllGatherPipelineFor91093Executor";
380 } else {394 } else {
381 if (!(algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING || algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB ||395 if (!(algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING || algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB ||
382 algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_WHOLE_RING || algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||396 algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_WHOLE_RING || algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
Msrc/framework/CMakeLists.txt+1-0
@@ -517,6 +517,7 @@ if(NOT BUILD_OPEN_PROJECT OR (BUILD_OPEN_PROJECT AND KERNEL_MODE))
517 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_all_gather/coll_all_gather_mid_count_for_910_93_executor.cc517 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_all_gather/coll_all_gather_mid_count_for_910_93_executor.cc
518 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_all_gather/coll_all_gather_mesh_graph_executor.cc518 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_all_gather/coll_all_gather_mesh_graph_executor.cc
519 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_all_gather/coll_all_gather_mesh_graph_pipeline_executor.cc519 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_all_gather/coll_all_gather_mesh_graph_pipeline_executor.cc
520+ ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_all_gather/coll_all_gather_pipeline_for_910_93_executor.cc
520 521 
521 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_all_gather_v/coll_all_gather_v_executor.cc522 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_all_gather_v/coll_all_gather_v_executor.cc
522 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_all_gather_v/coll_all_gather_v_ring_for_910_93_executor.cc523 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_all_gather_v/coll_all_gather_v_ring_for_910_93_executor.cc
Mtest/st/algorithm/testcase/testcase/main.cc+1-1
@@ -11,7 +11,7 @@
11 11 
12GTEST_API_ int main(int argc, char **argv) {12GTEST_API_ int main(int argc, char **argv) {
13 // testcase测试代码，只跑特定的用例13 // testcase测试代码，只跑特定的用例
14- //testing::GTEST_FLAG(filter) = "AllReduceTest.allreduce_cyw_test";14+ // testing::GTEST_FLAG(filter) = "AllReduceTest.allreduce_cyw_test";
15 testing::InitGoogleTest(&argc, argv);15 testing::InitGoogleTest(&argc, argv);
16 return RUN_ALL_TESTS();16 return RUN_ALL_TESTS();
17}17}
Mtest/st/algorithm/testcase/testcase/testcase_all_gather.cc+188-1
@@ -1227,11 +1227,198 @@ TEST_F(AllGatherTest, allgather_91093_AllGathermidcountExecutor)
1227 checkerOpParam.opType = CheckerOpType::ALLGATHER;1227 checkerOpParam.opType = CheckerOpType::ALLGATHER;
1228 checkerOpParam.tag = "AllGather";1228 checkerOpParam.tag = "AllGather";
1229 checkerOpParam.opMode = CheckerOpMode::OPBASE;1229 checkerOpParam.opMode = CheckerOpMode::OPBASE;
1230- checkerOpParam.DataDes.count = 1024 * 1024; // 64K 1230+ checkerOpParam.DataDes.count = 1024 * 1024; // 64K
1231 checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_INT8;1231 checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_INT8;
1232 checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;1232 checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
1233 checkerOpParam.algName = "AllGatherMidCountFor91093Executor";1233 checkerOpParam.algName = "AllGatherMidCountFor91093Executor";
1234 1234 
1235+ Checker checker;
1236+ HcclResult ret;
1237+ ret = checker.Check(checkerOpParam, topoMeta);
1238+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
1239+}
1240+ 
1241+// ==================== Pipeline AllGather (A3, 910_93, CollAllGatherPipelineFor91093Executor) ====================
1242+ 
1243+TEST_F(AllGatherTest, ag_pipeline_DR_2pod_2server_int32_4blk_equal)
1244+{
1245+ // DoubleRing + L1/L2 默认 NHR；4 块等大（偶数），完整预热->稳态->排空
1246+ setenv("HCCL_BUFFSIZE", "1", 1);
1247+ RankTable_For_LLT gen;
1248+ TopoMeta topoMeta;
1249+ gen.GenTopoMeta(topoMeta, 2, 2, 8);
1250+ 
1251+ CheckerOpParam checkerOpParam;
1252+ checkerOpParam.opType = CheckerOpType::ALLGATHER;
1253+ checkerOpParam.tag = "AllGather";
1254+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
1255+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
1256+ checkerOpParam.DataDes.count = 16384;
1257+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_INT32;
1258+ checkerOpParam.algName = "AllGatherPipelineFor91093Executor";
1259+ 
1260+ Checker checker;
1261+ HcclResult ret;
1262+ ret = checker.Check(checkerOpParam, topoMeta);
1263+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
1264+}
1265+ 
1266+TEST_F(AllGatherTest, ag_pipeline_DR_3pod_1server_bfp16_3blk_smallA)
1267+{
1268+ // 3 SuperPod；3 块（奇数）尾块小落 BufferA（loopIdx=2, memIdx_==0）
1269+ setenv("HCCL_BUFFSIZE", "1", 1);
1270+ RankTable_For_LLT gen;
1271+ TopoMeta topoMeta;
1272+ gen.GenTopoMeta(topoMeta, 3, 1, 8);
1273+ 
1274+ CheckerOpParam checkerOpParam;
1275+ checkerOpParam.opType = CheckerOpType::ALLGATHER;
1276+ checkerOpParam.tag = "AllGather";
1277+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
1278+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
1279+ checkerOpParam.DataDes.count = 49104;
1280+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_BFP16;
1281+ checkerOpParam.algName = "AllGatherPipelineFor91093Executor";
1282+ 
1283+ Checker checker;
1284+ HcclResult ret;
1285+ ret = checker.Check(checkerOpParam, topoMeta);
1286+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
1287+}
1288+ 
1289+TEST_F(AllGatherTest, ag_pipeline_DR_fp32_4blk_smallB)
1290+{
1291+ // FP32；4 块尾块小落 BufferB（loopIdx=3, memIdx_==1），互补用例 2
1292+ setenv("HCCL_BUFFSIZE", "1", 1);
1293+ RankTable_For_LLT gen;
1294+ TopoMeta topoMeta;
1295+ gen.GenTopoMeta(topoMeta, 2, 1, 8);
1296+ 
1297+ CheckerOpParam checkerOpParam;
1298+ checkerOpParam.opType = CheckerOpType::ALLGATHER;
1299+ checkerOpParam.tag = "AllGather";
1300+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
1301+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
1302+ checkerOpParam.DataDes.count = 26624;
1303+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_FP32;
1304+ checkerOpParam.algName = "AllGatherPipelineFor91093Executor";
1305+ 
1306+ Checker checker;
1307+ HcclResult ret;
1308+ ret = checker.Check(checkerOpParam, topoMeta);
1309+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
1310+}
1311+ 
1312+TEST_F(AllGatherTest, ag_pipeline_DR_force_ring_int8)
1313+{
1314+ // 强制 RING 模板；覆盖默认不覆盖的 Level1/Level2 RING 路径
1315+ setenv("HCCL_BUFFSIZE", "1", 1);
1316+ setenv("HCCL_ALGO", "level1:ring;level2:ring", 1);
1317+ RankTable_For_LLT gen;
1318+ TopoMeta topoMeta;
1319+ gen.GenTopoMeta(topoMeta, 2, 2, 8);
1320+ 
1321+ CheckerOpParam checkerOpParam;
1322+ checkerOpParam.opType = CheckerOpType::ALLGATHER;
1323+ checkerOpParam.tag = "AllGather";
1324+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
1325+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
1326+ checkerOpParam.DataDes.count = 49152;
1327+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_INT8;
1328+ checkerOpParam.algName = "AllGatherPipelineFor91093Executor";
1329+ 
1330+ Checker checker;
1331+ HcclResult ret;
1332+ ret = checker.Check(checkerOpParam, topoMeta);
1333+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
1334+}
1335+ 
1336+TEST_F(AllGatherTest, ag_pipeline_DR_force_nb_int16)
1337+{
1338+ // 强制 NB 模板路径覆盖
1339+ setenv("HCCL_BUFFSIZE", "1", 1);
1340+ setenv("HCCL_ALGO", "level1:NB;level2:NB", 1);
1341+ RankTable_For_LLT gen;
1342+ TopoMeta topoMeta;
1343+ gen.GenTopoMeta(topoMeta, 2, 2, 8);
1344+ 
1345+ CheckerOpParam checkerOpParam;
1346+ checkerOpParam.opType = CheckerOpType::ALLGATHER;
1347+ checkerOpParam.tag = "AllGather";
1348+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
1349+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
1350+ checkerOpParam.DataDes.count = 24576;
1351+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_INT16;
1352+ checkerOpParam.algName = "AllGatherPipelineFor91093Executor";
1353+ 
1354+ Checker checker;
1355+ HcclResult ret;
1356+ ret = checker.Check(checkerOpParam, topoMeta);
1357+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
1358+}
1359+ 
1360+TEST_F(AllGatherTest, ag_pipeline_SR_3pod_7rank_bfp16)
1361+{
1362+ // SingleRing：奇数 rank(7) 触发 SINGLE_RING
1363+ setenv("HCCL_BUFFSIZE", "1", 1);
1364+ RankTable_For_LLT gen;
1365+ TopoMeta topoMeta;
1366+ gen.GenTopoMeta(topoMeta, 3, 1, 7);
1367+ 
1368+ CheckerOpParam checkerOpParam;
1369+ checkerOpParam.opType = CheckerOpType::ALLGATHER;
1370+ checkerOpParam.tag = "AllGather";
1371+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
1372+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
1373+ checkerOpParam.DataDes.count = 37440;
1374+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_BFP16;
1375+ checkerOpParam.algName = "AllGatherPipelineFor91093Executor";
1376+ 
1377+ Checker checker;
1378+ HcclResult ret;
1379+ ret = checker.Check(checkerOpParam, topoMeta);
1380+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
1381+}
1382+ 
1383+TEST_F(AllGatherTest, ag_pipeline_SR_2pod_3server_1rank_int32)
1384+{
1385+ // rank/server=1 边界：L0 退化（机内仅 1 rank 无 gather），Pipeline 流水主要覆盖 L1/L2
1386+ setenv("HCCL_BUFFSIZE", "1", 1);
1387+ RankTable_For_LLT gen;
1388+ TopoMeta topoMeta;
1389+ gen.GenTopoMeta(topoMeta, 2, 3, 1);
1390+ 
1391+ CheckerOpParam checkerOpParam;
1392+ checkerOpParam.opType = CheckerOpType::ALLGATHER;
1393+ checkerOpParam.tag = "AllGather";
1394+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
1395+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
1396+ checkerOpParam.DataDes.count = 65520;
1397+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_INT32;
1398+ checkerOpParam.algName = "AllGatherPipelineFor91093Executor";
1399+ 
1400+ Checker checker;
1401+ HcclResult ret;
1402+ ret = checker.Check(checkerOpParam, topoMeta);
1403+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
1404+}
1405+ 
1406+TEST_F(AllGatherTest, ag_pipeline_DR_one_block_degenerate_fp16)
1407+{
1408+ // 小数据量单 block，Pipeline 退化为串行时序
1409+ RankTable_For_LLT gen;
1410+ TopoMeta topoMeta;
1411+ gen.GenTopoMeta(topoMeta, 2, 1, 8);
1412+ 
1413+ CheckerOpParam checkerOpParam;
1414+ checkerOpParam.opType = CheckerOpType::ALLGATHER;
1415+ checkerOpParam.tag = "AllGather";
1416+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
1417+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
1418+ checkerOpParam.DataDes.count = 8192;
1419+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_FP16;
1420+ checkerOpParam.algName = "AllGatherPipelineFor91093Executor";
1421+ 
1235 Checker checker;1422 Checker checker;
1236 HcclResult ret;1423 HcclResult ret;
1237 ret = checker.Check(checkerOpParam, topoMeta);1424 ret = checker.Check(checkerOpParam, topoMeta);