From 48ed72974f74d330601ac4a9cdc60e4a8d8b43ef Mon Sep 17 00:00:00 2001
From: Yogthos <yogthos@gmail.com>
Date: Sat, 20 Jun 2026 13:48:10 -0400
Subject: [PATCH] Chez concurrency pt.2: clojure.core.async on real-thread
 blocking channels
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

No mature Chez fibers library exists and this is a threaded Chez build, so a go
block is an OS thread and a channel is a mutex+condition blocking queue: <! / >!
are the blocking <!! / >!! and work anywhere (no CPS transform), like the Janet
stackful-fiber model but with real parallelism and a shared heap.

host/chez/async.ss provides chan (unbuffered rendezvous / fixed / dropping /
sliding), <! >! <!! >!! close! alts! timeout put! take! buffer ctors, channel
transducers, and go-spawn, all def-var!'d into clojure.core.async; go/go-loop/
thread are macros (mark-macro!) expanding to go-spawn, mirroring src/jolt/
async.janet. Binding conveyance rides the thread-parameter binding stack from
pt.1. alts! polls with a 1ms backoff (no cross-channel wait-set yet) and is
take-only, matching the Janet impl.

(require '[clojure.core.async ...]) resolves it with no file load — the vars are
resident and require just registers the :as/:refer.

cli-test covers go/buffered-drain/nested-<!/alts!/transducer/timeout/binding-
conveyance (43/43). core.async isn't in the conformance corpus (feature-gated
:async/core-async), so coverage is the Chez cli-test plus the existing Janet
core-async-spec. Seed unchanged (no .clj touched). Prelude corpus 2534->2559,
zero-Janet 2569, 0 new divergences on either; Janet gate + JVM cert green.

jolt-byjr
---
 host/chez/async.ss                 | 216 +++++++++++++++++++++++++++++
 host/chez/rt.ss                    |   5 +
 test/chez/cli-test.janet           |  10 +-
 test/chez/run-corpus-prelude.janet |   4 +-
 4 files changed, 233 insertions(+), 2 deletions(-)
 create mode 100644 host/chez/async.ss

