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_ring.h"
13 :
14 : namespace hccl {
15 32 : AllReduceRing::AllReduceRing(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher)
16 : {
17 32 : }
18 :
19 59 : AllReduceRing::~AllReduceRing()
20 : {
21 59 : }
22 :
23 32 : HcclResult AllReduceRing::Prepare(u64 reduceAttrBitMap, HcomCollOpInfo *opInfo)
24 : {
25 32 : reduceAttr_ = reduceAttrBitMap;
26 32 : return HCCL_SUCCESS;
27 : }
28 :
29 : // ringallreduce算法的函数入口
30 0 : HcclResult AllReduceRing::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK> &links)
31 : {
32 0 : HcclResult ret = HCCL_SUCCESS;
33 0 : ret = PrepareRunAsync(rank, rankSize, links);
34 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceRing][RunAsync]rank[%u] count[%llu] "\
35 : "failed in PrepareRunAsync step", rank, count_), ret);
36 0 : CHK_PRT_RET(rankSize == 1, HCCL_INFO("[AllReduceRing][RunAsync] rankSize[%u], do nothing.",
37 : rankSize), HCCL_SUCCESS);
38 :
39 : // 先执行reducescater
40 0 : ret = RunReduceScatter(rank, rankSize, links);
41 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceRing][RunAsync]rank[%u] count[%llu] failed in reducescater "\
42 : "step", rank, count_), ret);
43 :
44 : // 再执行allgather
45 0 : ret = RunAllGather(rank, rankSize, links);
46 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceRing][RunAsync]rank[%u] count[%llu] failed in AllGather "\
47 : "step", rank, count_), ret);
48 :
49 0 : HCCL_INFO("AllReduceRing finished: rank[%u] ranksize[%u]", rank, rankSize);
50 0 : return HCCL_SUCCESS;
51 : }
52 :
53 0 : HcclResult AllReduceRing::RunAsyncStaged(const u32 rank, const u32 rankSize, const std::vector<LINK> &links,
54 : RunStage stage)
55 : {
56 0 : CHK_PRT_RET(rankSize == 1 && stage != RunStage::RUN_PREPARE,
57 : HCCL_INFO("[AllReduceRing][RunAsyncStaged] rankSize[%u], stage[%d], do nothing.",
58 : rankSize, stage), HCCL_SUCCESS);
59 :
60 0 : HcclResult ret = HCCL_SUCCESS;
61 0 : switch (stage) {
62 0 : case RunStage::RUN_PREPARE:
63 0 : ret = PrepareRunAsync(rank, rankSize, links);
64 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
65 : HCCL_ERROR("[AllReduceRing][RunAsyncStaged]rank[%u] count[%llu] failed in PrepareRunAsync step",
66 : rank, count_), ret);
67 0 : break;
68 0 : case RunStage::RUN_REDUCE_SCATTER:
69 : // 先执行reducescater
70 0 : ret = RunReduceScatter(rank, rankSize, links, true);
71 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceRing][RunAsyncStaged]rank[%u] count[%llu] "\
72 : "failed in reducescater step", rank, count_), ret);
73 0 : break;
74 0 : case RunStage::RUN_ALLGATHER:
75 : // 再执行allgather
76 0 : ret = RunAllGather(rank, rankSize, links);
77 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllReduceRing][RunAsyncStaged]rank[%u] count[%llu] "\
78 : "failed in AllGather step", rank, count_), ret);
79 0 : break;
80 0 : default:
81 0 : HCCL_ERROR("[AllReduceRing][RunAsyncStaged]stage[%d]is not support", stage);
82 0 : return HCCL_E_NOT_SUPPORT;
83 : }
84 0 : HCCL_INFO("AllReduceRing RunAsyncStaged stage[%d] finished: rank[%u] ranksize[%u]", stage, rank, rankSize);
85 0 : return HCCL_SUCCESS;
86 : }
87 :
88 0 : HcclResult AllReduceRing::PrepareRunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK> &links)
89 : {
90 0 : HcclResult ret = HCCL_SUCCESS;
91 0 : CHK_SMART_PTR_NULL(dispatcher_);
92 0 : CHK_PTR_NULL(stream_.ptr());
93 0 : if (!outputMem_ || !inputMem_) {
94 0 : HCCL_ERROR("[AllReduceRing][RunAsync]rank[%u] run_async inputmem or outputmem is null", rank);
95 0 : return HCCL_E_PTR;
96 : }
97 0 : HCCL_INFO("AllReduceRing run: rank[%u] ranksize[%u] inputMem[%p] outputMem[%p] count[%llu]", \
98 : rank, rankSize, inputMem_.ptr(), outputMem_.ptr(), count_);
99 :
100 0 : if (links.size() < rankSize) {
101 0 : HCCL_ERROR("[AllReduceRing][RunAsync]rank[%u] linksize[%llu] is less than rankSize[%u]", rank, links.size(),
102 : rankSize);
103 0 : return HCCL_E_INTERNAL;
104 : }
105 :
106 : // 如果ranksize为1, inline reduce和普通跨片reduce操作一致，从input->output
107 0 : if (rankSize == 1) {
108 0 : if (inputMem_ != outputMem_) {
109 0 : ret = HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_);
110 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
111 : HCCL_ERROR("[AllReduceRing][RunAsync]rank[%u] memcpy async failed", rank), ret);
112 : }
113 :
114 0 : return ret;
115 : }
116 : // 计算reducescatter 阶段每个rank结果上的offset和size
117 0 : if (slices_.size() == 0) {
118 0 : slices_.resize(rankSize);
119 0 : u64 totalSize = count_ * SIZE_TABLE[dataType_];
120 0 : u64 sliceSizeCalculated = (totalSize + (rankSize - 1)) / rankSize;
121 0 : u64 sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSizeCalculated, HCCL_MIN_SLICE_ALIGN);
122 0 : u64 residueSize = totalSize;
123 :
124 0 : for (u32 i = 0; i < rankSize; i++) {
125 0 : slices_[i].size = (residueSize > sliceSizeAligned) ? sliceSizeAligned : residueSize;
126 0 : slices_[i].offset = totalSize - residueSize;
127 0 : residueSize -= slices_[i].size;
128 : }
129 :
130 0 : if (HcclCheckLogLevel(DLOG_DEBUG)) {
131 0 : for (size_t j = 0; j < slices_.size(); j++) {
132 0 : HCCL_DEBUG("rank[%u] slice[%u]: offset[%llu] size[%llu]", rank, j, slices_[j].offset, slices_[j].size);
133 : }
134 : }
135 : }
136 0 : HCCL_INFO("AllReduceRing PrepareRunAsync finished: rank[%u] ranksize[%u]", rank, rankSize);
137 0 : return HCCL_SUCCESS;
138 : }
139 :
140 :
141 0 : HcclResult AllReduceRing::RunReduceScatter(u32 rank, u32 rankSize, const std::vector<LINK> &links, bool needBarrier)
142 : {
143 0 : if (links.size() < rankSize) {
144 0 : HCCL_ERROR("[AllReduceRing][RunReduceScatter]rank[%u] linkSize[%llu] is less than rankSize[%u]",
145 : rank, links.size(), rankSize);
146 0 : return HCCL_E_INTERNAL;
147 : }
148 :
149 : // 调用reducescattering算法
150 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
151 0 : TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
152 0 : CHK_SMART_PTR_NULL(tempAlg);
153 0 : tempAlg->Prepare(reduceAttr_);
154 0 : HCCL_INFO("rank[%u] tempAlg ReduceScattering inputMem[%p] outputMem[%p] mem_size[%llu] "\
155 : "count[%llu] planeID:[%d]", \
156 : rank, inputMem_.ptr(), outputMem_.ptr(), outputMem_.size(), count_, profilerInput_.planeID);
157 0 : if (!needBarrier) {
158 0 : tempAlg->CloseBarrier();
159 : }
160 0 : CHK_RET(tempAlg->Prepare(inputMem_, inputMem_, outputMem_, count_, dataType_, stream_,
161 : reductionOp_, root_, slices_, baseOffset_));
162 :
163 0 : CHK_RET(tempAlg->RegisterProfiler(
164 : profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_));
165 :
166 0 : HcclResult ret = tempAlg->RunAsync(rank, rankSize, links);
167 0 : if (ret != HCCL_SUCCESS) {
168 0 : return ret;
169 : }
170 0 : return HCCL_SUCCESS;
171 0 : }
172 :
173 0 : HcclResult AllReduceRing::RunAllGather(u32 rank, u32 rankSize, const std::vector<LINK> &links)
174 : {
175 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
176 0 : TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
177 0 : CHK_SMART_PTR_NULL(tempAlg);
178 0 : HCCL_INFO("rank[%u] tempAlg AllGathering inputMem[%p] outputMem[%p] mem_size[%llu] "\
179 : "count[%llu] planeID:[%d]", rank, inputMem_.ptr(), outputMem_.ptr(), outputMem_.size(),
180 : count_, profilerInput_.planeID);
181 : // 判断是否关闭allgather的barrier
182 0 : if (!barrierSwitchOn_) {
183 0 : tempAlg->CloseBarrier();
184 : }
185 : // 调用allgatherring的算法执行
186 0 : CHK_RET(tempAlg->Prepare(inputMem_, outputMem_, outputMem_, count_, dataType_, stream_,
187 : reductionOp_, root_, slices_, baseOffset_));
188 :
189 0 : CHK_RET(tempAlg->RegisterProfiler(
190 : profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_));
191 :
192 0 : HcclResult ret = tempAlg->RunAsync(rank, rankSize, links);
193 0 : if (ret != HCCL_SUCCESS) {
194 0 : return ret;
195 : }
196 0 : return HCCL_SUCCESS;
197 0 : }
198 0 : HcclResult AllReduceRing::GetNslbAdjInfo(const u32 rank, const u32 rankSize,
199 : const std::vector<LINK> &links, AdjInfo& nslbAdjInfo)
200 : {
201 0 : u32 ringNextRank = (rank + 1) % rankSize;
202 0 : LINK nslbNext = links[ringNextRank];
203 0 : CHK_SMART_PTR_NULL(nslbNext);
204 : // reduce_scatter 的ring
205 0 : NslbDpAdjInfo adjInfoStep = {0};
206 0 : nslbAdjInfo.dstRankNum = 1;
207 0 : adjInfoStep.dstLocalRankId = nslbNext->GetRemoteRank();
208 0 : adjInfoStep.phaseId = 1;
209 0 : adjInfoStep.rev = 0;
210 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
211 :
212 0 : LINK right = links[ringNextRank];
213 0 : CHK_SMART_PTR_NULL(right);
214 : // all_gather 的ring
215 0 : NslbDpAdjInfo nextAdjInfo = {0};
216 0 : nslbAdjInfo.dstRankNum = nslbAdjInfo.dstRankNum + 1;
217 0 : nextAdjInfo.dstLocalRankId = right->GetRemoteRank();
218 0 : nextAdjInfo.phaseId = nslbAdjInfo.nsAdjInfo[0].phaseId + 1;
219 0 : nextAdjInfo.rev = 0;
220 0 : nslbAdjInfo.nsAdjInfo.push_back(nextAdjInfo);
221 0 : return HCCL_SUCCESS;
222 0 : }
223 :
224 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_REDUCE_RING, AllReduceRing);
225 : } // namespace hccl
|
