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 <chrono>
12 : #include <thread>
13 : #include "hccl_common.h"
14 : #include "hccl_socket.h"
15 : #include "env_config.h"
16 : #include "externalinput_pub.h"
17 : #include "detect_connect_anomalies.h"
18 :
19 : namespace hccl {
20 1079 : DetectConnectionAnomalies &DetectConnectionAnomalies::GetInstance(s32 deviceLogicID)
21 : {
22 1924 : static DetectConnectionAnomalies dca[MAX_MODULE_DEVICE_NUM];
23 1079 : if (static_cast<u32>(deviceLogicID) >= MAX_MODULE_DEVICE_NUM) {
24 273 : HCCL_WARNING("[DetectConnectionAnomalies][GetInstance]deviceLogicID[%d] is invalid", deviceLogicID);
25 273 : return dca[0];
26 : }
27 806 : return dca[deviceLogicID];
28 : }
29 :
30 : // 创建单例，保存所有RankInfoList中的Ip地址
31 196 : void DetectConnectionAnomalies::Init(std::vector<RankInfo> &rankInfos, bool isNeedNic)
32 : {
33 196 : if (isNeedNic) {
34 0 : isNeedNic_ = isNeedNic;
35 : }
36 : // 直接用set保存，省掉查重
37 196 : int ref = initRef_.Ref();
38 196 : HCCL_INFO("DetectConnectionAnomalies[Init] initRef[%d]", ref);
39 1131 : for (auto &rankInfo : rankInfos) {
40 938 : if (!rankInfo.nicIp[0].IsInvalid()) {
41 938 : uniqueIps_.insert(rankInfo.nicIp[0]);
42 : }
43 :
44 935 : if (!rankInfo.deviceVnicIp.IsInvalid()) {
45 573 : uniqueIps_.insert(rankInfo.deviceVnicIp);
46 : }
47 : }
48 193 : return;
49 : }
50 :
51 : // 添加ipQueue
52 38 : void DetectConnectionAnomalies::AddIpQueue(RankInfo &localRankInfo, RankInfo &remoteRankInfo, NicType nicType,
53 : s32 deviceLogicId)
54 : {
55 : // 检查是否需要进行连接异常检测
56 38 : if (GetExternalInputDfsConnectionFaultDetectionTime() == 0 || !threadExit_) {
57 38 : HCCL_RUN_INFO("[Add][IpQueue]GetExternalInputDfsConnectionFaultDetectionTime is 0, no need to detect");
58 265 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>({"reason"}), \
59 : std::vector<std::string>({GET_SOCKET_TIMEOUT_REASON_CLOSE_DETECT}));
60 38 : return;
61 : }
62 :
63 : // 检查设备类型是否支持
64 0 : if (localRankInfo.deviceType != DevType::DEV_TYPE_910_93 && localRankInfo.deviceType != DevType::DEV_TYPE_910B) {
65 0 : HCCL_WARNING("[AddIpQueue] not support deviceType[%d]", localRankInfo.deviceType);
66 0 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>({"reason"}), \
67 : std::vector<std::string>({GET_SOCKET_TIMEOUT_REASON_CLOSE_DETECT}));
68 0 : return;
69 : }
70 :
71 : // 检查是否需要进行连接异常检测
72 0 : HcclIpAddress localIp = (nicType == NicType::VNIC_TYPE) ? localRankInfo.deviceVnicIp : localRankInfo.nicIp[0];
73 0 : HcclIpAddress remoteIp = (nicType == NicType::DEVICE_NIC_TYPE || nicType == NicType::HOST_NIC_TYPE) ?
74 0 : remoteRankInfo.nicIp[0] : remoteRankInfo.deviceVnicIp;
75 0 : if (localIp.IsInvalid() || remoteIp.IsInvalid()) {
76 0 : return;
77 : }
78 :
79 : // 多线程访问ipQueue需要加锁
80 0 : Detect();
81 0 : std::unique_lock<std::mutex> lock(ipNictypeQueueMutex_);
82 0 : ErrInfo errInfo;
83 0 : auto ip = ipMap_.find(remoteIp);
84 0 : if (ip == ipMap_.end()) {
85 0 : ipMap_.insert(std::make_pair(remoteIp, localIp));
86 0 : HCCL_INFO("[Add][IpQueue]localIp[%s], remoteIp[%s], nicType[%d], deviceLogicId[%d]",
87 : localIp.GetReadableAddress(), remoteIp.GetReadableAddress(), nicType, deviceLogicId);
88 0 : errInfo.localRankInfo = localRankInfo;
89 0 : errInfo.remoteRankInfo = remoteRankInfo;
90 0 : errInfo.nicType = nicType;
91 0 : errInfo.deviceLogicId = deviceLogicId;
92 0 : ipNictypeQueue_.push(errInfo); // 记录报错卡信息
93 : }
94 0 : lock.unlock();
95 0 : WaitForDectect();
96 0 : HCCL_INFO("[Add][IpQueue]ipNictypeQueue size[%d]", ipNictypeQueue_.size());
97 0 : return;
98 38 : }
99 0 : HcclResult DetectConnectionAnomalies::WaitForDectect()
100 : {
101 : // 计算等待时间
102 0 : auto waitTime = std::chrono::seconds(GetExternalInputDfsConnectionFaultDetectionTime()) +
103 0 : std::chrono::seconds(broadCastTime);
104 0 : std::unique_lock<std::mutex> timelock(time_mutex);
105 0 : startTime = std::chrono::steady_clock::now(); // 刷新时间
106 0 : std::chrono::steady_clock::time_point localStartTime = startTime;
107 0 : timelock.unlock();
108 :
109 0 : while (threadExit_ && (std::chrono::steady_clock::now() - localStartTime) <= waitTime) {
110 0 : std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 每次休眠100毫秒
111 0 : timelock.lock();
112 0 : localStartTime = startTime;
113 0 : timelock.unlock();
114 : }
115 : // 通过锁只进入一次
116 0 : std::lock_guard<std::mutex> printlock(print_mutex);
117 0 : if (!isPrint_) {
118 0 : ProcessDetectionResults();
119 : }
120 0 : isPrint_ = true;
121 0 : return HCCL_SUCCESS;
122 0 : }
123 :
124 4 : std::string DetectConnectionAnomalies::BuildGroupedDetectMessage()
125 : {
126 4 : std::ostringstream result;
127 : // key: <srcServer, srcDevice> value: <dstServer, dstDeviceList>
128 4 : std::map<std::pair<std::string, s32>, std::map<std::string, std::vector<s32>>> summary;
129 : // 聚合deviceID
130 10 : for (const auto &item : recvErrorInfoMap_) {
131 6 : const DetectInfo &info = item.second;
132 18 : summary[{info.localServerId, info.localDeviceId}][info.remoteServerId].push_back(info.remoteDeviceId);
133 : }
134 4 : bool firstMsg = true;
135 8 : for (auto &srcGroup : summary) {
136 4 : const std::string &srcServer = srcGroup.first.first;
137 4 : s32 srcDevice = srcGroup.first.second;
138 8 : for (auto &dstGroup : srcGroup.second) {
139 4 : auto &devices = dstGroup.second;
140 4 : std::sort(devices.begin(), devices.end());
141 4 : devices.erase(std::unique(devices.begin(), devices.end()), devices.end());
142 4 : std::ostringstream deviceList;
143 4 : deviceList << "[";
144 9 : for (size_t i = 0; i < devices.size(); ++i) {
145 5 : if (i != 0) {
146 1 : deviceList << ",";
147 : }
148 5 : deviceList << devices[i];
149 : }
150 4 : deviceList << "]";
151 4 : if (!firstMsg) {
152 1 : result << "\n";
153 : }
154 4 : firstMsg = false;
155 : result << "This node (server " << srcServer
156 4 : << ", device ID " << srcDevice
157 : << ") detects that srcRank (server " << srcServer
158 4 : << ", device ID " << srcDevice
159 4 : << ") fails to connect to dstRank (server " << dstGroup.first
160 4 : << ", device ID " << deviceList.str()
161 4 : << "). Continue to analyze the fault based on the logs of srcRank and dstRank.";
162 4 : }
163 : }
164 8 : return result.str();
165 4 : }
166 :
167 0 : HcclResult DetectConnectionAnomalies::ProcessDetectionResults()
168 : {
169 0 : std::string errMsg;
170 0 : HCCL_ERROR("-------------------CONNECT TIMEOUT DETECT RESULT-----------------------");
171 0 : if (!recvErrorInfoMap_.empty()) {
172 0 : errMsg = BuildGroupedDetectMessage();
173 0 : HCCL_ERROR("%s", errMsg.c_str());
174 0 : HCCL_ERROR("%s", GET_SOCKET_TIMEOUT_REASON_WITH_EVENT.c_str());
175 0 : errMsg += "\n" + GET_SOCKET_TIMEOUT_REASON_WITH_EVENT;
176 : } else {
177 : errMsg ="This node detects no exception event. The possible cause is that the behaviors of different ranks are inconsistent. "
178 0 : "The possible causes are as follows:";
179 0 : HCCL_ERROR("%s", errMsg.c_str());
180 0 : HCCL_ERROR("%s", GET_SOCKET_TIMEOUT_REASON_WITHOUT_EVENT.c_str());
181 0 : errMsg += "\n" + GET_SOCKET_TIMEOUT_REASON_WITHOUT_EVENT;
182 : }
183 :
184 0 : HCCL_ERROR("----------------------------------------------------------------------");
185 0 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>{"reason"}, std::vector<std::string>{errMsg});
186 0 : return HCCL_SUCCESS;
187 0 : }
188 : // 检测连接异常
189 11 : HcclResult DetectConnectionAnomalies::Detect()
190 : {
191 11 : std::unique_lock<std::mutex> lock(detectThreadMutex_);
192 11 : if (!isInitThread_ && threadExit_) {
193 : // 初始化线程,轮询ipNictypeQueue_
194 4 : getIpNictypeQueue_.reset(new (std::nothrow) std::thread(&DetectConnectionAnomalies::DetectMonitor, this));
195 4 : CHK_SMART_PTR_NULL(getIpNictypeQueue_);
196 4 : isInitThread_ = true;
197 : }
198 11 : lock.unlock();
199 11 : return HCCL_SUCCESS;
200 11 : }
201 :
202 :
203 4 : void DetectConnectionAnomalies::DetectMonitor()
204 : {
205 7 : while (threadExit_) {
206 3 : GetIpQueue();
207 3 : std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 每次休眠100毫秒
208 : }
209 4 : return;
210 : }
211 : // 心跳线程调用
212 3 : HcclResult DetectConnectionAnomalies::GetIpQueue()
213 : {
214 3 : if (ipNictypeQueue_.empty()) {
215 3 : return HCCL_SUCCESS;
216 : }
217 :
218 0 : HCCL_RUN_INFO("[GetIpQueue]ipNictypeQueue_ size[%d], start to detect", ipNictypeQueue_.size());
219 0 : std::unique_lock<std::mutex> lock(ipNictypeQueueMutex_);
220 0 : while (!ipNictypeQueue_.empty() && threadExit_) {
221 0 : auto& errInfo = ipNictypeQueue_.front();
222 0 : if (CreateServers(errInfo) != HCCL_SUCCESS ||
223 0 : CreateClients(errInfo, linkClientThreads_) != HCCL_SUCCESS) {
224 0 : ipNictypeQueue_.pop();
225 0 : HCCL_ERROR("[GetIpQueue]CreateServers or CreateClients fail");
226 0 : return HCCL_E_INTERNAL;
227 : }
228 0 : ipNictypeQueue_.pop();
229 : }
230 0 : HCCL_INFO("[GetIpQueue] completed[%d]", ipNictypeQueue_.size());
231 0 : lock.unlock();
232 0 : return HCCL_SUCCESS;
233 0 : }
234 :
235 1 : HcclResult DetectConnectionAnomalies::CreateDetectVnicLinks(struct ErrInfo errInfo)
236 : {
237 1 : SetThreadName("Hccl_Detect_vnic");
238 1 : if (errInfo.deviceLogicId != HOST_DEVICE_ID) {
239 1 : hrtSetDevice(errInfo.deviceLogicId);
240 : }
241 1 : CHK_RET(HcclNetOpenDev(&vnicCtx_, NicType::VNIC_TYPE, errInfo.localRankInfo.devicePhyId,
242 : errInfo.deviceLogicId, errInfo.localRankInfo.deviceVnicIp));
243 1 : CHK_PTR_NULL(vnicCtx_);
244 1 : std::string tag = GetTag(errInfo.localRankInfo.deviceVnicIp);
245 1 : u32 port = (errInfo.localRankInfo.deviceVnicPort== HCCL_INVALID_PORT) ? HETEROG_CCL_PORT : port;
246 :
247 : // 创建vnic socket服务端
248 1 : EXCEPTION_CATCH((vnicSocket_ = std::make_shared<HcclSocket>(vnicCtx_, port)), return HCCL_E_PTR);
249 1 : HCCL_RUN_INFO("[CreateDetectVnicLinks]tag[%s], localIpAddr[%s], remoteIpAddr[%u], port[%u]", tag.c_str(),
250 : errInfo.localRankInfo.deviceVnicIp.GetReadableIP(), errInfo.remoteRankInfo.deviceVnicIp.GetReadableIP(), port);
251 :
252 1 : CHK_RET(vnicSocket_->Init());
253 1 : CHK_RET(vnicSocket_->Listen());
254 1 : CHK_RET(AddWhiteList(vnicSocket_, NicType::VNIC_TYPE, tag));; // 添加白名单
255 1 : HCCL_INFO("[CreateDetectVnicLinks]AddWhiteList finished");
256 :
257 1 : u32 acceptTimeOut = 1; // accept 超时1s
258 1 : std::shared_ptr<HcclSocket> acceptSuccessSocket;
259 1 : auto detectTimeOut = std::chrono::seconds(GetExternalInputDfsConnectionFaultDetectionTime());
260 1 : startTime = std::chrono::steady_clock::now();
261 : HcclResult ret;
262 11 : while (threadExit_ && (std::chrono::steady_clock::now() - startTime) < std::chrono::seconds(detectTimeOut)) {
263 10 : ret = vnicSocket_->Accept(tag, acceptSuccessSocket, acceptTimeOut);
264 10 : if (ret == HCCL_SUCCESS) {
265 0 : HCCL_INFO("[CreateDetectVnicLinks]accept success, localIpAddr[%s], acceptSuccessSocket[%p]",
266 : errInfo.localRankInfo.deviceVnicIp.GetReadableIP(), acceptSuccessSocket.get());
267 0 : listenVnicVec_.push_back(acceptSuccessSocket); // 保存accept成功的socket
268 : }
269 10 : usleep(ACCEPT_TIME_OF_USLEEP); // 休眠100毫秒
270 : }
271 : // 循环发送检测信息
272 1 : startTime = std::chrono::steady_clock::now();
273 11 : while (threadExit_ && (std::chrono::steady_clock::now() - startTime) <= std::chrono::seconds(broadCastTime)) {
274 10 : std::unique_lock<std::mutex> lock(readRecvErrtInfo_);
275 20 : for (auto &recvError : recvErrorInfoMap_) {
276 10 : auto it = sendErrorInfoMap_.find(recvError.first);
277 10 : if (it != sendErrorInfoMap_.end() && !sendErrorInfoMap_[recvError.first].isSendVnic) {
278 2 : for (auto &socket : listenVnicVec_) {
279 1 : CHK_RET(socket->Send(&recvError.second, sizeof(recvError.second)));
280 : }
281 : // 给所有socket都发送完成后，才标记发送完成
282 1 : sendErrorInfoMap_[recvError.first].isSendVnic = true;
283 : }
284 : }
285 10 : lock.unlock();
286 10 : usleep(ACCEPT_TIME_OF_USLEEP); // 休眠1毫秒,将锁释放给IRecv
287 10 : }
288 1 : std::unique_lock<std::mutex> lock(whiteListMutex_);
289 1 : CHK_RET(DelWhiteList(errInfo.localRankInfo.deviceVnicIp, vnicWhiteListInfosVec_, vnicSocket_)); // 删除白名单
290 1 : lock.unlock();
291 1 : if (errInfo.deviceLogicId != HOST_DEVICE_ID) {
292 1 : hrtResetDevice(errInfo.deviceLogicId);
293 : }
294 1 : HCCL_INFO("[CreateDetectVnicLinks] completed");
295 1 : return HCCL_SUCCESS;
296 1 : }
297 :
298 2 : HcclResult DetectConnectionAnomalies::CreateDetectNicLinks(struct ErrInfo errInfo)
299 : {
300 2 : SetThreadName("Hccl_Detect_Nic");
301 2 : if (errInfo.deviceLogicId != HOST_DEVICE_ID) {
302 2 : hrtSetDevice(errInfo.deviceLogicId);
303 : }
304 2 : CHK_RET(HcclNetOpenDev(&nicCtx_, NicType::DEVICE_NIC_TYPE, errInfo.localRankInfo.devicePhyId,
305 : errInfo.deviceLogicId, errInfo.localRankInfo.nicIp[0]));
306 2 : CHK_PTR_NULL(nicCtx_);
307 2 : std::string tag = GetTag(errInfo.localRankInfo.nicIp[0]);
308 :
309 2 : u32 port = (errInfo.localRankInfo.deviceNicPort == HCCL_INVALID_PORT) ? HETEROG_CCL_PORT : port;
310 2 : EXCEPTION_CATCH((nicSocket_ = std::make_shared<HcclSocket>(nicCtx_, port)), return HCCL_E_PTR);
311 :
312 2 : HCCL_RUN_INFO("[CreateDetectNicLinks]tag[%s], localIp[%s], remoteIp[%u], port[%u]", tag.c_str(),
313 : errInfo.localRankInfo.nicIp[0].GetReadableIP(), errInfo.remoteRankInfo.nicIp[0].GetReadableIP(), port);
314 2 : CHK_RET(nicSocket_->Init());
315 2 : CHK_RET(nicSocket_->Listen());
316 1 : CHK_RET(AddWhiteList(nicSocket_, NicType::DEVICE_NIC_TYPE, tag)); // 添加白名单
317 1 : HCCL_INFO("[CreateDetectNicLinks]AddWhiteList finished");
318 :
319 1 : u32 acceptTimeOutAccept = 1;
320 1 : auto acceptTimeOut = std::chrono::seconds(GetExternalInputDfsConnectionFaultDetectionTime());
321 1 : std::shared_ptr<HcclSocket> acceptSuccessSocket;
322 1 : startTime = std::chrono::steady_clock::now();
323 : HcclResult ret;
324 11 : while (threadExit_ && (std::chrono::steady_clock::now() - startTime) <= acceptTimeOut) {
325 10 : ret = nicSocket_->Accept(tag, acceptSuccessSocket, acceptTimeOutAccept);
326 10 : if (ret == HCCL_SUCCESS) {
327 0 : HCCL_INFO("[CreateDetectNicLinks]accept success, localIpAddr[%s], acceptSuccessSocket[%p]",
328 : errInfo.localRankInfo.nicIp[0].GetReadableIP(), acceptSuccessSocket.get());
329 0 : listenNicVec_.push_back(acceptSuccessSocket);
330 : }
331 10 : usleep(ACCEPT_TIME_OF_USLEEP); // 休眠100毫秒
332 : }
333 : // 循环发送
334 1 : startTime = std::chrono::steady_clock::now();
335 11 : while (threadExit_ && (std::chrono::steady_clock::now() - startTime) <= std::chrono::seconds(broadCastTime)) {
336 10 : std::unique_lock<std::mutex> lock(readRecvErrtInfo_);
337 20 : for (auto &recvError : recvErrorInfoMap_) {
338 10 : auto it = sendErrorInfoMap_.find(recvError.first);
339 10 : if (it != sendErrorInfoMap_.end() && !sendErrorInfoMap_[recvError.first].isSendNic) {
340 2 : for (auto &socket : listenNicVec_) {
341 1 : CHK_RET(socket->Send(&recvError.second, sizeof(recvError.second)));
342 : }
343 : // 给所有socket都发送完成后，才标记发送完成
344 1 : sendErrorInfoMap_[recvError.first].isSendNic = true;
345 : }
346 : }
347 10 : lock.unlock();
348 10 : usleep(ACCEPT_TIME_OF_USLEEP); // 休眠10毫秒,将锁释放给IRecv
349 10 : }
350 1 : std::unique_lock<std::mutex> lock(whiteListMutex_);
351 1 : CHK_RET(DelWhiteList(errInfo.localRankInfo.nicIp[0], nicWhiteListInfosVec_, nicSocket_)); // 删除白名单
352 1 : lock.unlock();
353 1 : if (errInfo.deviceLogicId != HOST_DEVICE_ID) {
354 1 : hrtResetDevice(errInfo.deviceLogicId);
355 : }
356 1 : HCCL_INFO("[CreateDetectNicLinks] completed");
357 1 : return HCCL_SUCCESS;
358 2 : }
359 :
360 4 : HcclResult DetectConnectionAnomalies::CreateServers(struct ErrInfo errInfo)
361 : {
362 4 : if (threadExit_) {
363 3 : if (!isCreateLink_) {
364 1 : detectVnicThread_.reset(new (std::nothrow) std::thread(&DetectConnectionAnomalies::CreateDetectVnicLinks,
365 1 : this, errInfo));
366 1 : CHK_SMART_PTR_NULL(detectVnicThread_);
367 1 : isCreateLink_ = true;
368 : }
369 : // 多机场景，且vnic失败时, 这里得用nicIp，否则添加白名单无效
370 3 : if (isNeedNic_ && !isCreateNicLink_) {
371 1 : detectNicThread_.reset(new (std::nothrow) std::thread(&DetectConnectionAnomalies::CreateDetectNicLinks,
372 1 : this, errInfo));
373 1 : CHK_SMART_PTR_NULL(detectNicThread_);
374 1 : isCreateNicLink_ = true;
375 : }
376 : }
377 4 : return HCCL_SUCCESS;
378 : }
379 :
380 5 : std::string DetectConnectionAnomalies::GetTag(HcclIpAddress &Ip, int i)
381 : {
382 15 : return std::string(Ip.GetReadableIP()) + "_detect_" + std::to_string(i);
383 : }
384 :
385 2 : HcclResult DetectConnectionAnomalies::AddWhiteList(
386 : std::shared_ptr<HcclSocket> socket,
387 : NicType nicType,
388 : std::string& tag)
389 : {
390 : // 根据 NicType 处理白名单
391 : HcclResult ret;
392 2 : if (nicType == NicType::VNIC_TYPE) {
393 0 : for (const auto& ipAddr : uniqueIps_) {
394 0 : HcclResult res = AddWlistEntry(ipAddr, tag, whiteVnicSet_, vnicWhiteListInfosVec_);
395 0 : if (res != HCCL_SUCCESS) {
396 0 : return res;
397 : }
398 : }
399 0 : ret = socket->AddWhiteList(vnicWhiteListInfosVec_);
400 0 : if (ret != HCCL_SUCCESS) {
401 0 : HCCL_ERROR("[AddWhiteList] fail");
402 0 : return HCCL_E_NOT_FOUND;
403 : }
404 2 : } else if (isNeedNic_) {
405 2 : for (const auto& ipAddr : uniqueIps_) {
406 1 : HcclResult res = AddWlistEntry(ipAddr, tag, whiteNicSet_, nicWhiteListInfosVec_);
407 1 : if (res != HCCL_SUCCESS) {
408 0 : return res;
409 : }
410 : }
411 1 : ret = socket->AddWhiteList(nicWhiteListInfosVec_);
412 1 : if (ret != HCCL_SUCCESS) {
413 0 : HCCL_ERROR("[AddWhiteList] fail");
414 0 : return HCCL_E_NOT_FOUND;
415 : }
416 : }
417 2 : return HCCL_SUCCESS;
418 : }
419 :
420 2 : HcclResult DetectConnectionAnomalies::DelWhiteList(HcclIpAddress &localIpAddr,
421 : std::vector<struct SocketWlistInfo> whiteListInfos, std::shared_ptr<HcclSocket> socket)
422 : {
423 2 : if (!threadExit_ || whiteListInfos.size() == 0) {
424 2 : return HCCL_SUCCESS;
425 : }
426 0 : HcclResult ret = socket->DelWhiteList(whiteListInfos);
427 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[DelWhiteList]ip[%s] DelWhiteList fail", localIpAddr.GetReadableIP()),
428 : HCCL_E_NOT_FOUND);
429 0 : whiteListInfos.clear();
430 0 : return HCCL_SUCCESS;
431 : }
432 :
433 2 : HcclResult DetectConnectionAnomalies::ConstructErrorInfo(std::shared_ptr<HcclSocket> &clientSocket,
434 : RankInfo &localRankInfo, RankInfo &remoteRankInfo)
435 : {
436 2 : DetectInfo detectInfo{};
437 2 : detectInfo.localDeviceId = localRankInfo.devicePhyId;
438 2 : detectInfo.remoteDeviceId = remoteRankInfo.devicePhyId;
439 :
440 : // 获取本地设备IP并复制到错误信息中(直接获取，因为vnic场景从localRankInfo获得的IP可能是无效的)
441 2 : std::string localDeviceIp = clientSocket->GetLocalIp().GetReadableIP();
442 2 : CHK_SAFETY_FUNC_RET(memcpy_s(detectInfo.localDeviceIp, DEST_MAX_LEN, localDeviceIp.c_str(), localDeviceIp.size()));
443 2 : detectInfo.localDeviceIp[localDeviceIp.size()] = '\0';
444 : // 获取远程设备IP并复制到错误信息中
445 2 : std::string remoteDeviceIp = clientSocket->GetRemoteIp().GetReadableIP();
446 2 : CHK_SAFETY_FUNC_RET(
447 : memcpy_s(detectInfo.remoteDeviceIp, DEST_MAX_LEN, remoteDeviceIp.c_str(), remoteDeviceIp.size()));
448 2 : detectInfo.remoteDeviceIp[remoteDeviceIp.size()] = '\0';
449 : // 复制本地ServerId ID到错误信息中
450 2 : std::string localServerId = localRankInfo.serverId;
451 2 : CHK_SAFETY_FUNC_RET(memcpy_s(detectInfo.localServerId, DEST_MAX_LEN, localServerId.c_str(),
452 : localServerId.size()));
453 2 : detectInfo.localServerId[localServerId.size()] = '\0';
454 :
455 : // 复制远程serverId到错误信息中
456 2 : std::string remoteServerId = remoteRankInfo.serverId;
457 2 : CHK_SAFETY_FUNC_RET(memcpy_s(detectInfo.remoteServerId, DEST_MAX_LEN, remoteServerId.c_str(),
458 : remoteServerId.size()));
459 2 : detectInfo.remoteServerId[remoteServerId.size()] = '\0';
460 :
461 2 : std::unique_lock<std::mutex> lock(readRecvErrtInfo_);
462 2 : std::string ip = localDeviceIp + "-" + remoteDeviceIp;
463 2 : recvErrorInfoMap_.emplace(ip, detectInfo);
464 2 : sendErrorInfoMap_.emplace(ip, SendInfo{});
465 :
466 2 : lock.unlock();
467 : // 保存错误信息
468 2 : return HCCL_SUCCESS;
469 2 : }
470 :
471 3 : HcclResult DetectConnectionAnomalies::GetStatus(struct ErrInfo errInfo, std::shared_ptr<HcclSocket> &clientSocket)
472 : {
473 3 : startTime = std::chrono::steady_clock::now();
474 3 : auto timeout = std::chrono::seconds(GetExternalInputDfsConnectionFaultDetectionTime());
475 : // 等待时间不大于超时时间
476 3 : HcclSocketStatus status = HcclSocketStatus::SOCKET_INIT;
477 :
478 4 : while ((std::chrono::steady_clock::now() - startTime) < timeout) {
479 3 : status = clientSocket->GetStatus();
480 3 : if (status == HcclSocketStatus::SOCKET_OK) {
481 2 : HCCL_INFO("[Detect][ConnectionAnomalies]GetStatus success, remoteIpAddr[%s]",
482 : clientSocket->GetRemoteIp().GetReadableIP());
483 2 : return HCCL_SUCCESS;
484 : }
485 1 : SaluSleep(CLIENT_TIME_OF_USLEEP); // 休眠500毫秒
486 : }
487 1 : std::unique_lock<std::mutex> lock(ipConstuctMutex_);
488 1 : CHK_RET(ConstructErrorInfo(clientSocket, errInfo.localRankInfo, errInfo.remoteRankInfo));
489 1 : lock.unlock();
490 1 : return HCCL_E_TIMEOUT;
491 1 : }
492 :
493 2 : HcclResult DetectConnectionAnomalies::Connect(struct ErrInfo errInfo, std::shared_ptr<HcclSocket> &clientSocket)
494 : {
495 2 : HcclIpAddress localIp = (errInfo.nicType == NicType::VNIC_TYPE) ? errInfo.localRankInfo.deviceVnicIp :
496 2 : errInfo.localRankInfo.nicIp[0];
497 :
498 2 : HcclNetDevCtx Ctx = (errInfo.nicType == NicType::VNIC_TYPE) ? vnicCtx_ : nicCtx_;
499 2 : if (Ctx == nullptr) {
500 1 : CHK_RET(HcclNetOpenDev(&Ctx, errInfo.nicType, errInfo.localRankInfo.devicePhyId,
501 : errInfo.deviceLogicId, localIp));
502 1 : CHK_PTR_NULL(Ctx);
503 1 : std::lock_guard<std::mutex> lock(clientResourcesMutex_);
504 1 : clientNicCtxs_.push_back(Ctx);
505 1 : }
506 :
507 2 : u32 port = (errInfo.nicType == NicType::VNIC_TYPE) ? errInfo.remoteRankInfo.deviceVnicPort : errInfo.remoteRankInfo.deviceNicPort;
508 2 : port = (port == HCCL_INVALID_PORT) ? HETEROG_CCL_PORT : port;
509 2 : HcclIpAddress remoteIpAddr = (errInfo.nicType == NicType::VNIC_TYPE) ? errInfo.remoteRankInfo.deviceVnicIp : errInfo.remoteRankInfo.nicIp[0];
510 :
511 2 : std::string tag = GetTag(remoteIpAddr);
512 2 : HCCL_INFO("[Connect]tag[%s], port[%u], nicCtx[%p], remoteIpAddr[%s], role[%d]", tag.c_str(),
513 : port, Ctx, remoteIpAddr.GetReadableIP(), HcclSocketRole::SOCKET_ROLE_CLIENT);
514 2 : EXCEPTION_CATCH((clientSocket = std::make_shared<HcclSocket>(tag, Ctx, remoteIpAddr, port, HcclSocketRole::SOCKET_ROLE_CLIENT)),
515 : return HCCL_E_PTR);
516 2 : CHK_RET(clientSocket->Init());
517 :
518 2 : HcclResult ret = clientSocket->Connect();
519 2 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Detect][ConnectionAnomalies] connect fail, localIp[%s], remoteIp[%s]",
520 : localIp.GetReadableIP(), remoteIpAddr.GetReadableIP()), HCCL_E_INTERNAL);
521 2 : return HCCL_SUCCESS;
522 2 : }
523 :
524 2 : HcclResult DetectConnectionAnomalies::CreateClient(struct ErrInfo errInfo)
525 : {
526 2 : SetThreadName("Hccl_Detect_Client");
527 2 : if (errInfo.deviceLogicId != HOST_DEVICE_ID) {
528 1 : hrtSetDevice(errInfo.deviceLogicId);
529 : }
530 2 : std::shared_ptr<HcclSocket> clientSocket;
531 2 : CHK_RET(Connect(errInfo, clientSocket));
532 2 : HcclResult ret = GetStatus(errInfo, clientSocket);
533 2 : if (ret != HCCL_SUCCESS) {
534 1 : HCCL_ERROR("[CreateClientConnect]GetStatus fail, ret[%d]", ret);
535 1 : return ret;
536 : }
537 : // 将localServerId转换char，以便打印函数统一
538 1 : char localServerId[DEST_MAX_LEN]{};
539 1 : CHK_SAFETY_FUNC_RET(memcpy_s(localServerId, DEST_MAX_LEN, errInfo.localRankInfo.serverId.c_str(),
540 : errInfo.localRankInfo.serverId.size()));
541 1 : localServerId[errInfo.localRankInfo.serverId.size()] = '\0';
542 :
543 : // 保存clientSocket，在析构时join
544 1 : clientResourcesMutex_.lock();
545 1 : clientSockets_.push_back(clientSocket);
546 1 : clientResourcesMutex_.unlock();
547 :
548 : // 开始计时
549 1 : auto waitTime = std::chrono::seconds(GetExternalInputDfsConnectionFaultDetectionTime()) +
550 2 : std::chrono::seconds(broadCastTime);
551 1 : startTime = std::chrono::steady_clock::now();
552 :
553 1 : DetectInfo detectInfo{};
554 1 : u64 totalSize = sizeof(detectInfo);
555 1 : void *recvBuffer = reinterpret_cast<void *>(&detectInfo);
556 1 : u64 recvSize = 0;
557 1 : while (threadExit_ && (std::chrono::steady_clock::now() - startTime) < waitTime) {
558 0 : u64 compSize = 0; // 本次接收长度
559 0 : void *recvBufferTmp = reinterpret_cast<u8 *>(recvBuffer) + recvSize; // 偏移
560 0 : ret = clientSocket->IRecv(recvBufferTmp, totalSize - recvSize, compSize);
561 0 : if (ret == HCCL_SUCCESS && compSize > 0) {
562 0 : recvSize += compSize;
563 : }
564 0 : if ((totalSize - recvSize) == 0) {
565 0 : recvSize = 0;
566 0 : std::string loaclIp(detectInfo.localDeviceIp);
567 0 : std::string remoteIp(detectInfo.remoteDeviceIp);
568 0 : std::string ip = loaclIp + "-" + remoteIp;
569 0 : std::unique_lock<std::mutex> lock(readRecvErrtInfo_);
570 0 : auto it = recvErrorInfoMap_.find(ip);
571 0 : if (it == recvErrorInfoMap_.end()) {
572 0 : recvErrorInfoMap_.emplace(ip, detectInfo);
573 0 : sendErrorInfoMap_.emplace(ip, SendInfo{});
574 : }
575 0 : CHK_SAFETY_FUNC_RET(memset_s(&detectInfo, sizeof(DetectInfo), 0, sizeof(DetectInfo)));
576 0 : lock.unlock();
577 0 : }
578 0 : usleep(IRECV_TIME_OF_USLEEP); // 休眠500毫秒
579 : }
580 1 : if (errInfo.deviceLogicId != HOST_DEVICE_ID) {
581 0 : hrtResetDevice(errInfo.deviceLogicId);
582 : }
583 1 : HCCL_INFO("[CreateClient]completed");
584 1 : return HCCL_SUCCESS;
585 2 : }
586 :
587 1 : HcclResult DetectConnectionAnomalies::CreateClients(struct ErrInfo errInfo, std::vector<std::unique_ptr<std::thread>> &linkClientThreads)
588 : {
589 1 : std::unique_ptr<std::thread> linkClientThread;
590 1 : linkClientThread.reset(new (std::nothrow) std::thread(&DetectConnectionAnomalies::CreateClient, this, errInfo));
591 1 : CHK_SMART_PTR_NULL(linkClientThread);
592 1 : linkClientThreads.emplace_back(std::move(linkClientThread));
593 1 : return HCCL_SUCCESS;
594 1 : }
595 :
596 0 : std::string DetectConnectionAnomalies::FormatDetectMessage(const std::string &localServerId, s32 localDeviceId, const DetectInfo &detectInfo)
597 : {
598 0 : return std::string("This node (server ") +
599 0 : localServerId + ", device ID " + std::to_string(localDeviceId) +
600 0 : ") detects that srcRank (server " + detectInfo.localServerId +
601 0 : ", device ID " + std::to_string(detectInfo.localDeviceId) +
602 0 : ") fails to connect to dstRank (server " + detectInfo.remoteServerId +
603 0 : ", device ID " + std::to_string(detectInfo.remoteDeviceId) +
604 0 : "). Continue to analyze the fault based on the logs of srcRank and dstRank.";
605 : }
606 :
607 808 : void DetectConnectionAnomalies::ThreadDestroy()
608 : {
609 808 : HCCL_DEBUG("[DetectConnectionAnomalies]Destroy");
610 808 : threadExit_ = false;
611 :
612 : // 销毁client线程
613 808 : std::unique_lock<std::mutex> lock(clientThreadMutex_);
614 808 : for (u32 index = 0; index < linkClientThreads_.size(); index++) {
615 0 : if (linkClientThreads_[index] != nullptr && linkClientThreads_[index]->joinable()) {
616 0 : HCCL_INFO("[DetectConnectionAnomalies]Destroy linkClientThreads_[%p]", linkClientThreads_[index].get());
617 0 : linkClientThreads_[index]->join(); // 等待线程执行完毕
618 0 : linkClientThreads_[index] = nullptr;
619 : }
620 : }
621 808 : linkClientThreads_.clear();
622 808 : lock.unlock();
623 :
624 : // 先销毁线程，再释放资源
625 : // 销毁server线程
626 808 : if (detectVnicThread_ != nullptr && detectVnicThread_->joinable()) {
627 0 : detectVnicThread_->join();
628 0 : detectVnicThread_ = nullptr;
629 : }
630 :
631 808 : if (detectNicThread_ != nullptr && detectNicThread_->joinable()) {
632 0 : detectNicThread_->join();
633 0 : detectNicThread_ = nullptr;
634 : }
635 :
636 : // 销毁轮询线程
637 808 : if(getIpNictypeQueue_ != nullptr && getIpNictypeQueue_->joinable()) {
638 0 : getIpNictypeQueue_->join();
639 0 : getIpNictypeQueue_ = nullptr;
640 : }
641 :
642 : // 释放server侧资源
643 808 : if (vnicSocket_ != nullptr) {
644 0 : for (auto &socket : listenVnicVec_) {
645 0 : socket->DeInit();
646 0 : socket = nullptr;
647 : }
648 0 : vnicSocket_->DeInit();
649 0 : vnicSocket_ = nullptr;
650 : }
651 :
652 808 : if (nicSocket_ != nullptr) {
653 3 : for (auto &socket : listenNicVec_) {
654 1 : socket->DeInit();
655 : }
656 2 : nicSocket_->DeInit();
657 2 : nicSocket_ = nullptr;
658 : }
659 :
660 : // 释放client资源
661 808 : for (auto &socket : clientSockets_) {
662 0 : if (socket != nullptr) {
663 0 : socket->DeInit();
664 0 : socket = nullptr;
665 : }
666 : }
667 :
668 : // 销毁ctx
669 808 : if (nicCtx_ != nullptr) {
670 2 : HcclNetCloseDev(nicCtx_);
671 2 : nicCtx_ = nullptr;
672 : }
673 :
674 808 : if (vnicCtx_ != nullptr) {
675 0 : HcclNetCloseDev(vnicCtx_);
676 0 : vnicCtx_ = nullptr;
677 : }
678 :
679 809 : for (auto nicCtx : clientNicCtxs_) {
680 1 : if (nicCtx != nullptr) {
681 1 : HcclNetCloseDev(nicCtx);
682 : }
683 : }
684 808 : clientNicCtxs_.clear();
685 808 : clientSockets_.clear();
686 808 : listenVnicVec_.clear();
687 808 : listenNicVec_.clear();
688 808 : }
689 808 : void DetectConnectionAnomalies::Deinit()
690 : {
691 808 : int count = initRef_.Unref();
692 808 : if (count > 0) {
693 0 : HCCL_INFO("[DetectConnectionAnomalies]Deinit initRef_[%d]", count);
694 0 : return;
695 808 : } else if (count < 0) {
696 756 : HCCL_WARNING("[DetectConnectionAnomalies]Deinit failed");
697 : }
698 808 : ThreadDestroy();
699 807 : HCCL_INFO("DetectConnectionAnomalies[Deinit] count[%d]", count);
700 808 : return;
701 : }
702 :
703 0 : void AddIpQueue(RankInfo &localRankInfo, RankInfo &remoteRankInfo, NicType nicType,
704 : s32 deviceLogicId)
705 : {
706 0 : DetectConnectionAnomalies::GetInstance(deviceLogicId).AddIpQueue(localRankInfo, remoteRankInfo,
707 : nicType, deviceLogicId);
708 0 : return;
709 : }
710 :
711 41 : __attribute__((constructor)) void DetetcCallBackAddIpQueue()
712 : {
713 41 : DetectCallBack(AddIpQueue);
714 41 : }
715 : } // namespace hccl
|
