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 "adapter_hccp_common.h"
12 : #include "externalinput_pub.h"
13 : #include "hccl_socket.h"
14 : #include "sal_pub.h"
15 : #include "hccl_socket_manager.h"
16 :
17 : namespace hccl {
18 719 : HcclSocketManager::HcclSocketManager(NICDeployment nicDeployment, s32 deviceLogicId, u32 devicePhyId, u32 userRank)
19 719 : : nicDeployment_(nicDeployment),
20 719 : deviceLogicId_(deviceLogicId),
21 719 : devicePhyId_(devicePhyId),
22 719 : userRank_(userRank),
23 719 : wlistInfosMap_(),
24 719 : commSocketsMap_()
25 719 : {}
26 :
27 1426 : HcclSocketManager::~HcclSocketManager() { DestroySockets(); }
28 :
29 : std::map<PortInfo, std::shared_ptr<HcclSocket>> HcclSocketManager::serverSocketMap_;
30 : std::map<PortInfo, Referenced> HcclSocketManager::serverSocketRefMap_;
31 : std::mutex HcclSocketManager::serverMapMutex_;
32 : RegisterDetectCallBack g_RegisterDetectCallBack = nullptr;
33 :
34 12 : HcclResult HcclSocketManager::ServerInit(const HcclNetDevCtx netDevCtx, u32 port)
35 : {
36 12 : HcclIpAddress localIp{0};
37 12 : CHK_RET(HcclNetDevGetLocalIp(netDevCtx, localIp));
38 :
39 12 : PortInfo portInfo(localIp, port);
40 :
41 12 : std::unique_lock<std::mutex> lock(serverMapMutex_);
42 12 : auto serverSocketInMap = serverSocketMap_.find(portInfo);
43 12 : if (serverSocketInMap != serverSocketMap_.end()) {
44 0 : auto& serverSocketRef = serverSocketRefMap_[portInfo];
45 0 : serverSocketRef.Ref();
46 0 : return HCCL_SUCCESS;
47 : }
48 :
49 12 : std::shared_ptr<HcclSocket> tempSocket;
50 12 : EXCEPTION_CATCH((tempSocket = std::make_shared<HcclSocket>(netDevCtx, port)), return HCCL_E_PTR);
51 12 : CHK_RET(tempSocket->Init());
52 12 : CHK_RET(tempSocket->Listen());
53 6 : HCCL_INFO("[Init][Server]ip[%s] port[%u]", localIp.GetReadableAddress(), port);
54 6 : serverSocketMap_.insert(std::make_pair(portInfo, tempSocket));
55 :
56 6 : Referenced ref;
57 6 : ref.Ref();
58 6 : serverSocketRefMap_.insert(std::make_pair(portInfo, ref));
59 :
60 6 : return HCCL_SUCCESS;
61 12 : }
62 :
63 158 : HcclResult HcclSocketManager::ServerDeInit(const HcclNetDevCtx netDevCtx, u32 port)
64 : {
65 158 : HcclIpAddress localIp{0};
66 159 : CHK_RET(HcclNetDevGetLocalIp(netDevCtx, localIp));
67 159 : CHK_RET(ServerDeInit(localIp, port));
68 :
69 159 : return HCCL_SUCCESS;
70 159 : }
71 :
72 158 : HcclResult HcclSocketManager::ServerDeInit(const HcclIpAddress& localIp, u32 port)
73 : {
74 158 : PortInfo portInfo(localIp, port);
75 :
76 157 : std::unique_lock<std::mutex> lock(serverMapMutex_);
77 159 : auto res = serverSocketMap_.find(portInfo);
78 159 : if (res == serverSocketMap_.end()) {
79 155 : HCCL_INFO("[DeInit][Server]ip[%s] port[%u] not found", localIp.GetReadableAddress(), port);
80 155 : return HCCL_SUCCESS;
81 : }
82 :
83 4 : auto& serverSocketRef = serverSocketRefMap_[portInfo];
84 4 : serverSocketRef.Unref();
85 :
86 4 : HCCL_INFO(
87 : "[DeInit][Server]ip[%s] port[%u] serverSocketRef.Count() = %d", localIp.GetReadableAddress(), port,
88 : serverSocketRef.Count());
89 4 : if (serverSocketRef.Count() == 0) {
90 4 : HCCL_INFO("[DeInit][Server]ip[%s] port[%u]", localIp.GetReadableAddress(), port);
91 4 : serverSocketMap_[portInfo]->DeInit();
92 4 : serverSocketMap_.erase(portInfo);
93 4 : serverSocketRefMap_.erase(portInfo);
94 : }
95 :
96 4 : return HCCL_SUCCESS;
97 159 : }
98 :
99 : // public
100 : // 填加向本端创建链接的客户端"RANK+IP"白名单
101 48 : HcclResult HcclSocketManager::AddWhiteList(
102 : const std::string& commTag, bool isInterLink, NicType socketType, const HcclIpAddress& localIp,
103 : const std::map<u32, HcclRankLinkInfo>& whiteListMap, uint32_t connectMode)
104 : {
105 48 : if (whiteListMap.size() == 0) {
106 0 : HCCL_ERROR("[Add][WhiteList]client infos map or local Ip is empty.");
107 0 : return HCCL_E_PARA;
108 : }
109 :
110 : HcclResult ret;
111 48 : for (auto& res : serverSocketMap_) {
112 0 : if (res.second->GetLocalIp() == localIp) {
113 0 : std::vector<SocketWlistInfo> wlistInfosVec{};
114 0 : for (auto iter = whiteListMap.begin(); iter != whiteListMap.end(); iter++) {
115 0 : auto dstRankLinkInfo = iter->second;
116 0 : ret = ConstructWhiteList(commTag, isInterLink, socketType, dstRankLinkInfo, wlistInfosVec, connectMode);
117 0 : CHK_PRT_RET(
118 : ret != HCCL_SUCCESS, HCCL_ERROR("[Add][WhiteList]Construct white lists is failed. ret[%d]", ret),
119 : ret);
120 0 : }
121 :
122 0 : if (wlistInfosVec.size() > 0) {
123 0 : HCCL_INFO("[Add][WhiteList]wlist size[%u] ", wlistInfosVec.size());
124 0 : CHK_RET(res.second->AddWhiteList(wlistInfosVec));
125 0 : SaveWhiteListInfo(commTag, res.second, wlistInfosVec);
126 : }
127 0 : }
128 : }
129 :
130 48 : return HCCL_SUCCESS;
131 : }
132 :
133 0 : HcclResult HcclSocketManager::ConstructWhiteList(
134 : const std::string& commTag, bool isInterLink, NicType socketType, const HcclRankLinkInfo& dstRankLinkInfo,
135 : std::vector<SocketWlistInfo>& wlistInfosVec, uint32_t connectMode)
136 : {
137 : SocketWlistInfo wlistInfo;
138 0 : u32 userRank = dstRankLinkInfo.userRank;
139 0 : for (u32 i = 0; i < dstRankLinkInfo.socketsPerLink; i++) {
140 : // 使用Client Rank作为确定标识,保证Client和Server的Tag一致
141 0 : std::string tag;
142 0 : if (connectMode != 0) {
143 0 : tag = commTag;
144 : } else {
145 0 : tag = MakeUniqueConnTag(commTag, isInterLink, userRank, i);
146 : }
147 0 : wlistInfo.connLimit = GetConnLimit(socketType);
148 0 : s32 sRet = memcpy_s(&wlistInfo.tag[0], sizeof(wlistInfo.tag), tag.c_str(), tag.size() + 1);
149 0 : if (sRet != EOK) {
150 0 : HCCL_ERROR("[Construct][WhiteList]memory copy failed. errorno[%d]", sRet);
151 0 : return HCCL_E_MEMORY;
152 : }
153 :
154 0 : wlistInfo.remoteIp.addr = dstRankLinkInfo.ip.GetBinaryAddress().addr;
155 0 : wlistInfo.remoteIp.addr6 = dstRankLinkInfo.ip.GetBinaryAddress().addr6;
156 0 : HCCL_DEBUG(
157 : "[Construct][WhiteList]remoteIp[%s], tag[%s]", dstRankLinkInfo.ip.GetReadableAddress(), wlistInfo.tag);
158 0 : wlistInfosVec.push_back(wlistInfo);
159 0 : }
160 :
161 0 : return HCCL_SUCCESS;
162 : }
163 :
164 : // private
165 : // 移除白名单
166 26 : HcclResult HcclSocketManager::DelWhiteList(const std::string& commTag)
167 : {
168 26 : std::unique_lock<std::mutex> lock(wlistMapMutex_);
169 26 : auto it = wlistInfosMap_.find(commTag);
170 26 : if (it != wlistInfosMap_.end()) {
171 0 : auto nicWlistInfosMap_ = it->second;
172 0 : for (auto iter = nicWlistInfosMap_.begin(); iter != nicWlistInfosMap_.end(); iter++) {
173 0 : auto wlistInfosVec_ = iter->second;
174 0 : iter->first->DelWhiteList(wlistInfosVec_);
175 0 : }
176 0 : wlistInfosMap_.erase(it);
177 0 : }
178 26 : return HCCL_SUCCESS;
179 26 : }
180 :
181 : // public API
182 : // 与远端创建连接，异步接口
183 90 : HcclResult HcclSocketManager::CreateSockets(
184 : const std::string& commTag, bool isInterLink, const HcclNetDevCtx netDevCtx, NicType socketType,
185 : HcclSocketRole localRole, const HcclIpAddress& localIp, const std::map<u32, HcclRankLinkInfo>& remoteInfos,
186 : std::map<u32, std::vector<std::shared_ptr<HcclSocket>>>& socketsMap, std::map<u32, u32>& dstRankToUserRank,
187 : bool isSupportReuse, uint32_t connectMode)
188 : {
189 90 : if (remoteInfos.size() == 0) {
190 0 : HCCL_ERROR("[Create][Sockets]remote infos map or local Ip is empty.");
191 0 : return HCCL_E_PARA;
192 : }
193 :
194 : HcclResult ret;
195 144 : for (auto iter = remoteInfos.begin(); iter != remoteInfos.end(); iter++) {
196 93 : std::vector<std::shared_ptr<HcclSocket>> rankSockets{};
197 93 : auto remoteIpIter = iter->second;
198 93 : u32 remoteUserRank = remoteIpIter.userRank;
199 93 : ret = CreateSockets(
200 : commTag, isInterLink, netDevCtx, socketType, localRole, localIp, remoteIpIter, rankSockets, isSupportReuse,
201 : connectMode);
202 93 : CHK_PRT_RET(
203 : ret != HCCL_SUCCESS,
204 : HCCL_ERROR("[Create][Sockets]create to rank[%u] connection is failed. ret[%u]", iter->first, ret), ret);
205 :
206 54 : if (rankSockets.size() > 0) {
207 54 : socketsMap.insert(std::make_pair(iter->first, rankSockets));
208 54 : dstRankToUserRank.insert(std::make_pair(iter->first, remoteUserRank));
209 : }
210 132 : }
211 51 : return HCCL_SUCCESS;
212 : }
213 :
214 : // private
215 : // 与指定IP创建Socket链接
216 93 : HcclResult HcclSocketManager::CreateSockets(
217 : const std::string& commTag, bool isInterLink, const HcclNetDevCtx netDevCtx, NicType socketType,
218 : HcclSocketRole localRole, const HcclIpAddress& localIp, const HcclRankLinkInfo& remoteLinkInfo,
219 : std::vector<std::shared_ptr<HcclSocket>>& ipSockets, bool isSupportReuse, uint32_t connectMode)
220 : {
221 : HcclResult ret;
222 :
223 93 : u32 remoteRank = remoteLinkInfo.userRank;
224 93 : auto remoteIp = remoteLinkInfo.ip;
225 93 : u32 remotePort = remoteLinkInfo.port;
226 93 : u32 socketsPerLink = remoteLinkInfo.socketsPerLink;
227 :
228 : // 支持复用，则先找下是否与相同的远端IP创建过链接
229 93 : if (isSupportReuse) {
230 : // 先根据远端IP(暂未管本端IP)找，找到了就返回，没找到正常创建
231 87 : u32 gotLinkNum = 0;
232 87 : GetSocketsByRankIP(commTag, remoteRank, remoteIp, socketsPerLink, ipSockets, gotLinkNum);
233 87 : socketsPerLink -= gotLinkNum;
234 87 : if (socketsPerLink == 0) {
235 0 : HCCL_INFO(
236 : "[Create][Sockets]get reuse socket is success."
237 : "commTag[%s], remoteRank[%u], remoteIp[%s], localRank[%u].",
238 : commTag.c_str(), remoteRank, remoteIp.GetReadableIP(), userRank_);
239 0 : return HCCL_SUCCESS;
240 : }
241 : }
242 :
243 93 : ret = ConstructSockets(
244 : commTag, isInterLink, netDevCtx, socketsPerLink, socketType, remoteRank, remoteIp, remotePort, localIp,
245 : localRole, ipSockets, connectMode);
246 92 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Create][Sockets]construct socket is failed. ret[%d]", ret), ret);
247 :
248 : // 若作为客户端，需要发起Connect请求
249 : // 原实现是采用批量接口，当前使用Socket对象进行发起连接，无法使用批量接口，性能理论上可能有下降
250 : // 后继根据影响情况，看是否使用批量接口，若采用批量，则需要在更上层调用发起Connect请求才能最大程度地批量
251 92 : if (localRole == HcclSocketRole::SOCKET_ROLE_CLIENT) {
252 46 : for (u32 i = 0; i < ipSockets.size(); i++) {
253 42 : if (ipSockets[i]->GetStatus() != HcclSocketStatus::SOCKET_OK) {
254 41 : CHK_RET(ipSockets[i]->Connect());
255 : }
256 : }
257 : }
258 :
259 54 : SaveSockets(commTag, remoteRank, remoteIp, ipSockets);
260 :
261 54 : HCCL_INFO("[Create][Sockets]Create Sockets is success.");
262 54 : return HCCL_SUCCESS;
263 93 : }
264 :
265 : // private
266 : // 当前仅用于析构时, 关闭所有Socket; 后继若有需要开放接口，注意 isSupCloseSockImmed_ 需要处理
267 859 : void HcclSocketManager::DestroySockets()
268 : {
269 : // 不对 isSupCloseSockImmed_ 进行判断处理, 强制关闭所有Socket
270 859 : std::unique_lock<std::mutex> lock(socketsMapMutex_);
271 895 : for (auto it = commSocketsMap_.begin(); it != commSocketsMap_.end(); it++) {
272 24 : auto rankSocketmap = it->second;
273 74 : for (auto iter = rankSocketmap.begin(); iter != rankSocketmap.end(); iter++) {
274 50 : DestroySockets(iter->second);
275 : }
276 : // 删除链接时，自动删除 WhiteList
277 24 : DelWhiteList(it->first);
278 24 : }
279 867 : commSocketsMap_.clear();
280 1734 : return;
281 864 : }
282 :
283 : // public API
284 2 : void HcclSocketManager::DestroySockets(const std::string& commTag)
285 : {
286 2 : std::unique_lock<std::mutex> lock(socketsMapMutex_);
287 2 : auto it = commSocketsMap_.find(commTag);
288 2 : if (it != commSocketsMap_.end()) {
289 2 : auto rankSocketmap = it->second;
290 6 : for (auto iter = rankSocketmap.begin(); iter != rankSocketmap.end(); iter++) {
291 4 : DestroySockets(iter->second);
292 : }
293 2 : commSocketsMap_.erase(it);
294 2 : }
295 2 : DelWhiteList(commTag);
296 4 : return;
297 2 : }
298 :
299 : // public API
300 : // Destroy指定的远端Rank的所有的 IP 连接
301 : // 暂无使用, 考虑删除
302 2 : void HcclSocketManager::DestroySockets(const std::string& commTag, u32 rank)
303 : {
304 2 : std::unique_lock<std::mutex> lock(socketsMapMutex_);
305 2 : auto it = commSocketsMap_.find(commTag);
306 2 : if (it != commSocketsMap_.end()) {
307 2 : auto rankSocketmap = it->second;
308 2 : auto iter = rankSocketmap.find(rank);
309 2 : if (iter != rankSocketmap.end()) {
310 2 : DestroySockets(iter->second);
311 2 : rankSocketmap.erase(iter);
312 : }
313 2 : }
314 4 : return;
315 2 : }
316 :
317 : // private
318 56 : void HcclSocketManager::DestroySockets(std::vector<std::shared_ptr<HcclSocket>> rankSockets)
319 : {
320 112 : for (u32 j = 0; j < rankSockets.size(); j++) {
321 56 : auto temp = rankSockets[j];
322 56 : if (temp != nullptr) {
323 56 : temp->Close();
324 : }
325 56 : }
326 :
327 56 : return;
328 : }
329 :
330 : // public API
331 : // isWaitEstablished 为 true 时, 连接建立完成后返回; 为 flase时, 连接请求发起后即返回. 默认为 true.
332 : // 预留 调用时设置为 false, 通过多线程的方式提升建链性能.
333 0 : HcclResult HcclSocketManager::CreateSockets(
334 : const std::string& commTag, bool isInterLink, const HcclNetDevCtx netDevCtx,
335 : const std::map<u32, HcclRankLinkInfo>& dstServerMap, const std::map<u32, HcclRankLinkInfo>& dstClientMap,
336 : std::map<u32, std::vector<std::shared_ptr<HcclSocket>>>& serverSocketsMap,
337 : std::map<u32, std::vector<std::shared_ptr<HcclSocket>>>& clientSocketsMap, bool isSupportReuse,
338 : bool isWaitEstablished)
339 : {
340 0 : HCCL_DEBUG("[Create][Sockets]client map size[%u], server map size[%u]", dstClientMap.size(), dstServerMap.size());
341 :
342 0 : HCCL_DEBUG(
343 : "[Create][Sockets]commTag %s, isInterLink %d, isSupportReuse %d, "
344 : "isWaitEstablished %d",
345 : commTag.c_str(), isInterLink, isSupportReuse, isWaitEstablished);
346 :
347 : HcclResult ret;
348 :
349 : NicType socketType;
350 0 : CHK_RET(HcclNetDevGetNicType(netDevCtx, &socketType));
351 0 : HcclIpAddress localIp{0};
352 0 : CHK_RET(HcclNetDevGetLocalIp(netDevCtx, localIp));
353 :
354 0 : HCCL_DEBUG("[Create][Sockets]localIp %s", localIp.GetReadableAddress());
355 :
356 0 : for (auto it = dstServerMap.begin(); it != dstServerMap.end(); ++it) {
357 0 : HCCL_DEBUG("[Create][Sockets]dstServerMap rank %u", it->first);
358 0 : auto info = it->second;
359 0 : HCCL_DEBUG(
360 : "[Create][Sockets]dstServerMap userRank %u, devicePhyId %u, ip %s, port %u", info.userRank,
361 : info.devicePhyId, info.ip.GetReadableAddress(), info.port);
362 0 : }
363 0 : for (auto it = dstClientMap.begin(); it != dstClientMap.end(); ++it) {
364 0 : HCCL_DEBUG("[Create][Sockets]dstClientMap rank %u", it->first);
365 0 : auto info = it->second;
366 0 : HCCL_DEBUG(
367 : "[Create][Sockets]dstClientMap userRank %u, devicePhyId %u, ip %s, port %u", info.userRank,
368 : info.devicePhyId, info.ip.GetReadableAddress(), info.port);
369 0 : }
370 :
371 : // 作为服务端时，先填加白名单
372 0 : if (dstClientMap.size() > 0) {
373 0 : ret = AddWhiteList(commTag, isInterLink, socketType, localIp, dstClientMap);
374 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Create][Sockets]Add white list failed. ret[%d]", ret), ret);
375 : }
376 :
377 0 : std::map<u32, u32> serverDstRankToUserRank; // 子平面rank 映射 通信域 rank
378 0 : std::map<u32, u32> clientDstRankToUserRank; // 子平面rank 映射 通信域 rank
379 :
380 0 : if (dstServerMap.size() > 0) {
381 : // 作为客户端，创建 Socket，并向所有的服务端发起建链请求
382 0 : ret = CreateSockets(
383 : commTag, isInterLink, netDevCtx, socketType, HcclSocketRole::SOCKET_ROLE_CLIENT, localIp, dstServerMap,
384 : clientSocketsMap, clientDstRankToUserRank, isSupportReuse);
385 0 : CHK_PRT_RET(
386 : ret != HCCL_SUCCESS,
387 : HCCL_ERROR("[Create][Sockets]Create sockets failed, local role is client. ret[%d]", ret), ret);
388 : }
389 :
390 0 : if (dstClientMap.size() > 0) {
391 : // 作为服务端，创建 Socket，然后返回
392 0 : ret = CreateSockets(
393 : commTag, isInterLink, netDevCtx, socketType, HcclSocketRole::SOCKET_ROLE_SERVER, localIp, dstClientMap,
394 : serverSocketsMap, serverDstRankToUserRank, isSupportReuse);
395 0 : CHK_PRT_RET(
396 : ret != HCCL_SUCCESS,
397 : HCCL_ERROR(
398 : "[Create][Sockets]Create connection failed, local role is server."
399 : " ret[%d]",
400 : ret),
401 : ret);
402 : }
403 :
404 0 : HCCL_INFO(
405 : "[Create][Sockets]client socket map size %u, server socket map size %u", serverSocketsMap.size(),
406 : serverSocketsMap.size());
407 :
408 : // 需要等待连接建立成功时，则会阻塞
409 0 : if (isWaitEstablished) {
410 : // 等待所有作为客户端的链接建立成功
411 0 : ret = WaitLinksEstablishCompleted(HcclSocketRole::SOCKET_ROLE_CLIENT, clientSocketsMap);
412 0 : if (ret != HCCL_SUCCESS) {
413 0 : HCCL_ERROR("[Create][Sockets]Wait links establish completed failed, local role is client. ret[%d]", ret);
414 0 : TlsStatus tlsStatus = TlsStatus::UNKNOWN;
415 0 : CHK_PRT_CONT(
416 : HcclNetDevGetTlsStatus(netDevCtx, &tlsStatus),
417 : HCCL_WARNING("[HcclNetDevGetTlsStatus] Can not get TlsStatus"));
418 0 : PrintErrorConnection(
419 : HcclSocketRole::SOCKET_ROLE_CLIENT, clientSocketsMap, clientDstRankToUserRank, tlsStatus);
420 0 : return ret;
421 : }
422 :
423 : // 等待所有作为服务端的链接建立成功
424 0 : ret = WaitLinksEstablishCompleted(HcclSocketRole::SOCKET_ROLE_SERVER, serverSocketsMap);
425 0 : if (ret != HCCL_SUCCESS) {
426 0 : HCCL_ERROR("[Create][Sockets]Wait links establish completed failed, local role is server. ret[%d]", ret);
427 0 : TlsStatus tlsStatus = TlsStatus::UNKNOWN;
428 0 : CHK_PRT_CONT(
429 : HcclNetDevGetTlsStatus(netDevCtx, &tlsStatus),
430 : HCCL_WARNING("[HcclNetDevGetTlsStatus] Can not get TlsStatus"));
431 0 : PrintErrorConnection(
432 : HcclSocketRole::SOCKET_ROLE_SERVER, serverSocketsMap, serverDstRankToUserRank, tlsStatus);
433 0 : return ret;
434 : }
435 : }
436 :
437 0 : return HCCL_SUCCESS;
438 0 : }
439 :
440 0 : HcclResult HcclSocketManager::AddWhiteList(
441 : const std::string& commTag, const HcclNetDevCtx netDevCtx, HcclRankLinkInfo remoteRankInfo)
442 : {
443 : NicType socketType;
444 0 : CHK_RET(HcclNetDevGetNicType(netDevCtx, &socketType));
445 0 : HcclIpAddress localIp{0};
446 0 : CHK_RET(HcclNetDevGetLocalIp(netDevCtx, localIp));
447 :
448 0 : bool isInterLink{true};
449 0 : if (socketType == NicType::VNIC_TYPE) {
450 0 : isInterLink = false;
451 : }
452 :
453 : HcclResult ret;
454 0 : HcclSocketRole role
455 0 : = userRank_ < remoteRankInfo.userRank ? HcclSocketRole::SOCKET_ROLE_SERVER : HcclSocketRole::SOCKET_ROLE_CLIENT;
456 :
457 0 : std::map<u32, HcclRankLinkInfo> remoteMap;
458 0 : remoteMap.insert(std::make_pair(remoteRankInfo.userRank, remoteRankInfo));
459 :
460 : // 作为服务端时，先填加白名单
461 0 : if (role == HcclSocketRole::SOCKET_ROLE_SERVER) {
462 0 : ret = AddWhiteList(commTag, isInterLink, socketType, localIp, remoteMap);
463 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Create][Sockets]Add white list failed. ret[%d]", ret), ret);
464 : }
465 :
466 0 : return HCCL_SUCCESS;
467 0 : }
468 :
469 86 : HcclResult HcclSocketManager::CreateSingleLinkSocket(
470 : const std::string& commTag, const HcclNetDevCtx netDevCtx, HcclRankLinkInfo rmtRank,
471 : std::vector<std::shared_ptr<HcclSocket>>& connectSockets, bool isWaitEstablished, bool isSupportReuse, s32 timeout,
472 : uint32_t connectMode)
473 : {
474 : HcclResult ret;
475 : NicType socketType;
476 86 : CHK_RET(HcclNetDevGetNicType(netDevCtx, &socketType));
477 86 : HcclIpAddress localIp{0};
478 87 : CHK_RET(HcclNetDevGetLocalIp(netDevCtx, localIp));
479 :
480 86 : bool isInterLink{true};
481 86 : if (socketType == NicType::VNIC_TYPE) {
482 86 : isInterLink = false;
483 : }
484 86 : HcclSocketRole role
485 86 : = userRank_ < rmtRank.userRank ? HcclSocketRole::SOCKET_ROLE_SERVER : HcclSocketRole::SOCKET_ROLE_CLIENT;
486 :
487 86 : std::map<u32, HcclRankLinkInfo> remoteMap;
488 87 : remoteMap.insert(std::make_pair(rmtRank.userRank, rmtRank));
489 87 : if (role == HcclSocketRole::SOCKET_ROLE_SERVER) { // 作为服务端时，先填加白名单
490 48 : ret = AddWhiteList(commTag, isInterLink, socketType, localIp, remoteMap, connectMode);
491 48 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Create][Sockets]Add white list failed. ret[%d]", ret), ret);
492 : }
493 :
494 87 : std::map<u32, u32> remoteRankToUserRank; // 子平面rank 映射 通信域 rank
495 87 : std::map<u32, std::vector<std::shared_ptr<HcclSocket>>> socketsMap;
496 87 : ret = CreateSockets(
497 : commTag, isInterLink, netDevCtx, socketType, role, localIp, remoteMap, socketsMap, remoteRankToUserRank,
498 : isSupportReuse, connectMode);
499 87 : CHK_PRT_RET(
500 : ret != HCCL_SUCCESS,
501 : HCCL_ERROR(
502 : "[Create][Sockets]Create connection failed, local role is server."
503 : " ret[%d]",
504 : ret),
505 : ret);
506 48 : if (isWaitEstablished) {
507 0 : ret = WaitLinksEstablishCompleted(role, socketsMap, timeout);
508 0 : if (ret != HCCL_SUCCESS) {
509 0 : HCCL_ERROR("[Create][Sockets]Wait links establish completed failed, local role is client. ret[%d]", ret);
510 0 : TlsStatus tlsStatus = TlsStatus::UNKNOWN;
511 0 : CHK_PRT_CONT(HcclNetDevGetTlsStatus(netDevCtx, &tlsStatus), HCCL_WARNING("Can not get TlsStatus"));
512 0 : PrintErrorConnection(role, socketsMap, remoteRankToUserRank, tlsStatus);
513 0 : AbortAndDeleteSocket(commTag, role, socketsMap);
514 0 : return ret;
515 : }
516 : }
517 :
518 48 : connectSockets.clear();
519 96 : for (auto iter = socketsMap.begin(); iter != socketsMap.end(); iter++) {
520 48 : auto rankSockets = iter->second;
521 96 : for (u32 i = 0; i < rankSockets.size(); i++) {
522 48 : connectSockets.push_back(rankSockets[i]);
523 : }
524 48 : }
525 48 : return HCCL_SUCCESS;
526 87 : }
527 :
528 0 : void HcclSocketManager::AbortAndDeleteSocket(
529 : const std::string& commTag, HcclSocketRole role,
530 : const std::map<u32, std::vector<std::shared_ptr<HcclSocket>>>& socketsMap)
531 : {
532 : (void)role;
533 0 : for (auto iter = socketsMap.begin(); iter != socketsMap.end(); iter++) {
534 0 : auto rankSockets = iter->second;
535 0 : for (auto socket : rankSockets) {
536 : // 建链超时后终止当前的建链请求并删除socket防止复用
537 0 : (void)socket->Close();
538 0 : }
539 0 : }
540 0 : DestroySockets(commTag);
541 0 : return;
542 : }
543 :
544 : // public
545 0 : HcclResult HcclSocketManager::GetListenPortByIp(
546 : [[maybe_unused]] const NICDeployment nicDeployment, const HcclIpAddress& ipAddr, std::set<u32>& listenedPort)
547 : {
548 0 : for (auto& res : serverSocketMap_) {
549 0 : if (res.first.ip == ipAddr) {
550 0 : listenedPort.insert(res.first.listenPort);
551 : }
552 : }
553 :
554 0 : if (listenedPort.size() > 0) {
555 0 : return HCCL_SUCCESS;
556 : }
557 :
558 0 : return HCCL_E_NOT_FOUND;
559 : }
560 :
561 : // private
562 40 : void HcclSocketManager::TransformSocketStatus(HcclSocketStatus status, std::string& stringStatus) const
563 : {
564 40 : switch (status) {
565 2 : case HcclSocketStatus::SOCKET_CONNECTING:
566 2 : stringStatus = "connecting";
567 2 : break;
568 0 : case HcclSocketStatus::SOCKET_OK:
569 0 : stringStatus = "connected";
570 0 : break;
571 0 : case HcclSocketStatus::SOCKET_TIMEOUT:
572 0 : stringStatus = "time out";
573 0 : break;
574 38 : case HcclSocketStatus::SOCKET_ERROR:
575 38 : stringStatus = "connect failed";
576 38 : break;
577 0 : case HcclSocketStatus::SOCKET_INIT:
578 : default:
579 0 : stringStatus = "no connect";
580 0 : break;
581 : }
582 40 : }
583 :
584 : // private
585 44 : void HcclSocketManager::PrintSocketsInfo(
586 : const std::string& localRole, u32 rank, std::vector<std::shared_ptr<HcclSocket>> ipSockets,
587 : std::string& sTlsStatus) const
588 : {
589 88 : for (u32 j = 0; j < ipSockets.size(); j++) {
590 44 : std::shared_ptr<HcclSocket> tempSocket = ipSockets[j];
591 44 : if (tempSocket->GetStatus() != HcclSocketStatus::SOCKET_OK) {
592 40 : std::string connectStatus;
593 40 : TransformSocketStatus(tempSocket->GetStatus(), connectStatus);
594 40 : HCCL_ERROR(
595 : " | %s(%u) | %u | %s(%u) | %u | %s | %s | %s | %s",
596 : tempSocket->GetRemoteIp().GetReadableAddress(), rank, tempSocket->GetRemotePort(),
597 : tempSocket->GetLocalIp().GetReadableAddress(), userRank_, tempSocket->GetLocalPort(), localRole.c_str(),
598 : connectStatus.c_str(), sTlsStatus.c_str(), LOG_KEYWORDS_LINK_INFO.c_str());
599 40 : }
600 44 : }
601 44 : }
602 :
603 : // private
604 41 : void HcclSocketManager::PrintErrorConnectionInfo(
605 : HcclSocketRole localRole, std::map<u32, std::vector<std::shared_ptr<HcclSocket>>>& rankSocketsMap,
606 : std::map<u32, u32>& dstRankToUserRank, TlsStatus& tlsStatus) const
607 : {
608 41 : std::string sRole;
609 41 : switch (localRole) {
610 39 : case HcclSocketRole::SOCKET_ROLE_SERVER:
611 39 : sRole = " server ";
612 39 : break;
613 2 : case HcclSocketRole::SOCKET_ROLE_CLIENT:
614 2 : sRole = " client ";
615 2 : break;
616 0 : default:
617 0 : sRole = " NA ";
618 0 : break;
619 : }
620 : const std::map<TlsStatus, std::string> tlsStatusMap
621 246 : = {{TlsStatus::ENABLE, "ENABLE"}, {TlsStatus::DISABLE, "DISABLE"}, {TlsStatus::UNKNOWN, "UNKNOWN"}};
622 82 : std::string sTlsStatus = tlsStatusMap.at(tlsStatus);
623 85 : for (auto iter = rankSocketsMap.begin(); iter != rankSocketsMap.end(); iter++) {
624 44 : auto rankSockets = iter->second;
625 44 : PrintSocketsInfo(sRole, dstRankToUserRank[iter->first], rankSockets, sTlsStatus);
626 44 : }
627 82 : }
628 :
629 : // private
630 41 : void HcclSocketManager::PrintErrorConnection(
631 : HcclSocketRole localRole, std::map<u32, std::vector<std::shared_ptr<HcclSocket>>>& rankSocketsMap,
632 : std::map<u32, u32>& dstRankToUserRank, TlsStatus& tlsStatus) const
633 : {
634 : // 原实现中，打印输出了一个num，而实际调用中，这个num都是1，所以当前删除了
635 41 : HCCL_ERROR(" _________________________LINK_ERROR_INFO___________________________");
636 41 : HCCL_ERROR(" | comm error, device[%d] ", deviceLogicId_);
637 41 : HCCL_ERROR(" | dest_ip(user_rank) | dest_port | src_ip(user_rank) | src_port | MyRole "
638 : "| Status | TlsStatus |");
639 41 : HCCL_ERROR(" |----------------------|---------------|----------------------|--------------|------------"
640 : "|------------|----------------|");
641 :
642 41 : PrintErrorConnectionInfo(localRole, rankSocketsMap, dstRankToUserRank, tlsStatus);
643 41 : return;
644 : }
645 :
646 : // private
647 : // 获取SOCKET连接的 ConnLimit，临时写一个，后继看如何判定，后继考虑初始化到Listen的Socket中
648 0 : u32 HcclSocketManager::GetConnLimit(NicType socketType)
649 : {
650 0 : u32 connLimit = 1;
651 0 : switch (socketType) {
652 0 : case NicType::DEVICE_NIC_TYPE:
653 0 : connLimit = NIC_SOCKET_CONN_LIMIT;
654 0 : break;
655 0 : case NicType::VNIC_TYPE:
656 0 : connLimit = VNIC_SOCKET_CONN_LIMIT;
657 0 : break;
658 0 : case NicType::HOST_NIC_TYPE:
659 0 : connLimit = HOST_SOCKET_CONN_LIMIT;
660 0 : break;
661 0 : default:
662 0 : break;
663 : }
664 0 : return connLimit;
665 : }
666 :
667 : // 生成统一的 SocketTag, 用于标识Socket, 对于一对链接(Server <--> Client), SocketTag 需要相同
668 : std::string
669 93 : HcclSocketManager::MakeUniqueConnTag(const std::string& commTag, bool isInterLink, u32 userRank, u32 indexForLink)
670 : {
671 93 : std::string tmpStr = isInterLink ? "_Inter_" : "_Intra_";
672 : std::string socketTag
673 92 : = commTag + tmpStr + "MultiSocket_" + std::to_string(userRank) + "_" + std::to_string(indexForLink);
674 93 : return socketTag;
675 93 : }
676 :
677 : // 保存白名单，后继保存到Listen的Socket对象中
678 0 : void HcclSocketManager::SaveWhiteListInfo(
679 : const std::string& commTag, std::shared_ptr<HcclSocket>& socket, const std::vector<SocketWlistInfo> wlistInfos)
680 : {
681 : // 将Add成功的whiteList, 保存在Socket对象中, 方便后继Del.
682 : // 理论上, 接口 DelSocketWhiteList, 上层应该没有调用的必要性.
683 : // socket关闭时, 底层应该会自动清除, 而关闭前, 也没看到调用的必要性.
684 : // 综上, 即保存这个 wlistInfosMap_ 必要性可能不大.
685 0 : std::unique_lock<std::mutex> lock(wlistMapMutex_);
686 0 : auto it = wlistInfosMap_.find(commTag);
687 0 : if (it == wlistInfosMap_.end()) {
688 0 : std::map<std::shared_ptr<HcclSocket>, std::vector<SocketWlistInfo>> nicWlistInfosMap;
689 0 : nicWlistInfosMap.insert(std::make_pair(socket, wlistInfos));
690 0 : wlistInfosMap_.insert(std::make_pair(commTag, nicWlistInfosMap));
691 0 : } else {
692 0 : auto& nicWlistInfosMap = it->second;
693 0 : auto iter = nicWlistInfosMap.find(socket);
694 0 : if (iter == nicWlistInfosMap.end()) {
695 0 : nicWlistInfosMap.insert(std::make_pair(socket, wlistInfos));
696 : } else {
697 0 : auto& wlist = iter->second;
698 0 : wlist.insert(wlist.end(), wlistInfos.begin(), wlistInfos.end());
699 : }
700 : }
701 0 : }
702 :
703 : // private
704 : // 根据相关参数构造 HcclSocket 对象
705 93 : HcclResult HcclSocketManager::ConstructSockets(
706 : const std::string& commTag, bool isInterLink, const HcclNetDevCtx netDevCtx, u32 socketsPerLink,
707 : [[maybe_unused]] NicType socketType, u32 remoteUserRank, const HcclIpAddress& remoteIp, u32 remotePort,
708 : [[maybe_unused]] const HcclIpAddress& localIp, HcclSocketRole localRole,
709 : std::vector<std::shared_ptr<HcclSocket>>& socketList, uint32_t connectMode)
710 : {
711 : // 使用Client Rank作为确定标识,保证Client和Server的Tag一致
712 93 : u32 clientRank = localRole == HcclSocketRole::SOCKET_ROLE_CLIENT ? userRank_ : remoteUserRank;
713 185 : for (u32 i = 0; i < socketsPerLink; i++) {
714 93 : std::string socketTag = MakeUniqueConnTag(commTag, isInterLink, clientRank, i);
715 93 : std::shared_ptr<HcclSocket> tempSocket;
716 93 : if (connectMode > 0) {
717 0 : EXCEPTION_CATCH(
718 : (tempSocket = std::make_shared<HcclSocket>(commTag, netDevCtx, remoteIp, remotePort, localRole)),
719 : return HCCL_E_PTR);
720 : } else {
721 93 : EXCEPTION_CATCH(
722 : (tempSocket = std::make_shared<HcclSocket>(socketTag, netDevCtx, remoteIp, remotePort, localRole)),
723 : return HCCL_E_PTR);
724 : }
725 93 : CHK_RET(tempSocket->Init());
726 93 : socketList.push_back(tempSocket);
727 93 : }
728 92 : return HCCL_SUCCESS;
729 : }
730 :
731 : // private
732 : // 保存 HcclSocket 对象
733 54 : void HcclSocketManager::SaveSockets(
734 : const std::string& commTag, u32 remoteRank, const HcclIpAddress& remoteIp,
735 : std::vector<std::shared_ptr<HcclSocket>>& ipSockets)
736 : {
737 : // 将 ipSockets 累加到 commSocketsMap_ 中
738 54 : HCCL_DEBUG(
739 : "[Save][Sockets]commTag[%s], remoteRank[%u], remoteIp[%s], localRank[%u], save socket size[%u].",
740 : commTag.c_str(), remoteRank, remoteIp.GetReadableIP(), userRank_, ipSockets.size());
741 :
742 54 : std::unique_lock<std::mutex> lock(socketsMapMutex_);
743 54 : auto it = commSocketsMap_.find(commTag);
744 54 : if (it == commSocketsMap_.end()) {
745 26 : std::map<u32, std::vector<std::shared_ptr<HcclSocket>>> rankSocketmap{};
746 26 : rankSocketmap.insert(std::make_pair(remoteRank, ipSockets));
747 26 : commSocketsMap_.insert(std::make_pair(commTag, rankSocketmap));
748 26 : } else {
749 28 : auto& rankSocketmap = it->second;
750 28 : auto iter = rankSocketmap.find(remoteRank);
751 28 : if (iter == rankSocketmap.end()) {
752 28 : rankSocketmap.insert(std::make_pair(remoteRank, ipSockets));
753 28 : HCCL_DEBUG(
754 : "[Save][Sockets]rankSocketmap size[%u], ipSockets size[%u].", rankSocketmap.size(), ipSockets.size());
755 : } else {
756 0 : auto& rankSockets = iter->second;
757 : // 当前直接插入，不判断本端IP，远端IP是否相同
758 0 : for (u32 i = 0; i < ipSockets.size(); i++) {
759 0 : rankSockets.push_back(ipSockets[i]);
760 : }
761 0 : HCCL_DEBUG(
762 : "[Save][Sockets]rankSocketmap size[%u], rankSockets size[%u].", rankSocketmap.size(),
763 : rankSockets.size());
764 : }
765 : }
766 54 : HCCL_DEBUG("[Save][Sockets]commSocketsMap_ size[%u].", commSocketsMap_.size());
767 54 : }
768 :
769 : // private
770 86 : void HcclSocketManager::GetSocketsByRankIP(
771 : const std::string& commTag, u32 remoteRank, const HcclIpAddress& remoteIp, u32 socketsPerLink,
772 : std::vector<std::shared_ptr<HcclSocket>>& ipSockets, u32& gotLinkNum)
773 : {
774 86 : HCCL_DEBUG(
775 : "[Get][SocketsByRankIP]commTag[%s], remoteRank[%u], remoteIp[%s], localRank[%u], SocketsMap size[%u].",
776 : commTag.c_str(), remoteRank, remoteIp.GetReadableIP(), userRank_, commSocketsMap_.size());
777 :
778 87 : gotLinkNum = 0;
779 87 : auto it = commSocketsMap_.find(commTag);
780 86 : if (it != commSocketsMap_.end()) {
781 25 : auto& rankSocketmap = it->second;
782 25 : HCCL_INFO("[Get][SocketsByRankIP]rankSocketmap size[%u].", rankSocketmap.size());
783 25 : auto iter = rankSocketmap.find(remoteRank);
784 25 : if (iter != rankSocketmap.end()) {
785 0 : auto& rankSockets = iter->second;
786 0 : GetSocketsByRankIP(remoteIp, socketsPerLink, rankSockets, ipSockets, gotLinkNum);
787 : }
788 : }
789 87 : }
790 :
791 0 : void HcclSocketManager::GetSocketsByRankIP(
792 : const HcclIpAddress& remoteIp, u32 socketsPerLink, std::vector<std::shared_ptr<HcclSocket>>& rankSockets,
793 : std::vector<std::shared_ptr<HcclSocket>>& ipSockets, u32& gotLinkNum)
794 : {
795 0 : for (u32 idx = 0; idx < rankSockets.size(); idx++) {
796 0 : if (rankSockets[idx]->GetRemoteIp() == remoteIp) {
797 0 : ipSockets.push_back(rankSockets[idx]);
798 0 : gotLinkNum++;
799 0 : if (gotLinkNum == socketsPerLink) {
800 0 : break;
801 : }
802 : }
803 : }
804 0 : }
805 :
806 : // private
807 : // 同步接口，更新连接状态，并返回连接成功的连接数量
808 : HcclResult
809 40 : HcclSocketManager::WaitLinkEstablish(std::shared_ptr<HcclSocket> socket, std::function<bool()> needStop, s32 timeout)
810 : {
811 40 : CHK_SMART_PTR_NULL(socket);
812 40 : u32 count = 0;
813 40 : auto startTime = std::chrono::steady_clock::now();
814 40 : auto timeoutSec = std::chrono::seconds(timeout > 0 ? timeout : GetExternalInputHcclLinkTimeOut());
815 40 : HCCL_DEBUG("[Wait][LinkEstablish]waiting for sockets link up...");
816 : while (true) {
817 987 : CHK_PRT_RET(needStop(), HCCL_ERROR("Terminating operation due to external request"), HCCL_E_INTERNAL);
818 :
819 987 : if ((std::chrono::steady_clock::now() - startTime) >= timeoutSec) {
820 1 : HCCL_ERROR(
821 : "[Wait][LinkEstablish]wait socket establish timeout, role[%u] rank[%u] timeout[%lld s]",
822 : static_cast<u32>(socket->GetLocalRole()), userRank_, timeoutSec);
823 1 : socket->SetStatus(HcclSocketStatus::SOCKET_TIMEOUT);
824 1 : return HCCL_E_TIMEOUT;
825 : }
826 986 : HcclSocketStatus status = socket->GetStatus();
827 986 : if (status == HcclSocketStatus::SOCKET_OK) {
828 0 : HCCL_DEBUG(
829 : "[Wait][LinkEstablish]socket is establish. localIp[%s], remoteIp[%s]",
830 : socket->GetLocalIp().GetReadableIP(), socket->GetRemoteIp().GetReadableIP());
831 0 : return HCCL_SUCCESS;
832 986 : } else if (status == HcclSocketStatus::SOCKET_CONNECTING) {
833 947 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
834 : // 日志过滤, 50 次才打印一次
835 947 : if (count % 50 == 0) {
836 19 : HCCL_DEBUG("[Wait][LinkEstablish]socket is connecting ");
837 : }
838 947 : count++;
839 :
840 947 : continue;
841 39 : } else if (status == HcclSocketStatus::SOCKET_TIMEOUT) {
842 0 : return HCCL_E_TIMEOUT;
843 : } else {
844 39 : socket->SetStatus(HcclSocketStatus::SOCKET_ERROR);
845 39 : return HCCL_E_TCP_CONNECT;
846 : }
847 947 : }
848 : return HCCL_E_TCP_CONNECT;
849 : }
850 :
851 38 : HcclResult HcclSocketManager::WaitLinksEstablishCompleted(
852 : HcclSocketRole localRole, std::map<u32, std::vector<std::shared_ptr<HcclSocket>>>& socketsMap,
853 : std::map<u32, u32>& dstRankToUserRank, const RankInfo& loaclRankInfo, const RankInfo& remoteRankInfo,
854 : const HcclNetDevCtx& netDevCtx)
855 : {
856 : (void)loaclRankInfo;
857 : (void)remoteRankInfo;
858 38 : HcclResult ret = WaitLinksEstablishCompleted(localRole, socketsMap);
859 38 : if (ret != HCCL_SUCCESS) {
860 38 : TlsStatus tlsStatus = TlsStatus::UNKNOWN;
861 38 : CHK_PRT_CONT(
862 : HcclNetDevGetTlsStatus(netDevCtx, &tlsStatus),
863 : HCCL_WARNING("[HcclNetDevGetTlsStatus] Can not get TlsStatus"));
864 38 : PrintErrorConnection(localRole, socketsMap, dstRankToUserRank, tlsStatus);
865 38 : HCCL_ERROR("[Create][Sockets]Wait links establish completed failed, local role is client. ret[%d]", ret);
866 38 : return ret;
867 : }
868 0 : return HCCL_SUCCESS;
869 : }
870 :
871 0 : void HcclSocketManager::AddIpQueue(
872 : RankInfo& localRankInfo, RankInfo& remoteRankInfo, NicType nicType, s32 deviceLogicId)
873 : {
874 0 : if (g_RegisterDetectCallBack != nullptr) {
875 0 : g_RegisterDetectCallBack(localRankInfo, remoteRankInfo, nicType, deviceLogicId);
876 : } else {
877 0 : HCCL_RUN_WARNING("[AddIpQueue] g_RegisterDetectCallBack is nullptr");
878 : }
879 0 : }
880 :
881 : // private
882 : // 同步接口，等待连接建立完成
883 40 : HcclResult HcclSocketManager::WaitLinksEstablishCompleted(
884 : [[maybe_unused]] HcclSocketRole localRole, std::map<u32, std::vector<std::shared_ptr<HcclSocket>>>& rankSocketsMap,
885 : s32 timeout)
886 : {
887 40 : for (auto iter = rankSocketsMap.begin(); iter != rankSocketsMap.end(); iter++) {
888 40 : auto rankSockets = iter->second;
889 40 : for (u32 i = 0; i < rankSockets.size(); i++) {
890 40 : HcclResult ret = WaitLinkEstablish(
891 40 : rankSockets[i],
892 40 : [this]() -> bool {
893 987 : return this->GetStopFlag();
894 : },
895 : timeout);
896 40 : CHK_PRT_RET(
897 : ret != HCCL_SUCCESS, HCCL_ERROR("[Wait][LinksEstablishCompleted] is failed. ret[%d].", ret), ret);
898 : }
899 40 : }
900 0 : return HCCL_SUCCESS;
901 : }
902 :
903 0 : HcclResult HcclSocketManager::SetStopFlag(bool value)
904 : {
905 0 : stopFlag_.store(value);
906 :
907 0 : std::unique_lock<std::mutex> lock(socketsMapMutex_);
908 0 : for (auto& socketsMap : commSocketsMap_) { // map
909 0 : for (auto& socketMap : socketsMap.second) { // map
910 0 : for (auto& socket : socketMap.second) { // vector
911 0 : CHK_RET(socket->SetStopFlag(value));
912 : }
913 : }
914 : }
915 :
916 0 : return HCCL_SUCCESS;
917 0 : }
918 :
919 987 : bool HcclSocketManager::GetStopFlag() { return stopFlag_.load(); }
920 :
921 : #ifdef __cplusplus
922 : extern "C" {
923 : #endif // __cplusplus
924 47 : void DetectCallBack(RegisterDetectCallBack p1) { g_RegisterDetectCallBack = p1; }
925 : #ifdef __cplusplus
926 : }
927 : #endif // __cplusplus
928 :
929 : } // namespace hccl
|
