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_ring_zerocopy_executor.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 :
16 0 : CollReduceScatterRingZerocopyExecutor::CollReduceScatterRingZerocopyExecutor(const HcclDispatcher dispatcher,
17 0 : std::unique_ptr<TopoMatcher> &topoMatcher)
18 0 : : CollReduceScatterExecutor(dispatcher, topoMatcher)
19 : {
20 0 : DMAReduceFlag_ = true; // 设为true，以禁用RunLoop中的本地拷贝
21 0 : desc_.isZeroCopy = true;
22 0 : desc_.deterministic = 1;
23 0 : desc_.level1SupportedAlgos = {
24 : AlgTypeLevel1::ALG_LEVEL1_NHR,
25 : AlgTypeLevel1::ALG_LEVEL1_NB,
26 : AlgTypeLevel1::ALG_LEVEL1_RING,
27 : AlgTypeLevel1::ALG_LEVEL1_AHC,
28 : AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE
29 0 : };
30 0 : desc_.level2SupportedAlgos = {
31 : AlgTypeLevel2::ALG_LEVEL2_NHR,
32 : AlgTypeLevel2::ALG_LEVEL2_NB,
33 : AlgTypeLevel2::ALG_LEVEL2_RING
34 0 : };
35 0 : }
36 :
37 0 : void CollReduceScatterRingZerocopyExecutor::ParseParam(const OpParam& param)
38 : {
39 0 : tag_ = param.tag;
40 0 : totalSize_ = topoAttr_.userRankSize * param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
41 0 : aicpuUnfoldMode_ = param.aicpuUnfoldMode;
42 0 : }
43 :
44 0 : HcclResult CollReduceScatterRingZerocopyExecutor::CalcStreamNum(u32& streamNum)
45 : {
46 0 : u32 totalStreamNum = (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) ?
47 : (LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE + 1) : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
48 0 : streamNum = totalStreamNum - 1;
49 0 : HCCL_INFO("[CollReduceScatterRingZerocopyExecutor][CalcStreamNum] tag[%s] streamNum[%u]",
50 : tag_.c_str(), streamNum);
51 :
52 0 : return HCCL_SUCCESS;
53 : }
54 :
55 0 : HcclResult CollReduceScatterRingZerocopyExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
56 : {
57 0 : TransportMemType inputType = TransportMemType::RESERVED;
58 0 : TransportMemType outputType = TransportMemType::RESERVED;
59 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
60 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
61 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
62 0 : CHK_RET(CalcLevel2CommInfo(inputType, outputType, opTransport));
63 0 : return HCCL_SUCCESS;
64 : }
65 :
66 0 : HcclResult CollReduceScatterRingZerocopyExecutor::CalcTransportMemType(TransportMemType &inputType,
67 : TransportMemType &outputType)
68 : {
69 0 : inputType = TransportMemType::CCL_INPUT;
70 0 : if (scratchMemFlag_) {
71 0 : outputType = TransportMemType::SCRATCH;
72 : } else {
73 0 : outputType = TransportMemType::CCL_OUTPUT;
74 : }
75 :
76 0 : HCCL_INFO("[CollReduceScatterRingZerocopyExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]",
77 : tag_.c_str(), inputType, outputType);
78 0 : return HCCL_SUCCESS;
79 : }
80 :
81 0 : HcclResult CollReduceScatterRingZerocopyExecutor::CalcLevel0CommInfo(TransportMemType inputType,
82 : TransportMemType outputType,
83 : std::vector<LevelNSubCommTransport>& opTransport)
84 : {
85 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
86 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
87 0 : LevelNSubCommTransport &commTransportLevel0 = opTransport[COMM_LEVEL0];
88 0 : for (u32 subCommIndex = 0; subCommIndex < commTransportLevel0.size(); subCommIndex++) {
89 0 : commTransportLevel0[subCommIndex].isZeroCopy = true;
90 : }
91 0 : return HCCL_SUCCESS;
92 0 : }
93 :
94 0 : u64 CollReduceScatterRingZerocopyExecutor::CalcLoopMaxCount(const u32 unitSize)
95 : {
96 : // 中转内存单次最多能够接受的output count，放开ranksize限制
97 0 : u64 maxCountPerLoop = inCCLbufferSize_ / topoAttr_.serverNum / HCCL_MIN_SLICE_ALIGN
98 0 : * HCCL_MIN_SLICE_ALIGN / unitSize;
99 0 : return maxCountPerLoop;
100 : }
101 :
102 0 : HcclResult CollReduceScatterRingZerocopyExecutor::SemiRingReduceScatter(
103 : const std::string &tag, DeviceMem inputMem, DeviceMem outputMem,
104 : const u64 count, const HcclDataType dataType, const HcclReduceOp reductionOp,
105 : const std::vector<std::vector<Slice> > multRingsSliceZero, Stream stream, s32 profStage,
106 : const u64 baseOffset, const HcomCollOpInfo *opInfo,
107 : const std::vector<std::vector<Slice>> multRingsUserMemSlice)
108 : {
109 : (void) tag;
110 : (void) multRingsSliceZero;
111 : (void) baseOffset;
112 : (void) opInfo;
113 0 : HCCL_INFO("[CollReduceScatterRingZerocopyExecutor][SemiRingReduceScatter] SemiRingReduceScatter starts");
114 :
115 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
116 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
117 :
118 : //此处计算reduceAttr计算outputmem使用scratchmem
119 0 : u64 reduceAttr = GetReduceAttr(inputMem, outputMem, dataType, reductionOp);
120 : // 执行
121 0 : std::unique_ptr<AlgTemplateBase> executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
122 0 : TemplateType::TEMPLATE_REDUCESCATTER_UNIFIED_MARCH, dispatcher_);
123 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_UNIFIED_MARCH in COMM_LEVEL0", __func__);
124 0 : CHK_SMART_PTR_NULL(executor);
125 :
126 0 : CHK_RET(executor->Prepare(stream, level0CommInfo,
127 : algResResp_->paramInputMem, algResResp_->paramOutputMem, inputMem,
128 : outputMem, count, algResResp_->slaveStreams, algResResp_->notifiesMain,
129 : algResResp_->notifiesAux, dataType, reductionOp, multRingsUserMemSlice, reduceAttr));
130 0 : HCCL_DEBUG("[CollReduceScatterSemiRingExecutor][DoubleRingMidCountReduceScatter]reduceAttr is %llu", reduceAttr);
131 :
132 0 : HcclResult ret = executor->RegisterProfiler(
133 : ((COMM_INDEX_0 + 1) << PROF_RINGINDEX_OFFSET_OF_PLANEID) +
134 0 : (level0CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank,
135 : profStage, HCCL_EXEC_STEP_NOT_SET, stream);
136 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
137 : HCCL_ERROR("[CollReduceScatterRingZerocopyExecutor][SemiRingReduceScatter]"\
138 : "Double ring ReduceScatter failed,return[%d]", ret), ret);
139 :
140 0 : CHK_RET(executor->RunAsync());
141 :
142 0 : HCCL_INFO("[CollReduceScatterRingZerocopyExecutor][SemiRingReduceScatter] SemiRingReduceScatter run success");
143 0 : return ret;
144 0 : }
145 :
146 0 : HcclResult CollReduceScatterRingZerocopyExecutor::RunIntraSeverReduceScatter(
147 : const std::string &tag, DeviceMem &inputMem, DeviceMem &outputMem,
148 : const u64 count, const HcclDataType &dataType, const HcclReduceOp &reductionOp,
149 : const std::vector<std::vector<Slice>> &multRingsSliceZero, const Stream &stream, s32 profStage,
150 : const u64 baseOffset, const HcomCollOpInfo *opInfo,
151 : const std::vector<std::vector<Slice>> &multRingsUserMemSlice, const bool disableDMAReduce)
152 : {
153 : (void) disableDMAReduce;
154 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_SINGLE_RING) {
155 0 : CHK_RET(MultiRingReduceScatter(tag, inputMem, outputMem, count, dataType, reductionOp,
156 : multRingsSliceZero, stream, profStage, baseOffset, opInfo, multRingsUserMemSlice));
157 : } else {
158 0 : CHK_PRT_RET(topoType_ != TopoType::TOPO_TYPE_NP_DOUBLE_RING,
159 : HCCL_ERROR("[%s] unknown topoType: %u", __func__, topoType_), HCCL_E_NOT_SUPPORT);
160 0 : CHK_RET(SemiRingReduceScatter(tag, inputMem, outputMem, count, dataType, reductionOp,
161 : multRingsSliceZero, stream, profStage, baseOffset, opInfo, multRingsUserMemSlice));
162 : }
163 0 : return HCCL_SUCCESS;
164 : }
165 :
166 0 : HcclResult CollReduceScatterRingZerocopyExecutor::CalcLevel0DataSlices(const OpParam ¶m, const ExecMem &execMem,
167 : std::vector<Slice> &dataSegsSlice)
168 : {
169 0 : return CalcIntraServerDataSlicesDiscontinuous(param, execMem,
170 0 : level0RankSize_, level1RankSize_, level2RankSize_, dataSegsSlice);
171 : }
172 :
173 0 : HcclResult CollReduceScatterRingZerocopyExecutor::KernelRunIntraServerPre(const OpParam ¶m, ExecMem &execMem)
174 : {
175 0 : HCCL_CONFIG_INFO(HCCL_ALG,
176 : "[CollReduceScatterRingZerocopyExecutor][KernelRunIntraServerPre] The ReduceScatterDoubleRingExecutor starts");
177 0 : bool isAHCAlgo = algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
178 0 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE;
179 0 : CHK_RET(GetCommRankInfoNormal(level0Rank_, level0RankSize_, level1Rank_, level1RankSize_, level2Rank_,
180 : level2RankSize_, isAHCAlgo));
181 :
182 : // 计算slice信息
183 0 : std::vector<Slice> dataSegsSlice;
184 0 : CHK_RET(CalcLevel0DataSlices(param, execMem, dataSegsSlice));
185 :
186 : // 执行ReduceScatter
187 0 : u64 level0Count = (dataSegsSlice.size() > level0RankSize_) ? // 如果是非连续数据通信
188 0 : (execMem.count) : (execMem.count * level1RankSize_ * level2RankSize_);
189 0 : std::vector<std::vector<Slice>> multRingsUserMemSlice = {dataSegsSlice};
190 0 : CHK_RET(RunIntraSeverReduceScatter(param.tag, execMem.inputMem, execMem.scratchMem, level0Count,
191 : param.DataDes.dataType, param.reduceType,
192 : multRingsUserMemSlice, param.stream, PROF_STAGE_1, 0, nullptr, multRingsUserMemSlice, true));
193 :
194 0 : return HCCL_SUCCESS;
195 0 : }
196 :
197 0 : HcclResult CollReduceScatterRingZerocopyExecutor::KernelRunInterServerPreProcess(const OpParam ¶m, const ExecMem &execMem)
198 : {
199 0 : u32 unitSize = 0;
200 0 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, unitSize));
201 :
202 0 : DeviceMem dstMem;
203 0 : DeviceMem srcMem;
204 0 : u64 curSize = execMem.outputMem.size();
205 0 : Stream stream = param.stream;
206 0 : for (u32 i = 0; i < level1RankSize_; i++) {
207 0 : for (u32 j = 0; j < level2RankSize_; j++) {
208 : // 拷贝input上每个slice的数据到中转内存，源端每个slice的size固定为output的size
209 0 : u32 dstIndex = i * level2RankSize_ + j;
210 0 : u32 srcIndex = j * level1RankSize_ + i;
211 0 : dstMem = execMem.inputMem.range(dstIndex * curSize, curSize);
212 0 : srcMem = DeviceMem::create(static_cast<u8 *>(execMem.inputPtr)
213 0 : + param.DataDes.count * unitSize * level0RankSize_ * srcIndex
214 0 : + param.DataDes.count * unitSize * level0Rank_,
215 0 : curSize);
216 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream));
217 : }
218 : }
219 0 : return HCCL_SUCCESS;
220 0 : }
221 :
222 0 : HcclResult CollReduceScatterRingZerocopyExecutor::KernelRunInterServer(const OpParam ¶m, ExecMem &execMem)
223 : {
224 0 : u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.scratchMem, param.DataDes.dataType, param.reduceType);
225 :
226 : // 将数据从user input搬运到ccl input
227 0 : CHK_RET(KernelRunInterServerPreProcess(param, execMem));
228 :
229 : // 计算slice
230 0 : std::vector<Slice> level1DataSegsSlice;
231 0 : CalcLevel1DataSlices(execMem.outputMem.size(), level1RankSize_, level2RankSize_, level1DataSegsSlice);
232 :
233 : // 超节点内、节点间通信
234 0 : bool isAHCAlgo = algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC || algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE;
235 0 : if (level1RankSize_ > 1) {
236 : // 获取对应算法的Template
237 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
238 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
239 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
240 0 : TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
241 0 : CHK_SMART_PTR_NULL(level1TempAlg);
242 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
243 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RING in COMM_LEVEL1", __func__);
244 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
245 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
246 0 : TemplateType::TEMPLATE_REDUCESCATTER_NB, dispatcher_);
247 0 : CHK_SMART_PTR_NULL(level1TempAlg);
248 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
249 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NB in COMM_LEVEL1", __func__);
250 0 : } else if (isAHCAlgo) {
251 : // 获取通信域分组信息
252 0 : std::vector<std::vector<std::vector<u32>>> globalSubGroups;
253 0 : std::map<AHCConcOpType, TemplateType> ahcAlgOption;
254 0 : CHK_RET(topoMatcher_->GetGlobalSubGroups(COMM_LEVEL1_AHC, globalSubGroups));
255 0 : topoMatcher_->GetAHCAlgOption(ahcAlgOption);
256 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC) {
257 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_AHC, dispatcher_);
258 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_AHC in COMM_LEVEL1", __func__);
259 : } else {
260 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_AHC_BROKE, dispatcher_);
261 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_AHC_BROKE in COMM_LEVEL1", __func__);
262 : }
263 0 : CHK_SMART_PTR_NULL(level1TempAlg);
264 0 : CHK_RET(level1TempAlg->Prepare(execMem.count, globalSubGroups, ahcAlgOption));
265 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
266 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
267 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
268 0 : TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
269 0 : CHK_SMART_PTR_NULL(level1TempAlg);
270 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr, false));
271 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR in COMM_LEVEL1", __func__);
272 : } else {
273 0 : HCCL_ERROR("ReduceScatter ring: unsupported level1 algtype [%s]", AlgTypeToStr(algType_).c_str());
274 0 : return HCCL_E_NOT_SUPPORT;
275 : }
276 : // 执行算法编排
277 0 : CommPlane commPlaneLevel1 = isAHCAlgo ? COMM_LEVEL1_AHC : COMM_LEVEL1;
278 0 : CHK_RET(CheckCommSize(commPlaneLevel1, level0Rank_ + 1));
279 0 : SubCommInfo level1CommInfo = GetSubCommInfo(commPlaneLevel1, level0Rank_);
280 0 : CHK_RET(level1TempAlg->Prepare(execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count,
281 : param.DataDes.dataType, param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, level1DataSegsSlice));
282 0 : CHK_RET(level1TempAlg->RegisterProfiler(
283 : (level1RankSize_ << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1Rank_,
284 : PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET, param.stream));
285 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
286 0 : }
287 :
288 : // 超节点间通信
289 0 : if (level2RankSize_ > 1 && !isAHCAlgo) {
290 : // 获取对应算法的Template
291 0 : std::unique_ptr<AlgTemplateBase> level2TempAlg;
292 0 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
293 0 : level2TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
294 0 : TemplateType::TEMPLATE_REDUCESCATTER_NB, dispatcher_);
295 0 : CHK_SMART_PTR_NULL(level2TempAlg);
296 0 : CHK_RET(level2TempAlg->Prepare(reduceAttr));
297 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NB in COMM_LEVEL2", __func__);
298 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_RING) {
299 0 : level2TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
300 0 : TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
301 0 : CHK_SMART_PTR_NULL(level2TempAlg);
302 0 : CHK_RET(level2TempAlg->Prepare(reduceAttr));
303 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RING in COMM_LEVEL2", __func__);
304 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
305 0 : level2TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
306 0 : TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
307 0 : CHK_SMART_PTR_NULL(level2TempAlg);
308 0 : CHK_RET(level2TempAlg->Prepare(reduceAttr, false));
309 0 : if (algoAttr_.isSupportAtomicWrite) {
310 0 : level2TempAlg->CloseBarrier();
311 : }
312 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR in COMM_LEVEL2", __func__);
313 : } else {
314 0 : HCCL_ERROR("ReduceScatter ring: unsupported level2 algtype [%s]", AlgTypeToStr(algType_).c_str());
315 0 : return HCCL_E_NOT_SUPPORT;
316 : }
317 : // 执行算法编排
318 0 : DeviceMem level2InputMem = execMem.inputMem.range(level1DataSegsSlice[level1Rank_].offset,
319 0 : level1DataSegsSlice[level1Rank_].size);
320 0 : CHK_RET(level2TempAlg->Prepare(level2InputMem, level2InputMem, execMem.scratchMem, execMem.count,
321 : param.DataDes.dataType, param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID,
322 : std::vector<Slice>(0), level1DataSegsSlice[level1Rank_].offset));
323 0 : CHK_RET(level2TempAlg->RegisterProfiler(
324 : (level2RankSize_ << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level2Rank_,
325 : PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET, param.stream));
326 0 : CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1));
327 0 : SubCommInfo level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
328 0 : CHK_RET(RunTemplate(level2TempAlg, level2CommInfo));
329 0 : }
330 :
331 : // 后处理
332 0 : CHK_RET(KernelRunInterServerPostProcess(param, execMem));
333 :
334 0 : return HCCL_SUCCESS;
335 0 : }
336 :
337 0 : HcclResult CollReduceScatterRingZerocopyExecutor::KernelRunInterServerPostProcess(const OpParam ¶m, const ExecMem &execMem)
338 : {
339 0 : u32 dataIndex = level1Rank_ * level2RankSize_ + level2Rank_;
340 0 : u64 curSize = execMem.outputMem.size();
341 0 : DeviceMem srcMem = execMem.inputMem.range(curSize * dataIndex, curSize);
342 0 : DeviceMem dstMem = DeviceMem::create(static_cast<u8 *>(execMem.outputPtr), curSize);
343 0 : Stream stream = param.stream;
344 0 : return HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream);
345 0 : }
346 :
347 : REGISTER_EXEC("ReduceScatterRingZerocopyExecutor", ReduceScatterRingZerocopy, CollReduceScatterRingZerocopyExecutor);
348 : }
|
