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_DETECT_CONNECT_ANOMALIES_H
12 : #define HCCL_DETECT_CONNECT_ANOMALIES_H
13 : #include <queue>
14 : #include <map>
15 : #include <unordered_set>
16 : #include <mutex>
17 : #include <thread>
18 : #include "hccl_socket.h"
19 : #include "hccl_socket_manager.h"
20 : #include "hccl_ip_address.h"
21 : #include "common.h"
22 :
23 : namespace hccl {
24 : // todo 本端和对端的都得保存，并打印
25 : constexpr size_t DEST_MAX_LEN = 128;
26 : constexpr u32 MAX_WHITE_LIST_ENTRY = 16;
27 : constexpr u32 ACCEPT_TIME_OF_USLEEP = 100000;
28 : constexpr u32 IRECV_TIME_OF_USLEEP = 500000;
29 : constexpr u32 CLIENT_TIME_OF_USLEEP = 1500000;
30 : struct DetectInfo {
31 : s32 localDeviceId = 0XFFFFFFFF;
32 : s32 remoteDeviceId = 0XFFFFFFFF;
33 :
34 : char localDeviceIp[DEST_MAX_LEN]{}; // 用来查重
35 : char remoteDeviceIp[DEST_MAX_LEN]{}; // 用来查重
36 :
37 : char localServerId[DEST_MAX_LEN]{};
38 : char remoteServerId[DEST_MAX_LEN]{};
39 : };
40 :
41 : struct SendInfo {
42 : bool isSendNic = false;
43 : bool isSendVnic = false;
44 : };
45 :
46 12 : struct ErrInfo {
47 : RankInfo localRankInfo;
48 : RankInfo remoteRankInfo;
49 : NicType nicType;
50 : s32 deviceLogicId;
51 : };
52 :
53 : // 统一处理 IP 插入逻辑
54 : template <typename ListType>
55 1 : HcclResult AddWlistEntry(
56 : const HcclIpAddress& ipAddr,
57 : const std::string& tag,
58 : ListType& whiteList,
59 : std::vector<SocketWlistInfo>& wlistVec)
60 : {
61 : // 查找是否已存在
62 1 : if (std::find(whiteList.begin(), whiteList.end(), ipAddr) != whiteList.end()) {
63 0 : return HCCL_SUCCESS;
64 : }
65 :
66 : // 构造白名单信息，按照最大限制16下发
67 1 : SocketWlistInfo wlistInfo = {};
68 1 : wlistInfo.connLimit = MAX_WHITE_LIST_ENTRY;
69 1 : wlistInfo.remoteIp.addr = ipAddr.GetBinaryAddress().addr;
70 1 : wlistInfo.remoteIp.addr6 = ipAddr.GetBinaryAddress().addr6;
71 1 : CHK_SAFETY_FUNC_RET(memcpy_s(&wlistInfo.tag[0], sizeof(wlistInfo.tag), tag.c_str(), tag.size() + 1));
72 :
73 : // 记录白名单信息
74 1 : HCCL_INFO("[AddWlistEntry] ip[%s] tag[%s]", ipAddr.GetReadableIP(), wlistInfo.tag);
75 1 : wlistVec.push_back(wlistInfo);
76 1 : whiteList.insert(ipAddr);
77 1 : return HCCL_SUCCESS;
78 : }
79 :
80 : class DetectConnectionAnomalies {
81 : public:
82 : static DetectConnectionAnomalies &GetInstance(s32 deviceLogicID);
83 : void Init(std::vector<RankInfo> &rankInfos, bool isNeedNic);
84 : void AddIpQueue(RankInfo &localRankInfo, RankInfo &remoteRankInfo, NicType nicType, s32 deviceLogicId);
85 : HcclResult Detect();
86 : void Deinit();
87 : private:
88 : void DetectMonitor();
89 : HcclResult GetIpQueue();
90 : HcclResult CreateServers(struct ErrInfo errInfo);
91 : std::string GetTag(HcclIpAddress &Ip, int i = 0);
92 : HcclResult AddWhiteList(std::shared_ptr<HcclSocket> socket, NicType nicType, std::string& tag);
93 : HcclResult DelWhiteList(HcclIpAddress &localIpAddr,
94 : std::vector<struct SocketWlistInfo> whiteListInfos, std::shared_ptr<HcclSocket> socket);
95 : HcclResult GetStatus(struct ErrInfo errInfo, std::shared_ptr<HcclSocket> &clientSocket);
96 : HcclResult Connect(struct ErrInfo errInfo, std::shared_ptr<HcclSocket> &clientSocket);
97 : HcclResult CreateDetectVnicLinks(struct ErrInfo errInfo);
98 : HcclResult CreateDetectNicLinks(struct ErrInfo errInfo);
99 : HcclResult CreateClients(struct ErrInfo errInfo, std::vector<std::unique_ptr<std::thread>> &linkClientThreads);
100 : HcclResult ConstructErrorInfo(std::shared_ptr<HcclSocket> &clientSocket, RankInfo &localRankInfo, RankInfo &remoteRankInfo);
101 : HcclResult CreateClient(struct ErrInfo errInfo);
102 : HcclResult processWhiteList(const HcclIpAddress &ipAddr, HcclIpAddress &localIpAddr, std::shared_ptr<HcclSocket> socket, NicType nicType);
103 : HcclResult WaitForDectect();
104 : HcclResult ProcessDetectionResults();
105 : std::string FormatDetectMessage(const std::string &localServerId, s32 localDeviceId, const DetectInfo &detectInfo);
106 : std::string BuildGroupedDetectMessage();
107 : void ThreadDestroy();
108 845 : ~DetectConnectionAnomalies() = default;
109 845 : DetectConnectionAnomalies() = default;
110 : int broadCastTime = 10; // 故障广播时间
111 : std::set<HcclIpAddress> uniqueIps_;
112 : bool threadExit_ = true;
113 : bool isNeedNic_ = false;
114 : bool isInitThread_ = false;
115 : std::mutex ipNictypeQueueMutex_;
116 : std::mutex ipConstuctMutex_;
117 : std::mutex whiteListMutex_; //删除白名单需要加锁
118 : std::mutex clientThreadMutex_; //删除clients需要加锁
119 : std::mutex clientResourcesMutex_; // 保护clientNicCtxs_和clientSockets_的并发访问
120 : std::mutex printDetectInfoMutex_; // 打印锁
121 : std::mutex detectThreadMutex_;
122 : std::set<HcclIpAddress> whiteVnicSet_; // 保存vnic的白名单 whiteVnicSet_
123 : std::set<HcclIpAddress> whiteNicSet_; // 保存nic的白名单
124 : std::shared_ptr<HcclSocket> vnicSocket_ = nullptr;
125 : std::shared_ptr<HcclSocket> nicSocket_ = nullptr;
126 : std::vector<std::shared_ptr<HcclSocket>> clientSockets_; //保存clien端的socket
127 : std::map<HcclIpAddress, HcclIpAddress> ipMap_;
128 : std::map<HcclIpAddress, std::shared_ptr<HcclSocket>> socketMap_;
129 : std::map<HcclIpAddress, HcclNetDevCtx> nicNetDevCtxMap_;
130 : std::vector<std::shared_ptr<HcclSocket>> listenNicVec_;
131 : std::vector<std::shared_ptr<HcclSocket>> listenVnicVec_;
132 : std::queue<ErrInfo> ipNictypeQueue_;
133 : HcclNetDevCtx nicCtx_;
134 : HcclNetDevCtx vnicCtx_;
135 : std::vector<HcclNetDevCtx> clientNicCtxs_;
136 : std::unique_ptr<std::thread> getIpNictypeQueue_ = nullptr;
137 : std::unique_ptr<std::thread> detectVnicThread_ = nullptr;
138 : std::unique_ptr<std::thread> detectNicThread_ = nullptr;
139 : std::vector<struct SocketWlistInfo> vnicWhiteListInfosVec_; // 保存vnic白名单单信息，方便删除的时候使用
140 : std::vector<struct SocketWlistInfo> nicWhiteListInfosVec_; // 保存nic白名单单信息，方便删除的时候使用
141 :
142 : // 发送完成后添加，发送前查重
143 : std::unordered_map<std::string, SendInfo> sendErrorInfoMap_;
144 : // 接收到添加，接收前查重
145 : std::unordered_map<std::string, DetectInfo> recvErrorInfoMap_;
146 : std::mutex readRecvErrtInfo_;
147 :
148 : bool isCreateLink_ = false;
149 : bool isCreateNicLink_ = false;
150 : std::atomic<bool> isPrint_{false};
151 : std::atomic<int> errorCount_{0};
152 : std::vector<std::unique_ptr<std::thread>> linkClientThreads_; // 保存client拉起的线程
153 : Referenced initRef_;
154 : std::chrono::steady_clock::time_point startTime;
155 : std::mutex time_mutex;
156 : std::mutex print_mutex;
157 : };
158 : }
159 :
160 : #endif // HCCL_DETECT_CONNECT_ANOMALIES_H
|
