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 <fcntl.h>
12 : #include <sys/ioctl.h>
13 : #include <sys/mman.h>
14 : #include <stdlib.h>
15 : #include <unistd.h>
16 : #include <errno.h>
17 : #include "securec.h"
18 : #include "dl_hal_function.h"
19 : #include "ra_comm.h"
20 : #include "ra_rs_comm.h"
21 : #include "ra_rs_err.h"
22 : #include "rs.h"
23 : #include "ra_peer_nda.h"
24 : #include "ra_peer.h"
25 :
26 : #define PAGE_SHIFT 12
27 : int gNotifyFd = -1;
28 :
29 : static pthread_mutex_t gRaPeerMutex[RA_MAX_PHY_ID_NUM];
30 : int gRaInitCounter[RA_MAX_PHY_ID_NUM] = {0};
31 :
32 58 : void RaPeerMutexLock(unsigned int phyId)
33 : {
34 58 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[phyId]);
35 58 : }
36 :
37 58 : void RaPeerMutexUnlock(unsigned int phyId)
38 : {
39 58 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
40 58 : }
41 :
42 3 : int RaPeerSocketBatchClose(unsigned int devId, struct SocketCloseInfoT conn[], unsigned int num)
43 : {
44 : int ret;
45 : unsigned int i;
46 : int disuseLinger;
47 3 : unsigned int index = 0;
48 3 : unsigned int closeNum = 0;
49 : struct RsSocketCloseInfoT closeInfo[MAX_SOCKET_NUM];
50 :
51 3 : ret = memset_s(closeInfo, sizeof(struct RsSocketCloseInfoT) * MAX_SOCKET_NUM, 0,
52 : sizeof(struct RsSocketCloseInfoT) * MAX_SOCKET_NUM);
53 3 : CHK_PRT_RETURN(ret != 0,
54 : hccp_err("[batch_close][ra_peer_socket]memset_s close_info failed, ret(%d), phyId(%u)", ret, devId),
55 : -ESAFEFUNC);
56 :
57 4 : for (i = 0; i < num; i++) {
58 2 : if (conn[i].fdHandle != NULL) {
59 2 : closeInfo[closeNum].fd = ((struct SocketPeerInfo *)(conn[i].fdHandle))->fd;
60 2 : ++closeNum;
61 : }
62 : }
63 :
64 : // use attr disuse_linger of the fist conn as the common attr for all(0 by default)
65 2 : disuseLinger = conn[0].disuseLinger;
66 :
67 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[devId]);
68 2 : RsSetCtx(devId);
69 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[devId]);
70 2 : ret = RsSocketBatchClose(disuseLinger, &closeInfo[index], closeNum);
71 2 : if (ret != 0) {
72 1 : hccp_err("[batch_close][ra_peer_socket]ra close failed ret(%d), phyId(%u)", ret, devId);
73 : }
74 :
75 4 : for (i = 0; i < num; i++) {
76 2 : if (conn[i].fdHandle != NULL) {
77 2 : free(conn[i].fdHandle);
78 2 : conn[i].fdHandle = NULL;
79 : }
80 : }
81 2 : return ret;
82 : }
83 :
84 3 : int RaPeerSocketBatchAbort(unsigned int devId, struct SocketConnectInfoT conn[], unsigned int num)
85 : {
86 : struct SocketConnectInfo connOut[MAX_SOCKET_NUM];
87 3 : int ret = 0;
88 :
89 3 : ret = RaGetSocketConnectInfo(conn, num, connOut, MAX_SOCKET_NUM);
90 3 : CHK_PRT_RETURN(ret != 0,
91 : hccp_err("[batch_abort][ra_peer_socket]ra_get_socket_connect_info failed, ret(%d), phyId(%u)", ret, devId),
92 : ret);
93 :
94 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[devId]);
95 2 : RsSetCtx(devId);
96 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[devId]);
97 2 : ret = RsSocketBatchAbort(connOut, num);
98 2 : CHK_PRT_RETURN(ret != 0,
99 : hccp_err("[batch_abort][ra_peer_socket]abort failed ret(%d), phyId(%u), num(%u)", ret, devId, num), ret);
100 :
101 1 : return ret;
102 : }
103 :
104 4 : int RaPeerSocketBatchConnect(unsigned int devId, struct SocketConnectInfoT conn[], unsigned int num)
105 : {
106 : int ret;
107 : struct SocketConnectInfo connOut[MAX_SOCKET_NUM];
108 :
109 4 : ret = RaGetSocketConnectInfo(conn, num, connOut, MAX_SOCKET_NUM);
110 4 : CHK_PRT_RETURN(ret != 0,
111 : hccp_err("[batch_connect][ra_peer_socket]RaGetSocketConnectInfo failed,"
112 : " ret(%d), phyId(%u)",
113 : ret, devId),
114 : ret);
115 :
116 : /* In peer online mode the server port number is user-defined */
117 3 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[devId]);
118 3 : RsSetCtx(devId);
119 3 : ret = RsSocketSetScopeId(devId, ((struct RaSocketHandle *)conn[0].socketHandle)->scopeId);
120 3 : if (ret != 0) {
121 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[devId]);
122 1 : hccp_err("[set scope id][ra_peer_socket]ra_peer_socket_set_scope_id failed, ret(%d), phyId(%u)", ret, devId);
123 1 : return ret;
124 : }
125 :
126 2 : ret = RsSocketBatchConnect(connOut, num);
127 2 : if (ret) {
128 1 : hccp_err("[batch_connect][ra_peer_socket]ra client connect failed ret(%d), phyId(%u)", ret, devId);
129 : }
130 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[devId]);
131 2 : return ret;
132 : }
133 :
134 5 : int RaPeerSocketListenStart(unsigned int devId, struct SocketListenInfoT conn[], unsigned int num)
135 : {
136 5 : struct SocketListenInfo rsConn[MAX_SOCKET_NUM] = {0};
137 : unsigned int i;
138 : int ret;
139 :
140 14 : for (i = 0; i < num; i++) {
141 9 : CHK_PRT_RETURN(conn[i].port > MAX_PORT_NUM,
142 : hccp_err("[listen_start][ra_peer_socket]port(%u) of "
143 : "conn(%u) is invalid, phyId(%u)",
144 : conn[i].port, i, devId),
145 : -EINVAL);
146 : }
147 :
148 5 : ret = RaGetSocketListenInfo(conn, num, rsConn, MAX_SOCKET_NUM);
149 5 : CHK_PRT_RETURN(ret != 0,
150 : hccp_err("[listen_start][ra_peer_socket]ra_get_socket_listen_info failed "
151 : "ret(%d), phyId(%u)",
152 : ret, devId),
153 : ret);
154 :
155 4 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[devId]);
156 4 : RsSetCtx(devId);
157 4 : ret = RsSocketSetScopeId(devId, ((struct RaSocketHandle *)conn[0].socketHandle)->scopeId);
158 4 : if (ret != 0) {
159 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[devId]);
160 1 : hccp_err("[set scope id][ra_peer_socket]ra_peer_socket_set_scope_id failed ret(%d), phyId(%u)", ret, devId);
161 1 : return ret;
162 : }
163 :
164 3 : ret = RsSocketListenStart(rsConn, num);
165 : // listen node found, degrade log level make it consistent with inner call
166 3 : if (ret == -EEXIST) {
167 0 : hccp_info("[listen_start][ra_peer_socket]ra listen start unsuccessful ret(%d), phyId(%u)", ret, devId);
168 3 : } else if (ret == -EADDRINUSE) {
169 0 : hccp_warn("[listen_start][ra_peer_socket]ra listen start unsuccessful ret(%d), phyId(%u)", ret, devId);
170 3 : } else if (ret != 0) {
171 2 : hccp_err("[listen_start][ra_peer_socket]ra listen start failed ret(%d), phyId(%u)", ret, devId);
172 : }
173 3 : if (ret != 0) {
174 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[devId]);
175 2 : return ret;
176 : }
177 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[devId]);
178 :
179 1 : ret = RaGetSocketListenResult(rsConn, num, conn, MAX_SOCKET_NUM);
180 1 : CHK_PRT_RETURN(ret != 0,
181 : hccp_err("[listen_start][ra_peer_socket]ra_get_socket_listen_result failed ret(%d), phyId(%u)", ret, devId),
182 : ret);
183 :
184 1 : return ret;
185 : }
186 :
187 4 : int RaPeerSocketListenStop(unsigned int devId, struct SocketListenInfoT conn[], unsigned int num)
188 : {
189 4 : struct SocketListenInfo rsConn[MAX_SOCKET_NUM] = {0};
190 : unsigned int i;
191 : int ret;
192 :
193 12 : for (i = 0; i < num; i++) {
194 8 : CHK_PRT_RETURN(conn[i].port > MAX_PORT_NUM,
195 : hccp_err("[listen_stop][ra_peer_socket]port(%u) of "
196 : "conn(%u) is invalid, phyId(%u)",
197 : conn[i].port, i, devId),
198 : -EINVAL);
199 : }
200 :
201 4 : ret = RaGetSocketListenInfo(conn, num, rsConn, MAX_SOCKET_NUM);
202 4 : CHK_PRT_RETURN(ret != 0,
203 : hccp_err("[listen_stop][ra_peer_socket]ra_peer_get_socket_listen_info failed ret(%d), phyId(%u)", ret, devId),
204 : ret);
205 :
206 3 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[devId]);
207 3 : RsSetCtx(devId);
208 3 : ret = RsSocketListenStop(rsConn, num);
209 3 : if (ret == -ENODEV) {
210 0 : hccp_warn("[listen_stop][ra_peer_socket]ra socket listen stop unsuccessful ret(%d), phyId(%u)", ret, devId);
211 3 : } else if (ret != 0) {
212 2 : hccp_err("[listen_stop][ra_peer_socket]ra socket listen stop failed ret(%d), phyId(%u)", ret, devId);
213 : }
214 3 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[devId]);
215 3 : return ret;
216 : }
217 :
218 9 : STATIC int RaPeerSetRsConnParam(struct SocketInfoT conn[], unsigned int num, struct SocketFdData rsConn[],
219 : unsigned int rsNum)
220 : {
221 : int ret;
222 : unsigned int i;
223 9 : struct RaSocketHandle *socketHandle = NULL;
224 :
225 9 : CHK_PRT_RETURN(num > rsNum,
226 : hccp_err("[set][ra_peer_rs_conn_param]num(%u) must smaller than rs_num(%u)", num, rsNum), -EINVAL);
227 :
228 14 : for (i = 0; i < num; i++) {
229 8 : socketHandle = (struct RaSocketHandle *)conn[i].socketHandle;
230 8 : rsConn[i].phyId = socketHandle->rdevInfo.phyId;
231 8 : rsConn[i].family = socketHandle->rdevInfo.family;
232 8 : rsConn[i].status = conn[i].status;
233 8 : ret = memcpy_s(&(rsConn[i].localIp), sizeof(union HccpIpAddr), &(socketHandle->rdevInfo.localIp),
234 : sizeof(union HccpIpAddr));
235 8 : CHK_PRT_RETURN(ret != 0,
236 : hccp_err("[set][ra_peer_rs_conn_param]memcpy_s local_ip failed, ret(%d), phyId(%u)", ret,
237 : socketHandle->rdevInfo.phyId),
238 : -ESAFEFUNC);
239 7 : ret = memcpy_s(&(rsConn[i].remoteIp), sizeof(union HccpIpAddr), &(conn[i].remoteIp), sizeof(union HccpIpAddr));
240 7 : CHK_PRT_RETURN(ret != 0,
241 : hccp_err("[set][ra_peer_rs_conn_param]memcpy_s remote_ip failed, ret(%d), phyId(%u)", ret,
242 : socketHandle->rdevInfo.phyId),
243 : ret);
244 6 : ret = memcpy_s(rsConn[i].tag, sizeof(rsConn[i].tag), conn[i].tag, sizeof(conn[i].tag));
245 6 : CHK_PRT_RETURN(ret != 0,
246 : hccp_err("[set][ra_peer_rs_conn_param]memcpy_s tag failed, ret(%d), phyId(%u)", ret,
247 : socketHandle->rdevInfo.phyId),
248 : -ESAFEFUNC);
249 : }
250 6 : return 0;
251 : }
252 :
253 4 : STATIC int RaPeerSetConnParam(struct SocketInfoT conn[], struct SocketFdData rsConn[], unsigned int i,
254 : unsigned int sslEnable)
255 : {
256 : int ret;
257 4 : struct RaSocketHandle *socketHandle = NULL;
258 :
259 4 : socketHandle = (struct RaSocketHandle *)conn[i].socketHandle;
260 4 : socketHandle->rdevInfo.phyId = rsConn[i].phyId;
261 :
262 4 : ret = memcpy_s(&(socketHandle->rdevInfo.localIp), sizeof(union HccpIpAddr), &(rsConn[i].localIp),
263 : sizeof(union HccpIpAddr));
264 4 : CHK_PRT_RETURN(ret != 0,
265 : hccp_err("[set][ra_peer_conn_param]memcpy_s local_ip failed, ret(%d), phyId(%u)", ret, rsConn[i].phyId),
266 : -ESAFEFUNC);
267 4 : ret = memcpy_s(&(conn[i].remoteIp), sizeof(union HccpIpAddr), &(rsConn[i].remoteIp), sizeof(union HccpIpAddr));
268 4 : CHK_PRT_RETURN(ret != 0,
269 : hccp_err("[set][ra_peer_conn_param]memcpy_s remote_ip failed, ret(%d), phyId(%u)", ret, rsConn[i].phyId),
270 : -ESAFEFUNC);
271 :
272 4 : if (conn[i].fdHandle != NULL) {
273 1 : ((struct SocketPeerInfo *)conn[i].fdHandle)->phyId = (int)rsConn[i].phyId;
274 1 : ((struct SocketPeerInfo *)conn[i].fdHandle)->fd = rsConn[i].fd;
275 1 : ((struct SocketPeerInfo *)conn[i].fdHandle)->socketHandle = socketHandle;
276 1 : ((struct SocketPeerInfo *)conn[i].fdHandle)->sslEnable = sslEnable;
277 : }
278 4 : conn[i].status = rsConn[i].status;
279 4 : return 0;
280 : }
281 :
282 8 : int RaPeerGetSockets(unsigned int phyId, unsigned int role, struct SocketInfoT conn[], unsigned int num)
283 : {
284 8 : struct SocketFdData rsConn[MAX_SOCKET_NUM] = {0};
285 : unsigned int sslEnable;
286 : int connectedNum;
287 : unsigned int i;
288 : unsigned int j;
289 : int ret;
290 :
291 8 : ret = RaPeerSetRsConnParam(conn, num, rsConn, MAX_SOCKET_NUM);
292 8 : CHK_PRT_RETURN(ret != 0,
293 : hccp_err("[get][ra_peer_sockets]ra_peer_set_rs_conn_param failed, ret(%d), phyId(%u)", ret, phyId), ret);
294 :
295 6 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[phyId]);
296 6 : RsSetCtx(phyId);
297 6 : connectedNum = RsGetSockets(role, rsConn, num);
298 6 : if (connectedNum < 0) {
299 0 : hccp_err("[get][ra_peer_sockets]ra get socket failed ret(%d), phyId(%u)", connectedNum, phyId);
300 0 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
301 0 : return connectedNum;
302 : }
303 6 : ret = RsGetSslEnable(&sslEnable);
304 6 : if (ret < 0) {
305 1 : hccp_err("[get][ra_peer_sockets]rs_get_ssl_enable failed ret(%d), phyId(%u)", ret, phyId);
306 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
307 1 : return ret;
308 : }
309 5 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
310 :
311 8 : for (i = 0; i < num; i++) {
312 5 : if (rsConn[i].status == RS_SOCK_STATUS_OK) {
313 1 : conn[i].fdHandle = (struct SocketPeerInfo *)calloc(1, sizeof(struct SocketPeerInfo));
314 1 : if (conn[i].fdHandle == NULL) {
315 1 : hccp_err("[get][ra_peer_sockets]socket handle calloc failed, phyId(%u)", phyId);
316 1 : ret = -ENOMEM;
317 1 : goto err_out;
318 : }
319 : } else {
320 4 : conn[i].fdHandle = NULL;
321 : }
322 :
323 4 : ret = RaPeerSetConnParam(conn, rsConn, i, sslEnable);
324 4 : if (ret) {
325 1 : hccp_err("[get][ra_peer_sockets]ra_peer_set_conn_param failed, ret(%d), phyId(%u)", ret, phyId);
326 1 : goto err_out;
327 : }
328 3 : if (memcpy_s(conn[i].tag, sizeof(conn[i].tag), rsConn[i].tag, sizeof(rsConn[i].tag))) {
329 0 : hccp_err("[get][ra_peer_sockets]memcpy_s tag failed, phyId(%u)", phyId);
330 0 : ret = -ESAFEFUNC;
331 0 : goto err_out;
332 : }
333 : }
334 :
335 3 : return connectedNum;
336 :
337 2 : err_out:
338 4 : for (j = 0; j <= i; j++) {
339 2 : if (conn[j].fdHandle != NULL) {
340 0 : free(conn[j].fdHandle);
341 0 : conn[j].fdHandle = NULL;
342 : }
343 : }
344 :
345 2 : return ret;
346 : }
347 :
348 5 : int RaPeerSocketSend(unsigned int devId, const void *handle, const void *data, unsigned long long size)
349 : {
350 : int fd;
351 : int ret;
352 : unsigned int sslEnable;
353 :
354 5 : fd = ((const struct SocketPeerInfo *)handle)->fd;
355 5 : sslEnable = ((const struct SocketPeerInfo *)handle)->sslEnable;
356 5 : if (sslEnable != RA_SSL_DISABLE) {
357 1 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[devId]);
358 1 : RsSetCtx(devId);
359 : }
360 5 : ret = RsPeerSocketSend(sslEnable, fd, data, size);
361 5 : if (sslEnable != RA_SSL_DISABLE) {
362 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[devId]);
363 : }
364 5 : return ret;
365 : }
366 :
367 4 : int RaPeerSocketRecv(unsigned int devId, const void *handle, void *data, unsigned long long size)
368 : {
369 : int fd;
370 : int ret;
371 : unsigned int sslEnable;
372 :
373 4 : fd = ((const struct SocketPeerInfo *)handle)->fd;
374 4 : sslEnable = ((const struct SocketPeerInfo *)handle)->sslEnable;
375 4 : if (sslEnable != RA_SSL_DISABLE) {
376 1 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[devId]);
377 1 : RsSetCtx(devId);
378 : }
379 4 : ret = RsPeerSocketRecv(sslEnable, fd, data, size);
380 4 : if (sslEnable != RA_SSL_DISABLE) {
381 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[devId]);
382 : }
383 4 : return ret;
384 : }
385 :
386 3 : int RaPeerSocketWhiteListAdd(struct rdev rdevInfo, struct SocketWlistInfoT whiteList[], unsigned int num)
387 : {
388 : int ret;
389 : unsigned int i;
390 3 : char netAddr[MAX_IP_LEN] = {0};
391 :
392 3 : for (i = 0; i < num; i++) {
393 3 : CHK_PRT_RETURN(inet_ntop(rdevInfo.family, &whiteList[i].remoteIp, netAddr, sizeof(netAddr)) == NULL,
394 : hccp_err("[add][ra_peer_socket_white_list]remote ip is invalid! i(%u), phyId(%u)", i, rdevInfo.phyId),
395 : -EINVAL);
396 : }
397 0 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[rdevInfo.phyId]);
398 0 : RsSetCtx(rdevInfo.phyId);
399 0 : ret = RsSocketWhiteListAdd(rdevInfo, whiteList, num);
400 0 : if (ret) {
401 0 : hccp_err("[add][ra_peer_socket_white_list]rs_socket_white_list_add failed ret(%d), phyId(%u)", ret,
402 : rdevInfo.phyId);
403 0 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdevInfo.phyId]);
404 0 : return ret;
405 : }
406 0 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdevInfo.phyId]);
407 0 : return ret;
408 : }
409 :
410 2 : int RaPeerEpollCtlAdd(const void *fdHandle, enum RaEpollEvent event)
411 : {
412 : int ret;
413 :
414 2 : ret = RsEpollCtlAdd(fdHandle, event);
415 2 : if (ret) {
416 1 : hccp_err("[ra_peer_epoll_ctl_add]rs_epoll_ctl_add failed ret(%d)", ret);
417 : }
418 2 : return ret;
419 : }
420 :
421 2 : int RaPeerEpollCtlMod(const void *fdHandle, enum RaEpollEvent event)
422 : {
423 : int ret;
424 :
425 2 : ret = RsEpollCtlMod(fdHandle, event);
426 2 : if (ret) {
427 1 : hccp_err("[ra_peer_epoll_ctl_mod]rs_epoll_ctl_mod failed ret(%d)", ret);
428 : }
429 2 : return ret;
430 : }
431 :
432 2 : int RaPeerEpollCtlDel(const void *fdHandle)
433 : {
434 2 : int fd = -1;
435 : int ret;
436 :
437 2 : fd = ((const struct SocketPeerInfo *)fdHandle)->fd;
438 2 : ret = RsEpollCtlDel(fd);
439 2 : if (ret) {
440 1 : hccp_err("[ra_peer_epoll_ctl_del]rs_epoll_ctl_del failed ret(%d), phyId(%d)", ret,
441 : ((const struct SocketPeerInfo *)fdHandle)->phyId);
442 : }
443 2 : return ret;
444 : }
445 :
446 2 : void RaPeerSetTcpRecvCallback(unsigned int phyId, const void *callback)
447 : {
448 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[phyId]);
449 2 : RsSetCtx(phyId);
450 2 : RsSetTcpRecvCallback(callback);
451 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
452 2 : }
453 :
454 2 : int RaPeerSocketWhiteListDel(struct rdev rdevInfo, struct SocketWlistInfoT whiteList[], unsigned int num)
455 : {
456 : int ret;
457 : unsigned int i;
458 2 : char netAddr[MAX_IP_LEN] = {0};
459 :
460 2 : for (i = 0; i < num; i++) {
461 2 : CHK_PRT_RETURN(inet_ntop(rdevInfo.family, &whiteList[i].remoteIp, netAddr, sizeof(netAddr)) == NULL,
462 : hccp_err("[del][ra_peer_socket_white_list]remote ip is invalid! i(%u), phyId(%u)", i, rdevInfo.phyId),
463 : -EINVAL);
464 : }
465 :
466 0 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[rdevInfo.phyId]);
467 0 : RsSetCtx(rdevInfo.phyId);
468 0 : ret = RsSocketWhiteListDel(rdevInfo, whiteList, num);
469 0 : if (ret) {
470 0 : hccp_err("[del][ra_peer_socket_white_list]ra socket listen stop failed ret(%d), phyId(%u)", ret,
471 : rdevInfo.phyId);
472 : }
473 0 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdevInfo.phyId]);
474 0 : return ret;
475 : }
476 :
477 3 : int RaPeerSocketDeinit(struct rdev rdevInfo)
478 : {
479 : int ret;
480 :
481 3 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[rdevInfo.phyId]);
482 3 : RsSetCtx(rdevInfo.phyId);
483 3 : ret = RsSocketDeinit(rdevInfo);
484 3 : if (ret) {
485 1 : hccp_err("[deinit][ra_peer_socket]rs_socket_deinit failed, ret(%d), phyId(%u)", ret, rdevInfo.phyId);
486 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdevInfo.phyId]);
487 1 : return ret;
488 : }
489 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdevInfo.phyId]);
490 2 : return 0;
491 : }
492 :
493 5 : int RaPeerQpCreate(struct RaRdmaHandle *rdmaHandle, int flag, int qpMode, void **qpHandle)
494 : {
495 5 : unsigned int phyId = rdmaHandle->rdevInfo.phyId;
496 5 : struct RaQpHandle *qpPeer = NULL;
497 5 : struct RsQpResp qpResp = {0};
498 5 : struct RsQpNorm qpNorm = {0};
499 : int ret;
500 :
501 5 : qpPeer = (struct RaQpHandle *)calloc(1, sizeof(struct RaQpHandle));
502 5 : CHK_PRT_RETURN(qpPeer == NULL, hccp_err("[create][ra_peer_qp]qp_peer calloc failed, phyId[%u]", phyId), -ENOMEM);
503 :
504 4 : qpNorm.flag = flag;
505 4 : qpNorm.isExp = 1;
506 4 : qpNorm.isExt = 0;
507 4 : qpNorm.qpMode = qpMode;
508 :
509 4 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[phyId]);
510 4 : RsSetCtx(phyId);
511 4 : ret = RsQpCreate(phyId, rdmaHandle->rdevIndex, qpNorm, &qpResp);
512 4 : if (ret) {
513 1 : hccp_err("[create][ra_peer_qp]RsQpCreate failed ret[%d], phyId[%u]", ret, phyId);
514 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
515 1 : goto calloc_err;
516 : }
517 3 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
518 3 : qpPeer->phyId = phyId;
519 3 : qpPeer->qpn = qpResp.qpn;
520 3 : qpPeer->psn = qpResp.psn;
521 3 : qpPeer->gidIdx = qpResp.gidIdx;
522 3 : qpPeer->flag = flag;
523 3 : qpPeer->qpMode = qpMode;
524 3 : qpPeer->rdevIndex = rdmaHandle->rdevIndex;
525 3 : qpPeer->rdmaHandle = rdmaHandle;
526 3 : qpPeer->rdmaOps = rdmaHandle->rdmaOps;
527 :
528 3 : *qpHandle = qpPeer;
529 3 : return ret;
530 :
531 1 : calloc_err:
532 1 : free(qpPeer);
533 1 : qpPeer = NULL;
534 1 : return ret;
535 : }
536 :
537 3 : int RaPeerQpCreateWithAttrs(struct RaRdmaHandle *rdmaHandle, struct QpExtAttrs *extAttrs, void **qpHandle)
538 : {
539 3 : unsigned int phyId = rdmaHandle->rdevInfo.phyId;
540 3 : struct RsQpNormWithAttrs qpNorm = {0};
541 3 : struct RsQpRespWithAttrs qpResp = {0};
542 3 : struct RaQpHandle *qpPeer = NULL;
543 : int ret;
544 :
545 3 : qpPeer = (struct RaQpHandle *)calloc(1, sizeof(struct RaQpHandle));
546 3 : CHK_PRT_RETURN(qpPeer == NULL, hccp_err("[create][ra_peer_qp_with_attrs]qp_peer calloc failed, phyId[%u]", phyId),
547 : -ENOMEM);
548 :
549 2 : qpNorm.isExp = 1;
550 2 : qpNorm.isExt = 0;
551 2 : ret = memcpy_s(&qpNorm.extAttrs, sizeof(struct QpExtAttrs), extAttrs, sizeof(struct QpExtAttrs));
552 2 : if (ret) {
553 0 : hccp_err("[create][ra_peer_qp_with_attrs]memcpy_s for ext_attrs failed ret[%d], phyId[%u]", ret, phyId);
554 0 : ret = -ESAFEFUNC;
555 0 : goto calloc_err;
556 : }
557 :
558 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[phyId]);
559 2 : RsSetCtx(phyId);
560 2 : ret = RsQpCreateWithAttrs(phyId, rdmaHandle->rdevIndex, &qpNorm, &qpResp);
561 2 : if (ret) {
562 1 : hccp_err("[create][ra_peer_qp_with_attrs]RsQpCreateWithAttrs failed ret[%d], phyId[%u]", ret, phyId);
563 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
564 1 : goto calloc_err;
565 : }
566 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
567 1 : qpPeer->phyId = phyId;
568 1 : qpPeer->qpn = qpResp.qpn;
569 1 : qpPeer->psn = qpResp.psn;
570 1 : qpPeer->gidIdx = qpResp.gidIdx;
571 1 : qpPeer->flag = extAttrs->qpAttr.qp_type == IBV_QPT_RC ? 0 : 1;
572 1 : qpPeer->qpMode = extAttrs->qpMode;
573 1 : qpPeer->rdevIndex = rdmaHandle->rdevIndex;
574 1 : qpPeer->rdmaHandle = rdmaHandle;
575 1 : qpPeer->rdmaOps = rdmaHandle->rdmaOps;
576 1 : qpPeer->typicalQpAttr.udpSport = extAttrs->udpSport;
577 :
578 1 : *qpHandle = qpPeer;
579 1 : return ret;
580 :
581 1 : calloc_err:
582 1 : free(qpPeer);
583 1 : qpPeer = NULL;
584 1 : return ret;
585 : }
586 :
587 2 : int RaPeerMrReg(struct RaQpHandle *qpPeer, struct MrInfoT *info)
588 : {
589 : int ret;
590 2 : struct RdmaMrRegInfo mrRegInfo = {0};
591 :
592 2 : mrRegInfo.addr = info->addr;
593 2 : mrRegInfo.len = info->size;
594 2 : mrRegInfo.access = info->access;
595 :
596 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[qpPeer->phyId]);
597 2 : RsSetCtx(qpPeer->phyId);
598 2 : ret = RsMrReg(qpPeer->phyId, qpPeer->rdevIndex, qpPeer->qpn, &mrRegInfo);
599 2 : if (ret) {
600 1 : hccp_err("[reg][ra_peer_mr]ra_reg_mr failed ret(%d), phyId(%u)", ret, qpPeer->phyId);
601 : }
602 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[qpPeer->phyId]);
603 2 : info->lkey = mrRegInfo.lkey;
604 2 : info->rkey = mrRegInfo.rkey;
605 2 : return ret;
606 : }
607 :
608 2 : int RaPeerMrDereg(struct RaQpHandle *qpPeer, struct MrInfoT *info)
609 : {
610 : int ret;
611 :
612 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[qpPeer->phyId]);
613 2 : RsSetCtx(qpPeer->phyId);
614 2 : ret = RsMrDereg(qpPeer->phyId, qpPeer->rdevIndex, qpPeer->qpn, (char *)info->addr);
615 2 : if (ret) {
616 1 : hccp_err("[dereg][ra_peer_mr]ra_de_reg_mr failed ret(%d), phyId(%u)", ret, qpPeer->phyId);
617 : }
618 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[qpPeer->phyId]);
619 2 : return ret;
620 : }
621 :
622 2 : int RaPeerRegisterMr(struct RaRdmaHandle *rdmaPeer, struct MrInfoT *info, void **mrHandle)
623 : {
624 2 : struct RdmaMrRegInfo mrRegInfo = {0};
625 : int ret;
626 :
627 2 : mrRegInfo.addr = info->addr;
628 2 : mrRegInfo.len = info->size;
629 2 : mrRegInfo.access = info->access;
630 :
631 2 : RsSetCtx(rdmaPeer->rdevInfo.phyId);
632 2 : ret = RsRegisterMr(rdmaPeer->rdevInfo.phyId, rdmaPeer->rdevIndex, &mrRegInfo, mrHandle);
633 2 : if (ret) {
634 1 : hccp_err("[ra_peer_register_mr]rs_register_mr failed ret(%d), phyId(%u)", ret, rdmaPeer->rdevInfo.phyId);
635 : }
636 2 : info->lkey = mrRegInfo.lkey;
637 2 : info->rkey = mrRegInfo.rkey;
638 2 : return ret;
639 : }
640 :
641 2 : int RaPeerDeregisterMr(struct RaRdmaHandle *rdmaPeer, void *mrHandle)
642 : {
643 : int ret;
644 :
645 2 : RsSetCtx(rdmaPeer->rdevInfo.phyId);
646 2 : ret = RsDeregisterMr(rdmaPeer->rdevInfo.phyId, rdmaPeer->rdevIndex, mrHandle);
647 2 : if (ret != 0) {
648 1 : hccp_err("[ra_peer_deregister_mr]rs_deregister_mr failed ret(%d), phyId(%u)", ret, rdmaPeer->rdevInfo.phyId);
649 : }
650 2 : return ret;
651 : }
652 :
653 2 : int RaPeerTypicalQpModify(struct RaQpHandle *qpPeer, struct TypicalQp *localQpInfo, struct TypicalQp *remoteQpInfo)
654 : {
655 : int ret;
656 :
657 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[qpPeer->phyId]);
658 2 : RsSetCtx(qpPeer->phyId);
659 2 : ret = RsTypicalQpModify(qpPeer->phyId, qpPeer->rdevIndex, *localQpInfo, *remoteQpInfo, &(qpPeer->typicalQpAttr));
660 2 : if (ret != 0) {
661 0 : hccp_err("[modify][ra_peer_qp]rs_typical_qp_modify failed ret(%d) phyId(%u) qpn(%u)", ret, qpPeer->phyId,
662 : qpPeer->qpn);
663 : }
664 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[qpPeer->phyId]);
665 :
666 2 : return ret;
667 : }
668 :
669 2 : int RaPeerSetQpLbValue(struct RaQpHandle *qpHandle, int lbValue)
670 : {
671 2 : int ret = 0;
672 :
673 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[qpHandle->phyId]);
674 2 : RsSetCtx(qpHandle->phyId);
675 2 : ret = RsSetQpLbValue(qpHandle->phyId, qpHandle->rdevIndex, qpHandle->qpn, lbValue);
676 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[qpHandle->phyId]);
677 2 : if (ret != 0) {
678 1 : if (ret == -ENOTSUPP) {
679 0 : hccp_run_warn("[set][lbValue]RsSetQpLbValue unsuccessful ret:%d", ret);
680 : } else {
681 1 : hccp_err("[set][lbValue]RsSetQpLbValue failed ret:%d, phyId:%u", ret, qpHandle->phyId);
682 : }
683 : }
684 2 : return ret;
685 : }
686 :
687 2 : int RaPeerGetQpLbValue(struct RaQpHandle *qpHandle, int *lbValue)
688 : {
689 2 : int ret = 0;
690 :
691 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[qpHandle->phyId]);
692 2 : RsSetCtx(qpHandle->phyId);
693 2 : ret = RsGetQpLbValue(qpHandle->phyId, qpHandle->rdevIndex, qpHandle->qpn, lbValue);
694 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[qpHandle->phyId]);
695 2 : if (ret != 0) {
696 1 : hccp_err("[get][lbValue]RsGetQpLbValue failed ret:%d, phyId:%u", ret, qpHandle->phyId);
697 : }
698 2 : return ret;
699 : }
700 :
701 2 : int RaPeerQpConnectAsync(struct RaQpHandle *qpPeer, const void *sockHandle)
702 : {
703 : int ret;
704 2 : int fd = ((const struct SocketPeerInfo *)sockHandle)->fd;
705 :
706 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[qpPeer->phyId]);
707 2 : RsSetCtx(qpPeer->phyId);
708 2 : ret = RsQpConnectAsync(qpPeer->phyId, qpPeer->rdevIndex, qpPeer->qpn, fd);
709 2 : if (ret) {
710 1 : hccp_err("[connect_async][ra_peer_qp]ra qp info sync failed socket fd(%d) ret(%d), phyId(%u)", fd, ret,
711 : qpPeer->phyId);
712 : }
713 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[qpPeer->phyId]);
714 2 : return ret;
715 : }
716 :
717 2 : int RaPeerGetQpStatus(struct RaQpHandle *qpPeer, int *status)
718 : {
719 2 : struct RsQpStatusInfo qpInfo = {0};
720 : int ret;
721 :
722 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[qpPeer->phyId]);
723 2 : RsSetCtx(qpPeer->phyId);
724 2 : ret = RsGetQpStatus(qpPeer->phyId, qpPeer->rdevIndex, qpPeer->qpn, &qpInfo);
725 2 : if (ret) {
726 1 : hccp_err("[get][ra_peer_qp_status]ra get qp status failed ret(%d), phyId(%u)", ret, qpPeer->phyId);
727 : }
728 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[qpPeer->phyId]);
729 2 : *status = qpInfo.status;
730 2 : return ret;
731 : }
732 :
733 2 : STATIC int RaPeerLoopbackQpModifyPrepare(struct RaQpHandle *qpHandle, struct TypicalQp *qpInfo)
734 : {
735 2 : int ret = 0;
736 :
737 2 : qpInfo->qpn = qpHandle->qpn;
738 2 : qpInfo->psn = qpHandle->psn;
739 2 : qpInfo->gidIdx = qpHandle->gidIdx;
740 2 : qpInfo->retryCnt = QP_DEFAULT_MAX_ATTR_RETRY_CNT;
741 2 : qpInfo->retryTime = QP_DEFAULT_MAX_ATTR_TIMEOUT;
742 2 : ret = memcpy_s(qpInfo->gid, sizeof(qpInfo->gid), qpHandle->rdmaHandle->gid, sizeof(qpHandle->rdmaHandle->gid));
743 2 : CHK_PRT_RETURN(ret != 0,
744 : hccp_err("memcpy_s gid failed, ret:%d, dst_len:%u, src_len:%d, phyId:%u", ret, sizeof(qpInfo->gid),
745 : qpHandle->rdmaHandle->gid, qpHandle->phyId),
746 : -ESAFEFUNC);
747 :
748 2 : return ret;
749 : }
750 :
751 1 : STATIC int RaPeerLoopbackQpModify(struct RaQpHandle *qpHandle0, struct RaQpHandle *qpHandle1)
752 : {
753 1 : struct TypicalQp qp0Info = {0};
754 1 : struct TypicalQp qp1Info = {0};
755 1 : int ret = 0;
756 :
757 1 : ret = RaPeerLoopbackQpModifyPrepare(qpHandle0, &qp0Info);
758 1 : CHK_PRT_RETURN(ret != 0,
759 : hccp_err("ra_peer_loopback_qp_modify_prepare qp0 failed, ret:%d, phyId:%u", ret, qpHandle0->phyId), ret);
760 1 : ret = RaPeerLoopbackQpModifyPrepare(qpHandle1, &qp1Info);
761 1 : CHK_PRT_RETURN(ret != 0,
762 : hccp_err("ra_peer_loopback_qp_modify_prepare qp1 failed, ret:%d, phyId:%u", ret, qpHandle1->phyId), ret);
763 :
764 1 : ret = RaPeerTypicalQpModify(qpHandle0, &qp0Info, &qp1Info);
765 1 : CHK_PRT_RETURN(ret != 0, hccp_err("ra_peer_typical_qp_modify qp0 failed, ret:%d, phyId:%u", ret, qpHandle0->phyId),
766 : ret);
767 1 : ret = RaPeerTypicalQpModify(qpHandle1, &qp1Info, &qp0Info);
768 1 : CHK_PRT_RETURN(ret != 0, hccp_err("ra_peer_typical_qp_modify qp1 failed, ret:%d, phyId:%u", ret, qpHandle1->phyId),
769 : ret);
770 :
771 1 : return ret;
772 : }
773 :
774 4 : STATIC void RaPeerLoopbackSingleQpDestroy(struct RaQpHandle *qpHandle)
775 : {
776 4 : struct RaLoopbackInfo *loopbackInfo = qpHandle->loopbackInfo;
777 4 : struct RaRdmaHandle *rdmaHandle = qpHandle->rdmaHandle;
778 4 : struct CqAttr attr = {0};
779 :
780 4 : attr.qpContext = &(loopbackInfo->cqContext);
781 4 : attr.ibSendCq = &(loopbackInfo->ibSendCq);
782 4 : attr.ibRecvCq = &(loopbackInfo->ibRecvCq);
783 :
784 4 : (void)RaPeerNormalQpDestroy(qpHandle);
785 4 : (void)RaPeerCqDestroy(rdmaHandle, &attr);
786 4 : (void)RaPeerDestroyCompChannel((void *)loopbackInfo->compChannel);
787 :
788 4 : free(loopbackInfo);
789 4 : loopbackInfo = NULL;
790 4 : }
791 :
792 5 : STATIC void RaPeerLoopbackQpCreatePrepare(struct CqAttr *cqAttr, struct ibv_qp_init_attr *qpInitAttr)
793 : {
794 5 : qpInitAttr->qp_context = *(cqAttr->qpContext);
795 5 : qpInitAttr->send_cq = *(cqAttr->ibSendCq);
796 5 : qpInitAttr->recv_cq = *(cqAttr->ibRecvCq);
797 5 : qpInitAttr->qp_type = IBV_QPT_RC;
798 5 : qpInitAttr->cap.max_send_wr = QP_DEFAULT_MIN_CAP_SEND_WR;
799 5 : qpInitAttr->cap.max_recv_wr = QP_DEFAULT_MIN_CAP_RECV_WR;
800 5 : qpInitAttr->cap.max_send_sge = QP_DEFAULT_MIN_CAP_SEND_SGE;
801 5 : qpInitAttr->cap.max_recv_sge = QP_DEFAULT_MIN_CAP_RECV_SGE;
802 5 : qpInitAttr->cap.max_inline_data = QP_DEFAULT_MAX_CAP_INLINE_DATA;
803 5 : }
804 :
805 6 : STATIC int RaPeerLoopbackSingleQpCreate(struct RaRdmaHandle *rdmaHandle, struct RaQpHandle **qpHandle,
806 : struct ibv_qp **qp)
807 : {
808 6 : struct RaLoopbackInfo *loopbackInfo = NULL;
809 6 : struct ibv_qp_init_attr qpInitAttr = {0};
810 6 : struct CqAttr cqAttr = {0};
811 6 : int ret = 0;
812 :
813 6 : loopbackInfo = (struct RaLoopbackInfo *)calloc(1, sizeof(struct RaLoopbackInfo));
814 6 : CHK_PRT_RETURN(loopbackInfo == NULL, hccp_err("loopback_info calloc failed, phyId:%u", rdmaHandle->rdevInfo.phyId),
815 : -ENOMEM);
816 :
817 6 : ret = RaPeerCreateCompChannel(rdmaHandle, (void **)&loopbackInfo->compChannel);
818 6 : if (ret != 0) {
819 0 : hccp_err("RaPeerCreateCompChannel failed, ret:%d, phyId:%u", ret, rdmaHandle->rdevInfo.phyId);
820 0 : goto channel_create_err;
821 : }
822 :
823 6 : cqAttr.qpContext = &(loopbackInfo->cqContext);
824 6 : cqAttr.ibSendCq = &(loopbackInfo->ibSendCq);
825 6 : cqAttr.ibRecvCq = &(loopbackInfo->ibRecvCq);
826 6 : cqAttr.sendChannel = loopbackInfo->compChannel;
827 6 : cqAttr.recvChannel = loopbackInfo->compChannel;
828 6 : cqAttr.sendCqDepth = CQ_DEFAULT_MIN_SEND_DEPTH;
829 6 : cqAttr.recvCqDepth = CQ_DEFAULT_MIN_RECV_DEPTH;
830 6 : ret = RaPeerCqCreate(rdmaHandle, &cqAttr);
831 6 : if (ret != 0) {
832 1 : hccp_err("ra_peer_cq_create failed, ret:%d, phyId:%u", ret, rdmaHandle->rdevInfo.phyId);
833 1 : goto cq_create_err;
834 : }
835 :
836 5 : RaPeerLoopbackQpCreatePrepare(&cqAttr, &qpInitAttr);
837 5 : ret = RaPeerNormalQpCreate(rdmaHandle, &qpInitAttr, (void **)qpHandle, (void **)qp);
838 5 : if (ret != 0) {
839 1 : hccp_err("ra_peer_normal_qp_create failed, ret:%d, phyId:%u", ret, rdmaHandle->rdevInfo.phyId);
840 1 : goto qp_create_err;
841 : }
842 4 : (*qpHandle)->loopbackInfo = loopbackInfo;
843 4 : return ret;
844 :
845 1 : qp_create_err:
846 1 : (void)RaPeerCqDestroy(rdmaHandle, &cqAttr);
847 2 : cq_create_err:
848 2 : (void)RsDestroyCompChannel((void *)loopbackInfo->compChannel);
849 2 : channel_create_err:
850 2 : free(loopbackInfo);
851 2 : loopbackInfo = NULL;
852 2 : return ret;
853 : }
854 :
855 4 : int RaPeerLoopbackQpCreate(struct RaRdmaHandle *rdmaHandle, struct LoopbackQpPair *qpPair, void **qpHandle)
856 : {
857 4 : struct RaQpHandle *qpHandle0 = NULL;
858 4 : struct RaQpHandle *qpHandle1 = NULL;
859 4 : struct ibv_qp *qp0 = NULL;
860 4 : struct ibv_qp *qp1 = NULL;
861 : int ret;
862 :
863 4 : ret = RaPeerLoopbackSingleQpCreate(rdmaHandle, &qpHandle0, &qp0);
864 4 : CHK_PRT_RETURN(ret != 0,
865 : hccp_err("ra_peer_loopback_single_qp_create qp0 failed, ret:%d, phyId:%u", ret, rdmaHandle->rdevInfo.phyId),
866 : ret);
867 :
868 2 : ret = RaPeerLoopbackSingleQpCreate(rdmaHandle, &qpHandle1, &qp1);
869 2 : if (ret != 0) {
870 0 : hccp_err("ra_peer_loopback_single_qp_create qp1 failed, ret:%d, phyId:%u", ret, rdmaHandle->rdevInfo.phyId);
871 0 : goto qp1_create_err;
872 : }
873 :
874 2 : ret = RaPeerLoopbackQpModify(qpHandle0, qpHandle1);
875 2 : if (ret != 0) {
876 1 : hccp_err("ra_peer_loopback_qp_modify failed, ret:%d, phyId:%u", ret, rdmaHandle->rdevInfo.phyId);
877 1 : goto qp_modify_err;
878 : }
879 :
880 1 : qpPair->ibvQp0 = qp0;
881 1 : qpPair->ibvQp1 = qp1;
882 1 : qpHandle0->loopbackQpHandle = qpHandle1;
883 1 : qpHandle1->loopbackQpHandle = qpHandle0;
884 1 : *qpHandle = qpHandle0;
885 1 : return ret;
886 :
887 1 : qp_modify_err:
888 1 : RaPeerLoopbackSingleQpDestroy(qpHandle1);
889 1 : qp1_create_err:
890 1 : RaPeerLoopbackSingleQpDestroy(qpHandle0);
891 1 : return ret;
892 : }
893 :
894 1 : STATIC void RaPeerLoopbackQpDestroy(struct RaQpHandle *qpHandle0)
895 : {
896 1 : struct RaQpHandle *qpHandle1 = qpHandle0->loopbackQpHandle;
897 :
898 1 : RaPeerLoopbackSingleQpDestroy(qpHandle1);
899 1 : RaPeerLoopbackSingleQpDestroy(qpHandle0);
900 1 : }
901 :
902 5 : STATIC int RaPeerSingleQpDestroy(struct RaQpHandle *qpPeer)
903 : {
904 5 : int ret = 0;
905 :
906 5 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[qpPeer->phyId]);
907 5 : RsSetCtx(qpPeer->phyId);
908 5 : ret = RsQpDestroy(qpPeer->phyId, qpPeer->rdevIndex, qpPeer->qpn);
909 5 : if (ret != 0) {
910 1 : hccp_err("[destroy][ra_peer_qp]destroy failed ret(%d), phyId(%u)", ret, qpPeer->phyId);
911 : }
912 5 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[qpPeer->phyId]);
913 5 : free(qpPeer);
914 5 : qpPeer = NULL;
915 5 : return ret;
916 : }
917 :
918 6 : int RaPeerQpDestroy(struct RaQpHandle *qpPeer)
919 : {
920 6 : if (qpPeer->loopbackQpHandle != NULL) {
921 1 : RaPeerLoopbackQpDestroy(qpPeer);
922 1 : return 0;
923 5 : } else if (qpPeer->directFlag != DIRECT_FLAG_NOTSUPP) {
924 0 : return RaPeerNdaQpDestroy(qpPeer);
925 : } else {
926 5 : return RaPeerSingleQpDestroy(qpPeer);
927 : }
928 : }
929 :
930 1 : int RaPeerSendWr(struct RaQpHandle *qpPeer, struct SendWr *wr, struct SendWrRsp *wrRsp)
931 : {
932 : int ret;
933 :
934 1 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[qpPeer->phyId]);
935 1 : RsSetCtx(qpPeer->phyId);
936 1 : ret = RsSendWr(qpPeer->phyId, qpPeer->rdevIndex, qpPeer->qpn, wr, wrRsp);
937 1 : if (ret) {
938 0 : hccp_err("[send][ra_peer_wr]ra_send_wr failed ret(%d), phyId(%u)", ret, qpPeer->phyId);
939 : }
940 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[qpPeer->phyId]);
941 1 : return ret;
942 : }
943 :
944 4 : STATIC void RaInitWrlistBaseInfo(struct RsWrlistBaseInfo *baseInfo, struct RaQpHandle *qpHandle)
945 : {
946 4 : baseInfo->phyId = qpHandle->phyId;
947 4 : baseInfo->rdevIndex = qpHandle->rdevIndex;
948 4 : baseInfo->qpn = qpHandle->qpn;
949 4 : baseInfo->keyFlag = 0;
950 4 : }
951 :
952 2 : int RaPeerSendWrlist(struct RaQpHandle *qpHandle, struct SendWrlistData wr[], struct SendWrRsp opRsp[],
953 : struct WrlistSendCompleteNum wrlistNum)
954 : {
955 2 : int ret = 0;
956 2 : unsigned int completeCnt = 0;
957 2 : unsigned int sendCnt = 0;
958 : struct RsWrlistBaseInfo baseInfo;
959 : struct WrlistSendCompleteNum wrlistOnce;
960 : unsigned int compeletOnceCnt, i;
961 2 : struct WrInfo *wrList = NULL;
962 :
963 2 : RaInitWrlistBaseInfo(&baseInfo, qpHandle);
964 : CHK_PRT_RETURN(wrlistNum.sendNum > SIZE_MAX / sizeof(struct WrInfo),
965 : hccp_err("Sendnum is invalid, phyId[%u]", baseInfo.phyId), -EINVAL);
966 2 : wrList = calloc(wrlistNum.sendNum, sizeof(struct WrInfo));
967 2 : CHK_PRT_RETURN(wrList == NULL, hccp_err("wr_list calloc failed, phyId[%u]", baseInfo.phyId), -ENOMEM);
968 :
969 4 : for (i = 0; i < wrlistNum.sendNum; i++) {
970 2 : wrList[i].op = wr[i].op;
971 2 : wrList[i].sendFlags = wr[i].sendFlags;
972 2 : wrList[i].dstAddr = wr[i].dstAddr;
973 2 : wrList[i].memList.addr = wr[i].memList.addr;
974 2 : wrList[i].memList.len = wr[i].memList.len;
975 2 : wrList[i].memList.lkey = wr[i].memList.lkey;
976 : }
977 :
978 3 : while (sendCnt < wrlistNum.sendNum) {
979 2 : wrlistOnce.sendNum = (wrlistNum.sendNum - sendCnt) > MAX_WR_NUM ? MAX_WR_NUM : (wrlistNum.sendNum - sendCnt);
980 :
981 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[baseInfo.phyId]);
982 2 : RsSetCtx(baseInfo.phyId);
983 2 : ret = RsSendWrlist(baseInfo, &wrList[sendCnt], wrlistOnce.sendNum, &opRsp[sendCnt], &compeletOnceCnt);
984 2 : if (ret) {
985 1 : hccp_err("[send][ra_peer_wrlist]ra_peer_send_wrlist failed ret[%d], sendNum[%u], sendCnt[%u], phyId[%u]",
986 : ret, wrlistNum.sendNum, sendCnt, baseInfo.phyId);
987 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[baseInfo.phyId]);
988 1 : goto alloc_wr_list_fail;
989 : }
990 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[baseInfo.phyId]);
991 1 : sendCnt += wrlistOnce.sendNum;
992 1 : completeCnt += compeletOnceCnt;
993 : }
994 :
995 1 : if (sendCnt != completeCnt) {
996 1 : hccp_err("[send][ra_peer_wrlist]complete_cnt[%u] != send_cnt[%u], phyId[%u]", completeCnt, sendCnt,
997 : baseInfo.phyId);
998 1 : ret = -EINVAL;
999 : } else {
1000 0 : *(wrlistNum.completeNum) = completeCnt;
1001 : }
1002 :
1003 2 : alloc_wr_list_fail:
1004 2 : free(wrList);
1005 2 : wrList = NULL;
1006 2 : return ret;
1007 : }
1008 :
1009 2 : int RaPeerGetNotifyBaseAddr(struct RaRdmaHandle *handle, unsigned long long *va, unsigned long long *size)
1010 : {
1011 2 : struct MrInfoT info = {0};
1012 : int ret;
1013 :
1014 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[handle->rdevInfo.phyId]);
1015 2 : RsSetCtx(handle->rdevInfo.phyId);
1016 2 : ret = RsGetNotifyMrInfo(handle->rdevInfo.phyId, handle->rdevIndex, &info);
1017 2 : if (ret) {
1018 1 : hccp_err("[get][ra_peer_notify_base_addr]rs_get_notify_mr_info failed ret(%d), phyId(%u)", ret,
1019 : handle->rdevInfo.phyId);
1020 : }
1021 2 : *va = (unsigned long long)(uintptr_t)info.addr;
1022 2 : *size = info.size;
1023 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[handle->rdevInfo.phyId]);
1024 2 : return ret;
1025 : }
1026 :
1027 5 : int RaPeerInit(struct RaInitConfig *cfg, unsigned int whiteListStatus)
1028 : {
1029 : int ret;
1030 :
1031 5 : hccp_info("[init][ra_peer]ra_peer_init phyId[%d] start", cfg->phyId);
1032 :
1033 : /* In peer online mode chip id equals to phy id */
1034 5 : struct RsInitConfig rsPeerOnlineCfg = {
1035 5 : .chipId = cfg->phyId,
1036 5 : .hccpMode = cfg->nicPosition,
1037 : .whiteListStatus = whiteListStatus,
1038 : };
1039 5 : ret = DlHalInit();
1040 5 : if (ret) {
1041 0 : hccp_err("[init][ra_peer]dl_hal_init failed, ret = %d, phyId = %u", ret, cfg->phyId);
1042 0 : return ret;
1043 : }
1044 :
1045 5 : int counter = __sync_fetch_and_add(&(gRaInitCounter[cfg->phyId]), 1);
1046 5 : if (counter > 0) {
1047 1 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[cfg->phyId]);
1048 1 : RsSetCtx(cfg->phyId);
1049 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[cfg->phyId]);
1050 1 : hccp_warn("ra peer has been init for device %u!", cfg->phyId);
1051 1 : return 0;
1052 : }
1053 :
1054 4 : ret = pthread_mutex_init(&gRaPeerMutex[cfg->phyId], NULL);
1055 4 : CHK_PRT_RETURN(ret != 0, hccp_err("[init][ra_peer]pthread_mutex_init failed, ret(%d) phyId(%u)", ret, cfg->phyId),
1056 : -ESYSFUNC);
1057 :
1058 3 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[cfg->phyId]);
1059 3 : RsSetCtx(cfg->phyId);
1060 3 : ret = RsInit(&rsPeerOnlineCfg);
1061 3 : if (ret) {
1062 1 : hccp_err("[init][ra_peer]rs init failed(%d), phyId(%u)", ret, cfg->phyId);
1063 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[cfg->phyId]);
1064 1 : pthread_mutex_destroy(&gRaPeerMutex[cfg->phyId]);
1065 1 : return ret;
1066 : }
1067 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[cfg->phyId]);
1068 2 : hccp_info("[init][ra_peer]ra_peer_init phyId[%d] succ", cfg->phyId);
1069 2 : return ret;
1070 : }
1071 :
1072 1 : int RaPeerGetTlsEnable(unsigned int phyId, bool *tlsEnable)
1073 : {
1074 : int ret;
1075 :
1076 1 : RaPeerMutexLock(phyId);
1077 1 : RsSetCtx(phyId);
1078 1 : ret = RsGetTlsEnable(phyId, tlsEnable);
1079 1 : if (ret != 0) {
1080 0 : hccp_err("[get][tls_enable]rs_get_tls_enable failed, ret(%d) phyId(%u)", ret, phyId);
1081 : }
1082 1 : RaPeerMutexUnlock(phyId);
1083 1 : return ret;
1084 : }
1085 :
1086 1 : int RaPeerGetSecRandom(unsigned int *value)
1087 : {
1088 : int ret;
1089 :
1090 1 : ret = RsGetSecRandom(value);
1091 1 : if (ret != 0) {
1092 0 : hccp_run_warn("[get_random] unsuccessful, ret(%d)", ret);
1093 : }
1094 1 : return ret;
1095 : }
1096 :
1097 5 : int RaPeerDeinit(struct RaInitConfig *cfg)
1098 : {
1099 5 : int ret = 0;
1100 :
1101 5 : hccp_info("[deinit][ra_peer]ra_peer_deinit phyId[%d] start", cfg->phyId);
1102 :
1103 : /* In peer online mode chip id equals to phy id */
1104 5 : struct RsInitConfig rsPeerOnlineCfg = {
1105 5 : .chipId = cfg->phyId,
1106 5 : .hccpMode = cfg->nicPosition,
1107 : .whiteListStatus = WHITE_LIST_ENABLE,
1108 : };
1109 :
1110 5 : if (__sync_fetch_and_sub(&(gRaInitCounter[cfg->phyId]), 1) > 1) {
1111 1 : goto dl_deinit;
1112 : }
1113 :
1114 4 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[cfg->phyId]);
1115 4 : RsSetCtx(cfg->phyId);
1116 4 : ret = RsDeinit(&rsPeerOnlineCfg);
1117 : // no need to destroy lock & return immediately for retry
1118 4 : if (ret == -EAGAIN) {
1119 1 : hccp_warn("[deinit][ra_peer]rs deinit unsuccessful(%d), phyId(%u)", ret, cfg->phyId);
1120 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[cfg->phyId]);
1121 1 : return ret;
1122 : }
1123 :
1124 3 : if (ret) {
1125 1 : hccp_err("[deinit][ra_peer]rs deinit failed(%d), phyId(%u)", ret, cfg->phyId);
1126 : }
1127 3 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[cfg->phyId]);
1128 3 : pthread_mutex_destroy(&gRaPeerMutex[cfg->phyId]);
1129 :
1130 4 : dl_deinit:
1131 4 : DlHalDeinit();
1132 4 : hccp_info("[deinit][ra_peer]ra_peer_deinit phyId[%d] succ", cfg->phyId);
1133 4 : return ret;
1134 : }
1135 :
1136 3 : int RaPeerGetIfnum(unsigned int phyId, unsigned int *num)
1137 : {
1138 : int ret;
1139 :
1140 3 : hccp_info("[get][ra_peer_ifnum]ra_peer_get_ifnum phyId[%u] start", phyId);
1141 3 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[phyId]);
1142 3 : RsSetCtx(phyId);
1143 3 : ret = RsPeerGetIfnum(phyId, num);
1144 3 : if (ret) {
1145 1 : hccp_err("[get][ra_peer_ifnum]rs_peer_get_ifnum failed(%d) phyId[%u]", ret, phyId);
1146 : } else {
1147 2 : hccp_info("[get][ra_peer_ifnum]ra_peer_get_ifnum phyId[%u] succ", phyId);
1148 : }
1149 3 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
1150 :
1151 3 : return ret;
1152 : }
1153 :
1154 2 : int RaPeerGetIfaddrs(unsigned int phyId, struct InterfaceInfo interfaceInfos[], unsigned int *num)
1155 : {
1156 : int ret;
1157 2 : hccp_info("[get][ra_peer_ifaddrs] ra_peer_get_ifaddrs phyId[%u] start", phyId);
1158 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[phyId]);
1159 2 : RsSetCtx(phyId);
1160 2 : ret = RsPeerGetIfaddrs(interfaceInfos, num, phyId);
1161 2 : if (ret) {
1162 1 : hccp_err("[get][ra_peer_ifaddrs]rs_peer_get_ifaddrs failed(%d), phyId(%u)", ret, phyId);
1163 : }
1164 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
1165 2 : hccp_info("[get][ra_peer_ifaddrs] ra_peer_get_ifaddrs phyId[%u] succ", phyId);
1166 2 : return ret;
1167 : }
1168 :
1169 6 : int HostNotifyBaseAddrInit(unsigned int phyId)
1170 : {
1171 : int ret, retVal;
1172 6 : unsigned int notifySize = 0;
1173 6 : unsigned long long *notifyVa = NULL;
1174 6 : unsigned int logicId = 0;
1175 :
1176 6 : ret = DlDrvDeviceGetIndexByPhyId(phyId, &logicId);
1177 6 : CHK_PRT_RETURN(ret != 0,
1178 : hccp_err("[init][base_addr]drvDeviceGetIndexByPhyId failed, ret(%d), phyId(%u)", ret, phyId), ret);
1179 :
1180 5 : ret = DlHalNotifyGetInfo(logicId, 0, RA_NOTIFY_TYPE_TOTAL_SIZE, ¬ifySize);
1181 5 : CHK_PRT_RETURN(ret != 0,
1182 : hccp_err("[init][base_addr]halNotifyGetInfo failed, ret(%d), logicId(%u), phyId(%u)", ret, logicId, phyId),
1183 : ret);
1184 :
1185 4 : gNotifyFd = open(HOST_DEVICE_NAME, O_RDWR);
1186 4 : CHK_PRT_RETURN(gNotifyFd < 0,
1187 : hccp_err("[init][base_addr]Failed to open file_path[%s], err_code[%d], phyId[%u]", HOST_DEVICE_NAME, errno,
1188 : phyId),
1189 : -ENOENT);
1190 :
1191 3 : notifyVa = mmap(NULL, notifySize, PROT_READ | PROT_WRITE, MAP_SHARED, gNotifyFd,
1192 3 : (unsigned long long)logicId << PAGE_SHIFT);
1193 3 : if (notifyVa == MAP_FAILED) {
1194 0 : hccp_err("[init][base_addr]failed to mmap recv buf, fd[%d], err_code[%d], phyId[%u]", gNotifyFd, errno, phyId);
1195 0 : ret = -ENOMEM;
1196 0 : goto close_fd;
1197 : }
1198 :
1199 3 : ret = RsNotifyCfgSet(phyId, (uintptr_t)notifyVa, notifySize);
1200 3 : if (ret) {
1201 2 : hccp_err("[init][base_addr]ra_hdc_notify_cfg_set failed, ret(%d), phyId(%u)", ret, phyId);
1202 2 : goto unmmap_mem;
1203 : }
1204 1 : return 0;
1205 :
1206 2 : unmmap_mem:
1207 2 : retVal = munmap((void *)notifyVa, notifySize);
1208 2 : if (retVal) {
1209 1 : hccp_err("[init][base_addr]munmap buf munmap error, length:%lu, ret:%d, phyId:%u", notifySize, retVal, phyId);
1210 : }
1211 2 : close_fd:
1212 2 : HCCP_CLOSE_RETRY_FOR_EINTR(gNotifyFd);
1213 2 : return ret;
1214 : }
1215 :
1216 8 : int RaPeerNotifyBaseAddrInit(unsigned int notifyType, unsigned int phyId)
1217 : {
1218 8 : switch (notifyType) {
1219 4 : case NOTIFY:
1220 4 : return HostNotifyBaseAddrInit(phyId);
1221 1 : case EVENTID:
1222 1 : return 0;
1223 2 : case NO_USE:
1224 2 : return 0;
1225 1 : default: {
1226 1 : hccp_err("[init][base_addr]notify_type[%u] error, phyId[%u]", notifyType, phyId);
1227 1 : return -EINVAL;
1228 : }
1229 : }
1230 : }
1231 :
1232 6 : int HostNotifyBaseAddrUninit(unsigned int phyId)
1233 : {
1234 : int ret;
1235 : unsigned long long va, size;
1236 6 : unsigned int logicId = 0;
1237 6 : struct HostRoceNotifyInfo notifyNode = {0};
1238 :
1239 6 : ret = DlDrvDeviceGetIndexByPhyId(phyId, &logicId);
1240 6 : CHK_PRT_RETURN(ret != 0,
1241 : hccp_err("[uninit][base_addr]drvDeviceGetIndexByPhyId failed, ret(%d), phyId(%u)", ret, phyId), ret);
1242 :
1243 5 : ret = RsNotifyCfgGet(phyId, &va, &size);
1244 5 : CHK_PRT_RETURN(ret != 0, hccp_err("[uninit][base_addr]rs_notify_cfg_get failed, ret(%d), phyId(%u)", ret, phyId),
1245 : ret);
1246 4 : notifyNode.logicId = logicId;
1247 4 : notifyNode.va = va;
1248 4 : notifyNode.sz = size;
1249 :
1250 4 : CHK_PRT_RETURN(gNotifyFd < 0,
1251 : hccp_err("[uninit][base_addr]file_path[%s] has closed, phyId[%u]", HOST_DEVICE_NAME, phyId), -ENOENT);
1252 :
1253 1 : ret = ioctl(gNotifyFd, HOST_CDEV_IOC_FREE_NOTIFY, ¬ifyNode);
1254 1 : if (ret < 0) {
1255 0 : hccp_err("[uninit][base_addr]Failed to run ioctl, ret[%d], err_code[%d], phyId[%u]", ret, errno, phyId);
1256 0 : HCCP_CLOSE_RETRY_FOR_EINTR(gNotifyFd);
1257 0 : return ret;
1258 : }
1259 :
1260 1 : ret = munmap((void *)(uintptr_t)va, size);
1261 1 : if (ret) {
1262 1 : hccp_err("[uninit][base_addr]munmap buf munmap error, *size:%lu, ret:%d, phyId:%u", size, ret, phyId);
1263 1 : HCCP_CLOSE_RETRY_FOR_EINTR(gNotifyFd);
1264 1 : return ret;
1265 : }
1266 :
1267 0 : HCCP_CLOSE_RETRY_FOR_EINTR(gNotifyFd);
1268 0 : return 0;
1269 : }
1270 :
1271 8 : int NotifyBaseAddrUninit(unsigned int notifyType, unsigned int phyId)
1272 : {
1273 8 : switch (notifyType) {
1274 4 : case NOTIFY:
1275 4 : return HostNotifyBaseAddrUninit(phyId);
1276 1 : case EVENTID:
1277 1 : return 0;
1278 2 : case NO_USE:
1279 2 : return 0;
1280 1 : default: {
1281 1 : hccp_err("[uninit][base_addr]notify_type[%u] error, phyId[%u]", notifyType, phyId);
1282 1 : return -EINVAL;
1283 : }
1284 : }
1285 : }
1286 :
1287 5 : int RaPeerRdevInit(struct RaRdmaHandle *rdmaHandle, unsigned int notifyType, struct rdev rdevInfo,
1288 : unsigned int *rdevIndex)
1289 : {
1290 : int ret, retVal;
1291 :
1292 5 : hccp_run_info("[init][ra_peer_rdev]ra_peer_rdev_init phyId[%d] notify_type[%u] physical device id[%u]",
1293 : rdevInfo.phyId, notifyType, rdmaHandle->rdevInfo.phyId);
1294 :
1295 5 : RsSetCtx(rdevInfo.phyId);
1296 5 : ret = RaPeerNotifyBaseAddrInit(notifyType, rdevInfo.phyId);
1297 5 : CHK_PRT_RETURN(ret != 0,
1298 : hccp_err("[init][ra_peer_rdev] ra_peer_notify_base_addr_init failed[%d], phyId[%u]", ret, rdevInfo.phyId), ret);
1299 :
1300 4 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[rdevInfo.phyId]);
1301 4 : ret = RsRdevInit(rdevInfo, notifyType, rdevIndex);
1302 4 : if (ret) {
1303 2 : hccp_err("[init][ra_peer_rdev] rs_rdev_init failed[%d], phyId[%u]", ret, rdevInfo.phyId);
1304 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdevInfo.phyId]);
1305 2 : goto notify_base_addr_uninit;
1306 : }
1307 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdevInfo.phyId]);
1308 :
1309 2 : return 0;
1310 2 : notify_base_addr_uninit:
1311 2 : retVal = NotifyBaseAddrUninit(notifyType, rdevInfo.phyId);
1312 2 : CHK_PRT_RETURN(retVal,
1313 : hccp_err("[init][ra_peer_rdev] notify_base_addr_uninit failed, ret(%d), phyId(%u)", retVal, rdevInfo.phyId),
1314 : retVal);
1315 1 : return ret;
1316 : }
1317 :
1318 2 : int RaPeerRdevGetPortStatus(struct RaRdmaHandle *rdmaHandle, enum PortStatus *status)
1319 : {
1320 2 : unsigned int phyId = rdmaHandle->rdevInfo.phyId;
1321 : int ret;
1322 :
1323 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[phyId]);
1324 2 : RsSetCtx(phyId);
1325 2 : ret = RsRdevGetPortStatus(phyId, rdmaHandle->rdevIndex, status);
1326 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
1327 2 : if (ret != 0) {
1328 1 : hccp_err("[get][ra_peer_port_status]RsRdevGetPortStatus failed ret(%d) phyId(%u)", ret, phyId);
1329 : }
1330 2 : return ret;
1331 : }
1332 :
1333 2 : int RaPeerGetLbMax(struct RaRdmaHandle *rdmaHandle, int *lbMax)
1334 : {
1335 2 : unsigned int phyId = rdmaHandle->rdevInfo.phyId;
1336 2 : int ret = 0;
1337 :
1338 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[phyId]);
1339 2 : RsSetCtx(phyId);
1340 2 : ret = RsGetLbMax(phyId, rdmaHandle->rdevIndex, lbMax);
1341 2 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
1342 2 : if (ret != 0) {
1343 1 : hccp_err("[get][lbMax]RsGetLbMax failed ret:%d, phyId:%u", ret, phyId);
1344 : }
1345 2 : return ret;
1346 : }
1347 :
1348 4 : int RaPeerRdevDeinit(struct RaRdmaHandle *rdmaHandle, unsigned int notifyType)
1349 : {
1350 : int ret;
1351 :
1352 4 : hccp_info("[deinit][ra_peer_rdev]ra_peer_rdev_deinit phyId[%d]", rdmaHandle->rdevInfo.phyId);
1353 4 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[rdmaHandle->rdevInfo.phyId]);
1354 4 : RsSetCtx(rdmaHandle->rdevInfo.phyId);
1355 4 : ret = RsRdevDeinit(rdmaHandle->rdevInfo.phyId, notifyType, rdmaHandle->rdevIndex);
1356 4 : if (ret) {
1357 1 : hccp_err("[deinit][ra_peer_rdev] rs_rdev_deinit failed[%d], phyId[%u]", ret, rdmaHandle->rdevInfo.phyId);
1358 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdmaHandle->rdevInfo.phyId]);
1359 1 : return ret;
1360 : }
1361 3 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdmaHandle->rdevInfo.phyId]);
1362 :
1363 3 : ret = NotifyBaseAddrUninit(notifyType, rdmaHandle->rdevInfo.phyId);
1364 3 : CHK_PRT_RETURN(ret != 0,
1365 : hccp_err("[deinit][ra_peer_rdev] notify_base_addr_uninit failed, ret(%d), phyId(%u)", ret,
1366 : rdmaHandle->rdevInfo.phyId),
1367 : ret);
1368 :
1369 2 : return 0;
1370 : }
1371 :
1372 2 : int RaPeerSetTsqpDepth(struct RaRdmaHandle *rdmaHandle, unsigned int tempDepth, unsigned int *qpNum)
1373 : {
1374 : int ret;
1375 2 : hccp_info("[set][peer_set_tsqp_depth]ra_peer_set_tsqp_depth phyId[%d]", rdmaHandle->rdevInfo.phyId);
1376 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[rdmaHandle->rdevInfo.phyId]);
1377 2 : RsSetCtx(rdmaHandle->rdevInfo.phyId);
1378 2 : ret = RsSetTsqpDepth(rdmaHandle->rdevInfo.phyId, rdmaHandle->rdevIndex, tempDepth, qpNum);
1379 2 : if (ret) {
1380 1 : hccp_err("[set][peer_set_tsqp_depth] rs_set_tsqp_depth failed[%d], phyId[%u]", ret, rdmaHandle->rdevInfo.phyId);
1381 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdmaHandle->rdevInfo.phyId]);
1382 1 : return ret;
1383 : }
1384 :
1385 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdmaHandle->rdevInfo.phyId]);
1386 1 : return 0;
1387 : }
1388 :
1389 2 : int RaPeerGetTsqpDepth(struct RaRdmaHandle *rdmaHandle, unsigned int *tempDepth, unsigned int *qpNum)
1390 : {
1391 : int ret;
1392 :
1393 2 : hccp_info("[get][peer_get_tsqp_depth]ra_peer_get_tsqp_depth phyId[%d]", rdmaHandle->rdevInfo.phyId);
1394 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[rdmaHandle->rdevInfo.phyId]);
1395 2 : RsSetCtx(rdmaHandle->rdevInfo.phyId);
1396 2 : ret = RsGetTsqpDepth(rdmaHandle->rdevInfo.phyId, rdmaHandle->rdevIndex, tempDepth, qpNum);
1397 2 : if (ret) {
1398 1 : hccp_err("[get][peer_set_tsqp_depth]rs_get_tsqp_depth failed[%d], phyId[%u]", ret, rdmaHandle->rdevInfo.phyId);
1399 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdmaHandle->rdevInfo.phyId]);
1400 1 : return ret;
1401 : }
1402 :
1403 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[rdmaHandle->rdevInfo.phyId]);
1404 1 : return ret;
1405 : }
1406 :
1407 2 : int RaPeerRecvWrlist(struct RaQpHandle *qpHandle, struct RecvWrlistData *wr, unsigned int recvNum,
1408 : unsigned int *completeNum)
1409 : {
1410 : int ret;
1411 2 : struct RsWrlistBaseInfo baseInfo = {0};
1412 2 : unsigned int completeCnt = 0;
1413 2 : unsigned int recvCnt = 0;
1414 : unsigned int recvNumPer;
1415 : unsigned int compeletOnceCnt;
1416 :
1417 2 : RaInitWrlistBaseInfo(&baseInfo, qpHandle);
1418 :
1419 3 : while (recvCnt < recvNum) {
1420 2 : recvNumPer = (recvNum - recvCnt) > MAX_WR_NUM ? MAX_WR_NUM : (recvNum - recvCnt);
1421 :
1422 2 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[baseInfo.phyId]);
1423 2 : RsSetCtx(baseInfo.phyId);
1424 2 : ret = RsRecvWrlist(baseInfo, &wr[recvCnt], recvNumPer, &compeletOnceCnt);
1425 2 : if (ret) {
1426 1 : hccp_err("[recv][peer_recv_wrlist]ra_peer_recv_wrlist failed ret[%d], recvCnt[%u], recvNumPer[%u],"
1427 : " phyId[%u]",
1428 : ret, recvCnt, recvNumPer, baseInfo.phyId);
1429 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[baseInfo.phyId]);
1430 1 : return ret;
1431 : }
1432 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[baseInfo.phyId]);
1433 1 : recvCnt += recvNumPer;
1434 1 : completeCnt += compeletOnceCnt;
1435 : }
1436 :
1437 1 : CHK_PRT_RETURN(recvCnt != completeCnt,
1438 : hccp_err("[recv][peer_recv_wrlist]complete_cnt[%u] != recv_cnt[%u],"
1439 : " phyId[%u]",
1440 : completeCnt, recvCnt, baseInfo.phyId),
1441 : -EINVAL);
1442 :
1443 1 : *completeNum = completeCnt;
1444 1 : return 0;
1445 : }
1446 :
1447 1 : int RaPeerGetQpContext(struct RaQpHandle *qpPeer, void **qp, void **sendCq, void **recvCq)
1448 : {
1449 : int ret;
1450 :
1451 1 : RsSetCtx(qpPeer->phyId);
1452 1 : ret = RsGetQpContext(qpPeer->phyId, qpPeer->rdevIndex, qpPeer->qpn, qp, sendCq, recvCq);
1453 1 : if (ret) {
1454 0 : hccp_err("[get][rs_get_qp_context]ra_peer_get_qp_context failed ret(%d), phyId(%u)", ret, qpPeer->phyId);
1455 : }
1456 1 : return ret;
1457 : }
1458 :
1459 9 : int RaPeerCqCreate(struct RaRdmaHandle *rdmaHandle, struct CqAttr *attr)
1460 : {
1461 : int ret;
1462 9 : unsigned int phyId = rdmaHandle->rdevInfo.phyId;
1463 :
1464 9 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[phyId]);
1465 9 : RsSetCtx(phyId);
1466 9 : ret = RsCqCreate(phyId, rdmaHandle->rdevIndex, attr);
1467 9 : if (ret) {
1468 1 : hccp_err("[create][ra_peer_cq_create]rs_cq_create failed ret[%d], phyId[%u]", ret, phyId);
1469 : }
1470 9 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
1471 :
1472 9 : return ret;
1473 : }
1474 :
1475 8 : int RaPeerCqDestroy(struct RaRdmaHandle *rdmaHandle, struct CqAttr *attr)
1476 : {
1477 : int ret;
1478 8 : unsigned int phyId = rdmaHandle->rdevInfo.phyId;
1479 :
1480 8 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[phyId]);
1481 8 : RsSetCtx(phyId);
1482 8 : ret = RsCqDestroy(phyId, rdmaHandle->rdevIndex, attr);
1483 8 : if (ret) {
1484 1 : hccp_err("[destroy][ra_peer_cq_destroy]rs_cq_destroy failed ret[%d], phyId[%u]", ret, phyId);
1485 : }
1486 8 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
1487 :
1488 8 : return ret;
1489 : }
1490 :
1491 8 : int RaPeerNormalQpCreate(struct RaRdmaHandle *rdmaHandle, struct ibv_qp_init_attr *qpInitAttr, void **qpHandle,
1492 : void **qp)
1493 : {
1494 8 : unsigned int phyId = rdmaHandle->rdevInfo.phyId;
1495 8 : struct RaQpHandle *qpPeer = NULL;
1496 8 : struct RsQpResp qpResp = {0};
1497 : int ret;
1498 :
1499 8 : qpPeer = (struct RaQpHandle *)calloc(1, sizeof(struct RaQpHandle));
1500 8 : CHK_PRT_RETURN(qpPeer == NULL,
1501 : hccp_err("[create][ra_normal_peer_qp]normal_qp_peer calloc failed, phyId[%u]", phyId), -ENOMEM);
1502 :
1503 7 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[phyId]);
1504 7 : RsSetCtx(phyId);
1505 7 : ret = RsNormalQpCreate(phyId, rdmaHandle->rdevIndex, qpInitAttr, &qpResp, qp);
1506 7 : if (ret) {
1507 1 : hccp_err("[create][ra_normal_peer_qp]rs_normal_qp_create failed ret[%d], phyId[%u]", ret, phyId);
1508 1 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
1509 1 : goto calloc_err;
1510 : }
1511 6 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[phyId]);
1512 6 : qpPeer->phyId = phyId;
1513 6 : qpPeer->qpn = qpResp.qpn;
1514 6 : qpPeer->psn = qpResp.psn;
1515 6 : qpPeer->gidIdx = qpResp.gidIdx;
1516 6 : qpPeer->rdevIndex = rdmaHandle->rdevIndex;
1517 6 : qpPeer->rdmaHandle = rdmaHandle;
1518 6 : qpPeer->rdmaOps = rdmaHandle->rdmaOps;
1519 :
1520 6 : *qpHandle = qpPeer;
1521 6 : return ret;
1522 :
1523 1 : calloc_err:
1524 1 : free(qpPeer);
1525 1 : qpPeer = NULL;
1526 1 : return ret;
1527 : }
1528 :
1529 6 : int RaPeerNormalQpDestroy(struct RaQpHandle *qpPeer)
1530 : {
1531 : int ret;
1532 :
1533 6 : PEER_PTHREAD_MUTEX_LOCK(&gRaPeerMutex[qpPeer->phyId]);
1534 6 : RsSetCtx(qpPeer->phyId);
1535 6 : ret = RsNormalQpDestroy(qpPeer->phyId, qpPeer->rdevIndex, qpPeer->qpn);
1536 6 : if (ret) {
1537 1 : hccp_err("[destroy][ra_peer_normal_qp]ra close failed ret(%d), phyId(%u)", ret, qpPeer->phyId);
1538 : }
1539 6 : PEER_PTHREAD_MUTEX_UNLOCK(&gRaPeerMutex[qpPeer->phyId]);
1540 6 : free(qpPeer);
1541 6 : qpPeer = NULL;
1542 6 : return ret;
1543 : }
1544 :
1545 2 : int RaPeerSetQpAttrQos(struct RaQpHandle *qpPeer, struct QosAttr *attr)
1546 : {
1547 : int ret;
1548 :
1549 2 : RsSetCtx(qpPeer->phyId);
1550 2 : ret = RsSetQpAttrQos(qpPeer->phyId, qpPeer->rdevIndex, qpPeer->qpn, attr);
1551 2 : CHK_PRT_RETURN(ret != 0,
1552 : hccp_err("[ra_peer_set_qp_attr_qos]rs_set_qp_attr_qos failed ret(%d), phyId(%u)", ret, qpPeer->phyId), ret);
1553 2 : return ret;
1554 : }
1555 :
1556 2 : int RaPeerSetQpAttrTimeout(struct RaQpHandle *qpPeer, unsigned int *timeout)
1557 : {
1558 : int ret;
1559 :
1560 2 : RsSetCtx(qpPeer->phyId);
1561 2 : ret = RsSetQpAttrTimeout(qpPeer->phyId, qpPeer->rdevIndex, qpPeer->qpn, timeout);
1562 2 : CHK_PRT_RETURN(ret != 0,
1563 : hccp_err("[ra_peer_set_qp_attr_timeout]rs_set_qp_attr_timeout failed ret(%d), phyId(%u)", ret, qpPeer->phyId),
1564 : ret);
1565 2 : return ret;
1566 : }
1567 :
1568 2 : int RaPeerSetQpAttrRetryCnt(struct RaQpHandle *qpPeer, unsigned int *retryCnt)
1569 : {
1570 : int ret;
1571 2 : RsSetCtx(qpPeer->phyId);
1572 2 : ret = RsSetQpAttrRetryCnt(qpPeer->phyId, qpPeer->rdevIndex, qpPeer->qpn, retryCnt);
1573 2 : CHK_PRT_RETURN(ret != 0,
1574 : hccp_err("[ra_peer_set_qp_attr_retry_cnt]rs_set_qp_attr_retry_cnt failed ret(%d), phyId(%u)", ret,
1575 : qpPeer->phyId),
1576 : ret);
1577 2 : return ret;
1578 : }
1579 :
1580 10 : int RaPeerCreateCompChannel(struct RaRdmaHandle *rdmaHandle, void **compChannel)
1581 : {
1582 : int ret;
1583 10 : RsSetCtx(rdmaHandle->rdevInfo.phyId);
1584 10 : ret = RsCreateCompChannel(rdmaHandle->rdevInfo.phyId, rdmaHandle->rdevIndex, compChannel);
1585 10 : CHK_PRT_RETURN(ret != 0,
1586 : hccp_err("[ra_peer_create_comp_channel]rs_create_comp_channel failed ret(%d), phyId(%u)", ret,
1587 : rdmaHandle->rdevInfo.phyId),
1588 : ret);
1589 :
1590 9 : return ret;
1591 : }
1592 :
1593 8 : int RaPeerDestroyCompChannel(void *compChannel)
1594 : {
1595 : int ret;
1596 :
1597 8 : ret = RsDestroyCompChannel(compChannel);
1598 8 : CHK_PRT_RETURN(ret != 0, hccp_err("[ra_peer_destroy_comp_channel]rs_create_comp_channel failed ret(%d)", ret), ret);
1599 :
1600 7 : return ret;
1601 : }
1602 :
1603 4 : int RaPeerCreateSrq(struct RaRdmaHandle *rdmaHandle, struct SrqAttr *attr)
1604 : {
1605 : int ret;
1606 :
1607 : // 创建srq&srq cq
1608 4 : RsSetCtx(rdmaHandle->rdevInfo.phyId);
1609 4 : ret = RsCreateSrq(rdmaHandle->rdevInfo.phyId, rdmaHandle->rdevIndex, attr);
1610 4 : CHK_PRT_RETURN(ret != 0,
1611 : hccp_err("[ra_peer_create_srq]rs_create_srq failed ret(%d), phyId(%u)", ret, rdmaHandle->rdevInfo.phyId), ret);
1612 :
1613 3 : return ret;
1614 : }
1615 :
1616 4 : int RaPeerDestroySrq(struct RaRdmaHandle *rdmaHandle, struct SrqAttr *attr)
1617 : {
1618 : int ret;
1619 :
1620 : // 销毁srq&srq cq
1621 4 : RsSetCtx(rdmaHandle->rdevInfo.phyId);
1622 4 : ret = RsDestroySrq(rdmaHandle->rdevInfo.phyId, rdmaHandle->rdevIndex, attr);
1623 4 : CHK_PRT_RETURN(ret != 0,
1624 : hccp_err("[ra_peer_destroy_srq]rs_destroy_srq failed ret(%d), phyId(%u)", ret, rdmaHandle->rdevInfo.phyId),
1625 : ret);
1626 :
1627 3 : return ret;
1628 : }
1629 :
1630 2 : int RaPeerCreateEventHandle(int *eventHandle)
1631 : {
1632 : int ret;
1633 :
1634 2 : ret = RsCreateEventHandle(eventHandle);
1635 2 : CHK_PRT_RETURN(ret != 0, hccp_err("[ra_peer_create_event_handle]rs_create_event_handle failed ret(%d)", ret), ret);
1636 :
1637 2 : return ret;
1638 : }
1639 :
1640 3 : int RaPeerCtlEventHandle(int eventHandle, const void *fdHandle, int opcode, enum RaEpollEvent event)
1641 : {
1642 : int ret;
1643 :
1644 3 : ret = RsCtlEventHandle(eventHandle, fdHandle, opcode, event);
1645 3 : CHK_PRT_RETURN(ret != 0, hccp_err("[ra_peer_ctl_event_handle]rs_ctl_event_handle failed ret(%d)", ret), ret);
1646 :
1647 2 : return ret;
1648 : }
1649 :
1650 2 : int RaPeerWaitEventHandle(int eventHandle, struct SocketEventInfoT *eventInfos, int timeout, unsigned int maxevents,
1651 : unsigned int *eventsNum)
1652 : {
1653 : int ret;
1654 :
1655 2 : ret = RsWaitEventHandle(eventHandle, eventInfos, timeout, maxevents, eventsNum);
1656 2 : CHK_PRT_RETURN(ret != 0, hccp_err("[ra_peer_wait_event_handle]rs_wait_event_handle failed ret(%d)", ret), ret);
1657 :
1658 2 : return ret;
1659 : }
1660 :
1661 2 : int RaPeerDestroyEventHandle(int *eventHandle)
1662 : {
1663 : int ret;
1664 :
1665 2 : ret = RsDestroyEventHandle(eventHandle);
1666 2 : CHK_PRT_RETURN(ret != 0, hccp_err("[ra_peer_destroy_event_handle]rs_destroy_event_handle failed ret(%d)", ret),
1667 : ret);
1668 :
1669 2 : return ret;
1670 : }
|
