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