diff --git a/lib/u_async.c b/lib/u_async.c index 4cad19d2..3897bbfd 100644 --- a/lib/u_async.c +++ b/lib/u_async.c @@ -287,9 +287,51 @@ static struct socket_node* socket_array_get_by_sock(struct socket_array* sa, soc if (index == -1 || !sa->sockets[index].active) return NULL; if (sa->sockets[index].type != SOCKET_NODE_TYPE_SOCK) return NULL; - return &sa->sockets[index]; -} - + return &sa->sockets[index]; +} + +// ------ socket_node dispatch helpers (eliminate FD/SOCK type duplication) ------ + +static inline int socket_node_fd(const struct socket_node* n) { + if (n->type == SOCKET_NODE_TYPE_SOCK) { +#ifdef _WIN32 + return (int)(intptr_t)n->sock; +#else + return n->sock; +#endif + } + return n->fd; +} + +static inline void socket_node_dispatch_read(const struct socket_node* n) { + if (n->type == SOCKET_NODE_TYPE_SOCK) { + if (n->read_cbk_sock) n->read_cbk_sock(n->sock, n->user_data); + } else { + if (n->read_cbk) n->read_cbk(n->fd, n->user_data); + } +} + +static inline void socket_node_dispatch_write(const struct socket_node* n) { + if (n->type == SOCKET_NODE_TYPE_SOCK) { + if (n->write_cbk_sock) n->write_cbk_sock(n->sock, n->user_data); + } else { + if (n->write_cbk) n->write_cbk(n->fd, n->user_data); + } +} + +static inline short socket_node_events(const struct socket_node* n) { + short events = 0; + if (n->type == SOCKET_NODE_TYPE_SOCK) { + if (n->read_cbk_sock && n->enable_read) events |= POLLIN; + if (n->write_cbk_sock && n->enable_write) events |= POLLOUT; + } else { + if (n->read_cbk && n->enable_read) events |= POLLIN; + if (n->write_cbk && n->enable_write) events |= POLLOUT; + } + if (n->except_cbk) events |= POLLPRI; + return events; +} + // Simplified timeout handling without reference counting static void get_current_time(struct timeval* tv) { #ifdef _WIN32 @@ -349,17 +391,6 @@ uint64_t get_time_us(void) { -// Drain wakeup pipe - read all available bytes -static void drain_wakeup_pipe(struct UASYNC* ua) { - if (!ua || !ua->wakeup_initialized) return; - - char buf[64]; - while (1) { - ssize_t n = read(ua->wakeup_pipe[0], buf, sizeof(buf)); - if (n <= 0) break; - } -} - // Process posted tasks (lock-u_free during execution) static void process_posted_tasks(struct UASYNC* ua) { if (!ua) return; @@ -381,7 +412,7 @@ static void process_posted_tasks(struct UASYNC* ua) { #endif while (list) { - DEBUG_DEBUG(DEBUG_CATEGORY_TUN, "POSTed task get"); + DEBUG_DEBUG(DEBUG_CATEGORY_UASYNC, "POSTed task get"); struct posted_task* t = list; list = list->next; @@ -721,36 +752,27 @@ err_t uasync_remove_socket_t(struct UASYNC* ua, socket_t sock) { return ERR_FAIL; } -err_t uasync_set_socket_read(struct UASYNC* ua, void* s_id, int enable) { +static err_t socket_set_event(struct UASYNC* ua, void* s_id, int is_read, int enable) { if (!ua || !s_id) return ERR_FAIL; struct socket_node* node = (struct socket_node*)s_id; if (!node->active || node->fd < 0) return ERR_FAIL; int val = enable ? 1 : 0; - if (node->enable_read == val) return ERR_OK; - node->enable_read = val; + int* field = is_read ? &node->enable_read : &node->enable_write; + if (*field == val) return ERR_OK; + *field = val; #if HAS_EPOLL if (ua->use_epoll && ua->epoll_fd >= 0) { struct epoll_event ev; - ev.events = 0; - 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.events = socket_node_events(node); + ev.data.fd = socket_node_fd(node); #ifdef _WIN32 - int efd = (int)(intptr_t)ev.data.fd; + epoll_ctl(ua->epoll_fd, EPOLL_CTL_MOD, (int)(intptr_t)ev.data.fd, &ev); #else - int efd = ev.data.fd; + epoll_ctl(ua->epoll_fd, EPOLL_CTL_MOD, ev.data.fd, &ev); #endif - epoll_ctl(ua->epoll_fd, EPOLL_CTL_MOD, efd, &ev); } #endif @@ -758,41 +780,12 @@ err_t uasync_set_socket_read(struct UASYNC* ua, void* s_id, int enable) { return ERR_OK; } -err_t uasync_set_socket_write(struct UASYNC* ua, void* s_id, int enable) { - if (!ua || !s_id) return ERR_FAIL; - - struct socket_node* node = (struct socket_node*)s_id; - if (!node->active || node->fd < 0) return ERR_FAIL; - - int val = enable ? 1 : 0; - if (node->enable_write == val) return ERR_OK; - node->enable_write = val; - -#if HAS_EPOLL - if (ua->use_epoll && ua->epoll_fd >= 0) { - struct epoll_event ev; - ev.events = 0; - 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; -#ifdef _WIN32 - int efd = (int)(intptr_t)ev.data.fd; -#else - int efd = ev.data.fd; -#endif - epoll_ctl(ua->epoll_fd, EPOLL_CTL_MOD, efd, &ev); - } -#endif +err_t uasync_set_socket_read(struct UASYNC* ua, void* s_id, int enable) { + return socket_set_event(ua, s_id, 1, enable); +} - ua->poll_fds_dirty = 1; - return ERR_OK; +err_t uasync_set_socket_write(struct UASYNC* ua, void* s_id, int enable) { + return socket_set_event(ua, s_id, 0, enable); } // Helper function to rebuild cached pollfd array @@ -825,37 +818,16 @@ static void rebuild_poll_fds(struct UASYNC* ua) { idx++; } - // Add socket fds using active_indices for O(1) traversal - for (int i = 0; i < socket_count; i++) { - int socket_array_idx = ua->sockets->active_indices[i]; - struct socket_node* cur = &ua->sockets->sockets[socket_array_idx]; - - // Handle socket_t vs int fd - if (cur->type == SOCKET_NODE_TYPE_SOCK) { - // socket_t - cast to int for pollfd -#ifdef _WIN32 - ua->poll_fds[idx].fd = (int)(intptr_t)cur->sock; -#else - ua->poll_fds[idx].fd = cur->sock; -#endif - } else { - // Regular fd - ua->poll_fds[idx].fd = cur->fd; - } - ua->poll_fds[idx].events = 0; - ua->poll_fds[idx].revents = 0; - - if (cur->type == SOCKET_NODE_TYPE_SOCK) { - if (cur->read_cbk_sock && cur->enable_read) ua->poll_fds[idx].events |= POLLIN; - if (cur->write_cbk_sock && cur->enable_write) ua->poll_fds[idx].events |= POLLOUT; - } else { - if (cur->read_cbk && cur->enable_read) ua->poll_fds[idx].events |= POLLIN; - if (cur->write_cbk && cur->enable_write) ua->poll_fds[idx].events |= POLLOUT; - } - if (cur->write_cbk && cur->enable_write) ua->poll_fds[idx].events |= POLLOUT; - if (cur->except_cbk) ua->poll_fds[idx].events |= POLLPRI; - - idx++; + // Add socket fds using active_indices for O(1) traversal + for (int i = 0; i < socket_count; i++) { + int socket_array_idx = ua->sockets->active_indices[i]; + struct socket_node* cur = &ua->sockets->sockets[socket_array_idx]; + + ua->poll_fds[idx].fd = socket_node_fd(cur); + ua->poll_fds[idx].events = socket_node_events(cur); + ua->poll_fds[idx].revents = 0; + + idx++; } ua->poll_fds_count = total_fds; @@ -869,7 +841,7 @@ static void process_epoll_events(struct UASYNC* ua, struct epoll_event* events, // Check if this is the wakeup fd (data.fd is -1) if (events[i].data.fd < 0) { if (events[i].events & EPOLLIN) { - drain_wakeup_pipe(ua); + handle_wakeup(ua); } continue; } @@ -892,35 +864,21 @@ static void process_epoll_events(struct UASYNC* ua, struct epoll_event* events, /* Check for error conditions first */ if (events[i].events & (EPOLLERR | EPOLLHUP)) { - if (local_except) { - local_except(local_fd, local_ud); - } + if (local_except) local_except(local_fd, local_ud); } - - /* Read readiness - use appropriate callback based on socket type */ + + /* Read readiness */ if (events[i].events & EPOLLIN) { if (local_type == SOCKET_NODE_TYPE_SOCK) { - if (local_read_sock) { - local_read_sock(local_sock, local_ud); - } - } else { - if (local_read) { - local_read(local_fd, local_ud); - } - } + if (local_read_sock) local_read_sock(local_sock, local_ud); + } else if (local_read) local_read(local_fd, local_ud); } - - /* Write readiness - use appropriate callback based on socket type */ + + /* Write readiness */ if (events[i].events & EPOLLOUT) { if (local_type == SOCKET_NODE_TYPE_SOCK) { - if (local_write_sock) { - local_write_sock(local_sock, local_ud); - } - } else { - if (local_write) { - local_write(local_fd, local_ud); - } - } + if (local_write_sock) local_write_sock(local_sock, local_ud); + } else if (local_write) local_write(local_fd, local_ud); } } } @@ -1073,35 +1031,12 @@ void uasync_poll(struct UASYNC* ua, int timeout_tb) { if (!has_read && !has_write && !has_except) continue; DEBUG_DEBUG(DEBUG_CATEGORY_UASYNC, "select→fd=%d r=%d w=%d e=%d", (int)s, has_read, has_write, has_except); - if (has_except) { - if (node->except_cbk) { - node->except_cbk(node->fd, node->user_data); - } - } - - if (has_read) { - if (node->type == SOCKET_NODE_TYPE_SOCK) { - if (node->read_cbk_sock) { - node->read_cbk_sock(node->sock, node->user_data); - } - } else { - if (node->read_cbk) { - node->read_cbk(node->fd, node->user_data); - } - } - } - - if (has_write) { - if (node->type == SOCKET_NODE_TYPE_SOCK) { - if (node->write_cbk_sock) { - node->write_cbk_sock(node->sock, node->user_data); - } - } else { - if (node->write_cbk) { - node->write_cbk(node->fd, node->user_data); - } - } - } + if (has_except) { + if (node->except_cbk) node->except_cbk(node->fd, node->user_data); + } + + if (has_read) socket_node_dispatch_read(node); + if (has_write) socket_node_dispatch_write(node); } } #else @@ -1132,14 +1067,6 @@ void uasync_poll(struct UASYNC* ua, int timeout_tb) { if (ua->poll_fds[i].revents == 0) continue; DEBUG_DEBUG(DEBUG_CATEGORY_UASYNC, "poll→fd=%d rev=0x%x", ua->poll_fds[i].fd, ua->poll_fds[i].revents); - /* Handle wakeup fd separately */ - if (wakeup_fd_present && i == 0) { - if (ua->poll_fds[i].revents & POLLIN) { - drain_wakeup_pipe(ua); - } - continue; - } - /* Socket event - lookup by fd */ struct socket_node* node = socket_array_get(ua->sockets, ua->poll_fds[i].fd); if (!node) { // Try by socket_t (in case this is a socket) @@ -1148,46 +1075,18 @@ void uasync_poll(struct UASYNC* ua, int timeout_tb) { } if (!node) continue; // Socket may have been removed - /* Check for error conditions first */ - if (ua->poll_fds[i].revents & (POLLERR | POLLHUP | POLLNVAL)) { - /* Treat as exceptional condition */ - if (node->except_cbk) { - node->except_cbk(node->fd, node->user_data); - } - } - - /* Exceptional data (out-of-band) */ - if (ua->poll_fds[i].revents & POLLPRI) { - if (node->except_cbk) { - node->except_cbk(node->fd, node->user_data); - } - } - - /* Read readiness - use appropriate callback based on socket type */ - if (ua->poll_fds[i].revents & POLLIN) { - if (node->type == SOCKET_NODE_TYPE_SOCK) { - if (node->read_cbk_sock) { - node->read_cbk_sock(node->sock, node->user_data); - } - } else { - if (node->read_cbk) { - node->read_cbk(node->fd, node->user_data); - } - } - } - - /* Write readiness - use appropriate callback based on socket type */ - if (ua->poll_fds[i].revents & POLLOUT) { - if (node->type == SOCKET_NODE_TYPE_SOCK) { - if (node->write_cbk_sock) { - node->write_cbk_sock(node->sock, node->user_data); - } - } else { - if (node->write_cbk) { - node->write_cbk(node->fd, node->user_data); - } - } - } + /* Check for error conditions first */ + if (ua->poll_fds[i].revents & (POLLERR | POLLHUP | POLLNVAL)) { + if (node->except_cbk) node->except_cbk(node->fd, node->user_data); + } + + /* Exceptional data (out-of-band) */ + if (ua->poll_fds[i].revents & POLLPRI) { + if (node->except_cbk) node->except_cbk(node->fd, node->user_data); + } + + if (ua->poll_fds[i].revents & POLLIN) socket_node_dispatch_read(node); + if (ua->poll_fds[i].revents & POLLOUT) socket_node_dispatch_write(node); } } #endif @@ -1197,9 +1096,6 @@ void uasync_poll(struct UASYNC* ua, int timeout_tb) { } -// Put this near the top of u_async.c, after includes and before uasync_create - - #ifdef _WIN32 static void wakeup_read_callback_win(socket_t sock, void* arg) { (void)sock; // не нужен diff --git a/lib/u_async.h b/lib/u_async.h index a65dc712..495179de 100644 --- a/lib/u_async.h +++ b/lib/u_async.h @@ -6,6 +6,7 @@ #define UASYNC_H #include "platform_compat.h" +#include "memory_pool.h" #include #include #include "socket_compat.h" @@ -25,11 +26,9 @@ typedef int err_t; #define SOCKET_NODE_TYPE_FD 0 // Regular file descriptor (pipe, file) #define SOCKET_NODE_TYPE_SOCK 1 // Socket (socket_t) -typedef void (*uasync_post_callback_t)(void* user_arg); - -#include "memory_pool.h" +typedef void (*uasync_post_callback_t)(void* user_arg); -struct timeout_node; +struct timeout_node; struct twheels; struct posted_task { @@ -74,7 +73,6 @@ struct UASYNC { // Type definitions typedef struct UASYNC uasync_t; -typedef struct UASYNC UASYNC_t; // Instance API - основной API для работы с uasync struct UASYNC* uasync_create(void);