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