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(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 : InsTempReduceScatterMesh1DOpt::~InsTempReduceScatterMesh1DOpt()
24 : {
25 0 : }
26 :
27 0 : HcclResult InsTempReduceScatterMesh1DOpt::CalcRes(AlgTempResReq &tempResReq)
28 : {
29 : // Mesh 需要的 que Num 为 tempVTopo_[0].size()-1
30 0 : tempResReq.queNum = (tempVTopo_[0].size() > 1) ? tempVTopo_[0].size() : 1;
31 0 : tempResReq.streamNum = tempResReq.queNum;
32 0 : tempResReq.queNotifys = CreateMasterSlaveQueNotifiesRequest(tempResReq.queNum);
33 0 : QId centerQ = 0;
34 0 : tempResReq.localWaitGroupCntNotify.emplace_back(centerQ, 0);
35 0 : tempResReq.localBcastPostCntNotify.emplace_back(centerQ, 0);
36 : // linkNumBtwPeers_这个在没有绕路的情况下，是设置成1
37 0 : CHK_PRT_RET(CalcResLinksMesh(myRank_, tempRankSize_, tempVTopo_, linkNumBtwPeers_, tempResReq) != HcclResult::HCCL_SUCCESS,
38 : HCCL_ERROR("[CollAlgFactory] [InsTempReduceScatterMesh1DOpt] Rank [%d], resLinks calculation error!", myRank_),
39 : HcclResult::HCCL_E_INTERNAL);
40 :
41 0 : return HcclResult::HCCL_SUCCESS;
42 : }
43 :
44 0 : u64 InsTempReduceScatterMesh1DOpt::CalcScratchMultiple(const BufferType &inBuffType, const BufferType &outBuffType) const
45 : {
46 : (void)inBuffType;
47 : (void)outBuffType;
48 0 : u64 scratchMultiple = tempRankSize_;
49 0 : return scratchMultiple;
50 : }
51 :
52 0 : HcclResult InsTempReduceScatterMesh1DOpt::GenExtIns(const TempFuncs &tempFuncs, const TemplateDataParams &tempAlgParams,
53 : const ResLinks &tempLinks, 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 [%llu], 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(queNum_ != tempInsQues.size(),
63 : HCCL_ERROR("[CollAlgFactory] [InsTempReduceScatterMesh1DOpt] Rank [%d], requiredQue Error.", myRank_),
64 : HcclResult::HCCL_E_INTERNAL);
65 0 : if (queNum_ > 1) {
66 0 : CHK_RET(PreSyncInterQueues(tempInsQues));
67 : }
68 0 : CHK_RET(RunReduceScatter(tempLinks, tempInsQues, tempAlgParams));
69 0 : if (queNum_ > 1) {
70 0 : CHK_RET(PostSyncInterQueues(tempInsQues));
71 : }
72 0 : PostCopy(tempAlgParams, tempInsQues);
73 0 : return HcclResult::HCCL_SUCCESS;
74 : }
75 :
76 0 : HcclResult InsTempReduceScatterMesh1DOpt::PostCopy(const TemplateDataParams &tempAlgParams, std::vector<InsQuePtr> &tempInsQues)
77 : {
78 : // 通信结束之后，数据都在 inbuff 上，需要搬运到对应的输出位置。
79 0 : u32 rankIdx = tempVirtRankMap_[myRank_];
80 : // 如果是单算子模式, 并且是最后一步算子，需要将数据从 inBuff 拷贝到 userOut
81 : // 是否需要将数据搬运到 OutBuff 上再搬运到 UserOut 上？？
82 0 : HCCL_INFO("[InsTempReduceScatterMesh1DOpt][PostCopy], copy from outBuff to userOut");
83 : // 先把本卡的数据从input搬运到output
84 0 : HCCL_INFO("[InsTempReduceScatterMesh1DOpt][PostCopy]tempAlgParams.repeatNum=%llu", tempAlgParams.repeatNum);
85 0 : u64 sliceSize = ((rankIdx == tempRankSize_ - 1) && (tempAlgParams.tailSize != 0)) ? tempAlgParams.tailSize : tempAlgParams.sliceSize;
86 0 : for (u32 repeatIdx = 0; repeatIdx < tempAlgParams.repeatNum; repeatIdx++) {
87 0 : DataSlice myRankSlice = DataSlice(tempAlgParams.buffInfo.inBuffType, tempAlgParams.buffInfo.inBuffBaseOff +
88 0 : repeatIdx * tempAlgParams.inputRepeatStride + rankIdx * tempAlgParams.inputSliceStride, sliceSize);
89 0 : DataSlice outputSlice = DataSlice(tempAlgParams.buffInfo.outBuffType, tempAlgParams.buffInfo.outBuffBaseOff +
90 0 : repeatIdx * tempAlgParams.outputRepeatStride, sliceSize);
91 0 : CHK_RET(LocalCopy(tempInsQues[0], myRankSlice, outputSlice));
92 : // 把其他卡的数据input累加到output
93 0 : for (u32 tmpRank = 0; tmpRank < tempRankSize_; tmpRank++) {
94 0 : if (tmpRank != rankIdx) {
95 0 : DataSlice srcDataSlice = DataSlice(tempAlgParams.buffInfo.scratBuffType, tempAlgParams.buffInfo.scratchBuffBaseOff +
96 0 : repeatIdx * tempAlgParams.outputRepeatStride + tmpRank * sliceSize, sliceSize);
97 0 : DataSlice dstDataSlice = DataSlice(tempAlgParams.buffInfo.outBuffType, tempAlgParams.buffInfo.outBuffBaseOff +
98 0 : repeatIdx * tempAlgParams.outputRepeatStride, sliceSize);
99 0 : CHK_RET(LocalReduce(tempInsQues[0], srcDataSlice, dstDataSlice, dataType_, redOp_));
100 : }
101 : }
102 : }
103 0 : return HcclResult::HCCL_SUCCESS;
104 : }
105 :
106 0 : HcclResult InsTempReduceScatterMesh1DOpt::RunReduceScatter(const ResLinks &tempLinks, std::vector<InsQuePtr> &tempInsQues,
107 : const TemplateDataParams &tempAlgParams)
108 : {
109 : u32 myAlgRank;
110 0 : CHK_RET(GetAlgRank(myRank_, tempVTopo_[0], myAlgRank));
111 0 : for (u32 queIdx = 1; queIdx < queNum_; queIdx++) {
112 0 : u32 nextRank = (myAlgRank + queIdx) % tempRankSize_;
113 0 : RankId remoteRank = tempVTopo_[0][nextRank];
114 :
115 : u32 rmAlgRank;
116 0 : CHK_RET(GetAlgRank(remoteRank, tempVTopo_[0], rmAlgRank));
117 :
118 0 : HCCL_DEBUG("[InsTempReduceScatterMesh1DOpt][RunReduceScatter] myRank[%d], toRank[%d], fromRank[%d]",
119 : myRank_, remoteRank, remoteRank);
120 0 : const std::vector<LinkData> &linkRecv = tempLinks.at(remoteRank);
121 0 : const std::vector<LinkData> &linkSend = tempLinks.at(remoteRank);
122 0 : std::vector<DataSlice> txSrcSlices;
123 0 : std::vector<DataSlice> txDstSlices;
124 0 : std::vector<DataSlice> rxSrcSlices;
125 0 : std::vector<DataSlice> rxDstSlices;
126 :
127 0 : u64 sendSlice = ((rmAlgRank == tempRankSize_ - 1) && (tempAlgParams.tailSize != 0)) ? tempAlgParams.tailSize : tempAlgParams.sliceSize;
128 0 : u64 recvSlice = ((myAlgRank == tempRankSize_ - 1) && (tempAlgParams.tailSize != 0)) ? tempAlgParams.tailSize : tempAlgParams.sliceSize;
129 :
130 : // 在 inBuff 上进行 ReduceScatter 操作
131 : // 数据从其他卡，传输到本卡，接收数据
132 0 : for (u32 repeatIdx = 0; repeatIdx < tempAlgParams.repeatNum; repeatIdx++) {
133 0 : DataSlice rxSrcSlice = DataSlice(tempAlgParams.buffInfo.inBuffType, tempAlgParams.buffInfo.inBuffBaseOff +
134 0 : repeatIdx * tempAlgParams.inputRepeatStride + myAlgRank * tempAlgParams.inputSliceStride, recvSlice); // 接收源
135 0 : DataSlice rxDstSlice = DataSlice(tempAlgParams.buffInfo.scratBuffType, tempAlgParams.buffInfo.scratchBuffBaseOff +
136 0 : repeatIdx * tempAlgParams.outputRepeatStride + nextRank * recvSlice, recvSlice); // 接收目标
137 0 : DataSlice txSrcSlice = DataSlice(tempAlgParams.buffInfo.inBuffType, tempAlgParams.buffInfo.inBuffBaseOff +
138 0 : repeatIdx * tempAlgParams.inputRepeatStride + nextRank * tempAlgParams.inputSliceStride, sendSlice); // 发送源
139 0 : DataSlice txDstSlice = DataSlice(tempAlgParams.buffInfo.scratBuffType, tempAlgParams.buffInfo.scratchBuffBaseOff +
140 0 : repeatIdx * tempAlgParams.outputRepeatStride + myAlgRank * sendSlice, sendSlice); // 发送目标
141 :
142 0 : rxSrcSlices.push_back(rxSrcSlice);
143 0 : rxDstSlices.push_back(rxDstSlice);
144 0 : txSrcSlices.push_back(txSrcSlice);
145 0 : txDstSlices.push_back(txDstSlice);
146 : }
147 0 : SendRecvInfo sendRecvInfo{{linkSend[0],linkRecv[0]},
148 0 : {{txSrcSlices, txDstSlices},{rxSrcSlices, rxDstSlices}}};
149 :
150 0 : CHK_PRT_RET(SendRecv(sendRecvInfo, tempInsQues[queIdx], 0, true, DmaMode::PUT),
151 : HCCL_ERROR("[InsTempReduceScatterMesh1DOpt] RunReduceScatter SendReduce failed"),
152 : HcclResult::HCCL_E_INTERNAL);
153 0 : }
154 0 : return HcclResult::HCCL_SUCCESS;
155 : }
156 :
157 0 : RankId InsTempReduceScatterMesh1DOpt::GetRankFromMap(const u32 rankIdx)
158 : {
159 0 : RankId rank = -1;
160 0 : for (auto &pair : tempVirtRankMap_) {
161 0 : if (pair.second == rankIdx) {
162 0 : rank = pair.first;
163 0 : break;
164 : }
165 : }
166 0 : return rank;
167 : }
168 :
169 : } // namespace Hccl
|
