csp: n-way split via chan-classify-by + split-by alias
ober
d59898300f9be7b4c702704598610b871ca58c15
--- a/docs/clojure-remaining.md +++ b/docs/clojure-remaining.md @@ -615,6 +615,25 @@ chan-classify-by **Effort:** ~40 lines + tests. One hour. +**[landed]** `chan-classify-by` is in `(std csp ops)`; `(std csp clj)` +re-exports it as both `chan-classify-by` and the short Clojure-style +alias `split-by`. The implementation extends the sketch in two ways: +it is thread-safe (a mutex protects the hashtable so callers can read +while the classifier is writing), and it accepts an optional +`initial-keys` list that eagerly pre-creates channels so tests and +known-universe callers can look up channels before the classifier has +processed any values. Arities: + +```scheme +(chan-classify-by f ch) +(chan-classify-by f ch buf-fn) +(chan-classify-by f ch buf-fn initial-keys) +``` + +Default `buf-fn` makes an unbuffered channel; all output channels +(both pre-populated and lazily-created) are closed once the source +closes. Covered by 5 tests in `tests/test-csp.ss`. + ### 3.7 Mult slow-subscriber policy **Current behaviour.** `make-mult` at `lib/std/csp/ops.sls` fans a source @@ -1694,7 +1713,7 @@ in this doc. **[deferred]** items are non-goals. | `timeout` channel | [current] (thread-per-timeout) | §3.3 improves | | `go` / `go-loop` | [current] (OS threads) | §3.8 deferred | | `to-chan`/`onto-chan`/`chan-reduce` | [current] | — | -| `merge`/`split`/`pipe` | [current] | §3.6 n-way split | +| `merge`/`split`/`pipe` | [current] | §3.6 landed | | `mult`/`tap`/`untap` | [current] | §3.7 slow-sub policy | | `pub`/`sub`/`unsub` | [current] | — | | `pipeline`/`pipeline-async` | [current] | — | @@ -1704,7 +1723,7 @@ in this doc. **[deferred]** items are non-goals. | 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 | -| `split` n-way | [gap] | §3.6 | +| `split` n-way | [current] `(std csp ops)` | §3.6 landed | | Mult slow-sub policies | [gap] | §3.7 | | Parked `go` (CPS) | [deferred] | §3.8 | | Transducer error handler on chan | [current] `(std csp clj)` | §3.1 landed | --- a/lib/std/csp/clj.sls +++ b/lib/std/csp/clj.sls @@ -51,7 +51,7 @@ to-chan onto-chan onto-chan! onto-chan!! async-reduce ;; Composition - merge split pipe + merge split split-by pipe ;; Broadcast mult tap untap untap-all ;; Topic @@ -211,6 +211,10 @@ (define merge chan-merge) (define split chan-split) + ;; n-way classifier — not in core.async, but common enough in third + ;; party libs (e.g. dispatch / group-by-channel) to deserve a short + ;; name here. Returns a hashtable keyed by classifier output. + (define split-by chan-classify-by) (define pipe chan-pipe-to) ;; Clojure core.async's `async/reduce` — an async fold over a --- a/lib/std/csp/ops.sls +++ b/lib/std/csp/ops.sls @@ -15,7 +15,7 @@ ;; Non-blocking callbacks put! take! ;; Composition - chan-merge chan-split chan-pipe-to + chan-merge chan-split chan-classify-by chan-pipe-to ;; Broadcast: mult / tap / untap make-mult mult? mult-source tap! untap! untap-all! @@ -260,6 +260,61 @@ [else (chan-put! fch v) (loop)]))))) (list tch fch))) + ;; (chan-classify-by f ch) — n-way split by key + ;; (chan-classify-by f ch buf-fn) — per-class buffer factory + ;; (chan-classify-by f ch buf-fn initial-keys) + ;; + ;; Fan `ch` out to per-class channels. `f` is a classifier: it maps + ;; each incoming value to a key (any `equal?`-comparable value). For + ;; every new key the classifier sees, it calls `buf-fn` with the key + ;; to construct a fresh output channel; `buf-fn` defaults to making + ;; an unbuffered channel. `initial-keys` is a list of expected keys + ;; whose channels are pre-created eagerly, so callers can look them + ;; up before the classifier has processed any values (useful in + ;; tests and when you know the class universe in advance). + ;; + ;; Returns a Chez hashtable mapping keys → output channels. Every + ;; output channel is closed when the source `ch` closes. Access to + ;; the hashtable is serialised internally via a mutex, so callers + ;; may safely use `hashtable-ref` at any time — but new keys only + ;; become visible after the classifier has seen a value bearing + ;; that key. + (define chan-classify-by + (case-lambda + [(f ch) + (chan-classify-by f ch (lambda (_k) (make-channel)) '())] + [(f ch buf-fn) + (chan-classify-by f ch buf-fn '())] + [(f ch buf-fn initial-keys) + (let ([outs (make-hashtable equal-hash equal?)] + [lk (make-mutex)]) + ;; Eagerly create channels for any caller-declared keys. + (for-each + (lambda (k) (hashtable-set! outs k (buf-fn k))) + initial-keys) + (fork-thread + (lambda () + (let loop () + (let ([v (chan-get! ch)]) + (cond + [(eof-object? v) + ;; Close every output channel once the source is done. + (with-mutex lk + (let-values ([(_ks vs) (hashtable-entries outs)]) + (vector-for-each chan-close! vs)))] + [else + (let ([out-ch + (with-mutex lk + (let* ([k (f v)] + [existing (hashtable-ref outs k #f)]) + (or existing + (let ([c (buf-fn k)]) + (hashtable-set! outs k c) + c))))]) + (chan-put! out-ch v)) + (loop)]))))) + outs)])) + ;; (chan-pipe-to from to) → close `to` on EOF ;; (chan-pipe-to from to close?) → don't close `to` when ;; from closes if #f. --- a/tests/test-csp.ss +++ b/tests/test-csp.ss @@ -161,6 +161,83 @@ (list e o)) '((2 4 6) (1 3 5))) +;;; chan-classify-by — n-way classifier. Uses `initial-keys` so we can +;;; grab per-class channels immediately without racing the helper thread. + +(test "chan-classify-by — three classes with initial-keys" + (let* ([in (to-chan '(1 2 3 4 5 6 7 8 9))] + [classify + (lambda (n) + (cond [(zero? (modulo n 3)) 'third] + [(even? n) 'even] + [else 'odd]))] + [tbl (chan-classify-by classify in + (lambda (_k) (make-channel 16)) + '(third even odd))] + [e (list-sort < (chan->list (hashtable-ref tbl 'even #f)))] + [o (list-sort < (chan->list (hashtable-ref tbl 'odd #f)))] + [t (list-sort < (chan->list (hashtable-ref tbl 'third #f)))]) + (list e o t)) + '((2 4 8) (1 5 7) (3 6 9))) + +(test "chan-classify-by — default buf-fn, two classes" + (let* ([in (to-chan '("aa" "bbb" "cc" "dddd" "e"))] + [tbl (chan-classify-by string-length in)]) + ;; default buf-fn makes unbuffered channels, so we need to read on + ;; helper threads. Use chan->list synchronously per-key since the + ;; helper thread will block putting until we drain. + ;; Simpler: wait for the source to drain and read via the table. + (sleep (millis 30)) + (list (list-sort string<? (chan->list (hashtable-ref tbl 1 #f))) + (list-sort string<? (chan->list (hashtable-ref tbl 2 #f))) + (list-sort string<? (chan->list (hashtable-ref tbl 3 #f))) + (list-sort string<? (chan->list (hashtable-ref tbl 4 #f))))) + '(("e") ("aa" "cc") ("bbb") ("dddd"))) + +(test "chan-classify-by — initial-keys eager channels are closed on EOF" + (let* ([in (to-chan '())] ;; empty source + [tbl (chan-classify-by car in + (lambda (_k) (make-channel 4)) + '(a b c))]) + (sleep (millis 20)) + (list (eof-object? (chan-get! (hashtable-ref tbl 'a #f))) + (eof-object? (chan-get! (hashtable-ref tbl 'b #f))) + (eof-object? (chan-get! (hashtable-ref tbl 'c #f))))) + '(#t #t #t)) + +(test "chan-classify-by — new keys appear in table as they are seen" + (let* ([in (to-chan '(apple ant banana bear cherry))] + [tbl (chan-classify-by + (lambda (sym) (string-ref (symbol->string sym) 0)) + in + (lambda (_k) (make-channel 8)))]) + ;; Drain helper: wait until classifier has processed everything, then + ;; enumerate the keys in the table. + (sleep (millis 30)) + (list + (list-sort < (map char->integer (vector->list + (hashtable-keys tbl)))) + (chan->list (hashtable-ref tbl #\a #f)) + (chan->list (hashtable-ref tbl #\b #f)) + (chan->list (hashtable-ref tbl #\c #f)))) + (list (list-sort < (list (char->integer #\a) + (char->integer #\b) + (char->integer #\c))) + '(apple ant) + '(banana bear) + '(cherry))) + +;; Clojure-style split-by alias +(test "split-by — Clojure alias for chan-classify-by" + (let* ([in (to-chan '(1 2 3 4 5 6))] + [tbl (split-by (lambda (n) (if (even? n) 'even 'odd)) + in + (lambda (_k) (make-channel 8)) + '(even odd))]) + (list (list-sort < (chan->list (hashtable-ref tbl 'even #f))) + (list-sort < (chan->list (hashtable-ref tbl 'odd #f))))) + '((2 4 6) (1 3 5))) + (test "chan-pipe-to" (let* ([in (to-chan '(1 2 3))] [out (make-channel 8)])