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