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 "log.h"
12 :
13 : #include "alg_data_trans_wrapper.h"
14 : #include "ins_temp_reduce_scatter_aicpu_reduce.h"
15 :
16 : namespace Hccl {
17 0 : InsTempReduceScatterAicpuReduce::InsTempReduceScatterAicpuReduce(
18 : const RankId virtualRank, const u32 tempRankSize, const std::vector<std::vector<RankId>>& tempVTopo,
19 0 : const std::map<RankId, u32>& tempVirtRankMap)
20 0 : : InsAlgTemplateBase(virtualRank, tempRankSize, tempVTopo, tempVirtRankMap)
21 0 : {}
22 :
23 0 : InsTempReduceScatterAicpuReduce::~InsTempReduceScatterAicpuReduce() {}
24 :
25 0 : HcclResult InsTempReduceScatterAicpuReduce::CalcRes(AlgTempResReq& tempResReq)
26 : {
27 0 : tempResReq.queNum = tempVTopo_[0].size();
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 0 : CHK_RET(CalcResLinksMesh(myRank_, tempRankSize_, tempVTopo_, linkNumBtwPeers_, tempResReq));
34 0 : HCCL_DEBUG(
35 : "[InsTempReduceScatterAicpuReduce]CalcRes: queNum[%u], myRank[%d], tempRankSize[%u]", tempResReq.queNum,
36 : myRank_, tempRankSize_);
37 0 : return HcclResult::HCCL_SUCCESS;
38 : }
39 :
40 0 : u32 InsTempReduceScatterAicpuReduce::CalcScratchMultiple(BufferType inBuffType, BufferType outBuffType) const
41 : {
42 : (void)inBuffType;
43 : (void)outBuffType;
44 0 : return tempRankSize_;
45 : }
46 :
47 0 : HcclResult InsTempReduceScatterAicpuReduce::RunAlltoAllMesh(
48 : const TempFuncs& tempFuncs, const TemplateDataParams& templateDataParams, const ResLinks& tempLinks,
49 : std::vector<InsQuePtr>& tempInsQues)
50 : {
51 : (void)tempFuncs;
52 0 : CHK_RET(PreSyncInterQueues(tempInsQues));
53 : // 本端rank数据从本端input -> 本端scratch
54 0 : u64 srcOffset = templateDataParams.inputSliceStride * u32(myRank_) + templateDataParams.buffInfo.inBuffBaseOff;
55 0 : u64 srcSize = templateDataParams.sliceSize;
56 0 : u64 dstOffset = templateDataParams.sliceSize * u32(myRank_);
57 0 : DataSlice srcSlice = DataSlice(BufferType::INPUT, srcOffset, srcSize);
58 0 : DataSlice dstSlice = DataSlice(BufferType::SCRATCH, dstOffset, srcSize);
59 0 : std::unique_ptr<Instruction> insLocalCopy = std::make_unique<InsLocalCopy>(srcSlice, dstSlice);
60 0 : tempInsQues[0]->Append(std::move(insLocalCopy));
61 :
62 : // 本端rank数据从本端input -> 对端scratch
63 :
64 : u32 myAlgRank;
65 0 : CHK_RET(GetAlgRank(myRank_, tempVTopo_[0], myAlgRank));
66 0 : for (u32 queIdx = 1; queIdx < tempVTopo_[0].size(); queIdx++) {
67 0 : RankId neighborRank = tempVTopo_[0][(myAlgRank + queIdx) % tempRankSize_];
68 0 : LinkData neighborLinkData = tempLinks.at(neighborRank)[0];
69 0 : TxRxLinks sendRecvLinks(neighborLinkData, neighborLinkData);
70 : // send
71 0 : std::vector<DataSlice> txSrcSlices;
72 0 : std::vector<DataSlice> txDstSlices;
73 : DataSlice currSendSliceSrc = DataSlice(
74 : BufferType::INPUT,
75 0 : templateDataParams.inputSliceStride * u32(neighborRank) + templateDataParams.buffInfo.inBuffBaseOff,
76 0 : templateDataParams.sliceSize);
77 : DataSlice currSendSliceDst
78 0 : = DataSlice(BufferType::SCRATCH, templateDataParams.sliceSize * u32(myRank_), templateDataParams.sliceSize);
79 0 : txSrcSlices.push_back(currSendSliceSrc);
80 0 : txDstSlices.push_back(currSendSliceDst);
81 : // recv
82 : DataSlice currRecvSliceSrc = DataSlice(
83 0 : BufferType::SCRATCH, templateDataParams.inputSliceStride * u32(neighborRank), templateDataParams.sliceSize);
84 : DataSlice currRecvSliceDst = DataSlice(
85 : BufferType::SCRATCH,
86 0 : templateDataParams.inputSliceStride * u32(neighborRank) + templateDataParams.buffInfo.inBuffBaseOff,
87 0 : templateDataParams.sliceSize);
88 0 : std::vector<DataSlice> rxSrcSlices;
89 0 : std::vector<DataSlice> rxDstSlices;
90 0 : rxSrcSlices.push_back(currRecvSliceSrc);
91 0 : rxDstSlices.push_back(currRecvSliceDst);
92 0 : TxRxSlicesList sendRecvSlicesList({txSrcSlices, txDstSlices}, {rxSrcSlices, rxDstSlices});
93 0 : SendRecvInfo sendRecvInfo(sendRecvLinks, sendRecvSlicesList);
94 0 : CHK_PRT_RET(
95 : SendRecv(sendRecvInfo, tempInsQues[queIdx], 0, true, dmaMode_),
96 : HCCL_ERROR("[InsTempReduceScatterAicpuReduce] RunReduceScatter sendrecv failed"),
97 : HcclResult::HCCL_E_INTERNAL);
98 0 : }
99 0 : CHK_RET(PostSyncInterQueues(tempInsQues));
100 0 : return HCCL_SUCCESS;
101 0 : }
102 :
103 0 : HcclResult InsTempReduceScatterAicpuReduce::RunAicpuLocalReduce(
104 : const TemplateDataParams& templateDataParams, std::vector<InsQuePtr>& tempInsQues)
105 : {
106 0 : DataSlice dataSlice = DataSlice(BufferType::SCRATCH, 0, templateDataParams.sliceSize);
107 0 : for (u32 rankId = 1; rankId < tempRankSize_; rankId++) {
108 : DataSlice reduceSlice
109 0 : = DataSlice(BufferType::SCRATCH, templateDataParams.sliceSize * rankId, templateDataParams.sliceSize);
110 0 : AicpuReduce(tempInsQues[0], reduceSlice, dataSlice, dataType_, redOp_);
111 : }
112 : DataSlice outputSlice
113 0 : = DataSlice(BufferType::OUTPUT, templateDataParams.buffInfo.inBuffBaseOff, templateDataParams.sliceSize);
114 0 : LocalCopy(tempInsQues[0], dataSlice, outputSlice);
115 0 : return HCCL_SUCCESS;
116 : }
117 :
118 0 : HcclResult InsTempReduceScatterAicpuReduce::GenExtIns(
119 : const TempFuncs& tempFuncs, const TemplateDataParams& templateDataParams, const ResLinks& tempLinks,
120 : std::vector<InsQuePtr>& tempInsQues)
121 : {
122 0 : HCCL_INFO("[InsTempReduceScatterAicpuReduce] Run start");
123 0 : if (IsPcieLink(tempLinks)) {
124 0 : dmaMode_ = DmaMode::GET;
125 : } else {
126 0 : dmaMode_ = DmaMode::PUT;
127 : }
128 0 : if (tempVTopo_[0].size() == 1) {
129 0 : return HcclResult::HCCL_SUCCESS;
130 : }
131 0 : opMode_ = tempFuncs.opMode;
132 0 : queNum_ = tempVTopo_[0].size();
133 0 : CHK_PRT_RET(
134 : queNum_ != tempInsQues.size(),
135 : HCCL_ERROR("[CollAlgFactory] [InsTempReduceScatterAicpuReduce] Rank [%d], requiredQue Error.", myRank_),
136 : HcclResult::HCCL_E_INTERNAL);
137 :
138 0 : RunAlltoAllMesh(tempFuncs, templateDataParams, tempLinks, tempInsQues);
139 0 : StreamSync(tempInsQues);
140 0 : RunAicpuLocalReduce(templateDataParams, tempInsQues);
141 0 : HCCL_INFO("[InsTempReduceScatterAicpuReduce] Run finished");
142 0 : return HCCL_SUCCESS;
143 : }
144 :
145 : } // namespace Hccl
|