if (!conn->connected) {
printf("[连接 %03d] 连接超时\n", conn_id);
+ g_stress_test.failed_connections++;
+
+ /* 清理连接资源 */
if (conn->handle) {
tcp_manager_close(conn->handle);
conn->handle = NULL;
}
+ conn->connected = false;
conn->thread_running = false;
return;
}
- /* 数据传输循环 */
- uint8_t test_buffer[512];
- uint32_t packet_counter = 0;
- uint32_t last_send_time = sys_now();
+ printf("[连接 %03d] 连接成功，开始发送数据\n", conn_id);
+ g_stress_test.successful_connections++;
+
+ /* 数据发送循环 */
+ uint32_t packet_id = 0;
+ uint8_t test_buffer[TCP_SEND_BUFFER_SIZE];
while (g_stress_test.running && conn->connected && conn->thread_running) {
- uint32_t current_time = sys_now();
-
- /* 检查测试时间 */
- if ((current_time - g_stress_test.start_time) >= (g_stress_test.test_duration * 1000)) {
+ /* 检查连接状态 */
+ tcp_connection_state_t state = tcp_get_connection_state(conn->handle);
+ if (state != TCP_STATE_CONNECTED) {
+ printf("[连接 %03d] 连接状态异常: %d\n", conn_id, state);
break;
}
- /* 控制发送频率 - 每200ms发送一次，减少发送压力 */
- if ((current_time - last_send_time) >= 200) {
- /* 生成测试数据 */
- uint16_t data_size = generate_test_data(test_buffer, sizeof(test_buffer),
- conn_id, packet_counter);
- if (data_size > 0) {
- /* 发送数据 */
- tcp_manager_result_t result = tcp_send_data(conn->handle, test_buffer, data_size);
- if (result == TCP_MANAGER_OK) {
- conn->packets_sent++;
- conn->bytes_sent += data_size;
- conn->last_activity = current_time;
-
- g_stress_test.total_packets_sent++;
- g_stress_test.total_bytes_sent += data_size;
-
- printf("[连接 %03d] 发送数据包 #%d, 大小=%d字节\n",
- conn_id, packet_counter, data_size);
- } else {
- 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;
- }
- }
+ /* 生成测试数据 */
+ uint16_t data_len = generate_test_data(test_buffer, sizeof(test_buffer), conn_id, packet_id);
+
+ /* 发送数据 */
+ tcp_manager_result_t send_result = tcp_send_data(conn->handle, test_buffer, data_len);
+ if (send_result == TCP_MANAGER_OK) {
+ conn->packets_sent++;
+ conn->bytes_sent += data_len;
+ g_stress_test.total_packets_sent++;
+ g_stress_test.total_bytes_sent += data_len;
+
+ if (packet_id % 100 == 0) {
+ printf("[连接 %03d] 发送数据包 #%u, 大小=%u字节\n", conn_id, packet_id, data_len);
}
+ } else {
+ printf("[连接 %03d] 发送失败: %d\n", conn_id, send_result);
+ conn->errors++;
+ g_stress_test.total_errors++;
- packet_counter++;
- last_send_time = current_time;
+ /* 检查连接是否仍然有效 */
+ state = tcp_get_connection_state(conn->handle);
+ if (state != TCP_STATE_CONNECTED) {
+ printf("[连接 %03d] 连接已断开，停止发送\n", conn_id);
+ break;
+ }
}
- sys_msleep(20); /* 短暂延迟 */
+ packet_id++;
+ sys_msleep(200); /* 200ms间隔 */
}
- /* 关闭连接 */
+ printf("[连接 %03d] 工作线程结束，已发送 %u 个数据包\n", conn_id, conn->packets_sent);
+
+ /* 清理连接资源 */
if (conn->handle) {
+ printf("[连接 %03d] 关闭连接\n", conn_id);
tcp_manager_close(conn->handle);
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,
- conn->packets_received, conn->bytes_received, conn->errors);
}
/* 打印最终测试报告 */
printf("\n压力测试完成，正在关闭所有连接...\n");
+ /* 停止所有工作线程 */
+ g_stress_test.running = false;
+
/* 等待所有工作线程完成 */
uint32_t active_threads = 0;
+ uint32_t wait_count = 0;
+ const uint32_t max_wait = 30; /* 最多等待30秒 */
+
do {
active_threads = 0;
for (uint32_t i = 0; i < g_stress_test.connection_count; i++) {
}
}
if (active_threads > 0) {
- printf("等待 %u 个线程完成...\n", active_threads);
+ printf("等待 %u 个线程完成... (已等待 %u 秒)\n", active_threads, wait_count);
sys_msleep(1000);
+ wait_count++;
+
+ /* 强制停止超时的线程 */
+ if (wait_count >= max_wait) {
+ printf("等待超时，强制停止剩余线程\n");
+ break;
+ }
}
} while (active_threads > 0);
- /* 关闭所有剩余连接 */
+ /* 强制关闭所有剩余连接 */
+ printf("关闭所有连接...\n");
for (uint32_t i = 0; i < g_stress_test.connection_count; i++) {
if (g_stress_test.connections[i].handle) {
+ printf("关闭连接 %u: handle=%p\n", i, g_stress_test.connections[i].handle);
tcp_manager_close(g_stress_test.connections[i].handle);
+ g_stress_test.connections[i].handle = NULL;
+ g_stress_test.connections[i].connected = false;
+ g_stress_test.connections[i].thread_running = false;
}
}
- sys_msleep(1000); /* 等待关闭完成 */
+ /* 等待TCP管理器完成清理 */
+ printf("等待TCP管理器清理完成...\n");
+ sys_msleep(2000);
+
+ /* 重置连接状态 */
+ for (uint32_t i = 0; i < g_stress_test.connection_count; i++) {
+ memset(&g_stress_test.connections[i], 0, sizeof(tcp_stress_connection_t));
+ g_stress_test.connections[i].conn_id = i;
+ }
+
+ printf("所有连接已关闭\n");
+
+ /* 强制清理TCP管理器 */
+ printf("清理TCP管理器...\n");
+ tcp_simple_manager_deinit();
+
+ /* 等待一段时间确保清理完成 */
+ sys_msleep(1000);
+
+ /* 重新初始化TCP管理器 */
+ printf("重新初始化TCP管理器...\n");
+ if (tcp_simple_manager_init() != TCP_MANAGER_OK) {
+ printf("重新初始化失败\n");
+ }
/* 打印最终报告 */
print_final_stress_test_report();
}
/**
- * @brief 清理连接资源
+ * @brief 清理连接资源（线程不安全，调用者需要加锁）
*/
static void cleanup_connection_unsafe(tcp_connection_info_t* conn)
{
if (!conn || !conn->in_use) return;
+ printf("清理连接资源: handle=%p, sockfd=%d\n", conn->handle, conn->sockfd);
+
/* 关闭socket */
if (conn->sockfd >= 0) {
+ shutdown(conn->sockfd, SHUT_RDWR); /* 优雅关闭 */
close(conn->sockfd);
+ printf("Socket已关闭: %d\n", conn->sockfd);
+ }
+
+ /* 清理发送队列 */
+ if (conn->send_queue_count > 0) {
+ printf("清理发送队列: %d个待发送项\n", conn->send_queue_count);
+ for (int i = 0; i < TCP_SEND_QUEUE_SIZE; i++) {
+ conn->send_queue[i].in_use = false;
+ conn->send_queue[i].length = 0;
+ }
+ conn->send_queue_count = 0;
+ conn->send_queue_head = 0;
+ conn->send_queue_tail = 0;
+ }
+
+ /* 清理DNS状态 */
+ for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS * 2; i++) {
+ if (g_dns_states[i].in_use &&
+ strcmp(g_dns_states[i].hostname, conn->remote_ip) == 0) {
+ printf("清理DNS状态: %s\n", g_dns_states[i].hostname);
+ g_dns_states[i].in_use = false;
+ g_dns_states[i].resolved = false;
+ g_dns_states[i].failed = false;
+ if (g_dns_states[i].dns_sem) {
+ sys_sem_free(&g_dns_states[i].dns_sem);
+ g_dns_states[i].dns_sem = NULL;
+ }
+ }
}
/* 清理连接状态 */
memset(conn, 0, sizeof(tcp_connection_info_t));
conn->sockfd = -1;
conn->in_use = false;
+
+ printf("连接资源清理完成\n");
}
/**
static void tcp_manager_io_thread(void* arg)
{
(void)arg;
+ printf("TCP管理器IO线程启动\n");
- /* 减少栈上的大变量 */
- static uint8_t recv_buffer[TCP_RECV_BUFFER_SIZE];
- fd_set read_fds, write_fds, error_fds;
+ fd_set read_fds, write_fds, except_fds;
struct timeval timeout;
while (g_io_thread_running) {
- int max_fd = 0;
- bool has_sockets = false;
-
- /* 准备文件描述符集合 */
+ /* 初始化文件描述符集合 */
FD_ZERO(&read_fds);
FD_ZERO(&write_fds);
- FD_ZERO(&error_fds);
+ FD_ZERO(&except_fds);
- sys_mutex_lock(&g_connections_mutex);
+ int max_fd = -1;
+ int total_connections = 0;
- /* æ\89«æ\8f\8fæ\89\80æ\9c\89è¿\9eæ\8e¥ */
+ /* æ·»å\8a æ\89\80æ\9c\89æ´»è·\83è¿\9eæ\8e¥å\88°æ\96\87件æ\8f\8f述符é\9b\86å\90\88 */
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;
- FD_SET(conn->sockfd, &read_fds);
- FD_SET(conn->sockfd, &error_fds);
+ total_connections++;
- if (conn->state == TCP_STATE_CONNECTING) {
- FD_SET(conn->sockfd, &write_fds);
- }
+ /* 添加socket到读集合 */
+ FD_SET(conn->sockfd, &read_fds);
+ FD_SET(conn->sockfd, &except_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);
+ /* 设置超时时间 */
+ timeout.tv_sec = 0;
+ timeout.tv_usec = 100000; /* 100ms */
- if (!has_sockets) {
- sys_msleep(TCP_SELECT_TIMEOUT_MS);
+ /* 调用select，添加错误处理 */
+ int select_result = 0;
+ if (max_fd >= 0) {
+ select_result = select(max_fd + 1, &read_fds, &write_fds, &except_fds, &timeout);
+ } else {
+ /* 没有活跃连接，短暂休眠 */
+ sys_msleep(100);
continue;
}
- /* 设置超时 */
- timeout.tv_sec = 0;
- timeout.tv_usec = TCP_SELECT_TIMEOUT_MS * 1000;
+ if (select_result < 0) {
+ if (errno == EINTR) {
+ /* 被信号中断，继续 */
+ continue;
+ } else {
+ printf("select错误: %d, 继续运行\n", errno);
+ sys_msleep(100);
+ continue;
+ }
+ }
- /* 执行select */
- int select_result = select(max_fd + 1, &read_fds, &write_fds, &error_fds, &timeout);
+ /* 处理异常事件 */
+ 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;
+
+ if (FD_ISSET(conn->sockfd, &except_fds)) {
+ printf("连接异常: handle=%p, sockfd=%d\n", conn->handle, conn->sockfd);
+ conn->state = TCP_STATE_ERROR;
+ cleanup_connection_unsafe(conn);
+ continue;
+ }
+ }
- if (select_result > 0) {
- 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) continue;
- /* 处理就绪的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 sockfd = conn->sockfd;
+ if (FD_ISSET(conn->sockfd, &read_fds)) {
+ /* 处理数据接收 */
+ uint8_t recv_buffer[TCP_RECV_BUFFER_SIZE];
+ ssize_t recv_len = recv(conn->sockfd, recv_buffer, sizeof(recv_buffer), MSG_DONTWAIT);
- /* 处理连接完成 */
- if (conn->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) {
- /* 连接成功 */
- 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 {
- /* 连接失败 */
- printf("连接失败: %s:%d, 错误: %d\n", conn->remote_ip, conn->remote_port, error);
+ if (recv_len > 0) {
+ /* 数据接收成功 */
+ conn->last_activity = sys_now();
+ g_stats.bytes_received += recv_len;
+
+ if (conn->data_cb) {
+ conn->data_cb(conn->handle, recv_buffer, recv_len);
+ }
+ } else if (recv_len == 0) {
+ /* 连接关闭 */
+ printf("连接关闭: handle=%p\n", conn->handle);
+ if (conn->disconnect_cb) {
+ conn->disconnect_cb(conn->handle);
+ }
+ cleanup_connection_unsafe(conn);
+ } else if (recv_len < 0) {
+ if (errno != EAGAIN && errno != EWOULDBLOCK) {
+ printf("接收错误: %d, 连接可能已断开\n", errno);
+ conn->state = TCP_STATE_ERROR;
+ cleanup_connection_unsafe(conn);
+ }
+ }
+ }
+ }
+
+ /* 处理写事件 */
+ 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;
+
+ if (FD_ISSET(conn->sockfd, &write_fds) && conn->send_queue_count > 0) {
+ /* 处理发送队列 */
+ process_send_queue(conn);
+ }
+ }
+
+ /* 处理连接状态变化 */
+ for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
+ tcp_connection_info_t* conn = &g_connections[i];
+ if (!conn->in_use) continue;
+
+ switch (conn->state) {
+ case TCP_STATE_CONNECTING:
+ /* 检查连接超时 */
+ if ((sys_now() - conn->last_activity) > TCP_CONNECT_TIMEOUT_MS) {
+ printf("连接超时: handle=%p\n", conn->handle);
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);
+ printf("计划重连 #%d: handle=%p\n", conn->retry_count, conn->handle);
} else {
- /* 重试次数用完，关闭连接 */
+ /* 重连次数用尽，清理连接 */
+ printf("重连次数用尽，清理连接: handle=%p\n", conn->handle);
if (conn->connect_cb) {
conn->connect_cb(conn->handle, false);
}
cleanup_connection_unsafe(conn);
- g_stats.errors++;
}
}
- }
-
- /* 处理数据接收 */
- if (FD_ISSET(sockfd, &read_fds)) {
- ssize_t bytes_received = recv(sockfd, recv_buffer, sizeof(recv_buffer), 0);
+ break;
- if (bytes_received > 0) {
+ case TCP_STATE_CONNECTED:
+ /* 检查连接是否仍然活跃 */
+ if ((sys_now() - conn->last_activity) > 30000) { /* 30秒无活动 */
+ printf("连接无活动，发送心跳: handle=%p\n", conn->handle);
+ /* 可以在这里发送心跳包 */
conn->last_activity = sys_now();
- g_stats.bytes_received += bytes_received;
-
- if (conn->data_cb) {
- conn->data_cb(conn->handle, recv_buffer, (uint16_t)bytes_received);
- }
- } 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);
- }
- 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);
+ break;
+
+ case TCP_STATE_CLOSING:
+ /* 等待关闭完成 */
+ if (conn->sockfd < 0) {
+ cleanup_connection_unsafe(conn);
}
+ break;
+
+ case TCP_STATE_ERROR:
+ /* 清理错误连接 */
+ printf("清理错误连接: handle=%p\n", conn->handle);
cleanup_connection_unsafe(conn);
- if (g_stats.active_connections > 0) {
- g_stats.active_connections--;
- }
- g_stats.errors++;
- }
+ break;
+
+ default:
+ break;
}
-
- sys_mutex_unlock(&g_connections_mutex);
- } else if (select_result < 0 && errno != EINTR) {
- /* 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) continue;
+ if (!conn->in_use || !conn->retry_pending) 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);
+ if ((sys_now() - conn->last_retry_time) >= TCP_CONNECT_RETRY_DELAY_MS) {
+ printf("执行重连: handle=%p, 重试 #%d\n", conn->handle, conn->retry_count);
+
+ /* 关闭旧socket */
+ if (conn->sockfd >= 0) {
+ close(conn->sockfd);
+ conn->sockfd = -1;
+ }
+
+ /* 重新创建socket */
+ int sockfd = socket(AF_INET, SOCK_STREAM, 0);
+ if (sockfd < 0) {
+ printf("重连失败：无法创建socket\n");
+ conn->retry_pending = false;
+ if (conn->connect_cb) {
+ conn->connect_cb(conn->handle, false);
}
cleanup_connection_unsafe(conn);
- if (g_stats.active_connections > 0) {
- g_stats.active_connections--;
- }
+ continue;
}
- } 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++;
+
+ /* 设置socket选项 */
+ if (set_socket_nonblocking(sockfd) < 0 || setup_keepalive(sockfd) < 0) {
+ printf("重连失败：设置socket选项失败\n");
+ close(sockfd);
+ conn->retry_pending = false;
+ if (conn->connect_cb) {
+ conn->connect_cb(conn->handle, false);
}
+ cleanup_connection_unsafe(conn);
+ continue;
}
+
+ conn->sockfd = sockfd;
+ conn->state = TCP_STATE_CONNECTING;
+ conn->last_activity = sys_now();
+ conn->retry_pending = false;
+
+ printf("重连socket已创建: %d\n", sockfd);
}
}
- sys_mutex_unlock(&g_connections_mutex);
+
+ /* 防止CPU占用过高 */
+ if (total_connections == 0) {
+ sys_msleep(50);
+ }
}
+
+ printf("TCP管理器IO线程退出\n");
}
/**
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);
-
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;
+ conn->last_activity = sys_now();
g_stats.total_connections++;
g_stats.active_connections++;
+
+ printf("连接立即成功: %s:%d, handle=%p\n", host, port, conn->handle);
+
+ /* 设置Keep-Alive */
setup_keepalive(sockfd);
- sys_mutex_unlock(&g_connections_mutex);
+ /* 调用连接成功回调 */
if (connect_cb) {
connect_cb(conn->handle, true);
}
- } else if (errno == EINPROGRESS) {
- /* 连接进行中，将由IO线程处理 */
- printf("连接进行中: %s:%d\n", host, port);
+
+ sys_mutex_unlock(&g_connections_mutex);
+ return conn->handle;
+
+ } else if (errno == EINPROGRESS || errno == EAGAIN || errno == EWOULDBLOCK) {
+ /* 连接正在进行中 */
+ conn->state = TCP_STATE_CONNECTING;
+ conn->last_activity = sys_now();
+ g_stats.total_connections++;
+ g_stats.active_connections++;
+
+ printf("连接进行中: %s:%d, handle=%p\n", host, port, conn->handle);
+
+ sys_mutex_unlock(&g_connections_mutex);
+ return conn->handle;
+
} else {
/* 连接失败 */
printf("连接失败: %s:%d, 错误: %d\n", host, port, errno);
- sys_mutex_lock(&g_connections_mutex);
- cleanup_connection_unsafe(conn);
+
+ /* 清理资源 */
+ close(sockfd);
+ conn->in_use = false;
+ conn->sockfd = -1;
+
sys_mutex_unlock(&g_connections_mutex);
- if (connect_cb) {
- connect_cb(conn->handle, false);
- }
return NULL;
}
-
- return conn->handle;
}
/**
projects / mirrors / FD230.git / commitdiff