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_opt.h"
12 : #include "log.h"
13 : #include "alg_data_trans_wrapper.h"
14 :
15 : namespace Hccl {
16 0 : InsTempReduceScatterMesh1DOpt::InsTempReduceScatterMesh1DOpt(
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 : InsTempReduceScatterMesh1DOpt::~InsTempReduceScatterMesh1DOpt() {}
23 :
24 0 : HcclResult InsTempReduceScatterMesh1DOpt::CalcRes(AlgTempResReq& tempResReq)
25 : {
26 : // Mesh 需要的 que Num 为 tempVTopo_[0].size()-1
27 0 : tempResReq.queNum = (tempVTopo_[0].size() > 1) ? tempVTopo_[0].size() : 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("[CollAlgFactory] [InsTempReduceScatterMesh1DOpt] Rank [%d], resLinks calculation error!", myRank_),
37 : HcclResult::HCCL_E_INTERNAL);
38 :
39 0 : return HcclResult::HCCL_SUCCESS;
40 : }
41 :
42 0 : u64 InsTempReduceScatterMesh1DOpt::CalcScratchMultiple(
43 : const BufferType& inBuffType, const BufferType& outBuffType) const
44 : {
45 : (void)inBuffType;
46 : (void)outBuffType;
47 0 : u64 scratchMultiple = tempRankSize_;
48 0 : return scratchMultiple;
49 : }
50 :
51 0 : HcclResult InsTempReduceScatterMesh1DOpt::GenExtIns(
52 : const TempFuncs& tempFuncs, const TemplateDataParams& tempAlgParams, const ResLinks& tempLinks,
53 : std::vector<InsQuePtr>& tempInsQues)
54 : {
55 0 : opMode_ = tempFuncs.opMode;
56 0 : enableCounterNotify_ = tempFuncs.enableCounterNotify;
57 0 : queNum_ = tempVTopo_[0].size();
58 0 : HCCL_INFO("LGC tempVTopo_.size() is [%zu], ReduceScatter queNum is [%llu]", tempVTopo_.size(), queNum_);
59 0 : processSize_ = tempAlgParams.sliceSize;
60 0 : HCCL_INFO("[InsTempReduceScatterMesh1DOpt] Run Start");
61 : // 这里不支持绕路的时候，应该就用原始的tempInsQues就行
62 0 : CHK_PRT_RET(
63 : queNum_ != tempInsQues.size(),
64 : HCCL_ERROR("[CollAlgFactory] [InsTempReduceScatterMesh1DOpt] Rank [%d], requiredQue Error.", myRank_),
65 : HcclResult::HCCL_E_INTERNAL);
66 0 : if (queNum_ > 1) {
67 0 : CHK_RET(PreSyncInterQueues(tempInsQues));
68 : }
69 0 : CHK_RET(RunReduceScatter(tempLinks, tempInsQues, tempAlgParams));
70 0 : if (queNum_ > 1) {
71 0 : CHK_RET(PostSyncInterQueues(tempInsQues));
72 : }
73 0 : PostCopy(tempAlgParams, tempInsQues);
74 0 : return HcclResult::HCCL_SUCCESS;
75 : }
76 :
77 : HcclResult
78 0 : InsTempReduceScatterMesh1DOpt::PostCopy(const TemplateDataParams& tempAlgParams, std::vector<InsQuePtr>& tempInsQues)
79 : {
80 : // 通信结束之后，数据都在 inbuff 上，需要搬运到对应的输出位置。
81 0 : u32 rankIdx = tempVirtRankMap_[myRank_];
82 : // 如果是单算子模式, 并且是最后一步算子，需要将数据从 inBuff 拷贝到 userOut
83 : // 是否需要将数据搬运到 OutBuff 上再搬运到 UserOut 上？？
84 0 : HCCL_INFO("[InsTempReduceScatterMesh1DOpt][PostCopy], copy from outBuff to userOut");
85 : // 先把本卡的数据从input搬运到output
86 0 : HCCL_INFO("[InsTempReduceScatterMesh1DOpt][PostCopy]tempAlgParams.repeatNum=%llu", tempAlgParams.repeatNum);
87 0 : u64 sliceSize = ((rankIdx == tempRankSize_ - 1) && (tempAlgParams.tailSize != 0)) ? tempAlgParams.tailSize :
88 : tempAlgParams.sliceSize;
89 0 : for (u32 repeatIdx = 0; repeatIdx < tempAlgParams.repeatNum; repeatIdx++) {
90 : DataSlice myRankSlice = DataSlice(
91 : tempAlgParams.buffInfo.inBuffType,
92 0 : tempAlgParams.buffInfo.inBuffBaseOff + repeatIdx * tempAlgParams.inputRepeatStride
93 0 : + rankIdx * tempAlgParams.inputSliceStride,
94 0 : sliceSize);
95 : DataSlice outputSlice = DataSlice(
96 : tempAlgParams.buffInfo.outBuffType,
97 0 : tempAlgParams.buffInfo.outBuffBaseOff + repeatIdx * tempAlgParams.outputRepeatStride, sliceSize);
98 0 : CHK_RET(LocalCopy(tempInsQues[0], myRankSlice, outputSlice));
99 : // 把其他卡的数据input累加到output
100 0 : for (u32 tmpRank = 0; tmpRank < tempRankSize_; tmpRank++) {
101 0 : if (tmpRank != rankIdx) {
102 : DataSlice srcDataSlice = DataSlice(
103 : tempAlgParams.buffInfo.scratBuffType,
104 0 : tempAlgParams.buffInfo.scratchBuffBaseOff + repeatIdx * tempAlgParams.outputRepeatStride
105 0 : + tmpRank * sliceSize,
106 0 : sliceSize);
107 : DataSlice dstDataSlice = DataSlice(
108 : tempAlgParams.buffInfo.outBuffType,
109 0 : tempAlgParams.buffInfo.outBuffBaseOff + repeatIdx * tempAlgParams.outputRepeatStride, sliceSize);
110 0 : CHK_RET(LocalReduce(tempInsQues[0], srcDataSlice, dstDataSlice, dataType_, redOp_));
111 : }
112 : }
113 : }
114 0 : return HcclResult::HCCL_SUCCESS;
115 : }
116 :
117 0 : HcclResult InsTempReduceScatterMesh1DOpt::RunReduceScatter(
118 : const ResLinks& tempLinks, std::vector<InsQuePtr>& tempInsQues, const TemplateDataParams& tempAlgParams)
119 : {
120 : u32 myAlgRank;
121 0 : CHK_RET(GetAlgRank(myRank_, tempVTopo_[0], myAlgRank));
122 0 : for (u32 queIdx = 1; queIdx < queNum_; queIdx++) {
123 0 : u32 nextRank = (myAlgRank + queIdx) % tempRankSize_;
124 0 : RankId remoteRank = tempVTopo_[0][nextRank];
125 :
126 : u32 rmAlgRank;
127 0 : CHK_RET(GetAlgRank(remoteRank, tempVTopo_[0], rmAlgRank));
128 :
129 0 : HCCL_DEBUG(
130 : "[InsTempReduceScatterMesh1DOpt][RunReduceScatter] myRank[%d], toRank[%d], fromRank[%d]", myRank_,
131 : remoteRank, remoteRank);
132 0 : const std::vector<LinkData>& linkRecv = tempLinks.at(remoteRank);
133 0 : const std::vector<LinkData>& linkSend = tempLinks.at(remoteRank);
134 0 : std::vector<DataSlice> txSrcSlices;
135 0 : std::vector<DataSlice> txDstSlices;
136 0 : std::vector<DataSlice> rxSrcSlices;
137 0 : std::vector<DataSlice> rxDstSlices;
138 :
139 0 : u64 sendSlice = ((rmAlgRank == tempRankSize_ - 1) && (tempAlgParams.tailSize != 0)) ? tempAlgParams.tailSize :
140 : tempAlgParams.sliceSize;
141 0 : u64 recvSlice = ((myAlgRank == tempRankSize_ - 1) && (tempAlgParams.tailSize != 0)) ? tempAlgParams.tailSize :
142 : tempAlgParams.sliceSize;
143 :
144 : // 在 inBuff 上进行 ReduceScatter 操作
145 : // 数据从其他卡，传输到本卡，接收数据
146 0 : for (u32 repeatIdx = 0; repeatIdx < tempAlgParams.repeatNum; repeatIdx++) {
147 : DataSlice rxSrcSlice = DataSlice(
148 : tempAlgParams.buffInfo.inBuffType,
149 0 : tempAlgParams.buffInfo.inBuffBaseOff + repeatIdx * tempAlgParams.inputRepeatStride
150 0 : + myAlgRank * tempAlgParams.inputSliceStride,
151 0 : recvSlice); // 接收源
152 : DataSlice rxDstSlice = DataSlice(
153 : tempAlgParams.buffInfo.scratBuffType,
154 0 : tempAlgParams.buffInfo.scratchBuffBaseOff + repeatIdx * tempAlgParams.outputRepeatStride
155 0 : + nextRank * recvSlice,
156 0 : recvSlice); // 接收目标
157 : DataSlice txSrcSlice = DataSlice(
158 : tempAlgParams.buffInfo.inBuffType,
159 0 : tempAlgParams.buffInfo.inBuffBaseOff + repeatIdx * tempAlgParams.inputRepeatStride
160 0 : + nextRank * tempAlgParams.inputSliceStride,
161 0 : sendSlice); // 发送源
162 : DataSlice txDstSlice = DataSlice(
163 : tempAlgParams.buffInfo.scratBuffType,
164 0 : tempAlgParams.buffInfo.scratchBuffBaseOff + repeatIdx * tempAlgParams.outputRepeatStride
165 0 : + myAlgRank * sendSlice,
166 0 : sendSlice); // 发送目标
167 :
168 0 : rxSrcSlices.push_back(rxSrcSlice);
169 0 : rxDstSlices.push_back(rxDstSlice);
170 0 : txSrcSlices.push_back(txSrcSlice);
171 0 : txDstSlices.push_back(txDstSlice);
172 : }
173 0 : SendRecvInfo sendRecvInfo{{linkSend[0], linkRecv[0]}, {{txSrcSlices, txDstSlices}, {rxSrcSlices, rxDstSlices}}};
174 :
175 0 : CHK_PRT_RET(
176 : SendRecv(sendRecvInfo, tempInsQues[queIdx], 0, true, DmaMode::PUT),
177 : HCCL_ERROR("[InsTempReduceScatterMesh1DOpt] RunReduceScatter SendReduce failed"),
178 : HcclResult::HCCL_E_INTERNAL);
179 0 : }
180 0 : return HcclResult::HCCL_SUCCESS;
181 : }
182 :
183 0 : RankId InsTempReduceScatterMesh1DOpt::GetRankFromMap(const u32 rankIdx)
184 : {
185 0 : RankId rank = -1;
186 0 : for (auto& pair : tempVirtRankMap_) {
187 0 : if (pair.second == rankIdx) {
188 0 : rank = pair.first;
189 0 : break;
190 : }
191 : }
192 0 : return rank;
193 : }
194 :
195 : } // namespace Hccl
|
