feat(io): io_uring TCP path for HTTP/1.1 and HTTP/2

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

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`이면 한도를 끕니다. |

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. |

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) {

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

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) {