Star464
1030
代码
代码
Issues115
Pull Requests354
Actions
项目讨论
Wiki
分析
Star464
1030
已合并
[feat] add ReduceScatterPipelineFor91093 #2071
p_ch创建于 5月8日
[feat] add ReduceScatterPipelineFor91093 #2071
已合并
p_ch创建于 5月8日
从p_ch/hcommreduce_scatter_pipeline_parallel合入到cann/hcommmaster
共 10 个文件变更+940-27
Msrc/algorithm/impl/coll_alg_utils.cc+2-1
@@ -376,7 +376,8 @@ bool ExecutorCanSupportDMAReduce(const std::string& algName)
376 "AlignedAllReduceDoubleRingFor91093Executor",376 "AlignedAllReduceDoubleRingFor91093Executor",
377 "ReduceScatterRingFor91093Executor",377 "ReduceScatterRingFor91093Executor",
378 "ReduceScatterFastDoubleRingFor91093Executor",378 "ReduceScatterFastDoubleRingFor91093Executor",
379- "AlignedReduceScatterDoubleRingFor91093Executor"379+ "AlignedReduceScatterDoubleRingFor91093Executor",
380+ "ReduceScatterPipelineFor91093Executor"
380 };381 };
381 if (executorCanSupportDMAReduceSet.find(algName) != executorCanSupportDMAReduceSet.end()) {382 if (executorCanSupportDMAReduceSet.find(algName) != executorCanSupportDMAReduceSet.end()) {
382 return true;383 return true;
Msrc/algorithm/impl/coll_executor/coll_comm_executor.cc+6-1
@@ -26,7 +26,7 @@ HcclResult CollCommExecutor::GetSubStreamInfoOnOneRing(const u32 ringIndex,
26 std::vector<std::shared_ptr<LocalNotify>> &mainSignalsInOneRing,26 std::vector<std::shared_ptr<LocalNotify>> &mainSignalsInOneRing,
27 std::vector<std::shared_ptr<LocalNotify>> &subSignalsInOneRing)27 std::vector<std::shared_ptr<LocalNotify>> &subSignalsInOneRing)
28{28{
29- u32 ringNum = algResResp_->slaveStreams.size() + 1;29+ u32 ringNum = GetLevel0RingNum();
30 if (ringNum == LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE * STREAM_NUM_FOR_DMAREDUCE_ONE_RING) {30 if (ringNum == LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE * STREAM_NUM_FOR_DMAREDUCE_ONE_RING) {
31 // double ring31 // double ring
32 subStreamsInOneRing.push_back(algResResp_->slaveStreams[ringIndex + 1]);32 subStreamsInOneRing.push_back(algResResp_->slaveStreams[ringIndex + 1]);
@@ -41,6 +41,11 @@ HcclResult CollCommExecutor::GetSubStreamInfoOnOneRing(const u32 ringIndex,
41 return HCCL_SUCCESS;41 return HCCL_SUCCESS;
42}42}
43 43 
44+u32 CollCommExecutor::GetLevel0RingNum() const
45+{
46+ return algResResp_->slaveStreams.size() + 1;
47+}
48+ 
44HcclResult CollCommExecutor::MultiRingAllReduce(const std::string &tag, DeviceMem &inputMem, DeviceMem &outputMem,49HcclResult CollCommExecutor::MultiRingAllReduce(const std::string &tag, DeviceMem &inputMem, DeviceMem &outputMem,
45 const u64 count, const HcclDataType dataType, const HcclReduceOp reductionOp,50 const u64 count, const HcclDataType dataType, const HcclReduceOp reductionOp,
46 const std::vector<std::vector<Slice>> &multRingsSliceZero, Stream stream, s32 profStage,51 const std::vector<std::vector<Slice>> &multRingsSliceZero, Stream stream, s32 profStage,
Msrc/algorithm/impl/coll_executor/coll_comm_executor.h+2-1
@@ -136,10 +136,11 @@ 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);
143+ virtual u32 GetLevel0RingNum() const;
143 HcclResult CalUserMemSlices(const HcclDataType dataType, const HcomCollOpInfo *opInfo,144 HcclResult CalUserMemSlices(const HcclDataType dataType, const HcomCollOpInfo *opInfo,
144 const std::vector<Slice> &singleRingSliceZero, u32 ringIndex,145 const std::vector<Slice> &singleRingSliceZero, u32 ringIndex,
145 const std::vector<std::vector<u32>> &multiRingsOrder,146 const std::vector<std::vector<u32>> &multiRingsOrder,
Msrc/algorithm/impl/coll_executor/coll_reduce_scatter/CMakeLists.txt+1-0
@@ -12,6 +12,7 @@ set(src_list
12 ${CMAKE_CURRENT_SOURCE_DIR}/coll_reduce_scatter_deter_executor.cc12 ${CMAKE_CURRENT_SOURCE_DIR}/coll_reduce_scatter_deter_executor.cc
13 ${CMAKE_CURRENT_SOURCE_DIR}/coll_reduce_scatter_fast_double_ring_for_910_93_executor.cc13 ${CMAKE_CURRENT_SOURCE_DIR}/coll_reduce_scatter_fast_double_ring_for_910_93_executor.cc
14 ${CMAKE_CURRENT_SOURCE_DIR}/coll_aligned_reduce_scatter_double_ring_for_910_93_executor.cc14 ${CMAKE_CURRENT_SOURCE_DIR}/coll_aligned_reduce_scatter_double_ring_for_910_93_executor.cc
15+ ${CMAKE_CURRENT_SOURCE_DIR}/coll_reduce_scatter_pipeline_for_910_93_executor.cc
15 ${CMAKE_CURRENT_SOURCE_DIR}/coll_reduce_scatter_ring_for_910_93_executor.cc16 ${CMAKE_CURRENT_SOURCE_DIR}/coll_reduce_scatter_ring_for_910_93_executor.cc
16 ${CMAKE_CURRENT_SOURCE_DIR}/coll_reduce_scatter_executor.cc17 ${CMAKE_CURRENT_SOURCE_DIR}/coll_reduce_scatter_executor.cc
17 ${CMAKE_CURRENT_SOURCE_DIR}/coll_reduce_scatter_mesh_dma_elimination.cc18 ${CMAKE_CURRENT_SOURCE_DIR}/coll_reduce_scatter_mesh_dma_elimination.cc
Asrc/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_pipeline_for_910_93_executor.cc+531-0
@@ -0,0 +1,531 @@
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_reduce_scatter_pipeline_for_910_93_executor.h"
12+#include "alg_template_register.h"
13+ 
14+namespace hccl {
15+namespace {
16+constexpr u32 PIPELINE_BUFFER_NUM = 2;
17+constexpr u32 PIPELINE_PINGPONG_NOTIFY_PAIRS = PIPELINE_BUFFER_NUM;
18+// HCCL 按 stream:notify = 1:1 分配；pipeline 仅使用最后一条额外 stream 承载 L2，
19+// 其余额外 stream 仅用于获取 ping-pong notify 对。
20+constexpr u32 PIPELINE_EXTRA_STREAM_NUM = PIPELINE_PINGPONG_NOTIFY_PAIRS;
21+}
22+ 
23+CollReduceScatterPipelineFor91093Executor::
24+ CollReduceScatterPipelineFor91093Executor(
25+ const HcclDispatcher dispatcher,
26+ std::unique_ptr<TopoMatcher> &topoMatcher)
27+ : CollReduceScatterRingFor91093Executor(dispatcher, topoMatcher)
28+{
29+}
30+ 
31+HcclResult CollReduceScatterPipelineFor91093Executor::CalcStreamNum(u32 &streamNum)
32+{
33+ CHK_RET(CollReduceScatterRingFor91093Executor::CalcStreamNum(streamNum));
34+ streamNum += PIPELINE_EXTRA_STREAM_NUM;
35+ HCCL_INFO("[CollReduceScatterPipelineFor91093Executor][CalcStreamNum] tag[%s] streamNum[%u]",
36+ tag_.c_str(), streamNum);
37+ return HCCL_SUCCESS;
38+}
39+ 
40+u64 CollReduceScatterPipelineFor91093Executor::CalcLoopMaxCount(const u32 unitSize)
41+{
42+ const u64 maxSizePerLoopUnaligned = inCCLbufferSize_ / topoAttr_.userRankSize / PIPELINE_BUFFER_NUM;
43+ const u64 maxSizePerLoop = maxSizePerLoopUnaligned / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN;
44+ const u64 maxCountPerLoop = maxSizePerLoop / unitSize;
45+ HCCL_INFO("[CollReduceScatterPipelineFor91093Executor][CalcLoopMaxCount] "
46+ "maxCountPerLoop[%llu], maxSizePerLoop[%llu]", maxCountPerLoop, maxSizePerLoop);
47+ return maxCountPerLoop;
48+}
49+ 
50+HcclResult CollReduceScatterPipelineFor91093Executor::RunLoop(OpParam &param, AlgResourceResponse &algRes)
51+{
52+ if (param.DataDes.count == 0) {
53+ return HCCL_SUCCESS;
54+ }
55+ 
56+ const u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
57+ 
58+ Stream streamL0L1 = param.stream;
59+ Stream streamL2 = algResResp_->slaveStreams.back();
60+ const u32 baseNotifyIndex = algResResp_->notifiesMain.size() - PIPELINE_PINGPONG_NOTIFY_PAIRS;
61+ auto notifyL0L1toL2A = algResResp_->notifiesMain[baseNotifyIndex];
62+ auto notifyL0L1toL2B = algResResp_->notifiesMain[baseNotifyIndex + 1];
63+ auto notifyL2toL0L1A = algResResp_->notifiesAux[baseNotifyIndex];
64+ auto notifyL2toL0L1B = algResResp_->notifiesAux[baseNotifyIndex + 1];
65+ PipelineLoopContext ctx;
66+ CHK_RET(BuildPipelineLoopContext(param, algRes, unitSize, ctx));
67+ CHK_RET(GetLevelCommInfo());
68+ 
69+ auto getForwardNotify = [&](u64 blockIdx) -> std::shared_ptr<LocalNotify> {
70+ return (blockIdx % PIPELINE_PINGPONG_NOTIFY_PAIRS == 0) ? notifyL0L1toL2A : notifyL0L1toL2B;
71+ };
72+ auto getBackwardNotify = [&](u64 blockIdx) -> std::shared_ptr<LocalNotify> {
73+ return (blockIdx % PIPELINE_PINGPONG_NOTIFY_PAIRS == 0) ? notifyL2toL0L1A : notifyL2toL0L1B;
74+ };
75+ 
76+ const u64 numLoopTotal = ctx.numBlockTotal + 1;
77+ for (u64 i = 0; i < numLoopTotal; ++i) {
78+ if (i < ctx.numBlockTotal) {
79+ if (i >= PIPELINE_BUFFER_NUM) {
80+ CHK_RET(LocalNotify::Wait(streamL0L1, dispatcher_, getBackwardNotify(i)));
81+ }
82+ CHK_RET(RunL0L1Phase(param, ctx, i, streamL0L1));
83+ CHK_RET(LocalNotify::Post(streamL0L1, dispatcher_, getForwardNotify(i)));
84+ }
85+ if (i >= 1 && i <= ctx.numBlockTotal) {
86+ const u64 blockIdx = i - 1;
87+ CHK_RET(LocalNotify::Wait(streamL2, dispatcher_, getForwardNotify(blockIdx)));
88+ CHK_RET(RunL2Phase(param, ctx, blockIdx, streamL2));
89+ CHK_RET(LocalNotify::Post(streamL2, dispatcher_, getBackwardNotify(blockIdx)));
90+ }
91+ CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
92+ }
93+ 
94+ CHK_RET(WaitForRemainingL2Signals(param, ctx.numBlockTotal, streamL0L1, notifyL2toL0L1A, notifyL2toL0L1B));
95+ CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
96+ HCCL_INFO("[CollReduceScatterPipelineFor91093Executor][RunLoop] Pipeline run success");
97+ return HCCL_SUCCESS;
98+}
99+ 
100+// 由 RunLoop 调用
101+HcclResult CollReduceScatterPipelineFor91093Executor::BuildPipelineLoopContext(
102+ OpParam &param, AlgResourceResponse &algRes, const u32 unitSize,
103+ PipelineLoopContext &ctx)
104+{
105+ u8 *curInputPtr = static_cast<u8 *>(param.inputPtr);
106+ u8 *curOutputPtr = static_cast<u8 *>(param.outputPtr);
107+ CHK_PTR_NULL(curInputPtr);
108+ CHK_PTR_NULL(curOutputPtr);
109+ 
110+ const u64 maxCountDataPerLoop = CalcLoopMaxCount(unitSize);
111+ const u64 targetCountDataPerLoop = HCCL_SMALL_COUNT_8_MB / unitSize;
112+ const u64 countDataPerLoop =
113+ maxCountDataPerLoop < targetCountDataPerLoop ? maxCountDataPerLoop : targetCountDataPerLoop;
114+ CHK_PRT_RET(countDataPerLoop == 0,
115+ HCCL_ERROR("[CollReduceScatterPipelineFor91093Executor][BuildPipelineLoopContext]"
116+ " countDataPerLoop is zero."),
117+ HCCL_E_INTERNAL);
118+ 
119+ const u64 countDataLastLoopTemp = param.DataDes.count % countDataPerLoop;
120+ const u64 countDataLastLoop = countDataLastLoopTemp > 0 ? countDataLastLoopTemp : countDataPerLoop;
121+ const u64 cclInputBufferSize = algRes.cclInputMem.size() / PIPELINE_BUFFER_NUM;
122+ const u64 cclOutputBufferSize = algRes.cclOutputMem.size() / PIPELINE_BUFFER_NUM;
123+ ctx.countDataPerLoop = countDataPerLoop;
124+ ctx.countDataLastLoop = countDataLastLoop;
125+ ctx.sizeDataPerLoop = countDataPerLoop * unitSize;
126+ ctx.numBlockTotal = (param.DataDes.count - countDataLastLoop) / countDataPerLoop + 1;
127+ ctx.cclInputBufferSize = cclInputBufferSize;
128+ ctx.cclInputAMem = algRes.cclInputMem.range(0, cclInputBufferSize);
129+ ctx.cclInputBMem = algRes.cclInputMem.range(cclInputBufferSize, cclInputBufferSize);
130+ ctx.cclOutputAMem = algRes.cclOutputMem.range(0, cclOutputBufferSize);
131+ ctx.cclOutputBMem = algRes.cclOutputMem.range(cclOutputBufferSize, cclOutputBufferSize);
132+ ctx.curInputPtr = curInputPtr;
133+ ctx.curOutputPtr = curOutputPtr;
134+ 
135+ HCCL_INFO("[CollReduceScatterPipelineFor91093Executor][BuildPipelineLoopContext] "
136+ "tag[%s] numBlockTotal[%llu] numLoopTotal[%llu] maxCountDataPerLoop[%llu] "
137+ "targetCountDataPerLoop[%llu] countDataPerLoop[%llu] countDataLastLoop[%llu]",
138+ param.tag.c_str(), ctx.numBlockTotal, ctx.numBlockTotal + 1, maxCountDataPerLoop, targetCountDataPerLoop,
139+ ctx.countDataPerLoop, ctx.countDataLastLoop);
140+ return HCCL_SUCCESS;
141+}
142+ 
143+// 由 RunLoop 调用
144+HcclResult CollReduceScatterPipelineFor91093Executor::WaitForRemainingL2Signals(
145+ const OpParam &param, u64 numBlockTotal, Stream &streamL0L1,
146+ const std::shared_ptr<LocalNotify> &notifyL2toL0L1A,
147+ const std::shared_ptr<LocalNotify> &notifyL2toL0L1B)
148+{
149+ const u64 remainingSignals = (numBlockTotal >= PIPELINE_PINGPONG_NOTIFY_PAIRS) ?
150+ PIPELINE_PINGPONG_NOTIFY_PAIRS : numBlockTotal;
151+ const u64 firstBlockIdx = numBlockTotal - remainingSignals;
152+ for (u64 blockIdx = firstBlockIdx; blockIdx < numBlockTotal; ++blockIdx) {
153+ auto notify = (blockIdx % PIPELINE_PINGPONG_NOTIFY_PAIRS == 0) ? notifyL2toL0L1A : notifyL2toL0L1B;
154+ HcclResult ret = LocalNotify::Wait(streamL0L1, dispatcher_, notify);
155+ CHK_PRT_RET(ret != HCCL_SUCCESS,
156+ HCCL_ERROR("[CollReduceScatterPipelineFor91093Executor][WaitForRemainingL2Signals] "
157+ "PostSync wait error, tag[%s] blockIdx[%llu]", param.tag.c_str(), blockIdx), ret);
158+ }
159+ return HCCL_SUCCESS;
160+}
161+ 
162+// 由 RunLoop 循环体调用
163+HcclResult CollReduceScatterPipelineFor91093Executor::RunL0L1Phase(
164+ OpParam &param, const PipelineLoopContext &ctx, u64 blockIdx, Stream &streamL0L1)
165+{
166+ HCCL_CONFIG_INFO(HCCL_ALG,
167+ "[CollReduceScatterPipelineFor91093Executor][RunL0L1Phase] blockIdx[%llu] useBufferA[%d]",
168+ blockIdx, (blockIdx % PIPELINE_BUFFER_NUM == 0));
169+ 
170+ const bool useBufferA = (blockIdx % PIPELINE_BUFFER_NUM == 0);
171+ const bool isLastBlock = (blockIdx == ctx.numBlockTotal - 1);
172+ ExecMem execMem;
173+ execMem.count = isLastBlock ? ctx.countDataLastLoop : ctx.countDataPerLoop;
174+ execMem.inputMem = useBufferA ? ctx.cclInputAMem : ctx.cclInputBMem;
175+ execMem.outputMem = useBufferA ? ctx.cclOutputAMem : ctx.cclOutputBMem;
176+ execMem.scratchMem = execMem.outputMem;
177+ execMem.inputPtr = ctx.curInputPtr + blockIdx * ctx.sizeDataPerLoop;
178+ execMem.outputPtr = ctx.curOutputPtr + blockIdx * ctx.sizeDataPerLoop;
179+ 
180+ const u64 bufferBaseOffset = useBufferA ? 0 : ctx.cclInputBufferSize;
181+ SliceExecMem(param, execMem);
182+ 
183+ HCCL_CONFIG_INFO(HCCL_ALG,
184+ "[CollReduceScatterPipelineFor91093Executor][RunL0L1Phase] chunk starts");
185+ 
186+ HcclResult ret = KernelRunLevel0To1(param, execMem, streamL0L1, bufferBaseOffset);
187+ CHK_PRT_RET(ret != HCCL_SUCCESS,
188+ HCCL_ERROR("[CollReduceScatterPipelineFor91093Executor][RunL0L1Phase] kernel run error, tag[%s]",
189+ param.tag.c_str()), ret);
190+ return HCCL_SUCCESS;
191+}
192+ 
193+// 由 RunLoop 循环体调用
194+HcclResult CollReduceScatterPipelineFor91093Executor::RunL2Phase(
195+ OpParam &param, const PipelineLoopContext &ctx, u64 blockIdx, Stream &streamL2)
196+{
197+ HCCL_CONFIG_INFO(HCCL_ALG,
198+ "[CollReduceScatterPipelineFor91093Executor][RunL2Phase] blockIdx[%llu] L2 phase", blockIdx);
199+ 
200+ const bool useBufferA = (blockIdx % PIPELINE_BUFFER_NUM == 0);
201+ const bool isLastBlock = (blockIdx == ctx.numBlockTotal - 1);
202+ ExecMem execMem;
203+ execMem.count = isLastBlock ? ctx.countDataLastLoop : ctx.countDataPerLoop;
204+ execMem.inputMem = useBufferA ? ctx.cclInputAMem : ctx.cclInputBMem;
205+ execMem.outputMem = useBufferA ? ctx.cclOutputAMem : ctx.cclOutputBMem;
206+ execMem.scratchMem = execMem.outputMem;
207+ execMem.inputPtr = ctx.curInputPtr + blockIdx * ctx.sizeDataPerLoop;
208+ execMem.outputPtr = ctx.curOutputPtr + blockIdx * ctx.sizeDataPerLoop;
209+ 
210+ const u64 l2BaseOffset = useBufferA ? 0 : ctx.cclInputBufferSize;
211+ SliceExecMem(param, execMem);
212+ 
213+ HCCL_CONFIG_INFO(HCCL_ALG,
214+ "[CollReduceScatterPipelineFor91093Executor][RunL2Phase] chunk starts");
215+ 
216+ HcclResult ret = KernelRunLevel2(param, execMem, streamL2, l2BaseOffset);
217+ CHK_PRT_RET(ret != HCCL_SUCCESS,
218+ HCCL_ERROR("[CollReduceScatterPipelineFor91093Executor][RunL2Phase] kernel run error, tag[%s]",
219+ param.tag.c_str()), ret);
220+ return HCCL_SUCCESS;
221+}
222+ 
223+// 由 RunL0L1Phase、RunL2Phase 调用
224+void CollReduceScatterPipelineFor91093Executor::SliceExecMem(
225+ const OpParam &param, ExecMem &execMem)
226+{
227+ u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
228+ u64 curSize = execMem.count * unitSize;
229+ u32 sliceNum = topoAttr_.userRankSize;
230+ execMem.inputMem = execMem.inputMem.range(0, curSize * sliceNum);
231+ execMem.outputMem = execMem.outputMem.range(0, curSize);
232+}
233+ 
234+// 由 KernelRunLevel0To1、KernelRunLevel2 调用
235+// Pipeline 约束 !isAHCAlgo，AHC 分支不可达，直接走 COMM_LEVEL2。
236+HcclResult CollReduceScatterPipelineFor91093Executor::GetLevel2CommInfo(
237+ SubCommInfo &level2CommInfo)
238+{
239+ CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1));
240+ level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
241+ return HCCL_SUCCESS;
242+}
243+ 
244+u32 CollReduceScatterPipelineFor91093Executor::GetLevel0RingNum() const
245+{
246+ // 排除尾部 Pipeline 专用资源后，与基类 ring 数语义一致。
247+ return algResResp_->slaveStreams.size() + 1 - PIPELINE_EXTRA_STREAM_NUM;
248+}
249+ 
250+HcclResult CollReduceScatterPipelineFor91093Executor::RunIntraSeverReduceScatter(
251+ const std::string &tag, DeviceMem &inputMem, DeviceMem &outputMem,
252+ const u64 count, const HcclDataType &dataType, const HcclReduceOp &reductionOp,
253+ const std::vector<std::vector<Slice>> &multRingsSliceZero, const Stream &stream, s32 profStage,
254+ const u64 baseOffset, const HcomCollOpInfo *opInfo,
255+ const std::vector<std::vector<Slice>> &multRingsUserMemSlice, const bool disableDMAReduce)
256+{
257+ // SemiRing（IsUnifiedMarch）分支不可达：Pipeline 约束 workflowMode_==OP_BASE 排除图模式，
258+ // superPodNum>1 排除单 server，IsUnifiedMarch 恒为 false。
259+ HcclResult ret = HCCL_SUCCESS;
260+ if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
261+ ret = DoubleRingReduceScatter(tag, inputMem, outputMem, count, dataType, reductionOp,
262+ multRingsSliceZero, stream, profStage, baseOffset, opInfo,
263+ multRingsUserMemSlice, disableDMAReduce);
264+ } else {
265+ ret = CollReduceScatterRingFor91093Executor::RunIntraSeverReduceScatter(
266+ tag, inputMem, outputMem, count, dataType, reductionOp,
267+ multRingsSliceZero, stream, profStage, baseOffset, opInfo,
268+ multRingsUserMemSlice, disableDMAReduce);
269+ }
270+ 
271+ CHK_RET(ret);
272+ return HCCL_SUCCESS;
273+}
274+ 
275+// 逻辑与 CollAlignedReduceScatterDoubleRingFor91093Executor::DoubleRingReduceScatter 一致
276+HcclResult CollReduceScatterPipelineFor91093Executor::DoubleRingReduceScatter(
277+ const std::string &tag, DeviceMem inputMem, DeviceMem outputMem,
278+ const u64 count, const HcclDataType dataType, const HcclReduceOp reductionOp,
279+ const std::vector<std::vector<Slice>> multRingsSliceZero, Stream stream, s32 profStage,
280+ const u64 baseOffset, const HcomCollOpInfo *opInfo,
281+ const std::vector<std::vector<Slice>> multRingsUserMemSlice, const bool disableDMAReduce)
282+{
283+ (void)tag;
284+ HCCL_CONFIG_INFO(HCCL_ALG,
285+ "[CollReduceScatterPipelineFor91093Executor][DoubleRingReduceScatter] DoubleRingReduceScatter starts");
286+ u32 ringNum = multRingsSliceZero.size();
287+ CHK_RET(CheckCommSize(COMM_LEVEL0, ringNum));
288+ 
289+ u64 reduceAttr = GetReduceAttr(inputMem, outputMem, dataType, reductionOp);
290+ SubCommInfo level0RingCommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
291+ 
292+ std::vector<std::vector<Slice>> userMemInputSlicesOfDoubleRing;
293+ std::vector<std::vector<u32>> rankOrders;
294+ CHK_RET(PrepareDoubleRingSlices(ringNum, dataType, opInfo, multRingsSliceZero,
295+ multRingsUserMemSlice, userMemInputSlicesOfDoubleRing, rankOrders));
296+ 
297+ std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
298+ TemplateType::TEMPLATE_REDUCESCATTER_DB_RING, dispatcher_);
299+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_DB_RING in COMM_LEVEL0", __func__);
300+ CHK_SMART_PTR_NULL(tempAlg);
301+ // 排除尾部 Pipeline 专用资源流（NotifyReserve + StreamL2），使模板看到与基类一致的流数
302+ std::vector<Stream> baseSlaveStreams(algResResp_->slaveStreams.begin(),
303+ algResResp_->slaveStreams.end() - PIPELINE_EXTRA_STREAM_NUM);
304+ std::vector<std::shared_ptr<LocalNotify>> baseNotifiesMain(algResResp_->notifiesMain.begin(),
305+ algResResp_->notifiesMain.end() - PIPELINE_PINGPONG_NOTIFY_PAIRS);
306+ std::vector<std::shared_ptr<LocalNotify>> baseNotifiesAux(algResResp_->notifiesAux.begin(),
307+ algResResp_->notifiesAux.end() - PIPELINE_PINGPONG_NOTIFY_PAIRS);
308+ HcclResult ret = tempAlg->Prepare(inputMem, inputMem, outputMem, count, dataType, stream, multRingsSliceZero,
309+ reductionOp, LEVEL0_BRIDGE_RANK_ID, baseOffset, disableDMAReduce,
310+ reduceAttr, opInfo, topoAttr_.userRank, baseSlaveStreams,
311+ baseNotifiesMain, baseNotifiesAux, rankOrders, userMemInputSlicesOfDoubleRing);
312+ CHK_PRT_RET(ret != HCCL_SUCCESS,
313+ HCCL_ERROR("[CollReduceScatterPipelineFor91093Executor][DoubleRingReduceScatter] "
314+ "Double ring ReduceScatter failed,return[%d]", ret), ret);
315+ 
316+ u32 ringIndexOp = COMM_INDEX_0;
317+ u32 rankSize = level0RingCommInfo.localRankSize;
318+ ret = tempAlg->RegisterProfiler(
319+ ((ringIndexOp + 1) << PROF_RINGINDEX_OFFSET_OF_PLANEID) +
320+ (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0RingCommInfo.localRank,
321+ profStage, HCCL_EXEC_STEP_NOT_SET, stream);
322+ CHK_PRT_RET(ret != HCCL_SUCCESS,
323+ HCCL_ERROR("[CollReduceScatterPipelineFor91093Executor][DoubleRingReduceScatter] "
324+ "Double ring ReduceScatter RegisterProfiler failed,return[%d]", ret), ret);
325+ 
326+ CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem, outputMem, stream, dispatcher_));
327+ ret = RunTemplate(tempAlg, level0RingCommInfo);
328+ CHK_PRT_RET(ret != HCCL_SUCCESS,
329+ HCCL_ERROR("[CollReduceScatterPipelineFor91093Executor][DoubleRingReduceScatter] "
330+ "Double ring ReduceScatter RunTemplate failed,return[%d]", ret), ret);
331+ 
332+ CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem, outputMem, stream, dispatcher_));
333+ return HCCL_SUCCESS;
334+}
335+ 
336+// 由 DoubleRingReduceScatter 调用
337+HcclResult CollReduceScatterPipelineFor91093Executor::PrepareDoubleRingSlices(
338+ u32 ringNum, const HcclDataType dataType, const HcomCollOpInfo *opInfo,
339+ const std::vector<std::vector<Slice>> &multRingsSliceZero,
340+ const std::vector<std::vector<Slice>> &multRingsUserMemSlice,
341+ std::vector<std::vector<Slice>> &userMemInputSlicesOfDoubleRing,
342+ std::vector<std::vector<u32>> &rankOrders)
343+{
344+ SubCommInfo level0ZeroCommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
345+ auto nicList = topoAttr_.nicList;
346+ std::vector<std::vector<u32>> multiRingsOrder =
347+ GetRingsOrderByTopoType(level0ZeroCommInfo.localRankSize, topoType_, nicList);
348+ CHK_RET(CollectMultiRingsUserMemSlices(ringNum, dataType,
349+ opInfo, multRingsSliceZero, multiRingsOrder, multRingsUserMemSlice,
350+ userMemInputSlicesOfDoubleRing));
351+ CHK_RET(CollectMultiRingsRankOrder(ringNum, multiRingsOrder, rankOrders));
352+ return HCCL_SUCCESS;
353+}
354+ 
355+// 拆分自 CollReduceScatterRingFor91093Executor::KernelRun 的 L0+L1 部分
356+HcclResult CollReduceScatterPipelineFor91093Executor::KernelRunLevel0To1(
357+ const OpParam &param, ExecMem &execMem, Stream &streamL0L1, const u64 baseOffset)
358+{
359+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] executor starts", __func__);
360+ u32 perDataSize = 0;
361+ const HcclDataType dataType = param.GetDataType();
362+ CHK_RET(SalGetDataTypeSize(dataType, perDataSize));
363+ 
364+ u32 ringNum;
365+ u32 sliceNum = logicalLevel0CommInfo_.localRankSize;
366+ u32 commIndex = logicalLevel0CommInfo_.localRank;
367+ 
368+ if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
369+ ringNum = LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE;
370+ } else {
371+ ringNum = LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
372+ }
373+ 
374+ SubCommInfo level2CommInfo;
375+ CHK_RET(GetLevel2CommInfo(level2CommInfo));
376+ const u32 level2RankSize = level2CommInfo.localRankSize;
377+ const u32 level1RankSize = logicalLevel1CommInfo_.localRankSize;
378+ 
379+ std::vector<std::vector<Slice>> multiStreamSlice;
380+ std::vector<std::vector<Slice>> level0DataSegsSlice;
381+ CHK_RET(CalLevel0DataSegsSlice(execMem, multiStreamSlice, param, ringNum, sliceNum, level1RankSize, level2RankSize,
382+ dataType, level0DataSegsSlice));
383+ 
384+ HcomCollOpInfo opInfo = GetHcomCollOpInfo(param, execMem);
385+ HcomCollOpInfo *opInfoPtr = &opInfo;
386+ 
387+ bool disableDMAReduce = algOpContext_.opRetryHandler.retryEnable &&
388+ (algOpContext_.opRetryHandler.inPlaceSupportRetryStatus ==
389+ InplaceSupportRetryStatus::RETRY_1_ALLOW_NO_DMA_REDUCE_CASE1 ||
390+ algOpContext_.opRetryHandler.inPlaceSupportRetryStatus ==
391+ InplaceSupportRetryStatus::RETRY_1_ALLOW_NO_DMA_REDUCE_CASE2);
392+ std::vector<std::vector<Slice>> multRingsUserMemSlice;
393+ CHK_RET(CalUserMemDataSegsSlice(execMem, level0DataSegsSlice, multiStreamSlice, param, ringNum, sliceNum,
394+ level1RankSize, level2RankSize, dataType, perDataSize, opInfoPtr, disableDMAReduce, multRingsUserMemSlice));
395+ 
396+ HcomCollOpInfo opInfoByReduceScatterDMAreduce = *opInfoPtr;
397+ opInfoByReduceScatterDMAreduce.outputAddr = nullptr;
398+ CHK_RET(RunIntraSeverReduceScatter(param.tag, execMem.inputMem, execMem.scratchMem, execMem.count,
399+ dataType, param.reduceType, level0DataSegsSlice, streamL0L1, PROF_STAGE_1, baseOffset,
400+ &opInfoByReduceScatterDMAreduce, multRingsUserMemSlice, disableDMAReduce));
401+ 
402+ if (level1RankSize > 1) {
403+ CHK_RET(RunLevel1Template(param, execMem, streamL0L1, baseOffset,
404+ commIndex, sliceNum, level1RankSize, level2RankSize, perDataSize));
405+ }
406+ 
407+ return HCCL_SUCCESS;
408+}
409+ 
410+// 由 KernelRunLevel0To1 调用
411+HcclResult CollReduceScatterPipelineFor91093Executor::RunLevel1Template(
412+ const OpParam &param, ExecMem &execMem, Stream &streamL0L1, u64 baseOffset,
413+ u32 commIndex, u32 sliceNum, u32 level1RankSize, u32 level2RankSize,
414+ u32 perDataSize)
415+{
416+ const HcclDataType dataType = param.GetDataType();
417+ u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.scratchMem, dataType, param.reduceType);
418+ std::unique_ptr<AlgTemplateBase> level1TempAlg;
419+ 
420+ std::vector<Slice> level1DataSegsSlice;
421+ CHK_RET(CalLevel1DataSegsSlice(execMem, param, logicalLevel1plane_, commIndex, sliceNum, level1RankSize,
422+ level2RankSize, perDataSize, level1DataSegsSlice));
423+ 
424+ if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
425+ level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
426+ TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
427+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RING in COMM_LEVEL1", __func__);
428+ CHK_SMART_PTR_NULL(level1TempAlg);
429+ CHK_RET(level1TempAlg->Prepare(reduceAttr));
430+ } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
431+ level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
432+ TemplateType::TEMPLATE_REDUCESCATTER_NB, dispatcher_);
433+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NB in COMM_LEVEL1", __func__);
434+ CHK_SMART_PTR_NULL(level1TempAlg);
435+ CHK_RET(level1TempAlg->Prepare(reduceAttr));
436+ } else {
437+ level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
438+ TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
439+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR in COMM_LEVEL1", __func__);
440+ CHK_SMART_PTR_NULL(level1TempAlg);
441+ CHK_RET(level1TempAlg->Prepare(reduceAttr, false));
442+ }
443+ 
444+ CHK_RET(level1TempAlg->Prepare(execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count,
445+ dataType, streamL0L1, param.reduceType, LEVEL0_BRIDGE_RANK_ID, level1DataSegsSlice, baseOffset));
446+ CHK_RET(level1TempAlg->RegisterProfiler(
447+ (level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + logicalLevel1CommInfo_.localRank,
448+ PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET, streamL0L1));
449+ CHK_RET(RunTemplate(level1TempAlg, logicalLevel1CommInfo_));
450+ return HCCL_SUCCESS;
451+}
452+ 
453+// 拆分自 CollReduceScatterRingFor91093Executor::KernelRun 的 L2+copyOut 部分，stream 替换为 streamL2
454+HcclResult CollReduceScatterPipelineFor91093Executor::KernelRunLevel2(
455+ const OpParam &param, ExecMem &execMem, Stream &streamL2, const u64 baseOffset)
456+{
457+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] executor starts", __func__);
458+ u32 perDataSize = 0;
459+ const HcclDataType dataType = param.GetDataType();
460+ CHK_RET(SalGetDataTypeSize(dataType, perDataSize));
461+ 
462+ SubCommInfo level2CommInfo;
463+ CHK_RET(GetLevel2CommInfo(level2CommInfo));
464+ const u32 level2RankSize = level2CommInfo.localRankSize;
465+ 
466+ CHK_RET(RunLevel2Template(param, execMem, streamL2, baseOffset,
467+ level2CommInfo, level2RankSize, perDataSize));
468+ 
469+ HcomCollOpInfo opInfo = GetHcomCollOpInfo(param, execMem);
470+ HcomCollOpInfo *opInfoPtr = &opInfo;
471+ 
472+ const u64 offset = CalcSrcMemOffset(execMem, param, perDataSize);
473+ DeviceMem srcMem = execMem.inputMem.range(offset, execMem.outputMem.size());
474+ if (opInfoPtr != nullptr) {
475+ DeviceMem dstMem = DeviceMem::create(static_cast<u8 *>(opInfoPtr->outputAddr), execMem.outputMem.size());
476+ CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, streamL2));
477+ } else {
478+ CHK_RET(HcclD2DMemcpyAsync(dispatcher_, execMem.outputMem, srcMem, streamL2));
479+ }
480+ 
481+ return HCCL_SUCCESS;
482+}
483+ 
484+// 由 KernelRunLevel2 调用
485+HcclResult CollReduceScatterPipelineFor91093Executor::RunLevel2Template(
486+ const OpParam &param, ExecMem &execMem, Stream &streamL2, u64 baseOffset,
487+ const SubCommInfo &level2CommInfo, u32 level2RankSize, u32 perDataSize)
488+{
489+ const HcclDataType dataType = param.GetDataType();
490+ u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.scratchMem, dataType, param.reduceType);
491+ 
492+ std::vector<Slice> level2DataSegsSlice;
493+ CHK_RET(CalLevel2DataSegsSlice(execMem, param, level2RankSize, perDataSize, level2DataSegsSlice));
494+ 
495+ std::unique_ptr<AlgTemplateBase> level2TempAlg;
496+ if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
497+ level2TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
498+ TemplateType::TEMPLATE_REDUCESCATTER_NB, dispatcher_);
499+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NB in COMM_LEVEL2", __func__);
500+ CHK_SMART_PTR_NULL(level2TempAlg);
501+ CHK_RET(level2TempAlg->Prepare(reduceAttr));
502+ } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
503+ level2TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
504+ TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
505+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR in COMM_LEVEL2", __func__);
506+ CHK_SMART_PTR_NULL(level2TempAlg);
507+ CHK_RET(level2TempAlg->Prepare(reduceAttr, false));
508+ if (algoAttr_.isSupportAtomicWrite) {
509+ level2TempAlg->CloseBarrier();
510+ }
511+ } else {
512+ level2TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
513+ TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
514+ HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RING in COMM_LEVEL2", __func__);
515+ CHK_SMART_PTR_NULL(level2TempAlg);
516+ CHK_RET(level2TempAlg->Prepare(reduceAttr));
517+ }
518+ 
519+ CHK_RET(level2TempAlg->Prepare(execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count,
520+ dataType, streamL2, param.reduceType, LEVEL0_BRIDGE_RANK_ID, level2DataSegsSlice, baseOffset));
521+ CHK_RET(level2TempAlg->RegisterProfiler(
522+ (level2RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level2CommInfo.localRank,
523+ PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET, streamL2));
524+ CHK_RET(RunTemplate(level2TempAlg, level2CommInfo));
525+ return HCCL_SUCCESS;
526+}
527+ 
528+REGISTER_EXEC("ReduceScatterPipelineFor91093Executor",
529+ ReduceScatterPipelineFor91093,
530+ CollReduceScatterPipelineFor91093Executor);
531+}
Asrc/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_pipeline_for_910_93_executor.h+89-0
@@ -0,0 +1,89 @@
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_REDUCESCATTER_PIPELINE_FOR_910_93_EXECUTOR_H
12+#define COLL_REDUCESCATTER_PIPELINE_FOR_910_93_EXECUTOR_H
13+#include "coll_reduce_scatter_ring_for_910_93_executor.h"
14+ 
15+namespace hccl {
16+class CollReduceScatterPipelineFor91093Executor
17+ : public CollReduceScatterRingFor91093Executor {
18+public:
19+ explicit CollReduceScatterPipelineFor91093Executor(
20+ const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher> &topoMatcher);
21+ ~CollReduceScatterPipelineFor91093Executor() override = default;
22+ 
23+private:
24+ struct PipelineLoopContext {
25+ u64 countDataPerLoop;
26+ u64 countDataLastLoop;
27+ u64 sizeDataPerLoop;
28+ u64 numBlockTotal;
29+ u64 cclInputBufferSize;
30+ DeviceMem cclInputAMem;
31+ DeviceMem cclInputBMem;
32+ DeviceMem cclOutputAMem;
33+ DeviceMem cclOutputBMem;
34+ u8 *curInputPtr;
35+ u8 *curOutputPtr;
36+ };
37+ 
38+ HcclResult CalcStreamNum(u32 &streamNum) override;
39+ u64 CalcLoopMaxCount(const u32 unitSize) override;
40+ HcclResult RunLoop(OpParam &param, AlgResourceResponse &algRes) override;
41+ HcclResult BuildPipelineLoopContext(OpParam &param, AlgResourceResponse &algRes,
42+ const u32 unitSize, PipelineLoopContext &ctx);
43+ HcclResult WaitForRemainingL2Signals(const OpParam &param, u64 numBlockTotal,
44+ Stream &streamL0L1, const std::shared_ptr<LocalNotify> &notifyL2toL0L1A,
45+ const std::shared_ptr<LocalNotify> &notifyL2toL0L1B);
46+ HcclResult RunIntraSeverReduceScatter(const std::string &tag, DeviceMem &inputMem, DeviceMem &outputMem,
47+ const u64 count, const HcclDataType &dataType, const HcclReduceOp &reductionOp,
48+ const std::vector<std::vector<Slice>> &multRingsSliceZero, const Stream &stream, s32 profStage,
49+ const u64 baseOffset = 0, const HcomCollOpInfo *opInfo = nullptr,
50+ const std::vector<std::vector<Slice>> &multRingsUserMemSlice = std::vector<std::vector<Slice>>(0),
51+ const bool disableDMAReduce = false) override;
52+ 
53+ void SliceExecMem(const OpParam &param, ExecMem &execMem);
54+ 
55+ HcclResult GetLevel2CommInfo(SubCommInfo &level2CommInfo);
56+ 
57+ HcclResult RunL0L1Phase(OpParam &param, const PipelineLoopContext &ctx, u64 blockIdx, Stream &streamL0L1);
58+ HcclResult RunL2Phase(OpParam &param, const PipelineLoopContext &ctx, u64 blockIdx, Stream &streamL2);
59+ 
60+ HcclResult KernelRunLevel0To1(const OpParam &param, ExecMem &execMem, Stream &streamL0L1, const u64 baseOffset);
61+ HcclResult KernelRunLevel2(const OpParam &param, ExecMem &execMem, Stream &streamL2, const u64 baseOffset);
62+ 
63+ HcclResult PrepareDoubleRingSlices(u32 ringNum, const HcclDataType dataType,
64+ const HcomCollOpInfo *opInfo,
65+ const std::vector<std::vector<Slice>> &multRingsSliceZero,
66+ const std::vector<std::vector<Slice>> &multRingsUserMemSlice,
67+ std::vector<std::vector<Slice>> &userMemInputSlicesOfDoubleRing,
68+ std::vector<std::vector<u32>> &rankOrders);
69+ 
70+ HcclResult RunLevel1Template(const OpParam &param, ExecMem &execMem,
71+ Stream &streamL0L1, u64 baseOffset, u32 commIndex, u32 sliceNum,
72+ u32 level1RankSize, u32 level2RankSize, u32 perDataSize);
73+ 
74+ HcclResult RunLevel2Template(const OpParam &param, ExecMem &execMem,
75+ Stream &streamL2, u64 baseOffset, const SubCommInfo &level2CommInfo,
76+ u32 level2RankSize, u32 perDataSize);
77+ 
78+ HcclResult DoubleRingReduceScatter(const std::string &tag, DeviceMem inputMem, DeviceMem outputMem,
79+ const u64 count, const HcclDataType dataType, const HcclReduceOp reductionOp,
80+ const std::vector<std::vector<Slice>> multRingsSliceZero, Stream stream, s32 profStage,
81+ const u64 baseOffset, const HcomCollOpInfo *opInfo,
82+ const std::vector<std::vector<Slice>> multRingsUserMemSlice, const bool disableDMAReduce);
83+ 
84+ u32 GetLevel0RingNum() const override;
85+};
86+ 
87+} // namespace hccl
88+ 
89+#endif
Msrc/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_ring_for_910_93_executor.h+25-24
@@ -21,23 +21,7 @@ public:
21 21 
22protected:22protected:
23 u64 CalcTotalCount(const OpParam &param) const;23 u64 CalcTotalCount(const OpParam &param) const;
24- 
25-private:
26- void ParseParam(const OpParam& param) override;
27- /* *************** 资源计算 *************** */
28- bool isZeroCopy_= false;
29- HcclResult CalcScratchMemSize(u64& scratchMemSize) override;
30 HcclResult CalcStreamNum(u32& streamNum) override;24 HcclResult CalcStreamNum(u32& streamNum) override;
31- HcclResult CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport) override;
32- HcclResult CalcLevel0CommInfo(TransportMemType inputType,
33- TransportMemType outputType,
34- std::vector<LevelNSubCommTransport>& opTransport) override;
35- HcclResult CalcLevel2CommInfo(TransportMemType inputType,
36- TransportMemType outputType,
37- std::vector<LevelNSubCommTransport>& opTransport) override;
38- HcclResult CalcTransportMemType(TransportMemType &inputType, TransportMemType &outputType);
39- 
40- /* *************** 算法编排 *************** */
41 u64 CalcLoopMaxCount(const u32 unitSize) override;25 u64 CalcLoopMaxCount(const u32 unitSize) override;
42 bool IsHugeData(const u64 curSize, OpParam *param = nullptr) override;26 bool IsHugeData(const u64 curSize, OpParam *param = nullptr) override;
43 virtual HcclResult RunIntraSeverReduceScatter(const std::string &tag, DeviceMem &inputMem, DeviceMem &outputMem,27 virtual HcclResult RunIntraSeverReduceScatter(const std::string &tag, DeviceMem &inputMem, DeviceMem &outputMem,
@@ -47,15 +31,9 @@ private:
47 const std::vector<std::vector<Slice>> &multRingsUserMemSlice = std::vector<std::vector<Slice>>(0),31 const std::vector<std::vector<Slice>> &multRingsUserMemSlice = std::vector<std::vector<Slice>>(0),
48 const bool disableDMAReduce = false);32 const bool disableDMAReduce = false);
49 virtual HcclResult GetLevelCommInfo();33 virtual HcclResult GetLevelCommInfo();
50- HcclResult KernelRun(const OpParam &param, ExecMem &execMem) override;
51- HcclResult Getlevel1CommRank(SubCommInfo& level1CommInfo) override;
52- HcclResult SelectTempAlg(std::unique_ptr<AlgTemplateBase> &level1TempAlg, u32 level1RankSize) override;
53- virtual bool IsUnifiedMarch(const OpParam &param) const;
54 HcomCollOpInfo GetHcomCollOpInfo(const OpParam &param, const ExecMem &execMem) const;34 HcomCollOpInfo GetHcomCollOpInfo(const OpParam &param, const ExecMem &execMem) const;
35+ virtual bool IsUnifiedMarch(const OpParam &param) const;
55 u64 CalcSrcMemOffset(const ExecMem &execMem, const OpParam &param, u32 perDataSize) const;36 u64 CalcSrcMemOffset(const ExecMem &execMem, const OpParam &param, u32 perDataSize) const;
56- /* **************** 数据准备*************** */
57- virtual void FillMultiRingSlice(const ExecMem &execMem, const std::vector<std::vector<Slice>> &multiStreamSlice,
58- u32 sliceNum, u32 level1RankSize, u32 level2RankSize, const u32 ringIndex, std::vector<Slice> &dataSlice);
59 virtual HcclResult CalLevel0DataSegsSlice(const ExecMem &execMem, std::vector<std::vector<Slice>> &multiStreamSlice,37 virtual HcclResult CalLevel0DataSegsSlice(const ExecMem &execMem, std::vector<std::vector<Slice>> &multiStreamSlice,
60 const OpParam &param, u32 ringNum, u32 sliceNum, u32 level1RankSize, u32 level2RankSize, HcclDataType dataType,38 const OpParam &param, u32 ringNum, u32 sliceNum, u32 level1RankSize, u32 level2RankSize, HcclDataType dataType,
61 std::vector<std::vector<Slice>> &level0DataSegsSlice);39 std::vector<std::vector<Slice>> &level0DataSegsSlice);
@@ -70,6 +48,28 @@ private:
70 virtual HcclResult CalLevel2DataSegsSlice(const ExecMem &execMem, const OpParam &param, u32 level2RankSize,48 virtual HcclResult CalLevel2DataSegsSlice(const ExecMem &execMem, const OpParam &param, u32 level2RankSize,
71 u32 perDataSize, std::vector<Slice> &level2DataSegsSlice);49 u32 perDataSize, std::vector<Slice> &level2DataSegsSlice);
72 50 
51+private:
52+ void ParseParam(const OpParam& param) override;
53+ /* *************** 资源计算 *************** */
54+ bool isZeroCopy_= false;
55+ HcclResult CalcScratchMemSize(u64& scratchMemSize) override;
56+ HcclResult CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport) override;
57+ HcclResult CalcLevel0CommInfo(TransportMemType inputType,
58+ TransportMemType outputType,
59+ std::vector<LevelNSubCommTransport>& opTransport) override;
60+ HcclResult CalcLevel2CommInfo(TransportMemType inputType,
61+ TransportMemType outputType,
62+ std::vector<LevelNSubCommTransport>& opTransport) override;
63+ HcclResult CalcTransportMemType(TransportMemType &inputType, TransportMemType &outputType);
64+ 
65+ /* *************** 算法编排 *************** */
66+ HcclResult KernelRun(const OpParam &param, ExecMem &execMem) override;
67+ HcclResult Getlevel1CommRank(SubCommInfo& level1CommInfo) override;
68+ HcclResult SelectTempAlg(std::unique_ptr<AlgTemplateBase> &level1TempAlg, u32 level1RankSize) override;
69+ /* **************** 数据准备*************** */
70+ virtual void FillMultiRingSlice(const ExecMem &execMem, const std::vector<std::vector<Slice>> &multiStreamSlice,
71+ u32 sliceNum, u32 level1RankSize, u32 level2RankSize, const u32 ringIndex, std::vector<Slice> &dataSlice);
72+ 
73 using Level0SlicesCalculator = void(*)(const OpParam &param, u32 sliceNum, u32 level1RankSize, u32 level1Index,73 using Level0SlicesCalculator = void(*)(const OpParam &param, u32 sliceNum, u32 level1RankSize, u32 level1Index,
74 u32 level2Index, u32 perDataSize, std::vector<Slice> &segSlices);74 u32 level2Index, u32 perDataSize, std::vector<Slice> &segSlices);
75 static void PrepareLevel0Slices(const OpParam &param, u32 sliceNum, u32 level1RankSize, u32 level1Index,75 static void PrepareLevel0Slices(const OpParam &param, u32 sliceNum, u32 level1RankSize, u32 level1Index,
@@ -92,6 +92,7 @@ private:
92 u32 sliceNum, u32 level1RankSize, u32 level2RankSize, u32 perDataSize, std::vector<Slice> &level1DataSegsSlice);92 u32 sliceNum, u32 level1RankSize, u32 level2RankSize, u32 perDataSize, std::vector<Slice> &level1DataSegsSlice);
93 virtual HcclResult CalLevel2DataSegsSliceV(const OpParam &param, u32 level2RankSize, u32 perDataSize,93 virtual HcclResult CalLevel2DataSegsSliceV(const OpParam &param, u32 level2RankSize, u32 perDataSize,
94 std::vector<Slice> &level2DataSegsSlice);94 std::vector<Slice> &level2DataSegsSlice);
95+ 
95protected:96protected:
96 SubCommInfo logicalLevel0CommInfo_ = {0, 0, {}, {}};97 SubCommInfo logicalLevel0CommInfo_ = {0, 0, {}, {}};
97 SubCommInfo logicalLevel1CommInfo_ = {0, 0, {}, {}};98 SubCommInfo logicalLevel1CommInfo_ = {0, 0, {}, {}};
@@ -101,4 +102,4 @@ protected:
101 102 
102} // namespace hccl103} // namespace hccl
103 104 
104-#endif105+#endif
Msrc/algorithm/impl/operator/reduce_scatter_operator.cc+28-0
@@ -23,7 +23,12 @@ constexpr u32 HCCL_310P_DATA_SIZE_MID_COUNT = 320 * 1024;
23constexpr u32 HCCL_310P_DATA_SIZE_SMALL_COUNT = 1024;23constexpr u32 HCCL_310P_DATA_SIZE_SMALL_COUNT = 1024;
24constexpr u32 HCCL_310P_SLIM_RING_MAX_SIZE = 8;24constexpr u32 HCCL_310P_SLIM_RING_MAX_SIZE = 8;
25 25 
26+// Pipeline并行比串行更优的总数据量临界点（基于910_93 2 SuperPod, 1 Server, 12 Rank 性能基线测试，
27+// 数据量低于此值时调度开销超过流水收益）
28+constexpr u64 HCCL_PIPELINE_TOTAL_DATA_SIZE_THRESHOLD = 608 * 1024 * 1024;
29+ 
26namespace hccl {30namespace hccl {
31+ 
27ReduceScatterOperator::ReduceScatterOperator(AlgConfigurator* algConfigurator, CCLBufferManager &cclBufferManager,32ReduceScatterOperator::ReduceScatterOperator(AlgConfigurator* algConfigurator, CCLBufferManager &cclBufferManager,
28 HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher> &topoMatcher) :33 HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher> &topoMatcher) :
29 CollAlgOperator(algConfigurator, cclBufferManager, dispatcher, topoMatcher, HcclCMDType::HCCL_CMD_REDUCE_SCATTER)34 CollAlgOperator(algConfigurator, cclBufferManager, dispatcher, topoMatcher, HcclCMDType::HCCL_CMD_REDUCE_SCATTER)
@@ -458,6 +463,16 @@ HcclResult ReduceScatterOperator::SelectAlgfor91093(const OpParam& param, std::s
458 u64 smallCountSingleServerThreshold = (hccsPortNum_ == HCCS_PORT_NUM_910_93_7) ? HCCL_SMALL_COUNT_512_KB : HCCL_SMALL_COUNT_1_MB;463 u64 smallCountSingleServerThreshold = (hccsPortNum_ == HCCS_PORT_NUM_910_93_7) ? HCCL_SMALL_COUNT_512_KB : HCCL_SMALL_COUNT_1_MB;
459 u64 smallCountMultiServerThreshold = (hccsPortNum_ == HCCS_PORT_NUM_910_93_7) ? HCCL_SMALL_COUNT_1_MB : HCCL_SMALL_COUNT_2_MB;464 u64 smallCountMultiServerThreshold = (hccsPortNum_ == HCCS_PORT_NUM_910_93_7) ? HCCL_SMALL_COUNT_1_MB : HCCL_SMALL_COUNT_2_MB;
460 CHK_RET(cclBufferManager_.GetInCCLbuffer(commInputPtr, commInputSize));465 CHK_RET(cclBufferManager_.GetInCCLbuffer(commInputPtr, commInputSize));
466+ u64 maxPipelineBlockSize = 0;
467+ if (userRankSize_ != 0) {
468+ maxPipelineBlockSize = commInputSize / userRankSize_ / HCCL_DEVICE_NUM_TWO /
469+ HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN;
470+ }
471+ bool isSupportPipelineFor91093 = (maxPipelineBlockSize >= HCCL_SMALL_COUNT_4_MB) &&
472+ (dataSize * userRankSize_ > HCCL_PIPELINE_TOTAL_DATA_SIZE_THRESHOLD);
473+ HCCL_INFO("[ReduceScatterOperator][SelectAlgfor91093] dataSize[%llu] commInputSize[%llu] "
474+ "userRankSize[%u] maxPipelineBlockSize[%llu] isSupportPipelineFor91093[%d]",
475+ dataSize, commInputSize, userRankSize_, maxPipelineBlockSize, isSupportPipelineFor91093);
461 bool dmaReduceLimit = (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) && isPowOfTwo &&476 bool dmaReduceLimit = (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) && isPowOfTwo &&
462 ((commInputSize * HCCL_DEVICE_NUM_TWO < param.DataDes.count * SIZE_TABLE[param.DataDes.dataType] * userRankSize_) ||477 ((commInputSize * HCCL_DEVICE_NUM_TWO < param.DataDes.count * SIZE_TABLE[param.DataDes.dataType] * userRankSize_) ||
463 retryEnable_);478 retryEnable_);
@@ -513,6 +528,14 @@ HcclResult ReduceScatterOperator::SelectAlgfor91093(const OpParam& param, std::s
513 } else {528 } else {
514 algName = "ReduceScatterRingZerocopyExchangeExecutor"; // 连续数据通信+数据交换（AHC不支持）529 algName = "ReduceScatterRingZerocopyExchangeExecutor"; // 连续数据通信+数据交换（AHC不支持）
515 }530 }
531+ } else if (isOpbase && superPodNum_ > 1 &&
532+ !isAHCAlgo &&
533+ !multiSuperPodDiffDeviceNumMode_ &&
534+ isSupportInlineReduce &&
535+ (topoType_ == TopoType::TOPO_TYPE_NP_SINGLE_RING ||
536+ topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) &&
537+ isSupportPipelineFor91093) {
538+ algName = "ReduceScatterPipelineFor91093Executor";
516 } else {539 } else {
517 if (topoType_ == TopoType::TOPO_TYPE_NP_SINGLE_RING) {540 if (topoType_ == TopoType::TOPO_TYPE_NP_SINGLE_RING) {
518 algName = "ReduceScatterRingFor91093Executor";541 algName = "ReduceScatterRingFor91093Executor";
@@ -551,6 +574,11 @@ HcclResult ReduceScatterOperator::SelectAlgfor91093(const OpParam& param, std::s
551 return HCCL_E_NOT_SUPPORT;574 return HCCL_E_NOT_SUPPORT;
552 }575 }
553 HCCL_INFO("[SelectAlgfor91093] ReduceScatter SelectAlgfor91093 is algName [%s]", algName.c_str());576 HCCL_INFO("[SelectAlgfor91093] ReduceScatter SelectAlgfor91093 is algName [%s]", algName.c_str());
577+
578+ HCCL_INFO("[SelectAlgfor91093] isOpbase[%d] superPodNum_[%u] isAHCAlgo[%d] multiSuperPodDiffDeviceNumMode_[%d] "
579+ "isSupportInlineReduce[%d] topoType_[%d] dataSize[%llu]",
580+ isOpbase, superPodNum_, isAHCAlgo, multiSuperPodDiffDeviceNumMode_,
581+ isSupportInlineReduce, topoType_, dataSize);
554 return HCCL_SUCCESS;582 return HCCL_SUCCESS;
555}583}
556 584 
Msrc/framework/CMakeLists.txt+1-0
@@ -539,6 +539,7 @@ if(NOT BUILD_OPEN_PROJECT OR (BUILD_OPEN_PROJECT AND KERNEL_MODE))
539 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_deter_executor.cc539 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_deter_executor.cc
540 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_fast_double_ring_for_910_93_executor.cc540 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_fast_double_ring_for_910_93_executor.cc
541 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_aligned_reduce_scatter_double_ring_for_910_93_executor.cc541 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_aligned_reduce_scatter_double_ring_for_910_93_executor.cc
542+ ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_pipeline_for_910_93_executor.cc
542 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_ring_for_910_93_executor.cc543 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_ring_for_910_93_executor.cc
543 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_executor.cc544 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_executor.cc
544 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_mesh_dma_elimination.cc545 ${HCCL_BASE_DIR}/algorithm/impl/coll_executor/coll_reduce_scatter/coll_reduce_scatter_mesh_dma_elimination.cc
Mtest/st/algorithm/testcase/testcase/testcase_reduce_scatter.cc+255-0
@@ -2203,3 +2203,258 @@ TEST_F(ReduceScatterTest, ReduceScatterMeshOpbaseSmallCountDeterministicExecutor
2203 ret = checker.Check(checkerOpParam, topoMeta);2203 ret = checker.Check(checkerOpParam, topoMeta);
2204 EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);2204 EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
2205}2205}
2206+ 
2207+// Case 1: DR + RING + 多server; 4块等大(偶数), numLoopTotal=5
2208+TEST_F(ReduceScatterTest, reduce_scatter_910_93_pipeline_DR_2pod_2server_4blk_equal)
2209+{
2210+ constexpr u32 p = 2, s = 2, r = 8, rankSize = p * s * r;
2211+ constexpr u32 buffSizeMB = 1, unitSize = 4;
2212+ constexpr u64 countDataPerLoop = buffSizeMB * 1024 * 1024 / rankSize / 2 / 128 * 128 / unitSize; // 4096
2213+ 
2214+ RankTable_For_LLT gen;
2215+ TopoMeta topoMeta;
2216+ gen.GenTopoMeta(topoMeta, p, s, r);
2217+ 
2218+ setenv("HCCL_BUFFSIZE", std::to_string(buffSizeMB).c_str(), 1);
2219+ 
2220+ CheckerOpParam checkerOpParam;
2221+ checkerOpParam.opType = CheckerOpType::REDUCE_SCATTER;
2222+ checkerOpParam.tag = "ReduceScatter";
2223+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
2224+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
2225+ checkerOpParam.DataDes.count = 4 * countDataPerLoop; // 16384
2226+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_INT32;
2227+ checkerOpParam.reduceType = CheckerReduceOp::REDUCE_SUM;
2228+ checkerOpParam.algName = "ReduceScatterPipelineFor91093Executor";
2229+ 
2230+ Checker checker;
2231+ HcclResult ret;
2232+ ret = checker.Check(checkerOpParam, topoMeta);
2233+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
2234+ 
2235+ unsetenv("HCCL_BUFFSIZE");
2236+}
2237+ 
2238+// Case 2: DR + 3SuperPod + 4rank + BFP16/MAX; 3块(奇数) 尾块小落bufferA, numLoopTotal=4
2239+TEST_F(ReduceScatterTest, reduce_scatter_910_93_pipeline_DR_3pod_4rank_bfp16_max_3blk_smallA)
2240+{
2241+ constexpr u32 p = 3, s = 1, r = 4, rankSize = p * s * r;
2242+ constexpr u32 buffSizeMB = 1, unitSize = 2;
2243+ constexpr u64 countDataPerLoop = buffSizeMB * 1024 * 1024 / rankSize / 2 / 128 * 128 / unitSize; // 21824
2244+ constexpr u64 numFullBlocks = 2;
2245+ constexpr u64 countDataLastLoop = countDataPerLoop / 4; // 5456
2246+ 
2247+ RankTable_For_LLT gen;
2248+ TopoMeta topoMeta;
2249+ gen.GenTopoMeta(topoMeta, p, s, r);
2250+ 
2251+ setenv("HCCL_BUFFSIZE", std::to_string(buffSizeMB).c_str(), 1);
2252+ 
2253+ CheckerOpParam checkerOpParam;
2254+ checkerOpParam.opType = CheckerOpType::REDUCE_SCATTER;
2255+ checkerOpParam.tag = "ReduceScatter";
2256+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
2257+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
2258+ checkerOpParam.DataDes.count = numFullBlocks * countDataPerLoop + countDataLastLoop; // 49104
2259+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_BFP16;
2260+ checkerOpParam.reduceType = CheckerReduceOp::REDUCE_MAX;
2261+ checkerOpParam.algName = "ReduceScatterPipelineFor91093Executor";
2262+ 
2263+ Checker checker;
2264+ HcclResult ret;
2265+ ret = checker.Check(checkerOpParam, topoMeta);
2266+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
2267+ 
2268+ unsetenv("HCCL_BUFFSIZE");
2269+}
2270+ 
2271+// Case 3: DR + FP32/MIN; 4块(偶数) 尾块小落bufferB, numLoopTotal=5
2272+TEST_F(ReduceScatterTest, reduce_scatter_910_93_pipeline_DR_fp32_min_4blk_smallB)
2273+{
2274+ constexpr u32 p = 2, s = 1, r = 8, rankSize = p * s * r;
2275+ constexpr u32 buffSizeMB = 1, unitSize = 4;
2276+ constexpr u64 countDataPerLoop = buffSizeMB * 1024 * 1024 / rankSize / 2 / 128 * 128 / unitSize; // 8192
2277+ constexpr u64 numFullBlocks = 3;
2278+ constexpr u64 countDataLastLoop = countDataPerLoop / 4; // 2048
2279+ 
2280+ RankTable_For_LLT gen;
2281+ TopoMeta topoMeta;
2282+ gen.GenTopoMeta(topoMeta, p, s, r);
2283+ 
2284+ setenv("HCCL_BUFFSIZE", std::to_string(buffSizeMB).c_str(), 1);
2285+ 
2286+ CheckerOpParam checkerOpParam;
2287+ checkerOpParam.opType = CheckerOpType::REDUCE_SCATTER;
2288+ checkerOpParam.tag = "ReduceScatter";
2289+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
2290+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
2291+ checkerOpParam.DataDes.count = numFullBlocks * countDataPerLoop + countDataLastLoop; // 26624
2292+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_FP32;
2293+ checkerOpParam.reduceType = CheckerReduceOp::REDUCE_MIN;
2294+ checkerOpParam.algName = "ReduceScatterPipelineFor91093Executor";
2295+ 
2296+ Checker checker;
2297+ HcclResult ret;
2298+ ret = checker.Check(checkerOpParam, topoMeta);
2299+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
2300+ 
2301+ unsetenv("HCCL_BUFFSIZE");
2302+}
2303+ 
2304+// Case 4: DR + force RING L1/L2 + INT8/SUM; 覆盖默认不覆盖的Level1/Level2 RING路径, numLoopTotal=4
2305+TEST_F(ReduceScatterTest, reduce_scatter_910_93_pipeline_DR_force_ring_int8_sum)
2306+{
2307+ constexpr u32 p = 2, s = 2, r = 8, rankSize = p * s * r;
2308+ constexpr u32 buffSizeMB = 1, unitSize = 1;
2309+ constexpr u64 countDataPerLoop = buffSizeMB * 1024 * 1024 / rankSize / 2 / 128 * 128 / unitSize; // 16384
2310+ 
2311+ RankTable_For_LLT gen;
2312+ TopoMeta topoMeta;
2313+ gen.GenTopoMeta(topoMeta, p, s, r);
2314+ 
2315+ setenv("HCCL_BUFFSIZE", std::to_string(buffSizeMB).c_str(), 1);
2316+ setenv("HCCL_ALGO", "level1:ring;level2:ring", 1);
2317+ 
2318+ CheckerOpParam checkerOpParam;
2319+ checkerOpParam.opType = CheckerOpType::REDUCE_SCATTER;
2320+ checkerOpParam.tag = "ReduceScatter";
2321+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
2322+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
2323+ checkerOpParam.DataDes.count = 3 * countDataPerLoop; // 49152
2324+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_INT8;
2325+ checkerOpParam.reduceType = CheckerReduceOp::REDUCE_SUM;
2326+ checkerOpParam.algName = "ReduceScatterPipelineFor91093Executor";
2327+ 
2328+ Checker checker;
2329+ HcclResult ret;
2330+ ret = checker.Check(checkerOpParam, topoMeta);
2331+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
2332+ 
2333+ unsetenv("HCCL_ALGO");
2334+ unsetenv("HCCL_BUFFSIZE");
2335+}
2336+ 
2337+// Case 5: DR + force NB L1/L2 + INT16/MAX; 覆盖Level1/Level2 NB路径, numLoopTotal=4
2338+TEST_F(ReduceScatterTest, reduce_scatter_910_93_pipeline_DR_force_nb_int16_max)
2339+{
2340+ constexpr u32 p = 2, s = 2, r = 8, rankSize = p * s * r;
2341+ constexpr u32 buffSizeMB = 1, unitSize = 2;
2342+ constexpr u64 countDataPerLoop = buffSizeMB * 1024 * 1024 / rankSize / 2 / 128 * 128 / unitSize; // 8192
2343+ 
2344+ RankTable_For_LLT gen;
2345+ TopoMeta topoMeta;
2346+ gen.GenTopoMeta(topoMeta, p, s, r);
2347+ 
2348+ setenv("HCCL_BUFFSIZE", std::to_string(buffSizeMB).c_str(), 1);
2349+ setenv("HCCL_ALGO", "level1:NB;level2:NB", 1);
2350+ 
2351+ CheckerOpParam checkerOpParam;
2352+ checkerOpParam.opType = CheckerOpType::REDUCE_SCATTER;
2353+ checkerOpParam.tag = "ReduceScatter";
2354+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
2355+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
2356+ checkerOpParam.DataDes.count = 3 * countDataPerLoop; // 24576
2357+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_INT16;
2358+ checkerOpParam.reduceType = CheckerReduceOp::REDUCE_MAX;
2359+ checkerOpParam.algName = "ReduceScatterPipelineFor91093Executor";
2360+ 
2361+ Checker checker;
2362+ HcclResult ret;
2363+ ret = checker.Check(checkerOpParam, topoMeta);
2364+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
2365+ 
2366+ unsetenv("HCCL_ALGO");
2367+ unsetenv("HCCL_BUFFSIZE");
2368+}
2369+ 
2370+// Case 6: SR + 3SuperPod + 奇数rank(7) + BFP16/SUM; Level1默认RING(server=1) + Level2默认NHR, numLoopTotal=4
2371+TEST_F(ReduceScatterTest, reduce_scatter_910_93_pipeline_SR_3pod_7rank_bfp16_sum)
2372+{
2373+ constexpr u32 p = 3, s = 1, r = 7, rankSize = p * s * r;
2374+ constexpr u32 buffSizeMB = 1, unitSize = 2;
2375+ constexpr u64 countDataPerLoop = buffSizeMB * 1024 * 1024 / rankSize / 2 / 128 * 128 / unitSize; // 12480
2376+ 
2377+ RankTable_For_LLT gen;
2378+ TopoMeta topoMeta;
2379+ gen.GenTopoMeta(topoMeta, p, s, r);
2380+ 
2381+ setenv("HCCL_BUFFSIZE", std::to_string(buffSizeMB).c_str(), 1);
2382+ 
2383+ CheckerOpParam checkerOpParam;
2384+ checkerOpParam.opType = CheckerOpType::REDUCE_SCATTER;
2385+ checkerOpParam.tag = "ReduceScatter";
2386+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
2387+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
2388+ checkerOpParam.DataDes.count = 3 * countDataPerLoop; // 37440
2389+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_BFP16;
2390+ checkerOpParam.reduceType = CheckerReduceOp::REDUCE_SUM;
2391+ checkerOpParam.algName = "ReduceScatterPipelineFor91093Executor";
2392+ 
2393+ Checker checker;
2394+ HcclResult ret;
2395+ ret = checker.Check(checkerOpParam, topoMeta);
2396+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
2397+ 
2398+ unsetenv("HCCL_BUFFSIZE");
2399+}
2400+ 
2401+// Case 7: SR + 2pod + 3server + rank/server=1边界 + INT32/SUM; L0退化, L1默认RING(3server) + L2默认NHR, numLoopTotal=4
2402+TEST_F(ReduceScatterTest, reduce_scatter_910_93_pipeline_SR_2pod_3server_1rank_int32_sum)
2403+{
2404+ constexpr u32 p = 2, s = 3, r = 1, rankSize = p * s * r;
2405+ constexpr u32 buffSizeMB = 1, unitSize = 4;
2406+ constexpr u64 countDataPerLoop = buffSizeMB * 1024 * 1024 / rankSize / 2 / 128 * 128 / unitSize; // 21824
2407+ 
2408+ RankTable_For_LLT gen;
2409+ TopoMeta topoMeta;
2410+ gen.GenTopoMeta(topoMeta, p, s, r);
2411+ 
2412+ setenv("HCCL_BUFFSIZE", std::to_string(buffSizeMB).c_str(), 1);
2413+ 
2414+ CheckerOpParam checkerOpParam;
2415+ checkerOpParam.opType = CheckerOpType::REDUCE_SCATTER;
2416+ checkerOpParam.tag = "ReduceScatter";
2417+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
2418+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
2419+ checkerOpParam.DataDes.count = 3 * countDataPerLoop; // 65472
2420+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_INT32;
2421+ checkerOpParam.reduceType = CheckerReduceOp::REDUCE_SUM;
2422+ checkerOpParam.algName = "ReduceScatterPipelineFor91093Executor";
2423+ 
2424+ Checker checker;
2425+ HcclResult ret;
2426+ ret = checker.Check(checkerOpParam, topoMeta);
2427+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
2428+ 
2429+ unsetenv("HCCL_BUFFSIZE");
2430+}
2431+ 
2432+// Case 8: DR + 小数据量退化串行 + FP16/SUM; numBlockTotal=1, Pipeline内部单块无并行重叠
2433+TEST_F(ReduceScatterTest, reduce_scatter_910_93_pipeline_DR_one_block_degenerate_fp16_sum)
2434+{
2435+ constexpr u32 p = 2, s = 1, r = 8;
2436+ constexpr u32 buffSizeMB = 200;
2437+ 
2438+ RankTable_For_LLT gen;
2439+ TopoMeta topoMeta;
2440+ gen.GenTopoMeta(topoMeta, p, s, r);
2441+ 
2442+ setenv("HCCL_BUFFSIZE", std::to_string(buffSizeMB).c_str(), 1);
2443+ 
2444+ CheckerOpParam checkerOpParam;
2445+ checkerOpParam.opType = CheckerOpType::REDUCE_SCATTER;
2446+ checkerOpParam.tag = "ReduceScatter";
2447+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
2448+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
2449+ checkerOpParam.DataDes.count = 8192;
2450+ checkerOpParam.DataDes.dataType = CheckerDataType::DATA_TYPE_FP16;
2451+ checkerOpParam.reduceType = CheckerReduceOp::REDUCE_SUM;
2452+ checkerOpParam.algName = "ReduceScatterPipelineFor91093Executor";
2453+ 
2454+ Checker checker;
2455+ HcclResult ret;
2456+ ret = checker.Check(checkerOpParam, topoMeta);
2457+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
2458+ 
2459+ unsetenv("HCCL_BUFFSIZE");
2460+}