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(const RankId virtualRank, const u32 tempRankSize,
18 : const std::vector<std::vector<RankId>> &tempVTopo,
19 0 : const std::map<RankId, u32> &tempVirtRankMap)
20 0 : : InsAlgTemplateBase(virtualRank, tempRankSize, tempVTopo, tempVirtRankMap)
21 : {
22 0 : }
23 :
24 0 : InsTempReduceScatterAicpuReduce::~InsTempReduceScatterAicpuReduce()
25 : {
26 0 : }
27 :
28 0 : HcclResult InsTempReduceScatterAicpuReduce::CalcRes(AlgTempResReq &tempResReq)
29 : {
30 0 : tempResReq.queNum = tempVTopo_[0].size();
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 0 : CHK_RET(CalcResLinksMesh(myRank_, tempRankSize_, tempVTopo_, linkNumBtwPeers_, tempResReq));
37 0 : HCCL_DEBUG("[InsTempReduceScatterAicpuReduce]CalcRes: queNum[%u], myRank[%d], tempRankSize[%u]", tempResReq.queNum, myRank_, tempRankSize_);
38 0 : return HcclResult::HCCL_SUCCESS;
39 : }
40 :
41 0 : u32 InsTempReduceScatterAicpuReduce::CalcScratchMultiple(BufferType inBuffType, BufferType outBuffType) const
42 : {
43 : (void) inBuffType;
44 : (void) outBuffType;
45 0 : return tempRankSize_;
46 : }
47 :
48 0 : HcclResult InsTempReduceScatterAicpuReduce::RunAlltoAllMesh(const TempFuncs &tempFuncs, const TemplateDataParams &templateDataParams,
49 : const ResLinks &tempLinks, 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
74 0 : = DataSlice(BufferType::INPUT, templateDataParams.inputSliceStride * u32(neighborRank) + templateDataParams.buffInfo.inBuffBaseOff,
75 0 : templateDataParams.sliceSize);
76 : DataSlice currSendSliceDst
77 0 : = DataSlice(BufferType::SCRATCH, templateDataParams.sliceSize * u32(myRank_), templateDataParams.sliceSize);
78 0 : txSrcSlices.push_back(currSendSliceSrc);
79 0 : txDstSlices.push_back(currSendSliceDst);
80 : //recv
81 : DataSlice currRecvSliceSrc
82 0 : = DataSlice(BufferType::SCRATCH, templateDataParams.inputSliceStride * u32(neighborRank),
83 0 : templateDataParams.sliceSize);
84 : DataSlice currRecvSliceDst
85 0 : = DataSlice(BufferType::SCRATCH, templateDataParams.inputSliceStride * u32(neighborRank) + templateDataParams.buffInfo.inBuffBaseOff,
86 0 : templateDataParams.sliceSize);
87 0 : std::vector<DataSlice> rxSrcSlices;
88 0 : std::vector<DataSlice> rxDstSlices;
89 0 : rxSrcSlices.push_back(currRecvSliceSrc);
90 0 : rxDstSlices.push_back(currRecvSliceDst);
91 0 : TxRxSlicesList sendRecvSlicesList({txSrcSlices, txDstSlices}, {rxSrcSlices, rxDstSlices});
92 0 : SendRecvInfo sendRecvInfo(sendRecvLinks, sendRecvSlicesList);
93 0 : CHK_PRT_RET(SendRecv(sendRecvInfo, tempInsQues[queIdx], 0, true, dmaMode_), HCCL_ERROR("[InsTempReduceScatterAicpuReduce] RunReduceScatter sendrecv failed"),
94 : HcclResult::HCCL_E_INTERNAL);
95 0 : }
96 0 : CHK_RET(PostSyncInterQueues(tempInsQues));
97 0 : return HCCL_SUCCESS;
98 0 : }
99 :
100 0 : HcclResult InsTempReduceScatterAicpuReduce::RunAicpuLocalReduce(const TemplateDataParams &templateDataParams, std::vector<InsQuePtr> &tempInsQues)
101 : {
102 0 : DataSlice dataSlice = DataSlice(BufferType::SCRATCH, 0, templateDataParams.sliceSize);
103 0 : for (u32 rankId = 1; rankId < tempRankSize_; rankId++) {
104 0 : DataSlice reduceSlice = DataSlice(BufferType::SCRATCH, templateDataParams.sliceSize * rankId, templateDataParams.sliceSize);
105 0 : AicpuReduce(tempInsQues[0], reduceSlice, dataSlice, dataType_, redOp_);
106 : }
107 0 : DataSlice outputSlice = DataSlice(BufferType::OUTPUT, templateDataParams.buffInfo.inBuffBaseOff, templateDataParams.sliceSize);
108 0 : LocalCopy(tempInsQues[0], dataSlice, outputSlice);
109 0 : return HCCL_SUCCESS;
110 : }
111 :
112 0 : HcclResult InsTempReduceScatterAicpuReduce::GenExtIns(const TempFuncs &tempFuncs, const TemplateDataParams &templateDataParams,
113 : const ResLinks &tempLinks, std::vector<InsQuePtr> &tempInsQues)
114 : {
115 0 : HCCL_INFO("[InsTempReduceScatterAicpuReduce] Run start");
116 0 : if (IsPcieLink(tempLinks)) {
117 0 : dmaMode_ = DmaMode::GET;
118 : } else {
119 0 : dmaMode_ = DmaMode::PUT;
120 : }
121 0 : if (tempVTopo_[0].size() == 1) {
122 0 : return HcclResult::HCCL_SUCCESS;
123 : }
124 0 : opMode_ = tempFuncs.opMode;
125 0 : queNum_ = tempVTopo_[0].size();
126 0 : CHK_PRT_RET(queNum_ != tempInsQues.size(),
127 : HCCL_ERROR("[CollAlgFactory] [InsTempReduceScatterAicpuReduce] Rank [%d], requiredQue Error.", myRank_),
128 : HcclResult::HCCL_E_INTERNAL);
129 :
130 0 : RunAlltoAllMesh(tempFuncs, templateDataParams, tempLinks, tempInsQues);
131 0 : StreamSync(tempInsQues);
132 0 : RunAicpuLocalReduce(templateDataParams, tempInsQues);
133 0 : HCCL_INFO("[InsTempReduceScatterAicpuReduce] Run finished");
134 0 : return HCCL_SUCCESS;
135 : }
136 :
137 : } // namespace Hccl
|