Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "coll_reduce_scatter_mix_executor.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 0 : CollReduceScatterMixExecutor::CollReduceScatterMixExecutor(
16 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
17 0 : : CollReduceScatterExecutor(dispatcher, topoMatcher)
18 : {
19 0 : DMAReduceFlag_ = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
20 0 : && topoAttr_.deviceType == DevType::DEV_TYPE_910_93;
21 0 : }
22 :
23 0 : void CollReduceScatterMixExecutor::ParseParam(const OpParam& param)
24 : {
25 0 : tag_ = param.tag;
26 0 : aicpuUnfoldMode_ = param.aicpuUnfoldMode;
27 :
28 : // 是否需要scratch memory
29 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE && isSupportSDMAReduce_
30 0 : && IsSupportRDMAReduce(param.DataDes.dataType, param.reduceType)) {
31 0 : scratchMemFlag_ = false;
32 : } else {
33 0 : scratchMemFlag_ = true;
34 : }
35 :
36 : // 记录图模式总数据量
37 0 : totalSize_ = topoAttr_.userRankSize * param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
38 :
39 : // 910B 图模式非确定计算，inlineReduce使能，MESH拓扑场景下，创建一个mesh平面
40 : bool isInlineReduce
41 0 : = IsSupportSDMAReduce(param.inputPtr, param.outputPtr, param.DataDes.dataType, param.reduceType);
42 0 : meshSinglePlane_ = (topoAttr_.deviceType == DevType::DEV_TYPE_910B)
43 0 : && topoMatcher_->GetExternalInputHcclDeterministic() == DETERMINISTIC_DISABLE && isInlineReduce
44 0 : && (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE);
45 :
46 0 : bool isAlsoSupportDMAReduce = topoAttr_.deviceType == DevType::DEV_TYPE_910B && isInlineReduce
47 0 : && workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
48 0 : && topoMatcher_->GetExternalInputHcclDeterministic() == DETERMINISTIC_DISABLE
49 0 : && param.DataDes.dataType != HCCL_DATA_TYPE_INT64
50 0 : && param.reduceType != HCCL_REDUCE_PROD;
51 0 : if (isAlsoSupportDMAReduce) {
52 0 : DMAReduceFlag_ = true;
53 : }
54 0 : }
55 :
56 0 : HcclResult CollReduceScatterMixExecutor::CalcScratchMemSize(u64& scratchMemSize)
57 : {
58 0 : if (scratchMemFlag_) {
59 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
60 0 : scratchMemSize = inCCLbufferSize_;
61 : } else {
62 0 : scratchMemSize = totalSize_;
63 : }
64 : } else {
65 0 : scratchMemSize = 0U;
66 : }
67 0 : HCCL_INFO(
68 : "[CollReduceScatterMixExecutor][CalcScratchMemSize] tag[%s] scratchMemSize[%llu]", tag_.c_str(),
69 : scratchMemSize);
70 0 : return HCCL_SUCCESS;
71 : }
72 :
73 0 : HcclResult CollReduceScatterMixExecutor::CalcStreamNum(u32& streamNum)
74 : {
75 0 : u32 totalStreamNum = 0;
76 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) { // mesh
77 0 : totalStreamNum = topoAttr_.deviceNumPerAggregation;
78 0 : HCCL_DEBUG("[CollReduceScatterMixExecutor][CalcStreamNum]totalStreamNum is %u", totalStreamNum);
79 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) { // dbring
80 : totalStreamNum
81 0 : = (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING ? LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE :
82 : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE);
83 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
84 0 : totalStreamNum *= STREAM_NUM_FOR_DMAREDUCE_ONE_RING;
85 : }
86 : }
87 :
88 0 : streamNum = totalStreamNum - 1U;
89 0 : HCCL_INFO("[CollReduceScatterMixExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
90 0 : return HCCL_SUCCESS;
91 : }
92 :
93 0 : bool CollReduceScatterMixExecutor::IsHugeData(const u64 curSize, OpParam* param)
94 : {
95 0 : const u64 TBE_REDUCE_MAX_COUNT = INT32_MAX;
96 :
97 0 : u64 curCount = curSize / SIZE_TABLE[param->DataDes.dataType];
98 0 : bool issupportRDMAInlineReduce = IsSupportRDMAReduce(param->DataDes.dataType, param->reduceType);
99 : // 这里如果CheckCommSize返回ERROR，相当于HugeData true，防止GetSubCommInfo越界
100 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
101 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
102 0 : u32 level0RankSize = level0CommInfo.localRankSize;
103 :
104 0 : bool hugeData = (curSize * level0RankSize / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE)
105 0 : || (curSize > SDMA_SEND_MAX_SIZE) || ((!isSupportSDMAReduce_) && (curCount > TBE_REDUCE_MAX_COUNT))
106 0 : || ((!issupportRDMAInlineReduce)
107 0 : && (curCount * level0RankSize / HCCL_INTERNODE_MAX_DATA_RATE > TBE_REDUCE_MAX_COUNT));
108 :
109 0 : return hugeData;
110 0 : }
111 :
112 0 : bool CollReduceScatterMixExecutor::IsSmallData(const u64 totalSize, [[maybe_unused]] const u64 curSize)
113 : {
114 0 : bool smallData = totalSize <= HCCL_SMALL_COUNT_32_KB;
115 0 : return smallData;
116 : }
117 :
118 0 : HcclResult CollReduceScatterMixExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
119 : {
120 0 : TransportMemType inputType = TransportMemType::RESERVED;
121 0 : TransportMemType outputType = TransportMemType::RESERVED;
122 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
123 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
124 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
125 :
126 : // mix在server间使用NHR通信域，并在多机A+X场景下当未设置使用RDMA时，默认使用RDMA
127 0 : std::vector<SingleSubCommTransport>& commTransportLevel1 = opTransport[COMM_LEVEL1];
128 0 : for (u32 ringIndex = 0; ringIndex < commTransportLevel1.size(); ringIndex++) {
129 0 : for (auto& transportRequest : commTransportLevel1[ringIndex].transportRequests) {
130 0 : transportRequest.isUsedRdma = true;
131 : }
132 : }
133 0 : return HCCL_SUCCESS;
134 : }
135 :
136 0 : HcclResult CollReduceScatterMixExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
137 : {
138 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
139 0 : inputType = TransportMemType::CCL_INPUT;
140 0 : if (scratchMemFlag_) {
141 0 : outputType = TransportMemType::SCRATCH;
142 : } else {
143 0 : outputType = TransportMemType::CCL_OUTPUT;
144 : }
145 : } else {
146 0 : inputType = TransportMemType::PARAM_INPUT;
147 0 : if (scratchMemFlag_) {
148 0 : outputType = TransportMemType::SCRATCH;
149 : } else {
150 0 : outputType = TransportMemType::PARAM_OUTPUT;
151 : }
152 : }
153 0 : HCCL_INFO(
154 : "[CollReduceScatterMixExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]", tag_.c_str(),
155 : inputType, outputType);
156 0 : return HCCL_SUCCESS;
157 : }
158 :
159 0 : HcclResult CollReduceScatterMixExecutor::CalcLevel0CommInfo(
160 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
161 : {
162 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
163 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
164 0 : commParaLevel0.meshSinglePlane = meshSinglePlane_;
165 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
166 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
167 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
168 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
169 0 : }
170 :
171 0 : return HCCL_SUCCESS;
172 : }
173 :
174 0 : u64 CollReduceScatterMixExecutor::CalcLoopMaxCount(const u32 unitSize)
175 : {
176 0 : u64 maxCountPerLoop = inCCLbufferSize_ / (topoAttr_.userRankSize * unitSize);
177 0 : return maxCountPerLoop;
178 : }
179 :
180 0 : void CollReduceScatterMixExecutor::CalLevel0DataSegsSlice(
181 : const ExecMem& execMem, const std::vector<std::vector<Slice>>& multiStreamSlice, u32 sliceNum, u32 level1RankSize,
182 : std::vector<std::vector<Slice>>& level0DataSegsSlice)
183 : {
184 0 : for (u32 ringIndex = 0; ringIndex < multiStreamSlice.size(); ringIndex++) {
185 0 : std::vector<Slice> dataSlice;
186 0 : FillMultiRingSlice(execMem, multiStreamSlice, sliceNum, level1RankSize, ringIndex, dataSlice);
187 0 : level0DataSegsSlice.push_back(dataSlice);
188 0 : }
189 0 : }
190 :
191 0 : void CollReduceScatterMixExecutor::FillMultiRingSlice(
192 : const ExecMem& execMem, const std::vector<std::vector<Slice>>& multiStreamSlice, u32 sliceNum, u32 level1RankSize,
193 : const u32 ringIndex, std::vector<Slice>& dataSlice)
194 : {
195 0 : for (u32 level0Idx = 0; level0Idx < sliceNum; level0Idx++) {
196 0 : Slice sliceTemp;
197 0 : for (u32 level1Idx = 0; level1Idx < level1RankSize; level1Idx++) {
198 0 : sliceTemp.size = multiStreamSlice[ringIndex][level0Idx].size;
199 : sliceTemp.offset
200 0 : = multiStreamSlice[ringIndex][level0Idx].offset + level1Idx * sliceNum * execMem.outputMem.size();
201 0 : dataSlice.push_back(sliceTemp);
202 0 : HCCL_DEBUG(
203 : "rank[%u] sliceTemp.size[%zu], sliceTemp.offset[%llu]", topoAttr_.userRank, sliceTemp.size,
204 : sliceTemp.offset);
205 : }
206 : }
207 0 : }
208 :
209 0 : HcclResult CollReduceScatterMixExecutor::CalLevel1DataSegsSlice(
210 : const ExecMem& execMem, const u32& commIndex, u32 sliceNum, u32 level1RankSize,
211 : std::vector<Slice>& level1DataSegsSlice)
212 : {
213 : (void)sliceNum;
214 0 : for (u32 i = 0; i < level1RankSize; i++) {
215 0 : Slice sliceTemp;
216 : u32 level1UserRank;
217 0 : CHK_RET(GetUserRankByRank(COMM_LEVEL1, commIndex, i, level1UserRank));
218 0 : sliceTemp.size = execMem.outputMem.size();
219 0 : sliceTemp.offset = level1UserRank * execMem.outputMem.size();
220 0 : level1DataSegsSlice.push_back(sliceTemp);
221 0 : HCCL_DEBUG(
222 : "rank[%u], level1DataSegsSlice[%u].offset=%llu, size=[%llu]", topoAttr_.userRank, i, sliceTemp.offset,
223 : sliceTemp.size);
224 : }
225 0 : return HCCL_SUCCESS;
226 : }
227 :
228 0 : HcclResult CollReduceScatterMixExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
229 : {
230 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollReduceScatterMixExecutor][KernelRun] The ReduceScatterMixExecutor starts.");
231 0 : u32 perDataSize = 0;
232 0 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, perDataSize));
233 :
234 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
235 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
236 0 : u32 level0RankSize = level0CommInfo.localRankSize;
237 0 : u32 commIndex = level0CommInfo.localRank; // 找到rank所在的节点间平面
238 :
239 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
240 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
241 0 : u32 serverIndex = level1CommInfo.localRank;
242 0 : u32 level1RankSize = level1CommInfo.localRankSize;
243 0 : HCCL_DEBUG(
244 : "ReduceScatterMixExecutor inputSize=%llu, level0RankSize=%u,commIndex=%u, level1RankSize=%u, serverIndex=%u",
245 : execMem.inputMem.size(), level0RankSize, commIndex, level1RankSize, serverIndex);
246 :
247 0 : HcomCollOpInfo opInfo = {"",
248 0 : execMem.inputPtr,
249 0 : execMem.outputPtr,
250 0 : param.DataDes.count,
251 0 : param.DataDes.dataType,
252 0 : param.root,
253 0 : param.reduceType,
254 0 : 0};
255 0 : HCCL_DEBUG(
256 : "[CollReduceScatterMixExecutor][KernelRun] execMem.inputPtr[%p], execMem.outputPtr[%p], "
257 : "execMem.inputMem[%p], execMem.outputMem[%p]",
258 : execMem.inputPtr, execMem.outputPtr, execMem.inputMem.ptr(), execMem.outputMem.ptr());
259 0 : HcomCollOpInfo* opInfoPtr = nullptr;
260 0 : if (DMAReduceFlag_) {
261 0 : opInfoPtr = &opInfo;
262 : }
263 :
264 : // 第一步，AI server内reduce scatter mesh/dbring
265 0 : u32 sliceNum = level0CommInfo.localRankSize;
266 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
267 : u32 ringNum;
268 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
269 0 : ringNum = LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE;
270 : } else {
271 0 : ringNum = LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
272 : }
273 :
274 0 : Slice sliceTemp;
275 :
276 0 : std::vector<Slice> dataSegsSlice; // 数据分成ranksize份，每份的起始偏移和大小
277 0 : std::vector<std::vector<Slice>> multiStreamSlice; // 每个stream使用的数据基于用户buffer的偏移
278 :
279 0 : CHK_RET(ActiveSlaveStreams(param.stream));
280 :
281 : // 计算slice
282 0 : std::vector<std::vector<Slice>> level0DataSegsSlice;
283 0 : bool useInlineRduce = false;
284 0 : bool isInlineReduce = IsSupportSDMAReduce(
285 0 : execMem.inputMem.ptr(), execMem.scratchMem.ptr(), param.DataDes.dataType, param.reduceType);
286 0 : useInlineRduce = isInlineReduce && algoAttr_.inlineReduceSwitchOn;
287 0 : multiStreamSlice = ReduceScatterRingSlicePrepare(
288 0 : ringNum, sliceNum, useInlineRduce, execMem.outputMem, dataSegsSlice,
289 0 : param.tag); // 2个ring，每条ring上数据的偏移和大小
290 :
291 0 : CalLevel0DataSegsSlice(execMem, multiStreamSlice, sliceNum, level1RankSize, level0DataSegsSlice);
292 :
293 0 : std::vector<std::vector<Slice>> multRingsUserMemSlice;
294 :
295 0 : if (opInfoPtr == nullptr
296 0 : && (!(
297 0 : topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING
298 0 : && workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB))) {
299 0 : multRingsUserMemSlice = level0DataSegsSlice;
300 : } else {
301 0 : for (u32 ringIndex = 0; ringIndex < level0DataSegsSlice.size(); ringIndex++) {
302 0 : std::vector<Slice> level1UserMemSlice;
303 0 : for (auto& cclSlice : level0DataSegsSlice[ringIndex]) {
304 0 : Slice tmpSlice;
305 0 : tmpSlice.size = cclSlice.size;
306 0 : tmpSlice.offset = (cclSlice.offset / execMem.outputMem.size()) * param.DataDes.count * perDataSize
307 0 : + multiStreamSlice[ringIndex][0].offset;
308 0 : level1UserMemSlice.push_back(tmpSlice);
309 0 : HCCL_DEBUG(
310 : "rank[%u], ringIndex[%u], tmpSlice.offset=[%llu], size=[%llu]", topoAttr_.userRank, ringIndex,
311 : tmpSlice.offset, tmpSlice.size);
312 : }
313 0 : multRingsUserMemSlice.push_back(level1UserMemSlice);
314 0 : }
315 : }
316 : // 区分消减拷贝场景
317 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING
318 0 : && workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) {
319 : // 图模式opinfo不为空
320 0 : HcomCollOpInfo graphModeOpInfo
321 0 : = {"", execMem.inputMem.ptr(), nullptr, param.DataDes.count, param.DataDes.dataType,
322 0 : param.root, param.reduceType, 0};
323 0 : CHK_RET(MultiRingReduceScatter(
324 : param.tag, execMem.inputMem, execMem.scratchMem, execMem.count, param.DataDes.dataType,
325 : param.reduceType, level0DataSegsSlice, param.stream, PROF_STAGE_0, 0, &graphModeOpInfo,
326 : multRingsUserMemSlice));
327 0 : } else if (opInfoPtr != nullptr) {
328 0 : HcomCollOpInfo opInfoByReduceScatterDMAreduce = *opInfoPtr;
329 0 : opInfoByReduceScatterDMAreduce.outputAddr = nullptr;
330 0 : CHK_RET(MultiRingReduceScatter(
331 : param.tag, execMem.inputMem, execMem.scratchMem, execMem.count, param.DataDes.dataType,
332 : param.reduceType, level0DataSegsSlice, param.stream, PROF_STAGE_0, 0, &opInfoByReduceScatterDMAreduce,
333 : multRingsUserMemSlice));
334 : } else {
335 0 : CHK_RET(MultiRingReduceScatter(
336 : param.tag, execMem.inputMem, execMem.scratchMem, execMem.count, param.DataDes.dataType,
337 : param.reduceType, level0DataSegsSlice, param.stream, PROF_STAGE_0, 0, opInfoPtr,
338 : multRingsUserMemSlice));
339 : }
340 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
341 0 : CHK_RET(ActiveSlaveStreams(param.stream));
342 :
343 : // 根据数据量算每个环上数据的偏移和大小，把做完hd的slice均分成RankSize份
344 0 : std::vector<Slice> dataSegsSlice;
345 0 : CHK_RET(PrepareReduceScatterSliceData(execMem.count, perDataSize, sliceNum, dataSegsSlice));
346 :
347 0 : if (opInfoPtr != nullptr) {
348 : u64 reduceAttr
349 0 : = GetReduceAttr(execMem.inputMem, execMem.scratchMem, param.DataDes.dataType, param.reduceType);
350 0 : std::unique_ptr<AlgTemplateBase> level0Executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
351 0 : TemplateType::TEMPLATE_REDUCESCATTER_MESH_MIX, dispatcher_);
352 :
353 0 : CHK_SMART_PTR_NULL(level0Executor);
354 0 : CHK_RET(level0Executor->Prepare(
355 : execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count, param.DataDes.dataType,
356 : param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, dataSegsSlice, 0, reduceAttr,
357 : algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux, serverIndex,
358 : level1RankSize, opInfoPtr));
359 0 : CHK_RET(level0Executor->RegisterProfiler(
360 : (level0RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + commIndex, PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET,
361 : param.stream));
362 0 : CHK_RET(RunTemplate(level0Executor, level0CommInfo));
363 0 : } else {
364 0 : std::vector<std::vector<Slice>> multiStreamSlice; // 每个stream使用的数据基于用户buffer的偏移
365 : // mesh算法stream数量为rank数减1
366 0 : CHK_RET(AlgTemplateBase::PrepareSliceMeshStreams(dataSegsSlice, sliceNum - 1, multiStreamSlice));
367 :
368 : // 计算slice
369 0 : std::vector<std::vector<Slice>> level0DataSegsSlice;
370 0 : CalLevel0DataSegsSlice(execMem, multiStreamSlice, sliceNum, level1RankSize, level0DataSegsSlice);
371 :
372 0 : CHK_RET(MultiStreamReduceScatterMesh(
373 : param.tag, execMem.inputMem, execMem.scratchMem, execMem.count, param.DataDes.dataType,
374 : param.reduceType, level0DataSegsSlice, param.stream, COMM_LEVEL0, 0));
375 0 : }
376 0 : }
377 :
378 : // 第二步，节点间reduce scatter
379 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.scratchMem, param.DataDes.dataType, param.reduceType);
380 :
381 0 : std::unique_ptr<AlgTemplateBase> level1Executor;
382 :
383 : // 计算slice
384 0 : std::vector<Slice> level1DataSegsSlice;
385 :
386 0 : CHK_RET(CalLevel1DataSegsSlice(execMem, commIndex, sliceNum, level1RankSize, level1DataSegsSlice));
387 :
388 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
389 : level1Executor
390 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
391 0 : CHK_SMART_PTR_NULL(level1Executor);
392 0 : CHK_RET(level1Executor->Prepare(reduceAttr));
393 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RING in COMM_LEVEL1", __func__);
394 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
395 : level1Executor
396 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
397 0 : CHK_SMART_PTR_NULL(level1Executor);
398 0 : CHK_RET(level1Executor->Prepare(reduceAttr, false));
399 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR in COMM_LEVEL1", __func__);
400 : } else {
401 0 : HCCL_ERROR(
402 : "[CollReduceScatterMixExecutor][KernelRun]ReduceScatter mix: algType[%u] is not supported.",
403 : algType_.algoLevel1);
404 0 : return HCCL_E_NOT_SUPPORT;
405 : }
406 :
407 0 : CHK_RET(level1Executor->Prepare(
408 : execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count, param.DataDes.dataType, param.stream,
409 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, level1DataSegsSlice));
410 0 : CHK_RET(level1Executor->RegisterProfiler(
411 : (level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank, PROF_STAGE_1,
412 : HCCL_EXEC_STEP_NOT_SET, param.stream));
413 0 : CHK_RET(RunTemplate(level1Executor, level1CommInfo));
414 :
415 : // 区分消减拷贝场景（消减拷贝数据需要拷贝到user output上）
416 0 : DeviceMem srcMem = execMem.inputMem.range(topoAttr_.userRank * execMem.outputMem.size(), execMem.outputMem.size());
417 0 : if (opInfoPtr != nullptr) {
418 0 : DeviceMem dstMem = DeviceMem::create(static_cast<u8*>(opInfoPtr->outputAddr), execMem.outputMem.size());
419 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
420 0 : } else {
421 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, execMem.outputMem, srcMem, const_cast<Stream&>(param.stream)));
422 : }
423 :
424 0 : HCCL_INFO("[CollReduceScatterMixExecutor][KernelRun]ReduceScatter mix run success");
425 0 : return HCCL_SUCCESS;
426 0 : }
427 :
428 : REGISTER_EXEC("ReduceScatterMixExecutor", ReduceScatterMix, CollReduceScatterMixExecutor);
429 : } // namespace hccl
|
