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 : #ifndef CLUSTER_MONITOR_H
11 : #define CLUSTER_MONITOR_H
12 : #include <thread>
13 : #include <map>
14 : #include <deque>
15 : #include <mutex>
16 : #include "hcclCommDfx.h"
17 : #include "ring_buffer.h"
18 : #include "coll_comm.h"
19 : #include "reference_map.h"
20 : #include "log.h"
21 : #include "hccl/hccl_types.h"
22 : #include "hccl_common.h"
23 : #include "hccl_comm_socket_c_adpt.h"
24 : #include "hccl_communicator.h"
25 : #include "../../common/loggers/comm_addr_logger.h"
26 :
27 :
28 : namespace hcomm {
29 : using ClusterUIDType = struct HcclClusterMonitorUID {
30 : char id[2048] = {0}; // netInstanceId + localId 最大不超过2048字节
31 211 : bool operator == (const HcclClusterMonitorUID &that) const
32 : {
33 844 : return std::string(this->id) == std::string(that.id);
34 : }
35 6 : bool operator != (const HcclClusterMonitorUID &that) const
36 : {
37 24 : return std::string(this->id) != std::string(that.id);
38 : }
39 62 : bool operator < (const HcclClusterMonitorUID &that) const
40 : {
41 248 : return std::string(this->id) < std::string(that.id);
42 : }
43 : };
44 : }
45 :
46 : namespace std {
47 : template <> class hash<hcomm::HcclClusterMonitorUID> {
48 : public:
49 241 : size_t operator () (const hcomm::HcclClusterMonitorUID &uid) const
50 : {
51 482 : return hash<string>()(string(uid.id));
52 : }
53 : };
54 : }
55 :
56 : namespace hcomm {
57 : enum class ClusterMonitorStatus {
58 : CLUSTER_MONITOR_OK,
59 : CLUSTER_MONITOR_LOST,
60 : CLUSTER_MONITOR_NOTIFY,
61 : CLUSTER_MONITOR_CQE_ERR,
62 : CLUSTER_MONITOR_OPRETRY_NOT_SUPPORT,
63 : CLUSTER_MONITOR_STUCK,
64 : CLUSTER_MONITOR_INCONSISTENT
65 : };
66 :
67 : struct ErrorCqeInfo {
68 : u32 cqeLocalId = 0;
69 : u32 cqeRemoteLocalId = 0;
70 : uint16_t cqeStatus = 0;
71 : std::string cqeLocalEid = "";
72 : std::string cqeRemoteEid = "";
73 : std::string cqeRemoteInsId = "";
74 : std::string cqeLocalInsId = "";
75 : };
76 : const std::map<ClusterMonitorStatus, std::string> CLUSTER_MONITOR_STATUS_STR_MAP{
77 : {ClusterMonitorStatus::CLUSTER_MONITOR_OK, "OK"},
78 : {ClusterMonitorStatus::CLUSTER_MONITOR_LOST, "LOST"},
79 : {ClusterMonitorStatus::CLUSTER_MONITOR_NOTIFY, "NOTIFY"},
80 : {ClusterMonitorStatus::CLUSTER_MONITOR_CQE_ERR, "CQE ERROR"}
81 : };
82 :
83 1 : inline std::string GetClusterMonitorStatusStr(ClusterMonitorStatus status)
84 : {
85 1 : auto iter = CLUSTER_MONITOR_STATUS_STR_MAP.find(status);
86 1 : if (iter == CLUSTER_MONITOR_STATUS_STR_MAP.end()) {
87 0 : return "Unknown";
88 : } else {
89 1 : return iter->second;
90 : }
91 : }
92 :
93 : struct ClusterMonitorFrame {
94 : ClusterUIDType src{}; // 心跳建链的本端
95 : ClusterUIDType dst{}; // 心跳建链的远端
96 : ClusterUIDType crimer{}; // 异常的节点
97 : ClusterUIDType informer{}; // 把异常传输给自己的节点
98 : ClusterMonitorStatus status = ClusterMonitorStatus::CLUSTER_MONITOR_OK;
99 : HcclUs TOARelative{}; // time of arrival (Relative)
100 : HcclSystemTime TOASystem{}; // time of arrival (System)
101 : char reserved[256] = {0}; // 预留256个字段，后续扩展可存储其他信息
102 2 : ClusterMonitorFrame() {}
103 6 : ClusterMonitorFrame(ClusterUIDType &crimer, ClusterUIDType &informer, ClusterMonitorStatus status, HcclUs TOARelativeIn,
104 : HcclSystemTime TOASystemIn)
105 6 : : crimer(crimer), informer(informer), status(status), TOARelative(TOARelativeIn),
106 6 : TOASystem(TOASystemIn)
107 6 : {}
108 12 : ClusterMonitorFrame(ClusterUIDType &src, ClusterUIDType &dst, ClusterUIDType &crimer, ClusterUIDType &informer, ClusterMonitorStatus status)
109 12 : : src(src), dst(dst), crimer(crimer), informer(informer), status(status)
110 12 : {}
111 : };
112 :
113 : struct ClusterMonitorSocketCtx { // 原ConnInfo
114 : SocketDesc socketDesc; // 与对端连接的描述符
115 : SocketHandle socketHandler; // 引用头文件定义
116 : std::queue<ClusterMonitorFrame> sendBuffer; // 用来发送的帧队列
117 : u32 restSize = 0; // 剩余待发送的帧长度
118 : hccl::RingBuffer recvBuffer; // 用来接收的环形帧队列
119 : u32 lostNum = 0; // 丢失的心跳个数
120 : bool newConn = false; // 是否是新增的连接
121 5 : ClusterMonitorSocketCtx() {}
122 1 : ClusterMonitorSocketCtx(SocketDesc &socketDesc, bool newConn)
123 1 : : socketDesc(socketDesc), socketHandler(nullptr), newConn(newConn)
124 1 : {}
125 :
126 6 : void PrintSocketDesc(std::string tag) const
127 : {
128 6 : std::string localAddr = hcomm::logger::CommAddrLogger::ToString(socketDesc.localEndpoint.commAddr);
129 6 : std::string remoteAddr = hcomm::logger::CommAddrLogger::ToString(socketDesc.remoteEndpoint.commAddr);
130 6 : HCCL_DEBUG("[%s] socketDesc: localEndpoint: {commAddr: %s, EndpointLocType: %d}, "
131 : "remoteEndpoint: {commAddr: %s, EndpointLocType: %d}, tag: %s, role: %d, listenPort: %u",
132 : tag.c_str(), localAddr.c_str(), socketDesc.localEndpoint.loc.locType, remoteAddr.c_str(),
133 : socketDesc.remoteEndpoint.loc.locType, socketDesc.tag, socketDesc.role, socketDesc.listenPort);
134 6 : }
135 : };
136 :
137 : struct UIDContext {
138 : ClusterUIDType uid;
139 : uint32_t netLayer{0};
140 : uint32_t rankId{0};
141 : uint32_t localId{0}; // 用来netLayer=0的时候排序使用
142 : std::string netInstId{}; // 用来netLayer>1的时候排序使用
143 : UIDContext() {}
144 21 : UIDContext(ClusterUIDType &uid, uint32_t netLayer, uint32_t rankId, uint32_t localId, std::string netInstId)
145 21 : : uid(uid), netLayer(netLayer), rankId(rankId), localId(localId), netInstId(netInstId)
146 21 : {}
147 : };
148 :
149 : struct ClusterUIDCxt {
150 : std::string netInstId;
151 : uint32_t localId;
152 : ClusterUIDCxt() {}
153 20 : ClusterUIDCxt(std::string &netInstId, uint32_t localId)
154 20 : : netInstId(netInstId), localId(localId)
155 20 : {}
156 : };
157 :
158 : class ClusterMonitor {
159 : public:
160 : HcclResult RegisterToClusterMonitor(HcclComm comm);
161 : HcclResult UnRegisterToClusterMonitor(const hccl::CollComm* collComm);
162 : ClusterUIDType FormatUID(ClusterUIDCxt cxt) const;
163 : std::string GetUID(const ClusterUIDType &uid) const;
164 : std::string FormatConnTag(HcommSocketRole role, std::pair<ClusterUIDType, ClusterUIDType> uidPair) const;
165 : HcclResult InsertClusterMonitorCtx(HcclComm comm, UIDContext remoteCtx, std::map<ClusterUIDType, ClusterMonitorSocketCtx> &needConnectRank);
166 : HcclResult GetSocketDescFromRankInfo(HcclComm comm, uint32_t remoteRank, uint32_t netLayer, const ClusterUIDType &remoteUID, SocketDesc &socketDesc);
167 : HcclResult GetSamePlaneRank(HcclComm comm, std::vector<UIDContext> singlePlaneCtx, std::map<ClusterUIDType, ClusterMonitorSocketCtx> &needConnectRank);
168 : HcclResult GetConnectRank(HcclComm comm, std::map<ClusterUIDType, ClusterMonitorSocketCtx> &needConnectRank, std::map<uint32_t,
169 : std::vector<UIDContext>> uidCtxs, std::vector<uint32_t> &netLayersVector);
170 : void CreateHBLinksAsync();
171 : void SetStatus(ClusterUIDType &crimer, ClusterUIDType &informer, ClusterMonitorStatus status, bool needBroadcast = true);
172 : void MonitorThread();
173 : HcclResult RunMonitorThread();
174 : HcclResult SendFrame(ClusterUIDType &dst, ClusterUIDType &crimer, ClusterUIDType &informer, ClusterMonitorStatus status);
175 : void DelErrorSocket();
176 : void ProcessExceptionEvent();
177 : HcclResult RecvFrame(ClusterUIDType rem);
178 : HcclResult ParseFrame(ClusterMonitorFrame &cmFrame, ClusterUIDType &src);
179 : HcclResult DeInit();
180 : static ClusterMonitor& GetInstance(u32 deviceId);
181 : void GetCqeErrInfoFromTaskException(u32 remoteLocalId, uint16_t status, std::string localEid, std::string remoteEid, std::string remoteInsId);
182 : std::vector<std::string> GetErrStatusVecFromCluserMonitor();
183 : std::vector<std::string> PrintEvents(std::map<ClusterMonitorStatus, std::queue<ClusterMonitorFrame>> &keyEvents) const;
184 : void MakeErrMsg(std::queue<ClusterMonitorFrame> &keyEvents, std::vector<std::string> &errStatusVec) const;
185 3186 : ClusterMonitor() = default;
186 : ~ClusterMonitor();
187 :
188 : private:
189 : HcclResult GetRemEndpointDescs(HcclComm comm, std::map<uint32_t, std::vector<UIDContext>> &uidCtxs,
190 : std::vector<uint32_t> &netLayersVector);
191 : void GetRemEndpointDescsPerLayer(uint32_t netLayer, HcclComm comm, const Hccl::RankGraph *rankGraph,
192 : const hccl::CollComm* collComm, std::map<uint32_t, std::vector<UIDContext>> &uidCtxs, std::set<uint32_t> &rankIdsSet);
193 :
194 : HcclResult ProcessConnectRanks(const std::string &commId, std::map<ClusterUIDType, ClusterMonitorSocketCtx> &needConnectRank);
195 : void ClearClusterLinkContext(const std::string &commId, std::set<ClusterUIDType> &remInQueue);
196 : bool UnregisterCommIdFromMaps(const std::string &commId, const std::set<ClusterUIDType> &remInQueue);
197 : HcclResult CreateTransportHandle(ClusterMonitorSocketCtx &info) const;
198 : HcclResult OnConnectionEstablished(const std::string &commId, const ClusterUIDType &rem, ClusterMonitorSocketCtx &needConnectRank);
199 : HcclResult SendFrameFromBuffer(ClusterUIDType &dst, ClusterMonitorFrame &cmFrame);
200 :
201 : void CreateLinkWithRemotePonit(std::string commId, ClusterUIDType rem, ClusterMonitorSocketCtx needConnectRank);
202 :
203 : struct FrameStatus { // 专门用来给frame设置对应的状态
204 : ClusterMonitorStatus status = ClusterMonitorStatus::CLUSTER_MONITOR_OK;
205 : ClusterUIDType informer;
206 : bool needBroadcast = false;
207 7 : FrameStatus() {}
208 : };
209 :
210 : enum class MonitorLinkStatus {
211 : MONITOR_LINK_NOT_START,
212 : MONITOR_LINK_BUILDING,
213 : MONITOR_LINK_COMPLETED,
214 : };
215 :
216 : uint32_t myRankLocalId_;
217 : std::string myRankNetInstId_;
218 : ClusterUIDType myRankUID_;
219 : s32 deviceLogicId_{0};
220 :
221 : bool clusterMonitorThreadFlag_ = false;
222 : std::unique_ptr<std::thread> clusterMonitorThread_;
223 : // 防止重复初始化
224 : bool initialized_ = false;
225 : uint32_t lostThreshold_ = 0;
226 : bool isDeInit_ = false;
227 : std::atomic<bool> linkThreadRunning_{false};
228 :
229 : // 防止多线程同时初始化的线程锁
230 : std::mutex threadLock_;
231 : std::vector<ClusterUIDType> errorSocket_;
232 : std::queue<ClusterMonitorFrame> errStatusQueue_;
233 :
234 : // 通信域名称为key，ClusterUIDType表示1个节点, bool表示0或1，是否连接上，用来指示是否有过这个通信域以及对应通信域里待连接的节点是否以及连接上，原groupMap_
235 : std::map<std::string, std::map<ClusterUIDType, bool>> commIdMap_;
236 :
237 : // 通信域名称为key, 一个通信域有多个待连接心跳的connInfo，存储到该结构体，待monitor线程轮询拿到, 原hbLinkConnInfo_
238 : std::map<std::string, std::queue<std::pair<ClusterUIDType, ClusterMonitorSocketCtx>>> clusterLinkContext_{};
239 : std::mutex clusertMonitorLinkMtx_; // 用来锁住clusterLinkContext_，原clusterLinkContext_
240 :
241 : // 存储UID与监控连接状态的map, NOT_START/BUILDING/COMPILETED，原rankId2LinkStatusMap_
242 : std::map<ClusterUIDType, MonitorLinkStatus> monitorLinkStatusMap_;
243 :
244 : // 存储UID与连接上下文的计数map，由于多个通信域都有可能使用同一个context去连接远端，需要计数处理，解注册时计数--，原rankId2SocketMap_
245 : hccl::ReferenceMap<ClusterUIDType, ClusterMonitorSocketCtx> uid2SocketRefMap_;
246 :
247 : // 用来做帧的统计计数，设置对应帧的状态，原rankId2StatusMap_
248 : hccl::ReferenceMap<ClusterUIDType, FrameStatus> uid2FrameStatusMap_;
249 :
250 : // uid与thread的维护关系，不同的remote起不同的异步建链线程，原linkThreadMap_
251 : std::map<ClusterUIDType, std::unique_ptr<std::thread>> linkThreadMap_{};
252 :
253 : // UnRegister 摘下、延后到 DeInit(join 之后) 再 SocketDestroy 的句柄
254 : std::vector<SocketHandle> pendingDestroySockets_;
255 :
256 : // 保存错误的节点
257 : std::queue<ClusterUIDType> errRankQueue_;
258 :
259 : ErrorCqeInfo cqeErrInfo_;
260 : };
261 : } // namespace hcomm
262 : #endif // CLUSTER_MONITOR_H
|
