From 187bcd80b92d156bfa90ef5cab8d90b5af31a23f Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 14 Sep 2026 12:56:52 +0300 Subject: [PATCH] =?UTF-8?q?uasync:=20=D1=80=D0=B5-=D0=B2=D0=B0=D0=BB=D0=B8?= =?UTF-8?q?=D0=B4=D0=B0=D1=86=D0=B8=D1=8F=20node=20=D0=BF=D0=B5=D1=80?= =?UTF-8?q?=D0=B5=D0=B4=20=D0=BF=D0=BE=D1=81=D0=BB=D0=B5=D0=B4=D1=83=D1=8E?= =?UTF-8?q?=D1=89=D0=B8=D0=BC=D0=B8=20=D0=BA=D0=BE=D0=BB=D0=BB=D0=B1=D1=8D?= =?UTF-8?q?=D0=BA=D0=B0=D0=BC=D0=B8=20(=D1=81=D1=87=D1=91=D1=82=D1=87?= =?UTF-8?q?=D0=B8=D0=BA=20cb=5Fcalled)=20=D0=B2=D0=BC=D0=B5=D1=81=D1=82?= =?UTF-8?q?=D0=BE=20=D1=81=D0=BD=D0=B0=D0=BF=D1=88=D0=BE=D1=82=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lib/u_async.c | 122 ++++++++++++++++++++++++++++---------------------- 1 file changed, 69 insertions(+), 53 deletions(-) diff --git a/lib/u_async.c b/lib/u_async.c index 3d54ccdc..e4ff1b0e 100644 --- a/lib/u_async.c +++ b/lib/u_async.c @@ -1158,22 +1158,11 @@ void uasync_poll(struct UASYNC* ua, int timeout_tb) { struct socket_node* node = &ua->sockets->sockets[idx]; if (!node->active) continue; - /* Снапшот полей — коллбэки могут реаллоцировать ua->sockets (node станет висячим) */ - int local_fd = node->fd; - socket_t local_sock = node->sock; - int local_type = node->type; - void* local_ud = node->user_data; - socket_callback_t local_except = node->except_cbk; - socket_callback_t local_read = node->read_cbk; - socket_callback_t local_write = node->write_cbk; - socket_t_callback_t local_read_sock = node->read_cbk_sock; - socket_t_callback_t local_write_sock = node->write_cbk_sock; - SOCKET s; - if (local_type == SOCKET_NODE_TYPE_SOCK) { - s = local_sock; + if (node->type == SOCKET_NODE_TYPE_SOCK) { + s = node->sock; } else { - s = (SOCKET)local_fd; + s = (SOCKET)node->fd; } int has_read = FD_ISSET(s, &read_fds); @@ -1183,34 +1172,49 @@ void uasync_poll(struct UASYNC* ua, int timeout_tb) { if (!has_read && !has_write && !has_except) continue; DEBUG_DEBUG(DEBUG_CATEGORY_SYS, "select→fd=%d r=%d w=%d e=%d", (int)s, has_read, has_write, has_except); + /* Коллбэки могут удалить/добавить сокет (realloc массива) — перед + * каждым последующим коллбэком пере-валидируем node по индексу. */ + int cb_called = 0; + if (has_except) { - if (local_except) { - local_except(local_fd, local_ud); + if (node->except_cbk) { + node->except_cbk(node->fd, node->user_data); } + cb_called++; } if (has_read) { - if (local_type == SOCKET_NODE_TYPE_SOCK) { - if (local_read_sock) { - local_read_sock(local_sock, local_ud); + if (cb_called) { + node = &ua->sockets->sockets[idx]; + if (!node->active) continue; + } + if (node->type == SOCKET_NODE_TYPE_SOCK) { + if (node->read_cbk_sock) { + node->read_cbk_sock(node->sock, node->user_data); } } else { - if (local_read) { - local_read(local_fd, local_ud); + if (node->read_cbk) { + node->read_cbk(node->fd, node->user_data); } } + cb_called++; } if (has_write) { - if (local_type == SOCKET_NODE_TYPE_SOCK) { - if (local_write_sock) { - local_write_sock(local_sock, local_ud); + if (cb_called) { + node = &ua->sockets->sockets[idx]; + if (!node->active) continue; + } + if (node->type == SOCKET_NODE_TYPE_SOCK) { + if (node->write_cbk_sock) { + node->write_cbk_sock(node->sock, node->user_data); } } else { - if (local_write) { - local_write(local_fd, local_ud); + if (node->write_cbk) { + node->write_cbk(node->fd, node->user_data); } } + cb_called++; } } } @@ -1254,63 +1258,75 @@ void uasync_poll(struct UASYNC* ua, int timeout_tb) { } /* Socket event - lookup by fd */ - struct socket_node* node = socket_array_get(ua->sockets, ua->poll_fds[i].fd); + int fd = ua->poll_fds[i].fd; + struct socket_node* node = socket_array_get(ua->sockets, fd); if (!node) { // Try by socket_t (in case this is a socket) - socket_t lookup_sock = ua->poll_fds[i].fd; - node = socket_array_get_by_sock(ua->sockets, lookup_sock); + node = socket_array_get_by_sock(ua->sockets, fd); } if (!node) continue; // Socket may have been removed - /* Снапшот полей — коллбэки могут реаллоцировать ua->sockets (node станет висячим) */ - int local_fd = node->fd; - socket_t local_sock = node->sock; - int local_type = node->type; - void* local_ud = node->user_data; - socket_callback_t local_except = node->except_cbk; - socket_callback_t local_read = node->read_cbk; - socket_callback_t local_write = node->write_cbk; - socket_t_callback_t local_read_sock = node->read_cbk_sock; - socket_t_callback_t local_write_sock = node->write_cbk_sock; + /* Коллбэки могут удалить/добавить сокет (realloc массива) — перед + * каждым последующим коллбэком пере-валидируем node по fd. */ + int cb_called = 0; /* Read readiness BEFORE error — avoid losing data on combined IN+ERR/HUP events */ if (ua->poll_fds[i].revents & POLLIN) { - if (local_type == SOCKET_NODE_TYPE_SOCK) { - if (local_read_sock) { - local_read_sock(local_sock, local_ud); + if (node->type == SOCKET_NODE_TYPE_SOCK) { + if (node->read_cbk_sock) { + node->read_cbk_sock(node->sock, node->user_data); } } else { - if (local_read) { - local_read(local_fd, local_ud); + if (node->read_cbk) { + node->read_cbk(node->fd, node->user_data); } } + cb_called++; } /* Write readiness BEFORE error — flush pending writes before handling HUP */ if (ua->poll_fds[i].revents & POLLOUT) { - if (local_type == SOCKET_NODE_TYPE_SOCK) { - if (local_write_sock) { - local_write_sock(local_sock, local_ud); + if (cb_called) { + node = socket_array_get(ua->sockets, fd); + if (!node) node = socket_array_get_by_sock(ua->sockets, fd); + if (!node) continue; + } + if (node->type == SOCKET_NODE_TYPE_SOCK) { + if (node->write_cbk_sock) { + node->write_cbk_sock(node->sock, node->user_data); } } else { - if (local_write) { - local_write(local_fd, local_ud); + if (node->write_cbk) { + node->write_cbk(node->fd, node->user_data); } } + cb_called++; } /* Check for error conditions LAST — I/O handlers drain/process data first */ if (ua->poll_fds[i].revents & (POLLERR | POLLHUP | POLLNVAL)) { + if (cb_called) { + node = socket_array_get(ua->sockets, fd); + if (!node) node = socket_array_get_by_sock(ua->sockets, fd); + if (!node) continue; + } /* Treat as exceptional condition */ - if (local_except) { - local_except(local_fd, local_ud); + if (node->except_cbk) { + node->except_cbk(node->fd, node->user_data); } + cb_called++; } /* Exceptional data (out-of-band) */ if (ua->poll_fds[i].revents & POLLPRI) { - if (local_except) { - local_except(local_fd, local_ud); + if (cb_called) { + node = socket_array_get(ua->sockets, fd); + if (!node) node = socket_array_get_by_sock(ua->sockets, fd); + if (!node) continue; + } + if (node->except_cbk) { + node->except_cbk(node->fd, node->user_data); } + cb_called++; } } }