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