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