uasync: рефакторинг — хелперы dispatch FD/SOCK, баг process_posted_tasks, чистка

lib/u_async.c

@@ -290,6 +290,48 @@ static struct socket_node* socket_array_get_by_sock(struct socket_array* sa, soc
     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;
+    int* field = is_read ? &node->enable_read : &node->enable_write;
-    node->enable_read = val;
+    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;
+        ev.events = socket_node_events(node);
-        if (node->type == SOCKET_NODE_TYPE_SOCK) {
+        ev.data.fd = socket_node_fd(node);
-            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;
+        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) {
+err_t uasync_set_socket_read(struct UASYNC* ua, void* s_id, int enable) {
-    if (!ua || !s_id) return ERR_FAIL;
+    return socket_set_event(ua, s_id, 1, enable);
-
-    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
 
-    ua->poll_fds_dirty = 1;
+err_t uasync_set_socket_write(struct UASYNC* ua, void* s_id, int enable) {
-    return ERR_OK;
+    return socket_set_event(ua, s_id, 0, enable);
 }
 
 // Helper function to rebuild cached pollfd array
@@ -830,31 +823,10 @@ static void rebuild_poll_fds(struct UASYNC* ua) {
         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
+        ua->poll_fds[idx].fd = socket_node_fd(cur);
-        if (cur->type == SOCKET_NODE_TYPE_SOCK) {
+        ua->poll_fds[idx].events = socket_node_events(cur);
-            // 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++;
     }
     
@@ -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) {
+            if (local_except) local_except(local_fd, local_ud);
-                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) {
+                if (local_read_sock) local_read_sock(local_sock, local_ud);
-                    local_read_sock(local_sock, local_ud);
+            } else if (local_read) local_read(local_fd, 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) {
+                if (local_write_sock) local_write_sock(local_sock, local_ud);
-                    local_write_sock(local_sock, local_ud);
+            } else if (local_write) local_write(local_fd, local_ud);
-                }
-            } else {
-                if (local_write) {
-                    local_write(local_fd, local_ud);
-                }
-            }
         }
     }
 }
@@ -1074,34 +1032,11 @@ void uasync_poll(struct UASYNC* ua, int timeout_tb) {
             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) {
+                if (node->except_cbk) node->except_cbk(node->fd, node->user_data);
-                    node->except_cbk(node->fd, node->user_data);
-                }
             }
 
-            if (has_read) {
+            if (has_read)  socket_node_dispatch_read(node);
-                if (node->type == SOCKET_NODE_TYPE_SOCK) {
+            if (has_write) socket_node_dispatch_write(node);
-                    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);
-                    }
-                }
-            }
         }
     }
 #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)
@@ -1150,44 +1077,16 @@ void uasync_poll(struct UASYNC* ua, int timeout_tb) {
             
             /* 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);
-                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) {
+                if (node->except_cbk) node->except_cbk(node->fd, node->user_data);
-                    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 & POLLIN)  socket_node_dispatch_read(node);
-            if (ua->poll_fds[i].revents & POLLOUT) {
+            if (ua->poll_fds[i].revents & POLLOUT) socket_node_dispatch_write(node);
-                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);
-                    }
-                }
-            }
         }
     }
 #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;                     // не нужен

lib/u_async.h

@@ -6,6 +6,7 @@
 #define UASYNC_H
 
 #include "platform_compat.h"
+#include "memory_pool.h"
 #include <stddef.h>
 #include <signal.h>
 #include "socket_compat.h"
@@ -27,8 +28,6 @@ typedef int err_t;
 
 typedef void (*uasync_post_callback_t)(void* user_arg);
 
-#include "memory_pool.h"
-
 struct timeout_node;
 struct twheels;
 
@@ -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);