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 "opretry_connection.h"
12 : #include <fstream>
13 : #include <nlohmann/json.hpp>
14 : #include "sal_pub.h"
15 : #include "hccl_network_pub.h"
16 : #include "externalinput_pub.h"
17 : #include "adapter_rts_common.h"
18 : #include "opretry_connection_pub.h"
19 :
20 : namespace hccl {
21 : UniversalConcurrentMap<std::string, OpRetryConnection::OpRetryConnectionPtr> *OpRetryConnection::instance_ = nullptr;
22 : std::mutex OpRetryConnection::lock_;
23 : bool OpRetryConnection::enable_ = true;
24 :
25 : /** OpRetryConnectionPub 封装一层OpRetryConnection接口 */
26 5 : void OpRetryConnectionPub::SetOpRetryConnEnable(bool enable)
27 : {
28 5 : OpRetryConnection::SetOpRetryConnEnable(enable);
29 5 : }
30 :
31 4 : bool OpRetryConnectionPub::IsOpRetryConnEnable()
32 : {
33 4 : return OpRetryConnection::IsOpRetryConnEnable();
34 : }
35 :
36 6 : HcclResult OpRetryConnectionPub::Init(const std::string &group, u32 rankSize, const OpRetryServerInfo& serverInfo,
37 : const OpRetryAgentInfo& agentInfo, u32 rootRank)
38 : {
39 6 : return OpRetryConnection::Init(group, rankSize, serverInfo, agentInfo, rootRank);
40 : }
41 :
42 4 : void OpRetryConnectionPub::DeInit(const std::string &group)
43 : {
44 4 : OpRetryConnection::DelInstance(group);
45 4 : }
46 :
47 2 : HcclResult OpRetryConnectionPub::GetConns(const std::string &group, bool &isRoot, std::shared_ptr<HcclSocket> &agent,
48 : std::map<u32, std::shared_ptr<HcclSocket>> &server)
49 : {
50 2 : if (!IsOpRetryConnEnable()) {
51 1 : HCCL_INFO("[OpRetryConnection][Init] op retry is disable, so don't need get conns");
52 1 : return HCCL_SUCCESS;
53 : }
54 :
55 1 : OpRetryConnection::OpRetryConnectionPtr conn;
56 1 : CHK_RET(OpRetryConnection::GetInstance(group, conn));
57 :
58 1 : isRoot = conn->IsRoot();
59 1 : CHK_RET(conn->GetAgentSocket(agent));
60 1 : if (isRoot) {
61 1 : CHK_RET(conn->GetServerSockets(server));
62 : }
63 :
64 1 : return HCCL_SUCCESS;
65 1 : }
66 : /** OpRetryConnectionPub 封装层结束 */
67 :
68 9 : OpRetryConnection::OpRetryConnection()
69 : {
70 9 : }
71 :
72 9 : OpRetryConnection::~OpRetryConnection()
73 : {
74 9 : DeInit();
75 9 : }
76 :
77 5 : void OpRetryConnection::SetOpRetryConnEnable(bool enable)
78 : {
79 5 : enable_ = enable;
80 5 : }
81 :
82 11 : bool OpRetryConnection::IsOpRetryConnEnable()
83 : {
84 11 : return enable_;
85 : }
86 :
87 7 : HcclResult OpRetryConnection::Init(const std::string &group, u32 rankSize, const OpRetryServerInfo& serverInfo,
88 : const OpRetryAgentInfo& agentInfo, u32 rootRank)
89 : {
90 7 : if (!IsOpRetryConnEnable()) {
91 1 : HCCL_INFO("[OpRetryConnection][Init] op retry is disable");
92 1 : return HCCL_SUCCESS;
93 : }
94 6 : u32 rankId = agentInfo.userRank;
95 6 : HcclIpAddress serverIp = serverInfo.hostIP;
96 6 : u32 serverPort = serverInfo.hostPort == HCCL_INVALID_PORT
97 6 : ? GetServerPort() + serverInfo.devId : serverInfo.hostPort;
98 6 : HcclIpAddress localIp = agentInfo.hostIP;
99 6 : if (serverIp.IsInvalid() || localIp.IsInvalid()) {
100 2 : HCCL_ERROR("[OpRetryConnection][Init] serverIp [%s] or localIp [%s] is invalid, "
101 : "check whether the value of host_ip in ranktable is correct.",
102 : serverIp.GetReadableIP(), localIp.GetReadableIP());
103 2 : return HCCL_E_PARA;
104 : }
105 4 : if (rankId >= rankSize || rankSize == 0 || rootRank >= rankSize) {
106 0 : HCCL_ERROR("[OpRetryConnection][Init] opRetryConnection input params invalid,"
107 : "rankId [%u] rankSize [%u] serverIp [%s] localIp [%s] rootRank [%u]",
108 : rankId, rankSize, serverIp.GetReadableIP(), localIp.GetReadableIP(), rootRank);
109 0 : return HCCL_E_PARA;
110 : }
111 :
112 4 : HCCL_INFO("[OpRetryConnection][Init] group[%s] rankId [%u] rankSize [%u] serverIp [%s] localIp [%s] rootRank [%u]",
113 : group.c_str(), rankId, rankSize, serverIp.GetReadableIP(), localIp.GetReadableIP(), rootRank);
114 :
115 4 : OpRetryConnectionPtr conn;
116 4 : CHK_RET(GetInstance(group, conn, true));
117 4 : conn->SetGroup(group);
118 4 : if (conn->Init(rankId, rankSize, serverIp, serverPort, serverInfo.devId, localIp, rootRank) != HCCL_SUCCESS) {
119 3 : HCCL_ERROR(
120 : "[OpRetryConnection][Init] group[%s] rankId [%u] rankSize [%u] serverIp [%s] localIp [%s] rootRank [%u] failed",
121 : group.c_str(), rankId, rankSize, serverIp.GetReadableIP(), localIp.GetReadableIP(), rootRank);
122 3 : HCCL_ERROR("There maybe some reasons to cause this error:");
123 3 : HCCL_ERROR("1. The port may have been used so we will bind error. OpRetry used port range [%u-%u]",
124 : serverPort, serverPort + OP_RETRY_CONN_PORT_MAX_RANGE);
125 3 : HCCL_ERROR("2. Somebody may have already listen on those ports, so we connect to wrong server "
126 : "and we will meet 'Recv unmatched ack' error");
127 3 : HCCL_ERROR("You may can set system reserved port to avoid this error by:");
128 3 : HCCL_ERROR("sysctl -w net.ipv4.ip_local_reserved_ports=%u-%u",
129 : serverPort, serverPort + OP_RETRY_CONN_PORT_MAX_RANGE);
130 3 : return HCCL_E_INTERNAL;
131 : }
132 :
133 1 : return HCCL_SUCCESS;
134 6 : }
135 :
136 5 : HcclResult OpRetryConnection::GetInstance(const std::string &group, OpRetryConnectionPtr &conn, bool forceNew)
137 : {
138 : // instance_本身是个指针，需要lock_锁来保护
139 5 : std::lock_guard<std::mutex> lockGaurd(lock_);
140 5 : if (instance_ == nullptr) {
141 4 : instance_ = new (std::nothrow) UniversalConcurrentMap<std::string, OpRetryConnectionPtr>();
142 4 : CHK_PTR_NULL(instance_);
143 : }
144 :
145 5 : std::lock_guard<std::shared_timed_mutex> guard(instance_->GetMtx());
146 5 : if (forceNew || instance_->FindLockFree(group) == instance_->EndLockFree()) {
147 4 : OpRetryConnectionPtr tmpConn;
148 4 : EXCEPTION_CATCH(tmpConn = std::make_shared<OpRetryConnection>(), return HCCL_E_PTR);
149 4 : instance_->EmplaceLockFree(group, tmpConn);
150 4 : }
151 :
152 5 : auto it = instance_->FindLockFree(group);
153 5 : CHK_PRT_RET(it == instance_->EndLockFree(),
154 : HCCL_ERROR("[OpRetryConnection][GetInstance] create connection failed in group [%s]", group.c_str()), HCCL_E_MEMORY);
155 5 : conn = it->second;
156 5 : CHK_SMART_PTR_NULL(conn);
157 :
158 5 : return HCCL_SUCCESS;
159 5 : }
160 :
161 4 : HcclResult OpRetryConnection::DelInstance(const std::string &group)
162 : {
163 : // instance_本身是个指针，需要lock_锁来保护
164 4 : std::lock_guard<std::mutex> lockGuard(lock_);
165 4 : if (instance_) {
166 4 : instance_->Erase(group);
167 4 : if (instance_->Size() == 0) {
168 4 : delete instance_;
169 4 : instance_ = nullptr;
170 : }
171 : }
172 :
173 4 : return HCCL_SUCCESS;
174 4 : }
175 :
176 7 : HcclResult OpRetryConnection::Init(u32 rankId, u32 rankSize, const HcclIpAddress &serverIp, u32 serverPort,
177 : s32 serverDevId, const HcclIpAddress &localIp, u32 rootRank)
178 : {
179 7 : if (rankId >= rankSize || rankSize == 0 || serverIp.IsInvalid() || rootRank >= rankSize || localIp.IsInvalid()) {
180 0 : HCCL_ERROR(
181 : "[OpRetryConnection][Init] Invalid params, rankId [%u] rankSize [%u] serverIp [%s] localIp [%s] rootRank [%u]",
182 : rankId, rankSize, serverIp.GetReadableIP(), localIp.GetReadableIP(), rootRank);
183 0 : return HCCL_E_PARA;
184 : }
185 :
186 7 : rankId_ = rankId;
187 7 : rankSize_ = rankSize;
188 7 : serverIp_ = serverIp;
189 7 : serverPort_ = serverPort;
190 7 : localIp_ = localIp;
191 7 : rootRank_ = rootRank;
192 :
193 7 : HCCL_INFO("[OpRetryConnection][Init] rankId[%u] rankSize[%d] rootRank[%u] serverIp[%s:%u] serverDevId[%d]", rankId_,
194 : rankSize_, rootRank_, serverIp.GetReadableIP(), serverPort_, serverDevId);
195 :
196 7 : CHK_RET(InitHcclNet());
197 :
198 7 : if (IsRoot()) {
199 7 : CHK_PRT_RET(StartListen() != HCCL_SUCCESS,
200 : HCCL_ERROR("[OpRetryConnection][Init] Start listen failed, serverIp_[%s] serverPort_[%u]",
201 : serverIp_.GetReadableIP(), serverPort_), HCCL_E_TCP_CONNECT);
202 : }
203 :
204 7 : auto startTime = std::chrono::steady_clock::now();
205 7 : auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
206 7 : u32 retryConnectTimes = 0;
207 : do {
208 10 : CHK_PRT_RET((std::chrono::steady_clock::now() - startTime) > timeout,
209 : HCCL_ERROR("[OpRetryConnection][Init] Connect to server timeout [%ld s], serverIp_[%s] serverPort_[%u]",
210 : timeout, serverIp_.GetReadableIP(), serverPort_), HCCL_E_TCP_CONNECT);
211 :
212 10 : auto ret = Connect();
213 10 : if (ret == HCCL_SUCCESS) {
214 5 : break;
215 5 : } else if (ret == HCCL_E_AGAIN) {
216 3 : HCCL_ERROR("[OpRetryConnection][Init] Connect to server failed, serverIp_[%s] serverPort_[%u], we try again",
217 : serverIp_.GetReadableIP(), serverPort_);
218 3 : retryConnectTimes++;
219 3 : SaluSleep(TEN_MILLISECOND_OF_USLEEP);
220 3 : continue;
221 : } else {
222 2 : HCCL_ERROR("[OpRetryConnection][Init] Connect to server failed, serverIp_[%s] serverPort_[%u]",
223 : serverIp_.GetReadableIP(), serverPort_);
224 2 : return ret;
225 : }
226 3 : } while (true);
227 :
228 5 : if (retryConnectTimes > 0) {
229 1 : HCCL_ERROR("[OpRetryConnection][Init] Client reconnect %u times to success, "
230 : "so the above or this error log can be ignored", retryConnectTimes);
231 : }
232 :
233 5 : if (IsRoot()) {
234 5 : CHK_RET(WaitAcceptFinish());
235 : }
236 :
237 4 : HCCL_INFO("[OpRetryConnection][Init] success rankId[%u] rankSize[%d] rootRank[%u] serverIp[%s:%u]", rankId_,
238 : rankSize_, rootRank_, serverIp.GetReadableIP(), serverPort_);
239 :
240 4 : return HCCL_SUCCESS;
241 : }
242 :
243 12 : HcclResult OpRetryConnection::DeInit()
244 : {
245 12 : HCCL_INFO("[OpRetryConnection][Deinit] tag[%s] ready to deinit", tag_.c_str());
246 : /* 等待后台线程结束 */
247 12 : backgroudThreadStop_ = true;
248 12 : if (backgroudThread_.joinable()) {
249 7 : backgroudThread_.join();
250 : }
251 :
252 : /* 关闭所有连接 */
253 17 : for (auto &socket: connectionSockets_) {
254 5 : socket.second->Close();
255 : }
256 12 : connectionSockets_.clear();
257 12 : StopListen();
258 :
259 12 : if (socket_) {
260 5 : HCCL_INFO("[OpRetryConnection] close socket");
261 5 : socket_->Close();
262 5 : socket_ = nullptr;
263 : }
264 :
265 : /* 释放NetCtx资源 */
266 12 : if (serverNetCtx_) {
267 7 : HcclNetCloseDev(serverNetCtx_);
268 7 : serverNetCtx_ = nullptr;
269 : }
270 :
271 12 : if (clientNetCtx_) {
272 5 : HcclNetCloseDev(clientNetCtx_);
273 5 : clientNetCtx_ = nullptr;
274 : }
275 :
276 12 : if (hcclNetInit_) {
277 7 : HcclNetDeInit(NICDeployment::NIC_DEPLOYMENT_HOST, devicePhysicID_, deviceLogicalID_);
278 7 : hcclNetInit_ = false;
279 : }
280 :
281 12 : return HCCL_SUCCESS;
282 : }
283 :
284 1 : HcclResult OpRetryConnection::GetAgentSocket(std::shared_ptr<HcclSocket> &sock)
285 : {
286 1 : if (socket_) {
287 1 : sock = socket_;
288 1 : return HCCL_SUCCESS;
289 : }
290 :
291 0 : HCCL_ERROR("[OpRetryConnection][GetAgentSocket] socket is nullptr");
292 0 : return HCCL_E_UNAVAIL;
293 : }
294 :
295 1 : HcclResult OpRetryConnection::GetServerSockets(std::map<u32, std::shared_ptr<HcclSocket>> &socks)
296 : {
297 1 : if (!IsRoot()) {
298 0 : HCCL_ERROR("[OpRetryConnection][GetServerSockets] rank[%u] is not root rank [%u], so no server sockets",
299 : rankId_, rootRank_);
300 0 : return HCCL_E_UNAVAIL;
301 : }
302 :
303 1 : if (!connectionSockets_.empty() && connectionSockets_.size() == rankSize_) {
304 1 : socks = connectionSockets_;
305 1 : return HCCL_SUCCESS;
306 : }
307 :
308 0 : HCCL_ERROR("[OpRetryConnection][GetServerSockets] connection sockets count [%u] rankSize [%u]",
309 : connectionSockets_.size(), rankSize_);
310 0 : return HCCL_E_UNAVAIL;
311 : }
312 :
313 7 : HcclResult OpRetryConnection::InitHcclNet()
314 : {
315 7 : CHK_RET(hrtGetDevice(&deviceLogicalID_));
316 7 : CHK_RET(hrtGetDevicePhyIdByIndex(deviceLogicalID_, devicePhysicID_));
317 7 : HCCL_INFO("[OpRetryConnection][InitHcclNet] deviceLogicalID_[%d] devicePhysicID_[%u]", deviceLogicalID_, devicePhysicID_);
318 :
319 7 : CHK_RET(HcclNetInit(NICDeployment::NIC_DEPLOYMENT_HOST, devicePhysicID_, deviceLogicalID_, true));
320 7 : hcclNetInit_ = true;
321 :
322 7 : return HCCL_SUCCESS;
323 : }
324 :
325 1 : HcclResult OpRetryConnection::LoadHostWhiteList(const std::string &whiteListFile)
326 : {
327 1 : if (!whitelist_.empty()) {
328 0 : return HCCL_SUCCESS;
329 : }
330 :
331 1 : nlohmann::json fileContent;
332 1 : std::ifstream infile(whiteListFile.c_str(), std::ifstream::in);
333 1 : CHK_PRT_RET(!infile, HCCL_ERROR("[OpRetryConnection][LoadHostWhiteList]open file %s failed", whiteListFile.c_str()),
334 : HCCL_E_PARA);
335 :
336 : try {
337 1 : infile >> fileContent; // 将文件内容读取到json对象内
338 0 : } catch (...) {
339 0 : HCCL_ERROR("[OpRetryConnection][LoadHostWhiteList]load file[%s] to json fail. please check json file format.",
340 : whiteListFile.c_str());
341 0 : infile.close();
342 0 : return HCCL_E_INTERNAL;
343 0 : }
344 1 : infile.close();
345 :
346 1 : CHK_PRT_RET(fileContent.find("host_ip") == fileContent.end(),
347 : HCCL_ERROR("[OpRetryConnection][LoadHostWhiteList] whitelist don't have host_ip"), HCCL_E_INTERNAL);
348 1 : nlohmann::json hostWhitelist = fileContent["host_ip"];
349 2 : for (auto& ipJson : hostWhitelist) {
350 1 : std::string ipStr;
351 : try {
352 1 : ipStr = ipJson.get<std::string>();
353 0 : } catch (...) {
354 0 : HCCL_ERROR("[OpRetryConnection][LoadHostWhiteList]get ipStr from ipJson failed, please check host white list");
355 0 : return HCCL_E_PARA;
356 0 : }
357 1 : HcclIpAddress ip(ipStr);
358 1 : CHK_PRT_RET(ip.IsInvalid(),
359 : HCCL_ERROR("[OpRetryConnection][LoadHostWhiteList]string[%s] is invalid ip", ipStr.c_str()), HCCL_E_PARA);
360 1 : whitelist_.push_back(ip);
361 1 : }
362 :
363 1 : return HCCL_SUCCESS;
364 1 : }
365 :
366 7 : HcclResult OpRetryConnection::StartListen()
367 : {
368 7 : CHK_RET(HcclNetOpenDev(&serverNetCtx_, NicType::HOST_NIC_TYPE, devicePhysicID_, deviceLogicalID_, localIp_));
369 7 : CHK_PTR_NULL(serverNetCtx_);
370 :
371 7 : auto enableWhiteList_ = GetExternalInputHcclEnableWhitelist();
372 7 : if (enableWhiteList_ != 0) {
373 1 : CHK_RET(GetHostSocketWhiteList());
374 : }
375 :
376 7 : EXCEPTION_CATCH((listenSocket_ = std::make_shared<HcclSocket>(serverNetCtx_, serverPort_)), return HCCL_E_PTR);
377 7 : CHK_SMART_PTR_NULL(listenSocket_);
378 7 : CHK_RET(listenSocket_->Init());
379 7 : CHK_RET(listenSocket_->Listen());
380 :
381 7 : if (enableWhiteList_ != 0) {
382 1 : CHK_RET(AddListenSocketWhiteList());
383 : }
384 :
385 7 : HCCL_INFO("[OpRetryConnection] Server start with host ip[%s] and port[%u]", serverIp_.GetReadableAddress(), serverPort_);
386 :
387 : /* 拉起后台线程，在线程中进行异步接收，这样不会阻塞当前主线程，可以进行后续
388 : * 用户需要主动调用WaitAcceptFinished()去等待建链结束
389 : */
390 7 : acceptFinished_ = false;
391 7 : backgroudThreadStop_ = false;
392 7 : std::thread thread(&OpRetryConnection::RunAccept, this);
393 7 : backgroudThread_ = std::move(thread);
394 :
395 7 : return HCCL_SUCCESS;
396 7 : }
397 :
398 17 : HcclResult OpRetryConnection::StopListen()
399 : {
400 17 : if (listenSocket_) {
401 7 : HCCL_INFO("[OpRetryConnection] Server stop listen socket");
402 7 : if (enableWhitelist_ && !wlistInfosVec_.empty()) {
403 0 : listenSocket_->DelWhiteList(wlistInfosVec_);
404 0 : enableWhitelist_ = false;
405 0 : wlistInfosVec_.clear();
406 : }
407 :
408 7 : listenSocket_->DeInit();
409 7 : listenSocket_ = nullptr;
410 : }
411 :
412 17 : return HCCL_SUCCESS;
413 : }
414 :
415 7 : HcclResult OpRetryConnection::Accept()
416 : {
417 7 : auto startTime = std::chrono::steady_clock::now();
418 7 : auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
419 7 : auto tag = GetTag();
420 7 : std::shared_ptr<HcclSocket> socket = nullptr;
421 :
422 : /* 这里我们按照通信域中rank数目依次accept所有连接 */
423 7 : u32 expectAcceptNum = rankSize_;
424 21 : while (expectAcceptNum > 0) {
425 9 : if (backgroudThreadStop_) {
426 2 : HCCL_ERROR("[OpRetryConnection][Accept] OpRetryConnection in acceptting but stop, rankSize_[%u] accept link[%u]",
427 : rankSize_, rankSize_ - expectAcceptNum);
428 2 : return HCCL_E_INTERNAL;
429 : }
430 :
431 7 : if ((std::chrono::steady_clock::now() - startTime) > timeout) {
432 0 : HCCL_ERROR("[OpRetryConnection][Accept] OpRetryConnection accept timeout! timeout[%d s]",
433 : GetExternalInputHcclLinkTimeOut());
434 0 : return HCCL_E_TIMEOUT;
435 : }
436 :
437 7 : if (listenSocket_->Accept(tag, socket) == HCCL_SUCCESS) {
438 5 : HCCL_INFO("[OpRetryConnection][Accept] accept peer ip[%s]", socket->GetRemoteIp().GetReadableIP());
439 : // 因为socket是全双工的，所以Server与Client侧均可以先发送，再接收
440 5 : CHK_RET(SendAckInfo(socket));
441 5 : CHK_RET(RecvMetaInfo(socket));
442 5 : expectAcceptNum--;
443 : }
444 : }
445 :
446 5 : return HCCL_SUCCESS;
447 7 : }
448 :
449 5 : HcclResult OpRetryConnection::WaitAcceptFinish()
450 : {
451 5 : auto startTime = std::chrono::steady_clock::now();
452 5 : auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
453 :
454 : while (true) {
455 14685393 : if (backgroudThreadStop_) {
456 1 : HCCL_ERROR("[OpRetryConnection][WaitAcceptFinish] background thread stopped! may some error happened");
457 1 : return HCCL_E_INTERNAL;
458 : }
459 :
460 14685392 : if (acceptFinished_) {
461 4 : HCCL_INFO("[OpRetryConnection][WaitAcceptFinish] wait background thread accept finished");
462 4 : return HCCL_SUCCESS;
463 : }
464 :
465 14685388 : if ((std::chrono::steady_clock::now() - startTime) > timeout) {
466 0 : HCCL_ERROR("[OpRetryConnection][WaitAcceptFinish] wait accept timeout! timeout[%d s]",
467 : GetExternalInputHcclLinkTimeOut());
468 0 : return HCCL_E_TIMEOUT;
469 : }
470 : }
471 :
472 : HCCL_ERROR("[OpRetryConnection][WaitAcceptFinish] code should not execute in there");
473 : return HCCL_E_INTERNAL;
474 : }
475 :
476 5 : HcclResult OpRetryConnection::RecvMetaInfo(std::shared_ptr<HcclSocket> &peerSocket)
477 : {
478 5 : u32 peerRankId = INVALID_UINT;
479 5 : auto ret = peerSocket->Recv(&peerRankId, sizeof(peerRankId));
480 5 : CHK_PRT_RET(ret != HCCL_SUCCESS,
481 : HCCL_ERROR("[OpRetryConnection][RecvMetaInfo] Recv peer meta info failed. ret [%d]", ret), ret);
482 :
483 5 : CHK_PRT_RET(peerRankId >= rankSize_,
484 : HCCL_ERROR("[OpRetryConnection][RecvMetaInfo] Recv peer meta info invalid peerRankId [%u]. rankSize_[%u]",
485 : peerRankId, rankSize_), HCCL_E_INTERNAL);
486 :
487 5 : if (connectionSockets_.find(peerRankId) != connectionSockets_.end()) {
488 0 : HCCL_ERROR("[OpRetryConnection][RecvMetaInfo] Recv same rankId [%u]", peerRankId);
489 0 : return HCCL_E_INTERNAL;
490 : }
491 :
492 5 : connectionSockets_.insert({peerRankId, peerSocket});
493 5 : HCCL_INFO("[OpRetryConnection][RecvMetaInfo] Recv peer rankId [%u]. rankSize_[%u] success", peerRankId, rankSize_);
494 5 : return HCCL_SUCCESS;
495 : }
496 :
497 : /* 这里我们使用Server与Client侧约定好的rankSize信息作为Server侧的ACK报文 */
498 5 : HcclResult OpRetryConnection::SendAckInfo(std::shared_ptr<HcclSocket> &peerSocket)
499 : {
500 5 : auto ret = peerSocket->Send(&rankSize_, sizeof(rankSize_));
501 5 : CHK_PRT_RET(ret != HCCL_SUCCESS,
502 : HCCL_ERROR("[OpRetryConnection][SendAckInfo] Send peer meta ack failed, ret %d", ret), ret);
503 :
504 5 : HCCL_INFO("[OpRetryConnection][SendAckInfo] rank [%u] send ack [%u] success", rankId_, rankSize_);
505 5 : return HCCL_SUCCESS;
506 : }
507 :
508 7 : void OpRetryConnection::RunAccept()
509 : {
510 7 : SetThreadName("Hccl_RetryConn");
511 :
512 : /* 这里我们跟所有Common进程建链，包括自己 */
513 7 : if (Accept() != HCCL_SUCCESS) {
514 2 : HCCL_ERROR("[OpRetryConnection][Run] Accept failed, serverIp_[%s] serverPort_[%u]", serverIp_.GetReadableIP(), serverPort_);
515 2 : backgroudThreadStop_ = true;
516 2 : return;
517 : }
518 :
519 : /* 这里停止listen，是因为后续不再接收新的建链，同时也能够释放该端口，让后续通信域使用 */
520 5 : if (StopListen() != HCCL_SUCCESS) {
521 0 : HCCL_ERROR("[OpRetryConnection][Run] Stop listen failed, serverIp_[%s] serverPort_[%u]", serverIp_.GetReadableIP(), serverPort_);
522 0 : backgroudThreadStop_ = true;
523 0 : return;
524 : }
525 :
526 5 : acceptFinished_ = true;
527 5 : HCCL_INFO("[OpRetryConnection][Run] Accept all client success");
528 : }
529 :
530 6 : HcclResult OpRetryConnection::Connect()
531 : {
532 : // 因为Connect可能会被多次调用，因此clientNetCtx不需要多次创建
533 6 : if (clientNetCtx_ == nullptr) {
534 5 : CHK_RET(HcclNetOpenDev(&clientNetCtx_, NicType::HOST_NIC_TYPE, devicePhysicID_, deviceLogicalID_, localIp_));
535 5 : CHK_PTR_NULL(clientNetCtx_);
536 : }
537 :
538 : // 因为Connect可能会被多次调用，因此如果socket已经存在则直接关闭，并创建新的
539 6 : if (socket_ != nullptr) {
540 1 : socket_->Close();
541 1 : socket_ = nullptr;
542 : }
543 :
544 6 : auto tag = GetTag();
545 6 : EXCEPTION_CATCH((socket_ = std::make_shared<HcclSocket>(tag,
546 : clientNetCtx_, serverIp_, serverPort_, HcclSocketRole::SOCKET_ROLE_CLIENT)), return HCCL_E_PTR);
547 6 : CHK_SMART_PTR_NULL(socket_);
548 6 : CHK_RET(socket_->Init());
549 6 : CHK_RET(socket_->Connect());
550 :
551 6 : auto startTime = std::chrono::steady_clock::now();
552 6 : auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
553 : while (true) {
554 16 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
555 1 : HCCL_ERROR("[OpRetryConnection][Connect] Get socket timeout! timeout [%ld s]", timeout);
556 1 : return HCCL_E_TIMEOUT;
557 : }
558 :
559 15 : auto status = socket_->GetStatus();
560 15 : if (status == HcclSocketStatus::SOCKET_CONNECTING) {
561 10 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
562 10 : continue;
563 5 : } else if (status != HcclSocketStatus::SOCKET_OK) {
564 0 : HCCL_ERROR("[OpRetryConnection][Connect] Get socket failed, ret [%d]", status);
565 0 : return HCCL_E_TCP_CONNECT;
566 : } else {
567 5 : HCCL_INFO("[OpRetryConnection][Connect] Get socket success with server [%s] port [%u]",
568 : serverIp_.GetReadableIP(), serverPort_);
569 5 : break;
570 : }
571 10 : }
572 :
573 5 : CHK_RET(SendMetaInfo());
574 5 : CHK_RET(RecvAckInfo());
575 4 : return HCCL_SUCCESS;
576 6 : }
577 :
578 : /* 目前meta信息主要是rankId */
579 5 : HcclResult OpRetryConnection::SendMetaInfo()
580 : {
581 5 : auto ret = socket_->Send(&rankId_, sizeof(rankId_));
582 5 : CHK_PRT_RET(ret != HCCL_SUCCESS,
583 : HCCL_WARNING("[OpRetryConnection][SendMetaInfo] Send peer meta info failed, ret %d", ret),
584 : HCCL_E_AGAIN);
585 :
586 5 : HCCL_INFO("[OpRetryConnection][SendMetaInfo] rank [%u] send meta info success", rankId_);
587 5 : return HCCL_SUCCESS;
588 : }
589 :
590 4 : HcclResult OpRetryConnection::RecvAckInfo()
591 : {
592 4 : u32 ackRankSize = 0;
593 4 : auto ret = socket_->Recv(&ackRankSize, sizeof(ackRankSize));
594 4 : CHK_PRT_RET(ret != HCCL_SUCCESS,
595 : HCCL_WARNING("[OpRetryConnection][RecvAckTag] Recv peer ack info failed. ret [%d]", ret),
596 : HCCL_E_AGAIN);
597 :
598 4 : if (ackRankSize != rankSize_) {
599 0 : HCCL_ERROR("[OpRetryConnection][RecvAckTag] Recv unmatched ack [%u] expect [%u]", ackRankSize, rankSize_);
600 0 : return HCCL_E_INTERNAL;
601 : }
602 :
603 4 : HCCL_INFO("[OpRetryConnection][RecvAckTag] Recv ack [%u] success", ackRankSize);
604 4 : return HCCL_SUCCESS;
605 : }
606 :
607 1 : HcclResult OpRetryConnection::GetHostSocketWhiteList()
608 : {
609 1 : auto whiteListFile = GetExternalInputHcclWhiteListFile();
610 1 : CHK_PRT_RET((whiteListFile.length() == 0),
611 : HCCL_ERROR("[OpRetryConnection][GetHostSocketWhitelist]environment variable HCCL_WHITELIST_FILE is not set or not exist"), HCCL_E_PARA);
612 :
613 1 : HcclResult ret = LoadHostWhiteList(whiteListFile);
614 1 : CHK_PRT_RET(ret != HCCL_SUCCESS,
615 : HCCL_ERROR("[OpRetryConnection][GetHostSocketWhitelist]hccl whitelist load config file[%s] failed. ret[%u].",
616 : whiteListFile.c_str(), ret), ret);
617 :
618 1 : CHK_PRT_RET(whitelist_.empty(), HCCL_ERROR("[OpRetryConnection][GetHostSocketWhitelist]whitelist file[%s] have no valid host ip.",
619 : whiteListFile.c_str()), HCCL_E_UNAVAIL);
620 :
621 1 : HCCL_INFO("[OpRetry][GetHostSocketWhiteList]Get host socket whitelist success. there are %zu host ip in the whitelist.",
622 : whitelist_.size());
623 1 : return HCCL_SUCCESS;
624 1 : }
625 :
626 1 : HcclResult OpRetryConnection::AddListenSocketWhiteList()
627 : {
628 1 : wlistInfosVec_.clear();
629 2 : for (auto ip : whitelist_) {
630 : SocketWlistInfo wlistInfo;
631 1 : wlistInfo.connLimit = HOST_SOCKET_CONN_LIMIT;
632 1 : wlistInfo.remoteIp.addr = ip.GetBinaryAddress().addr;
633 1 : wlistInfo.remoteIp.addr6 = ip.GetBinaryAddress().addr6;
634 1 : std::string tag = GetTag();
635 1 : s32 sRet = memcpy_s(&wlistInfo.tag[0], sizeof(wlistInfo.tag), tag.c_str(), tag.size() + 1);
636 1 : if (sRet != EOK) {
637 0 : HCCL_ERROR("[OpRetryConnection][AddListenSocketWhiteList]memory copy failed. errorno[%d]", sRet);
638 0 : return HCCL_E_MEMORY;
639 : }
640 1 : wlistInfosVec_.push_back(wlistInfo);
641 1 : HCCL_INFO("[OpRetryConnection][AddListenSocketWhiteList] add white ip %s", ip.GetReadableIP());
642 1 : }
643 :
644 1 : CHK_RET(listenSocket_->AddWhiteList(wlistInfosVec_));
645 :
646 1 : HCCL_INFO("[OpRetryConnection][AddListenSocketWhiteList] add socket white list success. total: %zu", whitelist_.size());
647 1 : return HCCL_SUCCESS;
648 : }
649 :
650 0 : u32 OpRetryConnection::GetServerPort()
651 : {
652 0 : u32 serverPort = HCCL_INVALID_PORT;
653 0 : if (GetExternalInputHcclIfBasePort() == HCCL_INVALID_PORT) {
654 0 : serverPort = HOST_CONTROL_BASE_PORT;
655 : } else {
656 0 : serverPort = GetExternalInputHcclIfBasePort();
657 : }
658 :
659 0 : return serverPort;
660 : }
661 : }
|
