Line data Source code
1 : /**
2 : * Copyright (c) 2026 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "rank_info_detect_service.h"
12 :
13 : #include <stdio.h>
14 : #include "rank_info_dispatcher.h"
15 : #include "env_config/env_config.h"
16 : #include "host_buffer.h"
17 : #include "root_handle_v2.h"
18 : #include "hccp_peer_manager.h"
19 : #include "orion_adapter_rts.h"
20 : #include "../../legacy/ascend950/framework/topo/rank_info_detect/preempt_port_manager.h"
21 : #include "host_socket_handle_manager.h"
22 : #include "adapter_error_manager_pub.h"
23 :
24 : namespace Hccl {
25 :
26 : const u32 DISPLAY_RANKNUM_PERLINE = 8;
27 : const u32 SOCKET_ACCEPT_TIMEOUT = 60; // Server调用Accept等待的最大超时时间 60s
28 : const u32 SOCKET_PRINT_COUNT = 3; // 未建链打印的数量
29 : const u32 MAX_AGENT_BUF_SIZE = 256;
30 :
31 0 : void RankInfoDetectService::Setup()
32 : {
33 : // 1. 连接所有rank
34 0 : GetConnections();
35 :
36 : // 2. 接收所有rank发来的localRankTable并整合为全局RankTable
37 0 : GetRankTable();
38 :
39 : // 3. 将完整RankTable广播给所有rank
40 0 : BroadcastRankTable();
41 0 : }
42 :
43 4 : void RankInfoDetectService::GetConnections()
44 : {
45 4 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
46 :
47 : // 超时参数
48 4 : auto startTime = std::chrono::steady_clock::now();
49 4 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
50 4 : bool isFirstAcceptTimeOut = false;
51 :
52 : // 期望等待连接的rank数量
53 4 : u32 expectedSocketNum = 1;
54 :
55 : // 首个connect获取到的rankSize
56 4 : u32 previousRankNum = 0;
57 :
58 : // 获取server端socket信息
59 4 : u32 hostPort = serverSocket_->GetListenPort();
60 4 : auto hccpHostSocketHandle = HostSocketHandleManager::GetInstance().Get(devPhyId_, hostIp_);
61 4 : CHK_PRT_THROW(hccpHostSocketHandle == nullptr,
62 : HCCL_ERROR("[RankInfoDetectService::%s] Get hccpHostSocketHandle fail.", __func__),
63 : InternalException, "get socket handle error");
64 4 : std::string connSocketTag = RANK_INFO_DETECT_TAG + "_" + identifier_ + "_" + std::to_string(hostPort);
65 4 : SocketStatus status = SocketStatus::INVALID;
66 :
67 : // 连接rankSize个client
68 21 : while (expectedSocketNum > 0) {
69 21 : auto topoExUsedTime = std::chrono::steady_clock::now() - startTime;
70 21 : if (topoExUsedTime >= timeout) {
71 14 : RPT_INPUT_ERR(true, "EI0015", std::vector<std::string>({"error_reason"}),
72 : std::vector<std::string>({StringFormat("Receiving message from the root node timed out. "
73 : "Timeout was set to %lld seconds. expected %u nodes, received %u nodes. "
74 : "Check whether worker nodes are reachable and report errors.",
75 : static_cast<long long>(timeout.count()),
76 : expectedSocketNum + previousRankNum, previousRankNum)}));
77 2 : HCCL_ERROR("[RankInfoDetectService::%s] server get sockets timeout[%lld s]", __func__, timeout);
78 4 : break;
79 : }
80 :
81 : // duration_cast<seconds> 会进行向下取整，不足 1s 时提前跳出，确保建链超时场景，server端在client端退出前发送临终遗言
82 19 : auto topoExResTime = timeout - topoExUsedTime;
83 19 : u32 topoExRes_i = std::chrono::duration_cast<std::chrono::seconds>(topoExResTime).count();
84 19 : if (topoExRes_i == 0) {
85 1 : HCCL_ERROR("[RankInfoDetectService::%s] timeout[%lld s] is exhausted", __func__, timeout);
86 1 : break;
87 : }
88 : std::shared_ptr<Socket> connSocket = std::make_shared<Socket>(
89 18 : hccpHostSocketHandle, hostIp_, hostPort, hostIp_, connSocketTag, SocketRole::SERVER, NicType::HOST_NIC_TYPE);
90 : // GetStatus 是阻塞接口，传入剩余时间作为超时上限，避免其内部超时导致外层循环超时处理失效
91 18 : EXCEPTION_CATCH(status = connSocket->GetStatus(topoExRes_i),
92 : {
93 : // 非本端client首次连接异常，直接重试
94 : if(status == SocketStatus::OK) {
95 : status = SocketStatus::CONNECTING;
96 : }
97 : HCCL_ERROR("[RankInfoDetectService::%s] server get socket fail", __func__);
98 : });
99 18 : if (status == SocketStatus::OK) {
100 6 : if(!RecvAndVerifyRemoteAgentIdAndRankSize(connSocket, expectedSocketNum, previousRankNum)) {
101 1 : break;
102 : }
103 5 : expectedSocketNum--;
104 5 : isFirstAcceptTimeOut = false;
105 5 : HCCL_INFO("[RankInfoDetectService::%s] socket[%s] connect ok.", __func__, connSocket->Describe().c_str());
106 12 : } else if (status == SocketStatus::CONNECTING) {
107 7 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
108 5 : } else if (status == SocketStatus::TIMEOUT) {
109 : // 避免重复打印
110 5 : if (isFirstAcceptTimeOut) {
111 3 : continue;
112 : }
113 2 : HCCL_ERROR("[RankInfoDetectService::%s] rank info detect server get socket timeout[%lld s]", __func__, timeout);
114 2 : DisplayConnectingStatus(previousRankNum, expectedSocketNum);
115 2 : isFirstAcceptTimeOut = true;
116 : } else {
117 0 : HCCL_ERROR("[RankInfoDetectService::%s] SocketStatus[%s] error", __func__, status.Describe().c_str());
118 0 : break;
119 : }
120 18 : }
121 :
122 : // 如果没有连接成功的rank则退出
123 6 : CHK_PRT_THROW(connSockets_.size() == 0, HCCL_ERROR("[RankInfoDetectService::%s] no rank connection success.", __func__),
124 : InternalException, "no rank connection success");
125 :
126 : // 处理异常流程
127 3 : if (expectedSocketNum > 0) {
128 : // 将建立连接超时的client信息添加到failedAgentIdList_
129 3 : FailedConnectionAgentIdString(previousRankNum);
130 3 : DisplayConnectedRanks();
131 3 : HCCL_INFO("[RankInfoDetectService::%s] end, there exist non-connected ranks.", __func__);
132 : } else {
133 0 : HCCL_INFO("[RankInfoDetectService::%s] end, all agentId get connection socket success.", __func__);
134 : }
135 6 : }
136 :
137 1 : void RankInfoDetectService::GetRankTable()
138 : {
139 1 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
140 :
141 : // 接收localRankTable并组全局RankTableInfo
142 1 : rankTable_ = RankTableInfo{};
143 2 : for (auto &iter : connSockets_) {
144 1 : vector<char> rankInfoMsg{};
145 1 : SocketAgent socketAgent(iter.second.get());
146 1 : RecvRankInfoMsg(socketAgent, rankInfoMsg);
147 1 : ParseRankTable(rankInfoMsg);
148 1 : }
149 :
150 : // 按照rankid排序
151 1 : SortRankTable();
152 :
153 : // 更新当前阶段
154 1 : currentStep_++;
155 :
156 1 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
157 1 : }
158 :
159 2 : void RankInfoDetectService::BroadcastRankTable()
160 : {
161 2 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
162 :
163 : // 广播全局ranktable
164 2 : std::shared_ptr<RankInfoDispather> dispatcher = std::make_shared<RankInfoDispather>(this);
165 2 : dispatcher->BroadcastRankTable(connSockets_, rankTable_, failedAgentIdList_, currentStep_);
166 :
167 2 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
168 2 : }
169 :
170 30 : void RankInfoDetectService::Disconnect()
171 : {
172 39 : for (auto iter = connSockets_.begin(); iter != connSockets_.end();) {
173 9 : iter->second.get()->Close();
174 9 : iter = connSockets_.erase(iter);
175 : }
176 30 : }
177 :
178 2 : bool RankInfoDetectService::RecvRemoteAgentId(SocketAgent &connSocketAgent, std::string &agentId)
179 : {
180 2 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
181 :
182 : // 接收消息
183 2 : u64 revMsgLen = 0;
184 2 : char msg[MAX_AGENT_BUF_SIZE] = {0};
185 2 : bool ret = connSocketAgent.RecvMsg(msg, revMsgLen);
186 2 : CHK_PRT_RET(!ret || revMsgLen >= MAX_AGENT_BUF_SIZE,
187 : HCCL_ERROR("[RankInfoDetectService::%s] recv error, revMsgLen[%llu].", __func__, revMsgLen), false);
188 :
189 : // 解析agentId
190 2 : msg[revMsgLen] = '\0';
191 2 : agentId = msg;
192 :
193 2 : HCCL_INFO("[RankInfoDetectService::%s] agentId[%s]", __func__, agentId.c_str());
194 2 : return true;
195 : }
196 :
197 1 : bool RankInfoDetectService::RecvRemoteRankSize(SocketAgent &connSocketAgent, u32 &rankSize)
198 : {
199 1 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
200 :
201 : // 接收rankSize
202 1 : u64 revMsgLen = 0;
203 1 : bool ret = connSocketAgent.RecvMsg(&rankSize, revMsgLen);
204 1 : CHK_PRT_RET(!ret, HCCL_ERROR("[RankInfoDetectService::%s] RecvMsg fail, revMsgLen[%llu].", __func__, revMsgLen), false);
205 :
206 1 : HCCL_INFO("[RankInfoDetectService::%s] rankSize[%u]", __func__, rankSize);
207 1 : return true;
208 : }
209 :
210 : // 接收客户端发送的字节流形式的rankinfo消息
211 1 : void RankInfoDetectService::RecvRankInfoMsg(SocketAgent &connSocketAgent, vector<char> &rankInfoMsg)
212 : {
213 1 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
214 :
215 1 : u64 revMsgLen = 0;
216 1 : std::unique_ptr<HostBuffer> msg = std::make_unique<HostBuffer>(MAX_BUFFER_LEN);
217 1 : char *msgAddr = reinterpret_cast<char *>(msg->GetAddr());
218 1 : CHK_PRT_THROW(!connSocketAgent.RecvMsg(msgAddr, revMsgLen),
219 : HCCL_ERROR("[RankInfoDetectService::%s] RecvMsg fail, revMsgLen[%llu]", __func__, revMsgLen),
220 : InvalidParamsException, "RecvMsg fail");
221 :
222 : // 以vector<char>格式保存
223 1 : rankInfoMsg.resize(revMsgLen);
224 1 : rankInfoMsg.assign(msgAddr, msgAddr + revMsgLen);
225 :
226 1 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
227 1 : }
228 :
229 : // 解析接收到的rank table信息
230 2 : void RankInfoDetectService::ParseRankTable(vector<char> &rankInfoMsg)
231 : {
232 2 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
233 :
234 : // 消息格式: [ranktable数据(n字节)][step(4字节)]
235 2 : BinaryStream binStream(rankInfoMsg);
236 :
237 : // 解析localRankInfo
238 2 : RankTableInfo localRankInfo(binStream);
239 2 : localRankInfo.Dump();
240 :
241 : // 解析step
242 : u32 receivedStep;
243 2 : binStream >> receivedStep;
244 :
245 : // 校验step是否匹配
246 2 : CHK_PRT_THROW(receivedStep != currentStep_,
247 : HCCL_ERROR("[RankInfoDetectService::%s] Step mismatch: received %u, expected %u", __func__, receivedStep, currentStep_),
248 : InvalidParamsException, "Step mismatch");
249 :
250 : // 添加到rankTable_
251 2 : rankTable_.UpdateRankTable(localRankInfo);
252 :
253 2 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
254 2 : }
255 :
256 0 : bool RankIdCompare(const NewRankInfo &i, const NewRankInfo &j)
257 : {
258 0 : return (i.rankId < j.rankId);
259 : }
260 :
261 1 : void RankInfoDetectService::SortRankTable()
262 : {
263 1 : std::sort(rankTable_.ranks.begin(), rankTable_.ranks.end(), RankIdCompare);
264 1 : }
265 :
266 3 : void RankInfoDetectService::FailedConnectionAgentIdString(u32 rankSize)
267 : {
268 3 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
269 :
270 3 : std::vector<bool> connectedRank(rankSize, false);
271 7 : for (auto it : connSockets_) {
272 5 : u32 rankid = 0;
273 5 : HcclResult ret = SalStrToULong(it.first, HCCL_BASE_DECIMAL, rankid);
274 5 : CHK_PRT_RET(ret != HCCL_SUCCESS,
275 : HCCL_ERROR("[RankInfoDetectService::%s] agentId[%s] strToULong fail.", __func__, it.first.c_str()),);
276 4 : CHK_PRT_RET(rankid >= rankSize,
277 : HCCL_ERROR("[RankInfoDetectService::%s] invalid rank id[%u], rankSize[%u].", __func__, rankid, rankSize),);
278 4 : connectedRank[rankid] = true;
279 5 : }
280 :
281 10 : for (u32 i = 0; i < rankSize; i++) {
282 8 : if (!connectedRank[i]) {
283 4 : if (!failedAgentIdList_.empty()) {
284 2 : failedAgentIdList_ += ',';
285 : }
286 4 : failedAgentIdList_ += std::to_string(i);
287 : }
288 : }
289 :
290 2 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
291 3 : }
292 :
293 : // 校验相关方法
294 6 : bool RankInfoDetectService::RecvAndVerifyRemoteAgentIdAndRankSize(
295 : std::shared_ptr<Socket> connSocket, u32 &expectedSocketNum, u32 &previousRankSize)
296 : {
297 6 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
298 6 : SocketAgent socketAgent(connSocket.get());
299 :
300 : // 接收AgentId
301 6 : std::string agentId = "";
302 6 : bool ret = RecvRemoteAgentId(socketAgent, agentId);
303 6 : CHK_PRT_RET(!ret, HCCL_ERROR("[RankInfoDetectService::%s] RecvRemoteAgentId fail.", __func__), false);
304 :
305 : // 保存connSocket
306 6 : auto iter = connSockets_.find(agentId);
307 6 : CHK_PRT_RET(iter != connSockets_.end(),
308 : HCCL_ERROR("[RankInfoDetectService::%s] agentId[%s] has been connected.", __func__, agentId.c_str()),
309 : false);
310 5 : connSockets_.insert({agentId, connSocket});
311 :
312 : // 接收RankSize
313 5 : u32 rankSize = 0;
314 5 : ret = RecvRemoteRankSize(socketAgent, rankSize);
315 5 : CHK_PRT_RET(!ret, HCCL_ERROR("[RankInfoDetectService::%s] RecvRemoteAgentId fail.", __func__), false);
316 :
317 : // 校验
318 5 : expectedSocketNum = (previousRankSize == 0) ? rankSize : expectedSocketNum;
319 5 : CHK_PRT_RET(!VerifyRemoteRankSize(previousRankSize, rankSize),
320 : HCCL_ERROR("[RankInfoDetectService::%s] VerifyRemoteRankSize fail, rankSize[%u]", __func__, rankSize),
321 : false);
322 :
323 5 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
324 5 : return true;
325 6 : }
326 :
327 5 : bool RankInfoDetectService::VerifyRemoteRankSize(u32 &previousRankSize, u32 remoteRankSize) const
328 : {
329 5 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
330 :
331 5 : if (previousRankSize == 0) {
332 3 : previousRankSize = remoteRankSize;
333 : } else {
334 2 : if (previousRankSize != remoteRankSize) {
335 0 : HCCL_ERROR("[RankInfoDetectService::%s] VerifyRemoteRankSize failed. remoteRankSize[%u] is different "
336 : "from previousRankSize[%u].", __func__, remoteRankSize, previousRankSize);
337 0 : return false;
338 : }
339 : }
340 :
341 5 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
342 5 : return true;
343 : }
344 :
345 : // DFX相关方法
346 3 : void RankInfoDetectService::DisplayConnectedRanks()
347 : {
348 3 : vector<std::string> ranksInfo;
349 8 : for (const auto &it : connSockets_) {
350 5 : ranksInfo.push_back(it.first);
351 : }
352 3 : u64 ranksLen = ranksInfo.size();
353 3 : u64 lineNum = (ranksInfo.size() % DISPLAY_RANKNUM_PERLINE == 0) ? (ranksInfo.size() / DISPLAY_RANKNUM_PERLINE)
354 3 : : (ranksInfo.size() / DISPLAY_RANKNUM_PERLINE + 1);
355 3 : HCCL_ERROR("[RankInfoDetectService::%s] total connected num is [%llu],line num is [%llu]", __func__, ranksLen, lineNum);
356 6 : for (u64 i = 0; i < lineNum; i++) {
357 3 : std::string tmpRankList;
358 8 : for (u32 j = 0; j < DISPLAY_RANKNUM_PERLINE; j++) {
359 8 : u32 ranksInfoIndex = i * DISPLAY_RANKNUM_PERLINE + j;
360 8 : if (ranksInfoIndex < ranksInfo.size()) {
361 5 : tmpRankList += "[" + ranksInfo[ranksInfoIndex] + "]";
362 : } else {
363 3 : break;
364 : }
365 5 : tmpRankList += ((j == DISPLAY_RANKNUM_PERLINE - 1 || ranksInfoIndex == ranksInfo.size() - 1) ? ";" : ",");
366 : }
367 3 : HCCL_ERROR("[RankInfoDetectService::%s] connected rankinfo[LINE %llu]: %s", __func__, i, tmpRankList.c_str());
368 3 : }
369 3 : }
370 :
371 2 : void RankInfoDetectService::DisplayConnectingStatus(u32 totalSockets, u32 waitSockets)
372 : {
373 2 : if (totalSockets == 0 && waitSockets == 1) {
374 0 : HCCL_INFO("[RankInfoDetectService::%s] wait for first connection.", __func__);
375 0 : return;
376 : }
377 :
378 2 : std::vector<bool> rankinfos(totalSockets, false);
379 5 : for (auto it : connSockets_) { // 建立映射
380 3 : u32 rankid = 0;
381 3 : HcclResult ret = SalStrToULong(it.first, HCCL_BASE_DECIMAL, rankid);
382 3 : CHK_PRT_RET(ret != HCCL_SUCCESS,
383 : HCCL_ERROR("[RankInfoDetectService::%s] agentId[%s] strToULong fail.", __func__, it.first.c_str()),);
384 3 : CHK_PRT_RET(rankid >= totalSockets,
385 : HCCL_ERROR("[RankInfoDetectService::%s] invalid rankid[%u], rankSize[%u].", __func__, rankid, totalSockets),);
386 3 : rankinfos[rankid] = true;
387 3 : }
388 :
389 2 : u32 unRankCount = 0; // 只打印前三条未建链的rank
390 2 : std::vector<std::string> unsocketinfos;
391 9 : for (u32 rankid = 0; rankid < totalSockets; rankid++) {
392 8 : if (unRankCount >= SOCKET_PRINT_COUNT) {
393 1 : break;
394 : }
395 7 : if (!rankinfos[rankid]) {
396 5 : unRankCount++;
397 5 : std::string rankID = std::to_string(rankid);
398 5 : std::string agentID = std::string(16 - rankID.length(), '0') + rankID;
399 5 : unsocketinfos.push_back(agentID);
400 5 : }
401 : }
402 :
403 4 : std::string infoStr = "succ sockets is [" + std::to_string((totalSockets - waitSockets)) +
404 6 : "], waiting sockets is [" + std::to_string(waitSockets) + "], wait sockets rankid: ";
405 7 : for (u32 index = 0; index < unsocketinfos.size(); index++) {
406 5 : if (index == (unsocketinfos.size() - 1)) {
407 2 : infoStr += "[" + unsocketinfos[index] + "]";
408 : } else {
409 3 : infoStr += "[" + unsocketinfos[index] + "],";
410 : }
411 : }
412 :
413 2 : HCCL_INFO("[RankInfoDetectService::%s] %s", __func__, infoStr.c_str());
414 2 : }
415 :
416 29 : void RankInfoDetectService::TearDown()
417 : {
418 29 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
419 :
420 29 : CHK_PRT_RET(!serverSocket_, HCCL_INFO("[RankInfoDetectService::%s] serverSocket is null", __func__),);
421 :
422 : // close socket
423 29 : Disconnect();
424 :
425 : // 如果白名单使能则删除白名单
426 29 : if (!EnvConfig::GetInstance().GetHostNicConfig().GetWhitelistDisable()) {
427 0 : CHK_PRT_CONT(wlistInfo_.size() == 0, HCCL_ERROR("whitelist is empty");break);
428 0 : SocketHandle hostSocketHandle = HostSocketHandleManager::GetInstance().Get(devPhyId_, hostIp_);
429 0 : HrtRaSocketWhiteListDel(hostSocketHandle, wlistInfo_);
430 : }
431 :
432 29 : s32 deviceLogicId = HrtGetDevice();
433 58 : if (EnvConfig::GetInstance().GetHostNicConfig().GetHostSocketPortRange().size() > 0 ||
434 29 : EnvConfig::GetInstance().GetHostNicConfig().GetIfBasePort() == HCCL_INVALID_PORT) {
435 : // 若开启抢占监听端口
436 29 : PreemptPortManager::GetInstance(deviceLogicId).Release(serverSocket_);
437 : } else {
438 : // 停止监听
439 0 : serverSocket_->StopListen();
440 : }
441 :
442 : // deinit handle
443 29 : HostSocketHandleManager::GetInstance().Destroy(devPhyId_, hostIp_);
444 :
445 : // deinit ra
446 29 : HccpPeerManager::GetInstance().DeInit(deviceLogicId);
447 :
448 29 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
449 : }
450 :
451 29 : RankInfoDetectService::~RankInfoDetectService()
452 : {
453 29 : DECTOR_TRY_CATCH("RankInfoDetectService", TearDown());
454 29 : }
455 :
456 : } // namespace Hccl
|
