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(const RankId virtualRank, const u32 tempRankSize,
17 0 : const std::vector<std::vector<RankId>> &tempVTopo, const std::map<RankId, u32> &tempVirtRankMap)
18 0 : : InsAlgTemplateBase(virtualRank, tempRankSize, tempVTopo, tempVirtRankMap)
19 0 : {}
20 :
21 0 : InsTempReduceScatterMesh1D::~InsTempReduceScatterMesh1D()
22 0 : {}
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("[InsTempReduceScatterMesh1D] Inconsistency pathNum to remoteRanks, Previous consistent pathNum=[%u], mismatched "
42 : "remoteRank=[%d], pathNum=[%u]",
43 : pathNum,
44 : remoteRank,
45 : 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(const TempFuncs &tempFuncs, const TemplateDataParams &tempAlgParams,
71 : const ResLinks &tempLinks, std::vector<InsQuePtr> &tempInsQues)
72 : {
73 0 : opMode_ = tempFuncs.opMode;
74 0 : enableCounterNotify_ = tempFuncs.enableCounterNotify;
75 0 : queNum_ = tempInsQues.size();
76 0 : HCCL_INFO("[InsTempReduceScatterMesh1D] Run Start");
77 0 : uint32_t linkNum = tempLinks.begin()->second.size();
78 0 : CHK_PRT_RET(linkNum > tempInsQues.size(), HCCL_ERROR("[CollAlgFactory] [InsTempReduceScatterMesh1D] Rank [%d], requiredQue Error.", myRank_),
79 : HcclResult::HCCL_E_INTERNAL);
80 :
81 0 : if (queNum_ > 1) {
82 0 : CHK_RET(PreSyncInterQueues(tempInsQues));
83 : }
84 0 : CHK_RET(RunReduceScatter(tempLinks, tempInsQues, tempAlgParams));
85 0 : HCCL_INFO("[InsTempReduceScatterMesh1D][PostCopy] Rank [%d].", myRank_);
86 : // 流间后同步，从流通知主流
87 0 : if (queNum_ > 1) {
88 0 : CHK_RET(PostSyncInterQueues(tempInsQues));
89 : }
90 0 : PostCopy(tempAlgParams, tempInsQues);
91 0 : return HcclResult::HCCL_SUCCESS;
92 : }
93 :
94 0 : HcclResult InsTempReduceScatterMesh1D::PostCopy(
95 : const TemplateDataParams &tempAlgParams, std::vector<InsQuePtr> &tempInsQues)
96 : {
97 : // 通信结束之后，数据都在 inbuff 上，需要搬运到对应的输出位置。
98 0 : u32 rankIdx = tempVirtRankMap_[myRank_];
99 : // 如果是单算子模式, 并且是最后一步算子，需要将数据从 inBuff 拷贝到 userOut
100 : // 是否需要将数据搬运到 OutBuff 上再搬运到 UserOut 上？？
101 0 : HCCL_INFO("[InsTempReduceScatterMesh1D][PostCopy], copy from outBuff to userOut");
102 : // 先把本卡的数据从input搬运到output
103 0 : HCCL_INFO("[InsTempReduceScatterMesh1D][PostCopy]tempAlgParams.repeatNum=%llu", tempAlgParams.repeatNum);
104 0 : u64 sliceSize = ((rankIdx == tempRankSize_ - 1) && (tempAlgParams.tailSize != 0)) ? tempAlgParams.tailSize : tempAlgParams.sliceSize;
105 0 : for (u32 repeatIdx = 0; repeatIdx < tempAlgParams.repeatNum; repeatIdx++) {
106 0 : DataSlice myRankSlice = DataSlice(tempAlgParams.buffInfo.inBuffType, tempAlgParams.buffInfo.inBuffBaseOff +
107 0 : repeatIdx * tempAlgParams.inputRepeatStride + rankIdx * tempAlgParams.inputSliceStride, sliceSize);
108 0 : DataSlice outputSlice = DataSlice(tempAlgParams.buffInfo.outBuffType, tempAlgParams.buffInfo.outBuffBaseOff +
109 0 : repeatIdx * tempAlgParams.outputRepeatStride, sliceSize);
110 : // myRankSlice与outputSlice一致就不进行LocalCopy
111 0 : if (!(tempAlgParams.buffInfo.inBuffType == tempAlgParams.buffInfo.outBuffType
112 0 : && myRankSlice.GetOffset() == outputSlice.GetOffset())) {
113 0 : CHK_RET(LocalCopy(tempInsQues[0], myRankSlice, outputSlice));
114 : }
115 :
116 : // 把其他卡的数据input累加到output
117 0 : for (u32 tmpRank = 0; tmpRank < tempRankSize_; tmpRank++) {
118 0 : if (tmpRank != rankIdx) {
119 0 : DataSlice srcDataSlice = DataSlice(tempAlgParams.buffInfo.scratBuffType, tempAlgParams.buffInfo.scratchBuffBaseOff +
120 0 : repeatIdx * tempAlgParams.outputRepeatStride + tmpRank * sliceSize, sliceSize);
121 0 : DataSlice dstDataSlice = DataSlice(tempAlgParams.buffInfo.outBuffType, tempAlgParams.buffInfo.outBuffBaseOff +
122 0 : repeatIdx * tempAlgParams.outputRepeatStride, sliceSize);
123 0 : CHK_RET(LocalReduce(tempInsQues[0], srcDataSlice, dstDataSlice, dataType_, redOp_));
124 : }
125 : }
126 : }
127 0 : return HcclResult::HCCL_SUCCESS;
128 : }
129 :
130 0 : HcclResult InsTempReduceScatterMesh1D::RunReduceScatter(const ResLinks &tempLinks, std::vector<InsQuePtr> &tempInsQues,
131 : const TemplateDataParams &tempAlgParams)
132 : {
133 : u32 myAlgRank;
134 0 : CHK_RET(GetAlgRank(myRank_, tempVTopo_[0], myAlgRank));
135 : // 控制mesh通信的rankSize - 1个对端
136 0 : u32 queIdx = 0;
137 0 : for (u32 rankIdx = 0; rankIdx < tempRankSize_ - 1; rankIdx++) {
138 0 : u32 nextRank = (myAlgRank + 1 + rankIdx) % tempRankSize_;
139 0 : RankId remoteRank = tempVTopo_[0][nextRank];
140 : u32 rmAlgRank;
141 0 : CHK_RET(GetAlgRank(remoteRank, tempVTopo_[0], rmAlgRank));
142 0 : HCCL_DEBUG("[InsTempReduceScatterMesh1D][RunReduceScatter] myRank[%d], toRank[%d], fromRank[%d], rmAlgRank[%u]", myRank_, remoteRank, remoteRank, rmAlgRank);
143 0 : CHK_PRT_RET(tempLinks.at(remoteRank).empty(), HCCL_ERROR("[InsTempReduceScatterMesh1D][RunReduceScatter] Rank [%d], remoteRank[%d] required links Error.", myRank_, remoteRank),
144 : HcclResult::HCCL_E_INTERNAL);
145 0 : const std::vector<LinkData> &neighborLinkDatas = tempLinks.at(remoteRank);
146 0 : u32 linkNum = rank2PathNumMap_.at(remoteRank);
147 0 : CHK_PRT_RET(linkNum != neighborLinkDatas.size(), HCCL_ERROR("[InsTempReduceScatterMesh1D][RunReduceScatter] Rank [%d], remoteRank[%d] linkNum != neighborLinkDatas.size().", myRank_, remoteRank),
148 : HcclResult::HCCL_E_INTERNAL);
149 0 : std::vector<float> dataSplitRate(linkNum);
150 0 : CHK_RET(CalcDataSplitRateForLinks(neighborLinkDatas, dataSplitRate));
151 0 : for (u32 linkIdx = 0; linkIdx < linkNum; linkIdx++) {
152 0 : CHK_PRT_RET(queIdx >= tempInsQues.size(), HCCL_ERROR("[InsTempReduceScatterMesh1D][RunReduceScatter] queIdx [%u] >= tempInsQues.size() [%zu].", queIdx, tempInsQues.size()),
153 : HcclResult::HCCL_E_INTERNAL);
154 0 : InsQuePtr currQue = tempInsQues[queIdx + 1];
155 0 : queIdx++;
156 0 : const LinkData &neighborLinkData = neighborLinkDatas[linkIdx];
157 0 : std::vector<DataSlice> txSrcSlices;
158 0 : std::vector<DataSlice> txDstSlices;
159 0 : std::vector<DataSlice> rxSrcSlices;
160 0 : std::vector<DataSlice> rxDstSlices;
161 0 : u64 sendSlice = ((rmAlgRank == tempRankSize_ - 1) && (tempAlgParams.tailSize != 0)) ? tempAlgParams.tailSize : tempAlgParams.sliceSize;
162 0 : for (u32 repeatIdx = 0; repeatIdx < tempAlgParams.repeatNum; repeatIdx++) {
163 0 : DataSlice rxSrcSlice = DataSlice(tempAlgParams.buffInfo.inBuffType, tempAlgParams.buffInfo.inBuffBaseOff +
164 0 : repeatIdx * tempAlgParams.inputRepeatStride + myAlgRank * tempAlgParams.inputSliceStride, tempAlgParams.sliceSize); // 接收源
165 0 : DataSlice rxDstSlice = DataSlice(tempAlgParams.buffInfo.scratBuffType, tempAlgParams.buffInfo.scratchBuffBaseOff +
166 0 : repeatIdx * tempAlgParams.outputRepeatStride + nextRank * tempAlgParams.sliceSize, tempAlgParams.sliceSize); // 接收目标
167 0 : DataSlice txSrcSlice = DataSlice(tempAlgParams.buffInfo.inBuffType, tempAlgParams.buffInfo.inBuffBaseOff +
168 0 : repeatIdx * tempAlgParams.inputRepeatStride + nextRank * tempAlgParams.inputSliceStride, sendSlice); // 发送源
169 0 : DataSlice txDstSlice = DataSlice(tempAlgParams.buffInfo.scratBuffType, tempAlgParams.buffInfo.scratchBuffBaseOff +
170 0 : repeatIdx * tempAlgParams.outputRepeatStride + myAlgRank * sendSlice, sendSlice); // 发送目标
171 :
172 0 : txSrcSlices.push_back(CalcDataSliceForLinks(txSrcSlice, dataSplitRate, linkIdx, dataType_));
173 0 : txDstSlices.push_back(CalcDataSliceForLinks(txDstSlice, dataSplitRate, linkIdx, dataType_));
174 0 : rxSrcSlices.push_back(CalcDataSliceForLinks(rxSrcSlice, dataSplitRate, linkIdx, dataType_));
175 0 : rxDstSlices.push_back(CalcDataSliceForLinks(rxDstSlice, dataSplitRate, linkIdx, dataType_));
176 : }
177 0 : SendRecvInfo sendRecvInfo{{neighborLinkData, neighborLinkData}, {{txSrcSlices, txDstSlices},{rxSrcSlices, rxDstSlices}}};
178 0 : CHK_PRT_RET(SendRecv(sendRecvInfo, currQue, 0, true, DmaMode::PUT),
179 : HCCL_ERROR("[InsTempReduceScatterMesh1D] RunReduceScatter SendReduce failed"), HcclResult::HCCL_E_INTERNAL);
180 0 : }
181 0 : }
182 0 : return HcclResult::HCCL_SUCCESS;
183 : }
184 :
185 0 : RankId InsTempReduceScatterMesh1D::GetRankFromMap(const u32 rankIdx)
186 : {
187 0 : RankId rank = -1;
188 0 : HCCL_INFO("[InsTempReduceScatterMesh1D] GetRankFromMap");
189 0 : for (auto &pair : tempVirtRankMap_) {
190 0 : if (pair.second == rankIdx) {
191 0 : rank = pair.first;
192 0 : break;
193 : }
194 : }
195 0 : return rank;
196 : }
197 :
198 : } // namespace Hccl
|
