diff --git a/Makefile b/Makefile index f40c276d..d3be54a7 100644 --- a/Makefile +++ b/Makefile @@ -897,6 +897,7 @@ TEST_TARGETS = test_worker_affinity \ test_reactor_wake \ test_reactor_timer \ test_rx_epoll \ + test_reactor_accept \ test_reactor_drain_chunk \ test_latency_probe \ test_sstring \ @@ -931,6 +932,7 @@ TEST_TARGETS = test_worker_affinity \ test_webtransport \ test_webtransport_client_api \ test_http \ + test_http_uring_hammer \ test_http_stringify \ test_siphash \ test_mux \ @@ -1126,6 +1128,16 @@ test_rx_epoll: tests/test_rx_epoll.c src/sys/io/reactor.c src/sys/io/reactor_rx. $(CC) $(CFLAGS) -o $@ tests/test_rx_epoll.c -pthread ./$@ +# Multishot-accept contract test (issue #406): functional run on the default +# backend, the knob-off arm refusal, and the forced-epoll fallback. +test_reactor_accept: tests/test_reactor_accept.c src/sys/io/reactor.c src/sys/io/reactor_rx.h + $(CC) $(CFLAGS) -o $@ tests/test_reactor_accept.c -pthread + ./$@ + ./$@ off +ifeq ($(UNAME_S),Linux) + CWIST_REACTOR_BACKEND=epoll ./$@ +endif + # Cooperative-queuing correctness test (issue #25): CWIST_REACTOR_DRAIN_CHUNK # interleaves foreign-thread post draining into a big CQE batch instead of # only at the batch's end. See tests/bench_cooperative_queuing.c for the @@ -1250,6 +1262,17 @@ test_http: $(LIB_NAME) tests/test_http.c $(CC) $(CFLAGS) -o test_http tests/test_http.c $(LIB_NAME) $(LIBS) ./test_http +# HTTP/1.1 keep-alive hammer on the async server (issue #406): default +# backend accepts through the multishot IORING_OP_ACCEPT path where the +# kernel supports it; the epoll and knob-off runs exercise the fallback. +test_http_uring_hammer: $(LIB_NAME) tests/test_http_uring_hammer.c + $(CC) $(CFLAGS) -o $@ tests/test_http_uring_hammer.c $(LIB_NAME) $(LIBS) + ./$@ +ifeq ($(UNAME_S),Linux) + CWIST_ACCEPT_URING=0 ./$@ + CWIST_REACTOR_BACKEND=epoll ./$@ +endif + test_http_stringify: $(LIB_NAME) tests/test_http_stringify.c $(CC) $(CFLAGS) -o test_http_stringify tests/test_http_stringify.c $(LIB_NAME) $(LIBS) ./test_http_stringify diff --git a/docs/ko/runtime-config.md b/docs/ko/runtime-config.md index e4e5e16a..cab7b665 100644 --- a/docs/ko/runtime-config.md +++ b/docs/ko/runtime-config.md @@ -89,6 +89,7 @@ |------|------|----|--------|------| | `CWIST_REACTOR_BACKEND` | tuning | `epoll` | io_uring | Linux: A/B 측정용으로 epoll 백엔드를 강제합니다. | | `CWIST_RX_URING` | tuning | `0`이면 끔 | 켜짐 | POLL + `recv` 대신 `IORING_OP_RECV`로 받습니다. | +| `CWIST_ACCEPT_URING` | tuning | `0`이면 끔 | 켜짐 | one-shot POLL + `accept4` 반복 대신 multishot `IORING_OP_ACCEPT` 하나로 새 연결을 받습니다. multishot accept를 지원하지 않는 커널은 조용히 poll 경로로 돌아갑니다. | | `CWIST_RX_EPOLL` | tuning | `0`이면 끔 | 켜짐 | epoll reactor(io_uring ring이 없거나 `CWIST_REACTOR_BACKEND=epoll`)에서 keep-alive 연결을 edge-triggered로 등록해 둔 채 reactor가 직접 받습니다. 요청마다 `epoll_ctl`로 one-shot poll을 다시 거는 대신입니다. | | `CWIST_REACTOR_DRAIN_CHUNK` | tuning | 0 이상 정수 | `8` (`CWIST_PROFILE`이 설정, 없으면 `64`) | 게시된 작업을 비우는 사이에 처리하는 연결 콜백 수. 모든 reactor 백엔드(io_uring, epoll, kqueue)에 적용됩니다. `0`이면 묶음 중간의 비우기를 끕니다(기존 동작). [협력적 큐잉](cooperative-queuing.md) 참고. | | `CWIST_REACTOR_ROUND_BUDGET_US` | tuning | 마이크로초, 0 이상 | `2000` | reactor 라운드가 이 시간을 넘기면, 대기 중인 re-arm을 라운드 끝이 아니라 콜백 64개마다 커널에 넘깁니다. 라운드는 끝까지 실행됩니다. `0`이면 한도를 끕니다. | diff --git a/docs/runtime-config.md b/docs/runtime-config.md index ea9738be..4d38a360 100644 --- a/docs/runtime-config.md +++ b/docs/runtime-config.md @@ -89,6 +89,7 @@ Each variable has one status. The status says what v4 promises about it. |----------|--------|--------|---------|--------| | `CWIST_REACTOR_BACKEND` | tuning | `epoll` | io_uring | Linux: force the epoll backend, for A/B measurement. | | `CWIST_RX_URING` | tuning | `0` disables | on | Receive with `IORING_OP_RECV` instead of POLL + `recv`. | +| `CWIST_ACCEPT_URING` | tuning | `0` disables | on | Accept new connections with a multishot `IORING_OP_ACCEPT` instead of a one-shot POLL + `accept4` drain loop. Kernels without multishot accept fall back to the poll loop silently. | | `CWIST_RX_EPOLL` | tuning | `0` disables | on | On the epoll reactor (no io_uring ring, or `CWIST_REACTOR_BACKEND=epoll`), keep keep-alive connections registered edge-triggered and receive from the reactor, instead of re-arming a one-shot poll with `epoll_ctl` per request. | | `CWIST_REACTOR_DRAIN_CHUNK` | tuning | integer ≥ 0 | `8` (set by `CWIST_PROFILE`; `64` without it) | Connection callbacks handled between drains of posted work, on every reactor backend (io_uring, epoll, kqueue). `0` disables the mid-batch drain (legacy). See [cooperative queuing](cooperative-queuing.md). | | `CWIST_REACTOR_ROUND_BUDGET_US` | tuning | microseconds ≥ 0 | `2000` | Once a reactor round runs longer than this, parked re-arms are flushed to the kernel every 64 callbacks instead of at the end of the round. The round still runs to completion. `0` disables the bound. | diff --git a/src/net/http/async_server.c b/src/net/http/async_server.c index 97eafcf7..779cfcbb 100644 --- a/src/net/http/async_server.c +++ b/src/net/http/async_server.c @@ -35,14 +35,75 @@ static bool app_use_https(const cwist_app *app) { return app && app->use_ssl && app->ssl_ctx && app->https_request_handler; } +/** + * @brief Hand one accepted socket to the HTTPS dispatcher or the HTTP pool. + * + * Shared by the one-shot POLL accept loop and the io_uring multishot accept + * path; the socket is already non-blocking. Connections accepted while SSL is + * requested but not fully configured are closed with a warning. + */ +static void async_accept_dispatch(cwist_app *app, int client_fd) { + /* The fd number may have been a closed socket's: reset its epoll + * registration hint before the first arm. */ + cwist_reactor_fd_forget(client_fd); + int nodelay = 1; + setsockopt(client_fd, IPPROTO_TCP, TCP_NODELAY, &nodelay, sizeof(nodelay)); +#if defined(__linux__) && defined(TCP_QUICKACK) + int quickack = 1; + setsockopt(client_fd, IPPROTO_TCP, TCP_QUICKACK, &quickack, sizeof(quickack)); +#endif + + if (app_use_https(app)) { + cwist_https_dispatch(client_fd, app->ssl_ctx, app->https_request_handler, app); + } else if (app && !app->use_ssl) { + cwist_http_pool_submit_async(client_fd, cwist_app_http_handler_async, app); + } else { + fprintf( + stderr, + "[async] SSL request accepted but HTTPS not ready (use_ssl=%d ssl_ctx=%p handler=%p), closing fd=%d\n", + app ? app->use_ssl : -1, app ? (void *)app->ssl_ctx : NULL, + app ? (void *)app->https_request_handler : NULL, client_fd); + close(client_fd); + } +} + +#if defined(__linux__) +/* Multishot-accept arm context: outlives the registration, so it is static + * (one async server loop per process, like g_reactor). */ +typedef struct { + cwist_app *app; + int listen_fd; +} async_accept_ctx_t; +static async_accept_ctx_t g_accept_ctx; + +static void async_accept_cb(int fd, void *ctx); + +/** + * @brief io_uring multishot-accept completion; one call per accepted fd. + * + * A negative fd is a terminal status (-errno): the multishot accept ended + * (e.g. IORING_ACCEPT_MULTISHOT unsupported on this kernel), so the proven + * one-shot POLL + accept4 loop takes over the listener. Runs on the reactor + * thread; the reactor has already released the accept slot when this returns. + */ +static void async_accept_uring_cb(void *ctx, int fd) { + async_accept_ctx_t *ac = (async_accept_ctx_t *)ctx; + if (fd < 0) { + if (g_reactor && atomic_load(&g_cwist_running)) { + cwist_reactor_add(g_reactor, ac->listen_fd, async_accept_cb, &ac->app, + sizeof(ac->app)); + } + return; + } + async_accept_dispatch(ac->app, fd); +} +#endif + /** * @brief Reactor callback: accept pending connections on a listening socket. * - * Accepts all currently queued connections in a loop (non-blocking), disables - * Nagle (and enables TCP_QUICKACK on Linux where available) on each accepted - * socket, then hands it either to the HTTPS dispatcher or to the HTTP - * connection pool, depending on the app configuration. Connections accepted - * while SSL is requested but not fully configured are closed with a warning. + * Accepts all currently queued connections in a loop (non-blocking), then + * hands each to async_accept_dispatch(). * * After the accept loop drains, the listening socket is re-armed so the worker * keeps accepting: transient reactor re-add failures (e.g. a full io_uring SQ) @@ -65,28 +126,7 @@ static void async_accept_cb(int fd, void *ctx) { socklen_t len = sizeof(addr); while ((client_fd = accept(fd, (struct sockaddr *)&addr, &len)) >= 0) { #endif - /* The fd number may have been a closed socket's: reset its epoll - * registration hint before the first arm. */ - cwist_reactor_fd_forget(client_fd); - int nodelay = 1; - setsockopt(client_fd, IPPROTO_TCP, TCP_NODELAY, &nodelay, sizeof(nodelay)); -#if defined(__linux__) && defined(TCP_QUICKACK) - int quickack = 1; - setsockopt(client_fd, IPPROTO_TCP, TCP_QUICKACK, &quickack, sizeof(quickack)); -#endif - - if (app_use_https(app)) { - cwist_https_dispatch(client_fd, app->ssl_ctx, app->https_request_handler, app); - } else if (app && !app->use_ssl) { - cwist_http_pool_submit_async(client_fd, cwist_app_http_handler_async, app); - } else { - fprintf( - stderr, - "[async] SSL request accepted but HTTPS not ready (use_ssl=%d ssl_ctx=%p handler=%p), closing fd=%d\n", - app ? app->use_ssl : -1, app ? (void *)app->ssl_ctx : NULL, - app ? (void *)app->https_request_handler : NULL, client_fd); - close(client_fd); - } + async_accept_dispatch(app, client_fd); } /* Re-arm the listening socket so we can accept the next batch. A @@ -178,7 +218,14 @@ cwist_error_t cwist_async_server_loop(int server_fd, cwist_app *app) { return err; } - cwist_reactor_add(g_reactor, server_fd, async_accept_cb, &app, sizeof(app)); +#if defined(__linux__) + /* io_uring reactors: one multishot ACCEPT SQE replaces the one-shot + * POLL + accept4 drain loop. Arm failure (no ring, knob off) and a + * terminal completion both land on the legacy path. */ + g_accept_ctx = (async_accept_ctx_t){.app = app, .listen_fd = server_fd}; + if (!cwist_reactor_accept_arm(g_reactor, server_fd, async_accept_uring_cb, &g_accept_ctx)) +#endif + cwist_reactor_add(g_reactor, server_fd, async_accept_cb, &app, sizeof(app)); #if defined(__linux__) cwist_sqpoll_ring_t sq_probe; if (cwist_io_uring_init_sqpoll(&sq_probe, 64, 1000) == 0) { diff --git a/src/net/http/http.c b/src/net/http/http.c index 80956329..5005d329 100644 --- a/src/net/http/http.c +++ b/src/net/http/http.c @@ -914,18 +914,20 @@ static bool http_async_stash_grow(cwist_http_async_conn_t *conn, size_t need); * CWIST_HTTP_KEEP_ALIVE_TIMEOUT_SEC default; cached after first use. */ static uint32_t cwist_http_keep_alive_timeout_sec(void) { - static int cached_timeout = -1; - if (cached_timeout < 0) { + static _Atomic int cached_timeout = -1; + int v = atomic_load_explicit(&cached_timeout, memory_order_relaxed); + if (v < 0) { const char *env = getenv("CWIST_HTTP_KEEP_ALIVE_TIMEOUT"); int val = 0; if (env && *env) { char *end = NULL; - long v = strtol(env, &end, 10); - if (end != env && *end == '\0' && v > 0 && v <= INT_MAX) val = (int)v; + long parsed = strtol(env, &end, 10); + if (end != env && *end == '\0' && parsed > 0 && parsed <= INT_MAX) val = (int)parsed; } - cached_timeout = (val > 0) ? val : CWIST_HTTP_KEEP_ALIVE_TIMEOUT_SEC; + v = (val > 0) ? val : CWIST_HTTP_KEEP_ALIVE_TIMEOUT_SEC; + atomic_store_explicit(&cached_timeout, v, memory_order_relaxed); } - return (uint32_t)cached_timeout; + return (uint32_t)v; } /* Default hysteresis window (seconds) before an async connection's stash is diff --git a/src/sys/io/reactor.c b/src/sys/io/reactor.c index 8e1eddf6..3ee090e4 100644 --- a/src/sys/io/reactor.c +++ b/src/sys/io/reactor.c @@ -79,6 +79,22 @@ struct io_uring_getevents_arg { #define IORING_ASYNC_CANCEL_ANY (1U << 2) #endif +#ifndef IORING_ACCEPT_MULTISHOT +#define IORING_ACCEPT_MULTISHOT (1U << 0) +#endif + +#ifndef IORING_CQE_F_MORE +#define IORING_CQE_F_MORE (1U << 1) +#endif + +/* accept4-style flags for the SQE; on Linux these equal the O_* values. */ +#ifndef SOCK_CLOEXEC +#define SOCK_CLOEXEC O_CLOEXEC +#endif +#ifndef SOCK_NONBLOCK +#define SOCK_NONBLOCK O_NONBLOCK +#endif + struct __kernel_timespec; /** @brief Raw io_uring_setup(2) syscall wrapper (no libc wrapper exists). @@ -230,6 +246,21 @@ typedef struct reactor_slot_chunk { reactor_event_ctx_t slots[]; } reactor_slot_chunk_t; +#ifdef __linux__ +/* Persistent multishot-accept registration (issue #406). Unlike the + * one-shot ev_ctx slots this object lives across completions: one + * IORING_OP_ACCEPT+MULTISHOT keeps producing CQEs until it terminates + * (completion without IORING_CQE_F_MORE). Dispatched via bit 1 of + * user_data (bit 0 is the RX-uring tag); cwist_alloc alignment keeps both + * tag bits clear. */ +typedef struct reactor_accept_slot { + struct reactor_accept_slot *next; + int listen_fd; + cwist_accept_cb_t cb; + void *ctx; +} reactor_accept_slot_t; +#endif + struct cwist_reactor { reactor_impl_t impl; bool running; @@ -270,6 +301,9 @@ struct cwist_reactor { /* RX-uring completion handler for tagged (low-bit set) RECV SQEs; * registered by the HTTP layer once per worker reactor. */ cwist_rx_cb_t rx_cb; + /* Persistent multishot-accept registrations (bit-1-tagged ACCEPT SQEs); + * freed on terminal completion or at destroy. */ + reactor_accept_slot_t *accept_slots; /* RX-epoll: fds whose armed receive may already have bytes waiting (the * last recv filled the buffer, or an edge arrived while unarmed). An * edge-triggered registration will not report those again, so the run @@ -685,6 +719,64 @@ void cwist_reactor_probe_record(cwist_reactor_t *reactor, bool queue, uint64_t s &reactor->probe[phase].sum_us, &reactor->probe[phase].max_us, &reactor->probe[phase].over_5ms, sample_us); } + +/* --- Multishot accept (issue #406) ------------------------------------------ + * CWIST_ACCEPT_URING: unset = auto (multishot accept whenever the reactor + * has a real io_uring ring), "0" = force off (one-shot POLL + accept4 loop). + * Cached after the first read like the other env knobs in this file. */ +static bool accept_uring_env_enabled(void) { + static _Atomic int cached = -1; + int v = atomic_load_explicit(&cached, memory_order_relaxed); + if (v < 0) { + const char *s = getenv("CWIST_ACCEPT_URING"); + v = (s && strcmp(s, "0") == 0) ? 0 : 1; + atomic_store_explicit(&cached, v, memory_order_relaxed); + } + return v == 1; +} + +static void reactor_accept_unlink(cwist_reactor_t *reactor, reactor_accept_slot_t *slot) { + reactor_accept_slot_t **pp = &reactor->accept_slots; + while (*pp) { + if (*pp == slot) { + *pp = slot->next; + return; + } + pp = &(*pp)->next; + } +} + +bool cwist_reactor_accept_arm(cwist_reactor_t *reactor, int listen_fd, cwist_accept_cb_t cb, + void *ctx) { + if (!reactor || listen_fd < 0 || !cb) return false; + if (reactor->impl.use_epoll || !reactor->impl.active) return false; + if (!accept_uring_env_enabled()) return false; + + reactor_accept_slot_t *slot = cwist_alloc(sizeof(*slot)); + if (!slot) return false; + slot->listen_fd = listen_fd; + slot->cb = cb; + slot->ctx = ctx; + + struct io_uring_sqe sqe; + memset(&sqe, 0, sizeof(sqe)); + sqe.opcode = IORING_OP_ACCEPT; + sqe.fd = listen_fd; + sqe.accept_flags = SOCK_CLOEXEC | SOCK_NONBLOCK; + sqe.ioprio = IORING_ACCEPT_MULTISHOT; + sqe.user_data = (uint64_t)(uintptr_t)slot | 2u; + + /* Link before submitting: a fast completion can dispatch on the run + * thread as soon as the SQE is visible, and destroy must find the slot. */ + slot->next = reactor->accept_slots; + reactor->accept_slots = slot; + if (!uring_submit(reactor, &sqe)) { + reactor_accept_unlink(reactor, slot); + cwist_free(slot); + return false; + } + return true; +} #endif static reactor_event_ctx_t *alloc_reactor_ctx(cwist_reactor_t *r, int fd, cwist_reactor_cb_t cb, @@ -1056,6 +1148,13 @@ void cwist_reactor_destroy(cwist_reactor_t *reactor) { pthread_mutex_destroy(&reactor->pool_lock); #ifdef __linux__ pthread_mutex_destroy(&reactor->impl.sq_lock); + /* Multishot accepts terminated with the ring; free their registrations. */ + reactor_accept_slot_t *as = reactor->accept_slots; + while (as) { + reactor_accept_slot_t *next = as->next; + cwist_free(as); + as = next; + } #endif cwist_free(reactor->timer_heap); reactor_slot_chunk_t *chunk = reactor->chunks; @@ -1611,6 +1710,22 @@ void cwist_reactor_run(cwist_reactor_t *reactor) { if (reactor->rx_cb) { reactor->rx_cb((void *)(user_data & ~(uint64_t)1u), cqe->res); } + } else if (user_data & 2u) { + /* Multishot-accept slot (issue #406): cqe->res is the + * accepted fd, already CLOEXEC|NONBLOCK. A completion + * without IORING_CQE_F_MORE ends the registration: notify + * once with the terminal status (-errno, or -ECANCELED on + * an error-free end) so the caller can resume the + * one-shot POLL accept loop, then free the slot. */ + reactor_accept_slot_t *slot = + (reactor_accept_slot_t *)(uintptr_t)(user_data & ~(uint64_t)3u); + int res = cqe->res; + if (res >= 0) slot->cb(slot->ctx, res); + if (!(cqe->flags & IORING_CQE_F_MORE)) { + reactor_accept_unlink(reactor, slot); + slot->cb(slot->ctx, res < 0 ? res : -ECANCELED); + cwist_free(slot); + } } else { reactor_event_ctx_t *ev_ctx = (reactor_event_ctx_t *)user_data; if (ev_ctx) { diff --git a/src/sys/io/reactor_rx.h b/src/sys/io/reactor_rx.h index 21ac0457..e5c44ee4 100644 --- a/src/sys/io/reactor_rx.h +++ b/src/sys/io/reactor_rx.h @@ -15,7 +15,8 @@ * * ev_ctx pointers come from cwist_alloc (malloc-backed, so at least * max_align_t aligned) and can never carry the tag bit; reactor.c asserts - * this when arming. + * this when arming. Multishot-accept slots (see the bottom of this header) + * tag bit 1 instead, so the run loop can tell all three apart. * * RX-epoll: on an epoll reactor (no io_uring ring, e.g. under a seccomp * profile that blocks it) the same contract is served without SQEs. The fd @@ -89,6 +90,32 @@ bool cwist_reactor_recv_arm(cwist_reactor_t *reactor, int fd, void *buf, unsigne * wrapper in cwist_reactor_run. */ void cwist_reactor_probe_record(cwist_reactor_t *reactor, bool queue, uint64_t sample_us); +/* --- Multishot accept (issue #406) ----------------------------------------- + * The listen socket's accept can ride one IORING_OP_ACCEPT with + * IORING_ACCEPT_MULTISHOT instead of a one-shot POLL + accept4 drain loop: + * each completion IS an accepted connection (SOCK_CLOEXEC|SOCK_NONBLOCK + * already set). Accept slots tag bit 1 of user_data (the RX path uses bit + * 0); both live outside the ev_ctx namespace, whose slots are malloc-aligned. + * + * The slot is persistent, unlike the one-shot ev_ctx slots: it survives + * across completions until the multishot accept terminates (completion + * without IORING_CQE_F_MORE) or the reactor is destroyed. */ + +/* Completion handler for multishot accepts. fd >= 0 is a freshly accepted + * socket (CLOEXEC+NONBLOCK set). A negative fd is -errno and is delivered + * exactly once, when the accept terminates: unsupported opcode on an old + * kernel (-EINVAL), listener error, or teardown. The caller should resume + * the one-shot POLL accept loop; the reactor has already freed the slot when + * the callback returns. Runs on the reactor owner thread while dispatching. */ +typedef void (*cwist_accept_cb_t)(void *ctx, int fd); + +/* Arm one multishot accept on the reactor ring. At most one per listen fd. + * Returns false when this reactor has no io_uring ring (epoll fallback) or + * CWIST_ACCEPT_URING=0; the caller then uses the one-shot POLL path. Arm + * before cwist_reactor_run() or from the run thread. */ +bool cwist_reactor_accept_arm(cwist_reactor_t *reactor, int listen_fd, cwist_accept_cb_t cb, + void *ctx); + #else /* !__linux__ */ static inline bool cwist_rx_uring_env_enabled(void) { @@ -122,6 +149,15 @@ static inline void cwist_reactor_probe_record(cwist_reactor_t *reactor, bool que (void)queue; (void)sample_us; } +typedef void (*cwist_accept_cb_t)(void *ctx, int fd); +static inline bool cwist_reactor_accept_arm(cwist_reactor_t *reactor, int listen_fd, + cwist_accept_cb_t cb, void *ctx) { + (void)reactor; + (void)listen_fd; + (void)cb; + (void)ctx; + return false; +} #endif /* __linux__ */ diff --git a/tests/test_http_uring_hammer.c b/tests/test_http_uring_hammer.c new file mode 100644 index 00000000..7f2d7d3e --- /dev/null +++ b/tests/test_http_uring_hammer.c @@ -0,0 +1,165 @@ +/** + * @file test_http_uring_hammer.c + * @brief HTTP/1.1 keep-alive hammer against the async (C1M) server. + * + * Forks a child running a cwist app on the default backend (io_uring where + * available, accepting through the multishot IORING_OP_ACCEPT path) and + * drives it with concurrent keep-alive clients. The Makefile also runs this + * with CWIST_REACTOR_BACKEND=epoll, which exercises the silent fallback to + * the one-shot POLL + accept4 loop, and with CWIST_ACCEPT_URING=0, which + * refuses the multishot arm on the uring backend. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#define TEST_PORT 19989 +#define TEST_HOST "127.0.0.1" +#define THREADS 8 +#define REQS_PER_THREAD 200 + +/* --- Server side --------------------------------------------------------- */ + +static void hello_handler(cwist_http_request *req, cwist_http_response *res) { + (void)req; + cwist_sstring_assign(res->body, "hammer-ok"); + cwist_http_header_add(&res->headers, "Content-Type", "text/plain"); + res->status_code = CWIST_HTTP_OK; +} + +/* --- Client side --------------------------------------------------------- */ + +static int connect_to_server(void) { + int fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) return -1; + struct sockaddr_in addr = { + .sin_family = AF_INET, + .sin_port = htons(TEST_PORT), + }; + inet_pton(AF_INET, TEST_HOST, &addr.sin_addr); + if (connect(fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) { + close(fd); + return -1; + } + return fd; +} + +static int send_all(int fd, const char *req, size_t len) { + size_t sent = 0; + while (sent < len) { + ssize_t n = send(fd, req + sent, len - sent, 0); + if (n <= 0) return -1; + sent += (size_t)n; + } + return 0; +} + +/* Read one full HTTP/1.x response (headers + Content-Length body). One + * request in flight per connection, so nothing is pipelined. */ +static int read_response(int fd, char *buf, size_t buf_size) { + size_t total = 0; + size_t need = 0; /* header_end + content_length once parsed */ + for (;;) { + if (need == 0) { + buf[total] = '\0'; + char *he = strstr(buf, "\r\n\r\n"); + if (he) { + size_t header_end = (size_t)(he - buf) + 4; + char *cl = strcasestr(buf, "Content-Length:"); + if (!cl || (size_t)(cl - buf) > header_end) return -1; + need = header_end + (size_t)atoi(cl + 15); + } + } + if (need > 0 && total >= need) return (int)total; + if (total >= buf_size - 1) return -1; + ssize_t n = recv(fd, buf + total, buf_size - 1 - total, 0); + if (n <= 0) return -1; + total += (size_t)n; + } +} + +static _Atomic int g_failures; + +static void *client_thread(void *arg) { + (void)arg; + static const char req[] = "GET /hello HTTP/1.1\r\nHost: localhost\r\n\r\n"; + char buf[4096]; + int fd = connect_to_server(); + if (fd < 0) { + atomic_fetch_add(&g_failures, 1); + return NULL; + } + for (int i = 0; i < REQS_PER_THREAD; i++) { + if (send_all(fd, req, sizeof(req) - 1) < 0 || + read_response(fd, buf, sizeof(buf)) < 0 || !strstr(buf, "HTTP/1.1 200") || + !strstr(buf, "hammer-ok")) { + atomic_fetch_add(&g_failures, 1); + break; + } + } + close(fd); + return NULL; +} + +int main(void) { + printf("Testing HTTP/1.1 keep-alive hammer (reactor accept path)...\n"); + + pid_t pid = fork(); + if (pid < 0) { + perror("fork"); + return 1; + } + + if (pid == 0) { + /* Single worker process: the accept path under test is per-process. */ + setenv("CWIST_WORKERS", "1", 1); + cwist_app *app = cwist_app_create(); + if (!app) _exit(1); + cwist_app_get(app, "/hello", hello_handler); + g_cwist_drain_timeout_sec = 1; + int rc = cwist_app_listen(app, TEST_PORT); + cwist_app_destroy(app); + _exit(rc == 0 ? 0 : 1); + } + + usleep(400000); + + pthread_t threads[THREADS]; + for (int i = 0; i < THREADS; i++) { + if (pthread_create(&threads[i], NULL, client_thread, NULL) != 0) { + atomic_fetch_add(&g_failures, 1); + threads[i] = 0; + } + } + for (int i = 0; i < THREADS; i++) { + if (threads[i]) pthread_join(threads[i], NULL); + } + + kill(pid, SIGTERM); + int status; + waitpid(pid, &status, 0); + if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) { + fprintf(stderr, "FAIL: server child exited abnormally (status=%d)\n", status); + atomic_fetch_add(&g_failures, 1); + } + + int failures = atomic_load(&g_failures); + if (failures == 0) { + printf("HTTP/1.1 keep-alive hammer (%d threads x %d requests): PASS\n", THREADS, + REQS_PER_THREAD); + return 0; + } + fprintf(stderr, "%d hammer failure(s).\n", failures); + return 1; +} diff --git a/tests/test_reactor_accept.c b/tests/test_reactor_accept.c new file mode 100644 index 00000000..0e4fa8c8 --- /dev/null +++ b/tests/test_reactor_accept.c @@ -0,0 +1,195 @@ +#ifndef _GNU_SOURCE +#define _GNU_SOURCE +#endif +/* Multishot-accept contract test (issue #406). Same approach as + * test_reactor_wake.c: include the implementation so the accept slot and + * ring internals are inspectable, replacing only the allocator and the + * process shutdown flag. + * + * "./test_reactor_accept" functional accept run (needs the uring ring) + * "./test_reactor_accept off" CWIST_ACCEPT_URING=0 must refuse the arm + * CWIST_REACTOR_BACKEND=epoll the arm must refuse on the epoll fallback + * + * CWIST_TEST_REQUIRE_IO_URING=1 fails rather than skipping a denied backend. + */ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "../src/sys/io/reactor.c" + +void *cwist_alloc(size_t size) { + return calloc(1, size); +} +void cwist_free(void *ptr) { + free(ptr); +} +atomic_int g_cwist_running = 1; + +enum { TARGET = 64 }; +static cwist_reactor_t *loop; +static pthread_t run_tid; +static _Atomic int accepted; +static _Atomic int terminals; +static _Atomic int last_terminal_err; + +static void *run_thread(void *arg) { + (void)arg; + run_tid = pthread_self(); + cwist_reactor_run(loop); + return NULL; +} + +static void accept_cb(void *ctx, int fd) { + (void)ctx; + assert(pthread_equal(pthread_self(), run_tid)); + if (fd < 0) { + atomic_fetch_add(&terminals, 1); + atomic_store(&last_terminal_err, fd); + return; + } + /* accept_flags must have produced a non-blocking socket. */ + int fl = fcntl(fd, F_GETFL, 0); + assert(fl >= 0 && (fl & O_NONBLOCK)); + close(fd); + if (atomic_fetch_add(&accepted, 1) + 1 == TARGET) cwist_reactor_stop(loop); +} + +static int make_listener(void) { + int fd = socket(AF_INET, SOCK_STREAM, 0); + assert(fd >= 0); + int one = 1; + setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)); + struct sockaddr_in addr = { + .sin_family = AF_INET, + .sin_port = htons(0), + .sin_addr.s_addr = htonl(INADDR_LOOPBACK), + }; + assert(bind(fd, (struct sockaddr *)&addr, sizeof(addr)) == 0); + assert(listen(fd, 128) == 0); + int fl = fcntl(fd, F_GETFL, 0); + assert(fl >= 0 && fcntl(fd, F_SETFL, fl | O_NONBLOCK) == 0); + return fd; +} + +static int listener_port(int fd) { + struct sockaddr_in addr; + socklen_t len = sizeof(addr); + assert(getsockname(fd, (struct sockaddr *)&addr, &len) == 0); + return ntohs(addr.sin_port); +} + +static void connect_n(int port, int fds[], int n) { + for (int i = 0; i < n; i++) { + int c = socket(AF_INET, SOCK_STREAM, 0); + assert(c >= 0); + struct sockaddr_in addr = { + .sin_family = AF_INET, + .sin_port = htons((uint16_t)port), + .sin_addr.s_addr = htonl(INADDR_LOOPBACK), + }; + assert(connect(c, (struct sockaddr *)&addr, sizeof(addr)) == 0); + fds[i] = c; + } +} + +/* On a real ring with the knob on, one multishot arm accepts TARGET + * connections without a single re-arm. */ +static void check_functional(void) { + int lfd = make_listener(); + assert(cwist_reactor_accept_arm(loop, lfd, accept_cb, NULL)); + pthread_t thread; + assert(pthread_create(&thread, NULL, run_thread, NULL) == 0); + int clients[TARGET]; + connect_n(listener_port(lfd), clients, TARGET); + assert(pthread_join(thread, NULL) == 0); + assert(atomic_load(&accepted) == TARGET); + assert(atomic_load(&terminals) == 0); + /* The multishot registration outlives the run: destroy must free it. */ + cwist_reactor_destroy(loop); + loop = NULL; + for (int i = 0; i < TARGET; i++) close(clients[i]); + close(lfd); +} + +/* An accept armed on a socket that is not listening completes with a + * terminal error, exactly once, so the caller can fall back to the + * one-shot POLL loop. */ +static void check_terminal(void) { + loop = cwist_reactor_create(); + assert(loop != NULL); + if (loop->impl.use_epoll) { + cwist_reactor_destroy(loop); + loop = NULL; + return; + } + int fd = socket(AF_INET, SOCK_STREAM, 0); + assert(fd >= 0); /* never listen()ed: accept must fail */ + assert(cwist_reactor_accept_arm(loop, fd, accept_cb, NULL)); + pthread_t thread; + assert(pthread_create(&thread, NULL, run_thread, NULL) == 0); + for (int i = 0; i < 2000 && atomic_load(&terminals) == 0; i++) { + struct timespec ts = {.tv_sec = 0, .tv_nsec = 1000 * 1000}; + nanosleep(&ts, NULL); + } + assert(atomic_load(&terminals) == 1); + assert(atomic_load(&last_terminal_err) < 0); + cwist_reactor_stop(loop); + assert(pthread_join(thread, NULL) == 0); + cwist_reactor_destroy(loop); + loop = NULL; + close(fd); +} + +int main(int argc, char **argv) { + alarm(30); /* Watchdog only, not the correctness oracle. */ + bool knob_off = argc == 2 && strcmp(argv[1], "off") == 0; + if (knob_off) setenv("CWIST_ACCEPT_URING", "0", 1); + + loop = cwist_reactor_create(); + assert(loop != NULL); +#ifdef __linux__ + printf("backend=%s\n", loop->impl.use_epoll ? "epoll" : "io_uring"); + if (loop->impl.use_epoll && getenv("CWIST_TEST_REQUIRE_IO_URING")) { + fprintf(stderr, "required io_uring backend unavailable\n"); + cwist_reactor_destroy(loop); + exit(1); + } + if (loop->impl.use_epoll) { + /* Epoll fallback reactor: the multishot arm must refuse, leaving the + * one-shot POLL accept path in charge. */ + int lfd = make_listener(); + assert(!cwist_reactor_accept_arm(loop, lfd, accept_cb, NULL)); + close(lfd); + cwist_reactor_destroy(loop); + puts("multishot accept refuses the epoll backend: PASS"); + return 0; + } + if (knob_off) { + int lfd = make_listener(); + assert(!cwist_reactor_accept_arm(loop, lfd, accept_cb, NULL)); + close(lfd); + cwist_reactor_destroy(loop); + puts("CWIST_ACCEPT_URING=0 refuses the arm: PASS"); + return 0; + } + check_functional(); + check_terminal(); + puts("multishot accept, terminal error, slot teardown: PASS"); + return 0; +#else + assert(!cwist_reactor_accept_arm(loop, -1, accept_cb, NULL)); + cwist_reactor_destroy(loop); + puts("multishot accept is a no-op off Linux: PASS"); + return 0; +#endif +}