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