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.h"
12 : #include "log.h"
13 : #include "alg_data_trans_wrapper.h"
14 :
15 : namespace Hccl {
16 0 : InsTempReduceScatterMesh1D::InsTempReduceScatterMesh1D(
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 : InsTempReduceScatterMesh1D::~InsTempReduceScatterMesh1D() {}
23 :
24 0 : HcclResult InsTempReduceScatterMesh1D::CalcRes(AlgTempResReq& tempResReq)
25 : {
26 0 : CHK_PRT_RET(
27 : CalcResLinksMesh(myRank_, tempRankSize_, tempVTopo_, linkNumBtwPeers_, tempResReq) != HcclResult::HCCL_SUCCESS,
28 : HCCL_ERROR("[CollAlgFactory] [InsTempReduceScatterMesh1D] Rank [%d], resLinks calculation error!", myRank_),
29 : HcclResult::HCCL_E_INTERNAL);
30 0 : auto& linkReq = tempResReq.links;
31 0 : u32 pathNum = 0;
32 0 : for (auto resReqIter = linkReq.begin(); resReqIter != linkReq.end(); resReqIter++) {
33 0 : auto remoteRank = resReqIter->first;
34 0 : if (rank2PathNumMap_.find(remoteRank) == rank2PathNumMap_.end() || rank2PathNumMap_[remoteRank] == 0) {
35 0 : HCCL_ERROR("[InsTempReduceScatterMesh1D] No path to remoteRank[%d]", remoteRank);
36 0 : return HcclResult::HCCL_E_INTERNAL;
37 : }
38 0 : if (pathNum == 0) {
39 0 : pathNum = rank2PathNumMap_[remoteRank];
40 0 : } else if (rank2PathNumMap_[remoteRank] != pathNum) {
41 0 : HCCL_ERROR(
42 : "[InsTempReduceScatterMesh1D] Inconsistency pathNum to remoteRanks, Previous consistent pathNum=[%u], "
43 : "mismatched "
44 : "remoteRank=[%d], pathNum=[%u]",
45 : pathNum, remoteRank, rank2PathNumMap_[remoteRank]);
46 0 : return HcclResult::HCCL_E_INTERNAL;
47 : }
48 0 : resReqIter->second = pathNum;
49 : }
50 :
51 : // Mesh 需要的 que Num 为 tempVTopo_[0].size()-1
52 0 : tempResReq.queNum = (tempVTopo_[0].size() > 1) ? (tempVTopo_[0].size()) * pathNum : pathNum;
53 0 : tempResReq.streamNum = tempResReq.queNum;
54 0 : tempResReq.queNotifys = CreateMasterSlaveQueNotifiesRequest(tempResReq.queNum);
55 0 : QId centerQ = 0;
56 0 : tempResReq.localWaitGroupCntNotify.emplace_back(centerQ, 0);
57 0 : tempResReq.localBcastPostCntNotify.emplace_back(centerQ, 0);
58 :
59 0 : return HcclResult::HCCL_SUCCESS;
60 : }
61 :
62 0 : u64 InsTempReduceScatterMesh1D::CalcScratchMultiple(const BufferType& inBuffType, const BufferType& outBuffType) const
63 : {
64 : (void)inBuffType;
65 : (void)outBuffType;
66 0 : u64 scratchMultiple = tempRankSize_;
67 0 : return scratchMultiple;
68 : }
69 :
70 0 : HcclResult InsTempReduceScatterMesh1D::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_ = tempInsQues.size();
77 0 : HCCL_INFO("[InsTempReduceScatterMesh1D] Run Start");
78 0 : uint32_t linkNum = tempLinks.begin()->second.size();
79 0 : CHK_PRT_RET(
80 : linkNum > tempInsQues.size(),
81 : HCCL_ERROR("[CollAlgFactory] [InsTempReduceScatterMesh1D] Rank [%d], requiredQue Error.", myRank_),
82 : HcclResult::HCCL_E_INTERNAL);
83 :
84 0 : if (queNum_ > 1) {
85 0 : CHK_RET(PreSyncInterQueues(tempInsQues));
86 : }
87 0 : CHK_RET(RunReduceScatter(tempLinks, tempInsQues, tempAlgParams));
88 0 : HCCL_INFO("[InsTempReduceScatterMesh1D][PostCopy] Rank [%d].", myRank_);
89 : // 流间后同步，从流通知主流
90 0 : if (queNum_ > 1) {
91 0 : CHK_RET(PostSyncInterQueues(tempInsQues));
92 : }
93 0 : PostCopy(tempAlgParams, tempInsQues);
94 0 : return HcclResult::HCCL_SUCCESS;
95 : }
96 :
97 : HcclResult
98 0 : InsTempReduceScatterMesh1D::PostCopy(const TemplateDataParams& tempAlgParams, std::vector<InsQuePtr>& tempInsQues)
99 : {
100 : // 通信结束之后，数据都在 inbuff 上，需要搬运到对应的输出位置。
101 0 : u32 rankIdx = tempVirtRankMap_[myRank_];
102 : // 如果是单算子模式, 并且是最后一步算子，需要将数据从 inBuff 拷贝到 userOut
103 : // 是否需要将数据搬运到 OutBuff 上再搬运到 UserOut 上？？
104 0 : HCCL_INFO("[InsTempReduceScatterMesh1D][PostCopy], copy from outBuff to userOut");
105 : // 先把本卡的数据从input搬运到output
106 0 : HCCL_INFO("[InsTempReduceScatterMesh1D][PostCopy]tempAlgParams.repeatNum=%llu", tempAlgParams.repeatNum);
107 0 : u64 sliceSize = ((rankIdx == tempRankSize_ - 1) && (tempAlgParams.tailSize != 0)) ? tempAlgParams.tailSize :
108 : tempAlgParams.sliceSize;
109 0 : for (u32 repeatIdx = 0; repeatIdx < tempAlgParams.repeatNum; repeatIdx++) {
110 : DataSlice myRankSlice = DataSlice(
111 : tempAlgParams.buffInfo.inBuffType,
112 0 : tempAlgParams.buffInfo.inBuffBaseOff + repeatIdx * tempAlgParams.inputRepeatStride
113 0 : + rankIdx * tempAlgParams.inputSliceStride,
114 0 : sliceSize);
115 : DataSlice outputSlice = DataSlice(
116 : tempAlgParams.buffInfo.outBuffType,
117 0 : tempAlgParams.buffInfo.outBuffBaseOff + repeatIdx * tempAlgParams.outputRepeatStride, sliceSize);
118 : // myRankSlice与outputSlice一致就不进行LocalCopy
119 0 : if (!(tempAlgParams.buffInfo.inBuffType == tempAlgParams.buffInfo.outBuffType
120 0 : && myRankSlice.GetOffset() == outputSlice.GetOffset())) {
121 0 : CHK_RET(LocalCopy(tempInsQues[0], myRankSlice, outputSlice));
122 : }
123 :
124 : // 把其他卡的数据input累加到output
125 0 : for (u32 tmpRank = 0; tmpRank < tempRankSize_; tmpRank++) {
126 0 : if (tmpRank != rankIdx) {
127 : DataSlice srcDataSlice = DataSlice(
128 : tempAlgParams.buffInfo.scratBuffType,
129 0 : tempAlgParams.buffInfo.scratchBuffBaseOff + repeatIdx * tempAlgParams.outputRepeatStride
130 0 : + tmpRank * sliceSize,
131 0 : sliceSize);
132 : DataSlice dstDataSlice = DataSlice(
133 : tempAlgParams.buffInfo.outBuffType,
134 0 : tempAlgParams.buffInfo.outBuffBaseOff + repeatIdx * tempAlgParams.outputRepeatStride, sliceSize);
135 0 : CHK_RET(LocalReduce(tempInsQues[0], srcDataSlice, dstDataSlice, dataType_, redOp_));
136 : }
137 : }
138 : }
139 0 : return HcclResult::HCCL_SUCCESS;
140 : }
141 :
142 0 : HcclResult InsTempReduceScatterMesh1D::RunReduceScatter(
143 : const ResLinks& tempLinks, std::vector<InsQuePtr>& tempInsQues, const TemplateDataParams& tempAlgParams)
144 : {
145 : u32 myAlgRank;
146 0 : CHK_RET(GetAlgRank(myRank_, tempVTopo_[0], myAlgRank));
147 : // 控制mesh通信的rankSize - 1个对端
148 0 : u32 queIdx = 0;
149 0 : for (u32 rankIdx = 0; rankIdx < tempRankSize_ - 1; rankIdx++) {
150 0 : u32 nextRank = (myAlgRank + 1 + rankIdx) % tempRankSize_;
151 0 : RankId remoteRank = tempVTopo_[0][nextRank];
152 : u32 rmAlgRank;
153 0 : CHK_RET(GetAlgRank(remoteRank, tempVTopo_[0], rmAlgRank));
154 0 : HCCL_DEBUG(
155 : "[InsTempReduceScatterMesh1D][RunReduceScatter] myRank[%d], toRank[%d], fromRank[%d], rmAlgRank[%u]",
156 : myRank_, remoteRank, remoteRank, rmAlgRank);
157 0 : CHK_PRT_RET(
158 : tempLinks.at(remoteRank).empty(),
159 : HCCL_ERROR(
160 : "[InsTempReduceScatterMesh1D][RunReduceScatter] Rank [%d], remoteRank[%d] required links Error.",
161 : myRank_, remoteRank),
162 : HcclResult::HCCL_E_INTERNAL);
163 0 : const std::vector<LinkData>& neighborLinkDatas = tempLinks.at(remoteRank);
164 0 : u32 linkNum = rank2PathNumMap_.at(remoteRank);
165 0 : CHK_PRT_RET(
166 : linkNum != neighborLinkDatas.size(),
167 : HCCL_ERROR(
168 : "[InsTempReduceScatterMesh1D][RunReduceScatter] Rank [%d], remoteRank[%d] linkNum != "
169 : "neighborLinkDatas.size().",
170 : myRank_, remoteRank),
171 : HcclResult::HCCL_E_INTERNAL);
172 0 : std::vector<float> dataSplitRate(linkNum);
173 0 : CHK_RET(CalcDataSplitRateForLinks(neighborLinkDatas, dataSplitRate));
174 0 : for (u32 linkIdx = 0; linkIdx < linkNum; linkIdx++) {
175 0 : CHK_PRT_RET(
176 : queIdx >= tempInsQues.size(),
177 : HCCL_ERROR(
178 : "[InsTempReduceScatterMesh1D][RunReduceScatter] queIdx [%u] >= tempInsQues.size() [%zu].", queIdx,
179 : tempInsQues.size()),
180 : HcclResult::HCCL_E_INTERNAL);
181 0 : InsQuePtr currQue = tempInsQues[queIdx + 1];
182 0 : queIdx++;
183 0 : const LinkData& neighborLinkData = neighborLinkDatas[linkIdx];
184 0 : std::vector<DataSlice> txSrcSlices;
185 0 : std::vector<DataSlice> txDstSlices;
186 0 : std::vector<DataSlice> rxSrcSlices;
187 0 : std::vector<DataSlice> rxDstSlices;
188 0 : u64 sendSlice = ((rmAlgRank == tempRankSize_ - 1) && (tempAlgParams.tailSize != 0)) ?
189 : tempAlgParams.tailSize :
190 : tempAlgParams.sliceSize;
191 0 : for (u32 repeatIdx = 0; repeatIdx < tempAlgParams.repeatNum; repeatIdx++) {
192 : DataSlice rxSrcSlice = DataSlice(
193 : tempAlgParams.buffInfo.inBuffType,
194 0 : tempAlgParams.buffInfo.inBuffBaseOff + repeatIdx * tempAlgParams.inputRepeatStride
195 0 : + myAlgRank * tempAlgParams.inputSliceStride,
196 0 : tempAlgParams.sliceSize); // 接收源
197 : DataSlice rxDstSlice = DataSlice(
198 : tempAlgParams.buffInfo.scratBuffType,
199 0 : tempAlgParams.buffInfo.scratchBuffBaseOff + repeatIdx * tempAlgParams.outputRepeatStride
200 0 : + nextRank * tempAlgParams.sliceSize,
201 0 : tempAlgParams.sliceSize); // 接收目标
202 : DataSlice txSrcSlice = DataSlice(
203 : tempAlgParams.buffInfo.inBuffType,
204 0 : tempAlgParams.buffInfo.inBuffBaseOff + repeatIdx * tempAlgParams.inputRepeatStride
205 0 : + nextRank * tempAlgParams.inputSliceStride,
206 0 : sendSlice); // 发送源
207 : DataSlice txDstSlice = DataSlice(
208 : tempAlgParams.buffInfo.scratBuffType,
209 0 : tempAlgParams.buffInfo.scratchBuffBaseOff + repeatIdx * tempAlgParams.outputRepeatStride
210 0 : + myAlgRank * sendSlice,
211 0 : sendSlice); // 发送目标
212 :
213 0 : txSrcSlices.push_back(CalcDataSliceForLinks(txSrcSlice, dataSplitRate, linkIdx, dataType_));
214 0 : txDstSlices.push_back(CalcDataSliceForLinks(txDstSlice, dataSplitRate, linkIdx, dataType_));
215 0 : rxSrcSlices.push_back(CalcDataSliceForLinks(rxSrcSlice, dataSplitRate, linkIdx, dataType_));
216 0 : rxDstSlices.push_back(CalcDataSliceForLinks(rxDstSlice, dataSplitRate, linkIdx, dataType_));
217 : }
218 : SendRecvInfo sendRecvInfo{
219 0 : {neighborLinkData, neighborLinkData}, {{txSrcSlices, txDstSlices}, {rxSrcSlices, rxDstSlices}}};
220 0 : CHK_PRT_RET(
221 : SendRecv(sendRecvInfo, currQue, 0, true, DmaMode::PUT),
222 : HCCL_ERROR("[InsTempReduceScatterMesh1D] RunReduceScatter SendReduce failed"),
223 : HcclResult::HCCL_E_INTERNAL);
224 0 : }
225 0 : }
226 0 : return HcclResult::HCCL_SUCCESS;
227 : }
228 :
229 0 : RankId InsTempReduceScatterMesh1D::GetRankFromMap(const u32 rankIdx)
230 : {
231 0 : RankId rank = -1;
232 0 : HCCL_INFO("[InsTempReduceScatterMesh1D] GetRankFromMap");
233 0 : for (auto& pair : tempVirtRankMap_) {
234 0 : if (pair.second == rankIdx) {
235 0 : rank = pair.first;
236 0 : break;
237 : }
238 : }
239 0 : return rank;
240 : }
241 :
242 : } // namespace Hccl
|
