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_bcast.h"
14 :
15 : namespace hccl {
16 0 : AllReduceLocalReduceBcast::AllReduceLocalReduceBcast(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher) {}
17 :
18 0 : AllReduceLocalReduceBcast::~AllReduceLocalReduceBcast() {}
19 :
20 0 : HcclResult AllReduceLocalReduceBcast::Prepare(
21 : u64 reduceAttrBitMap, std::vector<Stream>& meshStreams, std::vector<std::shared_ptr<LocalNotify>>& meshSignal,
22 : std::vector<std::shared_ptr<LocalNotify>>& meshSignalAux, u32 interRank, u32 interRankSize, u32 userRank,
23 : HcomCollOpInfo* opInfo)
24 : {
25 0 : reduceAttr_ = reduceAttrBitMap;
26 0 : localRank_ = interRank;
27 0 : localRankSize_ = interRankSize;
28 0 : userRank_ = userRank;
29 0 : meshStreams_ = meshStreams;
30 0 : meshSignal_ = &meshSignal;
31 0 : meshSignalAux_ = &meshSignalAux;
32 0 : opInfo_ = opInfo;
33 0 : return HCCL_SUCCESS;
34 : }
35 :
36 0 : HcclResult AllReduceLocalReduceBcast::MainRecordSub(u32 streamNum)
37 : {
38 0 : if (streamNum == 0) {
39 0 : for (u32 signalIndex = 0; signalIndex < meshSignalAux_->size(); signalIndex++) {
40 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, (*meshSignalAux_)[signalIndex], profilerInput_.stage));
41 : }
42 : } else {
43 0 : for (u32 signalIndex = 0; signalIndex < streamNum; signalIndex++) {
44 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, (*meshSignalAux_)[signalIndex], profilerInput_.stage));
45 : }
46 : }
47 0 : return HCCL_SUCCESS;
48 : }
49 :
50 0 : HcclResult AllReduceLocalReduceBcast::SubWaitMain(u32 streamNum)
51 : {
52 0 : if (streamNum == 0) {
53 0 : for (u32 streamIndex = 0; streamIndex < meshSignalAux_->size(); streamIndex++) {
54 0 : CHK_RET(LocalNotify::Wait(
55 : meshStreams_[streamIndex], dispatcher_, (*meshSignalAux_)[streamIndex], profilerInput_.stage));
56 : }
57 : } else {
58 0 : for (u32 streamIndex = 0; streamIndex < streamNum; streamIndex++) {
59 0 : CHK_RET(LocalNotify::Wait(
60 : meshStreams_[streamIndex], dispatcher_, (*meshSignalAux_)[streamIndex], profilerInput_.stage));
61 : }
62 : }
63 0 : return HCCL_SUCCESS;
64 : }
65 :
66 0 : HcclResult AllReduceLocalReduceBcast::MainWaitSub(u32 streamNum)
67 : {
68 0 : if (streamNum == 0) {
69 0 : for (u32 signalIndex = 0; signalIndex < meshSignal_->size(); signalIndex++) {
70 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, (*meshSignal_)[signalIndex], profilerInput_.stage));
71 : }
72 : } else {
73 0 : for (u32 signalIndex = 0; signalIndex < streamNum; signalIndex++) {
74 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, (*meshSignal_)[signalIndex], profilerInput_.stage));
75 : }
76 : }
77 0 : return HCCL_SUCCESS;
78 : }
79 :
80 0 : HcclResult AllReduceLocalReduceBcast::SubRecordMain(u32 streamNum)
81 : {
82 0 : if (streamNum == 0) {
83 0 : for (u32 streamIndex = 0; streamIndex < meshSignal_->size(); streamIndex++) {
84 0 : CHK_RET(LocalNotify::Post(
85 : meshStreams_[streamIndex], dispatcher_, (*meshSignal_)[streamIndex], profilerInput_.stage));
86 : }
87 : } else {
88 0 : for (u32 streamIndex = 0; streamIndex < streamNum; streamIndex++) {
89 0 : CHK_RET(LocalNotify::Post(
90 : meshStreams_[streamIndex], dispatcher_, (*meshSignal_)[streamIndex], profilerInput_.stage));
91 : }
92 : }
93 0 : return HCCL_SUCCESS;
94 : }
95 :
96 : // 将数据均分，最小单位是128
97 :
98 : // ringallreduce算法的函数入口
99 0 : HcclResult AllReduceLocalReduceBcast::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
100 : {
101 0 : HcclResult ret = HCCL_SUCCESS;
102 0 : CHK_SMART_PTR_NULL(dispatcher_);
103 0 : CHK_PTR_NULL(stream_.ptr());
104 0 : HCCL_INFO(
105 : "AllReduceLocalReduceBcast run: rank[%u] ranksize[%u] inputMem[%p] outputMem[%p] count[%llu]", rank, rankSize,
106 : inputMem_.ptr(), outputMem_.ptr(), count_);
107 :
108 0 : CHK_PRT_RET(
109 : links.size() < rankSize,
110 : HCCL_ERROR(
111 : "[AllReduceLocalReduceBcast][RunAsync]rank[%u] linksize[%llu] is less than rankSize[%u]", rank,
112 : links.size(), rankSize),
113 : HCCL_E_INTERNAL);
114 :
115 : // 如果ranksize为1, 从input->output
116 0 : if (rankSize == 1) {
117 0 : if (inputMem_ != outputMem_) {
118 0 : DeviceMem userMemIn = DeviceMem::create(inputMem_.ptr(), count_ * DataUnitSize(dataType_));
119 0 : DeviceMem userMemOut = DeviceMem::create(outputMem_.ptr(), count_ * DataUnitSize(dataType_));
120 0 : ret = HcclD2DMemcpyAsync(dispatcher_, userMemOut, userMemIn, stream_);
121 0 : CHK_PRT_RET(
122 : ret != HCCL_SUCCESS,
123 : HCCL_ERROR("[AllReduceLocalReduceBcast][RunAsync]rank[%u] memcpy async failed", rank), ret);
124 0 : }
125 0 : return ret;
126 : }
127 :
128 0 : ret = RunReduce(rank, rankSize, links);
129 0 : CHK_PRT_RET(
130 : ret != HCCL_SUCCESS,
131 : HCCL_ERROR("[AllReduceLocalReduceBcast][RunAsync]rank[%u] count[%llu] failed in Reduce step", rank, count_),
132 : ret);
133 :
134 0 : ret = RunBroadcast(rank, rankSize, links);
135 0 : CHK_PRT_RET(
136 : ret != HCCL_SUCCESS,
137 : HCCL_ERROR(
138 : "[AllReduceLocalReduceBcast][RunAsync]rank[%u] count[%llu] failed in Broadcast "
139 : "step",
140 : rank, count_),
141 : ret);
142 :
143 0 : HCCL_INFO("AllReduceLocalReduceBcast finished: rank[%u] ranksize[%u].", rank, rankSize);
144 0 : return HCCL_SUCCESS;
145 : }
146 :
147 0 : HcclResult AllReduceLocalReduceBcast::RunReduce(u32 rank, u32 rankSize, const std::vector<LINK>& links)
148 : {
149 0 : HCCL_INFO("AllReduceLocalReduceBcast RunReduce: rank[%u] totalrank[%u] count[%llu].", rank, rankSize, count_);
150 :
151 0 : u32 unitSize = SIZE_TABLE[dataType_];
152 :
153 0 : DeviceMem userMemIn = DeviceMem::create(opInfo_->inputAddr, count_ * unitSize);
154 0 : DeviceMem commMemOut = DeviceMem::create(outputMem_.ptr(), outputMem_.size());
155 :
156 0 : if (rank == 0) {
157 0 : DeviceMem src = userMemIn.range(0, count_ * unitSize);
158 0 : DeviceMem dst = commMemOut.range(0, count_ * unitSize);
159 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_));
160 0 : }
161 :
162 : // 数据准备
163 : HcclResult ret;
164 0 : if (rank == 0) {
165 0 : CHK_RET(RunAllReduceBDReduceReceive(rank, 0, links));
166 0 : ret = RunLocalReduce(rank, rankSize);
167 : } else {
168 0 : ret = RunAllReduceBDReduceSend(rank, 0, links);
169 : }
170 0 : CHK_PRT_RET(
171 : ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceLocalReduceBcastReduce]rank[%u] RunAllReduceBDReduce failed", rank),
172 : ret);
173 0 : return HCCL_SUCCESS;
174 0 : }
175 :
176 0 : HcclResult AllReduceLocalReduceBcast::RunBroadcast(u32 rank, u32 rankSize, const std::vector<LINK>& links)
177 : {
178 0 : HCCL_INFO("AllReduceLocalReduceBcast RunBroadcast: rank[%u] totalrank[%u] count[%llu].", rank, rankSize, count_);
179 :
180 : HcclResult ret;
181 0 : if (rank == 0) {
182 0 : ret = RunAllReduceBDMemcpySend(rank, 0, links);
183 : } else {
184 0 : ret = RunAllReduceBDMemcpyReceive(rank, 0, links);
185 : }
186 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceLocalReduceBcast]rank[%u] RunBroadcast failed", rank), ret);
187 :
188 0 : HCCL_INFO("AllReduceLocalReduceBcast RunBroadcast: rank[%u]", rank);
189 0 : return HCCL_SUCCESS;
190 : }
191 :
192 0 : HcclResult AllReduceLocalReduceBcast::RunAllReduceBDReduceSend(u32 rank, u32 peer, const std::vector<LINK>& links)
193 : {
194 0 : HCCL_INFO("AllReduceLocalReduceBcast RunAllReduceBDReduceSend: rank[%u] peer[%u] count[%llu].", rank, peer, count_);
195 :
196 : // 数据准备
197 0 : u32 unitSize = SIZE_TABLE[dataType_];
198 0 : u32 totalSize = count_ * unitSize;
199 :
200 0 : DeviceMem userMemIn = DeviceMem::create(opInfo_->inputAddr, count_ * unitSize);
201 :
202 0 : CHK_RET(links[peer]->RxAck(stream_));
203 :
204 0 : void* remMemPtr = nullptr;
205 0 : CHK_RET(links[peer]->GetRemoteMem(UserMemType::OUTPUT_MEM, &remMemPtr));
206 :
207 0 : DeviceMem src = userMemIn;
208 0 : DeviceMem dst = DeviceMem::create(static_cast<char*>(remMemPtr) + (rank - 1) * totalSize, totalSize);
209 :
210 0 : if (rank != 1) {
211 0 : CHK_RET(HcclD2DMemcpyAsync(
212 : dispatcher_, dst, src, stream_, links[peer]->GetRemoteRank(), links[peer]->GetLinkType()));
213 : } else {
214 0 : CHK_RET(HcclReduceAsync(
215 : dispatcher_, static_cast<void*>(src.ptr()), count_, dataType_, reductionOp_, stream_,
216 : static_cast<void*>(dst.ptr()), links[peer]->GetRemoteRank(), links[peer]->GetLinkType(),
217 : INLINE_REDUCE_BIT));
218 : }
219 :
220 0 : CHK_RET(links[peer]->TxDataSignal(stream_));
221 0 : return HCCL_SUCCESS;
222 0 : }
223 :
224 0 : HcclResult AllReduceLocalReduceBcast::RunAllReduceBDReduceReceive(u32 rank, u32 peer, const std::vector<LINK>& links)
225 : {
226 0 : HCCL_INFO(
227 : "AllReduceLocalReduceBcast RunAllReduceBDReduceReceive: rank[%u] peer[%u] count[%llu].", rank, peer, count_);
228 :
229 0 : CHK_RET(MainRecordSub());
230 0 : CHK_RET(SubWaitMain());
231 :
232 0 : for (u32 round = 1; round < localRankSize_; round++) {
233 0 : Stream& subStream = (round == localRankSize_ - 1) ? stream_ : meshStreams_[round - 1];
234 0 : CHK_RET(links[round]->TxAck(subStream));
235 0 : CHK_RET(links[round]->RxDataSignal(subStream));
236 : }
237 :
238 0 : CHK_RET(SubRecordMain());
239 0 : CHK_RET(MainWaitSub());
240 0 : HCCL_DEBUG("[AllReduceLocalReduceBcast]RunAllReduceBDReduceReceive success");
241 0 : return HCCL_SUCCESS;
242 : }
243 :
244 0 : HcclResult AllReduceLocalReduceBcast::RunLocalReduce(u32 rank, u32 rankSize)
245 : {
246 : (void)rank;
247 0 : DeviceMem commMemOut = DeviceMem::create(outputMem_.ptr(), outputMem_.size());
248 0 : u32 power = static_cast<u32>(log2(rankSize - 1));
249 0 : u32 rankPower = static_cast<u32>(pow(2, power));
250 0 : u32 unitSize = SIZE_TABLE[dataType_];
251 0 : u32 totalSize = count_ * unitSize;
252 0 : DeviceMem src;
253 0 : DeviceMem dst;
254 0 : if (rankPower < rankSize - 1) {
255 0 : src = commMemOut.range(rankPower * totalSize, (rankSize - rankPower - 1) * totalSize);
256 0 : dst = commMemOut.range(0, (rankSize - rankPower - 1) * totalSize);
257 0 : CHK_RET(HcclReduceAsync(
258 : dispatcher_, static_cast<void*>(src.ptr()), count_ * (rankSize - rankPower - 1), dataType_, reductionOp_,
259 : stream_, static_cast<void*>(dst.ptr()), INVALID_VALUE_RANKID, LinkType::LINK_ONCHIP, INLINE_REDUCE_BIT));
260 : }
261 0 : for (u32 round = 0; round < power; round++) {
262 0 : u32 sliceNum = rankPower / static_cast<u32>(pow(2, round + 1));
263 0 : src = commMemOut.range(sliceNum * totalSize, sliceNum * totalSize);
264 0 : dst = commMemOut.range(0, sliceNum * totalSize);
265 0 : CHK_RET(HcclReduceAsync(
266 : dispatcher_, static_cast<void*>(src.ptr()), count_ * sliceNum, dataType_, reductionOp_, stream_,
267 : static_cast<void*>(dst.ptr()), INVALID_VALUE_RANKID, LinkType::LINK_ONCHIP, INLINE_REDUCE_BIT));
268 : }
269 0 : return HCCL_SUCCESS;
270 0 : }
271 :
272 0 : HcclResult AllReduceLocalReduceBcast::RunAllReduceBDMemcpyReceive(u32 rank, u32 peer, const std::vector<LINK>& links)
273 : {
274 0 : HCCL_INFO(
275 : "AllReduceLocalReduceBcast RunAllReduceBDMemcpyReceive: rank[%u] peer[%u] count[%llu]", rank, peer, count_);
276 0 : u32 unitSize = SIZE_TABLE[dataType_];
277 :
278 0 : CHK_RET(links[peer]->RxAck(stream_));
279 0 : DeviceMem userMemOut = DeviceMem::create(opInfo_->outputAddr, count_ * unitSize);
280 :
281 0 : u32 totalSize = count_ * unitSize;
282 :
283 0 : void* remMemPtr = nullptr;
284 0 : CHK_RET(links[peer]->GetRemoteMem(UserMemType::OUTPUT_MEM, &remMemPtr));
285 0 : DeviceMem src;
286 :
287 0 : src = DeviceMem::create(static_cast<char*>(remMemPtr), totalSize);
288 :
289 0 : DeviceMem dst = userMemOut;
290 0 : CHK_RET(
291 : HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_, links[peer]->GetRemoteRank(), links[peer]->GetLinkType()));
292 0 : CHK_RET(links[peer]->TxDataSignal(stream_));
293 :
294 0 : HCCL_INFO("AllReduceLocalReduceBcast RunAllReduceBDMemcpyReceive finished: rank[%u]", rank);
295 0 : return HCCL_SUCCESS;
296 0 : }
297 :
298 0 : HcclResult AllReduceLocalReduceBcast::RunAllReduceBDMemcpySend(u32 rank, u32 peer, const std::vector<LINK>& links)
299 : {
300 0 : HCCL_INFO("AllReduceLocalReduceBcast RunAllReduceBDMemcpySend: rank[%u] peer[%u] count[%llu]", rank, peer, count_);
301 :
302 0 : u32 unitSize = SIZE_TABLE[dataType_];
303 0 : u32 totalSize = count_ * unitSize;
304 :
305 0 : if (opInfo_->outputAddr != outputMem_.ptr()) {
306 0 : DeviceMem dst = DeviceMem::create(opInfo_->outputAddr, totalSize);
307 0 : DeviceMem src = DeviceMem::create(outputMem_.ptr(), totalSize);
308 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_));
309 0 : }
310 0 : CHK_RET(MainRecordSub());
311 0 : CHK_RET(SubWaitMain());
312 :
313 0 : for (u32 round = 1; round < localRankSize_; round++) {
314 0 : Stream& subStream = (round == localRankSize_ - 1) ? stream_ : meshStreams_[round - 1];
315 0 : CHK_RET(links[round]->TxAck(subStream));
316 0 : CHK_RET(links[round]->RxDataSignal(subStream));
317 : }
318 :
319 0 : CHK_RET(SubRecordMain());
320 0 : CHK_RET(MainWaitSub());
321 :
322 0 : HCCL_INFO("AllReduceLocalReduceBcast RunAllReduceBDMemcpySend finished: rank[%u]", rank);
323 0 : return HCCL_SUCCESS;
324 : }
325 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_REDUCE_LOCAL_REDUCE_BCAST, AllReduceLocalReduceBcast);
326 : } // namespace hccl
|
