csp: add timer-wheel for timeout behind JERBOA_CSP_TIMER_WHEEL
ober
f4be24a479d1cea3380931db15368d8d66edcc39
--- a/docs/clojure-remaining.md +++ b/docs/clojure-remaining.md @@ -380,7 +380,22 @@ Covered by 10 tests in `tests/test-csp.ss`. ### 3.3 Timer wheel for `timeout` -**Current behaviour.** `(timeout ms)` at `lib/std/csp/select.sls:259` +**[landed]** Both implementations live in `lib/std/csp/select.sls`. The +default `timeout` is still thread-per-deadline; setting +`JERBOA_CSP_TIMER_WHEEL=1` in the environment at Scheme start flips +`timeout` to the wheel-backed dispatch. A new `wheel-timeout` export +always routes through the wheel so callers (and tests) can opt in +without restarting the process. The wheel is a single long-lived +thread that owns a min-heap of absolute deadlines (via +`(std misc pqueue)`) plus a size-1 wake-up channel. Enqueue is +O(log n), the main loop sleeps in 5ms chunks, and the wake-up channel +short-circuits the sleep when a new shorter deadline arrives. The +singleton wheel is built lazily under a double-checked lock so code +that never calls `timeout` (or `wheel-timeout`) pays nothing for the +timer thread. Covered by the timer-wheel subsection of +`tests/test-csp.ss`. + +**Current behaviour.** `(timeout ms)` at `lib/std/csp/select.sls` creates a fresh channel and spawns one helper thread that sleeps `ms` then closes the channel. For low-rate timeouts (tens per second) this is fine. For high-rate short-timeout workloads (rate limiting, retry @@ -1746,7 +1761,7 @@ in this doc. **[deferred]** items are non-goals. | Fixed/sliding/dropping buffers | [current] | — | | `>!!`/`<!!`/`close!`/`poll!`/`offer!` | [current] | — | | `alts!`/`alt!` with priority/default | [current] | — | -| `timeout` channel | [current] (thread-per-timeout) | §3.3 improves | +| `timeout` channel | [current] (thread-per-timeout + opt-in wheel) | §3.3 landed | | `go` / `go-loop` | [current] (OS threads) | §3.8 deferred | | `to-chan`/`onto-chan`/`chan-reduce` | [current] | — | | `merge`/`split`/`pipe` | [current] | §3.6 landed | @@ -1756,7 +1771,7 @@ in this doc. **[deferred]** items are non-goals. | `promise-chan` | [current] | — | | `(chan n xform)` | [current] `(std csp clj)` | §3.1 landed | | `mix`/`admix`/`toggle` | [current] `(std csp mix)` | §3.2 landed | -| Timer wheel | [gap] | §3.3 | +| Timer wheel | [current] `(std csp select)` | §3.3 landed | | `put!`/`take!` with callbacks | [current] `(std csp ops)` | §3.4 landed | | `async/reduce`, `onto-chan!` | [current] `(std csp ops)` | §3.5 landed | | `split` n-way | [current] `(std csp ops)` | §3.6 landed | --- a/lib/std/csp/select.sls +++ b/lib/std/csp/select.sls @@ -53,11 +53,15 @@ ;; (it's re-exported from (std csp clj) for convenience). default ;; Timeout channel - timeout timeout-channel) + timeout timeout-channel + ;; Force the timer-wheel implementation regardless of env var. + ;; Useful for tests and for callers that always want the wheel. + wheel-timeout) (import (chezscheme) (std csp) - (std event)) + (std event) + (std misc pqueue)) ;; ========================================================== ;; Event bridge — expose a channel as a (std event) event. @@ -251,12 +255,138 @@ ;; ========================================================== ;; timeout — channel that closes itself after N milliseconds. ;; - ;; Spawns one helper thread per timeout. At the scale of a few - ;; hundred outstanding timeouts per second that's fine; above - ;; that, use a timer-wheel (see Phase 4 in core-async.md). + ;; Two implementations selected at library load time by the + ;; `JERBOA_CSP_TIMER_WHEEL` environment variable. + ;; + ;; Default (env unset or !=1) + ;; -------------------------- + ;; One helper thread per timeout. Cheap to spin up and fine up to + ;; a few hundred outstanding timeouts per second. Each `(timeout N)` + ;; allocates a thread that sleeps N ms then closes the channel. + ;; + ;; Wheel mode (JERBOA_CSP_TIMER_WHEEL=1) + ;; ------------------------------------- + ;; A single long-lived timer thread owns a min-heap of absolute + ;; deadlines. Enqueue is O(log n), dispatch is O(log n) per fire, + ;; and there is no per-deadline thread — appropriate for high-rate + ;; short-timeout workloads (rate limiting, retry back-off). + ;; + ;; Chez's `condition-wait` has no timed variant, so the wheel's + ;; "wait until next deadline" path sleeps in 5ms chunks and + ;; short-circuits through a size-1 wake-up channel when a new + ;; shorter deadline is enqueued. 5ms is the practical granularity + ;; floor; a deadline 3ms away may fire at 5ms. ;; ========================================================== - (define (timeout ms) + (define (ms->time ms) + (let* ([whole-secs (quotient ms 1000)] + [rem-ms (remainder ms 1000)] + [nanos (* rem-ms 1000000)]) + (make-time 'time-duration nanos whole-secs))) + + (define (%now-ms) + (let ([t (current-time 'time-monotonic)]) + (+ (* (time-second t) 1000) + (quotient (time-nanosecond t) 1000000)))) + + ;; -- Timer wheel -------------------------------------------- + + (define-record-type %timer-wheel + (fields (immutable heap) ;; pqueue of (deadline . chan) + (immutable lock) ;; mutex guarding heap + (immutable wake-ch))) ;; size-1 wake-up channel + + (define (%make-timer-wheel) + (let ([w (make-%timer-wheel + (make-pqueue (lambda (a b) (< (car a) (car b)))) + (make-mutex) + (make-channel 1))]) + (fork-thread (lambda () (%timer-wheel-loop w))) + w)) + + ;; Push (deadline . ch) onto the heap and poke the wake-up channel + ;; so a sleeping timer thread rechecks. The poke is non-blocking; + ;; if the wake channel is already full (previous poke not yet + ;; consumed) we drop the new poke — one is enough. + (define (%timer-wheel-enqueue! w deadline ch) + (with-mutex (%timer-wheel-lock w) + (pqueue-push! (%timer-wheel-heap w) (cons deadline ch))) + (chan-try-put! (%timer-wheel-wake-ch w) 'poke)) + + ;; Sleep in 5ms chunks until `until` (absolute deadline in ms) has + ;; passed, or a wake-up message arrives. Returns 'expired if the + ;; deadline was reached, 'woken if short-circuited. + (define %wheel-chunk-ms 5) + (define (%wait-until until wake-ch) + (let loop () + (let* ([now (%now-ms)] + [rem (- until now)]) + (cond + [(<= rem 0) 'expired] + [(chan-try-get wake-ch) 'woken] + [else + (let ([chunk (if (< rem %wheel-chunk-ms) rem %wheel-chunk-ms)]) + (sleep (ms->time chunk)) + (loop))])))) + + ;; Pop every entry whose deadline is <= now. Returns the list of + ;; channels to close, in fire order. + (define (%drain-expired! w now) + (with-mutex (%timer-wheel-lock w) + (let loop ([acc '()]) + (cond + [(pqueue-empty? (%timer-wheel-heap w)) (reverse acc)] + [(<= (car (pqueue-peek (%timer-wheel-heap w))) now) + (loop (cons (cdr (pqueue-pop! (%timer-wheel-heap w))) acc))] + [else (reverse acc)])))) + + ;; Main loop. If the heap is empty, block on the wake-up channel. + ;; Otherwise peek the min deadline: if due, drain+close; if not, + ;; chunk-sleep until it fires or a wake arrives. + (define (%timer-wheel-loop w) + (let loop () + (let ([next + (with-mutex (%timer-wheel-lock w) + (cond + [(pqueue-empty? (%timer-wheel-heap w)) #f] + [else (car (pqueue-peek (%timer-wheel-heap w)))]))]) + (cond + [(not next) + (chan-get! (%timer-wheel-wake-ch w)) + (loop)] + [else + (let ([now (%now-ms)]) + (cond + [(<= next now) + (for-each (lambda (ch) + (guard (_ [else (void)]) + (chan-close! ch))) + (%drain-expired! w now)) + (loop)] + [else + (%wait-until next (%timer-wheel-wake-ch w)) + (loop)]))])))) + + ;; -- Dispatch ----------------------------------------------- + + (define %use-wheel? + (equal? (getenv "JERBOA_CSP_TIMER_WHEEL") "1")) + + ;; Lazily built so libraries that never call `timeout` don't pay + ;; for a timer thread. The initial `#f` is replaced on first use + ;; under the singleton lock. + (define %timer-wheel-singleton #f) + (define %timer-wheel-init-lock (make-mutex)) + + (define (%ensure-timer-wheel!) + (or %timer-wheel-singleton + (with-mutex %timer-wheel-init-lock + (or %timer-wheel-singleton + (let ([w (%make-timer-wheel)]) + (set! %timer-wheel-singleton w) + w))))) + + (define (%thread-timeout ms) (let ([ch (make-channel)]) (fork-thread (lambda () @@ -264,12 +394,19 @@ (chan-close! ch))) ch)) - (define timeout-channel timeout) + ;; Always uses the timer wheel, regardless of JERBOA_CSP_TIMER_WHEEL. + ;; Lazily spins up the singleton on first call. + (define (wheel-timeout ms) + (let ([ch (make-channel)] + [w (%ensure-timer-wheel!)]) + (%timer-wheel-enqueue! w (+ (%now-ms) ms) ch) + ch)) - (define (ms->time ms) - (let* ([whole-secs (quotient ms 1000)] - [rem-ms (remainder ms 1000)] - [nanos (* rem-ms 1000000)]) - (make-time 'time-duration nanos whole-secs))) + (define (timeout ms) + (if %use-wheel? + (wheel-timeout ms) + (%thread-timeout ms))) + + (define timeout-channel timeout) ) ;; end library --- a/tests/test-csp.ss +++ b/tests/test-csp.ss @@ -127,6 +127,56 @@ (eof-object? (car pick))) #t) +;; ---- Timer wheel (§3.3) -------------------------------------- +;; +;; `wheel-timeout` always uses the timer-wheel implementation, +;; regardless of JERBOA_CSP_TIMER_WHEEL. These tests exercise the +;; wheel directly so CI doesn't need two runs. + +(test "wheel-timeout fires" + (let ([pick (alts!! (list (wheel-timeout 15)))]) + (eof-object? (car pick))) + #t) + +(test "wheel-timeout two deadlines fire in order" + ;; Start two wheel timeouts simultaneously and verify the + ;; shorter one is observable-closed first. + (let* ([t1 (wheel-timeout 10)] + [t2 (wheel-timeout 50)]) + (sleep (millis 20)) + (list (chan-closed? t1) (chan-closed? t2))) + '(#t #f)) + +(test "wheel-timeout new-shorter-deadline wakes sleeper" + ;; Enqueue a long deadline first, then a short one; the wake-ch + ;; should let the wheel pick up the short one before the long. + (let* ([long (wheel-timeout 200)]) + (sleep (millis 5)) ;; let wheel start sleeping on long + (let ([short (wheel-timeout 15)]) + (sleep (millis 35)) + (list (chan-closed? short) (chan-closed? long)))) + '(#t #f)) + +(test "wheel-timeout many concurrent fire" + ;; 20 concurrent deadlines, all short. Confirm every channel + ;; ends up closed after one global wait. + (let ([chs (let loop ([n 0] [acc '()]) + (if (= n 20) + (reverse acc) + (loop (+ n 1) + (cons (wheel-timeout (+ 5 (random 20))) acc))))]) + (sleep (millis 60)) + (for-all chan-closed? chs)) + #t) + +(test "wheel-timeout integrates with alts!!" + ;; Race a normal channel against the wheel. + (let* ([c (make-channel)] + [t (wheel-timeout 10)] + [pick (alts!! (list c t))]) + (and (eq? (cadr pick) t) (eof-object? (car pick)))) + #t) + (test "alt!! dispatches to winning channel" (let ([c1 (make-channel 1)] [c2 (make-channel 1)]) (chan-put! c1 42)