Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 "acl/acl_rt.h"
12 : #include "rank_info_detect.h"
13 : #include <thread>
14 : #include <stdio.h>
15 : #include "sal.h"
16 : #include "rank_info_detect_service.h"
17 : #include "hccp_peer_manager.h"
18 : #include "hccp_hdc_manager.h"
19 : #include "internal_exception.h"
20 : #include "network_api_exception.h"
21 : #include "null_ptr_exception.h"
22 : #include "orion_adapter_hccp.h"
23 : #include "orion_adapter_rts.h"
24 : #include "whitelist.h"
25 : #include "socket.h"
26 : #include "host_socket_handle_manager.h"
27 : #include "env_config/env_config_v2.h"
28 : #include "root_handle_v2.h"
29 : #include "bootstrap_ip.h"
30 : #include "preempt_port_manager_v2.h"
31 : #include "adapter_error_manager_pub.h"
32 :
33 : namespace Hccl {
34 :
35 : constexpr u32 HOST_CONTROL_BASE_PORT = 60000; // 控制面起始port
36 : constexpr u32 HOST_CONTROL_PORT_COUNT = 15;
37 : constexpr u32 HCCL_WHITELIST_ON = 1;
38 : constexpr u32 HOST_SOCKET_CONN_LIMIT = 8; // HCCL_AISERVER_DEVICE_NUM (8)
39 :
40 : UniversalConcurrentMap<u32, u32> RankInfoDetect::g_detectServerStatus_;
41 :
42 21 : RankInfoDetect::RankInfoDetect()
43 : {
44 21 : userDevId_ = HrtGetDevice();
45 21 : aclError aclRet = aclrtGetLogicDevIdByUserDevId(userDevId_, &devLogicId_); // userDevId 转 logicDevId
46 21 : CHK_PRT_THROW(
47 : aclRet != ACL_SUCCESS,
48 : HCCL_ERROR(
49 : "[RankInfoDetect::%s] aclrtGetLogicDevIdByUserDevId failed, userDevId[%d], ret[%d]", __func__, userDevId_,
50 : aclRet),
51 : InternalException, "get logicDevId fail");
52 21 : s32 deviceNum = HrtGetDeviceCount();
53 21 : CHK_PRT_THROW(
54 : userDevId_ >= deviceNum,
55 : HCCL_ERROR("[RankInfoDetect::%s] userDevId[%d] is invalid, deviceNum[%d].", __func__, userDevId_, deviceNum),
56 : InternalException, "get hostIp fail");
57 : // 临时处理，当前拓扑探测阶段本质上都是使用的userDevId_，当前获取到devLogicId_是为了后续整改
58 : // 如果使用devLogicId_会在多进程各自设置了RT_VISIBLE_DEVICES环境变量时，可能会因为当前进行无法访问某些devLogicId_代表的rank而报错
59 21 : devLogicId_ = userDevId_;
60 :
61 21 : s32 phyDevId = 0;
62 21 : aclRet = aclrtGetPhyDevIdByUserDevId(userDevId_, &phyDevId); // userDevId 转 phyDevId
63 21 : CHK_PRT_THROW(
64 : aclRet != ACL_SUCCESS,
65 : HCCL_ERROR(
66 : "[RankInfoDetect::%s] aclrtGetPhyDevIdByUserDevId failed, userDevId[%d], ret[%d]", __func__, userDevId_,
67 : aclRet),
68 : InternalException, "get phyDevId fail");
69 21 : devPhyId_ = static_cast<u32>(phyDevId);
70 :
71 21 : HCCL_INFO(
72 : "[RankInfoDetect::%s] end, deviceNum[%d], userDevId_[%d], devLogicId_[%d], devPhyId_[%u].", __func__, deviceNum,
73 : userDevId_, devLogicId_, devPhyId_);
74 21 : }
75 :
76 4 : void RankInfoDetect::SetupServer(HcclRootHandleV2& rootHandle)
77 : {
78 4 : HCCL_DEBUG("[RankInfoDetect::%s] setup server start.", __func__);
79 :
80 : // host网卡使能
81 4 : HccpPeerManager::GetInstance().Init(devLogicId_);
82 :
83 : // 获取LocalHostIP
84 4 : hostIp_ = GetBootstrapIp(devPhyId_);
85 6 : CHK_PRT_THROW(
86 : hostIp_.IsInvalid(), HCCL_ERROR("[RankInfoDetect::%s] get hostIp fail.", __func__), InternalException,
87 : "get hostIp fail");
88 :
89 : // 获取端口号port
90 3 : hostPort_ = GetHostListenPort();
91 :
92 : // 1. 创建serverSocket，启动监听并获取实际端口
93 3 : shared_ptr<Socket> serverSocket = ServerInit();
94 :
95 : // 2. 构建rootHandle
96 3 : GetRootHandle(rootHandle);
97 :
98 : // 3. 实际端口和identifier就绪后，使用最终tag下发白名单
99 3 : HcclResult ret = GetHandleAndAddHostSocketWhitelist();
100 3 : if (ret == HCCL_E_PTR) {
101 1 : THROW<NullPtrException>("[RankInfoDetect::%s] get host socket handle failed, ret[%d].", __func__, ret);
102 : }
103 2 : if (ret == HCCL_E_NETWORK) {
104 1 : THROW<NetworkApiException>("[RankInfoDetect::%s] add host socket whitelist failed, ret[%d].", __func__, ret);
105 : }
106 1 : if (ret != HCCL_SUCCESS) {
107 1 : THROW<InternalException>(
108 : "[RankInfoDetect::%s] configure host socket whitelist failed, ret[%d].", __func__, ret);
109 : }
110 :
111 : // 4. 拉起线程，调用RankInfoDetectService.Run()，注意新线程中需要HrtSetDevice
112 : std::thread threadHandle(
113 0 : &RankInfoDetect::SetupRankInfoDetectService, this, serverSocket, devLogicId_, devPhyId_, identifier_,
114 0 : wlistInfo_);
115 0 : threadHandle.detach();
116 :
117 0 : HCCL_DEBUG("[RankInfoDetect::%s] setup server end.", __func__);
118 3 : }
119 :
120 4 : SocketHandle RankInfoDetect::GetHostSocketHandle()
121 : {
122 4 : HCCL_DEBUG("[RankInfoDetect::%s] server get host socket handle start.", __func__);
123 :
124 : // 获取socket句柄
125 4 : SocketHandle hostSocketHandle = HostSocketHandleManager::GetInstance().Create(devPhyId_, hostIp_);
126 :
127 : // 白名单tag依赖实际监听端口和identifier，在GetRootHandle后再构造并下发
128 4 : hostSocketWlist_.clear();
129 4 : if (!EnvConfig::GetInstance().GetHostNicConfig().GetWhitelistDisable()) {
130 1 : Whitelist::GetInstance().GetHostWhiteList(hostSocketWlist_);
131 1 : CHK_PRT_THROW(
132 : hostSocketWlist_.empty(), HCCL_ERROR("[%s] whitelist file have no valid host ip.", __func__),
133 : InternalException, "get host ip error");
134 1 : u32 whiteListEnable = 1;
135 1 : HrtRaSocketSetWhiteListStatus(whiteListEnable);
136 : }
137 :
138 4 : HCCL_DEBUG("[RankInfoDetect::%s] get host socket handle success, socketHandle[%p].", __func__, hostSocketHandle);
139 4 : return hostSocketHandle;
140 : }
141 :
142 3 : shared_ptr<Socket> RankInfoDetect::ServerInit()
143 : {
144 3 : HCCL_DEBUG("[RankInfoDetect::%s] server init start.", __func__);
145 :
146 3 : SocketHandle hccpHostSocketHandle = GetHostSocketHandle();
147 : std::shared_ptr<Socket> serverSocket = std::make_shared<Socket>(
148 3 : hccpHostSocketHandle, hostIp_, hostPort_, hostIp_, "server", SocketRole::SERVER, NicType::HOST_NIC_TYPE);
149 3 : if (hostPort_ == HCCL_INVALID_PORT) {
150 3 : auto portRange = EnvConfig::GetInstance().GetHostNicConfig().GetHostSocketPortRange();
151 3 : if (portRange.empty()) {
152 3 : SocketPortRange defaultRange = {HOST_CONTROL_BASE_PORT, HOST_CONTROL_BASE_PORT + HOST_CONTROL_PORT_COUNT};
153 3 : portRange.push_back(defaultRange);
154 : }
155 3 : PreemptPortManager::GetInstance(devLogicId_).ListenPreempt(serverSocket, portRange, hostPort_);
156 3 : } else {
157 0 : serverSocket->Listen();
158 : }
159 :
160 1 : HCCL_INFO("[RankInfoDetect::%s] serverSocket[%s] listen success.", __func__, serverSocket->Describe().c_str());
161 1 : return serverSocket;
162 2 : }
163 :
164 2 : HcclResult RankInfoDetect::GetHandleAndAddHostSocketWhitelist()
165 : {
166 : // 同一对象重试SetupServer时，先清理上一次生成的条目，避免旧tag被重复下发
167 2 : wlistInfo_.clear();
168 2 : if (hostSocketWlist_.empty()) {
169 1 : return HCCL_SUCCESS;
170 : }
171 :
172 1 : SocketHandle hostSocketHandle = HostSocketHandleManager::GetInstance().Get(devPhyId_, hostIp_);
173 1 : CHK_PRT_RET(
174 : hostSocketHandle == nullptr,
175 : HCCL_ERROR(
176 : "[RankInfoDetect::%s] get host socket handle failed before adding whitelist, "
177 : "devPhyId[%u], hostIp[%s], identifier[%s], listenPort[%u].",
178 : __func__, devPhyId_, hostIp_.GetIpStr().c_str(), identifier_.c_str(), hostPort_),
179 : HCCL_E_PTR);
180 0 : CHK_RET(AddHostSocketWhitelist(hostSocketHandle, hostSocketWlist_));
181 0 : return HCCL_SUCCESS;
182 : }
183 :
184 2 : HcclResult RankInfoDetect::AddHostSocketWhitelist(SocketHandle& socketHandle, const vector<IpAddress>& hostSocketWlist)
185 : {
186 2 : HCCL_DEBUG("[RankInfoDetect::%s] start, hostSocketWlist size[%zu].", __func__, hostSocketWlist.size());
187 :
188 4 : for (auto& ipAddress : hostSocketWlist) {
189 2 : RaSocketWhitelist info{};
190 2 : info.remoteIp = ipAddress;
191 2 : info.connLimit = HOST_SOCKET_CONN_LIMIT;
192 2 : info.tag = RANK_INFO_DETECT_TAG + "_" + identifier_ + "_" + std::to_string(hostPort_);
193 2 : wlistInfo_.push_back(info);
194 2 : }
195 :
196 2 : TRY_CATCH_RETURN(HrtRaSocketWhiteListAdd(socketHandle, wlistInfo_));
197 :
198 1 : HCCL_DEBUG("[RankInfoDetect::%s] end, add wlistInfo size[%zu] success.", __func__, wlistInfo_.size());
199 1 : return HCCL_SUCCESS;
200 : }
201 :
202 2 : std::shared_ptr<Socket> RankInfoDetect::ClientInit(const HcclRootHandleV2& rootHandle)
203 : {
204 2 : HCCL_DEBUG("[RankInfoDetect::%s] client init start devPhyId_[%u].", __func__, devPhyId_);
205 :
206 : // 获取socket句柄
207 2 : SocketHandle hostSocketHandle = HostSocketHandleManager::GetInstance().Create(devPhyId_, hostIp_);
208 :
209 : // 获取server端ip和port
210 2 : IpAddress serverIp = IpAddress(std::string(rootHandle.ip));
211 2 : u32 serverPort = rootHandle.listenPort;
212 :
213 : // 创建clientSocket
214 2 : std::string tag = RANK_INFO_DETECT_TAG + "_" + rootHandle.identifier + "_" + std::to_string(serverPort);
215 : std::shared_ptr<Socket> clientSocket = std::make_shared<Socket>(
216 2 : hostSocketHandle, hostIp_, serverPort, serverIp, tag, SocketRole::CLIENT, NicType::HOST_NIC_TYPE);
217 :
218 2 : HCCL_INFO("[RankInfoDetect::%s] clientSocket[%s] init end.", __func__, clientSocket->Describe().c_str());
219 2 : return clientSocket;
220 2 : }
221 :
222 1 : void RankInfoDetect::SetupAgent(u32 rankSize, u32 rankId, const HcclRootHandleV2& rootHandle)
223 : {
224 1 : HCCL_DEBUG("[RankInfoDetect::%s] setup agent start.", __func__);
225 :
226 : // 网卡使能
227 1 : HccpPeerManager::GetInstance().Init(devLogicId_);
228 1 : HccpHdcManager::GetInstance().Init(devLogicId_);
229 :
230 : // 获取LocalHostIP
231 1 : hostIp_ = GetBootstrapIp(devPhyId_);
232 1 : CHK_PRT_THROW(
233 : hostIp_.IsInvalid(), HCCL_ERROR("[RankInfoDetect::%s] get hostIp fail.", __func__), InternalException,
234 : "get hostIp fail");
235 :
236 : // 创建clientSocket
237 1 : std::shared_ptr<Socket> clientSocket = ClientInit(rootHandle);
238 :
239 : // 1. 创建RankInfoDetectClient对象
240 1 : rankInfoDetectClient = std::make_shared<RankInfoDetectClient>(devPhyId_, rankSize, rankId, clientSocket);
241 :
242 : // 2. 调用RankInfoDetectClient.Setup, 获取rankTable
243 1 : rankInfoDetectClient->Setup(rankTable_);
244 :
245 1 : HCCL_DEBUG("[RankInfoDetect::%s] setup agent end.", __func__);
246 1 : }
247 :
248 2 : void RankInfoDetect::SetupRankInfoDetectService(
249 : shared_ptr<Socket> serverSocket, s32 devLogicId, u32 devPhyId, std::string identifier,
250 : vector<RaSocketWhitelist> wlistInfo)
251 : {
252 2 : HCCL_INFO(
253 : "[RankInfoDetect::%s] start, devLogicId[%d], devPhyId[%u], identifier[%s].", __func__, devLogicId, devPhyId,
254 : identifier.c_str());
255 :
256 : // 拓扑探测server开始状态
257 2 : u32 hostPort = serverSocket->GetListenPort();
258 2 : HCCL_INFO("[RankInfoDetect::%s] listen port[%u].", __func__, hostPort);
259 :
260 2 : g_detectServerStatus_.EmplaceAndUpdate(hostPort, [](u32& status) {
261 2 : status = RANKINFO_DETECT_SERVER_STATUS_RUNING;
262 2 : });
263 :
264 2 : HrtSetDevice(devLogicId);
265 : std::shared_ptr<RankInfoDetectService> rankInfoDetectService
266 2 : = make_shared<RankInfoDetectService>(devPhyId, serverSocket, identifier, wlistInfo);
267 :
268 2 : bool hasException = false;
269 2 : EXCEPTION_CATCH(rankInfoDetectService->Setup(), hasException = true);
270 :
271 : // 若有异常则设置error状态退出
272 2 : if (hasException == true) {
273 1 : g_detectServerStatus_.EmplaceAndUpdate(hostPort, [](u32& status) {
274 1 : status = RANKINFO_DETECT_SERVER_STATUS_ERROR;
275 1 : });
276 1 : HCCL_ERROR("[RankInfoDetect::%s] end, status error.", __func__);
277 1 : return;
278 : }
279 :
280 : // 正常结束则设置为idle状态
281 1 : g_detectServerStatus_.EmplaceAndUpdate(hostPort, [](u32& status) {
282 1 : status = RANKINFO_DETECT_SERVER_STATUS_IDLE;
283 1 : });
284 :
285 1 : HCCL_DEBUG("[RankInfoDetect::%s] end, status idle.", __func__);
286 :
287 : // 确保root info流程先销毁server socket 再返回
288 : // 可能失败，需要将错误状态带出
289 1 : EXCEPTION_CATCH(serverSocket->Destroy(), hasException = true);
290 1 : HrtResetDevice(devLogicId);
291 :
292 : // 若有异常则设置error状态退出
293 1 : if (hasException == true) {
294 0 : g_detectServerStatus_.EmplaceAndUpdate(hostPort, [](u32& status) {
295 0 : status = RANKINFO_DETECT_SERVER_STATUS_ERROR;
296 0 : });
297 0 : HCCL_ERROR("[RankInfoDetect::%s] Destroy end, status error.", __func__);
298 0 : return;
299 : }
300 :
301 1 : HCCL_DEBUG("[RankInfoDetect::%s] end.", __func__);
302 2 : }
303 :
304 5 : u32 RankInfoDetect::GetHostListenPort()
305 : {
306 : // 端口监听范围配置
307 5 : u32 listenPort = HCCL_INVALID_PORT;
308 5 : auto portRange = EnvConfig::GetInstance().GetHostNicConfig().GetHostSocketPortRange();
309 5 : if (portRange.size() > 0) {
310 1 : HCCL_INFO("[RankInfoDetect::%s] SocketPortRange is configured.", __func__);
311 1 : return listenPort;
312 : }
313 :
314 : // Host网卡起始端口号
315 4 : u32 basePort = EnvConfig::GetInstance().GetHostNicConfig().GetIfBasePort();
316 4 : if (basePort != HCCL_INVALID_PORT) {
317 1 : listenPort = basePort + devPhyId_;
318 1 : HCCL_INFO("[RankInfoDetect::%s] BasePort is configured, listenPort[%u].", __func__, listenPort);
319 1 : return listenPort;
320 : }
321 :
322 : // 无环境变量设置，返回HCCL_INVALID_PORT触发PreemptPortManager轮询查找端口[60000, 60015]
323 3 : listenPort = HCCL_INVALID_PORT;
324 3 : HCCL_INFO(
325 : "[RankInfoDetect::%s] No port configuration, using default port range[%u, %u]", __func__,
326 : HOST_CONTROL_BASE_PORT, HOST_CONTROL_BASE_PORT + HOST_CONTROL_PORT_COUNT);
327 3 : return listenPort;
328 5 : }
329 :
330 1 : void RankInfoDetect::GetRootHandle(HcclRootHandleV2& rootHandle)
331 : {
332 1 : u64 timestamp = 0;
333 1 : HcclResult ret = SalGetCurrentTimestamp(timestamp);
334 1 : CHK_PRT_THROW(
335 : ret != HCCL_SUCCESS,
336 : HCCL_ERROR("[RankInfoDetect::%s] get timestamp failed, ret[%d].", __func__, static_cast<int>(ret)),
337 : InternalException, "get timestamp failed");
338 1 : identifier_ = hostIp_.GetIpStr();
339 1 : identifier_.append("_");
340 1 : identifier_.append(to_string(hostPort_));
341 1 : identifier_.append("_");
342 1 : identifier_.append(to_string(devPhyId_));
343 1 : identifier_.append("_");
344 1 : identifier_.append(to_string(timestamp));
345 1 : CHK_PRT_THROW(
346 : (identifier_.length() >= ROOTINFO_IDENTIFIER_MAX_LENGTH),
347 : HCCL_ERROR("[RankInfoDetect::%s] rootInfo identifier len[%u] is invalid.", __func__, identifier_.length()),
348 : InternalException, "identifier error");
349 :
350 3 : s32 sRet = memcpy_s(
351 1 : &rootHandle.identifier[0], sizeof(rootHandle.identifier), identifier_.c_str(), (identifier_.length() + 1));
352 1 : CHK_PRT_THROW(
353 : sRet != EOK,
354 : HCCL_ERROR(
355 : "[RankInfoDetect::%s] memcpy failed. ret[%d], params: destMaxSize[%zu], count[%zu]", __func__, sRet,
356 : sizeof(rootHandle.identifier), (identifier_.length() + 1)),
357 : InternalException, "memcpy failed");
358 :
359 2 : sRet = strncpy_s(
360 3 : rootHandle.ip, sizeof(rootHandle.ip), hostIp_.GetIpStr().c_str(), strlen(hostIp_.GetIpStr().c_str()));
361 1 : CHK_PRT_THROW(
362 : sRet != EOK, HCCL_ERROR("[RankInfoDetect::%s] strncpy failed [%d]", __func__, sRet), InternalException,
363 : "strncpy failed");
364 :
365 1 : rootHandle.listenPort = hostPort_;
366 1 : rootHandle.netMode = HrtNetworkMode::HDC;
367 :
368 1 : HCCL_INFO(
369 : "[RankInfoDetect::%s] rootInfo: ip[%s] port[%u] identifier[%s]", __func__, rootHandle.ip, rootHandle.listenPort,
370 : identifier_.c_str());
371 1 : }
372 :
373 0 : void RankInfoDetect::GetRankTable(RankTableInfo& ranktable) const { ranktable = rankTable_; }
374 :
375 3 : void RankInfoDetect::WaitComplete(u32 listenPort, u32 listenStatus) const
376 : {
377 : // 若server拓扑探测已正常结束则退出
378 3 : auto iter = g_detectServerStatus_.Find(listenPort);
379 3 : HCCL_INFO("[RankInfoDetect::%s] detect server listenPort[%u] status[%u].", __func__, listenPort, iter.second);
380 3 : CHK_PRT_RET(
381 : !iter.second,
382 : HCCL_INFO("[RankInfoDetect::%s] detect server listenPort[%u] status idle.", __func__, listenPort), );
383 :
384 3 : const auto start = chrono::steady_clock::now();
385 3 : const auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
386 :
387 3 : u32 status = RANKINFO_DETECT_SERVER_STATUS_RUNING;
388 : while (true) {
389 953 : auto it = g_detectServerStatus_.Find(listenPort);
390 953 : if (it.second) {
391 953 : status = it.first->second;
392 : }
393 953 : if (status == RANKINFO_DETECT_SERVER_STATUS_ERROR) {
394 1 : THROW<InternalException>(
395 3 : StringFormat("[RankInfoDetect::%s] topo detect failed, port[%u].", __func__, listenPort));
396 952 : } else if (status == listenStatus) {
397 1 : HCCL_INFO("[RankInfoDetect::%s] topoExchangeServer port[%u] compeleted.", __func__, listenPort);
398 1 : return;
399 : } else {
400 951 : const auto elapsed = chrono::duration_cast<chrono::seconds>(chrono::steady_clock::now() - start);
401 951 : if (elapsed > timeout) {
402 8 : RPT_INPUT_ERR(
403 : true, "EI0015", std::vector<std::string>({"error_reason"}),
404 : std::vector<std::string>({StringFormat(
405 : "Receiving message from the root node timed out "
406 : "after %lld seconds. Timeout was set to %lld seconds. Check whether node %s reports an error.",
407 : static_cast<long long>(elapsed.count()), static_cast<long long>(timeout.count()),
408 : identifier_.c_str())}));
409 2 : THROW<TimeoutException>(StringFormat(
410 : "[RankInfoDetect::%s] wait port[%u] complete timeout[%lld s]", __func__, listenPort, elapsed));
411 : }
412 950 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
413 950 : continue;
414 950 : }
415 950 : };
416 2 : }
417 :
418 : } // namespace Hccl
|
