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