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