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 "rank_info_detect_service.h"
12 :
13 : #include <stdio.h>
14 : #include "rank_info_dispatcher.h"
15 : #include "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 "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 12 : 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 2 : 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 6 : 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 3 : 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 36 : 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 15 : 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 6 : 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 9 : 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 9 : 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 4 : }
136 :
137 1 : void RankInfoDetectService::GetRankTable()
138 : {
139 3 : 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 3 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
157 1 : }
158 :
159 2 : void RankInfoDetectService::BroadcastRankTable()
160 : {
161 6 : 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 6 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
168 2 : }
169 :
170 31 : void RankInfoDetectService::Disconnect()
171 : {
172 40 : for (auto iter = connSockets_.begin(); iter != connSockets_.end();) {
173 9 : iter->second.get()->Close();
174 9 : iter = connSockets_.erase(iter);
175 : }
176 31 : }
177 :
178 2 : bool RankInfoDetectService::RecvRemoteAgentId(SocketAgent &connSocketAgent, std::string &agentId)
179 : {
180 6 : 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 6 : 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 3 : 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 3 : 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 3 : 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 3 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
227 1 : }
228 :
229 : // 解析接收到的rank table信息
230 2 : void RankInfoDetectService::ParseRankTable(vector<char> &rankInfoMsg)
231 : {
232 6 : 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 6 : 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 4 : void RankInfoDetectService::FailedConnectionAgentIdString(u32 rankSize)
267 : {
268 12 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
269 :
270 4 : std::vector<bool> connectedRank(rankSize, false);
271 8 : for (auto it : connSockets_) {
272 5 : u32 rankid = 0;
273 8 : CHK_PRT_RET_NULL(SalStrToULong(it.first, HCCL_BASE_DECIMAL, rankid),
274 : HCCL_ERROR("[RankInfoDetectService::%s] agentId[%s] strToULong fail.", __func__, it.first.c_str()));
275 4 : CHK_PRT_RET_NULL(rankid >= rankSize,
276 : HCCL_ERROR("[RankInfoDetectService::%s] invalid rank id[%u], rankSize[%u].", __func__, rankid, rankSize));
277 4 : connectedRank[rankid] = true;
278 5 : }
279 :
280 14 : for (u32 i = 0; i < rankSize; i++) {
281 11 : if (!connectedRank[i]) {
282 7 : if (!failedAgentIdList_.empty()) {
283 4 : failedAgentIdList_ += ",";
284 : }
285 7 : failedAgentIdList_ += std::to_string(i);
286 : }
287 : }
288 :
289 9 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
290 4 : }
291 :
292 : // 校验相关方法
293 6 : bool RankInfoDetectService::RecvAndVerifyRemoteAgentIdAndRankSize(
294 : std::shared_ptr<Socket> connSocket, u32 &expectedSocketNum, u32 &previousRankSize)
295 : {
296 18 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
297 6 : SocketAgent socketAgent(connSocket.get());
298 :
299 : // 接收AgentId
300 6 : std::string agentId = "";
301 6 : bool ret = RecvRemoteAgentId(socketAgent, agentId);
302 6 : CHK_PRT_RET(!ret, HCCL_ERROR("[RankInfoDetectService::%s] RecvRemoteAgentId fail.", __func__), false);
303 :
304 : // 保存connSocket
305 6 : auto iter = connSockets_.find(agentId);
306 9 : CHK_PRT_RET(iter != connSockets_.end(),
307 : HCCL_ERROR("[RankInfoDetectService::%s] agentId[%s] has been connected.", __func__, agentId.c_str()),
308 : false);
309 5 : connSockets_.insert({agentId, connSocket});
310 :
311 : // 接收RankSize
312 5 : u32 rankSize = 0;
313 5 : ret = RecvRemoteRankSize(socketAgent, rankSize);
314 5 : CHK_PRT_RET(!ret, HCCL_ERROR("[RankInfoDetectService::%s] RecvRemoteAgentId fail.", __func__), false);
315 :
316 : // 校验
317 5 : expectedSocketNum = (previousRankSize == 0) ? rankSize : expectedSocketNum;
318 5 : CHK_PRT_RET(!VerifyRemoteRankSize(previousRankSize, rankSize),
319 : HCCL_ERROR("[RankInfoDetectService::%s] VerifyRemoteRankSize fail, rankSize[%u]", __func__, rankSize),
320 : false);
321 :
322 15 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
323 5 : return true;
324 6 : }
325 :
326 5 : bool RankInfoDetectService::VerifyRemoteRankSize(u32 &previousRankSize, u32 remoteRankSize) const
327 : {
328 15 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
329 :
330 5 : if (previousRankSize == 0) {
331 3 : previousRankSize = remoteRankSize;
332 : } else {
333 2 : if (previousRankSize != remoteRankSize) {
334 0 : HCCL_ERROR("[RankInfoDetectService::%s] VerifyRemoteRankSize failed. remoteRankSize[%u] is different "
335 : "from previousRankSize[%u].", __func__, remoteRankSize, previousRankSize);
336 0 : return false;
337 : }
338 : }
339 :
340 15 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
341 5 : return true;
342 : }
343 :
344 : // DFX相关方法
345 3 : void RankInfoDetectService::DisplayConnectedRanks()
346 : {
347 3 : vector<std::string> ranksInfo;
348 8 : for (const auto &it : connSockets_) {
349 5 : ranksInfo.push_back(it.first);
350 : }
351 3 : u64 ranksLen = ranksInfo.size();
352 3 : u64 lineNum = (ranksInfo.size() % DISPLAY_RANKNUM_PERLINE == 0) ? (ranksInfo.size() / DISPLAY_RANKNUM_PERLINE)
353 3 : : (ranksInfo.size() / DISPLAY_RANKNUM_PERLINE + 1);
354 9 : HCCL_ERROR("[RankInfoDetectService::%s] total connected num is [%llu],line num is [%llu]", __func__, ranksLen, lineNum);
355 6 : for (u64 i = 0; i < lineNum; i++) {
356 3 : std::string tmpRankList;
357 8 : for (u32 j = 0; j < DISPLAY_RANKNUM_PERLINE; j++) {
358 8 : u32 ranksInfoIndex = i * DISPLAY_RANKNUM_PERLINE + j;
359 8 : if (ranksInfoIndex < ranksInfo.size()) {
360 5 : tmpRankList += "[" + ranksInfo[ranksInfoIndex] + "]";
361 : } else {
362 3 : break;
363 : }
364 5 : tmpRankList += ((j == DISPLAY_RANKNUM_PERLINE - 1 || ranksInfoIndex == ranksInfo.size() - 1) ? ";" : ",");
365 : }
366 9 : HCCL_ERROR("[RankInfoDetectService::%s] connected rankinfo[LINE %llu]: %s", __func__, i, tmpRankList.c_str());
367 3 : }
368 3 : }
369 :
370 2 : void RankInfoDetectService::DisplayConnectingStatus(u32 totalSockets, u32 waitSockets)
371 : {
372 2 : if (totalSockets == 0 && waitSockets == 1) {
373 0 : HCCL_INFO("[RankInfoDetectService::%s] wait for first connection.", __func__);
374 0 : return;
375 : }
376 :
377 2 : std::vector<bool> rankinfos(totalSockets, false);
378 5 : for (auto it : connSockets_) { // 建立映射
379 3 : u32 rankid = 0;
380 3 : CHK_PRT_RET_NULL(SalStrToULong(it.first, HCCL_BASE_DECIMAL, rankid),
381 : HCCL_ERROR("[RankInfoDetectService::%s] agentId[%s] strToULong fail.", __func__, it.first.c_str()));
382 3 : CHK_PRT_RET_NULL(rankid >= totalSockets,
383 : HCCL_ERROR("[RankInfoDetectService::%s] invalid rankid[%u], rankSize[%u].", __func__, rankid, totalSockets));
384 3 : rankinfos[rankid] = true;
385 3 : }
386 :
387 2 : u32 unRankCount = 0; // 只打印前三条未建链的rank
388 2 : std::vector<std::string> unsocketinfos;
389 9 : for (u32 rankid = 0; rankid < totalSockets; rankid++) {
390 8 : if (unRankCount >= SOCKET_PRINT_COUNT) {
391 1 : break;
392 : }
393 7 : if (!rankinfos[rankid]) {
394 5 : unRankCount++;
395 5 : std::string rankID = std::to_string(rankid);
396 5 : std::string agentID = std::string(16 - rankID.length(), '0') + rankID;
397 5 : unsocketinfos.push_back(agentID);
398 5 : }
399 : }
400 :
401 4 : std::string infoStr = "succ sockets is [" + std::to_string((totalSockets - waitSockets)) +
402 6 : "], waiting sockets is [" + std::to_string(waitSockets) + "], wait sockets rankid: ";
403 7 : for (u32 index = 0; index < unsocketinfos.size(); index++) {
404 5 : if (index == (unsocketinfos.size() - 1)) {
405 2 : infoStr += "[" + unsocketinfos[index] + "]";
406 : } else {
407 3 : infoStr += "[" + unsocketinfos[index] + "],";
408 : }
409 : }
410 :
411 6 : HCCL_INFO("[RankInfoDetectService::%s] %s", __func__, infoStr.c_str());
412 2 : }
413 :
414 30 : void RankInfoDetectService::TearDown()
415 : {
416 90 : HCCL_INFO("[RankInfoDetectService::%s] start.", __func__);
417 :
418 30 : CHK_PRT_RET_NULL(!serverSocket_,
419 : HCCL_INFO("[RankInfoDetectService::%s] serverSocket is null", __func__));
420 :
421 : // close socket
422 30 : Disconnect();
423 :
424 : // 如果白名单使能则删除白名单
425 30 : if (!EnvConfig::GetInstance().GetHostNicConfig().GetWhitelistDisable()) {
426 0 : CHK_PRT_CONT(wlistInfo_.size() == 0, HCCL_ERROR("whitelist is empty");break);
427 0 : SocketHandle hostSocketHandle = HostSocketHandleManager::GetInstance().Get(devPhyId_, hostIp_);
428 0 : HrtRaSocketWhiteListDel(hostSocketHandle, wlistInfo_);
429 : }
430 :
431 30 : s32 deviceLogicId = HrtGetDevice();
432 60 : if (EnvConfig::GetInstance().GetHostNicConfig().GetHostSocketPortRange().size() > 0 ||
433 30 : EnvConfig::GetInstance().GetHostNicConfig().GetIfBasePort() == HCCL_INVALID_PORT) {
434 : // 若开启抢占监听端口
435 30 : PreemptPortManager::GetInstance(deviceLogicId).Release(serverSocket_);
436 : } else {
437 : // 停止监听
438 0 : serverSocket_->StopListen();
439 : }
440 :
441 : // deinit handle
442 30 : HostSocketHandleManager::GetInstance().Destroy(devPhyId_, hostIp_);
443 :
444 : // deinit ra
445 30 : HccpPeerManager::GetInstance().DeInit(deviceLogicId);
446 :
447 90 : HCCL_INFO("[RankInfoDetectService::%s] end.", __func__);
448 : }
449 :
450 30 : RankInfoDetectService::~RankInfoDetectService()
451 : {
452 30 : DECTOR_TRY_CATCH("RankInfoDetectService", TearDown());
453 30 : }
454 :
455 : } // namespace Hccl
|
