csp: dynamic fan-in via make-mix / admix! / toggle!
ober
e0318e58246f894bed6861cc1d270bc36a39c67e
--- a/docs/clojure-remaining.md +++ b/docs/clojure-remaining.md @@ -344,6 +344,30 @@ Clojure names: **Effort:** ~200 lines for the mix module + 80 lines of tests. One day. +**[landed]** Implemented in `lib/std/csp/mix.sls` and re-exported from +`(std csp ops)` as `make-mix`, `mix?`, `mix-out`, `mix-solo-mode`, +`admix!`, `unmix!`, `unmix-all!`, `toggle!`, `solo-mode!`. Clojure +short names in `(std csp clj)`: `mix`, `admix`, `unmix`, `unmix-all`, +`toggle`, `solo-mode`. + +Implementation notes: + +- The fan-in loop uses `alts!!` over `[control-ch + active-inputs]`. + The control channel is a size-1 channel poked non-blockingly on + every reconfigure (`admix!`, `unmix!`, `toggle!`, `solo-mode!`). +- After `alts!!` returns, the loop **re-snapshots** before forwarding + to handle the race where an input was unmixed/paused/muted after + the snapshot but before `alts!!` picked it. Values that no longer + belong to an effective-active, non-muted sub are dropped. +- Solo semantics: if any sub has `solo?` set, only solo'd subs are + active; non-solo subs get treated per `solo-mode` — either `'mute` + (default: still read, dropped) or `'pause` (not read at all). +- `toggle!` accepts the per-input flag map as an alist of alists, + an alist of plists, or a hashtable → hashtable/plist/alist. Flag + keys are `'mute`, `'pause`, `'solo`. + +Covered by 10 tests in `tests/test-csp.ss`. + **Risks:** - **Control-channel back-pressure.** The control channel needs to be @@ -1731,7 +1755,7 @@ in this doc. **[deferred]** items are non-goals. | `pipeline`/`pipeline-async` | [current] | — | | `promise-chan` | [current] | — | | `(chan n xform)` | [current] `(std csp clj)` | §3.1 landed | -| `mix`/`admix`/`toggle` | [gap] | §3.2 | +| `mix`/`admix`/`toggle` | [current] `(std csp mix)` | §3.2 landed | | Timer wheel | [gap] | §3.3 | | `put!`/`take!` with callbacks | [current] `(std csp ops)` | §3.4 landed | | `async/reduce`, `onto-chan!` | [current] `(std csp ops)` | §3.5 landed | --- a/lib/std/csp/clj.sls +++ b/lib/std/csp/clj.sls @@ -54,6 +54,8 @@ merge split split-by pipe ;; Broadcast mult tap untap untap-all + ;; Dynamic fan-in (mix) + mix admix unmix unmix-all toggle solo-mode ;; Topic pub sub unsub unsub-all ;; Pipelines @@ -226,6 +228,14 @@ (define untap untap!) (define untap-all untap-all!) + ;; Dynamic fan-in — Clojure names for (std csp mix) + (define mix make-mix) + (define admix admix!) + (define unmix unmix!) + (define unmix-all unmix-all!) + (define toggle toggle!) + (define solo-mode solo-mode!) + (define pub make-pub) (define sub sub!) (define unsub unsub!) new file mode 100644 --- /dev/null +++ b/lib/std/csp/mix.sls @@ -0,0 +1,300 @@ +#!chezscheme +;;; (std csp mix) — Clojure core.async `mix` / `admix` / `toggle` +;;; +;;; A `mix` is a dynamic fan-in. You create one with `(make-mix out)` +;;; pointing to a destination channel, then add inputs with +;;; `(admix! m ch)`, remove them with `(unmix! m ch)`, and per-input +;;; mute / pause / solo with `(toggle! m state-map)`. +;;; +;;; Per-input state flags +;;; --------------------- +;;; muted? read from the input but drop its values +;;; paused? skip the input entirely +;;; solo? mark the input as solo +;;; +;;; If ANY input has `solo?` set, the mix behaves as if only solo'd +;;; inputs exist, plus non-solo'd inputs get treated according to the +;;; mix's `solo-mode` (`'mute` by default, `'pause` is the other +;;; option). This matches core.async's solo semantics. +;;; +;;; Control channel +;;; --------------- +;;; Every reconfig (admix, unmix, toggle, solo-mode) pokes a size-1 +;;; control channel so the mix loop's `alts!!` unblocks and re-reads +;;; its input set. The poke is non-blocking — if the control channel +;;; is already full the mix loop will re-read anyway on its next pass, +;;; so dropped control signals are benign. +;;; +;;; This module is layered under (std csp ops) which re-exports the +;;; public API. Clojure-style short names (mix, admix, unmix, toggle, +;;; solo-mode) live in (std csp clj). + +(library (std csp mix) + (export + make-mix mix? mix-out mix-solo-mode + admix! unmix! unmix-all! + toggle! solo-mode!) + + (import (chezscheme) + (std csp) + (std csp select)) + + ;; ====================================================== + ;; Per-input state + ;; ====================================================== + + (define-record-type %mix-input-state + (fields (mutable muted?) + (mutable paused?) + (mutable solo?))) + + (define (%make-default-state) + (make-%mix-input-state #f #f #f)) + + ;; ====================================================== + ;; Mix record + ;; ====================================================== + + (define-record-type %mix + (fields (immutable out) + (mutable inputs) ;; alist: (ch . %mix-input-state) + (immutable lock) ;; guards inputs + solo-mode + (immutable control-ch) ;; size-1 reconfig signal + (mutable solo-mode))) ;; 'mute | 'pause + + (define (mix? x) (%mix? x)) + (define (mix-out m) (%mix-out m)) + (define (mix-solo-mode m) (%mix-solo-mode m)) + + (define (make-mix out) + (let ([m (make-%mix out '() (make-mutex) (make-channel 1) 'mute)]) + (fork-thread (lambda () (%mix-loop m))) + m)) + + ;; ====================================================== + ;; Internal helpers + ;; ====================================================== + + ;; Non-blocking poke of the control channel so a running alts!! + ;; unblocks and re-reads the input set. Failures (channel full) + ;; are fine — the loop re-reads anyway on each iteration. + (define (%poke-control! m) + (chan-try-put! (%mix-control-ch m) 'reconfigure)) + + (define (%find-entry m ch) + (assq ch (%mix-inputs m))) + + ;; Snapshot of current configuration under the lock. Returns + ;; (list solo-mode inputs-alist) where inputs-alist is a shallow + ;; copy so the loop can safely iterate without further locking. + (define (%snapshot m) + (with-mutex (%mix-lock m) + (list (%mix-solo-mode m) + (map (lambda (e) (cons (car e) (cdr e))) + (%mix-inputs m))))) + + ;; Given a snapshot, compute: + ;; active — channels to read from on this pass + ;; muted — subset of `active` whose values should be dropped + (define (%effective-sets snap) + (let* ([solo-mode (car snap)] + [inputs (cadr snap)] + [any-solo? + (let loop ([xs inputs]) + (cond + [(null? xs) #f] + [(%mix-input-state-solo? (cdr (car xs))) #t] + [else (loop (cdr xs))]))]) + (let loop ([xs inputs] [active '()] [muted '()]) + (cond + [(null? xs) (list (reverse active) (reverse muted))] + [else + (let* ([entry (car xs)] + [ch (car entry)] + [st (cdr entry)] + [paused? (%mix-input-state-paused? st)] + [muted? (%mix-input-state-muted? st)] + [solo? (%mix-input-state-solo? st)] + [eff-paused? + (or paused? + (and any-solo? + (not solo?) + (eq? solo-mode 'pause)))] + [eff-muted? + (or muted? + (and any-solo? + (not solo?) + (eq? solo-mode 'mute)))]) + (cond + [eff-paused? (loop (cdr xs) active muted)] + [eff-muted? (loop (cdr xs) (cons ch active) (cons ch muted))] + [else (loop (cdr xs) (cons ch active) muted)]))])))) + + ;; Remove a closed input from the alist. Idempotent. + (define (%unmix-internal! m ch) + (with-mutex (%mix-lock m) + (%mix-inputs-set! m + (remp (lambda (e) (eq? (car e) ch)) (%mix-inputs m))))) + + ;; ====================================================== + ;; Fan-in loop + ;; ====================================================== + ;; + ;; Every iteration: snapshot state, compute active+muted sets, + ;; alts!! over [control-ch + active inputs]. If nothing is tapped, + ;; still block on control-ch so we wake up when someone admixes + ;; the first input. + + (define (%mix-loop m) + (let loop () + (let* ([snap (%snapshot m)] + [sets (%effective-sets snap)] + [active (car sets)] + [specs (cons (%mix-control-ch m) active)]) + (let* ([pick (alts!! specs)] + [v (car pick)] + [ch (cadr pick)]) + (cond + ;; control-channel fired — just re-read state. + [(eq? ch (%mix-control-ch m)) + (cond + [(eof-object? v) + ;; Caller closed the control channel — tear down. + (void)] + [else (loop)])] + ;; an input closed — drop it from the alist and continue. + [(eof-object? v) + (%unmix-internal! m ch) + (loop)] + ;; normal value. We may have lost a race with a concurrent + ;; unmix/toggle: the input was active when we snapshotted + ;; but might have been removed, muted, or paused by the + ;; time alts!! returned. Re-snapshot and drop values that + ;; no longer belong to an effective-active, non-muted sub. + [else + (let* ([now-sets (%effective-sets (%snapshot m))] + [now-active (car now-sets)] + [now-muted (cadr now-sets)]) + (cond + [(not (memq ch now-active)) + ;; input was removed or paused since we snapshotted + (loop)] + [(memq ch now-muted) + ;; input became muted since we snapshotted + (loop)] + [else + (guard (_ [else (void)]) + (chan-put! (%mix-out m) v)) + (loop)]))]))))) + + ;; ====================================================== + ;; Public mutators + ;; ====================================================== + + (define (admix! m ch) + (with-mutex (%mix-lock m) + (unless (%find-entry m ch) + (%mix-inputs-set! m + (cons (cons ch (%make-default-state)) + (%mix-inputs m))))) + (%poke-control! m) + m) + + (define (unmix! m ch) + (with-mutex (%mix-lock m) + (%mix-inputs-set! m + (remp (lambda (e) (eq? (car e) ch)) (%mix-inputs m)))) + (%poke-control! m) + m) + + (define (unmix-all! m) + (with-mutex (%mix-lock m) + (%mix-inputs-set! m '())) + (%poke-control! m) + m) + + ;; toggle! m state-map + ;; + ;; `state-map` is an association list / hashtable mapping + ;; `channel → flags-alist`, where flags-alist is an alist of + ;; 'mute / 'pause / 'solo → boolean. Unspecified flags are left + ;; unchanged. If a channel isn't currently in the mix, it is added + ;; first (this matches Clojure's `toggle`, which accepts channels + ;; you haven't explicitly admixed yet). + ;; + ;; Accepts either an alist or a hashtable for state-map. + (define (toggle! m state-map) + (let ([pairs (%state-map->pairs state-map)]) + (with-mutex (%mix-lock m) + (for-each + (lambda (pair) + (let* ([ch (car pair)] + [flags (cdr pair)] + [entry (assq ch (%mix-inputs m))]) + (unless entry + (%mix-inputs-set! m + (cons (cons ch (%make-default-state)) (%mix-inputs m))) + (set! entry (assq ch (%mix-inputs m)))) + (let ([st (cdr entry)]) + (%apply-flags! st flags)))) + pairs)) + (%poke-control! m) + m)) + + (define (%state-map->pairs x) + (cond + [(hashtable? x) + (let-values ([(ks vs) (hashtable-entries x)]) + (let loop ([i 0] [acc '()]) + (cond + [(= i (vector-length ks)) (reverse acc)] + [else (loop (+ i 1) + (cons (cons (vector-ref ks i) (vector-ref vs i)) + acc))])))] + [(pair? x) x] + [(null? x) '()] + [else (error 'toggle! "state-map must be alist or hashtable" x)])) + + (define (%apply-flags! st flags) + (let ([apply-one! + (lambda (key val) + (case key + [(mute) (%mix-input-state-muted?-set! st val)] + [(pause) (%mix-input-state-paused?-set! st val)] + [(solo) (%mix-input-state-solo?-set! st val)] + [else (error 'toggle! "unknown flag key" key)]))]) + (cond + [(hashtable? flags) + (let-values ([(ks vs) (hashtable-entries flags)]) + (let loop ([i 0]) + (unless (= i (vector-length ks)) + (apply-one! (vector-ref ks i) (vector-ref vs i)) + (loop (+ i 1)))))] + [(list? flags) + ;; accept either alist ((mute . #t) ...) or property list + ;; (mute #t pause #f ...) + (cond + [(and (pair? flags) (pair? (car flags))) + (for-each + (lambda (p) (apply-one! (car p) (cdr p))) + flags)] + [else + (let loop ([xs flags]) + (cond + [(null? xs) (void)] + [(null? (cdr xs)) + (error 'toggle! "odd-length flags plist" flags)] + [else (apply-one! (car xs) (cadr xs)) (loop (cddr xs))]))])] + [else (error 'toggle! "flags must be alist, plist, or hashtable" flags)]))) + + ;; (solo-mode! m 'mute) — non-solo inputs get muted when any sub is solo + ;; (solo-mode! m 'pause) — non-solo inputs get paused when any sub is solo + (define (solo-mode! m mode) + (unless (memq mode '(mute pause)) + (error 'solo-mode! "mode must be 'mute or 'pause" mode)) + (with-mutex (%mix-lock m) + (%mix-solo-mode-set! m mode)) + (%poke-control! m) + m) + +) ;; end library --- a/lib/std/csp/ops.sls +++ b/lib/std/csp/ops.sls @@ -19,6 +19,10 @@ ;; Broadcast: mult / tap / untap make-mult mult? mult-source mult-policy tap! untap! untap-all! + ;; Dynamic fan-in: mix / admix / toggle + make-mix mix? mix-out mix-solo-mode + admix! unmix! unmix-all! + toggle! solo-mode! ;; Topic routing: pub / sub make-pub pub? pub-source sub! unsub! unsub-all! @@ -28,7 +32,8 @@ promise-channel-put! promise-channel-get!) (import (chezscheme) - (std csp)) + (std csp) + (std csp mix)) ;;; ====================================================== ;;; Collection bridges --- a/tests/test-csp.ss +++ b/tests/test-csp.ss @@ -35,6 +35,14 @@ ;; helper for sleeps in tests — chez make-time takes nanos + whole secs (make-time 'time-duration (* ms 1000000) 0)) +;; Drain whatever is currently buffered on `ch` without blocking. +;; Useful for tests that pump values into a channel from a background +;; thread but never close that channel (e.g. the mix out channel). +(define (chan->list-nowait ch) + (let loop ([acc '()]) + (let ([v (chan-try-get ch)]) + (if v (loop (cons v acc)) (reverse acc))))) + (printf "--- CSP / core.async ---~%~%") ;;; ======== Phase 0: buffered + buffer policies ======== @@ -330,6 +338,144 @@ (list (chan->list s1) (chan->list s2))) '((x y z) (x y z))) +;;; mix / admix / toggle — dynamic fan-in + +(test "mix — basic fan-in from two sources" + (let* ([out (make-channel 16)] + [m (make-mix out)] + [a (make-channel 4)] + [b (make-channel 4)]) + (admix m a) (admix m b) + (chan-put! a 1) (chan-put! a 2) + (chan-put! b 3) (chan-put! b 4) + (chan-close! a) (chan-close! b) + (sleep (millis 100)) + (list-sort < (chan->list-nowait out))) + '(1 2 3 4)) + +(test "mix — unmix stops forwarding from a source" + (let* ([out (make-channel 16)] + [m (make-mix out)] + [a (make-channel 4)] + [b (make-channel 4)]) + (admix m a) (admix m b) + (chan-put! a 'a1) + (sleep (millis 30)) + (unmix m a) + (chan-put! a 'a2) ;; should NOT appear in out + (chan-put! b 'b1) + (sleep (millis 60)) + (chan-close! b) + (sleep (millis 60)) + ;; a is still open but not in the mix, so a2 should not land. + (list-sort (lambda (x y) (string<? (symbol->string x) (symbol->string y))) + (chan->list-nowait out))) + '(a1 b1)) + +(test "mix — unmix-all clears all inputs" + (let* ([out (make-channel 16)] + [m (make-mix out)] + [a (make-channel 4)] + [b (make-channel 4)]) + (admix m a) (admix m b) + (chan-put! a 'x) + (sleep (millis 30)) + (unmix-all! m) + (chan-put! a 'nope) (chan-put! b 'nope) + (sleep (millis 60)) + (chan->list-nowait out)) + '(x)) + +(test "mix — mute drops values from one source" + (let* ([out (make-channel 16)] + [m (make-mix out)] + [a (make-channel 4)] + [b (make-channel 4)]) + (admix m a) (admix m b) + (toggle m (list (cons a '((mute . #t))))) + (chan-put! a 'muted-1) (chan-put! a 'muted-2) + (chan-put! b 'kept-1) + (chan-close! a) (chan-close! b) + (sleep (millis 120)) + (chan->list-nowait out)) + '(kept-1)) + +(test "mix — pause skips reading from a source" + (let* ([out (make-channel 16)] + [m (make-mix out)] + [a (make-channel 4)] + [b (make-channel 4)]) + (admix m a) (admix m b) + (toggle m (list (cons a '((pause . #t))))) + (chan-put! a 'paused-a) ;; sits in buffer, not drained + (chan-put! b 'from-b) + (sleep (millis 80)) + ;; unpause a, flush + (toggle m (list (cons a '((pause . #f))))) + (sleep (millis 80)) + (chan-close! a) (chan-close! b) + (sleep (millis 60)) + (list-sort (lambda (x y) (string<? (symbol->string x) (symbol->string y))) + (chan->list-nowait out))) + '(from-b paused-a)) + +(test "mix — solo defaults to mute for non-solo inputs" + (let* ([out (make-channel 16)] + [m (make-mix out)] + [a (make-channel 4)] + [b (make-channel 4)]) + (admix m a) (admix m b) + (toggle m (list (cons a '((solo . #t))))) + ;; a is solo, so b is effectively muted (solo-mode default = 'mute) + (chan-put! a 'solo-a) (chan-put! b 'muted-b) + (chan-close! a) (chan-close! b) + (sleep (millis 120)) + (chan->list-nowait out)) + '(solo-a)) + +(test "mix — solo-mode 'pause blocks non-solo reads" + (let* ([out (make-channel 16)] + [m (make-mix out)] + [a (make-channel 4)] + [b (make-channel 4)]) + (solo-mode m 'pause) + (admix m a) (admix m b) + (toggle m (list (cons a '((solo . #t))))) + (chan-put! a 'solo-a) (chan-put! b 'blocked-b) + (sleep (millis 80)) + ;; now remove solo; b's value should finally drain + (toggle m (list (cons a '((solo . #f))))) + (sleep (millis 80)) + (chan-close! a) (chan-close! b) + (sleep (millis 60)) + (list (mix-solo-mode m) + (list-sort (lambda (x y) (string<? (symbol->string x) (symbol->string y))) + (chan->list-nowait out)))) + '(pause (blocked-b solo-a))) + +(test "mix — toggle with plist syntax" + (let* ([out (make-channel 16)] + [m (make-mix out)] + [a (make-channel 4)]) + (admix m a) + (toggle m (list (cons a '(mute #t)))) + (chan-put! a 'dropped) + (chan-close! a) + (sleep (millis 80)) + (chan->list-nowait out)) + '()) + +(test "mix? predicate" + (list (mix? (make-mix (make-channel))) + (mix? (make-channel)) + (mix? 42)) + '(#t #f #f)) + +(test "solo-mode! validates arg" + (guard (exn [#t 'err]) + (solo-mode! (make-mix (make-channel)) 'bogus)) + 'err) + (test "pub / sub by topic" (let* ([src (make-channel 10)] [p (make-pub src car)]