feat: core.async Phase 3 — channel transducers + dropping/sliding buffers
(chan n xform) applies a transducer on the put side: a jolt transducer is a directly-callable closure, composed over a reducing fn whose step gives each output value into the channel (honoring its buffer kind). One put may yield zero or more values; a reduced result (e.g. from take) closes the channel; close! runs the transducer completion arity to flush stateful remainders. Works with map/filter/mapcat/take/comp/etc. Buffers: (buffer n) fixed, (dropping-buffer n) drops new values when full, (sliding-buffer n) drops the oldest. Implemented via a non-blocking give — (ev/select [ch v] closed-chan) detects a full buffer without parking. harness: run-spec flushes per suite. spec: core.async/channel-transducers (5), core.async/buffers (3). jpm test green. Note: distinct/dedupe/partition-all/partition-by still lack a 0-coll transducer arity in core (separate gap), so they can't yet be used as channel xforms.
This commit is contained in:
Yogthos
parent
e8cb4605ac
commit
54db79e927
5 changed files with 94 additions and 9 deletions
|
|
@ -5,7 +5,7 @@
|
|||
{"_type":"issue","id":"jolt-x8s","title":"CRITICAL: collections not callable as IFn (vector/map/set)","status":"closed","priority":1,"issue_type":"bug","owner":"yogthos@gmail.com","created_at":"2026-06-04T17:46:14Z","created_by":"Yogthos","updated_at":"2026-06-04T17:50:54Z","closed_at":"2026-06-04T17:50:54Z","dependency_count":0,"dependent_count":0,"comment_count":0}
|
||||
{"_type":"issue","id":"jolt-alz","title":"CRITICAL: self-referential lazy-seq/lazy-cat yields only literal prefix","status":"closed","priority":1,"issue_type":"bug","owner":"yogthos@gmail.com","created_at":"2026-06-04T17:46:13Z","created_by":"Yogthos","updated_at":"2026-06-04T18:09:15Z","closed_at":"2026-06-04T18:09:15Z","dependency_count":0,"dependent_count":0,"comment_count":0}
|
||||
{"_type":"issue","id":"jolt-wec","title":"CRITICAL: multi-collection map returns unrealized thunk","status":"closed","priority":1,"issue_type":"bug","owner":"yogthos@gmail.com","created_at":"2026-06-04T17:46:12Z","created_by":"Yogthos","updated_at":"2026-06-04T18:06:02Z","closed_at":"2026-06-04T18:06:02Z","dependency_count":0,"dependent_count":0,"comment_count":0}
|
||||
{"_type":"issue","id":"jolt-5i5","title":"core.async Phase 2: dynamic var binding conveyance","description":"Snapshot/restore the global binding-stack per fiber so dynamic var bindings don't interleave across concurrent go blocks (Clojure binding conveyance).","status":"open","priority":2,"issue_type":"feature","owner":"yogthos@gmail.com","created_at":"2026-06-05T18:55:08Z","created_by":"Yogthos","updated_at":"2026-06-05T18:55:08Z","dependency_count":0,"dependent_count":0,"comment_count":0}
|
||||
{"_type":"issue","id":"jolt-5i5","title":"core.async Phase 2: dynamic var binding conveyance","description":"Snapshot/restore the global binding-stack per fiber so dynamic var bindings don't interleave across concurrent go blocks (Clojure binding conveyance).","status":"closed","priority":2,"issue_type":"feature","owner":"yogthos@gmail.com","created_at":"2026-06-05T18:55:08Z","created_by":"Yogthos","updated_at":"2026-06-05T19:22:50Z","closed_at":"2026-06-05T19:22:50Z","close_reason":"Binding stack moved to fiber-local dyn; go-spawn snapshots+installs for conveyance; concurrent isolation + inner shadowing verified (4 spec cases)","dependency_count":0,"dependent_count":0,"comment_count":0}
|
||||
{"_type":"issue","id":"jolt-tkw","title":"core.async on Janet fibers — Phase 1: API layer","description":"Implement clojure.core.async API (chan/go/\u003c!/\u003e!/\u003c!!/\u003e!!/close!/alts!/timeout/put!/take!/buffers) on top of Janet ev/ channels and fibers. go = run body in a fiber (stackful, no CPS macro needed).","status":"closed","priority":2,"issue_type":"feature","owner":"yogthos@gmail.com","created_at":"2026-06-05T18:55:07Z","created_by":"Yogthos","updated_at":"2026-06-05T19:17:25Z","closed_at":"2026-06-05T19:17:25Z","close_reason":"Phase 1 complete: chan/go/\u003c!/\u003e!/close!/alts!/timeout/put!/take! on Janet fibers; two-channel design (values + done) gives drain-then-nil with no leaked fibers; 16 spec cases","dependency_count":0,"dependent_count":0,"comment_count":0}
|
||||
{"_type":"issue","id":"jolt-lko","title":"clojure-test-suite: strictness — throw on malformed calls like Clojure","description":"Make jolt throw where Clojure throws (bad-shape conj!/assoc!, arithmetic on non-numbers, out-of-range indices, etc.) to satisfy the suite's ~292 p/thrown? assertions. Targeted input validation across affected fns; guard against regressing jolt's spec/conformance.","status":"closed","priority":2,"issue_type":"task","assignee":"Yogthos","owner":"yogthos@gmail.com","created_at":"2026-06-05T14:45:20Z","created_by":"Yogthos","updated_at":"2026-06-05T18:07:56Z","started_at":"2026-06-05T14:45:39Z","closed_at":"2026-06-05T18:07:56Z","close_reason":"Implemented Clojure-strict argument validation across ~30 fns; suite pass 3691-\u003e3898, p/thrown? bucket 292-\u003e~60 (remainder is platform: int 32-bit range, subs Unicode, min-key NaN, subvec floats, map-entry-as-2-vector)","dependency_count":0,"dependent_count":0,"comment_count":0}
|
||||
{"_type":"issue","id":"jolt-kxb","title":"clojure-test-suite: transducers don't short-circuit over infinite seqs (hangs)","description":"(into [] (take 5) (range)) and similar transducer-over-infinite forms hang jolt's eager evaluator (no reduced/early-termination). 7 suite files time out. Make transducing honor reduced.","status":"closed","priority":2,"issue_type":"bug","assignee":"Yogthos","owner":"yogthos@gmail.com","created_at":"2026-06-05T13:05:18Z","created_by":"Yogthos","updated_at":"2026-06-05T13:18:37Z","started_at":"2026-06-05T13:09:59Z","closed_at":"2026-06-05T13:18:37Z","close_reason":"reduce-with-reduced steps lazy seqs incrementally; transducers short-circuit over infinite seqs","dependency_count":0,"dependent_count":0,"comment_count":0}
|
||||
|
|
|
|||
|
|
@ -78,7 +78,7 @@ Jolt targets Clojure semantics but runs on Janet, not the JVM. The notable diver
|
|||
- **Collections.** By default Jolt uses immutable persistent data structures: vectors are 32-way branching tries (structural-sharing persistent vectors with O(log₃₂ n) `conj`/`assoc`/`nth`), lists are persistent singly-linked cons cells (O(1) `conj`/`cons` prepend with structural sharing), and maps/sets are persistent hash structures. Value equality and sequence operations are Clojure-compatible, but hash-map/hash-set iteration order is unspecified and differs from Clojure — use `sorted-map`/`sorted-set` when order matters.
|
||||
- **Mutable build mode.** Jolt can be compiled to use fast Janet-native *mutable* collections instead, via a build-time flag: `JOLT_MUTABLE=1 jpm build` (default `jpm build` is immutable). In mutable mode vectors and lists share one mutable array representation (so `conj` mutates in place and appends, and `vector?`/`list?` no longer distinguish them) — a performance/looseness trade-off. The default immutable build has full Clojure value semantics.
|
||||
- **Concurrency / STM.** Single-threaded. No refs, `dosync`, agents, or `send`; `locking` evaluates its body without real locking. Atoms, volatiles, and delays are supported.
|
||||
- **core.async.** `clojure.core.async` runs on Janet fibers and channels (`chan`, `go`, `go-loop`, `<!`/`>!`/`<!!`/`>!!`, `close!`, `alts!`, `timeout`, `put!`/`take!`). Because Janet fibers are stackful coroutines, a `go` block is just its body run in a fiber — no CPS/state-machine rewrite — so `<!`/`>!` work *anywhere*, including inside `try`, nested `fn`s, and loops (positions Clojure's `go` macro forbids). Go blocks are cooperatively scheduled on one OS thread, so parking (`<!`) and blocking (`<!!`) coincide; `thread` runs cooperatively too. Dynamic-var binding conveyance and channel transducers are not yet implemented.
|
||||
- **core.async.** `clojure.core.async` runs on Janet fibers and channels (`chan`, `go`, `go-loop`, `<!`/`>!`/`<!!`/`>!!`, `close!`, `alts!`, `timeout`, `put!`/`take!`, `buffer`/`dropping-buffer`/`sliding-buffer`, and channel transducers via `(chan n xform)`). Because Janet fibers are stackful coroutines, a `go` block is just its body run in a fiber — no CPS/state-machine rewrite — so `<!`/`>!` work *anywhere*, including inside `try`, nested `fn`s, and loops (positions Clojure's `go` macro forbids). Go blocks are cooperatively scheduled on one OS thread, so parking (`<!`) and blocking (`<!!`) coincide; `thread` runs cooperatively too. Dynamic-var bindings are conveyed into `go` blocks (each go block sees the bindings in effect when it was spawned).
|
||||
- **Regex.** Compiled to Janet's PEG engine (Janet has no regex). Supported: capturing groups (`[whole g1 …]`), greedy and lazy quantifiers with backtracking, `(?:…)`, lookahead `(?=…)`/`(?!…)`, alternation, anchors `^ $ \b \B`, character classes, and the `(?i)` flag. Not supported: lookbehind, backreferences (`\1`), and named groups (`(?<name>…)`).
|
||||
- **Arrays.** Java-style arrays map onto Janet's native types: `byte-array` is a Janet buffer (contiguous, C-backed); `object-array`/`int-array`/`double-array`/etc. are Janet arrays. `aget`/`aset`/`alength`/`aclone` work over both.
|
||||
- **Transients.** `transient`/`conj!`/`assoc!`/`dissoc!`/`disj!`/`pop!`/`persistent!` are real mutable scratch collections backed by Janet's native arrays and tables (vectors → arrays, maps/sets → tables), so building a collection with them avoids the per-step copying of the persistent path (notably for maps/sets). `persistent!` freezes back to a persistent value.
|
||||
|
|
|
|||
|
|
@ -28,14 +28,61 @@
|
|||
(defn- vchan [x]
|
||||
(if (jolt-chan? x) (x :ch) (error (string "expected a channel, got " (type x)))))
|
||||
|
||||
# (chan) unbuffered, (chan n) fixed buffer of n. A transducer 3rd arg is
|
||||
# accepted but ignored until Phase 3.
|
||||
(defn async-chan [&opt n xform]
|
||||
(wrap (if (and (number? n) (> n 0)) (ev/chan n) (ev/chan)) (ev/chan)))
|
||||
(defn- reduced? [x] (and (table? x) (= :jolt/reduced (get x :jolt/type))))
|
||||
|
||||
# Buffer specs: (buffer n) fixed, (dropping-buffer n) drops new values when full,
|
||||
# (sliding-buffer n) drops the oldest when full.
|
||||
(defn async-buffer [n] @{:jolt/type :jolt/buffer :kind :fixed :n n})
|
||||
(defn async-dropping-buffer [n] @{:jolt/type :jolt/buffer :kind :dropping :n n})
|
||||
(defn async-sliding-buffer [n] @{:jolt/type :jolt/buffer :kind :sliding :n n})
|
||||
(defn- buffer-spec? [x] (and (table? x) (= :jolt/buffer (get x :jolt/type))))
|
||||
|
||||
# An always-ready channel, used as the non-blocking fallback in ev/select so a
|
||||
# give can detect "buffer full" without parking.
|
||||
(def- full-signal (let [c (ev/chan)] (ev/chan-close c) c))
|
||||
|
||||
# Put one value into the channel's value chan honoring its buffer kind. Returns
|
||||
# true (the put "succeeds" even when dropped, like Clojure's dropping/sliding).
|
||||
(defn- buf-give [ch v]
|
||||
(case (ch :bufkind)
|
||||
:dropping (do (ev/select [(ch :ch) v] full-signal) true) # give if room, else drop
|
||||
:sliding (let [r (ev/select [(ch :ch) v] full-signal)]
|
||||
(when (= :close (in r 0)) # full: drop oldest, then add
|
||||
(protect (ev/take (ch :ch))) (protect (ev/give (ch :ch) v)))
|
||||
true)
|
||||
(if (in (protect (ev/give (ch :ch) v)) 0) true false))) # fixed/unbuffered: may park
|
||||
|
||||
# A channel transducer is applied on the put side. We build a reducing fn whose
|
||||
# step gives each output value into the channel (honoring its buffer kind); the
|
||||
# accumulator is the chan, threaded through but unused (output is the
|
||||
# side-effecting give). A jolt transducer/rf is a jolt closure, directly
|
||||
# callable as a Janet function.
|
||||
(defn- make-add-rf [w]
|
||||
(fn [& args]
|
||||
(case (length args)
|
||||
0 (w :ch) # init
|
||||
1 (in args 0) # completion: nothing extra to do
|
||||
(do (buf-give w (in args 1)) (in args 0))))) # step: give output
|
||||
|
||||
# (chan) unbuffered; (chan n) / (chan (buffer n)) fixed; (chan (dropping-buffer
|
||||
# n)) / (chan (sliding-buffer n)); a 2nd arg transducer composes over the buffer.
|
||||
(defn async-chan [&opt buf xform]
|
||||
(def spec (cond
|
||||
(buffer-spec? buf) buf
|
||||
(and (number? buf) (> buf 0)) {:kind :fixed :n buf}
|
||||
nil))
|
||||
(def vc (if spec (ev/chan (spec :n)) (ev/chan)))
|
||||
(def w (wrap vc (ev/chan)))
|
||||
(when spec (put w :bufkind (spec :kind)))
|
||||
(when (and xform (not (nil? xform)))
|
||||
(put w :xrf (xform (make-add-rf w))))
|
||||
w)
|
||||
|
||||
(defn async-close! [ch]
|
||||
(when (not (in (ch :closed) 0))
|
||||
(put (ch :closed) 0 true)
|
||||
# flush any buffered state of a stateful transducer (completion arity)
|
||||
(when (ch :xrf) (protect ((ch :xrf) (ch :ch))))
|
||||
(protect (ev/chan-close (ch :done))))
|
||||
nil)
|
||||
|
||||
|
|
@ -47,10 +94,17 @@
|
|||
|
||||
# >! / >!! — put, parking the fiber. Returns true if delivered, false if the
|
||||
# channel is closed. nil may not be put on a channel (it is the closed value).
|
||||
# With a transducer, the value is run through it (so one put may yield zero or
|
||||
# more values on the channel); a `reduced` result (e.g. from `take`) closes it.
|
||||
(defn async-give [ch v]
|
||||
(when (nil? v) (error "Can't put nil on a channel"))
|
||||
(if (in (ch :closed) 0) false
|
||||
(if (in (protect (ev/give (ch :ch) v)) 0) true false)))
|
||||
(cond
|
||||
(in (ch :closed) 0) false
|
||||
(ch :xrf)
|
||||
(let [r ((ch :xrf) (ch :ch) v)]
|
||||
(when (reduced? r) (async-close! ch))
|
||||
true)
|
||||
(buf-give ch v)))
|
||||
|
||||
# Run thunk (a jolt 0-arg closure, directly callable) in a fiber; return a
|
||||
# buffered(1) channel that conveys its value once, then closes. A nil result
|
||||
|
|
@ -132,6 +186,9 @@
|
|||
">!" async-give ">!!" async-give
|
||||
"alts!" async-alts "alts!!" async-alts
|
||||
"timeout" async-timeout
|
||||
"buffer" async-buffer
|
||||
"dropping-buffer" async-dropping-buffer
|
||||
"sliding-buffer" async-sliding-buffer
|
||||
"put!" async-put!
|
||||
"take!" async-take!
|
||||
"go-spawn" async-go-spawn
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@
|
|||
(use ../support/harness)
|
||||
|
||||
(def REQ
|
||||
"(require '[clojure.core.async :refer [go go-loop chan <! >! close! alts! timeout put! take! chan?]]) ")
|
||||
"(require '[clojure.core.async :refer [go go-loop chan <! >! close! alts! timeout put! take! chan? buffer dropping-buffer sliding-buffer]]) ")
|
||||
(defn- a [body] (string "(do " REQ body ")"))
|
||||
|
||||
(defspec "core.async / go & channels"
|
||||
|
|
@ -69,3 +69,30 @@
|
|||
["go's own binding shadows conveyed"
|
||||
":inner"
|
||||
(d "(<! (binding [*x* :outer] (go (binding [*x* :inner] (<! (timeout 5)) *x*))))")])
|
||||
|
||||
# Channel transducers (Phase 3): a transducer is applied on the put side, so one
|
||||
# put may yield zero or more values; `take` closes the channel early.
|
||||
(defn- drain [setup]
|
||||
(string "(do " REQ setup
|
||||
" (<! (go-loop [o []] (let [v (<! c)] (if (nil? v) o (recur (conj o v)))))))"))
|
||||
(defspec "core.async / channel transducers"
|
||||
["map transducer"
|
||||
"[2 3 4]" (drain "(def c (chan 10 (map inc))) (go (>! c 1) (>! c 2) (>! c 3) (close! c))")]
|
||||
["filter transducer"
|
||||
"[0 2 4]" (drain "(def c (chan 10 (filter even?))) (go (doseq [x (range 6)] (>! c x)) (close! c))")]
|
||||
["mapcat expands"
|
||||
"[1 1 2 2]" (drain "(def c (chan 10 (mapcat (fn [x] [x x])))) (go (>! c 1) (>! c 2) (close! c))")]
|
||||
["take closes early"
|
||||
"[:a :b]" (drain "(def c (chan 10 (take 2))) (go (>! c :a) (>! c :b) (>! c :c) (>! c :d) (close! c))")]
|
||||
["comp of transducers"
|
||||
"[10 30 50]" (drain "(def c (chan 10 (comp (filter odd?) (map (fn [x] (* x 10)))))) (go (doseq [x (range 6)] (>! c x)) (close! c))")])
|
||||
|
||||
# Buffers: fixed (default), dropping (drops new when full), sliding (drops oldest
|
||||
# when full). Filled synchronously on this fiber (dropping/sliding never park).
|
||||
(defn- fill [bufexpr]
|
||||
(string "(do " REQ "(def c (chan " bufexpr ")) (doseq [x [1 2 3 4 5]] (>! c x)) (close! c)"
|
||||
" (<! (go-loop [o []] (let [v (<! c)] (if (nil? v) o (recur (conj o v)))))))"))
|
||||
(defspec "core.async / buffers"
|
||||
["dropping-buffer keeps first" "[1 2]" (fill "(dropping-buffer 2)")]
|
||||
["sliding-buffer keeps last" "[4 5]" (fill "(sliding-buffer 2)")]
|
||||
["fixed (buffer n) holds all" "[1 2 3 4 5]" (fill "(buffer 5)")])
|
||||
|
|
|
|||
|
|
@ -54,6 +54,7 @@
|
|||
(= (r 1) true) (++ pass)
|
||||
(array/push fails [label (string "want " expected ", got " (show actual))])))))
|
||||
(printf " %s: %d/%d" suite pass (length cases))
|
||||
(flush)
|
||||
(each [l m] fails (printf " FAIL [%s] %s" l m))
|
||||
(when (> (length fails) 0)
|
||||
(error (string suite ": " (length fails) " failing behavior(s)")))
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue