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