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