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 "hccl_comm_conn.h"
12 : #include <algorithm>
13 : #include "dlhal_function.h"
14 : #include "hccl_comm_conn_mgr.h"
15 : #include "transport_heterog_raw_roce.h"
16 :
17 : using namespace std;
18 :
19 : namespace hccl {
20 :
21 : static const string CONNECT_TAG = "COMMCONN_";
22 :
23 0 : HcclCommConn::HcclCommConn()
24 : {
25 0 : }
26 :
27 0 : HcclCommConn::~HcclCommConn()
28 : {
29 0 : HcclResult ret = HCCL_SUCCESS;
30 0 : if (role_ == SERVER_ROLE_SOCKET && isListen_) {
31 0 : (void)StopListen();
32 : }
33 :
34 0 : if (memBlocksManager_ != nullptr) {
35 0 : HcclResult ret = MrManager::GetInstance().ReleaseKey(memBlocksManager_->GetMemAddr(),
36 : memBlocksManager_->GetMemSize());
37 0 : if (ret != HCCL_SUCCESS) {
38 0 : HCCL_ERROR("Comm connection ip[%s], ReleaseKey failed!",
39 : HcclIpAddress(localAddr_.info.tcp.ipv4Addr).GetReadableIP());
40 : }
41 : }
42 :
43 0 : if (transport_.get() != nullptr && rdmaHandle_ != nullptr) {
44 0 : (void)MrManager::GetInstance().DeInit(rdmaHandle_);
45 : }
46 :
47 0 : if (transport_.get() != nullptr) {
48 0 : transport_->Deinit();
49 : }
50 :
51 : // 用户使用Connect()但是底层链路未建链成功场景使用abort强行停止
52 0 : if (role_ == CLIENT_ROLE_SOCKET && socketInfo_.fdHandle == nullptr) {
53 0 : ret = hrtRaSocketNonBlockBatchAbort(&connectInfo_, 1);
54 0 : if (ret != HCCL_SUCCESS) {
55 0 : HCCL_ERROR("hrtRaSocketNonBlockBatchAbort failed");
56 : }
57 : }
58 0 : if (socketHandle_ != nullptr) {
59 0 : (void)hrtRaSocketDeInitRef(socketHandle_);
60 0 : socketHandle_ = nullptr;
61 : }
62 :
63 0 : if (rdmaHandle_ != nullptr) {
64 0 : (void)HrtRaRdmaDeInitRef(rdmaHandle_, NO_USE);
65 0 : rdmaHandle_ = nullptr;
66 : }
67 :
68 0 : HcclCommConnMgr::GetInstance().DeleteConnectCommMap(remoteAddr_);
69 0 : }
70 :
71 0 : HcclResult HcclCommConn::SetAddr(HcclAddr &bindAddr, u32 opType)
72 : {
73 0 : if (opType == INIT_LOCAL_IP) {
74 0 : localAddr_ = bindAddr;
75 0 : } else if (opType == INIT_REMOTE_IP) {
76 0 : remoteAddr_ = bindAddr;
77 : } else {
78 0 : HCCL_ERROR("This op[%u] is not supported currently.", opType);
79 0 : return HCCL_E_PARA;
80 : }
81 :
82 0 : return HCCL_SUCCESS;
83 : }
84 :
85 : // 在client端，由于hccp接口不支持，当前Bind接口不支持指定socket的本地port
86 0 : HcclResult HcclCommConn::Bind(HcclAddr &bindAddr)
87 : {
88 0 : HcclResult ret = HCCL_SUCCESS;
89 : // 增加一个锁，防止同一个comm出现并发情况
90 0 : lock_guard<mutex> lock(bindMutex_);
91 0 : if (socketHandle_ != nullptr && rdmaHandle_ != nullptr) {
92 0 : HCCL_ERROR("Duplicate bind, please check!");
93 0 : return HCCL_E_PARA;
94 : }
95 :
96 0 : CHK_RET(SetAddr(bindAddr, INIT_LOCAL_IP));
97 :
98 0 : u32 &localIpv4Addr = localAddr_.info.tcp.ipv4Addr;
99 0 : HCCL_RUN_INFO("HcclCommConn Bind localIpv4Addr[%s], port[%u]",
100 : HcclIpAddress(localIpv4Addr).GetReadableIP(), localAddr_.info.tcp.port);
101 :
102 0 : struct rdev nicRdevInfo{};
103 0 : nicRdevInfo.phyId = devId_;
104 0 : nicRdevInfo.family = AF_INET;
105 0 : nicRdevInfo.localIp.addr.s_addr = localIpv4Addr;
106 :
107 0 : if (socketHandle_ == nullptr) {
108 0 : ret = hrtRaSocketInitRef(NETWORK_PEER_ONLINE, nicRdevInfo, socketHandle_);
109 0 : CHK_PTR_NULL(socketHandle_);
110 0 : if (ret != HCCL_SUCCESS) {
111 0 : HCCL_ERROR("hrtRaSocketInit failed! ip[%s], port[%u], ret[%d]",
112 : HcclIpAddress(localIpv4Addr).GetReadableIP(), localAddr_.info.tcp.port, ret);
113 0 : return HCCL_E_ROCE_CONNECT;
114 : }
115 : }
116 :
117 0 : ret = HrtRaRdmaInitRef(NETWORK_PEER_ONLINE, NO_USE, nicRdevInfo, rdmaHandle_);
118 0 : CHK_PRT_RET(ret == HCCL_E_AGAIN , HCCL_WARNING("HcclCommConn Bind rdma init need retry."), HCCL_E_AGAIN);
119 0 : CHK_PTR_NULL(rdmaHandle_);
120 0 : if (ret != HCCL_SUCCESS) {
121 0 : HCCL_ERROR("hrtRaRdmaInit failed! ip[%s], ret[%d]", HcclIpAddress(localIpv4Addr).GetReadableIP(), ret);
122 0 : return HCCL_E_ROCE_CONNECT;
123 : }
124 :
125 0 : return HCCL_SUCCESS;
126 0 : }
127 :
128 0 : HcclResult HcclCommConn::Listen(int backLog)
129 : {
130 0 : if (isListen_) {
131 0 : HCCL_ERROR("This conn has been listened ip[%s], port[%u]",
132 : HcclIpAddress(localAddr_.info.tcp.ipv4Addr).GetReadableIP(), localAddr_.info.tcp.port);
133 0 : return HCCL_E_PARA;
134 : }
135 :
136 0 : if (UNLIKELY(role_ == CLIENT_ROLE_SOCKET)) {
137 0 : HCCL_ERROR("this HcclCommConn has been configured as client, cannot use listen as server.");
138 0 : return HCCL_E_INTERNAL;
139 : }
140 :
141 0 : CHK_PTR_NULL(socketHandle_);
142 : struct SocketListenInfoT serverInfo;
143 0 : serverInfo.socketHandle = socketHandle_;
144 0 : serverInfo.port = localAddr_.info.tcp.port;
145 0 : HCCL_RUN_INFO("HcclCommConn Listen localIpv4Addr[%s], port[%u]",
146 : HcclIpAddress(localAddr_.info.tcp.ipv4Addr).GetReadableIP(), localAddr_.info.tcp.port);
147 0 : HcclResult ret = hrtRaSocketNonBlockListenStart(&serverInfo, 1);
148 0 : std::string errormessage = "The IP address " + std::string(HcclIpAddress(localAddr_.info.tcp.ipv4Addr).GetReadableIP()) +
149 0 : " and port " + std::to_string(localAddr_.info.tcp.port) + " have already been bound.";
150 0 : RPT_INPUT_ERR(ret == HCCL_E_UNAVAIL, "EI0019", std::vector<std::string>({"reason"}),
151 : std::vector<std::string>({errormessage}));
152 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("HcclCommConn start listen socket fail. "), ret);
153 0 : CHK_RET(hrtRaSocketAcceptCreditAdd(&serverInfo, 1, MAX_CONCURRENCY_LINK_NUM));
154 0 : isListen_ = true;
155 0 : return HCCL_SUCCESS;
156 0 : }
157 :
158 0 : HcclResult HcclCommConn::StopListen()
159 : {
160 : struct SocketListenInfoT serverInfo;
161 0 : serverInfo.socketHandle = socketHandle_;
162 0 : serverInfo.port = localAddr_.info.tcp.port;
163 0 : CHK_RET(hrtRaSocketListenStop(&serverInfo, 1));
164 0 : isListen_ = false;
165 0 : HCCL_RUN_INFO("HcclCommConn ip[%s] port[%u] StopListen success.",
166 : HcclIpAddress(localAddr_.info.tcp.ipv4Addr).GetReadableIP(), localAddr_.info.tcp.port);
167 0 : return HCCL_SUCCESS;
168 : }
169 :
170 0 : HcclResult HcclCommConn::Accept(HcclAddr &acceptAddr, HcclCommConn *&acceptConn)
171 : {
172 0 : HcclResult ret = HCCL_SUCCESS;
173 0 : AcceptCommConn acceptComConn;
174 0 : std::queue<AcceptCommConn> connHandleTmpQueue{};
175 0 : bool isNeedCreditAdd = false;
176 0 : u32 creditNum = 0;
177 :
178 0 : std::unique_lock<std::mutex> lock(connHandleQueueMutex_);
179 0 : if (connHandleQueue_.size() == MAX_CONCURRENCY_LINK_NUM) {
180 0 : HCCL_RUN_WARNING("The maximum number of concurrent link setups is %u. cur link num[%u]",
181 : MAX_CONCURRENCY_LINK_NUM, connHandleQueue_.size());
182 0 : ret = HCCL_E_AGAIN;
183 0 : } else if (HcclCommConnMgr::GetInstance().IsExceedMaxLinkNum(SERVER_ROLE_SOCKET)) {
184 0 : HCCL_RUN_WARNING("The maximum number of communication connections that can be created is %u.",
185 : MAX_CONN_LINK_NUM);
186 0 : ret = HCCL_E_AGAIN;
187 : } else {
188 0 : ret = PrepareSocketInfoForServer(acceptComConn.socketInfo);
189 0 : if (ret != HCCL_SUCCESS) {
190 0 : return ret;
191 : }
192 0 : ret = GetSocket(acceptComConn.socketInfo);
193 0 : if (ret != HCCL_SUCCESS && ret != HCCL_E_AGAIN) {
194 0 : HCCL_ERROR("HcclCommConn Accept GetSocket fail error[%d]", ret);
195 0 : return HCCL_E_TCP_CONNECT;
196 0 : } else if (ret == HCCL_SUCCESS) {
197 0 : HCCL_RUN_INFO("Server Got new socket, ipv4Addr[%s] socketHandle[%p] fdHandle[%p]",
198 : HcclIpAddress(acceptComConn.socketInfo.remoteIp.addr.s_addr).GetReadableIP(),
199 : acceptComConn.socketInfo.socketHandle, acceptComConn.socketInfo.fdHandle);
200 0 : acceptComConn.newCommConn = new(nothrow) HcclCommConn();
201 0 : CHK_PTR_NULL(acceptComConn.newCommConn);
202 0 : acceptComConn.newCommConn->SetStartTime();
203 0 : connHandleQueue_.push(acceptComConn);
204 : }
205 : }
206 :
207 0 : while (!connHandleQueue_.empty()) {
208 0 : acceptComConn = connHandleQueue_.front();
209 0 : connHandleQueue_.pop();
210 0 : ret = acceptComConn.newCommConn->InitTransport(role_, localAddr_, acceptComConn.socketInfo);
211 0 : if (ret == HCCL_SUCCESS) {
212 0 : acceptConn = acceptComConn.newCommConn;
213 0 : acceptComConn.newCommConn = nullptr;
214 0 : acceptAddr = acceptConn->GetRemoteAddr();
215 0 : isNeedCreditAdd = true;
216 0 : creditNum++;
217 0 : HCCL_RUN_INFO("Server Got new socket finally, ipv4Addr[%s], port[%u]",
218 : HcclIpAddress(acceptAddr.info.tcp.ipv4Addr).GetReadableIP(), acceptAddr.info.tcp.port);
219 0 : break;
220 0 : } else if (ret != HCCL_E_AGAIN) {
221 0 : HCCL_RUN_WARNING("Accept Error Result[%d], Need Reset Conn ipv4Addr[%s]",
222 : ret, HcclIpAddress(acceptComConn.socketInfo.remoteIp.addr.s_addr).GetReadableIP());
223 0 : CHK_RET(ResetCurrentErrorConnection(acceptComConn.newCommConn));
224 0 : isNeedCreditAdd = true;
225 0 : creditNum++;
226 0 : break;
227 : } else {
228 : // 增加防吊死功能
229 0 : auto endTime = std::chrono::steady_clock::now();
230 0 : std::chrono::time_point<std::chrono::steady_clock> startTime;
231 0 : acceptComConn.newCommConn->GetStartTime(startTime);
232 0 : auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime).count();
233 0 : if (duration > ACCEPT_MAX_TIME) {
234 0 : HCCL_RUN_WARNING("accept time duration > %ums, Need Reset Conn ipv4Addr[%s]",
235 : ACCEPT_MAX_TIME, HcclIpAddress(acceptComConn.socketInfo.remoteIp.addr.s_addr).GetReadableIP());
236 0 : CHK_RET(ResetCurrentErrorConnection(acceptComConn.newCommConn));
237 0 : isNeedCreditAdd = true;
238 0 : creditNum++;
239 0 : continue;
240 0 : }
241 0 : connHandleTmpQueue.push(acceptComConn);
242 : }
243 : }
244 :
245 0 : while (!connHandleTmpQueue.empty()) {
246 0 : connHandleQueue_.push(connHandleTmpQueue.front());
247 0 : connHandleTmpQueue.pop();
248 : }
249 :
250 0 : if (isNeedCreditAdd) {
251 : // 当建链成功、qp交换信息返回不可恢复错误、触发防吊死三种情况都需要进程accept credit add
252 : struct SocketListenInfoT serverInfo;
253 0 : serverInfo.socketHandle = socketHandle_;
254 0 : serverInfo.port = localAddr_.info.tcp.port;
255 0 : CHK_RET(hrtRaSocketAcceptCreditAdd(&serverInfo, 1, creditNum));
256 : }
257 0 : return ret;
258 0 : }
259 :
260 0 : HcclResult HcclCommConn::ResetCurrentErrorConnection(HcclCommConn *&newCommConn)
261 : {
262 0 : if (newCommConn == nullptr) {
263 0 : HCCL_INFO("No Connection is being processed.");
264 0 : return HCCL_SUCCESS;
265 : }
266 :
267 0 : if (transport_ != nullptr) {
268 0 : transport_->SetForceClose();
269 : }
270 0 : delete newCommConn;
271 0 : newCommConn = nullptr;
272 :
273 0 : return HCCL_SUCCESS;
274 : }
275 :
276 0 : void HcclCommConn::SetForceClose()
277 : {
278 0 : if (transport_ != nullptr) {
279 0 : transport_->SetForceClose();
280 : }
281 0 : }
282 :
283 0 : const HcclAddr &HcclCommConn::GetRemoteAddr() const
284 : {
285 0 : return remoteAddr_;
286 : }
287 :
288 0 : HcclResult HcclCommConn::PrepareSocketInfoForServer(struct SocketInfoT &socketInfo)
289 : {
290 0 : string linkTag = CONNECT_TAG + to_string(0) + "_" + to_string(localAddr_.info.tcp.ipv4Addr) +
291 0 : "_" + to_string(localAddr_.info.tcp.port);
292 :
293 0 : socketInfo.socketHandle = socketHandle_;
294 0 : socketInfo.fdHandle = nullptr;
295 0 : socketInfo.status = CONNECT_FAIL;
296 0 : CHK_SAFETY_FUNC_RET(strncpy_s(socketInfo.tag, SOCK_CONN_TAG_SIZE, linkTag.c_str(), linkTag.length() + 1));
297 0 : return HCCL_SUCCESS;
298 0 : }
299 :
300 0 : HcclResult HcclCommConn::GetSocket(struct SocketInfoT &socketInfo)
301 : {
302 0 : u32 connectedNum = 0;
303 :
304 0 : HcclResult ret = hrtRaNonBlockGetSockets(role_, &socketInfo, 1, &connectedNum);
305 0 : if (ret == HCCL_SUCCESS) {
306 0 : if (connectedNum == 0) {
307 0 : ret = HCCL_E_AGAIN;
308 0 : } else if (connectedNum != 1 || socketInfo.status != CONNECT_OK || socketInfo.fdHandle == nullptr) {
309 0 : HCCL_ERROR("GetSocket fail linkTag linkTag[%s], connectedNum[%u] != 1, status[%d] != CONNECT_OK, "
310 : "or fdHandle is nullptr", socketInfo.tag, connectedNum, socketInfo.status);
311 0 : return HCCL_E_TCP_CONNECT;
312 : }
313 : }
314 :
315 0 : if (ret == HCCL_E_AGAIN) {
316 0 : SaluSleep(DELAY_TIME);
317 : }
318 :
319 0 : return ret;
320 : }
321 :
322 0 : HcclResult HcclCommConn::PrepareConnectSocketInfoForClient(HcclAddr &bindAddr)
323 : {
324 0 : CHK_RET(SetAddr(bindAddr, INIT_REMOTE_IP));
325 :
326 0 : HcclIpAddress remoteIp(remoteAddr_.info.tcp.ipv4Addr);
327 0 : string linkTag = CONNECT_TAG + to_string(0) + "_" + to_string(remoteAddr_.info.tcp.ipv4Addr) +
328 0 : "_" + to_string(remoteAddr_.info.tcp.port);
329 :
330 0 : connectInfo_.socketHandle = socketHandle_;
331 0 : connectInfo_.remoteIp.addr = remoteIp.GetBinaryAddress().addr;
332 0 : connectInfo_.remoteIp.addr6 = remoteIp.GetBinaryAddress().addr6;
333 0 : connectInfo_.port = remoteAddr_.info.tcp.port;
334 0 : CHK_SAFETY_FUNC_RET(strncpy_s(connectInfo_.tag, SOCK_CONN_TAG_SIZE, linkTag.c_str(), linkTag.length() + 1));
335 :
336 0 : socketInfo_.socketHandle = socketHandle_;
337 0 : socketInfo_.fdHandle = nullptr;
338 0 : socketInfo_.remoteIp.addr.s_addr = remoteAddr_.info.tcp.ipv4Addr;
339 0 : socketInfo_.status = CONNECT_FAIL;
340 0 : CHK_SAFETY_FUNC_RET(strncpy_s(socketInfo_.tag, SOCK_CONN_TAG_SIZE, linkTag.c_str(), linkTag.length() + 1));
341 0 : return HCCL_SUCCESS;
342 0 : }
343 :
344 0 : HcclResult HcclCommConn::InitMsgAndRequestBuffer()
345 : {
346 : {
347 0 : lock_guard<mutex> lock(msgInfosMutex_);
348 0 : if (msgInfosMem_ == nullptr) {
349 0 : msgInfosMem_.reset(new (nothrow) LocklessRingMemoryAllocate<HcclMessageInfo>(RESOURCE_MEMORY_CAPACITY));
350 0 : CHK_SMART_PTR_NULL(msgInfosMem_);
351 0 : CHK_RET(msgInfosMem_->Init());
352 0 : HCCL_INFO("InitRecvMsgBuffer Success!");
353 : }
354 0 : }
355 :
356 : {
357 0 : lock_guard<mutex> lock(reqInfosMutex_);
358 0 : if (reqInfosMem_ == nullptr) {
359 0 : reqInfosMem_.reset(new (nothrow) LocklessRingMemoryAllocate<HcclRequestInfo>(RESOURCE_MEMORY_CAPACITY));
360 0 : CHK_SMART_PTR_NULL(reqInfosMem_);
361 0 : CHK_RET(reqInfosMem_->Init());
362 0 : HCCL_INFO("InitRequestBuffer Success!");
363 : }
364 0 : }
365 :
366 0 : return HCCL_SUCCESS;
367 : }
368 :
369 0 : HcclResult HcclCommConn::InitMemBlocksAndRecvWrMem()
370 : {
371 0 : u32 memBlockNum = MEM_BLOCK_CAPACITY; // MEM_BLOCK_NUM_BIGER
372 0 : u32 info = 0;
373 0 : CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
374 0 : CHK_RET(hrtDrvGetPlatformInfo(&info));
375 :
376 : // 初始化信封内存
377 0 : if (memBlocksManager_ == nullptr) {
378 0 : memBlocksManager_.reset(new (nothrow) HeterogMemBlocksManager());
379 0 : CHK_SMART_PTR_NULL(memBlocksManager_);
380 0 : CHK_RET(memBlocksManager_->Init(memBlockNum));
381 : }
382 :
383 : // 初始化wr内存
384 : {
385 0 : lock_guard<mutex> lock(recvWrInfosMutex_);
386 0 : if (recvWrInfosMem_ == nullptr) {
387 0 : recvWrInfosMem_.reset(new (nothrow) LocklessRingMemoryAllocate<RecvWrInfo>(RESOURCE_MEMORY_CAPACITY));
388 0 : CHK_SMART_PTR_NULL(recvWrInfosMem_);
389 0 : CHK_RET(recvWrInfosMem_->Init());
390 : }
391 0 : }
392 :
393 : // 注册mr
394 0 : CHK_RET(MrManager::GetInstance().GetKey(memBlocksManager_->GetMemAddr(),
395 : memBlocksManager_->GetMemSize(), transportResourceInfo_.lkey));
396 0 : HCCL_INFO("InitMemBlocksAndRecvWrMem Success!");
397 :
398 0 : return HCCL_SUCCESS;
399 : }
400 :
401 0 : HcclResult HcclCommConn::InitTransport(u32 role, HcclAddr &localAddr, SocketInfoT &tmpInfo)
402 : {
403 0 : if (transport_ != nullptr) {
404 0 : return transport_->CheckAndPushBuildLink();
405 : }
406 :
407 0 : if (role == SERVER_ROLE_SOCKET) {
408 0 : role_ = role;
409 0 : localAddr_ = localAddr;
410 0 : remoteAddr_.info.tcp.ipv4Addr = tmpInfo.remoteIp.addr.s_addr;
411 0 : remoteAddr_.info.tcp.port = 0; // 不感知对端端口号，默认填0
412 :
413 0 : struct rdev nicRdevInfo{};
414 0 : nicRdevInfo.phyId = devId_;
415 0 : nicRdevInfo.family = AF_INET;
416 0 : nicRdevInfo.localIp.addr.s_addr = localAddr_.info.tcp.ipv4Addr;
417 0 : CHK_RET(hrtRaSocketInitRef(NETWORK_PEER_ONLINE, nicRdevInfo, socketHandle_));
418 0 : CHK_RET(HrtRaRdmaInitRef(NETWORK_PEER_ONLINE, NO_USE, nicRdevInfo, rdmaHandle_));
419 : }
420 :
421 0 : if (localAddr_.type != HCCL_ADDR_TYPE_ROCE) {
422 0 : HCCL_ERROR("HcclCommConn: This type[%d] is not supported currently.", localAddr_.type);
423 0 : return HCCL_E_NOT_SUPPORT;
424 : }
425 :
426 0 : CHK_RET(MrManager::GetInstance().Init(rdmaHandle_));
427 0 : CHK_RET(InitMsgAndRequestBuffer());
428 0 : CHK_RET(InitMemBlocksAndRecvWrMem());
429 :
430 0 : const string &linkTag = CONNECT_TAG;
431 0 : HcclIpAddress selfIp(localAddr_.info.tcp.ipv4Addr);
432 0 : HcclIpAddress peerIp(remoteAddr_.info.tcp.ipv4Addr);
433 :
434 0 : HCCL_RUN_INFO("role[%u], local ipv4[%s], port[%u], remote ipv4[%s], port[%u] init TransportRoce", role_,
435 : HcclIpAddress(localAddr_.info.tcp.ipv4Addr).GetReadableIP(), localAddr_.info.tcp.port,
436 : HcclIpAddress(remoteAddr_.info.tcp.ipv4Addr).GetReadableIP(), remoteAddr_.info.tcp.port);
437 :
438 0 : transportResourceInfo_.isRawConn = true;
439 0 : EXCEPTION_CATCH((transport_ = make_unique<TransportHeterogRawRoce>(linkTag, selfIp, peerIp,
440 : remoteAddr_.info.tcp.port, localAddr_.info.tcp.port, transportResourceInfo_)), return HCCL_E_PTR);
441 :
442 0 : CHK_SMART_PTR_NULL(transport_);
443 0 : CHK_RET(transport_->Init(tmpInfo, rdmaHandle_, &MrManager::GetInstance()));
444 :
445 0 : return transport_->CheckAndPushBuildLink();
446 0 : }
447 :
448 0 : HcclResult HcclCommConn::Connect(HcclAddr &connectAddr)
449 : {
450 0 : if (UNLIKELY(isListen_)) {
451 0 : HCCL_ERROR("this HcclCommConn has been listened as server, cannot use connect as client.");
452 0 : return HCCL_E_INTERNAL;
453 : }
454 :
455 0 : HcclResult ret = HCCL_SUCCESS;
456 0 : switch (connectState_) {
457 0 : case OpStatus::START:
458 0 : role_ = CLIENT_ROLE_SOCKET;
459 0 : ret = PrepareConnectSocketInfoForClient(connectAddr);
460 0 : if (ret != HCCL_SUCCESS) {
461 0 : break;
462 : }
463 : [[fallthrough]];
464 : case OpStatus::CONNECT:
465 0 : connectState_ = OpStatus::CONNECT;
466 0 : ret = hrtRaSocketNonBlockBatchConnect(&connectInfo_, 1);
467 0 : if (ret != HCCL_SUCCESS) {
468 0 : break;
469 : }
470 : [[fallthrough]];
471 : case OpStatus::GETSOCKET:
472 0 : connectState_ = OpStatus::GETSOCKET;
473 0 : ret = GetSocket(socketInfo_);
474 0 : if (ret != HCCL_SUCCESS) {
475 0 : break;
476 : }
477 : [[fallthrough]];
478 : case OpStatus::BUILDTRANSPORT:
479 0 : connectState_ = OpStatus::BUILDTRANSPORT;
480 0 : ret = InitTransport(role_, localAddr_, socketInfo_);
481 0 : if (ret == HCCL_SUCCESS) {
482 0 : connectState_ = OpStatus::END;
483 : }
484 0 : break;
485 0 : case OpStatus::END:
486 0 : HCCL_WARNING("Connect: This conn has been Connected ip[%s]",
487 : HcclIpAddress(localAddr_.info.tcp.ipv4Addr).GetReadableIP());
488 0 : break;
489 0 : default:
490 0 : HCCL_ERROR("Connect: op Invalid connectState[%u].", connectState_);
491 0 : return HCCL_E_INTERNAL;
492 : }
493 :
494 0 : HCCL_DEBUG("Connect: op connectState[%u] ret[%d].", connectState_, ret);
495 0 : return ret;
496 : }
497 :
498 0 : HcclResult HcclCommConn::Isend(const void* buf, int count, HcclDataType dataType, HcclRequest &request)
499 : {
500 0 : CheckDataType(dataType);
501 :
502 0 : if ((buf == nullptr) && (count != 0)) {
503 0 : HCCL_ERROR("[Check][Buffer]errNo[0x%016llx] or count[%d] is invalid",
504 : HCCL_ERROR_CODE(HCCL_E_PARA), count);
505 0 : return HCCL_E_PARA;
506 : }
507 :
508 0 : CHK_PRT_RET(transport_ == nullptr,
509 : HCCL_ERROR("[Get][transportPtr]errNo[0x%016llx] transportPtr is nullptr", HCCL_ERROR_CODE(HCCL_E_PARA)),
510 : HCCL_E_PARA);
511 :
512 0 : TransportEndPointInfo srcEp(0, DEFAULT_LOCAL_RANK, DEFAULT_TAG);
513 0 : TransportEndPointInfo dstEp(0, DEFAULT_REMOTE_RANK, DEFAULT_TAG);
514 0 : TransportEndPointParam epParam(srcEp, dstEp);
515 :
516 0 : TransData sendData(reinterpret_cast<u64>(buf), reinterpret_cast<u64>(nullptr), count, dataType, false, 0);
517 0 : HcclRequestInfo* requestHandle = nullptr;
518 0 : CHK_RET(transport_->Isend(sendData, epParam, requestHandle));
519 0 : request = requestHandle;
520 0 : return HCCL_SUCCESS;
521 : }
522 :
523 0 : HcclResult HcclCommConn::Improbe(int &flag, HcclMessage &msg, HcclStatus &status)
524 : {
525 0 : CHK_PRT_RET(transport_ == nullptr,
526 : HCCL_ERROR("[Get][transportPtr]errNo[0x%016llx] transportPtr is nullptr", HCCL_ERROR_CODE(HCCL_E_PARA)),
527 : HCCL_E_PARA);
528 :
529 0 : TransportEndPointInfo srcEp(0, DEFAULT_REMOTE_RANK, DEFAULT_TAG);
530 0 : TransportEndPointInfo dstEp(0, DEFAULT_LOCAL_RANK, DEFAULT_TAG);
531 0 : TransportEndPointParam epParam(srcEp, dstEp);
532 0 : HcclMessageInfo *msgHandle = nullptr;
533 :
534 0 : transport_->Improbe(epParam, flag, msgHandle, status);
535 0 : msg = msgHandle;
536 0 : return HCCL_SUCCESS;
537 : }
538 :
539 0 : HcclResult HcclCommConn::Imrecv(void* buf, int count, HcclDataType dataType, HcclMessage msg, HcclRequest &request)
540 : {
541 0 : CheckDataType(dataType);
542 :
543 0 : HcclMessageInfo* msgHandle = static_cast<HcclMessageInfo *>(msg);
544 0 : CHK_PTR_NULL(msgHandle);
545 0 : CHK_PRT_RET(transport_ == nullptr,
546 : HCCL_ERROR("[Get][transportPtr]errNo[0x%016llx] transportPtr is nullptr", HCCL_ERROR_CODE(HCCL_E_PARA)),
547 : HCCL_E_PARA);
548 :
549 0 : HcclRequestInfo* requestHandle = nullptr;
550 0 : TransData recvData(reinterpret_cast<u64>(nullptr), reinterpret_cast<u64>(buf), count, dataType);
551 0 : CHK_RET(transport_->Imrecv(recvData, *msgHandle, requestHandle));
552 0 : request = requestHandle;
553 0 : return HCCL_SUCCESS;
554 : }
555 :
556 0 : HcclResult HcclCommConn::ImrecvScatter(void *buf[], int count[], int bufCount, HcclDataType datatype, HcclMessage msg,
557 : HcclRequest &request)
558 : {
559 0 : CheckDataType(datatype);
560 :
561 0 : HcclMessageInfo *msgHandle = static_cast<HcclMessageInfo *>(msg);
562 0 : CHK_PTR_NULL(msgHandle);
563 0 : CHK_PRT_RET(transport_ == nullptr,
564 : HCCL_ERROR("[Get][transportPtr]errNo[0x%016llx] transportPtr is nullptr", HCCL_ERROR_CODE(HCCL_E_PARA)),
565 : HCCL_E_PARA);
566 :
567 0 : HcclRequestInfo *requestHandle = nullptr;
568 0 : CHK_RET(transport_->ImrecvScatter(buf, count, bufCount, datatype, *msgHandle, requestHandle));
569 0 : request = requestHandle;
570 0 : return HCCL_SUCCESS;
571 : }
572 :
573 0 : HcclResult HcclCommConn::Test(HcclRequest requestHandle, s32 &flag, HcclStatus &compState)
574 : {
575 0 : HcclRequestInfo *request = reinterpret_cast<HcclRequestInfo *>(requestHandle);
576 0 : CHK_PTR_NULL(request->transportHandle);
577 :
578 0 : TransportHeterog *transportPtr = reinterpret_cast<TransportHeterog *>(request->transportHandle);
579 0 : return transportPtr->Test(*request, flag, compState);
580 : }
581 :
582 0 : HcclResult HcclCommConn::CheckDataType(const HcclDataType dataType)
583 : {
584 0 : if ((dataType >= HCCL_DATA_TYPE_RESERVED) || (dataType < HCCL_DATA_TYPE_INT8)) {
585 0 : HCCL_ERROR("[Check][DataType]errNo[0x%016llx] data type[%s] not supported",
586 : HCCL_ERROR_CODE(HCCL_E_NOT_SUPPORT), GetDataTypeEnumStr(dataType).c_str());
587 0 : return HCCL_E_NOT_SUPPORT;
588 : }
589 0 : return HCCL_SUCCESS;
590 : }
591 :
592 0 : HcclResult HcclCommConn::SocketForceClose(SocketInfoT &socketInfo)
593 : {
594 0 : if (socketInfo.socketHandle == nullptr || socketInfo.fdHandle == nullptr) {
595 0 : HCCL_ERROR("SocketForceClose socketInfo is invalid socketHandle[%p] fdHandle[%p]",
596 : socketInfo.socketHandle, socketInfo.fdHandle);
597 0 : return HCCL_E_PARA;
598 : }
599 :
600 0 : SocketCloseInfoT conns[1]{};
601 0 : conns[0].socketHandle = socketInfo.socketHandle;
602 0 : conns[0].fdHandle = socketInfo.fdHandle;
603 0 : conns[0].disuseLinger = static_cast<s32>(true);
604 :
605 0 : HcclResult ret = hrtRaSocketBatchClose(conns, 1);
606 0 : if (ret != HCCL_SUCCESS) {
607 0 : HCCL_ERROR("SocketForceClose ra socket batch close failed socketHandle[%p] fdHandle[%p]",
608 : socketInfo.socketHandle, socketInfo.fdHandle);
609 0 : return ret;
610 : }
611 0 : socketInfo.socketHandle = nullptr;
612 0 : socketInfo.fdHandle = nullptr;
613 0 : return HCCL_SUCCESS;
614 : }
615 :
616 0 : void HcclCommConn::SetStartTime()
617 : {
618 0 : startTime_ = chrono::steady_clock::now();
619 0 : }
620 :
621 0 : void HcclCommConn::GetStartTime(std::chrono::time_point<std::chrono::steady_clock> &startTime)
622 : {
623 0 : startTime = startTime_;
624 0 : }
625 : }
|
