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_mesh_1D_meshchunk.h"
12 : #include "log.h"
13 : #include "alg_data_trans_wrapper.h"
14 :
15 : namespace Hccl {
16 0 : InsTempReduceScatterMesh1DMeshChunk::InsTempReduceScatterMesh1DMeshChunk(
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 : InsTempReduceScatterMesh1DMeshChunk::~InsTempReduceScatterMesh1DMeshChunk() {}
23 :
24 0 : HcclResult InsTempReduceScatterMesh1DMeshChunk::CalcRes(AlgTempResReq& tempResReq)
25 : {
26 : // Mesh 需要的 que Num 为 tempVTopo_[0].size()-1
27 0 : tempResReq.queNum = (tempVTopo_[0].size() > 1) ? tempVTopo_[0].size() - 1 : 1;
28 0 : tempResReq.streamNum = tempResReq.queNum;
29 0 : tempResReq.queNotifys = CreateMasterSlaveQueNotifiesRequest(tempResReq.queNum);
30 0 : QId centerQ = 0;
31 0 : tempResReq.localWaitGroupCntNotify.emplace_back(centerQ, 0);
32 0 : tempResReq.localBcastPostCntNotify.emplace_back(centerQ, 0);
33 : // linkNumBtwPeers_这个在没有绕路的情况下，是设置成1
34 0 : CHK_PRT_RET(
35 : CalcResLinksMesh(myRank_, tempRankSize_, tempVTopo_, linkNumBtwPeers_, tempResReq) != HcclResult::HCCL_SUCCESS,
36 : HCCL_ERROR(
37 : "[CollAlgFactory] [InsTempReduceScatterMesh1DMeshChunk] Rank [%d], resLinks calculation error!", myRank_),
38 : HcclResult::HCCL_E_INTERNAL);
39 :
40 0 : return HcclResult::HCCL_SUCCESS;
41 : }
42 :
43 0 : u64 InsTempReduceScatterMesh1DMeshChunk::CalcScratchMultiple(
44 : const BufferType& inBuffType, const BufferType& outBuffType) const
45 : {
46 : (void)inBuffType;
47 : (void)outBuffType;
48 0 : u64 scratchMultiple = tempRankSize_ - 1;
49 0 : return scratchMultiple;
50 : }
51 :
52 0 : HcclResult InsTempReduceScatterMesh1DMeshChunk::CalcSliceInfoVec(const u64& dataSize, RankSliceInfo& sliceInfoVec)
53 : {
54 0 : std::vector<SliceInfo> tmp(tempVTopo_.size());
55 0 : sliceInfoVec.resize(tempRankSize_, tmp);
56 0 : u64 accumOff = 0;
57 0 : for (u32 rankIdx = 0; rankIdx < sliceInfoVec.size(); rankIdx++) {
58 0 : SliceInfo slice = {accumOff, dataSize};
59 0 : sliceInfoVec[rankIdx][0] = slice;
60 0 : accumOff += dataSize;
61 : }
62 0 : CHK_PRT_RET(
63 : (sliceInfoVec[tempRankSize_ - 1][0].offset + sliceInfoVec[tempRankSize_ - 1][0].size
64 : != dataSize * tempRankSize_),
65 : HCCL_ERROR("[CollAlgFactory] Rank [%d], SliceInfo calculation error!", myRank_), HcclResult::HCCL_E_INTERNAL);
66 :
67 0 : return HcclResult::HCCL_SUCCESS;
68 0 : }
69 :
70 0 : HcclResult InsTempReduceScatterMesh1DMeshChunk::GenExtIns(
71 : const TempFuncs& tempFuncs, const TemplateDataParams& tempAlgParams, const ResLinks& tempLinks,
72 : std::vector<InsQuePtr>& tempInsQues)
73 : {
74 0 : opMode_ = tempFuncs.opMode;
75 0 : enableCounterNotify_ = tempFuncs.enableCounterNotify;
76 0 : queNum_ = tempVTopo_[0].size() - 1;
77 0 : processSize_ = tempAlgParams.sliceSize;
78 0 : rankIdx_ = tempVirtRankMap_[myRank_];
79 0 : RankSliceInfo sliceInfoVec;
80 0 : CHK_RET(CalcSliceInfoVec(tempAlgParams.sliceSize, sliceInfoVec));
81 0 : HCCL_INFO("[InsTempReduceScatterMesh1DMeshChunk] Run Start");
82 : // 这里不支持绕路的时候，应该就用原始的tempInsQues就行
83 0 : CHK_PRT_RET(
84 : queNum_ != tempInsQues.size(),
85 : HCCL_ERROR("[CollAlgFactory] [InsTempReduceScatterMesh1DMeshChunk] Rank [%d], requiredQue Error.", myRank_),
86 : HcclResult::HCCL_E_INTERNAL);
87 0 : PreCopy(tempAlgParams, tempInsQues);
88 0 : if (queNum_ > 1) {
89 0 : CHK_RET(PreSyncInterQueues(tempInsQues));
90 : }
91 :
92 0 : CHK_RET(RunReduceScatter(tempLinks, tempInsQues, tempAlgParams, sliceInfoVec));
93 :
94 0 : if (queNum_ > 1) {
95 0 : CHK_RET(PostSyncInterQueues(tempInsQues));
96 : }
97 0 : PostCopy(tempAlgParams, tempInsQues);
98 0 : return HcclResult::HCCL_SUCCESS;
99 0 : }
100 :
101 0 : HcclResult InsTempReduceScatterMesh1DMeshChunk::PreCopy(
102 : const TemplateDataParams& tempAlgParams, std::vector<InsQuePtr>& tempInsQues) const
103 : {
104 0 : HCCL_INFO("[InsTempReduceScatterMesh1DMeshChunk][PreCopy], copy from userIn to scratch");
105 0 : for (u32 repeatIdx = 0; repeatIdx < tempAlgParams.repeatNum; repeatIdx++) {
106 : DataSlice srcSlice = DataSlice(
107 : tempAlgParams.buffInfo.inBuffType,
108 0 : tempAlgParams.buffInfo.inBuffBaseOff + repeatIdx * tempAlgParams.inputRepeatStride
109 0 : + rankIdx_ * tempAlgParams.inputSliceStride,
110 0 : processSize_);
111 : DataSlice dstSlice
112 0 : = DataSlice(tempAlgParams.buffInfo.scratBuffType, tempAlgParams.buffInfo.scratchBuffBaseOff, processSize_);
113 0 : CHK_RET(LocalCopy(tempInsQues[0], srcSlice, dstSlice));
114 : }
115 0 : return HcclResult::HCCL_SUCCESS;
116 : }
117 :
118 0 : HcclResult InsTempReduceScatterMesh1DMeshChunk::RunReduceScatter(
119 : const ResLinks& tempLinks, std::vector<InsQuePtr>& tempInsQues, const TemplateDataParams& tempAlgParams,
120 : RankSliceInfo& sliceInfoVec)
121 : {
122 0 : HCCL_INFO("[InsTempReduceScatterMesh1DMeshChunk][RunReduceScatter] myRank[%d]", myRank_);
123 : u32 myAlgRank;
124 0 : CHK_RET(GetAlgRank(myRank_, tempVTopo_[0], myAlgRank));
125 :
126 0 : uint64_t sliceNum = tempRankSize_ - 1;
127 0 : uint64_t mySliceSize = sliceInfoVec[myAlgRank][0].size; // 获取本rank需要处理的数据量
128 0 : uint64_t mySliceCount = mySliceSize / DataTypeSizeGet(op_.dataType);
129 : // 数据切分为sliceNum块，当数据量不能均匀切分时，后面smallDataSliceNum个数据块比前面bigDataSliceNum个数据块每块少1个数据
130 0 : uint64_t bigDataSliceNum = mySliceCount % sliceNum;
131 0 : uint64_t bigDataSliceSize = (mySliceCount / sliceNum + 1) * DataTypeSizeGet(op_.dataType);
132 0 : uint64_t smallDataSliceNum = sliceNum - mySliceCount % sliceNum;
133 0 : uint64_t smallDataSliceSize = mySliceCount / sliceNum * DataTypeSizeGet(op_.dataType);
134 :
135 0 : std::vector<uint64_t> sliceSize;
136 0 : for (uint64_t i = 0; i < bigDataSliceNum; i++) {
137 0 : sliceSize.push_back(bigDataSliceSize);
138 : }
139 0 : for (uint64_t i = 0; i < smallDataSliceNum; i++) {
140 0 : sliceSize.push_back(smallDataSliceSize);
141 : }
142 0 : uint64_t sliceRecvBaseOffset = 0;
143 0 : uint16_t rankNum = 2;
144 0 : for (uint16_t i = 0; i < (tempRankSize_ - rankNum); i++) {
145 0 : sliceRecvBaseOffset += sliceSize[i];
146 : }
147 0 : for (u32 repeatIdx = 0; repeatIdx < tempAlgParams.repeatNum; repeatIdx++) {
148 : uint64_t sliceSendOffset_;
149 : uint64_t sliceRecvOffset_;
150 0 : DoMeshChunk(
151 : tempLinks, tempInsQues, tempAlgParams, sliceSize, repeatIdx, myAlgRank, sliceSendOffset_, sliceRecvOffset_,
152 : sliceRecvBaseOffset);
153 : }
154 0 : return HcclResult::HCCL_SUCCESS;
155 0 : }
156 :
157 0 : HcclResult InsTempReduceScatterMesh1DMeshChunk::DoMeshChunk(
158 : const ResLinks& tempLinks, std::vector<InsQuePtr>& tempInsQues, const TemplateDataParams& tempAlgParams,
159 : const std::vector<uint64_t>& sliceSize, const u32& repeatIdx, const u32& myAlgRank, uint64_t& sliceSendOffset_,
160 : uint64_t& sliceRecvOffset_, const uint64_t& sliceRecvBaseOffset)
161 : {
162 0 : for (uint16_t stepIdx = 0; stepIdx < (tempRankSize_ - 1); stepIdx++) {
163 0 : sliceSendOffset_ = 0;
164 0 : sliceRecvOffset_ = sliceRecvBaseOffset;
165 0 : uint16_t rankNum = 2;
166 0 : uint16_t tempNum = 3;
167 0 : for (uint16_t i = 0; i < (tempRankSize_ - 1); i++) {
168 0 : uint16_t nextNum = stepIdx + i + 1;
169 0 : if (nextNum >= tempRankSize_) {
170 0 : nextNum += 1;
171 : }
172 0 : uint16_t nextRank = (myAlgRank + nextNum) % tempRankSize_;
173 0 : uint16_t frontNum = 2 * myAlgRank - nextRank + tempRankSize_;
174 0 : uint16_t frontRank = frontNum % tempRankSize_;
175 0 : RankId toRank = tempVTopo_[0][frontRank];
176 : uint16_t queIdx;
177 0 : if (frontRank < myAlgRank) {
178 0 : queIdx = frontRank;
179 : } else {
180 0 : queIdx = frontRank - 1;
181 : }
182 : DataSlice rxSrcSlice = DataSlice(
183 : tempAlgParams.buffInfo.inBuffType,
184 0 : tempAlgParams.buffInfo.inBuffBaseOff + repeatIdx * tempAlgParams.inputRepeatStride
185 0 : + myAlgRank * tempAlgParams.inputSliceStride + sliceRecvOffset_,
186 0 : sliceSize[i]); // 接收源
187 : DataSlice rxDstSlice = DataSlice(
188 0 : tempAlgParams.buffInfo.scratBuffType, tempAlgParams.buffInfo.scratchBuffBaseOff + sliceRecvOffset_,
189 0 : sliceSize[i]); // 接收目标
190 : DataSlice txSrcSlice = DataSlice(
191 : tempAlgParams.buffInfo.inBuffType,
192 0 : tempAlgParams.buffInfo.inBuffBaseOff + repeatIdx * tempAlgParams.inputRepeatStride
193 0 : + frontRank * tempAlgParams.inputSliceStride + sliceSendOffset_,
194 0 : sliceSize[i]); // 发送源
195 : DataSlice txDstSlice = DataSlice(
196 0 : tempAlgParams.buffInfo.scratBuffType, tempAlgParams.buffInfo.scratchBuffBaseOff + sliceSendOffset_,
197 0 : sliceSize[i]); // 发送目标
198 :
199 0 : u32 rankFromRank = GetRankFromMap(toRank);
200 0 : auto it = tempLinks.find(rankFromRank);
201 0 : if (it == tempLinks.end()) {
202 0 : HCCL_ERROR("rankFromRank [%u] not in tempLinks.", rankFromRank);
203 0 : return HcclResult::HCCL_E_PARA;
204 : }
205 0 : const std::vector<LinkData>& linkRecv = tempLinks.at(GetRankFromMap(toRank));
206 0 : const std::vector<LinkData>& linkSend = tempLinks.at(GetRankFromMap(toRank));
207 0 : std::vector<DataSlice> txSrcSlices;
208 0 : std::vector<DataSlice> txDstSlices;
209 0 : std::vector<DataSlice> rxSrcSlices;
210 0 : std::vector<DataSlice> rxDstSlices;
211 0 : rxSrcSlices.push_back(rxSrcSlice);
212 0 : rxDstSlices.push_back(rxDstSlice);
213 0 : txSrcSlices.push_back(txSrcSlice);
214 0 : txDstSlices.push_back(txDstSlice);
215 :
216 : SendRecvReduceInfo sendRecvReduceInfo{
217 0 : {linkSend[0], linkRecv[0]},
218 : {{txSrcSlices, txDstSlices}, {rxSrcSlices, rxDstSlices}},
219 : dataType_,
220 0 : redOp_};
221 :
222 0 : CHK_PRT_RET(
223 : SendRecvReduce(sendRecvReduceInfo, tempInsQues[queIdx], 0, true, DmaMode::PUT),
224 : HCCL_ERROR("[InsTempReduceScatterMesh1DMeshChunk] RunReduceScatter SendRecvReduce failed"),
225 : HcclResult::HCCL_E_INTERNAL);
226 :
227 0 : sliceSendOffset_ += sliceSize[i];
228 0 : if (tempRankSize_ > rankNum && i < (tempRankSize_ - rankNum)) {
229 0 : sliceRecvOffset_ -= sliceSize[tempRankSize_ - tempNum - i];
230 : }
231 0 : }
232 0 : if (queNum_ > 1 && stepIdx < (tempRankSize_ - rankNum)) {
233 0 : CHK_RET(PostSyncInterQueues(tempInsQues));
234 0 : CHK_RET(PreSyncInterQueues(tempInsQues));
235 : }
236 : }
237 0 : return HcclResult::HCCL_SUCCESS;
238 : }
239 :
240 0 : HcclResult InsTempReduceScatterMesh1DMeshChunk::PostCopy(
241 : const TemplateDataParams& tempAlgParams, std::vector<InsQuePtr>& tempInsQues)
242 : {
243 : // 如果是单算子模式, 并且是最后一步算子，需要将数据从 scratch 拷贝到 userOut
244 0 : HCCL_INFO("[InsTempReduceScatterMesh1DMeshChunk][PostCopy], copy from scratch to userOut");
245 : u32 myAlgRank;
246 0 : CHK_RET(GetAlgRank(myRank_, tempVTopo_[0], myAlgRank));
247 : // 先把本卡的数据从input搬运到output
248 0 : for (u32 repeatIdx = 0; repeatIdx < tempAlgParams.repeatNum; repeatIdx++) {
249 : DataSlice myRankSlice
250 0 : = DataSlice(tempAlgParams.buffInfo.scratBuffType, tempAlgParams.buffInfo.scratchBuffBaseOff, processSize_);
251 : DataSlice outputSlice
252 0 : = DataSlice(tempAlgParams.buffInfo.outBuffType, tempAlgParams.buffInfo.outBuffBaseOff, processSize_);
253 0 : CHK_RET(LocalCopy(tempInsQues[0], myRankSlice, outputSlice));
254 : }
255 0 : return HcclResult::HCCL_SUCCESS;
256 : }
257 :
258 0 : RankId InsTempReduceScatterMesh1DMeshChunk::GetRankFromMap(const u32 rankIdx)
259 : {
260 0 : RankId rank = -1;
261 0 : for (auto& pair : tempVirtRankMap_) {
262 0 : if (pair.second == rankIdx) {
263 0 : rank = pair.first;
264 0 : break;
265 : }
266 : }
267 0 : return rank;
268 : }
269 : } // namespace Hccl
|
