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