From 54db79e92746deecd8f5ae83e44465ed44fc3be5 Mon Sep 17 00:00:00 2001 From: Yogthos Date: Fri, 5 Jun 2026 15:40:24 -0400 Subject: [PATCH] =?UTF-8?q?feat:=20core.async=20Phase=203=20=E2=80=94=20ch?= =?UTF-8?q?annel=20transducers=20+=20dropping/sliding=20buffers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit (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. --- .beads/issues.jsonl | 2 +- README.md | 2 +- src/jolt/async.janet | 69 ++++++++++++++++++++++++++++++--- test/spec/core-async-spec.janet | 29 +++++++++++++- test/support/harness.janet | 1 + 5 files changed, 94 insertions(+), 9 deletions(-) diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index 32fccc1..34a56cf 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -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} diff --git a/README.md b/README.md index 65b0c3d..7918dd4 100644 --- a/README.md +++ b/README.md @@ -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 (`!`/`!!`, `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 (`…)`). - **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. diff --git a/src/jolt/async.janet b/src/jolt/async.janet index 2d1ba93..8f9c385 100644 --- a/src/jolt/async.janet +++ b/src/jolt/async.janet @@ -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 diff --git a/test/spec/core-async-spec.janet b/test/spec/core-async-spec.janet index 726e483..dfaec9f 100644 --- a/test/spec/core-async-spec.janet +++ b/test/spec/core-async-spec.janet @@ -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 "(! 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)" + " ( (length fails) 0) (error (string suite ": " (length fails) " failing behavior(s)")))