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