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 "log.h"
12 : #include "alg_data_trans_wrapper.h"
13 : #include "ins_temp_reduce_nhr.h"
14 :
15 : namespace Hccl {
16 0 : InsTempReduceNHR::InsTempReduceNHR(
17 : const RankId virtualRank, const u32 tempRankSize, const std::vector<std::vector<RankId>>& tempVTopo,
18 0 : const std::map<RankId, u32>& tempVirtRankMap)
19 0 : : InsAlgTemplateBase(virtualRank, tempRankSize, tempVTopo, tempVirtRankMap)
20 0 : {}
21 :
22 0 : InsTempReduceNHR::~InsTempReduceNHR() {}
23 :
24 0 : HcclResult InsTempReduceNHR::CalcRes(AlgTempResReq& tempResReq)
25 : {
26 : // NHR 需要的 que Num 为 1
27 0 : tempResReq.queNum = 1;
28 0 : tempResReq.streamNum = tempResReq.queNum;
29 0 : tempResReq.queNotifys = CreateMasterSlaveQueNotifiesRequest(tempResReq.queNum);
30 :
31 0 : CHK_PRT_RET(
32 : CalcResLinksNHR(myRank_, tempRankSize_, tempVTopo_, tempResReq) != HcclResult::HCCL_SUCCESS,
33 : HCCL_ERROR("[CollAlgFactory] [InsTempReduceNHR] Rank [%d], resLinks calculation error!", myRank_),
34 : HcclResult::HCCL_E_INTERNAL);
35 :
36 0 : return HcclResult::HCCL_SUCCESS;
37 : }
38 :
39 : /*
40 : * Desc: 将数据按照rank切分为chuck 块，给后续的reduce操作使用
41 : * param: dataSize: 待处理的输入数据大小
42 : * return: sliceInfoVec: 存储数据切分结果
43 : * return: HcclResult
44 : */
45 0 : HcclResult InsTempReduceNHR::CalcSlice(const u64 dataSize, RankSliceInfo& sliceInfoVec)
46 : {
47 : // 按 rank 切分数据（与 AllReduceNHR 保持一致）
48 0 : std::vector<SliceInfo> tmp(tempVTopo_.size());
49 0 : sliceInfoVec.resize(tempRankSize_, tmp);
50 :
51 0 : u64 unitAlignSize = DataTypeSizeGet(dataType_);
52 0 : u64 chunkSize = RoundUp(dataSize, (tempRankSize_ * unitAlignSize)) * unitAlignSize;
53 :
54 0 : u64 accumOff = 0;
55 0 : for (u32 rankIdx = 0; rankIdx < tempRankSize_; rankIdx++) {
56 0 : u64 currChunkSize = ((dataSize - accumOff) > chunkSize) ? chunkSize : (dataSize - accumOff);
57 0 : SliceInfo slice = {accumOff, currChunkSize};
58 0 : sliceInfoVec[rankIdx][0] = slice;
59 0 : accumOff += currChunkSize;
60 : }
61 :
62 0 : CHK_PRT_RET(
63 : (sliceInfoVec[tempRankSize_ - 1][0].offset + sliceInfoVec[tempRankSize_ - 1][0].size != dataSize),
64 : HCCL_ERROR("[InsTempReduceNHR] chunkSize:[%llu], Rank:[%d], SliceInfo calculation error!", chunkSize, myRank_),
65 : HcclResult::HCCL_E_INTERNAL);
66 :
67 0 : return HcclResult::HCCL_SUCCESS;
68 0 : }
69 :
70 : /*
71 : * Desc: 返回当前rank能处理的数据量和scratch buffer之间的比例关系
72 : * param: input: 输入数据位置
73 : * param: output 输出数据位置
74 : */
75 0 : u32 InsTempReduceNHR::CalcScratchMultiple(BufferType input, BufferType output)
76 : {
77 : (void)input;
78 : (void)output;
79 : // 单算子模式下需要 1 倍的 scratch（ccl buffer），图/流水（OFFLOAD）模式下不需要
80 0 : u32 multiple = 0;
81 0 : if (op_.opMode == OpMode::OPBASE) {
82 0 : multiple = 1;
83 : }
84 0 : return multiple;
85 : }
86 :
87 0 : HcclResult InsTempReduceNHR::GenExtIns(
88 : const TempFuncs& tempFuncs, const TemplateDataParams& tempAlgParams, const ResLinks& tempLinks,
89 : std::vector<InsQuePtr>& tempInsQues)
90 : {
91 0 : HCCL_INFO("[InsTempReduceNHR][GenExtIns] ReduceNHR begin: rank[%d] start", myRank_);
92 0 : if (IsPcieLink(tempLinks)) {
93 0 : dmaMode_ = DmaMode::GET;
94 : }
95 0 : opMode_ = tempFuncs.opMode;
96 0 : enableCounterNotify_ = tempFuncs.enableCounterNotify;
97 0 : queNum_ = tempVTopo_.size();
98 :
99 0 : CHK_PRT_RET(
100 : queNum_ != tempInsQues.size(),
101 : HCCL_ERROR("[CollAlgFactory] [InsTempReduceNHR] Rank [%d], requiredQue Error.", myRank_),
102 : HcclResult::HCCL_E_INTERNAL);
103 :
104 : // 1. 切片
105 0 : RankSliceInfo sliceInfoVec;
106 0 : CHK_RET(CalcSlice(tempAlgParams.sliceSize, sliceInfoVec));
107 :
108 : // 2. PreCopy (OPBASE 模式下将 userIn -> scratch)
109 0 : CHK_RET(PreCopy(tempAlgParams, tempInsQues));
110 :
111 : // 3. ReduceScatter 阶段 (pairwise reduce)
112 0 : CHK_RET(RunReduceScatter(sliceInfoVec, tempLinks, tempInsQues));
113 :
114 : // 4. PrepareDataForGather 阶段
115 0 : CHK_RET(PrepareDataForGather(sliceInfoVec, tempInsQues));
116 :
117 : // 5. Gather 阶段 (将每个 chunk 聚合到 root)
118 0 : CHK_RET(RunGather(sliceInfoVec, tempLinks, tempInsQues));
119 :
120 : // 6. PostCopy (OPBASE 且在 root 上将 scratch -> userOut)
121 0 : CHK_RET(PostCopy(tempAlgParams, tempInsQues));
122 :
123 0 : HCCL_INFO("[InsTempReduceNHR][GenExtIns] ReduceNHR finished: rank[%d] end", myRank_);
124 0 : return HcclResult::HCCL_SUCCESS;
125 0 : }
126 :
127 0 : HcclResult InsTempReduceNHR::PreCopy(const TemplateDataParams& tempAlgParams, std::vector<InsQuePtr>& tempInsQues)
128 : {
129 : // 单算子模式，需要先将数据拷贝到cclBuffer
130 0 : if (opMode_ == OpMode::OPBASE) {
131 0 : reduceInBuffType_ = BufferType::SCRATCH;
132 0 : reduceInBuffBaseOff_ = tempAlgParams.buffInfo.inBuffBaseOff;
133 :
134 0 : if (tempAlgParams.buffInfo.inBuffType != BufferType::SCRATCH) {
135 0 : HCCL_INFO("[InsTempReduceNHR][PreCopy] Opbase copy from userIn to scratchBuffer");
136 : DataSlice usrInSlices = DataSlice(
137 0 : tempAlgParams.buffInfo.inBuffType, tempAlgParams.buffInfo.inBuffBaseOff, tempAlgParams.sliceSize);
138 : DataSlice scratchSlices
139 0 : = DataSlice(BufferType::SCRATCH, tempAlgParams.buffInfo.scratchBuffBaseOff, tempAlgParams.sliceSize);
140 0 : CHK_RET(LocalCopy(tempInsQues[0], usrInSlices, scratchSlices));
141 0 : reduceInBuffBaseOff_ = tempAlgParams.buffInfo.scratchBuffBaseOff;
142 : } else {
143 0 : HCCL_INFO("[InsTempReduceNHR][PreCopy] skip precopy");
144 : }
145 : } else {
146 : // OFFLOAD 图模式直接在用户 buffer 上操作
147 0 : HCCL_INFO("[InsTempReduceNHR][PreCopy] offload skip precopy");
148 0 : reduceInBuffType_ = tempAlgParams.buffInfo.inBuffType;
149 0 : reduceInBuffBaseOff_ = tempAlgParams.buffInfo.inBuffBaseOff;
150 : }
151 :
152 0 : reduceOutBuffType_ = tempAlgParams.buffInfo.outBuffType;
153 0 : reduceOutBuffBaseOff_ = tempAlgParams.buffInfo.outBuffBaseOff;
154 :
155 0 : return HcclResult::HCCL_SUCCESS;
156 : }
157 :
158 : // 将reduceScatter之后的数据先放到usrOut
159 : HcclResult
160 0 : InsTempReduceNHR::PrepareDataForGather(const RankSliceInfo& sliceInfoVec, std::vector<InsQuePtr>& tempInsQues)
161 : {
162 : // 如果是单算子模式，在原来的位置要先做完Gather，然后postCopy把数据放到usrOut
163 : // 如果是图模式，直接把数据放到usrOUt，然后在usrOut上做Gather
164 0 : HCCL_INFO("[InsTempReduceNHR][PrepareDataForGather] prepare data for Gather");
165 :
166 0 : if (opMode_ == OpMode::OFFLOAD) {
167 0 : u64 size = sliceInfoVec[tempVirtRankMap_[myRank_]][0].size;
168 0 : u64 srcOffset = sliceInfoVec[tempVirtRankMap_[myRank_]][0].offset;
169 0 : u64 dstOffset = sliceInfoVec[tempVirtRankMap_[myRank_]][0].offset;
170 0 : DataSlice srcSlice = DataSlice(reduceInBuffType_, reduceInBuffBaseOff_ + srcOffset, size);
171 0 : DataSlice dstSlice = DataSlice(reduceOutBuffType_, reduceOutBuffBaseOff_ + dstOffset, size);
172 0 : CHK_RET(LocalCopy(tempInsQues[0], srcSlice, dstSlice));
173 0 : reduceInBuffType_ = reduceOutBuffType_;
174 0 : reduceInBuffBaseOff_ = reduceOutBuffBaseOff_;
175 : }
176 :
177 0 : return HcclResult::HCCL_SUCCESS;
178 : }
179 :
180 0 : HcclResult InsTempReduceNHR::PostCopy(const TemplateDataParams& tempAlgParams, std::vector<InsQuePtr>& tempInsQues)
181 : {
182 : // PostCopy 仅在 OPBASE 并且在 root 上执行（root 收到完整结果后写回用户 out）
183 0 : RankId rootRank = this->root_; // Executor 在 CreateTemplates 时已调用 SetRoot(op_.root)
184 :
185 0 : if (myRank_ != rootRank) {
186 0 : HCCL_DEBUG("[InsTempReduceNHR][PostCopy] not root, skip postcopy rank[%d]", myRank_);
187 0 : return HcclResult::HCCL_SUCCESS;
188 : }
189 :
190 0 : if (opMode_ == OpMode::OPBASE) {
191 0 : HCCL_INFO("[InsTempReduceNHR][PostCopy] Opbase root copy from scratchBuffer to userOut");
192 0 : DataSlice scratchSlices = DataSlice(reduceInBuffType_, reduceInBuffBaseOff_, tempAlgParams.sliceSize);
193 0 : DataSlice usrOutSlices = DataSlice(reduceOutBuffType_, reduceOutBuffBaseOff_, tempAlgParams.sliceSize);
194 0 : CHK_RET(LocalCopy(tempInsQues[0], scratchSlices, usrOutSlices));
195 : } else {
196 0 : HCCL_INFO("[InsTempReduceNHR][PostCopy] offload skip postcopy");
197 : }
198 :
199 0 : return HcclResult::HCCL_SUCCESS;
200 : }
201 :
202 0 : HcclResult InsTempReduceNHR::RunReduceScatter(
203 : const RankSliceInfo& sliceInfoVec, const ResLinks& tempLinks, std::vector<InsQuePtr>& tempInsQues)
204 : {
205 0 : std::vector<AicpuNHRStepInfo> stepInfoList;
206 0 : CHK_RET(GetStepInfoList(stepInfoList));
207 :
208 0 : for (auto& stepInfo : stepInfoList) {
209 0 : HCCL_DEBUG(
210 : "[InsTempReduceNHR][RunReduceScatter] step[%u], myRank[%u], toRank[%u], fromRank[%u], nSlices[%u].",
211 : stepInfo.step, stepInfo.myRank, stepInfo.toRank, stepInfo.fromRank, stepInfo.nSlices);
212 :
213 0 : const std::vector<LinkData>& linkRecv = tempLinks.at(GetRankFromMap(stepInfo.fromRank));
214 0 : const std::vector<LinkData>& linkSend = tempLinks.at(GetRankFromMap(stepInfo.toRank));
215 :
216 0 : std::vector<DataSlice> txSlices;
217 0 : std::vector<DataSlice> rxSlices;
218 :
219 : // 发送和接收 slice 都发生在 reduceInBuffType_ 上（scratch 或用户 buffer）
220 0 : for (u32 i = 0; i < stepInfo.nSlices; i++) {
221 0 : u64 txOffset = sliceInfoVec[stepInfo.txSliceIdxs[i]][0].offset + reduceInBuffBaseOff_;
222 0 : u64 txSize = sliceInfoVec[stepInfo.txSliceIdxs[i]][0].size;
223 0 : u64 rxOffset = sliceInfoVec[stepInfo.rxSliceIdxs[i]][0].offset + reduceInBuffBaseOff_;
224 0 : u64 rxSize = sliceInfoVec[stepInfo.rxSliceIdxs[i]][0].size;
225 :
226 0 : txSlices.push_back(DataSlice(reduceInBuffType_, txOffset, txSize));
227 0 : rxSlices.push_back(DataSlice(reduceInBuffType_, rxOffset, rxSize));
228 : }
229 :
230 : SendRecvReduceInfo sendRecvReduceInfo{
231 0 : {linkSend[0], linkRecv[0]}, {{txSlices, txSlices}, {rxSlices, rxSlices}}, dataType_, redOp_};
232 :
233 0 : CHK_PRT_RET(
234 : SendRecvReduce(sendRecvReduceInfo, tempInsQues[0], 0, true, dmaMode_) != HcclResult::HCCL_SUCCESS,
235 : HCCL_ERROR("[InsTempReduceNHR] RunReduceScatter SendRecvReduce failed"), HcclResult::HCCL_E_INTERNAL);
236 0 : }
237 :
238 0 : return HcclResult::HCCL_SUCCESS;
239 0 : }
240 :
241 0 : HcclResult InsTempReduceNHR::RunGather(
242 : const RankSliceInfo& sliceInfoVec, const ResLinks& tempLinks, std::vector<InsQuePtr>& tempInsQues)
243 : {
244 0 : u32 nSteps = GetNHRStepNum(tempRankSize_);
245 0 : for (u32 step = 0; step < nSteps; step++) {
246 0 : AicpuNHRStepInfo stepInfo;
247 0 : CHK_RET(GetStepInfo(step, nSteps, stepInfo));
248 :
249 0 : const std::vector<LinkData>& linkRecv = tempLinks.at(GetRankFromMap(stepInfo.fromRank));
250 0 : const std::vector<LinkData>& linkSend = tempLinks.at(GetRankFromMap(stepInfo.toRank));
251 :
252 0 : std::vector<DataSlice> txSlices;
253 0 : std::vector<DataSlice> rxSlices;
254 0 : for (u32 i = 0; i < stepInfo.nSlices; i++) {
255 0 : u64 txOffset = sliceInfoVec[stepInfo.txSliceIdxs[i]][0].offset + reduceInBuffBaseOff_;
256 0 : u64 txSize = sliceInfoVec[stepInfo.txSliceIdxs[i]][0].size;
257 0 : u64 rxOffset = sliceInfoVec[stepInfo.rxSliceIdxs[i]][0].offset + reduceInBuffBaseOff_;
258 0 : u64 rxSize = sliceInfoVec[stepInfo.rxSliceIdxs[i]][0].size;
259 :
260 0 : txSlices.push_back(DataSlice(reduceInBuffType_, txOffset, txSize));
261 0 : rxSlices.push_back(DataSlice(reduceInBuffType_, rxOffset, rxSize));
262 : }
263 :
264 0 : TxRxLinks sendRecvLinks(linkSend[0], linkRecv[0]);
265 0 : TxRxSlicesList sendRecvSlicesList({txSlices, txSlices}, {rxSlices, rxSlices});
266 :
267 0 : SendRecvInfo sendRecvInfo(sendRecvLinks, sendRecvSlicesList);
268 0 : CHK_PRT_RET(
269 : SendRecv(sendRecvInfo, tempInsQues[0], 0, true, dmaMode_) != HcclResult::HCCL_SUCCESS,
270 : HCCL_ERROR("[InsTempReduceNHR] RunGather send/recv failed"), HcclResult::HCCL_E_INTERNAL);
271 0 : }
272 :
273 0 : return HcclResult::HCCL_SUCCESS;
274 : }
275 :
276 0 : HcclResult InsTempReduceNHR::GetStepInfo(u32 step, u32 nSteps, AicpuNHRStepInfo& stepInfo)
277 : {
278 0 : u32 rankIdx = tempVirtRankMap_[myRank_];
279 0 : stepInfo.txSliceIdxs.clear();
280 0 : stepInfo.rxSliceIdxs.clear();
281 0 : stepInfo.step = step;
282 0 : stepInfo.myRank = rankIdx;
283 :
284 : // 计算通信对象
285 0 : u32 deltaRank = 1 << (nSteps - 1 - step);
286 0 : u32 recvFrom = (rankIdx + tempRankSize_ - deltaRank) % tempRankSize_;
287 0 : u32 sendTo = (rankIdx + deltaRank) % tempRankSize_;
288 :
289 : // 数据份数和数据编号增量
290 0 : u32 nSlices = (tempRankSize_ - 1 + (1 << (nSteps - 1 - step))) / (1 << (nSteps - step));
291 0 : u32 deltaSliceIndex = 1 << (nSteps - step);
292 0 : u32 txSliceIdx = rankIdx;
293 0 : u32 rxSliceIdx = (rankIdx - (1 << (nSteps - 1 - step)) + tempRankSize_) % tempRankSize_;
294 :
295 0 : stepInfo.nSlices = nSlices;
296 0 : stepInfo.toRank = sendTo;
297 0 : stepInfo.fromRank = recvFrom;
298 :
299 0 : for (u32 i = 0; i < nSlices; i++) {
300 0 : stepInfo.txSliceIdxs.push_back(txSliceIdx);
301 0 : stepInfo.rxSliceIdxs.push_back(rxSliceIdx);
302 :
303 0 : HCCL_DEBUG("[InsTempReduceNHR][GetStepInfo] i[%u] txSliceIdx[%u] rxSliceIdx[%u]", i, txSliceIdx, rxSliceIdx);
304 :
305 0 : txSliceIdx = (txSliceIdx + tempRankSize_ - deltaSliceIndex) % tempRankSize_;
306 0 : rxSliceIdx = (rxSliceIdx + tempRankSize_ - deltaSliceIndex) % tempRankSize_;
307 : }
308 0 : return HcclResult::HCCL_SUCCESS;
309 : }
310 :
311 : // 计算每轮收发的对端以及slice编号
312 0 : HcclResult InsTempReduceNHR::GetStepInfoList(std::vector<AicpuNHRStepInfo>& stepInfoList)
313 : {
314 : // 将本 rank 号转换成算法使用的索引号
315 0 : u32 rankIdx = tempVirtRankMap_[myRank_];
316 0 : stepInfoList.clear();
317 :
318 0 : u32 nSteps = GetNHRStepNum(tempRankSize_);
319 0 : stepInfoList.resize(nSteps);
320 0 : for (u32 step = 0; step < nSteps; step++) {
321 : // 计算通信对象
322 0 : u32 deltaRank = 1 << step;
323 0 : u32 sendTo = (rankIdx + tempRankSize_ - deltaRank) % tempRankSize_;
324 0 : u32 recvFrom = (rankIdx + deltaRank) % tempRankSize_;
325 :
326 : // 数据份数和数据编号增量
327 0 : u32 nSlices = (tempRankSize_ - 1 + (1 << step)) / (1 << (step + 1));
328 0 : u32 deltaSliceIndex = 1 << (step + 1);
329 0 : u32 txSliceIdx = sendTo;
330 0 : u32 rxSliceIdx = rankIdx;
331 :
332 0 : AicpuNHRStepInfo& currStepInfo = stepInfoList[step];
333 0 : currStepInfo.step = step;
334 0 : currStepInfo.myRank = rankIdx;
335 0 : currStepInfo.nSlices = nSlices;
336 0 : currStepInfo.toRank = sendTo;
337 0 : currStepInfo.fromRank = recvFrom;
338 :
339 : // 计算本rank在每轮收/发中的slice编号
340 0 : currStepInfo.txSliceIdxs.reserve(nSlices);
341 0 : currStepInfo.rxSliceIdxs.reserve(nSlices);
342 0 : for (u32 i = 0; i < nSlices; i++) {
343 0 : currStepInfo.txSliceIdxs.push_back(txSliceIdx);
344 0 : currStepInfo.rxSliceIdxs.push_back(rxSliceIdx);
345 0 : HCCL_DEBUG(
346 : "[InsTempReduceNHR][GetStepInfoList] i[%u] txSliceIdx[%u] rxSliceIdx[%u]", i, txSliceIdx, rxSliceIdx);
347 0 : txSliceIdx = (txSliceIdx + tempRankSize_ - deltaSliceIndex) % tempRankSize_;
348 0 : rxSliceIdx = (rxSliceIdx + tempRankSize_ - deltaSliceIndex) % tempRankSize_;
349 : }
350 : }
351 0 : return HcclResult::HCCL_SUCCESS;
352 : }
353 :
354 0 : RankId InsTempReduceNHR::GetRankFromMap(const u32 rankIdx)
355 : {
356 0 : RankId rank = -1;
357 0 : for (auto& pair : tempVirtRankMap_) {
358 0 : if (pair.second == rankIdx) {
359 0 : rank = pair.first;
360 0 : break;
361 : }
362 : }
363 0 : return rank;
364 : }
365 :
366 : } // namespace Hccl
|
