From 2cd06a24d08a7967c4316fa4622924760fb981cd Mon Sep 17 00:00:00 2001 From: =?utf8?q?=E5=AD=99=E6=98=A5=E9=98=B3?= Date: Mon, 11 Aug 2025 15:11:11 +0800 Subject: [PATCH] =?utf8?q?ai=E5=AF=B9=E8=AF=9D=E5=8F=AF=E4=BB=A5=E8=BF=90?= =?utf8?q?=E8=A1=8C=EF=BC=8C=E4=BD=86=E8=BF=98=E4=B8=8D=E8=83=BD=E4=B8=8A?= =?utf8?q?=E4=BC=A0=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit --- .../brandy_evb/linker/standard/acore.prelds | 2 + .../slip/cat1Service/bk_websocket.h | 242 ++--- .../slip/cat1Service/cat1_ai_server.cpp | 54 +- .../peripheral/slip/cat1Service/cat1_ws.cpp | 21 +- .../slip/tcpmanager/tcp_simple_manager.c | 8 + .../slip/tcpmanager/tcp_simple_manager.h | 9 +- .../tcpmanager/tcp_simple_manager_broken.c | 942 ------------------ .../slip/tcpmanager/tcp_simple_manager_full.c | 942 ------------------ 8 files changed, 194 insertions(+), 2026 deletions(-) delete mode 100644 code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager_broken.c delete mode 100644 code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager_full.c diff --git a/code/sdk/drivers/boards/brandy_evb/linker/standard/acore.prelds b/code/sdk/drivers/boards/brandy_evb/linker/standard/acore.prelds index 7ba4a2b5d..f91df4686 100755 --- a/code/sdk/drivers/boards/brandy_evb/linker/standard/acore.prelds +++ b/code/sdk/drivers/boards/brandy_evb/linker/standard/acore.prelds @@ -96,6 +96,7 @@ SECTIONS SORT(*libmsg_center*.a:*.o*) (.data .data* .sdata .sdata*) SORT(*ohos*.a:*.o*) (.data .data* .sdata .sdata*) SORT(*liblwIP*.a:*.o*) (.data .data* .sdata .sdata*) + SORT(*libslip*.a:*.o*) (.data .data* .sdata .sdata*) SORT(*xts_test*.a:*.o*) (.data .data* .sdata .sdata*) *libnativelauncher*.a:*.o* (.data .data* .sdata .sdata*) SORT(*application/wearable*.a:*.o*)(.data .data* .sdata .sdata*) @@ -121,6 +122,7 @@ SECTIONS SORT(*libmsg_center*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON) SORT(*ohos*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON) SORT(*liblwIP*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON) + SORT(*libslip*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON) SORT(*xts_test*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON) *libnativelauncher*.a:*.o* (.bss .bss* .sbss .sbss* COMMON) SORT(*application/wearable*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON) diff --git a/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h b/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h index dbf6e3367..4e6cda40d 100644 --- a/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h +++ b/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h @@ -1,121 +1,121 @@ - -#ifndef __BK_WEBSOCKET_H__ -#define __BK_WEBSOCKET_H__ -// #include "aic_osa.h" -// #include -// WebSocket 消息类型枚举 - -#ifdef __cplusplus -#if __cplusplus -extern "C" { -#endif /* __cplusplus */ -#endif /* __cplusplus */ - -typedef enum { - WS_CONT = 0x00, - WS_TEXT = 0x01, - WS_BINARY = 0x02, - WS_CLOSE = 0x8, - WS_PING = 0x9, - WS_PONG = 0xa -} WebSocketMessageType; - -/* WebSocket数据包回调函数类型 */ -typedef void (*on_ws_packet_callback)(uint8_t* data, size_t datalen, WebSocketMessageType type); - -/* WebSocket组包缓冲区配置 */ -#define WS_FRAME_BUFFER_SIZE 256 -#define WS_PAYLOAD_BUFFER_SIZE 128 - -/* WebSocket组包状态 */ -typedef enum { - WS_PARSE_HEADER, /* 解析帧头 */ - WS_PARSE_EXTENDED, /* 解析扩展长度 */ - WS_PARSE_MASK, /* 解析掩码 */ - WS_PARSE_PAYLOAD /* 解析载荷数据 */ -} ws_parse_state_t; - -/* WebSocket组包缓冲区结构 */ -typedef struct { - uint8_t frame_buffer[WS_FRAME_BUFFER_SIZE]; /* 帧缓冲区 */ - uint32_t frame_buffer_len; /* 帧缓冲区当前长度 */ - - uint8_t payload_buffer[WS_PAYLOAD_BUFFER_SIZE]; /* 载荷缓冲区 */ - uint32_t payload_buffer_len; /* 载荷缓冲区当前长度 */ - - ws_parse_state_t parse_state; /* 解析状态 */ - uint8_t fin; /* FIN标志 */ - WebSocketMessageType opcode; /* 操作码 */ - uint8_t mask; /* 掩码标志 */ - uint64_t payload_len; /* 载荷长度 */ - uint32_t payload_received; /* 已接收载荷长度 */ - uint8_t masking_key[4]; /* 掩码密钥 */ - - on_ws_packet_callback packet_callback; /* 数据包回调函数 */ -} ws_packet_parser_t; - -/* WebSocket消息队列系统 */ -#define WS_QUEUE_SIZE 4 -#define WS_MESSAGE_MAX_SIZE 512 - -typedef struct { - uint8_t data[WS_MESSAGE_MAX_SIZE]; - uint16_t length; - WebSocketMessageType type; - uint32_t timestamp; - bool in_use; -} ws_message_item_t; - -typedef struct { - ws_message_item_t messages[WS_QUEUE_SIZE]; - uint8_t head; - uint8_t tail; - uint8_t count; - osMutexId_t mutex; - osSemaphoreId_t semaphore; - bool initialized; -} ws_message_queue_t; - - -// WebSocket 客户端结构体 -typedef struct { - void* tcp_handle; - ws_message_queue_t *msg_queue; - char host[128]; - char path[128]; - char query[128]; - int port; - - /* 组包解析器 */ - ws_packet_parser_t parser; -} WebSocketClient; - - -// 初始化 WebSocket 客户端 -WebSocketClient *init_websocket_client(const char *url, const char *headers,uint8_t use_dns); - -// 设置WebSocket数据包回调函数 -void ws_set_packet_callback(WebSocketClient *client, on_ws_packet_callback callback); - -// 处理接收到的原始数据 -int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_t len); - -// 关闭 WebSocket 客户端 -void lwip_close_websocket_client(WebSocketClient *client) ; - -// 接收 WebSocket 消息 -int receive_websocket_message(WebSocketClient *client, unsigned char *buf,int len,WebSocketMessageType *code); -int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type) ; - - -#ifdef __cplusplus -#if __cplusplus -} -#endif /* __cplusplus */ -#endif /* __cplusplus */ - -// 发送 WebSocket 消息 -// int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type) ; - - -#endif + +#ifndef __BK_WEBSOCKET_H__ +#define __BK_WEBSOCKET_H__ +// #include "aic_osa.h" +// #include +// WebSocket 消息类型枚举 + +#ifdef __cplusplus +#if __cplusplus +extern "C" { +#endif /* __cplusplus */ +#endif /* __cplusplus */ + +typedef enum { + WS_CONT = 0x00, + WS_TEXT = 0x01, + WS_BINARY = 0x02, + WS_CLOSE = 0x8, + WS_PING = 0x9, + WS_PONG = 0xa +} WebSocketMessageType; + +/* WebSocket数据包回调函数类型 */ +typedef void (*on_ws_packet_callback)(uint8_t* data, size_t datalen, WebSocketMessageType type); + +/* WebSocket组包缓冲区配置 */ +#define WS_FRAME_BUFFER_SIZE 256 +#define WS_PAYLOAD_BUFFER_SIZE 128 + +/* WebSocket组包状态 */ +typedef enum { + WS_PARSE_HEADER, /* 解析帧头 */ + WS_PARSE_EXTENDED, /* 解析扩展长度 */ + WS_PARSE_MASK, /* 解析掩码 */ + WS_PARSE_PAYLOAD /* 解析载荷数据 */ +} ws_parse_state_t; + +/* WebSocket组包缓冲区结构 */ +typedef struct { + uint8_t frame_buffer[WS_FRAME_BUFFER_SIZE]; /* 帧缓冲区 */ + uint32_t frame_buffer_len; /* 帧缓冲区当前长度 */ + + uint8_t payload_buffer[WS_PAYLOAD_BUFFER_SIZE]; /* 载荷缓冲区 */ + uint32_t payload_buffer_len; /* 载荷缓冲区当前长度 */ + + ws_parse_state_t parse_state; /* 解析状态 */ + uint8_t fin; /* FIN标志 */ + WebSocketMessageType opcode; /* 操作码 */ + uint8_t mask; /* 掩码标志 */ + uint64_t payload_len; /* 载荷长度 */ + uint32_t payload_received; /* 已接收载荷长度 */ + uint8_t masking_key[4]; /* 掩码密钥 */ + + on_ws_packet_callback packet_callback; /* 数据包回调函数 */ +} ws_packet_parser_t; + +/* WebSocket消息队列系统 */ +#define WS_QUEUE_SIZE 4 +#define WS_MESSAGE_MAX_SIZE 512 + +typedef struct { + uint8_t data[WS_MESSAGE_MAX_SIZE]; + uint16_t length; + WebSocketMessageType type; + uint32_t timestamp; + bool in_use; +} ws_message_item_t; + +typedef struct { + ws_message_item_t messages[WS_QUEUE_SIZE]; + uint8_t head; + uint8_t tail; + uint8_t count; + osMutexId_t mutex; + osSemaphoreId_t semaphore; + bool initialized; +} ws_message_queue_t; + + +// WebSocket 客户端结构体 +typedef struct { + void* tcp_handle; + ws_message_queue_t *msg_queue; + char host[128]; + char path[128]; + char query[128]; + int port; + + /* 组包解析器 */ + ws_packet_parser_t parser; +} WebSocketClient; + + +// 初始化 WebSocket 客户端 +WebSocketClient *init_websocket_client(const char *url, const char *headers,uint8_t use_dns); + +// 设置WebSocket数据包回调函数 +void ws_set_packet_callback(WebSocketClient *client, on_ws_packet_callback callback); + +// 处理接收到的原始数据 +int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_t len); + +// 关闭 WebSocket 客户端 +void lwip_close_websocket_client(WebSocketClient *client) ; + +// 接收 WebSocket 消息 +int receive_websocket_message(WebSocketClient *client, unsigned char *buf,int len,WebSocketMessageType *code); +int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type) ; + + +#ifdef __cplusplus +#if __cplusplus +} +#endif /* __cplusplus */ +#endif /* __cplusplus */ + +// 发送 WebSocket 消息 +// int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type) ; + + +#endif diff --git a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp index 28827b42e..bb0cb1048 100644 --- a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp +++ b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp @@ -1940,9 +1940,15 @@ void asr_server_deinit(void) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"asr_server_deinit"); - if(send_websocket_message(asr_client,(uint8_t *)"WS_CLOSE",strlen("WS_CLOSE"),WS_CLOSE)) - { - WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"WS_CLOSE"); + // 检查TCP管理器是否仍然可用再发送关闭消息 + extern bool tcp_manager_is_initialized(void); + if (asr_client && tcp_manager_is_initialized()) { + if(send_websocket_message(asr_client,(uint8_t *)"WS_CLOSE",strlen("WS_CLOSE"),WS_CLOSE)) + { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"WS_CLOSE"); + } + } else { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"跳过发送WS_CLOSE: TCP管理器不可用"); } } @@ -1951,9 +1957,15 @@ void chat_server_deinit(void) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"chat_server_deinit"); - if(send_websocket_message(chat_client,(uint8_t *)"WS_CLOSE",strlen("WS_CLOSE"),WS_CLOSE)) - { - WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"WS_CLOSE"); + // 检查TCP管理器是否仍然可用再发送关闭消息 + extern bool tcp_manager_is_initialized(void); + if (chat_client && tcp_manager_is_initialized()) { + if(send_websocket_message(chat_client,(uint8_t *)"WS_CLOSE",strlen("WS_CLOSE"),WS_CLOSE)) + { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"WS_CLOSE"); + } + } else { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"跳过发送WS_CLOSE: TCP管理器不可用"); } } @@ -1962,9 +1974,16 @@ void tts_server_deinit(void) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"tts_server_deinit"); - if(send_websocket_message(tts_client,(uint8_t *)"WS_CLOSE",strlen("WS_CLOSE"),WS_CLOSE)) - { - WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"WS_CLOSE"); + + // 检查TCP管理器是否仍然可用再发送关闭消息 + extern bool tcp_manager_is_initialized(void); + if (tts_client && tcp_manager_is_initialized()) { + if(send_websocket_message(tts_client,(uint8_t *)"WS_CLOSE",strlen("WS_CLOSE"),WS_CLOSE)) + { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"WS_CLOSE"); + } + } else { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"跳过发送WS_CLOSE: TCP管理器不可用"); } } @@ -2611,6 +2630,21 @@ errcode_t voice_chat_server_init(void) void ai_cleanup_resources(void) { AI_LOCK(); + // 首先关闭所有WebSocket连接 + extern void lwip_close_websocket_client(WebSocketClient *client); + if (chat_client) { + lwip_close_websocket_client(chat_client); + chat_client = NULL; + } + if (asr_client) { + lwip_close_websocket_client(asr_client); + asr_client = NULL; + } + if (tts_client) { + lwip_close_websocket_client(tts_client); + tts_client = NULL; + } + // 清理WebSocket消息队列 ws_queues_deinit_all(); @@ -2631,7 +2665,7 @@ void ai_cleanup_resources(void) { AI_UNLOCK(); - // 清理TCP管理器 + // 最后清理TCP管理器 tcp_simple_manager_deinit(); WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "AI resources cleaned up"); diff --git a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp index f1130d953..9a87bd364 100644 --- a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp +++ b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp @@ -478,17 +478,20 @@ int send_handshake(WebSocketClient *client,const char *headers,char *response,in printf("tcp_send_data_directly failed: %d", tcp_result); ret = -1; } - - int retlen = tcp_recv_data_directly(client->tcp_handle, (uint8_t *)response, response_len); - - if(retlen<0) - { - printf("tcp_recv_data_directly failed"); - ret = -1; - } else { - ret = retlen; + // 发送成功，尝试接收响应 + int retlen = tcp_recv_data_directly(client->tcp_handle, (uint8_t *)response, response_len); + + if(retlen<0) + { + printf("tcp_recv_data_directly failed"); + ret = -1; + } + else + { + ret = retlen; + } } free(request); diff --git a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.c b/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.c index 2960a3bc9..ca9e34cec 100644 --- a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.c +++ b/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.c @@ -1615,4 +1615,12 @@ void tcp_set_debug_level(int level) int tcp_get_debug_level(void) { return g_tcp_debug_level; +} + +/** + * @brief 检查TCP管理器是否已初始化 + */ +bool tcp_manager_is_initialized(void) +{ + return g_manager_initialized; } \ No newline at end of file diff --git a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.h b/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.h index b1f35b430..91786e31d 100644 --- a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.h +++ b/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.h @@ -53,8 +53,8 @@ extern int tcp_get_debug_level(void); /* 连接参数定义 */ #define TCP_CONNECT_TIMEOUT_MS 10000 #define TCP_DNS_TIMEOUT_MS 8000 -#define TCP_SEND_QUEUE_SIZE 4 -#define TCP_SEND_BUFFER_SIZE 128 +#define TCP_SEND_QUEUE_SIZE 16 +#define TCP_SEND_BUFFER_SIZE 1024 #define TCP_CONNECT_RETRY_COUNT 3 #define TCP_CONNECT_RETRY_DELAY_MS 1000 @@ -325,6 +325,11 @@ void tcp_set_debug_level(int level); */ int tcp_get_debug_level(void); +/** + * @brief 检查TCP管理器是否已初始化 + */ +bool tcp_manager_is_initialized(void); + #ifdef __cplusplus } #endif diff --git a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager_broken.c b/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager_broken.c deleted file mode 100644 index 2a5e7f3d1..000000000 --- a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager_broken.c +++ /dev/null @@ -1,942 +0,0 @@ -/** - * @file tcp_simple_manager.c - * @brief TCP简单管理器实现 - */ - -#include "tcp_simple_manager.h" -#include "lwip/sockets.h" -#include "lwip/netdb.h" -#include "lwip/sys.h" -#include "lwip/err.h" -#include -#include -#include -#include -#include -#include -#include -#include - -/* lwIP socket函数重定义，确保使用lwIP接口 */ -#define socket lwip_socket -#define connect lwip_connect -#define close lwip_close -#define send lwip_send -#define recv lwip_recv -#define bind lwip_bind -#define listen lwip_listen -#define accept lwip_accept -#define getsockopt lwip_getsockopt -#define setsockopt lwip_setsockopt - -/* 内部状态标志 */ -static bool g_manager_initialized = false; -static bool g_manager_running = false; - -/* 线程相关 */ -static sys_thread_t g_io_thread; -static sys_sem_t g_shutdown_sem; - -/* 队列相关 */ -static sys_mbox_t g_cmd_queue; -static sys_mbox_t g_recv_queue; - -/* 连接池 */ -static tcp_connection_info_t g_connections[TCP_MANAGER_MAX_CONNECTIONS]; -static sys_mutex_t g_connections_mutex; - -/* 统计信息 */ -static tcp_manager_stats_t g_stats; -static sys_mutex_t g_stats_mutex; - -/* 当前线程ID检查 - 用于tcp_send_data_directly的安全检查 */ -static sys_thread_t g_io_thread_id; - -/* 简单的线程ID获取函数 - 适配lwIP系统 */ -static sys_thread_t sys_thread_self(void) { - /* 对于嵌入式系统，这里可以返回当前任务ID */ - /* 暂时使用简化实现，实际环境中应该调用系统API */ - extern UINT32 LOS_CurTaskIDGet(void); - return (sys_thread_t)LOS_CurTaskIDGet(); -} - -/* 简单的消息队列计数函数 - 暂时返回0 */ -static int sys_mbox_num_msgs(sys_mbox_t *mbox) { - /* 简化实现，实际环境中应该查询队列状态 */ - (void)mbox; - return 0; -} - -/* 内部函数声明 */ -static void tcp_manager_io_thread(void* arg); -static tcp_connection_info_t* tcp_manager_find_free_connection(void); -static tcp_connection_info_t* tcp_manager_find_connection_by_handle(tcp_handle_t handle); -static tcp_connection_info_t* tcp_manager_find_connection_by_pcb(struct tcp_pcb* pcb); -static void tcp_manager_cleanup_connection(tcp_connection_info_t* conn); -static err_t tcp_manager_lwip_connected_cb(void* arg, struct tcp_pcb* pcb, err_t err); -static err_t tcp_manager_lwip_recv_cb(void* arg, struct tcp_pcb* pcb, struct pbuf* p, err_t err); -static err_t tcp_manager_lwip_sent_cb(void* arg, struct tcp_pcb* pcb, u16_t len); -static void tcp_manager_lwip_err_cb(void* arg, err_t err); -static void tcp_manager_process_command(tcp_command_t* cmd); -static void tcp_manager_process_recv_data(tcp_recv_data_t* recv_data); -static tcp_manager_result_t tcp_manager_send_internal(tcp_connection_info_t* conn, uint8_t* data, uint16_t datalen); -static tcp_manager_result_t tcp_manager_close_internal(tcp_connection_info_t* conn); -static void tcp_manager_check_connection_health(void); -static void tcp_manager_handle_socket_read(tcp_connection_info_t* conn); -static void tcp_manager_handle_socket_error(tcp_connection_info_t* conn); -static void tcp_manager_process_connect_command(tcp_command_t* cmd); - -/* 工具函数 */ -static void tcp_manager_update_stats_locked(void (*update_func)(void)) { - sys_mutex_lock(&g_stats_mutex); - update_func(); - sys_mutex_unlock(&g_stats_mutex); -} - -static void stats_inc_errors(void) { - g_stats.errors++; -} - -static void stats_inc_bytes_sent_value(uint16_t bytes) { - g_stats.bytes_sent += bytes; -} - -static void stats_inc_bytes_received_value(uint16_t bytes) { - g_stats.bytes_received += bytes; -} - -/* 全局变量用于传递参数给统计函数 */ -static uint16_t g_temp_bytes_count = 0; - -static void stats_inc_bytes_sent(void) { - g_stats.bytes_sent += g_temp_bytes_count; -} - -static void stats_inc_bytes_received(void) { - g_stats.bytes_received += g_temp_bytes_count; -} - -/** - * @brief 初始化TCP管理器 - */ -tcp_manager_result_t tcp_simple_manager_init(void) -{ - if (g_manager_initialized) { - return TCP_MANAGER_OK; - } - - /* 清零统计信息 */ - memset(&g_stats, 0, sizeof(g_stats)); - - /* 初始化连接池 */ - memset(g_connections, 0, sizeof(g_connections)); - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - g_connections[i].handle = (tcp_handle_t)(uintptr_t)(i + 1); // 避免NULL handle - g_connections[i].sockfd = -1; // 初始化为无效socket - g_connections[i].in_use = false; - g_connections[i].state = TCP_STATE_DISCONNECTED; - } - - /* 创建互斥锁 */ - if (sys_mutex_new(&g_connections_mutex) != ERR_OK) { - return TCP_MANAGER_ERROR; - } - - if (sys_mutex_new(&g_stats_mutex) != ERR_OK) { - sys_mutex_free(&g_connections_mutex); - return TCP_MANAGER_ERROR; - } - - /* 创建信号量 */ - if (sys_sem_new(&g_shutdown_sem, 0) != ERR_OK) { - sys_mutex_free(&g_connections_mutex); - sys_mutex_free(&g_stats_mutex); - return TCP_MANAGER_ERROR; - } - - /* 创建队列 */ - if (sys_mbox_new(&g_cmd_queue, TCP_MANAGER_CMD_QUEUE_SIZE) != ERR_OK) { - sys_mutex_free(&g_connections_mutex); - sys_mutex_free(&g_stats_mutex); - sys_sem_free(&g_shutdown_sem); - return TCP_MANAGER_ERROR; - } - - if (sys_mbox_new(&g_recv_queue, TCP_MANAGER_RECV_QUEUE_SIZE) != ERR_OK) { - sys_mutex_free(&g_connections_mutex); - sys_mutex_free(&g_stats_mutex); - sys_sem_free(&g_shutdown_sem); - sys_mbox_free(&g_cmd_queue); - return TCP_MANAGER_ERROR; - } - - /* 启动IO线程 */ - g_manager_running = true; - g_io_thread = sys_thread_new("tcp_manager_io", tcp_manager_io_thread, NULL, - TCP_MANAGER_THREAD_STACK_SIZE, TCP_MANAGER_THREAD_PRIORITY); - if (g_io_thread == (sys_thread_t)-1) { - g_manager_running = false; - sys_mutex_free(&g_connections_mutex); - sys_mutex_free(&g_stats_mutex); - sys_sem_free(&g_shutdown_sem); - sys_mbox_free(&g_cmd_queue); - sys_mbox_free(&g_recv_queue); - return TCP_MANAGER_THREAD_ERROR; - } - - g_manager_initialized = true; - return TCP_MANAGER_OK; -} - -/** - * @brief 反初始化TCP管理器 - */ -tcp_manager_result_t tcp_simple_manager_deinit(void) -{ - if (!g_manager_initialized) { - return TCP_MANAGER_NOT_INITIALIZED; - } - - /* 发送关闭命令 */ - tcp_command_t cmd = {0}; - cmd.type = TCP_CMD_SHUTDOWN; - - if (sys_mbox_trypost(&g_cmd_queue, &cmd) != ERR_OK) { - /* 如果队列满了，强制关闭 */ - g_manager_running = false; - } - - /* 等待线程结束 */ - sys_sem_wait(&g_shutdown_sem); - - /* 清理资源 */ - sys_mbox_free(&g_cmd_queue); - sys_mbox_free(&g_recv_queue); - sys_sem_free(&g_shutdown_sem); - sys_mutex_free(&g_connections_mutex); - sys_mutex_free(&g_stats_mutex); - - g_manager_initialized = false; - return TCP_MANAGER_OK; -} - -/** - * @brief 连接到指定地址 - */ -tcp_handle_t tcp_open_url(const char* host, uint16_t port, - tcp_on_connect_cb connect_cb, - tcp_on_disconnect_cb disconnect_cb, - tcp_on_data_received_cb data_cb) -{ - if (!g_manager_initialized || !host) { - return NULL; - } - - /* 查找空闲连接 */ - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_free_connection(); - if (!conn) { - sys_mutex_unlock(&g_connections_mutex); - return NULL; - } - - /* 初始化连接信息 */ - conn->in_use = true; - conn->state = TCP_STATE_CONNECTING; - conn->remote_port = port; - conn->connect_cb = connect_cb; - conn->disconnect_cb = disconnect_cb; - conn->data_cb = data_cb; - conn->last_activity = sys_now(); - - tcp_handle_t handle = conn->handle; - sys_mutex_unlock(&g_connections_mutex); - - /* 创建连接命令 */ - tcp_command_t* cmd = (tcp_command_t*)malloc(sizeof(tcp_command_t)); - if (!cmd) { - /* 清理连接 */ - sys_mutex_lock(&g_connections_mutex); - conn->in_use = false; - conn->state = TCP_STATE_DISCONNECTED; - sys_mutex_unlock(&g_connections_mutex); - return NULL; - } - - cmd->type = TCP_CMD_CONNECT; - cmd->handle = handle; - cmd->params.connect.host = strdup(host); - cmd->params.connect.port = port; - cmd->params.connect.connect_cb = connect_cb; - cmd->params.connect.disconnect_cb = disconnect_cb; - cmd->params.connect.data_cb = data_cb; - - if (sys_mbox_trypost(&g_cmd_queue, cmd) != ERR_OK) { - /* 队列满了 */ - free(cmd->params.connect.host); - free(cmd); - sys_mutex_lock(&g_connections_mutex); - conn->in_use = false; - conn->state = TCP_STATE_DISCONNECTED; - sys_mutex_unlock(&g_connections_mutex); - return NULL; - } - - return handle; -} - -/** - * @brief 发送数据 - */ -tcp_manager_result_t tcp_send_data(tcp_handle_t handle, uint8_t* data, uint16_t datalen) -{ - if (!g_manager_initialized || !handle || !data || datalen == 0) { - return TCP_MANAGER_ERROR; - } - - /* 创建发送命令 */ - tcp_command_t* cmd = (tcp_command_t*)malloc(sizeof(tcp_command_t)); - if (!cmd) { - return TCP_MANAGER_MEMORY_ERROR; - } - - uint8_t* data_copy = (uint8_t*)malloc(datalen); - if (!data_copy) { - free(cmd); - return TCP_MANAGER_MEMORY_ERROR; - } - - memcpy(data_copy, data, datalen); - - cmd->type = TCP_CMD_SEND; - cmd->handle = handle; - cmd->params.send.data = data_copy; - cmd->params.send.length = datalen; - cmd->params.send.need_free = true; - - if (sys_mbox_trypost(&g_cmd_queue, cmd) != ERR_OK) { - free(data_copy); - free(cmd); - return TCP_MANAGER_QUEUE_FULL; - } - - return TCP_MANAGER_OK; -} - -/** - * @brief 直接发送数据 (仅在回调中使用) - */ -tcp_manager_result_t tcp_send_data_directly(tcp_handle_t handle, uint8_t* data, uint16_t datalen) -{ - if (!g_manager_initialized || !handle || !data || datalen == 0) { - return TCP_MANAGER_ERROR; - } - - /* 检查是否在IO线程中调用 */ - if (sys_thread_self() != g_io_thread_id) { - /* 不在IO线程中，使用普通发送 */ - return tcp_send_data(handle, data, datalen); - } - - /* 在IO线程中，可以直接发送 */ - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(handle); - if (!conn || conn->state != TCP_STATE_CONNECTED) { - sys_mutex_unlock(&g_connections_mutex); - return TCP_MANAGER_INVALID_HANDLE; - } - - tcp_manager_result_t result = tcp_manager_send_internal(conn, data, datalen); - sys_mutex_unlock(&g_connections_mutex); - - return result; -} - -/** - * @brief 关闭TCP连接 - */ -tcp_manager_result_t tcp_manager_close(tcp_handle_t handle) -{ - if (!g_manager_initialized || !handle) { - return TCP_MANAGER_ERROR; - } - - /* 创建关闭命令 */ - tcp_command_t* cmd = (tcp_command_t*)malloc(sizeof(tcp_command_t)); - if (!cmd) { - return TCP_MANAGER_MEMORY_ERROR; - } - - cmd->type = TCP_CMD_CLOSE; - cmd->handle = handle; - - if (sys_mbox_trypost(&g_cmd_queue, cmd) != ERR_OK) { - free(cmd); - return TCP_MANAGER_QUEUE_FULL; - } - - return TCP_MANAGER_OK; -} - -/** - * @brief 获取连接状态 - */ -tcp_connection_state_t tcp_get_connection_state(tcp_handle_t handle) -{ - if (!g_manager_initialized || !handle) { - return TCP_STATE_ERROR; - } - - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(handle); - tcp_connection_state_t state = conn ? conn->state : TCP_STATE_ERROR; - sys_mutex_unlock(&g_connections_mutex); - - return state; -} - -/** - * @brief 获取管理器统计信息 - */ -tcp_manager_result_t tcp_get_manager_stats(tcp_manager_stats_t* stats) -{ - if (!stats) { - return TCP_MANAGER_ERROR; - } - - sys_mutex_lock(&g_stats_mutex); - memcpy(stats, &g_stats, sizeof(tcp_manager_stats_t)); - - /* 更新实时统计 */ - stats->recv_queue_count = sys_mbox_valid(&g_recv_queue) ? - sys_mbox_num_msgs(&g_recv_queue) : 0; - stats->cmd_queue_count = sys_mbox_valid(&g_cmd_queue) ? - sys_mbox_num_msgs(&g_cmd_queue) : 0; - - sys_mutex_lock(&g_connections_mutex); - stats->active_connections = 0; - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use && g_connections[i].state == TCP_STATE_CONNECTED) { - stats->active_connections++; - } - } - sys_mutex_unlock(&g_connections_mutex); - - sys_mutex_unlock(&g_stats_mutex); - return TCP_MANAGER_OK; -} - -/** - * @brief 直接关闭TCP连接 (仅在回调中使用) - */ -tcp_manager_result_t tcp_manager_close_directly(tcp_handle_t handle) -{ - if (!g_manager_initialized || !handle) { - return TCP_MANAGER_ERROR; - } - - /* 检查是否在IO线程中调用 */ - if (sys_thread_self() != g_io_thread_id) { - /* 不在IO线程中，使用普通关闭 */ - return tcp_manager_close(handle); - } - - /* 在IO线程中，可以直接关闭 */ - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(handle); - if (!conn) { - sys_mutex_unlock(&g_connections_mutex); - return TCP_MANAGER_INVALID_HANDLE; - } - - tcp_manager_result_t result = tcp_manager_close_internal(conn); - sys_mutex_unlock(&g_connections_mutex); - - return result; -} - -/** - * @brief 设置连接超时时间 - */ -tcp_manager_result_t tcp_set_timeout(tcp_handle_t handle, uint32_t timeout_ms) -{ - /* 简化实现，可以在连接信息中添加超时字段 */ - (void)handle; - (void)timeout_ms; - return TCP_MANAGER_OK; -} - -/* ========== 内部函数实现 ========== */ - -/** - * @brief IO线程主函数 - 使用select机制监控socket - */ -static void tcp_manager_io_thread(void* arg) -{ - (void)arg; - - g_io_thread_id = sys_thread_self(); - fd_set read_fds, write_fds, error_fds; - int max_fd = 0; - struct timeval timeout; - - while (g_manager_running) { - /* 重置文件描述符集合 */ - FD_ZERO(&read_fds); - FD_ZERO(&write_fds); - FD_ZERO(&error_fds); - max_fd = 0; - - /* 添加所有活跃连接的socket到监控集合 */ - sys_mutex_lock(&g_connections_mutex); - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use && g_connections[i].sockfd >= 0) { - FD_SET(g_connections[i].sockfd, &read_fds); - FD_SET(g_connections[i].sockfd, &error_fds); - if (g_connections[i].state == TCP_STATE_CONNECTING) { - FD_SET(g_connections[i].sockfd, &write_fds); - } - if (g_connections[i].sockfd > max_fd) { - max_fd = g_connections[i].sockfd; - } - } - } - sys_mutex_unlock(&g_connections_mutex); - - /* 设置超时时间 */ - timeout.tv_sec = 0; - timeout.tv_usec = 50000; /* 50ms */ - - /* 使用select进行多路复用监控 */ - int select_result = select(max_fd + 1, &read_fds, &write_fds, &error_fds, &timeout); - - if (select_result > 0) { - /* 有socket活动，处理事件 */ - sys_mutex_lock(&g_connections_mutex); - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use && g_connections[i].sockfd >= 0) { - int sockfd = g_connections[i].sockfd; - - /* 检查连接完成事件 */ - if (g_connections[i].state == TCP_STATE_CONNECTING && FD_ISSET(sockfd, &write_fds)) { - int error = 0; - socklen_t len = sizeof(error); - if (getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &error, &len) == 0 && error == 0) { - /* 连接成功 */ - g_connections[i].state = TCP_STATE_CONNECTED; - g_connections[i].last_activity = sys_now(); - if (g_connections[i].connect_cb) { - g_connections[i].connect_cb(g_connections[i].handle, true); - } - } else { - /* 连接失败 */ - g_connections[i].state = TCP_STATE_ERROR; - if (g_connections[i].connect_cb) { - g_connections[i].connect_cb(g_connections[i].handle, false); - } - tcp_manager_cleanup_connection(&g_connections[i]); - } - } - - /* 检查读事件 */ - if (FD_ISSET(sockfd, &read_fds)) { - tcp_manager_handle_socket_read(&g_connections[i]); - } - - /* 检查错误事件 */ - if (FD_ISSET(sockfd, &error_fds)) { - tcp_manager_handle_socket_error(&g_connections[i]); - } - } - } - sys_mutex_unlock(&g_connections_mutex); - } else if (select_result < 0 && errno != EINTR) { - /* select错误 */ - tcp_manager_update_stats_locked(stats_inc_errors); - } - - /* 处理命令队列 - 非阻塞处理 */ - void* msg; - while (sys_mbox_tryfetch(&g_cmd_queue, &msg) == ERR_OK) { - tcp_command_t* cmd = (tcp_command_t*)msg; - tcp_manager_process_command(cmd); - - if (cmd->type == TCP_CMD_SHUTDOWN) { - g_manager_running = false; - goto cleanup; - } - } - - /* 处理接收队列 - 非阻塞处理 */ - while (sys_mbox_tryfetch(&g_recv_queue, &msg) == ERR_OK) { - tcp_recv_data_t* recv_data = (tcp_recv_data_t*)msg; - tcp_manager_process_recv_data(recv_data); - free(recv_data); - } - - /* 检查连接健康状态和超时 */ - tcp_manager_check_connection_health(); - } - -cleanup: - /* 清理所有连接 */ - sys_mutex_lock(&g_connections_mutex); - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use) { - tcp_manager_cleanup_connection(&g_connections[i]); - } - } - sys_mutex_unlock(&g_connections_mutex); - - /* 通知主线程可以退出 */ - sys_sem_signal(&g_shutdown_sem); -} - -/** - * @brief 查找空闲连接 - */ -static tcp_connection_info_t* tcp_manager_find_free_connection(void) -{ - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (!g_connections[i].in_use) { - return &g_connections[i]; - } - } - return NULL; -} - -/** - * @brief 根据句柄查找连接 - */ -static tcp_connection_info_t* tcp_manager_find_connection_by_handle(tcp_handle_t handle) -{ - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use && g_connections[i].handle == handle) { - return &g_connections[i]; - } - } - return NULL; -} - -/** - * @brief 根据PCB查找连接 - */ -static tcp_connection_info_t* tcp_manager_find_connection_by_pcb(struct tcp_pcb* pcb) -{ - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use && g_connections[i].pcb == pcb) { - return &g_connections[i]; - } - } - return NULL; -} - -/** - * @brief 清理连接 - */ -static void tcp_manager_cleanup_connection(tcp_connection_info_t* conn) -{ - if (!conn) return; - - if (conn->sockfd >= 0) { - close(conn->sockfd); - conn->sockfd = -1; - } - - conn->in_use = false; - conn->state = TCP_STATE_DISCONNECTED; - conn->connect_cb = NULL; - conn->disconnect_cb = NULL; - conn->data_cb = NULL; -} - -/* ========== 移除了lwIP内部回调函数，使用标准socket接口 ========== */ - -/* ========== 命令处理函数 ========== */ - -static void tcp_manager_process_command(tcp_command_t* cmd) -{ - switch (cmd->type) { - case TCP_CMD_CONNECT: { - tcp_manager_process_connect_command(cmd); - break; - } - - case TCP_CMD_SEND: { - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(cmd->handle); - if (conn && conn->state == TCP_STATE_CONNECTED && conn->sockfd >= 0) { - tcp_manager_send_internal(conn, cmd->params.send.data, cmd->params.send.length); - } - sys_mutex_unlock(&g_connections_mutex); - - /* 释放命令资源 */ - if (cmd->params.send.need_free) { - free(cmd->params.send.data); - } - free(cmd); - break; - } - - case TCP_CMD_CLOSE: { - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(cmd->handle); - if (conn) { - conn->state = TCP_STATE_CLOSING; - if (conn->disconnect_cb) { - conn->disconnect_cb(conn->handle); - } - tcp_manager_cleanup_connection(conn); - } - sys_mutex_unlock(&g_connections_mutex); - - free(cmd); - break; - } - - case TCP_CMD_SHUTDOWN: - /* 不释放cmd，因为它在栈上 */ - break; - - default: - free(cmd); - break; - } -} - -static void tcp_manager_process_recv_data(tcp_recv_data_t* recv_data) -{ - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(recv_data->handle); - if (conn && conn->data_cb) { - tcp_on_data_received_cb callback = conn->data_cb; - sys_mutex_unlock(&g_connections_mutex); - - /* 调用用户回调 */ - callback(recv_data->handle, recv_data->data, recv_data->length); - } else { - sys_mutex_unlock(&g_connections_mutex); - } -} - -static tcp_manager_result_t tcp_manager_send_internal(tcp_connection_info_t* conn, uint8_t* data, uint16_t datalen) -{ - if (!conn || conn->sockfd < 0 || !data || datalen == 0) { - return TCP_MANAGER_ERROR; - } - - ssize_t sent = send(conn->sockfd, data, datalen, 0); - if (sent > 0) { - conn->last_activity = sys_now(); - g_temp_bytes_count = (uint16_t)sent; - tcp_manager_update_stats_locked(stats_inc_bytes_sent); - return TCP_MANAGER_OK; - } - - tcp_manager_update_stats_locked(stats_inc_errors); - return TCP_MANAGER_ERROR; -} - -/** - * @brief 处理连接命令 - */ -static void tcp_manager_process_connect_command(tcp_command_t* cmd) -{ - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(cmd->handle); - if (!conn) { - sys_mutex_unlock(&g_connections_mutex); - goto cleanup; - } - - /* 创建socket */ - int sockfd = socket(AF_INET, SOCK_STREAM, 0); - if (sockfd < 0) { - conn->state = TCP_STATE_ERROR; - if (conn->connect_cb) { - conn->connect_cb(conn->handle, false); - } - tcp_manager_cleanup_connection(conn); - tcp_manager_update_stats_locked(stats_inc_errors); - sys_mutex_unlock(&g_connections_mutex); - goto cleanup; - } - - /* 设置非阻塞模式 */ - int flags = fcntl(sockfd, F_GETFL, 0); - if (flags >= 0) { - fcntl(sockfd, F_SETFL, flags | O_NONBLOCK); - } - - /* 设置socket选项 */ - int opt = 1; - setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); - - conn->sockfd = sockfd; - conn->state = TCP_STATE_CONNECTING; - strcpy(conn->remote_ip, cmd->params.connect.host); - conn->remote_port = cmd->params.connect.port; - - /* 解析地址并连接 */ - struct sockaddr_in server_addr; - memset(&server_addr, 0, sizeof(server_addr)); - server_addr.sin_family = AF_INET; - server_addr.sin_port = htons(cmd->params.connect.port); - - /* 尝试解析IP地址 */ - if (inet_aton(cmd->params.connect.host, &server_addr.sin_addr) == 0) { - /* 不是IP地址，尝试域名解析 */ - struct hostent *he = gethostbyname(cmd->params.connect.host); - if (he && he->h_addr_list[0]) { - memcpy(&server_addr.sin_addr, he->h_addr_list[0], he->h_length); - } else { - /* 解析失败 */ - conn->state = TCP_STATE_ERROR; - if (conn->connect_cb) { - conn->connect_cb(conn->handle, false); - } - tcp_manager_cleanup_connection(conn); - tcp_manager_update_stats_locked(stats_inc_errors); - sys_mutex_unlock(&g_connections_mutex); - goto cleanup; - } - } - - /* 非阻塞连接 */ - int result = connect(sockfd, (struct sockaddr*)&server_addr, sizeof(server_addr)); - if (result == 0) { - /* 立即连接成功 */ - conn->state = TCP_STATE_CONNECTED; - conn->last_activity = sys_now(); - if (conn->connect_cb) { - conn->connect_cb(conn->handle, true); - } - } else if (errno == EINPROGRESS) { - /* 连接正在进行中，等待select检测 */ - /* 状态已经设置为TCP_STATE_CONNECTING */ - } else { - /* 连接失败 */ - conn->state = TCP_STATE_ERROR; - if (conn->connect_cb) { - conn->connect_cb(conn->handle, false); - } - tcp_manager_cleanup_connection(conn); - tcp_manager_update_stats_locked(stats_inc_errors); - } - - sys_mutex_unlock(&g_connections_mutex); - -cleanup: - /* 释放命令资源 */ - free(cmd->params.connect.host); - free(cmd); -} - -/** - * @brief 处理socket读事件 - */ -static void tcp_manager_handle_socket_read(tcp_connection_info_t* conn) -{ - if (!conn || conn->sockfd < 0) return; - - uint8_t buffer[TCP_MANAGER_RECV_BUFFER_SIZE]; - ssize_t bytes_read = recv(conn->sockfd, buffer, sizeof(buffer), MSG_DONTWAIT); - - if (bytes_read > 0) { - /* 分配接收数据结构 */ - tcp_recv_data_t* recv_data = (tcp_recv_data_t*)malloc(sizeof(tcp_recv_data_t)); - if (!recv_data) { - tcp_manager_update_stats_locked(stats_inc_errors); - return; - } - - recv_data->handle = conn->handle; - recv_data->length = (uint16_t)bytes_read; - recv_data->timestamp = sys_now(); - memcpy(recv_data->data, buffer, bytes_read); - - /* 更新统计 */ - g_temp_bytes_count = recv_data->length; - tcp_manager_update_stats_locked(stats_inc_bytes_received); - - /* 投递到接收队列 */ - if (sys_mbox_trypost(&g_recv_queue, recv_data) != ERR_OK) { - free(recv_data); - tcp_manager_update_stats_locked(stats_inc_errors); - } - - conn->last_activity = sys_now(); - } else if (bytes_read == 0) { - /* 连接被对端关闭 */ - conn->state = TCP_STATE_DISCONNECTED; - if (conn->disconnect_cb) { - conn->disconnect_cb(conn->handle); - } - tcp_manager_cleanup_connection(conn); - } else if (errno != EAGAIN && errno != EWOULDBLOCK) { - /* 读取错误 */ - tcp_manager_handle_socket_error(conn); - } -} - -/** - * @brief 处理socket错误事件 - */ -static void tcp_manager_handle_socket_error(tcp_connection_info_t* conn) -{ - if (!conn) return; - - conn->state = TCP_STATE_ERROR; - if (conn->disconnect_cb) { - conn->disconnect_cb(conn->handle); - } - - tcp_manager_cleanup_connection(conn); - tcp_manager_update_stats_locked(stats_inc_errors); -} - -/** - * @brief 检查连接健康状态 - */ -static void tcp_manager_check_connection_health(void) -{ - uint32_t current_time = sys_now(); - - sys_mutex_lock(&g_connections_mutex); - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use) { - /* 检查连接超时 (30秒无活动) */ - if (current_time - g_connections[i].last_activity > 30000) { - if (g_connections[i].state == TCP_STATE_CONNECTED) { - /* 发送心跳包或标记为超时 */ - g_connections[i].state = TCP_STATE_ERROR; - if (g_connections[i].disconnect_cb) { - g_connections[i].disconnect_cb(g_connections[i].handle); - } - tcp_manager_cleanup_connection(&g_connections[i]); - } - } - } - } - sys_mutex_unlock(&g_connections_mutex); -} - -/** - * @brief 内部关闭连接实现 - */ -static tcp_manager_result_t tcp_manager_close_internal(tcp_connection_info_t* conn) -{ - if (!conn) return TCP_MANAGER_ERROR; - - conn->state = TCP_STATE_CLOSING; - - /* 如果有断开回调，先调用 */ - if (conn->disconnect_cb) { - conn->disconnect_cb(conn->handle); - } - - /* 清理连接资源 */ - tcp_manager_cleanup_connection(conn); - - return TCP_MANAGER_OK; -} \ No newline at end of file diff --git a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager_full.c b/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager_full.c deleted file mode 100644 index 2a5e7f3d1..000000000 --- a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager_full.c +++ /dev/null @@ -1,942 +0,0 @@ -/** - * @file tcp_simple_manager.c - * @brief TCP简单管理器实现 - */ - -#include "tcp_simple_manager.h" -#include "lwip/sockets.h" -#include "lwip/netdb.h" -#include "lwip/sys.h" -#include "lwip/err.h" -#include -#include -#include -#include -#include -#include -#include -#include - -/* lwIP socket函数重定义，确保使用lwIP接口 */ -#define socket lwip_socket -#define connect lwip_connect -#define close lwip_close -#define send lwip_send -#define recv lwip_recv -#define bind lwip_bind -#define listen lwip_listen -#define accept lwip_accept -#define getsockopt lwip_getsockopt -#define setsockopt lwip_setsockopt - -/* 内部状态标志 */ -static bool g_manager_initialized = false; -static bool g_manager_running = false; - -/* 线程相关 */ -static sys_thread_t g_io_thread; -static sys_sem_t g_shutdown_sem; - -/* 队列相关 */ -static sys_mbox_t g_cmd_queue; -static sys_mbox_t g_recv_queue; - -/* 连接池 */ -static tcp_connection_info_t g_connections[TCP_MANAGER_MAX_CONNECTIONS]; -static sys_mutex_t g_connections_mutex; - -/* 统计信息 */ -static tcp_manager_stats_t g_stats; -static sys_mutex_t g_stats_mutex; - -/* 当前线程ID检查 - 用于tcp_send_data_directly的安全检查 */ -static sys_thread_t g_io_thread_id; - -/* 简单的线程ID获取函数 - 适配lwIP系统 */ -static sys_thread_t sys_thread_self(void) { - /* 对于嵌入式系统，这里可以返回当前任务ID */ - /* 暂时使用简化实现，实际环境中应该调用系统API */ - extern UINT32 LOS_CurTaskIDGet(void); - return (sys_thread_t)LOS_CurTaskIDGet(); -} - -/* 简单的消息队列计数函数 - 暂时返回0 */ -static int sys_mbox_num_msgs(sys_mbox_t *mbox) { - /* 简化实现，实际环境中应该查询队列状态 */ - (void)mbox; - return 0; -} - -/* 内部函数声明 */ -static void tcp_manager_io_thread(void* arg); -static tcp_connection_info_t* tcp_manager_find_free_connection(void); -static tcp_connection_info_t* tcp_manager_find_connection_by_handle(tcp_handle_t handle); -static tcp_connection_info_t* tcp_manager_find_connection_by_pcb(struct tcp_pcb* pcb); -static void tcp_manager_cleanup_connection(tcp_connection_info_t* conn); -static err_t tcp_manager_lwip_connected_cb(void* arg, struct tcp_pcb* pcb, err_t err); -static err_t tcp_manager_lwip_recv_cb(void* arg, struct tcp_pcb* pcb, struct pbuf* p, err_t err); -static err_t tcp_manager_lwip_sent_cb(void* arg, struct tcp_pcb* pcb, u16_t len); -static void tcp_manager_lwip_err_cb(void* arg, err_t err); -static void tcp_manager_process_command(tcp_command_t* cmd); -static void tcp_manager_process_recv_data(tcp_recv_data_t* recv_data); -static tcp_manager_result_t tcp_manager_send_internal(tcp_connection_info_t* conn, uint8_t* data, uint16_t datalen); -static tcp_manager_result_t tcp_manager_close_internal(tcp_connection_info_t* conn); -static void tcp_manager_check_connection_health(void); -static void tcp_manager_handle_socket_read(tcp_connection_info_t* conn); -static void tcp_manager_handle_socket_error(tcp_connection_info_t* conn); -static void tcp_manager_process_connect_command(tcp_command_t* cmd); - -/* 工具函数 */ -static void tcp_manager_update_stats_locked(void (*update_func)(void)) { - sys_mutex_lock(&g_stats_mutex); - update_func(); - sys_mutex_unlock(&g_stats_mutex); -} - -static void stats_inc_errors(void) { - g_stats.errors++; -} - -static void stats_inc_bytes_sent_value(uint16_t bytes) { - g_stats.bytes_sent += bytes; -} - -static void stats_inc_bytes_received_value(uint16_t bytes) { - g_stats.bytes_received += bytes; -} - -/* 全局变量用于传递参数给统计函数 */ -static uint16_t g_temp_bytes_count = 0; - -static void stats_inc_bytes_sent(void) { - g_stats.bytes_sent += g_temp_bytes_count; -} - -static void stats_inc_bytes_received(void) { - g_stats.bytes_received += g_temp_bytes_count; -} - -/** - * @brief 初始化TCP管理器 - */ -tcp_manager_result_t tcp_simple_manager_init(void) -{ - if (g_manager_initialized) { - return TCP_MANAGER_OK; - } - - /* 清零统计信息 */ - memset(&g_stats, 0, sizeof(g_stats)); - - /* 初始化连接池 */ - memset(g_connections, 0, sizeof(g_connections)); - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - g_connections[i].handle = (tcp_handle_t)(uintptr_t)(i + 1); // 避免NULL handle - g_connections[i].sockfd = -1; // 初始化为无效socket - g_connections[i].in_use = false; - g_connections[i].state = TCP_STATE_DISCONNECTED; - } - - /* 创建互斥锁 */ - if (sys_mutex_new(&g_connections_mutex) != ERR_OK) { - return TCP_MANAGER_ERROR; - } - - if (sys_mutex_new(&g_stats_mutex) != ERR_OK) { - sys_mutex_free(&g_connections_mutex); - return TCP_MANAGER_ERROR; - } - - /* 创建信号量 */ - if (sys_sem_new(&g_shutdown_sem, 0) != ERR_OK) { - sys_mutex_free(&g_connections_mutex); - sys_mutex_free(&g_stats_mutex); - return TCP_MANAGER_ERROR; - } - - /* 创建队列 */ - if (sys_mbox_new(&g_cmd_queue, TCP_MANAGER_CMD_QUEUE_SIZE) != ERR_OK) { - sys_mutex_free(&g_connections_mutex); - sys_mutex_free(&g_stats_mutex); - sys_sem_free(&g_shutdown_sem); - return TCP_MANAGER_ERROR; - } - - if (sys_mbox_new(&g_recv_queue, TCP_MANAGER_RECV_QUEUE_SIZE) != ERR_OK) { - sys_mutex_free(&g_connections_mutex); - sys_mutex_free(&g_stats_mutex); - sys_sem_free(&g_shutdown_sem); - sys_mbox_free(&g_cmd_queue); - return TCP_MANAGER_ERROR; - } - - /* 启动IO线程 */ - g_manager_running = true; - g_io_thread = sys_thread_new("tcp_manager_io", tcp_manager_io_thread, NULL, - TCP_MANAGER_THREAD_STACK_SIZE, TCP_MANAGER_THREAD_PRIORITY); - if (g_io_thread == (sys_thread_t)-1) { - g_manager_running = false; - sys_mutex_free(&g_connections_mutex); - sys_mutex_free(&g_stats_mutex); - sys_sem_free(&g_shutdown_sem); - sys_mbox_free(&g_cmd_queue); - sys_mbox_free(&g_recv_queue); - return TCP_MANAGER_THREAD_ERROR; - } - - g_manager_initialized = true; - return TCP_MANAGER_OK; -} - -/** - * @brief 反初始化TCP管理器 - */ -tcp_manager_result_t tcp_simple_manager_deinit(void) -{ - if (!g_manager_initialized) { - return TCP_MANAGER_NOT_INITIALIZED; - } - - /* 发送关闭命令 */ - tcp_command_t cmd = {0}; - cmd.type = TCP_CMD_SHUTDOWN; - - if (sys_mbox_trypost(&g_cmd_queue, &cmd) != ERR_OK) { - /* 如果队列满了，强制关闭 */ - g_manager_running = false; - } - - /* 等待线程结束 */ - sys_sem_wait(&g_shutdown_sem); - - /* 清理资源 */ - sys_mbox_free(&g_cmd_queue); - sys_mbox_free(&g_recv_queue); - sys_sem_free(&g_shutdown_sem); - sys_mutex_free(&g_connections_mutex); - sys_mutex_free(&g_stats_mutex); - - g_manager_initialized = false; - return TCP_MANAGER_OK; -} - -/** - * @brief 连接到指定地址 - */ -tcp_handle_t tcp_open_url(const char* host, uint16_t port, - tcp_on_connect_cb connect_cb, - tcp_on_disconnect_cb disconnect_cb, - tcp_on_data_received_cb data_cb) -{ - if (!g_manager_initialized || !host) { - return NULL; - } - - /* 查找空闲连接 */ - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_free_connection(); - if (!conn) { - sys_mutex_unlock(&g_connections_mutex); - return NULL; - } - - /* 初始化连接信息 */ - conn->in_use = true; - conn->state = TCP_STATE_CONNECTING; - conn->remote_port = port; - conn->connect_cb = connect_cb; - conn->disconnect_cb = disconnect_cb; - conn->data_cb = data_cb; - conn->last_activity = sys_now(); - - tcp_handle_t handle = conn->handle; - sys_mutex_unlock(&g_connections_mutex); - - /* 创建连接命令 */ - tcp_command_t* cmd = (tcp_command_t*)malloc(sizeof(tcp_command_t)); - if (!cmd) { - /* 清理连接 */ - sys_mutex_lock(&g_connections_mutex); - conn->in_use = false; - conn->state = TCP_STATE_DISCONNECTED; - sys_mutex_unlock(&g_connections_mutex); - return NULL; - } - - cmd->type = TCP_CMD_CONNECT; - cmd->handle = handle; - cmd->params.connect.host = strdup(host); - cmd->params.connect.port = port; - cmd->params.connect.connect_cb = connect_cb; - cmd->params.connect.disconnect_cb = disconnect_cb; - cmd->params.connect.data_cb = data_cb; - - if (sys_mbox_trypost(&g_cmd_queue, cmd) != ERR_OK) { - /* 队列满了 */ - free(cmd->params.connect.host); - free(cmd); - sys_mutex_lock(&g_connections_mutex); - conn->in_use = false; - conn->state = TCP_STATE_DISCONNECTED; - sys_mutex_unlock(&g_connections_mutex); - return NULL; - } - - return handle; -} - -/** - * @brief 发送数据 - */ -tcp_manager_result_t tcp_send_data(tcp_handle_t handle, uint8_t* data, uint16_t datalen) -{ - if (!g_manager_initialized || !handle || !data || datalen == 0) { - return TCP_MANAGER_ERROR; - } - - /* 创建发送命令 */ - tcp_command_t* cmd = (tcp_command_t*)malloc(sizeof(tcp_command_t)); - if (!cmd) { - return TCP_MANAGER_MEMORY_ERROR; - } - - uint8_t* data_copy = (uint8_t*)malloc(datalen); - if (!data_copy) { - free(cmd); - return TCP_MANAGER_MEMORY_ERROR; - } - - memcpy(data_copy, data, datalen); - - cmd->type = TCP_CMD_SEND; - cmd->handle = handle; - cmd->params.send.data = data_copy; - cmd->params.send.length = datalen; - cmd->params.send.need_free = true; - - if (sys_mbox_trypost(&g_cmd_queue, cmd) != ERR_OK) { - free(data_copy); - free(cmd); - return TCP_MANAGER_QUEUE_FULL; - } - - return TCP_MANAGER_OK; -} - -/** - * @brief 直接发送数据 (仅在回调中使用) - */ -tcp_manager_result_t tcp_send_data_directly(tcp_handle_t handle, uint8_t* data, uint16_t datalen) -{ - if (!g_manager_initialized || !handle || !data || datalen == 0) { - return TCP_MANAGER_ERROR; - } - - /* 检查是否在IO线程中调用 */ - if (sys_thread_self() != g_io_thread_id) { - /* 不在IO线程中，使用普通发送 */ - return tcp_send_data(handle, data, datalen); - } - - /* 在IO线程中，可以直接发送 */ - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(handle); - if (!conn || conn->state != TCP_STATE_CONNECTED) { - sys_mutex_unlock(&g_connections_mutex); - return TCP_MANAGER_INVALID_HANDLE; - } - - tcp_manager_result_t result = tcp_manager_send_internal(conn, data, datalen); - sys_mutex_unlock(&g_connections_mutex); - - return result; -} - -/** - * @brief 关闭TCP连接 - */ -tcp_manager_result_t tcp_manager_close(tcp_handle_t handle) -{ - if (!g_manager_initialized || !handle) { - return TCP_MANAGER_ERROR; - } - - /* 创建关闭命令 */ - tcp_command_t* cmd = (tcp_command_t*)malloc(sizeof(tcp_command_t)); - if (!cmd) { - return TCP_MANAGER_MEMORY_ERROR; - } - - cmd->type = TCP_CMD_CLOSE; - cmd->handle = handle; - - if (sys_mbox_trypost(&g_cmd_queue, cmd) != ERR_OK) { - free(cmd); - return TCP_MANAGER_QUEUE_FULL; - } - - return TCP_MANAGER_OK; -} - -/** - * @brief 获取连接状态 - */ -tcp_connection_state_t tcp_get_connection_state(tcp_handle_t handle) -{ - if (!g_manager_initialized || !handle) { - return TCP_STATE_ERROR; - } - - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(handle); - tcp_connection_state_t state = conn ? conn->state : TCP_STATE_ERROR; - sys_mutex_unlock(&g_connections_mutex); - - return state; -} - -/** - * @brief 获取管理器统计信息 - */ -tcp_manager_result_t tcp_get_manager_stats(tcp_manager_stats_t* stats) -{ - if (!stats) { - return TCP_MANAGER_ERROR; - } - - sys_mutex_lock(&g_stats_mutex); - memcpy(stats, &g_stats, sizeof(tcp_manager_stats_t)); - - /* 更新实时统计 */ - stats->recv_queue_count = sys_mbox_valid(&g_recv_queue) ? - sys_mbox_num_msgs(&g_recv_queue) : 0; - stats->cmd_queue_count = sys_mbox_valid(&g_cmd_queue) ? - sys_mbox_num_msgs(&g_cmd_queue) : 0; - - sys_mutex_lock(&g_connections_mutex); - stats->active_connections = 0; - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use && g_connections[i].state == TCP_STATE_CONNECTED) { - stats->active_connections++; - } - } - sys_mutex_unlock(&g_connections_mutex); - - sys_mutex_unlock(&g_stats_mutex); - return TCP_MANAGER_OK; -} - -/** - * @brief 直接关闭TCP连接 (仅在回调中使用) - */ -tcp_manager_result_t tcp_manager_close_directly(tcp_handle_t handle) -{ - if (!g_manager_initialized || !handle) { - return TCP_MANAGER_ERROR; - } - - /* 检查是否在IO线程中调用 */ - if (sys_thread_self() != g_io_thread_id) { - /* 不在IO线程中，使用普通关闭 */ - return tcp_manager_close(handle); - } - - /* 在IO线程中，可以直接关闭 */ - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(handle); - if (!conn) { - sys_mutex_unlock(&g_connections_mutex); - return TCP_MANAGER_INVALID_HANDLE; - } - - tcp_manager_result_t result = tcp_manager_close_internal(conn); - sys_mutex_unlock(&g_connections_mutex); - - return result; -} - -/** - * @brief 设置连接超时时间 - */ -tcp_manager_result_t tcp_set_timeout(tcp_handle_t handle, uint32_t timeout_ms) -{ - /* 简化实现，可以在连接信息中添加超时字段 */ - (void)handle; - (void)timeout_ms; - return TCP_MANAGER_OK; -} - -/* ========== 内部函数实现 ========== */ - -/** - * @brief IO线程主函数 - 使用select机制监控socket - */ -static void tcp_manager_io_thread(void* arg) -{ - (void)arg; - - g_io_thread_id = sys_thread_self(); - fd_set read_fds, write_fds, error_fds; - int max_fd = 0; - struct timeval timeout; - - while (g_manager_running) { - /* 重置文件描述符集合 */ - FD_ZERO(&read_fds); - FD_ZERO(&write_fds); - FD_ZERO(&error_fds); - max_fd = 0; - - /* 添加所有活跃连接的socket到监控集合 */ - sys_mutex_lock(&g_connections_mutex); - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use && g_connections[i].sockfd >= 0) { - FD_SET(g_connections[i].sockfd, &read_fds); - FD_SET(g_connections[i].sockfd, &error_fds); - if (g_connections[i].state == TCP_STATE_CONNECTING) { - FD_SET(g_connections[i].sockfd, &write_fds); - } - if (g_connections[i].sockfd > max_fd) { - max_fd = g_connections[i].sockfd; - } - } - } - sys_mutex_unlock(&g_connections_mutex); - - /* 设置超时时间 */ - timeout.tv_sec = 0; - timeout.tv_usec = 50000; /* 50ms */ - - /* 使用select进行多路复用监控 */ - int select_result = select(max_fd + 1, &read_fds, &write_fds, &error_fds, &timeout); - - if (select_result > 0) { - /* 有socket活动，处理事件 */ - sys_mutex_lock(&g_connections_mutex); - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use && g_connections[i].sockfd >= 0) { - int sockfd = g_connections[i].sockfd; - - /* 检查连接完成事件 */ - if (g_connections[i].state == TCP_STATE_CONNECTING && FD_ISSET(sockfd, &write_fds)) { - int error = 0; - socklen_t len = sizeof(error); - if (getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &error, &len) == 0 && error == 0) { - /* 连接成功 */ - g_connections[i].state = TCP_STATE_CONNECTED; - g_connections[i].last_activity = sys_now(); - if (g_connections[i].connect_cb) { - g_connections[i].connect_cb(g_connections[i].handle, true); - } - } else { - /* 连接失败 */ - g_connections[i].state = TCP_STATE_ERROR; - if (g_connections[i].connect_cb) { - g_connections[i].connect_cb(g_connections[i].handle, false); - } - tcp_manager_cleanup_connection(&g_connections[i]); - } - } - - /* 检查读事件 */ - if (FD_ISSET(sockfd, &read_fds)) { - tcp_manager_handle_socket_read(&g_connections[i]); - } - - /* 检查错误事件 */ - if (FD_ISSET(sockfd, &error_fds)) { - tcp_manager_handle_socket_error(&g_connections[i]); - } - } - } - sys_mutex_unlock(&g_connections_mutex); - } else if (select_result < 0 && errno != EINTR) { - /* select错误 */ - tcp_manager_update_stats_locked(stats_inc_errors); - } - - /* 处理命令队列 - 非阻塞处理 */ - void* msg; - while (sys_mbox_tryfetch(&g_cmd_queue, &msg) == ERR_OK) { - tcp_command_t* cmd = (tcp_command_t*)msg; - tcp_manager_process_command(cmd); - - if (cmd->type == TCP_CMD_SHUTDOWN) { - g_manager_running = false; - goto cleanup; - } - } - - /* 处理接收队列 - 非阻塞处理 */ - while (sys_mbox_tryfetch(&g_recv_queue, &msg) == ERR_OK) { - tcp_recv_data_t* recv_data = (tcp_recv_data_t*)msg; - tcp_manager_process_recv_data(recv_data); - free(recv_data); - } - - /* 检查连接健康状态和超时 */ - tcp_manager_check_connection_health(); - } - -cleanup: - /* 清理所有连接 */ - sys_mutex_lock(&g_connections_mutex); - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use) { - tcp_manager_cleanup_connection(&g_connections[i]); - } - } - sys_mutex_unlock(&g_connections_mutex); - - /* 通知主线程可以退出 */ - sys_sem_signal(&g_shutdown_sem); -} - -/** - * @brief 查找空闲连接 - */ -static tcp_connection_info_t* tcp_manager_find_free_connection(void) -{ - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (!g_connections[i].in_use) { - return &g_connections[i]; - } - } - return NULL; -} - -/** - * @brief 根据句柄查找连接 - */ -static tcp_connection_info_t* tcp_manager_find_connection_by_handle(tcp_handle_t handle) -{ - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use && g_connections[i].handle == handle) { - return &g_connections[i]; - } - } - return NULL; -} - -/** - * @brief 根据PCB查找连接 - */ -static tcp_connection_info_t* tcp_manager_find_connection_by_pcb(struct tcp_pcb* pcb) -{ - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use && g_connections[i].pcb == pcb) { - return &g_connections[i]; - } - } - return NULL; -} - -/** - * @brief 清理连接 - */ -static void tcp_manager_cleanup_connection(tcp_connection_info_t* conn) -{ - if (!conn) return; - - if (conn->sockfd >= 0) { - close(conn->sockfd); - conn->sockfd = -1; - } - - conn->in_use = false; - conn->state = TCP_STATE_DISCONNECTED; - conn->connect_cb = NULL; - conn->disconnect_cb = NULL; - conn->data_cb = NULL; -} - -/* ========== 移除了lwIP内部回调函数，使用标准socket接口 ========== */ - -/* ========== 命令处理函数 ========== */ - -static void tcp_manager_process_command(tcp_command_t* cmd) -{ - switch (cmd->type) { - case TCP_CMD_CONNECT: { - tcp_manager_process_connect_command(cmd); - break; - } - - case TCP_CMD_SEND: { - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(cmd->handle); - if (conn && conn->state == TCP_STATE_CONNECTED && conn->sockfd >= 0) { - tcp_manager_send_internal(conn, cmd->params.send.data, cmd->params.send.length); - } - sys_mutex_unlock(&g_connections_mutex); - - /* 释放命令资源 */ - if (cmd->params.send.need_free) { - free(cmd->params.send.data); - } - free(cmd); - break; - } - - case TCP_CMD_CLOSE: { - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(cmd->handle); - if (conn) { - conn->state = TCP_STATE_CLOSING; - if (conn->disconnect_cb) { - conn->disconnect_cb(conn->handle); - } - tcp_manager_cleanup_connection(conn); - } - sys_mutex_unlock(&g_connections_mutex); - - free(cmd); - break; - } - - case TCP_CMD_SHUTDOWN: - /* 不释放cmd，因为它在栈上 */ - break; - - default: - free(cmd); - break; - } -} - -static void tcp_manager_process_recv_data(tcp_recv_data_t* recv_data) -{ - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(recv_data->handle); - if (conn && conn->data_cb) { - tcp_on_data_received_cb callback = conn->data_cb; - sys_mutex_unlock(&g_connections_mutex); - - /* 调用用户回调 */ - callback(recv_data->handle, recv_data->data, recv_data->length); - } else { - sys_mutex_unlock(&g_connections_mutex); - } -} - -static tcp_manager_result_t tcp_manager_send_internal(tcp_connection_info_t* conn, uint8_t* data, uint16_t datalen) -{ - if (!conn || conn->sockfd < 0 || !data || datalen == 0) { - return TCP_MANAGER_ERROR; - } - - ssize_t sent = send(conn->sockfd, data, datalen, 0); - if (sent > 0) { - conn->last_activity = sys_now(); - g_temp_bytes_count = (uint16_t)sent; - tcp_manager_update_stats_locked(stats_inc_bytes_sent); - return TCP_MANAGER_OK; - } - - tcp_manager_update_stats_locked(stats_inc_errors); - return TCP_MANAGER_ERROR; -} - -/** - * @brief 处理连接命令 - */ -static void tcp_manager_process_connect_command(tcp_command_t* cmd) -{ - sys_mutex_lock(&g_connections_mutex); - tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(cmd->handle); - if (!conn) { - sys_mutex_unlock(&g_connections_mutex); - goto cleanup; - } - - /* 创建socket */ - int sockfd = socket(AF_INET, SOCK_STREAM, 0); - if (sockfd < 0) { - conn->state = TCP_STATE_ERROR; - if (conn->connect_cb) { - conn->connect_cb(conn->handle, false); - } - tcp_manager_cleanup_connection(conn); - tcp_manager_update_stats_locked(stats_inc_errors); - sys_mutex_unlock(&g_connections_mutex); - goto cleanup; - } - - /* 设置非阻塞模式 */ - int flags = fcntl(sockfd, F_GETFL, 0); - if (flags >= 0) { - fcntl(sockfd, F_SETFL, flags | O_NONBLOCK); - } - - /* 设置socket选项 */ - int opt = 1; - setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); - - conn->sockfd = sockfd; - conn->state = TCP_STATE_CONNECTING; - strcpy(conn->remote_ip, cmd->params.connect.host); - conn->remote_port = cmd->params.connect.port; - - /* 解析地址并连接 */ - struct sockaddr_in server_addr; - memset(&server_addr, 0, sizeof(server_addr)); - server_addr.sin_family = AF_INET; - server_addr.sin_port = htons(cmd->params.connect.port); - - /* 尝试解析IP地址 */ - if (inet_aton(cmd->params.connect.host, &server_addr.sin_addr) == 0) { - /* 不是IP地址，尝试域名解析 */ - struct hostent *he = gethostbyname(cmd->params.connect.host); - if (he && he->h_addr_list[0]) { - memcpy(&server_addr.sin_addr, he->h_addr_list[0], he->h_length); - } else { - /* 解析失败 */ - conn->state = TCP_STATE_ERROR; - if (conn->connect_cb) { - conn->connect_cb(conn->handle, false); - } - tcp_manager_cleanup_connection(conn); - tcp_manager_update_stats_locked(stats_inc_errors); - sys_mutex_unlock(&g_connections_mutex); - goto cleanup; - } - } - - /* 非阻塞连接 */ - int result = connect(sockfd, (struct sockaddr*)&server_addr, sizeof(server_addr)); - if (result == 0) { - /* 立即连接成功 */ - conn->state = TCP_STATE_CONNECTED; - conn->last_activity = sys_now(); - if (conn->connect_cb) { - conn->connect_cb(conn->handle, true); - } - } else if (errno == EINPROGRESS) { - /* 连接正在进行中，等待select检测 */ - /* 状态已经设置为TCP_STATE_CONNECTING */ - } else { - /* 连接失败 */ - conn->state = TCP_STATE_ERROR; - if (conn->connect_cb) { - conn->connect_cb(conn->handle, false); - } - tcp_manager_cleanup_connection(conn); - tcp_manager_update_stats_locked(stats_inc_errors); - } - - sys_mutex_unlock(&g_connections_mutex); - -cleanup: - /* 释放命令资源 */ - free(cmd->params.connect.host); - free(cmd); -} - -/** - * @brief 处理socket读事件 - */ -static void tcp_manager_handle_socket_read(tcp_connection_info_t* conn) -{ - if (!conn || conn->sockfd < 0) return; - - uint8_t buffer[TCP_MANAGER_RECV_BUFFER_SIZE]; - ssize_t bytes_read = recv(conn->sockfd, buffer, sizeof(buffer), MSG_DONTWAIT); - - if (bytes_read > 0) { - /* 分配接收数据结构 */ - tcp_recv_data_t* recv_data = (tcp_recv_data_t*)malloc(sizeof(tcp_recv_data_t)); - if (!recv_data) { - tcp_manager_update_stats_locked(stats_inc_errors); - return; - } - - recv_data->handle = conn->handle; - recv_data->length = (uint16_t)bytes_read; - recv_data->timestamp = sys_now(); - memcpy(recv_data->data, buffer, bytes_read); - - /* 更新统计 */ - g_temp_bytes_count = recv_data->length; - tcp_manager_update_stats_locked(stats_inc_bytes_received); - - /* 投递到接收队列 */ - if (sys_mbox_trypost(&g_recv_queue, recv_data) != ERR_OK) { - free(recv_data); - tcp_manager_update_stats_locked(stats_inc_errors); - } - - conn->last_activity = sys_now(); - } else if (bytes_read == 0) { - /* 连接被对端关闭 */ - conn->state = TCP_STATE_DISCONNECTED; - if (conn->disconnect_cb) { - conn->disconnect_cb(conn->handle); - } - tcp_manager_cleanup_connection(conn); - } else if (errno != EAGAIN && errno != EWOULDBLOCK) { - /* 读取错误 */ - tcp_manager_handle_socket_error(conn); - } -} - -/** - * @brief 处理socket错误事件 - */ -static void tcp_manager_handle_socket_error(tcp_connection_info_t* conn) -{ - if (!conn) return; - - conn->state = TCP_STATE_ERROR; - if (conn->disconnect_cb) { - conn->disconnect_cb(conn->handle); - } - - tcp_manager_cleanup_connection(conn); - tcp_manager_update_stats_locked(stats_inc_errors); -} - -/** - * @brief 检查连接健康状态 - */ -static void tcp_manager_check_connection_health(void) -{ - uint32_t current_time = sys_now(); - - sys_mutex_lock(&g_connections_mutex); - for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) { - if (g_connections[i].in_use) { - /* 检查连接超时 (30秒无活动) */ - if (current_time - g_connections[i].last_activity > 30000) { - if (g_connections[i].state == TCP_STATE_CONNECTED) { - /* 发送心跳包或标记为超时 */ - g_connections[i].state = TCP_STATE_ERROR; - if (g_connections[i].disconnect_cb) { - g_connections[i].disconnect_cb(g_connections[i].handle); - } - tcp_manager_cleanup_connection(&g_connections[i]); - } - } - } - } - sys_mutex_unlock(&g_connections_mutex); -} - -/** - * @brief 内部关闭连接实现 - */ -static tcp_manager_result_t tcp_manager_close_internal(tcp_connection_info_t* conn) -{ - if (!conn) return TCP_MANAGER_ERROR; - - conn->state = TCP_STATE_CLOSING; - - /* 如果有断开回调，先调用 */ - if (conn->disconnect_cb) { - conn->disconnect_cb(conn->handle); - } - - /* 清理连接资源 */ - tcp_manager_cleanup_connection(conn); - - return TCP_MANAGER_OK; -} \ No newline at end of file -- 2.17.1