csp: async-reduce + onto-chan! / onto-chan!!
ober
2275d26e9bd472eb8263628a018ec888eb096a3f
--- a/docs/clojure-remaining.md +++ b/docs/clojure-remaining.md @@ -499,6 +499,16 @@ Re-export in `(std csp clj)`. ### 3.5 `async/reduce` and `onto-chan!`/`onto-chan!!` +**Status.** [landed] `(std csp ops)` now exports `chan-reduce-async`, +`onto-chan!`, and `onto-chan!!`. `(std csp clj)` re-exports them under +the Clojure names `async-reduce`, `onto-chan!`, `onto-chan!!`. +Matching Clojure's actual implementation, `async-reduce` returns a +plain size-1 channel (NOT a promise-channel) — the first taker gets +the folded value, the channel then closes, and subsequent takers see +`(eof-object)`. Callers who need caching semantics should wrap the +result channel in a mult or promise-chan. Covered by eight tests in +`tests/test-csp.ss`. + **What Clojure does.** `clojure.core.async/reduce` is a go-block reduce that reads from a channel until it closes and returns a promise-chan with the final result: @@ -1683,7 +1693,7 @@ in this doc. **[deferred]** items are non-goals. | `alts!`/`alt!` with priority/default | [current] | — | | `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] | §3.5 async variant | +| `to-chan`/`onto-chan`/`chan-reduce` | [current] | — | | `merge`/`split`/`pipe` | [current] | §3.6 n-way split | | `mult`/`tap`/`untap` | [current] | §3.7 slow-sub policy | | `pub`/`sub`/`unsub` | [current] | — | @@ -1693,7 +1703,7 @@ in this doc. **[deferred]** items are non-goals. | `mix`/`admix`/`toggle` | [gap] | §3.2 | | Timer wheel | [gap] | §3.3 | | `put!`/`take!` with callbacks | [current] `(std csp ops)` | §3.4 landed | -| `async/reduce`, `onto-chan!` | [gap] | §3.5 | +| `async/reduce`, `onto-chan!` | [current] `(std csp ops)` | §3.5 landed | | `split` n-way | [gap] | §3.6 | | Mult slow-sub policies | [gap] | §3.7 | | Parked `go` (CPS) | [deferred] | §3.8 | --- a/lib/std/csp/clj.sls +++ b/lib/std/csp/clj.sls @@ -48,7 +48,8 @@ ;; Go / thread go go-loop clj-thread ;; Collection bridges - to-chan onto-chan + to-chan onto-chan onto-chan! onto-chan!! + async-reduce ;; Composition merge split pipe ;; Broadcast @@ -212,6 +213,10 @@ (define split chan-split) (define pipe chan-pipe-to) + ;; Clojure core.async's `async/reduce` — an async fold over a + ;; channel that returns a promise-channel holding the final value. + (define async-reduce chan-reduce-async) + (define mult make-mult) (define tap tap!) (define untap untap!) --- a/lib/std/csp/ops.sls +++ b/lib/std/csp/ops.sls @@ -10,8 +10,8 @@ (library (std csp ops) (export ;; Collection bridges - to-chan onto-chan - chan-reduce chan-into + to-chan onto-chan onto-chan! onto-chan!! + chan-reduce chan-reduce-async chan-into ;; Non-blocking callbacks put! take! ;; Composition @@ -61,6 +61,29 @@ (when close? (chan-close! ch)))) ch])) + ;; Clojure core.async aliases: + ;; + ;; (onto-chan! ch lst) — non-blocking: returns ch and + ;; feeds items from a helper thread. + ;; Same behavior as onto-chan above. + ;; (onto-chan!! ch lst) — blocking: runs on the caller's + ;; thread and only returns once every + ;; item has been delivered (and the + ;; channel closed, if close? is #t). + ;; + ;; Use the blocking form when you want back-pressure to reach the + ;; caller (e.g. tests, pipelines where the next stage can't start + ;; until the feed has completed). + (define onto-chan! onto-chan) + + (define onto-chan!! + (case-lambda + [(ch lst) (onto-chan!! ch lst #t)] + [(ch lst close?) + (for-each (lambda (x) (chan-put! ch x)) lst) + (when close? (chan-close! ch)) + ch])) + ;; (chan-reduce f init ch) → fold ch into a single value. ;; ;; Runs on the caller's thread — this is the blocking variant. @@ -72,6 +95,30 @@ acc (loop (f acc v)))))) + ;; (chan-reduce-async f init ch) → channel with the final acc. + ;; + ;; Clojure core.async's `async/reduce`: spawns a helper thread that + ;; folds `ch` with `f` starting from `init`, and delivers the + ;; single final value onto a fresh size-1 channel which is then + ;; closed. The caller can `chan-get!` / `<!!` to await the result. + ;; Any number of takers can read the value back — once the put + ;; lands the channel sits with one value until drained, and after + ;; close a taker on the drained channel receives `(eof-object)`. + ;; + ;; If `f` raises, the helper thread swallows the exception and + ;; closes the result channel so a pending taker observes eof + ;; rather than hanging. This matches the documented semantics in + ;; core-async.md §3.5 and mirrors how `go` treats body exceptions. + (define (chan-reduce-async f init ch) + (let ([out (make-channel 1)]) + (fork-thread + (lambda () + (guard (_ [else (chan-close! out)]) + (let ([acc (chan-reduce f init ch)]) + (chan-put! out acc) + (chan-close! out))))) + out)) + ;; (chan-into container ch) → collect into a list or vector. ;; ;; The only containers supported for now are `'()` (list) and --- a/tests/test-csp.ss +++ b/tests/test-csp.ss @@ -395,6 +395,66 @@ (list a b)))) '(after-put-returned second-delivered)) +;;; ======== async-reduce + onto-chan! / onto-chan!! (Phase C.2) ======== + +(test "async-reduce sum" + (let* ([input (to-chan '(1 2 3 4 5))] + [p (async-reduce + 0 input)]) + (<!! p)) + 15) + +(test "async-reduce conj into list" + (let* ([input (to-chan '(a b c d))] + [p (async-reduce (lambda (acc v) (cons v acc)) '() input)]) + (<!! p)) + '(d c b a)) + +(test "async-reduce on empty channel returns init" + (let* ([ch (make-channel)] + [p (async-reduce + 100 ch)]) + (chan-close! ch) + (<!! p)) + 100) + +(test "async-reduce second get returns eof (channel drains then closes)" + ;; Matches Clojure's async/reduce which uses (chan 1) internally — + ;; the first taker gets the value, the channel then closes, and + ;; subsequent takers see (eof-object). Users who need caching + ;; semantics should wrap the result channel in a mult or promise. + (let* ([input (to-chan '(10 20 30))] + [p (async-reduce + 0 input)]) + (let ([a (<!! p)] [b (<!! p)]) + (list a (eof-object? b)))) + '(60 #t)) + +(test "onto-chan! (async) feeds and closes" + (let ([ch (make-channel 16)]) + (onto-chan! ch '(1 2 3)) + (chan->list ch)) + '(1 2 3)) + +(test "onto-chan! with close?=#f leaves channel open" + (let ([ch (make-channel 16)]) + (onto-chan! ch '(a b c) #f) + (sleep (millis 20)) ;; let feeder finish + (list (chan-get! ch) (chan-get! ch) (chan-get! ch) + (chan-closed? ch))) + '(a b c #f)) + +(test "onto-chan!! (blocking) returns only after everything landed" + ;; With a big enough buffer the blocking variant returns + ;; synchronously; we then drain. + (let ([ch (make-channel 16)]) + (onto-chan!! ch '(x y z)) + (chan->list ch)) + '(x y z)) + +(test "onto-chan!! followed by async-reduce is deterministic" + (let ([ch (make-channel 64)]) + (onto-chan!! ch '(1 2 3 4 5 6 7 8 9 10)) + (<!! (async-reduce + 0 ch))) + 55) + ;;; Summary (printf "~%CSP: ~a passed, ~a failed~%" pass fail)