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 : #ifndef HCCL_HEARTBEAT_H
12 : #define HCCL_HEARTBEAT_H
13 :
14 : #include <thread>
15 : #include <map>
16 : #include <deque>
17 : #include <mutex>
18 :
19 : #include "hccl/hccl_types.h"
20 : #include "log.h"
21 : #include "reference_map.h"
22 : #include "ring_buffer.h"
23 : #include "common.h"
24 : #include "sal_pub.h"
25 : #include "hccl_socket_manager.h"
26 : #include "transport_pub.h"
27 : #include "topoinfo_struct.h"
28 : #include "comm_config_pub.h"
29 : namespace hccl {
30 : using RankId = u32;
31 : constexpr u32 BROADCAST_INTERVAL = 50; // 背景线程执行周期为50 ms
32 : constexpr u32 BROADCAST_INTERVAL_WITH_CHECK = 25; // 背景线程执行周期为25 ms
33 : constexpr u32 STUCK_INTERVAL = 300000; // 5min监控一次,默认 300000 ms
34 : constexpr u32 STUCK_COUNT = STUCK_INTERVAL / BROADCAST_INTERVAL;
35 : constexpr u32 OPINFO_SEND_NUM_BY_TAG = 500; // 一次心跳帧发送的算子信息个数
36 : constexpr u32 OPINFO_TAG_QUEUE_NUM = 10; // 一次心跳帧发送的算子信息个数
37 : constexpr u32 HEARTBEAT_INTERVAL = 1000; // 心跳帧发送周期为1000 ms
38 : constexpr u32 HEARTBEAT_COUNT = HEARTBEAT_INTERVAL / BROADCAST_INTERVAL; // 心跳帧发送间隔数
39 : constexpr u32 BASE_NUMBER = 2;
40 : constexpr u32 RETRY_CQE_ARRAY_SIZE = 128; // 重执行时获取的CQE数组的最大数量，最大128
41 : constexpr u32 JITTER_TIME = 300; // 关键事件允许的误差事件范围±300s。误差来源：EVENT和NOTIFY差异、传播耗时、计时误差
42 : constexpr u32 EVENT_MAX_CNT = 5000; // 防止内存泄漏，同时不能太短，防止正常事件被冲掉
43 : constexpr u32 THROUND_MILS = 1000; // 1000ms
44 : constexpr u32 OPINFO_QUEUE_MAX_SIZE = 131072; // 算子下发校验队列最大算子个数，防止内存占用
45 : constexpr u32 MAX_SENDBUFF_SIZE = 3072; // SendBuff[dst] 最大数量
46 : constexpr u32 SR_TAG_MAP_MAX_NUM = 65536;
47 : constexpr u32 HBFRAME_SEND_LOOP_MAX_NUM = 120;
48 : constexpr s32 HCCL_STUCK_DETECT_TIME_MIN = 60; // 卡住检测最短时间
49 : constexpr s32 HCCL_STUCK_DETECT_TIME_BASE = 3; // 卡住检测时间为execTime/3
50 : constexpr s32 HCCL_LOST_THRESHOLD = 30; // 心跳丢失阈值为30s
51 :
52 4 : using UIDType = struct HcclHeartBeatUid {
53 : char id[512] = {0}; // ip[IP_ADDRESS_BUFFER_LEN] + ifname[MAX_INTERFACE_NAME_LEN] + devid 最大不超过512字节
54 1911 : bool operator==(const HcclHeartBeatUid& that) const { return std::string(this->id) == std::string(that.id); }
55 45 : bool operator!=(const HcclHeartBeatUid& that) const { return std::string(this->id) != std::string(that.id); }
56 2135 : bool operator<(const HcclHeartBeatUid& that) const { return std::string(this->id) < std::string(that.id); }
57 : };
58 : } // namespace hccl
59 :
60 : namespace std {
61 : template <>
62 : class hash<hccl::HcclHeartBeatUid> {
63 : public:
64 1642 : size_t operator()(const hccl::HcclHeartBeatUid& uid) const { return hash<string>()(string(uid.id)); }
65 : };
66 : } // namespace std
67 :
68 : namespace hccl {
69 : constexpr u8 HAS_CONN = 1;
70 : constexpr u8 NO_CONN = 0;
71 : constexpr u32 TIME_FROM_1900 = 1900;
72 :
73 : enum class HeartBeatStatus {
74 : HEARTBEAT_OK,
75 : HEARTBEAT_LOST,
76 : HEARTBEAT_NOTIFY,
77 : HEARTBEAT_CQE_ERR,
78 : HEARTBEAT_OPRETRY_NOT_SUPPORT,
79 : HEARTBEAT_STUCK,
80 : HEARTBEAT_INCONSISTENT
81 : };
82 : const std::map<HeartBeatStatus, std::string> HEARTBEAT_STATUS_STR_MAP{
83 : {HeartBeatStatus::HEARTBEAT_OK, "OK"},
84 : {HeartBeatStatus::HEARTBEAT_LOST, "LOST"},
85 : {HeartBeatStatus::HEARTBEAT_NOTIFY, "NOTIFY"},
86 : {HeartBeatStatus::HEARTBEAT_CQE_ERR, "ERROR CQE"},
87 : {HeartBeatStatus::HEARTBEAT_OPRETRY_NOT_SUPPORT, "OPRETRY NOT SUPPORT"},
88 : {HeartBeatStatus::HEARTBEAT_STUCK, "STUCK"},
89 : {HeartBeatStatus::HEARTBEAT_INCONSISTENT, "INCONSISTENT"}};
90 5 : inline std::string GetHeartBeatStatusStr(HeartBeatStatus status)
91 : {
92 5 : auto iter = HEARTBEAT_STATUS_STR_MAP.find(status);
93 5 : if (iter == HEARTBEAT_STATUS_STR_MAP.end()) {
94 0 : return "Unknown";
95 : } else {
96 5 : return iter->second;
97 : }
98 : }
99 :
100 : struct CounterStat {
101 : std::pair<int32_t, int32_t> oldCounter{0, 0};
102 : std::pair<int32_t, int32_t> newCounter{0, 0};
103 : std::uint64_t issueCnt = 0;
104 : bool isNeedDetect = false;
105 : bool isFirst = true;
106 : std::uint64_t couterPrintInter = STUCK_COUNT;
107 9 : CounterStat() {};
108 : };
109 :
110 : enum class InconsistentType {
111 : NO_INCONSISTENT = 0,
112 : OPTYPE_INCONSISTENT = 1,
113 : DATATYPE_INCONSISTENT = 2,
114 : REDUCETYPE_INCONSISTENT = 3,
115 : ROOT_INCONSISTENT = 4,
116 : COUNT_INCONSISTENT = 5
117 : };
118 :
119 : const std::map<InconsistentType, std::string> OP_INCONSISTENT_STR_MAP{
120 : {InconsistentType::NO_INCONSISTENT, "no exist op inconsistent"},
121 : {InconsistentType::OPTYPE_INCONSISTENT, "op type inconsistent"},
122 : {InconsistentType::DATATYPE_INCONSISTENT, "op data type inconsistent"},
123 : {InconsistentType::REDUCETYPE_INCONSISTENT, "op reduce type inconsistent"},
124 : {InconsistentType::ROOT_INCONSISTENT, "op root inconsistent"},
125 : {InconsistentType::COUNT_INCONSISTENT, "op count inconsistent"}};
126 :
127 0 : inline std::string GetInconsistentTypeStr(InconsistentType status)
128 : {
129 0 : auto iter = OP_INCONSISTENT_STR_MAP.find(status);
130 0 : if (iter == OP_INCONSISTENT_STR_MAP.end()) {
131 0 : return "Unknown";
132 : } else {
133 0 : return iter->second;
134 : }
135 : };
136 :
137 : struct OpInfoDesc {
138 : HcclCMDType opType = HcclCMDType::HCCL_CMD_INVALID;
139 : HcclDataType dataType = HcclDataType::HCCL_DATA_TYPE_RESERVED;
140 : HcclReduceOp reduceOp = HcclReduceOp::HCCL_REDUCE_RESERVED;
141 : uint32_t root = 0;
142 : uint64_t count = 0;
143 : uint64_t index = 0;
144 : bool isValid = false;
145 : };
146 :
147 : struct OpInconsistentInfo {
148 : InconsistentType inconsistentType;
149 : std::string localInfo;
150 : std::string remoteInfo;
151 : OpInfoDesc opInfoDesc;
152 0 : OpInconsistentInfo(
153 : InconsistentType inconsistentType, const std::string& localInfo, const std::string& remoteInfo,
154 : const OpInfoDesc& opInfoDesc)
155 0 : : inconsistentType(inconsistentType),
156 0 : localInfo(localInfo),
157 0 : remoteInfo(remoteInfo),
158 0 : opInfoDesc(opInfoDesc)
159 0 : {}
160 : };
161 :
162 : struct OpInfoTagQueue {
163 : OpInfoDesc opInfoList[OPINFO_SEND_NUM_BY_TAG];
164 : char identifier[ROOTINFO_INDENTIFIER_MAX_LENGTH] = {};
165 : u32 opInfoNum = 0;
166 : };
167 :
168 : struct OpInfoTagQueueFrame {
169 : OpInfoTagQueue opInfoTagQueue[OPINFO_TAG_QUEUE_NUM];
170 : };
171 :
172 : struct HeartBeatFrame {
173 : UIDType src;
174 : UIDType dst;
175 : UIDType crimer;
176 : UIDType informer;
177 : HeartBeatStatus status = HeartBeatStatus::HEARTBEAT_OK;
178 : HcclUs TOARelative; // time of arrival (Relative)
179 : HcclSystemTime TOASystem; // time of arrival (System)
180 2 : HeartBeatFrame() {}
181 5 : HeartBeatFrame(
182 : UIDType& crimer, UIDType& informer, HeartBeatStatus status, HcclUs TOARelativeIn, HcclSystemTime TOASystemIn)
183 10 : : crimer(crimer),
184 5 : informer(informer),
185 5 : status(status),
186 5 : TOARelative(TOARelativeIn),
187 5 : TOASystem(TOASystemIn)
188 5 : {}
189 7 : HeartBeatFrame(UIDType& src, UIDType& dst, UIDType& crimer, UIDType& informer, HeartBeatStatus status)
190 7 : : src(src),
191 7 : dst(dst),
192 7 : crimer(crimer),
193 7 : informer(informer),
194 7 : status(status)
195 7 : {}
196 : };
197 :
198 : struct HeartBeatFrameWithOpCheck {
199 : UIDType src;
200 : UIDType dst;
201 : UIDType crimer;
202 : UIDType informer;
203 : HeartBeatStatus status = HeartBeatStatus::HEARTBEAT_OK;
204 : HcclUs TOARelative; // time of arrival (Relative)
205 : HcclSystemTime TOASystem; // time of arrival (System)
206 : OpInfoTagQueueFrame opInfoTagQueueFrame;
207 2 : HeartBeatFrameWithOpCheck() {}
208 : HeartBeatFrameWithOpCheck(
209 : UIDType& crimer, UIDType& informer, HeartBeatStatus status, HcclUs TOARelativeIn, HcclSystemTime TOASystemIn)
210 : : crimer(crimer),
211 : informer(informer),
212 : status(status),
213 : TOARelative(TOARelativeIn),
214 : TOASystem(TOASystemIn)
215 : {}
216 0 : HeartBeatFrameWithOpCheck(UIDType& src, UIDType& dst, UIDType& crimer, UIDType& informer, HeartBeatStatus status)
217 0 : : src(src),
218 0 : dst(dst),
219 0 : crimer(crimer),
220 0 : informer(informer),
221 0 : status(status)
222 0 : {}
223 : };
224 :
225 : struct ConnInfo {
226 : std::shared_ptr<HcclSocket> socket = nullptr;
227 : std::queue<HeartBeatFrame> sendBuffer;
228 : std::queue<HeartBeatFrameWithOpCheck> sendBufferWithOpCheck;
229 : u32 restSize = 0;
230 : RingBuffer recvBuffer;
231 : u32 lostNum = 0;
232 : u32 lostReportCnt = 0;
233 : bool newConn = false;
234 : std::vector<SocketWlistInfo> wlistInfosVec;
235 18 : ConnInfo() {}
236 8 : ConnInfo(bool newConn, std::shared_ptr<HcclSocket>& socket) : socket(socket), newConn(newConn) {}
237 : };
238 :
239 : struct LinkInfo {
240 : std::string identifier;
241 : RankId localRank;
242 : std::string localServerId;
243 : s32 localDevicePhyId;
244 : RankId remoteRank;
245 : std::string remoteServerId;
246 : s32 remoteDevicePhyId;
247 11 : LinkInfo() {}
248 0 : LinkInfo(
249 : std::string& identifier, RankId localRank, std::string& localServerId, s32 localDevicePhyId, RankId remoteRank,
250 : std::string& remoteServerId, s32 remoteDevicePhyId)
251 0 : : identifier(identifier),
252 0 : localRank(localRank),
253 0 : localServerId(localServerId),
254 0 : localDevicePhyId(localDevicePhyId),
255 0 : remoteRank(remoteRank),
256 0 : remoteServerId(remoteServerId),
257 0 : remoteDevicePhyId(remoteDevicePhyId)
258 0 : {}
259 : };
260 :
261 2 : using ErrCqeInfo = struct TagErrCqeInfo {
262 : CqeInfo cqeInfo;
263 : LinkInfo linkInfo;
264 : u32 qpn;
265 5 : TagErrCqeInfo() {}
266 4 : TagErrCqeInfo(CqeInfo& cqeInfo, LinkInfo& linkInfo, u32 qpn) : cqeInfo(cqeInfo), linkInfo(linkInfo), qpn(qpn) {}
267 7 : bool operator<(const TagErrCqeInfo& other) const { return qpn < other.qpn; }
268 : };
269 :
270 : class Heartbeat {
271 : public:
272 : static Heartbeat& GetInstance(s32 deviceLogicID);
273 : HcclResult RegisterToHeartBeat(
274 : u32 userRank, DevType devType, std::vector<RankInfo>& rankInfoList, const u32 port, const bool isNeedNic,
275 : const std::string& commIdentifier, bool useSuperPodMode, bool isUsedRdmaLevel0, bool retryEnable = false,
276 : bool backupEnable = false);
277 : HcclResult RegisterToHeartBeat(
278 : u32 userRank, DevType devType, std::vector<RankInfo>& rankInfoList, const u32 port, const bool isNeedNic,
279 : u32 peerRankId, const std::string& commIdentifier, const std::string& tag, bool useSuperPodMode,
280 : bool isUsedRdmaLevel0, bool retryEnable = false, bool backupEnable = false);
281 : HcclResult AddOpInfoToHeartBeat(const std::string& identifier, const OpInfoDesc& opInfo, const std::string& newTag);
282 : HcclResult DeleteOpInfoToHeartBeat(const std::string& identifier, const std::string& newTag);
283 : HcclResult UnRegisterRanks(const std::string& group = HCCL_WORLD_GROUP);
284 : // 集合通信，解开注册
285 : void UnRegisterToHeartBeat(DevType devType, const std::string& commIdentifier);
286 : // 非点对点通信，解开注册
287 : void UnRegisterToHeartBeat(DevType devType, const std::string& commIdentifier, const std::string& tag);
288 : HcclResult CheckErrorCqe(const std::string& identifier, HcclResult& result);
289 : HcclResult CheckOpInconsistentError(const std::string& identifier, HcclResult& result);
290 : HcclResult SetRankPortInfo(
291 : bool isUseRankPort, std::vector<u32>& nicRanksPorts, std::vector<u32>& vnicRanksPorts, bool devPortSwitchOn);
292 : std::vector<std::string> GetErrStatusVec(const std::string& group = HCCL_WORLD_GROUP);
293 : HcclResult GetQpnErr(const std::string& identifier, std::set<std::tuple<u32, u32, u32>>& qpErrSet);
294 : HcclResult BroadcastCqeErr(const std::string& identifier);
295 : HcclResult ClearAllCqeErr(const std::string& identifier);
296 : HcclResult ClearCqeErr(const std::string& identifier, u32 remoteRank, u32 qpn = 0);
297 : void SetOpretryErr();
298 : void GetIpQueue();
299 : bool IsPaused() const;
300 : bool IsResumed() const;
301 :
302 : private:
303 845 : Heartbeat() = default;
304 : ~Heartbeat();
305 : HcclResult Init(
306 : const RankInfo& locRank, const bool useSuperPodMode, const bool isNeedNic, const u32 port,
307 : const std::string& group = HCCL_WORLD_GROUP);
308 : HcclResult DeInit();
309 : HcclResult RegisterRanks(
310 : DevType devType, const RankInfo& locRank, std::vector<RankInfo>& rankInfos, const u32 port,
311 : const bool isNeedNic, const std::string& group = HCCL_WORLD_GROUP, bool useSuperPodMode = false,
312 : bool isUsedRdma = false);
313 : std::string GetConnTag(HcclSocketRole role, UIDType& rem);
314 : HcclResult GetConnInfo(
315 : RankInfo& remRank, bool useSuperPodMode, HcclSocketRole role, HcclSocketType type,
316 : std::map<UIDType, ConnInfo>& needConnectRank);
317 : template <typename T>
318 : HcclResult GetSamePlaneConnInfo(
319 : HcclSocketType type, std::vector<std::pair<T, u32>>& connVec, T& locId, std::vector<RankInfo>& rankInfos,
320 : std::map<UIDType, ConnInfo>& needConnectRank, bool useSuperPodMode, u32 worldRank);
321 : HcclResult GetConnectRank(
322 : const RankInfo& locRank, std::vector<RankInfo>& rankInfos, std::map<UIDType, ConnInfo>& needConnectRank,
323 : bool useSuperPodMode, bool isUsedRdma = false);
324 : UIDType GetUId(const RankInfo& rankInfo) const;
325 : std::string FormatUId(const UIDType& uid) const;
326 : HcclResult SendFrame(UIDType& dst, UIDType& crimer, UIDType& informer, HeartBeatStatus status);
327 : HcclResult SendFrameWithOpCheck(
328 : UIDType& dst, UIDType& crimer, UIDType& informer, HeartBeatStatus status,
329 : const OpInfoTagQueueFrame& opInfoTagQueueFrame);
330 : HcclResult RecvFrame(UIDType& src);
331 : HcclResult RecvFrameWithOpCheck(UIDType& src);
332 : HcclResult ParseFrame(HeartBeatFrame& bf, UIDType& src);
333 : HcclResult ParseFrameWithOpCheck(HeartBeatFrameWithOpCheck& bf, UIDType& src);
334 : void SetStatus(UIDType& crimer, UIDType& informer, HeartBeatStatus status, bool needBroadcast = true);
335 : void HeartbeatStatusMonitor();
336 : void ProcessExceptionEvent();
337 : void ProcessCqeErrInfo();
338 : void DelErrorSocket();
339 : bool IsKeyEvent(HeartBeatFrame& event, HcclUs curTime, const std::string& group = HCCL_WORLD_GROUP);
340 : void MakeErrMsg(std::queue<HeartBeatFrame>& keyEvents, std::vector<std::string>& errStatusVec);
341 : std::vector<std::string> PrintEvents(std::map<HeartBeatStatus, std::queue<HeartBeatFrame>>& keyEvents);
342 : void StuckDetection(uint64_t& cnt, CounterStat& counterStat);
343 : void InitStuckDetection(CounterStat& counterStat);
344 : void PrintAndBroadCastErrorCqe(const ErrCqeInfo& info);
345 : void SaveQpnForOpRetry(const ErrCqeInfo& info);
346 : void OpRetryCQEHandle(const HcclNetDevCtx netDevCtx);
347 : bool GetRetryEnable(const ErrCqeInfo& info);
348 : HcclResult ClearRetryEnableMapItem(const std::string& identifier);
349 : void ProcessCqeErrInfoByNetDevCtx(const HcclIpAddress& nicIp);
350 : bool IsEnableBackupLink();
351 : void RegisterRetryInfo(const std::string& commIdentifier, bool retryEnable, bool backupEnable);
352 : HcclResult InitNic(
353 : const NicType nicType, const s32 devicePhyId, const s32 deviceLogicId, const hccl::HcclIpAddress ip,
354 : const u32 port, const bool isBackUp = false);
355 : HcclResult InitDeviceNic(const RankInfo& locRank, bool isNeedNic, u32 port);
356 : HcclResult InitHostNic(const RankInfo& locRank, bool isNeedNic, u32 port);
357 : u32 GetPort(HcclSocketType type, u32 remoteUserRank, u32 remoteDeviceId);
358 : u32 GetHostPort(s32 devicePhyId);
359 : HcclResult PrepareConnect(ConnInfo& info);
360 : void CreateLinkWithRemote(std::string group, UIDType rem, ConnInfo needConnectRank);
361 : void CreateHBLinksAsync();
362 : void AddOpInfo(const std::string& identifier, const OpInfoDesc& opInfo, const std::string& paramTag);
363 : void GetOneOpInfo(std::string& tag, OpInfoDesc& opInfo);
364 : void GetSendOpInfoList(OpInfoTagQueueFrame& opInfoTagQueueFrame);
365 : void SaveOpInfo(const OpInfoTagQueueFrame& opInfoTagQueueFrame, UIDType& src);
366 : HcclResult CheckIsSameOp(const OpInfoDesc& localOpInfo, const OpInfoDesc& remoteOpInfo, InconsistentType& status);
367 : void CheckRecvOpInfoList();
368 : void RegisterSROpIdentifier(const std::string& identifier, const std::string& paramTag);
369 : void AddInconsistentOpRecord(
370 : const std::string& identifier, const OpInfoDesc& localOpInfo, InconsistentType status,
371 : const std::string& localInfo, const std::string& remoteInfo);
372 : void CheckSnapshotStatus();
373 : struct Status {
374 : HeartBeatStatus status = HeartBeatStatus::HEARTBEAT_OK;
375 : UIDType informer;
376 : bool needBroadcast = false;
377 63 : Status() {}
378 : };
379 :
380 : enum class HBLinkStatus { HEARTBEAT_LINK_NOT_START, HEARTBEAT_LINK_BUILDING, HEARTBEAT_LINK_COMPLETED };
381 :
382 : HcclIpAddress vnicIp_;
383 : HcclIpAddress nicIp_;
384 : HcclIpAddress backupNicIp_;
385 : u32 devicePhyId_;
386 : u32 deviceBackUpPhyId_;
387 : u32 superDeviceId_;
388 : NICDeployment nicDeploy_;
389 : UIDType uid_;
390 : bool initialized_ = false;
391 : u32 lostThreshold_ = 0;
392 : bool isDeInit_ = false;
393 : bool startSendRecvTask_ = false;
394 : std::map<std::string, std::queue<std::pair<UIDType, ConnInfo>>> hbLinkConnInfo_{};
395 : std::mutex hbLinkConnInfoMtx_;
396 : std::map<std::string, std::map<UIDType, u8>> groupMap_;
397 : ReferenceMap<UIDType, ConnInfo> rankId2SocketMap_;
398 : ReferenceMap<UIDType, Status> rankId2StatusMap_;
399 : std::map<UIDType, HBLinkStatus> rankId2LinkStatusMap_;
400 : std::map<UIDType, std::unique_ptr<std::thread>> linkThreadMap_{};
401 : std::atomic<bool> linkThreadRunning_{false};
402 : std::atomic<u32> linkThreadCount_{0};
403 : std::unique_ptr<std::thread> sendRecvThread_;
404 : std::queue<HeartBeatFrame> errStatusQueue_;
405 : std::queue<UIDType> errRankQueue_;
406 : std::mutex ProcessLock_;
407 : u32 deviceLogicId_;
408 : u32 deviceBackupLogicId_;
409 : std::map<std::string, std::set<ErrCqeInfo>> remoteIpMap;
410 : std::set<u32> qpnDissociativeSet;
411 : std::mutex remoteIpMutex_;
412 : bool isUseRankPort_{false};
413 : bool devPortSwitchOn_{false};
414 : std::vector<u32> nicRanksPorts_;
415 : std::vector<u32> vnicRanksPorts_;
416 : std::vector<UIDType> errorSocket_;
417 : std::map<std::string, std::map<u32, std::set<ErrCqeInfo>>> rankMapForRetryAgent;
418 : std::mutex qpnMapMutexForRetry_;
419 : std::map<std::string, bool> retryEnableTable_;
420 : std::mutex retryEnableMutex_;
421 : std::set<std::string> backupEnableTable_;
422 : std::mutex backupEnableMutex_;
423 : std::mutex ctxMapMutex_;
424 : std::map<HcclIpAddress, HcclNetDevCtx> netDevCtxMap_;
425 : std::map<HcclIpAddress, std::shared_ptr<HcclSocket>> listenSocketMap_;
426 : s32 stuckDetectTime_;
427 : std::mutex opInfoQueueMutex_;
428 : std::deque<std::pair<std::string, OpInfoDesc>> opInfoQueue_;
429 : std::deque<std::pair<std::string, OpInfoDesc>> opInfoQueueForSend_;
430 : std::unordered_map<std::string, u64> opInfoIndexMap_;
431 : std::mutex opInfoMapMutex_;
432 : std::unordered_map<std::string, std::map<u64, OpInfoDesc>> opInfoMap_;
433 : std::list<std::tuple<OpInfoDesc, std::string, UIDType>> recvOpInfoList_;
434 : std::mutex inconsistentOpMutex_;
435 : std::map<std::string, OpInconsistentInfo> inconsistentOpMap_;
436 : std::mutex srTagMutex_;
437 : std::map<std::string, std::string> srTagMap_; // SR算子tag->identifier映射
438 : bool isPaused_{false}; // heartbeat need to be paused when snapshot
439 : };
440 : } // namespace hccl
441 :
442 : #endif // HCCL_HEARTBEAT_H
|
