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