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