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