Line data Source code
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 : } // namespace
22 :
23 0 : CollReduceScatterPipelineFor91093Executor::CollReduceScatterPipelineFor91093Executor(
24 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
25 0 : : CollReduceScatterRingFor91093Executor(dispatcher, topoMatcher)
26 0 : {}
27 :
28 0 : HcclResult CollReduceScatterPipelineFor91093Executor::CalcStreamNum(u32& streamNum)
29 : {
30 0 : CHK_RET(CollReduceScatterRingFor91093Executor::CalcStreamNum(streamNum));
31 0 : streamNum += PIPELINE_EXTRA_STREAM_NUM;
32 0 : HCCL_INFO(
33 : "[CollReduceScatterPipelineFor91093Executor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
34 0 : return HCCL_SUCCESS;
35 : }
36 :
37 0 : u64 CollReduceScatterPipelineFor91093Executor::CalcLoopMaxCount(const u32 unitSize)
38 : {
39 0 : const u64 maxSizePerLoopUnaligned = inCCLbufferSize_ / topoAttr_.userRankSize / PIPELINE_BUFFER_NUM;
40 0 : const u64 maxSizePerLoop = maxSizePerLoopUnaligned / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN;
41 0 : const u64 maxCountPerLoop = maxSizePerLoop / unitSize;
42 0 : HCCL_INFO(
43 : "[CollReduceScatterPipelineFor91093Executor][CalcLoopMaxCount] "
44 : "maxCountPerLoop[%llu], maxSizePerLoop[%llu]",
45 : maxCountPerLoop, maxSizePerLoop);
46 0 : return maxCountPerLoop;
47 : }
48 :
49 0 : HcclResult CollReduceScatterPipelineFor91093Executor::RunLoop(OpParam& param, AlgResourceResponse& algRes)
50 : {
51 0 : if (param.DataDes.count == 0) {
52 0 : return HCCL_SUCCESS;
53 : }
54 :
55 0 : const u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
56 :
57 0 : Stream streamL0L1 = param.stream;
58 0 : Stream streamL2 = algResResp_->slaveStreams.back();
59 0 : const u32 baseNotifyIndex = algResResp_->notifiesMain.size() - PIPELINE_PINGPONG_NOTIFY_PAIRS;
60 0 : auto notifyL0L1toL2A = algResResp_->notifiesAux[baseNotifyIndex];
61 0 : auto notifyL0L1toL2B = algResResp_->notifiesAux[baseNotifyIndex + 1];
62 0 : auto notifyL2toL0L1A = algResResp_->notifiesMain[baseNotifyIndex];
63 0 : auto notifyL2toL0L1B = algResResp_->notifiesMain[baseNotifyIndex + 1];
64 0 : HCCL_INFO(
65 : "[CollReduceScatterPipelineFor91093Executor][RunLoop] NotifyIds: "
66 : "L0L1toL2A: Aux[%u], L0L1toL2B: Aux[%u], L2toL0L1A: Main[%u], L2toL0L1B: Main[%u]",
67 : baseNotifyIndex, baseNotifyIndex + 1, baseNotifyIndex, baseNotifyIndex + 1);
68 0 : PipelineLoopContext ctx;
69 0 : CHK_RET(BuildPipelineLoopContext(param, algRes, unitSize, ctx));
70 0 : CHK_RET(GetLevelCommInfo());
71 :
72 0 : auto getForwardNotify = [&](u64 blockIdx) -> std::shared_ptr<LocalNotify> {
73 0 : return (blockIdx % PIPELINE_PINGPONG_NOTIFY_PAIRS == 0) ? notifyL0L1toL2A : notifyL0L1toL2B;
74 0 : };
75 0 : auto getBackwardNotify = [&](u64 blockIdx) -> std::shared_ptr<LocalNotify> {
76 0 : return (blockIdx % PIPELINE_PINGPONG_NOTIFY_PAIRS == 0) ? notifyL2toL0L1A : notifyL2toL0L1B;
77 0 : };
78 :
79 0 : const u64 numLoopTotal = ctx.numBlockTotal + 1;
80 0 : for (u64 i = 0; i < numLoopTotal; ++i) {
81 0 : if (i < ctx.numBlockTotal) {
82 0 : if (i >= PIPELINE_BUFFER_NUM) {
83 0 : CHK_RET(LocalNotify::Wait(streamL0L1, dispatcher_, getBackwardNotify(i)));
84 : }
85 0 : CHK_RET(RunL0L1Phase(param, ctx, i, streamL0L1));
86 0 : CHK_RET(LocalNotify::Post(streamL0L1, dispatcher_, getForwardNotify(i)));
87 : }
88 0 : if (i >= 1 && i <= ctx.numBlockTotal) {
89 0 : const u64 blockIdx = i - 1;
90 0 : CHK_RET(LocalNotify::Wait(streamL2, dispatcher_, getForwardNotify(blockIdx)));
91 0 : CHK_RET(RunL2Phase(param, ctx, blockIdx, streamL2));
92 0 : CHK_RET(LocalNotify::Post(streamL2, dispatcher_, getBackwardNotify(blockIdx)));
93 : }
94 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
95 : }
96 :
97 0 : CHK_RET(WaitForRemainingL2Signals(param, ctx.numBlockTotal, streamL0L1, notifyL2toL0L1A, notifyL2toL0L1B));
98 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
99 0 : HCCL_INFO("[CollReduceScatterPipelineFor91093Executor][RunLoop] Pipeline run success");
100 0 : return HCCL_SUCCESS;
101 0 : }
102 :
103 : // 由 RunLoop 调用
104 0 : HcclResult CollReduceScatterPipelineFor91093Executor::BuildPipelineLoopContext(
105 : OpParam& param, AlgResourceResponse& algRes, const u32 unitSize, PipelineLoopContext& ctx)
106 : {
107 0 : u8* curInputPtr = static_cast<u8*>(param.inputPtr);
108 0 : u8* curOutputPtr = static_cast<u8*>(param.outputPtr);
109 0 : CHK_PTR_NULL(curInputPtr);
110 0 : CHK_PTR_NULL(curOutputPtr);
111 :
112 0 : const u64 maxCountDataPerLoop = CalcLoopMaxCount(unitSize);
113 0 : const u64 targetCountDataPerLoop = HCCL_SMALL_COUNT_8_MB / unitSize;
114 0 : const u64 countDataPerLoop
115 0 : = maxCountDataPerLoop < targetCountDataPerLoop ? maxCountDataPerLoop : targetCountDataPerLoop;
116 0 : CHK_PRT_RET(
117 : countDataPerLoop == 0,
118 : HCCL_ERROR("[CollReduceScatterPipelineFor91093Executor][BuildPipelineLoopContext]"
119 : " countDataPerLoop is zero."),
120 : HCCL_E_INTERNAL);
121 :
122 0 : const u64 countDataLastLoopTemp = param.DataDes.count % countDataPerLoop;
123 0 : const u64 countDataLastLoop = countDataLastLoopTemp > 0 ? countDataLastLoopTemp : countDataPerLoop;
124 0 : const u64 cclInputBufferSize = algRes.cclInputMem.size() / PIPELINE_BUFFER_NUM;
125 0 : const u64 cclOutputBufferSize = algRes.cclOutputMem.size() / PIPELINE_BUFFER_NUM;
126 0 : ctx.countDataPerLoop = countDataPerLoop;
127 0 : ctx.countDataLastLoop = countDataLastLoop;
128 0 : ctx.sizeDataPerLoop = countDataPerLoop * unitSize;
129 0 : ctx.numBlockTotal = (param.DataDes.count - countDataLastLoop) / countDataPerLoop + 1;
130 0 : ctx.cclInputBufferSize = cclInputBufferSize;
131 0 : ctx.cclInputAMem = algRes.cclInputMem.range(0, cclInputBufferSize);
132 0 : ctx.cclInputBMem = algRes.cclInputMem.range(cclInputBufferSize, cclInputBufferSize);
133 0 : ctx.cclOutputAMem = algRes.cclOutputMem.range(0, cclOutputBufferSize);
134 0 : ctx.cclOutputBMem = algRes.cclOutputMem.range(cclOutputBufferSize, cclOutputBufferSize);
135 0 : ctx.curInputPtr = curInputPtr;
136 0 : ctx.curOutputPtr = curOutputPtr;
137 :
138 0 : HCCL_INFO(
139 : "[CollReduceScatterPipelineFor91093Executor][BuildPipelineLoopContext] "
140 : "tag[%s] numBlockTotal[%llu] numLoopTotal[%llu] maxCountDataPerLoop[%llu] "
141 : "targetCountDataPerLoop[%llu] countDataPerLoop[%llu] countDataLastLoop[%llu]",
142 : param.tag.c_str(), ctx.numBlockTotal, ctx.numBlockTotal + 1, maxCountDataPerLoop, targetCountDataPerLoop,
143 : ctx.countDataPerLoop, ctx.countDataLastLoop);
144 0 : return HCCL_SUCCESS;
145 : }
146 :
147 : // 由 RunLoop 调用
148 0 : HcclResult CollReduceScatterPipelineFor91093Executor::WaitForRemainingL2Signals(
149 : const OpParam& param, u64 numBlockTotal, Stream& streamL0L1, const std::shared_ptr<LocalNotify>& notifyL2toL0L1A,
150 : const std::shared_ptr<LocalNotify>& notifyL2toL0L1B)
151 : {
152 0 : const u64 remainingSignals
153 : = (numBlockTotal >= PIPELINE_PINGPONG_NOTIFY_PAIRS) ? PIPELINE_PINGPONG_NOTIFY_PAIRS : numBlockTotal;
154 0 : const u64 firstBlockIdx = numBlockTotal - remainingSignals;
155 0 : for (u64 blockIdx = firstBlockIdx; blockIdx < numBlockTotal; ++blockIdx) {
156 0 : auto notify = (blockIdx % PIPELINE_PINGPONG_NOTIFY_PAIRS == 0) ? notifyL2toL0L1A : notifyL2toL0L1B;
157 0 : HcclResult ret = LocalNotify::Wait(streamL0L1, dispatcher_, notify);
158 0 : CHK_PRT_RET(
159 : ret != HCCL_SUCCESS,
160 : HCCL_ERROR(
161 : "[CollReduceScatterPipelineFor91093Executor][WaitForRemainingL2Signals] "
162 : "PostSync wait error, tag[%s] blockIdx[%llu]",
163 : param.tag.c_str(), blockIdx),
164 : ret);
165 0 : }
166 0 : return HCCL_SUCCESS;
167 : }
168 :
169 : // 由 RunLoop 循环体调用
170 0 : HcclResult CollReduceScatterPipelineFor91093Executor::RunL0L1Phase(
171 : OpParam& param, const PipelineLoopContext& ctx, u64 blockIdx, Stream& streamL0L1)
172 : {
173 0 : HCCL_CONFIG_INFO(
174 : HCCL_ALG, "[CollReduceScatterPipelineFor91093Executor][RunL0L1Phase] blockIdx[%llu] useBufferA[%d]", blockIdx,
175 : (blockIdx % PIPELINE_BUFFER_NUM == 0));
176 :
177 0 : const bool useBufferA = (blockIdx % PIPELINE_BUFFER_NUM == 0);
178 0 : const bool isLastBlock = (blockIdx == ctx.numBlockTotal - 1);
179 0 : ExecMem execMem;
180 0 : execMem.count = isLastBlock ? ctx.countDataLastLoop : ctx.countDataPerLoop;
181 0 : execMem.inputMem = useBufferA ? ctx.cclInputAMem : ctx.cclInputBMem;
182 0 : execMem.outputMem = useBufferA ? ctx.cclOutputAMem : ctx.cclOutputBMem;
183 0 : execMem.scratchMem = execMem.outputMem;
184 0 : execMem.inputPtr = ctx.curInputPtr + blockIdx * ctx.sizeDataPerLoop;
185 0 : execMem.outputPtr = ctx.curOutputPtr + blockIdx * ctx.sizeDataPerLoop;
186 :
187 0 : const u64 bufferBaseOffset = useBufferA ? 0 : ctx.cclInputBufferSize;
188 0 : SliceExecMem(param, execMem);
189 :
190 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollReduceScatterPipelineFor91093Executor][RunL0L1Phase] chunk starts");
191 :
192 0 : HcclResult ret = KernelRunLevel0To1(param, execMem, streamL0L1, bufferBaseOffset);
193 0 : CHK_PRT_RET(
194 : ret != HCCL_SUCCESS,
195 : HCCL_ERROR(
196 : "[CollReduceScatterPipelineFor91093Executor][RunL0L1Phase] kernel run error, tag[%s]", param.tag.c_str()),
197 : ret);
198 0 : return HCCL_SUCCESS;
199 0 : }
200 :
201 : // 由 RunLoop 循环体调用
202 0 : HcclResult CollReduceScatterPipelineFor91093Executor::RunL2Phase(
203 : OpParam& param, const PipelineLoopContext& ctx, u64 blockIdx, Stream& streamL2)
204 : {
205 0 : HCCL_CONFIG_INFO(
206 : HCCL_ALG, "[CollReduceScatterPipelineFor91093Executor][RunL2Phase] blockIdx[%llu] L2 phase", blockIdx);
207 :
208 0 : const bool useBufferA = (blockIdx % PIPELINE_BUFFER_NUM == 0);
209 0 : const bool isLastBlock = (blockIdx == ctx.numBlockTotal - 1);
210 0 : ExecMem execMem;
211 0 : execMem.count = isLastBlock ? ctx.countDataLastLoop : ctx.countDataPerLoop;
212 0 : execMem.inputMem = useBufferA ? ctx.cclInputAMem : ctx.cclInputBMem;
213 0 : execMem.outputMem = useBufferA ? ctx.cclOutputAMem : ctx.cclOutputBMem;
214 0 : execMem.scratchMem = execMem.outputMem;
215 0 : execMem.inputPtr = ctx.curInputPtr + blockIdx * ctx.sizeDataPerLoop;
216 0 : execMem.outputPtr = ctx.curOutputPtr + blockIdx * ctx.sizeDataPerLoop;
217 :
218 0 : const u64 l2BaseOffset = useBufferA ? 0 : ctx.cclInputBufferSize;
219 0 : SliceExecMem(param, execMem);
220 :
221 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollReduceScatterPipelineFor91093Executor][RunL2Phase] chunk starts");
222 :
223 0 : HcclResult ret = KernelRunLevel2(param, execMem, streamL2, l2BaseOffset);
224 0 : CHK_PRT_RET(
225 : ret != HCCL_SUCCESS,
226 : HCCL_ERROR(
227 : "[CollReduceScatterPipelineFor91093Executor][RunL2Phase] kernel run error, tag[%s]", param.tag.c_str()),
228 : ret);
229 0 : return HCCL_SUCCESS;
230 0 : }
231 :
232 : // 由 RunL0L1Phase、RunL2Phase 调用
233 0 : void CollReduceScatterPipelineFor91093Executor::SliceExecMem(const OpParam& param, ExecMem& execMem)
234 : {
235 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
236 0 : u64 curSize = execMem.count * unitSize;
237 0 : u32 sliceNum = topoAttr_.userRankSize;
238 0 : execMem.inputMem = execMem.inputMem.range(0, curSize * sliceNum);
239 0 : execMem.outputMem = execMem.outputMem.range(0, curSize);
240 0 : }
241 :
242 : // 由 KernelRunLevel0To1、KernelRunLevel2 调用
243 : // Pipeline 约束 !isAHCAlgo，AHC 分支不可达，直接走 COMM_LEVEL2。
244 0 : HcclResult CollReduceScatterPipelineFor91093Executor::GetLevel2CommInfo(SubCommInfo& level2CommInfo)
245 : {
246 0 : CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1));
247 0 : level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
248 0 : return HCCL_SUCCESS;
249 : }
250 :
251 0 : u32 CollReduceScatterPipelineFor91093Executor::GetLevel0RingNum() const
252 : {
253 : // slaveStreams 不包含主流，先 +1 补回主流，再减去 2 条 Pipeline 专用流，得到基类 ring 数。
254 0 : return algResResp_->slaveStreams.size() + 1 - PIPELINE_EXTRA_STREAM_NUM;
255 : }
256 :
257 0 : HcclResult CollReduceScatterPipelineFor91093Executor::RunIntraSeverReduceScatter(
258 : const std::string& tag, DeviceMem& inputMem, DeviceMem& outputMem, const u64 count, const HcclDataType& dataType,
259 : const HcclReduceOp& reductionOp, const std::vector<std::vector<Slice>>& multRingsSliceZero, const Stream& stream,
260 : s32 profStage, const u64 baseOffset, const HcomCollOpInfo* opInfo,
261 : const std::vector<std::vector<Slice>>& multRingsUserMemSlice, const bool disableDMAReduce)
262 : {
263 : // SemiRing（IsUnifiedMarch）分支不可达：Pipeline 约束 workflowMode_==OP_BASE 排除图模式，
264 : // superPodNum>1 排除单 server，IsUnifiedMarch 恒为 false。
265 0 : HcclResult ret = HCCL_SUCCESS;
266 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
267 0 : ret = DoubleRingReduceScatter(
268 : tag, inputMem, outputMem, count, dataType, reductionOp, multRingsSliceZero, stream, profStage, baseOffset,
269 : opInfo, multRingsUserMemSlice, disableDMAReduce);
270 : } else {
271 0 : ret = CollReduceScatterRingFor91093Executor::RunIntraSeverReduceScatter(
272 : tag, inputMem, outputMem, count, dataType, reductionOp, multRingsSliceZero, stream, profStage, baseOffset,
273 : opInfo, multRingsUserMemSlice, disableDMAReduce);
274 : }
275 :
276 0 : CHK_RET(ret);
277 0 : return HCCL_SUCCESS;
278 : }
279 :
280 : // 逻辑与 CollAlignedReduceScatterDoubleRingFor91093Executor::DoubleRingReduceScatter 一致
281 0 : HcclResult CollReduceScatterPipelineFor91093Executor::DoubleRingReduceScatter(
282 : const std::string& tag, DeviceMem inputMem, DeviceMem outputMem, const u64 count, const HcclDataType dataType,
283 : const HcclReduceOp reductionOp, const std::vector<std::vector<Slice>> multRingsSliceZero, Stream stream,
284 : s32 profStage, const u64 baseOffset, const HcomCollOpInfo* opInfo,
285 : const std::vector<std::vector<Slice>> multRingsUserMemSlice, const bool disableDMAReduce)
286 : {
287 : (void)tag;
288 0 : HCCL_CONFIG_INFO(
289 : HCCL_ALG,
290 : "[CollReduceScatterPipelineFor91093Executor][DoubleRingReduceScatter] DoubleRingReduceScatter starts");
291 0 : u32 ringNum = multRingsSliceZero.size();
292 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, ringNum));
293 :
294 0 : u64 reduceAttr = GetReduceAttr(inputMem, outputMem, dataType, reductionOp);
295 0 : SubCommInfo level0RingCommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
296 :
297 0 : std::vector<std::vector<Slice>> userMemInputSlicesOfDoubleRing;
298 0 : std::vector<std::vector<u32>> rankOrders;
299 0 : CHK_RET(PrepareDoubleRingSlices(
300 : ringNum, dataType, opInfo, multRingsSliceZero, multRingsUserMemSlice, userMemInputSlicesOfDoubleRing,
301 : rankOrders));
302 :
303 : std::unique_ptr<AlgTemplateBase> tempAlg
304 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_DB_RING, dispatcher_);
305 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_DB_RING in COMM_LEVEL0", __func__);
306 0 : CHK_SMART_PTR_NULL(tempAlg);
307 : // 排除尾部 Pipeline 专用资源流（NotifyReserve + StreamL2），使模板看到与基类一致的流数
308 : std::vector<Stream> baseSlaveStreams(
309 0 : algResResp_->slaveStreams.begin(), algResResp_->slaveStreams.end() - PIPELINE_EXTRA_STREAM_NUM);
310 : std::vector<std::shared_ptr<LocalNotify>> baseNotifiesMain(
311 0 : algResResp_->notifiesMain.begin(), algResResp_->notifiesMain.end() - PIPELINE_PINGPONG_NOTIFY_PAIRS);
312 : std::vector<std::shared_ptr<LocalNotify>> baseNotifiesAux(
313 0 : algResResp_->notifiesAux.begin(), algResResp_->notifiesAux.end() - PIPELINE_PINGPONG_NOTIFY_PAIRS);
314 0 : HcclResult ret = tempAlg->Prepare(
315 : inputMem, inputMem, outputMem, count, dataType, stream, multRingsSliceZero, reductionOp, LEVEL0_BRIDGE_RANK_ID,
316 0 : baseOffset, disableDMAReduce, reduceAttr, opInfo, topoAttr_.userRank, baseSlaveStreams, baseNotifiesMain,
317 : baseNotifiesAux, rankOrders, userMemInputSlicesOfDoubleRing);
318 0 : CHK_PRT_RET(
319 : ret != HCCL_SUCCESS,
320 : HCCL_ERROR(
321 : "[CollReduceScatterPipelineFor91093Executor][DoubleRingReduceScatter] "
322 : "Double ring ReduceScatter failed,return[%d]",
323 : ret),
324 : ret);
325 :
326 0 : u32 ringIndexOp = COMM_INDEX_0;
327 0 : u32 rankSize = level0RingCommInfo.localRankSize;
328 0 : ret = tempAlg->RegisterProfiler(
329 0 : ((ringIndexOp + 1) << PROF_RINGINDEX_OFFSET_OF_PLANEID) + (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID)
330 0 : + level0RingCommInfo.localRank,
331 : profStage, HCCL_EXEC_STEP_NOT_SET, stream);
332 0 : CHK_PRT_RET(
333 : ret != HCCL_SUCCESS,
334 : HCCL_ERROR(
335 : "[CollReduceScatterPipelineFor91093Executor][DoubleRingReduceScatter] "
336 : "Double ring ReduceScatter RegisterProfiler failed,return[%d]",
337 : ret),
338 : ret);
339 :
340 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem, outputMem, stream, dispatcher_));
341 0 : ret = RunTemplate(tempAlg, level0RingCommInfo);
342 0 : CHK_PRT_RET(
343 : ret != HCCL_SUCCESS,
344 : HCCL_ERROR(
345 : "[CollReduceScatterPipelineFor91093Executor][DoubleRingReduceScatter] "
346 : "Double ring ReduceScatter RunTemplate failed,return[%d]",
347 : ret),
348 : ret);
349 :
350 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem, outputMem, stream, dispatcher_));
351 0 : return HCCL_SUCCESS;
352 0 : }
353 :
354 : // 由 DoubleRingReduceScatter 调用
355 0 : HcclResult CollReduceScatterPipelineFor91093Executor::PrepareDoubleRingSlices(
356 : u32 ringNum, const HcclDataType dataType, const HcomCollOpInfo* opInfo,
357 : const std::vector<std::vector<Slice>>& multRingsSliceZero,
358 : const std::vector<std::vector<Slice>>& multRingsUserMemSlice,
359 : std::vector<std::vector<Slice>>& userMemInputSlicesOfDoubleRing, std::vector<std::vector<u32>>& rankOrders)
360 : {
361 0 : SubCommInfo level0ZeroCommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
362 0 : auto nicList = topoAttr_.nicList;
363 : std::vector<std::vector<u32>> multiRingsOrder
364 0 : = GetRingsOrderByTopoType(level0ZeroCommInfo.localRankSize, topoType_, nicList);
365 0 : CHK_RET(CollectMultiRingsUserMemSlices(
366 : ringNum, dataType, opInfo, multRingsSliceZero, multiRingsOrder, multRingsUserMemSlice,
367 : userMemInputSlicesOfDoubleRing));
368 0 : CHK_RET(CollectMultiRingsRankOrder(ringNum, multiRingsOrder, rankOrders));
369 0 : return HCCL_SUCCESS;
370 0 : }
371 :
372 : // 拆分自 CollReduceScatterRingFor91093Executor::KernelRun 的 L0+L1 部分
373 0 : HcclResult CollReduceScatterPipelineFor91093Executor::KernelRunLevel0To1(
374 : const OpParam& param, ExecMem& execMem, Stream& streamL0L1, const u64 baseOffset)
375 : {
376 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] executor starts", __func__);
377 0 : u32 perDataSize = 0;
378 0 : const HcclDataType dataType = param.GetDataType();
379 0 : CHK_RET(SalGetDataTypeSize(dataType, perDataSize));
380 :
381 : u32 ringNum;
382 0 : u32 sliceNum = logicalLevel0CommInfo_.localRankSize;
383 0 : u32 commIndex = logicalLevel0CommInfo_.localRank;
384 :
385 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
386 0 : ringNum = LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE;
387 : } else {
388 0 : ringNum = LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
389 : }
390 :
391 0 : SubCommInfo level2CommInfo;
392 0 : CHK_RET(GetLevel2CommInfo(level2CommInfo));
393 0 : const u32 level2RankSize = level2CommInfo.localRankSize;
394 0 : const u32 level1RankSize = logicalLevel1CommInfo_.localRankSize;
395 :
396 0 : std::vector<std::vector<Slice>> multiStreamSlice;
397 0 : std::vector<std::vector<Slice>> level0DataSegsSlice;
398 0 : CHK_RET(CalLevel0DataSegsSlice(
399 : execMem, multiStreamSlice, param, ringNum, sliceNum, level1RankSize, level2RankSize, dataType,
400 : level0DataSegsSlice));
401 :
402 0 : HcomCollOpInfo opInfo = GetHcomCollOpInfo(param, execMem);
403 0 : HcomCollOpInfo* opInfoPtr = &opInfo;
404 :
405 0 : bool disableDMAReduce = algOpContext_.opRetryHandler.retryEnable
406 0 : && (algOpContext_.opRetryHandler.inPlaceSupportRetryStatus
407 : == InplaceSupportRetryStatus::RETRY_1_ALLOW_NO_DMA_REDUCE_CASE1
408 0 : || algOpContext_.opRetryHandler.inPlaceSupportRetryStatus
409 : == InplaceSupportRetryStatus::RETRY_1_ALLOW_NO_DMA_REDUCE_CASE2);
410 0 : std::vector<std::vector<Slice>> multRingsUserMemSlice;
411 0 : CHK_RET(CalUserMemDataSegsSlice(
412 : execMem, level0DataSegsSlice, multiStreamSlice, param, ringNum, sliceNum, level1RankSize, level2RankSize,
413 : dataType, perDataSize, opInfoPtr, disableDMAReduce, multRingsUserMemSlice));
414 :
415 0 : HcomCollOpInfo opInfoByReduceScatterDMAreduce = *opInfoPtr;
416 0 : opInfoByReduceScatterDMAreduce.outputAddr = nullptr;
417 0 : CHK_RET(RunIntraSeverReduceScatter(
418 : param.tag, execMem.inputMem, execMem.scratchMem, execMem.count, dataType, param.reduceType, level0DataSegsSlice,
419 : streamL0L1, PROF_STAGE_1, baseOffset, &opInfoByReduceScatterDMAreduce, multRingsUserMemSlice,
420 : disableDMAReduce));
421 :
422 0 : if (level1RankSize > 1) {
423 0 : CHK_RET(RunLevel1Template(
424 : param, execMem, streamL0L1, baseOffset, commIndex, sliceNum, level1RankSize, level2RankSize, perDataSize));
425 : }
426 :
427 0 : return HCCL_SUCCESS;
428 0 : }
429 :
430 : // 由 KernelRunLevel0To1 调用
431 0 : HcclResult CollReduceScatterPipelineFor91093Executor::RunLevel1Template(
432 : const OpParam& param, ExecMem& execMem, Stream& streamL0L1, u64 baseOffset, u32 commIndex, u32 sliceNum,
433 : u32 level1RankSize, u32 level2RankSize, u32 perDataSize)
434 : {
435 0 : const HcclDataType dataType = param.GetDataType();
436 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.scratchMem, dataType, param.reduceType);
437 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
438 :
439 0 : std::vector<Slice> level1DataSegsSlice;
440 0 : CHK_RET(CalLevel1DataSegsSlice(
441 : execMem, param, logicalLevel1plane_, commIndex, sliceNum, level1RankSize, level2RankSize, perDataSize,
442 : level1DataSegsSlice));
443 :
444 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
445 : level1TempAlg
446 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
447 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RING in COMM_LEVEL1", __func__);
448 0 : CHK_SMART_PTR_NULL(level1TempAlg);
449 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
450 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
451 : level1TempAlg
452 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NB, dispatcher_);
453 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NB in COMM_LEVEL1", __func__);
454 0 : CHK_SMART_PTR_NULL(level1TempAlg);
455 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
456 : } else {
457 : level1TempAlg
458 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
459 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR in COMM_LEVEL1", __func__);
460 0 : CHK_SMART_PTR_NULL(level1TempAlg);
461 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr, false));
462 : }
463 :
464 0 : CHK_RET(level1TempAlg->Prepare(
465 : execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count, dataType, streamL0L1, param.reduceType,
466 : LEVEL0_BRIDGE_RANK_ID, level1DataSegsSlice, baseOffset));
467 0 : CHK_RET(level1TempAlg->RegisterProfiler(
468 : (level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + logicalLevel1CommInfo_.localRank, PROF_STAGE_2,
469 : HCCL_EXEC_STEP_NOT_SET, streamL0L1));
470 0 : CHK_RET(RunTemplate(level1TempAlg, logicalLevel1CommInfo_));
471 0 : return HCCL_SUCCESS;
472 0 : }
473 :
474 : // 拆分自 CollReduceScatterRingFor91093Executor::KernelRun 的 L2+copyOut 部分，stream 替换为 streamL2
475 0 : HcclResult CollReduceScatterPipelineFor91093Executor::KernelRunLevel2(
476 : const OpParam& param, ExecMem& execMem, Stream& streamL2, const u64 baseOffset)
477 : {
478 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] executor starts", __func__);
479 0 : u32 perDataSize = 0;
480 0 : const HcclDataType dataType = param.GetDataType();
481 0 : CHK_RET(SalGetDataTypeSize(dataType, perDataSize));
482 :
483 0 : SubCommInfo level2CommInfo;
484 0 : CHK_RET(GetLevel2CommInfo(level2CommInfo));
485 0 : const u32 level2RankSize = level2CommInfo.localRankSize;
486 :
487 0 : CHK_RET(RunLevel2Template(param, execMem, streamL2, baseOffset, level2CommInfo, level2RankSize, perDataSize));
488 :
489 0 : HcomCollOpInfo opInfo = GetHcomCollOpInfo(param, execMem);
490 0 : HcomCollOpInfo* opInfoPtr = &opInfo;
491 :
492 0 : const u64 offset = CalcSrcMemOffset(execMem, param, perDataSize);
493 0 : DeviceMem srcMem = execMem.inputMem.range(offset, execMem.outputMem.size());
494 0 : if (opInfoPtr != nullptr) {
495 0 : DeviceMem dstMem = DeviceMem::create(static_cast<u8*>(opInfoPtr->outputAddr), execMem.outputMem.size());
496 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, streamL2));
497 0 : } else {
498 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, execMem.outputMem, srcMem, streamL2));
499 : }
500 :
501 0 : return HCCL_SUCCESS;
502 0 : }
503 :
504 : // 由 KernelRunLevel2 调用
505 0 : HcclResult CollReduceScatterPipelineFor91093Executor::RunLevel2Template(
506 : const OpParam& param, ExecMem& execMem, Stream& streamL2, u64 baseOffset, const SubCommInfo& level2CommInfo,
507 : u32 level2RankSize, u32 perDataSize)
508 : {
509 0 : const HcclDataType dataType = param.GetDataType();
510 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.scratchMem, dataType, param.reduceType);
511 :
512 0 : std::vector<Slice> level2DataSegsSlice;
513 0 : CHK_RET(CalLevel2DataSegsSlice(execMem, param, level2RankSize, perDataSize, level2DataSegsSlice));
514 :
515 0 : std::unique_ptr<AlgTemplateBase> level2TempAlg;
516 0 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
517 : level2TempAlg
518 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NB, dispatcher_);
519 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NB in COMM_LEVEL2", __func__);
520 0 : CHK_SMART_PTR_NULL(level2TempAlg);
521 0 : CHK_RET(level2TempAlg->Prepare(reduceAttr));
522 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
523 : level2TempAlg
524 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
525 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR in COMM_LEVEL2", __func__);
526 0 : CHK_SMART_PTR_NULL(level2TempAlg);
527 0 : CHK_RET(level2TempAlg->Prepare(reduceAttr, false));
528 0 : if (algoAttr_.isSupportAtomicWrite) {
529 0 : level2TempAlg->CloseBarrier();
530 : }
531 : } else {
532 : level2TempAlg
533 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
534 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RING in COMM_LEVEL2", __func__);
535 0 : CHK_SMART_PTR_NULL(level2TempAlg);
536 0 : CHK_RET(level2TempAlg->Prepare(reduceAttr));
537 : }
538 :
539 0 : CHK_RET(level2TempAlg->Prepare(
540 : execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count, dataType, streamL2, param.reduceType,
541 : LEVEL0_BRIDGE_RANK_ID, level2DataSegsSlice, baseOffset));
542 0 : CHK_RET(level2TempAlg->RegisterProfiler(
543 : (level2RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level2CommInfo.localRank, PROF_STAGE_2,
544 : HCCL_EXEC_STEP_NOT_SET, streamL2));
545 0 : CHK_RET(RunTemplate(level2TempAlg, level2CommInfo));
546 0 : return HCCL_SUCCESS;
547 0 : }
548 :
549 : REGISTER_EXEC(
550 : "ReduceScatterPipelineFor91093Executor", ReduceScatterPipelineFor91093, CollReduceScatterPipelineFor91093Executor);
551 : } // namespace hccl
|
