jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fix: use-after-free in pkt_normalizer + ll_queue waiter cancel + uasync gen counter for epoll fd-reuse

Browse Source 
- pkt_normalizer: save recvpart->len before queue_data_put (may free synchronously)
- ll_queue: save uasync_call_soon handle in queue_waiter_handle, cancel on free
- uasync: generation counter in epoll data.u64 to skip stale events after fd reuse
- uasync: keep fd_to_index mapping after socket_array_remove for inactive nodes
- test: test_uasync_socket_race — 4 children × 200 iterations, currently flaky
This commit is contained in:
Evgeny committed 2026-07-01 23:51:31 +03:00
1 parent a01e2f0899
commit ccdbab95a8
6 files changed
+228 -59
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

lib/u_async.c
+55 -52
View File
@@ -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;