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