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