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 <unistd.h>
12 : #include <stdlib.h>
13 : #include <sys/types.h>
14 : #include <netinet/in.h>
15 : #include <arpa/inet.h>
16 : #include <dlfcn.h>
17 : #include <errno.h>
18 : #include <urma_opcode.h>
19 : #include "securec.h"
20 : #include "dl_hal_function.h"
21 : #include "dl_urma_function.h"
22 : #include "dl_net_function.h"
23 : #include "hccp_common.h"
24 : #include "rs.h"
25 : #include "ra_rs_err.h"
26 : #include "ra_rs_comm.h"
27 : #include "rs_inner.h"
28 : #include "rs_epoll.h"
29 : #include "rs_ub.h"
30 : #include "rs_ping_inner.h"
31 : #include "rs_ping_urma.h"
32 :
33 : urma_cr_t gPingJettyRecvCr[RS_PING_URMA_RECV_WC_NUM] = {0};
34 : urma_cr_t gPongJettyRecvCr[RS_PING_URMA_RECV_WC_NUM] = {0};
35 :
36 4 : STATIC bool RsPingUrmaCheckFd(struct RsPingCtxCb *pingCb, int fd)
37 : {
38 4 : if (pingCb->pingJetty.jfce != NULL && pingCb->pingJetty.jfce->fd == fd) {
39 2 : hccp_dbg("ping_jetty jfce->fd:%d poll jfc", fd);
40 2 : return true;
41 : }
42 2 : return false;
43 : }
44 :
45 3 : STATIC bool RsPongUrmaCheckFd(struct RsPingCtxCb *pingCb, int fd)
46 : {
47 3 : if (pingCb->pongJetty.jfce != NULL && pingCb->pongJetty.jfce->fd == fd) {
48 2 : hccp_dbg("pong_jetty jfce->fd:%d poll jfc", fd);
49 2 : return true;
50 : }
51 1 : return false;
52 : }
53 :
54 18 : STATIC void RsGetJettyInfo(struct PingQpInfo *qpInfo, urma_jetty_id_t *jettyId, urma_eid_t *eid)
55 : {
56 18 : urma_jetty_id_t jettyKeyInfo = {0};
57 18 : int ret = 0;
58 :
59 18 : ret = memcpy_s(&jettyKeyInfo, sizeof(urma_jetty_id_t), qpInfo->ub.key, qpInfo->ub.size);
60 18 : if (ret != 0) {
61 0 : hccp_err("memcpy jetty_key_info failed, ret:%d", ret);
62 0 : return;
63 : }
64 :
65 18 : if (jettyId != NULL) {
66 18 : *jettyId = jettyKeyInfo;
67 : }
68 18 : if (eid != NULL) {
69 15 : *eid = jettyKeyInfo.eid;
70 : }
71 : }
72 :
73 2 : STATIC bool RsPingCommonCompareUbInfo(struct PingQpInfo *a, struct PingQpInfo *b)
74 : {
75 2 : if (a->ub.size != b->ub.size) {
76 0 : return false;
77 : }
78 2 : if (memcmp(&a->ub.key, &b->ub.key, sizeof(a->ub.key)) != 0) {
79 0 : return false;
80 : }
81 2 : return true;
82 : }
83 :
84 1 : STATIC int RsPingCbGetUrmaContextAndIndex(struct RsPingCtxCb *pingCb, struct PingInitAttr *attr)
85 : {
86 1 : struct DevBaseAttr devAttr = {0};
87 : int ret;
88 :
89 1 : pingCb->udevCb.urmaCtx = RsUrmaCreateContext(pingCb->udevCb.urmaDev, attr->dev.ub.eidIndex);
90 1 : CHK_PRT_RETURN(pingCb->udevCb.urmaCtx == NULL, hccp_err("urma_create_context failed, errno:%d, "
91 : "eidIndex:%u", errno, attr->dev.ub.eidIndex), -ENODEV);
92 :
93 1 : ret = RsUbGetUeInfo(pingCb->udevCb.urmaCtx, &devAttr);
94 1 : if (ret != 0) {
95 0 : hccp_err("rs_ub_get_ue_info failed, ret:%d errno:%d", ret, errno);
96 0 : ret = -EOPENSRC;
97 0 : goto free_urma_ctx;
98 : }
99 :
100 1 : pingCb->devIndex = RsGenerateDevIndex(PING_URMA_DEV_CNT, devAttr.ub.dieId, devAttr.ub.funcId);
101 1 : return 0;
102 :
103 0 : free_urma_ctx:
104 0 : (void)RsUrmaDeleteContext(pingCb->udevCb.urmaCtx);
105 0 : pingCb->udevCb.urmaCtx = NULL;
106 0 : return ret;
107 : }
108 :
109 2 : STATIC int RsPingCommonInitJfce(struct RsPingCtxCb *pingCb, struct RsPingLocalJettyCb *jettyCb)
110 : {
111 2 : jettyCb->jfce = RsUrmaCreateJfce(pingCb->udevCb.urmaCtx);
112 2 : CHK_PRT_RETURN(jettyCb->jfce == NULL, hccp_err("urma_create_jfce failed, errno:%d", errno), -EOPENSRC);
113 :
114 2 : hccp_run_info("eid:%016llx:%016llx init jfce success, fd:%d", pingCb->udevCb.eidInfo.eid.in6.subnetPrefix,
115 : pingCb->udevCb.eidInfo.eid.in6.interfaceId, jettyCb->jfce->fd);
116 2 : return 0;
117 : }
118 :
119 2 : STATIC int RsPingCommonInitSendJfcWithAttr(struct rs_cb *rscb, struct RsPingCtxCb *pingCb,
120 : union PingQpAttr *attr, struct RsPingLocalJettyCb *jettyCb)
121 : {
122 2 : urma_jfc_cfg_t sendJfcCfg = {
123 2 : .depth = attr->ub.cqAttr.sendCqDepth,
124 : .flag = {.value = 0},
125 : .jfce = NULL,
126 : .user_ctx = 0,
127 : };
128 2 : jettyCb->sendJfc.depth = attr->ub.cqAttr.sendCqDepth;
129 2 : jettyCb->sendJfc.numEvents = 0;
130 2 : jettyCb->sendJfc.maxRecvWcNum = RS_PING_URMA_RECV_WC_NUM;
131 2 : jettyCb->sendJfc.jfc = RsUrmaCreateJfc(pingCb->udevCb.urmaCtx, &sendJfcCfg);
132 2 : CHK_PRT_RETURN(jettyCb->sendJfc.jfc == NULL, hccp_err("urma_create_jfc failed, errno:%d", errno), -EOPENSRC);
133 :
134 2 : hccp_run_info("eid:%016llx:%016llx init send jfc success, jfc_id:%u",
135 : pingCb->udevCb.eidInfo.eid.in6.subnetPrefix, pingCb->udevCb.eidInfo.eid.in6.interfaceId,
136 : jettyCb->sendJfc.jfc->jfc_id.id);
137 2 : return 0;
138 : }
139 :
140 2 : STATIC int RsPingCommonInitRecvJfcWithAttr(struct RsPingCtxCb *pingCb,
141 : union PingQpAttr *attr, struct RsPingLocalJettyCb *jettyCb)
142 : {
143 2 : urma_jfc_cfg_t recvJfcCfg = {
144 2 : .depth = attr->ub.cqAttr.recvCqDepth,
145 : .flag = {.value = 0},
146 2 : .jfce = jettyCb->jfce,
147 : .user_ctx = 0,
148 : };
149 2 : jettyCb->recvJfc.depth = attr->ub.cqAttr.recvCqDepth;
150 2 : jettyCb->recvJfc.numEvents = 0;
151 2 : jettyCb->recvJfc.maxRecvWcNum = RS_PING_URMA_RECV_WC_NUM;
152 2 : jettyCb->recvJfc.jfc = RsUrmaCreateJfc(pingCb->udevCb.urmaCtx, &recvJfcCfg);
153 2 : CHK_PRT_RETURN(jettyCb->recvJfc.jfc == NULL, hccp_err("urma_create_jfc failed, errno:%d", errno), -EOPENSRC);
154 :
155 2 : hccp_run_info("eid:%016llx:%016llx init recv jfc success, jfc_id:%u",
156 : pingCb->udevCb.eidInfo.eid.in6.subnetPrefix, pingCb->udevCb.eidInfo.eid.in6.interfaceId,
157 : jettyCb->recvJfc.jfc->jfc_id.id);
158 2 : return 0;
159 : }
160 :
161 2 : STATIC int RsPingCommonInitJettyWithAttr(struct RsPingCtxCb *pingCb,
162 : union PingQpAttr *attr, struct RsPingLocalJettyCb *jettyCb)
163 : {
164 2 : urma_jetty_cfg_t jettyCfg = {0};
165 2 : urma_jfs_cfg_t jfsCfg = {0};
166 2 : urma_jfr_cfg_t jfrCfg = {0};
167 : int ret;
168 :
169 2 : jfsCfg.depth = attr->ub.qpAttr.cap.maxSendWr;
170 2 : jfsCfg.trans_mode = URMA_TM_UM;
171 2 : jfsCfg.max_sge = (uint8_t)attr->ub.qpAttr.cap.maxSendSge;
172 2 : jfsCfg.max_inline_data = attr->ub.qpAttr.cap.maxInlineData;
173 2 : jfsCfg.rnr_retry = URMA_TYPICAL_RNR_RETRY;
174 2 : jfsCfg.jfc = jettyCb->sendJfc.jfc;
175 2 : jfsCfg.user_ctx = 0;
176 :
177 2 : jfrCfg.depth = attr->ub.qpAttr.cap.maxRecvWr;
178 2 : jfrCfg.flag.bs.token_policy = URMA_TOKEN_PLAIN_TEXT;
179 2 : jfrCfg.trans_mode = URMA_TM_UM;
180 2 : jfrCfg.max_sge = (uint8_t)attr->ub.qpAttr.cap.maxRecvSge;
181 2 : jfrCfg.min_rnr_timer = URMA_TYPICAL_MIN_RNR_TIMER;
182 2 : jfrCfg.jfc = jettyCb->recvJfc.jfc;
183 2 : jfrCfg.token_value.token = attr->ub.qpAttr.tokenValue;
184 :
185 2 : jettyCb->jfr = RsUrmaCreateJfr(pingCb->udevCb.urmaCtx, &jfrCfg);
186 2 : CHK_PRT_RETURN(jettyCb->jfr == NULL, hccp_err("urma_create_jfr failed, errno:%d", errno), -ENOMEM);
187 :
188 2 : jettyCfg.flag.bs.share_jfr = URMA_SHARE_JFR;
189 2 : jettyCfg.jfs_cfg = jfsCfg;
190 2 : jettyCfg.shared.jfr = jettyCb->jfr;
191 2 : jettyCfg.shared.jfc = jettyCb->recvJfc.jfc;
192 :
193 2 : jettyCb->jetty = RsUrmaCreateJetty(pingCb->udevCb.urmaCtx, &jettyCfg);
194 2 : if (jettyCb->jetty == NULL) {
195 0 : hccp_err("urma_create_jetty failed, errno:%d", errno);
196 0 : ret = -ENOMEM;
197 0 : goto create_jetty_fail;
198 : }
199 :
200 2 : jettyCb->tokenValue = attr->ub.qpAttr.tokenValue;
201 2 : hccp_run_info("eid:%016llx:%016llx init jetty success, jetty_id:%u",
202 : pingCb->udevCb.eidInfo.eid.in6.subnetPrefix, pingCb->udevCb.eidInfo.eid.in6.interfaceId,
203 : jettyCb->jetty->jetty_id.id);
204 2 : return 0;
205 :
206 0 : create_jetty_fail:
207 0 : (void)RsUrmaDeleteJfr(jettyCb->jfr);
208 0 : jettyCb->jfr = NULL;
209 0 : return ret;
210 : }
211 :
212 2 : STATIC int RsPingCommonInitLocalJetty(struct rs_cb *rscb, struct RsPingCtxCb *pingCb, union PingQpAttr *attr,
213 : struct RsPingLocalJettyCb *jettyCb)
214 : {
215 : int ret;
216 :
217 2 : hccp_info("eid:%016llx:%016llx cap{%u %u %u %u %u} start init local jettys",
218 : pingCb->udevCb.eidInfo.eid.in6.subnetPrefix, pingCb->udevCb.eidInfo.eid.in6.interfaceId,
219 : attr->ub.qpAttr.cap.maxSendWr, attr->ub.qpAttr.cap.maxRecvWr, attr->ub.qpAttr.cap.maxSendSge,
220 : attr->ub.qpAttr.cap.maxRecvSge, attr->ub.qpAttr.cap.maxInlineData);
221 :
222 2 : ret = RsPingCommonInitJfce(pingCb, jettyCb);
223 2 : if (ret != 0) {
224 0 : hccp_err("init jfce failed, ret:%d", ret);
225 0 : goto init_jfce_fail;
226 : }
227 :
228 2 : ret = RsEpollCtl(rscb->connCb.epollfd, EPOLL_CTL_ADD, jettyCb->jfce->fd, EPOLLIN | EPOLLRDHUP);
229 2 : if (ret != 0) {
230 0 : hccp_err("RsEpollCtl failed! epollfd:%d fd:%d ret:%d", rscb->connCb.epollfd, jettyCb->jfce->fd, ret);
231 0 : goto epoll_ctl_fail;
232 : }
233 :
234 2 : ret = RsPingCommonInitSendJfcWithAttr(rscb, pingCb, attr, jettyCb);
235 2 : if (ret != 0) {
236 0 : hccp_err("init send jfc failed, ret:%d", ret);
237 0 : goto init_send_jfc_fail;
238 : }
239 :
240 2 : ret = RsPingCommonInitRecvJfcWithAttr(pingCb, attr, jettyCb);
241 2 : if (ret != 0) {
242 0 : hccp_err("init recv jfc failed, ret:%d", ret);
243 0 : goto init_recv_jfc_fail;
244 : }
245 :
246 2 : ret = RsPingCommonInitJettyWithAttr(pingCb, attr, jettyCb);
247 2 : if (ret != 0) {
248 0 : hccp_err("init jetty failed, ret:%d", ret);
249 0 : goto init_jetty_fail;
250 : }
251 2 : return 0;
252 :
253 0 : init_jetty_fail:
254 0 : (void)RsUrmaDeleteJfc(jettyCb->recvJfc.jfc);
255 0 : jettyCb->recvJfc.jfc = NULL;
256 0 : init_recv_jfc_fail:
257 0 : (void)RsUrmaDeleteJfc(jettyCb->sendJfc.jfc);
258 0 : jettyCb->sendJfc.jfc = NULL;
259 0 : init_send_jfc_fail:
260 0 : (void)RsEpollCtl(rscb->connCb.epollfd, EPOLL_CTL_DEL, jettyCb->jfce->fd, EPOLLIN | EPOLLRDHUP);
261 0 : epoll_ctl_fail:
262 0 : (void)RsUrmaDeleteJfce(jettyCb->jfce);
263 0 : jettyCb->jfce = NULL;
264 0 : init_jfce_fail:
265 0 : return ret;
266 : }
267 :
268 2 : STATIC void RsPingCommonDeinitLocalJetty(struct rs_cb *rscb, struct RsPingCtxCb *pingCb,
269 : struct RsPingLocalJettyCb *jettyCb)
270 : {
271 2 : (void)RsUrmaDeleteJetty(jettyCb->jetty);
272 2 : jettyCb->jetty = NULL;
273 :
274 2 : (void)RsUrmaDeleteJfr(jettyCb->jfr);
275 2 : jettyCb->jfr = NULL;
276 :
277 2 : (void)RsUrmaDeleteJfc(jettyCb->recvJfc.jfc);
278 2 : jettyCb->recvJfc.jfc = NULL;
279 :
280 2 : (void)RsUrmaDeleteJfc(jettyCb->sendJfc.jfc);
281 2 : jettyCb->sendJfc.jfc = NULL;
282 :
283 2 : (void)RsEpollCtl(rscb->connCb.epollfd, EPOLL_CTL_DEL, jettyCb->jfce->fd, EPOLLIN | EPOLLRDHUP);
284 :
285 2 : (void)RsUrmaDeleteJfce(jettyCb->jfce);
286 2 : jettyCb->jfce = NULL;
287 :
288 2 : return;
289 : }
290 :
291 4 : STATIC int RsPingCommonInitSegCb(struct rs_cb *rscb, struct RsPingCtxCb *pingCb, struct RsPingSegCb *segCb)
292 : {
293 4 : unsigned long flag = 0;
294 4 : uint32_t idx = 0;
295 : int ret;
296 :
297 8 : urma_reg_seg_flag_t segFlag = {
298 : .bs.token_policy = URMA_TOKEN_PLAIN_TEXT,
299 : .bs.cacheable = URMA_NON_CACHEABLE,
300 : .bs.access = URMA_ACCESS_LOCAL_ONLY,
301 4 : .bs.non_pin = (RaRsHasCapability(RA_CAP_DRV_SHAREPOOL_NON_PIN, 0, RsNetGetApiVersion())) ? 1 : 0,
302 : .bs.token_id_valid = URMA_TOKEN_ID_INVALID,
303 : .bs.reserved = 0
304 : };
305 4 : urma_seg_cfg_t segCfg = {
306 : .va = 0,
307 4 : .len = segCb->len,
308 : .token_value = segCb->tokenValue,
309 : .flag = segFlag,
310 : .user_ctx = (uintptr_t)NULL,
311 : .iova = 0
312 : };
313 :
314 4 : hccp_info("payload_offset:%u len:0x%llx sge_num:%u grp_id:%u",
315 : segCb->payloadOffset, segCb->len, segCb->sgeNum, rscb->grpId);
316 :
317 4 : ret = pthread_mutex_init(&segCb->mutex, NULL);
318 4 : CHK_PRT_RETURN(ret != 0, hccp_err("pthread_mutex_init seg_cb mutex failed, ret:%d", ret), ret);
319 :
320 4 : flag = ((unsigned long)pingCb->logicDevid << BUFF_FLAGS_DEVID_OFFSET) | BUFF_SP_SVM;
321 4 : ret = DlHalBuffAllocAlignEx(segCb->len, (unsigned int)RA_RS_4K_PAGE_SIZE,
322 4 : flag, (int)rscb->grpId, (void **)&segCb->addr);
323 4 : if (ret != 0) {
324 0 : hccp_err("DlHalBuffAllocAlignEx failed, length:0x%llx, dev_id:0x%x, flag:0x%lx, grp_id:%u, ret:%d",
325 : segCb->len, pingCb->logicDevid, flag, rscb->grpId, ret);
326 0 : goto alloc_fail;
327 : }
328 :
329 4 : segCfg.va = segCb->addr;
330 4 : segCb->segment = RsUrmaRegisterSeg(pingCb->udevCb.urmaCtx, &segCfg);
331 4 : if (segCb->segment == NULL) {
332 0 : ret = -errno;
333 0 : hccp_err("urma_register_seg failed, ret:%d addr:0x%llx len:0x%llx", ret, segCb->addr, segCb->len);
334 0 : goto segment_reg_fail;
335 : }
336 :
337 : // init sge list
338 4 : segCb->sgeList = calloc(segCb->sgeNum, sizeof(urma_sge_t));
339 4 : if (segCb->sgeList == NULL) {
340 0 : ret = -errno;
341 0 : hccp_err("calloc failed, ret:%d sgeNum:%u", ret, segCb->sgeNum);
342 0 : goto calloc_fail;
343 : }
344 :
345 4 : for (idx = 0; idx < segCb->sgeNum; idx++) {
346 0 : segCb->sgeList[idx].tseg = segCb->segment;
347 0 : segCb->sgeList[idx].len = segCb->payloadOffset;
348 0 : if (idx == 0) {
349 0 : segCb->sgeList[idx].addr = segCb->addr;
350 : } else {
351 0 : segCb->sgeList[idx].addr = segCb->sgeList[idx - 1].addr + segCb->payloadOffset;
352 : }
353 : }
354 4 : segCb->sgeIdx = 0;
355 :
356 4 : hccp_info("eid:%016llx:%016llx segment register success, addr:0x%llx len:%u",
357 : pingCb->udevCb.eidInfo.eid.in6.subnetPrefix, pingCb->udevCb.eidInfo.eid.in6.interfaceId,
358 : segCb->addr, segCb->len);
359 :
360 4 : return 0;
361 :
362 0 : calloc_fail:
363 0 : (void)RsUrmaUnregisterSeg(segCb->segment);
364 0 : segCb->segment = NULL;
365 0 : segment_reg_fail:
366 0 : (void)DlHalBuffFree((void *)(uintptr_t)segCb->addr);
367 0 : alloc_fail:
368 0 : (void)pthread_mutex_destroy(&segCb->mutex);
369 0 : return ret;
370 : }
371 :
372 4 : STATIC void RsPingCommonDeinitSegCb(struct RsPingSegCb *segCb)
373 : {
374 4 : hccp_dbg("addr:0x%llx len:%llu", segCb->addr, segCb->len);
375 :
376 4 : free(segCb->sgeList);
377 4 : segCb->sgeList = NULL;
378 :
379 4 : (void)RsUrmaUnregisterSeg(segCb->segment);
380 4 : segCb->segment = NULL;
381 :
382 4 : (void)DlHalBuffFree((void *)(uintptr_t)segCb->addr);
383 :
384 4 : (void)pthread_mutex_destroy(&segCb->mutex);
385 4 : }
386 :
387 1 : STATIC int RsPingPongInitLocalJettyBuffer(struct rs_cb *rscb, struct PingInitAttr *attr,
388 : struct PingInitInfo *info, struct RsPingCtxCb *pingCb)
389 : {
390 : int ret;
391 :
392 : // prepare ping_jetty send segment
393 1 : pingCb->pingJetty.sendSegCb.payloadOffset = PING_TOTAL_PAYLOAD_MAX_SIZE;
394 1 : pingCb->pingJetty.sendSegCb.len =
395 1 : attr->client.ub.qpAttr.cap.maxSendWr * pingCb->pingJetty.sendSegCb.payloadOffset;
396 1 : pingCb->pingJetty.sendSegCb.sgeNum = attr->client.ub.qpAttr.cap.maxSendWr;
397 1 : pingCb->pingJetty.sendSegCb.tokenValue.token = attr->client.ub.segAttr.tokenValue;
398 1 : ret = RsPingCommonInitSegCb(rscb, pingCb, &pingCb->pingJetty.sendSegCb);
399 1 : CHK_PRT_RETURN(ret != 0, hccp_err("rs_ping_common_init_seg_cb ping_jetty send_seg_cb failed, ret %d", ret), ret);
400 :
401 : // prepare ping_jetty recv segment
402 1 : pingCb->pingJetty.recvSegCb.payloadOffset = PING_TOTAL_PAYLOAD_MAX_SIZE;
403 1 : pingCb->pingJetty.recvSegCb.len =
404 1 : attr->client.ub.qpAttr.cap.maxRecvWr * pingCb->pingJetty.recvSegCb.payloadOffset;
405 1 : pingCb->pingJetty.recvSegCb.sgeNum = attr->client.ub.qpAttr.cap.maxRecvWr;
406 1 : pingCb->pingJetty.recvSegCb.tokenValue.token = attr->client.ub.segAttr.tokenValue;
407 1 : ret = RsPingCommonInitSegCb(rscb, pingCb, &pingCb->pingJetty.recvSegCb);
408 1 : if (ret != 0) {
409 0 : hccp_err("rs_ping_common_init_seg_cb ping_jetty recv_seg_cb failed, ret %d", ret);
410 0 : goto init_ping_jetty_recv_seg_fail;
411 : }
412 :
413 : // prepare pong_jetty send segment
414 1 : pingCb->pongJetty.sendSegCb.payloadOffset = PING_TOTAL_PAYLOAD_MAX_SIZE;
415 1 : pingCb->pongJetty.sendSegCb.len =
416 1 : attr->server.ub.qpAttr.cap.maxSendWr * pingCb->pongJetty.sendSegCb.payloadOffset;
417 1 : pingCb->pongJetty.sendSegCb.sgeNum = attr->server.ub.qpAttr.cap.maxSendWr;
418 1 : pingCb->pongJetty.sendSegCb.tokenValue.token = attr->server.ub.segAttr.tokenValue;
419 1 : ret = RsPingCommonInitSegCb(rscb, pingCb, &pingCb->pongJetty.sendSegCb);
420 1 : if (ret != 0) {
421 0 : hccp_err("rs_ping_common_init_seg_cb pong_jetty send_seg_cb failed, ret %d", ret);
422 0 : goto init_pong_jetty_send_seg_fail;
423 : }
424 : // prepare pong_jetty recv segment
425 1 : pingCb->pongJetty.recvSegCb.payloadOffset = PING_TOTAL_PAYLOAD_MAX_SIZE;
426 1 : pingCb->pongJetty.recvSegCb.len = attr->bufferSize;
427 1 : pingCb->pongJetty.recvSegCb.sgeNum = attr->bufferSize / pingCb->pongJetty.recvSegCb.payloadOffset;
428 1 : pingCb->pongJetty.recvSegCb.tokenValue.token = attr->server.ub.segAttr.tokenValue;
429 1 : ret = RsPingCommonInitSegCb(rscb, pingCb, &pingCb->pongJetty.recvSegCb);
430 1 : if (ret != 0) {
431 0 : hccp_err("rs_ping_common_init_seg_cb pong_jetty recv_seg_cb failed, ret %d", ret);
432 0 : goto init_pong_jetty_recv_seg_fail;
433 : }
434 1 : info->result.bufferVa = pingCb->pongJetty.recvSegCb.addr;
435 1 : info->result.bufferSize = attr->bufferSize;
436 1 : info->result.payloadOffset = pingCb->pongJetty.recvSegCb.payloadOffset;
437 1 : info->result.headerSize = RS_PING_PAYLOAD_HEADER_RESV_CUSTOM;
438 1 : return 0;
439 :
440 0 : init_pong_jetty_recv_seg_fail:
441 0 : RsPingCommonDeinitSegCb(&pingCb->pongJetty.sendSegCb);
442 0 : init_pong_jetty_send_seg_fail:
443 0 : RsPingCommonDeinitSegCb(&pingCb->pingJetty.recvSegCb);
444 0 : init_ping_jetty_recv_seg_fail:
445 0 : RsPingCommonDeinitSegCb(&pingCb->pingJetty.sendSegCb);
446 0 : return ret;
447 : }
448 :
449 1 : STATIC void RsPingCommonDeinitLocalJettyBuffer(struct RsPingCtxCb *pingCb)
450 : {
451 1 : RsPingCommonDeinitSegCb(&pingCb->pongJetty.recvSegCb);
452 1 : RsPingCommonDeinitSegCb(&pingCb->pongJetty.sendSegCb);
453 1 : RsPingCommonDeinitSegCb(&pingCb->pingJetty.recvSegCb);
454 1 : RsPingCommonDeinitSegCb(&pingCb->pingJetty.sendSegCb);
455 1 : }
456 :
457 1 : STATIC int RsPingCommonJfrPostRecv(struct RsPingLocalJettyCb *jettyCb)
458 : {
459 1 : urma_jfr_wr_t *jfrBadWr = NULL;
460 1 : urma_jfr_wr_t jfrWr = {0};
461 1 : urma_sge_t list = {0};
462 : uint32_t sgeIdx;
463 : int ret;
464 :
465 1 : RS_PTHREAD_MUTEX_LOCK(&jettyCb->recvSegCb.mutex);
466 1 : sgeIdx = jettyCb->recvSegCb.sgeIdx;
467 1 : (void)memcpy_s(&list, sizeof(urma_sge_t), &jettyCb->recvSegCb.sgeList[sgeIdx], sizeof(urma_sge_t));
468 1 : jettyCb->recvSegCb.sgeIdx = (sgeIdx + 1) % jettyCb->recvSegCb.sgeNum;
469 1 : RS_PTHREAD_MUTEX_ULOCK(&jettyCb->recvSegCb.mutex);
470 :
471 1 : jfrWr.user_ctx = (uintptr_t)sgeIdx;
472 1 : jfrWr.next = NULL;
473 1 : jfrWr.src.sge = &list;
474 1 : jfrWr.src.num_sge = 1;
475 :
476 1 : ret = RsUrmaPostJettyRecvWr(jettyCb->jetty, &jfrWr, &jfrBadWr);
477 1 : if (ret != 0) {
478 0 : hccp_err("urma_post_jetty_recv_wr failed, ret:%d", ret);
479 0 : return ret;
480 : }
481 :
482 1 : return 0;
483 : }
484 :
485 2 : STATIC int RsPingCommonInitJettyPostRecvAll(struct RsPingLocalJettyCb *jettyCb)
486 : {
487 2 : int ret = 0;
488 : uint32_t i;
489 :
490 : // reset recv jfc notify
491 2 : (void)RsUrmaRearmJfc(jettyCb->recvJfc.jfc, false);
492 :
493 : // prepare jfr wqe
494 2 : for (i = jettyCb->recvSegCb.sgeIdx;
495 2 : i < jettyCb->recvSegCb.sgeNum && i < jettyCb->jfr->jfr_cfg.depth; i++) {
496 0 : ret = RsPingCommonJfrPostRecv(jettyCb);
497 0 : if (ret != 0) {
498 0 : hccp_err("rs_ping_common_jfr_post_recv %u-th rqe failed, ret:%d", i, ret);
499 0 : break;
500 : }
501 : }
502 :
503 2 : return ret;
504 : }
505 :
506 1 : STATIC int RsPingPongInitLocalUbResources(struct rs_cb *rscb, struct PingInitAttr *attr,
507 : struct PingInitInfo *info, struct RsPingCtxCb *pingCb)
508 : {
509 1 : urma_jetty_id_t jettyKey = {0};
510 : int ret;
511 :
512 1 : ret = RsPingCommonInitLocalJetty(rscb, pingCb, &attr->client, &pingCb->pingJetty);
513 1 : CHK_PRT_RETURN(ret != 0, hccp_err("init ping_jetty failed, ret:%d", ret), ret);
514 1 : info->client.version = 0;
515 1 : info->client.ub.tokenValue = attr->client.ub.qpAttr.tokenValue;
516 1 : info->client.ub.size = (uint8_t)sizeof(urma_jetty_id_t);
517 1 : jettyKey = pingCb->pingJetty.jetty->jetty_id;
518 :
519 1 : ret = memcpy_s(info->client.ub.key, sizeof(info->client.ub.key), &jettyKey, sizeof(urma_jetty_id_t));
520 1 : if (ret != 0) {
521 0 : hccp_err("memcpy_s urma_jetty_id_t to PingQpInfo.ub.key failed, ret:%d", ret);
522 0 : goto init_pong_jetty_fail;
523 : }
524 :
525 1 : ret = RsPingCommonInitLocalJetty(rscb, pingCb, &attr->server, &pingCb->pongJetty);
526 1 : if (ret != 0) {
527 0 : hccp_err("init pong_jetty failed, ret:%d", ret);
528 0 : goto init_pong_jetty_fail;
529 : }
530 1 : info->server.version = 0;
531 1 : info->server.ub.tokenValue = attr->server.ub.qpAttr.tokenValue;
532 1 : info->server.ub.size = (uint8_t)sizeof(urma_jetty_id_t);
533 1 : jettyKey = pingCb->pongJetty.jetty->jetty_id;
534 1 : (void)memcpy_s(info->server.ub.key, sizeof(info->server.ub.key), &jettyKey, sizeof(urma_jetty_id_t));
535 :
536 1 : ret = RsPingPongInitLocalJettyBuffer(rscb, attr, info, pingCb);
537 1 : if (ret != 0) {
538 0 : hccp_err("init jetty buffer failed, ret:%d", ret);
539 0 : goto init_buffer_fail;
540 : }
541 :
542 1 : ret = RsPingCommonInitJettyPostRecvAll(&pingCb->pingJetty);
543 1 : if (ret != 0) {
544 0 : hccp_err("ping_jetty post recv failed, ret:%d", ret);
545 0 : goto post_recv_fail;
546 : }
547 1 : ret = RsPingCommonInitJettyPostRecvAll(&pingCb->pongJetty);
548 1 : if (ret != 0) {
549 0 : hccp_err("pong_jetty post recv failed, ret:%d", ret);
550 0 : goto post_recv_fail;
551 : }
552 1 : return 0;
553 :
554 0 : post_recv_fail:
555 0 : RsPingCommonDeinitLocalJettyBuffer(pingCb);
556 0 : init_buffer_fail:
557 0 : RsPingCommonDeinitLocalJetty(rscb, pingCb, &pingCb->pongJetty);
558 0 : init_pong_jetty_fail:
559 0 : RsPingCommonDeinitLocalJetty(rscb, pingCb, &pingCb->pingJetty);
560 0 : return ret;
561 : }
562 :
563 1 : STATIC int RsPingUrmaPingCbInit(unsigned int phyId, struct PingInitAttr *attr, struct PingInitInfo *info,
564 : unsigned int *devIndex, struct RsPingCtxCb *pingCb)
565 : {
566 1 : struct rs_cb *rscb = NULL;
567 : union urma_eid eid;
568 : int ret;
569 :
570 1 : ret = RsGetRsCb(phyId, &rscb);
571 1 : CHK_PRT_RETURN(ret != 0, hccp_err("RsGetRsCb failed, phyId[%u] invalid, ret %d", phyId, ret), ret);
572 :
573 : // prepare input attr
574 1 : pingCb->udevCb.eidInfo.eidIndex = attr->dev.ub.eidIndex;
575 1 : pingCb->udevCb.eidInfo.eid = attr->dev.ub.eid;
576 1 : (void)memcpy_s(&pingCb->commInfo, sizeof(struct PingLocalCommInfo), &attr->commInfo,
577 : sizeof(struct PingLocalCommInfo));
578 :
579 1 : (void)memcpy_s(eid.raw, sizeof(eid.raw), attr->dev.ub.eid.raw, sizeof(attr->dev.ub.eid.raw));
580 1 : pingCb->udevCb.urmaDev = RsUrmaGetDeviceByEid(eid, URMA_TRANSPORT_UB);
581 1 : if (pingCb->udevCb.urmaDev == NULL) {
582 0 : hccp_err("urma_get_device_by_eid failed, urmaDev is NULL, errno:%d eid:%016llx:%016llx", errno,
583 : eid.in6.subnet_prefix, eid.in6.interface_id);
584 0 : ret = -ENODEV;
585 0 : goto get_urma_dev_fail;
586 : }
587 :
588 1 : ret = RsPingCbGetUrmaContextAndIndex(pingCb, attr);
589 1 : if (ret != 0) {
590 0 : hccp_err("rs_ping_cb_get_urma_context_and_index failed, ret:%d", ret);
591 0 : goto get_urma_dev_fail;
592 : }
593 :
594 1 : info->version = 0;
595 1 : ret = RsPingPongInitLocalUbResources(rscb, attr, info, pingCb);
596 1 : if (ret != 0) {
597 0 : hccp_err("rs_ping_pong_init_local_ub_resources failed, ret:%d phyId:%u", ret, phyId);
598 0 : goto init_local_resources_fail;
599 : }
600 1 : *devIndex = pingCb->devIndex;
601 1 : return 0;
602 :
603 0 : init_local_resources_fail:
604 0 : (void)RsUrmaDeleteContext(pingCb->udevCb.urmaCtx);
605 0 : pingCb->udevCb.urmaCtx = NULL;
606 0 : get_urma_dev_fail:
607 0 : (void)pthread_mutex_destroy(&pingCb->pingMutex);
608 0 : (void)pthread_mutex_destroy(&pingCb->pongMutex);
609 0 : return ret;
610 : }
611 :
612 5 : STATIC int RsPingUrmaFindTargetNode(struct RsPingCtxCb *pingCb, struct PingQpInfo *target,
613 : struct RsPingTargetInfo **node)
614 : {
615 5 : struct RsPingTargetInfo *targetNext = NULL;
616 5 : struct RsPingTargetInfo *targetCurr = NULL;
617 5 : urma_jetty_id_t targetJettyId = {0};
618 5 : urma_eid_t targetEid = {0};
619 :
620 5 : RsGetJettyInfo(target, &targetJettyId, &targetEid);
621 :
622 5 : RS_PTHREAD_MUTEX_LOCK(&pingCb->pingMutex);
623 5 : RS_LIST_GET_HEAD_ENTRY(targetCurr, targetNext, &pingCb->pingList, list, struct RsPingTargetInfo);
624 5 : for (; (&targetCurr->list) != &pingCb->pingList;
625 0 : targetCurr = targetNext, targetNext = list_entry(targetNext->list.next, struct RsPingTargetInfo, list)) {
626 1 : if (RsPingCommonCompareUbInfo(&targetCurr->qpInfo, target)) {
627 1 : *node = targetCurr;
628 1 : RS_PTHREAD_MUTEX_ULOCK(&pingCb->pingMutex);
629 1 : return 0;
630 : }
631 : }
632 4 : RS_PTHREAD_MUTEX_ULOCK(&pingCb->pingMutex);
633 :
634 4 : hccp_info("ping target node for jetty_id:%u eid:%016llx:%016llx not found", targetJettyId.id,
635 : targetEid.in6.subnet_prefix, targetEid.in6.interface_id);
636 4 : return -ENODEV;
637 : }
638 :
639 3 : STATIC int RsPingCommonImportJetty(urma_context_t *urmaCtx, struct PingQpInfo *target,
640 : urma_target_jetty_t **importTjetty)
641 : {
642 3 : urma_token_t tokenValue = {0};
643 3 : urma_eid_t remoteEid = {0};
644 3 : urma_rjetty_t rjetty = {0};
645 :
646 3 : RsGetJettyInfo(target, &rjetty.jetty_id, &remoteEid);
647 :
648 3 : tokenValue.token = target->ub.tokenValue;
649 3 : rjetty.trans_mode = URMA_TM_UM;
650 3 : rjetty.type = URMA_JETTY;
651 3 : rjetty.flag.bs.token_policy = URMA_TOKEN_PLAIN_TEXT;
652 3 : rjetty.tp_type = URMA_UTP;
653 :
654 3 : *importTjetty = RsUrmaImportJetty(urmaCtx, &rjetty, &tokenValue);
655 3 : if (*importTjetty == NULL) {
656 0 : hccp_err("urma_import_jetty failed, errno:%d remote eid:%016llx:%016llx", errno,
657 : remoteEid.in6.subnet_prefix, remoteEid.in6.interface_id);
658 0 : return -EOPENSRC;
659 : }
660 3 : return 0;
661 : }
662 :
663 4 : STATIC int RsPingUrmaAllocTargetNode(struct RsPingCtxCb *pingCb, struct PingTargetInfo *target,
664 : struct RsPingTargetInfo **node)
665 : {
666 4 : struct RsPingTargetInfo *targetInfo = NULL;
667 : int ret;
668 :
669 4 : targetInfo = (struct RsPingTargetInfo *)calloc(1, sizeof(struct RsPingTargetInfo));
670 4 : CHK_PRT_RETURN(targetInfo == NULL, hccp_err("calloc target_info failed! errno:%d", errno), -ENOMEM);
671 :
672 3 : ret = pthread_mutex_init(&targetInfo->tripMutex, NULL);
673 3 : if (ret != 0) {
674 0 : hccp_err("pthread_mutex_init tripMutex failed, ret:%d", ret);
675 0 : goto free_target_info;
676 : }
677 :
678 3 : targetInfo->payloadSize = target->payload.size;
679 3 : if (target->payload.size > 0) {
680 2 : targetInfo->payloadBuffer = (char *)calloc(1, target->payload.size);
681 2 : if (targetInfo->payloadBuffer == NULL) {
682 0 : hccp_err("calloc payloadBuffer failed! size:%u errno:%d", target->payload.size, errno);
683 0 : ret = -ENOMEM;
684 0 : goto free_trip_mutex;
685 : }
686 2 : (void)memcpy_s(targetInfo->payloadBuffer, target->payload.size, target->payload.buffer, target->payload.size);
687 : }
688 :
689 3 : (void)memcpy_s(&targetInfo->qpInfo, sizeof(struct PingQpInfo),
690 3 : &target->remoteInfo.qpInfo, sizeof(struct PingQpInfo));
691 :
692 3 : ret = RsPingCommonImportJetty(pingCb->udevCb.urmaCtx, &target->remoteInfo.qpInfo,
693 : &targetInfo->importTjetty);
694 3 : if (ret != 0) {
695 1 : hccp_err("rs_ping_import_jetty failed, ret:%d", ret);
696 1 : goto free_payload_buffer;
697 : }
698 :
699 2 : targetInfo->resultSummary.rttMin = ~0;
700 2 : targetInfo->state = RS_PING_PONG_TARGET_READY;
701 2 : *node = targetInfo;
702 2 : return 0;
703 :
704 1 : free_payload_buffer:
705 1 : if (target->payload.size > 0 && targetInfo->payloadBuffer != NULL) {
706 1 : free(targetInfo->payloadBuffer);
707 1 : targetInfo->payloadBuffer = NULL;
708 : }
709 0 : free_trip_mutex:
710 1 : (void)pthread_mutex_destroy(&targetInfo->tripMutex);
711 1 : free_target_info:
712 1 : free(targetInfo);
713 1 : targetInfo = NULL;
714 1 : return ret;
715 : }
716 :
717 1 : STATIC void RsPingUrmaResetRecvBuffer(struct RsPingCtxCb *pingCb)
718 : {
719 1 : RS_PTHREAD_MUTEX_LOCK(&pingCb->pongJetty.recvSegCb.mutex);
720 1 : (void)memset_s((void *)(uintptr_t)pingCb->pongJetty.recvSegCb.addr, pingCb->pongJetty.recvSegCb.len,
721 : 0, pingCb->pongJetty.recvSegCb.len);
722 1 : RS_PTHREAD_MUTEX_ULOCK(&pingCb->pongJetty.recvSegCb.mutex);
723 1 : }
724 :
725 1 : STATIC void RsPingFillSendHeader(struct RsPingPayloadHeader *header, urma_jetty_id_t *serverJettyKey,
726 : struct RsPingLocalJettyCb *pongJetty, struct RsPingTargetInfo *target)
727 : {
728 1 : header->type = RS_PING_TYPE_URMA_DETECT;
729 1 : (void)memcpy_s(header->server.ub.key, sizeof(header->server.ub.key), serverJettyKey, sizeof(urma_jetty_id_t));
730 1 : header->server.ub.size = sizeof(urma_jetty_id_t);
731 1 : header->server.ub.tokenValue = pongJetty->tokenValue;
732 1 : (void)memcpy_s(&header->target, sizeof(struct PingQpInfo), &target->qpInfo, sizeof(struct PingQpInfo));
733 1 : }
734 :
735 1 : STATIC void RsPingJettyBuildUpWr(struct RsPingCtxCb *pingCb, struct RsPingTargetInfo *target,
736 : urma_sge_t *list, urma_jfs_wr_t *wr)
737 : {
738 1 : wr->opcode = URMA_OPC_SEND;
739 1 : wr->flag.bs.complete_enable = 1;
740 1 : wr->tjetty = target->importTjetty;
741 1 : wr->user_ctx = target->uuid;
742 1 : wr->send.src.sge = list;
743 1 : wr->send.src.num_sge = 1;
744 1 : wr->send.imm_data = 0;
745 1 : wr->next = NULL;
746 1 : }
747 :
748 1 : STATIC int RsPingUrmaPostSend(struct RsPingCtxCb *pingCb, struct RsPingTargetInfo *target)
749 : {
750 1 : urma_jetty_id_t serverJettyKey = {0};
751 1 : struct RsPingPayloadHeader *header = NULL;
752 1 : urma_jetty_id_t targetJettyId = {0};
753 1 : struct timeval timestamp = {0};
754 1 : urma_jfs_wr_t *badWr = NULL;
755 1 : urma_eid_t targetEid = {0};
756 1 : urma_jfs_wr_t wr = {0};
757 1 : urma_sge_t list = {0};
758 : uint32_t sgeIdx;
759 1 : int ret = 0;
760 :
761 1 : RsGetJettyInfo(&target->qpInfo, &targetJettyId, &targetEid);
762 1 : hccp_dbg("target uuid:0x%llx state:%d payload_size:%u jetty_id:%u eid:%016llx:%016llx", target->uuid, target->state,
763 : target->payloadSize, targetJettyId.id, targetEid.in6.subnet_prefix, targetEid.in6.interface_id);
764 :
765 1 : RS_PTHREAD_MUTEX_LOCK(&pingCb->pingJetty.sendSegCb.mutex);
766 1 : sgeIdx = pingCb->pingJetty.sendSegCb.sgeIdx;
767 1 : (void)memcpy_s(&list, sizeof(urma_sge_t), &pingCb->pingJetty.sendSegCb.sgeList[sgeIdx], sizeof(urma_sge_t));
768 1 : pingCb->pingJetty.sendSegCb.sgeIdx = (sgeIdx + 1) % pingCb->pingJetty.sendSegCb.sgeNum;
769 1 : RS_PTHREAD_MUTEX_ULOCK(&pingCb->pingJetty.sendSegCb.mutex);
770 :
771 : // prepare ping_jetty send buffer
772 1 : serverJettyKey = pingCb->pongJetty.jetty->jetty_id;
773 1 : (void)memset_s((void *)(uintptr_t)list.addr, list.len, 0, list.len);
774 1 : header = (struct RsPingPayloadHeader *)(uintptr_t)list.addr;
775 1 : RsPingFillSendHeader(header, &serverJettyKey, &pingCb->pongJetty, target);
776 :
777 1 : if (target->payloadSize > 0) {
778 1 : ret = memcpy_s((void *)(uintptr_t)(list.addr + RS_PING_PAYLOAD_HEADER_RESV_CUSTOM),
779 1 : (list.len - RS_PING_PAYLOAD_HEADER_RESV_CUSTOM), (void *)target->payloadBuffer, target->payloadSize);
780 1 : CHK_PRT_RETURN(ret != 0, hccp_err("memcpy_s buffer payload_size:%u list.len:%u failed, ret:%d",
781 : target->payloadSize, (list.len - RS_PING_PAYLOAD_HEADER_RESV_CUSTOM), ret), -ESAFEFUNC);
782 : }
783 1 : list.len = RS_PING_PAYLOAD_HEADER_RESV_CUSTOM + target->payloadSize;
784 :
785 1 : RsPingJettyBuildUpWr(pingCb, target, &list, &wr);
786 :
787 : // record timestamp t1
788 1 : (void)gettimeofday(×tamp, NULL);
789 1 : header->timestamp.tvSec1 = (uint64_t)timestamp.tv_sec;
790 1 : header->timestamp.tvUsec1 = (uint64_t)timestamp.tv_usec;
791 1 : header->taskId = pingCb->taskId;
792 1 : header->magic = 0x55AA;
793 :
794 1 : ret = RsUrmaPostJettySendWr(pingCb->pingJetty.jetty, &wr, &badWr);
795 1 : if (ret != 0) {
796 0 : hccp_err("rs_urma_post_jetty_send_wr jetty_id:%u failed, ret:%d", serverJettyKey.id, ret);
797 0 : RS_PTHREAD_MUTEX_LOCK(&target->tripMutex);
798 0 : target->state = RS_PING_PONG_TARGET_ERROR;
799 0 : RS_PTHREAD_MUTEX_ULOCK(&target->tripMutex);
800 : }
801 1 : return ret;
802 : }
803 :
804 3 : STATIC int RsPingUrmaPollScq(struct RsPingCtxCb *pingCb, struct RsPingTargetInfo *target)
805 : {
806 3 : urma_cr_t sendCr = {0};
807 : int polledCnt;
808 :
809 3 : polledCnt = RsUrmaPollJfc(pingCb->pingJetty.sendJfc.jfc, 1, &sendCr);
810 3 : if (polledCnt != 1) {
811 1 : hccp_err("uuid:0x%llx rs_urma_poll_jfc polled_cnt:%d", target->uuid, polledCnt);
812 1 : target->state = RS_PING_PONG_TARGET_ERROR;
813 1 : return -ENODATA;
814 : }
815 2 : if (sendCr.status != URMA_CR_SUCCESS) {
816 1 : target->state = RS_PING_PONG_TARGET_ERROR;
817 1 : hccp_err("wr_id:0x%llx error cqe cr_status(%d)", sendCr.user_ctx, sendCr.status);
818 1 : return -EOPENSRC;
819 : }
820 1 : return 0;
821 : }
822 :
823 7 : STATIC int RsPingUrmaPollRcq(struct RsPingCtxCb *pingCb, int *polledCnt, struct timeval *timestamp2)
824 : {
825 7 : urma_jfc_t *evJfc = NULL;
826 7 : uint32_t ackCnt = 1;
827 : int waitCnt;
828 :
829 : // record timestamp t2
830 7 : (void)gettimeofday(timestamp2, NULL);
831 :
832 7 : waitCnt = RsUrmaWaitJfc(pingCb->pingJetty.jfce, 1, 0, &evJfc);
833 7 : if (waitCnt == 0) {
834 1 : return -EAGAIN;
835 : }
836 6 : if (waitCnt != 1) {
837 1 : hccp_err("urma_wait_jfc failed, ret:%d", waitCnt);
838 1 : return -EOPENSRC;
839 : }
840 5 : RsUrmaAckJfc((urma_jfc_t **)&evJfc, &ackCnt, 1);
841 :
842 5 : if (evJfc != pingCb->pingJetty.recvJfc.jfc) {
843 0 : hccp_err("urma_wait_jfc returned unknown jfc");
844 0 : return -EOPENSRC;
845 : }
846 5 : pingCb->pingJetty.recvJfc.numEvents++;
847 :
848 5 : *polledCnt = RsUrmaPollJfc(evJfc, pingCb->pingJetty.recvJfc.maxRecvWcNum, gPingJettyRecvCr);
849 5 : CHK_PRT_RETURN(*polledCnt > pingCb->pingJetty.recvJfc.maxRecvWcNum || *polledCnt < 0,
850 : hccp_err("urma_poll_jfc failed, ret:%d", *polledCnt), -EOPENSRC);
851 :
852 4 : return 0;
853 : }
854 :
855 2 : STATIC int RsPingCommonPollSendJfc(struct RsPingLocalJettyCb *jettyCb)
856 : {
857 2 : urma_cr_t cr = {0};
858 : int polledCnt;
859 :
860 2 : polledCnt = RsUrmaPollJfc(jettyCb->sendJfc.jfc, 1, &cr);
861 2 : if (polledCnt < 0) {
862 1 : hccp_warn("urma_poll_jfc unsuccessful, polledCnt:%d", polledCnt);
863 1 : } else if (polledCnt > 0) {
864 1 : if (cr.status != URMA_CR_SUCCESS) {
865 0 : hccp_err("wr_id:0x%llx error cqe status(%d)", cr.user_ctx, cr.status);
866 0 : return -EOPENSRC;
867 : }
868 : }
869 :
870 2 : return 0;
871 : }
872 :
873 2 : STATIC int RsPongJettyFindTargetNode(struct RsPingCtxCb *pingCb, struct PingQpInfo *target,
874 : struct RsPongTargetInfo **node)
875 : {
876 2 : struct RsPongTargetInfo *targetNext = NULL;
877 2 : struct RsPongTargetInfo *targetCurr = NULL;
878 2 : urma_jetty_id_t targetJettyId = {0};
879 2 : urma_eid_t targetEid = {0};
880 :
881 2 : RsGetJettyInfo(target, &targetJettyId, &targetEid);
882 :
883 2 : RS_CHECK_POINTER_NULL_WITH_RET(pingCb);
884 2 : RS_PTHREAD_MUTEX_LOCK(&pingCb->pongMutex);
885 2 : RS_LIST_GET_HEAD_ENTRY(targetCurr, targetNext, &pingCb->pongList, list, struct RsPongTargetInfo);
886 2 : for (; (&targetCurr->list) != &pingCb->pongList;
887 0 : targetCurr = targetNext, targetNext = list_entry(targetNext->list.next, struct RsPongTargetInfo, list)) {
888 1 : if (RsPingCommonCompareUbInfo(&targetCurr->qpInfo, target)) {
889 1 : *node = targetCurr;
890 1 : RS_PTHREAD_MUTEX_ULOCK(&pingCb->pongMutex);
891 1 : return 0;
892 : }
893 : }
894 1 : RS_PTHREAD_MUTEX_ULOCK(&pingCb->pongMutex);
895 :
896 1 : hccp_info("pong target node for jetty_id:%u eid:%016llX:%016llX not found", targetJettyId.id,
897 : targetEid.in6.subnet_prefix, targetEid.in6.interface_id);
898 1 : return -ENODEV;
899 : }
900 :
901 2 : STATIC int RsPongJettyFindAllocTargetNode(struct RsPingCtxCb *pingCb, struct PingQpInfo *target,
902 : struct RsPongTargetInfo **node)
903 : {
904 2 : struct RsPongTargetInfo *targetInfo = NULL;
905 : int ret;
906 :
907 2 : ret = RsPongJettyFindTargetNode(pingCb, target, node);
908 2 : if (ret == 0 && (*node)->state == RS_PING_PONG_TARGET_READY) {
909 0 : return 0;
910 2 : } else if (ret == 0) {
911 1 : targetInfo = *node;
912 1 : hccp_info("delete pong target uuid:0x%llx state:%d, realloc again", targetInfo->uuid, targetInfo->state);
913 1 : RsListDel(&targetInfo->list);
914 1 : if (targetInfo->importTjetty != NULL) {
915 0 : (void)RsUrmaUnimportJetty(targetInfo->importTjetty);
916 0 : targetInfo->importTjetty = NULL;
917 : }
918 1 : free(targetInfo);
919 1 : targetInfo = NULL;
920 : }
921 :
922 2 : targetInfo = (struct RsPongTargetInfo *)calloc(1, sizeof(struct RsPongTargetInfo));
923 2 : CHK_PRT_RETURN(targetInfo == NULL, hccp_err("calloc target_info failed! errno:%d", errno), -ENOMEM);
924 :
925 2 : (void)memcpy_s(&targetInfo->qpInfo, sizeof(struct PingQpInfo), target, sizeof(struct PingQpInfo));
926 :
927 2 : ret = RsPingCommonImportJetty(pingCb->udevCb.urmaCtx, target, &targetInfo->importTjetty);
928 2 : if (ret != 0) {
929 1 : hccp_err("rs_pong_import_jetty failed, ret:%d", ret);
930 1 : goto free_target_info;
931 : }
932 :
933 1 : targetInfo->state = RS_PING_PONG_TARGET_READY;
934 1 : *node = targetInfo;
935 :
936 1 : RS_PTHREAD_MUTEX_LOCK(&pingCb->pongMutex);
937 1 : targetInfo->uuid = (uint64_t)pingCb->pongNum << 32U;
938 1 : RsListAddTail(&targetInfo->list, &pingCb->pongList);
939 1 : pingCb->pongNum++;
940 1 : RS_PTHREAD_MUTEX_ULOCK(&pingCb->pongMutex);
941 :
942 1 : return 0;
943 :
944 1 : free_target_info:
945 1 : free(targetInfo);
946 1 : return ret;
947 : }
948 :
949 4 : STATIC int RsPongJettyPostSend(struct RsPingCtxCb *pingCb, urma_cr_t *cr, struct timeval *timestamp2)
950 : {
951 4 : struct RsPongTargetInfo *targetInfo = NULL;
952 4 : struct RsPingPayloadHeader *header = NULL;
953 4 : struct timeval timestamp3 = {0};
954 4 : urma_sge_t recvList = {0};
955 4 : urma_sge_t sendList = {0};
956 4 : urma_jfs_wr_t *badWr = NULL;
957 4 : urma_jfs_wr_t wr = {0};
958 : uint32_t recvSgeIdx;
959 : uint32_t sendSgeIdx;
960 4 : int ret = 0;
961 :
962 : // poll send jfc
963 4 : (void)RsPingCommonPollSendJfc(&pingCb->pongJetty);
964 :
965 : // handle detect packet & send response packet
966 4 : recvSgeIdx = (uint32_t)cr->user_ctx;
967 4 : if (recvSgeIdx >= pingCb->pingJetty.recvSegCb.sgeNum) {
968 1 : hccp_err("param err recv_sge_idx:%u > sge_num:%u", recvSgeIdx, pingCb->pingJetty.recvSegCb.sgeNum);
969 1 : return -EIO;
970 : }
971 3 : (void)memcpy_s(&recvList, sizeof(urma_sge_t),
972 3 : &pingCb->pingJetty.recvSegCb.sgeList[recvSgeIdx], sizeof(urma_sge_t));
973 :
974 3 : RS_PTHREAD_MUTEX_LOCK(&pingCb->pongJetty.sendSegCb.mutex);
975 3 : sendSgeIdx = pingCb->pongJetty.sendSegCb.sgeIdx;
976 3 : (void)memcpy_s(&sendList, sizeof(urma_sge_t),
977 3 : &pingCb->pongJetty.sendSegCb.sgeList[sendSgeIdx], sizeof(urma_sge_t));
978 3 : pingCb->pongJetty.sendSegCb.sgeIdx = (sendSgeIdx + 1) % pingCb->pongJetty.sendSegCb.sgeNum;
979 3 : RS_PTHREAD_MUTEX_ULOCK(&pingCb->pongJetty.sendSegCb.mutex);
980 :
981 3 : ret = memcpy_s((void *)(uintptr_t)sendList.addr, sendList.len,
982 3 : (void *)(uintptr_t)recvList.addr, cr->completion_len);
983 3 : CHK_PRT_RETURN(ret != 0, hccp_err("memcpy_s buffer cr->completion_len:%u send_list.length:%u failed, ret:%d",
984 : cr->completion_len, sendList.len, ret), -ESAFEFUNC);
985 3 : sendList.len = cr->completion_len;
986 3 : header = (struct RsPingPayloadHeader *)(uintptr_t)sendList.addr;
987 3 : header->type = RS_PING_TYPE_URMA_RESPONSE;
988 :
989 3 : ret = RsPongJettyFindAllocTargetNode(pingCb, &header->server, &targetInfo);
990 3 : if (ret != 0) {
991 1 : hccp_err("rs_pong_jetty_find_alloc_target_node failed, ret:%d", ret);
992 1 : return ret;
993 : }
994 :
995 2 : wr.opcode = URMA_OPC_SEND;
996 2 : wr.flag.bs.complete_enable = 1;
997 2 : wr.tjetty = targetInfo->importTjetty;
998 2 : wr.user_ctx = targetInfo->uuid;
999 2 : wr.send.src.sge = &sendList;
1000 2 : wr.send.src.num_sge = 1;
1001 2 : wr.send.imm_data = 0;
1002 2 : wr.next = NULL;
1003 :
1004 : // record timestamp t3
1005 2 : (void)gettimeofday(×tamp3, NULL);
1006 2 : header->timestamp.tvSec2 = (uint64_t)timestamp2->tv_sec;
1007 2 : header->timestamp.tvUsec2 = (uint64_t)timestamp2->tv_usec;
1008 2 : header->timestamp.tvSec3 = (uint64_t)timestamp3.tv_sec;
1009 2 : header->timestamp.tvUsec3 = (uint64_t)timestamp3.tv_usec;
1010 2 : header->magic = 0xAA55;
1011 :
1012 2 : ret = RsUrmaPostJettySendWr(pingCb->pongJetty.jetty, &wr, &badWr);
1013 2 : if (ret != 0) {
1014 1 : targetInfo->state = RS_PING_PONG_TARGET_ERROR;
1015 1 : hccp_err("urma_post_jetty_send_wr failed, ret:%d", ret);
1016 1 : return ret;
1017 : }
1018 :
1019 1 : return ret;
1020 : }
1021 :
1022 3 : STATIC void RsPongUrmaHandleSend(struct RsPingCtxCb *pingCb, int polledCnt, struct timeval *timestamp2)
1023 : {
1024 3 : urma_cr_t *cr = NULL;
1025 : int ret, i;
1026 :
1027 3 : cr = gPingJettyRecvCr;
1028 6 : for (i = 0; i < polledCnt; i++) {
1029 3 : if (cr[i].status != URMA_CR_SUCCESS) {
1030 0 : hccp_err("wr_id:0x%llx error cqe status(%d)", cr[i].user_ctx, cr[i].status);
1031 0 : continue;
1032 : }
1033 :
1034 3 : ret = RsPongJettyPostSend(pingCb, &cr[i], timestamp2);
1035 3 : if (ret != 0) {
1036 1 : hccp_err("rs_pong_jetty_post_send failed, wrId:0x%llx", cr[i].user_ctx);
1037 1 : continue;
1038 : }
1039 :
1040 2 : ret = RsPingCommonJfrPostRecv(&pingCb->pingJetty);
1041 2 : if (ret != 0) {
1042 2 : hccp_err("rs_ping_common_jfr_post_recv failed, ret:%d", ret);
1043 2 : continue;
1044 : }
1045 : }
1046 :
1047 3 : ret = RsUrmaRearmJfc(pingCb->pingJetty.recvJfc.jfc, false);
1048 3 : if (ret != 0) {
1049 1 : hccp_err("urma_rearm_jfc failed, ret:%d", ret);
1050 : }
1051 :
1052 3 : return;
1053 : }
1054 :
1055 4 : STATIC int RsPongJettyResolveResponsePacket(struct RsPingCtxCb *pingCb, uint32_t sgeIdx, struct timeval *timestamp4)
1056 : {
1057 4 : struct RsPingTargetInfo *targetInfo = NULL;
1058 4 : struct RsPingPayloadHeader *header = NULL;
1059 4 : urma_jetty_id_t targetJettyId = {0};
1060 4 : urma_eid_t targetEid = {0};
1061 4 : urma_sge_t *recvList = NULL;
1062 : uint32_t rtt;
1063 : int ret;
1064 :
1065 4 : recvList = &pingCb->pongJetty.recvSegCb.sgeList[sgeIdx];
1066 4 : header = (struct RsPingPayloadHeader *)(uintptr_t)(recvList->addr);
1067 4 : if (header->taskId != pingCb->taskId) {
1068 1 : hccp_warn("drop received packet, recv_taskId:%u, curr_taskId:%u", header->taskId, pingCb->taskId);
1069 1 : return 0;
1070 : }
1071 3 : RsGetJettyInfo(&header->target, &targetJettyId, &targetEid);
1072 :
1073 3 : header->timestamp.tvSec4 = (uint64_t)timestamp4->tv_sec;
1074 3 : header->timestamp.tvUsec4 = (uint64_t)timestamp4->tv_usec;
1075 3 : rtt = RsPingGetTripTime(&header->timestamp);
1076 3 : ret = RsPingUrmaFindTargetNode(pingCb, &header->target, &targetInfo);
1077 3 : if (ret != 0) {
1078 1 : hccp_err("rs_ping_urma_find_target_node failed, ret:%d jettyId:%u eid:%016llX:%016llX rtt:%u", ret,
1079 : targetJettyId.id, targetEid.in6.subnet_prefix, targetEid.in6.interface_id, rtt);
1080 1 : return ret;
1081 : }
1082 :
1083 2 : (void)memset_s((void *)header, RS_PING_PAYLOAD_HEADER_MASK_SIZE, 0, RS_PING_PAYLOAD_HEADER_MASK_SIZE);
1084 2 : RS_PTHREAD_MUTEX_LOCK(&targetInfo->tripMutex);
1085 2 : targetInfo->resultSummary.recvCnt++;
1086 2 : targetInfo->resultSummary.taskId = header->taskId;
1087 : // rtt timeout, increase timeoutCnt
1088 2 : if (((uint64_t)targetInfo->resultSummary.taskAttr.timeoutInterval * RS_PING_MSEC_TO_USEC) < rtt) {
1089 1 : targetInfo->resultSummary.timeoutCnt++;
1090 1 : hccp_dbg("recvCnt:%u timeoutInterval:%u rtt:%u timeoutCnt:%u", targetInfo->resultSummary.recvCnt,
1091 : targetInfo->resultSummary.taskAttr.timeoutInterval, rtt, targetInfo->resultSummary.timeoutCnt);
1092 1 : RS_PTHREAD_MUTEX_ULOCK(&targetInfo->tripMutex);
1093 1 : return 0;
1094 : }
1095 :
1096 : // handle rtt_min, rtt_max, rtt_avg
1097 1 : if (targetInfo->resultSummary.rttMin > rtt) {
1098 0 : targetInfo->resultSummary.rttMin = rtt;
1099 : }
1100 1 : if (targetInfo->resultSummary.rttMax < rtt) {
1101 1 : targetInfo->resultSummary.rttMax = rtt;
1102 : }
1103 1 : if (targetInfo->resultSummary.rttAvg == 0) {
1104 1 : targetInfo->resultSummary.rttAvg = rtt;
1105 : }
1106 1 : targetInfo->resultSummary.rttAvg = (targetInfo->resultSummary.rttAvg + rtt) / 2U;
1107 1 : RS_PTHREAD_MUTEX_ULOCK(&targetInfo->tripMutex);
1108 1 : return 0;
1109 : }
1110 :
1111 6 : STATIC void RsPongUrmaPollRcq(struct RsPingCtxCb *pingCb)
1112 : {
1113 6 : struct timeval timestamp = {0};
1114 6 : urma_jfc_t *evJfc = NULL;
1115 : uint32_t recvSgeIdx;
1116 6 : urma_cr_t *cr = NULL;
1117 6 : uint32_t ackCnt = 1;
1118 : int polledCnt, i;
1119 : int waitCnt;
1120 : int ret;
1121 :
1122 : // record timestamp t4
1123 6 : (void)gettimeofday(×tamp, NULL);
1124 :
1125 6 : waitCnt = RsUrmaWaitJfc(pingCb->pongJetty.jfce, 1, 0, &evJfc);
1126 6 : if (waitCnt == 0) {
1127 1 : return;
1128 : }
1129 5 : if (waitCnt != 1) {
1130 1 : hccp_err("urma_wait_jfc failed, ret:%d", waitCnt);
1131 1 : return;
1132 : }
1133 4 : RsUrmaAckJfc((urma_jfc_t **)&evJfc, &ackCnt, 1);
1134 :
1135 4 : if (evJfc != pingCb->pongJetty.recvJfc.jfc) {
1136 0 : hccp_err("urma_wait_jfc returned unknown jfc");
1137 0 : return;
1138 : }
1139 4 : pingCb->pongJetty.recvJfc.numEvents++;
1140 :
1141 4 : polledCnt = RsUrmaPollJfc(evJfc, pingCb->pongJetty.recvJfc.maxRecvWcNum, gPongJettyRecvCr);
1142 4 : if (polledCnt > pingCb->pongJetty.recvJfc.maxRecvWcNum || polledCnt < 0) {
1143 1 : hccp_err("urma_poll_jfc failed, ret:%d", polledCnt);
1144 1 : goto rearm_jfc;
1145 : }
1146 :
1147 3 : cr = gPongJettyRecvCr;
1148 6 : for (i = 0; i < polledCnt; i++) {
1149 3 : if (cr[i].status != URMA_CR_SUCCESS) {
1150 0 : hccp_err("wr_id:0x%llx error cqe status(%d)", cr[i].user_ctx, cr[i].status);
1151 0 : continue;
1152 : }
1153 3 : recvSgeIdx = (uint32_t)cr[i].user_ctx;
1154 3 : if (recvSgeIdx >= pingCb->pongJetty.recvSegCb.sgeNum) {
1155 3 : hccp_err("param err recv_sge_idx:%u > sge_num:%u", recvSgeIdx, pingCb->pongJetty.recvSegCb.sgeNum);
1156 3 : continue;
1157 : }
1158 :
1159 : // handle response packet result
1160 0 : ret = RsPongJettyResolveResponsePacket(pingCb, recvSgeIdx, ×tamp);
1161 0 : if (ret != 0) {
1162 0 : continue;
1163 : }
1164 :
1165 0 : ret = RsPingCommonJfrPostRecv(&pingCb->pongJetty);
1166 0 : if (ret != 0) {
1167 0 : continue;
1168 : }
1169 : }
1170 :
1171 3 : rearm_jfc:
1172 4 : ret = RsUrmaRearmJfc(evJfc, false);
1173 4 : if (ret != 0) {
1174 1 : hccp_err("urma_rearm_jfc failed, ret:%d", ret);
1175 : }
1176 :
1177 4 : return;
1178 : }
1179 :
1180 2 : STATIC int RsPingUrmaGetTargetResult(struct RsPingCtxCb *pingCb, struct PingTargetCommInfo *target,
1181 : struct PingResultInfo *result)
1182 : {
1183 2 : struct RsPingTargetInfo *targetInfo = NULL;
1184 2 : urma_jetty_id_t targetJettyId = {0};
1185 2 : urma_eid_t targetEid = {0};
1186 : int ret;
1187 :
1188 2 : RsGetJettyInfo(&target->qpInfo, &targetJettyId, &targetEid);
1189 :
1190 2 : ret = RsPingUrmaFindTargetNode(pingCb, &target->qpInfo, &targetInfo);
1191 2 : if (ret != 0) {
1192 1 : hccp_err("rs_ping_urma_find_target_node failed, ret:%d jettyId:%u eid:%016llx:%016llx", ret,
1193 : targetJettyId.id, targetEid.in6.subnet_prefix, targetEid.in6.interface_id);
1194 1 : return ret;
1195 : }
1196 :
1197 1 : (void)memcpy_s(&result->summary, sizeof(struct PingResultSummary), &targetInfo->resultSummary,
1198 : sizeof(struct PingResultSummary));
1199 1 : if (targetInfo->state == RS_PING_PONG_TARGET_FINISH) {
1200 0 : result->state = PING_RESULT_STATE_VALID;
1201 : } else {
1202 1 : result->state = PING_RESULT_STATE_INVALID;
1203 : }
1204 1 : hccp_dbg("eid:%016llx:%016llx jetty_id:%u, state:%d send_cnt:%u recv_cnt:%u timeout_cnt:%u rtt_min:%u rtt_max:%u "
1205 : "rtt_avg:%u", targetEid.in6.subnet_prefix, targetEid.in6.interface_id, targetJettyId.id, result->state,
1206 : result->summary.sendCnt, result->summary.recvCnt, result->summary.timeoutCnt, result->summary.rttMin,
1207 : result->summary.rttMax, result->summary.rttAvg);
1208 :
1209 1 : return 0;
1210 : }
1211 :
1212 1 : STATIC void RsPingUrmaFreeTargetNode(struct RsPingCtxCb *pingCb, struct RsPingTargetInfo *targetInfo)
1213 : {
1214 1 : RS_PTHREAD_MUTEX_LOCK(&pingCb->pingMutex);
1215 1 : RsListDel(&targetInfo->list);
1216 1 : RS_PTHREAD_MUTEX_ULOCK(&pingCb->pingMutex);
1217 :
1218 1 : if (targetInfo->payloadSize > 0 && targetInfo->payloadBuffer != NULL) {
1219 0 : free(targetInfo->payloadBuffer);
1220 0 : targetInfo->payloadBuffer = NULL;
1221 : }
1222 :
1223 1 : if (targetInfo->importTjetty != NULL) {
1224 1 : (void)RsUrmaUnimportJetty(targetInfo->importTjetty);
1225 : }
1226 1 : return;
1227 : }
1228 :
1229 1 : STATIC void RsPingPongJettyDelTargetList(struct RsPingCtxCb *pingCb)
1230 : {
1231 1 : struct RsPongTargetInfo *pongNext = NULL;
1232 1 : struct RsPingTargetInfo *pingNext = NULL;
1233 1 : struct RsPongTargetInfo *pongCurr = NULL;
1234 1 : struct RsPingTargetInfo *pingCurr = NULL;
1235 :
1236 : // del ping_list
1237 1 : RS_PTHREAD_MUTEX_LOCK(&pingCb->pingMutex);
1238 1 : RS_LIST_GET_HEAD_ENTRY(pingCurr, pingNext, &pingCb->pingList, list, struct RsPingTargetInfo);
1239 2 : for (; (&pingCurr->list) != &pingCb->pingList;
1240 1 : pingCurr = pingNext, pingNext = list_entry(pingNext->list.next, struct RsPingTargetInfo, list)) {
1241 1 : RsListDel(&pingCurr->list);
1242 1 : if (pingCurr->payloadSize > 0 && pingCurr->payloadBuffer != NULL) {
1243 1 : free(pingCurr->payloadBuffer);
1244 1 : pingCurr->payloadBuffer = NULL;
1245 : }
1246 1 : if (pingCurr->importTjetty != NULL) {
1247 1 : (void)RsUrmaUnimportJetty(pingCurr->importTjetty);
1248 : }
1249 1 : (void)pthread_mutex_destroy(&pingCurr->tripMutex);
1250 1 : free(pingCurr);
1251 1 : pingCurr = NULL;
1252 : }
1253 1 : RS_PTHREAD_MUTEX_ULOCK(&pingCb->pingMutex);
1254 :
1255 : // del pong_list
1256 1 : RS_PTHREAD_MUTEX_LOCK(&pingCb->pongMutex);
1257 1 : RS_LIST_GET_HEAD_ENTRY(pongCurr, pongNext, &pingCb->pongList, list, struct RsPongTargetInfo);
1258 1 : for (; (&pongCurr->list) != &pingCb->pongList;
1259 0 : pongCurr = pongNext, pongNext = list_entry(pongNext->list.next, struct RsPongTargetInfo, list)) {
1260 0 : RsListDel(&pongCurr->list);
1261 0 : if (pongCurr->importTjetty != NULL) {
1262 0 : (void)RsUrmaUnimportJetty(pongCurr->importTjetty);
1263 : }
1264 0 : free(pongCurr);
1265 0 : pongCurr = NULL;
1266 : }
1267 1 : RS_PTHREAD_MUTEX_ULOCK(&pingCb->pongMutex);
1268 1 : }
1269 :
1270 1 : STATIC void RsPingUrmaPingCbDeinit(unsigned int phyId, struct RsPingCtxCb *pingCb)
1271 : {
1272 1 : struct rs_cb *rscb = NULL;
1273 : int ret;
1274 :
1275 1 : ret = RsGetRsCb(phyId, &rscb);
1276 1 : if (ret != 0) {
1277 0 : hccp_err("RsGetRsCb failed, phyId[%u] invalid, ret %d", phyId, ret);
1278 0 : return;
1279 : }
1280 :
1281 1 : RS_PTHREAD_MUTEX_LOCK(&pingCb->pingMutex);
1282 1 : pingCb->taskStatus = RS_PING_TASK_RESET;
1283 1 : RS_PTHREAD_MUTEX_ULOCK(&pingCb->pingMutex);
1284 :
1285 1 : RsPingPongJettyDelTargetList(pingCb);
1286 :
1287 1 : RsPingCommonDeinitLocalJettyBuffer(pingCb);
1288 1 : RsPingCommonDeinitLocalJetty(rscb, pingCb, &pingCb->pongJetty);
1289 1 : RsPingCommonDeinitLocalJetty(rscb, pingCb, &pingCb->pingJetty);
1290 :
1291 1 : (void)RsUrmaDeleteContext(pingCb->udevCb.urmaCtx);
1292 1 : pingCb->udevCb.urmaCtx = NULL;
1293 : }
1294 :
1295 : // ping_pong_dfx function
1296 2 : STATIC void RsPingUrmaAddTargetSuccess(struct PingTargetInfo *target, struct RsPingTargetInfo *targetInfo)
1297 : {
1298 2 : urma_jetty_id_t jettyId = {0};
1299 2 : RsGetJettyInfo(&targetInfo->qpInfo, &jettyId, NULL);
1300 2 : hccp_info("target eid:%016llx:%016llx payload_size:%u add success, jettyId:%u uuid:0x%llx",
1301 : target->remoteInfo.eid.in6.subnetPrefix, target->remoteInfo.eid.in6.interfaceId,
1302 : target->payload.size, jettyId.id, targetInfo->uuid);
1303 2 : }
1304 :
1305 1 : STATIC void RsPingUrmaPingCbInitSuccess(unsigned int phyId, struct PingInitAttr *attr, unsigned int rdevIndex)
1306 : {
1307 1 : hccp_run_info("ping_cb init success, phyId:%u, eid:%016llx:%016llx, rdevIndex:%u",
1308 : phyId, attr->dev.ub.eid.in6.subnetPrefix, attr->dev.ub.eid.in6.interfaceId, rdevIndex);
1309 1 : }
1310 :
1311 1 : STATIC void RsPingUrmaCannotFindTargetNode(unsigned int i, int ret, struct PingTargetCommInfo target,
1312 : unsigned int phyId)
1313 : {
1314 1 : urma_jetty_id_t jettyId = {0};
1315 1 : RsGetJettyInfo(&target.qpInfo, &jettyId, NULL);
1316 :
1317 1 : hccp_err("rs_ping_urma_find_target_node i:%u failed, ret:%d eid:%016llx:%016llx jettyId:%u phyId:%u",i, ret,
1318 : target.eid.in6.subnetPrefix, target.eid.in6.interfaceId, jettyId.id, phyId);
1319 1 : }
1320 :
1321 : struct RsPingPongOps gRsPingUrmaOps = {
1322 : .checkPingFd = RsPingUrmaCheckFd,
1323 : .checkPongFd = RsPongUrmaCheckFd,
1324 : .initPingCb = RsPingUrmaPingCbInit,
1325 : .pingFindTargetNode = RsPingUrmaFindTargetNode,
1326 : .pingAllocTargetNode = RsPingUrmaAllocTargetNode,
1327 : .resetRecvBuffer = RsPingUrmaResetRecvBuffer,
1328 : .pingPostSend = RsPingUrmaPostSend,
1329 : .pingPollScq = RsPingUrmaPollScq,
1330 : .pingPollRcq = RsPingUrmaPollRcq,
1331 : .pongHandleSend = RsPongUrmaHandleSend,
1332 : .pongPollRcq = RsPongUrmaPollRcq,
1333 : .getTargetResult = RsPingUrmaGetTargetResult,
1334 : .pingFreeTargetNode = RsPingUrmaFreeTargetNode,
1335 : .deinitPingCb = RsPingUrmaPingCbDeinit,
1336 : };
1337 :
1338 : struct RsPingPongDfx gRsPingUrmaDfx = {
1339 : .addTargetSuccess = RsPingUrmaAddTargetSuccess,
1340 : .initPingCbSuccess = RsPingUrmaPingCbInitSuccess,
1341 : .pingCannotFindTargetNode = RsPingUrmaCannotFindTargetNode,
1342 : };
1343 :
1344 14 : struct RsPingPongOps *RsPingUrmaGetOps(void) {
1345 14 : return &gRsPingUrmaOps;
1346 : }
1347 :
1348 14 : struct RsPingPongDfx *RsPingUrmaGetDfx(void) {
1349 14 : return &gRsPingUrmaDfx;
1350 : }
|
