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