From 7d1e1f42e1774b417eb57124c825837002c4c24f Mon Sep 17 00:00:00 2001 From: Yogthos Date: Fri, 5 Jun 2026 15:17:15 -0400 Subject: [PATCH] =?UTF-8?q?feat:=20clojure.core.async=20on=20Janet=20fiber?= =?UTF-8?q?s=20(Phase=201=20=E2=80=94=20API=20layer)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Janet fibers are stackful coroutines, so a go block is just its body run in a fiber that parks on channel ops by yielding to the event loop — the interpreter call stack rides along, no CPS/state-machine transform. So ! work anywhere (inside try, nested fns, loops), unlike Clojure's go macro. src/jolt/async.janet implements chan/chan?/close!/!/!!/go/go-loop/ thread/alts!/timeout/put!/take! over ev/ channels and fibers, installed as the clojure.core.async namespace (pre-populated in init, so require finds it). A channel is a pair of ev/chans (:ch values + :done close-signal); a take is (ev/select :ch :done), which drains buffered values before the close signal — giving Clojure's drain-then-nil semantics without the buffer loss of ev/chan-close, and with no leaked fibers (close! just closes :done). Single-threaded cooperative scheduling: !`/`!!`, `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 (`…)`). - **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/api.janet b/src/jolt/api.janet index 71526be..8e5ba4b 100644 --- a/src/jolt/api.janet +++ b/src/jolt/api.janet @@ -9,6 +9,7 @@ (use ./core) (use ./compiler) (use ./loader) +(use ./async) (defn normalize-pvecs "Deep-convert any sequential (pvec/tuple/array) to a Janet tuple. Test helper @@ -37,6 +38,9 @@ # via JOLT_MUTABLE (see config.janet); init-core! registers vec/vector/conj/ # etc. that produce the mode-appropriate values, so nothing extra to load. (init-core! ctx) + # clojure.core.async (channels + go blocks on Janet fibers); pre-populated + # so (require '[clojure.core.async ...]) finds it and applies :as/:refer. + (install-async! ctx) ctx)) (defn eval-string diff --git a/src/jolt/async.janet b/src/jolt/async.janet new file mode 100644 index 0000000..9a5c3c3 --- /dev/null +++ b/src/jolt/async.janet @@ -0,0 +1,145 @@ +# clojure.core.async on Janet fibers. +# +# Janet fibers are stackful coroutines, so a `go` block is just "run the body in +# a fiber" — the body parks on a channel op by yielding to the event loop, and +# the whole interpreter call stack rides along on the fiber's stack. No CPS/state +# machine transform (unlike Clojure's `go` macro), so ! work anywhere +# (inside try, nested fns, loops, …). +# +# A channel is a pair of Janet ev/chans wrapped in a tagged table: a `:ch` that +# carries values and a `:done` that is closed to signal channel close. A take is +# `(ev/select :ch :done)` — ev/select checks in order, so buffered values drain +# before the close signal is seen, giving Clojure's drain-then-nil semantics. We +# use a separate `:done` channel because Janet's ev/chan-close *discards* a +# channel's buffered values. close! just closes :done (idempotent, no fiber), so +# nothing leaks. +# +# Single OS thread: go blocks run cooperatively on the event loop, so n 0)) (ev/chan n) (ev/chan)) (ev/chan))) + +(defn async-close! [ch] + (when (not (in (ch :closed) 0)) + (put (ch :closed) 0 true) + (protect (ev/chan-close (ch :done)))) + nil) + +# ! / >!! — 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). +(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))) + +# 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 +# just closes. Buffered(1) so a fire-and-forget go leaves no parked fiber. +(defn async-go-spawn [thunk] + (def w (async-chan 1)) + (ev/go (fn [] + (def res (protect (thunk))) + (when (and (in res 0) (not (nil? (in res 1)))) + (async-give w (in res 1))) + (async-close! w))) + w) + +# (alts! [ch ...]) — take from whichever channel is ready first; returns +# [value channel] (value is nil if that channel closed). Take-only for v1. +(defn async-alts [chans] + (def cs (cond (pvec? chans) (pv->array chans) + (tuple? chans) chans + (array? chans) chans + (error "alts! expects a vector of channels"))) + (def raws @[]) + (def lookup @{}) # raw ev/chan -> [jolt-chan done?] + (each c cs + (array/push raws (c :ch)) (put lookup (c :ch) [c false]) + (array/push raws (c :done)) (put lookup (c :done) [c true])) + (def r (ev/select ;raws)) + (def info (get lookup (in r 1))) + (def jc (in info 0)) + (def val (if (or (in info 1) (not= :take (in r 0))) nil (in r 2))) + (pv-from-indexed @[val jc])) + +# (timeout ms) — a channel that closes after ms milliseconds. +(defn async-timeout [ms] + (def w (async-chan)) + (ev/go (fn [] (ev/sleep (/ ms 1000)) (async-close! w))) + w) + +# (put! ch v [cb]) — async put; (take! ch cb) — async take. Fire a fiber and +# call the optional callback with the result. +(defn async-put! [ch v &opt cb] + (ev/go (fn [] + (def ok (async-give ch v)) + (when (and cb (not (nil? cb))) (cb ok)))) + nil) +(defn async-take! [ch cb] + (ev/go (fn [] + (def val (async-take ch)) + (when (and cb (not (nil? cb))) (cb val)))) + nil) + +# --- macros (Janet macro-fns that return forms) --- + +(defn- sym [name &opt ns] {:jolt/type :symbol :ns ns :name name}) + +# (go body...) -> (go-spawn (fn* [] body...)) +(defn async-go [& body] + @[(sym "go-spawn" "clojure.core.async") (array (sym "fn*") [] ;body)]) + +# (go-loop bindings body...) -> (go (loop bindings body...)) +(defn async-go-loop [bindings & body] + @[(sym "go" "clojure.core.async") (array (sym "loop") bindings ;body)]) + +# (thread body...) — runs cooperatively in a fiber here (no OS thread); same +# shape as go (returns a result channel). +(defn async-thread [& body] + @[(sym "go-spawn" "clojure.core.async") (array (sym "fn*") [] ;body)]) + +(def- async-bindings + @{"chan" async-chan + "chan?" jolt-chan? + "close!" async-close! + "!" async-give ">!!" async-give + "alts!" async-alts "alts!!" async-alts + "timeout" async-timeout + "put!" async-put! + "take!" async-take! + "go-spawn" async-go-spawn + "go" async-go + "go-loop" async-go-loop + "thread" async-thread}) + +(def- async-macros @{"go" true "go-loop" true "thread" true}) + +(defn install-async! + "Create/populate the clojure.core.async namespace in ctx." + [ctx] + (let [ns (ctx-find-ns ctx "clojure.core.async")] + (loop [[name f] :pairs async-bindings] + (def v (ns-intern ns name f)) + (when (get async-macros name) (put v :macro true))) + ns)) diff --git a/src/jolt/core.janet b/src/jolt/core.janet index 079f742..bf959a5 100644 --- a/src/jolt/core.janet +++ b/src/jolt/core.janet @@ -1557,6 +1557,7 @@ (set? v) (pr-render-seq buf (phs-seq v) "#{" "}") (phm? v) (pr-render-pairs buf (phm-entries v)) (core-transient? v) (buffer/push-string buf (string "#")) + (and (table? v) (= :jolt/chan (get v :jolt/type))) (buffer/push-string buf "#") (pvec? v) (pr-render-seq buf (pv->array v) "[" "]") (plist? v) (pr-render-seq buf (pl->array v) "(" ")") (and (table? v) (get v :jolt/deftype)) (buffer/push-string buf (string v)) diff --git a/src/jolt/main.janet b/src/jolt/main.janet index 61f0d30..bdb22c1 100644 --- a/src/jolt/main.janet +++ b/src/jolt/main.janet @@ -104,6 +104,9 @@ (and (table? v) (= :jolt/transient (v :jolt/type))) (push-str buf (string "#")) + (and (table? v) (= :jolt/chan (v :jolt/type))) + (push-str buf "#") + (phm? v) (do (push-str buf "{") diff --git a/test/spec/core-async-spec.janet b/test/spec/core-async-spec.janet new file mode 100644 index 0000000..3170d49 --- /dev/null +++ b/test/spec/core-async-spec.janet @@ -0,0 +1,55 @@ +# Specification: clojure.core.async on Janet fibers (Phase 1 — API layer). +# Each case is self-contained: it requires the ns, sets up channels/go blocks, +# and ends with a take that pumps the event loop and yields the value compared. +(use ../support/harness) + +(def REQ + "(require '[clojure.core.async :refer [go go-loop chan ! close! alts! timeout put! take! chan?]]) ") +(defn- a [body] (string "(do " REQ body ")")) + +(defspec "core.async / go & channels" + ["go produce, ! c (+ 40 2))) ( channel closes" + "nil" (a "(! x 10)) (go (>! y 32)) (! c 1) (>! c 2) (>! c 3) (close! c)) (! c :a) (close! c)) (! to a closed channel is false" + "false" (a "(def c (chan 1)) (close! c) (>! c 1)")] + ["take from closed empty channel is nil" + "nil" (a "(def c (chan)) (close! c) (! in 1) (>! in 2) (>! in 3) (close! in)) (! mid (inc (! out (* 10 (! in 4)) (! y :v)) (! c 7)) (