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