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 "alg_template_register.h"
12 : #include "all_reduce_nhr_v1.h"
13 :
14 : namespace hccl {
15 0 : AllReduceNHRV1::AllReduceNHRV1(const HcclDispatcher dispatcher) : NHRV1Base(dispatcher)
16 : {
17 0 : }
18 :
19 0 : AllReduceNHRV1::~AllReduceNHRV1()
20 : {
21 0 : }
22 :
23 0 : HcclResult AllReduceNHRV1::Prepare(u64 reduceAttrBitMap, HcomCollOpInfo *opInfo)
24 : {
25 0 : reduceAttr_ = reduceAttrBitMap;
26 0 : return HCCL_SUCCESS;
27 : }
28 :
29 0 : HcclResult AllReduceNHRV1::RunAsync(const u32 rank, const u32 rankSize,
30 : const std::vector<std::shared_ptr<Transport> > &links)
31 : {
32 0 : CHK_RET(PrepareRunAsync(rank, rankSize, links));
33 0 : CHK_PRT_RET(rankSize == 1, HCCL_INFO("[AllReduceNHRV1][RunAsync] rankSize[%u], do nothing.",
34 : rankSize), HCCL_SUCCESS);
35 :
36 0 : HcclResult ret = HCCL_SUCCESS;
37 : // 获取通信关系
38 0 : RingInfo info = GetRingInfo(rankSize);
39 : // 水平方向做broken reducescatter ring
40 :
41 0 : ret = RunReduceScatterOnHorizontal(rank, links, info);
42 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceNHRV1][RunAsync]rank[%u] count[%llu] failed in "\
43 : "RunReduceScatterOnHorizontal step", rank, count_), ret);
44 :
45 : // 垂直方向做allreduce ring
46 0 : ret = RunAllReduceOnVertical(rank, links, info);
47 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceNHRV1][RunAsync]rank[%u] count[%llu] failed in "\
48 : "RunAllReduceOnVertical step", rank, count_), ret);
49 :
50 : // 水平方向做broken allgather ring
51 0 : ret = RunAllGatherOnHorizontal(rank, links, info);
52 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceNHRV1][RunAsync]rank[%u] count[%llu] failed in "\
53 : "RunAllGatherOnHorizontal step", rank, count_), ret);
54 :
55 0 : HCCL_INFO("AllReduceNHRV1 finished: rank[%u]", rank);
56 0 : return HCCL_SUCCESS;
57 0 : }
58 :
59 0 : HcclResult AllReduceNHRV1::RunAsyncStaged(const u32 rank, const u32 rankSize, const std::vector<LINK> &links,
60 : RunStage stage)
61 : {
62 0 : CHK_PRT_RET(rankSize == 1 && stage != RunStage::RUN_PREPARE,
63 : HCCL_INFO("[AllReduceNHRV1][RunAsyncStaged] rankSize[%u], stage[%d], do nothing.",
64 : rankSize, stage), HCCL_SUCCESS);
65 : // 获取通信关系
66 0 : RingInfo info = GetRingInfo(rankSize);
67 :
68 0 : HcclResult ret = HCCL_SUCCESS;
69 0 : switch (stage) {
70 0 : case RunStage::RUN_PREPARE:
71 0 : ret = PrepareRunAsync(rank, rankSize, links);
72 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
73 : HCCL_ERROR("[AllReduceNHRV1][RunAsyncStaged]rank[%u] count[%llu] failed in PrepareRunAsync step",
74 : rank, count_), ret);
75 0 : break;
76 0 : case RunStage::RUN_REDUCE_SCATTER:
77 : // 水平方向做broken reducescatter ring
78 0 : ret = RunReduceScatterOnHorizontal(rank, links, info);
79 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceNHRV1][RunAsync]rank[%u] count[%llu] failed in "\
80 : "RunReduceScatterOnHorizontal step", rank, count_), ret);
81 0 : break;
82 0 : case RunStage::RUN_ALLREDUCE:
83 : // 垂直方向做allreduce ring
84 0 : ret = RunAllReduceOnVertical(rank, links, info);
85 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceNHRV1][RunAsync]rank[%u] count[%llu] failed in "\
86 : "RunAllReduceOnVertical step", rank, count_), ret);
87 0 : break;
88 0 : case RunStage::RUN_ALLGATHER:
89 : // 水平方向做broken allgather ring
90 0 : ret = RunAllGatherOnHorizontal(rank, links, info);
91 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceNHRV1][RunAsync]rank[%u] count[%llu] failed in "\
92 : "RunAllGatherOnHorizontal step", rank, count_), ret);
93 0 : break;
94 0 : default:
95 0 : HCCL_ERROR("[AllReduceNHRV1][RunAsyncStaged]stage[%d]is not support", stage);
96 0 : return HCCL_E_NOT_SUPPORT;
97 : }
98 0 : HCCL_INFO("AllReduceNHRV1 RunAsyncStaged stage[%d] finished: rank[%u] ranksize[%u]", stage, rank, rankSize);
99 0 : return HCCL_SUCCESS;
100 0 : }
101 :
102 0 : HcclResult AllReduceNHRV1::PrepareRunAsync(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 : CHK_PRT_RET(!outputMem_ || !inputMem_,
108 : HCCL_ERROR("[AllReduceNHRV1][RunAsync]rank[%u] run_async inputmem or outputmem is null", rank), HCCL_E_PTR);
109 :
110 0 : HCCL_INFO("AllReduceNHRV1 run: rank[%u] ranksize[%u] inputMem[%p] outputMem[%p] count[%llu]", \
111 : rank, rankSize, inputMem_.ptr(), outputMem_.ptr(), count_);
112 :
113 0 : CHK_PRT_RET(links.size() < rankSize, HCCL_ERROR("[AllReduceNHRV1][RunAsync]rank[%u] linksize[%llu] is less "\
114 : "than rankSize[%u]", rank, links.size(), rankSize), HCCL_E_INTERNAL);
115 :
116 : // 如果ranksize为1, inline reduce和普通跨片reduce操作一致，从input->output
117 0 : if (rankSize == 1) {
118 0 : if (inputMem_ != outputMem_) {
119 0 : ret = HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_);
120 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
121 : HCCL_ERROR("[AllReduceNHRV1][RunAsync]rank[%u] memcpy async failed", rank), ret);
122 : }
123 0 : return ret;
124 : }
125 :
126 : // 检查、并清空slices_
127 0 : if (slices_.size() != 0) {
128 0 : HCCL_WARNING("[AllReduceNHRV1][RunAsync]AllReduceNHRV1 not supported passing in parameter slice_, "\
129 : "otherwise will be cleared");
130 0 : slices_.clear();
131 : }
132 0 : return HCCL_SUCCESS;
133 : }
134 :
135 0 : HcclResult AllReduceNHRV1::CalcHSlicesAndLinks(const u32 rank, const std::vector<LINK> &links, const RingInfo &info,
136 : std::vector<LINK> &hLinks, std::vector<Slice> &hSlices)
137 : {
138 0 : u32 ringSize = info.GetHSizeByRank(rank); // 查找自己所处的行长度，也即Ring的大小
139 0 : u32 vIndex = info.GetVIndex(rank); // 查找自己位于第几行
140 :
141 : // 计算水平方向每个rank结果上的offset和size
142 0 : u64 sliceSizeCalculated = (count_+ (info.GetRowSize() - 1)) / info.GetRowSize() * DataUnitSize(dataType_);
143 0 : u64 totalSize = count_ * DataUnitSize(dataType_);
144 0 : u64 residueSize = totalSize;
145 0 : u64 sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSizeCalculated, HCCL_MIN_SLICE_ALIGN);
146 :
147 : // 水平方向都为broken ring，故最后一列有可能不需要参与计算，此时size为0
148 0 : for (u32 hIdx = 0; hIdx < ringSize; hIdx++) {
149 0 : u32 oldRank = info.GetRank(vIndex, hIdx);
150 :
151 0 : CHK_PRT_RET(oldRank >= links.size(), HCCL_ERROR("[AllReduceNHRV1] rank[%u] out of range, "\
152 : "oldRank=%u, links.size=%u", rank, oldRank, links.size()), HCCL_E_INTERNAL);
153 0 : hLinks.push_back(links[oldRank]);
154 0 : Slice slice;
155 0 : if (info.GetVSizeByHIndex(hIdx) == info.GetVSizeByHIndex(0)) {
156 0 : slice.size = (residueSize > sliceSizeAligned) ? sliceSizeAligned : residueSize;
157 0 : slice.offset = totalSize - residueSize;
158 0 : residueSize -= slice.size;
159 : } else {
160 0 : slice.size = 0;
161 0 : slice.offset = 0;
162 : }
163 0 : HCCL_DEBUG("[AllReduceNHRV1][CalcHSlicesAndLinks] rank[%u], slices[%u].offset=%llu, slices[%u].size=%llu",
164 : rank, hIdx, slice.offset, hIdx, slice.size);
165 0 : hSlices.push_back(slice);
166 : }
167 0 : return HCCL_SUCCESS;
168 : }
169 :
170 0 : HcclResult AllReduceNHRV1::CalcVSlicesAndLinks(const u32 rank, const std::vector<LINK> &links, const RingInfo &info,
171 : std::vector<LINK> &vLinks, std::vector<Slice> &vSlices)
172 : {
173 0 : u32 ringSize = info.GetVSizeByRank(rank); // 查找自己所处的列长度，也即Ring的大小
174 0 : u32 hIndex = info.GetHIndex(rank); // 查找自己位于第几列
175 :
176 0 : std::vector<Slice> hSlices;
177 0 : std::vector<LINK> hLinks;
178 0 : CHK_RET(CalcHSlicesAndLinks(rank, links, info, hLinks, hSlices));
179 :
180 : // 计算垂直方向每个rank结果上的offset和size
181 : u64 sliceSizeCalculated =
182 0 : (hSlices[hIndex].size / DataUnitSize(dataType_) + (ringSize - 1)) / ringSize * DataUnitSize(dataType_);
183 0 : u64 totalSize = hSlices[hIndex].size;
184 0 : u64 residueSize = totalSize;
185 0 : u64 sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSizeCalculated, HCCL_MIN_SLICE_ALIGN);
186 :
187 0 : for (u32 vIdx = 0; vIdx < ringSize; vIdx++) {
188 0 : u32 oldRank = info.GetRank(vIdx, hIndex);
189 0 : CHK_PRT_RET(oldRank >= links.size(), HCCL_ERROR("[AllReduceNHRV1] rank[%u] out of range, "\
190 : "oldRank=%u, links.size=%u", rank, oldRank, links.size()), HCCL_E_INTERNAL);
191 0 : vLinks.push_back(links[oldRank]);
192 0 : Slice slice;
193 0 : slice.size = (residueSize > sliceSizeAligned) ? sliceSizeAligned : residueSize;
194 0 : slice.offset = hSlices[hIndex].offset + totalSize - residueSize;
195 0 : residueSize -= slice.size;
196 0 : HCCL_DEBUG("[AllReduceNHRV1][CalcVSlicesAndLinks] rank[%u], slices[%u].offset=%llu, slices[%u].size=%llu",
197 : rank, vIdx, slice.offset, vIdx, slice.size);
198 0 : vSlices.push_back(slice);
199 : }
200 0 : return HCCL_SUCCESS;
201 0 : }
202 :
203 0 : HcclResult AllReduceNHRV1::RunReduceScatterOnHorizontal(const u32 rank, const std::vector<LINK> &links,
204 : const RingInfo &info)
205 : {
206 0 : u32 ringRank = info.GetHIndex(rank); // 查找自己位于第几列，也即处于Ring中的第几个rank
207 :
208 : // 计算reducescatter每个rank结果上的offset和size
209 0 : std::vector<Slice> hSlices;
210 0 : std::vector<LINK> hLinks;
211 0 : CHK_RET(CalcHSlicesAndLinks(rank, links, info, hLinks, hSlices));
212 :
213 : // 长度不足2，直接跳过
214 0 : if (hLinks.size() < 2) {
215 0 : return HCCL_SUCCESS;
216 : }
217 :
218 0 : HCCL_DEBUG("[AllReduceNHRV1][ReduceScatter-H] rank[%u] ringRank=%u, ringSize=%u", rank, ringRank, hLinks.size());
219 0 : return RunReduceScatterBrokenRing(ringRank, hLinks, hSlices);
220 0 : }
221 :
222 0 : HcclResult AllReduceNHRV1::RunAllReduceOnVertical(const u32 rank, const std::vector<LINK> &links, const RingInfo &info)
223 : {
224 0 : u32 ringRank = info.GetVIndex(rank); // 查找自己位于第几行，也即处于Ring中的第几个rank
225 0 : u32 ringSize = info.GetVSizeByRank(rank); // 查找自己所处的列长度，也即Ring的大小
226 : // 若最后一列不完整，则不做allreduce操作直接返回success
227 0 : if (ringSize < info.GetVSizeByHIndex(0)) {
228 0 : return HCCL_SUCCESS;
229 : }
230 : // 计算allreduce 阶段每个rank结果上的offset和size
231 0 : std::vector<Slice> vSlices;
232 0 : std::vector<LINK> vLinks;
233 0 : CHK_RET(CalcVSlicesAndLinks(rank, links, info, vLinks, vSlices));
234 :
235 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
236 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RING, dispatcher_);
237 0 : CHK_SMART_PTR_NULL(tempAlg);
238 0 : CHK_RET(tempAlg->Prepare(reduceAttr_));
239 :
240 : // 判断是否关闭allreduce的barrier
241 0 : if (!barrierSwitchOn_) {
242 0 : tempAlg->CloseBarrier();
243 : }
244 :
245 0 : CHK_RET(tempAlg->Prepare(inputMem_, outputMem_, outputMem_, count_, dataType_,
246 : stream_, reductionOp_, root_, vSlices, baseOffset_));
247 :
248 0 : CHK_RET(tempAlg->RegisterProfiler(
249 : profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_));
250 :
251 0 : HCCL_DEBUG("[AllReduceNHRV1][AllReduce-V] rank[%u] ringRank=%u, ringSize=%u", rank, ringRank, ringSize);
252 0 : return tempAlg->RunAsync(ringRank, ringSize, vLinks);
253 0 : }
254 :
255 0 : HcclResult AllReduceNHRV1::RunAllGatherOnHorizontal(const u32 rank, const std::vector<LINK> &links,
256 : const RingInfo &info)
257 : {
258 0 : u32 ringRank = info.GetHIndex(rank); // 查找自己位于第几列，也即处于Ring中的第几个rank
259 :
260 : // 计算allgather阶段每个rank结果上的offset和size
261 0 : std::vector<Slice> hSlices;
262 0 : std::vector<LINK> hLinks;
263 0 : CHK_RET(CalcHSlicesAndLinks(rank, links, info, hLinks, hSlices));
264 :
265 : // 长度不足2，直接跳过
266 0 : if (hLinks.size() < 2)
267 0 : return HCCL_SUCCESS;
268 :
269 0 : HCCL_DEBUG("[AllReduceNHRV1][AllGather-H] rank[%u] ringRank=%u, ringSize=%u", rank, ringRank, hLinks.size());
270 0 : return RunAllGatherBrokenRing(ringRank, hLinks, hSlices);
271 0 : }
272 :
273 0 : HcclResult AllReduceNHRV1::RunReduceScatterBrokenRing(const u32 rank, const std::vector<LINK> &links,
274 : const std::vector<Slice> &slices)
275 : {
276 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
277 0 : TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
278 0 : CHK_SMART_PTR_NULL(tempAlg);
279 0 : CHK_RET(tempAlg->Prepare(reduceAttr_));
280 :
281 : // 判断是否关闭reducescatter的barrier
282 0 : if (!barrierSwitchOn_) {
283 0 : tempAlg->CloseBarrier();
284 : }
285 :
286 : // 调用reducescatter ring的算法执行
287 0 : CHK_RET(tempAlg->Prepare(inputMem_, inputMem_, outputMem_, count_, dataType_,
288 : stream_, reductionOp_, root_, slices, baseOffset_));
289 :
290 0 : CHK_RET(tempAlg->RegisterProfiler(
291 : profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_));
292 :
293 0 : return tempAlg->RunAsync(rank, links.size(), links);
294 0 : }
295 :
296 0 : HcclResult AllReduceNHRV1::RunAllGatherBrokenRing(const u32 rank, const std::vector<LINK> &links,
297 : const std::vector<Slice> &slices)
298 : {
299 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
300 0 : TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
301 0 : CHK_SMART_PTR_NULL(tempAlg);
302 : // 判断是否关闭allgather的barrier
303 0 : if (!barrierSwitchOn_) {
304 0 : tempAlg->CloseBarrier();
305 : }
306 :
307 : // 调用allgather ring的算法执行
308 0 : CHK_RET(tempAlg->Prepare(outputMem_, outputMem_, outputMem_, count_, dataType_, stream_,
309 : reductionOp_, root_, slices, baseOffset_));
310 :
311 0 : CHK_RET(tempAlg->RegisterProfiler(
312 : profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_));
313 :
314 0 : return tempAlg->RunAsync(rank, links.size(), links);
315 0 : }
316 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_REDUCE_NHR_V1, AllReduceNHRV1);
317 : }
|
