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 : #include "cluster_monitor.h"
11 : #include "hccl_types.h"
12 : #include "hccl_comm_pub.h"
13 : #include "env_config/env_config.h"
14 : #include "log.h"
15 :
16 : #include "hcclCommTaskException.h"
17 : #include "ccuTaskException.h"
18 : #include "coll_comm_mgr.h"
19 : #include "heartbeat.h"
20 : #include "comm_addr_logger.h"
21 :
22 : constexpr u32 ONE_SECOND_OF_SLEEP = 1; // 1s
23 : namespace hcomm {
24 :
25 2666 : ClusterMonitor::~ClusterMonitor()
26 : {
27 2666 : DeInit();
28 2666 : }
29 :
30 18 : ClusterUIDType ClusterMonitor::FormatUID(ClusterUIDCxt cxt) const
31 : {
32 18 : ClusterUIDType uid{};
33 : // 构造唯一的uid: netInstanceId + local_id
34 18 : (void)snprintf_s(uid.id, sizeof(uid.id), sizeof(uid.id) - 1, "%s/%s",
35 36 : cxt.netInstId.c_str(), std::to_string(cxt.localId).c_str());
36 :
37 18 : return uid;
38 : }
39 :
40 59 : std::string ClusterMonitor::GetUID(const ClusterUIDType &uid) const
41 : {
42 118 : return uid.id;
43 : }
44 :
45 1 : void ClusterMonitor::GetRemEndpointDescsPerLayer(uint32_t netLayer, HcclComm comm, const Hccl::RankGraph *rankGraph,
46 : const hccl::CollComm* collComm, std::map<uint32_t, std::vector<UIDContext>> &uidCtxs, std::set<uint32_t> &rankIdsSet)
47 : {
48 1 : uint32_t *ranksPerLayer = nullptr;
49 1 : uint32_t rankNum = 0;
50 1 : auto myRankId = collComm->GetMyRankId();
51 1 : HcclRankGraphGetRanksByLayer(comm, netLayer, &ranksPerLayer, &rankNum); // 获取每层netLayer的所有rank
52 3 : for (uint32_t rankIdx = 0; rankIdx < rankNum; rankIdx++) {
53 2 : uint32_t rankId = ranksPerLayer[rankIdx];
54 2 : if (rankIdsSet.find(rankId) != rankIdsSet.end()) {
55 0 : continue; // rankSet维护了所有的ranks，如果已经加到Set说明该rank已经在更低的netLayer层级加入
56 : }
57 2 : rankIdsSet.insert(rankId);
58 2 : auto *netInstance = rankGraph->GetNetInstanceByRankId(0, rankId); // 查询对应rankId在netLayer=0的netInsId
59 2 : if (netInstance == nullptr) {
60 0 : continue; // 如果没有查询到netInstance，不报错，不把该rank加入needConnectRank，直接跳过该rank
61 : }
62 2 : auto netInstanceId = netInstance->GetNetInstId();
63 2 : auto localId = rankGraph->GetLocalId(rankId); // 根据rank查localId
64 2 : ClusterUIDCxt uidcxt(netInstanceId, localId);
65 2 : ClusterUIDType uid = FormatUID(uidcxt);
66 2 : if (myRankId == rankId) {
67 1 : myRankUID_ = uid;
68 1 : myRankLocalId_ = localId;
69 1 : myRankNetInstId_ = netInstanceId;
70 : }
71 2 : uid2FrameStatusMap_.insert(uid, FrameStatus());
72 2 : commIdMap_[collComm->GetCommId()].insert(std::make_pair(uid, false)); //初始状态均为未连接，包含自己
73 2 : if (uidCtxs.find(netLayer) == uidCtxs.end()) {
74 1 : uidCtxs.insert(std::make_pair(netLayer, std::vector<UIDContext>()));
75 : }
76 2 : UIDContext uidCtx(uid, netLayer, rankId, localId, netInstanceId);
77 2 : uidCtxs[netLayer].emplace_back(uidCtx);
78 2 : HCCL_INFO("commId[%s] insert remoteUID[%s]", collComm->GetCommId().c_str(), GetUID(uid).c_str());
79 2 : }
80 1 : }
81 :
82 3 : HcclResult ClusterMonitor::GetRemEndpointDescs(HcclComm comm, std::map<uint32_t, std::vector<UIDContext>> &uidCtxs,
83 : std::vector<uint32_t> &netLayersVector)
84 : {
85 : // 将所有远端的rank都加入到状态维护map中
86 3 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
87 3 : CHK_PTR_NULL(hcclComm);
88 3 : hccl::CollComm* collComm = hcclComm->GetCollComm();
89 3 : CHK_PTR_NULL(collComm);
90 3 : Hccl::HcclCommunicator* commV2 = static_cast<Hccl::HcclCommunicator*>(collComm->GetCommunicatorV2());
91 3 : CHK_PTR_NULL(commV2); // 获取到legacy communicator，说明v2通信域
92 3 : void *rankGraphPtr = nullptr;
93 3 : CHK_RET(commV2->GetRankGraphV2(rankGraphPtr));
94 3 : CHK_PTR_NULL(rankGraphPtr);
95 1 : Hccl::RankGraph *rankGraph = static_cast<Hccl::RankGraph*>(rankGraphPtr);
96 :
97 : // 获取netLayer信息存入到netLayersVector中
98 1 : uint32_t *netLayers = nullptr;
99 1 : uint32_t netLayerNum = 0;
100 1 : CHK_RET(HcclRankGraphGetLayers(comm, &netLayers, &netLayerNum));
101 1 : if (netLayerNum == 0) {
102 0 : HCCL_WARNING("[%s] no netLayer in RankGraph", __func__);
103 0 : return HCCL_SUCCESS;
104 : }
105 1 : netLayersVector.assign(netLayers, netLayers + netLayerNum);
106 1 : std::sort(netLayersVector.begin(), netLayersVector.end());
107 1 : std::set<uint32_t> rankIdsSet; // 存放通信域的唯一标识ranks，防止在netLayer>=1的时候，查到了netLayer=0已经存放的ranks
108 2 : for (auto netLayer : netLayersVector) {
109 1 : GetRemEndpointDescsPerLayer(netLayer, comm, rankGraph, collComm, uidCtxs, rankIdsSet);
110 : }
111 :
112 1 : return HCCL_SUCCESS;
113 1 : }
114 :
115 0 : std::string ClusterMonitor::FormatConnTag(HcommSocketRole role,
116 : std::pair<ClusterUIDType, ClusterUIDType> uidPair) const
117 : {
118 0 : std::string tag;
119 0 : if (role == HcommSocketRole::HCOMM_SOCKET_ROLE_CLIENT) {
120 0 : tag = "HeartBeat_" + GetUID(uidPair.first) + "_to_" + GetUID(uidPair.second);
121 : } else {
122 0 : tag = "HeartBeat_" + GetUID(uidPair.second) + "_to_" + GetUID(uidPair.first);
123 : }
124 :
125 0 : return tag;
126 0 : }
127 :
128 2 : HcclResult ClusterMonitor::GetSocketDescFromRankInfo(HcclComm comm, uint32_t remoteRank,
129 : uint32_t netLayer, const ClusterUIDType &remoteUID, SocketDesc &socketDesc)
130 : {
131 2 : uint32_t rmtPort = 0;
132 2 : uint32_t listenPort = 0;
133 2 : hccl::CollComm* collComm = static_cast<hccl::hcclComm*>(comm)->GetCollComm();
134 2 : auto rankGraph = collComm->GetRankGraph();
135 2 : auto myRankId = collComm->GetMyRankId();
136 2 : CHK_PTR_NULL(rankGraph);
137 2 : CHK_RET(rankGraph->GetDevicePort(remoteRank, &rmtPort));
138 2 : if (rmtPort > Hccl::MAX_VALUE_TCPPORT) {
139 0 : HCCL_ERROR("[%s] Invalid port[%u] of Rank[%u]", __func__, rmtPort, remoteRank);
140 0 : return HCCL_E_PARA;
141 : }
142 2 : CommLink *links = nullptr;
143 2 : uint32_t linkNum = 0;
144 2 : HcclResult result = HcclRankGraphGetLinks(comm, netLayer, myRankId, remoteRank, &links, &linkNum);
145 2 : if (result != HCCL_SUCCESS) {
146 1 : HCCL_WARNING("[%s] Get links between myRank[%u] and remoteRank[%u] failed, ret:%d", __func__, myRankId, remoteRank, result);
147 1 : return HCCL_E_NOT_FOUND;
148 : }
149 : // 如果没有查询到任何链接，不报错，不把该link加入needConnectRank，直接返回成功
150 1 : if (linkNum == 0 || links[0].srcEndpointDesc.loc.locType == EndpointLocType::ENDPOINT_LOC_TYPE_HOST ||
151 0 : links[0].dstEndpointDesc.loc.locType == EndpointLocType::ENDPOINT_LOC_TYPE_HOST) {
152 1 : HCCL_INFO("[%s] no link between myRank[%u] and remoteRank[%u]", __func__, myRankId, remoteRank);
153 1 : return HCCL_E_NOT_FOUND;
154 : }
155 : // 查询该socket链接的server端监听的端口（监听方的选择策略需要跟SocketConfig中保持一致）
156 0 : Hccl::IpAddress localIpAddr{};
157 0 : Hccl::IpAddress remoteIpAddr{};
158 0 : CHK_RET(CommAddrToIpAddress(links[0].srcEndpointDesc.commAddr, localIpAddr));
159 0 : CHK_RET(CommAddrToIpAddress(links[0].dstEndpointDesc.commAddr, remoteIpAddr));
160 0 : if (localIpAddr < remoteIpAddr) { // local地址比remote地址小时，local作为server监听端
161 : // 查询localRankId对应的devPort
162 0 : CHK_RET(rankGraph->GetDevicePort(myRankId, &listenPort));
163 0 : socketDesc.role = HcommSocketRole::HCOMM_SOCKET_ROLE_SERVER;
164 0 : if (listenPort > Hccl::MAX_VALUE_TCPPORT) {
165 0 : HCCL_ERROR("[%s] Invalid port[%u] of Rank[%u]", __func__, listenPort, myRankId);
166 0 : return HCCL_E_PARA;
167 : }
168 0 : socketDesc.listenPort = static_cast<uint16_t>(listenPort); // socketDesc.port中填监听端口号
169 : } else {
170 0 : socketDesc.role = HcommSocketRole::HCOMM_SOCKET_ROLE_CLIENT;
171 0 : socketDesc.listenPort = static_cast<uint16_t>(rmtPort); // socketDesc.port中填对端端口号(此场景下对端端口号也就是监听端口号)
172 : }
173 : // socket建链需要心跳专用的tag，用来区分业务的socket以及心跳的sockt
174 0 : std::string tag = FormatConnTag(socketDesc.role, std::make_pair(myRankUID_, remoteUID));
175 0 : errno_t ret = memcpy_s(socketDesc.tag, sizeof(socketDesc.tag), tag.c_str(), tag.size());
176 0 : CHK_PRT_RET((ret != EOK),
177 : HCCL_ERROR("[%s] memcpy_s failed, errno:%d, error:%s", __func__, errno, strerror(errno)), HCCL_E_SYSCALL);
178 0 : socketDesc.localEndpoint = links[0].srcEndpointDesc;
179 0 : socketDesc.remoteEndpoint = links[0].dstEndpointDesc;
180 0 : return HCCL_SUCCESS;
181 0 : }
182 :
183 2 : HcclResult ClusterMonitor::InsertClusterMonitorCtx(HcclComm comm, UIDContext remoteCtx,
184 : std::map<ClusterUIDType, ClusterMonitorSocketCtx> &needConnectRank)
185 : {
186 2 : bool newConn = true;
187 2 : SocketDesc socketDesc{};
188 2 : auto remoteUID = remoteCtx.uid;
189 2 : auto remoteRank = remoteCtx.rankId;
190 2 : auto netLayer = remoteCtx.netLayer;
191 :
192 2 : std::unique_lock<std::mutex> lock(threadLock_);
193 2 : if (monitorLinkStatusMap_.find(remoteUID) == monitorLinkStatusMap_.end()) {
194 1 : monitorLinkStatusMap_[remoteUID] = MonitorLinkStatus::MONITOR_LINK_NOT_START;
195 2 : } else if (monitorLinkStatusMap_[remoteUID] == MonitorLinkStatus::MONITOR_LINK_BUILDING ||
196 1 : monitorLinkStatusMap_[remoteUID] == MonitorLinkStatus::MONITOR_LINK_COMPLETED) {
197 0 : newConn = false;// 说明之前已经有remoteUID在建链
198 : }
199 :
200 : // 获取端口号用来建链
201 2 : HcclResult ret = GetSocketDescFromRankInfo(comm, remoteRank, netLayer, remoteUID, socketDesc);
202 2 : if (ret == HCCL_E_NOT_FOUND) {
203 2 : return HCCL_SUCCESS; // 本rank无有效链接，不加入needConnectRank
204 : }
205 0 : ClusterMonitorSocketCtx ctx(socketDesc, newConn);
206 0 : needConnectRank.insert(std::make_pair(remoteUID, ctx));
207 0 : HCCL_INFO("[%s] InsertClusterMonitorCtx for myRankUID_[%s], remoteUID[%s], role[%s], localEndpoint[commAddr:0x%llx], "
208 : "remoteEndpoint[commAddr:0x%llx], tag[%s], listenPort [%u], newConn[%d]", __func__, GetUID(myRankUID_).c_str(), GetUID(remoteUID).c_str(),
209 : (socketDesc.role == HcommSocketRole::HCOMM_SOCKET_ROLE_SERVER) ? "SERVER" : "CLIENT",
210 : hcomm::logger::CommAddrLogger::ToString(socketDesc.localEndpoint.commAddr).c_str(),
211 : hcomm::logger::CommAddrLogger::ToString(socketDesc.remoteEndpoint.commAddr).c_str(),
212 : socketDesc.tag, socketDesc.listenPort, newConn);
213 0 : return HCCL_SUCCESS;
214 2 : }
215 :
216 2 : HcclResult ClusterMonitor::GetSamePlaneRank(HcclComm comm, std::vector<UIDContext> singlePlaneCtx,
217 : std::map<ClusterUIDType, ClusterMonitorSocketCtx> &needConnectRank)
218 : {
219 2 : uint32_t index = 0;
220 2 : for (; index < singlePlaneCtx.size();index++) {
221 2 : if (singlePlaneCtx[index].uid == this->myRankUID_) { // 找出myRank在vector中的下标
222 2 : break;
223 : }
224 : }
225 :
226 2 : uint32_t singlePlaneSize = singlePlaneCtx.size(); // 包含myRank自己，一个平面所有的节点
227 2 : if (singlePlaneSize <= 1) { // 待连接的节点个数为0或1，无需连接
228 0 : HCCL_INFO("[%s] no need to connect", __func__);
229 0 : return HCCL_SUCCESS;
230 2 : } else if (singlePlaneSize == 2) { // 待连接的节点个数为2，不需要双ring环，一条边就够了
231 1 : uint32_t nextIndex = (index + 1) % singlePlaneSize; // 算出与本Rank相连，对端的节点
232 1 : HCCL_INFO("[%s] singlePlaneSize is 2, only connect nextIndex[%u]", __func__, nextIndex);
233 1 : CHK_RET(InsertClusterMonitorCtx(comm, singlePlaneCtx[nextIndex], needConnectRank));
234 : } else {
235 1 : uint32_t nextIndex = (index + 1) % singlePlaneSize; // 算出与本Rank相连，右手的节点
236 1 : uint32_t preIndex = (index + singlePlaneSize - 1) % singlePlaneSize; // 算出与本Rank相连，左手或回绕环的节点
237 1 : HCCL_INFO("[%s] singlePlaneSize is %u, connect nextIndex[%u], preIndex[%u]", __func__, singlePlaneSize, nextIndex, preIndex);
238 1 : CHK_RET(InsertClusterMonitorCtx(comm, singlePlaneCtx[nextIndex], needConnectRank)); //以本rank为起点，环的右手
239 1 : CHK_RET(InsertClusterMonitorCtx(comm, singlePlaneCtx[preIndex], needConnectRank)); // 以本rank为起点，环的左手
240 : }
241 :
242 2 : return HCCL_SUCCESS;
243 : }
244 :
245 3 : HcclResult ClusterMonitor::GetConnectRank(HcclComm comm,
246 : std::map<ClusterUIDType, ClusterMonitorSocketCtx> &needConnectRank,
247 : std::map<uint32_t, std::vector<UIDContext>> uidCtxs, std::vector<uint32_t> &netLayersVector)
248 : {
249 3 : if (netLayersVector.empty() || uidCtxs.empty()) {
250 1 : HCCL_INFO("[%s] netLayersVector is empty, no netLayer in RankGraph", __func__);
251 1 : return HCCL_SUCCESS;
252 : }
253 :
254 2 : std::vector<UIDContext> layer0CommLinks; // 需要存入UIDContext，待后续查出对应的port/remoteUID
255 : // 先处理netLayer=0，按照netLayer=0全局唯一的localId升序排列，在level0不需要考虑host网卡的场景，host网卡只会在level1及以上的层级
256 2 : std::sort(uidCtxs[0].begin(), uidCtxs[0].end(), [&](const UIDContext& a, const UIDContext& b) {
257 4 : return a.localId < b.localId;
258 : });
259 6 : for (auto it = uidCtxs[0].begin(); it != uidCtxs[0].end(); ++it) {
260 4 : layer0CommLinks.push_back(*it); // netLayer为0
261 : }
262 :
263 : // 从layer=1开始，将commLinks存入vector中，找到所有与当前localId相同的节点
264 2 : std::vector<UIDContext> highLayerCommLinks;
265 7 : for (uint32_t netLayer : netLayersVector) {
266 16 : for (auto it = uidCtxs[netLayer].begin(); it != uidCtxs[netLayer].end(); ++it) {
267 11 : if (it->localId == this->myRankLocalId_) {
268 : // 在跨server、跨pod、跨超节点的场景，统一拿到local，打平处理为同一个平面，类似layer=0的情况
269 : // 由于A5上的devPhyId在64卡的场景下8个[0,7]，所以使用localId
270 6 : highLayerCommLinks.push_back(*it);
271 : }
272 : }
273 : }
274 2 : std::sort(highLayerCommLinks.begin(), highLayerCommLinks.end(), [&](const UIDContext& a, const UIDContext& b) {
275 8 : return a.netInstId < b.netInstId;
276 : });
277 :
278 : // 每个平面都分别成环
279 2 : CHK_RET(GetSamePlaneRank(comm, layer0CommLinks, needConnectRank));
280 2 : CHK_RET(GetSamePlaneRank(comm, highLayerCommLinks, needConnectRank));
281 2 : return HCCL_SUCCESS;
282 2 : }
283 :
284 550 : void ClusterMonitor::CreateHBLinksAsync()
285 : {
286 550 : std::unique_lock<std::mutex> linksLock(clusertMonitorLinkMtx_);
287 550 : if (clusterLinkContext_.empty()) {
288 529 : return;
289 : }
290 21 : linkThreadRunning_ = true;
291 21 : std::queue<std::tuple<std::string, ClusterUIDType, ClusterMonitorSocketCtx>> connInfoQueue;
292 63 : for (auto &pair : clusterLinkContext_) {
293 42 : const std::string &commId = pair.first;
294 42 : auto &commIdConnInfoQueue = pair.second;
295 45 : while (!commIdConnInfoQueue.empty()) {
296 3 : connInfoQueue.push(
297 6 : std::make_tuple(commId, commIdConnInfoQueue.front().first, commIdConnInfoQueue.front().second));
298 3 : commIdConnInfoQueue.pop();
299 : }
300 : }
301 21 : linksLock.unlock();
302 :
303 24 : while (!connInfoQueue.empty()) {
304 3 : const std::string commId = std::get<0>(connInfoQueue.front());
305 3 : const ClusterUIDType &remUID = std::get<1>(connInfoQueue.front());
306 3 : ClusterMonitorSocketCtx &connInfo = std::get<2>(connInfoQueue.front());
307 3 : connInfo.PrintSocketDesc("CreateHBLinksAsync");
308 3 : auto it = linkThreadMap_.find(remUID);
309 3 : if (it != linkThreadMap_.end() && it->second->joinable()) {
310 0 : it->second->join();
311 0 : HCCL_INFO("[CreateMonitorLinksAsync] monitor link thread has been joined. commId[%s], remote uid[%s].",
312 : commId.c_str(), GetUID(remUID).c_str());
313 : }
314 6 : linkThreadMap_[remUID].reset(
315 3 : new (std::nothrow) std::thread(&ClusterMonitor::CreateLinkWithRemotePonit, this, commId, remUID, connInfo));
316 3 : if (linkThreadMap_[remUID] == nullptr) {
317 0 : HCCL_RUN_WARNING("commId[%s] establish rank[%s] to rank[%s] heartbeat connection failed. Reason: "
318 : "create thread failed.",
319 : commId.c_str(), GetUID(myRankUID_).c_str(), GetUID(remUID).c_str());
320 : }
321 3 : connInfoQueue.pop();
322 3 : }
323 21 : return;
324 550 : }
325 :
326 3 : HcclResult ClusterMonitor::CreateTransportHandle(ClusterMonitorSocketCtx &info) const
327 : {
328 3 : info.PrintSocketDesc("CreateTransportHandle");
329 3 : if (info.socketHandler == nullptr) {
330 3 : return SocketCreate(&info.socketDesc, &info.socketHandler);
331 : }
332 :
333 0 : HCCL_WARNING("[CreateTransportHandle] socketHandler has been created, skip.");
334 0 : return HCCL_SUCCESS;
335 : }
336 :
337 4 : void ClusterMonitor::CreateLinkWithRemotePonit(
338 : std::string commId, ClusterUIDType rem, ClusterMonitorSocketCtx needConnectRank)
339 : {
340 : // 给当前线程添加名字
341 4 : const std::string threadName = "hb" + GetUID(rem);
342 4 : SetThreadName(threadName);
343 4 : hrtSetDevice(deviceLogicId_);
344 :
345 4 : HcclResult ret = CreateTransportHandle(needConnectRank);
346 4 : if (ret != HCCL_SUCCESS) {
347 0 : HCCL_RUN_WARNING("[CreateLinkWithRemote] CreateTransportHandle ret[%d], commId[%s], remote uid[%s].", ret,
348 : commId.c_str(), GetUID(rem).c_str());
349 0 : hrtResetDevice(deviceLogicId_);
350 0 : return;
351 : }
352 :
353 4 : auto createLinkTimeout = std::chrono::seconds(Hccl::EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
354 4 : auto startTime = std::chrono::steady_clock::now();
355 7 : while (linkThreadRunning_.load()) {
356 7 : if ((std::chrono::steady_clock::now() - startTime) >= createLinkTimeout) {
357 0 : HCCL_RUN_WARNING("establish rank[%s] to rank[%s] connection failed. Reason: link timeout,"
358 : "timeout[%llds], the HCCL_CONNECT_TIMEOUT may be insufficient. commId[%s].",
359 : GetUID(myRankUID_).c_str(), GetUID(rem).c_str(), createLinkTimeout.count(), commId.c_str());
360 4 : break;
361 : }
362 :
363 : SocketStates status;
364 7 : HcclResult ret = SocketGetStatus(needConnectRank.socketHandler, &status);
365 7 : if (ret != HCCL_SUCCESS) {
366 1 : HCCL_RUN_WARNING(
367 : "establish rank[%s] to rank[%s] connection failed. Reason: get socket status[%d] failed, commId[%s]",
368 : GetUID(myRankUID_).c_str(), GetUID(rem).c_str(), status, commId.c_str());
369 1 : SocketDestroy(needConnectRank.socketHandler);
370 1 : break;
371 : }
372 :
373 6 : if (status == SocketStates::SOCKET_TIMEOUT) {
374 0 : HCCL_RUN_WARNING(
375 : "establish rank[%s] to rank[%s] connection failed. Reason: get socket status timeout, commId[%s]",
376 : GetUID(myRankUID_).c_str(), GetUID(rem).c_str(), commId.c_str());
377 0 : SocketDestroy(needConnectRank.socketHandler);
378 0 : break;
379 6 : } else if (status == SocketStates::SOCKET_CONNECTING) {
380 3 : SalSleep(ONE_SECOND_OF_SLEEP);
381 3 : continue;
382 : }
383 :
384 3 : ret = OnConnectionEstablished(commId, rem, needConnectRank);
385 3 : if (ret != HCCL_SUCCESS) {
386 1 : HCCL_RUN_WARNING("OnConnectionEstablished not success, ret[%d]", ret);
387 : }
388 3 : break;
389 : }
390 4 : hrtResetDevice(deviceLogicId_);
391 :
392 4 : HCCL_INFO("[%s] Thread [%s] end...", __func__, threadName.c_str());
393 4 : return;
394 4 : }
395 :
396 3 : HcclResult ClusterMonitor::OnConnectionEstablished(
397 : const std::string &commId, const ClusterUIDType &rem, ClusterMonitorSocketCtx &needConnectRank)
398 : {
399 3 : std::unique_lock<std::mutex> lock(threadLock_);
400 3 : if (commIdMap_.find(commId) == commIdMap_.end()) {
401 1 : HCCL_RUN_WARNING(
402 : "establish rank[%s] to rank[%s] connection failed. Reason: commId[%s] has been Unregistered.",
403 : GetUID(myRankUID_).c_str(), GetUID(rem).c_str(), commId.c_str());
404 1 : SocketDestroy(needConnectRank.socketHandler);
405 1 : lock.unlock();
406 1 : return HCCL_E_INTERNAL;
407 : }
408 2 : needConnectRank.newConn = false;
409 2 : uid2SocketRefMap_.insert(rem, needConnectRank);
410 : // 心跳socket建链完成后，需要立即及激活其心跳收发能力
411 2 : auto frameSize = sizeof(ClusterMonitorFrame);
412 2 : if (uid2SocketRefMap_[rem].recvBuffer.Init(hccl::BASE_NUMBER * frameSize) != HCCL_SUCCESS) { // 2倍帧长，确保不会溢出
413 0 : HCCL_RUN_WARNING(
414 : "establish rank[%s] to rank[%s] connection failed. Reason: socket recv buffer init failed. commId[%s].",
415 : GetUID(myRankUID_).c_str(), GetUID(rem).c_str(), commId.c_str());
416 0 : SocketDestroy(needConnectRank.socketHandler);
417 0 : uid2SocketRefMap_.erase(rem);
418 0 : lock.unlock();
419 0 : return HCCL_E_INTERNAL;
420 : }
421 2 : monitorLinkStatusMap_[rem] = MonitorLinkStatus::MONITOR_LINK_COMPLETED;
422 2 : commIdMap_[commId][rem] = true; // 更新状态为已连接
423 2 : lock.unlock();
424 2 : HCCL_RUN_INFO("commId:[%s], establish rank[%s] to rank[%s] heartbeat connection success.", commId.c_str(),
425 : GetUID(myRankUID_).c_str(), GetUID(rem).c_str());
426 2 : return HCCL_SUCCESS;
427 3 : }
428 :
429 5 : HcclResult ClusterMonitor::SendFrameFromBuffer(ClusterUIDType &dst, ClusterMonitorFrame &cmFrame)
430 : {
431 10 : if (cmFrame.status != ClusterMonitorStatus::CLUSTER_MONITOR_OK
432 5 : && uid2SocketRefMap_[dst].sendBuffer.size() < hccl::MAX_SENDBUFF_SIZE) {
433 5 : uid2SocketRefMap_[dst].sendBuffer.push(cmFrame);
434 : }
435 5 : while (uid2SocketRefMap_[dst].sendBuffer.size() > 0) {
436 5 : ClusterMonitorFrame hbf = uid2SocketRefMap_[dst].sendBuffer.front();
437 5 : u64 sendDis = sizeof(ClusterMonitorFrame) - uid2SocketRefMap_[dst].restSize;
438 5 : uint64_t compSize = 0;
439 5 : HcclResult ret = SocketSendNb(uid2SocketRefMap_[dst].socketHandler,
440 5 : (reinterpret_cast<std::byte *>(&hbf) + sendDis), uid2SocketRefMap_[dst].restSize, &compSize);
441 5 : if (ret != HCCL_SUCCESS) {
442 2 : HCCL_WARNING("[CreateTransportHandle] SocketSendNb failed, ret[%d]", ret);
443 2 : return ret;
444 : }
445 3 : if (uid2SocketRefMap_[dst].restSize == compSize) {
446 0 : uid2SocketRefMap_[dst].sendBuffer.pop();
447 0 : uid2SocketRefMap_[dst].restSize = sizeof(ClusterMonitorFrame);
448 0 : HCCL_DEBUG("[Heartbeat][SendFrame] Send Success, from [%s] to [%s] about [%s] by [%s] status[%d]",
449 : GetUID(myRankUID_).c_str(), GetUID(dst).c_str(), GetUID(cmFrame.crimer).c_str(),
450 : GetUID(cmFrame.informer).c_str(), cmFrame.status);
451 : } else {
452 3 : uid2SocketRefMap_[dst].restSize = uid2SocketRefMap_[dst].restSize - compSize;
453 3 : break;
454 : }
455 : }
456 3 : return HCCL_SUCCESS;
457 : }
458 :
459 10 : HcclResult ClusterMonitor::SendFrame(
460 : ClusterUIDType &dst, ClusterUIDType &crimer, ClusterUIDType &informer, ClusterMonitorStatus status)
461 : {
462 10 : ClusterMonitorFrame cmFrame(myRankUID_, dst, crimer, informer, status);
463 10 : if (uid2SocketRefMap_[dst].sendBuffer.size() > 0) {
464 5 : HcclResult ret = SendFrameFromBuffer(dst, cmFrame);
465 5 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_WARNING("[SendFrameFromBuffer] failed, ret[%d]", ret), ret);
466 : } else {
467 5 : uint64_t compSize = 0;
468 5 : uint64_t expectSize = sizeof(ClusterMonitorFrame);
469 10 : HcclResult ret = SocketSendNb(
470 5 : uid2SocketRefMap_[dst].socketHandler, &cmFrame, expectSize, &compSize);
471 5 : if (ret != HCCL_SUCCESS) {
472 0 : HCCL_WARNING("[CreateTransportHandle] SocketSendNb failed, ret[%d]", ret);
473 0 : return ret;
474 : }
475 5 : if (compSize == expectSize) {
476 2 : HCCL_DEBUG("[Heartbeat][SendFrame] Send Success, from [%s] to [%s] about [%s] by [%s] status[%d]",
477 : GetUID(myRankUID_).c_str(), GetUID(dst).c_str(), GetUID(crimer).c_str(), GetUID(informer).c_str(), status);
478 : } else {
479 3 : HCCL_DEBUG("[Heartbeat][SendFrame] Send Not Complete, from [%s] to [%s] about [%s] by [%s] status[%d], \
480 : expectSize[%u], compSize[%u]",
481 : GetUID(myRankUID_).c_str(), GetUID(dst).c_str(), GetUID(crimer).c_str(), GetUID(informer).c_str(), status,
482 : expectSize, compSize);
483 3 : uid2SocketRefMap_[dst].restSize = expectSize - compSize;
484 3 : uid2SocketRefMap_[dst].sendBuffer.push(cmFrame);
485 : }
486 : }
487 8 : return HCCL_SUCCESS;
488 : }
489 :
490 2 : HcclResult ClusterMonitor::RecvFrame(ClusterUIDType rem)
491 : {
492 2 : ClusterMonitorFrame cmFrame;
493 2 : u64 compSize = 0;
494 2 : u64 expectSize = sizeof(ClusterMonitorFrame);
495 : // 此处while循环用于最大限度的从socket中读取数据，直到没有数据可读或者发生错误。
496 : // 因为心跳帧较小，理论上一次recv就能读完。但为了兼容可能存在的粘包情况，增加循环读取的逻辑。
497 : while (true) {
498 2 : compSize = 0;
499 4 : HcclResult ret = SocketRecvNb(
500 2 : uid2SocketRefMap_[rem].socketHandler, &cmFrame, expectSize, (reinterpret_cast<uint64_t *>(&compSize)));
501 2 : if (ret == HCCL_SUCCESS && compSize > 0) {
502 0 : uid2SocketRefMap_[rem].recvBuffer.PushSeg(reinterpret_cast<u8 *>(&cmFrame), compSize);
503 0 : if (uid2SocketRefMap_[rem].recvBuffer.Size() >= expectSize) {
504 0 : uid2SocketRefMap_[rem].recvBuffer.GetSeg(reinterpret_cast<u8 *>(&cmFrame), expectSize);
505 0 : uid2SocketRefMap_[rem].recvBuffer.PopSeg(expectSize);
506 0 : CHK_RET(ParseFrame(cmFrame, rem));
507 : }
508 2 : } else if (ret == HCCL_E_INTERNAL) {
509 0 : HCCL_WARNING("SocketRecvNb recv rem[%s] fail", GetUID(rem).c_str());
510 0 : return ret;
511 : } else {
512 : // 当没有数据可读时，SocketRecvNb会返回成功但compSize为0，此时退出循环，继续进行后续的心跳发送和异常处理等逻辑
513 2 : break;
514 : }
515 0 : }
516 2 : return HCCL_SUCCESS;
517 : }
518 :
519 0 : HcclResult ClusterMonitor::ParseFrame(ClusterMonitorFrame &cmFrame, ClusterUIDType &src)
520 : {
521 0 : if (cmFrame.src != src || cmFrame.dst != myRankUID_) {
522 0 : HCCL_WARNING("rank[%s] recv wrong frame", GetUID(myRankUID_).c_str());
523 0 : return HCCL_E_INTERNAL;
524 : }
525 :
526 0 : HCCL_DEBUG("[ClusterMonitor][ParseMonitorFrame] Recv Success, from [%s] to [%s] about [%s] by [%s] state[%d]",
527 : GetUID(cmFrame.src).c_str(), GetUID(cmFrame.dst).c_str(), GetUID(cmFrame.crimer).c_str(),
528 : GetUID(cmFrame.informer).c_str(), cmFrame.status);
529 :
530 : // 能够收到进程卡住表示心跳是正常的
531 0 : if (cmFrame.status == ClusterMonitorStatus::CLUSTER_MONITOR_OK) {
532 0 : uid2SocketRefMap_[src].lostNum = 0;
533 : }
534 :
535 : // 只有心跳非正常时才需要打印TRACE
536 0 : if (cmFrame.status != ClusterMonitorStatus::CLUSTER_MONITOR_OK) {
537 0 : SetStatus(cmFrame.crimer, cmFrame.informer, cmFrame.status); // 设置异常状态
538 : }
539 :
540 0 : return HCCL_SUCCESS;
541 : }
542 :
543 576 : void ClusterMonitor::DelErrorSocket()
544 : {
545 576 : for (auto rem : errorSocket_) {
546 0 : HCCL_RUN_INFO("rank[%s] Try to Send/recv HeartBeat to rank[%s]", GetUID(myRankUID_).c_str(),
547 : GetUID(rem).c_str());
548 0 : uid2FrameStatusMap_.erase(rem);
549 0 : if (uid2SocketRefMap_.has(rem)) {
550 0 : SocketDestroy(uid2SocketRefMap_[rem].socketHandler);
551 0 : while (uid2SocketRefMap_.erase(rem) != 0) {
552 : };
553 : }
554 : }
555 576 : errorSocket_.clear();
556 576 : }
557 :
558 :
559 1 : void ClusterMonitor::SetStatus(ClusterUIDType &crimer, ClusterUIDType &informer,
560 : ClusterMonitorStatus status, bool needBroadcast)
561 : {
562 1 : if (uid2FrameStatusMap_[crimer].status != status) {
563 1 : uid2FrameStatusMap_[crimer].informer = informer;
564 1 : uid2FrameStatusMap_[crimer].status = status;
565 1 : uid2FrameStatusMap_[crimer].needBroadcast = needBroadcast;
566 1 : if (needBroadcast) {
567 1 : errRankQueue_.push(crimer);
568 : }
569 :
570 1 : errStatusQueue_.push(ClusterMonitorFrame(crimer, informer, status, TIME_NOW(), std::chrono::system_clock::now()));
571 1 : if (errStatusQueue_.size() > hccl::EVENT_MAX_CNT) {
572 0 : errStatusQueue_.pop();
573 : }
574 1 : HCCL_RUN_INFO("[%s][%s]local rank [%s]: crimer rank [%s] status[%s] by informer rank [%s]",
575 : LOG_KEYWORDS_TASK_EXEC.c_str(), LOG_KEYWORDS_HEARTBEAT_EVETN.c_str(), GetUID(myRankUID_).c_str(),
576 : GetUID(crimer).c_str(), GetClusterMonitorStatusStr(status).c_str(), GetUID(informer).c_str());
577 : }
578 1 : }
579 :
580 1 : HcclResult ClusterMonitor::ProcessConnectRanks(const std::string &commId,
581 : std::map<ClusterUIDType, ClusterMonitorSocketCtx> &needConnectRank)
582 : {
583 : // 将双ring环的pair放入clusterLinkContext_管理多个通信域
584 1 : std::unique_lock<std::mutex> linkCtxlock(clusertMonitorLinkMtx_);
585 1 : for (auto &item : needConnectRank) {
586 0 : if (item.second.newConn == true) {
587 : // 一旦放入clusterLinkContext_中，就会被后台的异步建链线程推动建链
588 0 : clusterLinkContext_[commId].push(std::move(item));
589 : }
590 : }
591 1 : linkCtxlock.unlock();
592 :
593 1 : std::unique_lock<std::mutex> lock(threadLock_);
594 1 : for (auto &item : needConnectRank) {
595 0 : if (item.second.newConn == true) {
596 : // 由于newConn==true的item已经入队，后台推动异步建链，所以状态迁移为建链中
597 0 : monitorLinkStatusMap_[item.first] = MonitorLinkStatus::MONITOR_LINK_BUILDING;
598 0 : } else if (commIdMap_[commId].find(item.first) == commIdMap_[commId].end() ||
599 0 : (commIdMap_[commId].count(item.first) && !commIdMap_[commId][item.first])) {
600 : // 若newConn=false，说明不是新增的连接
601 : // 1. 通信域找不到，2.通信域内能找到但还没有连接，计数++
602 0 : uid2SocketRefMap_.ref(item.first);
603 0 : HCCL_RUN_INFO("commId:[%s], establish rank[%s] to rank[%s] heartbeat connection success.", commId.c_str(),
604 : GetUID(myRankUID_).c_str(), GetUID(item.first).c_str());
605 0 : commIdMap_[commId][item.first] = true; // 认为通信域中对应的连接已经建立
606 : }
607 : }
608 1 : lock.unlock();
609 :
610 1 : return HCCL_SUCCESS;
611 1 : }
612 :
613 2 : void ClusterMonitor::MonitorThread()
614 : {
615 : // 给当前线程添加名字
616 2 : SetThreadName("Hccl_HeartBeat");
617 :
618 2 : hrtSetDevice(deviceLogicId_);
619 2 : HcclResult ret = HCCL_SUCCESS;
620 2 : uint32_t count = 0;
621 550 : while (clusterMonitorThreadFlag_) {
622 549 : CreateHBLinksAsync(); // 内部起线程对所有的socket进行异步建链
623 549 : threadLock_.lock();
624 549 : count++;
625 549 : if (count >= hccl::HEARTBEAT_COUNT) {
626 27 : count = 0;
627 30 : for (auto iter = uid2SocketRefMap_.begin(); iter != uid2SocketRefMap_.end(); iter++) {
628 3 : ClusterUIDType rem = iter->first;
629 3 : uid2SocketRefMap_[rem].lostNum++;
630 : // 先发送心跳帧，触发对端回复，才能准确地判断链路状态
631 3 : ret = SendFrame(rem, myRankUID_, myRankUID_, ClusterMonitorStatus::CLUSTER_MONITOR_OK);
632 3 : ret == HCCL_E_INTERNAL ? errorSocket_.push_back(rem) : void(0);
633 : }
634 27 : DelErrorSocket(); // 处理socket错误的句柄
635 : }
636 :
637 609 : for (auto iter = uid2SocketRefMap_.begin(); iter != uid2SocketRefMap_.end(); iter++) {
638 60 : ClusterUIDType rem = iter->first;
639 60 : ret = RecvFrame(rem);
640 60 : if (ret == HCCL_E_INTERNAL) {
641 0 : errorSocket_.push_back(rem);
642 60 : } else if (uid2SocketRefMap_[rem].lostNum >= lostThreshold_) {
643 0 : SetStatus(rem, myRankUID_, ClusterMonitorStatus::CLUSTER_MONITOR_LOST);
644 : }
645 : }
646 549 : DelErrorSocket(); // 处理socket错误的句柄
647 549 : ProcessExceptionEvent(); // 处理error cqe
648 549 : threadLock_.unlock();
649 :
650 549 : std::this_thread::sleep_for(std::chrono::milliseconds(hccl::BROADCAST_INTERVAL));
651 : }
652 :
653 1 : linkThreadRunning_ = false;
654 : // 在心跳进程结束之前join所有的建链线程
655 4 : for (auto &pair : linkThreadMap_) {
656 3 : if (pair.second != nullptr && pair.second->joinable()) {
657 0 : pair.second->join();
658 0 : HCCL_INFO("[%s] thread has joined. Remote uid is [%s]", __func__, GetUID(pair.first).c_str());
659 : }
660 : }
661 :
662 1 : hrtResetDevice(deviceLogicId_);
663 1 : }
664 :
665 2 : HcclResult ClusterMonitor::RunMonitorThread()
666 : {
667 2 : HCCL_INFO("[%s] Start ClusterMonitorThread.", __func__);
668 2 : clusterMonitorThreadFlag_ = true;
669 2 : clusterMonitorThread_.reset(new (std::nothrow) std::thread(&ClusterMonitor::MonitorThread, this));
670 2 : CHK_SMART_PTR_NULL(clusterMonitorThread_);
671 2 : lostThreshold_ = hccl::HCCL_LOST_THRESHOLD; // 心跳丢失阈值为30s
672 2 : initialized_ = true;
673 2 : isDeInit_ = false;
674 2 : return HCCL_SUCCESS;
675 : }
676 :
677 3 : HcclResult ClusterMonitor::RegisterToClusterMonitor(HcclComm comm)
678 : {
679 3 : HCCL_INFO("[%s] RegisterToClusterMonitor begin.", __func__);
680 3 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PTR);
681 3 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
682 3 : CHK_PTR_NULL(hcclComm);
683 3 : hccl::CollComm* collComm = hcclComm->GetCollComm();
684 3 : CHK_PTR_NULL(collComm);
685 3 : deviceLogicId_ = collComm->GetDeviceLogicId();
686 :
687 : // 单rank无对端，不支持心跳检测
688 3 : const std::string &commId = collComm->GetCommId();
689 3 : uint32_t rankSize = collComm->GetRankSize();
690 3 : CHK_PRT_RET(rankSize == 1,
691 : HCCL_WARNING("[%s] commId[%s] rankSize[%u] no need to register to ClusterMonitor",
692 : __func__, commId.c_str(), rankSize), HCCL_SUCCESS);
693 :
694 : // 判断该通信域是否曾经添加到commIdMap_中
695 3 : std::unique_lock<std::mutex> lock(threadLock_);
696 3 : auto iter = commIdMap_.find(commId);
697 3 : if (iter != commIdMap_.end()) {
698 0 : HCCL_INFO("commId[%s] has Registered, skip.", commId.c_str());
699 0 : return HCCL_SUCCESS;
700 : }
701 :
702 3 : if (!initialized_) {
703 : // 开始起监控线程
704 1 : CHK_RET(RunMonitorThread());
705 : }
706 3 : lock.unlock();
707 :
708 : // 存放所有节点的上下文
709 3 : std::map<uint32_t, std::vector<UIDContext>> uidCtxs;
710 3 : std::vector<uint32_t> netLayersVector;
711 :
712 : // 获取从myRank出发，所有的对端，并维护commIdMap_及uid2FrameStatusMap_
713 3 : lock.lock();
714 3 : CHK_RET(GetRemEndpointDescs(comm, uidCtxs, netLayersVector));
715 1 : lock.unlock();
716 :
717 : // 解析heartbeat环境变量，如果配置为off则不去注册对应的rank
718 1 : auto clusterHeartBeatEnable = Hccl::EnvConfig::GetInstance().GetLogConfig().GetDfsConfig().clusterHeartBeatEnable;
719 1 : if (!clusterHeartBeatEnable) {
720 0 : HCCL_RUN_INFO("[%s] HCCL_DFS_CONFIG cluster_heartbeat is off. It's unnecessary to "
721 : "register Ranks. commId[%s]", __func__, commId.c_str());
722 0 : return HCCL_SUCCESS;
723 : }
724 :
725 : // 从所有连接中，选择双ring环，存放到needConnectRank
726 1 : std::map<ClusterUIDType, ClusterMonitorSocketCtx> needConnectRank;
727 1 : CHK_RET(GetConnectRank(comm, needConnectRank, uidCtxs, netLayersVector));
728 :
729 : // 处理双ring环的连接（入队、更新状态、更新引用计数等）
730 1 : CHK_RET(ProcessConnectRanks(commId, needConnectRank));
731 :
732 1 : HCCL_INFO("[%s] commId[%s] RegisterRanks Completed", __func__, commId.c_str());
733 1 : return HCCL_SUCCESS;
734 3 : }
735 :
736 2667 : HcclResult ClusterMonitor::DeInit()
737 : {
738 2667 : if (isDeInit_) {
739 1 : HCCL_INFO("[%s] already deinit, skip.", __func__);
740 1 : return HCCL_SUCCESS;
741 : }
742 2666 : isDeInit_ = true;
743 2666 : HCCL_INFO("[%s] heartbeat deinit begin.", __func__);
744 2666 : clusterMonitorThreadFlag_ = false;
745 2666 : linkThreadRunning_ = false;
746 :
747 2666 : if (clusterMonitorThread_) {
748 1 : if (clusterMonitorThread_->joinable()) {
749 0 : clusterMonitorThread_->join();
750 : }
751 : }
752 : {
753 2666 : std::unique_lock<std::mutex> lock(threadLock_);
754 2668 : for (SocketHandle handler : pendingDestroySockets_) {
755 2 : if (handler == nullptr) {
756 0 : continue;
757 : }
758 2 : HcclResult ret = SocketDestroy(handler);
759 2 : if (ret != HCCL_SUCCESS) {
760 0 : HCCL_WARNING("[DeInit] pending SocketDestroy failed, ret[%d]", ret);
761 : }
762 : }
763 2666 : pendingDestroySockets_.clear();
764 :
765 2667 : for (auto iter = uid2SocketRefMap_.begin(); iter != uid2SocketRefMap_.end(); iter++) {
766 1 : HcclResult ret = SocketDestroy(iter->second.socketHandler);
767 1 : if (ret != HCCL_SUCCESS) {
768 0 : HCCL_WARNING("[DeInit] SocketDestroy failed, ret[%d]", ret);
769 : }
770 : }
771 2666 : uid2SocketRefMap_.clear();
772 2666 : uid2FrameStatusMap_.clear();
773 2666 : }
774 2666 : std::queue<ClusterMonitorFrame> empty;
775 2666 : std::swap(errStatusQueue_, empty);
776 :
777 2666 : initialized_ = false;
778 2666 : HCCL_INFO("[%s] heartbeat deinit end.", __func__);
779 2666 : return HCCL_SUCCESS;
780 2666 : }
781 :
782 27 : void ClusterMonitor::ClearClusterLinkContext(const std::string &commId, std::set<ClusterUIDType> &remInQueue)
783 : {
784 27 : std::unique_lock<std::mutex> linkCtxlock(clusertMonitorLinkMtx_);
785 27 : auto ctxIter = clusterLinkContext_.find(commId);
786 27 : if (ctxIter != clusterLinkContext_.end()) {
787 1 : while (!ctxIter->second.empty()) {
788 0 : remInQueue.insert(ctxIter->second.front().first); // uid出队存入set中
789 0 : ctxIter->second.pop();
790 : }
791 : }
792 27 : clusterLinkContext_.erase(commId);
793 27 : }
794 :
795 27 : bool ClusterMonitor::UnregisterCommIdFromMaps(const std::string &commId, const std::set<ClusterUIDType> &remInQueue)
796 : {
797 27 : std::unique_lock<std::mutex> lock(threadLock_);
798 :
799 27 : for (const auto &rem : remInQueue) {
800 0 : if (monitorLinkStatusMap_[rem] == MonitorLinkStatus::MONITOR_LINK_BUILDING) {
801 0 : monitorLinkStatusMap_[rem] = MonitorLinkStatus::MONITOR_LINK_NOT_START;
802 0 : HCCL_INFO("[%s] commId[%s] rem[%s] is in clusterLinkContext_ deque. Status change to not start", __func__,
803 : commId.c_str(), GetUID(rem).c_str());
804 : }
805 : }
806 27 : auto iter = commIdMap_.find(commId);
807 27 : if (iter == commIdMap_.end()) {
808 26 : HCCL_INFO("commId[%s] hasn't Registered, skip", commId.c_str());
809 26 : return false;
810 : }
811 :
812 5 : for (const auto& remRank : commIdMap_[commId]) {
813 4 : ClusterUIDType rem = remRank.first;
814 4 : uid2FrameStatusMap_.erase(rem);
815 4 : if (remRank.second) {
816 3 : if (uid2SocketRefMap_.count(rem) == 1) {
817 : // 不在此处 SocketDestroy；摘入 pending，等 DeInit join 后再销毁
818 2 : SocketHandle handler = uid2SocketRefMap_[rem].socketHandler;
819 2 : if (handler != nullptr) {
820 2 : pendingDestroySockets_.push_back(handler);
821 : }
822 2 : monitorLinkStatusMap_[rem] = MonitorLinkStatus::MONITOR_LINK_NOT_START;
823 : }
824 3 : HCCL_INFO("[%s]commId[%s] socket erase remote:%s", __func__, commId.c_str(), GetUID(rem).c_str());
825 3 : uid2SocketRefMap_.erase(rem);
826 : }
827 4 : HCCL_INFO("[%s]commId[%s] status erase remote:%s", __func__, commId.c_str(), GetUID(rem).c_str());
828 : }
829 1 : commIdMap_.erase(iter);
830 1 : HCCL_INFO("[%s]commId[%s] UnregisterRanks Completed.", __func__, commId.c_str());
831 1 : return true;
832 27 : }
833 :
834 124 : HcclResult ClusterMonitor::UnRegisterToClusterMonitor(const hccl::CollComm* collComm)
835 : {
836 124 : CHK_PRT_RET(initialized_ == false, HCCL_WARNING("Heartbeat has been destroyed, or not initialized"), HCCL_SUCCESS);
837 27 : const std::string &commId = collComm->GetCommId();
838 27 : std::set<ClusterUIDType> remInQueue;
839 27 : ClearClusterLinkContext(commId, remInQueue);
840 27 : if (!UnregisterCommIdFromMaps(commId, remInQueue)) {
841 26 : return HCCL_SUCCESS;
842 : }
843 1 : if (commIdMap_.size() == 0) {
844 1 : HCCL_RUN_INFO("[%s]Entry HeartBeat DeInit.", __func__);
845 1 : CHK_RET(DeInit());
846 : }
847 1 : return HCCL_SUCCESS;
848 27 : }
849 :
850 551 : void ClusterMonitor::ProcessExceptionEvent()
851 : {
852 553 : while (errRankQueue_.size() > 0) {
853 2 : ClusterUIDType cur = errRankQueue_.front();
854 2 : uid2FrameStatusMap_[cur].needBroadcast = false;
855 8 : for (auto iterRem = uid2SocketRefMap_.begin(); iterRem != uid2SocketRefMap_.end(); iterRem++) {
856 6 : ClusterUIDType rem = iterRem->first;
857 12 : if (rem != uid2FrameStatusMap_[cur].informer &&
858 6 : uid2FrameStatusMap_[rem].status == ClusterMonitorStatus::CLUSTER_MONITOR_OK) {
859 6 : (void)SendFrame(rem, cur, uid2FrameStatusMap_[cur].informer, uid2FrameStatusMap_[cur].status);
860 : }
861 : }
862 2 : errRankQueue_.pop();
863 : }
864 551 : return;
865 : }
866 :
867 : constexpr u32 BASE_YEAR = 1900;
868 0 : void GetCqeErrInfoFromTaskException(unsigned int remoteLocalId, unsigned int locDeviceId, unsigned short int status,
869 : std::string localEid, std::string remoteEid, std::string remoteInsId)
870 : {
871 0 : if (!Hccl::EnvConfig::GetInstance().GetLogConfig().GetDfsConfig().clusterHeartBeatEnable) {
872 0 : HCCL_RUN_INFO("[%s] HCCL_DFS_CONFIG cluster_heartbeat is off. It's unnecessary to "
873 : "get cqe error info.", __func__);
874 0 : return;
875 : }
876 0 : return hccl::CollCommMgr::GetInstance()->GetClusterMonitor(locDeviceId).GetCqeErrInfoFromTaskException(remoteLocalId,
877 0 : status, localEid, remoteEid, remoteInsId);
878 : }
879 :
880 1 : void ClusterMonitor::GetCqeErrInfoFromTaskException(u32 remoteLocalId, uint16_t status, std::string localEid, std::string remoteEid, std::string remoteInsId)
881 : {
882 1 : cqeErrInfo_.cqeRemoteLocalId = remoteLocalId;
883 1 : cqeErrInfo_.cqeStatus = status;
884 1 : cqeErrInfo_.cqeLocalEid = localEid;
885 1 : cqeErrInfo_.cqeRemoteEid = remoteEid;
886 1 : cqeErrInfo_.cqeRemoteInsId = remoteInsId;
887 1 : ClusterUIDCxt remoteUIDcxt(remoteInsId, remoteLocalId);
888 1 : ClusterUIDType localUID = myRankUID_;
889 1 : ClusterUIDType remoteUID = FormatUID(remoteUIDcxt);
890 1 : SetStatus(localUID, remoteUID, ClusterMonitorStatus::CLUSTER_MONITOR_CQE_ERR, true);
891 1 : time_t tmpt = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
892 1 : auto duration_us = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::system_clock::now().time_since_epoch());
893 : // 提取总微秒数
894 1 : auto total_us = duration_us.count();
895 : // 分离秒和微秒部分
896 1 : auto microseconds = total_us % 1000000;
897 1 : struct tm *now = localtime(&tmpt);
898 : char errorLinkLogBuffer[LOG_TMPBUF_SIZE];
899 :
900 1 : s32 stringRet = snprintf_s(errorLinkLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE- 1U,
901 : "localInfo{local instanceId[%s], LocalId[%u], localEid[%s]}, remoteInfo{remote instanceId[%s], remoteLocalId[%u], remoteEid[%s]}",
902 : myRankNetInstId_.c_str(), myRankLocalId_, cqeErrInfo_.cqeLocalEid.c_str(), cqeErrInfo_.cqeRemoteInsId.c_str(), cqeErrInfo_.cqeRemoteLocalId,
903 : cqeErrInfo_.cqeRemoteEid.c_str());
904 1 : CHK_PRT_CONT( stringRet < 0, HCCL_ERROR("[ClusterMonitor][GetCqeErrInfoFromTaskException]snprintf error when log cqe error info") );
905 :
906 1 : if (now == nullptr) {
907 0 : HCCL_ERROR("[%s][%s][%s]localtime fail, cqe error status[%u], %s", LOG_KEYWORDS_TASK_EXEC.c_str(), LOG_KEYWORDS_HEARTBEAT_EVETN.c_str(), LOG_KEYWORDS_CQE_ERROR.c_str(), cqeErrInfo_.cqeStatus, errorLinkLogBuffer);
908 : } else {
909 1 : HCCL_ERROR("[%s][%s][%s]cqe error status[%u], time:[%04u-%02d-%02d %02d:%0d:%02d.%06u], %s", LOG_KEYWORDS_TASK_EXEC.c_str(), LOG_KEYWORDS_HEARTBEAT_EVETN.c_str(), LOG_KEYWORDS_CQE_ERROR.c_str(),
910 : cqeErrInfo_.cqeStatus, now->tm_year + BASE_YEAR, now->tm_mon + 1, now->tm_mday, now->tm_hour,
911 : now->tm_min, now->tm_sec, microseconds, errorLinkLogBuffer);
912 : }
913 2 : return;
914 1 : }
915 :
916 :
917 35 : void ClusterMonitor::MakeErrMsg(std::queue<ClusterMonitorFrame> &keyEvents, std::vector<std::string> &errStatusVec) const
918 : {
919 41 : while (keyEvents.size() > 0) {
920 6 : auto &tmp = keyEvents.front();
921 6 : std::string crimerStr = GetUID(tmp.crimer);
922 6 : std::string informerStr = GetUID(tmp.informer);
923 :
924 12 : std::string headStr = "[" + LOG_KEYWORDS_TASK_EXEC + "][" + LOG_KEYWORDS_HEARTBEAT_EVETN + "]" +
925 6 : "Cluster Exception Location[IP/ID]:[";
926 :
927 6 : time_t tm = std::chrono::system_clock::to_time_t(tmp.TOASystem);
928 6 : std::string timeStr(ctime(&tm));
929 6 : if (!timeStr.empty()) { // ctime()函数自带换行符，需要去掉
930 6 : timeStr.pop_back();
931 : }
932 6 : timeStr = ", Arrival Time:[" + timeStr + "]";
933 :
934 12 : std::string errStr = ", ExceptionType:";
935 6 : std::string reasonStr = ", Possible Reason:";
936 6 : switch (tmp.status) {
937 2 : case ClusterMonitorStatus::CLUSTER_MONITOR_LOST:
938 2 : errStr = errStr + "[Heartbeat Lost Occurred]";
939 2 : reasonStr = reasonStr + "1. Process has exited, 2. Network Disconnected";
940 : errStr =
941 2 : headStr + crimerStr + "]" + timeStr + ", Discoverer:[" + informerStr + "]" + errStr + reasonStr;
942 2 : break;
943 4 : case ClusterMonitorStatus::CLUSTER_MONITOR_CQE_ERR:
944 4 : errStr = errStr + "[Error cqe Occurred]";
945 4 : reasonStr = reasonStr + "1.Network Disconnected, 2.Remote Rank Coredown";
946 4 : errStr = headStr + crimerStr + "]" + timeStr + errStr + reasonStr;
947 4 : break;
948 0 : default:
949 0 : errStr = " Unknown";
950 : }
951 6 : errStatusVec.emplace_back(errStr);
952 6 : keyEvents.pop();
953 6 : }
954 35 : }
955 :
956 16 : std::vector<std::string> ClusterMonitor::PrintEvents(std::map<ClusterMonitorStatus, std::queue<ClusterMonitorFrame>> &keyEvents) const
957 : {
958 16 : std::vector<std::string> errStatusVec;
959 : // 打印优先级 opretry not support > error cqe > stuck > lost
960 16 : MakeErrMsg(keyEvents[ClusterMonitorStatus::CLUSTER_MONITOR_CQE_ERR], errStatusVec);
961 16 : MakeErrMsg(keyEvents[ClusterMonitorStatus::CLUSTER_MONITOR_LOST], errStatusVec);
962 16 : return errStatusVec;
963 0 : }
964 :
965 14 : std::vector<std::string> ClusterMonitor::GetErrStatusVecFromCluserMonitor()
966 : {
967 14 : std::unique_lock<std::mutex> lock(threadLock_);
968 14 : std::map<ClusterMonitorStatus, std::queue<ClusterMonitorFrame>> keyEvents;
969 16 : while (errStatusQueue_.size() > 0) {
970 2 : auto &tmp = errStatusQueue_.front();
971 2 : keyEvents[tmp.status].push(tmp);
972 2 : errStatusQueue_.pop();
973 : }
974 28 : return PrintEvents(keyEvents);
975 14 : }
976 :
977 12 : std::vector<std::string> GetErrStatusVecFromCluserMonitor(s32 deviceLogicID)
978 : {
979 12 : return hccl::CollCommMgr::GetInstance()->GetClusterMonitor(deviceLogicID).GetErrStatusVecFromCluserMonitor();
980 : }
981 :
982 41 : __attribute__((constructor)) void ClusterMonitorCallBackInit()
983 : {
984 41 : hcomm::RegisterGetAicpuCqeErrInfoCallBackHcomm(GetCqeErrInfoFromTaskException);
985 41 : hcomm::RegisterGetCcuCqeErrInfoCallBackHcomm(GetCqeErrInfoFromTaskException);
986 41 : hcomm::RegisterAicpuGetErrStatusVecCallBack(GetErrStatusVecFromCluserMonitor);
987 41 : hcomm::RegisterCcuGetErrStatusVecCallBack(GetErrStatusVecFromCluserMonitor);
988 41 : }
989 :
990 :
991 : } // namespace hcomm
|
