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 <thread>
12 : #include <cstdlib>
13 : #include <fstream>
14 : #include <limits.h>
15 : #include "rank_info_detect_client.h"
16 : #include "root_handle_v2.h"
17 : #include "env_config/env_config_v2.h"
18 : #include "host_buffer.h"
19 : #include "binary_stream.h"
20 : #include "hccp_peer_manager.h"
21 : #include "hcomm_res.h"
22 : #include "orion_adapter_hccp.h"
23 : #include "orion_adapter_rts.h"
24 : #include "host_socket_handle_manager.h"
25 : #include "socket_manager.h"
26 : #include "topo_addr_info.h"
27 : #include "adapter_error_manager_pub.h"
28 : #include "network_api_exception.h"
29 : #include "phy_topo_builder.h"
30 : #include "preempt_port_manager_v2.h"
31 :
32 : namespace Hccl {
33 : namespace {
34 : constexpr u32 HOST_CONTROL_PORT_COUNT = 15;
35 : constexpr u32 HOST_BACKUP_ADDR_NET_LAYER = 3;
36 : constexpr const char* BACKUP_ADDR_FIELD = "backup_addr";
37 : constexpr const char* ADDR_FIELD = "addr";
38 : constexpr const char* ADDR_TYPE_FIELD = "addr_type";
39 :
40 43 : void FillCommAddr(CommAddr& commAddr, const IpAddress& ipAddr)
41 : {
42 43 : const s32 family = ipAddr.GetFamily();
43 43 : if (family == AF_INET) {
44 40 : commAddr.type = COMM_ADDR_TYPE_IP_V4;
45 40 : commAddr.addr = ipAddr.GetBinaryAddress().addr;
46 3 : } else if (family == AF_INET6) {
47 3 : commAddr.type = COMM_ADDR_TYPE_IP_V6;
48 3 : commAddr.addr6 = ipAddr.GetBinaryAddress().addr6;
49 : } else {
50 0 : THROW<InvalidParamsException>(
51 0 : StringFormat("[%s] invalid commAddrType, hostAddr[%s].", __func__, ipAddr.Describe().c_str()));
52 : }
53 43 : }
54 :
55 18 : void BuildHostAddrCandidates(const nlohmann::json& addrJson, std::vector<IpAddress>& candidates)
56 : {
57 : // 候选顺序固定为主地址在前、备地址按配置顺序在后，选择时取首个探测成功的地址。
58 18 : std::string addrType;
59 18 : std::string primaryAddr;
60 18 : const std::string msgAddrType = "get host addr_type failed";
61 18 : TRY_CATCH_THROW(InvalidParamsException, msgAddrType, addrType = GetJsonProperty(addrJson, ADDR_TYPE_FIELD););
62 18 : const std::string msgPrimaryAddr = "get primary host addr failed";
63 18 : TRY_CATCH_THROW(InvalidParamsException, msgPrimaryAddr, primaryAddr = GetJsonProperty(addrJson, ADDR_FIELD););
64 18 : IpAddress primaryIpAddress;
65 18 : const std::string msgParsePrimaryAddr = "parse primary host addr failed";
66 19 : TRY_CATCH_THROW(InvalidParamsException, msgParsePrimaryAddr,
67 : AddressInfo::ParseAddrByType(addrType, primaryAddr, primaryIpAddress););
68 17 : candidates.clear();
69 17 : candidates.emplace_back(primaryIpAddress);
70 :
71 17 : std::vector<IpAddress> backupAddrs;
72 17 : AddressInfo::ParseBackupAddrs(addrJson.at(BACKUP_ADDR_FIELD), addrType, backupAddrs);
73 13 : candidates.insert(candidates.end(), backupAddrs.begin(), backupAddrs.end());
74 42 : }
75 :
76 22 : void CollectLayer3AddrJsons(nlohmann::json& localDevInfoJson, std::vector<nlohmann::json*>& addrJsons)
77 : {
78 : // 仅返回 netLayer3 及以上且配置了 backup_addr 的可写地址节点，其他地址直接沿用主 addr。
79 22 : addrJsons.clear();
80 22 : CHK_PRT_THROW(
81 : !localDevInfoJson.contains("level_list") || !localDevInfoJson.at("level_list").is_array(),
82 : HCCL_ERROR("[%s] level_list is missing or is not an array.", __func__), InvalidParamsException,
83 : "level_list is missing or is not an array");
84 44 : for (auto& levelJson : localDevInfoJson.at("level_list")) {
85 22 : if (!levelJson.contains("rank_addr_list") || !levelJson["rank_addr_list"].is_array()) {
86 4 : continue;
87 : }
88 44 : u32 netLayer = 0;
89 22 : const std::string msgNetLayer = "get net_layer failed";
90 22 : TRY_CATCH_THROW(InvalidParamsException, msgNetLayer,
91 : netLayer = GetJsonPropertyUInt(levelJson, "net_layer"););
92 22 : if (netLayer < HOST_BACKUP_ADDR_NET_LAYER) {
93 4 : continue;
94 : }
95 37 : for (auto& addrJson : levelJson["rank_addr_list"]) {
96 19 : if (!addrJson.contains(BACKUP_ADDR_FIELD)) {
97 1 : HCCL_WARNING("[%s] backup_addr is not configured, use primary addr without probing.", __func__);
98 1 : continue;
99 : }
100 18 : addrJsons.push_back(&addrJson);
101 : }
102 22 : }
103 22 : }
104 :
105 6 : std::string QueryTopoFilePathByDevice()
106 : {
107 6 : const size_t bufSize = 1024;
108 6 : auto devLogicId = HrtGetDevice();
109 6 : auto devPhyId = HrtGetDevicePhyIdByIndex(devLogicId);
110 6 : std::vector<char> buffer(bufSize, '\0');
111 6 : int result = TopoAddrInfoGetTopoFilePath(devPhyId, buffer.data(), buffer.size());
112 6 : CHK_PRT_THROW(
113 : result != 0, HCCL_ERROR("[%s] Get topo file path failed.", __func__), InvalidParamsException,
114 : "Get topo file path failed.");
115 12 : return std::string(buffer.data());
116 6 : }
117 :
118 6 : void CheckTopoFilePath(const std::string& topoFilePath)
119 : {
120 6 : char resolvedPath[PATH_MAX] = {0};
121 6 : CHK_PRT_THROW(
122 : realpath(topoFilePath.c_str(), resolvedPath) == nullptr,
123 : HCCL_ERROR("[%s] topo_file_path[%s] is not a valid real path", __func__, topoFilePath.c_str()),
124 : InvalidParamsException, "topo_file_path error");
125 6 : }
126 :
127 6 : std::string GetRootInfoTopoFilePath()
128 : {
129 6 : std::string filePath = "/etc/hccl_rootinfo.json";
130 : JsonParser jsonParser{};
131 6 : nlohmann::json parseJson{};
132 6 : std::string topoFilePath{};
133 6 : std::ifstream file(filePath);
134 6 : if (file.good()) {
135 0 : jsonParser.ParseFileToJson(filePath, parseJson);
136 0 : std::string msgRankTopoFile = "error occurs when parser object of propName \"topo_file_path\"";
137 0 : TRY_CATCH_THROW(InvalidParamsException, msgRankTopoFile,
138 : topoFilePath = GetJsonProperty(parseJson, "topo_file_path"););
139 0 : } else {
140 6 : topoFilePath = QueryTopoFilePathByDevice();
141 : }
142 :
143 6 : CheckTopoFilePath(topoFilePath);
144 6 : return topoFilePath;
145 6 : }
146 : } // namespace
147 :
148 0 : void RankInfoDetectClient::Setup(RankTableInfo& rankTable)
149 : {
150 : // 1. 构造localRankTable
151 0 : RankTableInfo localRankTable{};
152 0 : ConstructRankTable(localRankTable);
153 :
154 : // 若启用单卡多进程抢占端口则执行
155 0 : SocketManager::ServerInitAll(localRankTable.ranks[0]);
156 0 : HostListenPortDetect(localRankTable.ranks[0]);
157 :
158 : // 2. 连接root节点
159 0 : Connect();
160 :
161 : // 3. 发送本端agentId和rankSize
162 0 : SendAgentIdAndRankSize();
163 :
164 : // 4. 发送给root节点
165 0 : SendLocalRankTable(localRankTable);
166 :
167 : // 5. 接收完整rankTable
168 0 : RecvRankTable();
169 0 : rankTable = rankTable_;
170 0 : }
171 :
172 0 : void RankInfoDetectClient::Connect()
173 : {
174 0 : clientSocket_->Connect();
175 0 : CheckStatus();
176 0 : }
177 :
178 2 : void RankInfoDetectClient::CheckStatus()
179 : {
180 2 : HCCL_DEBUG("[RankInfoDetectClient::%s] start.", __func__);
181 :
182 2 : auto startTime = std::chrono::steady_clock::now();
183 2 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
184 :
185 : while (true) {
186 826865 : bool isTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
187 826865 : if (isTimeout) {
188 1 : HCCL_ERROR(
189 : "[RankInfoDetectClient::%s] get connected status socket timeout! timeout[%lld s]", __func__, timeout);
190 7 : RPT_INPUT_ERR(
191 : isTimeout, "EI0015", std::vector<std::string>({"error_reason"}),
192 : std::vector<std::string>({StringFormat(
193 : "Receiving message from the root node timed out "
194 : "Timeout was set to %lld seconds. Check whether node rankId[%u] reports an error.",
195 : static_cast<long long>(timeout.count()), rankId_)}));
196 : // 建链超时后，sleep 20s，避免上层应用提前退出，确保其他正常 client 能够收到 server 发出的临终遗言
197 1 : sleep(WAIT_ERROR_BROADCAST_TIME);
198 1 : THROW<TimeoutException>("client get connection timeout");
199 : }
200 :
201 826864 : if (clientSocket_->GetStatus() == SocketStatus::OK) {
202 1 : HCCL_DEBUG("[RankInfoDetectClient::%s] client get socket connection success.", __func__);
203 1 : break;
204 : }
205 826863 : }
206 :
207 1 : HCCL_INFO("[RankInfoDetectClient::%s] end, connect ok.", __func__);
208 2 : }
209 :
210 1 : void RankInfoDetectClient::SendAgentIdAndRankSize()
211 : {
212 1 : HCCL_DEBUG("[RankInfoDetectClient::%s] start.", __func__);
213 :
214 : // 发送agentId
215 1 : std::string rankID = std::to_string(rankId_);
216 1 : std::string agentID = std::string(16 - rankID.length(), '0') + rankID;
217 1 : socketAgent_.SendMsg(agentID.c_str(), agentID.size());
218 :
219 : // 发送rankSize
220 1 : socketAgent_.SendMsg(&rankSize_, sizeof(rankSize_));
221 :
222 1 : HCCL_INFO(
223 : "[RankInfoDetectClient::%s] send agentID[%s] and rankSize_[%u] end.", __func__, agentID.c_str(), rankSize_);
224 1 : }
225 :
226 0 : void RankInfoDetectClient::SendLocalRankTable(const RankTableInfo& localRankTable)
227 : {
228 0 : HCCL_DEBUG("[RankInfoDetectClient::%s] start.", __func__);
229 :
230 : // 消息格式: [ranktable数据(n字节)][step(4字节)]
231 0 : BinaryStream binaryStream;
232 0 : localRankTable.GetBinStream(true, binaryStream);
233 0 : binaryStream << currentStep_;
234 :
235 : // 字节流转换为vector<char>格式
236 0 : vector<char> sendMsg;
237 0 : binaryStream.Dump(sendMsg);
238 :
239 : // 发送
240 0 : socketAgent_.SendMsg(sendMsg.data(), sendMsg.size());
241 :
242 0 : HCCL_INFO("[RankInfoDetectClient::%s] end, currentStep_[%u].", __func__, currentStep_);
243 0 : currentStep_++;
244 0 : }
245 :
246 5 : void RankInfoDetectClient::ConstructSingleRank(RankTableInfo& localRankTable)
247 : {
248 5 : localRankTable.version = "2.0";
249 5 : localRankTable.rankCount = 1;
250 5 : NewRankInfo rankInfo{};
251 5 : rankInfo.rankId = rankId_;
252 5 : rankInfo.rankLevelInfos.emplace_back(RankLevelInfo{});
253 5 : CHK_PRT_CONT(GetLocalTlsStatus(rankInfo.tlsStatus), HCCL_WARNING("[GetLocalTlsStatus] Can not get TlsStatus"));
254 5 : CHK_PRT_CONT(
255 : GetLocalHostDpuTlsStatus(rankInfo.hostDpuTlsStatus),
256 : HCCL_WARNING("[GetLocalHostDpuTlsStatus] Can not get Host DPU TlsStatus"));
257 5 : localRankTable.ranks.emplace_back(rankInfo);
258 :
259 : // 打印
260 5 : localRankTable.Dump();
261 5 : HCCL_INFO(
262 : "[RankInfoDetectClient::%s] end, single rank, localRankTable[%s].", __func__,
263 : localRankTable.Describe().c_str());
264 5 : }
265 :
266 1 : void CheckRootInfoJson(const nlohmann::json& parseJson)
267 : {
268 : // check version
269 1 : std::string version{};
270 1 : std::string msgVersion = "error occurs when parser rootinfo object of propName \"version\"";
271 1 : TRY_CATCH_THROW(InvalidParamsException, msgVersion, version = GetJsonProperty(parseJson, "version"););
272 1 : if (version != "2.0") {
273 0 : RPT_INPUT_ERR(
274 : true, "EI0016", std::vector<std::string>({"value", "variable", "expect"}),
275 : std::vector<std::string>({version, "version", "2.0"}));
276 0 : HCCL_ERROR("[%s] failed with version [%s] is not \"2.0\".", __func__, version.c_str());
277 0 : THROW<InvalidParamsException>("version error");
278 : }
279 :
280 : // parser topo_file_path
281 1 : std::string topoFilePath{};
282 1 : std::string msgRankTopoFile = "error occurs when parser object of propName \"topo_file_path\"";
283 1 : TRY_CATCH_THROW(InvalidParamsException, msgRankTopoFile,
284 : topoFilePath = GetJsonProperty(parseJson, "topo_file_path"););
285 :
286 : // check topo_file_path
287 1 : char resolvedPath[PATH_MAX] = {0};
288 1 : bool isInvalidPath = (realpath(topoFilePath.c_str(), resolvedPath) == nullptr);
289 1 : if (isInvalidPath) {
290 0 : RPT_INPUT_ERR(
291 : true, "EI0016", std::vector<std::string>({"value", "variable", "expect"}),
292 : std::vector<std::string>({topoFilePath, "topo_file_path", "valid path"}));
293 0 : HCCL_ERROR("[%s] topo_file_path[%s] is not a valid real path", __func__, topoFilePath.c_str());
294 0 : THROW<InvalidParamsException>("topo_file_path error");
295 : }
296 :
297 : // parser rank_count
298 1 : u32 rankCount{};
299 1 : std::string msgRankcount = "error occurs when parser object of propName \"rank_count\"";
300 1 : TRY_CATCH_THROW(InvalidParamsException, msgRankcount, rankCount = GetJsonPropertyUInt(parseJson, "rank_count"););
301 :
302 : // parser rank_list
303 1 : nlohmann::json rankJsons{};
304 1 : std::string msgRanklist = "error occurs when parser object of propName \"rank_list\"";
305 1 : TRY_CATCH_THROW(InvalidParamsException, msgRanklist, GetJsonPropertyList(parseJson, "rank_list", rankJsons););
306 :
307 : // check rank_count
308 1 : bool isRankCountMismatch = (rankCount != rankJsons.size());
309 1 : if (isRankCountMismatch) {
310 0 : RPT_INPUT_ERR(
311 : true, "EI0016", std::vector<std::string>({"value", "variable", "expect"}),
312 : std::vector<std::string>({std::to_string(rankCount), "rankCount", std::to_string(rankJsons.size())}));
313 0 : HCCL_ERROR(
314 : "[%s] failed with rankCount is not equal to rank_list size. "
315 : "rankCount[%u], ranks.size[%u]",
316 : __func__, rankCount, rankJsons.size());
317 0 : THROW<InvalidParamsException>("rankCount error");
318 : }
319 1 : }
320 :
321 1 : void RankInfoDetectClient::ConstructRankTable(RankTableInfo& localRankTable)
322 : {
323 1 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
324 :
325 : // 单P场景处理
326 1 : CHK_PRT_RET((rankSize_ == 1), ConstructSingleRank(localRankTable), );
327 :
328 : // 1. 解析文件topoInfo.json
329 1 : std::string filePath = "/etc/hccl_rootinfo.json";
330 : JsonParser jsonParser{};
331 1 : nlohmann::json parseJson{};
332 1 : std::ifstream file(filePath);
333 1 : if (file.good()) {
334 0 : jsonParser.ParseFileToJson(filePath, parseJson);
335 : } else {
336 : size_t bufSize;
337 1 : s32 result = TopoAddrInfoGetSize(devPhyId_, &bufSize); // 获取rankInfo大小，用于提前分配内存
338 1 : CHK_PRT_THROW(
339 : result != 0 || bufSize > MAX_BUFFER_LEN,
340 : HCCL_ERROR("[RankInfoDetectClient::%s] Get rankinfo size failed.", __func__), InvalidParamsException,
341 : "Get rankinfo size failed.");
342 1 : std::vector<char> buffer(bufSize, '\0');
343 1 : result = TopoAddrInfoGet(devPhyId_, buffer.data(), &bufSize); // 获取rankInfo 并更新大小
344 1 : CHK_PRT_THROW(
345 : result != 0, HCCL_ERROR("[RankInfoDetectClient::%s] Get rankinfo failed.", __func__),
346 : InvalidParamsException, "Get rankinfo size failed.");
347 1 : std::string jsonString(buffer.data(), bufSize);
348 : // 将生成的info信息转换成json文件
349 1 : parseJson = nlohmann::json::parse(jsonString);
350 1 : }
351 1 : CheckRootInfoJson(parseJson);
352 :
353 : // 2. 获取当前devPhyId_对应的devInfo
354 1 : nlohmann::json localDevInfoJson{};
355 1 : GetLocalDevInfoJson(parseJson, localDevInfoJson);
356 : // 3. 在反序列化和上报本地 RankTable 前改写 addr，确保后续全局 RankTable 和 RankGraph 使用选中地址
357 1 : SelectLocalHostBackupAddr(localDevInfoJson);
358 :
359 : // 4. 组rankTable的json格式
360 1 : nlohmann::json localRankTableJson{};
361 1 : GetLocalRankTableJson(parseJson, localRankTableJson);
362 1 : localRankTableJson["rank_list"].push_back(localDevInfoJson); // 添加localDevInfoJson
363 :
364 : // 5. 反序列化获得RankTableInfo
365 1 : std::string msgDeserialize = "error occurs when localRankTable Deserialize";
366 1 : TRY_CATCH_THROW(InvalidParamsException, msgDeserialize, localRankTable.Deserialize(localRankTableJson, false););
367 :
368 1 : CHK_PRT_THROW(
369 : localRankTable.ranks.empty(), HCCL_ERROR("[RankInfoDetectClient::%s] local rank table has no rank.", __func__),
370 : InvalidParamsException, "local rank table has no rank");
371 1 : CHK_PRT_CONT(
372 : GetLocalTlsStatus(localRankTable.ranks[0].tlsStatus),
373 : HCCL_WARNING("[GetLocalTlsStatus] Can not get TlsStatus"));
374 1 : CHK_PRT_CONT(
375 : GetLocalHostDpuTlsStatus(localRankTable.ranks[0].hostDpuTlsStatus),
376 : HCCL_WARNING("[GetLocalHostDpuTlsStatus] Can not get Host DPU TlsStatus"));
377 1 : HCCL_INFO("[RankInfoDetectClient::%s] end.", __func__);
378 1 : }
379 :
380 43 : void RankInfoDetectClient::ProbeHostRoceAddr(const IpAddress& hostAddr, bool& isAvailable) const
381 : {
382 43 : isAvailable = false;
383 : // hostAddr 来自 netLayer3 的 rank_addr_list，是 RoCE 数据通信地址；它不同于 RootInfoDetect
384 : // 使用的 rootHandle.ip，后者只负责控制面 socket 建链。
385 43 : EndpointDesc endpointDesc{};
386 43 : const HcommResult initRet = EndpointDescInit(&endpointDesc, 1);
387 43 : CHK_PRT_THROW(
388 : initRet != HCCL_SUCCESS, HCCL_ERROR("[%s] EndpointDescInit failed, ret[%d].", __func__, initRet),
389 : InternalException, StringFormat("[%s] EndpointDescInit failed, ret[%d]", __func__, initRet));
390 43 : endpointDesc.protocol = COMM_PROTOCOL_ROCE;
391 43 : endpointDesc.loc.locType = ENDPOINT_LOC_TYPE_HOST;
392 43 : const std::string msgFillCommAddr = "fill host RoCE endpoint address failed";
393 43 : TRY_CATCH_THROW(InvalidParamsException, msgFillCommAddr, FillCommAddr(endpointDesc.commAddr, hostAddr););
394 :
395 43 : EndpointHandle endpointHandle = nullptr;
396 43 : const HcommResult createRet = HcommEndpointCreate(&endpointDesc, &endpointHandle);
397 : // 仅网络错误允许上层继续尝试备用地址，其他错误按不可恢复异常立即终止。
398 43 : if (createRet == HCCL_E_NETWORK) {
399 29 : HCCL_WARNING(
400 : "[%s] host addr is unavailable, hostAddr[%s], ret[%d].", __func__, hostAddr.Describe().c_str(), createRet);
401 29 : return;
402 14 : } else if (createRet != HCCL_SUCCESS) {
403 2 : HCCL_ERROR(
404 : "[%s] HcommEndpointCreate failed, hostAddr[%s], ret[%d].", __func__, hostAddr.Describe().c_str(),
405 : createRet);
406 4 : THROW<InternalException>(StringFormat("[%s] HcommEndpointCreate failed, ret[%d]", __func__, createRet));
407 : }
408 12 : if (endpointHandle != nullptr) {
409 : // Endpoint 仅用于可用性探测，不参与后续通信，探测成功后立即释放。
410 3 : const HcommResult destroyRet = HcommEndpointDestroy(endpointHandle);
411 3 : CHK_PRT_THROW(
412 : destroyRet != HCCL_SUCCESS,
413 : HCCL_ERROR(
414 : "[%s] HcommEndpointDestroy failed, hostAddr[%s], ret[%d].", __func__, hostAddr.Describe().c_str(),
415 : destroyRet),
416 : InternalException, StringFormat("[%s] HcommEndpointDestroy failed, ret[%d]", __func__, destroyRet));
417 : }
418 11 : isAvailable = true;
419 11 : HCCL_INFO(
420 : "[%s] host addr probe success, devPhyId[%u], rankId[%u], hostAddr[%s].", __func__, devPhyId_, rankId_,
421 : hostAddr.Describe().c_str());
422 43 : }
423 :
424 24 : void RankInfoDetectClient::SelectLocalHostBackupAddr(nlohmann::json& localDevInfoJson)
425 : {
426 48 : const bool isLevelListInvalid = localDevInfoJson.empty() || !localDevInfoJson.contains("level_list")
427 48 : || !localDevInfoJson["level_list"].is_array();
428 28 : CHK_PRT_THROW(
429 : isLevelListInvalid,
430 : HCCL_ERROR(
431 : "[%s] level_list is missing or is not an array, devPhyId[%u], rankId[%u].", __func__, devPhyId_, rankId_),
432 : InvalidParamsException, "level_list is missing or is not an array");
433 22 : std::vector<nlohmann::json*> addrJsons;
434 22 : const std::string msgCollectLayer3Addr = "collect netLayer3 addr failed";
435 22 : TRY_CATCH_THROW(InvalidParamsException, msgCollectLayer3Addr, CollectLayer3AddrJsons(localDevInfoJson, addrJsons););
436 22 : if (addrJsons.empty()) {
437 5 : HCCL_DEBUG("[%s] no netLayer3+ addr with backup_addr needs probing.", __func__);
438 5 : return;
439 : }
440 : // 主备选择只依赖 RootInfo 的 net_layer 字段，不读取或构建 PhyTopo。
441 : // 对每个 netLayer3+ 地址独立探测主 addr；主地址不可用时，再按配置顺序逐个尝试 backup_addr。
442 26 : for (auto* addrJson : addrJsons) {
443 18 : SelectAvailableHostAddr(*addrJson);
444 : }
445 8 : HCCL_INFO(
446 : "[%s] end, devPhyId[%u], rankId[%u], addrConfigNum[%zu].", __func__, devPhyId_, rankId_, addrJsons.size());
447 36 : }
448 :
449 18 : void RankInfoDetectClient::SelectAvailableHostAddr(nlohmann::json& addrJson)
450 : {
451 18 : std::vector<IpAddress> candidates;
452 18 : BuildHostAddrCandidates(addrJson, candidates);
453 13 : HCCL_INFO(
454 : "[%s] devPhyId[%u], rankId[%u], primaryAddr[%s], backupAddrSize[%zu], "
455 : "candidateSize[%zu].",
456 : __func__, devPhyId_, rankId_, candidates.front().Describe().c_str(), candidates.size() - 1, candidates.size());
457 :
458 : // HCCL_E_NETWORK 是可恢复错误，通过 isAvailable 继续尝试下一个候选地址；
459 : // 其他异常不在此处恢复。
460 39 : for (std::size_t idx = 0; idx < candidates.size(); ++idx) {
461 39 : bool isAvailable = false;
462 39 : ProbeHostRoceAddr(candidates[idx], isAvailable);
463 37 : if (isAvailable) {
464 9 : UpdateSelectedHostAddr(addrJson, candidates, idx);
465 9 : return;
466 : }
467 28 : if (idx == candidates.size() - 1) {
468 28 : RPT_INPUT_ERR(
469 : true, "EI0016", std::vector<std::string>({"value", "variable", "expect"}),
470 : std::vector<std::string>(
471 : {candidates[idx].Describe(), "host addr candidates",
472 : "at least one available host addr for rank connections"}));
473 4 : THROW<NetworkApiException>(StringFormat("[%s] all host addr candidates are unavailable", __func__));
474 : }
475 26 : HCCL_WARNING(
476 : "[%s] host addr is unavailable, try next candidate, "
477 : "devPhyId[%u], rankId[%u], candidateAddr[%s], candidateIndex[%zu].",
478 : __func__, devPhyId_, rankId_, candidates[idx].Describe().c_str(), idx);
479 : }
480 22 : }
481 :
482 9 : void RankInfoDetectClient::UpdateSelectedHostAddr(
483 : nlohmann::json& addrJson, const std::vector<IpAddress>& candidates, std::size_t selectedIndex) const
484 : {
485 9 : const std::string oldAddr = addrJson[ADDR_FIELD].get<std::string>();
486 : // 只改写当前有效 addr，保留 backup_addr 原始配置，并随本地 RankTable 一并上报。
487 9 : addrJson[ADDR_FIELD] = candidates[selectedIndex].GetIpStr();
488 9 : HCCL_RUN_INFO(
489 : "[%s] select host addr success, devPhyId[%u], rankId[%u], "
490 : "selectedNicRole[%s], oldHostAddr[%s], selectedHostAddr[%s], candidateIndex[%zu], tryCount[%zu].",
491 : __func__, devPhyId_, rankId_, selectedIndex == 0 ? "primary" : "backup", oldAddr.c_str(),
492 : candidates[selectedIndex].Describe().c_str(), selectedIndex, selectedIndex + 1);
493 9 : }
494 :
495 1 : void RankInfoDetectClient::GetLocalDevInfoJson(const nlohmann::json& parseJson, nlohmann::json& localDevInfoJson)
496 : {
497 1 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
498 :
499 : // rankList字段对应json内容
500 1 : nlohmann::json rankJsons;
501 1 : std::string msgRanklist = "error occurs when parser object of propName \"rank_list\"";
502 1 : TRY_CATCH_THROW(InvalidParamsException, msgRanklist, GetJsonPropertyList(parseJson, "rank_list", rankJsons););
503 :
504 : // 获取localrankJsons, 匹配deviceId字段与当前devPhyId_匹配的内容
505 1 : for (auto& rankJson : rankJsons) {
506 1 : u32 devId = 0;
507 1 : std::string msgDeviceId = "error occurs when parser object of propName \"device_id\"";
508 1 : TRY_CATCH_THROW(InvalidParamsException, msgDeviceId, devId = GetJsonPropertyUInt(rankJson, "device_id"););
509 1 : if (devId == devPhyId_) {
510 1 : HCCL_INFO("[RankInfoDetectClient::%s] find localDevInfoJson.", __func__);
511 1 : localDevInfoJson = rankJson;
512 1 : break;
513 : }
514 1 : }
515 :
516 1 : if (localDevInfoJson.empty()) {
517 0 : HCCL_ERROR("[%s] failed, no device_id matches devPhyId_[%u] in rank_list.", __func__, devPhyId_);
518 : }
519 :
520 : // 添加rankId
521 1 : localDevInfoJson["rank_id"] = rankId_;
522 :
523 1 : HCCL_INFO("[RankInfoDetectClient::%s] end.", __func__);
524 1 : }
525 :
526 1 : void RankInfoDetectClient::GetLocalRankTableJson(const nlohmann::json& parseJson, nlohmann::json& localRankTableJson)
527 : {
528 1 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
529 :
530 1 : std::string version;
531 1 : std::string msgVersion = "error occurs when parser object of propName \"version\"";
532 1 : TRY_CATCH_THROW(InvalidParamsException, msgVersion, version = GetJsonProperty(parseJson, "version"););
533 1 : localRankTableJson["version"] = version;
534 :
535 1 : std::string detour;
536 1 : std::string msgDetour = "error occurs when parser object of propName \"detour\"";
537 1 : TRY_CATCH_THROW(InvalidParamsException, msgDetour, detour = GetJsonProperty(parseJson, "detour", false););
538 1 : if (detour == "true") {
539 0 : localRankTableJson["detour"] = detour;
540 : }
541 :
542 1 : localRankTableJson["rank_count"] = rankSize_;
543 1 : HCCL_INFO("[RankInfoDetectClient::%s] end.", __func__);
544 1 : }
545 :
546 1 : void RankInfoDetectClient::RecvRankTableMsg(vector<char>& rankInfoMsg)
547 : {
548 1 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
549 :
550 : // 接收数据
551 1 : u64 revMsgLen = 0;
552 1 : std::unique_ptr<HostBuffer> msg = std::make_unique<HostBuffer>(MAX_BUFFER_LEN);
553 1 : char* msgAddr = reinterpret_cast<char*>(msg->GetAddr());
554 1 : CHK_PRT_THROW(
555 : !socketAgent_.RecvMsg(msgAddr, revMsgLen),
556 : HCCL_ERROR("RankInfoDetectClient::%s, recv rankTable error.", __func__), SocketException, "client recv fail");
557 :
558 : // 以vector<char>格式保存
559 1 : rankInfoMsg.resize(revMsgLen);
560 1 : rankInfoMsg.assign(msgAddr, msgAddr + revMsgLen);
561 :
562 1 : HCCL_INFO("[RankInfoDetectClient::%s] end, revMsgLen[%llu].", __func__, revMsgLen);
563 1 : }
564 :
565 : // 解析接收到的rank table信息
566 2 : void RankInfoDetectClient::ParseRankTable(vector<char>& rankInfoMsg)
567 : {
568 2 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
569 :
570 : // 消息格式: [ranktable大小(u32, 4字节)][ranktable数据(n字节)][step(4字节)][failedAgentIdList]
571 2 : BinaryStream binStream(rankInfoMsg);
572 :
573 : // 解析localRankInfo
574 2 : rankTable_ = RankTableInfo(binStream);
575 2 : rankTable_.Dump();
576 :
577 : // 解析step
578 : u32 receivedStep;
579 2 : binStream >> receivedStep;
580 :
581 : // 解析failedAgentIdList
582 2 : std::string failedAgentIdList;
583 2 : binStream >> failedAgentIdList;
584 2 : if (failedAgentIdList.size() > 0) {
585 : // 建链失败时，打印 root 节点发来的临终遗言
586 7 : RPT_INPUT_ERR(
587 : true, "EI0015", std::vector<std::string>({"error_reason"}),
588 : std::vector<std::string>({"rank connection failed, failedRankIdList: " + failedAgentIdList}));
589 1 : HCCL_ERROR(
590 : "[RankInfoDetectClient::%s] TopoDetect ERROR occur, failedRankIdList[%s]", __func__,
591 : failedAgentIdList.c_str());
592 : }
593 :
594 2 : HCCL_INFO("[RankInfoDetectClient::%s] end.", __func__);
595 3 : }
596 :
597 1 : void RankInfoDetectClient::RecvRankTable()
598 : {
599 : // 获取rankTable
600 1 : vector<char> rankInfoMsg{};
601 1 : RecvRankTableMsg(rankInfoMsg);
602 :
603 : // 解析rankTable
604 1 : ParseRankTable(rankInfoMsg);
605 :
606 : // 校验
607 1 : VerifyRankTable();
608 1 : }
609 :
610 1 : void RankInfoDetectClient::VerifyRankTable()
611 : {
612 1 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
613 :
614 : // 校验rankCount符合预期
615 1 : if (rankTable_.rankCount != rankSize_) {
616 0 : THROW<InvalidParamsException>(StringFormat(
617 : "[RankInfoDetectClient::%s] rank_count[%u] does not match"
618 : " rankSize_[%u].",
619 : __func__, rankTable_.rankCount, rankSize_));
620 : }
621 :
622 : // 校验rankTable内容
623 1 : rankTable_.Check();
624 : // TLS开关一致性校验
625 1 : HcclResult ret = VerifyTlsConsistency();
626 1 : CHK_PRT_THROW(
627 : ret != HCCL_SUCCESS,
628 : HCCL_ERROR("[RankInfoDetectClient::%s] tls consistency verify failed, ret[%d]", __func__, ret),
629 : InvalidParamsException, "tls consistency verify failed");
630 :
631 1 : HcclResult hostDpuRet = VerifyHostDpuTlsConsistency();
632 3 : CHK_PRT_THROW(
633 : hostDpuRet != HCCL_SUCCESS,
634 : HCCL_ERROR("[RankInfoDetectClient::%s] hostDpuTls consistency verify failed, ret[%d]", __func__, hostDpuRet),
635 : InvalidParamsException, "hostDpuTls consistency verify failed");
636 :
637 0 : HCCL_INFO("[RankInfoDetectClient::%s] end.", __func__);
638 0 : }
639 :
640 6 : HcclResult RankInfoDetectClient::GetLocalTlsStatus(TlsStatus& tlsStatus) const
641 : {
642 6 : struct RaInfo raInfo {};
643 6 : raInfo.mode = NetworkMode::NETWORK_OFFLINE;
644 6 : raInfo.phyId = devPhyId_;
645 12 : return HrtRaGetTlsStatus(&raInfo, tlsStatus);
646 : }
647 :
648 8 : HcclResult RankInfoDetectClient::GetLocalHostDpuTlsStatus(TlsStatus& tlsStatus) const
649 : {
650 8 : struct RaInfo raInfo {};
651 8 : raInfo.mode = NetworkMode::NETWORK_PEER_ONLINE;
652 8 : raInfo.phyId = devPhyId_;
653 16 : return HrtRaGetTlsStatus(&raInfo, tlsStatus);
654 : }
655 :
656 33 : void RankInfoDetectClient::GenerateTlsStatusStr(std::string& tlsStatusStr, const std::vector<u32>& tlsStatusRanks) const
657 : {
658 33 : tlsStatusStr.clear();
659 63 : for (const auto& rank : tlsStatusRanks) {
660 30 : tlsStatusStr += std::to_string(rank) + ",";
661 : }
662 33 : if (!tlsStatusStr.empty() && tlsStatusStr.back() == ',') {
663 23 : tlsStatusStr.pop_back();
664 : }
665 33 : }
666 :
667 5 : void RankInfoDetectClient::ReportTlsConfigurationError(
668 : const std::string& tlsType, const std::string& tlsInconsistentTlsType, const std::string& tlsEnableRankStr,
669 : const std::string& tlsDisableRankStr, const std::string& tlsUnknownRankStr) const
670 : {
671 : std::string expectMessage = "\"All ranks are consistent. Current status: rankList for enabled tls: "
672 10 : + tlsEnableRankStr + "; rankList for disabled tls: " + tlsDisableRankStr
673 5 : + "; rankList for query failure tls: " + tlsUnknownRankStr;
674 10 : std::string errormessage = "Value \"" + tlsInconsistentTlsType + "\" for config \"" + tlsType
675 5 : + "\" is invalid. Expected: " + expectMessage;
676 :
677 65 : RPT_INPUT_ERR(
678 : true, "EI0016", std::vector<std::string>({"value", "variable", "expect"}),
679 : std::vector<std::string>({tlsInconsistentTlsType, "\"" + tlsType + "\"", expectMessage}));
680 :
681 5 : HCCL_ERROR("[ReportTlsConfigurationError][RanktableCheck] %s", errormessage.c_str());
682 10 : }
683 :
684 6 : HcclResult RankInfoDetectClient::VerifyTlsConsistency() const
685 : {
686 14 : auto getTlsStatus = [](const NewRankInfo& rankInfo) {
687 14 : return rankInfo.tlsStatus;
688 : };
689 18 : return VerifyTlsConsistencyByTlsType("tls", getTlsStatus);
690 : }
691 :
692 7 : HcclResult RankInfoDetectClient::VerifyHostDpuTlsConsistency() const
693 : {
694 16 : auto getTlsStatus = [](const NewRankInfo& rankInfo) {
695 16 : return rankInfo.hostDpuTlsStatus;
696 : };
697 21 : return VerifyTlsConsistencyByTlsType("hostDpuTls", getTlsStatus);
698 : }
699 :
700 13 : HcclResult RankInfoDetectClient::VerifyTlsConsistencyByTlsType(
701 : const std::string& tlsType, const std::function<TlsStatus(const NewRankInfo&)>& getTlsStatus) const
702 : {
703 13 : bool isSupportCheckTlsStatus = true; // 用于标识是否存在不支持查询Tls开关状态的情况
704 13 : bool isTlsConsistent = true; // 用于标识TLS开关状态是否一致
705 13 : std::vector<u32> tlsEnableRank;
706 13 : std::vector<u32> tlsDisableRank;
707 13 : std::vector<u32> tlsUnknownRank;
708 :
709 43 : for (const auto& rankInfo : rankTable_.ranks) {
710 30 : const TlsStatus tlsStatus = getTlsStatus(rankInfo);
711 30 : if (tlsStatus == TlsStatus::ENABLE) {
712 11 : tlsEnableRank.push_back(rankInfo.rankId);
713 19 : } else if (tlsStatus == TlsStatus::DISABLE) {
714 9 : tlsDisableRank.push_back(rankInfo.rankId);
715 : } else {
716 10 : isSupportCheckTlsStatus = false;
717 10 : tlsUnknownRank.push_back(rankInfo.rankId);
718 : }
719 : }
720 :
721 : // 将卡的信息汇总成string
722 13 : std::string tlsEnableRankStr;
723 13 : std::string tlsDisableRankStr;
724 13 : std::string tlsUnknownRankStr;
725 13 : GenerateTlsStatusStr(tlsEnableRankStr, tlsEnableRank);
726 13 : GenerateTlsStatusStr(tlsDisableRankStr, tlsDisableRank);
727 13 : if (!isSupportCheckTlsStatus) {
728 7 : GenerateTlsStatusStr(tlsUnknownRankStr, tlsUnknownRank);
729 : }
730 :
731 13 : std::string tlsInconsistentTlsType;
732 13 : if (!tlsEnableRank.empty() && !tlsDisableRank.empty()) {
733 5 : isTlsConsistent = false;
734 5 : tlsInconsistentTlsType = (tlsDisableRank.size() <= tlsEnableRank.size()) ? "Disable" : "Enable";
735 : }
736 :
737 : // 四种不同情况
738 13 : if (isTlsConsistent && isSupportCheckTlsStatus) {
739 : // 1.通信域所有卡都支持查询TLS开关状态，并且TLS开关状态都是一致的。
740 4 : HCCL_INFO("[Verify][%sConsistency] All ranks %sStatus are consistent", tlsType.c_str(), tlsType.c_str());
741 9 : } else if (!isTlsConsistent && isSupportCheckTlsStatus) {
742 : // 2.通信域所有卡都支持查询TLS开关状态，但是TLS开关状态存在不一致，报错。
743 2 : ReportTlsConfigurationError(
744 : tlsType, tlsInconsistentTlsType, tlsEnableRankStr, tlsDisableRankStr, tlsUnknownRankStr);
745 2 : return HCCL_E_PARA;
746 7 : } else if (isTlsConsistent && !isSupportCheckTlsStatus) {
747 : // 3.通信域内的部分卡不支持查询TLS开关状态，目前能查询到的卡的TLS开关状态是一致的，打印warning提醒
748 4 : HCCL_WARNING(
749 : "[Verify][%sConsistency] Some ranks do not support to check %sStatus, "
750 : "not support rankId: [%s]",
751 : tlsType.c_str(), tlsType.c_str(), tlsUnknownRankStr.c_str());
752 : } else {
753 : // 4.通信域内的部分卡不支持查询TLS开关状态，但是目前能查询到的卡的TLS开关状态已经不一致，报错
754 3 : ReportTlsConfigurationError(
755 : tlsType, tlsInconsistentTlsType, tlsEnableRankStr, tlsDisableRankStr, tlsUnknownRankStr);
756 3 : return HCCL_E_PARA;
757 : }
758 :
759 8 : return HCCL_SUCCESS;
760 13 : }
761 :
762 6 : void RankInfoDetectClient::HostListenPortDetect(NewRankInfo& rankInfo)
763 : {
764 6 : std::string topoPath = GetRootInfoTopoFilePath();
765 6 : PhyTopoBuilder::GetInstance().Build(topoPath);
766 6 : auto devLogicId = HrtGetDevice();
767 6 : u32 devPhyId = rankInfo.deviceId;
768 12 : for (auto& rankLevelInfo : rankInfo.rankLevelInfos) {
769 : shared_ptr<Graph<PhyTopo::Node, PhyTopo::Link>> graph
770 7 : = PhyTopo::GetInstance()->GetTopoGraph(rankLevelInfo.netLayer);
771 7 : if (graph == nullptr) {
772 4 : HCCL_DEBUG("[RankInfoDetectClient::%s]Can't find the layout %u Graph!", __func__, rankLevelInfo.netLayer);
773 4 : continue;
774 : }
775 3 : std::vector<std::shared_ptr<PhyTopo::Link>> links = graph->GetEdges(rankInfo.localId);
776 5 : for (auto& link : links) {
777 3 : if (link->GetSourceIFace()->GetPos() != AddrPosition::HOST) {
778 1 : continue;
779 : }
780 2 : const std::set<LinkProtocol>& protocols = link->GetLinkProtocols();
781 3 : for (auto& protocol : protocols) {
782 2 : LinkProtoType protoType = LinkProtocol2LinkProtoType(protocol);
783 2 : if (protoType != LinkProtoType::RDMA || rankLevelInfo.rankAddrs.empty()) {
784 1 : continue;
785 : }
786 1 : HCCL_DEBUG("[SocketManager::%s] find the host rdma link %s", __func__, link->Describe().c_str());
787 1 : const IpAddress& hostIp = rankLevelInfo.rankAddrs[0].addr;
788 1 : uint32_t hostPort = 0;
789 1 : SetupHostListenPort(devLogicId, devPhyId, hostIp, hostPort);
790 1 : rankInfo.hostPort = hostPort;
791 1 : return;
792 : }
793 2 : }
794 8 : }
795 6 : }
796 :
797 2 : void RankInfoDetectClient::SetupHostListenPort(
798 : u32 devLogicId, u32 devPhyId, const IpAddress& hostIp, uint32_t& hostPort)
799 : {
800 2 : std::lock_guard<std::mutex> lock(hostSocketLock_);
801 2 : u32 listenPort = HCCL_INVALID_PORT;
802 2 : auto portRange = EnvConfig::GetInstance().GetHostNicConfig().GetHostSocketPortRange();
803 2 : u32 basePort = EnvConfig::GetInstance().GetHostNicConfig().GetIfBasePort();
804 2 : if (portRange.empty() && basePort != HCCL_INVALID_PORT) {
805 1 : listenPort = basePort + devPhyId;
806 1 : HCCL_INFO("[RankInfoDetectClient::%s] BasePort is configured, listenPort[%u].", __func__, listenPort);
807 1 : hostPort = listenPort;
808 1 : return;
809 : }
810 :
811 1 : if (portRange.empty()) {
812 1 : constexpr u32 HOST_CONTROL_BASE_PORT = 60000; // 控制面起始port
813 1 : HCCL_INFO(
814 : "[RankInfoDetectClient::%s] No port configuration, using default port range[%u, %u]", __func__,
815 : HOST_CONTROL_BASE_PORT, HOST_CONTROL_BASE_PORT + HOST_CONTROL_PORT_COUNT);
816 1 : SocketPortRange defaultRange = {HOST_CONTROL_BASE_PORT, HOST_CONTROL_BASE_PORT + HOST_CONTROL_PORT_COUNT};
817 1 : portRange.push_back(defaultRange);
818 : }
819 :
820 1 : SocketHandle hostSocketHandle = HostSocketHandleManager::GetInstance().Create(devPhyId, hostIp);
821 1 : hostSocket_ = std::make_shared<Socket>(
822 0 : hostSocketHandle, hostIp, HCCL_INVALID_PORT, hostIp, "hostport_preempt", SocketRole::SERVER,
823 1 : NicType::HOST_NIC_TYPE);
824 1 : PreemptPortManager::GetInstance(devLogicId).ListenPreempt(hostSocket_, portRange, listenPort);
825 1 : HCCL_INFO("[RankInfoDetectClient::%s] preempt hostPort[%u] success.", __func__, listenPort);
826 :
827 : // 登记到进程级 map，供算子下发阶段复用，避免跨阶段端口竞争
828 1 : DevNetPortType portType(ConnectProtoType::RDMA);
829 1 : PortData portData(static_cast<RankId>(devPhyId), portType, 0, hostIp);
830 1 : SocketManager socketMgr(0, devPhyId, static_cast<u32>(devLogicId), "hostport_preempt");
831 1 : hostSocketRegistered_ = socketMgr.RegisterHostListenSocket(portData, hostSocket_);
832 :
833 1 : hostPort = listenPort;
834 3 : }
835 :
836 63 : void RankInfoDetectClient::SocketTearDown(u32 devPhyId)
837 : {
838 63 : std::lock_guard<std::mutex> lock(hostSocketLock_);
839 63 : if (hostSocket_ == nullptr) {
840 61 : return;
841 : }
842 2 : const IpAddress& hostIp = hostSocket_->GetLocalIp();
843 2 : auto devLogicId = HrtGetDevice();
844 2 : if (hostSocketRegistered_) {
845 : // 已登记到 SocketManager::GetServerSocketMap()，所有权已转移给 map，
846 : // 由算子下发阶段 HostSocketStopListen refcount 归 0 时清理，此处跳过 Release/Destroy
847 2 : HCCL_INFO(
848 : "[RankInfoDetectClient::%s] hostSocket already registered to ServerSocketMap, "
849 : "skip Release/Destroy, only release local ref.",
850 : __func__);
851 0 : } else if (
852 0 : EnvConfig::GetInstance().GetHostNicConfig().GetHostSocketPortRange().size() > 0
853 0 : || EnvConfig::GetInstance().GetHostNicConfig().GetIfBasePort() == HCCL_INVALID_PORT) {
854 : // 若开启抢占监听端口
855 0 : PreemptPortManager::GetInstance(devLogicId).Release(hostSocket_);
856 0 : HostSocketHandleManager::GetInstance().Destroy(devPhyId, hostIp);
857 : }
858 2 : hostSocket_ = nullptr;
859 63 : }
860 :
861 59 : void RankInfoDetectClient::TearDown()
862 : {
863 59 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
864 59 : SocketTearDown(devPhyId_);
865 :
866 : // close socket
867 59 : clientSocket_->Close();
868 :
869 : // deinit handle
870 59 : HostSocketHandleManager::GetInstance().Destroy(devPhyId_, clientSocket_->GetLocalIp());
871 :
872 : // deinit ra in detach thread to avoid block main thread
873 59 : s32 deviceLogicId = HrtGetDevice();
874 59 : std::thread{[deviceLogicId]() {
875 59 : EXCEPTION_CATCH(
876 : HccpPeerManager::GetInstance().DeInit(deviceLogicId),
877 : HCCL_ERROR("[RankInfoDetectClient::TearDown] DeInit exception"));
878 118 : }}.detach();
879 :
880 59 : HCCL_INFO("[RankInfoDetectClient::%s] end.", __func__);
881 59 : }
882 :
883 60 : RankInfoDetectClient::~RankInfoDetectClient() { DECTOR_TRY_CATCH("RankInfoDetectClient", TearDown()); }
884 :
885 : } // namespace Hccl
|
