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