diff --git a/host/chez/async.ss b/host/chez/async.ss
new file mode 100644
index 0000000..0d2a1ca
--- /dev/null
+++ b/host/chez/async.ss
@@ -0,0 +1,216 @@
+;; async.ss (jolt-byjr) — clojure.core.async on real OS threads for the Chez host.
+;;
+;; No mature Chez fibers library exists, and this Chez is a threaded build, so a
+;; `go` block is just an OS thread and a channel is a mutex+condition blocking
+;; queue: <! / >! are the blocking <!! / >!! (they "park" by blocking the thread).
+;; Like the Janet stackful-fiber model, <! / >! work ANYWHERE (no CPS transform) —
+;; here because they are ordinary blocking calls. Real parallelism, shared heap.
+;; Trade-off: one OS thread per go block (fine for typical use / conformance, not
+;; for thousands of simultaneous go blocks).
+;;
+;; 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.
+;;
+;; The fns are def-var!'d into clojure.core.async; go/go-loop/thread are macros
+;; (mark-macro!) expanding to go-spawn, mirroring src/jolt/async.janet. Loaded after
+;; concurrency.ss (reuses ms->duration). Requires a threaded Chez build.
+
+;; --- buffers ----------------------------------------------------------------
+(define-record-type async-buffer (fields n kind) (nongenerative async-buffer-v1))
+(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))
+
+;; --- channels ---------------------------------------------------------------
+;; items: a FIFO list of (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.
+(define-record-type async-chan
+  (fields mu cv (mutable items) cap kind (mutable closed?) (mutable xrf))
+  (nongenerative async-chan-v1))
+
+(define (ac-qlen ch) (length (async-chan-items ch)))
+(define (ac-qempty? ch) (null? (async-chan-items ch)))
+(define (ac-qpush! ch entry) (async-chan-items-set! ch (append (async-chan-items ch) (list entry))))
+(define (ac-qpop! ch)
+  (let ((e (car (async-chan-items ch))))
+    (async-chan-items-set! ch (cdr (async-chan-items ch))) e))
+(define (ac-qdrop-oldest! ch) (async-chan-items-set! ch (cdr (async-chan-items ch))))
+
+;; enqueue honoring the buffer kind (used by the transducer step + buffered puts).
+(define (ac-buf-give! ch v)
+  (case (async-chan-kind ch)
+    ((dropping) (when (< (ac-qlen ch) (async-chan-cap ch)) (ac-qpush! ch (cons v #f))))
+    ((sliding)  (when (>= (ac-qlen ch) (async-chan-cap ch)) (ac-qdrop-oldest! ch))
+                (ac-qpush! ch (cons v #f)))
+    (else       (ac-qpush! ch (cons v #f))))      ; fixed: caller ensured room
+  (condition-broadcast (async-chan-cv ch)))
+
+;; A transducer is a jolt fn (xform); (xform add-rf) yields the channel's reducing
+;; fn. add-rf: 0-arg init, 1-arg completion, 2-arg step (enqueue the output). A
+;; `reduced` step result closes the channel. Mirrors async.janet make-add-rf.
+(define (ac-make-add-rf ch)
+  (lambda args
+    (cond ((null? args) ch)                                   ; init
+          ((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) '() cap kind #f xrf))
+
+;; (chan) | (chan n) | (chan buf) | (chan n|buf xform)
+(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)))
+    (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)))
+        (unless (jolt-nil? xform)
+          (async-chan-xrf-set! ch (jolt-invoke xform (ac-make-add-rf ch))))
+        ch))))
+
+;; close! (idempotent): mark closed, flush a stateful transducer's completion, and
+;; wake everyone. ac-close! assumes the lock is held; the public form takes it.
+(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)))
+    (condition-broadcast (async-chan-cv ch)))
+  jolt-nil)
+(define (jolt-async-close! ch) (with-mutex (async-chan-mu ch) (ac-close! ch)))
+
+;; >! / >!! — put, blocking. false if closed; nil may not be put. With a
+;; 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))))
+  (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)))
+         (when (jolt-reduced? r) (ac-close! ch))
+         #t))
+      (else
+       (case (async-chan-kind ch)
+         ((dropping sliding) (ac-buf-give! ch v) #t)
+         (else
+          (if (> (async-chan-cap ch) 0)
+              (let loop ()                                    ; buffered fixed: wait for room
+                (cond ((async-chan-closed? ch) #f)
+                      ((< (ac-qlen ch) (async-chan-cap ch))
+                       (ac-qpush! ch (cons v #f)) (condition-broadcast (async-chan-cv ch)) #t)
+                      (else (condition-wait (async-chan-cv ch) (async-chan-mu ch)) (loop))))
+              (let ((box (vector #f)))                        ; unbuffered: rendezvous
+                (ac-qpush! ch (cons v box))
+                (condition-broadcast (async-chan-cv ch))
+                (let loop ()
+                  (cond ((vector-ref box 0) #t)
+                        ((async-chan-closed? ch) #f)
+                        (else (condition-wait (async-chan-cv ch) (async-chan-mu ch)) (loop))))))))))))
+
+;; remove + return the head value, waking a parked rendezvous putter.
+(define (ac-take-head! ch)
+  (let* ((entry (ac-qpop! ch)) (v (car entry)) (box (cdr entry)))
+    (when box (vector-set! box 0 #t))
+    (condition-broadcast (async-chan-cv ch))
+    v))
+
+;; <! / <!! — take, blocking. Drains buffered values, then nil once closed + empty.
+(define (jolt-async-take ch)
+  (with-mutex (async-chan-mu ch)
+    (let loop ()
+      (cond ((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))))))
+
+;; non-blocking take for alts!: 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)
+    (cond ((not (ac-qempty? ch)) (ac-take-head! ch))
+          ((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 (like the Janet
+;; impl). 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))))
+    (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)))))))))
+
+;; (timeout ms) — a channel that closes after ms milliseconds.
+(define (jolt-async-timeout ms)
+  (let ((w (ac-make 0 'unbuffered #f)))
+    (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)
+
+;; (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
+;; are conveyed (Chez inherits the thread-parameter at fork; we install explicitly).
+(define (async-go-spawn thunk)
+  (let ((w (ac-make 1 'fixed #f)) (snap (dyn-binding-stack)))
+    (fork-thread
+     (lambda ()
+       (dyn-binding-stack snap)
+       (let ((r (guard (e (#t (cons #f e))) (cons #t (jolt-invoke thunk)))))
+         (when (and (car r) (not (jolt-nil? (cdr r)))) (jolt-async-give w (cdr r)))
+         (jolt-async-close! w))))
+    w))
+
+;; --- macros (expander fns over the reader forms) ----------------------------
+(define cca-go-spawn-sym (jolt-symbol "clojure.core.async" "go-spawn"))
+(define cca-go-sym (jolt-symbol "clojure.core.async" "go"))
+(define cca-fn*-sym (jolt-symbol #f "fn*"))
+(define cca-loop-sym (jolt-symbol #f "loop"))
+
+;; (go body...) -> (clojure.core.async/go-spawn (fn* [] body...))
+(define (cca-go-macro . body)
+  (jolt-list cca-go-spawn-sym (apply jolt-list cca-fn*-sym empty-pvec body)))
+;; (go-loop bindings body...) -> (go (loop bindings body...))
+(define (cca-go-loop-macro bindings . body)
+  (jolt-list cca-go-sym (apply jolt-list cca-loop-sym bindings body)))
+;; (thread body...) — a real OS thread (same shape as go here).
+(define (cca-thread-macro . body)
+  (jolt-list cca-go-spawn-sym (apply jolt-list cca-fn*-sym empty-pvec body)))
+
+;; --- install clojure.core.async ---------------------------------------------
+(define (cca-def! name v) (def-var! "clojure.core.async" name v))
+(cca-def! "chan" jolt-async-chan)
+(cca-def! "chan?" async-chan?)
+(cca-def! "buffer" jolt-async-buffer)
+(cca-def! "dropping-buffer" jolt-async-dropping-buffer)
+(cca-def! "sliding-buffer" jolt-async-sliding-buffer)
+(cca-def! "close!" jolt-async-close!)
+(cca-def! "<!" jolt-async-take)   (cca-def! "<!!" jolt-async-take)
+(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! "go-spawn" async-go-spawn)
+(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/rt.ss b/host/chez/rt.ss
index fc2a2ee..7e179eb 100644
--- a/host/chez/rt.ss
+++ b/host/chez/rt.ss
@@ -331,3 +331,8 @@
 ;; thread-local binding stack (dyn-binding.ss) into workers. pmap/pcalls/pvalues
 ;; (overlay, over `future`) light up once future-call exists here.
 (load "host/chez/concurrency.ss")
+
+;; clojure.core.async (jolt-byjr): real-thread blocking channels + go/go-loop/
+;; thread macros, def-var!'d into clojure.core.async. After concurrency.ss (reuses
+;; ms->duration) and the collection/seq layer.
+(load "host/chez/async.ss")
diff --git a/test/chez/cli-test.janet b/test/chez/cli-test.janet
index ea5d2a7..cd1fdb0 100644
--- a/test/chez/cli-test.janet
+++ b/test/chez/cli-test.janet
@@ -68,7 +68,15 @@
    ["(let [a (atom 0)] (dorun (pmap (fn [_] (swap! a inc)) [1 2 3 4])) @a)" "4"]
    # promise blocks until delivered (JVM), unlike the Janet atom-shim.
    ["(let [p (promise)] (deliver p 7) @p)" "7"]
-   ["(let [p (promise)] (future (deliver p :hi)) @p)" ":hi"]])
+   ["(let [p (promise)] (future (deliver p :hi)) @p)" ":hi"]
+   # jolt-byjr: clojure.core.async on real-thread blocking channels.
+   ["(require (quote [clojure.core.async :refer [chan go <! >! <!!]])) (def c (chan)) (go (>! c (+ 40 2))) (<!! c)" "42"]
+   ["(require (quote [clojure.core.async :refer [chan go go-loop <! >! <!! close!]])) (def c (chan 5)) (go (>! c 1) (>! c 2) (>! c 3) (close! c)) (<!! (go-loop [o []] (let [v (<! c)] (if (nil? v) o (recur (conj o v))))))" "[1 2 3]"]
+   ["(require (quote [clojure.core.async :refer [chan go <! >! <!!]])) (def x (chan)) (def y (chan)) (go (>! x 10)) (go (>! y 32)) (<!! (go (+ (<! x) (<! y))))" "42"]
+   ["(require (quote [clojure.core.async :refer [chan go <! >! <!! alts!]])) (def x (chan)) (def y (chan)) (go (>! y :v)) (<!! (go (let [[v ch] (alts! [x y])] (and (= v :v) (= ch y)))))" "true"]
+   ["(require (quote [clojure.core.async :refer [chan go go-loop <! >! <!! close!]])) (def c (chan 10 (map inc))) (go (>! c 1) (>! c 2) (>! c 3) (close! c)) (<!! (go-loop [o []] (let [v (<! c)] (if (nil? v) o (recur (conj o v))))))" "[2 3 4]"]
+   ["(require (quote [clojure.core.async :refer [timeout <!!]])) (<!! (timeout 10)) :done" ":done"]
+   ["(require (quote [clojure.core.async :refer [chan go <! >! <!!]])) (def ^:dynamic *x* 0) (<!! (binding [*x* 7] (go (<! (clojure.core.async/timeout 5)) *x*)))" "7"]])
 
 (each [src want] cases
   (def [code out err] (joltc src))
diff --git a/test/chez/run-corpus-prelude.janet b/test/chez/run-corpus-prelude.janet
index c1e6e6f..c7740f8 100644
--- a/test/chez/run-corpus-prelude.janet
+++ b/test/chez/run-corpus-prelude.janet
@@ -303,8 +303,10 @@
 # into the prelude as an expander fn + mark-macro!, so syntax-quote VALUE forms
 # (`(...)/`[...]/`{...}/`#{...} via __sqcat/__sqvec/__sqmap/__sqset) run at runtime.
 # 2280->2295, 0 new divergences.
+# jolt-byjr (real-thread futures/pmap on Chez): future-call resolves, so the
+# future/deref/pmap cases run instead of crashing. 2534->2559, 0 new divergences.
 # Strided runs scale down.
-(def base-floor (scan-number (or (os/getenv "JOLT_CHEZ_PRELUDE_FLOOR") "2534")))
+(def base-floor (scan-number (or (os/getenv "JOLT_CHEZ_PRELUDE_FLOOR") "2559")))
 (def floor (if (os/getenv "JOLT_CORPUS_LIMIT") 0 base-floor))
 (when (or (> (length diverged) 0) (< pass floor))
   (printf "REGRESSION: pass %d < floor %d or %d new divergence(s)" pass floor (length diverged)))