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