Star9
33
代码
代码
Issues5
Pull Requests7
项目讨论
Wiki
分析
Star9
33
已合并
knet合并kbdtoe api #185
penghd创建于 6月12日
knet合并kbdtoe api #185
已合并
penghd创建于 6月12日
从penghd/knetdtoe合入到openeuler/knetdtoe
共 18 个文件变更+2254-0
Mbuild.py+4-0
@@ -306,6 +306,10 @@ def build_rpm(stackType):
306 elif stackType == 'dtoe':306 elif stackType == 'dtoe':
307 shutil.copy2(f"{KNET_SOURCE_DIR}/src/knet/api/dtoe_api/include/knet_dtoe_api.h", f"{KNET_RPM_DIR}/BUILD/usr/include")307 shutil.copy2(f"{KNET_SOURCE_DIR}/src/knet/api/dtoe_api/include/knet_dtoe_api.h", f"{KNET_RPM_DIR}/BUILD/usr/include")
308 cmd = ["rpmbuild", f"-ba", f"--define", f"stack_type dtoe", f"{KNET_RPM_DIR}/SPECS/knet.spec"]308 cmd = ["rpmbuild", f"-ba", f"--define", f"stack_type dtoe", f"{KNET_RPM_DIR}/SPECS/knet.spec"]
309+ elif stackType == 'kbdtoe':
310+ shutil.copy2(f"{KNET_SOURCE_DIR}/src/knet/api/kbdtoe_api/include/kbdtoe.h", f"{KNET_RPM_DIR}/BUILD/usr/include")
311+ cmd = ["rpmbuild", f"-ba", f"--define", f"stack_type kbdtoe", f"{KNET_RPM_DIR}/SPECS/knet.spec"]
312+
309 output = subprocess.run(cmd, shell=False)313 output = subprocess.run(cmd, shell=False)
310 if output.returncode != 0:314 if output.returncode != 0:
311 logging.error(f"exec cmd fail. [{cmd}]")315 logging.error(f"exec cmd fail. [{cmd}]")
Aconf/kbdtoe_rsyslog.conf+9-0
@@ -0,0 +1,9 @@
1+# Restart rsyslog after updating this file: systemctl restart rsyslog
2+# outchannel format: name,file,size,rotate-cmd
3+$outchannel kbdtoe,/var/log/kbdtoe/kbdtoe.log,2097152,/usr/sbin/logrotate /etc/logrotate.d/kbdtoe
4+$FileCreateMode 0600
5+ 
6+if $programname contains 'libkbdtoe' then {
7+ :omfile:$kbdtoe
8+ stop
9+}
Aconf/logrotate/kbdtoe+18-0
@@ -0,0 +1,18 @@
1+/var/log/kbdtoe/kbdtoe.log
2+{
3+ daily
4+ compress
5+ copytruncate
6+ notifempty
7+ maxage 365
8+ rotate 10
9+ size 2048k
10+ missingok
11+ sharedscripts
12+ postrotate
13+ chmod 400 /var/log/kbdtoe/kbdtoe.log?* > /dev/null 2>&1 || true
14+ /usr/bin/systemctl kill -s HUP rsyslog.service >/dev/null 2>&1 || true
15+ endscript
16+}
17+ 
18+ 
Mpackage/knet.spec+10-0
@@ -42,6 +42,11 @@ cp -ar %{_topdir}/BUILD/usr/lib64/libknet_frame.so* %{buildroot}"/usr/lib64"
42 cp -rf %{_topdir}/BUILD/usr/bin/knet_mp_daemon %{buildroot}"/usr/bin"42 cp -rf %{_topdir}/BUILD/usr/bin/knet_mp_daemon %{buildroot}"/usr/bin"
43%elif "%{stack_type}" == "dtoe"43%elif "%{stack_type}" == "dtoe"
44 cp -ar %{_topdir}/BUILD/usr/include/* %{buildroot}"/usr/include"44 cp -ar %{_topdir}/BUILD/usr/include/* %{buildroot}"/usr/include"
45+%elif "%{stack_type}" == "kbdtoe"
46+ cp -ar %{_topdir}/BUILD/usr/lib64/libkbdtoe.so* %{buildroot}"/usr/lib64"
47+ cp -ar %{_topdir}/BUILD/usr/include/* %{buildroot}"/usr/include"
48+ cp -rf %{_topdir}/SOURCES/kbdtoe_rsyslog.conf %buildroot%{knetsysdir}/rsyslog.d/
49+ cp -rf %{_topdir}/SOURCES/logrotate/kbdtoe %buildroot%{knetsysdir}/logrotate.d/
45%endif50%endif
46cp -rf %{_topdir}/SOURCES/knet_comm.conf %buildroot%{knetsysdir}/knet/51cp -rf %{_topdir}/SOURCES/knet_comm.conf %buildroot%{knetsysdir}/knet/
47cp -rf %{_topdir}/SOURCES/knet_rsyslog.conf %buildroot%{knetsysdir}/rsyslog.d/52cp -rf %{_topdir}/SOURCES/knet_rsyslog.conf %buildroot%{knetsysdir}/rsyslog.d/
@@ -88,6 +93,11 @@ fi
88 %attr(550, root, root) /usr/bin/knet_mp_daemon93 %attr(550, root, root) /usr/bin/knet_mp_daemon
89%elif "%{stack_type}" == "dtoe"94%elif "%{stack_type}" == "dtoe"
90 %attr(644, root, root) /usr/include/knet_dtoe_api.h95 %attr(644, root, root) /usr/include/knet_dtoe_api.h
96+%elif "%{stack_type}" == "kbdtoe"
97+ %attr(550, root, root) /usr/lib64/libkbdtoe.so*
98+ %attr(644, root, root) /usr/include/kbdtoe.h
99+ %attr(640, root, root) %{knetsysdir}/rsyslog.d/kbdtoe_rsyslog.conf
100+ %attr(640, root, root) %{knetsysdir}/logrotate.d/kbdtoe
91%endif101%endif
92%attr(600, root, root) %{knetsysdir}/knet/knet_comm.conf102%attr(600, root, root) %{knetsysdir}/knet/knet_comm.conf
93%attr(640, root, root) %{knetsysdir}/rsyslog.d/knet_rsyslog.conf103%attr(640, root, root) %{knetsysdir}/rsyslog.d/knet_rsyslog.conf
Msrc/knet/CMakeLists.txt+49-0
@@ -144,6 +144,48 @@ if (${KNET_STACK_TYPE} STREQUAL "dtoe")
144 )144 )
145endif()145endif()
146 146 
147+if (${KNET_STACK_TYPE} STREQUAL "kbdtoe")
148+ add_library(${TARGET_CORE} SHARED)
149+ set_target_properties(${TARGET_CORE} PROPERTIES VERSION ${VERSION_NUM} SOVERSION ${VERSION_MAJOR})
150+ add_definitions(-DKNET_STATIC=static)
151+ 
152+ if(${KNET_BUILD_TYPE} STREQUAL "Release")
153+ add_custom_command(
154+ TARGET "${TARGET_CORE}" POST_BUILD
155+ COMMAND ${CMAKE_OBJCOPY} --only-keep-debug $<TARGET_FILE:${TARGET_CORE}> "${LIBRARY_DEBUG_SYMBOL_DIR}/$<TARGET_FILE_NAME:${TARGET_CORE}>.debug"
156+ COMMAND ${CMAKE_OBJCOPY} --strip-all $<TARGET_FILE:${TARGET_CORE}>
157+ COMMAND ${CMAKE_OBJCOPY} --add-gnu-debuglink=${LIBRARY_DEBUG_SYMBOL_DIR}/$<TARGET_FILE_NAME:${TARGET_CORE}>.debug $<TARGET_FILE:${TARGET_CORE}>
158+ COMMENT "Stripping debug symbols and generating .debug file"
159+ VERBATIM
160+ )
161+ endif()
162+ 
163+ target_link_libraries(${TARGET_CORE}
164+ base log
165+ libcjson.a
166+ -Wl,--whole-archive ${SECURE_LIB_NAME} -Wl,--no-whole-archive
167+ )
168+ 
169+ set(TARGET_KBDTOE kbdtoe)
170+ add_library(${TARGET_KBDTOE} SHARED)
171+ set_target_properties(${TARGET_KBDTOE} PROPERTIES VERSION ${VERSION_NUM} SOVERSION ${VERSION_MAJOR})
172+ if(${KNET_BUILD_TYPE} STREQUAL "Release")
173+ add_custom_command(
174+ TARGET "${TARGET_KBDTOE}" POST_BUILD
175+ COMMAND ${CMAKE_OBJCOPY} --only-keep-debug $<TARGET_FILE:${TARGET_KBDTOE}> "${LIBRARY_DEBUG_SYMBOL_DIR}/$<TARGET_FILE_NAME:${TARGET_KBDTOE}>.debug"
176+ COMMAND ${CMAKE_OBJCOPY} --strip-all $<TARGET_FILE:${TARGET_KBDTOE}>
177+ COMMAND ${CMAKE_OBJCOPY} --add-gnu-debuglink=${LIBRARY_DEBUG_SYMBOL_DIR}/$<TARGET_FILE_NAME:${TARGET_KBDTOE}>.debug $<TARGET_FILE:${TARGET_KBDTOE}>
178+ COMMENT "Stripping debug symbols and generating .debug file"
179+ VERBATIM
180+ )
181+ endif()
182+ target_link_libraries(${TARGET_KBDTOE}
183+ kbdtoe_api base log
184+ ${TARGET_CORE}
185+ dtoe
186+ )
187+endif()
188+ 
147add_library(${TARGET_FRAME} SHARED)189add_library(${TARGET_FRAME} SHARED)
148# 每次更新无需手动更新VERSION和RELEASE，自动在knet.spec文件中获取190# 每次更新无需手动更新VERSION和RELEASE，自动在knet.spec文件中获取
149set_target_properties(${TARGET_FRAME} PROPERTIES VERSION ${VERSION_NUM} SOVERSION ${VERSION_MAJOR})191set_target_properties(${TARGET_FRAME} PROPERTIES VERSION ${VERSION_NUM} SOVERSION ${VERSION_MAJOR})
@@ -175,6 +217,13 @@ if (${KNET_STACK_TYPE} STREQUAL "dtoe")
175 )217 )
176endif()218endif()
177 219 
220+if (${KNET_STACK_TYPE} STREQUAL "kbdtoe")
221+ target_link_libraries(${TARGET_FRAME}
222+ log
223+ ${TARGET_CORE}
224+ )
225+endif()
226+ 
178add_compile_definitions(KNET_VERSION="${VERSION_NUM}")227add_compile_definitions(KNET_VERSION="${VERSION_NUM}")
179if (${KNET_STACK_TYPE} STREQUAL "dp")228if (${KNET_STACK_TYPE} STREQUAL "dp")
180 add_executable(${TARGET_DAEMON} daemon/knet_mp_daemon.c)229 add_executable(${TARGET_DAEMON} daemon/knet_mp_daemon.c)
Msrc/knet/api/CMakeLists.txt+4-0
@@ -6,4 +6,8 @@ endif()
6 6 
7if (${KNET_STACK_TYPE} STREQUAL "dtoe")7if (${KNET_STACK_TYPE} STREQUAL "dtoe")
8 add_subdirectory(dtoe_api)8 add_subdirectory(dtoe_api)
9+endif()
10+ 
11+if (${KNET_STACK_TYPE} STREQUAL "kbdtoe")
12+ add_subdirectory(kbdtoe_api)
9endif()13endif()
Asrc/knet/api/kbdtoe_api/CMakeLists.txt+1-0
@@ -0,0 +1 @@
1+ADD_MODULE(kbdtoe_api)
Asrc/knet/api/kbdtoe_api/include/kbdtoe.h+151-0
@@ -0,0 +1,151 @@
1+/*
2+* Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
3+* kraio is licensed under the Mulan PSL v2.
4+* You can use this software according to the terms and conditions of the Mulan PSL v2.
5+* You may obtain a copy of Mulan PSL v2 at:
6+* http://license.coscl.org.cn/MulanPSL2
7+* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
8+* IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
9+* PURPOSE.
10+* See the Mulan PSL v2 for more details.
11+*
12+* Encapsulate dtoe interface
13+*/
14+#ifndef KB_DTOE_H
15+#define KB_DTOE_H
16+#ifdef __cplusplus
17+extern "C" {
18+#endif
19+#include <string.h>
20+#include <stdio.h>
21+#include <stdbool.h>
22+#include <stdint.h>
23+#include <sys/uio.h>
24+ 
25+/******************************************************************
26+ Prototype : kbdtoe_init
27+ Description : dtoe 初始化函数
28+ Input : const char* dtoe_ip dtoe网卡
29+ : unsigned int max_conn_num 最大连接数
30+ Return Value : 0 success， others-failed
31+ **************************************************************/
32+int kbdtoe_init(const char* dtoe_ip, unsigned int max_conn_num);
33+ 
34+/******************************************************************
35+ Prototype : kbdtoe_uninit
36+ Description : dtoe 销毁函数
37+ Return Value : None
38+ **************************************************************/
39+void kbdtoe_uninit();
40+ 
41+/******************************************************************
42+ Prototype : kbdtoe_conn_start_offload
43+ Description : 单个TCP连接开始卸载的函数，必须在accept之后调用
44+ Input : int sockfd TCP accept之后的sock fd
45+ Return Value : 0 success， others-failed
46+ **************************************************************/
47+int kbdtoe_conn_start_offload(int sockfd);
48+ 
49+typedef struct flexda_dtoe_iovec {
50+ void* iov_base;
51+ size_t iov_len;
52+} flexda_dtoe_iovec_s;
53+ 
54+enum kbdtoe_recv_event_type {
55+ KBDTOE_RX_EVENT_NORMAL,
56+ KBDTOE_RX_EVENT_LEAK,
57+};
58+ 
59+struct kbdtoe_recv_events {
60+ int sockfd;
61+ uint32_t iov_cnt;
62+ enum kbdtoe_recv_event_type type;
63+};
64+ 
65+/******************************************************************
66+ Prototype : kbdtoe_thread_poll
67+ Description : 指定线程的轮询事件函数
68+ Intput : int thread_index 线程的index
69+ Output ：1) struct kbdtoe_recv_events recv_events[] 线程可读的事件信息
70+ 2) int *nr_recv_event 线程可读的事事件个数
71+ Return Value : true-还有事件需要继续poll， false-没有事件了
72+ **************************************************************/
73+bool kbdtoe_thread_poll(int thread_index, struct kbdtoe_recv_events recv_events[], int *nr_recv_event);
74+ 
75+/******************************************************************
76+ Prototype : kbdtoe_read
77+ Description : 语义类似read函数
78+ Return Value : 读到的字节数
79+ **************************************************************/
80+ssize_t kbdtoe_read(int fd, void *buf, size_t nbyte);
81+ 
82+/******************************************************************
83+ Prototype : kbdtoe_write
84+ Description : 语义类似write函数
85+ Return Value : 写入的字节数
86+ **************************************************************/
87+ssize_t kbdtoe_write(int fd, const void *buf, size_t nbyte);
88+ 
89+/******************************************************************
90+ Prototype : kbdtoe_writev
91+ Description : 语义类似writev函数
92+ Return Value : 写入的字节数
93+ **************************************************************/
94+ssize_t kbdtoe_writev(int fd, const struct iovec *iov, int iovcnt);
95+ 
96+/******************************************************************
97+ Prototype : kbdtoe_close
98+ Description : 释放单个dtoe连接的资源
99+ Return Value : 0 success， others-failed
100+ **************************************************************/
101+int kbdtoe_close(int fd);
102+ 
103+/******************************************************************
104+ Prototype : kbdtoe_is_conn_offload_success
105+ Description : 判断单个dtoe连接是否卸载完成
106+ Input : int sockfd TCP accept之后的sock fd
107+ Return Value : 0 success， others-failed
108+ **************************************************************/
109+bool kbdtoe_is_conn_offload_success(int sockfd);
110+ 
111+/******************************************************************
112+ Prototype : kbdtoe_is_conn_offload
113+ Description : 判断单个socket fd 是否进行dtoe 卸载
114+ Input : int sockfd TCP accept之后的sock fd
115+ Return Value : 0 success， others-failed
116+ **************************************************************/
117+bool kbdtoe_is_conn_offload(int sockfd);
118+ 
119+/******************************************************************
120+ Prototype : kbdtoe_conn_status_for_close
121+ Description : 设置卸载TCP进行预上载
122+ Input : int sockfd TCP accept之后的sock fd
123+ Return Value : None
124+ **************************************************************/
125+void kbdtoe_conn_status_for_close(int sockfd);
126+ 
127+/******************************************************************
128+ Prototype : kbdtoe_is_channel_epoll_fd
129+ Description : 判断fd是否为dtoe通信的channel epoll fd
130+ Input : int thread_idx 线程index
131+ : int fd 待判断的fd
132+ Return Value : true-是channel epoll fd， false-非channel epoll fd
133+ **************************************************************/
134+bool kbdtoe_is_channel_epoll_fd(int thread_idx, int fd);
135+ 
136+/******************************************************************
137+ Prototype : kbdtoe_register_channel_fd_to_epoll
138+ Description : 将dtoe通信的channel fd注册到epoll中
139+ Input : int thread_idx 线程index
140+ : int epoll_fd epoll fd
141+ Return Value : 0 success， others-failed
142+ **************************************************************/
143+int kbdtoe_register_channel_fd_to_epoll(int thread_idx, int epoll_fd);
144+ 
145+typedef void (*dtoe_close_done_callback_t) (int sockfd);
146+void register_dtoe_close_done_callback(dtoe_close_done_callback_t cb);
147+#ifdef __cplusplus
148+}
149+#endif
150+#endif
151+ 
Asrc/knet/api/kbdtoe_api/kbdtoe_api.spec+4-0
@@ -0,0 +1,4 @@
1+{
2+ "depends" : "securec, cJSON, dtoe, base, log",
3+ "description" : "knet kbdtoe_api module"
4+}
Asrc/knet/api/kbdtoe_api/src/kbdtoe.c+599-0
@@ -0,0 +1,599 @@
1+/*
2+* Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
3+* redis dtoe is licensed under the Mulan PSL v2.
4+* You can use this software according to the terms and conditions of the Mulan PSL v2.
5+* You may obtain a copy of Mulan PSL v2 at:
6+* http://license.coscl.org.cn/MulanPSL2
7+* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
8+* IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
9+* PURPOSE.
10+* See the Mulan PSL v2 for more details.
11+*
12+* Encapsulate dtoe interface
13+*/
14+#include "kbdtoe.h"
15+#include <sys/resource.h>
16+#include <sys/epoll.h>
17+#include "kbdtoe_base.h"
18+#include "kbdtoe_mempool_mr.h"
19+#include "securec.h"
20+ 
21+static uint8_t g_thread_num = 1; // 默认单线程
22+static uint8_t g_channel_num = 1; // 默认每个线程1对信道,redis 单实例最大支持1W，单对信道支持8k
23+static libdtoe_thread_pool_s g_thread_pool[DTOE_THREAD_MAX];
24+dtoe_close_done_callback_t g_dtoe_close_done_callback;
25+struct libdtoe_conn_readable_event_head g_readable_event_head;
26+static unsigned int g_max_conn_num = 0;
27+static uint64_t g_dev_sn = 0;
28+static uint32_t g_numa_id = 0;
29+static void **g_fd_to_conn;
30+static int g_libdtoe_fd_max = 0;
31+ 
32+static int libdtoe_fd_init(void)
33+{
34+ if (g_libdtoe_fd_max > 0) {
35+ KBDTOE_INFO("Dtoe fd module reinit");
36+ return 0;
37+ }
38+ 
39+ struct rlimit limit = {0};
40+ if (getrlimit(RLIMIT_NOFILE, &limit) != 0) {
41+ KBDTOE_ERR("Get Linux fd limit fail");
42+ return -1;
43+ }
44+ 
45+ g_fd_to_conn = (void **)calloc(limit.rlim_cur, sizeof(void *));
46+ if (g_fd_to_conn == NULL) {
47+ KBDTOE_ERR("Alloc Dtoe fd mng failed, fd cur limit %u!", limit.rlim_cur);
48+ return -1;
49+ }
50+ g_libdtoe_fd_max = limit.rlim_cur;
51+ KBDTOE_INFO("Dtoe fd max %d", g_libdtoe_fd_max);
52+ 
53+ return 0;
54+}
55+ 
56+static void libdtoe_fd_dinit(void)
57+{
58+ if (g_fd_to_conn) {
59+ free(g_fd_to_conn);
60+ g_fd_to_conn = NULL;
61+ }
62+}
63+ 
64+static inline void set_conn_by_fd(int fd, void *conn)
65+{
66+ if (fd >= 0 && fd < g_libdtoe_fd_max) {
67+ if ((g_fd_to_conn[fd] && conn != NULL) || (g_fd_to_conn[fd] == NULL && conn == NULL)) {
68+ KBDTOE_ERR("set_conn_by_fd error, fd:%d, oldconn:%p, conn:%p\n", fd, g_fd_to_conn[fd], conn);
69+ }
70+ g_fd_to_conn[fd] = conn;
71+ } else {
72+ KBDTOE_ERR("set_conn_by_fd error, invalid fd:%d, conn:%p\n", fd, conn);
73+ }
74+}
75+ 
76+void* get_conn_by_fd(int fd)
77+{
78+ if (fd >= 0 && fd < g_libdtoe_fd_max) {
79+ return g_fd_to_conn[fd]; // maybe NULL
80+ }
81+ return NULL;
82+}
83+ 
84+void register_dtoe_close_done_callback(dtoe_close_done_callback_t cb)
85+{
86+ if (cb == NULL) {
87+ KBDTOE_ERR("dtoe close done callback is null");
88+ return;
89+ }
90+ g_dtoe_close_done_callback = cb;
91+}
92+ 
93+inline libdtoe_thread_pool_s* get_thread_pool(int idx)
94+{
95+ if (idx < 0 || idx >= DTOE_THREAD_MAX) {
96+ KBDTOE_ERR("invalid thread pool index, idx=%d", idx);
97+ return NULL;
98+ }
99+ return &g_thread_pool[idx];
100+}
101+ 
102+/***************************** callback function start *****************************/
103+void libdtoe_close_done(void *dtoe_conn)
104+{
105+ libdtoe_conn_s *conn = (libdtoe_conn_s *)flexda_dtoe_get_ulp_user_data(dtoe_conn);
106+ if (conn == NULL) {
107+ KBDTOE_ERR("dtoe close done, conn is null");
108+ return;
109+ }
110+ libdtoe_thread_pool_s *thread_info = (libdtoe_thread_pool_s *)conn->thread_pool_ptr;
111+ 
112+ if (g_dtoe_close_done_callback != NULL) {
113+ g_dtoe_close_done_callback(conn->fd);
114+ } else {
115+ KBDTOE_ERR("dtoe close done callback need register");
116+ }
117+ 
118+ pthread_spin_lock(&(thread_info->connection.offload_lock));
119+ set_conn_by_fd(conn->fd, NULL);
120+ TAILQ_INSERT_TAIL(&thread_info->connection.free_conns, conn, free_node);
121+ pthread_spin_unlock(&(thread_info->connection.offload_lock));
122+}
123+ 
124+void libdtoe_prepare_close_done(void *dtoe_conn)
125+{
126+ libdtoe_conn_s *conn = (libdtoe_conn_s *)flexda_dtoe_get_ulp_user_data(dtoe_conn);
127+ conn->conn_status = DTOE_CONN_CLOSED;
128+ libdtoe_thread_pool_s *thread_info = (libdtoe_thread_pool_s *)conn->thread_pool_ptr;
129+ thread_info->connection.offload_num--;
130+ if (conn->has_readable_event) {
131+ TAILQ_REMOVE(&g_readable_event_head, conn, readable_event_node);
132+ conn->has_readable_event = 0;
133+ }
134+ 
135+ pthread_spin_lock(&conn->send.pending_send_lock);
136+ libdtoe_pending_send_s *req = TAILQ_FIRST(&conn->send.unack_req);
137+ while (req != NULL) {
138+ if (likely(req->free_cb != NULL)) {
139+ req->free_cb(req->sockfd, req->wr_id);
140+ }
141+ TAILQ_REMOVE(&conn->send.unack_req, req, node);
142+ free(req);
143+ req = TAILQ_FIRST(&conn->send.unack_req);
144+ }
145+ req = TAILQ_FIRST(&conn->send.free_req);
146+ while (req != NULL) {
147+ TAILQ_REMOVE(&conn->send.free_req, req, node);
148+ free(req);
149+ req = TAILQ_FIRST(&conn->send.free_req);
150+ }
151+ pthread_spin_unlock(&conn->send.pending_send_lock);
152+ 
153+ flexda_dtoe_close(conn->fd, dtoe_conn);
154+}
155+ 
156+void libdtoe_conn_async_offload_done(void *dtoe_conn, uint8_t rsp_status)
157+{
158+ libdtoe_conn_s *conn = (libdtoe_conn_s*)flexda_dtoe_get_ulp_user_data(dtoe_conn);
159+ libdtoe_thread_pool_s *thread_info = (libdtoe_thread_pool_s*) conn->thread_pool_ptr;
160+ if (rsp_status != 0 || conn->conn_status > DTOE_CONN_WORKING) {
161+ KBDTOE_ERR("sockfd=%d async offload failed, rsp_status:%d conn_status:%d", conn->fd, rsp_status, conn->conn_status);
162+ flexda_dtoe_prepare_close(dtoe_conn);
163+ conn->offload_status = DTOE_OFFLOAD_FAIL;
164+ return;
165+ }
166+ if (conn->conn_status == DTOE_CONN_CREATING) {
167+ conn->conn_status = DTOE_CONN_WORKING;
168+ thread_info->connection.offload_num++;
169+ }
170+ conn->offload_status = DTOE_OFFLOAD_SUCCESS;
171+ if (flexda_dtoe_get_leaked_packet_size(conn->dtoe_conn)) {
172+ conn->has_readable_event = 1;
173+ conn->poll_mask = 0;
174+ TAILQ_INSERT_TAIL(&g_readable_event_head, conn, readable_event_node);
175+ }
176+}
177+ 
178+static inline bool is_send_complete(libdtoe_conn_s *conn, libdtoe_pending_send_s *rnode)
179+{
180+ bool complete = false;
181+ if (likely(conn->send.last_ack_sn < conn->send.comp_sn)) {
182+ if (rnode->send_sn > conn->send.last_ack_sn && rnode->send_sn <= conn->send.comp_sn) {
183+ complete = true;
184+ }
185+ } else if (conn->send.last_ack_sn > conn->send.comp_sn) {
186+ if (rnode->send_sn <= conn->send.comp_sn || rnode->send_sn > conn->send.last_ack_sn) {
187+ complete = true;
188+ }
189+ }
190+ return complete;
191+}
192+ 
193+static void libdtoe_send_complete(void* dtoe_conn, flexda_dtoe_tx_event_s* event)
194+{
195+ libdtoe_conn_s *conn = (libdtoe_conn_s *)flexda_dtoe_get_ulp_user_data(dtoe_conn);
196+ if (unlikely(conn == NULL || event == NULL)) {
197+ return;
198+ }
199+ 
200+ conn->send.comp_sn = event->finish_msn;
201+ 
202+ pthread_spin_lock(&conn->send.pending_send_lock);
203+ 
204+ libdtoe_pending_send_s *req = TAILQ_FIRST(&conn->send.unack_req);
205+ while (req != NULL) {
206+ if (!is_send_complete(conn, req)) {
207+ break;
208+ }
209+ 
210+ if (likely(req->free_cb != NULL)) {
211+ req->free_cb(req->sockfd, req->wr_id);
212+ }
213+ 
214+ conn->send.last_ack_sn = req->send_sn;
215+ 
216+ TAILQ_REMOVE(&conn->send.unack_req, req, node);
217+ TAILQ_INSERT_TAIL(&conn->send.free_req, req, node);
218+ 
219+ req = TAILQ_FIRST(&conn->send.unack_req);
220+ }
221+ 
222+ pthread_spin_unlock(&conn->send.pending_send_lock);
223+}
224+ 
225+static void libdtoe_receive_notify(void* dtoe_conn, int iov_cnt)
226+{
227+ libdtoe_conn_s *conn = (libdtoe_conn_s *)flexda_dtoe_get_ulp_user_data(dtoe_conn);
228+ if (unlikely(conn == NULL)) {
229+ return;
230+ }
231+ libdtoe_recv_channel_wrapper_s *recv_channel = (libdtoe_recv_channel_wrapper_s *)conn->recv_channel;
232+ 
233+ if (likely(iov_cnt > 0)) {
234+ if (conn->recv_event_index == -1) {
235+ recv_channel->events[recv_channel->next_event_idx].sockfd = conn->fd;
236+ recv_channel->events[recv_channel->next_event_idx].iov_cnt = iov_cnt;
237+ recv_channel->events[recv_channel->next_event_idx].type = KBDTOE_RX_EVENT_NORMAL;
238+ conn->recv_event_index = recv_channel->next_event_idx;
239+ ++recv_channel->next_event_idx;
240+ } else {
241+ recv_channel->events[conn->recv_event_index].iov_cnt += iov_cnt;
242+ }
243+ } else if (iov_cnt == 0) {
244+ recv_channel->events[recv_channel->next_event_idx].sockfd = conn->fd;
245+ recv_channel->events[recv_channel->next_event_idx].iov_cnt = iov_cnt;
246+ recv_channel->events[recv_channel->next_event_idx].type = KBDTOE_RX_EVENT_NORMAL;
247+ ++recv_channel->next_event_idx;
248+ }
249+}
250+ 
251+/***************************** callback function end *****************************/
252+ 
253+static flexda_dtoe_ulp_ops_s g_dtoe_ulp_ops = {
254+ .send_complete = libdtoe_send_complete,
255+ .recv_notify = libdtoe_receive_notify,
256+ .close_done = libdtoe_close_done,
257+ .prepare_close_done = libdtoe_prepare_close_done,
258+ .conn_async_offload_done = libdtoe_conn_async_offload_done,
259+};
260+ 
261+static int libdtoe_init_conn_pool_per_thread()
262+{
263+ int conn_num = g_max_conn_num;
264+ for (int i = 0; i < g_thread_num; i++) {
265+ g_thread_pool[i].connection.conn_pool = (libdtoe_conn_s*)malloc(conn_num * sizeof(libdtoe_conn_s));
266+ if (!g_thread_pool[i].connection.conn_pool) {
267+ return DTOE_FAIL;
268+ }
269+ (void)memset_s((void*)g_thread_pool[i].connection.conn_pool, conn_num * sizeof(libdtoe_conn_s), 0, conn_num * sizeof(libdtoe_conn_s));
270+ }
271+ libdtoe_conn_s *conn = NULL;
272+ for (int i = 0; i < g_thread_num; ++i) {
273+ TAILQ_INIT(&g_thread_pool[i].connection.free_conns);
274+ pthread_spin_init(&g_thread_pool[i].connection.offload_lock, PTHREAD_PROCESS_PRIVATE);
275+ for (int j = 0; j < conn_num; ++j) {
276+ conn = &g_thread_pool[i].connection.conn_pool[j];
277+ TAILQ_INSERT_TAIL(&g_thread_pool[i].connection.free_conns, conn, free_node);
278+ }
279+ }
280+ return DTOE_SUCCESS;
281+}
282+ 
283+static int libdtoe_destory_mbuf(libdtoe_thread_pool_s *thread_info)
284+{
285+ free(thread_info->send_mr->addr);
286+ return DTOE_SUCCESS;
287+}
288+ 
289+static int libdtoe_prepare_mbuf(libdtoe_thread_pool_s *thread_info)
290+{
291+ char *buf = NULL;
292+ int buf_size = (DTOE_SEND_BUF_LEN * DTOE_MAX_CONN_PER_THREAD);
293+ buf_size += DTOE_PAGE_SIZE;
294+ buf = (char* )aligned_alloc(DTOE_PAGE_SIZE, buf_size);
295+ if (buf == NULL) {
296+ return DTOE_FAIL;
297+ }
298+ (void)memset_s(buf, buf_size, 0, buf_size);
299+ thread_info->send_mr = (flexda_dtoe_mr_s*)malloc(sizeof(flexda_dtoe_mr_s));
300+ if (thread_info->send_mr == NULL) {
301+ free(buf);
302+ return DTOE_FAIL;
303+ }
304+ int ret = flexda_dtoe_reg_mr(g_dev_sn, buf, buf_size, thread_info->send_mr);
305+ if (ret != 0) {
306+ free(buf);
307+ free(thread_info->send_mr);
308+ thread_info->send_mr = NULL;
309+ return DTOE_FAIL;
310+ }
311+ KBDTOE_INFO("libdtoe_prepare_mbuf is success");
312+ return 0;
313+}
314+ 
315+static int libdtoe_all_threads_create_channel()
316+{
317+ int ret;
318+ for (int i = 0; i < g_thread_num; ++i) {
319+ ret = libdtoe_prepare_mbuf(&g_thread_pool[i]);
320+ if (ret != DTOE_SUCCESS) {
321+ KBDTOE_ERR("libdtoe prepare mbuf failed");
322+ return DTOE_FAIL;
323+ }
324+ for (int j = 0; j < (g_channel_num / g_thread_num); ++j) {
325+ g_thread_pool[i].send_channel[j] = (flexda_send_channel_s*)malloc(sizeof(flexda_send_channel_s));
326+ if (g_thread_pool[i].send_channel[j] == NULL) {
327+ KBDTOE_ERR("malloc send channel wrapper failed");
328+ goto cleanup;
329+ }
330+ ret = flexda_dtoe_create_send_channel(g_dev_sn, FLEXDA_EPOLL_SCHD, g_thread_pool[i].send_channel[j]);
331+ if (ret != 0) {
332+ KBDTOE_ERR("create send channel failed, ret %d", ret);
333+ free(g_thread_pool[i].send_channel[j]);
334+ g_thread_pool[i].send_channel[j] = NULL;
335+ goto cleanup;
336+ }
337+ g_thread_pool[i].send_channel_fd[j] = g_thread_pool[i].send_channel[j]->epoll_fd;
338+ g_thread_pool[i].recv_channel[j] = (libdtoe_recv_channel_wrapper_s*)malloc(sizeof(libdtoe_recv_channel_wrapper_s));
339+ if (g_thread_pool[i].recv_channel[j] == NULL) {
340+ KBDTOE_ERR("malloc recv channel wrapper failed");
341+ goto cleanup;
342+ }
343+ g_thread_pool[i].recv_channel[j]->next_event_idx = 0;
344+ ret = flexda_dtoe_create_receive_channel(g_dev_sn, FLEXDA_EPOLL_SCHD, &g_thread_pool[i].recv_channel[j]->channel);
345+ if (ret != 0) {
346+ KBDTOE_ERR("create recv channel failed, ret %d", ret);
347+ free(g_thread_pool[i].recv_channel[j]);
348+ g_thread_pool[i].recv_channel[j] = NULL;
349+ goto cleanup;
350+ }
351+ g_thread_pool[i].recv_channel_fd[j] = g_thread_pool[i].recv_channel[j]->channel.epoll_fd;
352+ 
353+ g_thread_pool[i].channel_num++;
354+ }
355+ }
356+ return DTOE_SUCCESS;
357+cleanup:
358+ for (int i = 0; i < g_thread_num; ++i) {
359+ for (int j = 0; j < (g_channel_num / g_thread_num); ++j) {
360+ if (g_thread_pool[i].send_channel[j]) {
361+ flexda_dtoe_destroy_send_channel(g_thread_pool[i].send_channel[j]);
362+ free(g_thread_pool[i].send_channel[j]);
363+ g_thread_pool[i].send_channel[j] = NULL;
364+ }
365+ if (g_thread_pool[i].recv_channel[j]) {
366+ flexda_dtoe_destroy_receive_channel(&g_thread_pool[i].recv_channel[j]->channel);
367+ free(g_thread_pool[i].recv_channel[j]);
368+ g_thread_pool[i].recv_channel[j] = NULL;
369+ }
370+ }
371+ libdtoe_destory_mbuf(&g_thread_pool[i]);
372+ }
373+ return DTOE_FAIL;
374+}
375+ 
376+int kbdtoe_init(const char* dtoe_ip, unsigned int max_conn_num)
377+{
378+ kbdtoe_log_init();
379+ int ret = 0;
380+ g_max_conn_num = max_conn_num;
381+ ret = libdtoe_fd_init();
382+ if (ret != 0) {
383+ KBDTOE_ERR("dtoe fd init failed, ret %d", ret);
384+ return DTOE_FAIL;
385+ }
386+ 
387+ flexda_dtoe_ulp_ops_register(&g_dtoe_ulp_ops);
388+ ret = flexda_dtoe_ulp_config_set(g_channel_num);
389+ if (ret != 0) {
390+ KBDTOE_ERR("dtoe ulp config set failed, ret %d", ret);
391+ goto fail_fd_init;
392+ }
393+ 
394+ ret = flexda_dtoe_init();
395+ if (ret != 0) {
396+ KBDTOE_ERR("dtoe init failed, ret %d", ret);
397+ goto fail_fd_init;
398+ }
399+ 
400+#define BIND_INTERVAL 1
401+#define BIND_TIMES 10
402+ uint8_t times = 0;
403+ do {
404+ sleep(BIND_INTERVAL);
405+ ret = flexda_dtoe_bind_addr(dtoe_ip, &g_dev_sn, &g_numa_id);
406+ ++times;
407+ } while (times <= BIND_TIMES && ret != 0);
408+ if (ret != 0) {
409+ KBDTOE_ERR("dtoe bind addr failed, ret %d, dtoe_ip %s", ret, dtoe_ip);
410+ goto fail_dtoe_init;
411+ }
412+ 
413+ ret = libdtoe_init_conn_pool_per_thread();
414+ if (ret != DTOE_SUCCESS) {
415+ KBDTOE_ERR("kbdtoe init conn pool failed");
416+ goto fail_dtoe_init;
417+ }
418+ TAILQ_INIT(&g_readable_event_head);
419+ ret = libdtoe_all_threads_create_channel();
420+ if (ret != DTOE_SUCCESS) {
421+ KBDTOE_ERR("kbdtoe init thread channel failed");
422+ goto fail_dtoe_init;
423+ }
424+ ret = kbdtoe_mempool_init();
425+ if (ret != DTOE_SUCCESS) {
426+ KBDTOE_ERR("kbdtoe init memory pool failed");
427+ goto fail_dtoe_init;
428+ }
429+ KBDTOE_INFO("kbdtoe_init success !!!\n");
430+ return DTOE_SUCCESS;
431+ 
432+fail_dtoe_init:
433+ flexda_dtoe_uninit();
434+fail_fd_init:
435+ libdtoe_fd_dinit();
436+ return DTOE_FAIL;
437+}
438+ 
439+uint64_t get_dtoe_dev_sn()
440+{
441+ return g_dev_sn;
442+}
443+ 
444+int libdtoe_conn_init(libdtoe_thread_pool_s *thread_info, libdtoe_conn_s* libdtoe_conn)
445+{
446+ libdtoe_conn->recv_desc_num = 0;
447+ libdtoe_conn->thread_pool_ptr = (void*) thread_info;
448+ libdtoe_conn->poll_mask = 0;
449+ libdtoe_conn->recv_event_index = -1;
450+ libdtoe_conn->send.last_send_sn = 0;
451+ libdtoe_conn->send.comp_sn = 0;
452+ libdtoe_conn->send.last_ack_sn = 0;
453+ TAILQ_INIT(&libdtoe_conn->send.unack_req);
454+ TAILQ_INIT(&libdtoe_conn->send.free_req);
455+ pthread_spin_init(&libdtoe_conn->send.pending_send_lock, PTHREAD_PROCESS_PRIVATE);
456+ return DTOE_SUCCESS;
457+}
458+ 
459+int kbdtoe_conn_start_offload(int sockfd)
460+{
461+ int ret;
462+ libdtoe_thread_pool_s *thread_info = (libdtoe_thread_pool_s *)get_thread_pool(0);
463+ pthread_spin_lock(&(thread_info->connection.offload_lock));
464+ libdtoe_conn_s *conn = TAILQ_FIRST(&thread_info->connection.free_conns);
465+ if (conn == NULL) {
466+ pthread_spin_unlock(&(thread_info->connection.offload_lock));
467+ KBDTOE_ERR("kbdtoe start offload failed for no free conns");
468+ return DTOE_FAIL;
469+ }
470+ TAILQ_REMOVE(&thread_info->connection.free_conns, conn, free_node);
471+ pthread_spin_unlock(&(thread_info->connection.offload_lock));
472+ libdtoe_conn_init(thread_info, conn);
473+ 
474+ thread_info->epoch %= thread_info->channel_num;
475+ 
476+ conn->offload_status = DTOE_OFFLOAD_START;
477+ conn->conn_status = DTOE_CONN_CREATING;
478+ conn->fd = sockfd;
479+ 
480+ flexda_dtoe_offload_in_s in = {0};
481+ in.user_data = conn;
482+ in.send_channel = thread_info->send_channel[thread_info->epoch];
483+ in.recv_channel = (flexda_recv_channel_s *)thread_info->recv_channel[thread_info->epoch];
484+ 
485+ flexda_dtoe_offload_out_s out = {0};
486+ ret = flexda_dtoe_start_conn_offload(sockfd, &in, &out);
487+ if (ret != 0) {
488+ KBDTOE_ERR("kbdtoe start offload failed, sockfd:%d, ret:%d", sockfd, ret);
489+ return DTOE_FAIL;
atomgit-bot
atomgit-botatomgit-bot6月12日

🟠 High Priority

conn 已在 line 470 从 free_conns 中移除，但 flexda_dtoe_start_conn_offload 失败后直接 return DTOE_FAIL，未将 conn 插回 free_conns，导致该连接槽位永久丢失。长时间运行会耗尽连接池，新连接无法卸载。

likedislike
不准确？
490+ }
491+ 
492+ set_conn_by_fd(sockfd, conn);
493+ conn->dtoe_conn = out.dtoe_conn;
494+ /* Initialize last_ack_sn and comp_sn for is_send_comlete() */
495+ conn->send.comp_sn = out.send_sn - 1;
496+ conn->send.last_ack_sn = out.send_sn - 1;
497+ conn->send_channel = in.send_channel;
498+ conn->recv_channel = (flexda_recv_channel_s *)in.recv_channel;
499+ conn->leaked_size = 0;
500+ conn->read_leaked_offset = 0;
501+ conn->has_readable_event = 0;
502+ return DTOE_SUCCESS;
503+}
504+ 
505+inline bool kbdtoe_is_conn_offload_success(int sockfd)
506+{
507+ libdtoe_conn_s *conn = (libdtoe_conn_s *)get_conn_by_fd(sockfd);
508+ if (conn == NULL) {
509+ return false;
510+ }
511+ return conn->offload_status == DTOE_OFFLOAD_SUCCESS;
512+}
513+ 
514+void kbdtoe_conn_status_for_close(int sockfd)
515+{
516+ libdtoe_conn_s *conn = (libdtoe_conn_s *)get_conn_by_fd(sockfd);
517+ if (conn == NULL) {
518+ return;
519+ }
520+ conn->offload_status = DTOE_OFFLOAD_PRECLOSE;
521+}
522+ 
523+bool kbdtoe_is_conn_offload(int sockfd)
524+{
525+ libdtoe_conn_s *conn = (libdtoe_conn_s *)get_conn_by_fd(sockfd);
526+ if (conn == NULL) {
527+ return false;
528+ }
529+ return true;
530+}
531+ 
532+int kbdtoe_close(int fd)
533+{
534+ libdtoe_conn_s *conn = (libdtoe_conn_s *)get_conn_by_fd(fd);
535+ if (conn != NULL) {
536+ flexda_dtoe_prepare_close(conn->dtoe_conn);
537+ }
538+ return 0;
539+}
540+ 
541+void kbdtoe_uninit()
542+{
543+ flexda_dtoe_uninit();
544+ kbdtoe_mempool_destroy();
545+ kbdtoe_log_uninit();
546+}
547+ 
548+bool kbdtoe_is_channel_epoll_fd(int thread_idx, int fd)
549+{
550+ if (thread_idx < 0 || thread_idx >= g_thread_num) {
551+ KBDTOE_ERR("Invalid thread_idx %d", thread_idx);
552+ return false;
553+ }
554+ 
555+ libdtoe_thread_pool_s* thread_pool = get_thread_pool(thread_idx);
556+ if (thread_pool == NULL) {
557+ KBDTOE_ERR("get thread pool failed for thread_idx %d", thread_idx);
558+ return false;
559+ }
560+ 
561+ for (int i = 0; i < thread_pool->channel_num; i++) {
562+ if (thread_pool->send_channel_fd[i] == fd || thread_pool->recv_channel_fd[i] == fd) {
563+ return true;
564+ }
565+ }
566+ 
567+ return false;
568+}
569+ 
570+int kbdtoe_register_channel_fd_to_epoll(int thread_idx, int epoll_fd)
571+{
572+ if (thread_idx < 0 || thread_idx >= g_thread_num) {
573+ KBDTOE_ERR("Invalid thread_idx %d", thread_idx);
574+ return DTOE_FAIL;
575+ }
576+ 
577+ libdtoe_thread_pool_s* thread_pool = get_thread_pool(thread_idx);
578+ if (thread_pool == NULL) {
579+ KBDTOE_ERR("get thread pool failed for thread_idx %d", thread_idx);
580+ return DTOE_FAIL;
581+ }
582+ 
583+ for (int i = 0; i < thread_pool->channel_num; i++) {
584+ struct epoll_event ee = {0};
585+ ee.events = EPOLLIN;
586+ ee.data.fd = thread_pool->send_channel_fd[i];
587+ if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, thread_pool->send_channel_fd[i], &ee) != 0) {
588+ KBDTOE_ERR("Failed to add send channel fd %d to epoll, error: %s", thread_pool->send_channel_fd[i], strerror(errno));
589+ return DTOE_FAIL;
590+ }
591+ ee.data.fd = thread_pool->recv_channel_fd[i];
592+ if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, thread_pool->recv_channel_fd[i], &ee) != 0) {
593+ KBDTOE_ERR("Failed to add recv channel fd %d to epoll, error: %s", thread_pool->recv_channel_fd[i], strerror(errno));
594+ return DTOE_FAIL;
595+ }
596+ }
597+ 
598+ return DTOE_SUCCESS;
599+}
Asrc/knet/api/kbdtoe_api/src/kbdtoe_base.h+156-0
@@ -0,0 +1,156 @@
1+/*
2+* Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
3+* redis dtoe is licensed under the Mulan PSL v2.
4+* You can use this software according to the terms and conditions of the Mulan PSL v2.
5+* You may obtain a copy of Mulan PSL v2 at:
6+* http://license.coscl.org.cn/MulanPSL2
7+* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
8+* IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
9+* PURPOSE.
10+* See the Mulan PSL v2 for more details.
11+*
12+* Encapsulate dtoe interface
13+*/
14+#ifndef KB_DTOE_BASE_H
15+#define KB_DTOE_BASE_H
16+#include <unistd.h>
17+#include <inttypes.h>
18+#include <stddef.h>
19+#include <sys/queue.h>
20+#include <pthread.h>
21+#include "flexda_dtoe_interface.h"
22+#include "kbdtoe.h"
23+#include "kbdtoe_log.h"
24+ 
25+#ifndef likely
26+#define likely(x) __builtin_expect(!!(x), 1)
27+#endif
28+#ifndef unlikely
29+#define unlikely(x) __builtin_expect(!!(x), 0)
30+#endif
31+ 
32+#define DTOE_FAIL (-1)
33+#define DTOE_SUCCESS (0)
34+#define DTOE_THREAD_MAX (16)
35+#define DTOE_SOCKET_IP_LEN (16)
36+#define DTOE_CHANNEL_NUM_MAX (16)
37+#define DTOE_SEND_BUF_LEN (4096)
38+#define DTOE_PAGE_SIZE (4096)
39+#define DTOE_SEND_POLL_CNT (32)
40+#define DTOE_RECV_POLL_CNT (32)
41+#define DTOE_ASYNC_OFFLOAD_MAX_NUM (1024)
42+#define DTOE_MAX_CONN_PER_THREAD 10000 // redis单个实例
43+#define DTOE_CONN_PER_CHNL (1024)
44+#define DTOE_RECV_MAX_DESC_NUM (DTOE_RECV_POLL_CNT)
45+#define DTOE_UNUSED(a) ((a)=(a))
46+ 
47+TAILQ_HEAD(libdtoe_conn_head, libdtoe_conn);
48+TAILQ_HEAD(libdtoe_conn_readable_event_head, libdtoe_conn);
49+typedef enum {
50+ DTOE_OFFLOAD_WAIT = 0,
51+ DTOE_OFFLOAD_START,
52+ DTOE_OFFLOAD_SUCCESS,
53+ DTOE_OFFLOAD_FAIL,
54+ DTOE_OFFLOAD_PRECLOSE,
55+} libdtoe_thread_offload_status_e;
56+ 
57+typedef enum {
58+ DTOE_CONN_CREATING = 0,
59+ DTOE_CONN_WORKING,
60+ DTOE_CONN_PRE_CLOSING,
61+ DTOE_CONN_CLOSING,
62+ DTOE_CONN_CLOSED,
63+} libdtoe_conn_status_e;
64+ 
65+typedef struct libdtoe_tx_desc_node {
66+ TAILQ_ENTRY(libdtoe_tx_desc_node) tx_desc_node;
67+ uint16_t send_sn;
68+} libdtoe_tx_desc_node_s;
69+ 
70+typedef struct libdtoe_req_node {
71+ uint64_t request_id;
72+ uint8_t try_id;
73+ uint64_t wr_id;
74+ uint16_t send_sn;
75+ uint32_t dlen;
76+ libdtoe_tx_desc_node_s *send_desc;
77+} libdtoe_req_node_s;
78+ 
79+TAILQ_HEAD(libdtoe_tx_desc_head, libdtoe_tx_desc_node);
80+ 
81+typedef struct libdtoe_recv_desc {
82+ flexda_dtoe_iovec_s iov;
83+ int data_remain;
84+ flexda_dtoe_iovec_s iov_origin;
85+}libdtoe_recv_desc_s;
86+ 
87+typedef void (*kbdtoe_tx_req_free_cb_t)(int sockfd, uint64_t wr_id);
88+typedef struct libdtoe_pending_send {
89+ TAILQ_ENTRY(libdtoe_pending_send) node;
90+ kbdtoe_tx_req_free_cb_t free_cb;
91+ uint64_t wr_id;
92+ int sockfd;
93+ uint16_t send_sn;
94+} __attribute__((packed)) libdtoe_pending_send_s;
95+TAILQ_HEAD(libdtoe_pending_send_head, libdtoe_pending_send);
96+ 
97+typedef struct libdtoe_conn {
98+ libdtoe_conn_status_e conn_status;
99+ uint16_t recv_desc_num;
100+ flexda_send_channel_s *send_channel;
101+ flexda_recv_channel_s *recv_channel;
102+ libdtoe_recv_desc_s recv_desc;
103+ void (*process_cb) (void *, int);
104+ void* thread_pool_ptr;
105+ uint32_t offload_status;
106+ TAILQ_ENTRY(libdtoe_conn) free_node;
107+ int fd;
108+ void *dtoe_conn;
109+ struct {
110+ uint32_t last_send_sn;
111+ uint32_t comp_sn;
112+ uint32_t last_ack_sn;
113+ struct libdtoe_pending_send_head unack_req;
114+ struct libdtoe_pending_send_head free_req;
115+ pthread_spinlock_t pending_send_lock;
116+ } send;
117+ int recv_event_index;
118+ ssize_t leaked_size;
119+ ssize_t read_leaked_offset;
120+ void *leaked_buff;
121+ int has_readable_event;
122+ int poll_mask;
123+ TAILQ_ENTRY(libdtoe_conn) readable_event_node;
124+} libdtoe_conn_s;
125+ 
126+typedef struct libdtoe_conn_pool {
127+ libdtoe_conn_s *conn_pool;
128+ uint32_t pool_idx;
129+ uint32_t offload_num;
130+ pthread_spinlock_t offload_lock;
131+ struct libdtoe_conn_head free_conns;
132+} libdtoe_conn_pool_s;
133+ 
134+typedef struct libdtoe_recv_channel_wrapper {
135+ flexda_recv_channel_s channel;
136+ struct kbdtoe_recv_events *events;
137+ uint32_t next_event_idx;
138+ uint32_t maxevents;
139+} libdtoe_recv_channel_wrapper_s;
140+ 
141+typedef struct libdtoe_thread_pool {
142+ void *node;
143+ libdtoe_conn_pool_s connection;
144+ uint32_t channel_num;
145+ flexda_send_channel_s* send_channel[DTOE_CHANNEL_NUM_MAX];
146+ libdtoe_recv_channel_wrapper_s* recv_channel[DTOE_CHANNEL_NUM_MAX];
147+ int send_channel_fd[DTOE_CHANNEL_NUM_MAX];
148+ int recv_channel_fd[DTOE_CHANNEL_NUM_MAX];
149+ flexda_dtoe_mr_s *send_mr;
150+ uint32_t epoch; /*当前轮询到的channel位置*/
151+} libdtoe_thread_pool_s;
152+ 
153+libdtoe_thread_pool_s* get_thread_pool(int idx);
154+void* get_conn_by_fd(int fd);
155+uint64_t get_dtoe_dev_sn();
156+#endif
Asrc/knet/api/kbdtoe_api/src/kbdtoe_log.c+244-0
@@ -0,0 +1,244 @@
1+/*
2+* Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
3+* redis dtoe is licensed under the Mulan PSL v2.
4+* You can use this software according to the terms and conditions of the Mulan PSL v2.
5+* You may obtain a copy of Mulan PSL v2 at:
6+* http://license.coscl.org.cn/MulanPSL2
7+* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
8+* IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
9+* PURPOSE.
10+* See the Mulan PSL v2 for more details.
11+*
12+*/
13+#include "kbdtoe_log.h"
14+#include <stdio.h>
15+#include <unistd.h>
16+#include <string.h>
17+#include <stdarg.h>
18+#include <errno.h>
19+#include <sys/syscall.h>
20+#include <sys/prctl.h>
21+#include <pthread.h>
22+#include <time.h>
23+#include "securec.h"
24+#include "kbdtoe_base.h"
25+/* Some toolchains require an explicit prototype for getpid() */
26+extern pid_t getpid(void);
27+ 
28+#define PROCESS_PATH_SIZE 1024
29+#define MAX_LOG_LEN 1024
30+#define INVALID_PROC_NAME "invalid proc name"
31+#define NUM_2 2
32+#define MILLISECONDS_PER_SECOND 1000
33+#define NANOS_PER_MILLISECOND 1000000
34+#define PRINT_ERR(fmt, ...) printf("[ERR] %s|%d: " fmt "\n", __func__, __LINE__, ##__VA_ARGS__)
35+ 
36+typedef struct {
37+ KBDTOE_LogLevel level;
38+ const char *level_name;
39+} dtoe_loglevel_name_s;
40+ 
41+static KBDTOE_LogLevel g_loglevel = KBDTOE_LOG_DEFAULT;
42+static pthread_mutex_t g_log_mutex = PTHREAD_MUTEX_INITIALIZER;
43+static char g_proc_name[DTOE_THREAD_NAME_LEN];
44+static const dtoe_loglevel_name_s g_loglevel_name[] = {
45+ {KBDTOE_LOG_ERR, "ERROR" },
46+ {KBDTOE_LOG_WARN, "WARNING"},
47+ {KBDTOE_LOG_INFO, "INFO"},
48+ {KBDTOE_LOG_DEBUG, "DEBUG"},
49+};
50+ 
51+static const char *get_self_thread_name(char *name, size_t len)
52+{
53+ if (name == NULL || len < DTOE_THREAD_NAME_LEN) {
54+ return "invalid parameter";
55+ }
56+ if (prctl(PR_GET_NAME, name) < 0) {
57+ return "invalid thread name";
58+ }
59+ return name;
60+}
61+ 
62+uint64_t kbdtoe_get_current_time_millis(void)
63+{
64+ struct timespec ts = {0};
65+ int ret = clock_gettime(CLOCK_MONOTONIC, &ts);
66+ if (ret != 0) {
67+ KBDTOE_ERR("DTOE clock gettime error, ret %d, errno %d, %s", ret, errno, strerror(errno));
68+ return DTOE_FAIL;
69+ }
70+ // 将时间转换为毫秒，timespec在系统时间正常的情况下不会为负值。
71+ uint64_t milliseconds = (uint64_t)ts.tv_sec * MILLISECONDS_PER_SECOND
72+ + (uint64_t)ts.tv_nsec / NANOS_PER_MILLISECOND;
73+ return milliseconds;
74+}
75+ 
76+void kbdtoe_log_mutex_lock(void)
77+{
78+ pthread_mutex_lock(&g_log_mutex);
79+}
80+ 
81+void kbdtoe_log_mutex_unlock(void)
82+{
83+ pthread_mutex_unlock(&g_log_mutex);
84+}
85+ 
86+void kbdtoe_loglevel_set(KBDTOE_LogLevel log_level)
87+{
88+ if (log_level >= KBDTOE_LOG_MAX) {
89+ KBDTOE_ERR("logLevel %u not support", log_level);
90+ return;
91+ }
92+ g_loglevel = log_level;
93+}
94+ 
95+KBDTOE_LogLevel kbdtoe_loglevel_get(void)
96+{
97+ return g_loglevel;
98+}
99+ 
100+static const char *get_self_process_name(char *name, size_t len)
101+{
102+ char path[PROCESS_PATH_SIZE] = {0};
103+ if (readlink("/proc/self/exe", path, sizeof(path) - 1) <= 0) {
104+ return INVALID_PROC_NAME;
105+ }
106+ 
107+ char *proc_name = strrchr(path, '/');
108+ if (proc_name == NULL || path + strlen(path) <= proc_name + NUM_2) {
109+ return INVALID_PROC_NAME;
110+ }
111+ if (strlen(proc_name + 1) > len) {
112+ return INVALID_PROC_NAME;
113+ }
114+ int ret = strcpy_s(name, len, proc_name + 1);
115+ if (ret != 0) {
116+ return INVALID_PROC_NAME;
117+ }
118+ 
119+ return name;
120+}
121+ 
122+static void kbdtoe_log_lock_unlock(void *arg)
123+{
124+ (void)arg;
125+ pthread_mutex_unlock(&g_log_mutex);
126+}
127+ 
128+static int thread_safe_syslog(int level, char log_msg[MAX_LOG_LEN])
129+{
130+ int orig_cancel_type;
131+ int ret = pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, &orig_cancel_type);
132+ if (ret != 0) {
133+ PRINT_ERR("Pthread set canceltype failed, ret %d", ret);
134+ return ret;
135+ }
136+ 
137+ int orig_cancel_state;
138+ pthread_cleanup_push(kbdtoe_log_lock_unlock, NULL);
139+ ret = pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &orig_cancel_state);
140+ if (ret != 0) {
141+ PRINT_ERR("Pthread set cancelstate failed, ret %d", ret);
142+ return ret;
143+ }
144+ 
145+ pthread_mutex_lock(&g_log_mutex);
146+ syslog(level, "%s", log_msg);
147+ pthread_mutex_unlock(&g_log_mutex);
148+ 
149+ ret = pthread_setcancelstate(orig_cancel_state, NULL);
150+ if (ret != 0) {
151+ PRINT_ERR("Pthread restore canceltype failed, ret %d", ret);
152+ }
153+ pthread_cleanup_pop(1); // Ensure push/pop are paired in the same lexical scope.
154+ 
155+ ret = pthread_setcanceltype(orig_cancel_type, NULL);
156+ if (ret != 0) {
157+ PRINT_ERR("Pthread restore cancelstate failed, ret %d", ret);
atomgit-bot
atomgit-botatomgit-bot6月12日

🔴 Critical

第 147 行已调用 pthread_mutex_unlock 释放 g_log_mutex，第 153 行 pthread_cleanup_pop(1) 会执行清理函数 kbdtoe_log_lock_unlock，该函数再次调用 pthread_mutex_unlock。对于 PTHREAD_MUTEX_INITIALIZER 初始化的默认互斥锁，重复解锁属于未定义行为，可能导致死锁、崩溃或数据损坏。正常路径应使用 pthread_cleanup_pop(0) 不执行清理函数，清理函数仅为线程取消场景保留。

同一段代码其他问题

likedislike
不准确？
158+ }
159+ 
160+ return 0;
161+}
162+ 
163+static int log_message(const char *function, int line, int level, const char *format, va_list va)
164+{
165+ char new_format[MAX_LOG_LEN] = {0};
166+ char thread_name[DTOE_THREAD_NAME_LEN] = {0};
167+ if (g_proc_name[0] == '\0') {
168+ (void)get_self_process_name(g_proc_name, DTOE_THREAD_NAME_LEN);
169+ }
170+ int ret = sprintf_s(new_format, MAX_LOG_LEN, "[%s:%d | %s:%ld] "
171+#ifdef DTOE_DEBUG_BUILD
172+ "%s[%d]|"
173+#endif
174+ "%s\n",
175+ g_proc_name, getpid(),
176+ get_self_thread_name(thread_name, DTOE_THREAD_NAME_LEN), syscall(__NR_gettid),
177+#ifdef DTOE_DEBUG_BUILD
178+ function, line,
179+#endif
180+ format);
181+ if (ret <= 0 || ret >= sizeof(new_format)) {
182+ PRINT_ERR("Sprintf failed ret %d, errno %d", ret, errno);
183+ return ret;
184+ }
185+ 
186+ char log_msg[MAX_LOG_LEN] = {0};
187+ ret = vsprintf_s(log_msg, MAX_LOG_LEN, new_format, va);
188+ if (ret == -1) {
189+ PRINT_ERR("LogMsg size exceeds MAX_LOG_LEN size %d, errno %d", MAX_LOG_LEN, errno);
190+ return ret;
191+ }
192+ 
193+ ret = thread_safe_syslog(level, log_msg);
194+ if (ret != 0) {
195+ PRINT_ERR("Failed to write log in a thread-safe manner, ret %d errno %d", ret, errno);
196+ return ret;
197+ }
198+ 
199+#ifdef DTOE_DEBUG_BUILD
200+ (void)printf("%s", log_msg);
201+#endif
202+ 
203+ return ret;
204+}
205+ 
206+void kbdtoe_log(const char *function, int line, int level, const char *format, ...)
207+{
208+ if (likely(g_loglevel < level)) {
209+ return;
210+ }
211+ 
212+ va_list va;
213+ int ori_errno = errno;
214+ va_start(va, format);
215+ (void)log_message(function, line, level, format, va);
216+ va_end(va);
217+ errno = ori_errno;
218+}
219+ 
220+void kbdtoe_loglevel_set_by_str(const char *level_str)
221+{
222+ kbdtoe_loglevel_set(KBDTOE_LOG_DEFAULT);
223+ 
224+ size_t len = sizeof(g_loglevel_name) / sizeof(g_loglevel_name[0]);
225+ for (size_t i = 0; i < len; ++i) {
226+ if (strcmp(level_str, g_loglevel_name[i].level_name) == 0) {
227+ kbdtoe_loglevel_set(g_loglevel_name[i].level);
228+ return;
229+ }
230+ }
231+ KBDTOE_ERR("Log level not support, use default log level %u", KBDTOE_LOG_DEFAULT);
232+}
233+ 
234+void kbdtoe_log_init()
235+{
236+ openlog(KBDTOE_LOG_MODULE_NAME, LOG_PID | LOG_CONS | LOG_NDELAY, LOG_USER);
237+ (void)setlogmask(LOG_MASK(LOG_ERR) | LOG_MASK(LOG_WARNING) | LOG_MASK(LOG_INFO) | LOG_MASK(LOG_DEBUG));
238+ get_self_process_name(g_proc_name, DTOE_THREAD_NAME_LEN);
239+}
240+ 
241+void kbdtoe_log_uninit(void)
242+{
243+ closelog();
244+}
Asrc/knet/api/kbdtoe_api/src/kbdtoe_log.h+82-0
@@ -0,0 +1,82 @@
1+/*
2+* Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
3+* redis dtoe is licensed under the Mulan PSL v2.
4+* You can use this software according to the terms and conditions of the Mulan PSL v2.
5+* You may obtain a copy of Mulan PSL v2 at:
6+* http://license.coscl.org.cn/MulanPSL2
7+* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
8+* IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
9+* PURPOSE.
10+* See the Mulan PSL v2 for more details.
11+*
12+*/
13+#ifndef KBDTOE_LOG_H
14+#define KBDTOE_LOG_H
15+#include <stdbool.h>
16+#include <stddef.h>
17+#include <stdint.h>
18+#include <syslog.h>
19+ 
20+#define DTOE_THREAD_NAME_LEN 128
21+#define KBDTOE_LOG_MODULE_NAME "libkbdtoe" // 定义输出日志的模块名字，使得在输出日志中统一
22+#define KBDTOE_LIMIT_LOG_DEFAULT_INTERNEL_IN_MS (100) /* 0.1s */
23+ 
24+typedef enum {
25+ KBDTOE_LOG_EMERG = LOG_EMERG,
26+ KBDTOE_LOG_ERR = LOG_ERR,
27+ KBDTOE_LOG_WARN = LOG_WARNING,
28+ KBDTOE_LOG_DEFAULT = KBDTOE_LOG_WARN,
29+ KBDTOE_LOG_INFO = LOG_INFO,
30+ KBDTOE_LOG_DEBUG = LOG_DEBUG,
31+ KBDTOE_LOG_MAX,
32+} KBDTOE_LogLevel;
33+ 
34+void kbdtoe_log(const char *function, int line, int level, const char *format, ...);
35+void kbdtoe_log_init(void);
36+void kbdtoe_log_uninit(void);
37+uint64_t kbdtoe_get_current_time_millis(void);
38+void kbdtoe_loglevel_set(KBDTOE_LogLevel log_level);
39+KBDTOE_LogLevel kbdtoe_loglevel_get(void);
40+void kbdtoe_loglevel_set_by_str(const char *level_str);
41+ 
42+#define KBDTOE_LOG(level, fmt, args...) kbdtoe_log(__func__, __LINE__, level, fmt, ##args)
43+#define KBDTOE_ERR(fmt, args...) kbdtoe_log(__func__, __LINE__, KBDTOE_LOG_ERR, "[ERR] " fmt, ##args)
44+#define KBDTOE_WARN(fmt, args...) kbdtoe_log(__func__, __LINE__, KBDTOE_LOG_WARN, "[WARN] " fmt, ##args)
45+#define KBDTOE_INFO(fmt, args...) kbdtoe_log(__func__, __LINE__, KBDTOE_LOG_INFO, "[INFO] " fmt, ##args)
46+#define KBDTOE_DEBUG(fmt, args...) kbdtoe_log(__func__, __LINE__, KBDTOE_LOG_DEBUG, "[DEBUG] " fmt, ##args)
47+ 
48+#define KBDTOE_LOG_LINE_LIMIT(log_level, fmt, args...) \
49+ do { \
50+ if (kbdtoe_loglevel_get() < (log_level)) { \
51+ break; \
52+ } \
53+ static __thread uint32_t limit_times##__func__##__LINE__ = 0; \
54+ static __thread uint64_t last_ms##__func__##__LINE__ = 0; \
55+ uint64_t now_ms = kbdtoe_get_current_time_millis(); \
56+ if ((now_ms - last_ms##__func__##__LINE__) < KBDTOE_LIMIT_LOG_DEFAULT_INTERNEL_IN_MS) { \
57+ ++limit_times##__func__##__LINE__; \
58+ break; \
59+ } \
60+ last_ms##__func__##__LINE__ = now_ms; \
61+ switch (log_level) { \
62+ case KBDTOE_LOG_ERR: \
63+ kbdtoe_log(__func__, __LINE__, (log_level), \
64+ "[ERR | limitTimes %u] " fmt, limit_times##__func__##__LINE__, ##args); \
65+ break; \
66+ case KBDTOE_LOG_WARN: \
67+ kbdtoe_log(__func__, __LINE__, (log_level), \
68+ "[WARN | limitTimes %u] " fmt, limit_times##__func__##__LINE__, ##args); \
69+ break; \
70+ case KBDTOE_LOG_INFO: \
71+ kbdtoe_log(__func__, __LINE__, (log_level), \
72+ "[INFO | limitTimes %u] " fmt, limit_times##__func__##__LINE__, ##args); \
73+ break; \
74+ default: \
75+ kbdtoe_log(__func__, __LINE__, (log_level), \
76+ "[DEBUG | limitTimes %u] " fmt, limit_times##__func__##__LINE__, ##args); \
77+ break; \
78+ } \
79+ limit_times##__func__##__LINE__ = 0; \
80+ } while (0)
81+ 
82+#endif // __KBDTOE_LOG_H__
Asrc/knet/api/kbdtoe_api/src/kbdtoe_mempool_mr.c+457-0
@@ -0,0 +1,457 @@
1+/*
2+* Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
3+* redis dtoe is licensed under the Mulan PSL v2.
4+* You can use this software according to the terms and conditions of the Mulan PSL v2.
5+* You may obtain a copy of Mulan PSL v2 at:
6+* http://license.coscl.org.cn/MulanPSL2
7+* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
8+* IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
9+* PURPOSE.
10+* See the Mulan PSL v2 for more details.
11+*
12+* Encapsulate dtoe interface
13+*/
14+#include "kbdtoe_mempool_mr.h"
15+#include <stdlib.h>
16+#include <pthread.h>
17+#include <sys/mman.h>
18+#include "kbdtoe_base.h"
19+#include "securec.h"
20+ 
21+static const size_t SLAB_OBJ_SIZES[] = {8, 16, 128, 512, 1024};
22+#define POOL_SIZE (1ULL * 1024 * 1024 * 1024)
23+#define MIN_BLOCK_SIZE 32
24+#define MAX_LEVEL 32
25+#define SLAB_SIZE 16384
26+#define NUM_SLAB_CACHES (sizeof(SLAB_OBJ_SIZES) / sizeof(SLAB_OBJ_SIZES[0]))
27+#define MAGIC_SLAB 0x534C4142u // "SLAB"
28+#define MAGIC_BUDDY 0x42444459u // "BDDY"
29+#define SUCCESS 0
30+#define FAIL -1
31+#define DTOE_PAGE_SIZE (4096)
32+ 
33+typedef struct BuddyBlock {
34+ struct BuddyBlock* next;
35+} BuddyBlock;
36+ 
37+typedef struct Slab {
38+ struct Slab* next;
39+ unsigned int free_count;
40+ unsigned int total_count;
41+ unsigned char* bitmap;
42+ void* mem;
43+ size_t obj_size;
44+ size_t slot_size;
45+} Slab;
46+ 
47+typedef struct SlabCache {
48+ size_t obj_size;
49+ Slab* slabs;
50+ pthread_mutex_t lock;
51+} SlabCache;
52+ 
53+typedef struct MpHeader {
54+ uint32_t magic;
55+ void* owner;
56+} MpHeader;
57+ 
58+static unsigned char* g_memory_pool = NULL;
59+static BuddyBlock* g_free_lists[MAX_LEVEL];
60+static SlabCache g_slab_caches[NUM_SLAB_CACHES];
61+static pthread_mutex_t g_buddy_lock;
62+static size_t g_buddy_total_alloc = 0;
63+static size_t g_buddy_peak_alloc = 0;
64+static flexda_dtoe_mr_s *g_dmr;
65+ 
66+static int size_to_level(size_t size)
67+{
68+ int level = 0;
69+ while ((size > (MIN_BLOCK_SIZE << level)) && (level < (MAX_LEVEL - 1))) {
70+ level++;
71+ }
72+ return level;
73+}
74+ 
75+static void* get_buddy(void* addr, size_t size)
76+{
77+ uintptr_t base = (uintptr_t)g_memory_pool;
78+ uintptr_t offset = (uintptr_t)addr - base;
79+ uintptr_t buddy_offset = offset ^ size;
80+ return (void*)(base + buddy_offset);
81+}
82+ 
83+static int buddy_init()
84+{
85+ int ret;
86+ g_memory_pool = mmap(NULL, POOL_SIZE, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
87+ if (g_memory_pool == MAP_FAILED) {
88+ KBDTOE_ERR("mmap failed!\n");
89+ g_memory_pool = NULL;
90+ return FAIL;
91+ }
92+ if (madvise(g_memory_pool, POOL_SIZE, MADV_DONTFORK) !=0) {
93+ KBDTOE_ERR("madvise failed!\n");
94+ munmap(g_memory_pool, POOL_SIZE);
95+ g_memory_pool = NULL;
96+ return FAIL;
97+ }
98+ 
99+ (void)memset_s(g_memory_pool, POOL_SIZE, 0, POOL_SIZE);
100+ g_dmr = (flexda_dtoe_mr_s*)malloc(sizeof(flexda_dtoe_mr_s));
101+ if (g_dmr == NULL) {
102+ munmap(g_memory_pool, POOL_SIZE);
103+ g_memory_pool = NULL;
104+ return FAIL;
105+ }
106+ int reg_ret = flexda_dtoe_reg_mr(get_dtoe_dev_sn(), g_memory_pool, POOL_SIZE, g_dmr);
107+ if (reg_ret != 0) {
108+ munmap(g_memory_pool, POOL_SIZE);
109+ g_memory_pool = NULL;
110+ free(g_dmr);
111+ g_dmr = NULL;
112+ return FAIL;
113+ }
114+ ret = pthread_mutex_init(&g_buddy_lock, NULL);
115+ if (ret != 0) {
116+ flexda_dtoe_unreg_mr(get_dtoe_dev_sn(), g_dmr);
117+ munmap(g_memory_pool, POOL_SIZE);
118+ g_memory_pool = NULL;
119+ free(g_dmr);
120+ g_dmr = NULL;
121+ return FAIL;
122+ }
123+ for (int i = 0; i < MAX_LEVEL; i++) {
124+ g_free_lists[i] = NULL;
125+ }
126+ int max_level = size_to_level(POOL_SIZE);
127+ g_free_lists[max_level] = (BuddyBlock*)g_memory_pool;
128+ g_free_lists[max_level]->next = NULL;
129+ g_buddy_total_alloc = 0;
130+ g_buddy_peak_alloc = 0;
131+ return SUCCESS;
132+}
133+ 
134+static void* buddy_alloc(size_t size)
135+{
136+ pthread_mutex_lock(&g_buddy_lock);
137+ size_t need = size + sizeof(size_t);
138+ int level = size_to_level(need);
139+ int curr = level;
140+ while ((curr < MAX_LEVEL) && (g_free_lists[curr] == NULL)) {
141+ curr++;
142+ }
143+ if (curr >= MAX_LEVEL) {
144+ pthread_mutex_unlock(&g_buddy_lock);
145+ KBDTOE_ERR("buddy alloc failed\n");
146+ return NULL;
147+ }
148+ 
149+ while (curr > level) {
150+ BuddyBlock* block = g_free_lists[curr];
151+ g_free_lists[curr] = block->next;
152+ size_t block_size = (size_t)MIN_BLOCK_SIZE << curr;
153+ size_t half = block_size >> 1;
154+ BuddyBlock* b1 = block;
155+ BuddyBlock* b2 = (BuddyBlock*) ((char*)block + half);
156+ int next_level = curr - 1;
157+ b1->next = b2;
158+ b2->next = g_free_lists[next_level];
159+ g_free_lists[next_level] = b1;
160+ curr--;
161+ }
162+ BuddyBlock* block = g_free_lists[level];
163+ g_free_lists[level] = block->next;
164+ size_t block_size = (size_t)MIN_BLOCK_SIZE << level;
165+ *(size_t*)block = block_size;
166+ g_buddy_total_alloc += block_size;
167+ if (g_buddy_total_alloc > g_buddy_peak_alloc) {
168+ g_buddy_peak_alloc = g_buddy_total_alloc;
169+ }
170+ pthread_mutex_unlock(&g_buddy_lock);
171+ return (char*)block + sizeof(size_t);
172+}
173+ 
174+static void buddy_free(void* ptr)
175+{
176+ if (!ptr) {
177+ return;
178+ }
179+ pthread_mutex_lock(&g_buddy_lock);
180+ BuddyBlock* block = (BuddyBlock*) ((char*)ptr - sizeof(size_t));
181+ size_t size = *(size_t*)block;
182+ int level = size_to_level(size);
183+ void* addr = block;
184+ while (level < MAX_LEVEL) {
185+ void* buddy = get_buddy(addr, size);
186+ BuddyBlock** list = &g_free_lists[level];
187+ BuddyBlock* prev = NULL;
188+ BuddyBlock* curr = *list;
189+ int merged = 0;
190+ while (curr) {
191+ if ((void*)curr == buddy) {
192+ if (prev) {
193+ prev->next = curr->next;
194+ } else {
195+ *list = curr->next;
196+ }
197+ if (buddy < addr) {
198+ addr = buddy;
199+ }
200+ size <<= 1;
201+ level++;
202+ merged = 1;
203+ break;
204+ }
205+ prev = curr;
206+ curr = curr->next;
207+ }
208+ if (!merged) {
209+ break;
210+ }
211+ }
212+ BuddyBlock* bb = (BuddyBlock*)addr;
213+ bb->next = g_free_lists[level];
214+ g_free_lists[level] = bb;
215+ g_buddy_total_alloc -= size ;
atomgit-bot
atomgit-botatomgit-bot6月12日

🟡 Medium Priority

buddy_free 第 184-211 行合并 buddy 块后，size 已倍增为合并后的大小（如 2S、4S），但第 215 行 g_buddy_total_alloc -= size 减去的是合并后的大小。被合并的 buddy 块此前释放时已将其 size 从 g_buddy_total_alloc 中减去，导致重复扣减，g_buddy_total_alloc（size_t 无符号）下溢为一个极大值，kbdtoe_mempool_stats 输出的统计数据将完全错误。

likedislike
不准确？
216+ pthread_mutex_unlock(&g_buddy_lock);
217+}
218+ 
219+static Slab* slab_create(SlabCache* cache)
220+{
221+ void* mem = buddy_alloc(SLAB_SIZE);
222+ if (!mem) {
223+ return NULL;
224+ }
225+ Slab* slab = (Slab*)malloc(sizeof(Slab));
226+ if (!slab) {
227+ buddy_free(mem);
228+ return NULL;
229+ }
230+ 
231+ slab->obj_size = cache->obj_size;
232+ slab->slot_size = cache->obj_size + sizeof(MpHeader);
233+ slab->total_count = SLAB_SIZE / slab->slot_size;
234+ if (slab->total_count == 0) {
235+ buddy_free(mem);
236+ free(slab);
237+ return NULL;
238+ }
239+ slab->free_count = slab->total_count;
240+ slab->mem = mem;
241+ size_t bitmap_bytes = (slab->total_count + 7) / 8; // 加7除8是向上取整的经典写法
242+ slab->bitmap = (unsigned char*) calloc(bitmap_bytes, 1);
243+ if (!slab->bitmap) {
244+ buddy_free(mem);
245+ free(slab);
246+ return NULL;
247+ }
248+ slab->next = cache->slabs;
249+ cache->slabs = slab;
250+ return slab;
251+}
252+ 
253+static void* slab_alloc_from_cache(SlabCache* cache)
254+{
255+ pthread_mutex_lock(&cache->lock);
256+ Slab* slab = cache->slabs;
257+ while(slab && slab->free_count == 0) {
258+ slab = slab->next;
259+ }
260+ if (!slab) {
261+ slab = slab_create(cache);
262+ if (!slab) {
263+ pthread_mutex_unlock(&cache->lock);
264+ return NULL;
265+ }
266+ }
267+ 
268+ for (unsigned int i = 0; i < slab->total_count; i++) {
269+ unsigned int byte_index = i / 8;
270+ unsigned int bit_index = i % 8;
271+ unsigned char mask = (unsigned char)(1u << bit_index);
272+ if ((slab->bitmap[byte_index] & mask) == 0) {
273+ slab->bitmap[byte_index] |= mask;
274+ slab->free_count--;
275+ char* base = (char*) slab->mem + (i * slab->slot_size);
276+ MpHeader* hdr = (MpHeader*)base;
277+ hdr->magic = MAGIC_SLAB;
278+ hdr->owner = slab;
279+ pthread_mutex_unlock(&cache->lock);
280+ return (void*)(base + sizeof(MpHeader));
281+ }
282+ }
283+ 
284+ pthread_mutex_unlock(&cache->lock);
285+ return NULL;
286+}
287+ 
288+static void slab_free_obj(void* ptr)
289+{
290+ if (!ptr) {
291+ return;
292+ }
293+ MpHeader* hdr = (MpHeader*) ((char*)ptr - sizeof(MpHeader));
294+ if (hdr->magic != MAGIC_SLAB) {
295+ buddy_free(hdr);
296+ return;
297+ }
298+ Slab* slab = (Slab*)hdr->owner;
299+ if (!slab) {
300+ return;
301+ }
302+ 
303+ SlabCache* cache = NULL;
304+ for (int i = 0; i < NUM_SLAB_CACHES; ++i) {
305+ if (g_slab_caches[i].obj_size == slab->obj_size) {
306+ cache = &g_slab_caches[i];
307+ break;
308+ }
309+ }
310+ if (!cache) {
311+ return;
312+ }
313+ pthread_mutex_lock(&cache->lock);
314+ uintptr_t base = (uintptr_t)slab->mem;
315+ uintptr_t header = (uintptr_t)hdr;
316+ if (header < base || (header >= (base + SLAB_SIZE))) {
317+ pthread_mutex_unlock(&cache->lock);
318+ return;
319+ }
320+ uintptr_t offset = header - base;
321+ unsigned int index = (unsigned int) (offset / slab->slot_size);
322+ if (index >= slab->total_count) {
323+ pthread_mutex_unlock(&cache->lock);
324+ return;
325+ }
326+ unsigned int byte_index = index / 8;
327+ unsigned int bit_index = index % 8;
328+ unsigned char mask = (unsigned char) (1u << bit_index);
329+ if (slab->bitmap[byte_index] & mask) {
330+ slab->bitmap[byte_index] &= (unsigned char)~mask;
331+ slab->free_count++;
332+ }
333+ if (slab->free_count == slab->total_count) {
334+ Slab** ps = &cache->slabs;
335+ while (*ps && *ps != slab) {
336+ ps = &((*ps)->next);
337+ }
338+ if (*ps == slab) {
339+ *ps = slab->next;
340+ }
341+ buddy_free(slab->mem);
342+ free(slab->bitmap);
343+ free(slab);
344+ }
345+ pthread_mutex_unlock(&cache->lock);
346+}
347+ 
348+int kbdtoe_mempool_init()
349+{
350+ if (buddy_init() != 0) {
351+ return FAIL;
352+ }
353+ for (int i = 0; i < NUM_SLAB_CACHES; ++i) {
354+ g_slab_caches[i].obj_size = SLAB_OBJ_SIZES[i];
355+ g_slab_caches[i].slabs = NULL;
356+ pthread_mutex_init(&g_slab_caches[i].lock, NULL);
atomgit-bot
atomgit-botatomgit-bot6月12日

🟠 High Priority

第 356 行 pthread_mutex_init(&g_slab_caches[i].lock, NULL) 的返回值未被检查。若 mutex 初始化失败（如系统资源不足），函数仍返回 SUCCESS(0)，调用方会认为内存池初始化成功。此后对该 mutex 的 lock/unlock 操作行为未定义，可能导致死锁或崩溃。对比 buddy_init（第 114 行）对 g_buddy_lock 的 pthread_mutex_init 做了完整的错误检查和回滚处理。

likedislike
不准确？
357+ }
358+ return SUCCESS;
359+}
360+ 
361+flexda_dtoe_mr_s *get_dtoe_mr_s()
362+{
363+ return g_dmr;
364+}
365+ 
366+void* kbdtoe_mempool_alloc(size_t size)
367+{
368+ for (int i = 0; i < NUM_SLAB_CACHES; ++i) {
369+ if (size <= g_slab_caches[i].obj_size) {
370+ void *p = slab_alloc_from_cache(&g_slab_caches[i]);
371+ if (p) {
372+ return p;
373+ }
374+ break;
375+ }
376+ }
377+ void* p = buddy_alloc(size + sizeof(MpHeader));
378+ if (!p) {
379+ return NULL;
380+ }
381+ MpHeader* hdr = (MpHeader*)p;
382+ hdr->magic = MAGIC_BUDDY;
383+ hdr->owner = NULL;
384+ return (char*)p + sizeof(MpHeader);
385+}
386+ 
387+void kbdtoe_mempool_free(int sockfd, uint64_t w_id)
388+{
389+ void *ptr = (void*)w_id;
390+ if (!ptr) {
391+ return;
392+ }
393+ MpHeader* hdr = (MpHeader*)((char*)ptr - sizeof(MpHeader));
394+ if (hdr->magic == MAGIC_SLAB) {
395+ slab_free_obj(ptr);
396+ } else if (hdr->magic == MAGIC_BUDDY) {
397+ buddy_free(hdr);
398+ } else {
399+ KBDTOE_ERR("unknown mempool type free \n");
400+ }
401+}
402+ 
403+void kbdtoe_mempool_stats()
404+{
405+ KBDTOE_INFO("=== Buddy stats ===\n");
406+ KBDTOE_INFO("Buddy total allocated:%zu bytes\n", g_buddy_total_alloc);
407+ KBDTOE_INFO("Buddy peak allocated:%zu bytes\n", g_buddy_peak_alloc);
408+ KBDTOE_INFO("=== SLAB stats ===\n");
409+ for (int i = 0; i < NUM_SLAB_CACHES; ++i) {
410+ SlabCache* sc = &g_slab_caches[i];
411+ unsigned int slabs = 0;
412+ unsigned int objs = 0;
413+ unsigned int free_objs = 0;
414+ pthread_mutex_lock(&sc->lock);
415+ for (Slab* slab = sc->slabs; slab; slab = slab->next) {
416+ slabs++;
417+ objs += slab->total_count;
418+ free_objs += slab->free_count;
419+ }
420+ pthread_mutex_unlock(&sc->lock);
421+ if (slabs > 0) {
422+ KBDTOE_INFO("Cache %zuB: slabs=%u,obj=%u, free=%u\n", sc->obj_size,
423+ slabs, objs, free_objs);
424+ }
425+ }
426+}
427+ 
428+void kbdtoe_mempool_destroy()
429+{
430+ if (g_memory_pool == NULL) {
431+ return;
432+ }
433+ for (int i = 0; i < NUM_SLAB_CACHES; ++i) {
434+ SlabCache* sc = &g_slab_caches[i];
435+ pthread_mutex_lock(&sc->lock);
436+ Slab* slab = sc->slabs;
437+ while (slab) {
438+ Slab* next = slab->next;
439+ buddy_free(slab->mem);
440+ free(slab->bitmap);
441+ free(slab);
442+ slab = next;
443+ }
444+ sc->slabs = NULL;
445+ pthread_mutex_unlock (&sc->lock);
446+ pthread_mutex_destroy(&sc->lock);
447+ }
448+ pthread_mutex_destroy(&g_buddy_lock);
449+ munmap(g_memory_pool, POOL_SIZE);
450+ g_memory_pool = NULL;
451+ if (g_dmr != NULL) {
452+ flexda_dtoe_unreg_mr(get_dtoe_dev_sn(), g_dmr);
453+ free(g_dmr);
454+ g_dmr = NULL;
455+ }
456+}
457+ 
Asrc/knet/api/kbdtoe_api/src/kbdtoe_mempool_mr.h+81-0
@@ -0,0 +1,81 @@
1+/*
2+* Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
3+* redis dtoe is licensed under the Mulan PSL v2.
4+* You can use this software according to the terms and conditions of the Mulan PSL v2.
5+* You may obtain a copy of Mulan PSL v2 at:
6+* http://license.coscl.org.cn/MulanPSL2
7+* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
8+* IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
9+* PURPOSE.
10+* See the Mulan PSL v2 for more details.
11+*
12+* Encapsulate dtoe interface
13+*/
14+#ifndef KBDTOE_MEMPOOL_MR_H
15+#define KBDTOE_MEMPOOL_MR_H
16+#ifdef __cplusplus
17+extern "C" {
18+#endif
19+#include <inttypes.h>
20+#include <stdint.h>
21+#include <stddef.h>
22+#include "kbdtoe_base.h"
23+ 
24+/******************************************************************
25+ Prototype : kbdtoe_mempool_init
26+ Description : 分配1G的内存并完成虚拟地址物理地址的映射
27+ Return Value : 0 success others-failed
28+ **************************************************************/
29+int kbdtoe_mempool_init();
30+ 
31+/******************************************************************
32+ Prototype : kbdtoe_mempool_destroy
33+ Description : 释放内存池
34+ Input : None
35+ Output : None
36+ Return Value : None
37+ **************************************************************/
38+void kbdtoe_mempool_destroy();
39+ 
40+/******************************************************************
41+ Prototype : kbdtoe_mempool_alloc
42+ Description : 从内存池获取所需要大小内存的起始地址
43+ Input : size_t size: 需要内存的大小
44+ Output : None
45+ Return Value : NULL failed, others return start address
46+ **************************************************************/
47+void* kbdtoe_mempool_alloc(size_t size);
48+ 
49+/******************************************************************
50+ Prototype : kbdtoe_mempool_free
51+ Description : 将内存归还到内存池
52+ Input : uint64_t wid: 内存池的地址
53+ Input : int sockfd: socket的fd
54+ Output : None
55+ Return Value : None
56+ **************************************************************/
57+void kbdtoe_mempool_free(int sockfd, uint64_t wid);
58+ 
59+ 
60+/******************************************************************
61+ Prototype : get_dtoe_mr_s
62+ Description : 注册成功后放回的MR信息，发生数据时需要用到,可选
63+ Input : None
64+ Output : None
65+ Return Value : struct flexda_dtoe_mr:注册成功后放回的MR信息
66+ **************************************************************/
67+flexda_dtoe_mr_s *get_dtoe_mr_s();
68+ 
69+/******************************************************************
70+ Prototype : kbdtoe_mempool_stats
71+ Description : 打印内存使用情况
72+ Input : None
73+ Output : None
74+ Return Value : None
75+ **************************************************************/
76+void kbdtoe_mempool_stats();
77+#ifdef __cplusplus
78+}
79+#endif
80+#endif
81+ 
Asrc/knet/api/kbdtoe_api/src/libdtoe_poll.c+70-0
@@ -0,0 +1,70 @@
1+/*
2+* Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
3+* redis dtoe is licensed under the Mulan PSL v2.
4+* You can use this software according to the terms and conditions of the Mulan PSL v2.
5+* You may obtain a copy of Mulan PSL v2 at:
6+* http://license.coscl.org.cn/MulanPSL2
7+* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
8+* IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
9+* PURPOSE.
10+* See the Mulan PSL v2 for more details.
11+*
12+* Encapsulate dtoe interface
13+*/
14+#include "kbdtoe_base.h"
15+ 
16+extern struct libdtoe_conn_readable_event_head g_readable_event_head;
17+ 
18+bool kbdtoe_thread_poll(int thread_idx, struct kbdtoe_recv_events recv_events[], int *nr_recv_event)
19+{
20+ libdtoe_thread_pool_s* thread_pool = get_thread_pool(thread_idx);
21+ int nr_send_event = flexda_dtoe_poll_send_channel(thread_pool->send_channel[0], DTOE_CONN_PER_CHNL);
22+ if (nr_send_event < 0) {
23+ KBDTOE_ERR("kbdtoe kbdtoe thread poll send channel failed, ret:%d\n", nr_send_event);
24+ } else if (nr_send_event >= 0) {
25+ KBDTOE_DEBUG("kbdtoe kbdtoe thread poll send channel, nr_send_event:%d\n", nr_send_event);
26+ }
27+ 
28+ libdtoe_recv_channel_wrapper_s *recv_channel = thread_pool->recv_channel[0];
29+ recv_channel->next_event_idx = 0;
30+ recv_channel->maxevents = (*nr_recv_event) != 0 ? (*nr_recv_event) : DTOE_RECV_MAX_DESC_NUM;
31+ *nr_recv_event = 0;
32+ recv_channel->events = recv_events;
33+ int poll_max_cnt = recv_channel->maxevents;
34+ if (poll_max_cnt > DTOE_RECV_MAX_DESC_NUM) {
35+ poll_max_cnt = DTOE_RECV_MAX_DESC_NUM;
36+ }
37+ 
38+ int remain_ceq_events = flexda_dtoe_poll_receive_channel(&recv_channel->channel, poll_max_cnt);
39+ 
40+ for (uint32_t i = 0; i < recv_channel->next_event_idx; ++i) {
41+ libdtoe_conn_s *conn = (libdtoe_conn_s *)get_conn_by_fd(recv_channel->events[i].sockfd);
42+ if (conn != NULL) {
43+ conn->recv_event_index = -1;
44+ __atomic_add_fetch(&conn->recv_desc_num, recv_channel->events[i].iov_cnt, __ATOMIC_RELAXED);
45+ conn->poll_mask = 1;
46+ }
47+ }
48+ 
49+ *nr_recv_event = recv_channel->next_event_idx;
50+ 
51+ libdtoe_conn_s *node = NULL;
52+ TAILQ_FOREACH(node, &g_readable_event_head, readable_event_node) {
53+ if (node->poll_mask) {
54+ node->poll_mask = 0;
55+ continue;
56+ }
57+ if (*nr_recv_event >= recv_channel->maxevents) {
58+ break;
59+ }
60+ recv_events[*nr_recv_event].sockfd = node->fd;
61+ recv_events[*nr_recv_event].iov_cnt = 0;
62+ node->poll_mask = 0;
63+ (*nr_recv_event)++;
64+ }
65+ 
66+ if (nr_send_event > 0 || *nr_recv_event > 0 || remain_ceq_events > 0) {
67+ return true;
68+ }
69+ return false;
70+}
Asrc/knet/api/kbdtoe_api/src/libdtoe_recv.c+156-0
@@ -0,0 +1,156 @@
1+/*
2+* Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
3+* redis dtoe is licensed under the Mulan PSL v2.
4+* You can use this software according to the terms and conditions of the Mulan PSL v2.
5+* You may obtain a copy of Mulan PSL v2 at:
6+* http://license.coscl.org.cn/MulanPSL2
7+* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
8+* IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
9+* PURPOSE.
10+* See the Mulan PSL v2 for more details.
11+*
12+* Encapsulate dtoe interface
13+*/
14+#include "kbdtoe_base.h"
15+#include "securec.h"
16+ 
17+extern struct libdtoe_conn_readable_event_head g_readable_event_head;
18+ 
19+ssize_t kbdtoe_read(int fd, void *buf, size_t nbyte)
20+{
21+ int ret = 0;
22+ int memcpy_ret;
23+ int iov_cnt = 0;
24+ struct iovec iovs[DTOE_RECV_MAX_DESC_NUM];
25+ ssize_t read_length = 0;
26+ struct iovec iov;
27+ libdtoe_conn_s *conn = (libdtoe_conn_s *)get_conn_by_fd(fd);
28+ libdtoe_recv_desc_s *recv_desc = NULL;
29+ recv_desc = &conn->recv_desc;
30+ if (conn->offload_status == DTOE_OFFLOAD_START) {
atomgit-bot
atomgit-botatomgit-bot6月12日

🔴 Critical

第 27 行 get_conn_by_fd(fd) 返回值直接强转为 libdtoe_conn_s *，第 29 行立即解引用 &amp;conn-&gt;recv_desc，第 30 行访问 conn-&gt;offload_status，均未检查 NULL。若 fd 无效或连接不存在，必将触发段错误崩溃。对比同批 libdtoe_send.c 第 40-46 行，kbdtoe_write/kbdtoe_writev 均有正确的 NULL 检查。

改动建议
30
- if (conn->offload_status == DTOE_OFFLOAD_START) {
30
+ libdtoe_conn_s *conn = (libdtoe_conn_s *)get_conn_by_fd(fd);
31
+ if (conn == NULL) {
32
+ KBDTOE_ERR("kbdtoe read, conn is null for fd %d", fd);
33
+ errno = EINVAL;
34
+ return -1;
35
+ }
36
+ libdtoe_recv_desc_s *recv_desc = NULL;
37
+ recv_desc = &conn->recv_desc;
应用建议
likedislike
不准确？
31+ errno = EAGAIN;
32+ return -1;
33+ }
34+ /** 解决卸载过程中数据丢包问题 */
35+ ssize_t leaked_size = flexda_dtoe_get_leaked_packet_size(conn->dtoe_conn);
36+ if (unlikely(leaked_size > 0)) {
37+ conn->leaked_buff = malloc(leaked_size);
38+ if (conn->leaked_buff == NULL) {
39+ KBDTOE_ERR("kbdtoe read malloc leaked buff failed");
40+ errno = ENOMEM;
41+ return -1;
42+ }
43+ ssize_t recved_bytes = flexda_dtoe_recv_leaked_packet(conn->dtoe_conn, conn->leaked_buff, leaked_size);
44+ if (recved_bytes < 0) {
45+ KBDTOE_ERR("kbdtoe read leaked packet failed");
46+ free(conn->leaked_buff);
47+ errno = -recved_bytes;
48+ return -1;
49+ }
50+ conn->leaked_size = leaked_size;
51+ conn->read_leaked_offset = 0;
52+ }
53+ 
54+ if (unlikely(conn->leaked_size > 0)) {
55+ size_t leaked_copy_len = (conn->leaked_size > nbyte) ? nbyte : (size_t)conn->leaked_size;
56+ memcpy_ret = memcpy_s(buf, leaked_copy_len, (char *)conn->leaked_buff + conn->read_leaked_offset, leaked_copy_len);
57+ if (memcpy_ret != EOK) {
58+ KBDTOE_ERR("kbdtoe read leaked memcpy_s failed");
59+ errno = EFAULT;
60+ return -1;
61+ }
62+ conn->read_leaked_offset += leaked_copy_len;
63+ conn->leaked_size -= leaked_copy_len;
64+ read_length += leaked_copy_len;
65+ if (conn->leaked_size == 0) {
66+ free(conn->leaked_buff);
67+ conn->leaked_buff = NULL;
68+ if (conn->has_readable_event) {
69+ TAILQ_REMOVE(&g_readable_event_head, conn, readable_event_node);
70+ conn->has_readable_event = 0;
71+ }
72+ }
73+ if (read_length == nbyte) {
74+ if (conn->leaked_size > 0) {
75+ conn->has_readable_event = 1;
76+ }
77+ if (conn->has_readable_event) {
78+ TAILQ_INSERT_TAIL(&g_readable_event_head, conn, readable_event_node);
79+ }
80+ return read_length;
81+ }
82+ }
83+ if (conn->has_readable_event) {
84+ TAILQ_REMOVE(&g_readable_event_head, conn, readable_event_node);
85+ conn->has_readable_event = 0;
86+ }
87+ 
88+ while (__atomic_load_n(&conn->recv_desc_num, __ATOMIC_RELAXED) && (read_length < nbyte) && (iov_cnt < DTOE_RECV_MAX_DESC_NUM)) {
89+ if (recv_desc->data_remain == 0) {
90+ ret = flexda_dtoe_recv(conn->dtoe_conn, &iov, 1);
91+ if (ret < 0) {
92+ flexda_dtoe_recv_mem_loopback(iovs, iov_cnt);
93+ KBDTOE_ERR("kbdtoe conn:%p, flexda_dtoe_recv failed, error =%d!", conn, ret);
94+ errno = -ret;
95+ return -1;
96+ } else if (iov.iov_base == NULL) {
97+ KBDTOE_ERR("flexda_dtoe_recv debug!!!");
98+ flexda_dtoe_recv_mem_loopback(iovs, iov_cnt);
99+ errno = ENOBUFS; // fix me
100+ return -1;
101+ }
102+ } else {
103+ iov.iov_base = recv_desc->iov.iov_base;
104+ iov.iov_len = recv_desc->iov.iov_len;
105+ }
106+ 
107+ if ((read_length + iov.iov_len) <= nbyte) {
108+ memcpy_ret = memcpy_s(buf + read_length, iov.iov_len, iov.iov_base, iov.iov_len);
109+ if (memcpy_ret != EOK) {
110+ KBDTOE_ERR("kbdtoe read memcpy_s scene3 failed");
111+ flexda_dtoe_recv_mem_loopback(iovs, iov_cnt);
112+ errno = EFAULT;
113+ return -1;
114+ }
115+ read_length += iov.iov_len;
116+ (void)__atomic_fetch_sub(&conn->recv_desc_num, 1, __ATOMIC_RELAXED);
117+ 
118+ if (recv_desc->data_remain != 0) {
119+ iovs[iov_cnt].iov_base = conn->recv_desc.iov_origin.iov_base;
120+ iovs[iov_cnt].iov_len = conn->recv_desc.iov_origin.iov_len;
121+ } else {
122+ iovs[iov_cnt].iov_base = iov.iov_base;
123+ iovs[iov_cnt].iov_len = iov.iov_len;
124+ }
125+ iov_cnt++;
126+ recv_desc->data_remain = 0;
127+ } else {
128+ size_t remain_len = nbyte - read_length;
129+ memcpy_ret = memcpy_s((buf + read_length), remain_len, iov.iov_base, remain_len);
130+ if (memcpy_ret != EOK) {
131+ KBDTOE_ERR("kbdtoe read memcpy_s scene4 failed");
132+ flexda_dtoe_recv_mem_loopback(iovs, iov_cnt);
133+ errno = EFAULT;
134+ return -1;
135+ }
136+ if (recv_desc->data_remain == 0) {
137+ recv_desc->data_remain = 1;
138+ recv_desc->iov_origin.iov_base = iov.iov_base;
139+ recv_desc->iov_origin.iov_len = iov.iov_len;
140+ }
141+ iov.iov_base = (char*)iov.iov_base + remain_len;
142+ iov.iov_len -= remain_len;
143+ recv_desc->iov.iov_base = iov.iov_base;
144+ recv_desc->iov.iov_len = iov.iov_len;
145+ read_length = nbyte;
146+ }
147+ }
148+ if (iov_cnt) {
149+ flexda_dtoe_recv_mem_loopback(iovs, iov_cnt);
150+ }
151+ if (__atomic_load_n(&conn->recv_desc_num, __ATOMIC_RELAXED)) {
152+ TAILQ_INSERT_TAIL(&g_readable_event_head, conn, readable_event_node);
153+ conn->has_readable_event = 1;
154+ }
155+ return read_length;
156+}
Asrc/knet/api/kbdtoe_api/src/libdtoe_send.c+159-0
@@ -0,0 +1,159 @@
1+/*
2+* Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
3+* redis dtoe is licensed under the Mulan PSL v2.
4+* You can use this software according to the terms and conditions of the Mulan PSL v2.
5+* You may obtain a copy of Mulan PSL v2 at:
6+* http://license.coscl.org.cn/MulanPSL2
7+* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
8+* IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
9+* PURPOSE.
10+* See the Mulan PSL v2 for more details.
11+*
12+* Encapsulate dtoe interface
13+*/
14+#include "kbdtoe_base.h"
15+#include "kbdtoe_mempool_mr.h"
16+#include "securec.h"
17+ 
18+ssize_t kbdtoe_write(int fd, const void *buf, size_t nbyte)
19+{
20+ int ret = 0;
21+ int memcpy_ret;
22+ unsigned char *ptr = NULL;
23+ ptr = (unsigned char*)kbdtoe_mempool_alloc(nbyte);
24+ if (ptr == NULL) {
25+ KBDTOE_ERR("kbdtoe write mempool alloc fail");
26+ errno = EAGAIN;
27+ return -1;
28+ }
29+ memcpy_ret = memcpy_s(ptr, nbyte, buf, nbyte);
30+ if (memcpy_ret != EOK) {
31+ KBDTOE_ERR("kbdtoe write memcpy_s failed");
32+ kbdtoe_mempool_free(0, (uint64_t)ptr);
33+ errno = EFAULT;
34+ return -1;
35+ }
36+ struct iovec iov[1];
37+ iov[0].iov_base = ptr;
38+ iov[0].iov_len = nbyte;
39+ 
40+ libdtoe_conn_s *conn = (libdtoe_conn_s *)get_conn_by_fd(fd);
41+ if (conn == NULL) {
42+ KBDTOE_ERR("kbdtoe write, conn is null for fd %d", fd);
43+ kbdtoe_mempool_free(0, (uint64_t)ptr);
44+ errno = EINVAL;
45+ return -1;
46+ }
47+ 
48+ flexda_dtoe_mr_s *mr = get_dtoe_mr_s();
49+ flexda_dtoe_tx_info_s info = {
50+ .tx_in.op_code = 0,
51+ .tx_in.lkey = mr ? mr->lkey : 0,
52+ };
53+ ret = flexda_dtoe_send(conn->dtoe_conn, iov, 1, &info);
54+ if (ret < 0) {
55+ kbdtoe_mempool_free(0, (uint64_t)ptr);
56+ errno = -ret;
57+ return -1;
58+ }
59+ 
60+ pthread_spin_lock(&conn->send.pending_send_lock);
61+ libdtoe_pending_send_s *pending = TAILQ_FIRST(&conn->send.free_req);
62+ if (pending != NULL) {
63+ TAILQ_REMOVE(&conn->send.free_req, pending, node);
64+ }
65+ pthread_spin_unlock(&conn->send.pending_send_lock);
66+ 
67+ if (pending == NULL) {
68+ pending = (libdtoe_pending_send_s*)malloc(sizeof(libdtoe_pending_send_s));
69+ }
70+ 
71+ if (pending != NULL) {
72+ pending->free_cb = kbdtoe_mempool_free;
73+ pending->wr_id = (uint64_t)ptr;
74+ pending->sockfd = fd;
75+ pending->send_sn = info.tx_out.curr_msn;
76+ pthread_spin_lock(&conn->send.pending_send_lock);
77+ TAILQ_INSERT_TAIL(&conn->send.unack_req, pending, node);
78+ pthread_spin_unlock(&conn->send.pending_send_lock);
79+ }
80+ 
81+ return ret;
82+}
83+ 
84+ssize_t kbdtoe_writev(int fd, const struct iovec *iov, int iovcnt)
85+{
86+ int ret = 0;
87+ int memcpy_ret;
88+ size_t total_size = 0;
89+ for (int i = 0; i < iovcnt; i++) {
90+ total_size += iov[i].iov_len;
91+ }
92+ unsigned char *ptr = NULL;
93+ ptr = (unsigned char*)kbdtoe_mempool_alloc(total_size);
94+ if (ptr == NULL) {
95+ KBDTOE_ERR("kbdtoe writev mempool alloc fail");
96+ errno = EAGAIN;
97+ return -1;
98+ }
99+ size_t offset = 0;
100+ for (int i = 0; i < iovcnt; ++i) {
101+ memcpy_ret = memcpy_s(ptr + offset, iov[i].iov_len, iov[i].iov_base, iov[i].iov_len);
102+ if (memcpy_ret != EOK) {
103+ KBDTOE_ERR("kbdtoe write memcpy_s failed");
104+ kbdtoe_mempool_free(0, (uint64_t)ptr);
105+ errno = EFAULT;
106+ return -1;
107+ }
108+ offset += iov[i].iov_len;
109+ }
110+ 
111+ struct iovec dtoe_iov[1];
112+ dtoe_iov[0].iov_base = ptr;
113+ dtoe_iov[0].iov_len = total_size;
114+ 
115+ libdtoe_conn_s *conn = (libdtoe_conn_s *)get_conn_by_fd(fd);
116+ if (conn == NULL) {
117+ KBDTOE_ERR("kbdtoe writev, conn is null for fd %d", fd);
118+ kbdtoe_mempool_free(0, (uint64_t)ptr);
119+ errno = EINVAL;
120+ return -1;
121+ }
122+ 
123+ flexda_dtoe_mr_s *mr = get_dtoe_mr_s();
124+ flexda_dtoe_tx_info_s info = {
125+ .tx_in.op_code = 0,
126+ .tx_in.lkey = mr ? mr->lkey : 0
127+ };
128+ ret = flexda_dtoe_send(conn->dtoe_conn, dtoe_iov, 1, &info);
129+ if (ret < 0) {
130+ kbdtoe_mempool_free(0, (uint64_t)ptr);
131+ errno = -ret;
132+ return -1;
133+ }
134+ 
135+ pthread_spin_lock(&conn->send.pending_send_lock);
136+ libdtoe_pending_send_s *pending = TAILQ_FIRST(&conn->send.free_req);
137+ if (pending != NULL) {
138+ TAILQ_REMOVE(&conn->send.free_req, pending, node);
139+ }
140+ pthread_spin_unlock(&conn->send.pending_send_lock);
141+ 
142+ if (pending == NULL) {
143+ pending = (libdtoe_pending_send_s*)malloc(sizeof(libdtoe_pending_send_s));
144+ }
145+ 
146+ if (pending != NULL) {
147+ pending->free_cb = kbdtoe_mempool_free;
148+ pending->wr_id = (uint64_t)ptr;
149+ pending->sockfd = fd;
150+ pending->send_sn = info.tx_out.curr_msn;
151+ conn->send.last_send_sn = info.tx_out.curr_msn;
152+ pthread_spin_lock(&conn->send.pending_send_lock);
153+ TAILQ_INSERT_TAIL(&conn->send.unack_req, pending, node);
154+ pthread_spin_unlock(&conn->send.pending_send_lock);
155+ }
156+ 
157+ return ret;
158+}
159+