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 <cmath>
12 : #include "alg_template_register.h"
13 : #include "all_reduce_local_reduce.h"
14 :
15 : namespace hccl {
16 2 : AllReduceLocalReduce::AllReduceLocalReduce(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher)
17 2 : {}
18 :
19 4 : AllReduceLocalReduce::~AllReduceLocalReduce()
20 4 : {}
21 :
22 2 : HcclResult AllReduceLocalReduce::Prepare(u64 reduceAttrBitMap, std::vector<Stream> &meshStreams,
23 : std::vector<std::shared_ptr<LocalNotify>> &meshSignal, std::vector<std::shared_ptr<LocalNotify>> &meshSignalAux,
24 : u32 interRank, u32 interRankSize, u32 userRank, HcomCollOpInfo *opInfo)
25 : {
26 2 : reduceAttr_ = reduceAttrBitMap;
27 2 : localRank_ = interRank;
28 2 : localRankSize_ = interRankSize;
29 2 : userRank_ = userRank;
30 2 : meshStreams_ = meshStreams;
31 2 : meshSignal_ = &meshSignal;
32 2 : meshSignalAux_ = &meshSignalAux;
33 2 : opInfo_ = opInfo;
34 2 : return HCCL_SUCCESS;
35 : }
36 0 : HcclResult AllReduceLocalReduce::MainRecordSub()
37 : {
38 0 : for (u32 signalIndex = 0; signalIndex < meshSignalAux_->size(); signalIndex++) {
39 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, (*meshSignalAux_)[signalIndex],
40 : profilerInput_.stage));
41 : }
42 0 : return HCCL_SUCCESS;
43 : }
44 :
45 0 : HcclResult AllReduceLocalReduce::SubWaitMain()
46 : {
47 0 : for (u32 streamIndex = 0; streamIndex < meshSignalAux_->size(); streamIndex++) {
48 0 : CHK_RET(LocalNotify::Wait(meshStreams_[streamIndex], dispatcher_, (*meshSignalAux_)[streamIndex],
49 : profilerInput_.stage));
50 : }
51 0 : return HCCL_SUCCESS;
52 : }
53 :
54 0 : HcclResult AllReduceLocalReduce::MainWaitSub()
55 : {
56 0 : for (u32 signalIndex = 0; signalIndex < meshSignal_->size(); signalIndex++) {
57 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, (*meshSignal_)[signalIndex], profilerInput_.stage));
58 : }
59 0 : return HCCL_SUCCESS;
60 : }
61 :
62 0 : HcclResult AllReduceLocalReduce::SubRecordMain()
63 : {
64 0 : for (u32 streamIndex = 0; streamIndex < meshSignal_->size(); streamIndex++) {
65 0 : CHK_RET(LocalNotify::Post(meshStreams_[streamIndex], dispatcher_, (*meshSignal_)[streamIndex],
66 : profilerInput_.stage));
67 : }
68 0 : return HCCL_SUCCESS;
69 : }
70 :
71 : // 将数据均分，最小单位是128
72 1 : HcclResult AllReduceLocalReduce::PrepareSlice(u64 dataCount, u32 unitSize, u32 sliceNum,
73 : std::vector<Slice> &dataSlice, std::vector<Slice> &startSlice)
74 : {
75 1 : Slice temp;
76 1 : Slice startTemp;
77 1 : u64 totalSize = dataCount * unitSize;
78 1 : dataSlice.clear();
79 1 : dataSlice.reserve(sliceNum);
80 1 : if (sliceNum == 0) {
81 0 : HCCL_ERROR("[Prepare][SliceData]data slice prepare, sliceNum is 0");
82 0 : return HCCL_E_PARA;
83 : }
84 1 : u64 sizePerSliceOri = (totalSize + sliceNum - 1) / sliceNum; /* 1是为了向上取整 */
85 1 : u64 sizeLimit = 0;
86 1 : if (outputMem_.ptr() == opInfo_->outputAddr) {
87 1 : sizeLimit = outputMem_.size();
88 : } else {
89 0 : sizeLimit = totalSize;
90 : }
91 :
92 1 : u64 sizePerSlice = RoundUpWithDivisor(sizePerSliceOri, HCCL_MIN_SLICE_ALIGN_910B); // 512B对齐
93 1 : if (sizePerSlice * (localRankSize_ - 1) > sizeLimit) {
94 1 : sizePerSlice = RoundUpWithDivisor(sizePerSliceOri, HCCL_MIN_SLICE_ALIGN_ONCHIP);
95 : }
96 1 : if (sizePerSlice * (localRankSize_ - 1) > sizeLimit) {
97 1 : sizePerSlice = RoundUpWithDivisor(sizePerSliceOri, unitSize);
98 : }
99 1 : u64 residueSize = totalSize;
100 1 : u32 i = 0;
101 7 : while (residueSize > 0) {
102 6 : u64 sliceSize = sizePerSlice < residueSize ? sizePerSlice : residueSize;
103 6 : temp.size = sliceSize;
104 6 : temp.offset = totalSize - residueSize;
105 6 : i++;
106 6 : if (sliceSize <= 0) {
107 0 : HCCL_ERROR("[Prepare][SliceData]data_slices_prepare sliceSize[%llu]", sliceSize);
108 0 : return HCCL_E_PARA;
109 : }
110 6 : residueSize -= sliceSize;
111 6 : dataSlice.push_back(temp);
112 6 : if (i != sliceNum) {
113 6 : startTemp.size = sizePerSlice;
114 6 : startTemp.offset = 0;
115 6 : startSlice.push_back(startTemp);
116 : } else {
117 0 : startTemp.size = sizePerSlice;
118 0 : startTemp.offset = sizePerSlice;
119 0 : startSlice.push_back(startTemp);
120 : }
121 : }
122 3 : while (i < sliceNum) {
123 2 : temp.size = 0;
124 2 : temp.offset = totalSize;
125 2 : i++;
126 2 : dataSlice.push_back(temp);
127 2 : startTemp.size = 0;
128 2 : startTemp.offset = 0;
129 2 : startSlice.push_back(startTemp);
130 : }
131 1 : return HCCL_SUCCESS;
132 : }
133 :
134 :
135 0 : HcclResult AllReduceLocalReduce::PrepareAllreduceSliceData()
136 : {
137 0 : return PrepareSlice(count_, DataUnitSize(dataType_), localRankSize_, slices_, startOffset);
138 : }
139 :
140 : // ringallreduce算法的函数入口
141 0 : HcclResult AllReduceLocalReduce::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK> &links)
142 : {
143 0 : HcclResult ret = HCCL_SUCCESS;
144 0 : CHK_SMART_PTR_NULL(dispatcher_);
145 0 : CHK_PTR_NULL(stream_.ptr());
146 0 : HCCL_INFO("AllReduceLocalReduce run: rank[%u] ranksize[%u] inputMem[%p] outputMem[%p] count[%llu]",
147 : rank, rankSize, inputMem_.ptr(), outputMem_.ptr(), count_);
148 :
149 0 : if (links.size() < rankSize) {
150 0 : HCCL_ERROR("[AllReduceLocalReduce][RunAsync]rank[%u] linksize[%llu] is less than rankSize[%u]",
151 : rank, links.size(), rankSize);
152 0 : return HCCL_E_INTERNAL;
153 : }
154 :
155 : // 如果ranksize为1, inline reduce和普通跨片reduce操作一致，从input->output
156 0 : if (rankSize == 1) {
157 0 : if (opInfo_->inputAddr != opInfo_->outputAddr) {
158 0 : DeviceMem userMemIn = DeviceMem::create(opInfo_->inputAddr, count_ * DataUnitSize(dataType_));
159 0 : DeviceMem userMemOut = DeviceMem::create(opInfo_->outputAddr, count_ * DataUnitSize(dataType_));
160 0 : ret = HcclD2DMemcpyAsync(dispatcher_, userMemOut, userMemIn, stream_);
161 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
162 : HCCL_ERROR("[AllReduceLocalReuce][RunAsync]rank[%u] memcpy async failed", rank), ret);
163 0 : }
164 0 : return ret;
165 : }
166 :
167 0 : ret = PrepareAllreduceSliceData();
168 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
169 : HCCL_ERROR("[AllReduceLocalReuce][RunAsync]rank[%u] count[%llu] failed in PrepareSliceData step",
170 : rank, count_),
171 : ret);
172 :
173 0 : ret = RunReduceScatter(rank, rankSize, links);
174 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
175 : HCCL_ERROR("[AllReduceLocalReuce][RunAsync]rank[%u] count[%llu] failed in reducescater step",
176 : rank, count_),
177 : ret);
178 :
179 0 : ret = RunAllGather(rank, rankSize, links);
180 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
181 : HCCL_ERROR("[AllReduceLocalReuce][RunAsync]rank[%u] count[%llu] failed in AllGather step",
182 : rank, count_),
183 : ret);
184 :
185 0 : HCCL_INFO("AllReduceLocalReduce finished: rank[%u] ranksize[%u]", rank, rankSize);
186 0 : return HCCL_SUCCESS;
187 : }
188 :
189 0 : HcclResult AllReduceLocalReduce::RunReduceScatter(u32 rank, u32 rankSize, const std::vector<LINK> &links)
190 : {
191 0 : HCCL_INFO("ReduceScatterMeshLocalReduce run: rank[%u] totalrank[%u] inputMem[%p] outputMem[%p] count[%llu]",
192 : rank, rankSize, inputMem_.ptr(), outputMem_.ptr(), count_);
193 :
194 : // 数据准备
195 0 : u32 unitSize = DataUnitSize(dataType_);
196 0 : DeviceMem userMemIn = DeviceMem::create(opInfo_->inputAddr, count_ * unitSize);
197 0 : DeviceMem commMemOut = DeviceMem::create(outputMem_.ptr(), outputMem_.size());
198 :
199 0 : DeviceMem src;
200 0 : DeviceMem dst;
201 :
202 0 : src = DeviceMem::create(static_cast<char *>(opInfo_->inputAddr) + slices_[rank].offset, slices_[rank].size);
203 0 : dst = commMemOut.range(slices_[rank].offset, slices_[rank].size);
204 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_));
205 :
206 0 : DeviceMem emptySrc = userMemIn.range(0, 0);
207 0 : DeviceMem emptyDst = commMemOut.range(0, 0);
208 :
209 0 : CHK_RET(MainRecordSub());
210 0 : CHK_RET(SubWaitMain());
211 :
212 0 : for (u32 round = 1; round < rankSize; round++) {
213 0 : u32 dstRank = (round + rank) % rankSize;
214 0 : Stream &subStream = (round == rankSize - 1) ? stream_ : meshStreams_[round - 1];
215 :
216 0 : CHK_RET(links[dstRank]->TxAck(subStream));
217 0 : CHK_RET(links[dstRank]->RxAck(subStream));
218 : }
219 0 : HCCL_DEBUG("[ReduceScatterMeshLocalReduce] D2DMemcpy start");
220 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, emptyDst, emptySrc, stream_));
221 :
222 0 : CHK_RET(SubRecordMain());
223 0 : CHK_RET(MainWaitSub());
224 :
225 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, emptyDst, emptySrc, stream_));
226 :
227 0 : CHK_RET(MainRecordSub());
228 0 : CHK_RET(SubWaitMain());
229 :
230 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, emptyDst, emptySrc, stream_));
231 :
232 0 : for (u32 round = 1; round < rankSize; round++) {
233 0 : Stream &subStream = (round == rankSize - 1) ? stream_ : meshStreams_[round - 1];
234 0 : void *remMemPtr = nullptr;
235 0 : u32 dstRank = (rank + round) % rankSize;
236 0 : u32 dstSlice = (dstRank + round) % (rankSize - 1);
237 0 : if (dstRank == (rankSize - 1)) {
238 0 : dstSlice = (dstSlice + rankSize - 1 - 1) % (rankSize - 1);
239 : }
240 0 : if (round == (rankSize - 1)) {
241 0 : CHK_RET(links[dstRank]->GetRemoteMem(UserMemType::OUTPUT_MEM, &remMemPtr));
242 :
243 0 : dstSlice = (dstRank == (rankSize - 1)) ? (dstRank - 1) : dstRank;
244 0 : dst = DeviceMem::create(
245 0 : static_cast<char *>(remMemPtr) + startOffset[dstRank].offset + dstSlice * startOffset[dstRank].size,
246 0 : slices_[dstRank].size);
247 0 : src = userMemIn.range(slices_[dstRank].offset, slices_[dstRank].size);
248 :
249 0 : HCCL_INFO("AllReducelocalreduce reduce dst offset1 %llu offset2 %llu size %llu, rank %u, dstrank %u",
250 : startOffset[dstRank].offset, dstSlice * startOffset[dstRank].size,
251 : slices_[dstRank].size, rank, dstRank);
252 :
253 0 : HCCL_INFO("AllReducelocalreduce reduce src offset %llu size %llu, rank %u, dstrank %u",
254 : slices_[dstRank].offset, slices_[dstRank].size, rank, dstRank);
255 :
256 0 : CHK_RET(HcclReduceAsync(dispatcher_, static_cast<void *>(src.ptr()),
257 : slices_[dstRank].size / unitSize, dataType_, reductionOp_, subStream,
258 : static_cast<void *>(dst.ptr()), links[dstRank]->GetRemoteRank(), links[dstRank]->GetLinkType(),
259 : INLINE_REDUCE_BIT));
260 : } else {
261 0 : CHK_RET(links[dstRank]->GetRemoteMem(UserMemType::OUTPUT_MEM, &remMemPtr));
262 :
263 0 : dst = DeviceMem::create(
264 0 : static_cast<char *>(remMemPtr) + startOffset[dstRank].offset + dstSlice * startOffset[dstRank].size,
265 0 : slices_[dstRank].size);
266 0 : src = userMemIn.range(slices_[dstRank].offset, slices_[dstRank].size);
267 :
268 0 : HCCL_INFO("AllReducelocalreduce memcpy dst offset1 %llu offset2 %llu size %llu, rank %u, dstrank %u",
269 : startOffset[dstRank].offset, dstSlice * startOffset[dstRank].size,
270 : slices_[dstRank].size, rank, dstRank);
271 :
272 0 : HCCL_INFO("AllReducelocalreduce memcpy src offset %llu size %llu, rank %u, dstrank %u",
273 : slices_[dstRank].offset, slices_[dstRank].size, rank, dstRank);
274 :
275 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, subStream,
276 : links[dstRank]->GetRemoteRank(), links[dstRank]->GetLinkType()));
277 : }
278 :
279 0 : CHK_RET(links[dstRank]->TxDataSignal(subStream));
280 0 : CHK_RET(links[dstRank]->RxDataSignal(subStream));
281 : }
282 :
283 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, emptyDst, emptySrc, stream_));
284 :
285 0 : CHK_RET(SubRecordMain());
286 0 : CHK_RET(MainWaitSub());
287 :
288 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, emptyDst, emptySrc, stream_));
289 :
290 0 : HcclResult ret = HCCL_SUCCESS;
291 0 : ret = RunLocalReduce(rank, rankSize);
292 :
293 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduclocalReduce]rank[%u] ReduceScatter failed", rank), ret);
294 0 : return HCCL_SUCCESS;
295 0 : }
296 :
297 0 : HcclResult AllReduceLocalReduce::RunLocalReduce(u32 rank, u32 rankSize)
298 : {
299 0 : DeviceMem commMemOut = DeviceMem::create(outputMem_.ptr(), outputMem_.size());
300 0 : u32 power = static_cast<u32>(log2(rankSize - 1));
301 0 : u32 rankPower = static_cast<u32>(pow(2, power));
302 0 : u32 unitSize = SIZE_TABLE[dataType_];
303 0 : u64 align = startOffset[rank].size;
304 0 : u64 totalSize = slices_[rank].size;
305 0 : DeviceMem src;
306 0 : DeviceMem dst;
307 0 : for(u32 i = 0u; i < rankSize - rankPower - 1; ++i){
308 0 : u64 size = totalSize;
309 0 : if (rank < rankPower) {
310 0 : src = commMemOut.range(startOffset[rank].offset + (rankPower + i) * align, size);
311 0 : dst = commMemOut.range(startOffset[rank].offset + i * align, size);
312 0 : HCCL_INFO("[RunLocalReduce]LocalReduce rank[%u] src[%llu], dst[%llu] size[%llu]", rank, startOffset[rank].offset + (rankPower + i) * align,
313 : startOffset[rank].offset + i * align, size);
314 : } else {
315 0 : dst = commMemOut.range(startOffset[rank].offset + (rankPower + i) * align, size);
316 0 : src = commMemOut.range(startOffset[rank].offset + i * align, size);
317 0 : HCCL_INFO("[RunLocalReduce]LocalReduce rank[%u] src[%llu], dst[%llu] size[%llu]", rank, startOffset[rank].offset + i * align,
318 : startOffset[rank].offset + (rankPower + i) * align, size);
319 : }
320 0 : CHK_RET(HcclReduceAsync(dispatcher_, static_cast<void *>(src.ptr()),
321 : size / unitSize, dataType_, reductionOp_, stream_, static_cast<void *>(dst.ptr()),
322 : INVALID_VALUE_RANKID, LinkType::LINK_ONCHIP, INLINE_REDUCE_BIT));
323 : }
324 0 : u32 center = rank < rankPower ? rank : (rank - rankPower + 1);
325 0 : center = std::min(center, rankPower - 1);
326 0 : u64 offset = rank < rankPower ? 0 : ((rankSize - rankPower - 1) * align);
327 0 : offset += startOffset[rank].offset;
328 0 : for (u32 round = 0; round < power; round++) {
329 0 : u32 slices_num = static_cast<u32>(rankPower / pow(2, round + 1));
330 0 : u64 size = totalSize;
331 0 : if (center < slices_num) {
332 0 : for(auto i = 0u; i < slices_num; ++i){
333 0 : src = commMemOut.range(offset + (slices_num + i) * align, size);
334 0 : dst = commMemOut.range(offset + i * align, size);
335 0 : HCCL_INFO("[RunLocalReduce]LocalReduce rank[%u] src[%llu], dst[%llu] size[%llu]", rank, offset + (slices_num + i) * align,
336 : offset + i * align, size);
337 0 : CHK_RET(HcclReduceAsync(dispatcher_, static_cast<void *>(src.ptr()),
338 : src.size()/unitSize,
339 : dataType_,
340 : reductionOp_,
341 : stream_,
342 : static_cast<void *>(dst.ptr()), INVALID_VALUE_RANKID, LinkType::LINK_ONCHIP, INLINE_REDUCE_BIT));
343 : }
344 : } else {
345 0 : for(auto i = 0u; i < slices_num; ++i){
346 0 : dst = commMemOut.range(offset + (slices_num + i) * align, size);
347 0 : src = commMemOut.range(offset + i * align, size);
348 0 : HCCL_INFO("[RunLocalReduce]LocalReduce rank[%u] src[%llu], dst[%llu] size[%llu]", rank, offset + i * align,
349 : offset + (slices_num + i) * align, size);
350 0 : CHK_RET(HcclReduceAsync(dispatcher_, static_cast<void *>(src.ptr()),
351 : src.size()/unitSize,
352 : dataType_,
353 : reductionOp_,
354 : stream_,
355 : static_cast<void *>(dst.ptr()), INVALID_VALUE_RANKID, LinkType::LINK_ONCHIP, INLINE_REDUCE_BIT));
356 : }
357 0 : offset = offset + slices_num * align;
358 0 : center -= slices_num;
359 : }
360 : }
361 0 : return HCCL_SUCCESS;
362 0 : }
363 :
364 0 : HcclResult AllReduceLocalReduce::RunAllGather(u32 rank, u32 rankSize, const std::vector<LINK> &links)
365 : {
366 0 : HCCL_INFO("AllGatherMesh run: rank[%u] totalrank[%u] inputMem[%p] outputMem[%p] count[%llu]",
367 : rank, rankSize, inputMem_.ptr(), outputMem_.ptr(), count_);
368 0 : u32 unitSize = DataUnitSize(dataType_);
369 0 : DeviceMem userMemOut = DeviceMem::create(opInfo_->outputAddr, count_ * unitSize);
370 0 : DeviceMem commMemOut = DeviceMem::create(outputMem_.ptr(), outputMem_.size());
371 :
372 0 : DeviceMem src;
373 0 : DeviceMem dst;
374 :
375 0 : DeviceMem emptySrc = commMemOut.range(0, 0);
376 0 : DeviceMem emptyDst = userMemOut.range(0, 0);
377 :
378 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, emptyDst, emptySrc, stream_));
379 :
380 0 : CHK_RET(MainRecordSub());
381 0 : CHK_RET(SubWaitMain());
382 :
383 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, emptyDst, emptySrc, stream_));
384 :
385 0 : for (u32 round = 1; round < rankSize; round++) {
386 0 : u32 dstRank = BackwardRank(rank, rankSize, round);
387 0 : Stream &subStream = (round == rankSize - 1) ? stream_ : meshStreams_[round - 1];
388 0 : CHK_RET(links[dstRank]->TxAck(subStream));
389 0 : CHK_RET(links[dstRank]->RxAck(subStream));
390 : }
391 :
392 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, emptyDst, emptySrc, stream_));
393 :
394 0 : CHK_RET(SubRecordMain());
395 0 : CHK_RET(MainWaitSub());
396 :
397 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, emptyDst, emptySrc, stream_));
398 :
399 0 : CHK_RET(MainRecordSub());
400 0 : CHK_RET(SubWaitMain());
401 :
402 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, emptyDst, emptySrc, stream_));
403 :
404 0 : if (userMemOut.ptr() != commMemOut.ptr()) {
405 0 : src = commMemOut.range(slices_[rank].offset, slices_[rank].size);
406 0 : dst = userMemOut.range(slices_[rank].offset, slices_[rank].size);
407 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, meshStreams_[meshStreams_.size() - 1]));
408 : }
409 :
410 0 : for (u32 round = 1; round < rankSize; round++) {
411 0 : u32 dstRank = BackwardRank(rank, rankSize, round);
412 0 : Stream &subStream = (round == rankSize - 1) ? stream_ : meshStreams_[round - 1];
413 0 : void *remMemPtr = nullptr;
414 0 : CHK_RET(links[dstRank]->GetRemoteMem(UserMemType::OUTPUT_MEM, &remMemPtr));
415 :
416 0 : src = DeviceMem::create(static_cast<char *>(remMemPtr) + dstRank * slices_[0].size, slices_[dstRank].size);
417 0 : dst = userMemOut.range(slices_[dstRank].offset, slices_[dstRank].size);
418 :
419 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, subStream,
420 : links[dstRank]->GetRemoteRank(), links[dstRank]->GetLinkType()));
421 0 : CHK_RET(links[dstRank]->TxDataSignal(subStream));
422 0 : CHK_RET(links[dstRank]->RxDataSignal(subStream));
423 : }
424 :
425 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, emptyDst, emptySrc, stream_));
426 :
427 0 : CHK_RET(SubRecordMain());
428 0 : CHK_RET(MainWaitSub());
429 :
430 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, emptyDst, emptySrc, stream_));
431 :
432 0 : HCCL_INFO("AllGatherMesh finished: rank[%u]", rank);
433 0 : return HCCL_SUCCESS;
434 0 : }
435 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_REDUCE_LOCAL_REDUCE, AllReduceLocalReduce);
436 : } // namespace hccl
|
