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 2796 : ClusterMonitor::~ClusterMonitor()
26 : {
27 2796 : DeInit();
28 2796 : }
29 :
30 20 : ClusterUIDType ClusterMonitor::FormatUID(ClusterUIDCxt cxt) const
31 : {
32 20 : ClusterUIDType uid{};
33 : // 构造唯一的uid: netInstanceId + local_id
34 20 : (void)snprintf_s(uid.id, sizeof(uid.id), sizeof(uid.id) - 1, "%s/%s",
35 40 : cxt.netInstId.c_str(), std::to_string(cxt.localId).c_str());
36 :
37 20 : return uid;
38 : }
39 :
40 64 : std::string ClusterMonitor::GetUID(const ClusterUIDType &uid) const
41 : {
42 128 : 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 2 : std::string ClusterMonitor::FormatConnTag(HcommSocketRole role,
116 : std::pair<ClusterUIDType, ClusterUIDType> uidPair) const
117 : {
118 2 : std::string tag;
119 2 : if (role == HcommSocketRole::HCOMM_SOCKET_ROLE_CLIENT) {
120 1 : tag = "HeartBeat_" + GetUID(uidPair.first) + "_to_" + GetUID(uidPair.second);
121 : } else {
122 1 : tag = "HeartBeat_" + GetUID(uidPair.second) + "_to_" + GetUID(uidPair.first);
123 : }
124 :
125 2 : 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() + 1);
176 0 : CHK_PRT_RET((ret != EOK),
177 : HCCL_ERROR("[%s] memcpy_s failed, ret:%d, errno:%d, error:%s", __func__, ret, 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:%s], "
208 : "remoteEndpoint[commAddr:%s], 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 44 : void ClusterMonitor::CreateHBLinksAsync()
285 : {
286 44 : std::unique_lock<std::mutex> linksLock(clusertMonitorLinkMtx_);
287 44 : if (clusterLinkContext_.empty()) {
288 23 : 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 44 : }
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 6 : 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 : void* sendPtr = static_cast<char*>(static_cast<void*>(&hbf)) + sendDis;
440 : HcclResult ret
441 5 : = SocketSendNb(uid2SocketRefMap_[dst].socketHandler, sendPtr, uid2SocketRefMap_[dst].restSize, &compSize);
442 5 : if (ret != HCCL_SUCCESS) {
443 2 : HCCL_WARNING("[CreateTransportHandle] SocketSendNb failed, ret[%d]", ret);
444 2 : return ret;
445 : }
446 3 : if (uid2SocketRefMap_[dst].restSize == compSize) {
447 0 : uid2SocketRefMap_[dst].sendBuffer.pop();
448 0 : uid2SocketRefMap_[dst].restSize = sizeof(ClusterMonitorFrame);
449 0 : HCCL_DEBUG("[Heartbeat][SendFrame] Send Success, from [%s] to [%s] about [%s] by [%s] status[%d]",
450 : GetUID(myRankUID_).c_str(), GetUID(dst).c_str(), GetUID(cmFrame.crimer).c_str(),
451 : GetUID(cmFrame.informer).c_str(), cmFrame.status);
452 : } else {
453 3 : uid2SocketRefMap_[dst].restSize = uid2SocketRefMap_[dst].restSize - compSize;
454 3 : break;
455 : }
456 : }
457 3 : return HCCL_SUCCESS;
458 : }
459 :
460 10 : HcclResult ClusterMonitor::SendFrame(
461 : ClusterUIDType &dst, ClusterUIDType &crimer, ClusterUIDType &informer, ClusterMonitorStatus status)
462 : {
463 10 : ClusterMonitorFrame cmFrame(myRankUID_, dst, crimer, informer, status);
464 10 : if (uid2SocketRefMap_[dst].sendBuffer.size() > 0) {
465 5 : HcclResult ret = SendFrameFromBuffer(dst, cmFrame);
466 5 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_WARNING("[SendFrameFromBuffer] failed, ret[%d]", ret), ret);
467 : } else {
468 5 : uint64_t compSize = 0;
469 5 : uint64_t expectSize = sizeof(ClusterMonitorFrame);
470 10 : HcclResult ret = SocketSendNb(
471 5 : uid2SocketRefMap_[dst].socketHandler, &cmFrame, expectSize, &compSize);
472 5 : if (ret != HCCL_SUCCESS) {
473 0 : HCCL_WARNING("[CreateTransportHandle] SocketSendNb failed, ret[%d]", ret);
474 0 : return ret;
475 : }
476 5 : if (compSize == expectSize) {
477 2 : HCCL_DEBUG("[Heartbeat][SendFrame] Send Success, from [%s] to [%s] about [%s] by [%s] status[%d]",
478 : GetUID(myRankUID_).c_str(), GetUID(dst).c_str(), GetUID(crimer).c_str(), GetUID(informer).c_str(), status);
479 : } else {
480 3 : HCCL_DEBUG("[Heartbeat][SendFrame] Send Not Complete, from [%s] to [%s] about [%s] by [%s] status[%d], \
481 : expectSize[%llu], compSize[%llu]",
482 : GetUID(myRankUID_).c_str(), GetUID(dst).c_str(), GetUID(crimer).c_str(), GetUID(informer).c_str(), status,
483 : expectSize, compSize);
484 3 : uid2SocketRefMap_[dst].restSize = expectSize - compSize;
485 3 : uid2SocketRefMap_[dst].sendBuffer.push(cmFrame);
486 : }
487 : }
488 8 : return HCCL_SUCCESS;
489 : }
490 :
491 2 : HcclResult ClusterMonitor::RecvFrame(ClusterUIDType rem)
492 : {
493 2 : ClusterMonitorFrame cmFrame;
494 2 : u64 compSize = 0;
495 2 : u64 expectSize = sizeof(ClusterMonitorFrame);
496 : // 此处while循环用于最大限度的从socket中读取数据，直到没有数据可读或者发生错误。
497 : // 因为心跳帧较小，理论上一次recv就能读完。但为了兼容可能存在的粘包情况，增加循环读取的逻辑。
498 : while (true) {
499 2 : compSize = 0;
500 4 : HcclResult ret = SocketRecvNb(
501 2 : uid2SocketRefMap_[rem].socketHandler, &cmFrame, expectSize, (reinterpret_cast<uint64_t *>(&compSize)));
502 2 : if (ret == HCCL_SUCCESS && compSize > 0) {
503 0 : uid2SocketRefMap_[rem].recvBuffer.PushSeg(reinterpret_cast<u8 *>(&cmFrame), compSize);
504 0 : if (uid2SocketRefMap_[rem].recvBuffer.Size() >= expectSize) {
505 0 : uid2SocketRefMap_[rem].recvBuffer.GetSeg(reinterpret_cast<u8 *>(&cmFrame), expectSize);
506 0 : uid2SocketRefMap_[rem].recvBuffer.PopSeg(expectSize);
507 0 : CHK_RET(ParseFrame(cmFrame, rem));
508 : }
509 2 : } else if (ret == HCCL_E_INTERNAL) {
510 0 : HCCL_WARNING("SocketRecvNb recv rem[%s] fail", GetUID(rem).c_str());
511 0 : return ret;
512 : } else {
513 : // 当没有数据可读时，SocketRecvNb会返回成功但compSize为0，此时退出循环，继续进行后续的心跳发送和异常处理等逻辑
514 2 : break;
515 : }
516 0 : }
517 2 : return HCCL_SUCCESS;
518 : }
519 :
520 0 : HcclResult ClusterMonitor::ParseFrame(ClusterMonitorFrame &cmFrame, ClusterUIDType &src)
521 : {
522 0 : if (cmFrame.src != src || cmFrame.dst != myRankUID_) {
523 0 : HCCL_WARNING("rank[%s] recv wrong frame", GetUID(myRankUID_).c_str());
524 0 : return HCCL_E_INTERNAL;
525 : }
526 :
527 0 : HCCL_DEBUG("[ClusterMonitor][ParseMonitorFrame] Recv Success, from [%s] to [%s] about [%s] by [%s] state[%d]",
528 : GetUID(cmFrame.src).c_str(), GetUID(cmFrame.dst).c_str(), GetUID(cmFrame.crimer).c_str(),
529 : GetUID(cmFrame.informer).c_str(), cmFrame.status);
530 :
531 : // 能够收到进程卡住表示心跳是正常的
532 0 : if (cmFrame.status == ClusterMonitorStatus::CLUSTER_MONITOR_OK) {
533 0 : uid2SocketRefMap_[src].lostNum = 0;
534 : }
535 :
536 : // 只有心跳非正常时才需要打印TRACE
537 0 : if (cmFrame.status != ClusterMonitorStatus::CLUSTER_MONITOR_OK) {
538 0 : SetStatus(cmFrame.crimer, cmFrame.informer, cmFrame.status); // 设置异常状态
539 : }
540 :
541 0 : return HCCL_SUCCESS;
542 : }
543 :
544 45 : void ClusterMonitor::DelErrorSocket()
545 : {
546 45 : for (auto rem : errorSocket_) {
547 0 : HCCL_RUN_INFO("rank[%s] Try to Send/recv HeartBeat to rank[%s]", GetUID(myRankUID_).c_str(),
548 : GetUID(rem).c_str());
549 0 : uid2FrameStatusMap_.erase(rem);
550 0 : if (uid2SocketRefMap_.has(rem)) {
551 0 : SocketDestroy(uid2SocketRefMap_[rem].socketHandler);
552 0 : while (uid2SocketRefMap_.erase(rem) != 0) {
553 : };
554 : }
555 : }
556 45 : errorSocket_.clear();
557 45 : }
558 :
559 :
560 1 : void ClusterMonitor::SetStatus(ClusterUIDType &crimer, ClusterUIDType &informer,
561 : ClusterMonitorStatus status, bool needBroadcast)
562 : {
563 1 : if (uid2FrameStatusMap_[crimer].status != status) {
564 1 : uid2FrameStatusMap_[crimer].informer = informer;
565 1 : uid2FrameStatusMap_[crimer].status = status;
566 1 : uid2FrameStatusMap_[crimer].needBroadcast = needBroadcast;
567 1 : if (needBroadcast) {
568 1 : errRankQueue_.push(crimer);
569 : }
570 :
571 1 : errStatusQueue_.push(ClusterMonitorFrame(crimer, informer, status, TIME_NOW(), std::chrono::system_clock::now()));
572 1 : if (errStatusQueue_.size() > hccl::EVENT_MAX_CNT) {
573 0 : errStatusQueue_.pop();
574 : }
575 1 : HCCL_RUN_INFO("[%s][%s]local rank [%s]: crimer rank [%s] status[%s] by informer rank [%s]",
576 : LOG_KEYWORDS_TASK_EXEC.c_str(), LOG_KEYWORDS_HEARTBEAT_EVETN.c_str(), GetUID(myRankUID_).c_str(),
577 : GetUID(crimer).c_str(), GetClusterMonitorStatusStr(status).c_str(), GetUID(informer).c_str());
578 : }
579 1 : }
580 :
581 1 : HcclResult ClusterMonitor::ProcessConnectRanks(const std::string &commId,
582 : std::map<ClusterUIDType, ClusterMonitorSocketCtx> &needConnectRank)
583 : {
584 : // 将双ring环的pair放入clusterLinkContext_管理多个通信域
585 1 : std::unique_lock<std::mutex> linkCtxlock(clusertMonitorLinkMtx_);
586 1 : for (auto &item : needConnectRank) {
587 0 : if (item.second.newConn == true) {
588 : // 一旦放入clusterLinkContext_中，就会被后台的异步建链线程推动建链
589 0 : clusterLinkContext_[commId].push(std::move(item));
590 : }
591 : }
592 1 : linkCtxlock.unlock();
593 :
594 1 : std::unique_lock<std::mutex> lock(threadLock_);
595 1 : for (auto &item : needConnectRank) {
596 0 : if (item.second.newConn == true) {
597 : // 由于newConn==true的item已经入队，后台推动异步建链，所以状态迁移为建链中
598 0 : monitorLinkStatusMap_[item.first] = MonitorLinkStatus::MONITOR_LINK_BUILDING;
599 0 : } else if (commIdMap_[commId].find(item.first) == commIdMap_[commId].end() ||
600 0 : (commIdMap_[commId].count(item.first) && !commIdMap_[commId][item.first])) {
601 : // 若newConn=false，说明不是新增的连接
602 : // 1. 通信域找不到，2.通信域内能找到但还没有连接，计数++
603 0 : uid2SocketRefMap_.ref(item.first);
604 0 : HCCL_RUN_INFO("commId:[%s], establish rank[%s] to rank[%s] heartbeat connection success.", commId.c_str(),
605 : GetUID(myRankUID_).c_str(), GetUID(item.first).c_str());
606 0 : commIdMap_[commId][item.first] = true; // 认为通信域中对应的连接已经建立
607 : }
608 : }
609 1 : lock.unlock();
610 :
611 1 : return HCCL_SUCCESS;
612 1 : }
613 :
614 2 : void ClusterMonitor::MonitorThread()
615 : {
616 : // 给当前线程添加名字
617 2 : SetThreadName("Hccl_HeartBeat");
618 :
619 2 : hrtSetDevice(deviceLogicId_);
620 2 : HcclResult ret = HCCL_SUCCESS;
621 2 : uint32_t count = 0;
622 44 : while (clusterMonitorThreadFlag_) {
623 43 : CreateHBLinksAsync(); // 内部起线程对所有的socket进行异步建链
624 43 : threadLock_.lock();
625 43 : count++;
626 43 : if (count >= hccl::HEARTBEAT_COUNT) {
627 2 : count = 0;
628 5 : for (auto iter = uid2SocketRefMap_.begin(); iter != uid2SocketRefMap_.end(); iter++) {
629 3 : ClusterUIDType rem = iter->first;
630 3 : uid2SocketRefMap_[rem].lostNum++;
631 : // 先发送心跳帧，触发对端回复，才能准确地判断链路状态
632 3 : ret = SendFrame(rem, myRankUID_, myRankUID_, ClusterMonitorStatus::CLUSTER_MONITOR_OK);
633 3 : ret == HCCL_E_INTERNAL ? errorSocket_.push_back(rem) : void(0);
634 : }
635 2 : DelErrorSocket(); // 处理socket错误的句柄
636 : }
637 :
638 103 : for (auto iter = uid2SocketRefMap_.begin(); iter != uid2SocketRefMap_.end(); iter++) {
639 60 : ClusterUIDType rem = iter->first;
640 60 : ret = RecvFrame(rem);
641 60 : if (ret == HCCL_E_INTERNAL) {
642 0 : errorSocket_.push_back(rem);
643 60 : } else if (uid2SocketRefMap_[rem].lostNum >= lostThreshold_) {
644 0 : SetStatus(rem, myRankUID_, ClusterMonitorStatus::CLUSTER_MONITOR_LOST);
645 : }
646 : }
647 43 : DelErrorSocket(); // 处理socket错误的句柄
648 43 : ProcessExceptionEvent(); // 处理error cqe
649 43 : threadLock_.unlock();
650 :
651 43 : std::this_thread::sleep_for(std::chrono::milliseconds(hccl::BROADCAST_INTERVAL));
652 : }
653 :
654 1 : linkThreadRunning_ = false;
655 : // 在心跳进程结束之前join所有的建链线程
656 4 : for (auto &pair : linkThreadMap_) {
657 3 : if (pair.second != nullptr && pair.second->joinable()) {
658 0 : pair.second->join();
659 0 : HCCL_INFO("[%s] thread has joined. Remote uid is [%s]", __func__, GetUID(pair.first).c_str());
660 : }
661 : }
662 :
663 1 : hrtResetDevice(deviceLogicId_);
664 1 : }
665 :
666 2 : HcclResult ClusterMonitor::RunMonitorThread()
667 : {
668 2 : HCCL_INFO("[%s] Start ClusterMonitorThread.", __func__);
669 2 : clusterMonitorThreadFlag_ = true;
670 2 : clusterMonitorThread_.reset(new (std::nothrow) std::thread(&ClusterMonitor::MonitorThread, this));
671 2 : CHK_SMART_PTR_NULL(clusterMonitorThread_);
672 2 : lostThreshold_ = hccl::HCCL_LOST_THRESHOLD; // 心跳丢失阈值为30s
673 2 : initialized_ = true;
674 2 : isDeInit_ = false;
675 2 : return HCCL_SUCCESS;
676 : }
677 :
678 3 : HcclResult ClusterMonitor::RegisterToClusterMonitor(HcclComm comm)
679 : {
680 3 : HCCL_INFO("[%s] RegisterToClusterMonitor begin.", __func__);
681 3 : CHK_PRT_RET(comm == nullptr, HCCL_ERROR("[%s] comm is null", __func__), HCCL_E_PTR);
682 3 : auto* hcclComm = static_cast<hccl::hcclComm*>(comm);
683 3 : CHK_PTR_NULL(hcclComm);
684 3 : hccl::CollComm* collComm = hcclComm->GetCollComm();
685 3 : CHK_PTR_NULL(collComm);
686 3 : deviceLogicId_ = collComm->GetDeviceLogicId();
687 :
688 : // 单rank无对端，不支持心跳检测
689 3 : const std::string &commId = collComm->GetCommId();
690 3 : uint32_t rankSize = collComm->GetRankSize();
691 3 : CHK_PRT_RET(rankSize == 1,
692 : HCCL_WARNING("[%s] commId[%s] rankSize[%u] no need to register to ClusterMonitor",
693 : __func__, commId.c_str(), rankSize), HCCL_SUCCESS);
694 :
695 : // 判断该通信域是否曾经添加到commIdMap_中
696 3 : std::unique_lock<std::mutex> lock(threadLock_);
697 3 : auto iter = commIdMap_.find(commId);
698 3 : if (iter != commIdMap_.end()) {
699 0 : HCCL_INFO("commId[%s] has Registered, skip.", commId.c_str());
700 0 : return HCCL_SUCCESS;
701 : }
702 :
703 3 : if (!initialized_) {
704 : // 开始起监控线程
705 1 : CHK_RET(RunMonitorThread());
706 : }
707 3 : lock.unlock();
708 :
709 : // 存放所有节点的上下文
710 3 : std::map<uint32_t, std::vector<UIDContext>> uidCtxs;
711 3 : std::vector<uint32_t> netLayersVector;
712 :
713 : // 获取从myRank出发，所有的对端，并维护commIdMap_及uid2FrameStatusMap_
714 3 : lock.lock();
715 3 : CHK_RET(GetRemEndpointDescs(comm, uidCtxs, netLayersVector));
716 1 : lock.unlock();
717 :
718 : // 解析heartbeat环境变量，如果配置为off则不去注册对应的rank
719 1 : auto clusterHeartBeatEnable = Hccl::EnvConfig::GetInstance().GetLogConfig().GetDfsConfig().clusterHeartBeatEnable;
720 1 : if (!clusterHeartBeatEnable) {
721 0 : HCCL_RUN_INFO("[%s] HCCL_DFS_CONFIG cluster_heartbeat is off. It's unnecessary to "
722 : "register Ranks. commId[%s]", __func__, commId.c_str());
723 0 : return HCCL_SUCCESS;
724 : }
725 :
726 : // 从所有连接中，选择双ring环，存放到needConnectRank
727 1 : std::map<ClusterUIDType, ClusterMonitorSocketCtx> needConnectRank;
728 1 : CHK_RET(GetConnectRank(comm, needConnectRank, uidCtxs, netLayersVector));
729 :
730 : // 处理双ring环的连接（入队、更新状态、更新引用计数等）
731 1 : CHK_RET(ProcessConnectRanks(commId, needConnectRank));
732 :
733 1 : HCCL_INFO("[%s] commId[%s] RegisterRanks Completed", __func__, commId.c_str());
734 1 : return HCCL_SUCCESS;
735 3 : }
736 :
737 2797 : HcclResult ClusterMonitor::DeInit()
738 : {
739 2797 : if (isDeInit_) {
740 1 : HCCL_INFO("[%s] already deinit, skip.", __func__);
741 1 : return HCCL_SUCCESS;
742 : }
743 2796 : isDeInit_ = true;
744 2796 : HCCL_INFO("[%s] heartbeat deinit begin.", __func__);
745 2796 : clusterMonitorThreadFlag_ = false;
746 2796 : linkThreadRunning_ = false;
747 :
748 2796 : if (clusterMonitorThread_) {
749 1 : if (clusterMonitorThread_->joinable()) {
750 0 : clusterMonitorThread_->join();
751 : }
752 : }
753 : {
754 2796 : std::unique_lock<std::mutex> lock(threadLock_);
755 2798 : for (SocketHandle handler : pendingDestroySockets_) {
756 2 : if (handler == nullptr) {
757 0 : continue;
758 : }
759 2 : HcclResult ret = SocketDestroy(handler);
760 2 : if (ret != HCCL_SUCCESS) {
761 0 : HCCL_WARNING("[DeInit] pending SocketDestroy failed, ret[%d]", ret);
762 : }
763 : }
764 2796 : pendingDestroySockets_.clear();
765 :
766 2797 : for (auto iter = uid2SocketRefMap_.begin(); iter != uid2SocketRefMap_.end(); iter++) {
767 1 : HcclResult ret = SocketDestroy(iter->second.socketHandler);
768 1 : if (ret != HCCL_SUCCESS) {
769 0 : HCCL_WARNING("[DeInit] SocketDestroy failed, ret[%d]", ret);
770 : }
771 : }
772 2796 : uid2SocketRefMap_.clear();
773 2796 : uid2FrameStatusMap_.clear();
774 2796 : }
775 2796 : std::queue<ClusterMonitorFrame> empty;
776 2796 : std::swap(errStatusQueue_, empty);
777 :
778 2796 : initialized_ = false;
779 2796 : HCCL_INFO("[%s] heartbeat deinit end.", __func__);
780 2796 : return HCCL_SUCCESS;
781 2796 : }
782 :
783 28 : void ClusterMonitor::ClearClusterLinkContext(const std::string &commId, std::set<ClusterUIDType> &remInQueue)
784 : {
785 28 : std::unique_lock<std::mutex> linkCtxlock(clusertMonitorLinkMtx_);
786 28 : auto ctxIter = clusterLinkContext_.find(commId);
787 28 : if (ctxIter != clusterLinkContext_.end()) {
788 1 : while (!ctxIter->second.empty()) {
789 0 : remInQueue.insert(ctxIter->second.front().first); // uid出队存入set中
790 0 : ctxIter->second.pop();
791 : }
792 : }
793 28 : clusterLinkContext_.erase(commId);
794 28 : }
795 :
796 28 : bool ClusterMonitor::UnregisterCommIdFromMaps(const std::string &commId, const std::set<ClusterUIDType> &remInQueue)
797 : {
798 28 : std::unique_lock<std::mutex> lock(threadLock_);
799 :
800 28 : for (const auto &rem : remInQueue) {
801 0 : if (monitorLinkStatusMap_[rem] == MonitorLinkStatus::MONITOR_LINK_BUILDING) {
802 0 : monitorLinkStatusMap_[rem] = MonitorLinkStatus::MONITOR_LINK_NOT_START;
803 0 : HCCL_INFO("[%s] commId[%s] rem[%s] is in clusterLinkContext_ deque. Status change to not start", __func__,
804 : commId.c_str(), GetUID(rem).c_str());
805 : }
806 : }
807 28 : auto iter = commIdMap_.find(commId);
808 28 : if (iter == commIdMap_.end()) {
809 27 : HCCL_INFO("commId[%s] hasn't Registered, skip", commId.c_str());
810 27 : return false;
811 : }
812 :
813 5 : for (const auto& remRank : commIdMap_[commId]) {
814 4 : ClusterUIDType rem = remRank.first;
815 4 : uid2FrameStatusMap_.erase(rem);
816 4 : if (remRank.second) {
817 3 : if (uid2SocketRefMap_.count(rem) == 1) {
818 : // 不在此处 SocketDestroy；摘入 pending，等 DeInit join 后再销毁
819 2 : SocketHandle handler = uid2SocketRefMap_[rem].socketHandler;
820 2 : if (handler != nullptr) {
821 2 : pendingDestroySockets_.push_back(handler);
822 : }
823 2 : monitorLinkStatusMap_[rem] = MonitorLinkStatus::MONITOR_LINK_NOT_START;
824 : }
825 3 : HCCL_INFO("[%s]commId[%s] socket erase remote:%s", __func__, commId.c_str(), GetUID(rem).c_str());
826 3 : uid2SocketRefMap_.erase(rem);
827 : }
828 4 : HCCL_INFO("[%s]commId[%s] status erase remote:%s", __func__, commId.c_str(), GetUID(rem).c_str());
829 : }
830 1 : commIdMap_.erase(iter);
831 1 : HCCL_INFO("[%s]commId[%s] UnregisterRanks Completed.", __func__, commId.c_str());
832 1 : return true;
833 28 : }
834 :
835 135 : HcclResult ClusterMonitor::UnRegisterToClusterMonitor(const hccl::CollComm* collComm)
836 : {
837 135 : CHK_PRT_RET(initialized_ == false, HCCL_WARNING("Heartbeat has been destroyed, or not initialized"), HCCL_SUCCESS);
838 28 : const std::string &commId = collComm->GetCommId();
839 28 : std::set<ClusterUIDType> remInQueue;
840 28 : ClearClusterLinkContext(commId, remInQueue);
841 28 : if (!UnregisterCommIdFromMaps(commId, remInQueue)) {
842 27 : return HCCL_SUCCESS;
843 : }
844 1 : if (commIdMap_.size() == 0) {
845 1 : HCCL_RUN_INFO("[%s]Entry HeartBeat DeInit.", __func__);
846 1 : CHK_RET(DeInit());
847 : }
848 1 : return HCCL_SUCCESS;
849 28 : }
850 :
851 45 : void ClusterMonitor::ProcessExceptionEvent()
852 : {
853 47 : while (errRankQueue_.size() > 0) {
854 2 : ClusterUIDType cur = errRankQueue_.front();
855 2 : uid2FrameStatusMap_[cur].needBroadcast = false;
856 8 : for (auto iterRem = uid2SocketRefMap_.begin(); iterRem != uid2SocketRefMap_.end(); iterRem++) {
857 6 : ClusterUIDType rem = iterRem->first;
858 12 : if (rem != uid2FrameStatusMap_[cur].informer &&
859 6 : uid2FrameStatusMap_[rem].status == ClusterMonitorStatus::CLUSTER_MONITOR_OK) {
860 6 : (void)SendFrame(rem, cur, uid2FrameStatusMap_[cur].informer, uid2FrameStatusMap_[cur].status);
861 : }
862 : }
863 2 : errRankQueue_.pop();
864 : }
865 45 : return;
866 : }
867 :
868 : constexpr u32 BASE_YEAR = 1900;
869 0 : void GetCqeErrInfoFromTaskException(unsigned int remoteLocalId, unsigned int locDeviceId, unsigned short int status,
870 : std::string localEid, std::string remoteEid, std::string remoteInsId)
871 : {
872 0 : if (!Hccl::EnvConfig::GetInstance().GetLogConfig().GetDfsConfig().clusterHeartBeatEnable) {
873 0 : HCCL_RUN_INFO("[%s] HCCL_DFS_CONFIG cluster_heartbeat is off. It's unnecessary to "
874 : "get cqe error info.", __func__);
875 0 : return;
876 : }
877 0 : return hccl::CollCommMgr::GetInstance()->GetClusterMonitor(locDeviceId).GetCqeErrInfoFromTaskException(remoteLocalId,
878 0 : status, localEid, remoteEid, remoteInsId);
879 : }
880 :
881 1 : void ClusterMonitor::GetCqeErrInfoFromTaskException(u32 remoteLocalId, uint16_t status, std::string localEid, std::string remoteEid, std::string remoteInsId)
882 : {
883 1 : cqeErrInfo_.cqeRemoteLocalId = remoteLocalId;
884 1 : cqeErrInfo_.cqeStatus = status;
885 1 : cqeErrInfo_.cqeLocalEid = localEid;
886 1 : cqeErrInfo_.cqeRemoteEid = remoteEid;
887 1 : cqeErrInfo_.cqeRemoteInsId = remoteInsId;
888 1 : ClusterUIDCxt remoteUIDcxt(remoteInsId, remoteLocalId);
889 1 : ClusterUIDType localUID = myRankUID_;
890 1 : ClusterUIDType remoteUID = FormatUID(remoteUIDcxt);
891 1 : SetStatus(localUID, remoteUID, ClusterMonitorStatus::CLUSTER_MONITOR_CQE_ERR, true);
892 1 : time_t tmpt = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
893 1 : auto duration_us = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::system_clock::now().time_since_epoch());
894 : // 提取总微秒数
895 1 : auto total_us = duration_us.count();
896 : // 分离秒和微秒部分
897 1 : auto microseconds = total_us % 1000000;
898 1 : struct tm *now = localtime(&tmpt);
899 : char errorLinkLogBuffer[LOG_TMPBUF_SIZE];
900 :
901 1 : s32 stringRet = snprintf_s(errorLinkLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE- 1U,
902 : "localInfo{local instanceId[%s], LocalId[%u], localEid[%s]}, remoteInfo{remote instanceId[%s], remoteLocalId[%u], remoteEid[%s]}",
903 : myRankNetInstId_.c_str(), myRankLocalId_, cqeErrInfo_.cqeLocalEid.c_str(), cqeErrInfo_.cqeRemoteInsId.c_str(), cqeErrInfo_.cqeRemoteLocalId,
904 : cqeErrInfo_.cqeRemoteEid.c_str());
905 1 : CHK_PRT_CONT( stringRet < 0, HCCL_ERROR("[ClusterMonitor][GetCqeErrInfoFromTaskException]snprintf error when log cqe error info") );
906 :
907 1 : if (now == nullptr) {
908 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);
909 : } else {
910 1 : HCCL_ERROR("[%s][%s][%s]cqe error status[%u], time:[%04d-%02d-%02d %02d:%02d:%02d.%06lld], %s", LOG_KEYWORDS_TASK_EXEC.c_str(), LOG_KEYWORDS_HEARTBEAT_EVETN.c_str(), LOG_KEYWORDS_CQE_ERROR.c_str(),
911 : cqeErrInfo_.cqeStatus, now->tm_year + BASE_YEAR, now->tm_mon + 1, now->tm_mday, now->tm_hour,
912 : now->tm_min, now->tm_sec, microseconds, errorLinkLogBuffer);
913 : }
914 2 : return;
915 1 : }
916 :
917 :
918 35 : void ClusterMonitor::MakeErrMsg(std::queue<ClusterMonitorFrame> &keyEvents, std::vector<std::string> &errStatusVec) const
919 : {
920 41 : while (keyEvents.size() > 0) {
921 6 : auto &tmp = keyEvents.front();
922 6 : std::string crimerStr = GetUID(tmp.crimer);
923 6 : std::string informerStr = GetUID(tmp.informer);
924 :
925 12 : std::string headStr = "[" + LOG_KEYWORDS_TASK_EXEC + "][" + LOG_KEYWORDS_HEARTBEAT_EVETN + "]" +
926 6 : "Cluster Exception Location[IP/ID]:[";
927 :
928 6 : time_t tm = std::chrono::system_clock::to_time_t(tmp.TOASystem);
929 6 : std::string timeStr(ctime(&tm));
930 6 : if (!timeStr.empty()) { // ctime()函数自带换行符，需要去掉
931 6 : timeStr.pop_back();
932 : }
933 6 : timeStr = ", Arrival Time:[" + timeStr + "]";
934 :
935 12 : std::string errStr = ", ExceptionType:";
936 6 : std::string reasonStr = ", Possible Reason:";
937 6 : switch (tmp.status) {
938 2 : case ClusterMonitorStatus::CLUSTER_MONITOR_LOST:
939 2 : errStr = errStr + "[Heartbeat Lost Occurred]";
940 2 : reasonStr = reasonStr + "1. Process has exited, 2. Network Disconnected";
941 : errStr =
942 2 : headStr + crimerStr + "]" + timeStr + ", Discoverer:[" + informerStr + "]" + errStr + reasonStr;
943 2 : break;
944 4 : case ClusterMonitorStatus::CLUSTER_MONITOR_CQE_ERR:
945 4 : errStr = errStr + "[Error cqe Occurred]";
946 4 : reasonStr = reasonStr + "1.Network Disconnected, 2.Remote Rank Coredown";
947 4 : errStr = headStr + crimerStr + "]" + timeStr + errStr + reasonStr;
948 4 : break;
949 0 : default:
950 0 : errStr = " Unknown";
951 : }
952 6 : errStatusVec.emplace_back(errStr);
953 6 : keyEvents.pop();
954 6 : }
955 35 : }
956 :
957 16 : std::vector<std::string> ClusterMonitor::PrintEvents(std::map<ClusterMonitorStatus, std::queue<ClusterMonitorFrame>> &keyEvents) const
958 : {
959 16 : std::vector<std::string> errStatusVec;
960 : // 打印优先级 opretry not support > error cqe > stuck > lost
961 16 : MakeErrMsg(keyEvents[ClusterMonitorStatus::CLUSTER_MONITOR_CQE_ERR], errStatusVec);
962 16 : MakeErrMsg(keyEvents[ClusterMonitorStatus::CLUSTER_MONITOR_LOST], errStatusVec);
963 16 : return errStatusVec;
964 0 : }
965 :
966 14 : std::vector<std::string> ClusterMonitor::GetErrStatusVecFromCluserMonitor()
967 : {
968 14 : std::unique_lock<std::mutex> lock(threadLock_);
969 14 : std::map<ClusterMonitorStatus, std::queue<ClusterMonitorFrame>> keyEvents;
970 16 : while (errStatusQueue_.size() > 0) {
971 2 : auto &tmp = errStatusQueue_.front();
972 2 : keyEvents[tmp.status].push(tmp);
973 2 : errStatusQueue_.pop();
974 : }
975 28 : return PrintEvents(keyEvents);
976 14 : }
977 :
978 12 : std::vector<std::string> GetErrStatusVecFromCluserMonitor(s32 deviceLogicID)
979 : {
980 12 : return hccl::CollCommMgr::GetInstance()->GetClusterMonitor(deviceLogicID).GetErrStatusVecFromCluserMonitor();
981 : }
982 :
983 43 : __attribute__((constructor)) void ClusterMonitorCallBackInit()
984 : {
985 43 : hcomm::RegisterGetAicpuCqeErrInfoCallBackHcomm(GetCqeErrInfoFromTaskException);
986 43 : hcomm::RegisterGetCcuCqeErrInfoCallBackHcomm(GetCqeErrInfoFromTaskException);
987 43 : hcomm::RegisterAicpuGetErrStatusVecCallBack(GetErrStatusVecFromCluserMonitor);
988 43 : hcomm::RegisterCcuGetErrStatusVecCallBack(GetErrStatusVecFromCluserMonitor);
989 43 : }
990 :
991 :
992 : } // namespace hcomm
|
