bool connected;
uint32_t start_time;
uint32_t last_activity;
+ sys_thread_t worker_thread;
+ bool thread_running;
+
+ /* 连接重试相关 */
+ uint8_t retry_count;
+ uint32_t last_retry_time;
+ bool retry_pending;
} tcp_stress_connection_t;
typedef struct {
uint32_t successful_connections;
uint32_t failed_connections;
uint32_t data_verification_errors;
+ uint32_t retry_attempts;
/* 连接池 */
tcp_stress_connection_t connections[TCP_MANAGER_MAX_CONNECTIONS];
conn->connected = true;
conn->start_time = sys_now();
conn->last_activity = conn->start_time;
+ conn->retry_count = 0;
+ conn->retry_pending = false;
g_stress_test.successful_connections++;
printf("[连接 %03d] 连接成功\n", conn->conn_id);
} else {
conn->connected = false;
g_stress_test.failed_connections++;
printf("[连接 %03d] 连接失败\n", conn->conn_id);
+
+ /* 尝试重连 */
+ if (conn->retry_count < 3) {
+ conn->retry_count++;
+ conn->retry_pending = true;
+ conn->last_retry_time = sys_now();
+ g_stress_test.retry_attempts++;
+ printf("[连接 %03d] 计划重连 #%d\n", conn->conn_id, conn->retry_count);
+ } else {
+ printf("[连接 %03d] 重连次数用完，停止连接\n", conn->conn_id);
+ conn->thread_running = false;
+ }
}
}
printf("活跃连接: %u/%u\n", active_connections, g_stress_test.connection_count);
printf("成功连接: %u\n", g_stress_test.successful_connections);
printf("失败连接: %u\n", g_stress_test.failed_connections);
+ printf("重连尝试: %u\n", g_stress_test.retry_attempts);
printf("发送包数: %u\n", g_stress_test.total_packets_sent);
printf("接收包数: %u\n", g_stress_test.total_packets_received);
printf("发送数据: %s\n", sent_str);
printf("================================================\n");
}
-/* 连接测试函数 */
-static void stress_test_connection_worker(uint32_t conn_id)
+/* 连接工作线程函数 */
+static void stress_test_connection_worker_thread(void* arg)
{
+ uint32_t conn_id = (uint32_t)(uintptr_t)arg;
if (conn_id >= TCP_MANAGER_MAX_CONNECTIONS) return;
tcp_stress_connection_t *conn = &g_stress_test.connections[conn_id];
conn->conn_id = conn_id;
+ conn->thread_running = true;
+ conn->retry_count = 0;
+ conn->retry_pending = false;
/* 创建连接 */
conn->handle = tcp_open_url(g_stress_test.host, g_stress_test.port,
if (!conn->handle) {
printf("[连接 %03d] 创建连接失败\n", conn_id);
g_stress_test.failed_connections++;
+ conn->thread_running = false;
return;
}
/* 等待连接完成或超时 */
uint32_t wait_start = sys_now();
- while (!conn->connected && (sys_now() - wait_start) < 10000) { /* 10秒超时 */
+ while (!conn->connected && (sys_now() - wait_start) < 15000) { /* 15秒超时 */
sys_msleep(100);
- if (!g_stress_test.running) return;
+ if (!g_stress_test.running || !conn->thread_running) {
+ conn->thread_running = false;
+ return;
+ }
+
+ /* 检查重连 */
+ if (conn->retry_pending && (sys_now() - conn->last_retry_time) >= 2000) {
+ printf("[连接 %03d] 执行重连 #%d\n", conn_id, conn->retry_count);
+
+ /* 关闭旧连接 */
+ if (conn->handle) {
+ tcp_manager_close(conn->handle);
+ conn->handle = NULL;
+ }
+
+ /* 重新创建连接 */
+ conn->handle = tcp_open_url(g_stress_test.host, g_stress_test.port,
+ stress_test_on_connect, stress_test_on_disconnect,
+ stress_test_on_data_received);
+
+ if (conn->handle) {
+ conn->retry_pending = false;
+ wait_start = sys_now(); /* 重置等待时间 */
+ printf("[连接 %03d] 重连请求已发送\n", conn_id);
+ } else {
+ printf("[连接 %03d] 重连失败\n", conn_id);
+ conn->thread_running = false;
+ return;
+ }
+ }
}
if (!conn->connected) {
printf("[连接 %03d] 连接超时\n", conn_id);
- tcp_manager_close(conn->handle);
- conn->handle = NULL;
+ if (conn->handle) {
+ tcp_manager_close(conn->handle);
+ conn->handle = NULL;
+ }
+ conn->thread_running = false;
return;
}
uint32_t packet_counter = 0;
uint32_t last_send_time = sys_now();
- while (g_stress_test.running && conn->connected) {
+ while (g_stress_test.running && conn->connected && conn->thread_running) {
uint32_t current_time = sys_now();
/* 检查测试时间 */
break;
}
- /* 控制发送频率 - 每100ms发送一次 */
- if ((current_time - last_send_time) >= 100) {
+ /* 控制发送频率 - 每200ms发送一次，减少发送压力 */
+ if ((current_time - last_send_time) >= 200) {
/* 生成测试数据 */
uint16_t data_size = generate_test_data(test_buffer, sizeof(test_buffer),
conn_id, packet_counter);
conn->errors++;
g_stress_test.total_errors++;
printf("[连接 %03d] 发送失败: %d\n", conn_id, result);
+
+ /* 发送失败时，检查连接状态 */
+ tcp_connection_state_t state = tcp_get_connection_state(conn->handle);
+ if (state != TCP_STATE_CONNECTED) {
+ printf("[连接 %03d] 连接状态异常: %d，停止发送\n", conn_id, state);
+ break;
+ }
}
}
last_send_time = current_time;
}
- sys_msleep(10); /* 短暂延迟 */
+ sys_msleep(20); /* 短暂延迟 */
}
/* 关闭连接 */
conn->handle = NULL;
}
conn->connected = false;
+ conn->thread_running = false;
printf("[连接 %03d] 测试完成 - 发送: %u包/%u字节, 接收: %u包/%u字节, 错误: %u\n",
conn_id, conn->packets_sent, conn->bytes_sent,
printf("\n连接统计:\n");
printf(" 成功连接: %u\n", g_stress_test.successful_connections);
printf(" 失败连接: %u\n", g_stress_test.failed_connections);
+ printf(" 重连尝试: %u\n", g_stress_test.retry_attempts);
if (g_stress_test.connection_count > 0) {
printf(" 连接成功率: %u.%u%%\n",
(g_stress_test.successful_connections * 100) / g_stress_test.connection_count,
/* 初始化随机数种子 */
srand(sys_now());
- /* 启动所有连接的测试 */
+ /* 并行启动所有连接的测试线程，增加启动间隔 */
+ printf("正在启动 %u 个并发连接...\n", g_stress_test.connection_count);
for (uint32_t i = 0; i < g_stress_test.connection_count; i++) {
- stress_test_connection_worker(i);
+ /* 为每个连接创建独立的工作线程 */
+ g_stress_test.connections[i].worker_thread = sys_thread_new("stress_worker",
+ stress_test_connection_worker_thread,
+ (void*)(uintptr_t)i,
+ 8192, 10);
+ if (g_stress_test.connections[i].worker_thread == (sys_thread_t)-1) {
+ printf("[连接 %03d] 创建线程失败\n", i);
+ g_stress_test.failed_connections++;
+ } else {
+ printf("[连接 %03d] 线程已启动\n", i);
+ }
- /* 错开连接启动时间 */
- sys_msleep(200);
+ /* 增加启动间隔，避免同时创建过多连接 */
+ sys_msleep(200); /* 200ms间隔 */
}
/* 监控循环 - 每5秒显示一次统计 */
printf("\n压力测试完成，正在关闭所有连接...\n");
+ /* 等待所有工作线程完成 */
+ uint32_t active_threads = 0;
+ do {
+ active_threads = 0;
+ for (uint32_t i = 0; i < g_stress_test.connection_count; i++) {
+ if (g_stress_test.connections[i].thread_running) {
+ active_threads++;
+ }
+ }
+ if (active_threads > 0) {
+ printf("等待 %u 个线程完成...\n", active_threads);
+ sys_msleep(1000);
+ }
+ } while (active_threads > 0);
+
/* 关闭所有剩余连接 */
for (uint32_t i = 0; i < g_stress_test.connection_count; i++) {
if (g_stress_test.connections[i].handle) {
/**
- * @file tcp_simple_manager.c
- * @brief TCP简单管理器 - 完善版本，简单、安全、稳定
- * @version 1.2
- * @date 2024-08-10
+ * TCP单线程管理器 v1.3 - 最终修复版本
+ * 修复内容:
+ * 1. 增大线程栈空间：2KB -> 8KB
+ * 2. 简化DNS解析逻辑，避免递归调用
+ * 3. 增强资源清理，防止内存泄漏
+ * 4. 优化错误处理，避免死循环
+ * 5. 使用正确的枚举值
+ * TCP单线程管理器 v1.4 - 压力测试优化版本
+ * 修复内容:
+ * 1. 改进DNS解析，支持并发解析
+ * 2. 添加发送队列，避免发送阻塞
+ * 3. 优化并发连接处理，增加连接间隔
+ * 4. 改进错误处理，支持重试机制
+ * 5. 增强连接状态管理
*/
#include "tcp_simple_manager.h"
-#include "lwip/sockets.h"
-#include "lwip/sys.h"
-
-#include <string.h>
+#include <stdio.h>
#include <stdlib.h>
-#include <netinet/in.h>
-#include <arpa/inet.h>
+#include <string.h>
+#include <unistd.h>
#include <errno.h>
#include <sys/select.h>
#include <fcntl.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include "lwip/sockets.h"
+#include "lwip/netdb.h"
+#include "lwip/sys.h"
+#include "lwip/dns.h"
+#include "lwip/ip.h"
+#include "lwip/err.h"
-/* 重定义socket函数为lwIP版本 */
+/* 重定义lwIP函数，确保使用标准接口 */
#ifndef socket
#define socket lwip_socket
#define connect lwip_connect
#define close lwip_close
#define send lwip_send
#define recv lwip_recv
-
#define select lwip_select
#define fcntl lwip_fcntl
+#define getsockopt lwip_getsockopt
+#define setsockopt lwip_setsockopt
#endif
-/* 常量定义 */
+#ifndef MSG_DONTWAIT
+#define MSG_DONTWAIT 0x40
+#endif
+
+/* 优化后的常量定义 */
#define TCP_RECV_BUFFER_SIZE 1024
-#define TCP_CONNECT_TIMEOUT_MS 5000
-#define TCP_SELECT_TIMEOUT_MS 100
-#define TCP_THREAD_STACK_SIZE 2048
+#define TCP_SEND_QUEUE_SIZE 16 /* 每个连接的发送队列大小 */
+#define TCP_SEND_BUFFER_SIZE 512 /* 发送缓冲区大小 */
+#define TCP_CONNECT_TIMEOUT_MS 10000 /* 增加连接超时时间 */
+#define TCP_SELECT_TIMEOUT_MS 50 /* 减少select超时，提高响应性 */
+#define TCP_THREAD_STACK_SIZE 8192
#define TCP_THREAD_PRIORITY 10
+#define TCP_DNS_TIMEOUT_MS 8000 /* 增加DNS超时时间 */
+#define TCP_KEEPALIVE_IDLE 30
+#define TCP_KEEPALIVE_INTERVAL 5
+#define TCP_KEEPALIVE_COUNT 3
+#define TCP_CONNECTION_IDLE_TIMEOUT_MS 60000
+#define TCP_CONNECT_RETRY_COUNT 3 /* 连接重试次数 */
+#define TCP_CONNECT_RETRY_DELAY_MS 1000 /* 重试延迟 */
+
+/* 结构体定义已在头文件中 */
+
+/* 改进的DNS解析状态 */
+typedef struct {
+ bool in_use;
+ bool resolved;
+ bool failed;
+ ip_addr_t resolved_ip;
+ sys_sem_t dns_sem;
+ uint32_t start_time;
+ char hostname[64]; /* 保存主机名 */
+} dns_resolve_state_t;
/* 内部状态 */
static bool g_manager_initialized = false;
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 dns_resolve_state_t g_dns_states[TCP_MANAGER_MAX_CONNECTIONS * 2]; /* 增加DNS状态槽位 */
/* IO线程相关 */
static bool g_io_thread_running = false;
static tcp_connection_info_t* find_free_connection_unsafe(void);
static tcp_connection_info_t* find_connection_by_handle_unsafe(tcp_handle_t handle);
static void cleanup_connection_unsafe(tcp_connection_info_t* conn);
-static int resolve_host(const char* host, struct in_addr* addr);
static int set_socket_nonblocking(int sockfd);
-static bool is_valid_ip_address(const char* ip);
+static int setup_keepalive(int sockfd);
+static void tcp_dns_found_callback(const char *name, const ip_addr_t *ipaddr, void *callback_arg);
+static int sync_dns_resolve(const char* hostname, struct in_addr* addr);
+static bool add_to_send_queue(tcp_connection_info_t* conn, const uint8_t* data, uint16_t length);
+static bool process_send_queue(tcp_connection_info_t* conn);
+static void reset_connection_retry(tcp_connection_info_t* conn);
/**
- * @brief 初始化TCP管理器
+ * @brief 改进的DNS解析回调函数
*/
-tcp_manager_result_t tcp_simple_manager_init(void)
+static void tcp_dns_found_callback(const char *name, const ip_addr_t *ipaddr, void *callback_arg)
{
- if (g_manager_initialized) {
- return TCP_MANAGER_OK;
+ (void)name;
+ dns_resolve_state_t* dns_state = (dns_resolve_state_t*)callback_arg;
+
+ if (!dns_state || !dns_state->in_use) return;
+
+ if (ipaddr != NULL) {
+ dns_state->resolved = true;
+ dns_state->failed = false;
+ dns_state->resolved_ip = *ipaddr;
+ printf("DNS解析成功: %s -> %s\n", dns_state->hostname,
+ inet_ntoa(*(struct in_addr*)&dns_state->resolved_ip.u_addr.ip4));
+ } else {
+ dns_state->resolved = false;
+ dns_state->failed = true;
+ printf("DNS解析失败: %s\n", dns_state->hostname);
}
+
+ /* 释放等待的信号量 */
+ sys_sem_signal(&dns_state->dns_sem);
+}
- /* 初始化连接池 */
- 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);
- g_connections[i].sockfd = -1;
- g_connections[i].in_use = false;
- g_connections[i].state = TCP_STATE_DISCONNECTED;
+/**
+ * @brief 改进的同步DNS解析函数
+ */
+static int sync_dns_resolve(const char* hostname, struct in_addr* addr)
+{
+ if (!hostname || !addr) return -1;
+
+ /* 首先检查是否为IP地址 */
+ if (inet_aton(hostname, addr) != 0) {
+ printf("直接使用IP地址: %s\n", hostname);
+ return 0;
}
-
- /* 初始化统计信息 */
- memset(&g_stats, 0, sizeof(g_stats));
-
- /* 创建互斥锁 */
- if (sys_mutex_new(&g_connections_mutex) != ERR_OK) {
- return TCP_MANAGER_ERROR;
+
+ /* 查找空闲的DNS状态槽位 */
+ int dns_index = -1;
+ for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS * 2; i++) {
+ if (!g_dns_states[i].in_use) {
+ dns_index = i;
+ break;
+ }
}
-
- /* 启动IO线程 */
- g_io_thread_running = true;
- g_io_thread = sys_thread_new("tcp_io", tcp_manager_io_thread, NULL,
- TCP_THREAD_STACK_SIZE, TCP_THREAD_PRIORITY);
- if (g_io_thread == (sys_thread_t)-1) {
- sys_mutex_free(&g_connections_mutex);
- g_io_thread_running = false;
- return TCP_MANAGER_ERROR;
+
+ if (dns_index < 0) {
+ printf("DNS解析错误: 没有可用的DNS状态槽位\n");
+ return -1;
}
-
- g_manager_initialized = true;
- return TCP_MANAGER_OK;
+
+ dns_resolve_state_t* dns_state = &g_dns_states[dns_index];
+
+ /* 初始化DNS状态 */
+ memset(dns_state, 0, sizeof(dns_resolve_state_t));
+ dns_state->in_use = true;
+ dns_state->resolved = false;
+ dns_state->failed = false;
+ dns_state->start_time = sys_now();
+ strncpy(dns_state->hostname, hostname, sizeof(dns_state->hostname) - 1);
+
+ /* 创建信号量 */
+ if (sys_sem_new(&dns_state->dns_sem, 0) != ERR_OK) {
+ dns_state->in_use = false;
+ return -1;
+ }
+
+ printf("开始DNS解析: %s\n", hostname);
+
+ /* 启动DNS查询 */
+ err_t err = dns_gethostbyname(hostname, &dns_state->resolved_ip,
+ tcp_dns_found_callback, dns_state);
+
+ int result = -1;
+
+ if (err == ERR_OK) {
+ /* 立即解析成功 (缓存命中) */
+ memcpy(&addr->s_addr, &dns_state->resolved_ip.u_addr.ip4, 4);
+ result = 0;
+ printf("DNS缓存命中: %s\n", hostname);
+ } else if (err == ERR_INPROGRESS) {
+ /* DNS查询进行中，等待完成 */
+ if (sys_arch_sem_wait(&dns_state->dns_sem, TCP_DNS_TIMEOUT_MS) != SYS_ARCH_TIMEOUT) {
+ if (dns_state->resolved && !dns_state->failed) {
+ /* DNS解析成功 */
+ memcpy(&addr->s_addr, &dns_state->resolved_ip.u_addr.ip4, 4);
+ result = 0;
+ }
+ } else {
+ printf("DNS解析超时: %s\n", hostname);
+ }
+ }
+
+ /* 清理DNS状态 */
+ sys_sem_free(&dns_state->dns_sem);
+ dns_state->in_use = false;
+
+ return result;
}
/**
- * @brief 反初始化TCP管理器
+ * @brief 设置socket为非阻塞模式
*/
-tcp_manager_result_t tcp_simple_manager_deinit(void)
+static int set_socket_nonblocking(int sockfd)
{
- if (!g_manager_initialized) {
- return TCP_MANAGER_NOT_INITIALIZED;
- }
-
- /* 停止IO线程 */
- g_io_thread_running = false;
- sys_msleep(200); /* 等待线程安全退出 */
+ int flags = fcntl(sockfd, F_GETFL, 0);
+ if (flags < 0) return -1;
+
+ return fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
+}
- /* 关闭所有连接 */
- sys_mutex_lock(&g_connections_mutex);
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use) {
- cleanup_connection_unsafe(&g_connections[i]);
- }
+/**
+ * @brief 设置socket的Keep-Alive选项
+ */
+static int setup_keepalive(int sockfd)
+{
+ int keepalive = 1;
+ int idle = TCP_KEEPALIVE_IDLE;
+ int interval = TCP_KEEPALIVE_INTERVAL;
+ int count = TCP_KEEPALIVE_COUNT;
+
+ /* 启用Keep-Alive */
+ if (setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, &keepalive, sizeof(keepalive)) < 0) {
+ return -1;
}
- sys_mutex_unlock(&g_connections_mutex);
-
- /* 清理资源 */
- sys_mutex_free(&g_connections_mutex);
- g_manager_initialized = false;
- return TCP_MANAGER_OK;
+
+ /* 设置Keep-Alive参数 */
+ setsockopt(sockfd, IPPROTO_TCP, TCP_KEEPIDLE, &idle, sizeof(idle));
+ setsockopt(sockfd, IPPROTO_TCP, TCP_KEEPINTVL, &interval, sizeof(interval));
+ setsockopt(sockfd, IPPROTO_TCP, TCP_KEEPCNT, &count, sizeof(count));
+
+ return 0;
}
/**
- * @brief 查找空闲连接 (非线程安全)
+ * @brief 查找空闲连接槽位
*/
static tcp_connection_info_t* find_free_connection_unsafe(void)
{
}
/**
- * @brief 根据句柄查找连接 (非线程安全)
+ * @brief 根据句柄查找连接
*/
static tcp_connection_info_t* find_connection_by_handle_unsafe(tcp_handle_t handle)
{
- if (!handle) return NULL;
-
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];
}
/**
- * @brief 清理连接资源 (非线程安全)
+ * @brief 清理连接资源
*/
static void cleanup_connection_unsafe(tcp_connection_info_t* conn)
{
- if (!conn) return;
+ if (!conn || !conn->in_use) return;
+ /* 关闭socket */
if (conn->sockfd >= 0) {
close(conn->sockfd);
- conn->sockfd = -1;
}
+ /* 清理连接状态 */
+ memset(conn, 0, sizeof(tcp_connection_info_t));
+ conn->sockfd = -1;
conn->in_use = false;
- conn->state = TCP_STATE_DISCONNECTED;
- conn->connect_cb = NULL;
- conn->disconnect_cb = NULL;
- conn->data_cb = NULL;
- conn->last_activity = 0;
- memset(conn->remote_ip, 0, sizeof(conn->remote_ip));
- conn->remote_port = 0;
}
/**
- * @brief 解析IP地址 (仅支持IP地址，不支持域名)
+ * @brief 重置连接重试状态
*/
-static int resolve_host(const char* host, struct in_addr* addr)
+static void reset_connection_retry(tcp_connection_info_t* conn)
{
- if (!host || !addr) return -1;
-
- /* 只支持IP地址解析 */
- if (inet_aton(host, addr) != 0) {
- return 0; /* 成功 */
- }
-
- return -1; /* 失败 - 不支持域名解析 */
+ if (!conn) return;
+ conn->retry_count = 0;
+ conn->retry_pending = false;
+ conn->last_retry_time = 0;
}
/**
- * @brief 设置socket为非阻塞模式
+ * @brief 添加数据到发送队列
*/
-static int set_socket_nonblocking(int sockfd)
+static bool add_to_send_queue(tcp_connection_info_t* conn, const uint8_t* data, uint16_t length)
{
- int flags = fcntl(sockfd, F_GETFL, 0);
- if (flags < 0) return -1;
+ if (!conn || !data || length == 0 || length > TCP_SEND_BUFFER_SIZE) {
+ return false;
+ }
- return fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
+ if (conn->send_queue_count >= TCP_SEND_QUEUE_SIZE) {
+ return false; /* 队列满 */
+ }
+
+ uint8_t index = conn->send_queue_tail;
+ memcpy(conn->send_queue[index].data, data, length);
+ conn->send_queue[index].length = length;
+ conn->send_queue[index].timestamp = sys_now();
+ conn->send_queue[index].in_use = true;
+
+ conn->send_queue_tail = (conn->send_queue_tail + 1) % TCP_SEND_QUEUE_SIZE;
+ conn->send_queue_count++;
+
+ return true;
}
/**
- * @brief 检查是否为有效IP地址
+ * @brief 处理发送队列
*/
-static bool is_valid_ip_address(const char* ip)
+static bool process_send_queue(tcp_connection_info_t* conn)
{
- if (!ip) return false;
+ if (!conn || conn->send_queue_count == 0) {
+ return true;
+ }
+
+ bool all_sent = true;
- struct sockaddr_in sa;
- return inet_pton(AF_INET, ip, &(sa.sin_addr)) != 0;
+ for (int i = 0; i < conn->send_queue_count; i++) {
+ uint8_t index = (conn->send_queue_head + i) % TCP_SEND_QUEUE_SIZE;
+ tcp_send_queue_item_t* item = &conn->send_queue[index];
+
+ if (!item->in_use) continue;
+
+ /* 尝试发送数据 */
+ ssize_t sent = send(conn->sockfd, item->data, item->length, MSG_NOSIGNAL);
+
+ if (sent == item->length) {
+ /* 发送成功 */
+ g_stats.bytes_sent += sent;
+ conn->last_activity = sys_now();
+ item->in_use = false;
+
+ /* 移动队列头 */
+ conn->send_queue_head = (conn->send_queue_head + 1) % TCP_SEND_QUEUE_SIZE;
+ conn->send_queue_count--;
+ i--; /* 重新检查当前位置 */
+
+ } else if (sent < 0) {
+ if (errno == EAGAIN || errno == EWOULDBLOCK) {
+ /* 发送缓冲区满，稍后重试 */
+ all_sent = false;
+ break;
+ } else {
+ /* 发送错误，标记为失败 */
+ printf("发送错误: %d, 连接可能已断开\n", errno);
+ item->in_use = false;
+ conn->send_queue_count--;
+ i--;
+ all_sent = false;
+ }
+ } else if (sent < item->length) {
+ /* 部分发送，更新剩余数据 */
+ memmove(item->data, item->data + sent, item->length - sent);
+ item->length -= sent;
+ g_stats.bytes_sent += sent;
+ conn->last_activity = sys_now();
+ all_sent = false;
+ }
+ }
+
+ return all_sent;
}
/**
- * @brief IO处理线程 - 使用select机制
+ * @brief 改进的IO处理线程
*/
static void tcp_manager_io_thread(void* arg)
{
(void)arg;
- uint8_t buffer[TCP_RECV_BUFFER_SIZE];
+
+ /* 减少栈上的大变量 */
+ static uint8_t recv_buffer[TCP_RECV_BUFFER_SIZE];
fd_set read_fds, write_fds, error_fds;
struct timeval timeout;
FD_ZERO(&error_fds);
sys_mutex_lock(&g_connections_mutex);
+
+ /* 扫描所有连接 */
for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
tcp_connection_info_t* conn = &g_connections[i];
- if (conn->in_use && conn->sockfd >= 0) {
- FD_SET(conn->sockfd, &read_fds);
- FD_SET(conn->sockfd, &error_fds);
-
- /* 连接中的socket需要监听写事件 */
- if (conn->state == TCP_STATE_CONNECTING) {
- FD_SET(conn->sockfd, &write_fds);
- }
-
- if (conn->sockfd > max_fd) {
- max_fd = conn->sockfd;
- }
- has_sockets = true;
+ if (!conn->in_use || conn->sockfd < 0) continue;
+
+ FD_SET(conn->sockfd, &read_fds);
+ FD_SET(conn->sockfd, &error_fds);
+
+ if (conn->state == TCP_STATE_CONNECTING) {
+ FD_SET(conn->sockfd, &write_fds);
}
+
+ /* 如果有待发送数据，也监听写事件 */
+ if (conn->send_queue_count > 0) {
+ FD_SET(conn->sockfd, &write_fds);
+ }
+
+ if (conn->sockfd > max_fd) {
+ max_fd = conn->sockfd;
+ }
+ has_sockets = true;
}
+
sys_mutex_unlock(&g_connections_mutex);
if (!has_sockets) {
if (select_result > 0) {
sys_mutex_lock(&g_connections_mutex);
+ /* 处理就绪的socket */
for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
tcp_connection_info_t* conn = &g_connections[i];
if (!conn->in_use || conn->sockfd < 0) continue;
int error = 0;
socklen_t len = sizeof(error);
if (getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &error, &len) == 0 && error == 0) {
+ /* 连接成功 */
conn->state = TCP_STATE_CONNECTED;
conn->last_activity = sys_now();
g_stats.total_connections++;
g_stats.active_connections++;
+ /* 重置重试状态 */
+ reset_connection_retry(conn);
+
+ /* 设置Keep-Alive */
+ setup_keepalive(sockfd);
+
+ printf("连接成功: %s:%d\n", conn->remote_ip, conn->remote_port);
+
if (conn->connect_cb) {
conn->connect_cb(conn->handle, true);
}
} else {
/* 连接失败 */
- if (conn->connect_cb) {
- conn->connect_cb(conn->handle, false);
+ printf("连接失败: %s:%d, 错误: %d\n", conn->remote_ip, conn->remote_port, error);
+
+ if (conn->retry_count < TCP_CONNECT_RETRY_COUNT) {
+ /* 尝试重连 */
+ conn->retry_count++;
+ conn->retry_pending = true;
+ conn->last_retry_time = sys_now();
+ printf("计划重连 #%d: %s:%d\n", conn->retry_count, conn->remote_ip, conn->remote_port);
+ } else {
+ /* 重试次数用完，关闭连接 */
+ if (conn->connect_cb) {
+ conn->connect_cb(conn->handle, false);
+ }
+ cleanup_connection_unsafe(conn);
+ g_stats.errors++;
}
- cleanup_connection_unsafe(conn);
- g_stats.errors++;
}
}
/* 处理数据接收 */
if (FD_ISSET(sockfd, &read_fds)) {
- ssize_t bytes_received = recv(sockfd, buffer, sizeof(buffer), 0);
+ ssize_t bytes_received = recv(sockfd, recv_buffer, sizeof(recv_buffer), 0);
if (bytes_received > 0) {
conn->last_activity = sys_now();
g_stats.bytes_received += bytes_received;
- /* 更新接收统计 */
if (conn->data_cb) {
- conn->data_cb(conn->handle, buffer, (uint16_t)bytes_received);
+ conn->data_cb(conn->handle, recv_buffer, (uint16_t)bytes_received);
}
- } else if (bytes_received == 0) {
- /* 连接被对端关闭 */
+ } else if (bytes_received == 0 || (bytes_received < 0 && errno != EAGAIN && errno != EWOULDBLOCK)) {
+ /* 连接关闭或出错 */
+ printf("连接断开: %s:%d\n", conn->remote_ip, conn->remote_port);
if (conn->disconnect_cb) {
conn->disconnect_cb(conn->handle);
}
if (g_stats.active_connections > 0) {
g_stats.active_connections--;
}
- } else {
- /* 接收错误 */
- if (errno != EAGAIN && errno != EWOULDBLOCK) {
- if (conn->disconnect_cb) {
- conn->disconnect_cb(conn->handle);
- }
- cleanup_connection_unsafe(conn);
- if (g_stats.active_connections > 0) {
- g_stats.active_connections--;
- }
+ if (bytes_received < 0) {
g_stats.errors++;
}
}
}
+ /* 处理数据发送 */
+ if (FD_ISSET(sockfd, &write_fds) && conn->send_queue_count > 0) {
+ process_send_queue(conn);
+ }
+
/* 处理错误事件 */
if (FD_ISSET(sockfd, &error_fds)) {
+ printf("连接错误: %s:%d\n", conn->remote_ip, conn->remote_port);
if (conn->disconnect_cb) {
conn->disconnect_cb(conn->handle);
}
sys_mutex_unlock(&g_connections_mutex);
} else if (select_result < 0 && errno != EINTR) {
- /* select错误 */
- sys_msleep(100);
- g_stats.errors++;
+ /* select出错 */
+ printf("TCP IO: select error: %d\n", errno);
+ sys_msleep(TCP_SELECT_TIMEOUT_MS);
}
- /* 检查连接健康状态 */
+ /* 检查连接健康状态和重连 */
uint32_t current_time = sys_now();
sys_mutex_lock(&g_connections_mutex);
for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
tcp_connection_info_t* conn = &g_connections[i];
- if (conn->in_use && conn->state == TCP_STATE_CONNECTED) {
- /* 检查连接超时 (30秒无活动) */
- if (current_time - conn->last_activity > 30000) {
+ if (!conn->in_use) continue;
+
+ if (conn->state == TCP_STATE_CONNECTED) {
+ /* 检查空闲超时 */
+ if (current_time - conn->last_activity > TCP_CONNECTION_IDLE_TIMEOUT_MS) {
+ printf("连接空闲超时: %s:%d\n", conn->remote_ip, conn->remote_port);
if (conn->disconnect_cb) {
conn->disconnect_cb(conn->handle);
}
g_stats.active_connections--;
}
}
+ } else if (conn->state == TCP_STATE_CONNECTING && conn->retry_pending) {
+ /* 检查重连时机 */
+ if (current_time - conn->last_retry_time >= TCP_CONNECT_RETRY_DELAY_MS) {
+ printf("执行重连 #%d: %s:%d\n", conn->retry_count, conn->remote_ip, conn->remote_port);
+
+ /* 关闭旧socket */
+ if (conn->sockfd >= 0) {
+ close(conn->sockfd);
+ conn->sockfd = -1;
+ }
+
+ /* 重新创建socket */
+ int sockfd = socket(AF_INET, SOCK_STREAM, 0);
+ if (sockfd >= 0) {
+ if (set_socket_nonblocking(sockfd) == 0) {
+ conn->sockfd = sockfd;
+ conn->retry_pending = false;
+
+ /* 重新发起连接 */
+ struct sockaddr_in server_addr;
+ memset(&server_addr, 0, sizeof(server_addr));
+ server_addr.sin_family = AF_INET;
+ server_addr.sin_port = htons(conn->remote_port);
+
+ /* 重新解析IP地址 */
+ struct in_addr addr;
+ if (sync_dns_resolve(conn->remote_ip, &addr) >= 0) {
+ server_addr.sin_addr = addr;
+ connect(sockfd, (struct sockaddr*)&server_addr, sizeof(server_addr));
+ } else {
+ printf("重连DNS解析失败: %s\n", conn->remote_ip);
+ cleanup_connection_unsafe(conn);
+ g_stats.errors++;
+ }
+ } else {
+ close(sockfd);
+ cleanup_connection_unsafe(conn);
+ g_stats.errors++;
+ }
+ } else {
+ cleanup_connection_unsafe(conn);
+ g_stats.errors++;
+ }
+ }
}
}
sys_mutex_unlock(&g_connections_mutex);
}
}
+/**
+ * @brief 初始化TCP管理器
+ */
+tcp_manager_result_t tcp_simple_manager_init(void)
+{
+ if (g_manager_initialized) {
+ return TCP_MANAGER_OK;
+ }
+
+ printf("TCP管理器初始化开始...\n");
+
+ /* 初始化连接数组 */
+ memset(g_connections, 0, sizeof(g_connections));
+ for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
+ g_connections[i].sockfd = -1;
+ g_connections[i].send_queue_head = 0;
+ g_connections[i].send_queue_tail = 0;
+ g_connections[i].send_queue_count = 0;
+ g_connections[i].retry_count = 0;
+ g_connections[i].retry_pending = false;
+ }
+
+ /* 初始化DNS状态数组 */
+ memset(g_dns_states, 0, sizeof(g_dns_states));
+
+ /* 初始化统计信息 */
+ memset(&g_stats, 0, sizeof(g_stats));
+
+ /* 创建互斥锁 */
+ if (sys_mutex_new(&g_connections_mutex) != ERR_OK) {
+ printf("TCP管理器初始化失败: 无法创建互斥锁\n");
+ return TCP_MANAGER_THREAD_ERROR;
+ }
+
+ /* 创建IO线程 */
+ g_io_thread_running = true;
+ g_io_thread = sys_thread_new("tcp_io", tcp_manager_io_thread, NULL,
+ TCP_THREAD_STACK_SIZE, TCP_THREAD_PRIORITY);
+
+ if (g_io_thread == (sys_thread_t)-1) {
+ printf("TCP管理器初始化失败: 无法创建IO线程\n");
+ sys_mutex_free(&g_connections_mutex);
+ return TCP_MANAGER_THREAD_ERROR;
+ }
+
+ g_manager_initialized = true;
+ printf("TCP管理器初始化成功\n");
+
+ return TCP_MANAGER_OK;
+}
+
+/**
+ * @brief 反初始化TCP管理器
+ */
+tcp_manager_result_t tcp_simple_manager_deinit(void)
+{
+ if (!g_manager_initialized) {
+ return TCP_MANAGER_OK;
+ }
+
+ printf("TCP管理器反初始化开始...\n");
+
+ /* 停止IO线程 */
+ g_io_thread_running = false;
+
+ /* 等待线程结束 */
+ if (g_io_thread != (sys_thread_t)-1) {
+ sys_msleep(1000); /* 等待1秒 */
+ }
+
+ /* 关闭所有连接 */
+ sys_mutex_lock(&g_connections_mutex);
+ for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
+ if (g_connections[i].in_use) {
+ cleanup_connection_unsafe(&g_connections[i]);
+ }
+ }
+ sys_mutex_unlock(&g_connections_mutex);
+
+ /* 清理DNS状态 */
+ for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS * 2; i++) {
+ if (g_dns_states[i].in_use) {
+ sys_sem_free(&g_dns_states[i].dns_sem);
+ }
+ }
+
+ /* 释放互斥锁 */
+ sys_mutex_free(&g_connections_mutex);
+
+ g_manager_initialized = false;
+ printf("TCP管理器反初始化完成\n");
+
+ return TCP_MANAGER_OK;
+}
+
/**
* @brief 连接到指定地址
*/
tcp_on_disconnect_cb disconnect_cb,
tcp_on_data_received_cb data_cb)
{
- if (!g_manager_initialized || !host || port == 0) {
- return NULL;
- }
-
- /* 参数验证 */
- if (strlen(host) >= sizeof(((tcp_connection_info_t*)0)->remote_ip)) {
+ if (!g_manager_initialized || !host) {
return NULL;
}
-
+
+ printf("尝试连接到: %s:%d\n", host, port);
+
sys_mutex_lock(&g_connections_mutex);
- /* 查找空闲连接 */
tcp_connection_info_t* conn = find_free_connection_unsafe();
if (!conn) {
sys_mutex_unlock(&g_connections_mutex);
+ printf("连接失败: 没有可用的连接槽位\n");
return NULL;
}
-
+
+ /* 初始化连接 */
+ memset(conn, 0, sizeof(tcp_connection_info_t));
+ conn->in_use = true;
+ conn->handle = conn;
+ conn->state = TCP_STATE_CONNECTING;
+ conn->sockfd = -1;
+ conn->remote_port = port;
+ conn->connect_cb = connect_cb;
+ conn->disconnect_cb = disconnect_cb;
+ conn->data_cb = data_cb;
+ conn->last_activity = sys_now();
+ conn->send_queue_head = 0;
+ conn->send_queue_tail = 0;
+ conn->send_queue_count = 0;
+ conn->retry_count = 0;
+ conn->retry_pending = false;
+
+ /* 复制主机名 */
+ strncpy(conn->remote_ip, host, sizeof(conn->remote_ip) - 1);
+ conn->remote_ip[sizeof(conn->remote_ip) - 1] = '\0';
+
+ sys_mutex_unlock(&g_connections_mutex);
+
+ /* DNS解析 */
+ struct in_addr addr;
+ if (sync_dns_resolve(host, &addr) < 0) {
+ printf("DNS解析失败: %s\n", host);
+ sys_mutex_lock(&g_connections_mutex);
+ cleanup_connection_unsafe(conn);
+ sys_mutex_unlock(&g_connections_mutex);
+ if (connect_cb) {
+ connect_cb(conn->handle, false);
+ }
+ return NULL;
+ }
+
/* 创建socket */
int sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0) {
+ printf("创建socket失败: %d\n", errno);
+ sys_mutex_lock(&g_connections_mutex);
+ cleanup_connection_unsafe(conn);
sys_mutex_unlock(&g_connections_mutex);
+ if (connect_cb) {
+ connect_cb(conn->handle, false);
+ }
return NULL;
}
-
- /* 设置非阻塞模式 */
+
+ /* 设置非阻塞 */
if (set_socket_nonblocking(sockfd) < 0) {
+ printf("设置非阻塞模式失败: %d\n", errno);
close(sockfd);
+ sys_mutex_lock(&g_connections_mutex);
+ cleanup_connection_unsafe(conn);
sys_mutex_unlock(&g_connections_mutex);
+ if (connect_cb) {
+ connect_cb(conn->handle, false);
+ }
return NULL;
}
-
- /* 解析地址 */
+
+ /* 更新连接信息 */
+ sys_mutex_lock(&g_connections_mutex);
+ conn->sockfd = sockfd;
+ sys_mutex_unlock(&g_connections_mutex);
+
+ /* 发起连接 */
struct sockaddr_in server_addr;
memset(&server_addr, 0, sizeof(server_addr));
server_addr.sin_family = AF_INET;
+ server_addr.sin_addr = addr;
server_addr.sin_port = htons(port);
- if (resolve_host(host, &server_addr.sin_addr) < 0) {
- close(sockfd);
- sys_mutex_unlock(&g_connections_mutex);
- return NULL;
- }
-
- /* 初始化连接信息 */
- conn->in_use = true;
- conn->sockfd = sockfd;
- conn->state = TCP_STATE_CONNECTING;
- conn->remote_port = port;
- strncpy(conn->remote_ip, host, sizeof(conn->remote_ip) - 1);
- conn->remote_ip[sizeof(conn->remote_ip) - 1] = '\0';
- 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;
-
- /* 发起连接 */
int result = connect(sockfd, (struct sockaddr*)&server_addr, sizeof(server_addr));
if (result == 0) {
/* 立即连接成功 */
+ printf("连接立即成功: %s:%d\n", host, port);
+ sys_mutex_lock(&g_connections_mutex);
conn->state = TCP_STATE_CONNECTED;
g_stats.total_connections++;
g_stats.active_connections++;
+ setup_keepalive(sockfd);
sys_mutex_unlock(&g_connections_mutex);
if (connect_cb) {
- connect_cb(handle, true);
+ connect_cb(conn->handle, true);
}
} else if (errno == EINPROGRESS) {
- /* 连接正在进行中 */
- sys_mutex_unlock(&g_connections_mutex);
+ /* 连接进行中，将由IO线程处理 */
+ printf("连接进行中: %s:%d\n", host, port);
} else {
/* 连接失败 */
+ printf("连接失败: %s:%d, 错误: %d\n", host, port, errno);
+ sys_mutex_lock(&g_connections_mutex);
cleanup_connection_unsafe(conn);
sys_mutex_unlock(&g_connections_mutex);
- g_stats.errors++;
-
if (connect_cb) {
- connect_cb(handle, false);
+ connect_cb(conn->handle, false);
}
return NULL;
}
- return handle;
+ return conn->handle;
}
/**
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;
+ return TCP_MANAGER_INVALID_HANDLE;
}
-
+
sys_mutex_lock(&g_connections_mutex);
+
tcp_connection_info_t* conn = find_connection_by_handle_unsafe(handle);
- if (!conn || conn->sockfd < 0 || conn->state != TCP_STATE_CONNECTED) {
+ if (!conn || conn->state != TCP_STATE_CONNECTED) {
sys_mutex_unlock(&g_connections_mutex);
- return TCP_MANAGER_INVALID_HANDLE;
+ return TCP_MANAGER_CONNECTION_FAILED;
}
-
- ssize_t sent = send(conn->sockfd, data, datalen, MSG_NOSIGNAL);
- if (sent > 0) {
- conn->last_activity = sys_now();
- g_stats.bytes_sent += sent;
- /* 更新发送统计 */
- sys_mutex_unlock(&g_connections_mutex);
- return TCP_MANAGER_OK;
- } else {
+
+ if (!add_to_send_queue(conn, data, datalen)) {
sys_mutex_unlock(&g_connections_mutex);
- g_stats.errors++;
- return TCP_MANAGER_ERROR;
+ return TCP_MANAGER_QUEUE_FULL;
}
+
+ sys_mutex_unlock(&g_connections_mutex);
+
+ return TCP_MANAGER_OK;
}
/**
- * @brief 直接发送数据 (线程安全)
+ * @brief 直接发送数据
*/
tcp_manager_result_t tcp_send_data_directly(tcp_handle_t handle, uint8_t* data, uint16_t datalen)
{
tcp_manager_result_t tcp_manager_close(tcp_handle_t handle)
{
if (!g_manager_initialized || !handle) {
- return TCP_MANAGER_ERROR;
+ return TCP_MANAGER_INVALID_HANDLE;
}
-
+
sys_mutex_lock(&g_connections_mutex);
+
tcp_connection_info_t* conn = find_connection_by_handle_unsafe(handle);
- if (conn) {
- cleanup_connection_unsafe(conn);
- if (g_stats.active_connections > 0) {
- g_stats.active_connections--;
- }
+ if (!conn) {
+ sys_mutex_unlock(&g_connections_mutex);
+ return TCP_MANAGER_CONNECTION_FAILED;
+ }
+
+ printf("关闭连接: %s:%d\n", conn->remote_ip, conn->remote_port);
+ cleanup_connection_unsafe(conn);
+ if (g_stats.active_connections > 0) {
+ g_stats.active_connections--;
}
+
sys_mutex_unlock(&g_connections_mutex);
-
+
return TCP_MANAGER_OK;
}
/**
- * @brief 直接关闭连接 (线程安全)
+ * @brief 直接关闭TCP连接
*/
tcp_manager_result_t tcp_manager_close_directly(tcp_handle_t handle)
{
tcp_connection_state_t tcp_get_connection_state(tcp_handle_t handle)
{
if (!g_manager_initialized || !handle) {
- return TCP_STATE_ERROR;
+ return TCP_STATE_DISCONNECTED;
}
-
+
sys_mutex_lock(&g_connections_mutex);
+
tcp_connection_info_t* conn = find_connection_by_handle_unsafe(handle);
- tcp_connection_state_t state = conn ? conn->state : TCP_STATE_ERROR;
+ tcp_connection_state_t state = conn ? conn->state : TCP_STATE_DISCONNECTED;
+
sys_mutex_unlock(&g_connections_mutex);
-
+
return state;
}
*/
tcp_manager_result_t tcp_get_manager_stats(tcp_manager_stats_t* stats)
{
- if (!stats) {
- return TCP_MANAGER_ERROR;
- }
-
- if (!g_manager_initialized) {
- memset(stats, 0, sizeof(tcp_manager_stats_t));
- return TCP_MANAGER_NOT_INITIALIZED;
+ if (!g_manager_initialized || !stats) {
+ return TCP_MANAGER_INVALID_HANDLE;
}
sys_mutex_lock(&g_connections_mutex);
- memcpy(stats, &g_stats, sizeof(tcp_manager_stats_t));
-
- /* 重新计算当前活跃连接数 */
- stats->active_connections = 0;
- stats->total_connections = 0;
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use) {
- stats->total_connections++;
- if (g_connections[i].state == TCP_STATE_CONNECTED) {
- stats->active_connections++;
- }
- }
- }
+ *stats = g_stats;
sys_mutex_unlock(&g_connections_mutex);
-
+
return TCP_MANAGER_OK;
}
/**
- * @brief 设置连接超时时间
+ * @brief 设置连接超时 (暂未实现)
*/
tcp_manager_result_t tcp_set_timeout(tcp_handle_t handle, uint32_t timeout_ms)
{
- if (!g_manager_initialized || !handle) {
- return TCP_MANAGER_ERROR;
- }
-
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = find_connection_by_handle_unsafe(handle);
- if (conn && conn->sockfd >= 0) {
- struct timeval tv;
- tv.tv_sec = timeout_ms / 1000;
- tv.tv_usec = (timeout_ms % 1000) * 1000;
-
- setsockopt(conn->sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
- setsockopt(conn->sockfd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
-
- sys_mutex_unlock(&g_connections_mutex);
- return TCP_MANAGER_OK;
- }
- sys_mutex_unlock(&g_connections_mutex);
-
- return TCP_MANAGER_INVALID_HANDLE;
+ (void)handle;
+ (void)timeout_ms;
+ return TCP_MANAGER_ERROR; /* 未实现 */
}
\ No newline at end of file
projects / mirrors / FD230.git / commitdiff