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_all_reduce_order_preserved_executor.h"
12 :
13 : namespace hccl {
14 :
15 0 : CollAllReduceOrderPreservedExecutor::CollAllReduceOrderPreservedExecutor(
16 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
17 0 : : CollAllReduceExecutor(dispatcher, topoMatcher)
18 : {
19 0 : DMAReduceFlag_ = true;
20 0 : CCLMemSlice_ = false;
21 0 : }
22 :
23 0 : void CollAllReduceOrderPreservedExecutor::ParseParam(const OpParam& param)
24 : {
25 0 : tag_ = param.tag;
26 :
27 0 : u64 sizePerBlock = (param.DataDes.count + topoAttr_.userRankSize - 1) / topoAttr_.userRankSize
28 0 : * SIZE_TABLE[param.DataDes.dataType];
29 0 : sizePerBlock = AlgTemplateBase::RoundUpWithDivisor(sizePerBlock, HCCL_MIN_SLICE_ALIGN);
30 :
31 : // 是否需要scratch memory（图模式没有cclbuffer，需要额外申请scratchMem）
32 0 : u64 inputSize = param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
33 0 : scratchMemFlag_ = (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE)
34 0 : && (inputSize < (topoAttr_.userRankSize - 1) * sizePerBlock);
35 :
36 0 : totalSize_ = std::max(sizePerBlock * topoAttr_.userRankSize, inputSize);
37 0 : }
38 :
39 0 : HcclResult CollAllReduceOrderPreservedExecutor::CalcScratchMemSize(u64& scratchMemSize)
40 : {
41 0 : scratchMemSize = scratchMemFlag_ ? totalSize_ : 0U;
42 0 : HCCL_INFO("[%s]tag[%s] scratchMemSize[%llu]", __func__, tag_.c_str(), scratchMemSize);
43 0 : return HCCL_SUCCESS;
44 : }
45 :
46 0 : u32 CollAllReduceOrderPreservedExecutor::CalReduceStreamNum(const u32& localRankSize)
47 : {
48 0 : return (1 << static_cast<int>(std::floor(log2(localRankSize))));
49 : }
50 :
51 0 : HcclResult CollAllReduceOrderPreservedExecutor::CalcStreamNum(u32& streamNum)
52 : {
53 0 : if (topoAttr_.deviceNumPerAggregation == 1) {
54 0 : u32 level1StreamNum = CalReduceStreamNum(topoAttr_.moduleNum);
55 0 : streamNum = std::min(level1StreamNum, DEVICE_EIGHT + DEVICE_EIGHT / FACTOR_NUM_TWO - 1);
56 0 : HCCL_INFO(
57 : "[%s]tag[%s] single rank per module, level1StreamNum[%u], streamNum[%u]", __func__, tag_.c_str(),
58 : level1StreamNum, streamNum);
59 0 : return HCCL_SUCCESS;
60 : }
61 :
62 : // Level0RankSize条流给alltoall，剩下的流给LocalReduce使用
63 0 : u32 level0StreamNum = topoAttr_.deviceNumPerAggregation - 1 + CalReduceStreamNum(topoAttr_.deviceNumPerAggregation);
64 : // level1主流分给alltoall，从流给LocalReduce使用
65 0 : u32 level1StreamNum = CalReduceStreamNum(topoAttr_.moduleNum);
66 : // 总流数上限：7（alltoall使用，提前的本地拷贝任务不需要并行）+ 4（LocalReduce使用）
67 : streamNum
68 0 : = std::min(std::max(level0StreamNum - 1, level1StreamNum), DEVICE_EIGHT + DEVICE_EIGHT / FACTOR_NUM_TWO - 1);
69 :
70 0 : HCCL_INFO(
71 : "[%s]tag[%s] level0StreamNum[%u], level1StreamNum[%u], streamNum[%u]", __func__, tag_.c_str(), level0StreamNum,
72 : level1StreamNum, streamNum);
73 0 : return HCCL_SUCCESS;
74 : }
75 :
76 0 : HcclResult CollAllReduceOrderPreservedExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
77 : {
78 0 : TransportMemType inputType = TransportMemType::RESERVED;
79 0 : TransportMemType outputType = TransportMemType::RESERVED;
80 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
81 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
82 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
83 0 : return HCCL_SUCCESS;
84 : }
85 :
86 : HcclResult
87 0 : CollAllReduceOrderPreservedExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
88 : {
89 : // 图模式场景使用PARAM_INPUT/OUTPUT -> userInput/userOutPut，不需要scrachMem
90 0 : inputType = workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE ? TransportMemType::PARAM_INPUT :
91 : TransportMemType::CCL_INPUT;
92 0 : outputType = workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE ? TransportMemType::PARAM_OUTPUT :
93 : TransportMemType::CCL_OUTPUT;
94 :
95 0 : if (scratchMemFlag_) {
96 0 : outputType = TransportMemType::SCRATCH;
97 : }
98 :
99 0 : HCCL_INFO("[%s]tag[%s] inputType[%d], outputType[%d]", __func__, tag_.c_str(), inputType, outputType);
100 0 : return HCCL_SUCCESS;
101 : }
102 :
103 0 : HcclResult CollAllReduceOrderPreservedExecutor::CalcLevel0CommInfo(
104 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
105 : {
106 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
107 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
108 0 : return HCCL_SUCCESS;
109 0 : }
110 :
111 0 : HcclResult CollAllReduceOrderPreservedExecutor::CalcLevel1CommInfo(
112 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
113 : {
114 0 : if (topoAttr_.moduleNum > 1) {
115 0 : CommParaInfo commParaLevel1(COMM_LEVEL1, CommType::COMM_TAG_MESH);
116 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel1, opTransport[COMM_LEVEL1], inputType, outputType));
117 0 : }
118 0 : return HCCL_SUCCESS;
119 : }
120 :
121 0 : bool CollAllReduceOrderPreservedExecutor::IsHugeData(const u64 curSize)
122 : {
123 0 : bool hugeData = curSize / topoAttr_.deviceNumPerAggregation / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE
124 0 : || curSize > SDMA_SEND_MAX_SIZE;
125 0 : HCCL_DEBUG(
126 : "[%s]isHugeData[%d], curSize[%llu], topoAttr_.deviceNumPerAggregation[%u]", __func__, hugeData, curSize,
127 : topoAttr_.deviceNumPerAggregation);
128 0 : return hugeData;
129 : }
130 :
131 0 : void CollAllReduceOrderPreservedExecutor::CalcSizePerBlock(const OpParam& param, ExecMem& execMem)
132 : {
133 : sizePerBlock_
134 0 : = (execMem.count + topoAttr_.userRankSize - 1) / topoAttr_.userRankSize * SIZE_TABLE[param.DataDes.dataType];
135 0 : sizePerBlock_ = AlgTemplateBase::RoundUpWithDivisor(sizePerBlock_, HCCL_MIN_SLICE_ALIGN);
136 0 : }
137 :
138 0 : void CollAllReduceOrderPreservedExecutor::CalGroupSlices(const OpParam& param, const ExecMem& execMem)
139 : {
140 0 : groupSize_.clear();
141 0 : u64 sizeRemain = execMem.count * SIZE_TABLE[param.DataDes.dataType];
142 0 : for (u32 rankId = 0; rankId < topoAttr_.userRankSize; rankId++) {
143 0 : u64 size = (sizeRemain > sizePerBlock_) ? sizePerBlock_ : sizeRemain;
144 0 : groupSize_.push_back(size);
145 0 : sizeRemain -= size;
146 : }
147 0 : }
148 :
149 0 : HcclResult CollAllReduceOrderPreservedExecutor::RunReduceScatterLevel0(
150 : const OpParam& param, ExecMem& execMem, SubCommInfo& level0CommInfo)
151 : {
152 0 : if (level0CommInfo.localRankSize == 1) {
153 0 : all2allOffset_ = topoAttr_.moduleNum > 1 ? 1 : 0;
154 0 : HCCL_INFO("[%s] single rank per module, skip L0 AllToAll and LocalReduce, tag[%s]", __func__, tag_.c_str());
155 0 : CHK_RET(RunReduceScatterLevel0SingleRank(param, execMem, level0CommInfo));
156 0 : return HCCL_SUCCESS;
157 : }
158 :
159 : // 切分数据(ReduceScatter分组，记录每组的起始偏移和大小)
160 0 : GroupSlicesInfo groupSlicesInfoLevel0;
161 0 : for (u32 groupId = 0; groupId < topoAttr_.moduleNum; groupId++) {
162 0 : MemBlockInfo memInfo;
163 0 : for (u32 dataId = 0; dataId < level0CommInfo.localRankSize; dataId++) {
164 0 : u64 globalDataId = groupId * level0CommInfo.localRankSize + dataId;
165 0 : u64 size = groupSize_[globalDataId];
166 0 : u64 offset = globalDataId * sizePerBlock_;
167 0 : memInfo.size.push_back(size);
168 0 : memInfo.userInputOffsets.push_back(offset);
169 0 : memInfo.inputOffsets.push_back(offset);
170 0 : memInfo.outputOffsets.push_back(offset);
171 : }
172 0 : groupSlicesInfoLevel0.push_back(memInfo);
173 0 : }
174 :
175 0 : CHK_RET(ActiveSlaveStreams(param.stream));
176 0 : all2allOffset_ = topoAttr_.moduleNum > 1 ? 1 : 0; // 多机场景需要偏移1（给L1预留计算位，减少拷贝次数）
177 :
178 0 : std::unique_ptr<AlgTemplateBase> level0TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
179 0 : TemplateType::TEMPLATE_REDUCESCATTER_PLANT_LOCAL_REDUCE, dispatcher_);
180 0 : CHK_SMART_PTR_NULL(level0TempAlg);
181 :
182 0 : DeviceMem outputMem = scratchMemFlag_ ? execMem.scratchMem : execMem.outputMem;
183 0 : CHK_RET(level0TempAlg->Prepare(
184 : execMem.inputPtr, execMem.inputMem, outputMem, param.stream, algResResp_->slaveStreams,
185 : algResResp_->notifiesMain, algResResp_->notifiesAux, groupSlicesInfoLevel0, param.reduceType, all2allOffset_,
186 : param.DataDes.dataType, true));
187 :
188 0 : CHK_RET(level0TempAlg->RegisterProfiler(
189 : (level0CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_2,
190 : HCCL_EXEC_STEP_NOT_SET, param.stream));
191 0 : CHK_RET(RunTemplate(level0TempAlg, level0CommInfo));
192 0 : return HCCL_SUCCESS;
193 0 : }
194 :
195 0 : HcclResult CollAllReduceOrderPreservedExecutor::RunReduceScatterLevel0SingleRank(
196 : const OpParam& param, ExecMem& execMem, const SubCommInfo& level0CommInfo) const
197 : {
198 : (void)level0CommInfo;
199 0 : u64 size = execMem.count * SIZE_TABLE[param.DataDes.dataType];
200 :
201 0 : DeviceMem srcMem = DeviceMem::create(execMem.inputPtr, size);
202 0 : DeviceMem dstMem = scratchMemFlag_ ? execMem.scratchMem.range(0, size) : execMem.inputMem.range(0, size);
203 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
204 :
205 0 : return HCCL_SUCCESS;
206 0 : }
207 :
208 0 : HcclResult CollAllReduceOrderPreservedExecutor::RunReduceScatterLevel1(
209 : const OpParam& param, ExecMem& execMem, SubCommInfo& level0CommInfo)
210 : {
211 0 : u32 commIndex = level0CommInfo.localRank;
212 0 : u32 level0Ranksize = level0CommInfo.localRankSize;
213 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
214 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
215 :
216 : // 切分数据，记录每组的起始偏移和大小（仅1组）
217 0 : MemBlockInfo memInfo;
218 0 : u32 inputBaseIndex
219 0 : = (all2allOffset_ + commIndex) % level0Ranksize; // 多机场景需要偏移1（给L1预留计算位，减少拷贝次数）
220 0 : for (u32 dataId = 0; dataId < level1CommInfo.localRankSize; dataId++) {
221 0 : u64 inputIndex = inputBaseIndex + dataId * level0Ranksize;
222 0 : memInfo.inputOffsets.push_back(inputIndex * sizePerBlock_);
223 0 : memInfo.size.push_back(groupSize_[commIndex + dataId * level0Ranksize]);
224 0 : u64 outputIndex = commIndex + dataId * level0Ranksize;
225 0 : memInfo.outputOffsets.push_back(outputIndex * sizePerBlock_);
226 0 : memInfo.userInputOffsets.push_back(outputIndex * sizePerBlock_);
227 : }
228 :
229 0 : DeviceMem outputMem = scratchMemFlag_ ? execMem.scratchMem : execMem.outputMem;
230 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
231 0 : TemplateType::TEMPLATE_REDUCESCATTER_PLANT_LOCAL_REDUCE_COMBINE, dispatcher_);
232 0 : CHK_SMART_PTR_NULL(level1TempAlg);
233 :
234 0 : u32 level0LastRank = level0Ranksize - 1;
235 0 : bool isUseCclIn = (level0Ranksize == 1) || (commIndex == level0LastRank - 1);
236 0 : CHK_RET(level1TempAlg->Prepare(
237 : execMem.inputMem, outputMem, param.stream, algResResp_->slaveStreams, algResResp_->notifiesMain,
238 : algResResp_->notifiesAux, memInfo, param.reduceType, param.DataDes.dataType, isUseCclIn,
239 : commIndex == level0LastRank, true));
240 :
241 0 : CHK_RET(level1TempAlg->RegisterProfiler(
242 : (level0Ranksize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_2,
243 : HCCL_EXEC_STEP_NOT_SET, param.stream));
244 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
245 0 : return HCCL_SUCCESS;
246 0 : }
247 :
248 0 : HcclResult CollAllReduceOrderPreservedExecutor::RunAllGatherLevel0(
249 : const OpParam& param, ExecMem& execMem, SubCommInfo& level0CommInfo)
250 : {
251 0 : u32 level0RankSize = level0CommInfo.localRankSize;
252 0 : u32 commIndex = level0CommInfo.localRank;
253 0 : u64 count = execMem.count / topoAttr_.userRankSize;
254 0 : u64 serverOffsetConut = topoAttr_.userRank / level0RankSize * level0RankSize;
255 :
256 : // allgather 计算slice，数据分成ranksize份，每份的起始偏移和大小
257 0 : std::vector<Slice> dataSegsSlice;
258 0 : for (u32 rank = 0; rank < level0RankSize; rank++) {
259 0 : Slice userslice;
260 0 : userslice.size = groupSize_[rank + serverOffsetConut];
261 0 : userslice.offset = userslice.size == 0 ? 0 : (rank + serverOffsetConut) * sizePerBlock_;
262 0 : dataSegsSlice.emplace_back(std::move(userslice));
263 : }
264 :
265 0 : DeviceMem outputMem = scratchMemFlag_ ? execMem.scratchMem : execMem.outputMem;
266 :
267 : std::unique_ptr<AlgTemplateBase> level0TempAlg
268 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_MESH_ATOMIC, dispatcher_);
269 :
270 0 : CHK_SMART_PTR_NULL(level0TempAlg);
271 0 : CHK_RET(level0TempAlg->Prepare(
272 : algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux, topoAttr_.userRank, nullptr,
273 : commIndex, level0RankSize));
274 0 : CHK_RET(level0TempAlg->Prepare(
275 : outputMem, outputMem, outputMem, count, param.DataDes.dataType, param.stream, HCCL_REDUCE_RESERVED,
276 : LEVEL0_BRIDGE_RANK_ID, dataSegsSlice, 0));
277 :
278 0 : CHK_RET(level0TempAlg->RegisterProfiler(
279 : (level0RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + commIndex, PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET,
280 : param.stream));
281 0 : CHK_RET(RunTemplate(level0TempAlg, level0CommInfo));
282 0 : return HCCL_SUCCESS;
283 0 : }
284 :
285 0 : HcclResult CollAllReduceOrderPreservedExecutor::RunAllGatherLevel1(
286 : const OpParam& param, ExecMem& execMem, const SubCommInfo& level0CommInfo)
287 : {
288 0 : u32 commIndex = level0CommInfo.localRank;
289 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
290 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
291 :
292 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg; // Level1Allgather(根据算法选择)
293 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
294 : level1TempAlg
295 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
296 0 : HCCL_INFO("AllGather mesh: using ring algo inter-server.");
297 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
298 : level1TempAlg
299 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
300 0 : HCCL_INFO("AllGather mesh: using nonuniform-bruck algo inter-server.");
301 : } else {
302 : level1TempAlg
303 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
304 0 : HCCL_INFO("AllGather mesh: using nhr algo inter-server.");
305 : }
306 :
307 0 : DeviceMem outputMem = scratchMemFlag_ ? execMem.scratchMem : execMem.outputMem;
308 :
309 0 : u32 level0RankSize = level0CommInfo.localRankSize;
310 0 : u64 count = execMem.count / level1CommInfo.localRankSize;
311 :
312 0 : std::vector<u64> level1GroupSize;
313 0 : for (u32 rank = 0; rank < level1CommInfo.localRankSize; rank++) {
314 0 : u64 size = 0;
315 0 : for (u32 level0RankId = 0; level0RankId < level0RankSize; level0RankId++) {
316 0 : size += groupSize_[rank * level0RankSize + level0RankId];
317 : }
318 0 : level1GroupSize.push_back(size);
319 : }
320 :
321 : // allgather 计算slice，数据分成ranksize份，每份的起始偏移和大小
322 0 : std::vector<Slice> dataSegsSlice;
323 0 : for (u32 rank = 0; rank < level1CommInfo.localRankSize; rank++) {
324 0 : Slice userslice;
325 0 : userslice.size = level1GroupSize[rank];
326 0 : userslice.offset = rank * level0RankSize * sizePerBlock_;
327 0 : dataSegsSlice.emplace_back(std::move(userslice));
328 : }
329 :
330 0 : CHK_SMART_PTR_NULL(level1TempAlg);
331 0 : CHK_RET(level1TempAlg->Prepare(
332 : outputMem, outputMem, outputMem, count, param.DataDes.dataType, param.stream,
333 : HcclReduceOp::HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID, dataSegsSlice));
334 :
335 0 : CHK_RET(level1TempAlg->RegisterProfiler(
336 : (level1CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank, PROF_STAGE_2,
337 : HCCL_EXEC_STEP_NOT_SET, param.stream));
338 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
339 0 : return HCCL_SUCCESS;
340 0 : }
341 :
342 0 : HcclResult CollAllReduceOrderPreservedExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
343 : {
344 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s]The CollAllReduceOrderPreservedExecutor starts, tag[%s]", __func__, tag_.c_str());
345 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
346 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
347 :
348 0 : CalcSizePerBlock(param, execMem);
349 0 : CalGroupSlices(param, execMem);
350 :
351 0 : u64 inputSize = param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
352 0 : scratchMemFlag_ = (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE)
353 0 : && (inputSize < (topoAttr_.userRankSize - 1) * sizePerBlock_);
354 :
355 : // L0 节点内 reduce scatter
356 0 : CHK_RET(RunReduceScatterLevel0(param, execMem, level0CommInfo));
357 : // L1 节点间 reduce scatter
358 0 : if (topoAttr_.moduleNum > 1) {
359 0 : CHK_RET(RunReduceScatterLevel1(param, execMem, level0CommInfo));
360 : }
361 :
362 : // Level0 节点内 AllGatherMeshAtomic
363 0 : CHK_RET(RunAllGatherLevel0(param, execMem, level0CommInfo));
364 0 : if (topoAttr_.moduleNum > 1) {
365 : // L1 节点间 allgather
366 0 : CHK_RET(RunAllGatherLevel1(param, execMem, level0CommInfo));
367 : }
368 :
369 : // 单算子需要 execMem.outputMem最后拷贝至UserOut
370 0 : if (scratchMemFlag_ || workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
371 0 : u64 dataSize = execMem.count * SIZE_TABLE[param.DataDes.dataType];
372 0 : void* srcPtr = scratchMemFlag_ ? execMem.scratchMem.ptr() : execMem.outputMem.ptr();
373 0 : DeviceMem srcMem = DeviceMem::create(srcPtr, dataSize);
374 0 : DeviceMem dstMem = DeviceMem::create(execMem.outputPtr, dataSize);
375 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
376 0 : }
377 :
378 0 : HCCL_INFO("[%s]order preserved AllReduce run success, tag[%s]", __func__, tag_.c_str());
379 0 : return HCCL_SUCCESS;
380 0 : }
381 :
382 : REGISTER_EXEC("AllReduceOrderPreservedExecutor", AllReduceOrderPreserved, CollAllReduceOrderPreservedExecutor);
383 : } // namespace hccl
|
