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 "ins_coll_alg_registry.h"
14 : #include "ins_recv_executor.h"
15 : #include "alg_data_trans_wrapper.h"
16 :
17 : using namespace std;
18 :
19 : namespace Hccl {
20 0 : InsRecvExecutor::InsRecvExecutor() : InsCollAlgBase() {}
21 :
22 0 : InsRecvExecutor::~InsRecvExecutor() {}
23 :
24 0 : HcclResult InsRecvExecutor::Orchestrate(
25 : const RankGraph* rankGraph, const CollAlgOperator& op, const CollAlgParams& params, InsQuePtr insQue)
26 : {
27 0 : dmaMode_ = DmaMode::DEFAULT;
28 0 : HCCL_DEBUG("[InsCollAlgFactory][InsRecvExecutor][Orchestrate] Begin to Generate Instruction Queue for RECV.");
29 0 : CHK_RET(Init(op, params, insQue));
30 : // 从集合通信算子op中获得remote rank和data type/count相关信息
31 0 : RankId remoteRank = op.sendRecvRemoteRank;
32 0 : u32 dataElemSize = DATA_TYPE_SIZE_MAP.at(op.dataType);
33 0 : u64 totalDataSize = static_cast<u64>(dataElemSize) * op.dataCount;
34 0 : if (totalDataSize == 0) {
35 0 : HCCL_WARNING("[InsCollAlgFactory][InsRecvExecutor][Orchestrate] totalDataSize is 0, do nothing.");
36 0 : return HcclResult::HCCL_SUCCESS;
37 : }
38 : // 判断是不是自发自收这种情况，若是，则什么都不做，直接返回
39 0 : if (myRank_ == remoteRank) {
40 0 : HCCL_WARNING("[InsCollAlgFactory][InsRecvExecutor][Orchestrate] Self send, Self recv, Do nothing");
41 0 : return HcclResult::HCCL_SUCCESS;
42 : }
43 0 : HCCL_DEBUG("[InsCollAlgFactory][InsRecvExecutor][Orchestrate] Other send, Self recv");
44 :
45 : // 从virtualTopo里面拿到 link，转成LinkData格式
46 0 : const std::vector<NetInstance::Path> recvPath = GetPathsFromRankGraph(rankGraph, myRank_, remoteRank);
47 0 : CHK_PRT_RET(
48 : recvPath.size() == 0,
49 : HCCL_ERROR("[InsCollAlgFactory] Unable to obtain valid link, srcRank [%d], dstRank [%d].", myRank_, remoteRank),
50 : HcclResult::HCCL_E_INTERNAL);
51 0 : LinkData recvLinkData(recvPath[0]);
52 0 : if (recvLinkData.GetType() == PortDeploymentType::P2P && recvLinkData.GetLinkProtocol() == LinkProtocol::PCIE) {
53 0 : dmaMode_ = DmaMode::GET;
54 : }
55 0 : HCCL_DEBUG(
56 : "[InsCollAlgFactory][InsRecvExecutor][Orchestrate] Total transfer data size [%llu], Max scratch buffer size "
57 : "[%u].",
58 : totalDataSize, params.maxTmpMemSize);
59 :
60 : // 初始化循环参数
61 0 : u64 resDataSize = totalDataSize;
62 0 : u64 currentOffset = 0;
63 0 : u32 roundIdx = 0;
64 : // 模式判断
65 0 : if (opMode_ == OpMode::OFFLOAD) {
66 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OFFLOAD Mode for HOST.", myRank_);
67 0 : u64 transferSize = resDataSize;
68 : // 根据本轮数据搬运量声明相关DataSlice--图模式一次搬运全部数据 从inputbuffer直接send到对面outputbuffer
69 0 : DataSlice outputBuffer(BufferType::OUTPUT, currentOffset, transferSize);
70 0 : DataSlice remoteInputBuffer(BufferType::INPUT, currentOffset, transferSize);
71 0 : SlicesList recvSlicesList({remoteInputBuffer}, {outputBuffer});
72 0 : DataInfo recvInfo(recvLinkData, recvSlicesList);
73 0 : CHK_RET(Recv(recvInfo, insQue, 0, true, dmaMode_));
74 0 : } else {
75 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OPBASE Mode for HOST.", myRank_);
76 : // 当需要多轮搬运时，需保证一次数据的搬运量需为单个数据size的整数倍
77 0 : u64 maxRoundTransferSize = params.maxTmpMemSize - params.maxTmpMemSize % dataElemSize;
78 0 : while (resDataSize > 0) {
79 : // 判断本轮需搬运的数据量
80 0 : u64 transferSize = resDataSize > params.maxTmpMemSize ? maxRoundTransferSize : resDataSize;
81 0 : HCCL_DEBUG(
82 : "[InsCollAlgFactory][InsRecvExecutor][Orchestrate] Recv round [%u], transfer data size [%llu]",
83 : roundIdx, transferSize);
84 : // 根据本轮数据搬运量创建相关DataSlice
85 0 : DataSlice outputBuffer(BufferType::OUTPUT, currentOffset, transferSize);
86 0 : DataSlice scratchBuffer(BufferType::SCRATCH, 0, transferSize);
87 0 : DataSlice remoteScratchBuffer(BufferType::SCRATCH, 0, transferSize);
88 0 : SlicesList recvSlicesList({remoteScratchBuffer}, {scratchBuffer});
89 0 : DataInfo recvInfo(recvLinkData, recvSlicesList);
90 0 : CHK_RET(Recv(recvInfo, insQue, 0, true, dmaMode_));
91 : // local copy
92 0 : CHK_RET(LocalCopy(insQue, scratchBuffer, outputBuffer));
93 : // 更新循环参数
94 0 : currentOffset = currentOffset + transferSize;
95 0 : resDataSize = resDataSize - transferSize;
96 0 : roundIdx = roundIdx + 1;
97 0 : }
98 : }
99 0 : return HcclResult::HCCL_SUCCESS;
100 0 : }
101 :
102 0 : HcclResult InsRecvExecutor::CalcResOffload(const RankGraph* rankGraph, const u64& dataSize, CollOffloadOpResReq& resReq)
103 : {
104 : (void)rankGraph;
105 : (void)dataSize;
106 0 : resReq.requiredScratchMemSize = 0; // 图模式不用scratchmemory
107 0 : resReq.requiredSubQueNum = 0;
108 :
109 0 : return HcclResult::HCCL_SUCCESS;
110 : }
111 :
112 0 : HcclResult InsRecvExecutor::CalcRes(const RankGraph* rankGraph, CollAlgResReq& algResReq)
113 : {
114 0 : u32 linkNumBtwPeers = 1;
115 0 : algResReq.primQueueNum = 1;
116 0 : AlgTempResReq tempResReq;
117 0 : tempResReq.queNum = 1;
118 0 : tempResReq.streamNum = tempResReq.streamNum;
119 0 : if (static_cast<u32>(sendRecvRemoteRank_) > rankSize_ - 1) {
120 0 : HCCL_ERROR(
121 : "[InsCollAlgFactory][InsRecvExecutor][CalcRes] Rank[%d] get dest[%d] is invalid", myRank_,
122 : sendRecvRemoteRank_);
123 0 : return HcclResult::HCCL_E_PARA;
124 : }
125 0 : tempResReq.links[sendRecvRemoteRank_] = linkNumBtwPeers;
126 0 : uint32_t linkNum = GetPathsFromRankGraph(rankGraph, myRank_, sendRecvRemoteRank_).size();
127 0 : if (linkNum == 0) {
128 0 : HCCL_ERROR(
129 : "[InsCollAlgFactory][InsRecvExecutor][CalcRes] Rank[%d] get path num to dest[%d] is zero", myRank_,
130 : sendRecvRemoteRank_);
131 : }
132 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, algResReq.links));
133 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, tempResReq.links, algResReq.levelRankPairs));
134 0 : return HcclResult::HCCL_SUCCESS;
135 0 : }
136 :
137 0 : HcclResult InsRecvExecutor::Orchestrate(
138 : const AlgTopoInfo& topoInfo, const CollAlgOperator& op, const CollAlgParams& params, ConnectedLinkMgr* linkMgr,
139 : InsQuePtr insQue)
140 : {
141 0 : dmaMode_ = DmaMode::DEFAULT;
142 : (void)topoInfo;
143 0 : HCCL_DEBUG(
144 : "[InsCollAlgFactory][InsRecvExecutor][Orchestrate] Begin to Generate Instruction Queue for RECV AICPU mode.");
145 0 : CHK_RET(Init(op, params, insQue));
146 : // 从集合通信算子op中获得remote rank和data type/count相关信息
147 0 : RankId remoteRank = op.sendRecvRemoteRank;
148 0 : u32 dataElemSize = DATA_TYPE_SIZE_MAP.at(op.dataType);
149 0 : u64 totalDataSize = static_cast<u64>(dataElemSize) * op.dataCount;
150 0 : if (totalDataSize == 0) {
151 0 : HCCL_WARNING("[InsCollAlgFactory][InsRecvExecutor][Orchestrate] totalDataSize is 0, do nothing.");
152 0 : return HcclResult::HCCL_SUCCESS;
153 : }
154 : // 判断是不是自发自收这种情况，若是，则什么都不做，直接返回
155 0 : if (myRank_ == remoteRank) {
156 0 : HCCL_WARNING("[InsCollAlgFactory][InsRecvExecutor][Orchestrate] Self send, Self recv, Do nothing");
157 0 : return HcclResult::HCCL_SUCCESS;
158 : }
159 0 : HCCL_DEBUG("[InsCollAlgFactory][InsRecvExecutor][Orchestrate] Other send, Self recv.");
160 :
161 : // 从linkMgr里面拿到 linkData
162 0 : const vector<LinkData> recvPath = linkMgr->GetLinks(remoteRank);
163 0 : CHK_PRT_RET(
164 : recvPath.size() == 0,
165 : HCCL_ERROR("[InsCollAlgFactory] Unable to obtain valid link, srcRank [%d], dstRank [%d].", myRank_, remoteRank),
166 : HcclResult::HCCL_E_INTERNAL);
167 0 : LinkData recvLinkData(recvPath[0]);
168 0 : if (recvLinkData.GetType() == PortDeploymentType::P2P && recvLinkData.GetLinkProtocol() == LinkProtocol::PCIE) {
169 0 : dmaMode_ = DmaMode::GET;
170 : }
171 0 : HCCL_DEBUG(
172 : "[InsCollAlgFactory][InsRecvExecutor][Orchestrate] Total transfer data size [%llu], Max scratch buffer size "
173 : "[%u].",
174 : totalDataSize, params.maxTmpMemSize);
175 :
176 : // 初始化循环参数
177 0 : u64 resDataSize = totalDataSize;
178 0 : u64 currentOffset = 0;
179 0 : u32 roundIdx = 0;
180 : // 模式判断
181 0 : if (opMode_ == OpMode::OFFLOAD) {
182 0 : u64 maxLoopOutputSize = 256 * 1024 * 1024; // 256m为一轮
183 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OFFLOAD Mode for HOST.", myRank_);
184 0 : while (resDataSize > 0) {
185 0 : u64 transferSize = resDataSize > maxLoopOutputSize ? maxLoopOutputSize : resDataSize;
186 0 : DataSlice outputBuffer(BufferType::OUTPUT, currentOffset, transferSize);
187 0 : DataSlice remoteInputBuffer(BufferType::INPUT, currentOffset, transferSize);
188 0 : SlicesList recvSlicesList({remoteInputBuffer}, {outputBuffer});
189 0 : DataInfo recvInfo(recvLinkData, recvSlicesList);
190 0 : CHK_RET(Recv(recvInfo, insQue, 0, true, dmaMode_));
191 0 : currentOffset = currentOffset + transferSize;
192 0 : resDataSize = resDataSize - transferSize;
193 0 : roundIdx = roundIdx + 1;
194 0 : }
195 : } else {
196 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OPBASE Mode for HOST.", myRank_);
197 : // 当需要多轮搬运时，需保证一次数据的搬运量需为单个数据size的整数倍
198 0 : u64 maxRoundTransferSize = params.maxTmpMemSize - params.maxTmpMemSize % dataElemSize;
199 0 : while (resDataSize > 0) {
200 : // 判断本轮需搬运的数据量
201 0 : u64 transferSize = resDataSize > params.maxTmpMemSize ? maxRoundTransferSize : resDataSize;
202 0 : HCCL_DEBUG(
203 : "[InsCollAlgFactory][InsRecvExecutor][Orchestrate] Recv round [%u], transfer data size [%llu]",
204 : roundIdx, transferSize);
205 : // 根据本轮数据搬运量创建相关DataSlice
206 0 : DataSlice outputBuffer(BufferType::OUTPUT, currentOffset, transferSize);
207 0 : DataSlice scratchBuffer(BufferType::SCRATCH, 0, transferSize);
208 0 : DataSlice remoteScratchBuffer(BufferType::SCRATCH, 0, transferSize);
209 :
210 0 : SlicesList recvSlicesList({remoteScratchBuffer}, {scratchBuffer});
211 0 : DataInfo recvInfo(recvLinkData, recvSlicesList);
212 0 : CHK_RET(Recv(recvInfo, insQue, 0, true, dmaMode_));
213 : // local copy
214 0 : CHK_RET(LocalCopy(insQue, scratchBuffer, outputBuffer));
215 : // 更新循环参数
216 0 : currentOffset = currentOffset + transferSize;
217 0 : resDataSize = resDataSize - transferSize;
218 0 : roundIdx = roundIdx + 1;
219 0 : }
220 : }
221 0 : return HcclResult::HCCL_SUCCESS;
222 0 : }
223 :
224 : // 注册
225 : INS_REGISTER_IMPL(OpType::RECV, InsRecv, InsRecvExecutor);
226 :
227 : } // namespace Hccl
|
