diff --git a/docs/libraries.md b/docs/libraries.md index 12f4ef9..7d42ab6 100644 --- a/docs/libraries.md +++ b/docs/libraries.md @@ -48,6 +48,10 @@ e.g. the [ring-app example](https://github.com/jolt-lang/examples/tree/main/ring [data.priority-map](https://github.com/clojure/data.priority-map). * [core.memoize](https://github.com/clojure/core.memoize) — function memoization over [core.cache](https://github.com/clojure/core.cache). +* [core.async](https://github.com/clojure/core.async) — CSP channels and `go` blocks + (`!`/`alts!`, `pipeline`, `mult`/`mix`/`pub`/`sub`) on real OS threads. +* [core.logic](https://github.com/clojure/core.logic) — relational logic programming + (unification, `run`/`fresh`/`conde`, finite domains). * [tick](https://github.com/juxt/tick) — date/time over Jolt's `java.time`; `#time/…` literals via `time-literals`. * [transit-jolt](https://github.com/jolt-lang/transit-jolt) — Transit (JSON) read/write diff --git a/host/chez/java/async.ss b/host/chez/java/async.ss index ee200a0..98a4a11 100644 --- a/host/chez/java/async.ss +++ b/host/chez/java/async.ss @@ -1,17 +1,20 @@ -;; async.ss — clojure.core.async on real OS threads for the Chez host. +;; async.ss — clojure.core.async channel primitives on real OS threads. ;; -;; A `go` block is an OS thread and a channel is a mutex+condition blocking -;; queue: ! are the blocking !! (they "park" by blocking the thread). -;; ! work ANYWHERE — no CPS transform — because they are ordinary blocking -;; calls. Real parallelism, shared heap. Trade-off: one OS thread per go block -;; (fine for typical use, not for thousands of simultaneous go blocks). +;; A `go` block is an OS thread and a channel is a Chez mutex+condition blocking +;; queue: ! are the blocking !! (they "park" by blocking the thread), +;; and work ANYWHERE — no CPS transform, no go-only restriction. Real parallelism, +;; shared heap. This is a superset of the JVM model: it has no fixed go-block +;; thread pool, no MAX-QUEUE-SIZE on pending ops, and parking ops are legal outside +;; a go block. One OS thread per go block (fine for typical use). ;; ;; Channel: an unbuffered channel is a rendezvous (the putter blocks until its ;; value is taken); a buffered (chan n) put blocks only when full; dropping/sliding -;; buffers never block the putter. A transducer is applied on the put side. +;; buffers never block the putter. A transducer is applied on the put side; an +;; optional ex-handler catches a throw from the transducer step. ;; -;; The fns are def-var!'d into clojure.core.async; go/go-loop/thread are macros -;; (mark-macro!) expanding to go-spawn. Loaded after +;; This file provides the primitives; the higher-level dataflow API (mult, mix, +;; pub/sub, pipeline, map, merge, reduce, …) is a Clojure overlay over them. +;; go/go-loop/thread are macros (mark-macro!) expanding to go-spawn. Loaded after ;; concurrency.ss (reuses ms->duration). Requires a threaded Chez build. ;; --- buffers ---------------------------------------------------------------- @@ -19,6 +22,8 @@ (define (jolt-async-buffer n) (make-async-buffer n 'fixed)) (define (jolt-async-dropping-buffer n) (make-async-buffer n 'dropping)) (define (jolt-async-sliding-buffer n) (make-async-buffer n 'sliding)) +(define (jolt-async-unblocking-buffer? b) + (if (and (async-buffer? b) (memq (async-buffer-kind b) '(dropping sliding promise))) #t #f)) ;; --- channels --------------------------------------------------------------- ;; items: an amortized-O(1) FIFO held as a mutable #(out in len) — `out` is the @@ -27,9 +32,12 @@ ;; Each entry is (value . box); box is #f for a buffered value or a 1-slot vector ;; for an unbuffered rendezvous put (set #t when taken, waking the putter). ;; cap 0 + kind 'unbuffered = rendezvous; cap>0 with kind fixed/dropping/sliding. +;; takew counts threads parked in a blocking take (so a non-blocking offer! to an +;; unbuffered channel can tell a taker is waiting). xrf is the transducer reducing +;; fn (or #f); exh the ex-handler (or #f). (define-record-type async-chan - (fields mu cv (mutable items) cap kind (mutable closed?) (mutable xrf)) - (nongenerative async-chan-v1)) + (fields mu cv (mutable items) cap kind (mutable closed?) (mutable xrf) (mutable takew) exh) + (nongenerative async-chan-v2)) (define (ac-qnew) (vector '() '() 0)) (define (ac-qlen ch) (vector-ref (async-chan-items ch) 2)) @@ -73,17 +81,30 @@ ((null? (cdr args)) (car args)) ; completion (else (ac-buf-give! ch (cadr args)) (car args))))) ; step -(define (ac-make cap kind xrf) (make-async-chan (make-mutex) (make-condition) (ac-qnew) cap kind #f xrf)) +;; run the transducer step (or completion) guarded by the channel's ex-handler: +;; if the xform throws and exh returns non-nil, that value is added to the buffer. +(define (ac-xrf-apply ch . v) + (let ((xrf (async-chan-xrf ch)) (exh (async-chan-exh ch))) + (guard (e (#t (if exh + (let ((else (jolt-invoke exh e))) + (unless (jolt-nil? else) (ac-buf-give! ch else)) + (async-chan-xrf ch)) ; treat as non-reduced + (raise e)))) + (apply jolt-invoke xrf ch v)))) -;; (chan) | (chan n) | (chan buf) | (chan n|buf xform) +(define (ac-make cap kind xrf) (make-async-chan (make-mutex) (make-condition) (ac-qnew) cap kind #f xrf 0 #f)) +(define (ac-make/exh cap kind exh) (make-async-chan (make-mutex) (make-condition) (ac-qnew) cap kind #f #f 0 exh)) + +;; (chan) | (chan n) | (chan buf) | (chan n|buf xform) | (chan n|buf xform exh) (define (jolt-async-chan . args) (let ((buf (if (pair? args) (car args) jolt-nil)) - (xform (if (and (pair? args) (pair? (cdr args))) (cadr args) jolt-nil))) + (xform (if (and (pair? args) (pair? (cdr args))) (cadr args) jolt-nil)) + (exh (if (and (pair? args) (pair? (cdr args)) (pair? (cddr args))) (caddr args) jolt-nil))) (let-values (((cap kind) (cond ((async-buffer? buf) (values (async-buffer-n buf) (async-buffer-kind buf))) ((and (number? buf) (> buf 0)) (values buf 'fixed)) (else (values 0 'unbuffered))))) - (let ((ch (ac-make cap kind #f))) + (let ((ch (ac-make/exh cap kind (if (jolt-nil? exh) #f exh)))) (unless (jolt-nil? xform) (async-chan-xrf-set! ch (jolt-invoke xform (ac-make-add-rf ch)))) ch)))) @@ -93,7 +114,7 @@ (define (ac-close! ch) (unless (async-chan-closed? ch) (async-chan-closed?-set! ch #t) - (when (async-chan-xrf ch) (guard (e (#t #f)) (jolt-invoke (async-chan-xrf ch) ch))) + (when (async-chan-xrf ch) (guard (e (#t #f)) (ac-xrf-apply ch))) (condition-broadcast (async-chan-cv ch))) jolt-nil) (define (jolt-async-close! ch) (with-mutex (async-chan-mu ch) (ac-close! ch))) @@ -102,12 +123,12 @@ ;; transducer the value is run through it (one put -> zero or more channel values); ;; a `reduced` result closes the channel. (define (jolt-async-give ch v) - (when (jolt-nil? v) (jolt-throw (jolt-ex-info "Can't put nil on a channel" (jolt-hash-map)))) + (when (jolt-nil? v) (jolt-throw (jolt-host-throwable "java.lang.IllegalArgumentException" "Can't put nil on a channel"))) (with-mutex (async-chan-mu ch) (cond ((async-chan-closed? ch) #f) ((async-chan-xrf ch) - (let ((r (jolt-invoke (async-chan-xrf ch) ch v))) + (let ((r (ac-xrf-apply ch v))) (when (jolt-reduced? r) (ac-close! ch)) #t)) (else @@ -154,12 +175,19 @@ (cond ((eq? (async-chan-kind ch) 'promise) (cond ((not (ac-qempty? ch)) (ac-peek ch)) ((async-chan-closed? ch) jolt-nil) - (else (condition-wait (async-chan-cv ch) (async-chan-mu ch)) (loop)))) + (else (ac-take-wait ch) (loop)))) ((not (ac-qempty? ch)) (ac-take-head! ch)) ((async-chan-closed? ch) jolt-nil) - (else (condition-wait (async-chan-cv ch) (async-chan-mu ch)) (loop)))))) + (else (ac-take-wait ch) (loop)))))) -;; non-blocking take for alts!: a value, jolt-nil (closed+empty), or ac-poll-empty. +;; park in a take, tracking the waiter count so a concurrent offer! to an +;; unbuffered channel can see that a taker is ready. +(define (ac-take-wait ch) + (async-chan-takew-set! ch (fx+ 1 (async-chan-takew ch))) + (condition-wait (async-chan-cv ch) (async-chan-mu ch)) + (async-chan-takew-set! ch (fx- (async-chan-takew ch) 1))) + +;; non-blocking take for alts!/poll!: a value, jolt-nil (closed+empty), or ac-poll-empty. (define ac-poll-empty (list 'empty)) (define (ac-poll! ch) (with-mutex (async-chan-mu ch) @@ -168,28 +196,40 @@ ((async-chan-closed? ch) jolt-nil) (else ac-poll-empty)))) -;; (alts! [ch ...]) — take from whichever channel is ready first; returns -;; [value channel] (value nil if that channel closed). Take-only: every port must -;; be a channel — put specs [ch val] and the :default option are not supported, so -;; reject them with a clear error instead of crashing inside ac-poll!. -;; Polls with a 1ms backoff — no cross-channel wait-set yet. -(define ac-1ms (make-time 'time-duration 1000000 0)) -(define (jolt-async-alts chans) - (let ((cs (seq->list (jolt-seq chans)))) - (for-each (lambda (c) - (unless (async-chan? c) - (jolt-throw (jolt-ex-info - "alts! supports channel ports only (put specs [ch val] and :default are not supported)" - (jolt-hash-map))))) - cs) - (let loop () - (let try ((rest cs)) - (if (null? rest) - (begin (sleep ac-1ms) (loop)) - (let ((r (ac-poll! (car rest)))) - (if (eq? r ac-poll-empty) - (try (cdr rest)) - (jolt-vector r (car rest))))))))) +;; non-blocking give: 'ok (accepted), 'full (would block), or 'closed. +(define (ac-try-give! ch v) + (when (jolt-nil? v) (jolt-throw (jolt-host-throwable "java.lang.IllegalArgumentException" "Can't put nil on a channel"))) + (with-mutex (async-chan-mu ch) + (cond + ((async-chan-closed? ch) 'closed) + ((async-chan-xrf ch) (let ((r (ac-xrf-apply ch v))) + (when (jolt-reduced? r) (ac-close! ch)) 'ok)) + (else + (case (async-chan-kind ch) + ((dropping sliding) (ac-buf-give! ch v) 'ok) + ((promise) (when (ac-qempty? ch) (ac-qpush! ch (cons v #f)) + (condition-broadcast (async-chan-cv ch))) 'ok) + (else + (cond + ((> (async-chan-cap ch) 0) + (if (< (ac-qlen ch) (async-chan-cap ch)) + (begin (ac-qpush! ch (cons v #f)) (condition-broadcast (async-chan-cv ch)) 'ok) + 'full)) + ;; unbuffered: only immediate if a taker is parked to receive it. + ((> (async-chan-takew ch) 0) + (let ((box (vector #f))) + (ac-qpush! ch (cons v box)) + (condition-broadcast (async-chan-cv ch)) + 'ok)) + (else 'full)))))))) + +;; offer! / poll! — never block. offer! returns #t/#f(closed) on completion, nil if +;; it would block; poll! returns a value, nil (closed+empty), or the ::none sentinel. +(define cca-none (keyword "clojure.core.async" "none")) +(define (jolt-async-offer! ch v) + (case (ac-try-give! ch v) ((ok) #t) ((closed) #f) (else jolt-nil))) +(define (jolt-async-poll! ch) + (let ((r (ac-poll! ch))) (if (eq? r ac-poll-empty) cca-none r))) ;; (timeout ms) — a channel that closes after ms milliseconds. (define (jolt-async-timeout ms) @@ -197,17 +237,28 @@ (fork-thread (lambda () (sleep (ms->duration ms)) (jolt-async-close! w))) w)) -;; (put! ch v [cb]) / (take! ch cb) — async put/take on a thread, optional callback. -(define (jolt-async-put! ch v . cb) - (fork-thread (lambda () - (let ((ok (jolt-async-give ch v))) - (when (and (pair? cb) (not (jolt-nil? (car cb)))) (jolt-invoke (car cb) ok))))) - jolt-nil) -(define (jolt-async-take! ch cb) - (fork-thread (lambda () - (let ((v (jolt-async-take ch))) - (unless (jolt-nil? cb) (jolt-invoke cb v))))) - jolt-nil) +;; (put! ch v [cb [on-caller?]]) — async put, optional completion callback. If the +;; put completes immediately and on-caller? (default #t), the callback runs on the +;; calling thread; otherwise on another thread. Returns true unless already closed. +(define (jolt-async-put! ch v . rest) + (let* ((cb (if (pair? rest) (car rest) jolt-nil)) + (on-caller? (if (and (pair? rest) (pair? (cdr rest))) (jolt-truthy? (cadr rest)) #t)) + (call-cb (lambda (ok) (unless (jolt-nil? cb) (jolt-invoke cb ok))))) + (case (ac-try-give! ch v) + ((ok) (if on-caller? (call-cb #t) (fork-thread (lambda () (call-cb #t)))) #t) + ((closed) (if on-caller? (call-cb #f) (fork-thread (lambda () (call-cb #f)))) #f) + (else (fork-thread (lambda () (call-cb (jolt-async-give ch v)))) #t)))) + +;; (take! ch cb [on-caller?]) — async take. Same on-caller? rule as put!. +(define (jolt-async-take! ch cb . rest) + (let* ((on-caller? (if (pair? rest) (jolt-truthy? (car rest)) #t)) + (call-cb (lambda (v) (unless (jolt-nil? cb) (jolt-invoke cb v)))) + (r (ac-poll! ch))) + (cond + ((eq? r ac-poll-empty) (fork-thread (lambda () (call-cb (jolt-async-take ch))))) + (on-caller? (call-cb r)) + (else (fork-thread (lambda () (call-cb r))))) + jolt-nil)) ;; (go-spawn thunk) — run thunk on a thread; return a buffered(1) channel that ;; conveys its value once then closes (a nil result just closes). Dynamic bindings @@ -246,14 +297,19 @@ (cca-def! "buffer" jolt-async-buffer) (cca-def! "dropping-buffer" jolt-async-dropping-buffer) (cca-def! "sliding-buffer" jolt-async-sliding-buffer) +(cca-def! "__promise-buffer" (lambda () (make-async-buffer 1 'promise))) +(cca-def! "unblocking-buffer?" jolt-async-unblocking-buffer?) (cca-def! "close!" jolt-async-close!) (cca-def! "!" jolt-async-give) (cca-def! ">!!" jolt-async-give) -(cca-def! "alts!" jolt-async-alts) (cca-def! "alts!!" jolt-async-alts) (cca-def! "timeout" jolt-async-timeout) (cca-def! "put!" jolt-async-put!) (cca-def! "take!" jolt-async-take!) +(cca-def! "offer!" jolt-async-offer!) (cca-def! "go-spawn" async-go-spawn) +;; non-blocking primitives the Clojure overlay's do-alts polls over. +(cca-def! "__poll!" jolt-async-poll!) +(cca-def! "__offer!" jolt-async-offer!) (cca-def! "go" cca-go-macro) (mark-macro! "clojure.core.async" "go") (cca-def! "go-loop" cca-go-loop-macro) (mark-macro! "clojure.core.async" "go-loop") (cca-def! "thread" cca-thread-macro) (mark-macro! "clojure.core.async" "thread") diff --git a/host/chez/loader.ss b/host/chez/loader.ss index b456642..87d9db7 100644 --- a/host/chez/loader.ss +++ b/host/chez/loader.ss @@ -141,6 +141,14 @@ (vector-for-each (lambda (c) (hashtable-set! loaded-ns (var-cell-ns c) #t)) (hashtable-values var-table)) +;; clojure.core.async ships native channel primitives (async.ss) AND a Clojure +;; overlay (stdlib/clojure/core/async.clj) with the higher-level dataflow API +;; (alts!, pipe, mult, mix, pub/sub, map, merge, …). The primitives pre-seed the +;; namespace above, which would make a `require` no-op and skip the overlay. Drop +;; it from the loaded set so a require pulls the overlay from the source roots +;; (like clojure.test); the primitives stay defined either way. +(hashtable-delete! loaded-ns "clojure.core.async") + ;; Does `name` already have vars in the var-table? A namespace baked into the ;; image after the snapshot above — an AOT'd app namespace in a `jolt build` ;; binary — exists in memory with no source file; a later `require` of it must diff --git a/host/chez/ns.ss b/host/chez/ns.ss index cad5286..f65315f 100644 --- a/host/chez/ns.ss +++ b/host/chez/ns.ss @@ -74,7 +74,8 @@ ;; :refer :all — bring in every public var (require :refer :all) ((and (keyword? v) (string=? (keyword-t-name v) "all")) (chez-register-refer-all! cns target)) - ((pvec? v) + ;; :refer [a b] or :refer (a b) — both forms list names to bring in. + ((or (pvec? v) (cseq? v) (empty-list-t? v)) (for-each (lambda (n) (when (symbol-t? n) (chez-register-refer! cns (symbol-t-name n) target))) (seq->list v)))))))) diff --git a/host/chez/seq.ss b/host/chez/seq.ss index 98401ad..9f70e75 100644 --- a/host/chez/seq.ss +++ b/host/chez/seq.ss @@ -249,10 +249,13 @@ (jolt-persistent! (reduce-seq (lambda (t x) (jolt-conj! t x)) (jolt-transient-new to) (jolt-seq from))))) (define (range-from n) (cseq-lazy n (lambda () (range-from (+ n 1))))) +;; An empty range is () (jolt-empty-list), NOT nil — (range 0) and (range 5 5) are +;; empty seqs in Clojure, so (= () (range 0)) holds. The same () terminates the +;; lazy tail of a non-empty range (jolt-empty-list seqs back to nil, see jolt-take). (define (range-bounded n end step) (if (if (> step 0.0) (< n end) (> n end)) (cseq-lazy n (lambda () (range-bounded (+ n step) end step))) - jolt-nil)) + jolt-empty-list)) ;; numeric tower: exact 0/1 defaults so (range 3) yields exact ints ;; (= JVM longs); flonum args still produce flonums (Scheme arithmetic preserves). (define jolt-range diff --git a/stdlib/clojure/core/async.clj b/stdlib/clojure/core/async.clj new file mode 100644 index 0000000..5965100 --- /dev/null +++ b/stdlib/clojure/core/async.clj @@ -0,0 +1,667 @@ +;; clojure.core.async — higher-level dataflow API over the channel primitives. +;; +;; The primitives (chan, !, !!, close!, put!, take!, offer!, timeout, +;; promise-chan, buffer/dropping-buffer/sliding-buffer, go/go-loop/thread, go-spawn) +;; are provided natively (host/chez/java/async.ss) on real OS threads. This overlay +;; adds the portable dataflow operators — alts!, pipe, pipeline, split, reduce, +;; transduce, mult, mix, pub/sub, map, merge, and the deprecated map/… — +;; ported from clojure.core.async over those primitives. Because go blocks are real +;; threads, parking ops are ordinary blocking ops and work anywhere; this is a +;; superset of the JVM model (no fixed thread pool, no pending-op limit). + +(ns clojure.core.async + (:refer-clojure :exclude [reduce transduce into merge map take partition partition-by])) + +;; --- alts ------------------------------------------------------------------- +;; do-alts polls each port non-blockingly under its own channel lock; the first +;; ready op wins. A take port is ready when a value (or closed nil) is available; +;; a put spec [ch val] is ready when the value can be offered. Polls with a 1ms +;; backoff (no cross-channel wait-set). + +(defn- alt-attempt [port] + (if (vector? port) + (let [ch (nth port 0) v (nth port 1)] + (assert (some? v) "Can't put nil on channel") + (let [r (clojure.core.async/__offer! ch v)] ; true | false (closed) | nil (would block) + (when (some? r) [r ch]))) + (let [r (clojure.core.async/__poll! port)] ; value | nil (closed) | ::none + (when (not= r ::none) [r port])))) + +(defn do-alts + "Returns [val port] for the first ready op among ports. ports is a vector of + take ports and/or [channel val] put specs. opts may include :priority true + (try in order) and :default val (return [val :default] if none ready)." + [ports opts] + (assert (pos? (count ports)) "alts must have at least one channel operation") + (let [ports (vec ports) + n (count ports) + priority (:priority opts) + has-default (contains? opts :default)] + ;; Scan ports from a random start (sequential, wrapping) so a non-priority alts + ;; is fair without allocating a fresh shuffle every poll. With :priority the scan + ;; starts at 0 (declared order). Returns the first ready op. + (loop [first? true] + (let [start (if priority 0 (rand-int n)) + hit (loop [k 0] + (when (< k n) + (let [j (+ start k) i (if (< j n) j (- j n))] + (or (alt-attempt (nth ports i)) + (recur (inc k))))))] + (cond + hit hit + (and first? has-default) [(:default opts) :default] + :else (do (Thread/sleep 1) (recur false))))))) + +(defn alts!! + "Completes at most one of several channel operations. ports is a vector of take + ports and/or [channel val] put specs. Returns [val port]. Blocks until ready." + [ports & {:as opts}] + (do-alts ports opts)) + +(defn alts! + "Like alts!!. In jolt a go block is a real thread, so parking and blocking alts + are the same operation." + [ports & {:as opts}] + (do-alts ports opts)) + +(defn poll! + "Takes a val from port if possible immediately. Never blocks. Returns the value + or nil." + [port] + (let [r (clojure.core.async/__poll! port)] + (when (not= r ::none) r))) + +;; --- thread variants -------------------------------------------------------- + +(defn thread-call + "Executes f in another thread, returning a channel that receives f's result then + closes." + ([f] (clojure.core.async/go-spawn f)) + ([f _workload] (clojure.core.async/go-spawn f))) + +(defmacro io-thread + "Executes body in another thread, returning a channel that receives the result + then closes." + [& body] + `(thread-call (fn [] ~@body) :io)) + +;; --- pipe / pipeline -------------------------------------------------------- + +(defn pipe + "Takes elements from the from channel and supplies them to the to channel. + Closes to when from closes unless close? is false." + ([from to] (pipe from to true)) + ([from to close?] + (go-loop [] + (let [v (! to v) + (recur))))) + to)) + +(defn- pipeline* + [n to xf from close? ex-handler type] + (assert (pos? n)) + (let [jobs (chan n) + results (chan n) + process (fn [job] + (if (nil? job) + (do (close! results) nil) + (let [v (nth job 0) p (nth job 1) + res (chan 1 xf ex-handler)] + (>!! res v) + (close! res) + (put! p res) + true))) + afn (fn [job] + (if (nil? job) + (do (close! results) nil) + (let [v (nth job 0) p (nth job 1) + res (chan 1)] + (xf v res) + (put! p res) + true)))] + (dotimes [_ n] + (case type + (:blocking :compute) (thread + (loop [] + (let [job (! jobs [v p]) + (>! results p) + (recur))))) + (go-loop [] + (let [p (! to v)) + (recur)))) + (recur))))))) + +(defn pipeline + "Takes elements from from, applies transducer xf with parallelism n, supplies to + to. Outputs are ordered relative to inputs." + ([n to xf from] (pipeline n to xf from true)) + ([n to xf from close?] (pipeline n to xf from close? nil)) + ([n to xf from close? ex-handler] (pipeline* n to xf from close? ex-handler :compute))) + +(defn pipeline-blocking + "Like pipeline, for blocking operations." + ([n to xf from] (pipeline-blocking n to xf from true)) + ([n to xf from close?] (pipeline-blocking n to xf from close? nil)) + ([n to xf from close? ex-handler] (pipeline* n to xf from close? ex-handler :blocking))) + +(defn pipeline-async + "Like pipeline, for async fns af of two args [input result-channel]." + ([n to af from] (pipeline-async n to af from true)) + ([n to af from close?] (pipeline* n to af from close? nil :async))) + +(defn split + "Splits ch by predicate p into [true-chan false-chan]." + ([p ch] (split p ch nil nil)) + ([p ch t-buf-or-n f-buf-or-n] + (let [tc (chan t-buf-or-n) + fc (chan f-buf-or-n)] + (go-loop [] + (let [v (! (if (p v) tc fc) v) + (recur))))) + [tc fc]))) + +;; --- reduce / transduce / collection sinks ---------------------------------- + +(defn reduce + "Returns a channel with the single result of reducing ch with f from init." + [f init ch] + (go-loop [ret init] + (let [v (! ch (first vs))) + (recur (next vs)) + (when close? + (close! ch)))))) + +(defn to-chan! + "Returns a channel containing the contents of coll, closing when exhausted." + [coll] + (let [c (bounded-count 100 coll)] + (if (pos? c) + (let [ch (chan c)] + (onto-chan! ch coll) + ch) + (let [ch (chan)] + (close! ch) + ch)))) + +(defn onto-chan!! + "Like onto-chan! for use when accessing coll might block." + ([ch coll] (onto-chan!! ch coll true)) + ([ch coll close?] + (thread + (loop [vs (seq coll)] + (if (and vs (>!! ch (first vs))) + (recur (next vs)) + (when close? + (close! ch))))))) + +(defn to-chan!! + "Like to-chan! for use when accessing coll might block." + [coll] + (let [c (bounded-count 100 coll)] + (if (pos? c) + (let [ch (chan c)] + (onto-chan!! ch coll) + ch) + (let [ch (chan)] + (close! ch) + ch)))) + +(defn onto-chan + "Deprecated - use onto-chan! or onto-chan!!" + ([ch coll] (onto-chan! ch coll true)) + ([ch coll close?] (onto-chan! ch coll close?))) + +(defn to-chan + "Deprecated - use to-chan! or to-chan!!" + [coll] + (to-chan! coll)) + +(defn into + "Returns a channel with the single collection result of conjoining items from ch + onto coll. ch must close first." + [coll ch] + (reduce conj coll ch)) + +(defn take + "Returns a channel that returns at most n items from ch, then closes." + ([n ch] (take n ch nil)) + ([n ch buf-or-n] + (let [out (chan buf-or-n)] + (go (loop [x 0] + (when (< x n) + (let [v (! out v) + (recur (inc x)))))) + (close! out)) + out))) + +;; --- mult / tap ------------------------------------------------------------- + +(defprotocol Mux + (muxch* [_])) + +(defprotocol Mult + (tap* [m ch close?]) + (untap* [m ch]) + (untap-all* [m])) + +(defn mult + "Creates a mult of ch. Copies can be created with tap and removed with untap. + Each item is distributed to all taps synchronously." + [ch] + (let [cs (atom {}) + m (reify + Mux + (muxch* [_] ch) + Mult + (tap* [_ ch close?] (swap! cs assoc ch close?) nil) + (untap* [_ ch] (swap! cs dissoc ch) nil) + (untap-all* [_] (reset! cs {}) nil)) + dchan (chan 1) + dctr (atom nil) + done (fn [_] (when (zero? (swap! dctr dec)) + (put! dchan true)))] + (go-loop [] + (let [val (! out v) + (recur state)) + (recur state))))) + m)) + +(defn admix + "Adds ch as an input to the mix." + [mix ch] + (admix* mix ch)) + +(defn unmix + "Removes ch as an input to the mix." + [mix ch] + (unmix* mix ch)) + +(defn unmix-all + "Removes all inputs from the mix." + [mix] + (unmix-all* mix)) + +(defn toggle + "Atomically sets the state of one or more channels in a mix." + [mix state-map] + (toggle* mix state-map)) + +(defn solo-mode + "Sets the solo mode of the mix (:mute or :pause)." + [mix mode] + (solo-mode* mix mode)) + +;; --- pub / sub -------------------------------------------------------------- + +(defprotocol Pub + (sub* [p v ch close?]) + (unsub* [p v ch]) + (unsub-all* [p] [p v])) + +(defn pub + "Creates a pub of ch partitioned by topic-fn. Subscribe with sub." + ([ch topic-fn] (pub ch topic-fn (constantly nil))) + ([ch topic-fn buf-fn] + (let [mults (atom {}) + ensure-mult (fn [topic] + (or (get @mults topic) + (get (swap! mults + #(if (% topic) % (assoc % topic (mult (chan (buf-fn topic)))))) + topic))) + p (reify + Mux + (muxch* [_] ch) + Pub + (sub* [_p topic ch close?] + (let [m (ensure-mult topic)] + (tap m ch close?))) + (unsub* [_p topic ch] + (when-let [m (get @mults topic)] + (untap m ch))) + (unsub-all* [_] (reset! mults {})) + (unsub-all* [_ topic] (swap! mults dissoc topic)))] + (go-loop [] + (let [val (! (muxch* m) val) + (swap! mults dissoc topic))) + (recur))))) + p))) + +(defn sub + "Subscribes ch to a topic of pub p." + ([p topic ch] (sub p topic ch true)) + ([p topic ch close?] (sub* p topic ch close?))) + +(defn unsub + "Unsubscribes ch from a topic of pub p." + [p topic ch] + (unsub* p topic ch)) + +(defn unsub-all + "Unsubscribes all channels from a pub, or from a topic." + ([p] (unsub-all* p)) + ([p topic] (unsub-all* p topic))) + +;; --- map / merge ------------------------------------------------------------ + +(defn map + "Applies f to the set of first items from each source channel, then second, etc. + Closes the output channel when any source closes." + ([f chs] (map f chs nil)) + ([f chs buf-or-n] + (let [chs (vec chs) + out (chan buf-or-n) + cnt (count chs) + rets (atom (vec (repeat cnt nil))) + dchan (chan 1) + dctr (atom nil) + done (mapv (fn [i] + (fn [ret] + (swap! rets assoc i ret) + (when (zero? (swap! dctr dec)) + (put! dchan @rets)))) + (range cnt))] + (if (zero? cnt) + (close! out) + (go-loop [] + (reset! dctr cnt) + (dotimes [i cnt] + (take! (nth chs i) (nth done i))) + (let [rets (! out (apply f rets)) + (recur)))))) + out))) + +(defn merge + "Returns a channel with all values taken from the source channels chs. Closes + after all sources close." + ([chs] (merge chs nil)) + ([chs buf-or-n] + (let [out (chan buf-or-n)] + (go-loop [cs (vec chs)] + (if (pos? (count cs)) + (let [[v c] (alts! cs)] + (if (nil? v) + (recur (filterv #(not= c %) cs)) + (do (>! out v) + (recur cs)))) + (close! out))) + out))) + +;; --- deprecated channel ops (rewritten as go-loops) ------------------------- + +(defn map< + "Deprecated - use a transducer. Returns a read-side channel mapping f over ch." + [f ch] + (let [out (chan)] + (go-loop [] + (let [v (! out (f v)) (recur))))) + out)) + +(defn map> + "Deprecated - use a transducer. Returns a write-side channel mapping f into out." + [f out] + (let [in (chan)] + (go-loop [] + (let [v (! out (f v)) (recur))))) + in)) + +(defn filter< + "Deprecated - use a transducer." + ([p ch] (filter< p ch nil)) + ([p ch buf-or-n] + (let [out (chan buf-or-n)] + (go-loop [] + (let [val (! out val)) + (recur))))) + out))) + +(defn remove< + "Deprecated - use a transducer." + ([p ch] (remove< p ch nil)) + ([p ch buf-or-n] (filter< (complement p) ch buf-or-n))) + +(defn filter> + "Deprecated - use a transducer." + [p out] + (let [in (chan)] + (go-loop [] + (let [v (! out v)) + (recur))))) + in)) + +(defn remove> + "Deprecated - use a transducer." + [p out] + (filter> (complement p) out)) + +(defn- mapcat* [f in out] + (go-loop [] + (let [val (! out v)) + (recur)))))) + +(defn mapcat< + "Deprecated - use a transducer." + ([f in] (mapcat< f in nil)) + ([f in buf-or-n] + (let [out (chan buf-or-n)] + (mapcat* f in out) + out))) + +(defn mapcat> + "Deprecated - use a transducer." + ([f out] (mapcat> f out nil)) + ([f out buf-or-n] + (let [in (chan buf-or-n)] + (mapcat* f in out) + in))) + +(defn unique + "Deprecated - use a transducer. Drops consecutive duplicates." + ([ch] (unique ch nil)) + ([ch buf-or-n] + (let [out (chan buf-or-n)] + (go (loop [last nil] + (let [v (! out v) + (recur v)))))) + (close! out)) + out))) + +(defn partition + "Deprecated - use a transducer. Partitions ch into vectors of n." + ([n ch] (partition n ch nil)) + ([n ch buf-or-n] + (let [out (chan buf-or-n)] + (go-loop [arr [] idx 0] + (let [v (! out arr) (recur [] 0)))) + (do (when (> idx 0) (>! out arr)) + (close! out))))) + out))) + +(defn partition-by + "Deprecated - use a transducer. Partitions ch by runs of (f v)." + ([f ch] (partition-by f ch nil)) + ([f ch buf-or-n] + (let [out (chan buf-or-n)] + (go-loop [lst [] last ::nothing] + (let [v (! out lst) (recur [v] new-itm)))) + (do (when (> (count lst) 0) (>! out lst)) + (close! out))))) + out))) diff --git a/stdlib/clojure/core/async/lab.clj b/stdlib/clojure/core/async/lab.clj new file mode 100644 index 0000000..b8c8890 --- /dev/null +++ b/stdlib/clojure/core/async/lab.clj @@ -0,0 +1,34 @@ +;; clojure.core.async.lab — experimental features over the channel primitives. +;; +;; multiplex/broadcast are ported as go-loops over jolt's primitives (the JVM +;; versions reify the impl handler protocol, which jolt does not expose). + +(ns clojure.core.async.lab + (:require [clojure.core.async :as async])) + +(defn multiplex + "Returns a read port that yields values from whichever of ports is ready. A + closed port is dropped; the multiplex port closes once all ports have closed." + [& ports] + (let [out (async/chan)] + (async/go-loop [cs (vec ports)] + (if (pos? (count cs)) + (let [[v c] (async/alts! cs)] + (if (nil? v) + (recur (filterv #(not= c %) cs)) + (do (async/>! out v) + (recur cs)))) + (async/close! out))) + out)) + +(defn broadcast + "Returns a write port that writes each value to all of ports. A write parks until + the value has been written to every port." + [& ports] + (let [in (async/chan)] + (async/go-loop [] + (let [v (async/! p v)) + (recur)))) + in)) diff --git a/test/chez/corpus.edn b/test/chez/corpus.edn index 04f385b..93fa003 100644 --- a/test/chez/corpus.edn +++ b/test/chez/corpus.edn @@ -3349,4 +3349,8 @@ {:suite "symbols / interning" :label "equal symbols share an interned name string" :expected "true" :actual "(let [a (quote ?foo) b (quote ?foo)] (identical? (name a) (name b)))"} {:suite "reader / unquote" :label "~x reads as clojure.core/unquote" :expected "true" :actual "(= (quote (clojure.core/unquote v)) (read-string \"~v\"))"} {:suite "reader / unquote" :label "~@x reads as clojure.core/unquote-splicing" :expected "true" :actual "(= (quote clojure.core/unquote-splicing) (first (read-string \"~@xs\")))"} + {:suite "clojure.core / range" :label "(range 0) is the empty seq ()" :expected "true" :actual "(= () (range 0))"} + {:suite "clojure.core / range" :label "(range 5 5) is the empty seq ()" :expected "true" :actual "(= () (range 5 5))"} + {:suite "clojure.core / range" :label "empty range is not nil" :expected "true" :actual "(some? (range 5 0))"} + {:suite "clojure.core / range" :label "empty range count" :expected "0" :actual "(count (range 0))"} ]