/********************************************************************************************************* * ------------------------------------------------------------------------------------------------------ * file description * ------------------------------------------------------------------------------------------------------ * \file coroutine_io.c * \unit coroutine * \brief High-throughput asynchronous I/O multiplexer and cooperative socket API * \author Lamdonn * \version v0.2.0 * \license GPL-2.0 * \copyright Copyright (C) 2025 Lamdonn. ********************************************************************************************************/ #if defined(_WIN32) && !defined(_WIN32_WINNT) #define _WIN32_WINNT 0x0600 #endif #include "coroutine_io.h" #include "coroutine_hook.h" #include #include #include /* ======================================================================================================= * Quad-Tier Platform Detection & Header Dispatching * ======================================================================================================= */ #if defined(COROUTINE_IO_BAREMETAL) #define CO_IO_PLATFORM_BAREMETAL 1 #elif defined(_WIN32) #define CO_IO_PLATFORM_WIN32 1 #elif defined(__linux__) || defined(__unix__) || defined(__APPLE__) || \ defined(__QNX__) || defined(__QNXNTO__) || defined(__ANDROID__) || \ defined(__VXWORKS__) || defined(__NuttX__) #define CO_IO_PLATFORM_POSIX 1 #elif defined(COROUTINE_IO_LWIP) || defined(LWIP_SOCKET) #define CO_IO_PLATFORM_LWIP 1 #else #define CO_IO_PLATFORM_BAREMETAL 1 #endif #if defined(CO_IO_PLATFORM_WIN32) #include #include #include typedef int socklen_t; typedef WSAPOLLFD PollFd_t; #elif defined(CO_IO_PLATFORM_POSIX) #include #include #include #include #include #include #include #include typedef struct pollfd PollFd_t; #elif defined(CO_IO_PLATFORM_LWIP) #include "lwip/sockets.h" #include "lwip/netdb.h" #include typedef struct pollfd PollFd_t; #elif defined(CO_IO_PLATFORM_BAREMETAL) /* Pure bare-metal mode: zero external socket headers */ #ifndef POLLIN #define POLLIN (0x0001) #endif #ifndef POLLOUT #define POLLOUT (0x0004) #endif #ifndef POLLERR #define POLLERR (0x0008) #endif #ifndef POLLHUP #define POLLHUP (0x0010) #endif #ifndef POLLNVAL #define POLLNVAL (0x0020) #endif typedef CoIoPollFd_t PollFd_t; static CoIoBaremetalPoller_t s_baremetal_poller = NULL; void CoIo_SetBaremetalPoller(CoIoBaremetalPoller_t poller) { s_baremetal_poller = poller; } #endif /* ======================================================================================================= * Small Buffer Optimization (SBO) Hybrid Poller Architecture * ======================================================================================================= */ #define CO_IO_STATIC_CAPACITY (64) typedef struct { int fd; uint32_t events; CoTask_t task; int readyEvents; } CoIoWatch; typedef struct { CoIoWatch staticWatches[CO_IO_STATIC_CAPACITY]; /**< Static fast cache (0 heap allocations) */ CoIoWatch *dynamicWatches; /**< Overflow dynamic buffer (heap, auto-expanding) */ uint32_t capacity; /**< Current total capacity (starts at CO_IO_STATIC_CAPACITY) */ uint32_t count; /**< Current number of active watched descriptors */ } CoSchedulerPoller; static CoSchedulerPoller s_pollers[COROUTINE_SCHEDULER_NUMBER] = {0}; static int s_io_inited = 0; /* Forward declarations */ static void CoIo_IdleHook(uint32_t schedulerId, uint32_t idleUs); static void CoIo_WakeupHook(uint32_t schedulerId); static inline CoIoWatch *CoIo_GetWatch(CoSchedulerPoller *poller, uint32_t index) { if (poller->dynamicWatches != NULL) { return &poller->dynamicWatches[index]; } return &poller->staticWatches[index]; } static int CoIo_EnsureCapacity(CoSchedulerPoller *poller) { if (poller->capacity == 0) { poller->capacity = CO_IO_STATIC_CAPACITY; } if (poller->count < poller->capacity) { return 0; /* Existing capacity has free slots */ } /* Verify if the coroutine system has a configured dynamic memory allocator */ if (!Coroutine_HasMemHook()) { return -1; /* Pure static mode: safely reject expansion beyond static limit */ } /* Needs expansion: double capacity */ uint32_t newCap = poller->capacity * 2; CoIoWatch *newBuf = (CoIoWatch *)Coroutine_Malloc(sizeof(CoIoWatch) * newCap); if (newBuf == NULL) return -1; memset(newBuf, 0, sizeof(CoIoWatch) * newCap); if (poller->dynamicWatches == NULL) { /* First overflow: promote from static cache to dynamic heap */ memcpy(newBuf, poller->staticWatches, sizeof(CoIoWatch) * poller->capacity); } else { /* Subsequent expansion: copy from previous dynamic buffer and free old */ memcpy(newBuf, poller->dynamicWatches, sizeof(CoIoWatch) * poller->capacity); Coroutine_Free(poller->dynamicWatches); } poller->dynamicWatches = newBuf; poller->capacity = newCap; return 0; } static void CoIo_TryShrink(CoSchedulerPoller *poller) { /* If all descriptors have finished and poller is completely empty, revert to static */ if (poller->dynamicWatches != NULL && poller->count == 0) { Coroutine_Free(poller->dynamicWatches); poller->dynamicWatches = NULL; poller->capacity = CO_IO_STATIC_CAPACITY; } } /* ======================================================================================================= * Subsystem Management * ======================================================================================================= */ int CoIo_Init(void) { if (s_io_inited) return 0; #if defined(CO_IO_PLATFORM_WIN32) WSADATA wsaData; int ret = WSAStartup(MAKEWORD(2, 2), &wsaData); if (ret != 0) return -1; #endif s_io_inited = 1; return 0; } void CoIo_Deinit(void) { if (!s_io_inited) return; for (uint32_t i = 0; i < COROUTINE_SCHEDULER_NUMBER; i++) { if (s_pollers[i].dynamicWatches != NULL) { Coroutine_Free(s_pollers[i].dynamicWatches); s_pollers[i].dynamicWatches = NULL; } s_pollers[i].capacity = CO_IO_STATIC_CAPACITY; s_pollers[i].count = 0; } #if defined(CO_IO_PLATFORM_WIN32) WSACleanup(); #endif s_io_inited = 0; } int CoIo_AttachScheduler(uint32_t schedulerId) { if (schedulerId >= COROUTINE_SCHEDULER_NUMBER) { return COROUTINE_E_INVALID_PARAMETER; } CoIo_Init(); s_pollers[schedulerId].capacity = CO_IO_STATIC_CAPACITY; s_pollers[schedulerId].count = 0; s_pollers[schedulerId].dynamicWatches = NULL; CoScheduler_SetIdleHook(schedulerId, CoIo_IdleHook); CoScheduler_SetWakeupHook(schedulerId, CoIo_WakeupHook); return COROUTINE_E_OK; } int CoIo_DetachScheduler(uint32_t schedulerId) { if (schedulerId >= COROUTINE_SCHEDULER_NUMBER) { return COROUTINE_E_INVALID_PARAMETER; } CoScheduler_SetIdleHook(schedulerId, NULL); CoScheduler_SetWakeupHook(schedulerId, NULL); if (s_pollers[schedulerId].dynamicWatches != NULL) { Coroutine_Free(s_pollers[schedulerId].dynamicWatches); s_pollers[schedulerId].dynamicWatches = NULL; } s_pollers[schedulerId].capacity = CO_IO_STATIC_CAPACITY; s_pollers[schedulerId].count = 0; return COROUTINE_E_OK; } int CoIo_SetNonblocking(int fd) { #if defined(CO_IO_PLATFORM_WIN32) u_long mode = 1; return ioctlsocket((SOCKET)fd, FIONBIO, &mode); #elif defined(CO_IO_PLATFORM_POSIX) int flags = fcntl(fd, F_GETFL, 0); if (flags < 0) return -1; return fcntl(fd, F_SETFL, flags | O_NONBLOCK); #elif defined(CO_IO_PLATFORM_LWIP) int flags = lwip_fcntl(fd, F_GETFL, 0); if (flags < 0) return -1; return lwip_fcntl(fd, F_SETFL, flags | O_NONBLOCK); #else (void)fd; return 0; #endif } /* ======================================================================================================= * Polling Engine & Reactor Idle Hook (WSAPoll / poll based, no FD limits) * ======================================================================================================= */ static void CoIo_IdleHook(uint32_t schedulerId, uint32_t idleUs) { if (schedulerId >= COROUTINE_SCHEDULER_NUMBER) return; CoSchedulerPoller *poller = &s_pollers[schedulerId]; /* If no I/O descriptors registered, fall back to standard low-power platform sleep hook */ if (poller->count == 0) { CoHook_Baremetal_Idle(schedulerId, idleUs); return; } /* Allocate pollfd array using Small Buffer Optimization */ PollFd_t staticPfds[CO_IO_STATIC_CAPACITY]; uint32_t staticMap[CO_IO_STATIC_CAPACITY]; PollFd_t *pfds = staticPfds; uint32_t *mapIndex = staticMap; uint32_t heapAllocated = 0; if (poller->count > CO_IO_STATIC_CAPACITY) { pfds = (PollFd_t *)Coroutine_Malloc(sizeof(PollFd_t) * poller->count); mapIndex = (uint32_t *)Coroutine_Malloc(sizeof(uint32_t) * poller->count); if (pfds == NULL || mapIndex == NULL) { if (pfds) Coroutine_Free(pfds); if (mapIndex) Coroutine_Free(mapIndex); CoHook_Baremetal_Idle(schedulerId, idleUs); return; } heapAllocated = 1; } uint32_t nfds = 0; for (uint32_t i = 0; i < poller->capacity && nfds < poller->count; i++) { CoIoWatch *w = CoIo_GetWatch(poller, i); if (w->task != NULL) { pfds[nfds].fd = w->fd; pfds[nfds].events = 0; if (w->events & CO_IO_READ) pfds[nfds].events |= POLLIN; if (w->events & CO_IO_WRITE) pfds[nfds].events |= POLLOUT; pfds[nfds].revents = 0; mapIndex[nfds] = i; nfds++; } } if (nfds == 0) { if (heapAllocated) { Coroutine_Free(pfds); Coroutine_Free(mapIndex); } CoHook_Baremetal_Idle(schedulerId, idleUs); return; } int timeoutMs = (idleUs == 0) ? -1 : (int)((idleUs + 999) / 1000); #if defined(CO_IO_PLATFORM_WIN32) int n = WSAPoll(pfds, nfds, timeoutMs); #elif defined(CO_IO_PLATFORM_POSIX) int n = poll(pfds, nfds, timeoutMs); #elif defined(CO_IO_PLATFORM_LWIP) int n = lwip_poll(pfds, nfds, timeoutMs); #elif defined(CO_IO_PLATFORM_BAREMETAL) int n = 0; if (s_baremetal_poller != NULL) { n = s_baremetal_poller((CoIoPollFd_t *)pfds, nfds, timeoutMs); } else { CoHook_Baremetal_Idle(schedulerId, idleUs); } #endif if (n > 0) { for (uint32_t i = 0; i < nfds; i++) { if (pfds[i].revents != 0) { uint32_t idx = mapIndex[i]; CoIoWatch *w = CoIo_GetWatch(poller, idx); int rev = 0; if (pfds[i].revents & POLLIN) rev |= CO_IO_READ; if (pfds[i].revents & POLLOUT) rev |= CO_IO_WRITE; if (pfds[i].revents & (POLLERR | POLLHUP | POLLNVAL)) rev |= CO_IO_ERROR; w->readyEvents = rev; CoTask_Wakeup(w->task); } } } if (heapAllocated) { Coroutine_Free(pfds); Coroutine_Free(mapIndex); } } static void CoIo_WakeupHook(uint32_t schedulerId) { (void)schedulerId; } int CoTask_WaitFd(int fd, uint32_t events, uint32_t timeoutMs) { int schedId = CoTask_SchedulerId(); if (schedId < 0 || schedId >= (int)COROUTINE_SCHEDULER_NUMBER) { return CO_IO_INVALID; } CoSchedulerPoller *poller = &s_pollers[schedId]; /* Ensure poller has enough capacity (auto expands from static to dynamic heap if needed) */ if (CoIo_EnsureCapacity(poller) != 0) { return CO_IO_INVALID; } /* Find an available watch slot within current capacity */ int slot = -1; for (uint32_t i = 0; i < poller->capacity; i++) { CoIoWatch *w = CoIo_GetWatch(poller, i); if (w->task == NULL) { slot = (int)i; break; } } if (slot < 0) { return CO_IO_INVALID; } CoIoWatch *watch = CoIo_GetWatch(poller, (uint32_t)slot); watch->fd = fd; watch->events = events; watch->task = CoTask_Self(); watch->readyEvents = 0; poller->count++; /* Suspend coroutine until polled event or timeout */ CoTask_Wait(.ms = timeoutMs); /* Retrieve ready events and clean up slot (re-evaluate watch pointer as poller might have realloc'd) */ CoIoWatch *w = CoIo_GetWatch(poller, (uint32_t)slot); int res = w->readyEvents; w->task = NULL; w->fd = -1; w->events = 0; w->readyEvents = 0; if (poller->count > 0) poller->count--; /* Shrink back to static cache if all overflow connections closed */ CoIo_TryShrink(poller); if (res != 0) { return res; } return CO_IO_TIMEOUT; } /* ======================================================================================================= * Cooperative Non-Blocking Socket Operations * ======================================================================================================= */ int CoSocket_Create(int domain, int type, int protocol) { CoIo_Init(); #if defined(CO_IO_PLATFORM_WIN32) SOCKET s = socket(domain, type, protocol); if (s == INVALID_SOCKET) return -1; CoIo_SetNonblocking((int)s); return (int)s; #elif defined(CO_IO_PLATFORM_POSIX) int s = socket(domain, type, protocol); if (s < 0) return -1; CoIo_SetNonblocking(s); return s; #elif defined(CO_IO_PLATFORM_LWIP) int s = lwip_socket(domain, type, protocol); if (s < 0) return -1; CoIo_SetNonblocking(s); return s; #else (void)domain; (void)type; (void)protocol; return -1; #endif } int CoSocket_Connect(int fd, const void *addr, int addrlen, uint32_t timeoutMs) { #if defined(CO_IO_PLATFORM_WIN32) int ret = connect((SOCKET)fd, (const struct sockaddr *)addr, addrlen); if (ret == 0) return 0; int err = WSAGetLastError(); if (err != WSAEWOULDBLOCK && err != WSAEINPROGRESS) return -1; #elif defined(CO_IO_PLATFORM_POSIX) int ret = connect(fd, (const struct sockaddr *)addr, (socklen_t)addrlen); if (ret == 0) return 0; if (errno != EINPROGRESS && errno != EWOULDBLOCK) return -1; #elif defined(CO_IO_PLATFORM_LWIP) int ret = lwip_connect(fd, (const struct sockaddr *)addr, (socklen_t)addrlen); if (ret == 0) return 0; if (errno != EINPROGRESS && errno != EWOULDBLOCK) return -1; #else (void)fd; (void)addr; (void)addrlen; (void)timeoutMs; return -1; #endif int waitRes = CoTask_WaitFd(fd, CO_IO_WRITE, timeoutMs); if (waitRes <= 0) return waitRes; /* Verify connect success status via SO_ERROR */ int sockErr = 0; #if defined(CO_IO_PLATFORM_WIN32) socklen_t optLen = sizeof(sockErr); getsockopt((SOCKET)fd, SOL_SOCKET, SO_ERROR, (char *)&sockErr, (int *)&optLen); #elif defined(CO_IO_PLATFORM_POSIX) socklen_t optLen = sizeof(sockErr); getsockopt(fd, SOL_SOCKET, SO_ERROR, &sockErr, &optLen); #elif defined(CO_IO_PLATFORM_LWIP) socklen_t optLen = sizeof(sockErr); lwip_getsockopt(fd, SOL_SOCKET, SO_ERROR, &sockErr, &optLen); #else sockErr = -1; #endif return (sockErr == 0) ? 0 : -1; } int CoSocket_Accept(int fd, void *addr, int *addrlen, uint32_t timeoutMs) { while (1) { #if defined(CO_IO_PLATFORM_WIN32) SOCKET client = accept((SOCKET)fd, (struct sockaddr *)addr, (int *)addrlen); if (client != INVALID_SOCKET) { CoIo_SetNonblocking((int)client); return (int)client; } int err = WSAGetLastError(); if (err != WSAEWOULDBLOCK) return -1; #elif defined(CO_IO_PLATFORM_POSIX) socklen_t slen = addrlen ? (socklen_t)*addrlen : 0; int client = accept(fd, (struct sockaddr *)addr, addrlen ? &slen : NULL); if (client >= 0) { if (addrlen) *addrlen = (int)slen; CoIo_SetNonblocking(client); return client; } if (errno != EAGAIN && errno != EWOULDBLOCK) return -1; #elif defined(CO_IO_PLATFORM_LWIP) socklen_t slen = addrlen ? (socklen_t)*addrlen : 0; int client = lwip_accept(fd, (struct sockaddr *)addr, addrlen ? &slen : NULL); if (client >= 0) { if (addrlen) *addrlen = (int)slen; CoIo_SetNonblocking(client); return client; } if (errno != EAGAIN && errno != EWOULDBLOCK) return -1; #else (void)fd; (void)addr; (void)addrlen; (void)timeoutMs; return -1; #endif int waitRes = CoTask_WaitFd(fd, CO_IO_READ, timeoutMs); if (waitRes <= 0) return waitRes; } } int CoSocket_Read(int fd, void *buf, size_t len, uint32_t timeoutMs) { while (1) { #if defined(CO_IO_PLATFORM_WIN32) int ret = recv((SOCKET)fd, (char *)buf, (int)len, 0); if (ret >= 0) return ret; int err = WSAGetLastError(); if (err != WSAEWOULDBLOCK) return -1; #elif defined(CO_IO_PLATFORM_POSIX) ssize_t ret = recv(fd, buf, len, 0); if (ret >= 0) return (int)ret; if (errno != EAGAIN && errno != EWOULDBLOCK) return -1; #elif defined(CO_IO_PLATFORM_LWIP) int ret = lwip_recv(fd, buf, len, 0); if (ret >= 0) return ret; if (errno != EAGAIN && errno != EWOULDBLOCK) return -1; #else (void)fd; (void)buf; (void)len; (void)timeoutMs; return -1; #endif int waitRes = CoTask_WaitFd(fd, CO_IO_READ, timeoutMs); if (waitRes <= 0) return waitRes; } } int CoSocket_Write(int fd, const void *buf, size_t len, uint32_t timeoutMs) { size_t total = 0; while (total < len) { #if defined(CO_IO_PLATFORM_WIN32) int ret = send((SOCKET)fd, (const char *)buf + total, (int)(len - total), 0); if (ret > 0) { total += ret; continue; } int err = WSAGetLastError(); if (err != WSAEWOULDBLOCK) return -1; #elif defined(CO_IO_PLATFORM_POSIX) ssize_t ret = send(fd, (const char *)buf + total, len - total, 0); if (ret > 0) { total += (size_t)ret; continue; } if (errno != EAGAIN && errno != EWOULDBLOCK) return -1; #elif defined(CO_IO_PLATFORM_LWIP) int ret = lwip_send(fd, (const char *)buf + total, len - total, 0); if (ret > 0) { total += (size_t)ret; continue; } if (errno != EAGAIN && errno != EWOULDBLOCK) return -1; #else (void)fd; (void)buf; (void)len; (void)timeoutMs; return -1; #endif int waitRes = CoTask_WaitFd(fd, CO_IO_WRITE, timeoutMs); if (waitRes <= 0) return waitRes; } return (int)total; } int CoSocket_Close(int fd) { #if defined(CO_IO_PLATFORM_WIN32) return closesocket((SOCKET)fd); #elif defined(CO_IO_PLATFORM_POSIX) return close(fd); #elif defined(CO_IO_PLATFORM_LWIP) return lwip_close(fd); #else (void)fd; return 0; #endif }