From 507bc2aedb5dc47ea655cecf97dd290acab930a0 Mon Sep 17 00:00:00 2001 From: Lee Yunjin Date: Tue, 13 Oct 2026 23:07:05 +0900 Subject: [PATCH] perf(reactor): serve events queued past 3.5 ms first (tail heuristic) Stamp every armed event with CLOCK_MONOTONIC at arm time (one clock read and store) and check queue age at delivery in both reactor backends: a round-start scan pulls already-aged events to the front of the dispatch batch, and a per-delivery check catches events that age behind a slow earlier callback. An event past CWIST_REACTOR_TAIL_US (default 3500 us, 0 disables) is dispatched first and sees cwist_reactor_dispatch_late() true, so the HTTP layer serves its buffered pipeline without yielding -- the same bypass late-round mode (CWIST_REACTOR_LATE_US) uses, so both backends share one code path. RX-uring/RX-epoll completions carry their arm stamp on the HTTP side; the RX completion handler runs the same age check and raises the bypass. Aged-event counter is reported by the latency probe as tail_elevations. --- Makefile | 9 ++ docs/ko/runtime-config.md | 1 + docs/runtime-config.md | 1 + src/net/http/http.c | 19 +-- src/sys/app/app.c | 7 +- src/sys/io/reactor.c | 252 +++++++++++++++++++++++++++++-------- src/sys/io/reactor_rx.h | 44 ++++++- tests/test_http_fairness.c | 5 + tests/test_reactor_late.c | 4 + tests/test_reactor_tail.c | 245 ++++++++++++++++++++++++++++++++++++ 10 files changed, 517 insertions(+), 70 deletions(-) create mode 100644 tests/test_reactor_tail.c diff --git a/Makefile b/Makefile index 7e5a36b17..462885c82 100644 --- a/Makefile +++ b/Makefile @@ -897,6 +897,7 @@ TEST_TARGETS = test_worker_affinity \ test_rx_epoll \ test_reactor_drain_chunk \ test_reactor_late \ + test_reactor_tail \ test_latency_probe \ test_sstring \ test_abi_layout \ @@ -1125,6 +1126,14 @@ test_reactor_late: tests/test_reactor_late.c src/sys/io/reactor.c $(CC) $(CFLAGS) -o $@ tests/test_reactor_late.c -pthread ./$@ +# Tail-elevation heuristic: an event armed past CWIST_REACTOR_TAIL_US is +# served before younger events of the same batch on both backends, and its +# callback observes cwist_reactor_dispatch_late(). Runs each knob value in a +# child process (the knob is cached on first read) on uring and epoll. +test_reactor_tail: tests/test_reactor_tail.c src/sys/io/reactor.c + $(CC) $(CFLAGS) -o $@ tests/test_reactor_tail.c -pthread + ./$@ + # Latency probe test (issue #166): run twice per build, with the probe # enabled (recorder must accumulate samples) and disabled (recorder must # stay at zero). CWIST_LATENCY_PROBE must be set before reactor creation. diff --git a/docs/ko/runtime-config.md b/docs/ko/runtime-config.md index 479d42f63..28bb8321f 100644 --- a/docs/ko/runtime-config.md +++ b/docs/ko/runtime-config.md @@ -91,6 +91,7 @@ | `CWIST_REACTOR_ROUND_BUDGET_US` | tuning | 마이크로초, 0 이상 | `2000` | reactor 라운드가 이 시간을 넘기면, 대기 중인 re-arm을 라운드 끝이 아니라 콜백 64개마다 커널에 넘깁니다. 라운드는 끝까지 실행됩니다. `0`이면 한도를 끕니다. | | `CWIST_REACTOR_CQ_GRACE_NS` | tuning | 나노초, 0 이상 | `20000` | 커널에서 잠들기 전에 completion ring을 확인하며 도는 시간. `0`이면 돌지 않습니다. | | `CWIST_REACTOR_LATE_US` | tuning | 마이크로초, 0 이상 | `3000` | 실험 기능(#319). reactor 라운드가 completion을 수확한 뒤 이 시간을 넘기면, 남은 콜백을 지연 모드로 처리합니다: 콜백마다 re-arm을 바로 커널에 넘기고, 게시된 작업을 콜백마다 비우며, 파이프라인 HTTP/1.1 연결은 양보하지 않고 쌓인 요청을 처리합니다. io_uring 전용. `0`이면 끕니다. | +| `CWIST_REACTOR_TAIL_US` | tuning | 마이크로초, 0 이상 | `3500` | tail 우선 처리 휴리스틱(io_uring과 epoll 둘 다). 이 시간보다 오래 전에 arm된 이벤트는 디스패치 배치의 맨 앞으로 끌려와 먼저 처리되고, 그 연결은 쌓인 파이프라인을 양보 없이 처리합니다. 느린 라운드가 아니라 느린 이벤트에 반응한다는 점에서 `CWIST_REACTOR_LATE_US`를 보완합니다. `0`이면 끕니다. | ## 진단 diff --git a/docs/runtime-config.md b/docs/runtime-config.md index 16e4990dc..98cb1e839 100644 --- a/docs/runtime-config.md +++ b/docs/runtime-config.md @@ -91,6 +91,7 @@ Each variable has one status. The status says what v4 promises about it. | `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. | | `CWIST_REACTOR_CQ_GRACE_NS` | tuning | nanoseconds ≥ 0 | `20000` | How long the reactor spins on the completion ring before it sleeps in the kernel. `0` disables the spin. | | `CWIST_REACTOR_LATE_US` | tuning | microseconds ≥ 0 | `3000` | Experiment (#319). Once a reactor round has run this long since its completions were reaped, the rest of the round runs in late mode: each callback's re-arm is submitted right after it, posted work is drained after each callback, and a pipelined HTTP/1.1 connection serves its buffered requests without yielding. io_uring only. `0` disables. | +| `CWIST_REACTOR_TAIL_US` | tuning | microseconds ≥ 0 | `3500` | Tail-elevation heuristic (io_uring and epoll). An event armed longer ago than this is pulled to the front of its dispatch batch and served first; its connection then serves its buffered pipeline without yielding. Complements `CWIST_REACTOR_LATE_US`, which reacts to a slow round rather than a slow event. `0` disables. | ## Diagnostics diff --git a/src/net/http/http.c b/src/net/http/http.c index fb0fbed84..06d406917 100644 --- a/src/net/http/http.c +++ b/src/net/http/http.c @@ -1101,16 +1101,17 @@ static void http_rx_recv_cb(void *conn_ptr, int res) { /* res == -EAGAIN: nothing staged; the POLL fallback below resumes the * wait without running the handler on an empty stash. */ - /* Latency probe: rx_armed_ns is only written while the probe is enabled, - * so a zero value means the probe is off and both clock reads below are - * skipped. Snapshot it now: the serve path re-arms the next RECV and - * would overwrite conn->rx_armed_ns before the queue delay is computed. */ - uint64_t t0 = 0; + /* rx_armed_ns is stamped at every arm. Snapshot it now: the serve path + * re-arms the next RECV and would overwrite conn->rx_armed_ns before the + * queue delay is computed. One clock read at delivery feeds the + * tail-elevation heuristic (CWIST_REACTOR_TAIL_US) and, when the probe + * is enabled, the queue-delay sample below. */ uint64_t armed_ns = conn->rx_armed_ns; + uint64_t t0 = 0; if (armed_ns) { - struct timespec ts; - clock_gettime(CLOCK_MONOTONIC, &ts); - t0 = (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; + t0 = cwist_reactor_now_ns(); + if (cwist_reactor_event_aged(t0, armed_ns, cwist_reactor_tail_ns())) + cwist_reactor_dispatch_age_bypass(); } conn->last_active_sec = cwist_fast_monotonic_sec(); @@ -1149,7 +1150,7 @@ static void http_rx_recv_cb(void *conn_ptr, int res) { conn->rx_prefers_poll = false; http_async_dispatch(conn->fd, conn, conn->handler, conn->user_ctx, conn->reactor); - if (armed_ns) { + if (armed_ns && cwist_reactor_latency_probe_enabled()) { uint64_t queue_us = (t0 - armed_ns) / 1000; struct timespec ts; clock_gettime(CLOCK_MONOTONIC, &ts); diff --git a/src/sys/app/app.c b/src/sys/app/app.c index cc359e5d2..554444c02 100644 --- a/src/sys/app/app.c +++ b/src/sys/app/app.c @@ -3314,9 +3314,10 @@ cwist_async_action_t cwist_app_http_handler_async(int client_fd, cwist_http_asyn uint32_t priority_weight = ((mixed_seed * 16777619U) >> 8) % 100; /* One pipeline must not monopolize this reactor's other connections, - * unless this callback runs in a late round (issue #319 experiment): the - * connection has already waited past CWIST_REACTOR_LATE_US, so it serves - * its buffered pipeline now instead of yielding behind the others. */ + * unless this callback runs in a late round (issue #319 experiment) or the + * tail heuristic aged its event out of the batch (CWIST_REACTOR_TAIL_US): + * the connection has already waited past the threshold, so it serves its + * buffered pipeline now instead of yielding behind the others. */ const bool late = cwist_reactor_dispatch_late(); const unsigned int yield_batch = app_http_yield_batch(); const unsigned int turn_max = diff --git a/src/sys/io/reactor.c b/src/sys/io/reactor.c index 3b4308480..64ad9142c 100644 --- a/src/sys/io/reactor.c +++ b/src/sys/io/reactor.c @@ -201,11 +201,11 @@ typedef struct { cwist_reactor_cb_t cb; void *ctx; #ifdef __linux__ - /* CLOCK_MONOTONIC timestamp taken in reactor_add_common right before the - * POLL_ADD submit/defer decision; approximates when the fd started - * waiting for readiness. Only set when CWIST_LATENCY_PROBE is enabled; - * zero means "armed before the probe was enabled" (defensive, see the - * dispatch loop). */ + /* CLOCK_MONOTONIC timestamp taken in alloc_reactor_ctx right before the + * arm is submitted or deferred; approximates when the fd started waiting + * for readiness. Always set (the tail-elevation heuristic and the latency + * probe both consume it); zero means "armed before the stamp existed" + * (defensive, see the dispatch loop). */ uint64_t armed_ns; #endif /* Inline caller payload; ctx always points here. Zeroed on checkout so @@ -265,10 +265,14 @@ struct cwist_reactor { bool dispatching; /* True while the run thread dispatches a callback of a round that has * run past CWIST_REACTOR_LATE_US (issue #319 experiment). Run thread - * only; read through cwist_reactor_dispatch_late(). */ + * only; read through cwist_reactor_dispatch_late(), which also reports + * per-event tail elevation (CWIST_REACTOR_TAIL_US). */ bool dispatch_late; uint64_t late_rounds; /* Rounds that crossed the threshold. */ uint64_t late_callbacks; /* Callbacks dispatched in late mode. */ + /* Events the tail heuristic pulled out of a batch and served first + * (see reactor_tail_ns). Run thread only. */ + uint64_t tail_elevations; /* 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; @@ -298,6 +302,13 @@ struct cwist_reactor { * threads would otherwise read that field while the run thread writes it. */ static __thread cwist_reactor_t *t_run_reactor; +/* True while the run thread dispatches an event the tail heuristic aged out + * of its batch: its callback sees cwist_reactor_dispatch_late() == true. Set + * by the dispatch loops' elevation pass and by the HTTP RX completion + * handler (which owns the RECV arm timestamp); cleared when the callback + * returns, so it never leaks into the next event of the batch. */ +static __thread bool t_dispatch_aged; + static inline bool reactor_on_run_thread(const cwist_reactor_t *reactor) { return t_run_reactor == reactor; } @@ -379,6 +390,9 @@ static void latency_probe_dump(const cwist_reactor_t *reactor) { fprintf(stderr, "[latency-probe] pid=%d late_rounds=%llu late_callbacks=%llu\n", (int)getpid(), (unsigned long long)reactor->late_rounds, (unsigned long long)reactor->late_callbacks); + if (reactor->tail_elevations) + fprintf(stderr, "[latency-probe] pid=%d tail_elevations=%llu\n", (int)getpid(), + (unsigned long long)reactor->tail_elevations); for (int phase = 0; phase < 2; phase++) { const uint64_t *b = reactor->probe[phase].buckets; uint64_t count = reactor->probe[phase].count; @@ -560,8 +574,13 @@ static void rx_epoll_recv(cwist_reactor_t *r, int fd, epoll_fd_slot_t *slot) { slot->maybe_more = n > 0 && (size_t)n == slot->len; slot->armed = false; /* rx_cb may close and free the connection, or re-arm (possibly queueing - * this fd again); the slot itself stays valid. */ - if (r->rx_cb) r->rx_cb(slot->conn, res); + * this fd again); the slot itself stays valid. The handler may raise the + * tail-heuristic bypass (cwist_reactor_dispatch_age_bypass); it is scoped + * to this callback, so clear it before the next event of the batch. */ + if (r->rx_cb) { + r->rx_cb(slot->conn, res); + t_dispatch_aged = false; + } } /** @brief Handle one readiness event of an RX-mode fd. Run thread only. */ @@ -639,11 +658,9 @@ bool cwist_reactor_recv_arm(cwist_reactor_t *reactor, int fd, void *buf, unsigne if (!reactor || fd < 0 || !buf || len == 0 || !conn) return false; if (!cwist_reactor_rx_supported(reactor)) return false; if (reactor->impl.use_epoll) { - if (armed_ns && latency_probe_enabled()) { - struct timespec ts; - clock_gettime(CLOCK_MONOTONIC, &ts); - *armed_ns = (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; - } + /* Always stamped: the HTTP completion handler runs the tail heuristic + * against it (and the latency probe on top). */ + if (armed_ns) *armed_ns = cwist_reactor_now_ns(); return rx_epoll_arm(reactor, fd, buf, len, conn); } /* Tag-bit discipline: ev_ctx pointers come from cwist_alloc (malloc @@ -659,11 +676,7 @@ bool cwist_reactor_recv_arm(cwist_reactor_t *reactor, int fd, void *buf, unsigne sqe.addr = (uint64_t)(uintptr_t)buf; sqe.len = len; sqe.user_data = (uint64_t)(uintptr_t)conn | 1u; - if (armed_ns && latency_probe_enabled()) { - struct timespec ts; - clock_gettime(CLOCK_MONOTONIC, &ts); - *armed_ns = (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; - } + if (armed_ns) *armed_ns = cwist_reactor_now_ns(); /* Run-thread re-arms defer into the batch flush; everything else * submits immediately so remote workers wake. Data SQEs carry no ev_ctx @@ -723,6 +736,16 @@ static reactor_event_ctx_t *alloc_reactor_ctx(cwist_reactor_t *r, int fd, cwist_ if (!ev_ctx) return NULL; memset(ev_ctx, 0, sizeof(*ev_ctx)); +#ifdef __linux__ + { + /* One clock read + store per arm: feeds the tail-elevation heuristic + * (dispatch-side compare against CWIST_REACTOR_TAIL_US) and the + * latency probe's queue_delay histogram. */ + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + ev_ctx->armed_ns = (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; + } +#endif ev_ctx->fd = fd; ev_ctx->cb = cb; ev_ctx->ctx = ev_ctx->payload; @@ -740,6 +763,37 @@ static void free_reactor_ctx(cwist_reactor_t *r, reactor_event_ctx_t *ev_ctx) { pthread_mutex_unlock(&r->pool_lock); } +#ifdef __linux__ +/* Dispatch one ev_ctx completion: probe-wrapped when CWIST_LATENCY_PROBE is + * enabled, plain otherwise. Shared by the uring batch loop and its + * tail-elevation pre-pass so both record identical probe samples. */ +static void dispatch_ev_ctx(cwist_reactor_t *reactor, reactor_event_ctx_t *ev_ctx) { + if (!latency_probe_enabled()) { + ev_ctx->cb(ev_ctx->fd, ev_ctx->ctx); + return; + } + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + uint64_t t0 = (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; + /* armed_ns == 0 means the slot predates the stamp; fold it into the + * first bucket rather than producing a garbage delay. */ + uint64_t delay_ns = ev_ctx->armed_ns ? t0 - ev_ctx->armed_ns : 0; + ev_ctx->cb(ev_ctx->fd, ev_ctx->ctx); + clock_gettime(CLOCK_MONOTONIC, &ts); + uint64_t t1 = (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; + latency_probe_record(reactor->probe[LATENCY_PROBE_QUEUE].buckets, + &reactor->probe[LATENCY_PROBE_QUEUE].count, + &reactor->probe[LATENCY_PROBE_QUEUE].sum_us, + &reactor->probe[LATENCY_PROBE_QUEUE].max_us, + &reactor->probe[LATENCY_PROBE_QUEUE].over_5ms, delay_ns / 1000); + latency_probe_record(reactor->probe[LATENCY_PROBE_SVC].buckets, + &reactor->probe[LATENCY_PROBE_SVC].count, + &reactor->probe[LATENCY_PROBE_SVC].sum_us, + &reactor->probe[LATENCY_PROBE_SVC].max_us, + &reactor->probe[LATENCY_PROBE_SVC].over_5ms, (t1 - t0) / 1000); +} +#endif + bool cwist_reactor_post(cwist_reactor_t *r, cwist_reactor_post_t *node) { if (!r || !node || !node->cb) return false; cwist_reactor_post_t *head = atomic_load_explicit(&r->post_head, memory_order_relaxed); @@ -1162,6 +1216,7 @@ static void flush_deferred(cwist_reactor_t *reactor) { /* Tagged RECV SQE: the connection owns the fd and shell, so * report the cancellation and let its handler clean up. */ reactor->rx_cb((void *)(sqe->user_data & ~(uint64_t)1u), -ECANCELED); + t_dispatch_aged = false; } } } @@ -1200,11 +1255,6 @@ static bool reactor_add_common(cwist_reactor_t *reactor, int fd, cwist_reactor_c if (!ev_ctx) return false; #ifdef __linux__ - if (latency_probe_enabled()) { - struct timespec ts; - clock_gettime(CLOCK_MONOTONIC, &ts); - ev_ctx->armed_ns = (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; - } if (!reactor->impl.use_epoll) { /* One-shot POLL_ADD: multishot (IORING_POLL_ADD_MULTI) was rejected * because the slot is recycled after firing, so a persistent poll @@ -1450,9 +1500,73 @@ static uint64_t reactor_late_ns(void) { return (uint64_t)v * 1000ull; } +/* Tail-elevation threshold (per-event, both reactor backends). An event + * armed this long ago has already queued past the knee of the arm-to-dispatch + * latency curve; when a dispatch batch holds it behind younger events, the + * run loop pulls it out and serves it first, and cwist_reactor_dispatch_late() + * reports true for its callback so the HTTP layer serves its buffered + * pipeline without yielding. Complements the round-duration late mode above: + * that one reacts to a slow round, this one to a slow *event* regardless of + * how young the round is. CWIST_REACTOR_TAIL_US overrides the 3500 us + * default; 0 disables. */ +static uint64_t reactor_tail_ns(void) { + static _Atomic long cached = -1; + long v = atomic_load_explicit(&cached, memory_order_relaxed); + if (v < 0) { + const char *s = getenv("CWIST_REACTOR_TAIL_US"); + if (s && *s) { + char *end = NULL; + long parsed = strtol(s, &end, 10); + v = (end != s && *end == '\0' && parsed >= 0 && parsed < 3600000000L) ? parsed : 3500; + } else { + v = 3500; + } + atomic_store_explicit(&cached, v, memory_order_relaxed); + } + return (uint64_t)v * 1000ull; +} + +uint64_t cwist_reactor_now_ns(void) { + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; +} + +bool cwist_reactor_event_aged(uint64_t now_ns, uint64_t armed_ns, uint64_t tail_ns) { + return tail_ns > 0 && armed_ns > 0 && now_ns - armed_ns >= tail_ns; +} + +uint64_t cwist_reactor_tail_ns(void) { + return reactor_tail_ns(); +} + +bool cwist_reactor_latency_probe_enabled(void) { + return latency_probe_enabled(); +} + +/* True while the run thread dispatches an event the tail heuristic aged out + * of its batch: its callback sees cwist_reactor_dispatch_late() == true. Set + * by the dispatch loops' elevation pass and by the HTTP RX completion + * handler (which owns the RECV arm timestamp); cleared when the callback + * returns, so it never leaks into the next event of the batch. */ +void cwist_reactor_dispatch_age_bypass(void) { + t_dispatch_aged = true; +} + +/* Delivery-time half of the tail heuristic (run thread): one clock read and + * compare against the slot's arm stamp; aged events raise the elevation + * counter and see cwist_reactor_dispatch_late() == true for the duration of + * their callback (the caller clears the flag after the callback returns). */ +static bool reactor_tail_mark_aged(cwist_reactor_t *reactor, uint64_t armed_ns, uint64_t tail_ns) { + if (!cwist_reactor_event_aged(cwist_reactor_now_ns(), armed_ns, tail_ns)) return false; + reactor->tail_elevations++; + t_dispatch_aged = true; + return true; +} + bool cwist_reactor_dispatch_late(void) { const cwist_reactor_t *r = t_run_reactor; - return r && r->dispatching && r->dispatch_late; + return r && r->dispatching && (r->dispatch_late || t_dispatch_aged); } #endif @@ -1555,12 +1669,40 @@ void cwist_reactor_run(cwist_reactor_t *reactor) { reactor_tsan_acquire(&reactor->impl); reactor->dispatching = true; const uint64_t round_budget_us = reactor_round_budget_us(); + const uint64_t tail_ns = reactor_tail_ns(); uint64_t round_start_ns = 0; if (round_budget_us || late_ns) { struct timespec ts; clock_gettime(CLOCK_MONOTONIC, &ts); round_start_ns = (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; } + /* Tail elevation (CWIST_REACTOR_TAIL_US, both backends): an event + * armed longer ago than the threshold has already waited out the + * tail; if this batch buries it behind younger completions it + * would wait out the whole round too. One clock read per round + * scans the batch for already-aged ev_ctx slots and dispatches + * them first; the main loop re-checks age at each delivery for + * events that age behind a slow earlier callback. Either way + * cwist_reactor_dispatch_late() reports true for the callback so + * the HTTP layer bypasses its yield. Consumed pre-pass slots are + * tagged user_data = 0 and skipped by the main loop. Tagged RX + * completions carry no reactor-side stamp; the HTTP RX handler + * runs the same check against the arm timestamp it owns. */ + if (tail_ns) { + for (uint32_t i = head; i != tail; i++) { + struct io_uring_cqe *cqe = + &reactor->impl.cqes[i & *reactor->impl.cq_ring_mask]; + const uint64_t ud = cqe->user_data; + if (ud == 0 || (ud & 1u)) continue; + reactor_event_ctx_t *ev_ctx = (reactor_event_ctx_t *)ud; + if (!reactor_tail_mark_aged(reactor, ev_ctx->armed_ns, tail_ns)) continue; + cqe->user_data = 0; + if (cqe->res >= 0) dispatch_ev_ctx(reactor, ev_ctx); + t_dispatch_aged = false; + free_reactor_ctx(reactor, ev_ctx); + __atomic_store_n(reactor->impl.cq_head, i + 1, __ATOMIC_RELEASE); + } + } /* Cooperative queuing: a busy round can carry hundreds of ready * connections in one CQE batch (the ring is 4096 deep). Foreign- * thread completions (cwist_async_defer, background jobs) queue @@ -1591,6 +1733,11 @@ void cwist_reactor_run(cwist_reactor_t *reactor) { } struct io_uring_cqe *cqe = &reactor->impl.cqes[head & *reactor->impl.cq_ring_mask]; uint64_t user_data = cqe->user_data; + if (user_data == 0) { + /* Served by the tail-elevation pre-pass. */ + head++; + continue; + } if (user_data & 1u) { /* Tagged data-path SQE (RX-uring RECV): the connection * pointer rides in the remaining bits. Dispatch on any @@ -1600,40 +1747,15 @@ void cwist_reactor_run(cwist_reactor_t *reactor) { * timestamp), not by the ev_ctx wrapper below. */ if (reactor->rx_cb) { reactor->rx_cb((void *)(user_data & ~(uint64_t)1u), cqe->res); + t_dispatch_aged = false; } } else { reactor_event_ctx_t *ev_ctx = (reactor_event_ctx_t *)user_data; if (ev_ctx) { - if (cqe->res >= 0) { - if (latency_probe_enabled()) { - struct timespec ts; - clock_gettime(CLOCK_MONOTONIC, &ts); - uint64_t t0 = - (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; - /* armed_ns == 0 means the slot was armed before - * the probe was enabled; fold it into the first - * bucket rather than producing a garbage delay. */ - uint64_t delay_ns = ev_ctx->armed_ns ? t0 - ev_ctx->armed_ns : 0; - ev_ctx->cb(ev_ctx->fd, ev_ctx->ctx); - clock_gettime(CLOCK_MONOTONIC, &ts); - uint64_t t1 = - (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; - latency_probe_record(reactor->probe[LATENCY_PROBE_QUEUE].buckets, - &reactor->probe[LATENCY_PROBE_QUEUE].count, - &reactor->probe[LATENCY_PROBE_QUEUE].sum_us, - &reactor->probe[LATENCY_PROBE_QUEUE].max_us, - &reactor->probe[LATENCY_PROBE_QUEUE].over_5ms, - delay_ns / 1000); - latency_probe_record(reactor->probe[LATENCY_PROBE_SVC].buckets, - &reactor->probe[LATENCY_PROBE_SVC].count, - &reactor->probe[LATENCY_PROBE_SVC].sum_us, - &reactor->probe[LATENCY_PROBE_SVC].max_us, - &reactor->probe[LATENCY_PROBE_SVC].over_5ms, - (t1 - t0) / 1000); - } else { - ev_ctx->cb(ev_ctx->fd, ev_ctx->ctx); - } - } + (void)(tail_ns && + reactor_tail_mark_aged(reactor, ev_ctx->armed_ns, tail_ns)); + if (cqe->res >= 0) dispatch_ev_ctx(reactor, ev_ctx); + t_dispatch_aged = false; free_reactor_ctx(reactor, ev_ctx); } } @@ -1689,6 +1811,24 @@ void cwist_reactor_run(cwist_reactor_t *reactor) { if (errno == EINTR) continue; break; } + /* Tail elevation (CWIST_REACTOR_TAIL_US): same heuristic as the + * uring batch above, on this backend's event array. Aged ev_ctx + * slots are served first and nulled out; the plain loop below + * skips them. RX-mode fds (low bit set) carry their arm stamp on + * the HTTP side, which runs the check in its completion handler. */ + reactor->dispatching = true; + const uint64_t tail_ns = reactor_tail_ns(); + if (tail_ns) { + for (int i = 0; i < n; i++) { + if ((events[i].data.u64 & 1u) || !events[i].data.ptr) continue; + reactor_event_ctx_t *ev_ctx = (reactor_event_ctx_t *)events[i].data.ptr; + if (!reactor_tail_mark_aged(reactor, ev_ctx->armed_ns, tail_ns)) continue; + events[i].data.ptr = NULL; + ev_ctx->cb(ev_ctx->fd, ev_ctx->ctx); + t_dispatch_aged = false; + free_reactor_ctx(reactor, ev_ctx); + } + } for (int i = 0; i < n; i++) { /* Low bit set: an RX-mode fd (fd << 1 | 1); ev_ctx slots are * at least 8-byte aligned and never carry it. */ @@ -1698,11 +1838,15 @@ void cwist_reactor_run(cwist_reactor_t *reactor) { } reactor_event_ctx_t *ev_ctx = (reactor_event_ctx_t *)events[i].data.ptr; if (ev_ctx) { + (void)(tail_ns && + reactor_tail_mark_aged(reactor, ev_ctx->armed_ns, tail_ns)); ev_ctx->cb(ev_ctx->fd, ev_ctx->ctx); + t_dispatch_aged = false; free_reactor_ctx(reactor, ev_ctx); } } if (reactor->rx_ready_n > 0) rx_epoll_drain_ready(reactor); + reactor->dispatching = false; } } #elif defined(__APPLE__) || defined(__FreeBSD__) || defined(__NetBSD__) || defined(__OpenBSD__) diff --git a/src/sys/io/reactor_rx.h b/src/sys/io/reactor_rx.h index 52edf5920..844b921f3 100644 --- a/src/sys/io/reactor_rx.h +++ b/src/sys/io/reactor_rx.h @@ -74,9 +74,10 @@ void cwist_reactor_set_rx_cb(cwist_reactor_t *reactor, cwist_rx_cb_t cb); /* Arm one IORING_OP_RECV on the reactor ring. The SQE references * [buf, buf+len) and must stay valid until the completion arrives, so the * caller must not move or consume that buffer while the SQE is in flight. - * At most one RECV per connection at a time. When the latency probe is - * enabled, *armed_ns (if non-NULL) receives the arm timestamp so the caller - * can compute queue delay. Run-thread calls during dispatch defer into the + * At most one RECV per connection at a time. *armed_ns (if non-NULL) always + * receives the arm timestamp: the HTTP completion handler runs the + * tail-elevation heuristic against it, and the latency probe records queue + * delay from it when enabled. Run-thread calls during dispatch defer into the * existing SQE batch; other threads submit immediately so parked workers * wake. Returns false when unsupported or on submission failure; the caller * then falls back to the legacy POLL path. */ @@ -92,14 +93,49 @@ void cwist_reactor_probe_record(cwist_reactor_t *reactor, bool queue, uint64_t s /* True while the calling thread dispatches a reactor callback in late mode: * the round it belongs to has run past CWIST_REACTOR_LATE_US (issue #319 * experiment), so the connection should finish its buffered work now - * instead of yielding. False off the run thread and outside dispatch. */ + * instead of yielding. Also true for a callback whose event was aged past + * CWIST_REACTOR_TAIL_US (tail-elevation heuristic, both backends): the event + * already queued that long, so it is served first and runs to completion. + * False off the run thread and outside dispatch. */ bool cwist_reactor_dispatch_late(void); +/* Tail-elevation heuristic (CWIST_REACTOR_TAIL_US, default 3500 us; 0 + * disables): CLOCK_MONOTONIC read, age check, and probe-enable query shared + * by the dispatch loops and the HTTP RX completion handler. */ +uint64_t cwist_reactor_now_ns(void); +uint64_t cwist_reactor_tail_ns(void); +bool cwist_reactor_event_aged(uint64_t now_ns, uint64_t armed_ns, uint64_t tail_ns); +bool cwist_reactor_latency_probe_enabled(void); + +/* Mark the remainder of the current dispatch callback as aged: used by the + * HTTP RX completion handler, which owns the RECV arm timestamp the reactor + * cannot see. cwist_reactor_dispatch_late() returns true until the callback + * returns to the run loop. */ +void cwist_reactor_dispatch_age_bypass(void); + #else /* !__linux__ */ static inline bool cwist_reactor_dispatch_late(void) { return false; } +static inline uint64_t cwist_reactor_now_ns(void) { + return 0; +} +static inline uint64_t cwist_reactor_tail_ns(void) { + return 0; +} +static inline bool cwist_reactor_event_aged(uint64_t now_ns, uint64_t armed_ns, + uint64_t tail_ns) { + (void)now_ns; + (void)armed_ns; + (void)tail_ns; + return false; +} +static inline bool cwist_reactor_latency_probe_enabled(void) { + return false; +} +static inline void cwist_reactor_dispatch_age_bypass(void) { +} static inline bool cwist_rx_uring_env_enabled(void) { return false; diff --git a/tests/test_http_fairness.c b/tests/test_http_fairness.c index 332a139db..8d609c0fb 100644 --- a/tests/test_http_fairness.c +++ b/tests/test_http_fairness.c @@ -167,6 +167,11 @@ int main(void) { setenv("CWIST_WORKERS", "1", 1); setenv("CWIST_WORKER_THREADS", "1", 1); setenv("CWIST_C1M_MODE", "1", 1); + /* Isolate cooperative yielding from the tail heuristic: a connection + * gated longer than CWIST_REACTOR_TAIL_US is *supposed* to finish its + * buffered pipeline without yielding, which would defeat this test's + * probe. The heuristic itself is covered by test_reactor_tail. */ + setenv("CWIST_REACTOR_TAIL_US", "0", 1); const char *delay = getenv("CWIST_TEST_HANDLER_DELAY_US"); delay_us = delay ? strtol(delay, NULL, 10) : 0; bool half_close = getenv("CWIST_TEST_HALF_CLOSE") != NULL; diff --git a/tests/test_reactor_late.c b/tests/test_reactor_late.c index 5fbb8b5d4..c676fd9a5 100644 --- a/tests/test_reactor_late.c +++ b/tests/test_reactor_late.c @@ -104,6 +104,10 @@ static result_t run_in_child(const char *late_env, int spin_us_each) { setenv("CWIST_REACTOR_LATE_US", late_env, 1); else unsetenv("CWIST_REACTOR_LATE_US"); + /* Isolate late-round mode from the tail heuristic: both feed + * cwist_reactor_dispatch_late(), and these batches always run past + * CWIST_REACTOR_TAIL_US because of the callback spins. */ + setenv("CWIST_REACTOR_TAIL_US", "0", 1); result_t r = run_batch(spin_us_each); ssize_t w = write(fds[1], &r, sizeof(r)); (void)w; diff --git a/tests/test_reactor_tail.c b/tests/test_reactor_tail.c new file mode 100644 index 000000000..8aa3be3cd --- /dev/null +++ b/tests/test_reactor_tail.c @@ -0,0 +1,245 @@ +#ifndef _GNU_SOURCE +#define _GNU_SOURCE +#endif +/* Tail-elevation heuristic (CWIST_REACTOR_TAIL_US): an event whose arm stamp + * is older than the threshold is served before younger events of the same + * batch and its callback observes cwist_reactor_dispatch_late() == true; + * below the threshold, with the knob at 0, and outside dispatch it is false. + * Both reactor backends. + * + * Same technique as test_reactor_late.c: include the reactor implementation + * directly, and run each configuration in its own process because the knob + * is cached on first read. + * + * Two mechanisms are exercised: + * - the round-start scan pulls an already-aged event to the front of the + * batch (arm one pipe, sleep past the threshold, arm the rest, run); + * - the delivery-time check catches events that age behind a slow earlier + * callback (first callback spins past the threshold). + */ +#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 { BATCH = 8 }; + +static cwist_reactor_t *loop; +static int pipes[BATCH][2]; +static int order[BATCH]; +static int late_seen[BATCH]; +static int handled; +static int spin_first_us; +static int bypass_in_first; /* model the HTTP RX handler raising the bypass */ + +static void spin_us(long us) { + struct timespec start, now; + clock_gettime(CLOCK_MONOTONIC, &start); + do { + clock_gettime(CLOCK_MONOTONIC, &now); + } while ((now.tv_sec - start.tv_sec) * 1000000L + (now.tv_nsec - start.tv_nsec) / 1000L < us); +} + +static void conn_cb(int fd, void *ctx) { + (void)fd; + int idx = *(int *)ctx; + char buf[8]; + ssize_t rc = read(fd, buf, sizeof(buf)); + (void)rc; + /* Model the HTTP RX handler: it may raise the bypass (its event aged on + * the HTTP side) before running the serve path that reads + * cwist_reactor_dispatch_late(). */ + if (handled == 0 && bypass_in_first) cwist_reactor_dispatch_age_bypass(); + if (handled == 0 && spin_first_us > 0) spin_us(spin_first_us); + order[handled] = idx; + late_seen[handled] = cwist_reactor_dispatch_late(); + handled++; + if (handled == BATCH) cwist_reactor_stop(loop); +} + +typedef struct { + int skipped; /* -1: backend unavailable */ + int order[BATCH]; + int late[BATCH]; + unsigned long long elevations; +} result_t; + +static void sleep_us(long us) { + struct timespec ts = {.tv_sec = us / 1000000, .tv_nsec = (us % 1000000) * 1000}; + nanosleep(&ts, NULL); +} + +/* Every pipe is readable before its arm, so all callbacks land in a single + * dispatch round. tail_env selects the knob (NULL = default); when + * sleep_before_rest is set, pipe 0 is armed first and the rest after a 2 ms + * sleep (pipe 0 is aged at the round-start scan, the rest are young). */ +static result_t run_batch(const char *tail_env, const char *backend_env, bool sleep_before_rest, + int spin_us_first, bool bypass_first) { + result_t res; + memset(&res, 0, sizeof(res)); + handled = 0; + spin_first_us = spin_us_first; + bypass_in_first = bypass_first ? 1 : 0; + memset(order, 0, sizeof(order)); + memset(late_seen, 0, sizeof(late_seen)); + if (tail_env) setenv("CWIST_REACTOR_TAIL_US", tail_env, 1); + if (backend_env) setenv("CWIST_REACTOR_BACKEND", backend_env, 1); + setenv("CWIST_REACTOR_LATE_US", "0", 1); /* isolate from late-round mode */ + + loop = cwist_reactor_create(); + assert(loop); + /* A kernel without io_uring support cannot run the uring backend. */ + if (!backend_env && loop->impl.use_epoll) { + cwist_reactor_destroy(loop); + res.skipped = -1; + return res; + } + for (int i = 0; i < BATCH; i++) { + assert(pipe(pipes[i]) == 0); + char b = 'x'; + assert(write(pipes[i][1], &b, 1) == 1); + } + assert(cwist_reactor_add(loop, pipes[0][0], conn_cb, &*(int[]){0}, sizeof(int))); + if (sleep_before_rest) sleep_us(2000); /* > the 1000 us test threshold */ + for (int i = 1; i < BATCH; i++) { + assert(cwist_reactor_add(loop, pipes[i][0], conn_cb, &*(int[]){i}, sizeof(int))); + } + cwist_reactor_run(loop); + assert(handled == BATCH); + memcpy(res.order, order, sizeof(order)); + memcpy(res.late, late_seen, sizeof(late_seen)); + res.elevations = (unsigned long long)loop->tail_elevations; + for (int i = 0; i < BATCH; i++) { + close(pipes[i][0]); + close(pipes[i][1]); + } + cwist_reactor_destroy(loop); + return res; +} + +static result_t run_in_child(const char *tail_env, const char *backend_env, + bool sleep_before_rest, int spin_us_first, bool bypass_first) { + int fds[2]; + assert(pipe(fds) == 0); + pid_t pid = fork(); + assert(pid >= 0); + if (pid == 0) { + close(fds[0]); + result_t r = + run_batch(tail_env, backend_env, sleep_before_rest, spin_us_first, bypass_first); + ssize_t w = write(fds[1], &r, sizeof(r)); + (void)w; + close(fds[1]); + _exit(0); + } + close(fds[1]); + result_t r; + memset(&r, 0, sizeof(r)); + r.skipped = -2; + ssize_t n = read(fds[0], &r, sizeof(r)); + close(fds[0]); + int status; + assert(waitpid(pid, &status, 0) == pid); + assert(WIFEXITED(status) && WEXITSTATUS(status) == 0); + assert(n == (ssize_t)sizeof(r)); + return r; +} + +/* Natural dispatch order is 0..BATCH-1 (uring: CQEs complete in submission + * order; the asserts that need an order are gated on strict_order). */ +static void expect_natural_order(const result_t *r, bool strict_order) { + if (strict_order) { + for (int i = 0; i < BATCH; i++) assert(r->order[i] == i); + } else { + assert(r->order[0] == 0); /* the aged pipe 0 must still be served first */ + } +} + +int main(void) { + /* Pure age check: threshold semantics. */ + assert(cwist_reactor_event_aged(4500, 1000, 3500)); /* age == tail */ + assert(cwist_reactor_event_aged(4501, 1000, 3500)); /* age > tail */ + assert(!cwist_reactor_event_aged(4499, 1000, 3500)); /* age < tail */ + assert(!cwist_reactor_event_aged(4500, 1000, 0)); /* disabled */ + assert(!cwist_reactor_event_aged(4500, 0, 3500)); /* unstamped */ + printf("pure age check passed\n"); + + const char *backends[2] = {NULL, "epoll"}; + for (int b = 0; b < 2; b++) { + const char *backend = backends[b]; + const bool strict = backend == NULL; /* uring CQEs complete in order */ + const char *tag = backend ? "epoll" : "uring"; + /* An ambient CWIST_REACTOR_BACKEND override is exercised by the + * explicit epoll block below. */ + if (!backend && getenv("CWIST_REACTOR_BACKEND")) continue; + + /* Round-start scan: pipe 0 aged (armed 2 ms ago, threshold 1 ms) is + * elevated; the rest are young both at the scan and at delivery. + * Exactly one elevation and the scan must not double-dispatch. */ + result_t scan = run_in_child("1000", backend, true, 0, false); + assert(scan.skipped == 0); + assert(scan.elevations == 1); + assert(scan.late[0] == 1); + for (int i = 1; i < BATCH; i++) assert(scan.late[i] == 0); + expect_natural_order(&scan, strict); + printf("%s: round-start scan elevates the aged event\n", tag); + + /* Delivery-time check: nobody is aged at the scan; the first callback + * spins 2 ms, so every later delivery of the same batch is aged. */ + result_t deliv = run_in_child("1000", backend, false, 2000, false); + assert(deliv.skipped == 0); + assert(deliv.elevations >= BATCH - 1); + assert(deliv.late[0] == 0); + for (int i = 1; i < BATCH; i++) assert(deliv.late[i] == 1); + printf("%s: delivery-time check catches batch-aged events\n", tag); + + /* Default threshold (3500 us): a 2 ms stamp never qualifies. */ + result_t def = run_in_child(NULL, backend, true, 0, false); + assert(def.skipped == 0); + assert(def.elevations == 0); + for (int i = 0; i < BATCH; i++) assert(def.late[i] == 0); + expect_natural_order(&def, strict); + printf("%s: default 3500 us threshold not crossed by 2 ms\n", tag); + + /* 60 ms threshold: neither the 2 ms scan case nor a 2 ms spin case + * qualifies. */ + result_t high = run_in_child("60000", backend, true, 2000, false); + assert(high.skipped == 0); + assert(high.elevations == 0); + for (int i = 0; i < BATCH; i++) assert(high.late[i] == 0); + printf("%s: 60000 us threshold elevates nothing\n", tag); + + /* 0 disables the heuristic entirely. */ + result_t off = run_in_child("0", backend, true, 2000, false); + assert(off.skipped == 0); + assert(off.elevations == 0); + for (int i = 0; i < BATCH; i++) assert(off.late[i] == 0); + printf("%s: 0 disables\n", tag); + + /* Bypass scoping: a handler that raises the bypass (the HTTP RX + * completion does) affects only its own callback; with the flag + * leaking into the rest of the batch the later late[] entries below + * would flip to 1. */ + result_t scope = run_in_child("60000", backend, false, 0, true); + assert(scope.skipped == 0); + assert(scope.elevations == 0); + assert(scope.late[0] == 1); + for (int i = 1; i < BATCH; i++) assert(scope.late[i] == 0); + printf("%s: bypass is scoped to its own callback\n", tag); + } + + printf("tail-elevation heuristic (CWIST_REACTOR_TAIL_US) test passed\n"); + return 0; +}