/* SPDX-License-Identifier: GPL-3.0-or-later * Copyright © 2026 The TokTok team. */ #include "os_event.h" #include #include #include #include #ifdef _WIN32 #include #include #else #include #endif /* WIN32 */ #ifdef EV_USE_EPOLL #include #elif defined(ESP_PLATFORM) #include #elif !defined(_WIN32) #include #endif /* EV_USE_EPOLL */ #include "ccompat.h" #include "logger.h" #include "mem.h" #ifdef _WIN32 // Using the undocumented AFD driver directly (IOCTL_AFD_POLL). // This bypasses the 64-handle limit of WaitForMultipleEvents and the bugs of WSAPoll. // It provides true Level-Triggered notifications similar to poll/epoll. #define IOCTL_AFD_POLL 0x00012024 typedef struct AFD_Poll_Handle_Info { void *_Nullable handle; unsigned long events; uint32_t status; } AFD_Poll_Handle_Info; typedef struct AFD_Poll_Info { int64_t timeout; unsigned long number_of_handles; unsigned long exclusive; AFD_Poll_Handle_Info handles[1]; } AFD_Poll_Info; typedef enum Afd_Poll { AFD_POLL_RECEIVE = 0x0001, AFD_POLL_RECEIVE_EXPEDITED = 0x0002, AFD_POLL_SEND = 0x0004, AFD_POLL_DISCONNECT = 0x0008, AFD_POLL_ABORT = 0x0010, AFD_POLL_LOCAL_CLOSE = 0x0020, AFD_POLL_ACCEPT = 0x0080, AFD_POLL_CONNECT_FAIL = 0x0100, } Afd_Poll; #endif /* _WIN32 */ typedef struct OS_Ev_Registration { Socket sock; Ev_Events events; void *_Nullable data; } OS_Ev_Registration; typedef struct OS_Ev { const Memory *_Nonnull mem; const Logger *_Nonnull log; OS_Ev_Registration **_Nonnull regs; uint32_t regs_count; uint32_t regs_capacity; #ifdef EV_USE_EPOLL int epoll_fd; struct epoll_event *events; uint32_t events_capacity; #elif defined(_WIN32) AFD_Poll_Handle_Info *h_events; uint32_t h_events_capacity; SOCKET helper_sock; #else struct pollfd *pfds; uint32_t pfds_count; #endif /* EV_USE_EPOLL */ } OS_Ev; static OS_Ev_Registration *_Nullable os_ev_prepare_add(OS_Ev *_Nonnull os_ev, Socket sock, Ev_Events events, void *_Nullable data) { for (uint32_t i = 0; i < os_ev->regs_count; ++i) { if (net_socket_to_native(os_ev->regs[i]->sock) == net_socket_to_native(sock)) { return nullptr; } } if (os_ev->regs_count == os_ev->regs_capacity) { const uint32_t new_capacity = os_ev->regs_capacity == 0 ? 8 : os_ev->regs_capacity * 2; OS_Ev_Registration **new_regs = (OS_Ev_Registration **)mem_vrealloc(os_ev->mem, os_ev->regs, new_capacity, sizeof(OS_Ev_Registration *)); if (new_regs == nullptr) { return nullptr; } os_ev->regs = new_regs; os_ev->regs_capacity = new_capacity; #ifdef _WIN32 AFD_Poll_Handle_Info *new_afd = (AFD_Poll_Handle_Info *)mem_vrealloc( os_ev->mem, os_ev->h_events, new_capacity, sizeof(AFD_Poll_Handle_Info)); if (new_afd == nullptr) { return nullptr; } os_ev->h_events = new_afd; os_ev->h_events_capacity = new_capacity; #endif /* _WIN32 */ } OS_Ev_Registration *reg = (OS_Ev_Registration *)mem_alloc(os_ev->mem, sizeof(OS_Ev_Registration)); if (reg == nullptr) { return nullptr; } reg->sock = sock; reg->events = events; reg->data = data; return reg; } /** @brief Adds a registration from os_ev_prepare_add once the backend has accepted it. */ static void os_ev_commit_add(OS_Ev *_Nonnull os_ev, OS_Ev_Registration *_Nonnull reg) { os_ev->regs[os_ev->regs_count] = reg; ++os_ev->regs_count; } #ifdef EV_USE_EPOLL static uint32_t events_to_epoll(Ev_Events events) { uint32_t e = 0; if ((events & EV_READ) != 0) { e |= EPOLLIN; } if ((events & EV_WRITE) != 0) { e |= EPOLLOUT; } return e; } static Ev_Events epoll_to_events(uint32_t e) { Ev_Events events = 0; if ((e & EPOLLIN) != 0) { events |= EV_READ; } if ((e & EPOLLOUT) != 0) { events |= EV_WRITE; } if ((e & (EPOLLERR | EPOLLHUP | EPOLLRDHUP)) != 0) { events |= EV_ERROR; } return events; } static bool os_ev_add(void *_Nonnull self, Socket sock, Ev_Events events, void *_Nullable data) { OS_Ev *os_ev = (OS_Ev *)self; OS_Ev_Registration *reg = os_ev_prepare_add(os_ev, sock, events, data); if (reg == nullptr) { return false; } struct epoll_event ep_ev; ep_ev.events = events_to_epoll(events); ep_ev.data.ptr = reg; if (epoll_ctl(os_ev->epoll_fd, EPOLL_CTL_ADD, net_socket_to_native(sock), &ep_ev) == -1) { Net_Strerror error_buf; LOGGER_ERROR(os_ev->log, "epoll_ctl(ADD) failed: %s", net_strerror(errno, &error_buf)); mem_delete(os_ev->mem, reg); return false; } LOGGER_DEBUG(os_ev->log, "Added socket %d to epoll (events: %u)", net_socket_to_native(sock), events); os_ev_commit_add(os_ev, reg); return true; } static bool os_ev_mod(void *_Nonnull self, Socket sock, Ev_Events events, void *_Nullable data) { OS_Ev *os_ev = (OS_Ev *)self; for (uint32_t i = 0; i < os_ev->regs_count; ++i) { if (net_socket_to_native(os_ev->regs[i]->sock) == net_socket_to_native(sock)) { OS_Ev_Registration *reg = os_ev->regs[i]; reg->events = events; reg->data = data; struct epoll_event ep_ev; ep_ev.events = events_to_epoll(events); ep_ev.data.ptr = reg; if (epoll_ctl(os_ev->epoll_fd, EPOLL_CTL_MOD, net_socket_to_native(sock), &ep_ev) == -1) { Net_Strerror error_buf; LOGGER_ERROR(os_ev->log, "epoll_ctl(MOD) failed: %s", net_strerror(errno, &error_buf)); return false; } LOGGER_DEBUG(os_ev->log, "Modified socket %d in epoll (events: %u)", net_socket_to_native(sock), events); return true; } } return false; } static bool os_ev_del(void *_Nonnull self, Socket sock) { OS_Ev *os_ev = (OS_Ev *)self; for (uint32_t i = 0; i < os_ev->regs_count; ++i) { if (net_socket_to_native(os_ev->regs[i]->sock) == net_socket_to_native(sock)) { if (epoll_ctl(os_ev->epoll_fd, EPOLL_CTL_DEL, net_socket_to_native(sock), nullptr) == -1) { Net_Strerror error_buf; LOGGER_ERROR(os_ev->log, "epoll_ctl(DEL) failed: %s", net_strerror(errno, &error_buf)); } LOGGER_DEBUG(os_ev->log, "Removed socket %d from epoll", net_socket_to_native(sock)); mem_delete(os_ev->mem, os_ev->regs[i]); os_ev->regs[i] = os_ev->regs[os_ev->regs_count - 1]; --os_ev->regs_count; return true; } } return false; } static int32_t os_ev_run(void *_Nonnull self, Ev_Result *_Nonnull results, uint32_t max_results, int32_t timeout_ms) { OS_Ev *os_ev = (OS_Ev *)self; if (os_ev->events_capacity < max_results) { struct epoll_event *new_events = (struct epoll_event *)mem_vrealloc(os_ev->mem, os_ev->events, max_results, sizeof(struct epoll_event)); if (new_events == nullptr) { return -1; } os_ev->events = new_events; os_ev->events_capacity = max_results; } const int n = epoll_wait(os_ev->epoll_fd, os_ev->events, (int)max_results, (int)timeout_ms); if (n < 0) { if (errno == EINTR) { return 0; } Net_Strerror error_buf; LOGGER_ERROR(os_ev->log, "epoll_wait failed: %s", net_strerror(errno, &error_buf)); return -1; } if (n == 0) { return 0; } uint32_t count = 0; for (int i = 0; i < n; ++i) { OS_Ev_Registration *reg = (OS_Ev_Registration *)os_ev->events[i].data.ptr; if (reg != nullptr) { results[count].sock = reg->sock; results[count].events = epoll_to_events(os_ev->events[i].events); results[count].data = reg->data; ++count; } } return (int32_t)count; } #elif defined(_WIN32) static uint32_t events_to_afd(Ev_Events events) { uint32_t e = AFD_POLL_ABORT | AFD_POLL_LOCAL_CLOSE | AFD_POLL_CONNECT_FAIL; // Always monitor errors if (events & EV_READ) { e |= AFD_POLL_RECEIVE | AFD_POLL_RECEIVE_EXPEDITED | AFD_POLL_ACCEPT | AFD_POLL_DISCONNECT; } if (events & EV_WRITE) { e |= AFD_POLL_SEND; } return e; } static Ev_Events afd_to_events(uint32_t e) { Ev_Events events = 0; if (e & (AFD_POLL_RECEIVE | AFD_POLL_RECEIVE_EXPEDITED | AFD_POLL_ACCEPT | AFD_POLL_DISCONNECT)) { events |= EV_READ; } if (e & AFD_POLL_SEND) { events |= EV_WRITE; } if (e & (AFD_POLL_ABORT | AFD_POLL_LOCAL_CLOSE | AFD_POLL_CONNECT_FAIL)) { events |= EV_ERROR; } return events; } static bool os_ev_add(void *_Nonnull self, Socket sock, Ev_Events events, void *_Nullable data) { OS_Ev *os_ev = (OS_Ev *)self; OS_Ev_Registration *reg = os_ev_prepare_add(os_ev, sock, events, data); if (reg == nullptr) { return false; } AFD_Poll_Handle_Info *afd_handles = os_ev->h_events; afd_handles[os_ev->regs_count].handle = (HANDLE)(uintptr_t)net_socket_to_native(sock); afd_handles[os_ev->regs_count].events = events_to_afd(events); afd_handles[os_ev->regs_count].status = 0; LOGGER_DEBUG(os_ev->log, "Added socket %d to AFD (events: %u)", net_socket_to_native(sock), events); os_ev_commit_add(os_ev, reg); return true; } static bool os_ev_mod(void *_Nonnull self, Socket sock, Ev_Events events, void *_Nullable data) { OS_Ev *os_ev = (OS_Ev *)self; for (uint32_t i = 0; i < os_ev->regs_count; ++i) { if (net_socket_to_native(os_ev->regs[i]->sock) == net_socket_to_native(sock)) { os_ev->regs[i]->events = events; os_ev->regs[i]->data = data; AFD_Poll_Handle_Info *afd_handles = os_ev->h_events; afd_handles[i].events = events_to_afd(events); afd_handles[i].status = 0; LOGGER_DEBUG(os_ev->log, "Modified socket %d in AFD (events: %u)", net_socket_to_native(sock), events); return true; } } return false; } static bool os_ev_del(void *_Nonnull self, Socket sock) { OS_Ev *os_ev = (OS_Ev *)self; for (uint32_t i = 0; i < os_ev->regs_count; ++i) { if (net_socket_to_native(os_ev->regs[i]->sock) == net_socket_to_native(sock)) { mem_delete(os_ev->mem, os_ev->regs[i]); os_ev->regs[i] = os_ev->regs[os_ev->regs_count - 1]; AFD_Poll_Handle_Info *afd_handles = os_ev->h_events; afd_handles[i] = afd_handles[os_ev->regs_count - 1]; --os_ev->regs_count; LOGGER_DEBUG(os_ev->log, "Removed socket %d from AFD", net_socket_to_native(sock)); return true; } } return false; } static int32_t os_ev_run(void *_Nonnull self, Ev_Result *_Nonnull results, uint32_t max_results, int32_t timeout_ms) { OS_Ev *os_ev = (OS_Ev *)self; if (os_ev->regs_count == 0) { if (timeout_ms > 0) { Sleep(timeout_ms); } return 0; } const size_t info_size = sizeof(AFD_Poll_Info) + (os_ev->regs_count - 1) * sizeof(AFD_Poll_Handle_Info); AFD_Poll_Info *info = (AFD_Poll_Info *)HeapAlloc(GetProcessHeap(), 0, info_size); if (info == nullptr) { return -1; } info->timeout = timeout_ms < 0 ? INT64_MAX : (int64_t)timeout_ms * -10000; info->number_of_handles = os_ev->regs_count; info->exclusive = 0; memcpy(info->handles, os_ev->h_events, os_ev->regs_count * sizeof(AFD_Poll_Handle_Info)); DWORD bytes_ret = 0; if (!DeviceIoControl((HANDLE)os_ev->helper_sock, IOCTL_AFD_POLL, info, (DWORD)info_size, info, (DWORD)info_size, &bytes_ret, NULL)) { const int err = net_error(); HeapFree(GetProcessHeap(), 0, info); if (err == WAIT_TIMEOUT) { return 0; } Net_Strerror error_buf; LOGGER_ERROR(os_ev->log, "AFD_POLL failed: %s", net_strerror(err, &error_buf)); return -1; } uint32_t count = 0; for (uint32_t i = 0; i < info->number_of_handles && count < max_results; ++i) { if (info->handles[i].events != 0) { results[count].sock = os_ev->regs[i]->sock; results[count].data = os_ev->regs[i]->data; results[count].events = afd_to_events(info->handles[i].events); ++count; } } HeapFree(GetProcessHeap(), 0, info); return (int32_t)count; } #else // POLL static bool os_ev_add(void *_Nonnull self, Socket sock, Ev_Events events, void *_Nullable data) { OS_Ev *os_ev = (OS_Ev *)self; OS_Ev_Registration *reg = os_ev_prepare_add(os_ev, sock, events, data); if (reg == nullptr) { return false; } LOGGER_DEBUG(os_ev->log, "Added socket %d to poll (events: %u)", net_socket_to_native(sock), events); os_ev_commit_add(os_ev, reg); return true; } static bool os_ev_mod(void *_Nonnull self, Socket sock, Ev_Events events, void *_Nullable data) { OS_Ev *os_ev = (OS_Ev *)self; for (uint32_t i = 0; i < os_ev->regs_count; ++i) { if (net_socket_to_native(os_ev->regs[i]->sock) == net_socket_to_native(sock)) { os_ev->regs[i]->events = events; os_ev->regs[i]->data = data; LOGGER_DEBUG(os_ev->log, "Modified socket %d in poll (events: %u)", net_socket_to_native(sock), events); return true; } } return false; } static bool os_ev_del(void *_Nonnull self, Socket sock) { OS_Ev *os_ev = (OS_Ev *)self; for (uint32_t i = 0; i < os_ev->regs_count; ++i) { if (net_socket_to_native(os_ev->regs[i]->sock) == net_socket_to_native(sock)) { mem_delete(os_ev->mem, os_ev->regs[i]); os_ev->regs[i] = os_ev->regs[os_ev->regs_count - 1]; --os_ev->regs_count; LOGGER_DEBUG(os_ev->log, "Removed socket %d from poll", net_socket_to_native(sock)); return true; } } return false; } static int32_t os_ev_run(void *_Nonnull self, Ev_Result *_Nonnull results, uint32_t max_results, int32_t timeout_ms) { OS_Ev *os_ev = (OS_Ev *)self; if (os_ev->pfds_count < os_ev->regs_count) { struct pollfd *new_pfds = (struct pollfd *)mem_vrealloc(os_ev->mem, os_ev->pfds, os_ev->regs_count, sizeof(struct pollfd)); if (new_pfds == nullptr) { return -1; } os_ev->pfds = new_pfds; os_ev->pfds_count = os_ev->regs_count; } for (uint32_t i = 0; i < os_ev->regs_count; ++i) { os_ev->pfds[i].fd = net_socket_to_native(os_ev->regs[i]->sock); os_ev->pfds[i].events = 0; if ((os_ev->regs[i]->events & EV_READ) != 0) { os_ev->pfds[i].events |= POLLIN; } if ((os_ev->regs[i]->events & EV_WRITE) != 0) { os_ev->pfds[i].events |= POLLOUT; } os_ev->pfds[i].revents = 0; } const int n = poll(os_ev->pfds, (nfds_t)os_ev->regs_count, (int)timeout_ms); if (n < 0) { if (errno == EINTR) { return 0; } Net_Strerror error_buf; LOGGER_ERROR(os_ev->log, "poll failed: %s", net_strerror(errno, &error_buf)); return -1; } if (n == 0) { return 0; } uint32_t count = 0; for (uint32_t i = 0; i < os_ev->regs_count && count < max_results; ++i) { if (os_ev->pfds[i].revents == 0) { continue; } results[count].sock = os_ev->regs[i]->sock; results[count].events = 0; results[count].data = os_ev->regs[i]->data; if ((os_ev->pfds[i].revents & POLLIN) != 0) { results[count].events |= EV_READ; } if ((os_ev->pfds[i].revents & POLLOUT) != 0) { results[count].events |= EV_WRITE; } if ((os_ev->pfds[i].revents & (POLLERR | POLLHUP | POLLNVAL)) != 0) { results[count].events |= EV_ERROR; } ++count; } return (int32_t)count; } #endif /* EV_USE_EPOLL */ static void os_ev_kill(Ev *_Nullable ev) { if (ev == nullptr) { return; } OS_Ev *os_ev = (OS_Ev *)ev->user_data; const Memory *mem = os_ev->mem; #ifdef EV_USE_EPOLL close(os_ev->epoll_fd); mem_delete(mem, os_ev->events); #elif defined(_WIN32) closesocket(os_ev->helper_sock); mem_delete(mem, os_ev->h_events); #else mem_delete(mem, os_ev->pfds); #endif /* EV_USE_EPOLL */ for (uint32_t i = 0; i < os_ev->regs_count; ++i) { mem_delete(mem, os_ev->regs[i]); } mem_delete(mem, os_ev->regs); mem_delete(mem, os_ev); mem_delete(mem, ev); } static const Ev_Funcs os_ev_funcs = { os_ev_add, os_ev_mod, os_ev_del, os_ev_run, os_ev_kill, }; Ev *os_event_new(const Memory *mem, const Logger *log) { OS_Ev *os_ev = (OS_Ev *)mem_alloc(mem, sizeof(OS_Ev)); if (os_ev == nullptr) { return nullptr; } os_ev->mem = mem; os_ev->log = log; #ifdef EV_USE_EPOLL os_ev->epoll_fd = epoll_create1(EPOLL_CLOEXEC); if (os_ev->epoll_fd == -1) { Net_Strerror error_buf; LOGGER_ERROR(log, "Failed to create epoll fd: %s", net_strerror(errno, &error_buf)); mem_delete(mem, os_ev); return nullptr; } #elif defined(_WIN32) // Create a dummy socket for AFD communication os_ev->helper_sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); if (os_ev->helper_sock == INVALID_SOCKET) { Net_Strerror error_buf; LOGGER_ERROR(log, "Failed to create helper socket: %s", net_strerror(net_error(), &error_buf)); mem_delete(mem, os_ev); return nullptr; } #endif /* EV_USE_EPOLL */ Ev *ev = (Ev *)mem_alloc(mem, sizeof(Ev)); if (ev == nullptr) { #ifdef EV_USE_EPOLL close(os_ev->epoll_fd); #elif defined(_WIN32) closesocket(os_ev->helper_sock); #endif /* EV_USE_EPOLL */ mem_delete(mem, os_ev); return nullptr; } ev->funcs = &os_ev_funcs; ev->user_data = os_ev; LOGGER_DEBUG(log, "Created new os_event loop"); return ev; }