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 <thread>
12 : #include "rank_info_detect_client.h"
13 : #include "root_handle_v2.h"
14 : #include "env_config.h"
15 : #include "host_buffer.h"
16 : #include "binary_stream.h"
17 : #include "hccp_peer_manager.h"
18 : #include "orion_adapter_hccp.h"
19 : #include "orion_adapter_rts.h"
20 : #include "host_socket_handle_manager.h"
21 : #include "socket_manager.h"
22 : #include "topo_addr_info.h"
23 : #include "adapter_error_manager_pub.h"
24 : #include "communicator_impl.h"
25 : #include "phy_topo_builder.h"
26 : #include "preempt_port_manager.h"
27 :
28 : namespace Hccl {
29 : constexpr u32 HOST_CONTROL_PORT_COUNT = 15;
30 :
31 0 : void RankInfoDetectClient::Setup(RankTableInfo &rankTable)
32 : {
33 : // 1. 构造localRankTable
34 0 : RankTableInfo localRankTable{};
35 0 : ConstructRankTable(localRankTable);
36 :
37 : // 若启用单卡多进程抢占端口则执行
38 0 : SocketManager::ServerInitAll(localRankTable.ranks[0]);
39 0 : HostListenPortDetect(localRankTable.ranks[0]);
40 :
41 : // 2. 连接root节点
42 0 : Connect();
43 :
44 : // 3. 发送本端agentId和rankSize
45 0 : SendAgentIdAndRankSize();
46 :
47 : // 4. 发送给root节点
48 0 : SendLocalRankTable(localRankTable);
49 :
50 : // 5. 接收完整rankTable
51 0 : RecvRankTable();
52 0 : rankTable = rankTable_;
53 0 : }
54 :
55 0 : void RankInfoDetectClient::Connect()
56 : {
57 0 : clientSocket_->Connect();
58 0 : CheckStatus();
59 0 : }
60 :
61 2 : void RankInfoDetectClient::CheckStatus()
62 : {
63 6 : HCCL_DEBUG("[RankInfoDetectClient::%s] start.", __func__);
64 :
65 2 : auto startTime = std::chrono::steady_clock::now();
66 2 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
67 :
68 : while (true) {
69 836185 : bool isTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
70 836185 : if (isTimeout) {
71 3 : HCCL_ERROR("[RankInfoDetectClient::%s] get connected status socket timeout! timeout[%lld s]", __func__, timeout);
72 1 : RPT_INPUT_ERR(isTimeout, "EI0015", std::vector<std::string>({"error_reason"}),
73 : std::vector<std::string>({StringFormat("Receiving message from the root node timed out "
74 : "Timeout was set to %lld seconds. Check whether node rankId[%u] reports an error.",
75 : static_cast<long long>(timeout.count()), rankId_)}));
76 : // 建链超时后，sleep 20s，避免上层应用提前退出，确保其他正常 client 能够收到 server 发出的临终遗言
77 1 : sleep(WAIT_ERROR_BROADCAST_TIME);
78 1 : THROW<TimeoutException>("client get connection timeout");
79 : }
80 :
81 836184 : if (clientSocket_->GetStatus() == SocketStatus::OK) {
82 3 : HCCL_DEBUG("[RankInfoDetectClient::%s] client get socket connection success.", __func__);
83 1 : break;
84 : }
85 836183 : }
86 :
87 3 : HCCL_INFO("[RankInfoDetectClient::%s] end, connect ok.", __func__);
88 1 : }
89 :
90 1 : void RankInfoDetectClient::SendAgentIdAndRankSize()
91 : {
92 3 : HCCL_DEBUG("[RankInfoDetectClient::%s] start.", __func__);
93 :
94 : // 发送agentId
95 1 : std::string rankID = std::to_string(rankId_);
96 1 : std::string agentID = std::string(16 - rankID.length(), '0') + rankID;
97 1 : socketAgent_.SendMsg(agentID.c_str(), agentID.size());
98 :
99 : // 发送rankSize
100 1 : socketAgent_.SendMsg(&rankSize_, sizeof(rankSize_));
101 :
102 3 : HCCL_INFO("[RankInfoDetectClient::%s] send agentID[%s] and rankSize_[%u] end.",
103 : __func__, agentID.c_str(), rankSize_);
104 1 : }
105 :
106 0 : void RankInfoDetectClient::SendLocalRankTable(const RankTableInfo &localRankTable)
107 : {
108 0 : HCCL_DEBUG("[RankInfoDetectClient::%s] start.", __func__);
109 :
110 : // 消息格式: [ranktable数据(n字节)][step(4字节)]
111 0 : BinaryStream binaryStream;
112 0 : localRankTable.GetBinStream(true, binaryStream);
113 0 : binaryStream << currentStep_;
114 :
115 : // 字节流转换为vector<char>格式
116 0 : vector<char> sendMsg;
117 0 : binaryStream.Dump(sendMsg);
118 :
119 : // 发送
120 0 : socketAgent_.SendMsg(sendMsg.data(), sendMsg.size());
121 :
122 0 : HCCL_INFO("[RankInfoDetectClient::%s] end, currentStep_[%u].", __func__, currentStep_);
123 0 : currentStep_++;
124 0 : }
125 :
126 4 : void RankInfoDetectClient::ConstructSingleRank(RankTableInfo &localRankTable)
127 : {
128 4 : localRankTable.version = "2.0";
129 4 : localRankTable.rankCount = 1;
130 4 : NewRankInfo rankInfo{};
131 4 : rankInfo.rankId = rankId_;
132 4 : rankInfo.rankLevelInfos.emplace_back(RankLevelInfo{});
133 12 : CHK_PRT_CONT(GetLocalTlsStatus(rankInfo.tlsStatus),
134 : HCCL_WARNING("[GetLocalTlsStatus] Can not get TlsStatus"));
135 4 : localRankTable.ranks.emplace_back(rankInfo);
136 :
137 : // 打印
138 4 : localRankTable.Dump();
139 12 : HCCL_INFO("[RankInfoDetectClient::%s] end, single rank, localRankTable[%s].", __func__, localRankTable.Describe().c_str());
140 4 : }
141 :
142 1 : void CheckRootInfoJson(const nlohmann::json &parseJson)
143 : {
144 : // check version
145 1 : std::string version{};
146 1 : std::string msgVersion = "error occurs when parser rootinfo object of propName \"version\"";
147 1 : TRY_CATCH_THROW(InvalidParamsException, msgVersion, version = GetJsonProperty(parseJson, "version"););
148 1 : if (version != "2.0") {
149 0 : RPT_INPUT_ERR(true, "EI0014", std::vector<std::string>({"value", "variable", "expect"}),
150 : std::vector<std::string>({version, "version", "2.0"}));
151 0 : HCCL_ERROR("[%s] failed with version [%s] is not \"2.0\".", __func__ , version.c_str());
152 0 : THROW<InvalidParamsException>("version error");
153 : }
154 :
155 : // parser topo_file_path
156 1 : std::string topoFilePath{};
157 1 : std::string msgRankTopoFile = "error occurs when parser object of propName \"topo_file_path\"";
158 1 : TRY_CATCH_THROW(InvalidParamsException, msgRankTopoFile, topoFilePath = GetJsonProperty(parseJson, "topo_file_path"););
159 :
160 : // check topo_file_path
161 1 : char resolvedPath[PATH_MAX] = {0};
162 1 : bool isInvalidPath = (realpath(topoFilePath.c_str(), resolvedPath) == nullptr);
163 1 : if (isInvalidPath) {
164 0 : RPT_INPUT_ERR(true, "EI0014", std::vector<std::string>({"value", "variable", "expect"}),
165 : std::vector<std::string>({topoFilePath, "topo_file_path", "valid path"}));
166 0 : HCCL_ERROR("[%s] topo_file_path[%s] is not a valid real path", __func__, topoFilePath.c_str());
167 0 : THROW<InvalidParamsException>("topo_file_path error");
168 : }
169 :
170 : // parser rank_count
171 1 : u32 rankCount{};
172 1 : std::string msgRankcount = "error occurs when parser object of propName \"rank_count\"";
173 1 : TRY_CATCH_THROW(InvalidParamsException, msgRankcount, rankCount = GetJsonPropertyUInt(parseJson, "rank_count"););
174 :
175 : // parser rank_list
176 1 : nlohmann::json rankJsons{};
177 1 : std::string msgRanklist = "error occurs when parser object of propName \"rank_list\"";
178 1 : TRY_CATCH_THROW(InvalidParamsException, msgRanklist,
179 : GetJsonPropertyList(parseJson, "rank_list", rankJsons););
180 :
181 : // check rank_count
182 1 : bool isRankCountMismatch = (rankCount != rankJsons.size());
183 1 : if (isRankCountMismatch) {
184 0 : RPT_INPUT_ERR(true, "EI0014", std::vector<std::string>({"value", "variable", "expect"}),
185 : std::vector<std::string>({std::to_string(rankCount), "rankCount", std::to_string(rankJsons.size())}));
186 0 : HCCL_ERROR("[%s] failed with rankCount is not equal to rank_list size."
187 : "rankCount[%u], ranks.size[%u]", __func__, rankCount, rankJsons.size());
188 0 : THROW<InvalidParamsException>("rankCount error");
189 : }
190 1 : }
191 :
192 1 : void RankInfoDetectClient::ConstructRankTable(RankTableInfo &localRankTable)
193 : {
194 3 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
195 :
196 : // 单P场景处理
197 1 : CHK_PRT_RET_NULL((rankSize_ == 1), ConstructSingleRank(localRankTable));
198 :
199 : // 1. 解析文件topoInfo.json
200 1 : std::string filePath = "/etc/hccl_rootinfo.json";
201 : JsonParser jsonParser{};
202 1 : nlohmann::json parseJson{};
203 1 : std::ifstream file(filePath);
204 1 : if (file.good()) {
205 0 : jsonParser.ParseFileToJson(filePath, parseJson);
206 : } else {
207 : size_t bufSize;
208 1 : s32 result = TopoAddrInfoGetSize(devPhyId_, &bufSize); // 获取rankInfo大小，用于提前分配内存
209 1 : CHK_PRT_THROW(result != 0 || bufSize > MAX_BUFFER_LEN,
210 : HCCL_ERROR("[RankInfoDetectClient::%s] Get rankinfo size failed.", __func__),
211 : InvalidParamsException, "Get rankinfo size failed.");
212 1 : std::vector<char> buffer(bufSize, '\0');
213 1 : result = TopoAddrInfoGet(devPhyId_, buffer.data(), &bufSize); // 获取rankInfo 并更新大小
214 1 : CHK_PRT_THROW(result != 0,
215 : HCCL_ERROR("[RankInfoDetectClient::%s] Get rankinfo failed.", __func__),
216 : InvalidParamsException, "Get rankinfo size failed.");
217 1 : std::string jsonString(buffer.data(), bufSize);
218 : // 将生成的info信息转换成json文件
219 1 : parseJson = nlohmann::json::parse(jsonString);
220 1 : }
221 1 : CheckRootInfoJson(parseJson);
222 :
223 : // 2. 获取当前devPhyId_对应的devInfo
224 1 : nlohmann::json localDevInfoJson{};
225 1 : GetLocalDevInfoJson(parseJson, localDevInfoJson);
226 :
227 : // 3. 组rankTable的json格式
228 1 : nlohmann::json localRankTableJson{};
229 1 : GetLocalRankTableJson(parseJson, localRankTableJson);
230 1 : localRankTableJson["rank_list"].push_back(localDevInfoJson); // 添加localDevInfoJson
231 :
232 : // 4. 反序列化获得RankTableInfo
233 1 : std::string msgDeserialize = "error occurs when localRankTable Deserialize";
234 1 : TRY_CATCH_THROW(InvalidParamsException, msgDeserialize, localRankTable.Deserialize(localRankTableJson, false););
235 :
236 1 : CHK_PRT_THROW(localRankTable.ranks.empty(),
237 : HCCL_ERROR("[RankInfoDetectClient::%s] local rank table has no rank.", __func__),
238 : InvalidParamsException, "local rank table has no rank");
239 3 : CHK_PRT_CONT(GetLocalTlsStatus(localRankTable.ranks[0].tlsStatus),
240 : HCCL_WARNING("[GetLocalTlsStatus] Can not get TlsStatus"));
241 3 : HCCL_INFO("[RankInfoDetectClient::%s] end.", __func__);
242 1 : }
243 :
244 1 : void RankInfoDetectClient::GetLocalDevInfoJson(const nlohmann::json &parseJson, nlohmann::json &localDevInfoJson)
245 : {
246 3 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
247 :
248 : // rankList字段对应json内容
249 1 : nlohmann::json rankJsons;
250 1 : std::string msgRanklist = "error occurs when parser object of propName \"rank_list\"";
251 1 : TRY_CATCH_THROW(InvalidParamsException, msgRanklist,
252 : GetJsonPropertyList(parseJson, "rank_list", rankJsons););
253 :
254 : // 获取localrankJsons, 匹配deviceId字段与当前devPhyId_匹配的内容
255 1 : for (auto &rankJson : rankJsons) {
256 1 : u32 devId = 0;
257 1 : std::string msgDeviceId = "error occurs when parser object of propName \"device_id\"";
258 1 : TRY_CATCH_THROW(InvalidParamsException, msgDeviceId,
259 : devId = GetJsonPropertyUInt(rankJson, "device_id");
260 : );
261 1 : if (devId == devPhyId_) {
262 3 : HCCL_INFO("[RankInfoDetectClient::%s] find localDevInfoJson.", __func__);
263 1 : localDevInfoJson = rankJson;
264 1 : break;
265 : }
266 1 : }
267 :
268 1 : if (localDevInfoJson.empty()) {
269 0 : HCCL_ERROR("[%s] failed, no device_id matches devPhyId_[%u] in rank_list.", __func__, devPhyId_);
270 : }
271 :
272 : // 添加rankId
273 1 : localDevInfoJson["rank_id"] = rankId_;
274 :
275 3 : HCCL_INFO("[RankInfoDetectClient::%s] end.", __func__);
276 1 : }
277 :
278 1 : void RankInfoDetectClient::GetLocalRankTableJson(const nlohmann::json &parseJson, nlohmann::json &localRankTableJson)
279 : {
280 3 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
281 :
282 1 : std::string version;
283 1 : std::string msgVersion = "error occurs when parser object of propName \"version\"";
284 1 : TRY_CATCH_THROW(InvalidParamsException, msgVersion, version = GetJsonProperty(parseJson, "version"););
285 1 : localRankTableJson["version"] = version;
286 :
287 1 : std::string detour;
288 1 : std::string msgDetour = "error occurs when parser object of propName \"detour\"";
289 1 : TRY_CATCH_THROW(InvalidParamsException, msgDetour, detour = GetJsonProperty(parseJson, "detour", false););
290 1 : if (detour == "true") {
291 0 : localRankTableJson["detour"] = detour;
292 : }
293 :
294 1 : localRankTableJson["rank_count"] = rankSize_;
295 3 : HCCL_INFO("[RankInfoDetectClient::%s] end.", __func__);
296 1 : }
297 :
298 1 : void RankInfoDetectClient::RecvRankTableMsg(vector<char> &rankInfoMsg)
299 : {
300 3 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
301 :
302 : // 接收数据
303 1 : u64 revMsgLen = 0;
304 1 : std::unique_ptr<HostBuffer> msg = std::make_unique<HostBuffer>(MAX_BUFFER_LEN);
305 1 : char *msgAddr = reinterpret_cast<char *>(msg->GetAddr());
306 1 : CHK_PRT_THROW(!socketAgent_.RecvMsg(msgAddr, revMsgLen),
307 : HCCL_ERROR("RankInfoDetectClient::%s, recv rankTable error.", __func__),
308 : SocketException, "client recv fail");
309 :
310 : // 以vector<char>格式保存
311 1 : rankInfoMsg.resize(revMsgLen);
312 1 : rankInfoMsg.assign(msgAddr, msgAddr + revMsgLen);
313 :
314 3 : HCCL_INFO("[RankInfoDetectClient::%s] end, revMsgLen[%llu].", __func__, revMsgLen);
315 1 : }
316 :
317 : // 解析接收到的rank table信息
318 1 : void RankInfoDetectClient::ParseRankTable(vector<char> &rankInfoMsg)
319 : {
320 3 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
321 :
322 : // 消息格式: [ranktable大小(u32, 4字节)][ranktable数据(n字节)][step(4字节)][failedAgentIdList]
323 1 : BinaryStream binStream(rankInfoMsg);
324 :
325 : // 解析localRankInfo
326 1 : rankTable_ = RankTableInfo(binStream);
327 1 : rankTable_.Dump();
328 :
329 : // 解析step
330 : u32 receivedStep;
331 1 : binStream >> receivedStep;
332 :
333 : // 解析failedAgentIdList
334 1 : std::string failedAgentIdList;
335 1 : binStream >> failedAgentIdList;
336 1 : if (failedAgentIdList.size() > 0) {
337 : // 建链失败时，打印 root 节点发来的临终遗言
338 0 : HCCL_ERROR("[RankInfoDetectClient::%s] TopoDetect ERROR occur, failedRankIdList[%s]",
339 : __func__, failedAgentIdList.c_str());
340 : }
341 :
342 3 : HCCL_INFO("[RankInfoDetectClient::%s] end.", __func__);
343 1 : }
344 :
345 1 : void RankInfoDetectClient::RecvRankTable()
346 : {
347 : // 获取rankTable
348 1 : vector<char> rankInfoMsg{};
349 1 : RecvRankTableMsg(rankInfoMsg);
350 :
351 : // 解析rankTable
352 1 : ParseRankTable(rankInfoMsg);
353 :
354 : // 校验
355 1 : VerifyRankTable();
356 1 : }
357 :
358 0 : void RankInfoDetectClient::VerifyRankTable()
359 : {
360 0 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
361 :
362 : // 校验rankCount符合预期
363 0 : if (rankTable_.rankCount != rankSize_) {
364 0 : THROW<InvalidParamsException>(StringFormat("[RankInfoDetectClient::%s] rank_count[%u] does not match"
365 : " rankSize_[%u].", __func__, rankTable_.rankCount, rankSize_));
366 : }
367 :
368 : // 校验rankTable内容
369 0 : rankTable_.Check();
370 : // TLS开关一致性校验
371 0 : HcclResult ret = VerifyTlsConsistency();
372 0 : CHK_PRT_THROW(ret != HCCL_SUCCESS,
373 : HCCL_ERROR("[RankInfoDetectClient::%s] tls consistency verify failed, ret[%d]", __func__, ret),
374 : InvalidParamsException, "tls consistency verify failed");
375 :
376 0 : HCCL_INFO("[RankInfoDetectClient::%s] end.", __func__);
377 0 : }
378 :
379 5 : HcclResult RankInfoDetectClient::GetLocalTlsStatus(TlsStatus &tlsStatus) const
380 : {
381 : struct RaInfo raInfo;
382 5 : raInfo.mode = NetworkMode::NETWORK_OFFLINE;
383 5 : raInfo.phyId = devPhyId_;
384 10 : return HrtRaGetTlsStatus(&raInfo, tlsStatus);
385 : }
386 :
387 12 : void RankInfoDetectClient::GenerateTlsStatusStr(
388 : std::string &tlsStatusStr, const std::vector<u32> &tlsStatusRanks) const
389 : {
390 12 : tlsStatusStr.clear();
391 23 : for (const auto &rank : tlsStatusRanks) {
392 11 : tlsStatusStr += std::to_string(rank) + ",";
393 : }
394 12 : if (!tlsStatusStr.empty() && tlsStatusStr.back() == ',') {
395 9 : tlsStatusStr.pop_back();
396 : }
397 12 : }
398 :
399 2 : void RankInfoDetectClient::ReportTlsConfigurationError(const std::string &tlsInconsistentTlsType,
400 : const std::string &tlsEnableRankStr, const std::string &tlsDisableRankStr,
401 : const std::string &tlsUnknownRankStr) const
402 : {
403 2 : std::string expectMessage = "\"All ranks are consistent. Current status: rankList for enabled tls: " +
404 4 : tlsEnableRankStr + "; rankList for disabled tls: " + tlsDisableRankStr +
405 2 : "; rankList for query failure tls: " + tlsUnknownRankStr + ".\"";
406 4 : std::string errormessage = "Value \"" + tlsInconsistentTlsType +
407 2 : "\" for config \"tls\" is invalid. Expected: " + expectMessage;
408 :
409 2 : RPT_INPUT_ERR(true,
410 : "EI0016",
411 : std::vector<std::string>({"value", "variable", "expect"}),
412 : std::vector<std::string>({tlsInconsistentTlsType, "\"tls\"", expectMessage}));
413 :
414 6 : HCCL_ERROR("[ReportTlsConfigurationError][RanktableCheck] %s", errormessage.c_str());
415 2 : }
416 :
417 5 : HcclResult RankInfoDetectClient::VerifyTlsConsistency() const
418 : {
419 5 : bool isSupportCheckTlsStatus = true; // 用于标识是否存在不支持查询Tls开关状态的情况
420 5 : bool isTlsConsistent = true; // 用于标识TLS开关状态是否一致
421 5 : std::vector<u32> tlsEnableRank;
422 5 : std::vector<u32> tlsDisableRank;
423 5 : std::vector<u32> tlsUnknownRank;
424 :
425 16 : for (const auto &rankInfo : rankTable_.ranks) {
426 11 : if (rankInfo.tlsStatus == TlsStatus::ENABLE) {
427 5 : tlsEnableRank.push_back(rankInfo.rankId);
428 6 : } else if (rankInfo.tlsStatus == TlsStatus::DISABLE) {
429 4 : tlsDisableRank.push_back(rankInfo.rankId);
430 : } else {
431 2 : isSupportCheckTlsStatus = false;
432 2 : tlsUnknownRank.push_back(rankInfo.rankId);
433 : }
434 : }
435 :
436 : // 将卡的信息汇总成string
437 5 : std::string tlsEnableRankStr;
438 5 : std::string tlsDisableRankStr;
439 5 : std::string tlsUnknownRankStr;
440 5 : GenerateTlsStatusStr(tlsEnableRankStr, tlsEnableRank);
441 5 : GenerateTlsStatusStr(tlsDisableRankStr, tlsDisableRank);
442 5 : if (!isSupportCheckTlsStatus) {
443 2 : GenerateTlsStatusStr(tlsUnknownRankStr, tlsUnknownRank);
444 : }
445 :
446 5 : std::string tlsInconsistentTlsType;
447 5 : if (!tlsEnableRank.empty() && !tlsDisableRank.empty()) {
448 2 : isTlsConsistent = false;
449 2 : tlsInconsistentTlsType = (tlsDisableRank.size() <= tlsEnableRank.size()) ? "Disable" : "Enable";
450 : }
451 :
452 : // 四种不同情况
453 5 : if (isTlsConsistent && isSupportCheckTlsStatus) {
454 : // 1.通信域所有卡都支持查询TLS开关状态，并且TLS开关状态都是一致的。
455 6 : HCCL_INFO("[Verify][TlsConsistency] All ranks tlsStatus are consistent");
456 5 : } else if (!isTlsConsistent && isSupportCheckTlsStatus) {
457 : // 2.通信域所有卡都支持查询TLS开关状态，但是TLS开关状态存在不一致，报错。
458 1 : ReportTlsConfigurationError(
459 : tlsInconsistentTlsType, tlsEnableRankStr, tlsDisableRankStr, tlsUnknownRankStr);
460 1 : return HCCL_E_PARA;
461 2 : } else if (isTlsConsistent && !isSupportCheckTlsStatus) {
462 : // 3.通信域内的部分卡不支持查询TLS开关状态，目前能查询到的卡的TLS开关状态是一致的，打印warning提醒
463 3 : HCCL_WARNING("[Verify][TlsConsistency] Some ranks do not support to check tlsStatus, " \
464 : "not support rankId: [%s]", tlsUnknownRankStr.c_str());
465 1 : } else {
466 : // 4.通信域内的部分卡不支持查询TLS开关状态，但是目前能查询到的卡的TLS开关状态已经不一致，报错
467 1 : ReportTlsConfigurationError(
468 : tlsInconsistentTlsType, tlsEnableRankStr, tlsDisableRankStr, tlsUnknownRankStr);
469 1 : return HCCL_E_PARA;
470 : }
471 :
472 3 : return HCCL_SUCCESS;
473 5 : }
474 :
475 6 : void RankInfoDetectClient::HostListenPortDetect(NewRankInfo &rankInfo)
476 : {
477 6 : const std::string &topoPath = CommunicatorImpl::GetTopoFilePath();
478 6 : PhyTopoBuilder::GetInstance().Build(topoPath);
479 6 : auto devLogicId = HrtGetDevice();
480 6 : u32 devPhyId = rankInfo.deviceId;
481 12 : for (auto &rankLevelInfo : rankInfo.rankLevelInfos) {
482 7 : shared_ptr<Graph<PhyTopo::Node, PhyTopo::Link>> graph = PhyTopo::GetInstance()->GetTopoGraph(rankLevelInfo.netLayer);
483 7 : if (graph == nullptr) {
484 9 : HCCL_DEBUG("[RankInfoDetectClient::%s]Can't find the layout %u Graph!", __func__, rankLevelInfo.netLayer);
485 3 : continue;
486 3 : }
487 4 : std::vector<std::shared_ptr<PhyTopo::Link>> links = graph->GetEdges(rankInfo.localId);
488 7 : for (auto &link : links) {
489 4 : if (link->GetSourceIFace()->GetPos() != AddrPosition::HOST) {
490 2 : continue;
491 : }
492 2 : const std::set<LinkProtocol> &protocols = link->GetLinkProtocols();
493 3 : for (auto &protocol : protocols) {
494 2 : LinkProtoType protoType = LinkProtocol2LinkProtoType(protocol);
495 2 : if (protoType != LinkProtoType::RDMA || rankLevelInfo.rankAddrs.empty()) {
496 1 : continue;
497 : }
498 3 : HCCL_DEBUG("[SocketManager::%s] find the host rdma link %s", __func__, link->Describe().c_str());
499 1 : const IpAddress& hostIp = rankLevelInfo.rankAddrs[0].addr;
500 1 : uint32_t hostPort = 0;
501 1 : SetupHostListenPort(devLogicId, devPhyId, hostIp, hostPort);
502 1 : rankInfo.hostPort = hostPort;
503 1 : return;
504 : }
505 2 : }
506 8 : }
507 6 : }
508 :
509 1 : void RankInfoDetectClient::SetupHostListenPort(u32 devLogicId, u32 devPhyId, const IpAddress &hostIp, uint32_t &hostPort)
510 : {
511 1 : std::lock_guard<std::mutex> lock(hostSocketLock_);
512 1 : u32 listenPort = HCCL_INVALID_PORT;
513 1 : auto portRange = EnvConfig::GetInstance().GetHostNicConfig().GetHostSocketPortRange();
514 1 : u32 basePort = EnvConfig::GetInstance().GetHostNicConfig().GetIfBasePort();
515 1 : if (portRange.empty() && basePort != HCCL_INVALID_PORT) {
516 1 : listenPort = basePort + devPhyId;
517 3 : HCCL_INFO("[RankInfoDetectClient::%s] BasePort is configured, listenPort[%u].", __func__, listenPort);
518 1 : hostPort = listenPort;
519 1 : return;
520 : }
521 :
522 0 : if (portRange.empty()) {
523 0 : constexpr u32 HOST_CONTROL_BASE_PORT = 60000; // 控制面起始port
524 0 : HCCL_INFO("[RankInfoDetectClient::%s] No port configuration, using default port range[%u, %u]", __func__,
525 : HOST_CONTROL_BASE_PORT, HOST_CONTROL_BASE_PORT + HOST_CONTROL_PORT_COUNT);
526 0 : SocketPortRange defaultRange = {HOST_CONTROL_BASE_PORT, HOST_CONTROL_BASE_PORT + HOST_CONTROL_PORT_COUNT};
527 0 : portRange.push_back(defaultRange);
528 : }
529 :
530 0 : SocketHandle hostSocketHandle = HostSocketHandleManager::GetInstance().Create(devPhyId, hostIp);
531 0 : hostSocket_ = std::make_shared<Socket>(hostSocketHandle, hostIp, HCCL_INVALID_PORT, hostIp,
532 0 : "hostport_preempt", SocketRole::SERVER, NicType::HOST_NIC_TYPE);
533 0 : PreemptPortManager::GetInstance(devLogicId).ListenPreempt(hostSocket_, portRange, listenPort);
534 0 : HCCL_INFO("[RankInfoDetectClient::%s] preempt hostPort[%u] success.", __func__, listenPort);
535 0 : hostPort = listenPort;
536 2 : }
537 :
538 23 : void RankInfoDetectClient::SocketTearDown(u32 devPhyId)
539 : {
540 23 : std::lock_guard<std::mutex> lock(hostSocketLock_);
541 23 : if (hostSocket_ == nullptr) {
542 23 : return;
543 : }
544 0 : const IpAddress& hostIp = hostSocket_->GetLocalIp();
545 0 : auto devLogicId = HrtGetDevice();
546 0 : if (EnvConfig::GetInstance().GetHostNicConfig().GetHostSocketPortRange().size() > 0 ||
547 0 : EnvConfig::GetInstance().GetHostNicConfig().GetIfBasePort() == HCCL_INVALID_PORT) {
548 : // 若开启抢占监听端口
549 0 : PreemptPortManager::GetInstance(devLogicId).Release(hostSocket_);
550 : }
551 0 : hostSocket_ = nullptr;
552 0 : HostSocketHandleManager::GetInstance().Destroy(devPhyId, hostIp);
553 23 : }
554 :
555 21 : void RankInfoDetectClient::TearDown()
556 : {
557 63 : HCCL_INFO("[RankInfoDetectClient::%s] start.", __func__);
558 21 : SocketTearDown(devPhyId_);
559 :
560 : // close socket
561 21 : clientSocket_->Close();
562 :
563 : // deinit handle
564 21 : HostSocketHandleManager::GetInstance().Destroy(devPhyId_, clientSocket_->GetLocalIp());
565 :
566 : // deinit ra in detach thread to avoid block main thread
567 21 : s32 deviceLogicId = HrtGetDevice();
568 21 : std::thread{[deviceLogicId](){
569 21 : EXCEPTION_CATCH(HccpPeerManager::GetInstance().DeInit(deviceLogicId),
570 : HCCL_ERROR("[RankInfoDetectClient::TearDown] DeInit exception"));
571 42 : }}.detach();
572 :
573 63 : HCCL_INFO("[RankInfoDetectClient::%s] end.", __func__);
574 21 : }
575 :
576 22 : RankInfoDetectClient::~RankInfoDetectClient()
577 : {
578 22 : DECTOR_TRY_CATCH("RankInfoDetectClient", TearDown());
579 22 : }
580 :
581 : }
|
