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