diff --git a/lib/ll_queue.c b/lib/ll_queue.c index 24f8cc77..afa83443 100644 --- a/lib/ll_queue.c +++ b/lib/ll_queue.c @@ -130,11 +130,11 @@ void queue_free(struct ll_queue* q) { // ВАЖНО: Не освобождаем элементы в очереди - они должны быть извлечены отдельно // Это упрощает архитектуру и предотвращает double-u_free - // Освободить список waiters (сбросить handle->internal) + // Освободить список waiters (сбросить handle->internal, отменить отложенные вызовы) struct queue_waiter* w = q->waiter_head; while (w) { struct queue_waiter* next = w->next; - if (w->handle) w->handle->internal = NULL; + if (w->handle) { w->handle->internal = NULL; if (w->handle->call_soon_id) { uasync_call_soon_cancel(q->ua, w->handle->call_soon_id); w->handle->call_soon_id = NULL; } } u_free(w); w = next; } @@ -305,14 +305,17 @@ static uint32_t make_hash(const void* data, uint16_t len) {// алгоритм F static void waiter_defer_cb(void* arg) { struct queue_waiter_handle* h = (struct queue_waiter_handle*)arg; - h->defer_cb(h->defer_q, h->defer_arg); + h->call_soon_id = NULL; + if (h->defer_cb) h->defer_cb(h->defer_q, h->defer_arg); + else DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "waiter_defer_cb: NULL callback (stale?) q=%p", (void*)h->defer_q); } static inline void queue_waiter_call(struct ll_queue* q, queue_threshold_callback_fn cb, void* arg, struct queue_waiter_handle* h) { if (q->waiter_defer) { + if (h->call_soon_id) { DEBUG_WARN(DEBUG_CATEGORY_LL_QUEUE, "queue_waiter_call: overwriting existing call_soon_id=%p", h->call_soon_id); uasync_call_soon_cancel(q->ua, h->call_soon_id); } h->defer_q = q; h->defer_cb = cb; h->defer_arg = arg; - uasync_call_soon(q->ua, h, waiter_defer_cb); + h->call_soon_id = uasync_call_soon(q->ua, h, waiter_defer_cb); } else { cb(q, arg); } @@ -692,6 +695,7 @@ int queue_waiter_wait(struct ll_queue* q, struct queue_waiter_handle* h, void queue_waiter_cancel(struct ll_queue* q, struct queue_waiter_handle* h) { if (!q || !h || !h->internal) return; + if (h->call_soon_id) { uasync_call_soon_cancel(q->ua, h->call_soon_id); h->call_soon_id = NULL; } struct queue_waiter* waiter = h->internal; struct queue_waiter* prev = NULL; diff --git a/lib/ll_queue.h b/lib/ll_queue.h index 8ba2eb35..5d0653be 100644 --- a/lib/ll_queue.h +++ b/lib/ll_queue.h @@ -117,6 +117,7 @@ struct queue_waiter_handle { struct ll_queue* defer_q; queue_threshold_callback_fn defer_cb; void* defer_arg; + void* call_soon_id; ///< хендл отложенного вызова (для отмены) }; /** diff --git a/lib/u_async.c b/lib/u_async.c index c27759ef..2128986f 100644 --- a/lib/u_async.c +++ b/lib/u_async.c @@ -52,20 +52,22 @@ struct socket_node { socket_t_callback_t write_cbk_sock; // For SOCK type socket_callback_t except_cbk; void* user_data; - int active; // 1 if socket is active, 0 if u_freed (for reuse) + int active; // 1 if socket is active, 0 if freed (for reuse) int enable_read; // 1 if read monitoring is enabled int enable_write; // 1 if write monitoring is enabled + uint16_t gen; // generation counter for epoll event validation }; // Array-based socket management for O(1) operations -struct socket_array { - struct socket_node* sockets; // Dynamic array of socket nodes - int* fd_to_index; // FD to array index mapping - int* index_to_fd; // Array index to FD mapping - int* active_indices; // Array of indices of active sockets (for O(1) traversal) - int capacity; // Total allocated capacity - int count; // Number of active sockets - int max_fd; // Maximum FD for bounds checking +struct socket_array { + struct socket_node* sockets; // Dynamic array of socket nodes + int* fd_to_index; // FD to array index mapping + int* index_to_fd; // Array index to FD mapping + int* active_indices; // Array of indices of active sockets (for O(1) traversal) + int capacity; // Total allocated capacity + int count; // Number of active sockets + int max_fd; // Maximum FD for bounds checking + uint16_t gen_counter; // incrementing generation for epoll stale-event detection }; static struct socket_array* socket_array_create(int initial_capacity); @@ -80,8 +82,9 @@ static struct socket_node* socket_array_get(struct socket_array* sa, int fd); static struct socket_array* socket_array_create(int initial_capacity) { if (initial_capacity < 4) initial_capacity = 4; // Minimum capacity - struct socket_array* sa = u_malloc(sizeof(struct socket_array)); - if (!sa) return NULL; + struct socket_array* sa = u_malloc(sizeof(struct socket_array)); + if (!sa) return NULL; + sa->gen_counter = 0; sa->sockets = u_calloc(initial_capacity, sizeof(struct socket_node)); sa->fd_to_index = u_calloc(initial_capacity, sizeof(int)); @@ -167,20 +170,22 @@ static int socket_array_add_internal(struct socket_array* sa, int fd, socket_t s sa->capacity = new_capacity; } - // Check if FD already exists - if (sa->fd_to_index[fd] != -1) return -1; // FD already exists - - // Find first u_free slot - int index = -1; - for (int i = 0; i < sa->capacity; i++) { - if (!sa->sockets[i].active) { - index = i; - break; - } + // Check if FD already has a node — reuse inactive slot if present + int index; + int existing = sa->fd_to_index[fd]; + if (existing != -1) { + if (sa->sockets[existing].active) return -1; // FD уже занят активной нодой — ошибка + index = existing; // переиспользуем неактивный слот (после socket_array_remove) + } else { + // Find first free slot + index = -1; + for (int i = 0; i < sa->capacity; i++) { + if (!sa->sockets[i].active) { index = i; break; } + } + if (index == -1) return -1; // No free slots + sa->fd_to_index[fd] = index; // новая привязка fd→слот } - if (index == -1) return -1; // No u_free slots (shouldn't happen) - // Add the socket sa->sockets[index].fd = fd; sa->sockets[index].sock = sock; @@ -194,8 +199,8 @@ static int socket_array_add_internal(struct socket_array* sa, int fd, socket_t s sa->sockets[index].active = 1; sa->sockets[index].enable_read = (read_cbk_fd != NULL || read_cbk_sock != NULL) ? 1 : 0; sa->sockets[index].enable_write = (write_cbk_fd != NULL || write_cbk_sock != NULL) ? 1 : 0; + sa->sockets[index].gen = ++sa->gen_counter; - sa->fd_to_index[fd] = index; sa->index_to_fd[index] = fd; sa->active_indices[sa->count] = index; // Add to active list sa->count++; @@ -246,7 +251,8 @@ static int socket_array_remove(struct socket_array* sa, int fd) { sa->sockets[index].user_data = NULL; sa->sockets[index].enable_read = 0; sa->sockets[index].enable_write = 0; - sa->fd_to_index[fd] = -1; + // fd_to_index[fd] сохраняем — stale epoll события найдут неактивную ноду и пропустятся. + // При переиспользовании fd, socket_array_add_internal перезапишет этот же слот. sa->index_to_fd[index] = -1; // Remove from active_indices by swapping with last element @@ -651,14 +657,14 @@ void* uasync_add_socket(struct UASYNC* ua, int fd, socket_callback_t read_cbk, s ua->poll_fds_dirty = 1; // Mark poll_fds as needing rebuild #if HAS_EPOLL - // Add to epoll if using epoll - if (ua->use_epoll && ua->epoll_fd >= 0) { - struct epoll_event ev; - ev.events = 0; - if (read_cbk) ev.events |= EPOLLIN; - if (write_cbk) ev.events |= EPOLLOUT; - // Use level-triggered mode (default) for compatibility with UDP sockets - ev.data.fd = fd; + // Add to epoll if using epoll + if (ua->use_epoll && ua->epoll_fd >= 0) { + struct epoll_event ev; + ev.events = 0; + if (read_cbk) ev.events |= EPOLLIN; + if (write_cbk) ev.events |= EPOLLOUT; + // Embed gen in upper 32 bits of data.u64 for stale-event detection + ev.data.u64 = ((uint64_t)ua->sockets->sockets[index].gen << 32) | (uint32_t)fd; if (epoll_ctl(ua->epoll_fd, EPOLL_CTL_ADD, fd, &ev) < 0) { // Failed to add to epoll - remove from socket array and return error @@ -721,7 +727,7 @@ void* uasync_add_socket_t(struct UASYNC* ua, socket_t sock, socket_t_callback_t #else int fd = sock; #endif - ev.data.fd = fd; + ev.data.u64 = ((uint64_t)ua->sockets->sockets[index].gen << 32) | (uint32_t)fd; if (epoll_ctl(ua->epoll_fd, EPOLL_CTL_ADD, fd, &ev) < 0) { socket_array_remove(ua->sockets, fd); ua->socket_alloc_count--; @@ -786,13 +792,13 @@ err_t uasync_set_socket_read(struct UASYNC* ua, void* s_id, int enable) { } else { if (node->read_cbk && node->enable_read) ev.events |= EPOLLIN; if (node->write_cbk && node->enable_write) ev.events |= EPOLLOUT; - ev.data.fd = node->fd; } if (node->except_cbk) ev.events |= EPOLLPRI; + ev.data.u64 = ((uint64_t)node->gen << 32) | (uint32_t)(node->type == SOCKET_NODE_TYPE_SOCK ? (int)node->sock : node->fd); #ifdef _WIN32 - int efd = (int)(intptr_t)ev.data.fd; + int efd = (int)(intptr_t)(node->type == SOCKET_NODE_TYPE_SOCK ? node->sock : node->fd); #else - int efd = ev.data.fd; + int efd = (node->type == SOCKET_NODE_TYPE_SOCK ? (int)node->sock : node->fd); #endif epoll_ctl(ua->epoll_fd, EPOLL_CTL_MOD, efd, &ev); } @@ -819,17 +825,16 @@ err_t uasync_set_socket_write(struct UASYNC* ua, void* s_id, int enable) { if (node->type == SOCKET_NODE_TYPE_SOCK) { if (node->read_cbk_sock && node->enable_read) ev.events |= EPOLLIN; if (node->write_cbk_sock && node->enable_write) ev.events |= EPOLLOUT; - ev.data.fd = node->sock; } else { if (node->read_cbk && node->enable_read) ev.events |= EPOLLIN; if (node->write_cbk && node->enable_write) ev.events |= EPOLLOUT; - ev.data.fd = node->fd; } if (node->except_cbk) ev.events |= EPOLLPRI; + ev.data.u64 = ((uint64_t)node->gen << 32) | (uint32_t)(node->type == SOCKET_NODE_TYPE_SOCK ? (int)node->sock : node->fd); #ifdef _WIN32 - int efd = (int)(intptr_t)ev.data.fd; + int efd = (int)(intptr_t)(node->type == SOCKET_NODE_TYPE_SOCK ? node->sock : node->fd); #else - int efd = ev.data.fd; + int efd = (node->type == SOCKET_NODE_TYPE_SOCK ? (int)node->sock : node->fd); #endif epoll_ctl(ua->epoll_fd, EPOLL_CTL_MOD, efd, &ev); } @@ -908,22 +913,20 @@ static void rebuild_poll_fds(struct UASYNC* ua) { // Process events from epoll (Linux only) #if HAS_EPOLL -static void process_epoll_events(struct UASYNC* ua, struct epoll_event* events, int n_events) { - for (int i = 0; i < n_events; i++) { - // Check if this is the wakeup fd (data.fd is -1) +static void process_epoll_events(struct UASYNC* ua, struct epoll_event* events, int n_events) { + for (int i = 0; i < n_events; i++) { + // Wakeup fd detected by data.fd < 0 (lower 32 bits of data.u64 = -1) if (events[i].data.fd < 0) { - if (events[i].events & EPOLLIN) { - drain_wakeup_pipe(ua); - } + if (events[i].events & EPOLLIN) { drain_wakeup_pipe(ua); } continue; } - // Socket event — save node fields locally: callbacks may realloc - // the socket array, invalidating `node`. - struct socket_node* node = socket_array_get(ua->sockets, events[i].data.fd); - DEBUG_DEBUG(DEBUG_CATEGORY_UASYNC, "poll→fd=%d ev=0x%x%s", events[i].data.fd, events[i].events, - !node || !node->active ? " (inactive)" : ""); + int fd = (int)(events[i].data.u64 & 0xFFFFFFFF); + uint16_t ev_gen = (uint16_t)(events[i].data.u64 >> 32); + + struct socket_node* node = socket_array_get(ua->sockets, fd); if (!node || !node->active) continue; + if (node->gen != ev_gen) { DEBUG_ERROR(DEBUG_CATEGORY_UASYNC, "STALE epoll fd=%d ev_gen=%u node_gen=%u", fd, ev_gen, node->gen); continue; } int local_fd = node->fd; socket_t local_sock = node->sock; int local_type = node->type; diff --git a/src/pkt_normalizer.c b/src/pkt_normalizer.c index 2917ae4a..4e36ab01 100644 --- a/src/pkt_normalizer.c +++ b/src/pkt_normalizer.c @@ -394,11 +394,12 @@ static void pn_unpacker_cb(struct ll_queue* q, void* arg) { ptr += cp; if (pn->recvpart->len == pn->recvpart->memlen) { - DEBUG_DEBUG(DEBUG_CATEGORY_NORMALIZER, "unpacked dgram (size=%d)", pn->recvpart->len); - if (debug_should_output(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_DUMP)) log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_DUMP, "NORM->", pn->recvpart->dgram, pn->recvpart->len); + uint32_t recv_len = pn->recvpart->len; + DEBUG_DEBUG(DEBUG_CATEGORY_NORMALIZER, "unpacked dgram (size=%d)", recv_len); + if (debug_should_output(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_DUMP)) log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_DUMP, "NORM->", pn->recvpart->dgram, recv_len); queue_data_put(pn->output, pn->recvpart); pn->out_total_pkts++; - pn->out_total_bytes += pn->recvpart->len; + pn->out_total_bytes += recv_len; pn->recvpart = NULL; } } diff --git a/tests/Makefile.am b/tests/Makefile.am index 2200e62c..e6626fdd 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -50,6 +50,7 @@ check_PROGRAMS = \ test_bbr_integration \ test_intensive_memory_pool \ test_tcp_io \ + test_uasync_socket_race \ bench_timeout_heap \ bench_uasync_timeouts @@ -281,6 +282,10 @@ test_tcp_io_SOURCES = test_tcp_io.c test_tcp_io_CFLAGS = -I$(top_srcdir)/lib test_tcp_io_LDADD = $(COMMON_LIBS) +test_uasync_socket_race_SOURCES = test_uasync_socket_race.c +test_uasync_socket_race_CFLAGS = -I$(top_srcdir)/lib +test_uasync_socket_race_LDADD = $(COMMON_LIBS) + test_memory_pool_and_config_SOURCES = test_memory_pool_and_config.c test_memory_pool_and_config_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib test_memory_pool_and_config_LDADD = $(COMMON_LIBS) diff --git a/tests/test_uasync_socket_race.c b/tests/test_uasync_socket_race.c new file mode 100644 index 00000000..0cceecab --- /dev/null +++ b/tests/test_uasync_socket_race.c @@ -0,0 +1,155 @@ +// test_uasync_socket_race.c — тест гонки fd-reuse в epoll при быстром accept/close/accept +// 4 дочерних процесса, каждый 200 connect+send+close. Сервер на uasync+tcp_io. +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "../lib/tcp_io.h" +#include "../lib/u_async.h" +#include "../lib/ll_queue.h" +#include "../lib/debug_config.h" +#include "../lib/mem.h" + +#define CHILDREN 4 +#define ITER_PER_CHILD 200 +#define TIMEOUT_MS 30000 + +static struct UASYNC* g_ua = NULL; +static int g_ok = 0, g_done = 0; +static int g_conn_count = 0, g_read_count = 0, g_err_count = 0; +static pid_t g_children[CHILDREN]; +static int g_port = 0; +static void* g_mon_id = NULL; + +static void on_read_cb(struct ll_queue* q, void* arg) { + struct tcp_conn* tc = (struct tcp_conn*)arg; + struct ll_entry* e = queue_data_get(q); + if (!e) { queue_resume_callback(q); return; } + if (e->len == 3 && memcmp(e->dgram, "OK\n", 3) == 0) g_read_count++; + memory_pool_free(tc->data_pool, e->dgram); queue_entry_free(e); + queue_resume_callback(q); +} + +static void on_error_cb(struct tcp_conn* tc, int err, void* arg) { + (void)err; (void)arg; + g_err_count++; + tcp_conn_destroy(tc); +} + +static void on_fin_cb(struct tcp_conn* tc, void* arg) { + (void)arg; + tcp_conn_push_close(tc); +} + +static void on_closed_cb(struct tcp_conn* tc, void* arg) { + (void)arg; + tcp_conn_destroy(tc); +} + +static void on_accept_cb(int fd, void* arg) { + (void)arg; + struct sockaddr_in addr; socklen_t alen = sizeof(addr); + int csock = accept(fd, (struct sockaddr*)&addr, &alen); + if (csock < 0) return; + socket_set_nonblocking(csock); + struct tcp_conn* tc = tcp_conn_create(g_ua, csock, 512, 512, 4, 0, 0, on_fin_cb, on_error_cb, NULL); + if (!tc) { socket_close_wrapper(csock); return; } + tc->on_closed = on_closed_cb; + queue_set_callback(tc->read_queue, on_read_cb, tc); + g_conn_count++; +} + +static int child_main(int port, int id, int count) { + (void)id; + for (int i = 0; i < count; i++) { + int s = socket(AF_INET, SOCK_STREAM, 0); + if (s < 0) return 1; + struct sockaddr_in a; memset(&a, 0, sizeof(a)); + a.sin_family = AF_INET; a.sin_port = htons((uint16_t)port); + inet_pton(AF_INET, "127.0.0.1", &a.sin_addr); + if (connect(s, (struct sockaddr*)&a, sizeof(a)) < 0) { close(s); return 1; } + if (send(s, "OK\n", 3, 0) != 3) { close(s); return 1; } + close(s); + } + return 0; +} + +static void monitor(void* arg) { + (void)arg; + int alive = 0; + for (int i = 0; i < CHILDREN; i++) { + if (g_children[i] <= 0) continue; + int status; pid_t r = waitpid(g_children[i], &status, WNOHANG); + if (r == 0) { alive++; continue; } + if (WIFEXITED(status) && WEXITSTATUS(status) != 0) { g_done = -1; return; } + g_children[i] = 0; + } + if (alive == 0) { g_ok = 1; g_done = 1; } + if (!g_done) g_mon_id = uasync_set_timeout(g_ua, 500, NULL, monitor, "mon"); +} + +static void test_timeout(void* arg) { + (void)arg; + printf("[FAIL] timeout conn=%d read=%d err=%d\n", g_conn_count, g_read_count, g_err_count); + g_done = -1; +} + +int main(void) { + printf("=== test_uasync_socket_race ===\n"); fflush(stdout); + + debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); + srand((unsigned)getpid()); + close(0); open("/dev/null", O_RDONLY); + + g_port = 25000 + (rand() % 10000); + g_ua = uasync_create(); + if (!g_ua) { printf("[FAIL] uasync_create\n"); return 1; } + + int lsock = socket(AF_INET, SOCK_STREAM, 0); + if (lsock < 0) { printf("[FAIL] socket\n"); goto cleanup; } + socket_set_reuseaddr(lsock, 1); + socket_set_nonblocking(lsock); + struct sockaddr_in la; memset(&la, 0, sizeof(la)); + la.sin_family = AF_INET; la.sin_port = htons((uint16_t)g_port); + la.sin_addr.s_addr = inet_addr("127.0.0.1"); + if (bind(lsock, (struct sockaddr*)&la, sizeof(la)) < 0) { printf("[FAIL] bind: %s\n", strerror(errno)); goto cleanup; } + if (listen(lsock, 32) < 0) { printf("[FAIL] listen: %s\n", strerror(errno)); goto cleanup; } + uasync_add_socket(g_ua, lsock, on_accept_cb, NULL, NULL, NULL); + printf(" listen fd=%d epoll ok\n", lsock); fflush(stdout); + + for (int i = 0; i < CHILDREN; i++) { + pid_t pid = fork(); + if (pid < 0) { printf("[FAIL] fork\n"); goto cleanup; } + if (pid == 0) { _exit(child_main(g_port, i, ITER_PER_CHILD)); } + g_children[i] = pid; + } + printf(" %d children × %d iterations on port %d\n", CHILDREN, ITER_PER_CHILD, g_port); fflush(stdout); + + g_mon_id = uasync_set_timeout(g_ua, 100, NULL, monitor, "mon"); + void* to_id = uasync_set_timeout(g_ua, TIMEOUT_MS * 10, NULL, test_timeout, "to"); + + while (!g_done) uasync_poll(g_ua, 50); + + if (to_id) uasync_cancel_timeout(g_ua, to_id); + + int expected = CHILDREN * ITER_PER_CHILD; + if (g_ok && g_conn_count == expected && g_read_count == expected) { + printf("[PASS] test_uasync_socket_race — %d/%d/%d conn/read/err\n", g_conn_count, g_read_count, g_err_count); + } else { + printf("[FAIL] test_uasync_socket_race — expected %d, got %d/%d/%d\n", expected, g_conn_count, g_read_count, g_err_count); + g_ok = 0; + } + +cleanup: + for (int i = 0; i < CHILDREN; i++) if (g_children[i] > 0) { kill(g_children[i], SIGKILL); waitpid(g_children[i], NULL, 0); } + if (g_mon_id) uasync_cancel_timeout(g_ua, g_mon_id); + if (g_ua) uasync_destroy(g_ua, 0); + return g_ok ? 0 : 1; +}