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 "ins_temp_reduce_scatter_nhr.h"
12 : #include "log.h"
13 : #include "alg_data_trans_wrapper.h"
14 :
15 : namespace Hccl {
16 0 : InsTempReduceScatterNHR::InsTempReduceScatterNHR(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 : InsTempReduceScatterNHR::~InsTempReduceScatterNHR()
24 : {
25 0 : }
26 :
27 0 : HcclResult InsTempReduceScatterNHR::CalcRes(AlgTempResReq &tempResReq)
28 : {
29 0 : CHK_PRT_RET(CalcResLinksNHR(myRank_, tempRankSize_, tempVTopo_, tempResReq) != HcclResult::HCCL_SUCCESS,
30 : HCCL_ERROR("[CollAlgFactory] [InsTempReduceScatterNHR] Rank [%d], resLinks calculation error!", myRank_),
31 : HcclResult::HCCL_E_INTERNAL);
32 0 : auto &linkReq = tempResReq.links;
33 0 : u32 pathNum = 0;
34 0 : for (auto resReqIter = linkReq.begin(); resReqIter != linkReq.end(); resReqIter++) {
35 0 : auto remoteRank = resReqIter->first;
36 0 : if (rank2PathNumMap_.find(remoteRank) == rank2PathNumMap_.end() || rank2PathNumMap_[remoteRank] == 0) {
37 0 : HCCL_ERROR("[InsTempReduceScatterNHR] No path to remoteRank[%d]", remoteRank);
38 0 : return HcclResult::HCCL_E_INTERNAL;
39 : }
40 0 : if (pathNum == 0) {
41 0 : pathNum = rank2PathNumMap_[remoteRank];
42 0 : } else if (rank2PathNumMap_[remoteRank] != pathNum) {
43 0 : HCCL_ERROR("[InsTempReduceScatterNHR] Inconsistency pathNum to remoteRanks, Previous consistent pathNum=[%u], mismatched "
44 : "remoteRank=[%d], pathNum=[%u]",
45 : pathNum,
46 : remoteRank,
47 : rank2PathNumMap_[remoteRank]);
48 0 : return HcclResult::HCCL_E_INTERNAL;
49 : }
50 0 : resReqIter->second = pathNum;
51 : }
52 : // NHR 需要的 que Num 为 1 * pathNum
53 0 : tempResReq.queNum = pathNum;
54 0 : tempResReq.streamNum = tempResReq.queNum;
55 0 : tempResReq.queNotifys = CreateMasterSlaveQueNotifiesRequest(tempResReq.queNum);
56 0 : return HcclResult::HCCL_SUCCESS;
57 : }
58 :
59 : /*
60 : dataSize / (rankSize) --> chunkSize
61 : dataSize / (rankSize * queNum) --> sliceSize
62 :
63 : SliceInfoVecforNHR: [1st chunk: [1st Slice, 2nd Slice, ...], 2nd chunk: [1st Slice, 2nd Slice, ...], ...]
64 : */
65 0 : HcclResult InsTempReduceScatterNHR::CalcSliceInfo(const AllignInfo &allignInfo, const u64 dataSize, RankSliceInfo &sliceInfoVec)
66 : {
67 0 : std::vector<SliceInfo> tmp(tempVTopo_.size());
68 0 : sliceInfoVec.resize(tempRankSize_, tmp);
69 0 : CHK_RET(CalcRsAgSliceInfoNHR(myRank_, tempRankSize_, allignInfo, dataSize, sliceInfoVec));
70 :
71 0 : return HcclResult::HCCL_SUCCESS;
72 0 : }
73 :
74 0 : HcclResult InsTempReduceScatterNHR::Run(const TempFuncs &tempFuncs, const RankSliceInfo &sliceInfoVec,
75 : const BuffInfo &buffInfo, const ResLinks &tempLinks, std::vector<InsQuePtr> &tempInsQues)
76 : {
77 0 : if (IsPcieLink(tempLinks)) {
78 0 : dmaMode_ = DmaMode::GET;
79 : }
80 0 : opMode_ = tempFuncs.opMode;
81 0 : enableCounterNotify_ = tempFuncs.enableCounterNotify;
82 0 : buffInfo_ = buffInfo;
83 0 : queNum_ = tempVTopo_.size();
84 0 : CHK_PRT_RET(queNum_ != tempInsQues.size(),
85 : HCCL_ERROR("[CollAlgFactory] [InsTempReduceScatterNHR] Rank [%d], requiredQue Error.", myRank_),
86 : HcclResult::HCCL_E_INTERNAL);
87 0 : CHK_RET(PreCopy(tempFuncs, sliceInfoVec, tempInsQues));
88 0 : CHK_RET(RunReduceScatter(sliceInfoVec, tempLinks, tempInsQues));
89 0 : CHK_RET(PostCopy(tempFuncs, sliceInfoVec, tempInsQues));
90 0 : return HcclResult::HCCL_SUCCESS;
91 : }
92 :
93 0 : HcclResult InsTempReduceScatterNHR::PreCopy(const TempFuncs &tempFuncs, const RankSliceInfo &sliceInfoVec,
94 : std::vector<InsQuePtr> &tempInsQues)
95 : {
96 : (void) sliceInfoVec;
97 : // 通信前需要将所有的数据统一拷贝到 inBuff 上的对应位置。
98 0 : if (tempFuncs.isForepart && opMode_ == OpMode::OPBASE) {
99 : // 单算子模式下，第一个算子，需要将数据从 userIn 拷贝到 inBuff
100 0 : HCCL_INFO("[InsTempReduceScatterNHR][PreCopy] Opbase Forepart, copy from userIn to outBuff");
101 0 : CHK_RET(MultiSliceLocalCopy(tempInsQues[0], tempFuncs.usrData.usrInSlices,
102 : tempFuncs.usrData.scratchInSlices));
103 : } else {
104 : // 图模式或者单算子模式下非第一个算子，数据已经在 inbuff 上了，不需要拷贝
105 0 : HCCL_INFO("[InsTempReduceScatterNHR][PreCopy] not forpat and opbse, skip precopy");
106 : }
107 0 : return HcclResult::HCCL_SUCCESS;
108 : }
109 :
110 0 : HcclResult InsTempReduceScatterNHR::PostCopy(const TempFuncs &tempFuncs, const RankSliceInfo &sliceInfoVec,
111 : std::vector<InsQuePtr> &tempInsQues)
112 : {
113 : // 通信结束之后，数据都在 inbuff 上，需要搬运到对应的输出位置。
114 0 : if (tempFuncs.isBottom && opMode_ == OpMode::OPBASE) {
115 : // 如果是单算子模式, 并且是最后一步算子，需要将数据从 inBuff 拷贝到 userOut
116 : // 是否需要将数据搬运到 OutBuff 上再搬运到 UserOut 上？？
117 0 : HCCL_INFO("[InsTempReduceScatterNHR][PostCopy] Opbase Bottom, copy from outBuff to userOut");
118 0 : CHK_RET(
119 : MultiSliceLocalCopy(tempInsQues[0], tempFuncs.usrData.scratchOutSlices, tempFuncs.usrData.usrOutSlices));
120 0 : } else if (tempFuncs.forAllReduce) {
121 : // 如果是 forAllReduce 算子的前半部分需要将数据从 inBuff 拷贝到 outBuff 并且加上本rank的偏移
122 0 : if (buffInfo_.inBuffType != buffInfo_.outBuffType || buffInfo_.inBuffBaseOff != buffInfo_.outBuffBaseOff) {
123 0 : HCCL_INFO("[InsTempReduceScatterNHR][PostCopy] forAllReduce, copy from inBuff to outBuff");
124 0 : u64 size = sliceInfoVec[tempVirtRankMap_[myRank_]][0].size;
125 0 : u64 srcOffset = sliceInfoVec[tempVirtRankMap_[myRank_]][0].offset;
126 0 : u64 dstOffset = sliceInfoVec[tempVirtRankMap_[myRank_]][0].offset;
127 0 : DataSlice srcSlice = DataSlice(buffInfo_.inBuffType, srcOffset + buffInfo_.inBuffBaseOff, size);
128 0 : DataSlice dstSlice = DataSlice(buffInfo_.outBuffType, dstOffset + buffInfo_.outBuffBaseOff, size);
129 0 : CHK_RET(LocalCopy(tempInsQues[0], srcSlice, dstSlice));
130 : } else {
131 0 : HCCL_INFO("[InsTempReduceScatterNHR][PostCopy] forAllReduce, inBuff same as outBuff, skip copy");
132 : }
133 : } else {
134 : // 如果是图模式，或者单算子模式但不是最后一步算子需要将数据从 inBuff 拷贝到 outBuff 顶头放
135 0 : u64 size = sliceInfoVec[tempVirtRankMap_[myRank_]][0].size;
136 0 : u64 srcOffset = sliceInfoVec[tempVirtRankMap_[myRank_]][0].offset + buffInfo_.inBuffBaseOff;
137 0 : u64 dstOffset = buffInfo_.outBuffBaseOff;
138 0 : if (buffInfo_.inBuffType == buffInfo_.outBuffType && srcOffset == dstOffset) {
139 0 : HCCL_INFO(
140 : "[InsTempReduceScatterNHR][PostCopy] not forpat and opbse, inBuffType same as outBuffType, skip copy");
141 : } else {
142 0 : HCCL_INFO("[InsTempReduceScatterNHR][PostCopy] not forpat and opbse, copy from outBuff to userOut");
143 0 : DataSlice srcSlice = DataSlice(buffInfo_.inBuffType, srcOffset, size);
144 0 : DataSlice dstSlice = DataSlice(buffInfo_.outBuffType, dstOffset, size);
145 0 : CHK_RET(LocalCopy(tempInsQues[0], srcSlice, dstSlice));
146 : }
147 : }
148 0 : return HcclResult::HCCL_SUCCESS;
149 : }
150 :
151 0 : HcclResult InsTempReduceScatterNHR::MultiSliceLocalCopy(InsQuePtr &insQue, const std::vector<DataSlice> &srcList,
152 : const std::vector<DataSlice> &dstList) const
153 : {
154 0 : CHK_PRT_RET(srcList.size() != dstList.size(),
155 : HCCL_ERROR("[InsTempReduceScatterNHR] [LocalCopy] Rank [%d], srcList size[%llu] and "
156 : "dstList size[%llu] not same.",
157 : myRank_, srcList.size(), dstList.size()),
158 : HcclResult::HCCL_E_INTERNAL);
159 0 : CHK_RET(LocalCopySlices(insQue, srcList, dstList));
160 0 : return HcclResult::HCCL_SUCCESS;
161 : }
162 :
163 0 : HcclResult InsTempReduceScatterNHR::RunReduceScatter(const RankSliceInfo &sliceInfoVec, const ResLinks &tempLinks,
164 : std::vector<InsQuePtr> &tempInsQues)
165 : {
166 0 : std::vector<AicpuNHRStepInfo> stepInfoList;
167 0 : GetStepInfoList(stepInfoList);
168 0 : for(auto& stepInfo : stepInfoList) {
169 0 : HCCL_DEBUG("[InsTempReduceScatterNHR][RunReduceScatter] step[%u], myRank[%u], toRank[%u], fromRank[%u], nSlices[%u].",
170 : stepInfo.step, stepInfo.myRank, stepInfo.toRank, stepInfo.fromRank, stepInfo.nSlices);
171 :
172 0 : const std::vector<LinkData> &linkRecv = tempLinks.at(GetRankFromMap(stepInfo.fromRank));
173 0 : const std::vector<LinkData> &linkSend = tempLinks.at(GetRankFromMap(stepInfo.toRank));
174 0 : std::vector<DataSlice> txSlices;
175 0 : std::vector<DataSlice> rxSlices;
176 :
177 : // 在 inBuff 上进行 ReduceScatter 操作
178 0 : for (u32 i = 0; i < stepInfo.nSlices; i++) {
179 0 : u64 txOffset = sliceInfoVec[stepInfo.txSliceIdxs[i]][0].offset + buffInfo_.inBuffBaseOff;
180 0 : u64 txSize = sliceInfoVec[stepInfo.txSliceIdxs[i]][0].size;
181 0 : u64 rxOffset = sliceInfoVec[stepInfo.rxSliceIdxs[i]][0].offset + buffInfo_.inBuffBaseOff;
182 0 : u64 rxSize = sliceInfoVec[stepInfo.rxSliceIdxs[i]][0].size;
183 0 : DataSlice txSlice = DataSlice(buffInfo_.inBuffType, txOffset, txSize);
184 0 : DataSlice rxSlice = DataSlice(buffInfo_.inBuffType, rxOffset, rxSize);
185 0 : txSlices.push_back(txSlice);
186 0 : rxSlices.push_back(rxSlice);
187 : }
188 : SendRecvReduceInfo sendRecvReduceInfo{
189 0 : {linkSend[0],linkRecv[0]},
190 : {{txSlices, txSlices},{rxSlices, rxSlices}}, dataType_, redOp_
191 0 : };
192 0 : CHK_PRT_RET(SendRecvReduce(sendRecvReduceInfo, tempInsQues[0], 0, true, dmaMode_),
193 : HCCL_ERROR("[InsTempReduceScatterNHR] RunReduceScatter SendRecvReduce failed"),
194 : HcclResult::HCCL_E_INTERNAL);
195 0 : }
196 0 : return HcclResult::HCCL_SUCCESS;
197 0 : }
198 :
199 0 : HcclResult InsTempReduceScatterNHR::GenExtIns(const TempFuncs &tempFuncs,
200 : const TemplateDataParams &tempAlgParams,
201 : const ResLinks &tempLinks,
202 : std::vector<InsQuePtr> &tempInsQues)
203 : {
204 0 : HCCL_INFO("[InsTempReduceScatterNHR] GenExtIns start");
205 0 : if (IsPcieLink(tempLinks)) {
206 0 : dmaMode_ = DmaMode::GET;
207 : }
208 0 : opMode_ = tempFuncs.opMode;
209 0 : enableCounterNotify_ = tempFuncs.enableCounterNotify;
210 0 : tempAlgParams_ = tempAlgParams;
211 0 : tempLinks_ = tempLinks;
212 0 : buffInfo_ = tempAlgParams_.buffInfo;
213 :
214 0 : uint32_t linkNum = tempLinks.begin()->second.size();
215 : // 流的数量不能少于linkNum
216 0 : CHK_PRT_RET(linkNum > tempInsQues.size(), HCCL_ERROR("[CollAlgFactory] [InsTempAllReduceNHR] Rank [%d], requiredQue Error.", myRank_),
217 : HcclResult::HCCL_E_INTERNAL);
218 :
219 0 : std::vector<float> dataSplitRate(linkNum);
220 0 : CHK_RET(CalcDataSplitRateForLinks(tempLinks.begin()->second, dataSplitRate));
221 : //将一个RankSliceInfo，拆分成linkNum 个RankSliceInfo
222 0 : u64 typeSize = DataTypeSizeGet(dataType_);
223 0 : u64 dataCnt = tempAlgParams.sliceSize / typeSize;
224 0 : std::vector<RankSliceInfo> sliceInfoVecForAllLinks(linkNum);
225 0 : u64 linkSliceOffset = 0;
226 0 : for (u32 linkIdx = 0; linkIdx < linkNum; linkIdx++) {
227 0 : processSize_.emplace_back();
228 0 : linkSliceOffsetVec_.emplace_back();
229 0 : HCCL_INFO("[InsTempAllReduceNHR] linkIdx =%u",linkIdx);
230 0 : if (linkIdx != linkNum - 1) {
231 0 : processSize_[linkIdx] = static_cast<u64>(static_cast<float>(dataCnt) * dataSplitRate[linkIdx]) * typeSize;
232 : } else {
233 0 : processSize_[linkIdx] = tempAlgParams.sliceSize - linkSliceOffset;
234 : }
235 0 : linkSliceOffsetVec_[linkIdx] = linkSliceOffset;
236 0 : linkSliceOffset += processSize_[linkIdx];
237 : }
238 :
239 0 : u32 mainQueIdx = 0;
240 : // 流间前同步，主流通知从流，只有一个流则不做任何事
241 0 : CHK_RET(PreSyncQues(tempInsQues, mainQueIdx));
242 :
243 0 : for (uint32_t linkIdx = 0; linkIdx < linkNum; linkIdx++) {
244 0 : CHK_RET(LocalDataCopy(tempInsQues, tempFuncs, linkIdx));
245 :
246 0 : if (tempRankSize_ <= 1) {
247 0 : CHK_RET(PostLocalCopy(tempInsQues, linkIdx));//input->scratch
248 0 : return HcclResult::HCCL_SUCCESS;
249 : }
250 :
251 0 : CHK_RET(RunNHR(tempInsQues, linkIdx));
252 0 : CHK_RET(PostLocalCopy(tempInsQues, linkIdx));
253 : }
254 0 : CHK_RET(PostSyncQues(tempInsQues, mainQueIdx));
255 0 : return HcclResult::HCCL_SUCCESS;
256 0 : }
257 :
258 :
259 0 : HcclResult InsTempReduceScatterNHR::LocalDataCopy(std::vector<InsQuePtr> &tempInsQues, const TempFuncs &tempFuncs, u32 linkIdx)
260 : {
261 0 : CHK_PRT_RET(tempInsQues.empty(),
262 : HCCL_ERROR("[RS-NHR][LocalDataCopy] empty queue"), HcclResult::HCCL_E_INTERNAL);
263 0 : InsQuePtr q = tempInsQues[linkIdx];
264 : u64 inBaseOff;
265 : u64 inOff;
266 0 : const u64 rptNum = std::max<u64>(1, tempAlgParams_.repeatNum);
267 0 : for (u32 localRandId = 0; localRandId < tempRankSize_; ++localRandId) {
268 0 : for (u64 rpt = 0; rpt < rptNum; ++rpt) {
269 0 : if (tempFuncs.isBottom) { // 后nhr 前一半数据
270 0 : inBaseOff = tempAlgParams_.buffInfo.inBuffBaseOff +
271 0 : rpt * tempAlgParams_.inputRepeatStride;
272 0 : inOff = inBaseOff + localRandId * tempAlgParams_.inputSliceStride + linkSliceOffsetVec_[linkIdx];
273 : } else { // 前nhr，后一半数据
274 0 : inBaseOff = tempAlgParams_.buffInfo.inBuffBaseOff +
275 0 : localRandId * tempAlgParams_.inputRepeatStride;
276 0 : inOff = inBaseOff + rpt * tempAlgParams_.inputSliceStride + linkSliceOffsetVec_[linkIdx];
277 : }
278 0 : const u64 scratchBase = tempAlgParams_.buffInfo.scratchBuffBaseOff +
279 0 : rpt * tempAlgParams_.outputRepeatStride;
280 0 : const u64 scOff = scratchBase + localRandId * tempAlgParams_.sliceSize + linkSliceOffsetVec_[linkIdx];
281 : // 如果源地址和目标地址相同，则不需要做拷贝
282 0 : if (tempAlgParams_.buffInfo.inBuffType != tempAlgParams_.buffInfo.scratBuffType || inOff != scOff) {
283 0 : DataSlice src(tempAlgParams_.buffInfo.inBuffType, inOff, processSize_[linkIdx]);
284 0 : DataSlice dst(tempAlgParams_.buffInfo.scratBuffType, scOff, processSize_[linkIdx]);
285 0 : auto ins = std::make_unique<InsLocalCopy>(src, dst);
286 0 : q->Append(std::move(ins));
287 0 : }
288 : }
289 : }
290 0 : return HcclResult::HCCL_SUCCESS;
291 0 : }
292 :
293 :
294 0 : HcclResult InsTempReduceScatterNHR::PostLocalCopy(std::vector<InsQuePtr> &tempInsQues, u32 linkIdx)
295 : {
296 0 : CHK_PRT_RET(tempInsQues.empty(),
297 : HCCL_ERROR("[RS-NHR][PostLocalCopy] empty queue"), HcclResult::HCCL_E_INTERNAL);
298 :
299 0 : const u32 myAlgIdx = tempVirtRankMap_.at(myRank_);
300 0 : InsQuePtr q = tempInsQues[linkIdx];
301 :
302 0 : const u64 rptNum = std::max<u64>(1, tempAlgParams_.repeatNum);
303 0 : for (u64 rpt = 0; rpt < rptNum; ++rpt) {
304 0 : const u64 outBaseOff = tempAlgParams_.buffInfo.outBuffBaseOff
305 0 : + rpt * tempAlgParams_.outputRepeatStride;
306 0 : const u64 scratchBase = tempAlgParams_.buffInfo.scratchBuffBaseOff
307 0 : + rpt * tempAlgParams_.outputRepeatStride;
308 :
309 0 : const u64 scOff = scratchBase + tempAlgParams_.sliceSize * myAlgIdx + linkSliceOffsetVec_[linkIdx];
310 0 : const u64 outOff = outBaseOff + linkSliceOffsetVec_[linkIdx];
311 0 : DataSlice src(tempAlgParams_.buffInfo.scratBuffType, scOff, processSize_[linkIdx]);
312 0 : DataSlice dst(tempAlgParams_.buffInfo.outBuffType, outOff, processSize_[linkIdx]);
313 0 : if (tempAlgParams_.buffInfo.scratBuffType != tempAlgParams_.buffInfo.outBuffType || scOff != outOff) {
314 0 : auto ins = std::make_unique<InsLocalCopy>(src, dst);
315 0 : q->Append(std::move(ins));
316 0 : }
317 : }
318 0 : return HcclResult::HCCL_SUCCESS;
319 0 : }
320 :
321 0 : HcclResult InsTempReduceScatterNHR::RunNHR(std::vector<InsQuePtr> &tempInsQues, u32 linkIdx)
322 : {
323 0 : CHK_PRT_RET(tempInsQues.empty(),
324 : HCCL_ERROR("[RS-NHR][RunNHR] empty queue"), HcclResult::HCCL_E_INTERNAL);
325 :
326 0 : if (tempRankSize_ <= 1) return HcclResult::HCCL_SUCCESS;
327 :
328 0 : InsQuePtr q = tempInsQues[linkIdx];
329 :
330 : // 步进参数
331 0 : const u64 rptNum = std::max<u64>(1, tempAlgParams_.repeatNum);
332 :
333 : // 预计算步骤列表（算法序）
334 0 : std::vector<AicpuNHRStepInfo> steps;
335 0 : CHK_RET(GetStepInfoList(steps));
336 0 : for (u32 s = 0; s < steps.size(); ++s) {
337 0 : const auto &st = steps[s];
338 :
339 0 : const RankId recvFromRank = GetRankFromMap(st.fromRank);
340 0 : const RankId sendToRank = GetRankFromMap(st.toRank);
341 0 : CHK_PRT_RET(recvFromRank == static_cast<RankId>(-1) || sendToRank == static_cast<RankId>(-1),
342 : HCCL_ERROR("[RS-NHR][RunNHR] rank map failed: from[%u] to[%u]", st.fromRank, st.toRank),
343 : HcclResult::HCCL_E_INTERNAL);
344 :
345 0 : auto itRecv = tempLinks_.find(recvFromRank);
346 0 : auto itSend = tempLinks_.find(sendToRank);
347 0 : CHK_PRT_RET(itRecv == tempLinks_.end() || itRecv->second.empty() ||
348 : itSend == tempLinks_.end() || itSend->second.empty(),
349 : HCCL_ERROR("[RS-NHR][RunNHR] link missing: recvFrom=%d sendTo=%d", recvFromRank, sendToRank),
350 : HcclResult::HCCL_E_INTERNAL);
351 :
352 0 : const LinkData &linkRecv = itRecv->second[linkIdx];
353 0 : const LinkData &linkSend = itSend->second[linkIdx];
354 :
355 0 : std::vector<DataSlice> txSlices;
356 0 : std::vector<DataSlice> rxSlices;
357 0 : txSlices.reserve(st.nSlices);
358 0 : rxSlices.reserve(st.nSlices);
359 0 : for (u64 rpt = 0; rpt < rptNum; ++rpt) {
360 0 : const u64 scratchBase = tempAlgParams_.buffInfo.scratchBuffBaseOff
361 0 : + rpt * tempAlgParams_.outputRepeatStride;
362 :
363 : // RS：在 SCRATCH 上进行规约交换
364 0 : for (u32 i = 0; i < st.nSlices; ++i) {
365 0 : const u32 txIdx = st.txSliceIdxs[i]; // 算法序
366 0 : const u32 rxIdx = st.rxSliceIdxs[i];
367 0 : const u64 txScOff = scratchBase + tempAlgParams_.sliceSize * txIdx + linkSliceOffsetVec_[linkIdx];
368 0 : const u64 rxScOff = scratchBase + tempAlgParams_.sliceSize * rxIdx + linkSliceOffsetVec_[linkIdx];
369 0 : txSlices.emplace_back(tempAlgParams_.buffInfo.scratBuffType, txScOff, processSize_[linkIdx]);
370 0 : rxSlices.emplace_back(tempAlgParams_.buffInfo.scratBuffType, rxScOff, processSize_[linkIdx]);
371 : }
372 : }
373 : SendRecvReduceInfo info{
374 : { linkSend, linkRecv }, { { txSlices, txSlices }, { rxSlices, rxSlices } }, dataType_, redOp_
375 0 : };
376 :
377 0 : CHK_PRT_RET(SendRecvReduce(info, tempInsQues[linkIdx], 0, true, dmaMode_),
378 : HCCL_ERROR("[RS-NHR][RunNHR] SendRecvReduce failed (step=%u)", st.step), HcclResult::HCCL_E_INTERNAL);
379 0 : }
380 :
381 0 : return HcclResult::HCCL_SUCCESS;
382 0 : }
383 :
384 :
385 : // 计算每轮收发的对端以及slice编号
386 0 : HcclResult InsTempReduceScatterNHR::GetStepInfoList(std::vector<AicpuNHRStepInfo> &stepInfoList)
387 : {
388 : // 将本 rank 号转换成算法使用的索引号
389 0 : u32 rankIdx = tempVirtRankMap_[myRank_];
390 0 : stepInfoList.clear();
391 :
392 0 : u32 nSteps = GetNHRStepNum(tempRankSize_);
393 0 : stepInfoList.resize(nSteps);
394 0 : for (u32 step = 0; step < nSteps; step++) {
395 : // 计算通信对象
396 0 : u32 deltaRank = 1 << step;
397 0 : u32 sendTo = (rankIdx + tempRankSize_ - deltaRank) % tempRankSize_;
398 0 : u32 recvFrom = (rankIdx + deltaRank) % tempRankSize_;
399 :
400 : // 数据份数和数据编号增量
401 0 : u32 nSlices = (tempRankSize_ - 1 + (1 << step)) / (1 << (step + 1));
402 0 : u32 deltaSliceIndex = 1 << (step + 1);
403 0 : u32 txSliceIdx = sendTo;
404 0 : u32 rxSliceIdx = rankIdx;
405 :
406 0 : AicpuNHRStepInfo &currStepInfo = stepInfoList[step];
407 0 : currStepInfo.step = step;
408 0 : currStepInfo.myRank = rankIdx;
409 0 : currStepInfo.nSlices = nSlices;
410 0 : currStepInfo.toRank = sendTo;
411 0 : currStepInfo.fromRank = recvFrom;
412 :
413 : // 计算本rank在每轮收/发中的slice编号
414 0 : currStepInfo.txSliceIdxs.reserve(nSlices);
415 0 : currStepInfo.rxSliceIdxs.reserve(nSlices);
416 0 : for (u32 i = 0; i < nSlices; i++) {
417 0 : currStepInfo.txSliceIdxs.push_back(txSliceIdx);
418 0 : currStepInfo.rxSliceIdxs.push_back(rxSliceIdx);
419 0 : HCCL_DEBUG("[InsTempReduceScatterNHR][GetStepInfoList] i[%u] txSliceIdx[%u] rxSliceIdx[%u]", i, txSliceIdx, rxSliceIdx);
420 0 : txSliceIdx = (txSliceIdx + tempRankSize_ - deltaSliceIndex) % tempRankSize_;
421 0 : rxSliceIdx = (rxSliceIdx + tempRankSize_ - deltaSliceIndex) % tempRankSize_;
422 : }
423 : }
424 0 : return HcclResult::HCCL_SUCCESS;
425 : }
426 :
427 0 : RankId InsTempReduceScatterNHR::GetRankFromMap(const u32 rankIdx)
428 : {
429 0 : HCCL_INFO("[InsTempReduceScatterNHR] GetRankFromMap");
430 0 : RankId rank = -1;
431 0 : for (auto &pair : tempVirtRankMap_) {
432 0 : if (pair.second == rankIdx) {
433 0 : rank = pair.first;
434 0 : break;
435 : }
436 : }
437 0 : return rank;
438 : }
439 :
440 0 : HcclResult InsTempReduceScatterNHR::GetScratchBufferInfo(const uint64_t scratchBufferSize, DataType dataType) const
441 : {
442 : (void)scratchBufferSize;
443 : (void)dataType;
444 0 : return HcclResult::HCCL_SUCCESS;
445 : }
446 : } // namespace Hccl
|
