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