csp: mult slow-subscriber policies ('block / 'drop / 'timeout)
ober
d543a3189dc507f9c4ddcbaa322865c5e9706e36
--- a/docs/clojure-remaining.md +++ b/docs/clojure-remaining.md @@ -681,6 +681,18 @@ specs and a `timeout N` as the default-cutoff: time; if you want a different policy you create a new mult. Keep the same constraint in Jerboa — the policy is set once. +**[landed]** `make-mult` is extended to accept an optional policy +symbol (`'block`, `'drop`, `'timeout`) with the timeout ms as a third +argument. `'block` (default) preserves existing behaviour. `'drop` +uses `chan-try-put!` on every subscriber — slow subs silently miss +values. `'timeout N` polls `chan-try-put!` per subscriber with a +bounded deadline; a sub that still can't accept after N ms drops +that value. Arg validation raises on nonsense combinations +(`'timeout` without ms; non-`'timeout` with three args; unknown +policy symbol). The configured policy is fixed at creation and +exposed via the new `mult-policy` accessor. Covered by 8 tests in +`tests/test-csp.ss`. + ### 3.8 Parking go (research, likely deferred) **The problem.** Jerboa's `(go body ...)` spawns a real OS thread. Each @@ -1714,7 +1726,7 @@ in this doc. **[deferred]** items are non-goals. | `go` / `go-loop` | [current] (OS threads) | §3.8 deferred | | `to-chan`/`onto-chan`/`chan-reduce` | [current] | — | | `merge`/`split`/`pipe` | [current] | §3.6 landed | -| `mult`/`tap`/`untap` | [current] | §3.7 slow-sub policy | +| `mult`/`tap`/`untap` | [current] | §3.7 landed (drop / timeout) | | `pub`/`sub`/`unsub` | [current] | — | | `pipeline`/`pipeline-async` | [current] | — | | `promise-chan` | [current] | — | @@ -1724,7 +1736,7 @@ in this doc. **[deferred]** items are non-goals. | `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 | -| Mult slow-sub policies | [gap] | §3.7 | +| Mult slow-sub policies | [current] `(std csp ops)` | §3.7 landed | | Parked `go` (CPS) | [deferred] | §3.8 | | Transducer error handler on chan | [current] `(std csp clj)` | §3.1 landed | --- a/lib/std/csp/ops.sls +++ b/lib/std/csp/ops.sls @@ -17,7 +17,7 @@ ;; Composition chan-merge chan-split chan-classify-by chan-pipe-to ;; Broadcast: mult / tap / untap - make-mult mult? mult-source + make-mult mult? mult-source mult-policy tap! untap! untap-all! ;; Topic routing: pub / sub make-pub pub? pub-source @@ -344,42 +344,112 @@ ;; dynamically maintained set of subscriber channels. Subscribers ;; are added with `tap!` and removed with `untap!`. ;; - ;; Semantics trade-off: this implementation blocks the fan-out - ;; thread on the slowest subscriber. Clojure's core.async allows - ;; a timeout + drop policy per sub. For v1 we document the - ;; slow-subscriber hazard and expect users to hand slow subs a - ;; sliding / dropping buffer (see `make-channel/sliding`). + ;; Slow-subscriber policy + ;; ---------------------- + ;; The default policy ('block) preserves Clojure's core.async + ;; behaviour: the fan-out thread blocks on `chan-put!` until every + ;; subscriber has accepted the value, so a slow tap stalls the + ;; entire mult. Two additional policies are available at creation + ;; time for code that prefers drop-on-slow-consumer semantics: + ;; + ;; (make-mult src) ;; 'block (default) + ;; (make-mult src 'block) ;; same as (make-mult src) + ;; (make-mult src 'drop) ;; drop values for a slow sub + ;; ;; via chan-try-put! + ;; (make-mult src 'timeout 100) ;; give each sub up to 100ms; + ;; ;; drop if it still can't accept + ;; + ;; The policy is fixed at creation. `mult-policy` returns the + ;; configured policy symbol (useful for introspection and tests). (define-record-type %mult - (fields (immutable source) (mutable subs) (immutable lock))) - - (define (make-mult source) - (let ([m (make-%mult source '() (make-mutex))]) - (fork-thread - (lambda () + (fields (immutable source) + (mutable subs) + (immutable lock) + (immutable policy) ;; 'block | 'drop | 'timeout + (immutable send-fn))) ;; (lambda (sub val) ...) + + ;; Monotonic-ms clock for the 'timeout policy. + (define (%now-ms) + (let ([t (current-time 'time-monotonic)]) + (+ (* (time-second t) 1000) + (quotient (time-nanosecond t) 1000000)))) + + (define %mult-tick (make-time 'time-duration 1000000 0)) ;; 1 ms + + ;; Pre-built send functions per policy. + (define %mult-send-block + (lambda (s v) + ;; chan-put! errors if already closed elsewhere — guard so an + ;; untap of a still-open sub doesn't nuke the fan-out thread. + (guard (_ [else (void)]) (chan-put! s v)))) + + (define %mult-send-drop + (lambda (s v) + (guard (_ [else (void)]) (chan-try-put! s v)))) + + (define (%make-mult-send-timeout ms) + (lambda (s v) + (guard (_ [else (void)]) + (let ([deadline (+ (%now-ms) ms)]) (let loop () - (let ([v (chan-get! source)]) + (cond + [(chan-try-put! s v) #t] + [(>= (%now-ms) deadline) #f] ;; drop + [else (sleep %mult-tick) (loop)])))))) + + ;; Shared fan-out loop: reads from source, dispatches via send-fn, + ;; closes all subs on EOF. + (define (%mult-run! m) + (fork-thread + (lambda () + (let ([send-fn (%mult-send-fn m)]) + (let loop () + (let ([v (chan-get! (%mult-source m))]) (cond [(eof-object? v) (with-mutex (%mult-lock m) (for-each (lambda (s) - ;; chan-close! errors if already closed elsewhere - ;; — guard so untap of a still-open sub doesn't - ;; nuke the fan-out thread. (guard (_ [else (void)]) (chan-close! s))) (%mult-subs m)))] [else (let ([subs (with-mutex (%mult-lock m) (%mult-subs m))]) - (for-each - (lambda (s) - (guard (_ [else (void)]) (chan-put! s v))) - subs)) - (loop)]))))) - m)) + (for-each (lambda (s) (send-fn s v)) subs)) + (loop)])))))) + m) + + (define make-mult + (case-lambda + [(source) + (%mult-run! + (make-%mult source '() (make-mutex) 'block %mult-send-block))] + [(source policy) + (case policy + [(block) + (%mult-run! + (make-%mult source '() (make-mutex) 'block %mult-send-block))] + [(drop) + (%mult-run! + (make-%mult source '() (make-mutex) 'drop %mult-send-drop))] + [(timeout) + (error 'make-mult + "'timeout policy requires a timeout in milliseconds" policy)] + [else (error 'make-mult "unknown mult policy" policy)])] + [(source policy ms) + (unless (eq? policy 'timeout) + (error 'make-mult + "third argument only valid for 'timeout policy" policy)) + (unless (and (integer? ms) (positive? ms)) + (error 'make-mult + "timeout must be a positive integer (milliseconds)" ms)) + (%mult-run! + (make-%mult source '() (make-mutex) 'timeout + (%make-mult-send-timeout ms)))])) (define (mult? x) (%mult? x)) (define (mult-source m) (%mult-source m)) + (define (mult-policy m) (%mult-policy m)) (define (tap! m ch) (with-mutex (%mult-lock m) --- a/tests/test-csp.ss +++ b/tests/test-csp.ss @@ -259,6 +259,77 @@ (list (chan->list s1) (chan->list s2))) '((a b) (a b))) +;;; mult policies — default 'block, 'drop, 'timeout + +(test "mult-policy defaults to 'block" + (mult-policy (make-mult (make-channel))) + 'block) + +(test "mult 'block explicit" + (let ([m (make-mult (make-channel) 'block)]) + (mult-policy m)) + 'block) + +(test "mult 'drop — slow sub misses values without stalling fast sub" + (let* ([src (make-channel 8)] + [m (make-mult src 'drop)] + ;; fast sub: big buffer, drains everything + [fast (make-channel 16)] + ;; slow sub: unbuffered, we never read from it, so chan-try-put! + ;; returns #f and these values are dropped for the slow sub. + [slow (make-channel)]) + (tap! m fast) (tap! m slow) + (for-each (lambda (x) (chan-put! src x)) '(1 2 3 4 5 6 7 8)) + (chan-close! src) + (sleep (millis 120)) + (list (mult-policy m) + (chan->list fast))) + '(drop (1 2 3 4 5 6 7 8))) + +(test "mult 'drop — rejects 3-arg arity without 'timeout" + (guard (exn [#t 'err]) + (make-mult (make-channel) 'drop 100)) + 'err) + +(test "mult 'timeout — arity + ms validation" + (list + (guard (exn [#t 'no-ms]) (make-mult (make-channel) 'timeout)) + (guard (exn [#t 'bad-ms]) (make-mult (make-channel) 'timeout -5)) + (guard (exn [#t 'nonsense]) (make-mult (make-channel) 'bogus))) + '(no-ms bad-ms nonsense)) + +(test "mult 'timeout — mult-policy introspection" + (mult-policy (make-mult (make-channel) 'timeout 20)) + 'timeout) + +(test "mult 'timeout — slow sub misses values after deadline" + (let* ([src (make-channel 4)] + ;; 25ms deadline per value per sub — short enough to test, long + ;; enough that a buffered fast sub always accepts. + [m (make-mult src 'timeout 25)] + [fast (make-channel 16)] + [slow (make-channel)]) ;; never drained + (tap! m fast) (tap! m slow) + (for-each (lambda (x) (chan-put! src x)) '(a b c d)) + (chan-close! src) + ;; Give the fan-out thread enough time: ~25ms per dropped value, + ;; 4 values → up to 100ms of waiting on slow sub. Then close fan out. + (sleep (millis 250)) + (chan->list fast)) + '(a b c d)) + +(test "mult 'block — passes every value when all subs keep up" + (let* ([src (make-channel 4)] + [m (make-mult src 'block)] + [s1 (make-channel 8)] + [s2 (make-channel 8)]) + (tap! m s1) (tap! m s2) + (chan-put! src 'x) (chan-put! src 'y) (chan-put! src 'z) + (chan-close! src) + (sleep (millis 80)) + (list (chan->list s1) (chan->list s2))) + '((x y z) (x y z))) + (test "pub / sub by topic" (let* ([src (make-channel 10)] [p (make-pub src car)]