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