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