csp: transducer-backed channels (Clojure (chan n xform) + ex-handler)
ober
cd5c208044a80190d72129930e5b60b71b6fde50
--- a/docs/clojure-remaining.md +++ b/docs/clojure-remaining.md @@ -1673,7 +1673,7 @@ in this doc. **[deferred]** items are non-goals. | `pub`/`sub`/`unsub` | [current] | — | | `pipeline`/`pipeline-async` | [current] | — | | `promise-chan` | [current] | — | -| `(chan n xform)` | [gap] | §3.1 | +| `(chan n xform)` | [current] `(std csp clj)` | §3.1 landed | | `mix`/`admix`/`toggle` | [gap] | §3.2 | | Timer wheel | [gap] | §3.3 | | `put!`/`take!` with callbacks | [gap] | §3.4 | @@ -1681,7 +1681,7 @@ in this doc. **[deferred]** items are non-goals. | `split` n-way | [gap] | §3.6 | | Mult slow-sub policies | [gap] | §3.7 | | Parked `go` (CPS) | [deferred] | §3.8 | -| Transducer error handler on chan | [gap] | §3.1 | +| Transducer error handler on chan | [current] `(std csp clj)` | §3.1 landed | ### Persistent data structures and Clojure idioms --- a/lib/std/csp.sls +++ b/lib/std/csp.sls @@ -21,6 +21,20 @@ ;;; - sliding : `make-channel/sliding` — drops the oldest queued value ;;; - dropping: `make-channel/dropping` — drops the incoming value ;;; +;;; NOTE ON TRANSDUCERS: a channel can be "backed" by a transducer via +;;; the `channel-xform-fn-set!` hook. When set, `chan-put!` and +;;; `chan-try-put!` run the step procedure instead of a raw enqueue +;;; (the step procedure may itself call `%chan-enqueue-raw!` zero or +;;; more times to expand one input into many outputs). If the step +;;; returns the literal symbol `'stop`, the channel is closed +;;; immediately — this is how transducers like `(taking n)` signal +;;; early termination. On `chan-close!`, the done-fn is invoked first +;;; so stateful transducers (e.g. partitioning-by) can flush buffered +;;; state. The xform-fn is OPAQUE to (std csp) — it neither imports +;;; (std transducer) nor understands `reduced?`. The clj constructor +;;; in (std csp clj) wraps the user's transducer into a closure that +;;; translates `reduced` into `'stop`. +;;; ;;; API: ;;; (make-channel) — unbuffered channel ;;; (make-channel n) — buffered channel with capacity n @@ -55,7 +69,16 @@ chan-put! chan-get! chan-try-put! chan-try-get chan-close! chan-closed? go go-named yield csp-run - chan->list chan-pipe chan-map chan-filter) + chan->list chan-pipe chan-map chan-filter + ;; Transducer hook — used by (std csp clj) to back a channel + ;; with a transducer. `%chan-enqueue-raw!` is a low-level + ;; helper that enqueues one value unconditionally (caller must + ;; hold the channel's mutex). The xform-fn / xform-done-fn + ;; setters attach opaque procedures that (std csp) invokes + ;; during put / close — see the header comments. + %chan-enqueue-raw! + channel-xform-fn channel-xform-fn-set! + channel-xform-done-fn channel-xform-done-fn-set!) (import (chezscheme)) @@ -76,16 +99,21 @@ (mutable closed?) (immutable mutex) (immutable not-empty) ;; signaled on put or close - (immutable not-full)) ;; signaled on get or close + (immutable not-full) ;; signaled on get or close + (mutable xform-fn) ;; opaque (lambda (ch val) → 'stop | #f), or #f + (mutable xform-done-fn)) ;; opaque (lambda (ch) → _), or #f (protocol (lambda (new) (case-lambda [() (new '() '() 0 0 'fixed #f - (make-mutex) (make-condition) (make-condition))] + (make-mutex) (make-condition) (make-condition) + #f #f)] [(cap) (new '() '() 0 cap 'fixed #f - (make-mutex) (make-condition) (make-condition))] + (make-mutex) (make-condition) (make-condition) + #f #f)] [(cap kind) (new '() '() 0 cap kind #f - (make-mutex) (make-condition) (make-condition))])))) + (make-mutex) (make-condition) (make-condition) + #f #f)])))) (define (make-channel/buf n) (make-channel n)) @@ -144,6 +172,38 @@ ;; unbuffered case identically to fixed — the writer blocks until ;; a reader shows up. + ;; Low-level raw enqueue: add one value to ch's queue and notify + ;; any waiting reader. The caller MUST hold the channel's mutex. + ;; Returns ch so it can serve as a reducing-function accumulator. + ;; This is the only "back door" the transducer bridge needs from + ;; (std csp) — it lets the buffer-rf in (std csp clj) push items + ;; past a raw put-check. + (define (%chan-enqueue-raw! ch val) + (q-enqueue! ch val) + (condition-broadcast (channel-not-empty ch)) + ch) + + ;; Put one value into a channel with the mutex already held, running + ;; the channel's transducer step if one is attached. If the xform + ;; returns the literal symbol 'stop (the opaque signal for reduced), + ;; the channel is closed in place (without re-acquiring the mutex). + ;; Without a transducer this collapses to the raw enqueue. + (define (%chan-put-into! ch val) + (let ([xf (channel-xform-fn ch)]) + (cond + [xf + (let ([result (xf ch val)]) + (when (eq? result 'stop) + ;; Transducer signaled early termination. Mark closed + ;; directly — we already hold the mutex, so can't reuse + ;; chan-close! which would re-enter with-mutex. + (channel-closed?-set! ch #t) + (condition-broadcast (channel-not-empty ch)) + (condition-broadcast (channel-not-full ch))))] + [else + (q-enqueue! ch val) + (condition-broadcast (channel-not-empty ch))]))) + (define (chan-put! ch val) (with-mutex (channel-mutex ch) (let loop () @@ -151,15 +211,14 @@ [(channel-closed? ch) (error 'chan-put! "channel is closed")] [(not (q-full? ch)) - (q-enqueue! ch val) - (condition-broadcast (channel-not-empty ch))] + (%chan-put-into! ch val)] [(eq? 'sliding (channel-kind ch)) - ;; Drop the oldest value to make room, then enqueue. + ;; Drop the oldest value to make room, then put. (q-dequeue! ch) - (q-enqueue! ch val) - (condition-broadcast (channel-not-empty ch))] + (%chan-put-into! ch val)] [(eq? 'dropping (channel-kind ch)) - ;; Silently drop the incoming value. + ;; Silently drop the incoming value — do NOT run the xform, + ;; matching Clojure's behavior for a dropping buffer. (void)] [else ;; 'fixed (and the unbuffered degenerate case) (condition-wait (channel-not-full ch) (channel-mutex ch)) @@ -170,13 +229,11 @@ (cond [(channel-closed? ch) #f] [(not (q-full? ch)) - (q-enqueue! ch val) - (condition-broadcast (channel-not-empty ch)) + (%chan-put-into! ch val) #t] [(eq? 'sliding (channel-kind ch)) (q-dequeue! ch) - (q-enqueue! ch val) - (condition-broadcast (channel-not-empty ch)) + (%chan-put-into! ch val) #t] [(eq? 'dropping (channel-kind ch)) ;; Drop silently but report success (no block). @@ -212,6 +269,14 @@ (define (chan-close! ch) (with-mutex (channel-mutex ch) + ;; If a transducer is attached, flush its stateful tail first + ;; (e.g. partitioning-by's in-flight partition). The done-fn + ;; may itself call %chan-enqueue-raw! through its buffer-rf. + ;; We only flush once per channel — the `closed?` guard prevents + ;; a double-flush if chan-close! races with itself. + (let ([done (channel-xform-done-fn ch)]) + (when (and done (not (channel-closed? ch))) + (done ch))) (channel-closed?-set! ch #t) (condition-broadcast (channel-not-empty ch)) (condition-broadcast (channel-not-full ch)))) --- a/lib/std/csp/clj.sls +++ b/lib/std/csp/clj.sls @@ -61,7 +61,8 @@ (import (except (chezscheme) merge) (except (std csp) go go-named) (std csp select) - (std csp ops)) + (std csp ops) + (std transducer)) ;; ====================================================== ;; Buffer specs — opaque tags for chan to dispatch on @@ -77,25 +78,79 @@ ;; chan — Clojure-style constructor ;; ====================================================== ;; - ;; Clojure supports (chan), (chan n), (chan buf), (chan n xform), - ;; (chan buf xform). Transducers aren't implemented yet — passing a - ;; third arg raises an error. + ;; Clojure supports all of + ;; (chan) + ;; (chan n) (chan buf) + ;; (chan n xform) (chan buf xform) + ;; (chan n xform eh) (chan buf xform eh) + ;; + ;; With a transducer, every put runs through `xform` before landing + ;; in the buffer. If the transducer signals early termination + ;; (via `reduced`), the channel closes immediately. If a step raises + ;; an exception the optional `ex-handler` is called with the + ;; condition; its return value is enqueued verbatim (a #f return + ;; drops the value silently, matching Clojure semantics). + + ;; Build a bare channel from a buffer-size or buffer-spec argument. + (define (%make-bare-chan x) + (cond + [(buffer-spec? x) + (case (buffer-spec-kind x) + [(sliding) (make-channel/sliding (buffer-spec-size x))] + [(dropping) (make-channel/dropping (buffer-spec-size x))] + [else (make-channel (buffer-spec-size x))])] + [(and (integer? x) (zero? x)) (make-channel)] + [(integer? x) (make-channel x)] + [else (error 'chan "expected buffer spec or non-negative integer" x)])) + + ;; Attach a transducer to an already-constructed channel. Builds a + ;; buffer-rf that writes directly into the channel's queue, wraps it + ;; in the user's xform, and installs two closures (step / done) on + ;; the channel record. `reduced` is translated into the symbol 'stop + ;; so (std csp) does not need to know about the reduced type. + (define (%attach-xform! ch xform ex-handler) + (let* ([buffer-rf + (case-lambda + [() ch] + [(acc) acc] + [(acc val) (%chan-enqueue-raw! acc val) acc])] + [rf (apply-xf xform buffer-rf)]) + (channel-xform-fn-set! + ch + (lambda (c val) + (call/cc + (lambda (k) + (with-exception-handler + (lambda (exn) + (cond + [ex-handler + (let ([v (ex-handler exn)]) + (when v (%chan-enqueue-raw! c v)) + (k #f))] + [else (raise exn)])) + (lambda () + (let ([result (rf c val)]) + (if (reduced? result) 'stop #f)))))))) + (channel-xform-done-fn-set! + ch + (lambda (c) + (call/cc + (lambda (k) + (with-exception-handler + (lambda (exn) + (when ex-handler (ex-handler exn)) + (k #f)) + (lambda () (rf c) #f)))))) + ch)) (define chan (case-lambda [() (make-channel)] - [(x) - (cond - [(buffer-spec? x) - (case (buffer-spec-kind x) - [(sliding) (make-channel/sliding (buffer-spec-size x))] - [(dropping) (make-channel/dropping (buffer-spec-size x))] - [else (make-channel (buffer-spec-size x))])] - [(and (integer? x) (zero? x)) (make-channel)] - [(integer? x) (make-channel x)] - [else (error 'chan "expected buffer spec or non-negative integer" x)])] - [(_n _xform) - (error 'chan "transducers are not supported yet")])) + [(x) (%make-bare-chan x)] + [(x xform) + (%attach-xform! (%make-bare-chan x) xform #f)] + [(x xform ex-handler) + (%attach-xform! (%make-bare-chan x) xform ex-handler)])) ;; ====================================================== ;; Put / take / close / poll / offer --- a/lib/std/transducer.sls +++ b/lib/std/transducer.sls @@ -78,6 +78,7 @@ ;; Reduction transduce + apply-xf ;; Reducing functions rf-cons --- a/tests/test-csp.ss +++ b/tests/test-csp.ss @@ -13,7 +13,8 @@ (except (std csp) go go-named) (std csp select) (std csp ops) - (std csp clj)) + (std csp clj) + (std transducer)) (define pass 0) (define fail 0) @@ -247,6 +248,84 @@ (chan->list (merge (list (to-chan '(1 2)) (to-chan '(3 4))) 4))) '(1 2 3 4)) +;;; ======== Transducer-backed channels (Phase A.2) ======== + +(test "chan/mapping — increments every put" + (let ([ch (chan 8 (mapping (lambda (x) (+ x 1))))]) + (chan-put! ch 1) (chan-put! ch 2) (chan-put! ch 3) + (chan-close! ch) + (chan->list ch)) + '(2 3 4)) + +(test "chan/filtering — keeps only evens" + (let ([ch (chan 8 (filtering even?))]) + (for-each (lambda (x) (chan-put! ch x)) '(1 2 3 4 5 6)) + (chan-close! ch) + (chan->list ch)) + '(2 4 6)) + +(test "chan/taking — early-stop closes channel" + (let ([ch (chan 8 (taking 3))]) + ;; Put 5 values; only first 3 survive, and channel is auto-closed + ;; after the 3rd by the reduced? signal. + (chan-put! ch 'a) (chan-put! ch 'b) (chan-put! ch 'c) + ;; Subsequent try-puts should fail because ch is closed. + (let ([drained (chan->list ch)] + [after (chan-try-put! ch 'd)]) + (list drained after))) + '((a b c) #f)) + +(test "chan/flat-mapping — one input produces many outputs" + (let ([ch (chan 16 (flat-mapping (lambda (x) (list x (* x 10)))))]) + (chan-put! ch 1) (chan-put! ch 2) + (chan-close! ch) + (chan->list ch)) + '(1 10 2 20)) + +(test "chan/composed — filter then map then take" + (let ([ch (chan 32 + (compose-transducers + (filtering odd?) + (mapping (lambda (x) (* x x))) + (taking 3)))]) + ;; Use chan-try-put! so a closed channel (after taking stops) simply + ;; returns #f instead of raising. Then drain to see the survivors. + (for-each (lambda (x) (chan-try-put! ch x)) '(1 2 3 4 5 6 7 8 9 10)) + (chan->list ch)) + '(1 9 25)) + +(test "chan/partitioning-by — close flushes pending partition" + (let ([ch (chan 16 (partitioning-by even?))]) + (for-each (lambda (x) (chan-put! ch x)) '(1 3 2 4 5 7)) + (chan-close! ch) + (chan->list ch)) + '((1 3) (2 4) (5 7))) + +(test "chan/ex-handler — exception routed to handler, value dropped" + (let ([ch (chan 8 + (mapping (lambda (x) + (if (= x 0) (error 'bad "zero") x))) + ;; Handler returns #f ⇒ swallow the error; nothing put. + (lambda (exn) #f))]) + (chan-put! ch 1) + (chan-put! ch 0) ;; raises inside xform — ex-handler returns #f + (chan-put! ch 2) + (chan-close! ch) + (chan->list ch)) + '(1 2)) + +(test "chan/ex-handler — handler returns substitute value" + (let ([ch (chan 8 + (mapping (lambda (x) + (if (= x 0) (error 'bad "zero") x))) + (lambda (exn) 'oops))]) + (chan-put! ch 1) + (chan-put! ch 0) ;; raises → handler returns 'oops → enqueued raw + (chan-put! ch 2) + (chan-close! ch) + (chan->list ch)) + '(1 oops 2)) + ;;; Summary (printf "~%CSP: ~a passed, ~a failed~%" pass fail)