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 "topoinfo_exchange_agent.h"
12 : #include <iostream>
13 : #include <sstream>
14 : #include <cstring>
15 : #include "externalinput_pub.h"
16 : #include "adapter_error_manager_pub.h"
17 : #include "config.h"
18 : #include "sal_pub.h"
19 : #include "device_capacity.h"
20 :
21 : namespace hccl {
22 : constexpr s32 DEVICE_LOGIC_ID_LENGTH = 4;
23 : constexpr u32 AGENT_MAX_RETRY_TIME = 3;
24 :
25 21 : TopoInfoExchangeAgent::TopoInfoExchangeAgent(HcclIpAddress &serverIp, u32 serverPort, std::string identifier,
26 21 : HcclNetDevCtx netDevCtx, HcclBasicRankInfo localRankInfo)
27 21 : : serverIP_(serverIp),
28 21 : serverPort_(serverPort),
29 21 : identifier_(identifier),
30 21 : localRankInfo_(localRankInfo),
31 21 : clusterTopoInfo_(),
32 21 : netDevCtx_(netDevCtx),
33 42 : isRetry_(GetExternalInputInterSuperPodRetryEnable())
34 21 : {}
35 :
36 0 : TopoInfoExchangeAgent::TopoInfoExchangeAgent(HcclIpAddress &serverIp, u32 serverPort, std::string identifier,
37 0 : HcclNetDevCtx netDevCtx, HcclBasicRankInfo localRankInfo, u32 connSize, u32 connRank)
38 0 : : serverIP_(serverIp),
39 0 : serverPort_(serverPort),
40 0 : identifier_(identifier),
41 0 : localRankInfo_(localRankInfo),
42 0 : clusterTopoInfo_(),
43 0 : netDevCtx_(netDevCtx),
44 0 : connSize_(connSize),
45 0 : connRank_(connRank),
46 0 : isRetry_(GetExternalInputInterSuperPodRetryEnable())
47 0 : {}
48 :
49 0 : TopoInfoExchangeAgent::TopoInfoExchangeAgent(HcclIpAddress &serverIp, u32 serverPort, std::string identifier,
50 0 : HcclNetDevCtx netDevCtx, HcclBasicRankInfo localRankInfo, HcclRankHandle rankInfo)
51 0 : : serverIP_(serverIp),
52 0 : serverPort_(serverPort),
53 0 : identifier_(identifier),
54 0 : localRankInfo_(localRankInfo),
55 0 : localRankHandle_(rankInfo),
56 0 : clusterTopoInfo_(),
57 0 : netDevCtx_(netDevCtx),
58 0 : isRetry_(GetExternalInputInterSuperPodRetryEnable())
59 0 : {}
60 :
61 21 : TopoInfoExchangeAgent::~TopoInfoExchangeAgent()
62 : {
63 21 : Teardown();
64 21 : }
65 :
66 0 : HcclResult TopoInfoExchangeAgent::SetIsInterSuperPodRetryEnable(bool isInterSuperPodRetryEnable)
67 : {
68 0 : isRetry_ = isInterSuperPodRetryEnable;
69 0 : return HCCL_SUCCESS;
70 : }
71 :
72 0 : HcclResult TopoInfoExchangeAgent::Setup()
73 : {
74 0 : connSize_ = localRankInfo_.rankSize;
75 0 : connRank_ = localRankInfo_.rank;
76 : //填充要发送的localRankHandle的值
77 0 : localRankHandle_.rankId = localRankInfo_.rank;
78 0 : HcclResult ret = ConnectWithRetry(serverIP_, serverPort_, socket_);
79 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[TopoInfoExchangeAgent][Setup]TopoExchangeAgent: "\
80 : "connect server[%s : %u] failed", serverIP_.GetReadableAddress(), serverPort_), ret);
81 0 : HCCL_INFO("TopoExchangeAgent: client connect with server ip[%s] port[%u] success.",
82 : serverIP_.GetReadableAddress(), serverPort_);
83 :
84 0 : if (!isByMasterInfo_ && localRankInfo_.rankSize > TOPO_HIERARCHICAL_ENABLE_THRESHOLD) {
85 0 : ret = socket_->Send(&localRankHandle_, sizeof(localRankHandle_));
86 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
87 : HCCL_ERROR("[SendRankHandle]errNo[0x%016llx] rankID[%s] send localRankHandle to remote by"\
88 : "client fdHandle failed, ret[%u]", HCCL_ERROR_CODE(HCCL_E_TCP_TRANSFER), localRankInfo_.rank, ret), ret);
89 :
90 0 : CHK_RET(RecvGrpLeaderInfo(socket_, grpLeaderInfo_));
91 0 : u32 grpIndex = localRankInfo_.rank / TOPO_MAX_GROUP_SIZE;
92 0 : grpLeader_ = grpLeaderInfo_.GroupLeaderList[grpIndex];
93 0 : } else {
94 0 : CHK_RET(DetectClusterTopoInfo(socket_, clusterTopoInfo_));
95 0 : ret = VerifyClusterInfo(clusterTopoInfo_);
96 0 : if (ret != HCCL_SUCCESS) {
97 0 : auto current = g_broadcastStage.load(std::memory_order_acquire);
98 0 : if (current == BroadcastStage::Started) {
99 0 : std::unique_lock<std::mutex> lock(g_broadcast_stage_mutex);
100 0 : std::chrono::seconds timeout(MAX_WAIT_BROADCAST_SECONDS);
101 0 : g_broadcast_stage_cv.wait_for(lock, timeout, [] {
102 0 : return g_broadcastStage.load(std::memory_order_relaxed) == BroadcastStage::Completed;
103 : });
104 0 : }
105 0 : HCCL_ERROR("[TopoInfoExchangeAgent][Setup]VerifyCluseterInfo failed, g_broadcastStage[%d]", g_broadcastStage.load());
106 : }
107 :
108 0 : return ret;
109 : }
110 :
111 0 : return HCCL_SUCCESS;
112 : }
113 :
114 0 : HcclResult TopoInfoExchangeAgent::SetupRank(std::shared_ptr<HcclSocket> socket)
115 : {
116 0 : CHK_RET(RecvGrpLeaderInfo(socket, grpLeaderInfo_));
117 0 : u32 grpIndex = localRankInfo_.rank / TOPO_MAX_GROUP_SIZE;
118 0 : grpLeader_ = grpLeaderInfo_.GroupLeaderList[grpIndex];
119 0 : return HCCL_SUCCESS;
120 : }
121 :
122 0 : HcclResult TopoInfoExchangeAgent::SetupMember()
123 : {
124 0 : HcclResult ret = Connect(serverIP_, serverPort_, socket_);
125 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[TopoInfoExchangeAgent][Setup]SetupGroupMember: "\
126 : "connect server[%s : %u] failed", serverIP_.GetReadableAddress(), serverPort_), ret);
127 0 : HCCL_INFO("SetupGroupMember: client connect with server ip[%s] port[%u] success.",
128 : serverIP_.GetReadableAddress(), serverPort_);
129 :
130 0 : CHK_RET(DetectClusterTopoInfo(socket_, clusterTopoInfo_));
131 :
132 0 : CHK_RET(VerifyClusterInfo(clusterTopoInfo_));
133 :
134 0 : return HCCL_SUCCESS;
135 : }
136 :
137 21 : HcclResult TopoInfoExchangeAgent::Teardown()
138 : {
139 21 : CHK_RET(Disconnect(socket_));
140 21 : return HCCL_SUCCESS;
141 : }
142 :
143 0 : HcclResult TopoInfoExchangeAgent::GetConnection(std::shared_ptr<HcclSocket> &socket)
144 : {
145 0 : socket = socket_;
146 0 : return HCCL_SUCCESS;
147 : }
148 :
149 0 : HcclResult TopoInfoExchangeAgent::GetGroupLeader(HcclRankHandle &rankHandle)
150 : {
151 0 : rankHandle = grpLeader_;
152 0 : return HCCL_SUCCESS;
153 : }
154 :
155 0 : HcclResult TopoInfoExchangeAgent::SetupByMasterInfo()
156 : {
157 0 : isByMasterInfo_ = true;
158 0 : CHK_RET(Setup());
159 0 : return HCCL_SUCCESS;
160 : }
161 :
162 0 : HcclResult TopoInfoExchangeAgent::DetectClusterTopoInfo(
163 : std::shared_ptr<HcclSocket> socket, RankTable_t &clusterTopoInfo)
164 : {
165 0 : RankTable_t localBasicInfo;
166 0 : CHK_RET(ConstructRankTableMsg(localBasicInfo));
167 0 : CHK_RET(SendClusterInfo(socket, localBasicInfo));
168 0 : HCCL_INFO("topo exchange client send rank basic info success.");
169 :
170 0 : CHK_RET(RecvClusterInfo(socket, clusterTopoInfo));
171 0 : HCCL_INFO("topo exchange client get rank basic info success.");
172 :
173 : // 按照rankId排序
174 0 : std::vector<RankInfo_t> &rankList = clusterTopoInfo_.rankList;
175 0 : sort(rankList.begin(), rankList.end(), [](const RankInfo_t &a, const RankInfo_t &b) {
176 0 : return a.rankId < b.rankId; });
177 :
178 0 : CHK_RET(SetServerIdx(clusterTopoInfo));
179 0 : CHK_RET(GroupSuperPodsByRankContinuity(clusterTopoInfo));
180 0 : CHK_RET(SetSuperPodIdx(clusterTopoInfo));
181 0 : return HCCL_SUCCESS;
182 0 : }
183 :
184 0 : HcclResult TopoInfoExchangeAgent::GroupSuperPodsByRankContinuity(RankTable_t &clusterInfo) const
185 : {
186 : // 按照superPodId将节点分组，相同superPodId在一个组
187 : // clusterInfo已经按照rankId排好序，按顺序插入到新的subRankTable中，不需要再排序
188 0 : std::map<std::string, std::vector<RankInfo_t*>> podGroupClusters;
189 0 : for (auto& rankInfo : clusterInfo.rankList) {
190 0 : rankInfo.originalSuperPodId = rankInfo.superPodId; // 把用户配置的原始superPodId先保存下来
191 0 : podGroupClusters[rankInfo.superPodId].emplace_back(&rankInfo);
192 : }
193 0 : std::set<std::string> superPodIdSet;
194 0 : std::map<std::string, std::pair<u32, u32>> superPodIdRanges; // 记录每个逻辑超节点的rank id范围
195 0 : for (auto& subCluster : podGroupClusters) {
196 0 : auto& subClusterInfo = subCluster.second;
197 0 : if (subClusterInfo.size() <= 1) {
198 0 : continue;
199 : }
200 0 : u32 groupId = 0;
201 0 : superPodIdSet.insert(subCluster.first);
202 0 : RankInfo_t preRank = *(subClusterInfo[0]);
203 0 : superPodIdRanges[preRank.superPodId] = {preRank.rankId, preRank.rankId}; // 初始化范围
204 0 : for (u32 i = 1; i < subClusterInfo.size(); ++i) {
205 0 : RankInfo_t& curRank = *(subClusterInfo[i]);
206 : // 当前的curRank和上一个preRank的rankId不连续，分配新的逻辑超节点ID
207 0 : if (curRank.rankId != preRank.rankId + 1) {
208 0 : std::string newSuperPodId = curRank.originalSuperPodId + "_HCCLSPLIT_" + std::to_string(groupId);
209 0 : curRank.superPodId = newSuperPodId;
210 0 : groupId++;
211 0 : superPodIdRanges[curRank.superPodId] = {curRank.rankId, curRank.rankId}; // 初始化新的范围
212 0 : } else {
213 : // 同一个sub通信域两个rank原始逻辑超节点是一致的
214 : // rankId连续 上一个rank的superPodId可能已经重新分配，需要更新当前superPodId为上一个rank的
215 0 : curRank.superPodId = preRank.superPodId;
216 0 : superPodIdRanges[curRank.superPodId].second = curRank.rankId; // 更新最大rank id
217 : }
218 0 : superPodIdSet.insert(curRank.superPodId);
219 0 : preRank = curRank;
220 : }
221 0 : }
222 : // 打印每个逻辑超节点的rank id范围，只打印包含_HCCLSPLIT_的逻辑超节点
223 0 : for (const auto& entry : superPodIdRanges) {
224 0 : auto superPodId = entry.first;
225 0 : if (superPodId.find("_HCCLSPLIT_") != std::string::npos) {
226 0 : auto range = entry.second;
227 0 : HCCL_RUN_INFO("[TopoInfoExchangeAgent][%s]Split superPod, ID[%s], rank range[%u, %u]", __func__,
228 : superPodId.c_str(), range.first, range.second);
229 : }
230 0 : }
231 0 : clusterInfo.superPodNum = superPodIdSet.size();
232 0 : return HCCL_SUCCESS;
233 0 : }
234 :
235 0 : HcclResult TopoInfoExchangeAgent::SetServerIdx(RankTable_t &clusterInfo) const
236 : {
237 : struct ServerSortInfo {
238 : u32 serverPosition;
239 : u32 selectedRankId;
240 : };
241 0 : std::vector<ServerSortInfo> serverSortInfoVec;
242 0 : for (u32 i = 0; i < clusterInfo.serverList.size(); i++) {
243 0 : for (u32 j = 0; j < clusterInfo.rankList.size(); j++) {
244 0 : if (clusterInfo.rankList[j].serverId == clusterInfo.serverList[i].serverId) {
245 : // 每个server的rankid都是连续的，只需要取每个server里任意一个rankid进行排序
246 : ServerSortInfo serverSortInfo;
247 0 : serverSortInfo.serverPosition = i;
248 0 : serverSortInfo.selectedRankId = clusterInfo.rankList[j].rankId;
249 0 : serverSortInfoVec.push_back(serverSortInfo);
250 0 : break;
251 : }
252 : }
253 : }
254 0 : sort(serverSortInfoVec.begin(), serverSortInfoVec.end(), [](const ServerSortInfo &a,
255 0 : const ServerSortInfo &b) { return a.selectedRankId < b.selectedRankId; });
256 : // 遍历ranklist，根据serverid获取serveridx
257 0 : for (u32 serverIdx = 0; serverIdx < serverSortInfoVec.size(); serverIdx++) {
258 0 : for (u32 j = 0; j < clusterInfo.rankList.size(); j++) {
259 0 : if (clusterInfo.rankList[j].serverId ==
260 0 : clusterInfo.serverList[serverSortInfoVec[serverIdx].serverPosition].serverId) {
261 0 : clusterInfo.rankList[j].serverIdx = serverIdx;
262 : }
263 : }
264 : }
265 0 : return HCCL_SUCCESS;
266 0 : }
267 :
268 1 : HcclResult TopoInfoExchangeAgent::SetSuperPodIdx(RankTable_t &clusterInfo) const
269 : {
270 1 : std::map<std::string, u32> spodIdToIdx;
271 1 : bool isDiffDeviceType = false;
272 1 : DevType standardDevType = DevType::DEV_TYPE_NOSOC;
273 1 : if (clusterInfo.rankList.size() > 0) {
274 1 : standardDevType = clusterInfo.rankList[0].deviceInfo.deviceType;
275 : }
276 4 : for (u32 i = 0; i < clusterInfo.rankList.size(); ++i) {
277 3 : RankInfo_t& rankInfo = clusterInfo.rankList[i];
278 3 : if (rankInfo.deviceInfo.deviceType != standardDevType) {
279 0 : isDiffDeviceType = true;
280 : }
281 :
282 3 : if (isDiffDeviceType) {
283 0 : rankInfo.superPodIdx = spodIdToIdx.size();
284 3 : } else if (spodIdToIdx.find(rankInfo.superPodId) == spodIdToIdx.end()) {
285 2 : rankInfo.superPodIdx = spodIdToIdx.size();
286 2 : spodIdToIdx.insert({rankInfo.superPodId, rankInfo.superPodIdx});
287 1 : } else if (spodIdToIdx[rankInfo.superPodId] + 1 == spodIdToIdx.size()) {
288 0 : rankInfo.superPodIdx = spodIdToIdx[rankInfo.superPodId];
289 : } else {
290 1 : u32 preIndex = (i > 0) ? i - 1 : i;
291 1 : RankInfo_t& preRankInfo = clusterInfo.rankList[preIndex];
292 1 : u32 index = 0;
293 1 : for (; index < preIndex; index++) {
294 1 : RankInfo_t& tmpRankInfo = clusterInfo.rankList[index];
295 1 : if(tmpRankInfo.superPodId == rankInfo.superPodId) {
296 1 : break;
297 : }
298 : }
299 : // 超节点内rank id不连续
300 1 : HCCL_RUN_WARNING("rank in superPodId is not continuous, pre: rank[%u] superPodId[%s], "\
301 : "cur: rank[%u] superPodId[%s], ", preRankInfo.rankId, preRankInfo.superPodId.c_str(),
302 : rankInfo.rankId, rankInfo.superPodId.c_str());
303 1 : rankInfo.superPodIdx = spodIdToIdx[rankInfo.superPodId];
304 : }
305 3 : HCCL_INFO("SetSuperPodIdx rankList[%u]: rankId[%u], superPodId[%s], superPodIdx[%u], sdid[%u]",
306 : i, rankInfo.rankId, rankInfo.superPodId.c_str(), rankInfo.superPodIdx, rankInfo.superDeviceId);
307 : }
308 1 : return HCCL_SUCCESS;
309 1 : }
310 :
311 0 : HcclResult TopoInfoExchangeAgent::GetClusterTopoInfo(RankTable_t &clusterInfo)
312 : {
313 0 : clusterInfo.nicDeploy = clusterTopoInfo_.nicDeploy;
314 0 : clusterInfo.deviceNum = clusterTopoInfo_.deviceNum;
315 0 : clusterInfo.serverNum = clusterTopoInfo_.serverNum;
316 0 : clusterInfo.superPodNum = clusterTopoInfo_.superPodNum;
317 0 : clusterInfo.rankNum = clusterTopoInfo_.rankNum;
318 0 : clusterInfo.rankList = clusterTopoInfo_.rankList;
319 0 : clusterInfo.serverList = clusterTopoInfo_.serverList;
320 :
321 0 : return HCCL_SUCCESS;
322 : }
323 0 : HcclResult TopoInfoExchangeAgent::GetIdentifier(u32 &identify)
324 : {
325 0 : identify = identifierNum_;
326 0 : return HCCL_SUCCESS;
327 : }
328 0 : HcclResult TopoInfoExchangeAgent::Connect(HcclIpAddress &serverIp, u32 port,
329 : std::shared_ptr<HcclSocket> &socket)
330 : {
331 0 : std::string tag = TOPO_DETECT_TAG + "_" + identifier_ + "_" + std::to_string(port);
332 0 : EXCEPTION_CATCH((socket = std::make_shared<HcclSocket>(tag,
333 : netDevCtx_, serverIp, port, HcclSocketRole::SOCKET_ROLE_CLIENT)), return HCCL_E_PTR);
334 0 : CHK_SMART_PTR_NULL(socket);
335 0 : CHK_RET(socket->Init());
336 0 : CHK_RET(socket->Connect());
337 :
338 0 : return GetConnection(serverIp, port, socket);
339 0 : }
340 :
341 0 : HcclResult TopoInfoExchangeAgent::ConnectWithRetry(HcclIpAddress &serverIp, u32 port,
342 : std::shared_ptr<HcclSocket> &socket)
343 : {
344 0 : u32 retryTime = 1;
345 0 : HcclResult ret = HCCL_SUCCESS;
346 0 : while (retryTime <= AGENT_MAX_RETRY_TIME) {
347 0 : std::string tag = TOPO_DETECT_TAG + "_" + identifier_ + "_" + std::to_string(port);
348 0 : EXCEPTION_CATCH((socket = std::make_shared<HcclSocket>(tag,
349 : netDevCtx_, serverIp, port, HcclSocketRole::SOCKET_ROLE_CLIENT)), return HCCL_E_PTR);
350 0 : CHK_SMART_PTR_NULL(socket);
351 0 : CHK_RET(socket->Init());
352 0 : CHK_RET(socket->Connect());
353 :
354 0 : CHK_RET(GetConnection(serverIp, port, socket));
355 :
356 0 : ret = TryRecvFromServer(socket, retryTime);
357 0 : if (ret == HCCL_SUCCESS) {
358 0 : break;
359 : } else {
360 0 : retryTime++;
361 : }
362 0 : }
363 0 : return ret;
364 : }
365 :
366 3 : HcclResult TopoInfoExchangeAgent::TryRecvFromServer(std::shared_ptr<HcclSocket> &socket, u32 retryTime)
367 : {
368 : // client端获取socket之后尝试从server接收数据，若在一定时间内没有接收到，则重新发起建链请求
369 3 : u32 timeout = GetExternalInputHcclLinkTimeOut() / AGENT_MAX_RETRY_TIME;
370 3 : char recvMsgBuf[sizeof(TOPO_EXCHANGE_CHECK_MESSAGE)] = {0};
371 3 : auto ret = HCCL_SUCCESS;
372 3 : if (retryTime == AGENT_MAX_RETRY_TIME) {
373 3 : ret = socket->Recv(recvMsgBuf, sizeof(TOPO_EXCHANGE_CHECK_MESSAGE), timeout);
374 : } else {
375 : // 重试时打印RUN_WARN日志
376 0 : SetErrToWarnSwitch(true);
377 0 : ret = socket->Recv(recvMsgBuf, sizeof(TOPO_EXCHANGE_CHECK_MESSAGE), timeout);
378 0 : SetErrToWarnSwitch(false);
379 : }
380 :
381 3 : if (ret == HCCL_SUCCESS) {
382 2 : HCCL_RUN_INFO("[%s]recvMes %s", __func__, recvMsgBuf);
383 : // 校验收到的是否正确，server端使用的是固定消息
384 2 : if (strncmp(recvMsgBuf, TOPO_EXCHANGE_CHECK_MESSAGE, sizeof(TOPO_EXCHANGE_CHECK_MESSAGE)) != 0) {
385 1 : HCCL_ERROR("[%s]recv message check failed, expect [%s], but recv [%s]",
386 : __func__, TOPO_EXCHANGE_CHECK_MESSAGE, recvMsgBuf);
387 1 : return HCCL_E_INTERNAL;
388 : }
389 1 : } else if (retryTime < AGENT_MAX_RETRY_TIME) {
390 0 : HCCL_RUN_WARNING("[%s]client recv from server failed, will try to connect with server again.", __func__);
391 : } else {
392 1 : HCCL_ERROR("[%s]failed to recv messages from server with %u times", __func__, AGENT_MAX_RETRY_TIME);
393 : }
394 :
395 2 : return ret;
396 : }
397 :
398 0 : void TopoInfoExchangeAgent::PrintSocketTimeoutReasons(HcclIpAddress &serverIp, u32 port,
399 : std::shared_ptr<HcclSocket> &socket)
400 : {
401 0 : HCCL_ERROR("current rank connect to server timeout, maybe due to following reasons:");
402 0 : HCCL_ERROR("1. local host ip is [%s], server host ip and port is [%s:%u], Please check the network connectivity. "
403 : "If it is not connected, modify the network configuration or use HCCL_SOCKET_IFNAME and HCCL_IF_BASE_PORT to specify ifname and server port.",
404 : socket->GetLocalIp().GetReadableIP(), serverIp.GetReadableIP(), port);
405 0 : HCCL_ERROR("2. Check whether any other exceptions have occurred on server[%s] or "
406 : "whether the time difference between the execution of hcom on ranks exceeds the timeout threshold.",
407 : serverIp.GetReadableIP());
408 0 : }
409 :
410 0 : HcclResult TopoInfoExchangeAgent::GetConnection(HcclIpAddress &serverIp, u32 port,
411 : std::shared_ptr<HcclSocket> &socket)
412 : {
413 0 : auto startTime = std::chrono::steady_clock::now();
414 0 : auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
415 : while (true) {
416 0 : std::string errormessage = "1. The current node " + std::string(serverIp.GetReadableIP()) +
417 0 : " is disconnected from the host of the root node " + std::string(localRankHandle_.ip) + ". "\
418 0 : "2. the timeout set by the HCCL_CONNECT_TIMEOUT environment variable is too short";
419 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
420 0 : RPT_INPUT_ERR(true, "EI0015", std::vector<std::string>({"error_reason"}), \
421 : std::vector<std::string>({errormessage}));
422 0 : HCCL_ERROR("[%s][%s] topo exchange agent get socket timeout! timeout[%lld s]",
423 : LOG_KEYWORDS_INIT_GROUP.c_str(), LOG_KEYWORDS_RANKTABLE_DETECT.c_str(), timeout);
424 0 : PrintSocketTimeoutReasons(serverIp, port, socket);
425 0 : sleep(WAIT_ERROR_BROADCAST_TIME);
426 0 : return HCCL_E_TIMEOUT;
427 : }
428 0 : HcclSocketStatus status = socket->GetStatus();
429 0 : if (status == HcclSocketStatus::SOCKET_CONNECTING) {
430 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
431 0 : } else if (status != HcclSocketStatus::SOCKET_OK) {
432 0 : HCCL_ERROR("[Get][Connection]server: get socket failed ret[%d]", status);
433 0 : return HCCL_E_TCP_CONNECT;
434 : } else {
435 0 : HCCL_INFO("TopoInfoExchangeAgent get socket success.");
436 0 : std::string agentID;
437 0 : if (isByMasterInfo_) {
438 0 : agentID = localRankInfo_.superPodId + "/";
439 0 : GenerateAgentID(localRankInfo_, agentID);
440 : } else {
441 0 : std::string rankID = std::to_string(connRank_);
442 0 : agentID = std::string(16 - rankID.length(), '0') + rankID; // agent id为rank id，16位，左对齐补零
443 0 : }
444 0 : char agentBuf[MAX_AGENT_BUF_SIZE] = {0};
445 0 : s32 sRet = memcpy_s(agentBuf, sizeof(agentBuf), agentID.c_str(), agentID.size());
446 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("memcpy_s failed, errorno[%d]", sRet), HCCL_E_MEMORY);
447 0 : HcclResult ret = socket->Send(&agentBuf, sizeof(agentBuf));
448 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
449 : HCCL_ERROR("[Get][Connection]errNo[0x%016llx] agentID[%s] send local rank id to remote "\
450 : "by client fdHandle failed, ret[%u]", HCCL_ERROR_CODE(HCCL_E_TCP_TRANSFER), agentBuf, ret), ret);
451 0 : ret = socket->Send(&connSize_, sizeof(connSize_));
452 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
453 : HCCL_ERROR("[Get][Connection]errNo[0x%016llx] rank[%u] send local rank num[%u] to "\
454 : "remote by client fdHandle failed, ret[%u]", HCCL_ERROR_CODE(HCCL_E_TCP_TRANSFER),
455 : localRankInfo_.rank, localRankInfo_.rankSize, ret), ret);
456 0 : HCCL_INFO("local rank[%u] get socket connection with server[%s] port[%u] success.",
457 : localRankInfo_.rank, serverIp.GetReadableAddress(), port);
458 0 : break;
459 0 : }
460 0 : }
461 0 : return HCCL_SUCCESS;
462 0 : }
463 :
464 0 : std::string TopoInfoExchangeAgent::Dec2Hex(s32 i, u32 width)
465 : {
466 0 : std::string temp;
467 0 : std::stringstream ss;
468 0 : ss << std::hex << i;
469 0 : ss >> temp;
470 0 : if (width > temp.size()) {
471 0 : return std::string((width - temp.size()), '0') + temp;
472 : } else {
473 0 : HCCL_WARNING("Dec2Hex: length[%u] is over width[%u]", temp.size(), width);
474 : }
475 0 : return temp;
476 0 : }
477 :
478 0 : void TopoInfoExchangeAgent::GenerateAgentID(HcclBasicRankInfo &localRankInfo, std::string &agentID)
479 : {
480 0 : struct in_addr addr = localRankInfo.hostIP.GetBinaryAddress().addr;
481 0 : struct in6_addr addr6 = localRankInfo.hostIP.GetBinaryAddress().addr6;
482 0 : if (localRankInfo.hostIP.IsIPv6()) {
483 0 : for (size_t i = 0; i < sizeof(addr6.s6_addr); i++) {
484 0 : agentID += Dec2Hex(addr6.s6_addr[i], 2); // 转换为2位十六进制数据，左对齐补零
485 : }
486 : } else {
487 0 : for (size_t i = 0; i < sizeof(addr.s_addr) / sizeof(u8); i++) {
488 0 : agentID += Dec2Hex(*(reinterpret_cast<u8 *>(&addr.s_addr) + i), 2); // 转换为2位十六进制数据，左对齐补零
489 : }
490 : }
491 0 : agentID.append("/");
492 0 : std::string devID = std::to_string(localRankInfo.deviceLogicID);
493 0 : CHK_PRT_RET(devID.size() > DEVICE_LOGIC_ID_LENGTH, HCCL_ERROR("deviceLogicID[%s] is invalid", devID.c_str()),);
494 : // device id转换为4位十进制数字，左对齐补零
495 0 : agentID.append(std::string((DEVICE_LOGIC_ID_LENGTH - devID.size()), '0') + devID);
496 0 : HCCL_INFO("GenerateAgentID agentID[%s]", agentID.c_str());
497 0 : return;
498 0 : }
499 :
500 21 : HcclResult TopoInfoExchangeAgent::Disconnect(std::shared_ptr<HcclSocket> &socket)
501 : {
502 21 : CHK_RET(DisconnectSocket(socket));
503 21 : socket = nullptr;
504 :
505 21 : return HCCL_SUCCESS;
506 : }
507 :
508 0 : HcclResult TopoInfoExchangeAgent::RecvGrpLeaderInfo(std::shared_ptr<HcclSocket> socket, GroupLeader_t &leaderInfo)
509 : {
510 : //每次获取之前先清空 保证填充之后的数据是最新的
511 0 : leaderInfo.grpLeaderNum = 0;
512 0 : leaderInfo.GroupLeaderList.clear();
513 0 : CHK_RET(RecvGrpLeaderInfoMsg(socket, leaderInfo));
514 0 : return HCCL_SUCCESS;
515 : }
516 :
517 0 : HcclResult TopoInfoExchangeAgent::SendGroupLeaderPortInfo(std::shared_ptr<HcclSocket> socket, HcclRankHandle &rankHandle)
518 : {
519 0 : CHK_RET(GetConnection(socket));
520 0 : HcclResult ret = socket->Send(&rankHandle, sizeof(rankHandle));
521 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[TopoInfoExchangeAgent][SendGroupLeaderPortInfo]errNo[0x%016llx] " \
522 : "send grpleader port info fail", HCCL_ERROR_CODE(ret)), ret);
523 0 : return HCCL_SUCCESS;
524 : }
525 :
526 0 : HcclResult TopoInfoExchangeAgent::ConstructRankTableMsg(RankTable_t &clusterInfo)
527 : {
528 0 : RankInfo_t myRankInfo;
529 0 : myRankInfo.rankId = localRankInfo_.rank;
530 0 : myRankInfo.hostIp = localRankInfo_.hostIP;
531 0 : myRankInfo.hostPort = localRankInfo_.hostPort;
532 0 : myRankInfo.deviceInfo.devicePhyId = localRankInfo_.devicePhysicID;
533 0 : myRankInfo.deviceInfo.deviceIp = localRankInfo_.deviceIP;
534 0 : myRankInfo.deviceInfo.deviceType = localRankInfo_.deviceType;
535 0 : myRankInfo.deviceInfo.backupDeviceIp = localRankInfo_.backupDeviceIP;
536 0 : myRankInfo.deviceInfo.port = localRankInfo_.deviceNicPort;
537 0 : myRankInfo.deviceInfo.vnicPort = localRankInfo_.deviceVnicPort;
538 0 : myRankInfo.deviceInfo.backupPort = localRankInfo_.backupDevicePort;
539 0 : myRankInfo.superPodId = localRankInfo_.superPodId;
540 0 : myRankInfo.superDeviceId = localRankInfo_.superDeviceId;
541 0 : myRankInfo.tlsStatus = localRankInfo_.tlsStatus;
542 0 : ConstructRankTableServerId(myRankInfo.serverId);
543 :
544 0 : ServerInfo_t myServerInfo;
545 0 : myServerInfo.serverId = myRankInfo.serverId;
546 :
547 0 : clusterInfo.nicDeploy = localRankInfo_.nicDeploy;
548 0 : clusterInfo.rankList.push_back(myRankInfo);
549 0 : clusterInfo.serverList.push_back(myServerInfo);
550 0 : return HCCL_SUCCESS;
551 0 : }
552 :
553 0 : void TopoInfoExchangeAgent::ConstructRankTableServerId(std::string &serverId)
554 : {
555 0 : serverId = localRankInfo_.hostIP.GetReadableIP();
556 : // 配置逻辑超节点时, serverId要根据逻辑超节点划分
557 0 : if (localRankInfo_.deviceType == DevType::DEV_TYPE_910_93 && GetExternalInputLogicSuperPodId().empty() == false) {
558 0 : serverId += "_" + GetExternalInputLogicSuperPodId();
559 : }
560 0 : HCCL_INFO("ConstructRankTableServerId serverId %s", serverId.c_str());
561 0 : }
562 :
563 0 : HcclResult TopoInfoExchangeAgent::SetTransportInfo(RankTable_t &clusterInfo)
564 : {
565 0 : CHK_PRT_RET(clusterInfo.rankList.size() <= localRankInfo_.rank, HCCL_ERROR("[Set][TransportInfo]rank list is "\
566 : "invalid. size[%zu] should be greater than myRank[%u].", clusterInfo.rankList.size(), localRankInfo_.rank),
567 : HCCL_E_INTERNAL);
568 0 : RankInfo_t& myRankInfo = clusterInfo.rankList[localRankInfo_.rank];
569 0 : TransportInfo_t transportInfo = {0};
570 :
571 0 : for (u32 index = 0; index < clusterInfo.rankList.size(); index++) {
572 0 : transportInfo.dstRankId = clusterInfo.rankList[index].rankId;
573 0 : HcclResult ret = DetectTransportType(myRankInfo, clusterInfo.rankList[index], transportInfo.transportType);
574 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
575 : HCCL_ERROR("[Set][TransportInfo]rank[%u] detect transport type failed, ret[%u]. "\
576 : "remote[%u]", localRankInfo_.rank, ret, transportInfo.dstRankId), ret);
577 0 : myRankInfo.transportInfo.push_back(transportInfo);
578 : }
579 0 : return HCCL_SUCCESS;
580 : }
581 :
582 0 : HcclResult TopoInfoExchangeAgent::DetectTransportType(const RankInfo_t& localRankInfo,
583 : const RankInfo_t& remoteRankInfo, TransportType& transportType) const
584 : {
585 0 : if (remoteRankInfo.serverId == localRankInfo.serverId) {
586 0 : transportType = TransportType::TRANS_TYPE_P2P;
587 : }
588 0 : return HCCL_SUCCESS;
589 : }
590 :
591 4 : HcclResult TopoInfoExchangeAgent::VerifyClusterInfo(RankTable_t &clusterInfo)
592 : {
593 4 : std::string errormessage;
594 :
595 4 : if (clusterInfo.rankList.size() != localRankInfo_.rankSize) {
596 2 : errormessage = "The number of ranks[" + std::to_string(localRankInfo_.rankSize) +
597 4 : "]passed by the communicator initialization interface does not match the number of ranks[" + std::to_string(clusterInfo.rankList.size()) +
598 1 : "]obtained during cluster information negotiction.";
599 1 : HCCL_ERROR("[%s][%s]%s",
600 : LOG_KEYWORDS_INIT_GROUP.c_str(),
601 : LOG_KEYWORDS_RANKTABLE_DETECT.c_str(),
602 : errormessage.c_str());
603 1 : return HCCL_E_PARA;
604 : }
605 :
606 3 : if (clusterInfo.rankNum != localRankInfo_.rankSize) {
607 2 : errormessage = "The number of ranks[" + std::to_string(localRankInfo_.rankSize) +
608 4 : "]passed by the communicator initialization interface does not match the number of ranks[" + std::to_string(clusterInfo.rankNum) +
609 1 : "] obtained during cluster information negotiction.";
610 1 : HCCL_ERROR("[%s][%s]%s",
611 : LOG_KEYWORDS_INIT_GROUP.c_str(),
612 : LOG_KEYWORDS_RANKTABLE_DETECT.c_str(),
613 : errormessage.c_str());
614 1 : return HCCL_E_PARA;
615 : }
616 :
617 2 : if (clusterInfo.serverNum != clusterInfo.serverList.size()) {
618 4 : errormessage = "server num[" + std::to_string(clusterInfo.serverNum) + "] is different with server list size[" +
619 6 : std::to_string(clusterInfo.serverList.size()) + "] in total topo rank info";
620 14 : RPT_INPUT_ERR(true, "EI0015",
621 : std::vector<std::string>({ "error_reason"}),
622 : std::vector<std::string>({ errormessage }));
623 2 : HCCL_ERROR("[%s][%s]%s",
624 : LOG_KEYWORDS_INIT_GROUP.c_str(),
625 : LOG_KEYWORDS_RANKTABLE_DETECT.c_str(),
626 : errormessage.c_str());
627 2 : return HCCL_E_PARA;
628 : }
629 :
630 0 : if (clusterInfo.nicDeploy != localRankInfo_.nicDeploy) {
631 0 : errormessage = "nicDeploy[" + std::to_string(static_cast<int>(localRankInfo_.nicDeploy)) +
632 0 : "] is different with nicDeploy[" + std::to_string(static_cast<int>(clusterInfo.nicDeploy)) + "] in total topo rank info";
633 0 : RPT_INPUT_ERR(true, "EI0015",
634 : std::vector<std::string>({ "error_reason"}),
635 : std::vector<std::string>({ errormessage }));
636 0 : HCCL_ERROR("[%s][%s]%s",
637 : LOG_KEYWORDS_INIT_GROUP.c_str(),
638 : LOG_KEYWORDS_RANKTABLE_DETECT.c_str(),
639 : errormessage.c_str());
640 0 : return HCCL_E_PARA;
641 : }
642 :
643 0 : CHK_RET(VerifyClusterRankID(clusterInfo));
644 0 : if (localRankInfo_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
645 0 : CHK_RET(VerifyClusterDeviceIP(clusterInfo));
646 0 : CHK_RET(VerifyClusterBackupDeviceIP(clusterInfo));
647 : }
648 0 : std::map<std::string, std::vector<RankInfo_t>> serverMap;
649 0 : for (uint32_t i = 0; i < clusterInfo.rankList.size(); i++) {
650 0 : auto iter = serverMap.find(clusterInfo.rankList[i].serverId);
651 0 : if (iter == serverMap.end()) {
652 0 : std::vector<RankInfo_t> vec;
653 0 : vec.push_back(clusterInfo.rankList[i]);
654 0 : serverMap.insert({clusterInfo.rankList[i].serverId, vec});
655 0 : } else {
656 0 : serverMap[clusterInfo.rankList[i].serverId].push_back(clusterInfo.rankList[i]);
657 : }
658 : }
659 :
660 0 : if (clusterInfo.serverNum != serverMap.size()) {
661 0 : errormessage = "server num[" + std::to_string(clusterInfo.serverNum) +
662 0 : "] is different with server num[" +
663 0 : std::to_string(serverMap.size()) + "] in total topo rank info";
664 0 : RPT_INPUT_ERR(true,
665 : "EI0015",
666 : std::vector<std::string>({"error_reason"}),
667 : std::vector<std::string>({ errormessage }));
668 0 : HCCL_ERROR("[%s][%s]%s",
669 : LOG_KEYWORDS_INIT_GROUP.c_str(),
670 : LOG_KEYWORDS_RANKTABLE_DETECT.c_str(),
671 : errormessage.c_str());
672 0 : return HCCL_E_PARA;
673 : }
674 :
675 0 : uint32_t deviceNumInServer = 0;
676 0 : for (auto &server : serverMap) {
677 0 : CHK_PRT_RET((server.second.size() == 0),
678 : HCCL_ERROR("[%s][%s]server ip[%s] has %u device.",
679 : LOG_KEYWORDS_INIT_GROUP.c_str(),
680 : LOG_KEYWORDS_RANKTABLE_DETECT.c_str(),
681 : server.first.c_str(),
682 : server.second.size()),
683 : HCCL_E_PARA);
684 :
685 0 : if (deviceNumInServer != 0) {
686 0 : HCCL_WARNING("[%s][%s]server ip[%s] has %u devices, other server has %u.",
687 : LOG_KEYWORDS_INIT_GROUP.c_str(),
688 : LOG_KEYWORDS_RANKTABLE_DETECT.c_str(),
689 : server.first.c_str(),
690 : server.second.size(),
691 : deviceNumInServer);
692 : }
693 0 : deviceNumInServer = server.second.size();
694 0 : HcclResult ret = VerifyServerDevicePhysicID(server.second);
695 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
696 : HCCL_ERROR("[%s][%s]server id[%s] verify device physic id failed.",
697 : LOG_KEYWORDS_INIT_GROUP.c_str(),
698 : LOG_KEYWORDS_RANKTABLE_DETECT.c_str(),
699 : server.first.c_str()),
700 : HCCL_E_PARA);
701 : }
702 :
703 0 : bool useSuperPodMode = false;
704 0 : CHK_RET(IsSuperPodMode(useSuperPodMode));
705 0 : bool isSinglePodInterHccs = clusterInfo.superPodNum == 1 && GetExternalInputInterHccsDisable() == false && useSuperPodMode;
706 : // 单超节点，并且节点间走HCCS场景，不校验ip family
707 0 : if (clusterInfo.serverNum > 1 && !isSinglePodInterHccs) {
708 0 : CHK_RET(CheckRankIpFamily(clusterInfo.rankList));
709 : }
710 :
711 : // 超节点校验
712 0 : CHK_RET(VerifyClusterSuperPodInfo(clusterInfo.rankList));
713 :
714 : // TLS开关一致性校验
715 0 : CHK_RET(VerifyClusterTlsConsistency(clusterInfo));
716 0 : return HCCL_SUCCESS;
717 6 : }
718 :
719 1 : HcclResult TopoInfoExchangeAgent::VerifyClusterDeviceIP(const RankTable_t &clusterInfo)
720 : {
721 1 : if (clusterInfo.rankList.size() == 1) {
722 0 : return HCCL_SUCCESS;
723 : }
724 1 : if (clusterInfo.serverList.size() == 1) {
725 : // 单机场景对 device ip不做要求
726 0 : return HCCL_SUCCESS;
727 : }
728 1 : bool useSuperPodMode = false;
729 1 : CHK_RET(IsSuperPodMode(useSuperPodMode));
730 1 : if (clusterInfo.superPodNum == 1 && GetExternalInputInterHccsDisable() == false && useSuperPodMode) {
731 : // 单超节点，并且节点间走HCCS场景，device ip不做要求
732 0 : return HCCL_SUCCESS;
733 : }
734 1 : for (u32 i = 0; i < (clusterInfo.rankList.size() - 1); i++) {
735 1 : for (u32 j = (i + 1); j < clusterInfo.rankList.size(); j++) {
736 1 : bool isErr = HasRepeatedIP(clusterInfo.rankList[i].deviceInfo.deviceIp,
737 1 : clusterInfo.rankList[j].deviceInfo.deviceIp);
738 1 : if (isErr) {
739 2 : std::string errormessage = "The device IP address " + std::string(clusterInfo.rankList[i].deviceInfo.deviceIp[0].GetReadableIP()) +
740 4 : " of rank " + std::to_string(clusterInfo.rankList[i].rankId) +
741 3 : " on node " +clusterInfo.rankList[i].serverId +
742 4 : " is the same as the device IP address " + std::string(clusterInfo.rankList[j].deviceInfo.deviceIp[0].GetReadableIP()) +
743 4 : " of rank " + std::to_string(clusterInfo.rankList[j].rankId) +
744 2 : " on node " + clusterInfo.rankList[j].serverId;
745 7 : RPT_INPUT_ERR(true,
746 : "EI0015",
747 : std::vector<std::string>({"error_reason"}),
748 : std::vector<std::string>({errormessage}));
749 :
750 1 : HCCL_ERROR("[%s][%s]%s",
751 : LOG_KEYWORDS_INIT_GROUP.c_str(),
752 : LOG_KEYWORDS_RANKTABLE_DETECT.c_str(),
753 : errormessage.c_str());
754 1 : return HCCL_E_PARA;
755 1 : }
756 : }
757 : }
758 0 : return HCCL_SUCCESS;
759 1 : }
760 :
761 1 : HcclResult TopoInfoExchangeAgent::VerifyClusterBackupDeviceIP(RankTable_t &clusterInfo)
762 : {
763 1 : if (localRankInfo_.deviceType != DevType::DEV_TYPE_910_93 || !isRetry_) {
764 : // 未开启重执行，则无需 backup device ip
765 0 : return HCCL_SUCCESS;
766 : }
767 1 : bool useSuperPodMode = false;
768 1 : CHK_RET(IsSuperPodMode(useSuperPodMode));
769 1 : if (!useSuperPodMode || clusterInfo.superPodNum == 1) {
770 : // 非多超节点场景，backup device ip 不做要求
771 0 : return HCCL_SUCCESS;
772 : }
773 1 : if (clusterInfo.rankList.size() == 1 || clusterInfo.serverList.size() == 1) {
774 : // 单卡或单机场景对 device ip 不做要求
775 0 : return HCCL_SUCCESS;
776 : }
777 :
778 1 : std::unordered_map<std::string, s32> devIp2PhyId;
779 3 : for (auto &rankInfo : clusterInfo.rankList) {
780 4 : for (auto &devIp : rankInfo.deviceInfo.deviceIp) {
781 2 : devIp2PhyId.emplace(devIp.GetReadableIP(), rankInfo.deviceInfo.devicePhyId);
782 : }
783 : }
784 :
785 3 : for (auto &rankInfo : clusterInfo.rankList) {
786 4 : for (auto &backupDevIp : rankInfo.deviceInfo.backupDeviceIp) {
787 2 : if (backupDevIp.IsInvalid()) {
788 1 : continue;
789 : }
790 2 : std::string backupIpStr = std::string(backupDevIp.GetReadableIP());
791 2 : if (devIp2PhyId.find(backupIpStr) == devIp2PhyId.end()) {
792 1 : HCCL_RUN_WARNING("[Verify][ClusterBackupDeviceIP]"
793 : "backup devIp[%s] for devicePhyId[%d] is not in this comm. "
794 : "The validation of this backup ip could not be verified! "
795 : "Please notice it might be an invalid backup ip!",
796 : backupIpStr.c_str(), rankInfo.deviceInfo.devicePhyId);
797 1 : continue;
798 : }
799 :
800 1 : s32 backupDevPhyId = devIp2PhyId[backupIpStr];
801 1 : std::string errormessage;
802 1 : if (backupDevPhyId == rankInfo.deviceInfo.devicePhyId) {
803 0 : errormessage = "PhyId[" + std::to_string(backupDevPhyId) + "] for backup devIp[" + backupIpStr +
804 0 : "] is the same with self devicephyId[" +
805 0 : std::to_string(rankInfo.deviceInfo.devicePhyId) +
806 0 : "]. Please do not use self ip as backup ip";
807 0 : RPT_INPUT_ERR(true,
808 : "EI0015",
809 : std::vector<std::string>({"error_reason"}),
810 : std::vector<std::string>({ errormessage }));
811 0 : HCCL_ERROR("[%s][%s]errNo[0x%016llx], %s",
812 : LOG_KEYWORDS_INIT_GROUP.c_str(),
813 : LOG_KEYWORDS_RANKTABLE_DETECT.c_str(),
814 : HCOM_ERROR_CODE(HCCL_E_PARA),
815 : errormessage.c_str());
816 0 : return HCCL_E_PARA;
817 : }
818 :
819 1 : LinkTypeInServer linkType = LinkTypeInServer::RESERVED_LINK_TYPE;
820 1 : CHK_RET(hrtGetPairDeviceLinkType(rankInfo.deviceInfo.devicePhyId, backupDevPhyId, linkType));
821 1 : if (linkType != LinkTypeInServer::SIO_TYPE) {
822 0 : RPT_INPUT_ERR(true,
823 : "EI0014",
824 : std::vector<std::string>({ "value", "variable" ,"expect" }),
825 : std::vector<std::string>({ std::to_string(backupDevPhyId), " \"backup_device_ip of "\
826 : "rank " + std::to_string(rankInfo.rankId) + "\" ", " \"is device_ip another Die under the same NPU\" " }));
827 0 : errormessage = "Value " + std::to_string(backupDevPhyId) + " for rankTable variable \"backup_device_ip of "\
828 0 : "rank " + std::to_string(rankInfo.rankId) + "\" is invalid, expected value \"is device_ip another Die under the same NPU\".";
829 :
830 0 : HCCL_ERROR(
831 : "[%s][%s]errNo[0x%016llx], %s",
832 : LOG_KEYWORDS_INIT_GROUP.c_str(),
833 : LOG_KEYWORDS_RANKTABLE_CHECK.c_str(),
834 : HCOM_ERROR_CODE(HCCL_E_PARA),
835 : errormessage.c_str());
836 0 : return HCCL_E_PARA;
837 : }
838 2 : }
839 : }
840 1 : return HCCL_SUCCESS;
841 1 : }
842 :
843 1 : bool TopoInfoExchangeAgent::HasRepeatedIP(const std::vector<HcclIpAddress> &deviceAIP,
844 : const std::vector<HcclIpAddress> &deviceBIP) const
845 : {
846 1 : for (u32 i = 0; i < deviceAIP.size(); i++) {
847 1 : for (u32 j = 0; j < deviceBIP.size(); j++) {
848 1 : if (deviceAIP[i] == deviceBIP[j]) {
849 1 : HCCL_WARNING("device ip[%s] is repeated.", deviceAIP[i].GetReadableAddress());
850 1 : return true;
851 : }
852 : }
853 : }
854 0 : return false;
855 : }
856 :
857 1 : HcclResult TopoInfoExchangeAgent::VerifyClusterRankID(const RankTable_t &clusterInfo) const
858 : {
859 1 : if (clusterInfo.rankList.size() == 1) {
860 0 : return HCCL_SUCCESS;
861 : }
862 1 : for (u32 i = 0; i < (clusterInfo.rankList.size() - 1); i++) {
863 1 : for (u32 j = (i + 1); j < clusterInfo.rankList.size(); j++) {
864 1 : bool isErr = (clusterInfo.rankList[i].rankId == clusterInfo.rankList[j].rankId);
865 1 : if (isErr) {
866 2 : std::string errormessage = "Rank ID " + std::to_string(clusterInfo.rankList[i].rankId) +
867 4 : " of device ID " + std::to_string(clusterInfo.rankList[i].deviceInfo.devicePhyId) + " on node " + clusterInfo.rankList[i].serverId +
868 4 : " is the same as that of device ID " + std::to_string(clusterInfo.rankList[j].deviceInfo.devicePhyId) +
869 2 : " on node " + clusterInfo.rankList[j].serverId;
870 7 : RPT_INPUT_ERR(true,
871 : "EI0015",
872 : std::vector<std::string>({"error_reason"}),
873 : std::vector<std::string>({errormessage}));
874 1 : HCCL_ERROR("[%s][%s]%s",
875 : LOG_KEYWORDS_INIT_GROUP.c_str(),
876 : LOG_KEYWORDS_RANKTABLE_DETECT.c_str(),
877 : errormessage.c_str());
878 1 : return HCCL_E_PARA;
879 1 : }
880 : }
881 : }
882 0 : return HCCL_SUCCESS;
883 1 : }
884 :
885 1 : HcclResult TopoInfoExchangeAgent::VerifyServerDevicePhysicID(const std::vector<RankInfo_t> &serverInfo) const
886 : {
887 1 : if (serverInfo.size() == 1) {
888 0 : return HCCL_SUCCESS;
889 : }
890 1 : for (u32 i = 0; i < (serverInfo.size() - 1); i++) {
891 1 : for (u32 j = (i + 1); j < serverInfo.size(); j++) {
892 1 : bool isErr = (serverInfo[i].deviceInfo.devicePhyId == serverInfo[j].deviceInfo.devicePhyId);
893 1 : if (isErr) {
894 2 : std::string errormessage = "Rank " + std::to_string(serverInfo[i].rankId) + " of node " +
895 3 : serverInfo[i].serverId +
896 4 : " has the same physical device ID " + std::to_string(serverInfo[i].deviceInfo.devicePhyId) +
897 3 : " as the rank " + std::to_string(serverInfo[j].rankId);
898 7 : RPT_INPUT_ERR(true,
899 : "EI0015",
900 : std::vector<std::string>({"error_reason"}),
901 : std::vector<std::string>({errormessage}));
902 1 : HCCL_ERROR("[%s][%s]%s",
903 : LOG_KEYWORDS_INIT_GROUP.c_str(),
904 : LOG_KEYWORDS_RANKTABLE_DETECT.c_str(),
905 : errormessage.c_str());
906 1 : return HCCL_E_PARA;
907 1 : }
908 : }
909 : }
910 0 : return HCCL_SUCCESS;
911 1 : }
912 :
913 2 : HcclResult TopoInfoExchangeAgent::VerifyClusterSuperPodInfo(const std::vector<RankInfo_t> &rankInfo) const
914 : {
915 2 : DevType curDevType = rankInfo.begin()->deviceInfo.deviceType;
916 5 : for (auto curRankInfo : rankInfo) {
917 3 : if (curDevType != curRankInfo.deviceInfo.deviceType) {
918 0 : HCCL_DEBUG("[Verify][SuperPodInfo] mix device type, does not need verify superPod info");
919 0 : return HCCL_SUCCESS;
920 : }
921 3 : }
922 :
923 2 : bool useSuperPodMode = false;
924 2 : CHK_RET(IsSuperPodMode(useSuperPodMode));
925 2 : CHK_PRT_RET(useSuperPodMode == false,
926 : HCCL_DEBUG("[Verify][SuperPodInfo] does not need verify superPod info"), HCCL_SUCCESS);
927 :
928 2 : std::string errormessage = "";
929 : // 获取每个超节点内的serverId
930 2 : std::map<std::string, std::set<std::string>> superPodSrvIdMap; // super_pod_id -> serverId
931 2 : std::map<std::string, std::unordered_map<u32, u32>> superPodSdidMap; // super_pod_id -> superDeviceId
932 3 : for (u32 i = 0; i < rankInfo.size(); i++) {
933 : // 超节点模式下, 校验superPodId和sdid值有效
934 4 : if ((rankInfo[i].superPodId.empty() || rankInfo[i].superDeviceId == INVALID_UINT) &&
935 1 : rankInfo[i].deviceInfo.deviceType == DevType::DEV_TYPE_910_93) {
936 14 : RPT_INPUT_ERR(true,
937 : "EI0014",
938 : std::vector<std::string>({ "value", "variable" ,"expect" }),
939 : std::vector<std::string>({std::to_string(rankInfo[i].superDeviceId), "super_device_id",
940 : "is less than the communication size " + std::to_string(rankInfo.size()) + " and must be unique"}));
941 2 : errormessage = "Value " + std::to_string(rankInfo[i].superDeviceId) + " for rankTable variable superDeviceId is invalid, "\
942 3 : "expected value is less than the communication size " + std::to_string(rankInfo.size()) + " and must be unique.";
943 :
944 1 : HCCL_ERROR("[%s][%s]%s",
945 : LOG_KEYWORDS_INIT_GROUP.c_str(),
946 : LOG_KEYWORDS_RANKTABLE_CHECK.c_str(),
947 : errormessage.c_str());
948 2 : return HCCL_E_PARA;
949 : }
950 :
951 2 : auto iter = superPodSrvIdMap.find(rankInfo[i].superPodId);
952 2 : if (iter == superPodSrvIdMap.end()) {
953 1 : std::set<std::string> serverIdSet;
954 1 : serverIdSet.insert(rankInfo[i].serverId);
955 1 : superPodSrvIdMap.insert({rankInfo[i].superPodId, serverIdSet});
956 2 : } else if (iter->second.find(rankInfo[i].serverId) == iter->second.end()) {
957 0 : iter->second.insert(rankInfo[i].serverId);
958 : }
959 :
960 2 : auto it = superPodSdidMap.find(rankInfo[i].superPodId);
961 2 : if (it == superPodSdidMap.end()) {
962 1 : std::unordered_map<u32, u32> superDeviceIdSet;
963 1 : superDeviceIdSet.insert({rankInfo[i].superDeviceId, rankInfo[i].rankId});
964 1 : superPodSdidMap.insert({rankInfo[i].superPodId, superDeviceIdSet});
965 2 : } else if (it->second.find(rankInfo[i].superDeviceId) == it->second.end()) {
966 0 : it->second.insert({rankInfo[i].superDeviceId, rankInfo[i].rankId});
967 : } else {
968 : // 超节点内superDeviceId在超节点内唯一
969 1 : if (it->second.find(rankInfo[i].superDeviceId) != it->second.end()) {
970 15 : RPT_INPUT_ERR(true,
971 : "EI0014",
972 : std::vector<std::string>({ "value", "variable" ,"expect" }),
973 : std::vector<std::string>({std::to_string(rankInfo[i].superDeviceId), " \"Device Id of server Id " + rankInfo[i].serverId + "\" ", "is unique"}));
974 2 : errormessage = "Value " + std::to_string(rankInfo[i].superDeviceId) + " for rankTable "\
975 2 : "variable \"Device Id of server Id " + rankInfo[i].serverId + "\" is invalid, expected value is unique.";
976 1 : HCCL_ERROR("[%s][%s]%s",
977 : LOG_KEYWORDS_INIT_GROUP.c_str(),
978 : LOG_KEYWORDS_RANKTABLE_CHECK.c_str(),
979 : errormessage.c_str());
980 1 : return HCCL_E_PARA;
981 : }
982 : }
983 : }
984 :
985 : // 校验每个超节点内的server数量一致
986 0 : u32 serverNumPerPod = 0;
987 0 : for (auto iter = superPodSrvIdMap.begin(); iter != superPodSrvIdMap.end(); ++iter) {
988 0 : if (iter == superPodSrvIdMap.begin()) {
989 0 : serverNumPerPod = superPodSrvIdMap.begin()->second.size();
990 : }
991 0 : u32 serverNumCurPod = iter->second.size();
992 0 : if (serverNumPerPod != serverNumCurPod) {
993 0 : HCCL_DEBUG("[Verify][SuperPodInfo]serverNum[%u] in superPod[%s] and serverNum[%u] in superPod[%s] "\
994 : "are different.", serverNumPerPod, superPodSrvIdMap.begin()->first.c_str(),
995 : serverNumCurPod, iter->first.c_str());
996 : }
997 : }
998 :
999 0 : return HCCL_SUCCESS;
1000 8 : }
1001 :
1002 7 : HcclResult TopoInfoExchangeAgent::VerifyClusterTlsConsistency(const RankTable_t &clusterInfo)
1003 : {
1004 7 : bool isSupportCheckTlsStatus = true; // 用于标识是否存在不支持查询Tls开关状态的情况
1005 7 : bool isTlsConsistent = true; // 用于标识TLS开关状态是否一致
1006 7 : std::unordered_map<std::string, std::vector<u32>> tlsEnableRank;
1007 7 : std::unordered_map<std::string, std::vector<u32>> tlsDisableRank;
1008 7 : std::unordered_map<std::string, std::vector<u32>> tlsUnknownRank;
1009 35 : for (auto& rankInfo : clusterInfo.rankList) {
1010 28 : if (rankInfo.tlsStatus == TlsStatus::ENABLE) {
1011 15 : AddRankInfoToTlsStatusMap(rankInfo, tlsEnableRank);
1012 13 : } else if (rankInfo.tlsStatus == TlsStatus::DISABLE) {
1013 9 : AddRankInfoToTlsStatusMap(rankInfo, tlsDisableRank);
1014 : } else {
1015 4 : isSupportCheckTlsStatus = false;
1016 4 : AddRankInfoToTlsStatusMap(rankInfo, tlsUnknownRank);
1017 : }
1018 : }
1019 : // 将不一致的卡信息汇总成一个string
1020 14 : std::string tlsInconsistentEnableStr = "";
1021 14 : std::string tlsInconsistentDisableStr = "";
1022 7 : std::string tlsInconsistentTlsType = "";
1023 7 : if (!tlsEnableRank.empty() && !tlsDisableRank.empty()) {
1024 5 : isTlsConsistent = false;
1025 5 : tlsInconsistentTlsType = (tlsEnableRank.size() >= tlsDisableRank.size()) ? "Disable" : "Enable";
1026 5 : GenerateTlsStatusStr(tlsInconsistentEnableStr, tlsEnableRank);
1027 5 : GenerateTlsStatusStr(tlsInconsistentDisableStr, tlsDisableRank);
1028 : }
1029 : // 将不支持查询的卡的信息汇总成一个string
1030 7 : std::string tlsUnknownRankStr = "";
1031 7 : if (!isSupportCheckTlsStatus) {
1032 3 : GenerateTlsStatusStr(tlsUnknownRankStr, tlsUnknownRank);
1033 : }
1034 11 : tlsUnknownRankStr = tlsUnknownRankStr.empty() ? "N/A" : tlsUnknownRankStr;
1035 : // 四种不同情况
1036 7 : if (isTlsConsistent && isSupportCheckTlsStatus) {
1037 : // 1.通信域所有卡都支持查询TLS开关状态，并且TLS开关状态都是一致的。
1038 1 : HCCL_INFO("[Verify][TlsConsistency] All ranks tlsStatus are consistent");
1039 6 : } else if (!isTlsConsistent && isSupportCheckTlsStatus) {
1040 : // 2.通信域所有卡都支持查询TLS开关状态，但是TLS开关状态存在不一致，报错。
1041 3 : ReportTlsConfigurationError(tlsInconsistentTlsType, tlsInconsistentEnableStr, tlsInconsistentDisableStr, "N/A");
1042 3 : return HCCL_E_PARA;
1043 3 : } else if (isTlsConsistent && !isSupportCheckTlsStatus) {
1044 : // 3.通信域内的部分卡不支持查询TLS开关状态，目前能查询到的卡的TLS开关状态是一致的，打印warning提醒
1045 1 : HCCL_RUN_WARNING("[Verify][TlsConsistency] Some ranks do not support to check tlsStatus, " \
1046 : "not support serverId/rankId: %s", tlsUnknownRankStr.c_str());
1047 : } else {
1048 : // 4.通信域内的部分卡不支持查询TLS开关状态，但是目前能查询到的卡的TLS开关状态已经不一致，报错
1049 2 : ReportTlsConfigurationError(tlsInconsistentTlsType, tlsInconsistentEnableStr, tlsInconsistentDisableStr, tlsUnknownRankStr);
1050 2 : return HCCL_E_PARA;
1051 : }
1052 2 : return HCCL_SUCCESS;
1053 7 : }
1054 :
1055 28 : void TopoInfoExchangeAgent::AddRankInfoToTlsStatusMap(const RankInfo_t &rankInfo,
1056 : std::unordered_map<std::string, std::vector<u32>> &tlsStatusRankMap)
1057 : {
1058 28 : auto iter = tlsStatusRankMap.find(rankInfo.serverId);
1059 28 : if (iter == tlsStatusRankMap.end()) {
1060 15 : std::vector<u32> tlsStatusRankList;
1061 15 : tlsStatusRankList.push_back(rankInfo.rankId);
1062 15 : tlsStatusRankMap.insert({rankInfo.serverId, tlsStatusRankList});
1063 15 : } else {
1064 13 : iter->second.push_back(rankInfo.rankId);
1065 : }
1066 56 : return;
1067 : }
1068 :
1069 13 : void TopoInfoExchangeAgent::GenerateTlsStatusStr(std::string &tlsStatusStr,
1070 : const std::unordered_map<std::string, std::vector<u32>> &tlsStatusRankMap)
1071 : {
1072 26 : for (const auto& rankIt : tlsStatusRankMap) {
1073 13 : tlsStatusStr += ("[" + rankIt.first + "/");
1074 35 : for (const auto& rank : rankIt.second) {
1075 22 : tlsStatusStr += std::to_string(rank) + ",";
1076 : }
1077 13 : if (!tlsStatusStr.empty() && tlsStatusStr.back() == ',') {
1078 13 : tlsStatusStr = tlsStatusStr.substr(0, tlsStatusStr.size() - 1); // 删除逗号
1079 : }
1080 13 : tlsStatusStr += "];";
1081 : }
1082 13 : return;
1083 : }
1084 :
1085 5 : void TopoInfoExchangeAgent::ReportTlsConfigurationError(const std::string& tlsInconsistentTlsType,
1086 : const std::string& tlsInconsistentEnableStr, const std::string& tlsInconsistentDisableStr, const std::string& tlsUnknownRankStr)
1087 : {
1088 10 : std::string errormessage = "Value " + tlsInconsistentTlsType + " for config \"tls\" is invalid. Expected: \"All ranks are consistent. Current status: "\
1089 5 : "rankList for enabled tls: " + tlsInconsistentEnableStr + " rankList for disabled tls:" + tlsInconsistentDisableStr + " rankList for query failure tls:" + tlsUnknownRankStr + ".\"";
1090 70 : RPT_INPUT_ERR(true,
1091 : "EI0016",
1092 : std::vector<std::string>({"value", "variable", "expect"}),
1093 : std::vector<std::string>({tlsInconsistentTlsType, " \"tls\" ",
1094 : " \"All ranks are consistent. Current status: rankList for enabled tls:" + tlsInconsistentEnableStr + "; "\
1095 : "rankList for disabled tls:" + tlsInconsistentDisableStr + " rankList for query failure tls:" + tlsUnknownRankStr + ".\" "}));
1096 5 : HCCL_ERROR("[%s][%s] %s", LOG_KEYWORDS_INIT_GROUP.c_str(), LOG_KEYWORDS_RANKTABLE_CHECK.c_str(), errormessage.c_str());
1097 20 : }
1098 : }
|
