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(const HcclDispatcher dispatcher,
16 0 : 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 &&
30 0 : isSupportSDMAReduce_ && 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 0 : bool isInlineReduce = IsSupportSDMAReduce(param.inputPtr, param.outputPtr, param.DataDes.dataType,
41 0 : param.reduceType);
42 0 : meshSinglePlane_ = (topoAttr_.deviceType == DevType::DEV_TYPE_910B) &&
43 0 : topoMatcher_->GetExternalInputHcclDeterministic() == DETERMINISTIC_DISABLE &&
44 0 : isInlineReduce && (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 && param.reduceType != HCCL_REDUCE_PROD;
50 0 : if (isAlsoSupportDMAReduce) {
51 0 : DMAReduceFlag_ = true;
52 : }
53 0 : }
54 :
55 0 : HcclResult CollReduceScatterMixExecutor::CalcScratchMemSize(u64& scratchMemSize)
56 : {
57 0 : if (scratchMemFlag_) {
58 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
59 0 : scratchMemSize = inCCLbufferSize_;
60 : } else {
61 0 : scratchMemSize = totalSize_;
62 : }
63 : } else {
64 0 : scratchMemSize = 0U;
65 : }
66 0 : HCCL_INFO("[CollReduceScatterMixExecutor][CalcScratchMemSize] tag[%s] scratchMemSize[%llu]",
67 : tag_.c_str(), scratchMemSize);
68 0 : return HCCL_SUCCESS;
69 : }
70 :
71 0 : HcclResult CollReduceScatterMixExecutor::CalcStreamNum(u32& streamNum)
72 : {
73 0 : u32 totalStreamNum = 0;
74 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) { // mesh
75 0 : totalStreamNum = topoAttr_.deviceNumPerAggregation;
76 0 : HCCL_DEBUG("[CollReduceScatterMixExecutor][CalcStreamNum]totalStreamNum is %u", totalStreamNum);
77 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) { // dbring
78 0 : totalStreamNum = (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING ? LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE :
79 : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE);
80 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
81 0 : totalStreamNum *= STREAM_NUM_FOR_DMAREDUCE_ONE_RING;
82 : }
83 : }
84 :
85 0 : streamNum = totalStreamNum - 1U;
86 0 : HCCL_INFO("[CollReduceScatterMixExecutor][CalcStreamNum] tag[%s] streamNum[%u]",
87 : tag_.c_str(), streamNum);
88 0 : return HCCL_SUCCESS;
89 : }
90 :
91 :
92 0 : bool CollReduceScatterMixExecutor::IsHugeData(const u64 curSize, OpParam *param)
93 : {
94 0 : const u64 TBE_REDUCE_MAX_COUNT = INT32_MAX;
95 :
96 0 : u64 curCount = curSize / SIZE_TABLE[param->DataDes.dataType];
97 0 : bool issupportRDMAInlineReduce = IsSupportRDMAReduce(param->DataDes.dataType, param->reduceType);
98 : // 这里如果CheckCommSize返回ERROR，相当于HugeData true，防止GetSubCommInfo越界
99 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
100 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
101 0 : u32 level0RankSize = level0CommInfo.localRankSize;
102 :
103 0 : bool hugeData =
104 0 : (curSize * level0RankSize / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE) ||
105 0 : (curSize > SDMA_SEND_MAX_SIZE) ||
106 0 : ((!isSupportSDMAReduce_) && (curCount > TBE_REDUCE_MAX_COUNT)) ||
107 0 : ((!issupportRDMAInlineReduce) && (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, 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("[CollReduceScatterMixExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]",
154 : tag_.c_str(), inputType, outputType);
155 0 : return HCCL_SUCCESS;
156 : }
157 :
158 0 : HcclResult CollReduceScatterMixExecutor::CalcLevel0CommInfo(TransportMemType inputType, TransportMemType outputType,
159 : std::vector<LevelNSubCommTransport>& opTransport)
160 : {
161 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
162 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
163 0 : commParaLevel0.meshSinglePlane = meshSinglePlane_;
164 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
165 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
166 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
167 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
168 0 : }
169 :
170 0 : return HCCL_SUCCESS;
171 : }
172 :
173 0 : u64 CollReduceScatterMixExecutor::CalcLoopMaxCount(const u32 unitSize)
174 : {
175 0 : u64 maxCountPerLoop = inCCLbufferSize_ / (topoAttr_.userRankSize * unitSize);
176 0 : return maxCountPerLoop;
177 : }
178 :
179 0 : void CollReduceScatterMixExecutor::CalLevel0DataSegsSlice(
180 : const ExecMem &execMem, const std::vector<std::vector<Slice>> &multiStreamSlice,
181 : u32 sliceNum, u32 level1RankSize, std::vector<std::vector<Slice>> &level0DataSegsSlice)
182 : {
183 0 : for (u32 ringIndex = 0; ringIndex < multiStreamSlice.size(); ringIndex++) {
184 0 : std::vector<Slice> dataSlice;
185 0 : FillMultiRingSlice(execMem, multiStreamSlice, sliceNum, level1RankSize, ringIndex, dataSlice);
186 0 : level0DataSegsSlice.push_back(dataSlice);
187 0 : }
188 0 : }
189 :
190 0 : void CollReduceScatterMixExecutor::FillMultiRingSlice(
191 : const ExecMem &execMem, const std::vector<std::vector<Slice>> &multiStreamSlice,
192 : u32 sliceNum, u32 level1RankSize, const u32 ringIndex, std::vector<Slice> &dataSlice)
193 : {
194 0 : for (u32 level0Idx = 0; level0Idx < sliceNum; level0Idx++) {
195 0 : Slice sliceTemp;
196 0 : for (u32 level1Idx = 0; level1Idx < level1RankSize; level1Idx++) {
197 0 : sliceTemp.size = multiStreamSlice[ringIndex][level0Idx].size;
198 0 : sliceTemp.offset = multiStreamSlice[ringIndex][level0Idx].offset +
199 0 : level1Idx * sliceNum * execMem.outputMem.size();
200 0 : dataSlice.push_back(sliceTemp);
201 0 : HCCL_DEBUG("rank[%u] sliceTemp.size[%zu], sliceTemp.offset[%llu]", topoAttr_.userRank,
202 : sliceTemp.size, sliceTemp.offset);
203 : }
204 : }
205 0 : }
206 :
207 0 : HcclResult CollReduceScatterMixExecutor::CalLevel1DataSegsSlice(
208 : const ExecMem &execMem, const u32 &commIndex,
209 : u32 sliceNum, u32 level1RankSize, std::vector<Slice> &level1DataSegsSlice)
210 : {
211 : (void) sliceNum;
212 0 : for (u32 i = 0; i < level1RankSize; i++) {
213 0 : Slice sliceTemp;
214 : u32 level1UserRank;
215 0 : CHK_RET(GetUserRankByRank(COMM_LEVEL1, commIndex, i, level1UserRank));
216 0 : sliceTemp.size = execMem.outputMem.size();
217 0 : sliceTemp.offset = level1UserRank * execMem.outputMem.size();
218 0 : level1DataSegsSlice.push_back(sliceTemp);
219 0 : HCCL_DEBUG("rank[%u], level1DataSegsSlice[%u].offset=%llu, size=[%llu]", topoAttr_.userRank, i,
220 : sliceTemp.offset, sliceTemp.size);
221 : }
222 0 : return HCCL_SUCCESS;
223 : }
224 :
225 0 : HcclResult CollReduceScatterMixExecutor::KernelRun(const OpParam ¶m, ExecMem &execMem)
226 : {
227 0 : HCCL_CONFIG_INFO(HCCL_ALG,"[CollReduceScatterMixExecutor][KernelRun] The ReduceScatterMixExecutor starts.");
228 0 : u32 perDataSize = 0;
229 0 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, perDataSize));
230 :
231 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
232 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
233 0 : u32 level0RankSize = level0CommInfo.localRankSize;
234 0 : u32 commIndex = level0CommInfo.localRank; // 找到rank所在的节点间平面
235 :
236 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
237 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
238 0 : u32 serverIndex = level1CommInfo.localRank;
239 0 : u32 level1RankSize = level1CommInfo.localRankSize;
240 0 : HCCL_DEBUG("ReduceScatterMixExecutor inputSize=%llu, level0RankSize=%u,commIndex=%u, level1RankSize=%u, serverIndex=%u",
241 : execMem.inputMem.size(), level0RankSize, commIndex, level1RankSize, serverIndex);
242 :
243 0 : HcomCollOpInfo opInfo = {"", execMem.inputPtr, execMem.outputPtr, param.DataDes.count, param.DataDes.dataType,
244 0 : param.root, param.reduceType};
245 0 : HCCL_DEBUG("[CollReduceScatterMixExecutor][KernelRun] execMem.inputPtr[%p], execMem.outputPtr[%p], "
246 : "execMem.inputMem[%p], execMem.outputMem[%p]",
247 : execMem.inputPtr, execMem.outputPtr, execMem.inputMem.ptr(), execMem.outputMem.ptr());
248 0 : HcomCollOpInfo *opInfoPtr = nullptr;
249 0 : if (DMAReduceFlag_) {
250 0 : opInfoPtr = &opInfo;
251 : }
252 :
253 : // 第一步，AI server内reduce scatter mesh/dbring
254 0 : u32 sliceNum = level0CommInfo.localRankSize;
255 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
256 : u32 ringNum;
257 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
258 0 : ringNum = LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE;
259 : } else {
260 0 : ringNum = LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
261 : }
262 :
263 0 : Slice sliceTemp;
264 :
265 0 : std::vector<Slice> dataSegsSlice; // 数据分成ranksize份，每份的起始偏移和大小
266 0 : std::vector<std::vector<Slice>> multiStreamSlice; // 每个stream使用的数据基于用户buffer的偏移
267 :
268 0 : CHK_RET(ActiveSlaveStreams(param.stream));
269 :
270 : // 计算slice
271 0 : std::vector<std::vector<Slice>> level0DataSegsSlice;
272 0 : bool useInlineRduce = false;
273 0 : bool isInlineReduce = IsSupportSDMAReduce(execMem.inputMem.ptr(), execMem.scratchMem.ptr(),
274 0 : param.DataDes.dataType, param.reduceType);
275 0 : useInlineRduce = isInlineReduce && algoAttr_.inlineReduceSwitchOn;
276 0 : multiStreamSlice = ReduceScatterRingSlicePrepare(ringNum, sliceNum, useInlineRduce, execMem.outputMem,
277 0 : dataSegsSlice, param.tag); // 2个ring，每条ring上数据的偏移和大小
278 :
279 0 : CalLevel0DataSegsSlice(execMem, multiStreamSlice, sliceNum, level1RankSize, level0DataSegsSlice);
280 :
281 0 : std::vector<std::vector<Slice>> multRingsUserMemSlice;
282 :
283 0 : if (opInfoPtr == nullptr &&
284 0 : (!(topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING &&
285 0 : workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB))) {
286 0 : multRingsUserMemSlice = level0DataSegsSlice;
287 : } else {
288 0 : for (u32 ringIndex = 0; ringIndex < level0DataSegsSlice.size(); ringIndex++) {
289 0 : std::vector<Slice> level1UserMemSlice;
290 0 : for (auto &cclSlice : level0DataSegsSlice[ringIndex]) {
291 0 : Slice tmpSlice;
292 0 : tmpSlice.size = cclSlice.size;
293 0 : tmpSlice.offset =
294 0 : (cclSlice.offset / execMem.outputMem.size()) * param.DataDes.count * perDataSize +
295 0 : multiStreamSlice[ringIndex][0].offset;
296 0 : level1UserMemSlice.push_back(tmpSlice);
297 0 : HCCL_DEBUG("rank[%u], ringIndex[%u], tmpSlice.offset=[%llu], size=[%llu]",
298 : topoAttr_.userRank, ringIndex, tmpSlice.offset, tmpSlice.size);
299 : }
300 0 : multRingsUserMemSlice.push_back(level1UserMemSlice);
301 0 : }
302 : }
303 : // 区分消减拷贝场景
304 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING &&
305 0 : workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) {
306 : // 图模式opinfo不为空
307 0 : HcomCollOpInfo graphModeOpInfo = {
308 0 : "", execMem.inputMem.ptr(), nullptr, param.DataDes.count, param.DataDes.dataType,
309 0 : param.root, param.reduceType};
310 0 : CHK_RET(MultiRingReduceScatter(param.tag, execMem.inputMem, execMem.scratchMem, execMem.count,
311 : param.DataDes.dataType, param.reduceType, level0DataSegsSlice,
312 : param.stream, PROF_STAGE_0, 0, &graphModeOpInfo, multRingsUserMemSlice));
313 0 : } else if (opInfoPtr != nullptr) {
314 0 : HcomCollOpInfo opInfoByReduceScatterDMAreduce = *opInfoPtr;
315 0 : opInfoByReduceScatterDMAreduce.outputAddr = nullptr;
316 0 : CHK_RET(MultiRingReduceScatter(param.tag, execMem.inputMem, execMem.scratchMem, execMem.count,
317 : param.DataDes.dataType, param.reduceType, level0DataSegsSlice,
318 : param.stream, PROF_STAGE_0, 0, &opInfoByReduceScatterDMAreduce, multRingsUserMemSlice));
319 : } else {
320 0 : CHK_RET(MultiRingReduceScatter(param.tag, execMem.inputMem, execMem.scratchMem, execMem.count,
321 : param.DataDes.dataType, param.reduceType,
322 : level0DataSegsSlice, param.stream, PROF_STAGE_0, 0, opInfoPtr, multRingsUserMemSlice));
323 : }
324 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
325 0 : CHK_RET(ActiveSlaveStreams(param.stream));
326 :
327 : // 根据数据量算每个环上数据的偏移和大小，把做完hd的slice均分成RankSize份
328 0 : std::vector<Slice> dataSegsSlice;
329 0 : CHK_RET(PrepareReduceScatterSliceData(execMem.count, perDataSize, sliceNum, dataSegsSlice));
330 :
331 0 : if (opInfoPtr != nullptr) {
332 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.scratchMem, param.DataDes.dataType,
333 0 : param.reduceType);
334 0 : std::unique_ptr<AlgTemplateBase> level0Executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
335 0 : TemplateType::TEMPLATE_REDUCESCATTER_MESH_MIX, dispatcher_);
336 :
337 0 : CHK_SMART_PTR_NULL(level0Executor);
338 0 : CHK_RET(level0Executor->Prepare(execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count,
339 : param.DataDes.dataType, param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, dataSegsSlice, 0,
340 : reduceAttr, algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
341 : serverIndex, level1RankSize, opInfoPtr));
342 0 : CHK_RET(level0Executor->RegisterProfiler((level0RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + commIndex,
343 : PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
344 0 : CHK_RET(RunTemplate(level0Executor, level0CommInfo));
345 0 : } else {
346 0 : std::vector<std::vector<Slice> > multiStreamSlice; // 每个stream使用的数据基于用户buffer的偏移
347 : // mesh算法stream数量为rank数减1
348 0 : CHK_RET(AlgTemplateBase::PrepareSliceMeshStreams(dataSegsSlice, sliceNum - 1, multiStreamSlice));
349 :
350 : // 计算slice
351 0 : std::vector<std::vector<Slice>> level0DataSegsSlice;
352 0 : CalLevel0DataSegsSlice(execMem, multiStreamSlice, sliceNum, level1RankSize, level0DataSegsSlice);
353 :
354 0 : CHK_RET(MultiStreamReduceScatterMesh(param.tag, execMem.inputMem, execMem.scratchMem, execMem.count,
355 : param.DataDes.dataType, param.reduceType, level0DataSegsSlice, param.stream, COMM_LEVEL0, 0));
356 0 : }
357 0 : }
358 :
359 : // 第二步，节点间reduce scatter
360 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.scratchMem, param.DataDes.dataType, param.reduceType);
361 :
362 0 : std::unique_ptr<AlgTemplateBase> level1Executor;
363 :
364 : // 计算slice
365 0 : std::vector<Slice> level1DataSegsSlice;
366 :
367 0 : CHK_RET(CalLevel1DataSegsSlice(execMem, commIndex, sliceNum, level1RankSize, level1DataSegsSlice));
368 :
369 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
370 0 : level1Executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
371 0 : TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
372 0 : CHK_SMART_PTR_NULL(level1Executor);
373 0 : CHK_RET(level1Executor->Prepare(reduceAttr));
374 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RING in COMM_LEVEL1", __func__);
375 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
376 0 : level1Executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
377 0 : TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
378 0 : CHK_SMART_PTR_NULL(level1Executor);
379 0 : CHK_RET(level1Executor->Prepare(reduceAttr, false));
380 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR in COMM_LEVEL1", __func__);
381 : } else {
382 0 : HCCL_ERROR("[CollReduceScatterMixExecutor][KernelRun]ReduceScatter mix: algType[%u] is not supported.", algType_.algoLevel1);
383 0 : return HCCL_E_NOT_SUPPORT;
384 : }
385 :
386 0 : CHK_RET(level1Executor->Prepare(execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count,
387 : param.DataDes.dataType, param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, level1DataSegsSlice));
388 0 : CHK_RET(level1Executor->RegisterProfiler(
389 : (level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank,
390 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
391 0 : CHK_RET(RunTemplate(level1Executor, level1CommInfo));
392 :
393 : // 区分消减拷贝场景（消减拷贝数据需要拷贝到user output上）
394 0 : DeviceMem srcMem = execMem.inputMem.range(topoAttr_.userRank * execMem.outputMem.size(),
395 0 : execMem.outputMem.size());
396 0 : if (opInfoPtr != nullptr) {
397 0 : DeviceMem dstMem = DeviceMem::create(static_cast<u8 *>(opInfoPtr->outputAddr), execMem.outputMem.size());
398 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
399 0 : } else {
400 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, execMem.outputMem, srcMem, const_cast<Stream&>(param.stream)));
401 : }
402 :
403 0 : HCCL_INFO("[CollReduceScatterMixExecutor][KernelRun]ReduceScatter mix run success");
404 0 : return HCCL_SUCCESS;
405 0 : }
406 :
407 : REGISTER_EXEC("ReduceScatterMixExecutor", ReduceScatterMix, CollReduceScatterMixExecutor);
408 : } // namespace hccl
|
