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 "comm_star.h"
12 :
13 : constexpr s32 NORMAL_QP_MODE = 0;
14 : constexpr s32 OFFLINE_QP_MODE = 1;
15 : constexpr s32 OPBASE_QP_MODE = 2;
16 : constexpr s32 OFFLINE_QP_MODE_EXT = 3; // 下沉模式(910B/91093)QP
17 : constexpr s32 OPBASE_QP_MODE_EXT = 4; // 单算子模式(910B/91093)的QP
18 :
19 : namespace hccl {
20 : constexpr s32 MODULE_TYPE_SYSTEM = 0;
21 : constexpr s32 INFO_TYPE_VERSION = 1;
22 : constexpr u32 DEV_TYPE_DIGIT_NUM = 8;
23 : constexpr u32 DEV_TYPE_DIGIT_MASK = 0xff00;
24 :
25 1 : CommStar::CommStar(const std::string &collectiveId, const u32 userRank,
26 : const u32 userRankSize, const u32 rank, const u32 rankSize, const TopoType topoFlag,
27 : const HcclDispatcher dispatcher, const std::unique_ptr<NotifyPool> ¬ifyPool,
28 : std::map<HcclIpAddress, HcclNetDevCtx> &netDevCtxMap,
29 : const IntraExchanger &exchanger, const std::vector<RankInfo> paraVector,
30 : const DeviceMem& inputMem, const DeviceMem& outputMem, const bool isUsedRdmaLevel0,
31 : const std::string &tag,
32 1 : const NICDeployment nicDeployInner, const u32 subUserRankRoot, bool isHaveCpuRank) : CommBase(collectiveId,
33 : userRank, userRankSize, rank, rankSize, paraVector, topoFlag, dispatcher, notifyPool, netDevCtxMap, exchanger,
34 : inputMem, outputMem, isUsedRdmaLevel0, tag,
35 1 : nicDeployInner, false, false, false, subUserRankRoot, isHaveCpuRank)
36 : {
37 1 : IsHostUseDevNic(isHostUseDevNic_);
38 1 : HCCL_DEBUG("CommStar isSetHDCModeInfo_[%d] isHostUseDevNic_ is[%d]", isSetHDCModeInfo_, isHostUseDevNic_);
39 1 : }
40 :
41 2 : CommStar::~CommStar()
42 : {
43 2 : }
44 :
45 0 : HcclResult CommStar::CalcLink()
46 : {
47 0 : if (rank_ == subUserRankRoot_) {
48 0 : for (u32 dstRank = 0; dstRank < rankSize_; dstRank++) {
49 0 : if (dstRank != rank_) {
50 0 : HCCL_INFO("CommStar CalcLink i[%u] am root, dst rank %u", rank_, dstRank);
51 0 : HcclResult ret = CalcLinksNum(MachineType::MACHINE_SERVER_TYPE, dstRank);
52 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
53 : HCCL_ERROR("[Calc][Link]comm Star calc links num failed, type[%d], dstRank[%u]",
54 : static_cast<int32_t>(MachineType::MACHINE_SERVER_TYPE), dstRank), ret);
55 : }
56 : }
57 : } else {
58 0 : HCCL_INFO("CommStar CalcLink i[%u] am not root, dst rank %u", rank_, subUserRankRoot_);
59 0 : HcclResult ret = CalcLinksNum(MachineType::MACHINE_CLIENT_TYPE, subUserRankRoot_);
60 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Calc][Link]comm Star calc links num failed, type[%d], Root[%u]",
61 : static_cast<int32_t>(MachineType::MACHINE_CLIENT_TYPE), subUserRankRoot_), ret);
62 : }
63 :
64 0 : return HCCL_SUCCESS;
65 : }
66 :
67 0 : HcclResult CommStar::MakeClientInfo(const u32 dstRank, RankInfo &dstRankInfo, bool isInterRdma, bool isInterHccs)
68 : {
69 0 : if (isInterRdma && !isInterHccs) {
70 0 : HcclRankLinkInfo tempLinkInfo {};
71 0 : tempLinkInfo.userRank = dstRankInfo.userRank;
72 :
73 0 : std::string remoteHostIp(dstRankInfo.nicIp[0].GetReadableAddress());
74 0 : std::string LocalHostIp(paraVector_[rank_].nicIp[0].GetReadableAddress());
75 0 : if (rankDevicePhyIdNicInfoMap_.find(remoteHostIp) != rankDevicePhyIdNicInfoMap_.end() &&
76 0 : remoteHostIp != LocalHostIp && dstRankInfo.devicePhyId == HOST_DEVICE_ID) {
77 0 : tempLinkInfo.ip = rankDevicePhyIdNicInfoMap_[remoteHostIp][devicePhyId_];
78 0 : tempLinkInfo.devicePhyId = devicePhyId_;
79 : } else {
80 0 : tempLinkInfo.ip = dstRankInfo.nicIp[0];
81 0 : tempLinkInfo.devicePhyId = dstRankInfo.devicePhyId;
82 : }
83 :
84 0 : tempLinkInfo.socketsPerLink = GetSocketsPerLink();
85 :
86 0 : tempLinkInfo.port = GetInterRemotePort(tempLinkInfo.devicePhyId, dstRankInfo.userRank);
87 :
88 0 : auto iter = dstInterClientMap_.find(dstRank);
89 0 : bool check = (iter != dstInterClientMap_.end());
90 0 : CHK_PRT_RET(check, HCCL_ERROR("[Make][ClientInfo]dstRank[%u] already exists in dst inter client map",
91 : dstRank), HCCL_E_PARA);
92 0 : dstInterClientMap_.insert(std::make_pair(dstRank, tempLinkInfo));
93 0 : } else {
94 0 : dstIntraClientVec_.push_back(dstRank);
95 : }
96 0 : return HCCL_SUCCESS;
97 : }
98 :
99 0 : HcclResult CommStar::MakeServerInfo(const u32 dstRank, RankInfo &dstRankInfo, bool isInterRdma, bool isInterHccs)
100 : {
101 : // 节点间或者是节点内采用RDMA通信的，放至dst_inter_client_map_,采用rdma建链
102 0 : if (isInterRdma && !isInterHccs) {
103 0 : HcclRankLinkInfo tempLinkInfo {};
104 0 : tempLinkInfo.userRank = dstRankInfo.userRank;
105 :
106 0 : std::string remoteHostIp(dstRankInfo.nicIp[0].GetReadableAddress());
107 0 : std::string LocalHostIp(paraVector_[rank_].nicIp[0].GetReadableAddress());
108 0 : if (rankDevicePhyIdNicInfoMap_.find(remoteHostIp) != rankDevicePhyIdNicInfoMap_.end() &&
109 0 : remoteHostIp != LocalHostIp && dstRankInfo.devicePhyId == HOST_DEVICE_ID) {
110 0 : tempLinkInfo.ip = rankDevicePhyIdNicInfoMap_[remoteHostIp][devicePhyId_];
111 0 : tempLinkInfo.devicePhyId = devicePhyId_;
112 : } else {
113 0 : tempLinkInfo.ip = dstRankInfo.nicIp[0];
114 0 : tempLinkInfo.devicePhyId = dstRankInfo.devicePhyId;
115 : }
116 :
117 0 : tempLinkInfo.socketsPerLink = GetSocketsPerLink();
118 :
119 0 : tempLinkInfo.port = GetInterRemotePort(tempLinkInfo.devicePhyId, dstRankInfo.userRank);
120 :
121 0 : auto iter = dstInterServerMap_.find(dstRank);
122 0 : bool check = (iter != dstInterServerMap_.end());
123 0 : CHK_PRT_RET(check, HCCL_ERROR("[Make][ServerInfo]dstRank[%u] already exists in dst inter server map",
124 : dstRank), HCCL_E_PARA);
125 0 : dstInterServerMap_.insert(std::make_pair(dstRank, tempLinkInfo));
126 0 : } else {
127 0 : dstIntraServerVec_.push_back(dstRank);
128 : }
129 0 : HCCL_DEBUG("[CommStar]MakeServerInfo success");
130 0 : return HCCL_SUCCESS;
131 : }
132 :
133 0 : HcclResult CommStar::CreateInterLinks()
134 : {
135 0 : HcclResult ret = HCCL_SUCCESS;
136 : u32 targetDevicePhyId;
137 : u32 deviceLogicId;
138 0 : std::map <u32, std::vector<std::shared_ptr<HcclSocket> > > serverSocketsMap;
139 0 : std::map <u32, std::vector<std::shared_ptr<HcclSocket> > > clientSocketsMap;
140 :
141 0 : if (dstInterServerMap_.size() + dstInterClientMap_.size() == 0) {
142 0 : HCCL_DEBUG("[Create][InterLinks] do not need create links.");
143 0 : return HCCL_SUCCESS;
144 : }
145 :
146 0 : if (paraVector_[rank_].devicePhyId == HOST_DEVICE_ID && isHostUseDevNic_) {
147 0 : std::string hostIp(paraVector_[rank_].hostIp.GetReadableAddress());
148 0 : for (auto phyNicInfo : rankDevicePhyIdNicInfoMap_[hostIp]) {
149 0 : targetDevicePhyId = phyNicInfo.first;
150 0 : CHK_RET(hrtGetDeviceIndexByPhyId(targetDevicePhyId, deviceLogicId));
151 :
152 0 : pyhIdResourseSockets_[targetDevicePhyId].reset(
153 0 : new (std::nothrow) HcclSocketManager(nicDeployInner_, deviceLogicId, targetDevicePhyId, userRank_));
154 0 : CHK_PTR_NULL(pyhIdResourseSockets_[targetDevicePhyId]);
155 0 : HCCL_DEBUG("[Create][InterLinks] dstInterServerMap size[%u], dstInterClientMap size[%u]",
156 : dstInterServerMap_.size(), dstInterClientMap_.size());
157 :
158 0 : for (auto &serverInfo : dstInterServerMap_) {
159 0 : if (targetDevicePhyId == serverInfo.second.devicePhyId) {
160 0 : HCCL_DEBUG("[Create][InterLinks] targetDevicePhyId[%u], phyNicInfo.second[%s] serverInfo "
161 : "dstRank[%u] serverInfo.second.devicePhyId[%u]", targetDevicePhyId,
162 : phyNicInfo.second.GetReadableAddress(), serverInfo.first, serverInfo.second.devicePhyId);
163 0 : ret = pyhIdResourseSockets_[targetDevicePhyId]->CreateSockets(tag_, true,
164 0 : netDevCtxMap_[phyNicInfo.second], dstInterServerMap_, dstInterClientMap_,
165 : serverSocketsMap, clientSocketsMap);
166 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
167 : HCCL_ERROR("[Create][InterLinks] socket manager create connections failed, ret[%u]", ret), ret);
168 0 : break;
169 : }
170 : }
171 0 : }
172 0 : } else {
173 0 : interSocketManager_.reset(
174 0 : new (std::nothrow) HcclSocketManager(nicDeployInner_, deviceLogicId_, devicePhyId_, userRank_));
175 0 : CHK_PTR_NULL(interSocketManager_);
176 :
177 0 : HCCL_INFO("[Create][InterLinks] dstInterServerMap size[%u], dstInterClientMap size[%u]",
178 : dstInterServerMap_.size(), dstInterClientMap_.size());
179 :
180 0 : ret = interSocketManager_->CreateSockets(tag_, true, netDevCtxMap_[paraVector_[rank_].nicIp[0]],
181 0 : dstInterServerMap_, dstInterClientMap_, serverSocketsMap, clientSocketsMap);
182 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
183 : HCCL_ERROR("[Create][InterLinks] socket manager create connections failed, ret[%u]", ret), ret);
184 : }
185 :
186 0 : return CreateLinksThread(serverSocketsMap, clientSocketsMap);
187 0 : }
188 :
189 0 : HcclResult CommStar::CreateLinksThread(
190 : std::map<u32, std::vector<std::shared_ptr<HcclSocket>>> &serverSocketsMap,
191 : std::map<u32, std::vector<std::shared_ptr<HcclSocket>>> &clientSocketsMap)
192 : {
193 0 : HcclResult ret = HCCL_SUCCESS;
194 0 : for (auto &sockets : clientSocketsMap) {
195 0 : ret = CreateInterThread(CLIENT_ROLE_SOCKET, sockets.first, sockets.second);
196 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
197 : HCCL_ERROR("[Create][InterLinks] create inter thread failed, socket role[CLIENT_ROLE_SOCKET] "),
198 : ret);
199 : }
200 :
201 0 : for (auto &sockets : serverSocketsMap) {
202 0 : ret = CreateInterThread(SERVER_ROLE_SOCKET, sockets.first, sockets.second);
203 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
204 : HCCL_ERROR("[Create][InterLinks] create inter thread failed, socket role[SERVER_ROLE_SOCKET] "),
205 : ret);
206 : }
207 :
208 0 : return ret;
209 : }
210 :
211 0 : HcclResult CommStar::GetDevIP(const HcclIpAddress& hostIp, const u32& devicePhyId,
212 : HcclIpAddress& ip)
213 : {
214 0 : std::string hostIpStr(hostIp.GetReadableAddress());
215 0 : CHK_PRT_RET(rankDevicePhyIdNicInfoMap_.find(hostIpStr) == rankDevicePhyIdNicInfoMap_.end() ||
216 : rankDevicePhyIdNicInfoMap_[hostIpStr].find(devicePhyId) ==
217 : rankDevicePhyIdNicInfoMap_[hostIpStr].end(), HCCL_ERROR("Get available device nic info fail,"\
218 : "hostIp[%s] devicePhyId[%u]", hostIpStr.c_str(), devicePhyId), HCCL_E_PARA);
219 0 : ip = rankDevicePhyIdNicInfoMap_[hostIpStr][devicePhyId];
220 0 : HCCL_DEBUG("Get available device nic info success, hostIp[%s] devicePhyId[%u] device ip[%s]",
221 : hostIpStr.c_str(), devicePhyId, ip.GetReadableAddress());
222 :
223 0 : return HCCL_SUCCESS;
224 0 : }
225 :
226 0 : HcclResult CommStar::SetMachinePara(MachineType machineType, const std::string &serverId, u32 dstRank,
227 : const std::vector<std::shared_ptr<HcclSocket> > &sockets, MachinePara &machinePara)
228 : {
229 0 : CommBase::SetMachinePara(machineType, serverId, dstRank, sockets, machinePara);
230 0 : std::string localHostIpStr(paraVector_[rank_].hostIp.GetReadableAddress());
231 0 : std::string remoteHostIpStr(paraVector_[dstRank].hostIp.GetReadableAddress());
232 0 : if (paraVector_[rank_].devicePhyId == HOST_DEVICE_ID &&
233 0 : paraVector_[dstRank].devicePhyId != HOST_DEVICE_ID &&
234 0 : rankDevicePhyIdNicInfoMap_.find(localHostIpStr) != rankDevicePhyIdNicInfoMap_.end()) {
235 0 : CHK_PRT(GetDevIP(paraVector_[rank_].hostIp, paraVector_[dstRank].devicePhyId,
236 : machinePara.localIpAddr));
237 :
238 : u32 deviceLogicId;
239 0 : u32 phyId = static_cast<u32>(paraVector_[dstRank].devicePhyId);
240 0 : CHK_RET(hrtGetDeviceIndexByPhyId(phyId, deviceLogicId));
241 0 : machinePara.deviceLogicId = deviceLogicId;
242 : }
243 0 : if (paraVector_[dstRank].devicePhyId == HOST_DEVICE_ID &&
244 0 : paraVector_[rank_].devicePhyId != HOST_DEVICE_ID &&
245 0 : rankDevicePhyIdNicInfoMap_.find(remoteHostIpStr) != rankDevicePhyIdNicInfoMap_.end()) {
246 0 : CHK_PRT(GetDevIP(paraVector_[dstRank].hostIp, paraVector_[rank_].devicePhyId,
247 : machinePara.remoteIpAddr));
248 : }
249 :
250 0 : HCCL_INFO("selfIp[%s] selfPort[%u] peerIp[%s] peerPort[%u] deviceLogicId[%d].",
251 : machinePara.localIpAddr.GetReadableAddress(), machinePara.localSocketPort,
252 : machinePara.remoteIpAddr.GetReadableAddress(), machinePara.localSocketPort, machinePara.deviceLogicId);
253 0 : return HCCL_SUCCESS;
254 0 : }
255 :
256 1 : void CommStar::SetTransportParam(TransportPara ¶, MachinePara &machinePara)
257 : {
258 1 : CommBase::SetTransportParam(para, machinePara);
259 1 : para.selfIp = &machinePara.localIpAddr;
260 1 : para.selfPort = machinePara.localSocketPort;
261 1 : para.peerIp = &machinePara.remoteIpAddr;
262 1 : para.peerPort = machinePara.remoteSocketPort;
263 1 : para.proxyDevLogicId = machinePara.deviceLogicId;
264 1 : HCCL_INFO("SetTransportParam proxyDevLogicId[%u] deviceType is %u",
265 : para.proxyDevLogicId, paraVector_[rank_].deviceType);
266 :
267 1 : if (paraVector_[rank_].deviceType == DevType::DEV_TYPE_NOSOC) {
268 1 : para.qpMode = NORMAL_QP_MODE;
269 1 : para.devLogicId = HOST_DEVICE_ID;
270 1 : para.isHdcMode = false;
271 1 : para.remoteIsHdc = GetRemoteIsHdc();
272 1 : para.isESPs = true;
273 1 : HCCL_INFO("selfIp[%s] selfPort[%u] peerIp[%s] peerPort[%u] qpMode[%d].", para.selfIp->GetReadableAddress(),
274 : para.selfPort, para.peerIp->GetReadableAddress(), para.peerPort, para.qpMode);
275 1 : return;
276 : }
277 :
278 0 : if (paraVector_[rank_].devicePhyId == HOST_DEVICE_ID) {
279 0 : if (paraVector_[rank_].deviceType == DevType::DEV_TYPE_910B) {
280 0 : para.qpMode = OPBASE_QP_MODE_EXT;
281 : } else {
282 0 : para.qpMode = OPBASE_QP_MODE;
283 : }
284 0 : para.devLogicId = HOST_DEVICE_ID;
285 0 : para.remoteIsHdc = GetRemoteIsHdc();
286 0 : para.isESPs = true;
287 : } else {
288 0 : if (paraVector_[rank_].deviceType == DevType::DEV_TYPE_910B) {
289 0 : para.qpMode = (GetWorkflowMode() ==
290 0 : HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE ? OPBASE_QP_MODE_EXT : OFFLINE_QP_MODE_EXT);
291 : } else {
292 0 : para.qpMode = (GetWorkflowMode() ==
293 0 : HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE ? OPBASE_QP_MODE : OFFLINE_QP_MODE);
294 : }
295 0 : para.devLogicId = machinePara.deviceLogicId;
296 : }
297 0 : para.isHdcMode = true;
298 0 : HCCL_INFO("selfIp[%s] selfPort[%u] peerIp[%s] peerPort[%u] qpMode[%d].", para.selfIp->GetReadableAddress(),
299 : para.selfPort, para.peerIp->GetReadableAddress(), para.peerPort, para.qpMode);
300 : }
301 :
302 0 : HcclResult CommStar::CreateExchangerNetwork()
303 : {
304 0 : HCCL_DEBUG("CommStar do not need to Create ExchangerNetwork");
305 0 : return HCCL_SUCCESS;
306 : }
307 : } // namespace hccl
308 :
|
