From c719e5454342f832d939baffe1fda9f2213b6698 Mon Sep 17 00:00:00 2001 From: Yogthos Date: Sat, 20 Jun 2026 14:22:37 -0400 Subject: [PATCH] Chez concurrency pt.3: async agents (per-agent serialized dispatch) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the Janet synchronous agent shim (agent = atom, send applies inline) with JVM-style async agents: send/send-off enqueue an action and a single worker thread per agent applies them in order; deref reads the current (maybe not-yet-updated) state without blocking; await blocks until the queue drains. A validator rejection or a thrown action puts the agent in an error state (agent-error) and halts the queue; restart-agent clears it. send and send-off share one serialized worker (a superset of the JVM's fixed/cached pool split). Native versions re-asserted in post-prelude over the overlay; await/restart-agent are new. Corpus: the two "send/send-off applies" cases do (send a f) (deref a) with no await, so they now read state before the action runs — diverging like the JVM (the suite was literally "synchronous shim"). Allowlisted on both gates; floors -2 (zero-Janet 2569->2567, prelude 2559->2557). cli-test covers async agents via await (ordered 100-send dispatch, error capture) — 49/49. Janet gate + JVM cert green; 0 new divergences on either corpus. jolt-byjr --- host/chez/concurrency.ss | 84 ++++++++++++++++++++++++++- host/chez/post-prelude.ss | 10 ++++ test/chez/cli-test.janet | 9 ++- test/chez/run-corpus-prelude.janet | 11 +++- test/chez/run-corpus-zero-janet.janet | 13 ++++- 5 files changed, 120 insertions(+), 7 deletions(-) diff --git a/host/chez/concurrency.ss b/host/chez/concurrency.ss index 7386f91..a31de0e 100644 --- a/host/chez/concurrency.ss +++ b/host/chez/concurrency.ss @@ -143,9 +143,81 @@ (else (jolt-promise-delivered? p))))))) (if got (jolt-promise-value p) timeout-val))) +;; --- agents (async, per-agent serialized dispatch) -------------------------- +;; JVM semantics, not Janet's synchronous shim: send/send-off enqueue an action +;; and a single worker thread applies them to the state IN ORDER; deref reads the +;; (possibly not-yet-updated) state without blocking; await blocks until the queue +;; drains. An action error is captured (agent-error) and stops the queue. +(define-record-type jolt-agent + (fields (mutable state) (mutable err) (mutable validator) + (mutable queue) (mutable running?) mu cv) + (nongenerative jolt-agent-v1)) + +;; (agent state) / (agent state :validator f :error-mode m :meta x): only :validator +;; has runtime behaviour here; other opts are accepted/ignored. +(define (jolt-agent-new state . opts) + (let loop ((o opts) (validator jolt-nil)) + (cond + ((or (null? o) (null? (cdr o))) + (make-jolt-agent state jolt-nil validator '() #f (make-mutex) (make-condition))) + ((and (keyword-t? (car o)) (string=? (keyword-t-name (car o)) "validator")) + (loop (cddr o) (cadr o))) + (else (loop (cddr o) validator))))) + +;; Drain the queue, applying each action (f state arg*) outside the lock (an action +;; may send/deref the same agent). A validator rejection or a thrown action puts the +;; agent in an error state and halts the queue (JVM :fail mode). +(define (jolt-agent-worker a) + (let loop () + (let ((act (with-mutex (jolt-agent-mu a) + (if (or (not (jolt-nil? (jolt-agent-err a))) (null? (jolt-agent-queue a))) + (begin (jolt-agent-running?-set! a #f) + (condition-broadcast (jolt-agent-cv a)) #f) + (let ((x (car (jolt-agent-queue a)))) + (jolt-agent-queue-set! a (cdr (jolt-agent-queue a))) x))))) + (when act + (guard (e (#t (with-mutex (jolt-agent-mu a) + (jolt-agent-err-set! a e) + (condition-broadcast (jolt-agent-cv a))))) + (let ((nv (apply jolt-invoke (car act) (jolt-agent-state a) (cdr act)))) + (let ((vf (jolt-agent-validator a))) + (when (and (not (jolt-nil? vf)) (jolt-not (jolt-invoke vf nv))) + (error #f "Invalid reference state"))) + (jolt-agent-state-set! a nv))) + (loop))))) + +;; send / send-off: enqueue the action, start the worker if idle. (jolt treats them +;; identically — one serialized worker per agent — which is observably a superset of +;; the JVM's fixed/cached pool split.) +(define (jolt-agent-send a f . args) + (with-mutex (jolt-agent-mu a) + (jolt-agent-queue-set! a (append (jolt-agent-queue a) (list (cons f args)))) + (unless (jolt-agent-running? a) + (jolt-agent-running?-set! a #t) + (fork-thread (lambda () (jolt-agent-worker a))))) + a) + +;; (await & agents): block until each agent's queue has drained. +(define (jolt-agent-await . agents) + (for-each + (lambda (a) + (with-mutex (jolt-agent-mu a) + (let loop () + (when (or (jolt-agent-running? a) (pair? (jolt-agent-queue a))) + (condition-wait (jolt-agent-cv a) (jolt-agent-mu a)) (loop))))) + agents) + jolt-nil) + +(define (jolt-agent-error a) (jolt-agent-err a)) +(define (jolt-agent-restart a new-state . _opts) + (jolt-agent-err-set! a jolt-nil) + (jolt-agent-state-set! a new-state) + a) + ;; --- deref extension -------------------------------------------------------- -;; Chain the fully-built jolt-deref (atoms/vars/volatiles/reduced) with futures + -;; promises, and accept the timed (deref ref ms val) arity for both. +;; Chain the fully-built jolt-deref (atoms/vars/volatiles/reduced) with futures, +;; promises, and agents, and accept the timed (deref ref ms val) arity for the +;; blocking ref types. (define %pre-conc-deref jolt-deref) (set! jolt-deref (lambda (x . opts) @@ -156,6 +228,7 @@ ((jolt-promise? x) (if (null? opts) (jolt-promise-deref x) (jolt-promise-deref-timed x (car opts) (cadr opts)))) + ((jolt-agent? x) (jolt-agent-state x)) (else (apply %pre-conc-deref x opts))))) ;; realized? for a Chez future/promise (the overlay reads Janet map keys). Wrap the @@ -173,4 +246,11 @@ (def-var! "clojure.core" "future-cancelled?" jolt-native-future-cancelled?) (def-var! "clojure.core" "promise" jolt-promise-new) (def-var! "clojure.core" "deliver" jolt-deliver) +(def-var! "clojure.core" "agent" jolt-agent-new) +(def-var! "clojure.core" "agent?" jolt-agent?) +(def-var! "clojure.core" "send" jolt-agent-send) +(def-var! "clojure.core" "send-off" jolt-agent-send) +(def-var! "clojure.core" "await" jolt-agent-await) +(def-var! "clojure.core" "agent-error" jolt-agent-error) +(def-var! "clojure.core" "restart-agent" jolt-agent-restart) (def-var! "clojure.core" "deref" jolt-deref) diff --git a/host/chez/post-prelude.ss b/host/chez/post-prelude.ss index af69537..fae4a8a 100644 --- a/host/chez/post-prelude.ss +++ b/host/chez/post-prelude.ss @@ -48,6 +48,16 @@ (def-var! "clojure.core" "future?" jolt-future?) (def-var! "clojure.core" "promise" jolt-promise-new) (def-var! "clojure.core" "deliver" jolt-deliver) +;; agents: the overlay (50-io) is a synchronous shim (agent = atom, send applies +;; immediately). Re-assert the native async agents (per-agent serialized worker), +;; matching the JVM. await/restart-agent are new (the overlay has neither). +(def-var! "clojure.core" "agent" jolt-agent-new) +(def-var! "clojure.core" "agent?" jolt-agent?) +(def-var! "clojure.core" "send" jolt-agent-send) +(def-var! "clojure.core" "send-off" jolt-agent-send) +(def-var! "clojure.core" "await" jolt-agent-await) +(def-var! "clojure.core" "agent-error" jolt-agent-error) +(def-var! "clojure.core" "restart-agent" jolt-agent-restart) (def-var! "clojure.core" "deref" jolt-deref) (let ((overlay-realized? (var-deref "clojure.core" "realized?"))) (def-var! "clojure.core" "realized?" diff --git a/test/chez/cli-test.janet b/test/chez/cli-test.janet index cd1fdb0..cd2f5e7 100644 --- a/test/chez/cli-test.janet +++ b/test/chez/cli-test.janet @@ -76,7 +76,14 @@ ["(require (quote [clojure.core.async :refer [chan go ! ! y :v)) (! ! c 1) (>! c 2) (>! c 3) (close! c)) (! ! false), like the JVM; the Janet :expected relies on its # cooperative scheduler. Flaky/divergent. "cancel an in-flight future returns true" true - "future-cancelled? after cancel" true}) + "future-cancelled? after cancel" true + # agents are async on Chez (per-agent serialized dispatch) = JVM, not Janet's + # synchronous shim: (send a f) (deref a) reads state before the action runs, so + # these sync-shim cases diverge like the JVM. Per jvm-parity-north-star. + "send applies" true + "send-off applies" true}) # Cases that BLOCK forever on a shared-heap / JVM host (profile.edn :bucket # :timeout) — skip, like :throws, so a hung per-case process can't stall the gate. @@ -305,8 +310,10 @@ # 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. +# Then -2 when agents went async (the two "send/send-off applies" sync-shim cases +# match the JVM's async raciness and are allowlisted) -> 2557. # Strided runs scale down. -(def base-floor (scan-number (or (os/getenv "JOLT_CHEZ_PRELUDE_FLOOR") "2559"))) +(def base-floor (scan-number (or (os/getenv "JOLT_CHEZ_PRELUDE_FLOOR") "2557"))) (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))) diff --git a/test/chez/run-corpus-zero-janet.janet b/test/chez/run-corpus-zero-janet.janet index 7b364ae..446017b 100644 --- a/test/chez/run-corpus-zero-janet.janet +++ b/test/chez/run-corpus-zero-janet.janet @@ -89,7 +89,13 @@ # future usually completes before the cancel (cancel -> false), like the JVM. # The Janet :expected "true" relies on cooperative scheduling. Flaky/divergent. "cancel an in-flight future returns true" true - "future-cancelled? after cancel" true}) + "future-cancelled? after cancel" true + # agents are async on Chez (per-agent serialized dispatch) = JVM, not Janet's + # synchronous shim: (send a f) (deref a) reads the state BEFORE the action runs + # (use await to observe the result), so these sync-shim cases now diverge like + # the JVM. Deliberate per jvm-parity-north-star. + "send applies" true + "send-off applies" true}) # Cases that BLOCK forever on a shared-heap / JVM host (profile.edn :bucket # :timeout) — skip them, like :throws: a single hung case would stall the whole @@ -183,7 +189,10 @@ # Regression floor: raise as the Chez-hosted compiler closes gaps. The gate fails # on any NEW divergence or if pass drops below the floor. Strided runs scale to 0. -(def base-floor (scan-number (or (os/getenv "JOLT_CHEZ_ZJ_FLOOR") "2569"))) +# 2569 after futures/pmap (jolt-byjr pt.1); -2 when agents went async (the two +# "send/send-off applies" sync-shim cases now match the JVM's async raciness and +# are allowlisted) -> 2567. +(def base-floor (scan-number (or (os/getenv "JOLT_CHEZ_ZJ_FLOOR") "2567"))) (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)))