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