Line data Source code
1 : /**
2 : * Copyright (c) 2025 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 "heartbeat.h"
12 : #include <set>
13 : #include <tuple>
14 : #include "device_capacity.h"
15 : #include "externalinput_pub.h"
16 : #include "env_config.h"
17 : #include "opexecounter_pub.h"
18 : #include "hccl_communicator.h"
19 : #include "task_exception_handler_pub.h"
20 : #include "comm_configer.h"
21 : #include "snapshot_control.h"
22 :
23 : namespace hccl {
24 1652 : Heartbeat& Heartbeat::GetInstance(s32 deviceLogicID)
25 : {
26 2497 : static Heartbeat hb[MAX_MODULE_DEVICE_NUM];
27 1655 : if (static_cast<u32>(deviceLogicID) >= MAX_MODULE_DEVICE_NUM) {
28 531 : HCCL_WARNING("[Heartbeat][%s]deviceLogicID[%d] is invalid", __func__, deviceLogicID);
29 531 : return hb[0];
30 : }
31 1124 : return hb[deviceLogicID];
32 : }
33 :
34 845 : Heartbeat::~Heartbeat()
35 : {
36 845 : if (!groupMap_.empty()) {
37 1 : HCCL_RUN_INFO("[Heartbeat]groupMap_ size[%llu].", groupMap_.size());
38 2 : for (auto iter = groupMap_.begin(); iter != groupMap_.end(); iter++) {
39 1 : HCCL_RUN_WARNING("[Heartbeat]UnRegister group[%s].", iter->first.c_str());
40 : }
41 : }
42 845 : (void)DeInit();
43 845 : groupMap_.clear();
44 845 : retryEnableTable_.clear();
45 845 : backupEnableTable_.clear();
46 845 : opInfoIndexMap_.clear();
47 845 : opInfoQueue_.clear();
48 845 : opInfoMap_.clear();
49 845 : recvOpInfoList_.clear();
50 845 : inconsistentOpMap_.clear();
51 845 : srTagMap_.clear();
52 845 : }
53 :
54 25 : bool Heartbeat::IsEnableBackupLink()
55 : {
56 25 : std::lock_guard<std::mutex> lock(backupEnableMutex_);
57 : // 若backupEnableTable_不为空，则当前还有通信域使能借轨，需要获取备用的cqe
58 25 : auto isEmpty = backupEnableTable_.empty();
59 25 : return !isEmpty;
60 25 : }
61 :
62 5 : HcclResult Heartbeat::InitNic(
63 : const NicType nicType, const s32 devicePhyId, const s32 deviceLogicId, const hccl::HcclIpAddress ip, const u32 port,
64 : const bool isBackUp)
65 : {
66 : HcclNetDevCtx nicCtx;
67 5 : CHK_RET(HcclNetOpenDev(&nicCtx, nicType, devicePhyId, deviceLogicId, ip));
68 5 : CHK_PTR_NULL(nicCtx);
69 5 : netDevCtxMap_.insert(std::make_pair(ip, nicCtx));
70 :
71 5 : if (!isBackUp) {
72 5 : std::shared_ptr<HcclSocket> tempSocket;
73 5 : EXCEPTION_CATCH((tempSocket = std::make_shared<HcclSocket>(nicCtx, port)), return HCCL_E_PTR);
74 5 : CHK_RET(tempSocket->Init());
75 5 : CHK_RET(tempSocket->Listen());
76 :
77 0 : listenSocketMap_.insert(std::make_pair(ip, tempSocket));
78 5 : }
79 :
80 0 : HCCL_INFO(
81 : "[Heartbeat][%s]NicType[%d], devicePhyId[%d], deviceLogicId[%d], ip[%s], port[%u], isBackUp[%d].", __func__,
82 : nicType, devicePhyId, deviceLogicId, ip.GetReadableAddress(), port, isBackUp);
83 0 : return HCCL_SUCCESS;
84 : }
85 :
86 5 : HcclResult Heartbeat::InitDeviceNic(const RankInfo& locRank, bool isNeedNic, u32 port)
87 : {
88 5 : if (isNeedNic && locRank.nicIp.size() != 0) {
89 0 : nicIp_ = locRank.nicIp[0];
90 0 : u32 nicPort = (port == HCCL_INVALID_PORT) ? locRank.deviceNicPort : port;
91 0 : if (locRank.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE && !nicIp_.IsInvalid()
92 0 : && netDevCtxMap_.find(nicIp_) == netDevCtxMap_.end()) {
93 0 : CHK_RET(InitNic(NicType::DEVICE_NIC_TYPE, devicePhyId_, deviceLogicId_, nicIp_, nicPort));
94 : }
95 : }
96 :
97 5 : if (isNeedNic && locRank.backupNicIp.size() != 0 && IsEnableBackupLink()) {
98 0 : backupNicIp_ = locRank.backupNicIp[0];
99 0 : u32 backupPort = HCCL_INVALID_PORT; // 不初始化备用网卡上的Socket
100 0 : if (locRank.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE
101 0 : && netDevCtxMap_.find(backupNicIp_) == netDevCtxMap_.end()) {
102 0 : CHK_RET(InitNic(
103 : NicType::DEVICE_NIC_TYPE, deviceBackUpPhyId_, deviceBackupLogicId_, backupNicIp_, backupPort, true));
104 : }
105 : }
106 5 : return HCCL_SUCCESS;
107 : }
108 :
109 5 : HcclResult Heartbeat::InitHostNic(const RankInfo& locRank, bool isNeedNic, u32 port)
110 : {
111 5 : if (!isNeedNic || locRank.nicDeploy != NICDeployment::NIC_DEPLOYMENT_HOST) {
112 5 : return HCCL_SUCCESS;
113 : }
114 :
115 0 : if (!locRank.nicIp[0].IsInvalid()) {
116 0 : u32 nicPort = (port == HCCL_INVALID_PORT) ? locRank.deviceNicPort : port;
117 0 : nicIp_ = locRank.nicIp[0];
118 0 : if (netDevCtxMap_.find(nicIp_) == netDevCtxMap_.end()) {
119 0 : CHK_RET(InitNic(NicType::HOST_NIC_TYPE, devicePhyId_, deviceLogicId_, nicIp_, nicPort));
120 : }
121 : } else {
122 0 : if (netDevCtxMap_.find(locRank.hostIp) == netDevCtxMap_.end()) {
123 0 : u32 hostPort = GetHostPort(devicePhyId_);
124 0 : CHK_RET(InitNic(NicType::HOST_NIC_TYPE, devicePhyId_, deviceLogicId_, locRank.hostIp, hostPort));
125 : }
126 : }
127 0 : return HCCL_SUCCESS;
128 : }
129 :
130 5 : HcclResult Heartbeat::Init(
131 : const RankInfo& locRank, [[maybe_unused]] const bool useSuperPodMode, const bool isNeedNic, const u32 port,
132 : const std::string& group)
133 : {
134 5 : HCCL_INFO("[%s] heartbeat Init begin.", __func__);
135 5 : devicePhyId_ = locRank.devicePhyId;
136 5 : if (IsEnableBackupLink()) {
137 2 : CHK_RET(hrtGetPairDevicePhyId(devicePhyId_, deviceBackUpPhyId_));
138 : }
139 5 : superDeviceId_ = locRank.superDeviceId;
140 5 : if (devicePhyId_ == static_cast<u32>(HOST_DEVICE_ID)) {
141 1 : deviceLogicId_ = devicePhyId_;
142 1 : deviceBackupLogicId_ = deviceBackUpPhyId_;
143 : } else {
144 4 : CHK_RET(hrtGetDeviceIndexByPhyId(devicePhyId_, deviceLogicId_));
145 4 : if (IsEnableBackupLink()) {
146 1 : CHK_RET(hrtGetDeviceIndexByPhyId(deviceBackUpPhyId_, deviceBackupLogicId_));
147 : }
148 : }
149 5 : std::unique_lock<std::mutex> mapLock(ctxMapMutex_);
150 5 : CHK_RET(InitDeviceNic(locRank, isNeedNic, port));
151 5 : CHK_RET(InitHostNic(locRank, isNeedNic, port));
152 5 : mapLock.unlock();
153 5 : uid_ = GetUId(locRank);
154 5 : nicDeploy_ = locRank.nicDeploy;
155 5 : s32 hcclExecTimeOut = CommConfiger::GetInstance().GetCommConfigExecTimeOut(group);
156 5 : stuckDetectTime_ = std::max(hcclExecTimeOut / HCCL_STUCK_DETECT_TIME_BASE, HCCL_STUCK_DETECT_TIME_MIN);
157 5 : startSendRecvTask_ = true;
158 5 : sendRecvThread_.reset(new (std::nothrow) std::thread(&Heartbeat::HeartbeatStatusMonitor, std::ref(*this)));
159 5 : CHK_SMART_PTR_NULL(sendRecvThread_);
160 5 : lostThreshold_ = HCCL_LOST_THRESHOLD; // 心跳丢失阈值为30s
161 5 : initialized_ = true;
162 5 : isPaused_ = false;
163 5 : isDeInit_ = false;
164 5 : HCCL_INFO("[%s] heartbeat Init end, stuckDetectTime[%d s].", __func__, stuckDetectTime_);
165 5 : return HCCL_SUCCESS;
166 5 : }
167 :
168 850 : HcclResult Heartbeat::DeInit()
169 : {
170 850 : HCCL_INFO("[%s] heartbeat deinit begin.", __func__);
171 850 : isDeInit_ = true;
172 850 : startSendRecvTask_ = false;
173 850 : linkThreadRunning_ = false;
174 850 : isPaused_ = false;
175 850 : if (sendRecvThread_) {
176 6 : if (sendRecvThread_->joinable()) {
177 5 : sendRecvThread_->join();
178 : }
179 : }
180 : {
181 850 : std::unique_lock<std::mutex> lock(ProcessLock_);
182 850 : for (auto iter = rankId2SocketMap_.begin(); iter != rankId2SocketMap_.end(); iter++) {
183 0 : if (iter->second.socket->GetLocalRole() == HcclSocketRole::SOCKET_ROLE_SERVER) {
184 0 : CHK_PRT_RET(
185 : listenSocketMap_.find(iter->second.socket->GetLocalIp()) == listenSocketMap_.end(),
186 : HCCL_ERROR(
187 : "ip[%s] listenSocketMap is not found", iter->second.socket->GetLocalIp().GetReadableAddress()),
188 : HCCL_E_NOT_FOUND);
189 0 : listenSocketMap_[iter->second.socket->GetLocalIp()]->DelWhiteList(iter->second.wlistInfosVec);
190 : }
191 0 : iter->second.socket->Close();
192 : }
193 850 : rankId2SocketMap_.clear();
194 850 : rankId2StatusMap_.clear();
195 850 : }
196 850 : std::queue<HeartBeatFrame> empty;
197 850 : std::swap(errStatusQueue_, empty);
198 :
199 850 : std::unique_lock<std::mutex> mapLock(ctxMapMutex_);
200 850 : listenSocketMap_.clear();
201 855 : for (auto& iter : netDevCtxMap_) {
202 5 : HcclNetCloseDev(iter.second);
203 : }
204 850 : vnicIp_.clear();
205 850 : nicIp_.clear();
206 850 : backupNicIp_.clear();
207 :
208 850 : netDevCtxMap_.clear();
209 850 : mapLock.unlock();
210 :
211 850 : initialized_ = false;
212 850 : HCCL_INFO("[%s] heartbeat deinit end.", __func__);
213 850 : return HCCL_SUCCESS;
214 850 : }
215 :
216 0 : HcclResult Heartbeat::PrepareConnect(ConnInfo& info)
217 : {
218 0 : CHK_SMART_PTR_NULL(info.socket);
219 0 : if (info.socket->GetLocalRole() == HcclSocketRole::SOCKET_ROLE_SERVER) {
220 0 : CHK_PRT_RET(
221 : listenSocketMap_.find(info.socket->GetLocalIp()) == listenSocketMap_.end(),
222 : HCCL_ERROR("ip[%s] listenSocketMap is not found", info.socket->GetLocalIp().GetReadableAddress()),
223 : HCCL_E_NOT_FOUND);
224 0 : CHK_RET(listenSocketMap_[info.socket->GetLocalIp()]->AddWhiteList(info.wlistInfosVec));
225 : } else {
226 0 : if (info.socket->GetStatus() != HcclSocketStatus::SOCKET_OK) {
227 0 : CHK_RET(info.socket->Connect());
228 : }
229 : }
230 :
231 0 : return HCCL_SUCCESS;
232 : }
233 :
234 28 : HcclResult Heartbeat::RegisterRanks(
235 : [[maybe_unused]] DevType devType, const RankInfo& locRank, std::vector<RankInfo>& rankInfos, const u32 port,
236 : const bool isNeedNic, const std::string& group, bool useSuperPodMode, bool isUsedRdma)
237 : {
238 28 : HCCL_INFO(
239 : "[%s] group[%s] isUsedRdma[%d], isNeedNic[%d], RegisterRanks Start.", __func__, group.c_str(), isUsedRdma,
240 : isNeedNic);
241 : // 线程锁，防止多线程同时Init
242 28 : std::unique_lock<std::mutex> lock(ProcessLock_);
243 28 : auto iter = groupMap_.find(group);
244 28 : if (iter != groupMap_.end()) {
245 18 : HCCL_INFO("group[%s] has Registered, skip.", group.c_str());
246 18 : return HCCL_SUCCESS;
247 : }
248 :
249 10 : if (!initialized_) {
250 6 : CHK_RET(Init(locRank, useSuperPodMode, isNeedNic, port, group));
251 : }
252 :
253 : // 刷新uid_，防止不同通信域下serverId不一致问题
254 10 : uid_ = GetUId(locRank);
255 10 : lock.unlock();
256 :
257 10 : std::unique_lock<std::mutex> mapLock(ctxMapMutex_);
258 10 : if (devicePhyId_ != static_cast<u32>(HOST_DEVICE_ID) && rankInfos.size() > 1 && vnicIp_.IsInvalid()) {
259 6 : vnicIp_ = HcclIpAddress(useSuperPodMode ? superDeviceId_ : devicePhyId_);
260 6 : u32 vnicPort = (port == HCCL_INVALID_PORT) ? locRank.deviceVnicPort : port;
261 6 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
262 : devicePhyId_, (useSuperPodMode ? DeviceIdType::DEVICE_ID_TYPE_SDID : DeviceIdType::DEVICE_ID_TYPE_PHY_ID),
263 : (useSuperPodMode ? superDeviceId_ : devicePhyId_), vnicIp_));
264 6 : if (netDevCtxMap_.find(vnicIp_) == netDevCtxMap_.end()) {
265 5 : CHK_RET(InitNic(NicType::VNIC_TYPE, devicePhyId_, deviceLogicId_, vnicIp_, vnicPort));
266 : }
267 : }
268 :
269 : // 防止首次没有读到nicIp, 后续注册心跳的时候刷新上
270 5 : if (isNeedNic && nicIp_.IsInvalid() && locRank.nicIp.size() != 0) {
271 0 : nicIp_ = locRank.nicIp[0];
272 0 : u32 nicPort = (port == HCCL_INVALID_PORT) ? locRank.deviceNicPort : port;
273 0 : if (locRank.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE && !nicIp_.IsInvalid()
274 0 : && netDevCtxMap_.find(nicIp_) == netDevCtxMap_.end()) {
275 0 : CHK_RET(InitNic(NicType::DEVICE_NIC_TYPE, devicePhyId_, deviceLogicId_, nicIp_, nicPort));
276 : }
277 : }
278 :
279 5 : if (isNeedNic && backupNicIp_.IsInvalid() && locRank.backupNicIp.size() != 0) {
280 0 : backupNicIp_ = locRank.backupNicIp[0];
281 0 : u32 backupPort = HCCL_INVALID_PORT; // 不初始化备用网卡上的Socket
282 0 : if (IsEnableBackupLink()) {
283 0 : CHK_RET(hrtGetPairDevicePhyId(devicePhyId_, deviceBackUpPhyId_));
284 0 : CHK_RET(hrtGetDeviceIndexByPhyId(deviceBackUpPhyId_, deviceBackupLogicId_));
285 0 : if (locRank.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE
286 0 : && netDevCtxMap_.find(backupNicIp_) == netDevCtxMap_.end()) {
287 0 : CHK_RET(InitNic(
288 : NicType::DEVICE_NIC_TYPE, deviceBackUpPhyId_, deviceBackupLogicId_, backupNicIp_, backupPort,
289 : true));
290 : }
291 : }
292 : }
293 :
294 5 : u32 tcpPort = 0;
295 5 : if (isNeedNic && locRank.nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST) {
296 0 : if (!locRank.nicIp[0].IsInvalid()) {
297 0 : nicIp_ = locRank.nicIp[0];
298 0 : tcpPort = (port == HCCL_INVALID_PORT) ? locRank.deviceNicPort : port;
299 0 : if (netDevCtxMap_.find(nicIp_) == netDevCtxMap_.end()) {
300 0 : CHK_RET(InitNic(NicType::HOST_NIC_TYPE, devicePhyId_, deviceLogicId_, nicIp_, tcpPort));
301 : }
302 : } else {
303 0 : if (netDevCtxMap_.find(locRank.hostIp) == netDevCtxMap_.end()) {
304 0 : tcpPort = GetHostPort(devicePhyId_);
305 0 : CHK_RET(InitNic(NicType::HOST_NIC_TYPE, devicePhyId_, deviceLogicId_, locRank.hostIp, tcpPort));
306 : }
307 : }
308 : }
309 5 : mapLock.unlock();
310 :
311 5 : lock.lock();
312 61 : for (const auto& remRank : rankInfos) {
313 56 : UIDType rem = GetUId(remRank);
314 56 : rankId2StatusMap_.insert(rem, Status());
315 56 : groupMap_[group].insert(std::make_pair(rem, NO_CONN));
316 56 : HCCL_INFO("[%s]group[%s] remote:%s", __func__, group.c_str(), FormatUId(rem).c_str());
317 : }
318 5 : lock.unlock();
319 :
320 5 : if (!GetExternalInputHcclHeartBeatEnable()) {
321 0 : HCCL_RUN_INFO(
322 : "[Heartbeat][%s] Enable HcclHeartBeatLink is [%d]. It's unnecessary to "
323 : "register Ranks. Group[%s] isUsedRdma[%d], netDevCtxMap size[%llu]",
324 : __func__, GetExternalInputHcclHeartBeatEnable(), group.c_str(), isUsedRdma, netDevCtxMap_.size());
325 0 : return HCCL_SUCCESS;
326 : }
327 :
328 5 : std::map<UIDType, ConnInfo> needConnectRank;
329 5 : CHK_RET(GetConnectRank(locRank, rankInfos, needConnectRank, useSuperPodMode, isUsedRdma));
330 :
331 5 : std::unique_lock<std::mutex> linkInfolock(hbLinkConnInfoMtx_);
332 13 : for (auto& item : needConnectRank) {
333 8 : if (item.second.newConn == true) {
334 6 : hbLinkConnInfo_[group].push(std::move(item));
335 : }
336 : }
337 5 : linkInfolock.unlock();
338 :
339 5 : lock.lock();
340 13 : for (auto& item : needConnectRank) {
341 8 : if (item.second.newConn == true) {
342 6 : rankId2LinkStatusMap_[item.first] = HBLinkStatus::HEARTBEAT_LINK_BUILDING;
343 2 : } else if (
344 2 : groupMap_[group].find(item.first) == groupMap_[group].end()
345 2 : || (groupMap_[group].count(item.first) && groupMap_[group][item.first] == NO_CONN)) {
346 2 : rankId2SocketMap_.ref(item.first);
347 2 : HCCL_RUN_INFO(
348 : "group:[%s], establish rank[%s] to rank[%s] heartbeat connection success.", group.c_str(),
349 : FormatUId(uid_).c_str(), FormatUId(item.first).c_str());
350 2 : groupMap_[group][item.first] = HAS_CONN;
351 : }
352 : }
353 5 : lock.unlock();
354 :
355 5 : HCCL_INFO(
356 : "[%s]group[%s] isUsedRdma[%d], netDevCtxMap size[%llu], RegisterRanks Completed", __func__, group.c_str(),
357 : isUsedRdma, netDevCtxMap_.size());
358 5 : return HCCL_SUCCESS;
359 28 : }
360 :
361 0 : void Heartbeat::CreateLinkWithRemote(std::string group, UIDType rem, ConnInfo needConnectRank)
362 : {
363 : // 给当前线程添加名字
364 0 : const std::string threadName = "hb" + FormatUId(rem);
365 0 : SetThreadName(threadName);
366 :
367 0 : if (deviceLogicId_ != static_cast<u32>(HOST_DEVICE_ID)) {
368 0 : hrtSetDevice(deviceLogicId_);
369 : }
370 0 : HCCL_INFO("[Heartbeat][CreateLinkWithRemote] Group[%s], thread[%s] start...", group.c_str(), threadName.c_str());
371 :
372 0 : HcclResult ret = PrepareConnect(needConnectRank);
373 0 : if (ret != HCCL_SUCCESS) {
374 0 : HCCL_ERROR(
375 : "[CreateLinkWithRemote] PrepareConnect ret[%d], group[%s], remote uid[%s].", ret, group.c_str(),
376 : FormatUId(rem).c_str());
377 0 : if (deviceLogicId_ != static_cast<u32>(HOST_DEVICE_ID)) {
378 0 : hrtResetDevice(deviceLogicId_);
379 : }
380 0 : return;
381 : }
382 0 : auto HEART_CREATE_LINK_TIMEOUT = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
383 0 : auto startTime = std::chrono::steady_clock::now();
384 0 : while (linkThreadRunning_) {
385 0 : if ((std::chrono::steady_clock::now() - startTime) >= HEART_CREATE_LINK_TIMEOUT) {
386 0 : HCCL_RUN_WARNING(
387 : "establish rank[%s] to rank[%s] heartbeat connection failed. Reason: get rasocket timeout,"
388 : "timeout[%llds], the HCCL_CONNECT_TIMEOUT may be insufficient. Group[%s].",
389 : FormatUId(uid_).c_str(), FormatUId(rem).c_str(), HEART_CREATE_LINK_TIMEOUT, group.c_str());
390 0 : break;
391 : }
392 :
393 0 : if (needConnectRank.socket->GetStatus() == HcclSocketStatus::SOCKET_TIMEOUT
394 0 : || needConnectRank.socket->GetStatus() == HcclSocketStatus::SOCKET_ERROR) {
395 0 : HCCL_RUN_WARNING(
396 : "establish rank[%s] to rank[%s] heartbeat connection failed. Reason: socket status [%d]"
397 : "Group[%s]",
398 : FormatUId(uid_).c_str(), FormatUId(rem).c_str(), needConnectRank.socket->GetStatus(), group.c_str());
399 0 : needConnectRank.socket->Close();
400 0 : break;
401 : }
402 :
403 0 : if (needConnectRank.socket->GetStatus() == HcclSocketStatus::SOCKET_CONNECTING) {
404 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
405 0 : continue;
406 : }
407 :
408 0 : std::unique_lock<std::mutex> lock(ProcessLock_);
409 0 : if (groupMap_.find(group) == groupMap_.end()) {
410 0 : HCCL_RUN_WARNING(
411 : "establish rank[%s] to rank[%s] heartbeat connection failed. Reason: Group[%s] has been"
412 : "Unregistered.",
413 : FormatUId(uid_).c_str(), FormatUId(rem).c_str(), group.c_str());
414 0 : needConnectRank.socket->Close();
415 0 : lock.unlock();
416 0 : break;
417 : }
418 0 : needConnectRank.newConn = false;
419 0 : rankId2SocketMap_.insert(rem, needConnectRank);
420 : // 心跳socket建链完成后，需要立即及激活其心跳收发能力
421 0 : auto frameSize = GetExternalInconsistentCheckSwitch() == InconsistentCheckMode::ON ?
422 : sizeof(HeartBeatFrameWithOpCheck) :
423 0 : sizeof(HeartBeatFrame);
424 0 : if (rankId2SocketMap_[rem].recvBuffer.Init(BASE_NUMBER * frameSize) != HCCL_SUCCESS) { // 2倍帧长，确报不会溢出
425 0 : HCCL_RUN_WARNING(
426 : "establish rank[%s] to rank[%s] heartbeat connection failed. Reason: socket recv buffer init"
427 : "failed. Group[%s].",
428 : FormatUId(uid_).c_str(), FormatUId(rem).c_str(), group.c_str());
429 0 : rankId2SocketMap_.erase(rem);
430 0 : lock.unlock();
431 0 : break;
432 : }
433 0 : rankId2LinkStatusMap_[rem] = HBLinkStatus::HEARTBEAT_LINK_COMPLETED;
434 0 : groupMap_[group][rem] = HAS_CONN;
435 0 : lock.unlock();
436 0 : HCCL_RUN_INFO(
437 : "group:[%s], establish rank[%s] to rank[%s] heartbeat connection success.", group.c_str(),
438 : FormatUId(uid_).c_str(), FormatUId(rem).c_str());
439 0 : break;
440 0 : }
441 0 : if (deviceLogicId_ != static_cast<u32>(HOST_DEVICE_ID)) {
442 0 : hrtResetDevice(deviceLogicId_);
443 : }
444 :
445 0 : HCCL_INFO("[%s] Thread [%s] end...", __func__, threadName.c_str());
446 0 : return;
447 0 : }
448 :
449 20 : void Heartbeat::RegisterRetryInfo(const std::string& commIdentifier, bool retryEnable, bool backupEnable)
450 : {
451 : {
452 20 : std::lock_guard<std::mutex> retryEnablelock(retryEnableMutex_);
453 20 : auto search = retryEnableTable_.find(commIdentifier);
454 20 : if (search != retryEnableTable_.end()) {
455 19 : HCCL_INFO(
456 : "[%s]register identifier[%s] retryEnable[%d] has been registered", __func__, commIdentifier.c_str(),
457 : search->second);
458 : } else {
459 1 : retryEnableTable_.insert({commIdentifier, retryEnable});
460 1 : HCCL_RUN_INFO("[%s]register identifier[%s] retryEnable[%d]", __func__, commIdentifier.c_str(), retryEnable);
461 : }
462 20 : }
463 20 : if (backupEnable) {
464 : // 若当前通信域使能借轨，则加入到backupEnableTable_中
465 0 : std::lock_guard<std::mutex> backupEnablelock(backupEnableMutex_);
466 0 : if (backupEnableTable_.find(commIdentifier) == backupEnableTable_.end()) {
467 0 : backupEnableTable_.insert(commIdentifier);
468 0 : HCCL_RUN_INFO(
469 : "[%s]register identifier[%s] backupEnable[%d]", __func__, commIdentifier.c_str(), backupEnable);
470 : }
471 0 : }
472 20 : return;
473 : }
474 21 : HcclResult Heartbeat::RegisterToHeartBeat(
475 : u32 userRank, DevType devType, std::vector<RankInfo>& rankInfoList, const u32 port, const bool isNeedNic,
476 : const std::string& commIdentifier, bool useSuperPodMode, bool isUsedRdmaLevel0, bool retryEnable, bool backupEnable)
477 : {
478 21 : if (Is310PDevice() || devType == DevType::DEV_TYPE_310P3) {
479 0 : return HCCL_SUCCESS;
480 : }
481 :
482 21 : CHK_PRT_RET(
483 : rankInfoList.size() == 1,
484 : HCCL_WARNING(
485 : "[RegisterToHeartBeat]Identifier[%s] rankSize[%llu] needn't to register.", commIdentifier.c_str(),
486 : rankInfoList.size()),
487 : HCCL_SUCCESS);
488 :
489 20 : RankInfo locRank;
490 20 : for (auto rank : rankInfoList) {
491 20 : if (userRank == rank.userRank) {
492 20 : locRank = rank;
493 20 : break;
494 : }
495 20 : }
496 :
497 20 : RegisterRetryInfo(commIdentifier, retryEnable, backupEnable);
498 :
499 20 : CHK_RET(RegisterRanks(
500 : devType, locRank, rankInfoList, port, isNeedNic, commIdentifier, useSuperPodMode, isUsedRdmaLevel0));
501 19 : return HCCL_SUCCESS;
502 20 : }
503 :
504 0 : HcclResult Heartbeat::RegisterToHeartBeat(
505 : u32 userRank, DevType devType, std::vector<RankInfo>& rankInfoList, const u32 port, const bool isNeedNic,
506 : u32 peerRankId, const std::string& commIdentifier, const std::string& tag, bool useSuperPodMode,
507 : bool isUsedRdmaLevel0, bool retryEnable, bool backupEnable)
508 : {
509 0 : if (Is310PDevice() || devType == DevType::DEV_TYPE_310P3 || (rankInfoList[userRank].devicePhyId == HOST_DEVICE_ID)
510 0 : || (rankInfoList[peerRankId].devicePhyId == HOST_DEVICE_ID)) {
511 0 : return HCCL_SUCCESS;
512 : }
513 :
514 0 : CHK_PRT_RET(
515 : rankInfoList.size() == 1,
516 : HCCL_WARNING(
517 : "[RegisterToHeartBeat]Identifier[%s] rankSize[%llu] needn't to register.", commIdentifier.c_str(),
518 : rankInfoList.size()),
519 : HCCL_SUCCESS);
520 :
521 0 : RankInfo locRank;
522 0 : std::vector<RankInfo> peerRankInfoList;
523 0 : bool findLoc = false;
524 0 : bool findPeer = false;
525 0 : for (auto rank : rankInfoList) {
526 0 : if (userRank == rank.userRank) {
527 0 : locRank = rank;
528 0 : peerRankInfoList.push_back(rank);
529 0 : findLoc = true;
530 : }
531 :
532 0 : if (peerRankId == rank.userRank) {
533 0 : peerRankInfoList.push_back(rank);
534 0 : findPeer = true;
535 : }
536 :
537 0 : if (findLoc && findPeer) {
538 0 : break;
539 : }
540 0 : }
541 0 : RegisterRetryInfo(commIdentifier, retryEnable, backupEnable);
542 0 : CHK_RET(RegisterRanks(devType, locRank, peerRankInfoList, port, isNeedNic, tag, useSuperPodMode, isUsedRdmaLevel0));
543 0 : return HCCL_SUCCESS;
544 0 : }
545 :
546 : HcclResult
547 0 : Heartbeat::AddOpInfoToHeartBeat(const std::string& identifier, const OpInfoDesc& opInfo, const std::string& newTag)
548 : {
549 0 : AddOpInfo(identifier, opInfo, newTag);
550 0 : return HCCL_SUCCESS;
551 : }
552 :
553 649 : HcclResult Heartbeat::DeleteOpInfoToHeartBeat(const std::string& identifier, const std::string& newTag)
554 : {
555 649 : std::string tag;
556 649 : if (newTag != "") {
557 0 : tag = newTag;
558 : } else {
559 649 : tag = identifier;
560 : }
561 649 : CHK_PRT_RET(initialized_ == false, HCCL_WARNING("Heartbeat has been destroyed"), HCCL_SUCCESS);
562 0 : std::unique_lock<std::mutex> lock(opInfoMapMutex_);
563 0 : opInfoMap_.erase(tag);
564 0 : opInfoIndexMap_.erase(tag);
565 0 : return HCCL_SUCCESS;
566 649 : }
567 :
568 220 : HcclResult Heartbeat::UnRegisterRanks(const std::string& group)
569 : {
570 220 : CHK_PRT_RET(initialized_ == false, HCCL_WARNING("Heartbeat has been destroyed"), HCCL_SUCCESS);
571 7 : std::set<UIDType> remInQueue;
572 7 : std::unique_lock<std::mutex> connInfoLock(hbLinkConnInfoMtx_);
573 7 : if (hbLinkConnInfo_.find(group) != hbLinkConnInfo_.end()) {
574 9 : while (!hbLinkConnInfo_[group].empty()) {
575 6 : remInQueue.insert(hbLinkConnInfo_[group].front().first);
576 6 : hbLinkConnInfo_[group].pop();
577 : }
578 : }
579 7 : hbLinkConnInfo_.erase(group);
580 7 : connInfoLock.unlock();
581 :
582 : {
583 7 : std::unique_lock<std::mutex> lock(ProcessLock_);
584 :
585 13 : for (const auto& rem : remInQueue) {
586 6 : if (rankId2LinkStatusMap_[rem] == HBLinkStatus::HEARTBEAT_LINK_BUILDING) {
587 6 : rankId2LinkStatusMap_[rem] = HBLinkStatus::HEARTBEAT_LINK_NOT_START;
588 6 : HCCL_INFO(
589 : "[%s] group[%s] rem[%s] is in hbLinkConnInfo deque. Status change to not start", __func__,
590 : group.c_str(), FormatUId(rem).c_str());
591 : }
592 : }
593 7 : auto iter = groupMap_.find(group);
594 7 : if (iter == groupMap_.end()) {
595 3 : HCCL_INFO("group[%s] hasn't Registered, skip", group.c_str());
596 3 : return HCCL_SUCCESS;
597 : }
598 :
599 40 : for (const auto& remRank : groupMap_[group]) {
600 36 : UIDType rem = remRank.first;
601 36 : rankId2StatusMap_.erase(rem);
602 36 : if (remRank.second == HAS_CONN) {
603 2 : if (rankId2SocketMap_.count(rem) == 1) {
604 0 : if (rankId2SocketMap_[rem].socket->GetLocalRole() == HcclSocketRole::SOCKET_ROLE_SERVER) {
605 0 : CHK_PRT_RET(
606 : listenSocketMap_.find(rankId2SocketMap_[rem].socket->GetLocalIp())
607 : == listenSocketMap_.end(),
608 : HCCL_ERROR(
609 : "ip[%s] listenSocketMap is not found",
610 : rankId2SocketMap_[rem].socket->GetLocalIp().GetReadableAddress()),
611 : HCCL_E_NOT_FOUND);
612 0 : listenSocketMap_[rankId2SocketMap_[rem].socket->GetLocalIp()]->DelWhiteList(
613 0 : rankId2SocketMap_[rem].wlistInfosVec);
614 : }
615 0 : rankId2SocketMap_[rem].socket->Close();
616 0 : rankId2LinkStatusMap_[rem] = HBLinkStatus::HEARTBEAT_LINK_NOT_START;
617 : }
618 2 : HCCL_INFO("[%s]group[%s] socket erase remote:%s", __func__, group.c_str(), FormatUId(rem).c_str());
619 2 : rankId2SocketMap_.erase(rem);
620 : }
621 36 : HCCL_INFO("[%s]group[%s] status erase remote:%s", __func__, group.c_str(), FormatUId(rem).c_str());
622 : }
623 4 : groupMap_.erase(iter);
624 4 : HCCL_INFO("[%s]group[%s] UnregisterRanks Completed.", __func__, group.c_str());
625 7 : }
626 :
627 4 : if (groupMap_.size() == 0) {
628 3 : HCCL_RUN_INFO("[%s]Entry HeartBeat DeInit.", __func__);
629 3 : CHK_RET(DeInit());
630 : }
631 4 : return HCCL_SUCCESS;
632 7 : }
633 :
634 809 : void Heartbeat::UnRegisterToHeartBeat(DevType devType, const std::string& commIdentifier)
635 : {
636 809 : if (Is310PDevice() || devType == DevType::DEV_TYPE_310P3) {
637 4 : return;
638 : }
639 807 : ClearRetryEnableMapItem(commIdentifier);
640 804 : HcclResult ret = UnRegisterRanks(commIdentifier);
641 804 : if (ret != HCCL_SUCCESS) {
642 0 : HCCL_ERROR("UnRegisterToHeartBeat failed");
643 : }
644 : }
645 0 : void Heartbeat::UnRegisterToHeartBeat(DevType devType, const std::string& commIdentifier, const std::string& tag)
646 : {
647 0 : if (Is310PDevice() || devType == DevType::DEV_TYPE_310P3) {
648 0 : return;
649 : }
650 0 : ClearRetryEnableMapItem(commIdentifier);
651 0 : HcclResult ret = UnRegisterRanks(tag);
652 0 : if (ret != HCCL_SUCCESS) {
653 0 : HCCL_ERROR("UnRegisterToHeartBeat failed");
654 : }
655 : }
656 :
657 108 : UIDType Heartbeat::GetUId(const RankInfo& rankInfo) const
658 : {
659 108 : UIDType uid;
660 108 : s32 ret = snprintf_s(
661 : uid.id, sizeof(uid.id), sizeof(uid.id) - 1, "%s%s%s", rankInfo.serverId.c_str(), "/",
662 216 : std::to_string(rankInfo.devicePhyId).c_str());
663 108 : if (ret == -1) {
664 0 : HCCL_WARNING("[Heartbeat][%s] snprintf_s failed", __func__);
665 : }
666 108 : return uid;
667 : }
668 :
669 636 : std::string Heartbeat::FormatUId(const UIDType& uid) const { return uid.id; }
670 :
671 13 : std::string Heartbeat::GetConnTag(HcclSocketRole role, UIDType& rem)
672 : {
673 13 : std::string tag;
674 13 : if (role == HcclSocketRole::SOCKET_ROLE_CLIENT) {
675 7 : tag = "HeartBeat_" + FormatUId(uid_) + "_to_" + FormatUId(rem);
676 : } else {
677 6 : tag = "HeartBeat_" + FormatUId(rem) + "_to_" + FormatUId(uid_);
678 : }
679 :
680 13 : return tag;
681 0 : }
682 :
683 13 : HcclResult Heartbeat::GetConnInfo(
684 : RankInfo& remRank, bool useSuperPodMode, HcclSocketRole role, HcclSocketType type,
685 : std::map<UIDType, ConnInfo>& needConnectRank)
686 : {
687 13 : bool newConn = true;
688 13 : UIDType rem = GetUId(remRank);
689 : {
690 13 : std::unique_lock<std::mutex> lock(ProcessLock_);
691 13 : if (rankId2LinkStatusMap_.find(rem) == rankId2LinkStatusMap_.end()) {
692 8 : rankId2LinkStatusMap_[rem] = HBLinkStatus::HEARTBEAT_LINK_NOT_START;
693 5 : } else if (
694 5 : rankId2LinkStatusMap_[rem] == HBLinkStatus::HEARTBEAT_LINK_BUILDING
695 5 : || rankId2LinkStatusMap_[rem] == HBLinkStatus::HEARTBEAT_LINK_COMPLETED) {
696 2 : newConn = false;
697 : }
698 13 : }
699 13 : std::string tag = GetConnTag(role, rem);
700 13 : HcclIpAddress remNicIp;
701 13 : if (remRank.nicIp.size() > 0) {
702 13 : remNicIp = remRank.nicIp[0];
703 : }
704 :
705 13 : if (type == HcclSocketType::SOCKET_NIC && (nicIp_.IsInvalid() || remNicIp.IsInvalid())) {
706 5 : HCCL_INFO("No Invalid Nic, Skip");
707 5 : return HCCL_SUCCESS;
708 : }
709 :
710 : u32 remoteDeviceId;
711 : u32 localDeviceId;
712 : DeviceIdType deviceIdType;
713 8 : if (useSuperPodMode) {
714 0 : remoteDeviceId = remRank.superDeviceId;
715 0 : localDeviceId = superDeviceId_;
716 0 : deviceIdType = DeviceIdType::DEVICE_ID_TYPE_SDID;
717 : } else {
718 8 : remoteDeviceId = remRank.devicePhyId;
719 8 : localDeviceId = devicePhyId_;
720 8 : deviceIdType = DeviceIdType::DEVICE_ID_TYPE_PHY_ID;
721 : }
722 :
723 8 : HcclIpAddress locNicIp = nicIp_;
724 8 : if (type == HcclSocketType::SOCKET_VNIC) {
725 : // 获取本端vnic ip
726 8 : locNicIp = HcclIpAddress(localDeviceId);
727 8 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(devicePhyId_, deviceIdType, localDeviceId, locNicIp));
728 : // 获取远端vnic ip
729 8 : remNicIp = HcclIpAddress(remoteDeviceId);
730 8 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(devicePhyId_, deviceIdType, remoteDeviceId, remNicIp));
731 : }
732 :
733 8 : u32 port = HCCL_INVALID_PORT;
734 8 : if (remRank.nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST) {
735 0 : port = GetHostPort(remoteDeviceId);
736 : } else {
737 8 : port = GetPort(type, remRank.userRank, remoteDeviceId);
738 : }
739 :
740 8 : HCCL_INFO("remote userRank[%u], connect port[%u].", remRank.userRank, port);
741 :
742 8 : std::shared_ptr<HcclSocket> tempSocket;
743 8 : std::unique_lock<std::mutex> mapLock(ctxMapMutex_);
744 8 : CHK_PRT_RET(
745 : netDevCtxMap_.find(locNicIp) == netDevCtxMap_.end(),
746 : HCCL_ERROR("ip[%s] netDevCtx is not found, socket type[%d]", locNicIp.GetReadableAddress(), type),
747 : HCCL_E_NOT_FOUND);
748 8 : HcclNetDevCtx devCtx = netDevCtxMap_[locNicIp];
749 8 : mapLock.unlock();
750 8 : ConnInfo conn(newConn, tempSocket);
751 8 : if (role == HcclSocketRole::SOCKET_ROLE_SERVER) {
752 : SocketWlistInfo wlistInfo;
753 4 : wlistInfo.connLimit = 1;
754 4 : CHK_SAFETY_FUNC_RET(memcpy_s(&wlistInfo.tag[0], sizeof(wlistInfo.tag), tag.c_str(), tag.size() + 1));
755 :
756 4 : wlistInfo.remoteIp.addr = remNicIp.GetBinaryAddress().addr;
757 4 : wlistInfo.remoteIp.addr6 = remNicIp.GetBinaryAddress().addr6;
758 4 : conn.wlistInfosVec.push_back(wlistInfo);
759 : }
760 :
761 8 : EXCEPTION_CATCH((tempSocket = std::make_shared<HcclSocket>(tag, devCtx, remNicIp, port, role)), return HCCL_E_PTR);
762 8 : CHK_RET(tempSocket->Init());
763 :
764 8 : conn.socket = tempSocket;
765 :
766 8 : needConnectRank.insert(std::make_pair(rem, conn));
767 8 : return HCCL_SUCCESS;
768 13 : }
769 :
770 3 : HcclResult GetSocketTypeIn91093(
771 : const std::vector<RankInfo>& rankInfos, bool useSuperPodMode, u32 index, u32 nextOrPrevIndex, HcclSocketType& type)
772 : {
773 : // 910_93 Type要动态改一下 1. 同server vnic 2. 不同server 超结点内vnic 超结点间nic
774 3 : auto locRank = rankInfos[index];
775 3 : auto rankInfo = rankInfos[nextOrPrevIndex];
776 3 : bool localUseSuporPodModel = useSuperPodMode && locRank.superPodId.empty() == false;
777 3 : bool needSuperModeHb = localUseSuporPodModel && useSuperPodMode && rankInfo.superPodId.empty() == false;
778 3 : if (needSuperModeHb) {
779 0 : bool isInterServer = false;
780 0 : uint32_t userRankServerId = 0;
781 0 : uint32_t remoteRankServerId = 0;
782 0 : rtError_t ret = rtGetServerIDBySDID(locRank.superDeviceId, &userRankServerId);
783 0 : CHK_PRT_RET(
784 : ret != RT_ERROR_NONE,
785 : HCCL_ERROR(
786 : "[GetSocketTypeIn91093]rtGetServerIDBySDID failed sdid[0x%08x], serverID[%u], ret[%u]",
787 : locRank.superDeviceId, userRankServerId, ret),
788 : HCCL_E_RUNTIME);
789 0 : ret = rtGetServerIDBySDID(rankInfo.superDeviceId, &remoteRankServerId);
790 0 : CHK_PRT_RET(
791 : ret != RT_ERROR_NONE,
792 : HCCL_ERROR(
793 : "[GetSocketTypeIn91093]rtGetServerIDBySDID failed sdid[0x%08x], serverID[%u], ret[%u]",
794 : rankInfo.superDeviceId, remoteRankServerId, ret),
795 : HCCL_E_RUNTIME);
796 0 : isInterServer = (userRankServerId != remoteRankServerId) || (locRank.superPodId != rankInfo.superPodId);
797 0 : HCCL_INFO(
798 : "[GetSocketTypeIn91093]localSDID[0x%08x], localdevicePhyId[%d], localServerId[%s], "
799 : "localServerIdBySDID[%d], localSuperPodId[%s], "
800 : "remoteSDID[0x%08x], remotedevicePhyId[%d], remoteServerId[%s], remoteServerIdBySDID[%d], "
801 : "remoteSuperPodId[%s], "
802 : "isInterServer[%s]",
803 : locRank.superDeviceId, locRank.devicePhyId, locRank.serverId.c_str(), userRankServerId,
804 : locRank.superPodId.c_str(), rankInfo.superDeviceId, rankInfo.devicePhyId, rankInfo.serverId.c_str(),
805 : remoteRankServerId, rankInfo.superPodId.c_str(), isInterServer ? "true" : "false");
806 0 : if (!isInterServer) { // serverId相同表示同超结点同server
807 0 : type = HcclSocketType::SOCKET_VNIC;
808 0 : } else if (locRank.superPodId == rankInfo.superPodId) { // 同超结点
809 0 : type = (GetExternalInputInterHccsDisable() == true) ? HcclSocketType::SOCKET_NIC :
810 : HcclSocketType::SOCKET_VNIC;
811 : } else { // 表示不同超结点
812 0 : type = HcclSocketType::SOCKET_NIC;
813 : }
814 : }
815 3 : return HCCL_SUCCESS;
816 3 : }
817 :
818 : template <typename T>
819 10 : HcclResult Heartbeat::GetSamePlaneConnInfo(
820 : HcclSocketType type, std::vector<std::pair<T, u32>>& connVec, T& locId, std::vector<RankInfo>& rankInfos,
821 : std::map<UIDType, ConnInfo>& needConnectRank, bool useSuperPodMode, u32 worldRank)
822 : {
823 10 : u32 index = 0;
824 23 : for (; index < connVec.size(); index++) {
825 23 : if (connVec[index].first == locId) {
826 10 : break;
827 : }
828 : }
829 :
830 : DevType devType;
831 10 : CHK_RET(hrtGetDeviceType(devType));
832 10 : u32 connCount = connVec.size();
833 10 : if (connCount <= 1) {
834 3 : HCCL_INFO("nothing need to connect");
835 7 : } else if (connCount == 2) { // 2个rank, 只需建链一条连接
836 1 : u32 nextIndex = connVec[(index + 1) % connCount].second;
837 1 : if (devType == DevType::DEV_TYPE_910_93) {
838 1 : CHK_RET(GetSocketTypeIn91093(rankInfos, useSuperPodMode, connVec[index].second, nextIndex, type));
839 : }
840 1 : HCCL_INFO(
841 : "[GetSamePlaneConnInfo]local rank[%u], remote rank[%u], type[%d]", worldRank,
842 : rankInfos[nextIndex].worldRank, type);
843 1 : if (index == 0) {
844 1 : CHK_RET(GetConnInfo(
845 : rankInfos[nextIndex], useSuperPodMode, HcclSocketRole::SOCKET_ROLE_CLIENT, type, needConnectRank));
846 : } else {
847 0 : CHK_RET(GetConnInfo(
848 : rankInfos[nextIndex], useSuperPodMode, HcclSocketRole::SOCKET_ROLE_SERVER, type, needConnectRank));
849 : }
850 : } else {
851 6 : u32 nextIndex = connVec[(index + 1) % connCount].second;
852 6 : if (devType == DevType::DEV_TYPE_910_93) {
853 1 : CHK_RET(GetSocketTypeIn91093(rankInfos, useSuperPodMode, connVec[index].second, nextIndex, type));
854 : }
855 6 : HCCL_INFO(
856 : "[GetSamePlaneConnInfo][nextIndex]local rank[%u], remote rank[%u], type[%d]", worldRank,
857 : rankInfos[nextIndex].worldRank, type);
858 6 : CHK_RET(GetConnInfo(
859 : rankInfos[nextIndex], useSuperPodMode, HcclSocketRole::SOCKET_ROLE_CLIENT, type, needConnectRank));
860 :
861 6 : u32 prevIndex = connVec[(index + connCount - 1) % connCount].second;
862 6 : if (devType == DevType::DEV_TYPE_910_93) {
863 1 : CHK_RET(GetSocketTypeIn91093(rankInfos, useSuperPodMode, connVec[index].second, prevIndex, type));
864 : }
865 6 : HCCL_INFO(
866 : "[GetSamePlaneConnInfo][prevIndex]local rank[%u], remote rank[%u], type[%d]", worldRank,
867 : rankInfos[prevIndex].worldRank, type);
868 6 : CHK_RET(GetConnInfo(
869 : rankInfos[prevIndex], useSuperPodMode, HcclSocketRole::SOCKET_ROLE_SERVER, type, needConnectRank));
870 : }
871 :
872 10 : return HCCL_SUCCESS;
873 : }
874 :
875 5 : HcclResult Heartbeat::GetConnectRank(
876 : const RankInfo& locRank, std::vector<RankInfo>& rankInfos, std::map<UIDType, ConnInfo>& needConnectRank,
877 : bool useSuperPodMode, bool isUsedRdma)
878 : {
879 5 : std::vector<std::pair<u32, u32>> devVec;
880 5 : std::vector<std::pair<std::string, u32>> serVec;
881 : DevType devType;
882 5 : CHK_RET(hrtGetDeviceType(devType));
883 :
884 61 : for (u32 index = 0; index < rankInfos.size(); index++) {
885 56 : auto rankInfo = rankInfos[index];
886 56 : if (rankInfo.serverId == locRank.serverId) {
887 45 : devVec.push_back(std::make_pair(rankInfo.devicePhyId, index));
888 : }
889 56 : if (rankInfo.devicePhyId == locRank.devicePhyId) {
890 11 : serVec.push_back(std::make_pair(rankInfo.serverId, index));
891 : }
892 56 : }
893 : // server内单环dev排布, 为兼容310P(devId为0, 2, 4...), 扩展为16
894 : int* ringConfig;
895 5 : int ringConfig910A[16] = {0, 3, 1, 2, 7, 4, 6, 5, 4, 6, 2, 0, 3, 1, 5, 7};
896 5 : int ringConfig910B[16] = {0, 1, 2, 3, 4, 5, 6, 7, 15, 14, 13, 12, 11, 10, 9, 8};
897 :
898 5 : ringConfig = ringConfig910A;
899 5 : if (devType == DevType::DEV_TYPE_910B || devType == DevType::DEV_TYPE_310P3
900 5 : || devType == DevType::DEV_TYPE_910_93) {
901 1 : ringConfig = ringConfig910B;
902 : }
903 5 : std::sort(devVec.begin(), devVec.end(), [&](const std::pair<u32, u32> p1, const std::pair<u32, u32> p2) {
904 167 : return ringConfig[p1.first] < ringConfig[p2.first];
905 : });
906 :
907 5 : std::sort(
908 12 : serVec.begin(), serVec.end(), [](const std::pair<std::string, u32>& p1, const std::pair<std::string, u32>& p2) {
909 12 : return p1.first < p2.first;
910 : });
911 5 : u32 locDevId = locRank.devicePhyId;
912 5 : u32 worldRank = locRank.worldRank;
913 :
914 5 : HcclSocketType devSocketType = ((devType == DevType::DEV_TYPE_910B) && isUsedRdma) ? HcclSocketType::SOCKET_NIC :
915 : HcclSocketType::SOCKET_VNIC;
916 5 : CHK_RET(
917 : GetSamePlaneConnInfo(devSocketType, devVec, locDevId, rankInfos, needConnectRank, useSuperPodMode, worldRank));
918 :
919 5 : auto nodeId = locRank.serverId;
920 5 : CHK_RET(GetSamePlaneConnInfo(
921 : HcclSocketType::SOCKET_NIC, serVec, nodeId, rankInfos, needConnectRank, useSuperPodMode, worldRank));
922 5 : return HCCL_SUCCESS;
923 5 : }
924 :
925 1 : void Heartbeat::AddOpInfo(const std::string& identifier, const OpInfoDesc& opInfo, const std::string& paramTag)
926 : {
927 1 : if (!opInfo.isValid || opInfo.opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV) {
928 : // 若当前opInfo为无效值或者为batchsendrecv算子时，无需添加
929 0 : return;
930 : }
931 : // 添加一个opInfo到发送队列中
932 1 : OpInfoDesc opInfoTmp = opInfo;
933 1 : std::string tag;
934 1 : if (opInfo.opType == HcclCMDType::HCCL_CMD_SEND || opInfo.opType == HcclCMDType::HCCL_CMD_RECEIVE) {
935 0 : RegisterSROpIdentifier(identifier, paramTag);
936 0 : tag = paramTag;
937 : } else {
938 1 : tag = identifier;
939 : }
940 1 : std::lock_guard<std::mutex> lock(opInfoQueueMutex_);
941 :
942 1 : auto opInfoIndexIter = opInfoIndexMap_.find(tag);
943 1 : if (opInfoIndexIter == opInfoIndexMap_.end()) {
944 1 : opInfoIndexMap_.insert(std::make_pair(tag, 1));
945 1 : opInfoTmp.index = 1;
946 : } else {
947 0 : opInfoTmp.index = ++(opInfoIndexIter->second);
948 : }
949 1 : opInfoQueue_.push_back(std::make_pair(tag, opInfoTmp));
950 1 : HCCL_DEBUG(
951 : "[Heartbeat][AddOpInfo]opType[%d], dataType[%d], reduce[%d], count[%llu], root[%d], tag[%s], index[%llu] add "
952 : "success",
953 : opInfoTmp.opType, opInfoTmp.dataType, opInfoTmp.reduceOp, opInfoTmp.count, opInfoTmp.root, tag.c_str(),
954 : opInfoTmp.index);
955 :
956 : // 限制发送队列的长度，防止内存逐渐溢出
957 1 : if (opInfoQueue_.size() > OPINFO_QUEUE_MAX_SIZE) {
958 0 : opInfoQueue_.pop_front();
959 : }
960 1 : return;
961 1 : }
962 :
963 3 : void Heartbeat::GetOneOpInfo(std::string& tag, OpInfoDesc& opInfo)
964 : {
965 : // 从发送队列中获取一个opInfo发送给对端
966 3 : std::unique_lock<std::mutex> lock(opInfoQueueMutex_);
967 3 : if (opInfoQueue_.empty()) {
968 : static OpInfoDesc defaultOpInfo;
969 2 : opInfo = defaultOpInfo;
970 2 : return;
971 : }
972 1 : auto opInfoPair = opInfoQueue_.front();
973 1 : opInfoQueue_.pop_front();
974 1 : lock.unlock();
975 :
976 1 : tag = opInfoPair.first;
977 1 : opInfo = opInfoPair.second;
978 1 : std::unique_lock<std::mutex> mapLock(opInfoMapMutex_);
979 1 : if (opInfoMap_.find(tag) == opInfoMap_.end()) {
980 1 : std::map<u64, OpInfoDesc> opInfoList;
981 1 : opInfoList.insert(std::make_pair(opInfo.index, opInfo));
982 1 : opInfoMap_.insert(std::make_pair(tag, opInfoList));
983 1 : } else {
984 0 : opInfoMap_[tag].insert(std::make_pair(opInfo.index, opInfo));
985 : }
986 :
987 : // 限制发送队列的长度，防止内存逐渐溢出
988 1 : while (opInfoMap_[tag].size() > OPINFO_QUEUE_MAX_SIZE) {
989 : // 删除index最小的数据，防止内存不断增加
990 0 : auto smallIt = opInfoMap_[tag].begin();
991 0 : opInfoMap_[tag].erase(smallIt);
992 : }
993 :
994 1 : HCCL_DEBUG(
995 : "[Heartbeat][GetOneOpInfo]opType[%d], dataType[%d], reduce[%d], count[%llu], root[%d], tag[%s], "
996 : "index[%llu] get success",
997 : opInfo.opType, opInfo.dataType, opInfo.reduceOp, opInfo.count, opInfo.root, tag.c_str(), opInfo.index);
998 1 : return;
999 3 : }
1000 :
1001 1 : void Heartbeat::GetSendOpInfoList(OpInfoTagQueueFrame& opInfoTagQueueFrame)
1002 : {
1003 1 : if (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) {
1004 0 : return;
1005 : }
1006 1 : while (opInfoQueueForSend_.size() < OPINFO_TAG_QUEUE_NUM * OPINFO_SEND_NUM_BY_TAG) {
1007 1 : OpInfoDesc opInfo;
1008 1 : std::string tag;
1009 1 : GetOneOpInfo(tag, opInfo);
1010 1 : if (opInfo.isValid) {
1011 0 : opInfoQueueForSend_.push_back(std::make_pair(tag, opInfo));
1012 : } else {
1013 1 : break;
1014 : }
1015 1 : }
1016 :
1017 1 : HCCL_DEBUG("[%s] opInfoQueueForSend_.size[%d] begin", __func__, opInfoQueueForSend_.size());
1018 1 : auto& opInfoTagQueue = opInfoTagQueueFrame.opInfoTagQueue;
1019 2 : for (auto iter = opInfoQueueForSend_.begin(); iter != opInfoQueueForSend_.end();) {
1020 1 : bool isAdd = false;
1021 1 : for (u32 index = 0; index < OPINFO_TAG_QUEUE_NUM; index++) {
1022 : // 当前 index 对应的 opInfoTagQueue 为未初始化状态
1023 1 : if (strncmp(opInfoTagQueue[index].identifier, "\0", ROOTINFO_INDENTIFIER_MAX_LENGTH) == 0) {
1024 3 : s32 memcpyRet = memcpy_s(
1025 1 : opInfoTagQueue[index].identifier, ROOTINFO_INDENTIFIER_MAX_LENGTH, iter->first.c_str(),
1026 1 : iter->first.size() + 1);
1027 1 : if (memcpyRet != EOK) {
1028 0 : HCCL_WARNING(
1029 : "[%s]copy tag[%s] to opInfoTagQueue failed, ret[%d]", __func__, iter->first.c_str(), memcpyRet);
1030 0 : break;
1031 : }
1032 1 : opInfoTagQueue[index].opInfoList[opInfoTagQueue[index].opInfoNum] = iter->second;
1033 1 : opInfoTagQueue[index].opInfoNum++;
1034 1 : isAdd = true;
1035 1 : HCCL_DEBUG(
1036 : "[%s]opInfoTagQueue[%d] add success identifier[%s] ", __func__, index,
1037 : opInfoTagQueue[index].identifier);
1038 1 : break;
1039 : }
1040 : // 当前 index 对应的 opInfoTagQueue 已经被某个tag 的算子占用
1041 0 : else if (
1042 0 : strncmp(opInfoTagQueue[index].identifier, iter->first.c_str(), ROOTINFO_INDENTIFIER_MAX_LENGTH) == 0) {
1043 0 : if (opInfoTagQueue[index].opInfoNum < OPINFO_SEND_NUM_BY_TAG) {
1044 0 : opInfoTagQueue[index].opInfoList[opInfoTagQueue[index].opInfoNum] = iter->second;
1045 0 : opInfoTagQueue[index].opInfoNum++;
1046 0 : isAdd = true;
1047 0 : HCCL_DEBUG(
1048 : "[%s]opInfoTagQueue[%d] has exists and add success identifier[%s] ", __func__, index,
1049 : opInfoTagQueue[index].identifier);
1050 0 : break;
1051 : }
1052 : }
1053 : }
1054 1 : if (isAdd) {
1055 1 : iter = opInfoQueueForSend_.erase(iter);
1056 : } else {
1057 0 : iter++; // opInfoQueueForSend_ 残留数据会被保存到下一轮 GetSendOpInfoList
1058 : }
1059 : }
1060 1 : return;
1061 : }
1062 :
1063 6 : void Heartbeat::SaveOpInfo(const OpInfoTagQueueFrame& opInfoTagQueueFrame, UIDType& src)
1064 : {
1065 6 : const auto& opInfoTagQueue = opInfoTagQueueFrame.opInfoTagQueue;
1066 66 : for (u32 index = 0; index < OPINFO_TAG_QUEUE_NUM; index++) {
1067 60 : std::string tag = std::string(opInfoTagQueue[index].identifier);
1068 64 : for (u32 num = 0; num < opInfoTagQueue[index].opInfoNum; num++) {
1069 4 : std::unique_lock<std::mutex> lock(opInfoMapMutex_);
1070 : // 保存接收到的opInfo到接收队列中
1071 4 : auto& opInfo = opInfoTagQueue[index].opInfoList[num];
1072 4 : recvOpInfoList_.push_back(std::make_tuple(opInfo, tag, src));
1073 4 : HCCL_DEBUG(
1074 : "[Heartbeat][%s]tag[%s], opType[%d], dataType[%d], reduce[%d], count[%u], root[%d], index[%llu] get "
1075 : "success",
1076 : __func__, tag.c_str(), opInfo.opType, opInfo.dataType, opInfo.reduceOp, opInfo.count, opInfo.root,
1077 : opInfo.index);
1078 4 : }
1079 60 : }
1080 6 : std::unique_lock<std::mutex> lock(opInfoMapMutex_);
1081 6 : while (recvOpInfoList_.size() > OPINFO_QUEUE_MAX_SIZE) { // 可能存在误丢
1082 0 : recvOpInfoList_.pop_front();
1083 : }
1084 :
1085 12 : return;
1086 6 : }
1087 :
1088 : HcclResult
1089 6 : Heartbeat::CheckIsSameOp(const OpInfoDesc& localOpInfo, const OpInfoDesc& remoteOpInfo, InconsistentType& status)
1090 : {
1091 6 : if (localOpInfo.opType == HcclCMDType::HCCL_CMD_SEND) {
1092 2 : if (remoteOpInfo.opType != HcclCMDType::HCCL_CMD_RECEIVE) {
1093 1 : status = InconsistentType::OPTYPE_INCONSISTENT;
1094 1 : return HCCL_SUCCESS;
1095 : }
1096 4 : } else if (localOpInfo.opType == HcclCMDType::HCCL_CMD_RECEIVE) {
1097 1 : if (remoteOpInfo.opType != HcclCMDType::HCCL_CMD_SEND) {
1098 1 : status = InconsistentType::OPTYPE_INCONSISTENT;
1099 1 : return HCCL_SUCCESS;
1100 : }
1101 3 : } else if (localOpInfo.opType != remoteOpInfo.opType) {
1102 1 : status = InconsistentType::OPTYPE_INCONSISTENT;
1103 1 : return HCCL_SUCCESS;
1104 : }
1105 :
1106 3 : if (localOpInfo.dataType != remoteOpInfo.dataType) {
1107 1 : status = InconsistentType::DATATYPE_INCONSISTENT;
1108 1 : return HCCL_SUCCESS;
1109 : }
1110 :
1111 2 : if (localOpInfo.reduceOp != remoteOpInfo.reduceOp) {
1112 0 : status = InconsistentType::REDUCETYPE_INCONSISTENT;
1113 0 : return HCCL_SUCCESS;
1114 : }
1115 :
1116 2 : if (localOpInfo.root != remoteOpInfo.root) {
1117 0 : status = InconsistentType::ROOT_INCONSISTENT;
1118 0 : return HCCL_SUCCESS;
1119 : }
1120 :
1121 2 : if (localOpInfo.opType != HcclCMDType::HCCL_CMD_ALLGATHER_V && localOpInfo.opType != HcclCMDType::HCCL_CMD_ALLTOALLV
1122 2 : && localOpInfo.opType != HcclCMDType::HCCL_CMD_ALLTOALLVC
1123 2 : && localOpInfo.opType != HcclCMDType::HCCL_CMD_REDUCE_SCATTER_V) {
1124 : // 仅对数据量均等的算子进行校验数据量count
1125 2 : if (localOpInfo.count != remoteOpInfo.count) {
1126 1 : status = InconsistentType::COUNT_INCONSISTENT;
1127 1 : return HCCL_SUCCESS;
1128 : }
1129 : }
1130 1 : status = InconsistentType::NO_INCONSISTENT;
1131 1 : return HCCL_SUCCESS;
1132 : }
1133 :
1134 5 : void Heartbeat::CheckRecvOpInfoList()
1135 : {
1136 5 : if (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) {
1137 5 : return;
1138 : }
1139 : // 校验接收队列中接收到的opInfo
1140 0 : std::unique_lock<std::mutex> lock(opInfoMapMutex_);
1141 0 : for (auto it = recvOpInfoList_.begin(); it != recvOpInfoList_.end();) {
1142 0 : const auto& opInfoRecv = std::get<0>(*it);
1143 0 : const auto& identifier = std::get<1>(*it);
1144 0 : const auto& uid = std::get<2>(*it);
1145 0 : auto opInfoIndexMap = opInfoMap_.find(identifier);
1146 0 : if (opInfoIndexMap == opInfoMap_.end()) {
1147 0 : ++it;
1148 0 : HCCL_DEBUG(
1149 : "[Heartbeat]check recv not found. identifier[%s] index[%u]", identifier.c_str(), opInfoRecv.index);
1150 0 : continue;
1151 : }
1152 :
1153 0 : if (opInfoIndexMap->second.find(opInfoRecv.index) != opInfoIndexMap->second.end()) {
1154 0 : const auto& opInfo = opInfoIndexMap->second[opInfoRecv.index];
1155 0 : InconsistentType inconsistent = InconsistentType::NO_INCONSISTENT;
1156 0 : CheckIsSameOp(opInfo, opInfoRecv, inconsistent);
1157 0 : if (inconsistent != InconsistentType::NO_INCONSISTENT) {
1158 : // 当算子不匹配时，记录并打印ERROR日志并广播下发不一致错误给其他节点
1159 : char localInfo[LOG_TMPBUF_SIZE];
1160 0 : s32 ret = snprintf_s(
1161 : localInfo, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1162 0 : "node[%s] optype[%s] dataType[%s] reduceOp[%s] count[%d] root[%d]", FormatUId(uid_).c_str(),
1163 0 : GetCMDTypeEnumStr(opInfo.opType).c_str(), GetDataTypeEnumStr(opInfo.dataType).c_str(),
1164 0 : GetReduceOpEnumStr(opInfo.reduceOp).c_str(), opInfo.count, opInfo.root);
1165 0 : CHK_PRT_CONT(ret == -1, HCCL_ERROR("Failed to build log info"));
1166 : char remoteInfo[LOG_TMPBUF_SIZE];
1167 0 : ret = snprintf_s(
1168 : remoteInfo, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1169 0 : "node[%s] optype[%s] dataType[%s] reduceOp[%s] count[%lu] root[%u]", FormatUId(uid).c_str(),
1170 0 : GetCMDTypeEnumStr(opInfoRecv.opType).c_str(), GetDataTypeEnumStr(opInfoRecv.dataType).c_str(),
1171 0 : GetReduceOpEnumStr(opInfoRecv.reduceOp).c_str(), opInfoRecv.count, opInfoRecv.root);
1172 0 : CHK_PRT_CONT(ret == -1, HCCL_ERROR("Failed to build log info"));
1173 :
1174 0 : AddInconsistentOpRecord(
1175 0 : identifier, opInfo, inconsistent, std::string(localInfo), std::string(remoteInfo));
1176 0 : HCCL_ERROR(
1177 : "[Heartbeat]check opinfo inconsistent. identifier[%s] index[%u], "
1178 : "local(%s); remote(%s)",
1179 : identifier.c_str(), opInfoRecv.index, localInfo, remoteInfo);
1180 0 : SetStatus(uid_, uid_, HeartBeatStatus::HEARTBEAT_INCONSISTENT);
1181 : }
1182 : // 校验完成后删除收到的opInfo
1183 0 : it = recvOpInfoList_.erase(it);
1184 : } else {
1185 : // 若在opInfoIndexMap中没有找到相同index的算子，先跳到该记录，校验下一个收到的算子
1186 0 : ++it;
1187 : }
1188 : }
1189 0 : return;
1190 0 : }
1191 :
1192 4 : HcclResult Heartbeat::SendFrame(UIDType& dst, UIDType& crimer, UIDType& informer, HeartBeatStatus status)
1193 : {
1194 4 : HeartBeatFrame bf(uid_, dst, crimer, informer, status);
1195 4 : if (rankId2SocketMap_[dst].sendBuffer.size() > 0) {
1196 1 : if (status != HeartBeatStatus::HEARTBEAT_OK && rankId2SocketMap_[dst].sendBuffer.size() < MAX_SENDBUFF_SIZE) {
1197 1 : rankId2SocketMap_[dst].sendBuffer.push(bf);
1198 : }
1199 3 : while (rankId2SocketMap_[dst].sendBuffer.size() > 0) {
1200 2 : HeartBeatFrame hbf = rankId2SocketMap_[dst].sendBuffer.front();
1201 2 : u64 sendDis = sizeof(HeartBeatFrame) - rankId2SocketMap_[dst].restSize;
1202 2 : u64 compSize = 0;
1203 2 : HcclResult ret = rankId2SocketMap_[dst].socket->ISend(
1204 2 : reinterpret_cast<void*>(reinterpret_cast<uintptr_t>(&hbf) + sendDis), rankId2SocketMap_[dst].restSize,
1205 : compSize);
1206 2 : if (ret != HCCL_SUCCESS) {
1207 0 : return ret;
1208 : }
1209 2 : if (rankId2SocketMap_[dst].restSize == compSize) {
1210 2 : rankId2SocketMap_[dst].sendBuffer.pop();
1211 2 : rankId2SocketMap_[dst].restSize = sizeof(HeartBeatFrame);
1212 2 : HCCL_DEBUG(
1213 : "[Heartbeat][SendFrame] Send Success, from [%s] to [%s] about [%s] by [%s] status[%d]",
1214 : FormatUId(uid_).c_str(), FormatUId(dst).c_str(), FormatUId(crimer).c_str(),
1215 : FormatUId(informer).c_str(), status);
1216 : } else {
1217 0 : rankId2SocketMap_[dst].restSize = rankId2SocketMap_[dst].restSize - compSize;
1218 0 : break;
1219 : }
1220 : }
1221 : } else {
1222 3 : u64 compSize = 0;
1223 3 : u32 expectSize = sizeof(HeartBeatFrame);
1224 3 : HcclResult ret = rankId2SocketMap_[dst].socket->ISend(&bf, expectSize, compSize);
1225 3 : if (ret != HCCL_SUCCESS) {
1226 0 : return ret;
1227 : }
1228 3 : if (compSize == expectSize) {
1229 2 : HCCL_DEBUG(
1230 : "[Heartbeat][SendFrame] Send Success, from [%s] to [%s] about [%s] by [%s] status[%d]",
1231 : FormatUId(uid_).c_str(), FormatUId(dst).c_str(), FormatUId(crimer).c_str(), FormatUId(informer).c_str(),
1232 : status);
1233 : } else {
1234 1 : HCCL_DEBUG(
1235 : "[Heartbeat][SendFrame] Send Not Complete, from [%s] to [%s] about [%s] by [%s] status[%d], "
1236 : "expectSize[%u], compSize[%u]",
1237 : FormatUId(uid_).c_str(), FormatUId(dst).c_str(), FormatUId(crimer).c_str(), FormatUId(informer).c_str(),
1238 : status, expectSize, compSize);
1239 1 : rankId2SocketMap_[dst].restSize = expectSize - compSize;
1240 1 : rankId2SocketMap_[dst].sendBuffer.push(bf);
1241 : }
1242 : }
1243 4 : return HCCL_SUCCESS;
1244 : }
1245 :
1246 0 : HcclResult Heartbeat::SendFrameWithOpCheck(
1247 : UIDType& dst, UIDType& crimer, UIDType& informer, HeartBeatStatus status,
1248 : const OpInfoTagQueueFrame& opInfoTagQueueFrame)
1249 : {
1250 0 : HeartBeatFrameWithOpCheck bf(uid_, dst, crimer, informer, status);
1251 0 : bf.opInfoTagQueueFrame = opInfoTagQueueFrame;
1252 :
1253 0 : if (rankId2SocketMap_[dst].sendBufferWithOpCheck.size() > 0) {
1254 0 : if (status != HeartBeatStatus::HEARTBEAT_OK
1255 0 : && rankId2SocketMap_[dst].sendBufferWithOpCheck.size() < MAX_SENDBUFF_SIZE) {
1256 0 : rankId2SocketMap_[dst].sendBufferWithOpCheck.push(bf);
1257 : }
1258 : } else {
1259 0 : rankId2SocketMap_[dst].sendBufferWithOpCheck.push(bf);
1260 0 : rankId2SocketMap_[dst].restSize = sizeof(HeartBeatFrameWithOpCheck);
1261 : }
1262 : // 查询到某个Dst的发送缓冲数据量
1263 0 : u32 unCompletedCount = 0; // 已经发送的loop次数
1264 0 : while (rankId2SocketMap_[dst].sendBufferWithOpCheck.size() > 0) {
1265 0 : HeartBeatFrameWithOpCheck hbf = rankId2SocketMap_[dst].sendBufferWithOpCheck.front();
1266 0 : u64 sendDis = sizeof(HeartBeatFrameWithOpCheck) - rankId2SocketMap_[dst].restSize;
1267 0 : u64 compSize = 0;
1268 0 : HcclResult ret = rankId2SocketMap_[dst].socket->ISend(
1269 0 : reinterpret_cast<void*>(reinterpret_cast<uintptr_t>(&hbf) + sendDis), rankId2SocketMap_[dst].restSize,
1270 : compSize);
1271 0 : if (ret != HCCL_SUCCESS) {
1272 0 : return ret;
1273 : }
1274 0 : if (rankId2SocketMap_[dst].restSize == compSize) {
1275 0 : rankId2SocketMap_[dst].sendBufferWithOpCheck.pop();
1276 0 : rankId2SocketMap_[dst].restSize = sizeof(HeartBeatFrameWithOpCheck);
1277 0 : HCCL_DEBUG(
1278 : "[Heartbeat][%s] Send Success, from [%s] to [%s] about [%s] by [%s] status[%d]", __func__,
1279 : FormatUId(uid_).c_str(), FormatUId(dst).c_str(), FormatUId(crimer).c_str(), FormatUId(informer).c_str(),
1280 : status);
1281 : } else {
1282 0 : HCCL_DEBUG(
1283 : "[Heartbeat][%s] Send Not Complete, from [%s] to [%s] about [%s] by [%s] status[%d], expectSize[%u], "
1284 : "compSize[%u]",
1285 : __func__, FormatUId(uid_).c_str(), FormatUId(dst).c_str(), FormatUId(crimer).c_str(),
1286 : FormatUId(informer).c_str(), status, rankId2SocketMap_[dst].restSize, compSize);
1287 0 : rankId2SocketMap_[dst].restSize = rankId2SocketMap_[dst].restSize - compSize;
1288 0 : unCompletedCount++;
1289 0 : SaluSleep(ONE_HUNDRED_MICROSECOND_OF_USLEEP); // 100us
1290 : // 限制发送的循环此时，避免在send流程里死循环
1291 0 : if (unCompletedCount > HBFRAME_SEND_LOOP_MAX_NUM) {
1292 0 : break; // 120个loop约30毫秒
1293 : }
1294 : }
1295 : }
1296 0 : return HCCL_SUCCESS;
1297 : }
1298 :
1299 0 : HcclResult Heartbeat::RecvFrame(UIDType& src)
1300 : {
1301 0 : HeartBeatFrame bf;
1302 0 : u64 compSize = 0;
1303 0 : u64 expectSize = sizeof(HeartBeatFrame);
1304 : while (true) {
1305 0 : compSize = 0;
1306 0 : HcclResult retVal = rankId2SocketMap_[src].socket->IRecv(&bf, expectSize, compSize);
1307 0 : if (retVal == HCCL_SUCCESS && compSize > 0) {
1308 0 : rankId2SocketMap_[src].recvBuffer.PushSeg(reinterpret_cast<u8*>(&bf), compSize);
1309 0 : if (rankId2SocketMap_[src].recvBuffer.Size() >= expectSize) {
1310 0 : rankId2SocketMap_[src].recvBuffer.GetSeg(reinterpret_cast<u8*>(&bf), expectSize);
1311 0 : rankId2SocketMap_[src].recvBuffer.PopSeg(expectSize);
1312 0 : CHK_RET(ParseFrame(bf, src));
1313 : }
1314 0 : } else if (retVal == HCCL_E_INTERNAL) {
1315 0 : return HCCL_E_INTERNAL;
1316 : } else {
1317 0 : break;
1318 : }
1319 0 : }
1320 0 : return HCCL_SUCCESS;
1321 : }
1322 :
1323 0 : HcclResult Heartbeat::RecvFrameWithOpCheck(UIDType& src)
1324 : {
1325 0 : HeartBeatFrameWithOpCheck bf;
1326 0 : u64 compSize = 0;
1327 0 : u64 expectSize = sizeof(HeartBeatFrameWithOpCheck);
1328 : while (true) {
1329 0 : compSize = 0;
1330 0 : HcclResult retVal = rankId2SocketMap_[src].socket->IRecv(&bf, expectSize, compSize);
1331 0 : if (retVal == HCCL_SUCCESS && compSize > 0) {
1332 0 : rankId2SocketMap_[src].recvBuffer.PushSeg(reinterpret_cast<u8*>(&bf), compSize);
1333 : // 标识当前Recvbuf中已经存放了一个完整的帧
1334 0 : if (rankId2SocketMap_[src].recvBuffer.Size() >= expectSize) {
1335 0 : rankId2SocketMap_[src].recvBuffer.GetSeg(reinterpret_cast<u8*>(&bf), expectSize);
1336 0 : rankId2SocketMap_[src].recvBuffer.PopSeg(expectSize);
1337 0 : CHK_RET(ParseFrameWithOpCheck(bf, src));
1338 0 : break;
1339 : }
1340 0 : } else if (retVal == HCCL_E_INTERNAL) {
1341 0 : return HCCL_E_INTERNAL;
1342 : } else {
1343 0 : break;
1344 : }
1345 0 : }
1346 0 : return HCCL_SUCCESS;
1347 : }
1348 :
1349 2 : HcclResult Heartbeat::ParseFrame(HeartBeatFrame& bf, UIDType& src)
1350 : {
1351 2 : if (bf.src != src || bf.dst != uid_) {
1352 0 : HCCL_WARNING("rank[%s] recv wrong frame", FormatUId(uid_).c_str());
1353 0 : return HCCL_E_INTERNAL;
1354 : }
1355 :
1356 2 : HCCL_DEBUG(
1357 : "[Heartbeat][RecvFrame] Recv Success, from [%s] to [%s] about [%s] by [%s] state[%d]",
1358 : FormatUId(bf.src).c_str(), FormatUId(bf.dst).c_str(), FormatUId(bf.crimer).c_str(),
1359 : FormatUId(bf.informer).c_str(), bf.status);
1360 :
1361 : // 能够收到进程卡住表示心跳是正常的
1362 2 : if (bf.status == HeartBeatStatus::HEARTBEAT_OK || bf.status == HeartBeatStatus::HEARTBEAT_STUCK) {
1363 2 : rankId2SocketMap_[src].lostNum = 0;
1364 2 : rankId2SocketMap_[src].lostReportCnt = 0;
1365 : }
1366 :
1367 : // 只有心跳非正常时才需要打印TRACE
1368 2 : if (bf.status != HeartBeatStatus::HEARTBEAT_OK) {
1369 1 : SetStatus(bf.crimer, bf.informer, bf.status);
1370 : }
1371 :
1372 2 : return HCCL_SUCCESS;
1373 : }
1374 :
1375 2 : HcclResult Heartbeat::ParseFrameWithOpCheck(HeartBeatFrameWithOpCheck& bf, UIDType& src)
1376 : {
1377 2 : if (bf.src != src || bf.dst != uid_) {
1378 0 : HCCL_WARNING("rank[%s] recv wrong frame", FormatUId(uid_).c_str());
1379 0 : return HCCL_E_INTERNAL;
1380 : }
1381 :
1382 2 : HCCL_DEBUG(
1383 : "[Heartbeat][RecvFrame] Recv Success, from [%s] to [%s] about [%s] by [%s] state[%d]",
1384 : FormatUId(bf.src).c_str(), FormatUId(bf.dst).c_str(), FormatUId(bf.crimer).c_str(),
1385 : FormatUId(bf.informer).c_str(), bf.status);
1386 :
1387 2 : if (bf.status == HeartBeatStatus::HEARTBEAT_OK || bf.status == HeartBeatStatus::HEARTBEAT_STUCK) {
1388 2 : rankId2SocketMap_[src].lostNum = 0;
1389 2 : rankId2SocketMap_[src].lostReportCnt = 0;
1390 : }
1391 :
1392 2 : if (bf.status != HeartBeatStatus::HEARTBEAT_OK) {
1393 1 : SetStatus(bf.crimer, bf.informer, bf.status);
1394 : }
1395 :
1396 2 : SaveOpInfo(bf.opInfoTagQueueFrame, src);
1397 2 : return HCCL_SUCCESS;
1398 : }
1399 :
1400 7 : void Heartbeat::SetStatus(UIDType& crimer, UIDType& informer, HeartBeatStatus status, bool needBroadcast)
1401 : {
1402 7 : if (rankId2StatusMap_[crimer].status != status) {
1403 5 : rankId2StatusMap_[crimer].informer = informer;
1404 5 : rankId2StatusMap_[crimer].status = status;
1405 5 : rankId2StatusMap_[crimer].needBroadcast = needBroadcast;
1406 5 : if (needBroadcast) {
1407 1 : errRankQueue_.push(crimer);
1408 : }
1409 :
1410 5 : errStatusQueue_.push(HeartBeatFrame(crimer, informer, status, TIME_NOW(), std::chrono::system_clock::now()));
1411 5 : if (errStatusQueue_.size() > EVENT_MAX_CNT) {
1412 0 : errStatusQueue_.pop();
1413 : }
1414 5 : HCCL_RUN_INFO(
1415 : "[%s][%s]local rank [%s]: crimer rank [%s] status[%s] by informer rank [%s]",
1416 : LOG_KEYWORDS_TASK_EXEC.c_str(), LOG_KEYWORDS_HEARTBEAT_EVETN.c_str(), FormatUId(uid_).c_str(),
1417 : FormatUId(crimer).c_str(), GetHeartBeatStatusStr(status).c_str(), FormatUId(informer).c_str());
1418 : }
1419 7 : }
1420 :
1421 4 : bool Heartbeat::IsKeyEvent(HeartBeatFrame& event, HcclUs curTime, const std::string& group)
1422 : {
1423 4 : bool ret = false;
1424 4 : s64 intervalTime = DURATION_US(curTime - event.TOARelative).count() / (TIME_S_TO_MS * ONE_MILLISECOND_OF_USLEEP);
1425 4 : s32 hcclExecTimeout = CommConfiger::GetInstance().GetCommConfigExecTimeOut(group);
1426 4 : s64 execTimeout = hcclExecTimeout;
1427 4 : s64 detectionTime = 0;
1428 4 : switch (event.status) {
1429 1 : case HeartBeatStatus::HEARTBEAT_LOST:
1430 1 : detectionTime = (lostThreshold_ * HEARTBEAT_INTERVAL) / TIME_S_TO_MS;
1431 1 : break;
1432 3 : case HeartBeatStatus::HEARTBEAT_CQE_ERR:
1433 : case HeartBeatStatus::HEARTBEAT_INCONSISTENT:
1434 : case HeartBeatStatus::HEARTBEAT_OPRETRY_NOT_SUPPORT:
1435 3 : detectionTime = 0;
1436 3 : break;
1437 0 : case HeartBeatStatus::HEARTBEAT_STUCK:
1438 0 : detectionTime = 2 * stuckDetectTime_; // 最长探测时间为2倍的卡住检测时间
1439 0 : break;
1440 0 : case HeartBeatStatus::HEARTBEAT_NOTIFY:
1441 : default:
1442 0 : return false; // 当前不支持的事件，不做处理和展现
1443 : }
1444 8 : ret = ((execTimeout - intervalTime - detectionTime) < JITTER_TIME)
1445 4 : && ((intervalTime + detectionTime - execTimeout) < JITTER_TIME);
1446 4 : return ret;
1447 : }
1448 :
1449 96 : void Heartbeat::MakeErrMsg(std::queue<HeartBeatFrame>& keyEvents, std::vector<std::string>& errStatusVec)
1450 : {
1451 99 : while (keyEvents.size() > 0) {
1452 3 : auto& tmp = keyEvents.front();
1453 3 : std::string crimerStr = FormatUId(tmp.crimer);
1454 3 : std::string informerStr = FormatUId(tmp.informer);
1455 :
1456 6 : std::string headStr = "[" + LOG_KEYWORDS_TASK_EXEC + "][" + LOG_KEYWORDS_HEARTBEAT_EVETN + "]"
1457 3 : + "Cluster Exception Location[IP/ID]:[";
1458 :
1459 3 : time_t tm = std::chrono::system_clock::to_time_t(tmp.TOASystem);
1460 3 : std::string timeStr(ctime(&tm));
1461 3 : if (!timeStr.empty()) { // ctime()函数自带换行符，需要去掉
1462 3 : timeStr.pop_back();
1463 : }
1464 3 : timeStr = ", Arrival Time:[" + timeStr + "]";
1465 :
1466 6 : std::string errStr = ", ExceptionType:";
1467 3 : std::string reasonStr = ", Possible Reason:";
1468 3 : switch (tmp.status) {
1469 1 : case HeartBeatStatus::HEARTBEAT_LOST:
1470 1 : errStr = errStr + "[Heartbeat Lost Occurred]";
1471 1 : reasonStr = reasonStr + "1. Process has exited, 2. Network Disconnected";
1472 : errStr
1473 1 : = headStr + crimerStr + "]" + timeStr + ", Discoverer:[" + informerStr + "]" + errStr + reasonStr;
1474 1 : break;
1475 0 : case HeartBeatStatus::HEARTBEAT_NOTIFY:
1476 0 : errStr = errStr + "[Notify Wait Error Occurred]";
1477 0 : errStr = headStr + crimerStr + "]" + timeStr + errStr;
1478 0 : break;
1479 1 : case HeartBeatStatus::HEARTBEAT_OPRETRY_NOT_SUPPORT:
1480 1 : errStr = errStr + "[OpRetry Not Supported Occurred]";
1481 1 : reasonStr = reasonStr + "OpRetry is not supported";
1482 1 : errStr = headStr + crimerStr + "]" + timeStr + errStr + reasonStr;
1483 1 : break;
1484 1 : case HeartBeatStatus::HEARTBEAT_CQE_ERR:
1485 1 : errStr = errStr + "[Error cqe Occurred]";
1486 1 : reasonStr = reasonStr + "1.Network Disconnected, 2.Remote Rank Coredown";
1487 1 : errStr = headStr + crimerStr + "]" + timeStr + errStr + reasonStr;
1488 1 : break;
1489 0 : case HeartBeatStatus::HEARTBEAT_STUCK:
1490 0 : errStr = errStr + "[Stuck Occurred]";
1491 0 : reasonStr = reasonStr + "1.Host process is stuck, 2.Device task is stuck";
1492 0 : errStr = headStr + crimerStr + "]" + timeStr + errStr + reasonStr;
1493 0 : break;
1494 0 : case HeartBeatStatus::HEARTBEAT_INCONSISTENT:
1495 0 : errStr = errStr + "[Op Inconsistent Occurred]";
1496 0 : reasonStr = reasonStr + "communication operator is inconsistent";
1497 0 : errStr = headStr + crimerStr + "]" + timeStr + errStr + reasonStr;
1498 0 : break;
1499 0 : default:
1500 0 : errStr = " Unknown";
1501 : }
1502 3 : errStatusVec.emplace_back(errStr);
1503 3 : keyEvents.pop();
1504 3 : }
1505 96 : }
1506 19 : std::vector<std::string> Heartbeat::PrintEvents(std::map<HeartBeatStatus, std::queue<HeartBeatFrame>>& keyEvents)
1507 : {
1508 19 : std::vector<std::string> errStatusVec;
1509 : // 打印优先级 opretry not support > error cqe > stuck > lost
1510 19 : MakeErrMsg(keyEvents[HeartBeatStatus::HEARTBEAT_OPRETRY_NOT_SUPPORT], errStatusVec);
1511 19 : MakeErrMsg(keyEvents[HeartBeatStatus::HEARTBEAT_CQE_ERR], errStatusVec);
1512 19 : MakeErrMsg(keyEvents[HeartBeatStatus::HEARTBEAT_STUCK], errStatusVec);
1513 19 : MakeErrMsg(keyEvents[HeartBeatStatus::HEARTBEAT_LOST], errStatusVec);
1514 19 : MakeErrMsg(keyEvents[HeartBeatStatus::HEARTBEAT_INCONSISTENT], errStatusVec);
1515 19 : return errStatusVec;
1516 0 : }
1517 19 : std::vector<std::string> Heartbeat::GetErrStatusVec(const std::string& group)
1518 : {
1519 19 : std::unique_lock<std::mutex> lock(ProcessLock_);
1520 19 : HcclUs curTime = TIME_NOW();
1521 19 : std::map<HeartBeatStatus, std::queue<HeartBeatFrame>> keyEvents;
1522 22 : while (errStatusQueue_.size() > 0) {
1523 3 : auto& tmp = errStatusQueue_.front();
1524 3 : if (IsKeyEvent(tmp, curTime, group)) { // 非关键事件不处理
1525 2 : keyEvents[tmp.status].push(tmp);
1526 : }
1527 3 : errStatusQueue_.pop();
1528 : }
1529 38 : return PrintEvents(keyEvents);
1530 19 : }
1531 :
1532 4 : void Heartbeat::ProcessExceptionEvent()
1533 : {
1534 5 : while (errRankQueue_.size() > 0) {
1535 1 : UIDType cur = errRankQueue_.front();
1536 1 : rankId2StatusMap_[cur].needBroadcast = false;
1537 4989 : OpInfoTagQueueFrame opInfoTagQueueFrame;
1538 2 : for (auto iterRem = rankId2SocketMap_.begin(); iterRem != rankId2SocketMap_.end(); iterRem++) {
1539 1 : UIDType rem = iterRem->first;
1540 1 : if (rem != rankId2StatusMap_[cur].informer
1541 1 : && rankId2StatusMap_[rem].status == HeartBeatStatus::HEARTBEAT_OK) {
1542 1 : if (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) {
1543 1 : (void)SendFrame(rem, cur, rankId2StatusMap_[cur].informer, rankId2StatusMap_[cur].status);
1544 : } else {
1545 0 : (void)SendFrameWithOpCheck(
1546 0 : rem, cur, rankId2StatusMap_[cur].informer, rankId2StatusMap_[cur].status, opInfoTagQueueFrame);
1547 : }
1548 : }
1549 : }
1550 1 : errRankQueue_.pop();
1551 : }
1552 4 : }
1553 :
1554 3 : void Heartbeat::CreateHBLinksAsync()
1555 : {
1556 3 : std::unique_lock<std::mutex> infoLock(hbLinkConnInfoMtx_);
1557 3 : if (hbLinkConnInfo_.empty()) {
1558 3 : return;
1559 : }
1560 0 : linkThreadRunning_ = true;
1561 0 : std::queue<std::tuple<std::string, UIDType, ConnInfo>> connInfoQueue;
1562 0 : for (auto& pair : hbLinkConnInfo_) {
1563 0 : const std::string& groupName = pair.first;
1564 0 : auto& groupConnInfoQueue = pair.second;
1565 0 : while (!groupConnInfoQueue.empty()) {
1566 0 : connInfoQueue.push(
1567 0 : std::make_tuple(groupName, groupConnInfoQueue.front().first, groupConnInfoQueue.front().second));
1568 0 : groupConnInfoQueue.pop();
1569 : }
1570 : }
1571 0 : infoLock.unlock();
1572 0 : while (!connInfoQueue.empty()) {
1573 0 : const std::string groupName = std::get<0>(connInfoQueue.front());
1574 0 : const UIDType& remUid = std::get<1>(connInfoQueue.front());
1575 0 : ConnInfo& connInfo = std::get<2>(connInfoQueue.front());
1576 0 : auto it = linkThreadMap_.find(remUid);
1577 0 : if (it != linkThreadMap_.end() && it->second->joinable()) {
1578 0 : it->second->join();
1579 0 : HCCL_INFO(
1580 : "[CreateHBLinksAsync] Heartbeat link thread has been joined. Group[%s], remote uid[%s].",
1581 : groupName.c_str(), FormatUId(remUid).c_str());
1582 : }
1583 0 : linkThreadMap_[remUid].reset(new (std::nothrow) std::thread(
1584 0 : &Heartbeat::CreateLinkWithRemote, std::ref(*this), groupName, remUid, connInfo));
1585 0 : if (linkThreadMap_[remUid] == nullptr) {
1586 0 : HCCL_RUN_WARNING(
1587 : "Group[%s] establish rank[%s] to rank[%s] heartbeat connection failed. Reason: "
1588 : "create thread failed.",
1589 : groupName.c_str(), FormatUId(uid_).c_str(), FormatUId(remUid).c_str());
1590 : }
1591 0 : connInfoQueue.pop();
1592 0 : }
1593 0 : return;
1594 3 : }
1595 :
1596 7 : void Heartbeat::HeartbeatStatusMonitor()
1597 : {
1598 : // 给当前线程添加名字
1599 7 : SetThreadName("Hccl_HeartBeat");
1600 :
1601 7 : u32 count = 0;
1602 7 : if (deviceLogicId_ != static_cast<u32>(HOST_DEVICE_ID)) {
1603 3 : hrtSetDevice(deviceLogicId_);
1604 : }
1605 7 : uint64_t cnt = 0;
1606 : HcclResult ret;
1607 7 : auto counterStat = CounterStat();
1608 7 : InitStuckDetection(counterStat);
1609 16 : while (startSendRecvTask_) {
1610 9 : CheckSnapshotStatus();
1611 9 : if (isPaused_) {
1612 0 : std::this_thread::sleep_for(std::chrono::milliseconds(BROADCAST_INTERVAL));
1613 0 : continue;
1614 : }
1615 9 : CreateHBLinksAsync();
1616 9 : ProcessLock_.lock();
1617 9 : count++;
1618 9 : if (count >= HEARTBEAT_COUNT) {
1619 0 : count = 0;
1620 0 : OpInfoTagQueueFrame opInfoTagQueueFrame;
1621 0 : GetSendOpInfoList(opInfoTagQueueFrame);
1622 0 : for (auto iter = rankId2SocketMap_.begin(); iter != rankId2SocketMap_.end(); iter++) {
1623 0 : UIDType rem = iter->first;
1624 0 : HCCL_DEBUG(
1625 : "rank[%s] Try to Send HeartBeat to rank[%s]", FormatUId(uid_).c_str(), FormatUId(rem).c_str());
1626 0 : rankId2SocketMap_[rem].lostNum++;
1627 0 : HeartBeatStatus status = HeartBeatStatus::HEARTBEAT_OK;
1628 0 : if (counterStat.issueCnt != 0) {
1629 0 : status = HeartBeatStatus::HEARTBEAT_STUCK;
1630 : }
1631 0 : if (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) {
1632 0 : ret = SendFrame(rem, uid_, uid_, status);
1633 : } else {
1634 0 : ret = SendFrameWithOpCheck(rem, uid_, uid_, status, opInfoTagQueueFrame);
1635 : }
1636 0 : if (ret == HCCL_E_INTERNAL) {
1637 0 : errorSocket_.push_back(rem);
1638 : }
1639 : }
1640 0 : DelErrorSocket();
1641 0 : ProcessCqeErrInfo();
1642 0 : if (counterStat.issueCnt != 0) {
1643 0 : SetStatus(uid_, uid_, HeartBeatStatus::HEARTBEAT_STUCK);
1644 : }
1645 : }
1646 :
1647 21 : for (auto iter = rankId2SocketMap_.begin(); iter != rankId2SocketMap_.end(); iter++) {
1648 12 : UIDType rem = iter->first;
1649 12 : HCCL_DEBUG("rank[%s] Try to Recv from rank[%s]", FormatUId(uid_).c_str(), FormatUId(rem).c_str());
1650 12 : ret = (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) ? RecvFrame(rem) :
1651 0 : RecvFrameWithOpCheck(rem);
1652 12 : if (ret == HCCL_E_INTERNAL) {
1653 0 : errorSocket_.push_back(rem);
1654 0 : continue;
1655 : }
1656 12 : uint32_t threshold = lostThreshold_ << rankId2SocketMap_[rem].lostReportCnt; // LOST帧发送周期放长
1657 12 : if (rankId2SocketMap_[rem].lostNum >= threshold) {
1658 1 : SetStatus(rem, uid_, HeartBeatStatus::HEARTBEAT_LOST);
1659 1 : rankId2SocketMap_[rem].lostReportCnt++;
1660 : }
1661 : }
1662 9 : CheckRecvOpInfoList();
1663 9 : DelErrorSocket();
1664 9 : StuckDetection(cnt, counterStat);
1665 9 : ProcessExceptionEvent();
1666 9 : ProcessLock_.unlock();
1667 :
1668 9 : auto sleeptime = (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) ?
1669 : BROADCAST_INTERVAL :
1670 9 : BROADCAST_INTERVAL_WITH_CHECK;
1671 9 : std::this_thread::sleep_for(std::chrono::milliseconds(sleeptime));
1672 : }
1673 7 : linkThreadRunning_ = false;
1674 : // 在心跳进程结束之前join所有的建链线程
1675 7 : for (auto& pair : linkThreadMap_) {
1676 0 : if (pair.second != nullptr && pair.second->joinable()) {
1677 0 : pair.second->join();
1678 0 : HCCL_INFO("[HeartbeatStatusMonitor] thread has joined. Remote uid is [%s]", FormatUId(pair.first).c_str());
1679 : }
1680 : }
1681 :
1682 7 : if (deviceLogicId_ != static_cast<u32>(HOST_DEVICE_ID)) {
1683 3 : hrtResetDevice(deviceLogicId_);
1684 : }
1685 7 : }
1686 :
1687 7 : void Heartbeat::InitStuckDetection(CounterStat& counterStat)
1688 : {
1689 7 : counterStat.isNeedDetect = (GetExternalInputStuckDetect() == true) ? true : false;
1690 7 : counterStat.couterPrintInter = stuckDetectTime_ * THROUND_MILS / BROADCAST_INTERVAL;
1691 7 : }
1692 :
1693 5 : void Heartbeat::StuckDetection(uint64_t& cnt, CounterStat& counterStat)
1694 : {
1695 5 : HCCL_DEBUG(
1696 : "cnt: %d, isNeedDetect: %d, issueCnt:%llu, interTimes:%d", cnt, counterStat.isNeedDetect, counterStat.issueCnt,
1697 : counterStat.couterPrintInter);
1698 5 : cnt++;
1699 5 : HcclResult ret = HCCL_SUCCESS;
1700 5 : if (counterStat.isNeedDetect && cnt % counterStat.couterPrintInter == 0) {
1701 2 : if (counterStat.isFirst) {
1702 1 : OpExeCounter::GetInstance(deviceLogicId_).GetCounter(counterStat.oldCounter);
1703 1 : counterStat.isFirst = false;
1704 : } else {
1705 1 : ret = OpExeCounter::GetInstance(deviceLogicId_).GetCounter(counterStat.newCounter);
1706 1 : if (ret == HCCL_SUCCESS && counterStat.newCounter.first == counterStat.oldCounter.first
1707 1 : && counterStat.newCounter.first == counterStat.oldCounter.second
1708 1 : && counterStat.newCounter.first == counterStat.newCounter.second) {
1709 1 : HCCL_RUN_INFO(
1710 : "[HCCL_TRACE]rank:%s, count of currently executed operators:%d", FormatUId(uid_).c_str(),
1711 : counterStat.newCounter.first);
1712 1 : counterStat.couterPrintInter *= (BASE_NUMBER << counterStat.issueCnt); // 检测卡住后，把检测周期放长
1713 1 : counterStat.issueCnt++;
1714 : } else {
1715 : // 检测不卡之后，检测间隔恢复到默认间隔
1716 0 : counterStat.couterPrintInter = stuckDetectTime_ * THROUND_MILS / BROADCAST_INTERVAL;
1717 0 : counterStat.issueCnt = 0;
1718 : }
1719 1 : counterStat.oldCounter = counterStat.newCounter; // 更新旧的计数器
1720 : }
1721 : }
1722 5 : }
1723 :
1724 1 : void Heartbeat::PrintAndBroadCastErrorCqe(const ErrCqeInfo& info)
1725 : {
1726 : time_t tmpt;
1727 : struct tm* now;
1728 1 : if (info.cqeInfo.status == 0) {
1729 0 : return;
1730 : }
1731 :
1732 1 : SetStatus(uid_, uid_, HeartBeatStatus::HEARTBEAT_CQE_ERR);
1733 1 : tmpt = static_cast<time_t>(info.cqeInfo.time.tv_sec);
1734 1 : now = localtime(&tmpt);
1735 :
1736 : char errorLinkLogBuffer[LOG_TMPBUF_SIZE];
1737 3 : s32 stringRet = snprintf_s(
1738 : errorLinkLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1739 : "localInfo{server[%s],deviceId[%d],deviceIp[%s]}, remoteIP{server[%s],deviceId[%d],deviceIp[%s]}",
1740 1 : info.linkInfo.localServerId.c_str(), info.linkInfo.localDevicePhyId, nicIp_.GetReadableAddress(),
1741 1 : info.linkInfo.remoteServerId.c_str(), info.linkInfo.remoteDevicePhyId,
1742 : info.cqeInfo.remoteIp.GetReadableAddress());
1743 1 : CHK_PRT_CONT(stringRet == -1, HCCL_ERROR("[Create][DestLink]Transport init error! Failed to build log info"));
1744 :
1745 1 : if (now == nullptr) {
1746 0 : HCCL_ERROR(
1747 : "[%s][%s][%s]localtime fail, cqe error status[%u], %s", LOG_KEYWORDS_TASK_EXEC.c_str(),
1748 : LOG_KEYWORDS_HEARTBEAT_EVETN.c_str(), LOG_KEYWORDS_CQE_ERROR.c_str(), info.cqeInfo.status,
1749 : errorLinkLogBuffer);
1750 : } else {
1751 1 : HCCL_ERROR(
1752 : "[%s][%s][%s]cqe error status[%u], time:[%04u-%02d-%02d %02d:%0d:%02d.%06u], %s",
1753 : LOG_KEYWORDS_TASK_EXEC.c_str(), LOG_KEYWORDS_HEARTBEAT_EVETN.c_str(), LOG_KEYWORDS_CQE_ERROR.c_str(),
1754 : info.cqeInfo.status, now->tm_year + TIME_FROM_1900, now->tm_mon + 1, now->tm_mday, now->tm_hour,
1755 : now->tm_min, now->tm_sec, static_cast<u32>(info.cqeInfo.time.tv_usec), errorLinkLogBuffer);
1756 : }
1757 :
1758 1 : std::unique_lock<std::mutex> lock(remoteIpMutex_);
1759 1 : auto search = remoteIpMap.find(info.linkInfo.identifier);
1760 1 : if (search != remoteIpMap.end()) {
1761 0 : remoteIpMap[info.linkInfo.identifier].insert(info);
1762 : } else {
1763 1 : std::set<ErrCqeInfo> remoteInfoSet;
1764 1 : remoteInfoSet.insert(info);
1765 1 : remoteIpMap.insert(std::pair<std::string, std::set<ErrCqeInfo>>(info.linkInfo.identifier, remoteInfoSet));
1766 1 : }
1767 1 : }
1768 :
1769 6 : void Heartbeat::SaveQpnForOpRetry(const ErrCqeInfo& info)
1770 : {
1771 6 : if (info.cqeInfo.status == 0) {
1772 2 : return;
1773 : }
1774 :
1775 4 : HCCL_RUN_INFO(
1776 : "[Heartbeat][SaveQpnForOpRetry]receive a cqe error [%u][%u], dstrank[%u] identifier[%s]", info.cqeInfo.status,
1777 : info.qpn, info.linkInfo.remoteRank, info.linkInfo.identifier.c_str());
1778 4 : auto identiSearch = rankMapForRetryAgent.find(info.linkInfo.identifier);
1779 4 : if (identiSearch != rankMapForRetryAgent.end()) {
1780 3 : auto rankSearch = identiSearch->second.find(info.linkInfo.remoteRank);
1781 3 : if (rankSearch != identiSearch->second.end()) {
1782 2 : (*rankSearch).second.insert(info);
1783 : } else {
1784 4 : identiSearch->second.insert({info.linkInfo.remoteRank, {info}});
1785 : }
1786 : } else {
1787 1 : std::map<u32, std::set<ErrCqeInfo>> rankExtendMap;
1788 2 : rankExtendMap[info.linkInfo.remoteRank] = {info};
1789 1 : rankMapForRetryAgent.insert(std::make_pair(info.linkInfo.identifier, rankExtendMap));
1790 1 : }
1791 2 : }
1792 :
1793 0 : void Heartbeat::OpRetryCQEHandle(const HcclNetDevCtx netDevCtx)
1794 : {
1795 0 : u32 cqeNum = RETRY_CQE_ARRAY_SIZE;
1796 : do {
1797 0 : cqeNum = RETRY_CQE_ARRAY_SIZE;
1798 :
1799 0 : std::vector<ErrCqeInfo> infos;
1800 0 : HcclResult ret = HcclCommunicator::GetTransportCqeErrors(netDevCtx, infos, cqeNum);
1801 0 : if (ret != HCCL_SUCCESS || cqeNum == 0) {
1802 0 : return;
1803 : }
1804 0 : for (auto& info : infos) {
1805 0 : if (GetRetryEnable(info)
1806 0 : && CommConfiger::GetInstance().GetCommConfigInterSuperPodRetryEnable(info.linkInfo.identifier)) {
1807 0 : SaveQpnForOpRetry(info);
1808 : } else {
1809 0 : PrintAndBroadCastErrorCqe(info);
1810 : }
1811 : }
1812 0 : } while (cqeNum == RETRY_CQE_ARRAY_SIZE);
1813 : }
1814 :
1815 0 : bool Heartbeat::GetRetryEnable(const ErrCqeInfo& info)
1816 : {
1817 0 : std::lock_guard<std::mutex> retryEnablelock(retryEnableMutex_);
1818 0 : auto search = retryEnableTable_.find(info.linkInfo.identifier);
1819 0 : if (search != retryEnableTable_.end()) {
1820 0 : return search->second;
1821 : }
1822 0 : return false;
1823 0 : }
1824 806 : HcclResult Heartbeat::ClearRetryEnableMapItem(const std::string& identifier)
1825 : {
1826 806 : CHK_PRT_RET(initialized_ == false, HCCL_WARNING("Heartbeat has been destroyed"), HCCL_SUCCESS);
1827 0 : u32 delRes = 0;
1828 : {
1829 0 : std::lock_guard<std::mutex> retryEnablelock(retryEnableMutex_);
1830 0 : delRes = retryEnableTable_.erase(identifier);
1831 0 : if (delRes != 0) {
1832 0 : HCCL_INFO("[Heartbeat][ClearRetryEnableMapItem] del identifier[%s] succ", identifier.c_str());
1833 : } else {
1834 0 : HCCL_DEBUG("[Heartbeat][ClearRetryEnableMapItem] identifier[%s] is not found.", identifier.c_str());
1835 : }
1836 0 : }
1837 0 : std::lock_guard<std::mutex> bakcupEnablelock(backupEnableMutex_);
1838 0 : delRes = backupEnableTable_.erase(identifier);
1839 0 : if (delRes != 0) {
1840 0 : HCCL_INFO("[Heartbeat][ClearRetryEnableMapItem] del backup identifier[%s] succ", identifier.c_str());
1841 : } else {
1842 0 : HCCL_DEBUG("[Heartbeat][ClearRetryEnableMapItem] identifier[%s] is not found.", identifier.c_str());
1843 : }
1844 0 : return HCCL_SUCCESS;
1845 0 : }
1846 21 : void Heartbeat::ProcessCqeErrInfoByNetDevCtx(const HcclIpAddress& nicIp)
1847 : {
1848 21 : std::unique_lock<std::mutex> mapLock(ctxMapMutex_);
1849 21 : auto iter = netDevCtxMap_.find(nicIp);
1850 21 : if (iter == netDevCtxMap_.end() || netDevCtxMap_[nicIp] == nullptr) {
1851 7 : return;
1852 : }
1853 14 : mapLock.unlock();
1854 14 : const HcclNetDevCtx netDevCtx = iter->second;
1855 14 : std::vector<ErrCqeInfo> infos;
1856 14 : u32 cqeNum = 1;
1857 14 : HcclResult ret = HcclCommunicator::GetTransportCqeErrors(netDevCtx, infos, cqeNum);
1858 14 : if (ret != HCCL_SUCCESS || infos.size() == 0) {
1859 14 : return;
1860 : }
1861 0 : if (GetRetryEnable(infos[0])
1862 0 : && CommConfiger::GetInstance().GetCommConfigInterSuperPodRetryEnable(infos[0].linkInfo.identifier)) {
1863 0 : SaveQpnForOpRetry(infos[0]);
1864 : } else {
1865 0 : PrintAndBroadCastErrorCqe(infos[0]);
1866 : }
1867 : // infoList 处理
1868 0 : OpRetryCQEHandle(netDevCtx);
1869 35 : }
1870 :
1871 20 : void Heartbeat::ProcessCqeErrInfo()
1872 : {
1873 20 : ProcessCqeErrInfoByNetDevCtx(nicIp_);
1874 20 : if (IsEnableBackupLink()) {
1875 1 : ProcessCqeErrInfoByNetDevCtx(backupNicIp_);
1876 : }
1877 20 : }
1878 :
1879 3 : void Heartbeat::DelErrorSocket()
1880 : {
1881 3 : for (auto rem : errorSocket_) {
1882 0 : HCCL_RUN_INFO(
1883 : "rank[%s] Try to Send/recv HeartBeat to rank[%s]", FormatUId(uid_).c_str(), FormatUId(rem).c_str());
1884 0 : rankId2StatusMap_.erase(rem);
1885 0 : if (rankId2SocketMap_.has(rem)) {
1886 0 : if (rankId2SocketMap_[rem].socket->GetLocalRole() == HcclSocketRole::SOCKET_ROLE_SERVER
1887 0 : && listenSocketMap_.find(rankId2SocketMap_[rem].socket->GetLocalIp()) != listenSocketMap_.end()) {
1888 0 : listenSocketMap_[rankId2SocketMap_[rem].socket->GetLocalIp()]->DelWhiteList(
1889 0 : rankId2SocketMap_[rem].wlistInfosVec);
1890 : }
1891 0 : rankId2SocketMap_[rem].socket->Close();
1892 0 : while (rankId2SocketMap_.erase(rem)) {
1893 : };
1894 : }
1895 : }
1896 3 : errorSocket_.clear();
1897 3 : }
1898 :
1899 2 : HcclResult Heartbeat::GetQpnErr(const std::string& identifier, std::set<std::tuple<u32, u32, u32>>& qpErrSet)
1900 : {
1901 2 : std::unique_lock<std::mutex> lock(qpnMapMutexForRetry_);
1902 2 : auto search = rankMapForRetryAgent.find(identifier);
1903 2 : if (search == rankMapForRetryAgent.end()) {
1904 1 : HCCL_INFO("[GetQpnErr]identifier[%s] is not found", identifier.c_str());
1905 1 : return HCCL_SUCCESS;
1906 : }
1907 1 : if (search->second.size() > 0) {
1908 2 : for (auto iter : search->second) {
1909 1 : u32 dstRank = iter.first;
1910 2 : for (auto qpnInfo : iter.second) {
1911 1 : u32 status = qpnInfo.cqeInfo.status;
1912 1 : qpErrSet.insert(std::make_tuple(dstRank, status, qpnInfo.qpn));
1913 1 : }
1914 1 : }
1915 : }
1916 1 : HCCL_INFO("[GetQpnErr]identifier[%s] is found, qpErrSet size is %u", identifier.c_str(), qpErrSet.size());
1917 1 : return HCCL_SUCCESS;
1918 2 : }
1919 : // OpRetry 失败后，将进行广播操作
1920 1 : HcclResult Heartbeat::BroadcastCqeErr(const std::string& identifier)
1921 : {
1922 1 : u32 cqeSize = 0;
1923 1 : std::unique_lock<std::mutex> qpnMaplock(qpnMapMutexForRetry_);
1924 1 : auto search = rankMapForRetryAgent.find(identifier);
1925 1 : if (search != rankMapForRetryAgent.end()) {
1926 1 : if (search->second.size() > 0) {
1927 1 : cqeSize = search->second.size();
1928 2 : for (auto& qpInfo : search->second) {
1929 2 : for (auto qpnset : qpInfo.second) {
1930 1 : PrintAndBroadCastErrorCqe(qpnset);
1931 1 : HCCL_RUN_INFO(
1932 : "[BroadcastCqeErr][item]remoteIp[%s] remoteRank[%u] status[%u] qpn[%u]",
1933 : qpnset.cqeInfo.remoteIp.GetReadableAddress(), qpInfo.first, qpnset.cqeInfo.status, qpnset.qpn);
1934 1 : }
1935 : }
1936 1 : search->second.clear();
1937 : }
1938 : }
1939 : // 查询剩余量，一般为0
1940 1 : HCCL_RUN_INFO(
1941 : "[Heartbeat][BroadcastCqeErr]clear qpn err size from [%u] to [%u], identifier[%s] ", cqeSize,
1942 : search->second.size(), identifier.c_str());
1943 1 : return HCCL_SUCCESS;
1944 1 : }
1945 :
1946 : /* 非点对点通信 重执行成功后进行调用 */
1947 1 : HcclResult Heartbeat::ClearAllCqeErr(const std::string& identifier)
1948 : {
1949 1 : std::unique_lock<std::mutex> qpnMaplock(qpnMapMutexForRetry_);
1950 1 : u32 cqeSize = 0;
1951 1 : auto search = rankMapForRetryAgent.find(identifier);
1952 1 : if (search != rankMapForRetryAgent.end()) {
1953 1 : if (search->second.size() > 0) {
1954 0 : cqeSize = search->second.size();
1955 0 : search->second.clear();
1956 : }
1957 : }
1958 : // 查询剩余量，一般为0
1959 1 : HCCL_RUN_INFO(
1960 : "[Heartbeat][ClearAllCqeErr]clear qpn err size from [%u] to [%u], identifier[%s]", cqeSize,
1961 : search->second.size(), identifier.c_str());
1962 1 : return HCCL_SUCCESS;
1963 1 : }
1964 : /* 点对点通信 重执行成功后进行调用
1965 : */
1966 5 : HcclResult Heartbeat::ClearCqeErr(const std::string& identifier, u32 remoteRank, u32 qpn)
1967 : {
1968 5 : HCCL_RUN_INFO(
1969 : "[Heartbeat][ClearCqeErr] identifier[%s] remoteRank[%u] qpn[%u].", identifier.c_str(), remoteRank, qpn);
1970 5 : std::unique_lock<std::mutex> qpnMaplock(qpnMapMutexForRetry_);
1971 5 : const auto& search = rankMapForRetryAgent.find(identifier);
1972 5 : if (search == rankMapForRetryAgent.end()) {
1973 0 : return HCCL_SUCCESS;
1974 : }
1975 :
1976 5 : auto& ranksearch = rankMapForRetryAgent[identifier];
1977 : // 删除指定通信域内的固定 remoteRank 固定qpn的cqe err
1978 5 : if (ranksearch.find(remoteRank) != ranksearch.end()) {
1979 2 : if (ranksearch[remoteRank].size() == 1) {
1980 : // remotrank只有一个qpn err，直接删除map
1981 1 : ranksearch.erase(remoteRank);
1982 1 : HCCL_RUN_INFO("[ClearCqeErr][qpnClear] clear dstRank[%u] qpn[%u] now", remoteRank, qpn);
1983 1 : } else if (ranksearch[remoteRank].size() > 1) {
1984 3 : for (auto iter = ranksearch[remoteRank].begin(); iter != ranksearch[remoteRank].end();) {
1985 2 : if (iter->qpn == qpn) {
1986 1 : iter = ranksearch[remoteRank].erase(iter);
1987 1 : HCCL_RUN_INFO("[ClearCqeErr][qpnClear] clear dstRank[%u] qpn[%u] now", remoteRank, qpn);
1988 : } else {
1989 1 : ++iter;
1990 : }
1991 : }
1992 : }
1993 : }
1994 :
1995 : // 查询指定通信域剩余的 QP ERROR 数量
1996 5 : HCCL_RUN_INFO("[ClearCqeErr][qpnClear]identifier qpn err left [%u] now.", search->second.size());
1997 5 : return HCCL_SUCCESS;
1998 5 : }
1999 :
2000 3 : HcclResult Heartbeat::CheckErrorCqe(const std::string& identifier, HcclResult& result)
2001 : {
2002 3 : HcclIpAddress ip;
2003 3 : result = HCCL_SUCCESS;
2004 :
2005 3 : std::unique_lock<std::mutex> lock(remoteIpMutex_);
2006 3 : auto search = remoteIpMap.find(identifier);
2007 3 : if (search == remoteIpMap.end()) {
2008 2 : if (qpnDissociativeSet.size() != 0) { // 如果没有发生error cqe异常的通信域，则确认是否存在游离(Destroy)qpn
2009 0 : HCCL_ERROR(
2010 : "[Heartbeat]find cqe error [%d] num[%llu] dissociative. maybe its qp has already been destroyed",
2011 : result, qpnDissociativeSet.size());
2012 0 : qpnDissociativeSet.clear();
2013 0 : return HCCL_E_REMOTE;
2014 : }
2015 2 : return HCCL_SUCCESS;
2016 : }
2017 1 : if (search->second.size() > 0) {
2018 1 : result = HCCL_E_REMOTE;
2019 1 : HCCL_ERROR("[Heartbeat]find cqe error [%d], in comm [%s]", result, identifier.c_str());
2020 2 : for (auto& it : search->second) {
2021 1 : HCCL_ERROR(
2022 : "[Heartbeat]find cqe error, localIP[%s], remoteIP[%s]", nicIp_.GetReadableAddress(),
2023 : it.cqeInfo.remoteIp.GetReadableAddress());
2024 23 : RPT_INPUT_ERR(
2025 : true, "EI0013",
2026 : std::vector<std::string>(
2027 : {"localServerId", "localDeviceId", "localDeviceIp", "remoteServerId", "remoteDeviceId",
2028 : "remoteDeviceIp"}),
2029 : std::vector<std::string>(
2030 : {it.linkInfo.localServerId, std::to_string(it.linkInfo.localDevicePhyId),
2031 : std::string(nicIp_.GetReadableAddress()), it.linkInfo.remoteServerId,
2032 : std::to_string(it.linkInfo.remoteDevicePhyId),
2033 : std::string(it.cqeInfo.remoteIp.GetReadableAddress())}));
2034 : }
2035 : }
2036 1 : lock.unlock();
2037 :
2038 1 : return HCCL_SUCCESS;
2039 5 : }
2040 :
2041 0 : void Heartbeat::RegisterSROpIdentifier(const std::string& identifier, const std::string& paramTag)
2042 : {
2043 : // SR算子通信域映射关系注册
2044 0 : std::lock_guard<std::mutex> lock(srTagMutex_);
2045 0 : if (srTagMap_.size() > SR_TAG_MAP_MAX_NUM) {
2046 0 : srTagMap_.erase(srTagMap_.begin());
2047 : }
2048 :
2049 0 : auto iter = srTagMap_.find(paramTag);
2050 0 : if (iter == srTagMap_.end()) {
2051 0 : srTagMap_.insert(std::make_pair(paramTag, identifier));
2052 : }
2053 0 : }
2054 :
2055 0 : void Heartbeat::AddInconsistentOpRecord(
2056 : const std::string& identifier, const OpInfoDesc& localOpInfo, InconsistentType status, const std::string& localInfo,
2057 : const std::string& remoteInfo)
2058 : {
2059 0 : std::lock_guard<std::mutex> lock(inconsistentOpMutex_);
2060 0 : if (localOpInfo.opType == HcclCMDType::HCCL_CMD_SEND || localOpInfo.opType == HcclCMDType::HCCL_CMD_RECEIVE) {
2061 0 : auto iter = srTagMap_.find(identifier);
2062 0 : if (iter == srTagMap_.end()) {
2063 0 : HCCL_ERROR(
2064 : "[%s] SR tag[%s] may have already been deleted due to prolonged storage time", __func__,
2065 : identifier.c_str());
2066 0 : return;
2067 : }
2068 :
2069 0 : auto search = inconsistentOpMap_.find(iter->second); // SR tag
2070 0 : if (search == inconsistentOpMap_.end()) {
2071 0 : inconsistentOpMap_.insert(
2072 0 : std::make_pair(iter->second, OpInconsistentInfo(status, localInfo, remoteInfo, localOpInfo)));
2073 0 : HCCL_INFO(
2074 : "[%s] save record SR[%s] identifier[%s] index[%d]", __func__, identifier.c_str(), iter->second.c_str(),
2075 : localOpInfo.index);
2076 : }
2077 0 : } else {
2078 0 : auto search = inconsistentOpMap_.find(identifier); // AR identifier
2079 0 : if (search == inconsistentOpMap_.end()) {
2080 0 : inconsistentOpMap_.insert(
2081 0 : std::make_pair(identifier, OpInconsistentInfo(status, localInfo, remoteInfo, localOpInfo)));
2082 0 : HCCL_INFO("[%s] save record identifier[%s] index[%d]", __func__, identifier.c_str(), localOpInfo.index);
2083 : }
2084 : }
2085 0 : }
2086 :
2087 0 : HcclResult Heartbeat::CheckOpInconsistentError(const std::string& identifier, HcclResult& result)
2088 : {
2089 0 : if (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) {
2090 0 : return HCCL_SUCCESS;
2091 : }
2092 0 : std::lock_guard<std::mutex> lock(inconsistentOpMutex_);
2093 0 : auto search = inconsistentOpMap_.find(identifier);
2094 0 : if (search != inconsistentOpMap_.end()) {
2095 0 : result = HCCL_E_PARA;
2096 0 : const OpInconsistentInfo& inconsistentInfo = search->second;
2097 0 : std::string opInfo = "Unknown";
2098 0 : for (const auto& pair : HCCL_OPTYPE_NAME_MAP) {
2099 0 : if (pair.second == inconsistentInfo.opInfoDesc.opType) {
2100 0 : opInfo = std::string(pair.first);
2101 0 : break;
2102 : }
2103 : }
2104 0 : HCCL_ERROR(
2105 : "[%s]find inconsistent op [%s] error [%d], in comm [%s]", __func__, opInfo, result, identifier.c_str());
2106 0 : RPT_INPUT_ERR(
2107 : true, "EI0005", std::vector<std::string>({"ccl_op", "group", "para_name", "local_para", "remote_para"}),
2108 : std::vector<std::string>(
2109 : {opInfo, identifier, GetInconsistentTypeStr(search->second.inconsistentType), search->second.localInfo,
2110 : search->second.remoteInfo}));
2111 0 : }
2112 0 : return HCCL_SUCCESS;
2113 0 : }
2114 :
2115 24 : HcclResult Heartbeat::SetRankPortInfo(
2116 : bool isUseRankPort, std::vector<u32>& nicRanksPorts, std::vector<u32>& vnicRanksPorts, bool devPortSwitchOn)
2117 : {
2118 24 : isUseRankPort_ = isUseRankPort;
2119 24 : nicRanksPorts_ = nicRanksPorts;
2120 24 : vnicRanksPorts_ = vnicRanksPorts;
2121 24 : devPortSwitchOn_ = devPortSwitchOn;
2122 24 : return HCCL_SUCCESS;
2123 : }
2124 :
2125 0 : void Heartbeat::SetOpretryErr()
2126 : {
2127 : // 重执行约束场景，给errStatusQueue添加重执行失败心跳帧
2128 0 : SetStatus(uid_, uid_, HeartBeatStatus::HEARTBEAT_OPRETRY_NOT_SUPPORT);
2129 0 : }
2130 :
2131 8 : u32 Heartbeat::GetPort(HcclSocketType type, u32 remoteUserRank, [[maybe_unused]] u32 remoteDeviceId)
2132 : {
2133 8 : u32 port = HCCL_INVALID_PORT;
2134 8 : if (isUseRankPort_) {
2135 0 : if (devPortSwitchOn_ && type == HcclSocketType::SOCKET_VNIC && remoteUserRank < vnicRanksPorts_.size()
2136 0 : && vnicRanksPorts_[remoteUserRank] != HCCL_INVALID_PORT) {
2137 0 : port = vnicRanksPorts_[remoteUserRank];
2138 0 : HCCL_INFO("[Heartbeat][GetPort] use vnic ranks port[%u]", port);
2139 0 : } else if (remoteUserRank < nicRanksPorts_.size() && nicRanksPorts_[remoteUserRank] != HCCL_INVALID_PORT) {
2140 0 : port = nicRanksPorts_[remoteUserRank];
2141 0 : HCCL_INFO("[Heartbeat][GetPort] use nic ranks port[%u]", port);
2142 : } else {
2143 0 : port = HETEROG_CCL_PORT;
2144 : }
2145 : } else {
2146 8 : port = HETEROG_CCL_PORT;
2147 : }
2148 8 : return port;
2149 : }
2150 :
2151 2 : u32 Heartbeat::GetHostPort(s32 devicePhyId)
2152 : {
2153 2 : if (GetExternalInputHcclIfBasePort() == HCCL_INVALID_PORT) {
2154 1 : return (devicePhyId + HOST_PARA_BASE_PORT);
2155 : } else {
2156 1 : return (devicePhyId + GetExternalInputHcclIfBasePort() + HCCL_AISERVER_DEVICE_NUM);
2157 : }
2158 : }
2159 :
2160 0 : bool Heartbeat::IsPaused() const { return !startSendRecvTask_ || isPaused_; }
2161 :
2162 0 : bool Heartbeat::IsResumed() const { return !startSendRecvTask_ || !isPaused_; }
2163 :
2164 3 : void Heartbeat::CheckSnapshotStatus()
2165 : {
2166 3 : auto snapshotStatus = SnapshotControl::GetInstance(deviceLogicId_).GetStatus();
2167 3 : if (isPaused_ && snapshotStatus == SnapshotStatus::POST_SNAPSHOT) {
2168 0 : isPaused_ = false;
2169 0 : HCCL_RUN_INFO(
2170 : "[Heartbeat][CheckSnapshotStatus] detect snapshot post-processing, heart is resumed, "
2171 : "deviceLogicId[%u].",
2172 : deviceLogicId_);
2173 3 : } else if (!isPaused_ && snapshotStatus == SnapshotStatus::PRE_SNAPSHOT) {
2174 0 : isPaused_ = true;
2175 0 : HCCL_RUN_INFO(
2176 : "[Heartbeat][CheckSnapshotStatus] detect snapshot pre-processing, heart is paused, "
2177 : "deviceLogicId[%u].",
2178 : deviceLogicId_);
2179 : }
2180 3 : }
2181 :
2182 0 : HcclResult RegisterToHeartBeat(
2183 : s32 deviceLogicID, u32 userRank, DevType devType, std::vector<RankInfo>& rankInfoList, const u32 port,
2184 : const bool isNeedNic, u32 peerRankId, const std::string& commIdentifier, const std::string& tag,
2185 : bool useSuperPodMode, bool isUsedRdmaLevel0)
2186 : {
2187 0 : return peerRankId == INVALID_VALUE_RANKID ? Heartbeat::GetInstance(deviceLogicID)
2188 0 : .RegisterToHeartBeat(
2189 : userRank, devType, rankInfoList, port, isNeedNic,
2190 : commIdentifier, useSuperPodMode, isUsedRdmaLevel0) :
2191 0 : Heartbeat::GetInstance(deviceLogicID)
2192 0 : .RegisterToHeartBeat(
2193 : userRank, devType, rankInfoList, port, isNeedNic, peerRankId,
2194 0 : commIdentifier, tag, useSuperPodMode, isUsedRdmaLevel0);
2195 : }
2196 :
2197 0 : void UnRegisterRanks(s32 deviceLogicID, DevType devType, const std::string& commIdentifier, const std::string& tag)
2198 : {
2199 0 : return tag.empty() ? Heartbeat::GetInstance(deviceLogicID).UnRegisterToHeartBeat(devType, commIdentifier) :
2200 0 : Heartbeat::GetInstance(deviceLogicID).UnRegisterToHeartBeat(devType, commIdentifier, tag);
2201 : }
2202 :
2203 0 : HcclResult SetRankPortInfo(s32 deviceLogicID, bool isUseRankPort, std::vector<u32>& ranksPort)
2204 : {
2205 0 : return Heartbeat::GetInstance(deviceLogicID).SetRankPortInfo(isUseRankPort, ranksPort, ranksPort, false);
2206 : }
2207 :
2208 17 : std::vector<std::string> GetErrStatusVec(s32 deviceLogicID, const std::string& group)
2209 : {
2210 17 : return Heartbeat::GetInstance(deviceLogicID).GetErrStatusVec(group);
2211 : }
2212 :
2213 52 : __attribute__((constructor)) void HeartBeatCallBackInit()
2214 : {
2215 52 : RegisterHeartBeatCallBack(RegisterToHeartBeat, UnRegisterRanks, SetRankPortInfo);
2216 52 : RegisterGetErrStatusVecCallBack(GetErrStatusVec);
2217 52 : }
2218 : } // namespace hccl
|
