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 1645 : Heartbeat& Heartbeat::GetInstance(s32 deviceLogicID)
25 : {
26 2490 : static Heartbeat hb[MAX_MODULE_DEVICE_NUM];
27 1645 : if (static_cast<u32>(deviceLogicID) >= MAX_MODULE_DEVICE_NUM) {
28 527 : HCCL_WARNING("[Heartbeat][%s]deviceLogicID[%d] is invalid", __func__, deviceLogicID);
29 527 : return hb[0];
30 : }
31 1118 : 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 645 : HcclResult Heartbeat::DeleteOpInfoToHeartBeat(const std::string& identifier, const std::string& newTag)
554 : {
555 645 : std::string tag;
556 645 : if (newTag != "") {
557 0 : tag = newTag;
558 : } else {
559 645 : tag = identifier;
560 : }
561 645 : 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 645 : }
567 :
568 216 : HcclResult Heartbeat::UnRegisterRanks(const std::string& group)
569 : {
570 216 : 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 807 : void Heartbeat::UnRegisterToHeartBeat(DevType devType, const std::string& commIdentifier)
635 : {
636 807 : if (Is310PDevice() || devType == DevType::DEV_TYPE_310P3) {
637 4 : return;
638 : }
639 803 : ClearRetryEnableMapItem(commIdentifier);
640 803 : HcclResult ret = UnRegisterRanks(commIdentifier);
641 800 : 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 2 : void Heartbeat::GetOneOpInfo(std::string& tag, OpInfoDesc& opInfo)
964 : {
965 : // 从发送队列中获取一个opInfo发送给对端
966 2 : std::unique_lock<std::mutex> lock(opInfoQueueMutex_);
967 2 : if (opInfoQueue_.empty()) {
968 : static OpInfoDesc defaultOpInfo;
969 1 : opInfo = defaultOpInfo;
970 1 : 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 2 : }
1000 :
1001 0 : void Heartbeat::GetSendOpInfoList(OpInfoTagQueueFrame& opInfoTagQueueFrame)
1002 : {
1003 0 : if (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) {
1004 0 : return;
1005 : }
1006 0 : while (opInfoQueueForSend_.size() < OPINFO_TAG_QUEUE_NUM * OPINFO_SEND_NUM_BY_TAG) {
1007 0 : OpInfoDesc opInfo;
1008 0 : std::string tag;
1009 0 : GetOneOpInfo(tag, opInfo);
1010 0 : if (opInfo.isValid) {
1011 0 : opInfoQueueForSend_.push_back(std::make_pair(tag, opInfo));
1012 : } else {
1013 0 : break;
1014 : }
1015 0 : }
1016 :
1017 0 : HCCL_DEBUG("[%s] opInfoQueueForSend_.size[%d] begin", __func__, opInfoQueueForSend_.size());
1018 0 : auto& opInfoTagQueue = opInfoTagQueueFrame.opInfoTagQueue;
1019 0 : for (auto iter = opInfoQueueForSend_.begin(); iter != opInfoQueueForSend_.end();) {
1020 0 : bool isAdd = false;
1021 0 : for (u32 index = 0; index < OPINFO_TAG_QUEUE_NUM; index++) {
1022 : // 当前 index 对应的 opInfoTagQueue 为未初始化状态
1023 0 : if (strncmp(opInfoTagQueue[index].identifier, "\0", ROOTINFO_INDENTIFIER_MAX_LENGTH) == 0) {
1024 0 : memcpy_s(
1025 0 : opInfoTagQueue[index].identifier, iter->first.size() + 1, iter->first.c_str(),
1026 0 : iter->first.size() + 1);
1027 0 : opInfoTagQueue[index].opInfoList[opInfoTagQueue[index].opInfoNum] = iter->second;
1028 0 : opInfoTagQueue[index].opInfoNum++;
1029 0 : isAdd = true;
1030 0 : HCCL_DEBUG(
1031 : "[%s]opInfoTagQueue[%d] add success identifier[%s] ", __func__, index,
1032 : opInfoTagQueue[index].identifier);
1033 0 : break;
1034 : }
1035 : // 当前 index 对应的 opInfoTagQueue 已经被某个tag 的算子占用
1036 0 : else if (
1037 0 : strncmp(opInfoTagQueue[index].identifier, iter->first.c_str(), ROOTINFO_INDENTIFIER_MAX_LENGTH) == 0) {
1038 0 : if (opInfoTagQueue[index].opInfoNum < OPINFO_SEND_NUM_BY_TAG) {
1039 0 : opInfoTagQueue[index].opInfoList[opInfoTagQueue[index].opInfoNum] = iter->second;
1040 0 : opInfoTagQueue[index].opInfoNum++;
1041 0 : isAdd = true;
1042 0 : HCCL_DEBUG(
1043 : "[%s]opInfoTagQueue[%d] has exists and add success identifier[%s] ", __func__, index,
1044 : opInfoTagQueue[index].identifier);
1045 0 : break;
1046 : }
1047 : }
1048 : }
1049 0 : if (isAdd) {
1050 0 : iter = opInfoQueueForSend_.erase(iter);
1051 : } else {
1052 0 : iter++; // opInfoQueueForSend_ 残留数据会被保存到下一轮 GetSendOpInfoList
1053 : }
1054 : }
1055 0 : return;
1056 : }
1057 :
1058 6 : void Heartbeat::SaveOpInfo(const OpInfoTagQueueFrame& opInfoTagQueueFrame, UIDType& src)
1059 : {
1060 6 : const auto& opInfoTagQueue = opInfoTagQueueFrame.opInfoTagQueue;
1061 66 : for (u32 index = 0; index < OPINFO_TAG_QUEUE_NUM; index++) {
1062 60 : std::string tag = std::string(opInfoTagQueue[index].identifier);
1063 64 : for (u32 num = 0; num < opInfoTagQueue[index].opInfoNum; num++) {
1064 4 : std::unique_lock<std::mutex> lock(opInfoMapMutex_);
1065 : // 保存接收到的opInfo到接收队列中
1066 4 : auto& opInfo = opInfoTagQueue[index].opInfoList[num];
1067 4 : recvOpInfoList_.push_back(std::make_tuple(opInfo, tag, src));
1068 4 : HCCL_DEBUG(
1069 : "[Heartbeat][%s]tag[%s], opType[%d], dataType[%d], reduce[%d], count[%u], root[%d], index[%llu] get "
1070 : "success",
1071 : __func__, tag.c_str(), opInfo.opType, opInfo.dataType, opInfo.reduceOp, opInfo.count, opInfo.root,
1072 : opInfo.index);
1073 4 : }
1074 60 : }
1075 6 : std::unique_lock<std::mutex> lock(opInfoMapMutex_);
1076 6 : while (recvOpInfoList_.size() > OPINFO_QUEUE_MAX_SIZE) { // 可能存在误丢
1077 0 : recvOpInfoList_.pop_front();
1078 : }
1079 :
1080 12 : return;
1081 6 : }
1082 :
1083 : HcclResult
1084 6 : Heartbeat::CheckIsSameOp(const OpInfoDesc& localOpInfo, const OpInfoDesc& remoteOpInfo, InconsistentType& status)
1085 : {
1086 6 : if (localOpInfo.opType == HcclCMDType::HCCL_CMD_SEND) {
1087 2 : if (remoteOpInfo.opType != HcclCMDType::HCCL_CMD_RECEIVE) {
1088 1 : status = InconsistentType::OPTYPE_INCONSISTENT;
1089 1 : return HCCL_SUCCESS;
1090 : }
1091 4 : } else if (localOpInfo.opType == HcclCMDType::HCCL_CMD_RECEIVE) {
1092 1 : if (remoteOpInfo.opType != HcclCMDType::HCCL_CMD_SEND) {
1093 1 : status = InconsistentType::OPTYPE_INCONSISTENT;
1094 1 : return HCCL_SUCCESS;
1095 : }
1096 3 : } else if (localOpInfo.opType != remoteOpInfo.opType) {
1097 1 : status = InconsistentType::OPTYPE_INCONSISTENT;
1098 1 : return HCCL_SUCCESS;
1099 : }
1100 :
1101 3 : if (localOpInfo.dataType != remoteOpInfo.dataType) {
1102 1 : status = InconsistentType::DATATYPE_INCONSISTENT;
1103 1 : return HCCL_SUCCESS;
1104 : }
1105 :
1106 2 : if (localOpInfo.reduceOp != remoteOpInfo.reduceOp) {
1107 0 : status = InconsistentType::REDUCETYPE_INCONSISTENT;
1108 0 : return HCCL_SUCCESS;
1109 : }
1110 :
1111 2 : if (localOpInfo.root != remoteOpInfo.root) {
1112 0 : status = InconsistentType::ROOT_INCONSISTENT;
1113 0 : return HCCL_SUCCESS;
1114 : }
1115 :
1116 2 : if (localOpInfo.opType != HcclCMDType::HCCL_CMD_ALLGATHER_V && localOpInfo.opType != HcclCMDType::HCCL_CMD_ALLTOALLV
1117 2 : && localOpInfo.opType != HcclCMDType::HCCL_CMD_ALLTOALLVC
1118 2 : && localOpInfo.opType != HcclCMDType::HCCL_CMD_REDUCE_SCATTER_V) {
1119 : // 仅对数据量均等的算子进行校验数据量count
1120 2 : if (localOpInfo.count != remoteOpInfo.count) {
1121 1 : status = InconsistentType::COUNT_INCONSISTENT;
1122 1 : return HCCL_SUCCESS;
1123 : }
1124 : }
1125 1 : status = InconsistentType::NO_INCONSISTENT;
1126 1 : return HCCL_SUCCESS;
1127 : }
1128 :
1129 5 : void Heartbeat::CheckRecvOpInfoList()
1130 : {
1131 5 : if (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) {
1132 5 : return;
1133 : }
1134 : // 校验接收队列中接收到的opInfo
1135 0 : std::unique_lock<std::mutex> lock(opInfoMapMutex_);
1136 0 : for (auto it = recvOpInfoList_.begin(); it != recvOpInfoList_.end();) {
1137 0 : const auto& opInfoRecv = std::get<0>(*it);
1138 0 : const auto& identifier = std::get<1>(*it);
1139 0 : const auto& uid = std::get<2>(*it);
1140 0 : auto opInfoIndexMap = opInfoMap_.find(identifier);
1141 0 : if (opInfoIndexMap == opInfoMap_.end()) {
1142 0 : ++it;
1143 0 : HCCL_DEBUG(
1144 : "[Heartbeat]check recv not found. identifier[%s] index[%u]", identifier.c_str(), opInfoRecv.index);
1145 0 : continue;
1146 : }
1147 :
1148 0 : if (opInfoIndexMap->second.find(opInfoRecv.index) != opInfoIndexMap->second.end()) {
1149 0 : const auto& opInfo = opInfoIndexMap->second[opInfoRecv.index];
1150 0 : InconsistentType inconsistent = InconsistentType::NO_INCONSISTENT;
1151 0 : CheckIsSameOp(opInfo, opInfoRecv, inconsistent);
1152 0 : if (inconsistent != InconsistentType::NO_INCONSISTENT) {
1153 : // 当算子不匹配时，记录并打印ERROR日志并广播下发不一致错误给其他节点
1154 : char localInfo[LOG_TMPBUF_SIZE];
1155 0 : s32 ret = snprintf_s(
1156 : localInfo, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1157 0 : "node[%s] optype[%s] dataType[%s] reduceOp[%s] count[%d] root[%d]", FormatUId(uid_).c_str(),
1158 0 : GetCMDTypeEnumStr(opInfo.opType).c_str(), GetDataTypeEnumStr(opInfo.dataType).c_str(),
1159 0 : GetReduceOpEnumStr(opInfo.reduceOp).c_str(), opInfo.count, opInfo.root);
1160 0 : CHK_PRT_CONT(ret == -1, HCCL_ERROR("Failed to build log info"));
1161 : char remoteInfo[LOG_TMPBUF_SIZE];
1162 0 : ret = snprintf_s(
1163 : remoteInfo, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1164 0 : "node[%s] optype[%s] dataType[%s] reduceOp[%s] count[%lu] root[%u]", FormatUId(uid).c_str(),
1165 0 : GetCMDTypeEnumStr(opInfoRecv.opType).c_str(), GetDataTypeEnumStr(opInfoRecv.dataType).c_str(),
1166 0 : GetReduceOpEnumStr(opInfoRecv.reduceOp).c_str(), opInfoRecv.count, opInfoRecv.root);
1167 0 : CHK_PRT_CONT(ret == -1, HCCL_ERROR("Failed to build log info"));
1168 :
1169 0 : AddInconsistentOpRecord(
1170 0 : identifier, opInfo, inconsistent, std::string(localInfo), std::string(remoteInfo));
1171 0 : HCCL_ERROR(
1172 : "[Heartbeat]check opinfo inconsistent. identifier[%s] index[%u], "
1173 : "local(%s); remote(%s)",
1174 : identifier.c_str(), opInfoRecv.index, localInfo, remoteInfo);
1175 0 : SetStatus(uid_, uid_, HeartBeatStatus::HEARTBEAT_INCONSISTENT);
1176 : }
1177 : // 校验完成后删除收到的opInfo
1178 0 : it = recvOpInfoList_.erase(it);
1179 : } else {
1180 : // 若在opInfoIndexMap中没有找到相同index的算子，先跳到该记录，校验下一个收到的算子
1181 0 : ++it;
1182 : }
1183 : }
1184 0 : return;
1185 0 : }
1186 :
1187 4 : HcclResult Heartbeat::SendFrame(UIDType& dst, UIDType& crimer, UIDType& informer, HeartBeatStatus status)
1188 : {
1189 4 : HeartBeatFrame bf(uid_, dst, crimer, informer, status);
1190 4 : if (rankId2SocketMap_[dst].sendBuffer.size() > 0) {
1191 1 : if (status != HeartBeatStatus::HEARTBEAT_OK && rankId2SocketMap_[dst].sendBuffer.size() < MAX_SENDBUFF_SIZE) {
1192 1 : rankId2SocketMap_[dst].sendBuffer.push(bf);
1193 : }
1194 3 : while (rankId2SocketMap_[dst].sendBuffer.size() > 0) {
1195 2 : HeartBeatFrame hbf = rankId2SocketMap_[dst].sendBuffer.front();
1196 2 : u64 sendDis = sizeof(HeartBeatFrame) - rankId2SocketMap_[dst].restSize;
1197 2 : u64 compSize = 0;
1198 2 : HcclResult ret = rankId2SocketMap_[dst].socket->ISend(
1199 2 : reinterpret_cast<void*>(reinterpret_cast<uintptr_t>(&hbf) + sendDis), rankId2SocketMap_[dst].restSize,
1200 : compSize);
1201 2 : if (ret != HCCL_SUCCESS) {
1202 0 : return ret;
1203 : }
1204 2 : if (rankId2SocketMap_[dst].restSize == compSize) {
1205 2 : rankId2SocketMap_[dst].sendBuffer.pop();
1206 2 : rankId2SocketMap_[dst].restSize = sizeof(HeartBeatFrame);
1207 2 : HCCL_DEBUG(
1208 : "[Heartbeat][SendFrame] Send Success, from [%s] to [%s] about [%s] by [%s] status[%d]",
1209 : FormatUId(uid_).c_str(), FormatUId(dst).c_str(), FormatUId(crimer).c_str(),
1210 : FormatUId(informer).c_str(), status);
1211 : } else {
1212 0 : rankId2SocketMap_[dst].restSize = rankId2SocketMap_[dst].restSize - compSize;
1213 0 : break;
1214 : }
1215 : }
1216 : } else {
1217 3 : u64 compSize = 0;
1218 3 : u32 expectSize = sizeof(HeartBeatFrame);
1219 3 : HcclResult ret = rankId2SocketMap_[dst].socket->ISend(&bf, expectSize, compSize);
1220 3 : if (ret != HCCL_SUCCESS) {
1221 0 : return ret;
1222 : }
1223 3 : if (compSize == expectSize) {
1224 2 : HCCL_DEBUG(
1225 : "[Heartbeat][SendFrame] Send Success, from [%s] to [%s] about [%s] by [%s] status[%d]",
1226 : FormatUId(uid_).c_str(), FormatUId(dst).c_str(), FormatUId(crimer).c_str(), FormatUId(informer).c_str(),
1227 : status);
1228 : } else {
1229 1 : HCCL_DEBUG(
1230 : "[Heartbeat][SendFrame] Send Not Complete, from [%s] to [%s] about [%s] by [%s] status[%d], "
1231 : "expectSize[%u], compSize[%u]",
1232 : FormatUId(uid_).c_str(), FormatUId(dst).c_str(), FormatUId(crimer).c_str(), FormatUId(informer).c_str(),
1233 : status, expectSize, compSize);
1234 1 : rankId2SocketMap_[dst].restSize = expectSize - compSize;
1235 1 : rankId2SocketMap_[dst].sendBuffer.push(bf);
1236 : }
1237 : }
1238 4 : return HCCL_SUCCESS;
1239 : }
1240 :
1241 0 : HcclResult Heartbeat::SendFrameWithOpCheck(
1242 : UIDType& dst, UIDType& crimer, UIDType& informer, HeartBeatStatus status,
1243 : const OpInfoTagQueueFrame& opInfoTagQueueFrame)
1244 : {
1245 0 : HeartBeatFrameWithOpCheck bf(uid_, dst, crimer, informer, status);
1246 0 : bf.opInfoTagQueueFrame = opInfoTagQueueFrame;
1247 :
1248 0 : if (rankId2SocketMap_[dst].sendBufferWithOpCheck.size() > 0) {
1249 0 : if (status != HeartBeatStatus::HEARTBEAT_OK
1250 0 : && rankId2SocketMap_[dst].sendBufferWithOpCheck.size() < MAX_SENDBUFF_SIZE) {
1251 0 : rankId2SocketMap_[dst].sendBufferWithOpCheck.push(bf);
1252 : }
1253 : } else {
1254 0 : rankId2SocketMap_[dst].sendBufferWithOpCheck.push(bf);
1255 0 : rankId2SocketMap_[dst].restSize = sizeof(HeartBeatFrameWithOpCheck);
1256 : }
1257 : // 查询到某个Dst的发送缓冲数据量
1258 0 : u32 unCompletedCount = 0; // 已经发送的loop次数
1259 0 : while (rankId2SocketMap_[dst].sendBufferWithOpCheck.size() > 0) {
1260 0 : HeartBeatFrameWithOpCheck hbf = rankId2SocketMap_[dst].sendBufferWithOpCheck.front();
1261 0 : u64 sendDis = sizeof(HeartBeatFrameWithOpCheck) - rankId2SocketMap_[dst].restSize;
1262 0 : u64 compSize = 0;
1263 0 : HcclResult ret = rankId2SocketMap_[dst].socket->ISend(
1264 0 : reinterpret_cast<void*>(reinterpret_cast<uintptr_t>(&hbf) + sendDis), rankId2SocketMap_[dst].restSize,
1265 : compSize);
1266 0 : if (ret != HCCL_SUCCESS) {
1267 0 : return ret;
1268 : }
1269 0 : if (rankId2SocketMap_[dst].restSize == compSize) {
1270 0 : rankId2SocketMap_[dst].sendBufferWithOpCheck.pop();
1271 0 : rankId2SocketMap_[dst].restSize = sizeof(HeartBeatFrameWithOpCheck);
1272 0 : HCCL_DEBUG(
1273 : "[Heartbeat][%s] Send Success, from [%s] to [%s] about [%s] by [%s] status[%d]", __func__,
1274 : FormatUId(uid_).c_str(), FormatUId(dst).c_str(), FormatUId(crimer).c_str(), FormatUId(informer).c_str(),
1275 : status);
1276 : } else {
1277 0 : HCCL_DEBUG(
1278 : "[Heartbeat][%s] Send Not Complete, from [%s] to [%s] about [%s] by [%s] status[%d], expectSize[%u], "
1279 : "compSize[%u]",
1280 : __func__, FormatUId(uid_).c_str(), FormatUId(dst).c_str(), FormatUId(crimer).c_str(),
1281 : FormatUId(informer).c_str(), status, rankId2SocketMap_[dst].restSize, compSize);
1282 0 : rankId2SocketMap_[dst].restSize = rankId2SocketMap_[dst].restSize - compSize;
1283 0 : unCompletedCount++;
1284 0 : SaluSleep(ONE_HUNDRED_MICROSECOND_OF_USLEEP); // 100us
1285 : // 限制发送的循环此时，避免在send流程里死循环
1286 0 : if (unCompletedCount > HBFRAME_SEND_LOOP_MAX_NUM) {
1287 0 : break; // 120个loop约30毫秒
1288 : }
1289 : }
1290 : }
1291 0 : return HCCL_SUCCESS;
1292 : }
1293 :
1294 0 : HcclResult Heartbeat::RecvFrame(UIDType& src)
1295 : {
1296 0 : HeartBeatFrame bf;
1297 0 : u64 compSize = 0;
1298 0 : u64 expectSize = sizeof(HeartBeatFrame);
1299 : while (true) {
1300 0 : compSize = 0;
1301 0 : HcclResult retVal = rankId2SocketMap_[src].socket->IRecv(&bf, expectSize, compSize);
1302 0 : if (retVal == HCCL_SUCCESS && compSize > 0) {
1303 0 : rankId2SocketMap_[src].recvBuffer.PushSeg(reinterpret_cast<u8*>(&bf), compSize);
1304 0 : if (rankId2SocketMap_[src].recvBuffer.Size() >= expectSize) {
1305 0 : rankId2SocketMap_[src].recvBuffer.GetSeg(reinterpret_cast<u8*>(&bf), expectSize);
1306 0 : rankId2SocketMap_[src].recvBuffer.PopSeg(expectSize);
1307 0 : CHK_RET(ParseFrame(bf, src));
1308 : }
1309 0 : } else if (retVal == HCCL_E_INTERNAL) {
1310 0 : return HCCL_E_INTERNAL;
1311 : } else {
1312 0 : break;
1313 : }
1314 0 : }
1315 0 : return HCCL_SUCCESS;
1316 : }
1317 :
1318 0 : HcclResult Heartbeat::RecvFrameWithOpCheck(UIDType& src)
1319 : {
1320 0 : HeartBeatFrameWithOpCheck bf;
1321 0 : u64 compSize = 0;
1322 0 : u64 expectSize = sizeof(HeartBeatFrameWithOpCheck);
1323 : while (true) {
1324 0 : compSize = 0;
1325 0 : HcclResult retVal = rankId2SocketMap_[src].socket->IRecv(&bf, expectSize, compSize);
1326 0 : if (retVal == HCCL_SUCCESS && compSize > 0) {
1327 0 : rankId2SocketMap_[src].recvBuffer.PushSeg(reinterpret_cast<u8*>(&bf), compSize);
1328 : // 标识当前Recvbuf中已经存放了一个完整的帧
1329 0 : if (rankId2SocketMap_[src].recvBuffer.Size() >= expectSize) {
1330 0 : rankId2SocketMap_[src].recvBuffer.GetSeg(reinterpret_cast<u8*>(&bf), expectSize);
1331 0 : rankId2SocketMap_[src].recvBuffer.PopSeg(expectSize);
1332 0 : CHK_RET(ParseFrameWithOpCheck(bf, src));
1333 0 : break;
1334 : }
1335 0 : } else if (retVal == HCCL_E_INTERNAL) {
1336 0 : return HCCL_E_INTERNAL;
1337 : } else {
1338 0 : break;
1339 : }
1340 0 : }
1341 0 : return HCCL_SUCCESS;
1342 : }
1343 :
1344 2 : HcclResult Heartbeat::ParseFrame(HeartBeatFrame& bf, UIDType& src)
1345 : {
1346 2 : if (bf.src != src || bf.dst != uid_) {
1347 0 : HCCL_WARNING("rank[%s] recv wrong frame", FormatUId(uid_).c_str());
1348 0 : return HCCL_E_INTERNAL;
1349 : }
1350 :
1351 2 : HCCL_DEBUG(
1352 : "[Heartbeat][RecvFrame] Recv Success, from [%s] to [%s] about [%s] by [%s] state[%d]",
1353 : FormatUId(bf.src).c_str(), FormatUId(bf.dst).c_str(), FormatUId(bf.crimer).c_str(),
1354 : FormatUId(bf.informer).c_str(), bf.status);
1355 :
1356 : // 能够收到进程卡住表示心跳是正常的
1357 2 : if (bf.status == HeartBeatStatus::HEARTBEAT_OK || bf.status == HeartBeatStatus::HEARTBEAT_STUCK) {
1358 2 : rankId2SocketMap_[src].lostNum = 0;
1359 2 : rankId2SocketMap_[src].lostReportCnt = 0;
1360 : }
1361 :
1362 : // 只有心跳非正常时才需要打印TRACE
1363 2 : if (bf.status != HeartBeatStatus::HEARTBEAT_OK) {
1364 1 : SetStatus(bf.crimer, bf.informer, bf.status);
1365 : }
1366 :
1367 2 : return HCCL_SUCCESS;
1368 : }
1369 :
1370 2 : HcclResult Heartbeat::ParseFrameWithOpCheck(HeartBeatFrameWithOpCheck& bf, UIDType& src)
1371 : {
1372 2 : if (bf.src != src || bf.dst != uid_) {
1373 0 : HCCL_WARNING("rank[%s] recv wrong frame", FormatUId(uid_).c_str());
1374 0 : return HCCL_E_INTERNAL;
1375 : }
1376 :
1377 2 : HCCL_DEBUG(
1378 : "[Heartbeat][RecvFrame] Recv Success, from [%s] to [%s] about [%s] by [%s] state[%d]",
1379 : FormatUId(bf.src).c_str(), FormatUId(bf.dst).c_str(), FormatUId(bf.crimer).c_str(),
1380 : FormatUId(bf.informer).c_str(), bf.status);
1381 :
1382 2 : if (bf.status == HeartBeatStatus::HEARTBEAT_OK || bf.status == HeartBeatStatus::HEARTBEAT_STUCK) {
1383 2 : rankId2SocketMap_[src].lostNum = 0;
1384 2 : rankId2SocketMap_[src].lostReportCnt = 0;
1385 : }
1386 :
1387 2 : if (bf.status != HeartBeatStatus::HEARTBEAT_OK) {
1388 1 : SetStatus(bf.crimer, bf.informer, bf.status);
1389 : }
1390 :
1391 2 : SaveOpInfo(bf.opInfoTagQueueFrame, src);
1392 2 : return HCCL_SUCCESS;
1393 : }
1394 :
1395 7 : void Heartbeat::SetStatus(UIDType& crimer, UIDType& informer, HeartBeatStatus status, bool needBroadcast)
1396 : {
1397 7 : if (rankId2StatusMap_[crimer].status != status) {
1398 5 : rankId2StatusMap_[crimer].informer = informer;
1399 5 : rankId2StatusMap_[crimer].status = status;
1400 5 : rankId2StatusMap_[crimer].needBroadcast = needBroadcast;
1401 5 : if (needBroadcast) {
1402 1 : errRankQueue_.push(crimer);
1403 : }
1404 :
1405 5 : errStatusQueue_.push(HeartBeatFrame(crimer, informer, status, TIME_NOW(), std::chrono::system_clock::now()));
1406 5 : if (errStatusQueue_.size() > EVENT_MAX_CNT) {
1407 0 : errStatusQueue_.pop();
1408 : }
1409 5 : HCCL_RUN_INFO(
1410 : "[%s][%s]local rank [%s]: crimer rank [%s] status[%s] by informer rank [%s]",
1411 : LOG_KEYWORDS_TASK_EXEC.c_str(), LOG_KEYWORDS_HEARTBEAT_EVETN.c_str(), FormatUId(uid_).c_str(),
1412 : FormatUId(crimer).c_str(), GetHeartBeatStatusStr(status).c_str(), FormatUId(informer).c_str());
1413 : }
1414 7 : }
1415 :
1416 4 : bool Heartbeat::IsKeyEvent(HeartBeatFrame& event, HcclUs curTime, const std::string& group)
1417 : {
1418 4 : bool ret = false;
1419 4 : s64 intervalTime = DURATION_US(curTime - event.TOARelative).count() / (TIME_S_TO_MS * ONE_MILLISECOND_OF_USLEEP);
1420 4 : s32 hcclExecTimeout = CommConfiger::GetInstance().GetCommConfigExecTimeOut(group);
1421 4 : s64 execTimeout = hcclExecTimeout;
1422 4 : s64 detectionTime = 0;
1423 4 : switch (event.status) {
1424 1 : case HeartBeatStatus::HEARTBEAT_LOST:
1425 1 : detectionTime = (lostThreshold_ * HEARTBEAT_INTERVAL) / TIME_S_TO_MS;
1426 1 : break;
1427 3 : case HeartBeatStatus::HEARTBEAT_CQE_ERR:
1428 : case HeartBeatStatus::HEARTBEAT_INCONSISTENT:
1429 : case HeartBeatStatus::HEARTBEAT_OPRETRY_NOT_SUPPORT:
1430 3 : detectionTime = 0;
1431 3 : break;
1432 0 : case HeartBeatStatus::HEARTBEAT_STUCK:
1433 0 : detectionTime = 2 * stuckDetectTime_; // 最长探测时间为2倍的卡住检测时间
1434 0 : break;
1435 0 : case HeartBeatStatus::HEARTBEAT_NOTIFY:
1436 : default:
1437 0 : return false; // 当前不支持的事件，不做处理和展现
1438 : }
1439 8 : ret = ((execTimeout - intervalTime - detectionTime) < JITTER_TIME)
1440 4 : && ((intervalTime + detectionTime - execTimeout) < JITTER_TIME);
1441 4 : return ret;
1442 : }
1443 :
1444 96 : void Heartbeat::MakeErrMsg(std::queue<HeartBeatFrame>& keyEvents, std::vector<std::string>& errStatusVec)
1445 : {
1446 99 : while (keyEvents.size() > 0) {
1447 3 : auto& tmp = keyEvents.front();
1448 3 : std::string crimerStr = FormatUId(tmp.crimer);
1449 3 : std::string informerStr = FormatUId(tmp.informer);
1450 :
1451 6 : std::string headStr = "[" + LOG_KEYWORDS_TASK_EXEC + "][" + LOG_KEYWORDS_HEARTBEAT_EVETN + "]"
1452 3 : + "Cluster Exception Location[IP/ID]:[";
1453 :
1454 3 : time_t tm = std::chrono::system_clock::to_time_t(tmp.TOASystem);
1455 3 : std::string timeStr(ctime(&tm));
1456 3 : if (!timeStr.empty()) { // ctime()函数自带换行符，需要去掉
1457 3 : timeStr.pop_back();
1458 : }
1459 3 : timeStr = ", Arrival Time:[" + timeStr + "]";
1460 :
1461 6 : std::string errStr = ", ExceptionType:";
1462 3 : std::string reasonStr = ", Possible Reason:";
1463 3 : switch (tmp.status) {
1464 1 : case HeartBeatStatus::HEARTBEAT_LOST:
1465 1 : errStr = errStr + "[Heartbeat Lost Occurred]";
1466 1 : reasonStr = reasonStr + "1. Process has exited, 2. Network Disconnected";
1467 : errStr
1468 1 : = headStr + crimerStr + "]" + timeStr + ", Discoverer:[" + informerStr + "]" + errStr + reasonStr;
1469 1 : break;
1470 0 : case HeartBeatStatus::HEARTBEAT_NOTIFY:
1471 0 : errStr = errStr + "[Notify Wait Error Occurred]";
1472 0 : errStr = headStr + crimerStr + "]" + timeStr + errStr;
1473 0 : break;
1474 1 : case HeartBeatStatus::HEARTBEAT_OPRETRY_NOT_SUPPORT:
1475 1 : errStr = errStr + "[OpRetry Not Supported Occurred]";
1476 1 : reasonStr = reasonStr + "OpRetry is not supported";
1477 1 : errStr = headStr + crimerStr + "]" + timeStr + errStr + reasonStr;
1478 1 : break;
1479 1 : case HeartBeatStatus::HEARTBEAT_CQE_ERR:
1480 1 : errStr = errStr + "[Error cqe Occurred]";
1481 1 : reasonStr = reasonStr + "1.Network Disconnected, 2.Remote Rank Coredown";
1482 1 : errStr = headStr + crimerStr + "]" + timeStr + errStr + reasonStr;
1483 1 : break;
1484 0 : case HeartBeatStatus::HEARTBEAT_STUCK:
1485 0 : errStr = errStr + "[Stuck Occurred]";
1486 0 : reasonStr = reasonStr + "1.Host process is stuck, 2.Device task is stuck";
1487 0 : errStr = headStr + crimerStr + "]" + timeStr + errStr + reasonStr;
1488 0 : break;
1489 0 : case HeartBeatStatus::HEARTBEAT_INCONSISTENT:
1490 0 : errStr = errStr + "[Op Inconsistent Occurred]";
1491 0 : reasonStr = reasonStr + "communication operator is inconsistent";
1492 0 : errStr = headStr + crimerStr + "]" + timeStr + errStr + reasonStr;
1493 0 : break;
1494 0 : default:
1495 0 : errStr = " Unknown";
1496 : }
1497 3 : errStatusVec.emplace_back(errStr);
1498 3 : keyEvents.pop();
1499 3 : }
1500 96 : }
1501 19 : std::vector<std::string> Heartbeat::PrintEvents(std::map<HeartBeatStatus, std::queue<HeartBeatFrame>>& keyEvents)
1502 : {
1503 19 : std::vector<std::string> errStatusVec;
1504 : // 打印优先级 opretry not support > error cqe > stuck > lost
1505 19 : MakeErrMsg(keyEvents[HeartBeatStatus::HEARTBEAT_OPRETRY_NOT_SUPPORT], errStatusVec);
1506 19 : MakeErrMsg(keyEvents[HeartBeatStatus::HEARTBEAT_CQE_ERR], errStatusVec);
1507 19 : MakeErrMsg(keyEvents[HeartBeatStatus::HEARTBEAT_STUCK], errStatusVec);
1508 19 : MakeErrMsg(keyEvents[HeartBeatStatus::HEARTBEAT_LOST], errStatusVec);
1509 19 : MakeErrMsg(keyEvents[HeartBeatStatus::HEARTBEAT_INCONSISTENT], errStatusVec);
1510 19 : return errStatusVec;
1511 0 : }
1512 19 : std::vector<std::string> Heartbeat::GetErrStatusVec(const std::string& group)
1513 : {
1514 19 : std::unique_lock<std::mutex> lock(ProcessLock_);
1515 19 : HcclUs curTime = TIME_NOW();
1516 19 : std::map<HeartBeatStatus, std::queue<HeartBeatFrame>> keyEvents;
1517 22 : while (errStatusQueue_.size() > 0) {
1518 3 : auto& tmp = errStatusQueue_.front();
1519 3 : if (IsKeyEvent(tmp, curTime, group)) { // 非关键事件不处理
1520 2 : keyEvents[tmp.status].push(tmp);
1521 : }
1522 3 : errStatusQueue_.pop();
1523 : }
1524 38 : return PrintEvents(keyEvents);
1525 19 : }
1526 :
1527 4 : void Heartbeat::ProcessExceptionEvent()
1528 : {
1529 5 : while (errRankQueue_.size() > 0) {
1530 1 : UIDType cur = errRankQueue_.front();
1531 1 : rankId2StatusMap_[cur].needBroadcast = false;
1532 4989 : OpInfoTagQueueFrame opInfoTagQueueFrame;
1533 2 : for (auto iterRem = rankId2SocketMap_.begin(); iterRem != rankId2SocketMap_.end(); iterRem++) {
1534 1 : UIDType rem = iterRem->first;
1535 1 : if (rem != rankId2StatusMap_[cur].informer
1536 1 : && rankId2StatusMap_[rem].status == HeartBeatStatus::HEARTBEAT_OK) {
1537 1 : if (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) {
1538 1 : (void)SendFrame(rem, cur, rankId2StatusMap_[cur].informer, rankId2StatusMap_[cur].status);
1539 : } else {
1540 0 : (void)SendFrameWithOpCheck(
1541 0 : rem, cur, rankId2StatusMap_[cur].informer, rankId2StatusMap_[cur].status, opInfoTagQueueFrame);
1542 : }
1543 : }
1544 : }
1545 1 : errRankQueue_.pop();
1546 : }
1547 4 : }
1548 :
1549 3 : void Heartbeat::CreateHBLinksAsync()
1550 : {
1551 3 : std::unique_lock<std::mutex> infoLock(hbLinkConnInfoMtx_);
1552 3 : if (hbLinkConnInfo_.empty()) {
1553 3 : return;
1554 : }
1555 0 : linkThreadRunning_ = true;
1556 0 : std::queue<std::tuple<std::string, UIDType, ConnInfo>> connInfoQueue;
1557 0 : for (auto& pair : hbLinkConnInfo_) {
1558 0 : const std::string& groupName = pair.first;
1559 0 : auto& groupConnInfoQueue = pair.second;
1560 0 : while (!groupConnInfoQueue.empty()) {
1561 0 : connInfoQueue.push(
1562 0 : std::make_tuple(groupName, groupConnInfoQueue.front().first, groupConnInfoQueue.front().second));
1563 0 : groupConnInfoQueue.pop();
1564 : }
1565 : }
1566 0 : infoLock.unlock();
1567 0 : while (!connInfoQueue.empty()) {
1568 0 : const std::string groupName = std::get<0>(connInfoQueue.front());
1569 0 : const UIDType& remUid = std::get<1>(connInfoQueue.front());
1570 0 : ConnInfo& connInfo = std::get<2>(connInfoQueue.front());
1571 0 : auto it = linkThreadMap_.find(remUid);
1572 0 : if (it != linkThreadMap_.end() && it->second->joinable()) {
1573 0 : it->second->join();
1574 0 : HCCL_INFO(
1575 : "[CreateHBLinksAsync] Heartbeat link thread has been joined. Group[%s], remote uid[%s].",
1576 : groupName.c_str(), FormatUId(remUid).c_str());
1577 : }
1578 0 : linkThreadMap_[remUid].reset(new (std::nothrow) std::thread(
1579 0 : &Heartbeat::CreateLinkWithRemote, std::ref(*this), groupName, remUid, connInfo));
1580 0 : if (linkThreadMap_[remUid] == nullptr) {
1581 0 : HCCL_RUN_WARNING(
1582 : "Group[%s] establish rank[%s] to rank[%s] heartbeat connection failed. Reason: "
1583 : "create thread failed.",
1584 : groupName.c_str(), FormatUId(uid_).c_str(), FormatUId(remUid).c_str());
1585 : }
1586 0 : connInfoQueue.pop();
1587 0 : }
1588 0 : return;
1589 3 : }
1590 :
1591 7 : void Heartbeat::HeartbeatStatusMonitor()
1592 : {
1593 : // 给当前线程添加名字
1594 7 : SetThreadName("Hccl_HeartBeat");
1595 :
1596 7 : u32 count = 0;
1597 7 : if (deviceLogicId_ != static_cast<u32>(HOST_DEVICE_ID)) {
1598 3 : hrtSetDevice(deviceLogicId_);
1599 : }
1600 7 : uint64_t cnt = 0;
1601 : HcclResult ret;
1602 7 : auto counterStat = CounterStat();
1603 7 : InitStuckDetection(counterStat);
1604 16 : while (startSendRecvTask_) {
1605 9 : CheckSnapshotStatus();
1606 9 : if (isPaused_) {
1607 0 : std::this_thread::sleep_for(std::chrono::milliseconds(BROADCAST_INTERVAL));
1608 0 : continue;
1609 : }
1610 9 : CreateHBLinksAsync();
1611 9 : ProcessLock_.lock();
1612 9 : count++;
1613 9 : if (count >= HEARTBEAT_COUNT) {
1614 0 : count = 0;
1615 0 : OpInfoTagQueueFrame opInfoTagQueueFrame;
1616 0 : GetSendOpInfoList(opInfoTagQueueFrame);
1617 0 : for (auto iter = rankId2SocketMap_.begin(); iter != rankId2SocketMap_.end(); iter++) {
1618 0 : UIDType rem = iter->first;
1619 0 : HCCL_DEBUG(
1620 : "rank[%s] Try to Send HeartBeat to rank[%s]", FormatUId(uid_).c_str(), FormatUId(rem).c_str());
1621 0 : rankId2SocketMap_[rem].lostNum++;
1622 0 : HeartBeatStatus status = HeartBeatStatus::HEARTBEAT_OK;
1623 0 : if (counterStat.issueCnt != 0) {
1624 0 : status = HeartBeatStatus::HEARTBEAT_STUCK;
1625 : }
1626 0 : if (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) {
1627 0 : ret = SendFrame(rem, uid_, uid_, status);
1628 : } else {
1629 0 : ret = SendFrameWithOpCheck(rem, uid_, uid_, status, opInfoTagQueueFrame);
1630 : }
1631 0 : if (ret == HCCL_E_INTERNAL) {
1632 0 : errorSocket_.push_back(rem);
1633 : }
1634 : }
1635 0 : DelErrorSocket();
1636 0 : ProcessCqeErrInfo();
1637 0 : if (counterStat.issueCnt != 0) {
1638 0 : SetStatus(uid_, uid_, HeartBeatStatus::HEARTBEAT_STUCK);
1639 : }
1640 : }
1641 :
1642 21 : for (auto iter = rankId2SocketMap_.begin(); iter != rankId2SocketMap_.end(); iter++) {
1643 12 : UIDType rem = iter->first;
1644 12 : HCCL_DEBUG("rank[%s] Try to Recv from rank[%s]", FormatUId(uid_).c_str(), FormatUId(rem).c_str());
1645 12 : ret = (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) ? RecvFrame(rem) :
1646 0 : RecvFrameWithOpCheck(rem);
1647 12 : if (ret == HCCL_E_INTERNAL) {
1648 0 : errorSocket_.push_back(rem);
1649 0 : continue;
1650 : }
1651 12 : uint32_t threshold = lostThreshold_ << rankId2SocketMap_[rem].lostReportCnt; // LOST帧发送周期放长
1652 12 : if (rankId2SocketMap_[rem].lostNum >= threshold) {
1653 1 : SetStatus(rem, uid_, HeartBeatStatus::HEARTBEAT_LOST);
1654 1 : rankId2SocketMap_[rem].lostReportCnt++;
1655 : }
1656 : }
1657 9 : CheckRecvOpInfoList();
1658 9 : DelErrorSocket();
1659 9 : StuckDetection(cnt, counterStat);
1660 9 : ProcessExceptionEvent();
1661 9 : ProcessLock_.unlock();
1662 :
1663 9 : auto sleeptime = (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) ?
1664 : BROADCAST_INTERVAL :
1665 9 : BROADCAST_INTERVAL_WITH_CHECK;
1666 9 : std::this_thread::sleep_for(std::chrono::milliseconds(sleeptime));
1667 : }
1668 7 : linkThreadRunning_ = false;
1669 : // 在心跳进程结束之前join所有的建链线程
1670 7 : for (auto& pair : linkThreadMap_) {
1671 0 : if (pair.second != nullptr && pair.second->joinable()) {
1672 0 : pair.second->join();
1673 0 : HCCL_INFO("[HeartbeatStatusMonitor] thread has joined. Remote uid is [%s]", FormatUId(pair.first).c_str());
1674 : }
1675 : }
1676 :
1677 7 : if (deviceLogicId_ != static_cast<u32>(HOST_DEVICE_ID)) {
1678 3 : hrtResetDevice(deviceLogicId_);
1679 : }
1680 7 : }
1681 :
1682 7 : void Heartbeat::InitStuckDetection(CounterStat& counterStat)
1683 : {
1684 7 : counterStat.isNeedDetect = (GetExternalInputStuckDetect() == true) ? true : false;
1685 7 : counterStat.couterPrintInter = stuckDetectTime_ * THROUND_MILS / BROADCAST_INTERVAL;
1686 7 : }
1687 :
1688 5 : void Heartbeat::StuckDetection(uint64_t& cnt, CounterStat& counterStat)
1689 : {
1690 5 : HCCL_DEBUG(
1691 : "cnt: %d, isNeedDetect: %d, issueCnt:%llu, interTimes:%d", cnt, counterStat.isNeedDetect, counterStat.issueCnt,
1692 : counterStat.couterPrintInter);
1693 5 : cnt++;
1694 5 : HcclResult ret = HCCL_SUCCESS;
1695 5 : if (counterStat.isNeedDetect && cnt % counterStat.couterPrintInter == 0) {
1696 2 : if (counterStat.isFirst) {
1697 1 : OpExeCounter::GetInstance(deviceLogicId_).GetCounter(counterStat.oldCounter);
1698 1 : counterStat.isFirst = false;
1699 : } else {
1700 1 : ret = OpExeCounter::GetInstance(deviceLogicId_).GetCounter(counterStat.newCounter);
1701 1 : if (ret == HCCL_SUCCESS && counterStat.newCounter.first == counterStat.oldCounter.first
1702 1 : && counterStat.newCounter.first == counterStat.oldCounter.second
1703 1 : && counterStat.newCounter.first == counterStat.newCounter.second) {
1704 1 : HCCL_RUN_INFO(
1705 : "[HCCL_TRACE]rank:%s, count of currently executed operators:%d", FormatUId(uid_).c_str(),
1706 : counterStat.newCounter.first);
1707 1 : counterStat.couterPrintInter *= (BASE_NUMBER << counterStat.issueCnt); // 检测卡住后，把检测周期放长
1708 1 : counterStat.issueCnt++;
1709 : } else {
1710 : // 检测不卡之后，检测间隔恢复到默认间隔
1711 0 : counterStat.couterPrintInter = stuckDetectTime_ * THROUND_MILS / BROADCAST_INTERVAL;
1712 0 : counterStat.issueCnt = 0;
1713 : }
1714 1 : counterStat.oldCounter = counterStat.newCounter; // 更新旧的计数器
1715 : }
1716 : }
1717 5 : }
1718 :
1719 1 : void Heartbeat::PrintAndBroadCastErrorCqe(const ErrCqeInfo& info)
1720 : {
1721 : time_t tmpt;
1722 : struct tm* now;
1723 1 : if (info.cqeInfo.status == 0) {
1724 0 : return;
1725 : }
1726 :
1727 1 : SetStatus(uid_, uid_, HeartBeatStatus::HEARTBEAT_CQE_ERR);
1728 1 : tmpt = static_cast<time_t>(info.cqeInfo.time.tv_sec);
1729 1 : now = localtime(&tmpt);
1730 :
1731 : char errorLinkLogBuffer[LOG_TMPBUF_SIZE];
1732 3 : s32 stringRet = snprintf_s(
1733 : errorLinkLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1734 : "localInfo{server[%s],deviceId[%d],deviceIp[%s]}, remoteIP{server[%s],deviceId[%d],deviceIp[%s]}",
1735 1 : info.linkInfo.localServerId.c_str(), info.linkInfo.localDevicePhyId, nicIp_.GetReadableAddress(),
1736 1 : info.linkInfo.remoteServerId.c_str(), info.linkInfo.remoteDevicePhyId,
1737 : info.cqeInfo.remoteIp.GetReadableAddress());
1738 1 : CHK_PRT_CONT(stringRet == -1, HCCL_ERROR("[Create][DestLink]Transport init error! Failed to build log info"));
1739 :
1740 1 : if (now == nullptr) {
1741 0 : HCCL_ERROR(
1742 : "[%s][%s][%s]localtime fail, cqe error status[%u], %s", LOG_KEYWORDS_TASK_EXEC.c_str(),
1743 : LOG_KEYWORDS_HEARTBEAT_EVETN.c_str(), LOG_KEYWORDS_CQE_ERROR.c_str(), info.cqeInfo.status,
1744 : errorLinkLogBuffer);
1745 : } else {
1746 1 : HCCL_ERROR(
1747 : "[%s][%s][%s]cqe error status[%u], time:[%04u-%02d-%02d %02d:%0d:%02d.%06u], %s",
1748 : LOG_KEYWORDS_TASK_EXEC.c_str(), LOG_KEYWORDS_HEARTBEAT_EVETN.c_str(), LOG_KEYWORDS_CQE_ERROR.c_str(),
1749 : info.cqeInfo.status, now->tm_year + TIME_FROM_1900, now->tm_mon + 1, now->tm_mday, now->tm_hour,
1750 : now->tm_min, now->tm_sec, static_cast<u32>(info.cqeInfo.time.tv_usec), errorLinkLogBuffer);
1751 : }
1752 :
1753 1 : std::unique_lock<std::mutex> lock(remoteIpMutex_);
1754 1 : auto search = remoteIpMap.find(info.linkInfo.identifier);
1755 1 : if (search != remoteIpMap.end()) {
1756 0 : remoteIpMap[info.linkInfo.identifier].insert(info);
1757 : } else {
1758 1 : std::set<ErrCqeInfo> remoteInfoSet;
1759 1 : remoteInfoSet.insert(info);
1760 1 : remoteIpMap.insert(std::pair<std::string, std::set<ErrCqeInfo>>(info.linkInfo.identifier, remoteInfoSet));
1761 1 : }
1762 1 : }
1763 :
1764 6 : void Heartbeat::SaveQpnForOpRetry(const ErrCqeInfo& info)
1765 : {
1766 6 : if (info.cqeInfo.status == 0) {
1767 2 : return;
1768 : }
1769 :
1770 4 : HCCL_RUN_INFO(
1771 : "[Heartbeat][SaveQpnForOpRetry]receive a cqe error [%u][%u], dstrank[%u] identifier[%s]", info.cqeInfo.status,
1772 : info.qpn, info.linkInfo.remoteRank, info.linkInfo.identifier.c_str());
1773 4 : auto identiSearch = rankMapForRetryAgent.find(info.linkInfo.identifier);
1774 4 : if (identiSearch != rankMapForRetryAgent.end()) {
1775 3 : auto rankSearch = identiSearch->second.find(info.linkInfo.remoteRank);
1776 3 : if (rankSearch != identiSearch->second.end()) {
1777 2 : (*rankSearch).second.insert(info);
1778 : } else {
1779 4 : identiSearch->second.insert({info.linkInfo.remoteRank, {info}});
1780 : }
1781 : } else {
1782 1 : std::map<u32, std::set<ErrCqeInfo>> rankExtendMap;
1783 2 : rankExtendMap[info.linkInfo.remoteRank] = {info};
1784 1 : rankMapForRetryAgent.insert(std::make_pair(info.linkInfo.identifier, rankExtendMap));
1785 1 : }
1786 2 : }
1787 :
1788 0 : void Heartbeat::OpRetryCQEHandle(const HcclNetDevCtx netDevCtx)
1789 : {
1790 0 : u32 cqeNum = RETRY_CQE_ARRAY_SIZE;
1791 : do {
1792 0 : cqeNum = RETRY_CQE_ARRAY_SIZE;
1793 :
1794 0 : std::vector<ErrCqeInfo> infos;
1795 0 : HcclResult ret = HcclCommunicator::GetTransportCqeErrors(netDevCtx, infos, cqeNum);
1796 0 : if (ret != HCCL_SUCCESS || cqeNum == 0) {
1797 0 : return;
1798 : }
1799 0 : for (auto& info : infos) {
1800 0 : if (GetRetryEnable(info)
1801 0 : && CommConfiger::GetInstance().GetCommConfigInterSuperPodRetryEnable(info.linkInfo.identifier)) {
1802 0 : SaveQpnForOpRetry(info);
1803 : } else {
1804 0 : PrintAndBroadCastErrorCqe(info);
1805 : }
1806 : }
1807 0 : } while (cqeNum == RETRY_CQE_ARRAY_SIZE);
1808 : }
1809 :
1810 0 : bool Heartbeat::GetRetryEnable(const ErrCqeInfo& info)
1811 : {
1812 0 : std::lock_guard<std::mutex> retryEnablelock(retryEnableMutex_);
1813 0 : auto search = retryEnableTable_.find(info.linkInfo.identifier);
1814 0 : if (search != retryEnableTable_.end()) {
1815 0 : return search->second;
1816 : }
1817 0 : return false;
1818 0 : }
1819 800 : HcclResult Heartbeat::ClearRetryEnableMapItem(const std::string& identifier)
1820 : {
1821 800 : CHK_PRT_RET(initialized_ == false, HCCL_WARNING("Heartbeat has been destroyed"), HCCL_SUCCESS);
1822 0 : u32 delRes = 0;
1823 : {
1824 0 : std::lock_guard<std::mutex> retryEnablelock(retryEnableMutex_);
1825 0 : delRes = retryEnableTable_.erase(identifier);
1826 0 : if (delRes != 0) {
1827 0 : HCCL_INFO("[Heartbeat][ClearRetryEnableMapItem] del identifier[%s] succ", identifier.c_str());
1828 : } else {
1829 0 : HCCL_DEBUG("[Heartbeat][ClearRetryEnableMapItem] identifier[%s] is not found.", identifier.c_str());
1830 : }
1831 0 : }
1832 0 : std::lock_guard<std::mutex> bakcupEnablelock(backupEnableMutex_);
1833 0 : delRes = backupEnableTable_.erase(identifier);
1834 0 : if (delRes != 0) {
1835 0 : HCCL_INFO("[Heartbeat][ClearRetryEnableMapItem] del backup identifier[%s] succ", identifier.c_str());
1836 : } else {
1837 0 : HCCL_DEBUG("[Heartbeat][ClearRetryEnableMapItem] identifier[%s] is not found.", identifier.c_str());
1838 : }
1839 0 : return HCCL_SUCCESS;
1840 0 : }
1841 21 : void Heartbeat::ProcessCqeErrInfoByNetDevCtx(const HcclIpAddress& nicIp)
1842 : {
1843 21 : std::unique_lock<std::mutex> mapLock(ctxMapMutex_);
1844 21 : auto iter = netDevCtxMap_.find(nicIp);
1845 21 : if (iter == netDevCtxMap_.end() || netDevCtxMap_[nicIp] == nullptr) {
1846 7 : return;
1847 : }
1848 14 : mapLock.unlock();
1849 14 : const HcclNetDevCtx netDevCtx = iter->second;
1850 14 : std::vector<ErrCqeInfo> infos;
1851 14 : u32 cqeNum = 1;
1852 14 : HcclResult ret = HcclCommunicator::GetTransportCqeErrors(netDevCtx, infos, cqeNum);
1853 14 : if (ret != HCCL_SUCCESS || infos.size() == 0) {
1854 14 : return;
1855 : }
1856 0 : if (GetRetryEnable(infos[0])
1857 0 : && CommConfiger::GetInstance().GetCommConfigInterSuperPodRetryEnable(infos[0].linkInfo.identifier)) {
1858 0 : SaveQpnForOpRetry(infos[0]);
1859 : } else {
1860 0 : PrintAndBroadCastErrorCqe(infos[0]);
1861 : }
1862 : // infoList 处理
1863 0 : OpRetryCQEHandle(netDevCtx);
1864 35 : }
1865 :
1866 20 : void Heartbeat::ProcessCqeErrInfo()
1867 : {
1868 20 : ProcessCqeErrInfoByNetDevCtx(nicIp_);
1869 20 : if (IsEnableBackupLink()) {
1870 1 : ProcessCqeErrInfoByNetDevCtx(backupNicIp_);
1871 : }
1872 20 : }
1873 :
1874 3 : void Heartbeat::DelErrorSocket()
1875 : {
1876 3 : for (auto rem : errorSocket_) {
1877 0 : HCCL_RUN_INFO(
1878 : "rank[%s] Try to Send/recv HeartBeat to rank[%s]", FormatUId(uid_).c_str(), FormatUId(rem).c_str());
1879 0 : rankId2StatusMap_.erase(rem);
1880 0 : if (rankId2SocketMap_.has(rem)) {
1881 0 : if (rankId2SocketMap_[rem].socket->GetLocalRole() == HcclSocketRole::SOCKET_ROLE_SERVER
1882 0 : && listenSocketMap_.find(rankId2SocketMap_[rem].socket->GetLocalIp()) != listenSocketMap_.end()) {
1883 0 : listenSocketMap_[rankId2SocketMap_[rem].socket->GetLocalIp()]->DelWhiteList(
1884 0 : rankId2SocketMap_[rem].wlistInfosVec);
1885 : }
1886 0 : rankId2SocketMap_[rem].socket->Close();
1887 0 : while (rankId2SocketMap_.erase(rem)) {
1888 : };
1889 : }
1890 : }
1891 3 : errorSocket_.clear();
1892 3 : }
1893 :
1894 2 : HcclResult Heartbeat::GetQpnErr(const std::string& identifier, std::set<std::tuple<u32, u32, u32>>& qpErrSet)
1895 : {
1896 2 : std::unique_lock<std::mutex> lock(qpnMapMutexForRetry_);
1897 2 : auto search = rankMapForRetryAgent.find(identifier);
1898 2 : if (search == rankMapForRetryAgent.end()) {
1899 1 : HCCL_INFO("[GetQpnErr]identifier[%s] is not found", identifier.c_str());
1900 1 : return HCCL_SUCCESS;
1901 : }
1902 1 : if (search->second.size() > 0) {
1903 2 : for (auto iter : search->second) {
1904 1 : u32 dstRank = iter.first;
1905 2 : for (auto qpnInfo : iter.second) {
1906 1 : u32 status = qpnInfo.cqeInfo.status;
1907 1 : qpErrSet.insert(std::make_tuple(dstRank, status, qpnInfo.qpn));
1908 1 : }
1909 1 : }
1910 : }
1911 1 : HCCL_INFO("[GetQpnErr]identifier[%s] is found, qpErrSet size is %u", identifier.c_str(), qpErrSet.size());
1912 1 : return HCCL_SUCCESS;
1913 2 : }
1914 : // OpRetry 失败后，将进行广播操作
1915 1 : HcclResult Heartbeat::BroadcastCqeErr(const std::string& identifier)
1916 : {
1917 1 : u32 cqeSize = 0;
1918 1 : std::unique_lock<std::mutex> qpnMaplock(qpnMapMutexForRetry_);
1919 1 : auto search = rankMapForRetryAgent.find(identifier);
1920 1 : if (search != rankMapForRetryAgent.end()) {
1921 1 : if (search->second.size() > 0) {
1922 1 : cqeSize = search->second.size();
1923 2 : for (auto& qpInfo : search->second) {
1924 2 : for (auto qpnset : qpInfo.second) {
1925 1 : PrintAndBroadCastErrorCqe(qpnset);
1926 1 : HCCL_RUN_INFO(
1927 : "[BroadcastCqeErr][item]remoteIp[%s] remoteRank[%u] status[%u] qpn[%u]",
1928 : qpnset.cqeInfo.remoteIp.GetReadableAddress(), qpInfo.first, qpnset.cqeInfo.status, qpnset.qpn);
1929 1 : }
1930 : }
1931 1 : search->second.clear();
1932 : }
1933 : }
1934 : // 查询剩余量，一般为0
1935 1 : HCCL_RUN_INFO(
1936 : "[Heartbeat][BroadcastCqeErr]clear qpn err size from [%u] to [%u], identifier[%s] ", cqeSize,
1937 : search->second.size(), identifier.c_str());
1938 1 : return HCCL_SUCCESS;
1939 1 : }
1940 :
1941 : /* 非点对点通信 重执行成功后进行调用 */
1942 1 : HcclResult Heartbeat::ClearAllCqeErr(const std::string& identifier)
1943 : {
1944 1 : std::unique_lock<std::mutex> qpnMaplock(qpnMapMutexForRetry_);
1945 1 : u32 cqeSize = 0;
1946 1 : auto search = rankMapForRetryAgent.find(identifier);
1947 1 : if (search != rankMapForRetryAgent.end()) {
1948 1 : if (search->second.size() > 0) {
1949 0 : cqeSize = search->second.size();
1950 0 : search->second.clear();
1951 : }
1952 : }
1953 : // 查询剩余量，一般为0
1954 1 : HCCL_RUN_INFO(
1955 : "[Heartbeat][ClearAllCqeErr]clear qpn err size from [%u] to [%u], identifier[%s]", cqeSize,
1956 : search->second.size(), identifier.c_str());
1957 1 : return HCCL_SUCCESS;
1958 1 : }
1959 : /* 点对点通信 重执行成功后进行调用
1960 : */
1961 5 : HcclResult Heartbeat::ClearCqeErr(const std::string& identifier, u32 remoteRank, u32 qpn)
1962 : {
1963 5 : HCCL_RUN_INFO(
1964 : "[Heartbeat][ClearCqeErr] identifier[%s] remoteRank[%u] qpn[%u].", identifier.c_str(), remoteRank, qpn);
1965 5 : std::unique_lock<std::mutex> qpnMaplock(qpnMapMutexForRetry_);
1966 5 : const auto& search = rankMapForRetryAgent.find(identifier);
1967 5 : if (search == rankMapForRetryAgent.end()) {
1968 0 : return HCCL_SUCCESS;
1969 : }
1970 :
1971 5 : auto& ranksearch = rankMapForRetryAgent[identifier];
1972 : // 删除指定通信域内的固定 remoteRank 固定qpn的cqe err
1973 5 : if (ranksearch.find(remoteRank) != ranksearch.end()) {
1974 2 : if (ranksearch[remoteRank].size() == 1) {
1975 : // remotrank只有一个qpn err，直接删除map
1976 1 : ranksearch.erase(remoteRank);
1977 1 : HCCL_RUN_INFO("[ClearCqeErr][qpnClear] clear dstRank[%u] qpn[%u] now", remoteRank, qpn);
1978 1 : } else if (ranksearch[remoteRank].size() > 1) {
1979 3 : for (auto iter = ranksearch[remoteRank].begin(); iter != ranksearch[remoteRank].end();) {
1980 2 : if (iter->qpn == qpn) {
1981 1 : iter = ranksearch[remoteRank].erase(iter);
1982 1 : HCCL_RUN_INFO("[ClearCqeErr][qpnClear] clear dstRank[%u] qpn[%u] now", remoteRank, qpn);
1983 : } else {
1984 1 : ++iter;
1985 : }
1986 : }
1987 : }
1988 : }
1989 :
1990 : // 查询指定通信域剩余的 QP ERROR 数量
1991 5 : HCCL_RUN_INFO("[ClearCqeErr][qpnClear]identifier qpn err left [%u] now.", search->second.size());
1992 5 : return HCCL_SUCCESS;
1993 5 : }
1994 :
1995 3 : HcclResult Heartbeat::CheckErrorCqe(const std::string& identifier, HcclResult& result)
1996 : {
1997 3 : HcclIpAddress ip;
1998 3 : result = HCCL_SUCCESS;
1999 :
2000 3 : std::unique_lock<std::mutex> lock(remoteIpMutex_);
2001 3 : auto search = remoteIpMap.find(identifier);
2002 3 : if (search == remoteIpMap.end()) {
2003 2 : if (qpnDissociativeSet.size() != 0) { // 如果没有发生error cqe异常的通信域，则确认是否存在游离(Destroy)qpn
2004 0 : HCCL_ERROR(
2005 : "[Heartbeat]find cqe error [%d] num[%llu] dissociative. maybe its qp has already been destroyed",
2006 : result, qpnDissociativeSet.size());
2007 0 : qpnDissociativeSet.clear();
2008 0 : return HCCL_E_REMOTE;
2009 : }
2010 2 : return HCCL_SUCCESS;
2011 : }
2012 1 : if (search->second.size() > 0) {
2013 1 : result = HCCL_E_REMOTE;
2014 1 : HCCL_ERROR("[Heartbeat]find cqe error [%d], in comm [%s]", result, identifier.c_str());
2015 2 : for (auto& it : search->second) {
2016 1 : HCCL_ERROR(
2017 : "[Heartbeat]find cqe error, localIP[%s], remoteIP[%s]", nicIp_.GetReadableAddress(),
2018 : it.cqeInfo.remoteIp.GetReadableAddress());
2019 23 : RPT_INPUT_ERR(
2020 : true, "EI0013",
2021 : std::vector<std::string>(
2022 : {"localServerId", "localDeviceId", "localDeviceIp", "remoteServerId", "remoteDeviceId",
2023 : "remoteDeviceIp"}),
2024 : std::vector<std::string>(
2025 : {it.linkInfo.localServerId, std::to_string(it.linkInfo.localDevicePhyId),
2026 : std::string(nicIp_.GetReadableAddress()), it.linkInfo.remoteServerId,
2027 : std::to_string(it.linkInfo.remoteDevicePhyId),
2028 : std::string(it.cqeInfo.remoteIp.GetReadableAddress())}));
2029 : }
2030 : }
2031 1 : lock.unlock();
2032 :
2033 1 : return HCCL_SUCCESS;
2034 5 : }
2035 :
2036 0 : void Heartbeat::RegisterSROpIdentifier(const std::string& identifier, const std::string& paramTag)
2037 : {
2038 : // SR算子通信域映射关系注册
2039 0 : std::lock_guard<std::mutex> lock(srTagMutex_);
2040 0 : if (srTagMap_.size() > SR_TAG_MAP_MAX_NUM) {
2041 0 : srTagMap_.erase(srTagMap_.begin());
2042 : }
2043 :
2044 0 : auto iter = srTagMap_.find(paramTag);
2045 0 : if (iter == srTagMap_.end()) {
2046 0 : srTagMap_.insert(std::make_pair(paramTag, identifier));
2047 : }
2048 0 : }
2049 :
2050 0 : void Heartbeat::AddInconsistentOpRecord(
2051 : const std::string& identifier, const OpInfoDesc& localOpInfo, InconsistentType status, const std::string& localInfo,
2052 : const std::string& remoteInfo)
2053 : {
2054 0 : std::lock_guard<std::mutex> lock(inconsistentOpMutex_);
2055 0 : if (localOpInfo.opType == HcclCMDType::HCCL_CMD_SEND || localOpInfo.opType == HcclCMDType::HCCL_CMD_RECEIVE) {
2056 0 : auto iter = srTagMap_.find(identifier);
2057 0 : if (iter == srTagMap_.end()) {
2058 0 : HCCL_ERROR(
2059 : "[%s] SR tag[%s] may have already been deleted due to prolonged storage time", __func__,
2060 : identifier.c_str());
2061 0 : return;
2062 : }
2063 :
2064 0 : auto search = inconsistentOpMap_.find(iter->second); // SR tag
2065 0 : if (search == inconsistentOpMap_.end()) {
2066 0 : inconsistentOpMap_.insert(
2067 0 : std::make_pair(iter->second, OpInconsistentInfo(status, localInfo, remoteInfo, localOpInfo)));
2068 0 : HCCL_INFO(
2069 : "[%s] save record SR[%s] identifier[%s] index[%d]", __func__, identifier.c_str(), iter->second.c_str(),
2070 : localOpInfo.index);
2071 : }
2072 0 : } else {
2073 0 : auto search = inconsistentOpMap_.find(identifier); // AR identifier
2074 0 : if (search == inconsistentOpMap_.end()) {
2075 0 : inconsistentOpMap_.insert(
2076 0 : std::make_pair(identifier, OpInconsistentInfo(status, localInfo, remoteInfo, localOpInfo)));
2077 0 : HCCL_INFO("[%s] save record identifier[%s] index[%d]", __func__, identifier.c_str(), localOpInfo.index);
2078 : }
2079 : }
2080 0 : }
2081 :
2082 0 : HcclResult Heartbeat::CheckOpInconsistentError(const std::string& identifier, HcclResult& result)
2083 : {
2084 0 : if (GetExternalInconsistentCheckSwitch() != InconsistentCheckMode::ON) {
2085 0 : return HCCL_SUCCESS;
2086 : }
2087 0 : std::lock_guard<std::mutex> lock(inconsistentOpMutex_);
2088 0 : auto search = inconsistentOpMap_.find(identifier);
2089 0 : if (search != inconsistentOpMap_.end()) {
2090 0 : result = HCCL_E_PARA;
2091 0 : const OpInconsistentInfo& inconsistentInfo = search->second;
2092 0 : std::string opInfo = "Unknown";
2093 0 : for (const auto& pair : HCCL_OPTYPE_NAME_MAP) {
2094 0 : if (pair.second == inconsistentInfo.opInfoDesc.opType) {
2095 0 : opInfo = std::string(pair.first);
2096 0 : break;
2097 : }
2098 : }
2099 0 : HCCL_ERROR(
2100 : "[%s]find inconsistent op [%s] error [%d], in comm [%s]", __func__, opInfo, result, identifier.c_str());
2101 0 : RPT_INPUT_ERR(
2102 : true, "EI0005", std::vector<std::string>({"ccl_op", "group", "para_name", "local_para", "remote_para"}),
2103 : std::vector<std::string>(
2104 : {opInfo, identifier, GetInconsistentTypeStr(search->second.inconsistentType), search->second.localInfo,
2105 : search->second.remoteInfo}));
2106 0 : }
2107 0 : return HCCL_SUCCESS;
2108 0 : }
2109 :
2110 24 : HcclResult Heartbeat::SetRankPortInfo(
2111 : bool isUseRankPort, std::vector<u32>& nicRanksPorts, std::vector<u32>& vnicRanksPorts, bool devPortSwitchOn)
2112 : {
2113 24 : isUseRankPort_ = isUseRankPort;
2114 24 : nicRanksPorts_ = nicRanksPorts;
2115 24 : vnicRanksPorts_ = vnicRanksPorts;
2116 24 : devPortSwitchOn_ = devPortSwitchOn;
2117 24 : return HCCL_SUCCESS;
2118 : }
2119 :
2120 0 : void Heartbeat::SetOpretryErr()
2121 : {
2122 : // 重执行约束场景，给errStatusQueue添加重执行失败心跳帧
2123 0 : SetStatus(uid_, uid_, HeartBeatStatus::HEARTBEAT_OPRETRY_NOT_SUPPORT);
2124 0 : }
2125 :
2126 8 : u32 Heartbeat::GetPort(HcclSocketType type, u32 remoteUserRank, [[maybe_unused]] u32 remoteDeviceId)
2127 : {
2128 8 : u32 port = HCCL_INVALID_PORT;
2129 8 : if (isUseRankPort_) {
2130 0 : if (devPortSwitchOn_ && type == HcclSocketType::SOCKET_VNIC && remoteUserRank < vnicRanksPorts_.size()
2131 0 : && vnicRanksPorts_[remoteUserRank] != HCCL_INVALID_PORT) {
2132 0 : port = vnicRanksPorts_[remoteUserRank];
2133 0 : HCCL_INFO("[Heartbeat][GetPort] use vnic ranks port[%u]", port);
2134 0 : } else if (remoteUserRank < nicRanksPorts_.size() && nicRanksPorts_[remoteUserRank] != HCCL_INVALID_PORT) {
2135 0 : port = nicRanksPorts_[remoteUserRank];
2136 0 : HCCL_INFO("[Heartbeat][GetPort] use nic ranks port[%u]", port);
2137 : } else {
2138 0 : port = HETEROG_CCL_PORT;
2139 : }
2140 : } else {
2141 8 : port = HETEROG_CCL_PORT;
2142 : }
2143 8 : return port;
2144 : }
2145 :
2146 2 : u32 Heartbeat::GetHostPort(s32 devicePhyId)
2147 : {
2148 2 : if (GetExternalInputHcclIfBasePort() == HCCL_INVALID_PORT) {
2149 1 : return (devicePhyId + HOST_PARA_BASE_PORT);
2150 : } else {
2151 1 : return (devicePhyId + GetExternalInputHcclIfBasePort() + HCCL_AISERVER_DEVICE_NUM);
2152 : }
2153 : }
2154 :
2155 0 : bool Heartbeat::IsPaused() const { return !startSendRecvTask_ || isPaused_; }
2156 :
2157 0 : bool Heartbeat::IsResumed() const { return !startSendRecvTask_ || !isPaused_; }
2158 :
2159 3 : void Heartbeat::CheckSnapshotStatus()
2160 : {
2161 3 : auto snapshotStatus = SnapshotControl::GetInstance(deviceLogicId_).GetStatus();
2162 3 : if (isPaused_ && snapshotStatus == SnapshotStatus::POST_SNAPSHOT) {
2163 0 : isPaused_ = false;
2164 0 : HCCL_RUN_INFO(
2165 : "[Heartbeat][CheckSnapshotStatus] detect snapshot post-processing, heart is resumed, "
2166 : "deviceLogicId[%u].",
2167 : deviceLogicId_);
2168 3 : } else if (!isPaused_ && snapshotStatus == SnapshotStatus::PRE_SNAPSHOT) {
2169 0 : isPaused_ = true;
2170 0 : HCCL_RUN_INFO(
2171 : "[Heartbeat][CheckSnapshotStatus] detect snapshot pre-processing, heart is paused, "
2172 : "deviceLogicId[%u].",
2173 : deviceLogicId_);
2174 : }
2175 3 : }
2176 :
2177 0 : HcclResult RegisterToHeartBeat(
2178 : s32 deviceLogicID, u32 userRank, DevType devType, std::vector<RankInfo>& rankInfoList, const u32 port,
2179 : const bool isNeedNic, u32 peerRankId, const std::string& commIdentifier, const std::string& tag,
2180 : bool useSuperPodMode, bool isUsedRdmaLevel0)
2181 : {
2182 0 : return peerRankId == INVALID_VALUE_RANKID ? Heartbeat::GetInstance(deviceLogicID)
2183 0 : .RegisterToHeartBeat(
2184 : userRank, devType, rankInfoList, port, isNeedNic,
2185 : commIdentifier, useSuperPodMode, isUsedRdmaLevel0) :
2186 0 : Heartbeat::GetInstance(deviceLogicID)
2187 0 : .RegisterToHeartBeat(
2188 : userRank, devType, rankInfoList, port, isNeedNic, peerRankId,
2189 0 : commIdentifier, tag, useSuperPodMode, isUsedRdmaLevel0);
2190 : }
2191 :
2192 0 : void UnRegisterRanks(s32 deviceLogicID, DevType devType, const std::string& commIdentifier, const std::string& tag)
2193 : {
2194 0 : return tag.empty() ? Heartbeat::GetInstance(deviceLogicID).UnRegisterToHeartBeat(devType, commIdentifier) :
2195 0 : Heartbeat::GetInstance(deviceLogicID).UnRegisterToHeartBeat(devType, commIdentifier, tag);
2196 : }
2197 :
2198 0 : HcclResult SetRankPortInfo(s32 deviceLogicID, bool isUseRankPort, std::vector<u32>& ranksPort)
2199 : {
2200 0 : return Heartbeat::GetInstance(deviceLogicID).SetRankPortInfo(isUseRankPort, ranksPort, ranksPort, false);
2201 : }
2202 :
2203 17 : std::vector<std::string> GetErrStatusVec(s32 deviceLogicID, const std::string& group)
2204 : {
2205 17 : return Heartbeat::GetInstance(deviceLogicID).GetErrStatusVec(group);
2206 : }
2207 :
2208 47 : __attribute__((constructor)) void HeartBeatCallBackInit()
2209 : {
2210 47 : RegisterHeartBeatCallBack(RegisterToHeartBeat, UnRegisterRanks, SetRankPortInfo);
2211 47 : RegisterGetErrStatusVecCallBack(GetErrStatusVec);
2212 47 : }
2213 : } // namespace hccl
|
