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