agent: Clojure-style async state cells via (std agent)
ober
4b7a9ea2c2dc2475e21b9e75e162b7192def71f5
--- a/docs/clojure-remaining.md +++ b/docs/clojure-remaining.md @@ -1565,6 +1565,26 @@ limitation applies). ### 4.8 Agents +**[landed]** Phase E.5 shipped `(std agent)`. The module exports +`agent`, `agent?`, `send`, `send-off`, `agent-value`, `agent-error`, +`clear-agent-errors`, `restart-agent`, `await`, and `shutdown-agent!`. +Each agent wraps a single worker thread that reads actions off a +buffered CSP channel and applies them to the agent's value in order. +Because there's one worker per agent, actions are guaranteed +serialized regardless of how many threads call `send` concurrently. + +Error semantics match Clojure's `:fail` mode: if an action throws, +the exception lands in the agent's error slot and subsequent `send` +calls raise until `restart-agent` is called to clear the error and +reset the value. `send-off` is currently an alias for `send` — Jerboa +doesn't distinguish CPU and I/O thread pools. + +`(std agent)` is *not* in the prelude because `send` is a common name +that would conflict with `(std actor)`'s `send`. Users who want both +modules should import one of them with a rename. A new module header +note cross-references `(std actor)` for users who want supervised +hierarchies instead of a single state cell. + **The gap.** Clojure's agent is an asynchronous state cell with an action queue: you `(send agent fn args)` and the function is applied to the current agent value on a background thread pool. Errors put the @@ -1946,7 +1966,7 @@ in this doc. **[deferred]** items are non-goals. | `defprotocol`/`extend-type` | [landed] `(std protocol)` | §4.6 landed | | Atom watches | [current] `(std misc atom)` | §4.7 landed | | Volatiles | [current] `(std misc atom)` | §4.7 landed | -| Agents | [gap] | §4.8 | +| Agents | [landed] `(std agent)` | §4.8 landed | | Record-as-map | [gap] | §4.10 | | `#!clojure-reader` literal switch | [gap] (risky) | §4.9 | | `{}`/`#{}`/`[]`/`:kw` default reader | [deferred] | §4.9 | new file mode 100644 --- /dev/null +++ b/lib/std/agent.sls @@ -0,0 +1,193 @@ +#!chezscheme +;;; (std agent) — Clojure-style agents. +;;; +;;; An agent is an asynchronous state cell with a serialized action +;;; queue. You create one with an initial value and then dispatch +;;; actions against it: +;;; +;;; (def a (agent 0)) +;;; (send a + 1) ;; queues (+ 0 1) -> 1; returns a +;;; (send a * 3) ;; queues (* 1 3) -> 3; returns a +;;; (await a) ;; blocks until the queue drains +;;; (agent-value a) ;; => 3 +;;; +;;; Actions run one at a time on a dedicated worker thread per agent, +;;; so there's never contention on the value — readers with +;;; `agent-value` / `deref` see the most recent successfully-applied +;;; state, and writers via `send` queue without blocking on each +;;; other. +;;; +;;; Error handling +;;; -------------- +;;; If an action throws, the exception is captured and placed in the +;;; agent's error slot. Subsequent `send` calls raise an error until +;;; `restart-agent` is called to clear the error and optionally reset +;;; the value. This matches Clojure's default `:fail` error mode. +;;; +;;; Use `agent-error` to check for an error without raising: +;;; +;;; (send a (lambda (v) (error 'bad "boom"))) +;;; (await a) +;;; (agent-error a) ;; => a condition +;;; (send a + 1) ;; raises: agent has error, call restart-agent +;;; (restart-agent a 0) ;; clears error and resets value to 0 +;;; (send a + 1) ;; works again +;;; +;;; Distinction from `(std actor)` +;;; ------------------------------ +;;; `(std actor)` provides supervised message-passing hierarchies with +;;; behaviours, links, and monitors — full actor-model. `(std agent)` +;;; is a much smaller construct: a single state cell with a serialized +;;; action queue. Agents are good when you want one thing to hold +;;; state and coalesce updates without locking. Actors are good when +;;; you want a tree of supervised processes. They can coexist. +;;; +;;; Shutdown +;;; -------- +;;; `shutdown-agent!` closes the action queue and lets the worker thread +;;; finish naturally when the queue drains. There is no forcible kill. +;;; After shutdown, further sends raise an error. + +(library (std agent) + (export + agent agent? + send send-off + agent-value agent-error + clear-agent-errors restart-agent + await shutdown-agent!) + + (import (chezscheme) + (std csp)) + + ;; --- Agent record ------------------------------------------- + ;; + ;; `val` and `err` are mutable. Only the worker thread writes + ;; them; readers (deref / agent-value / agent-error) observe + ;; the most recent write. There's no lock on reads — we rely on + ;; Chez's atomic-word slot updates. + ;; + ;; The worker thread's handle is not stored in the record — the + ;; thread exits naturally when its action channel is closed, and + ;; Chez reclaims thread objects on GC. Nothing outside the library + ;; needs to poke the thread object directly. + + (define-record-type %agent + (fields (mutable val) + (mutable err) + (immutable action-ch)) + (sealed #t)) + + (define (agent? x) (%agent? x)) + + (define (agent-value a) + (unless (%agent? a) (error 'agent-value "not an agent" a)) + (%agent-val a)) + + (define (agent-error a) + (unless (%agent? a) (error 'agent-error "not an agent" a)) + (%agent-err a)) + + ;; --- Constructor -------------------------------------------- + + ;; (agent initial-value) — default queue capacity 1024 + ;; (agent initial-value n) — custom queue capacity + (define agent + (case-lambda + [(initial) (agent initial 1024)] + [(initial buf-size) + (let* ([ch (make-channel buf-size)] + [a (make-%agent initial #f ch)]) + (fork-thread (%make-worker-loop a ch)) + a)])) + + (define (%make-worker-loop a ch) + (lambda () + (let loop () + (let ([action (chan-get! ch)]) + (cond + [(eof-object? action) #f] ;; channel closed, exit + [else + ;; Skip actions if the agent is already in an error + ;; state — queued work drains but doesn't execute + ;; until restart-agent. + (unless (%agent-err a) + (guard (exn [else (%agent-err-set! a exn)]) + (let ([new-val (apply (car action) + (%agent-val a) + (cdr action))]) + (%agent-val-set! a new-val)))) + (loop)]))))) + + ;; --- Dispatch ----------------------------------------------- + + (define (send a fn . args) + (unless (%agent? a) (error 'send "not an agent" a)) + (unless (procedure? fn) (error 'send "action is not a procedure" fn)) + (when (%agent-err a) + (error 'send + "agent has error; call restart-agent to clear" + (%agent-err a))) + (when (chan-closed? (%agent-action-ch a)) + (error 'send "agent has been shut down" a)) + (chan-put! (%agent-action-ch a) (cons fn args)) + a) + + ;; In Clojure, send-off dispatches on a dedicated unbounded I/O + ;; thread pool so blocking operations don't starve the cpu pool. + ;; Jerboa's agent runs on a single dedicated worker, so send and + ;; send-off are operationally identical. The alias is kept so + ;; Clojure code doesn't need rewriting. + (define send-off send) + + ;; --- Error handling ----------------------------------------- + + (define (clear-agent-errors a) + (unless (%agent? a) (error 'clear-agent-errors "not an agent" a)) + (%agent-err-set! a #f) + a) + + ;; (restart-agent a new-value) — clears error and resets value. + ;; Clojure's restart-agent takes the new state as its required + ;; second argument. + (define (restart-agent a new-value) + (unless (%agent? a) (error 'restart-agent "not an agent" a)) + (%agent-err-set! a #f) + (%agent-val-set! a new-value) + a) + + ;; --- Synchronization ---------------------------------------- + + ;; (await a) — block until all currently-queued actions have been + ;; processed. Returns the agent. + ;; + ;; Implementation: send a sentinel action that signals a private + ;; channel. Because actions are processed in order, receiving on + ;; the sentinel channel guarantees all prior actions have run. + ;; + ;; Note: if the agent is in an error state the sentinel will be + ;; skipped along with all other actions, so await would hang + ;; forever. We detect this and bail out with an error. + (define (await a) + (unless (%agent? a) (error 'await "not an agent" a)) + (when (%agent-err a) + (error 'await "agent has error; call restart-agent to clear" + (%agent-err a))) + (when (chan-closed? (%agent-action-ch a)) + (error 'await "agent has been shut down" a)) + (let ([done (make-channel 1)]) + (chan-put! (%agent-action-ch a) + (cons (lambda (v) + (chan-put! done 'done) + v) + '())) + (chan-get! done) + a)) + + ;; (shutdown-agent! a) — close the action queue. The worker + ;; thread exits when it drains the queue. Subsequent sends raise. + (define (shutdown-agent! a) + (unless (%agent? a) (error 'shutdown-agent! "not an agent" a)) + (chan-close! (%agent-action-ch a)) + a) + +) ;; end library new file mode 100644 --- /dev/null +++ b/tests/test-agent.ss @@ -0,0 +1,241 @@ +#!chezscheme +;;; Tests for (std agent) — Clojure-style agents. + +(import (except (jerboa prelude) make-time) + (only (chezscheme) make-time sleep fork-thread) + (std agent)) + +(define pass 0) +(define fail 0) + +(define-syntax test + (syntax-rules () + [(_ name expr expected) + (guard (exn [#t (set! fail (+ fail 1)) + (printf "FAIL ~a: ~a~%" name + (if (message-condition? exn) (condition-message exn) exn))]) + (let ([got expr]) + (if (equal? got expected) + (begin (set! pass (+ pass 1)) (printf " ok ~a~%" name)) + (begin (set! fail (+ fail 1)) + (printf "FAIL ~a: got ~s expected ~s~%" name got expected)))))])) + +(define (brief-sleep) (sleep (make-time 'time-duration 50000000 0))) + +(printf "--- std/agent ---~%~%") + +;;; ---- Basic send + await ---------------------------------------- + +(test "agent? recognizes an agent" + (let ([a (agent 0)]) + (let ([r (agent? a)]) (shutdown-agent! a) r)) + #t) + +(test "agent? false for non-agents" + (list (agent? 42) (agent? "str") (agent? '())) + '(#f #f #f)) + +(test "initial value" + (let ([a (agent 42)]) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + 42) + +(test "single send + await" + (let ([a (agent 0)]) + (send a + 5) + (await a) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + 5) + +(test "send returns the agent" + (let ([a (agent 0)]) + (let ([r (eq? a (send a + 1))]) + (await a) + (shutdown-agent! a) + r)) + #t) + +(test "multiple sends are applied in order" + (let ([a (agent 0)]) + (send a + 1) ;; 1 + (send a * 3) ;; 3 + (send a - 1) ;; 2 + (await a) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + 2) + +(test "send-off is an alias for send" + (let ([a (agent 0)]) + (send-off a + 10) + (send-off a * 2) + (await a) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + 20) + +(test "sends with multiple args" + (let ([a (agent 0)]) + (send a + 1 2 3) + (send a - 0 0 1) ;; = v - 0 - 0 - 1 + (await a) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + 5) + +;;; ---- agent-value / agent-error ---------------------------------- + +(test "agent-error is #f when no error" + (let ([a (agent 0)]) + (send a + 1) + (await a) + (let ([e (agent-error a)]) (shutdown-agent! a) e)) + #f) + +(test "agent-error captures thrown exception" + (let ([a (agent 10)]) + (send a (lambda (v) (error 'boom "kaboom"))) + (brief-sleep) + (let ([has-err (and (agent-error a) #t)]) + (shutdown-agent! a) + has-err)) + #t) + +(test "value is preserved across failing action" + (let ([a (agent 10)]) + (send a (lambda (v) (error 'boom "kaboom"))) + (brief-sleep) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + 10) + +(test "send after error raises" + (let ([a (agent 10)]) + (send a (lambda (v) (error 'boom "kaboom"))) + (brief-sleep) + (let ([r (guard (_ [#t 'raised]) (send a + 1))]) + (shutdown-agent! a) + r)) + 'raised) + +(test "await after error raises" + (let ([a (agent 10)]) + (send a (lambda (v) (error 'boom "kaboom"))) + (brief-sleep) + (let ([r (guard (_ [#t 'raised]) (await a))]) + (shutdown-agent! a) + r)) + 'raised) + +;;; ---- clear-agent-errors / restart-agent ------------------------- + +(test "clear-agent-errors clears the error" + (let ([a (agent 10)]) + (send a (lambda (v) (error 'boom "kaboom"))) + (brief-sleep) + (clear-agent-errors a) + (let ([e (agent-error a)]) (shutdown-agent! a) e)) + #f) + +(test "clear-agent-errors preserves value" + (let ([a (agent 10)]) + (send a (lambda (v) (error 'boom "kaboom"))) + (brief-sleep) + (clear-agent-errors a) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + 10) + +(test "restart-agent resets value and clears error" + (let ([a (agent 10)]) + (send a (lambda (v) (error 'boom "kaboom"))) + (brief-sleep) + (restart-agent a 100) + (let ([e (agent-error a)] + [v (agent-value a)]) + (shutdown-agent! a) + (list e v))) + '(#f 100)) + +(test "send works after restart-agent" + (let ([a (agent 10)]) + (send a (lambda (v) (error 'boom "kaboom"))) + (brief-sleep) + (restart-agent a 0) + (send a + 5) + (await a) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + 5) + +;;; ---- Shutdown -------------------------------------------------- + +(test "send after shutdown raises" + (let ([a (agent 0)]) + (shutdown-agent! a) + (guard (_ [#t 'raised]) (send a + 1))) + 'raised) + +(test "agent-value readable after shutdown" + (let ([a (agent 42)]) + (shutdown-agent! a) + (agent-value a)) + 42) + +;;; ---- Concurrent sends ------------------------------------------- +;;; +;;; Clojure guarantees agent actions are serialized — the value +;;; seen by action N is the result of action N-1 regardless of +;;; which thread called send. This confirms per-agent serialization. + +(test "concurrent sends serialize through the agent" + (let ([a (agent 0)]) + (for ([i (in-range 100)]) + (send a + 1)) + (await a) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + 100) + +(test "concurrent sends from multiple threads serialize" + (let ([a (agent 0)] + [threads '()]) + (for ([t (in-range 10)]) + (set! threads + (cons (fork-thread + (lambda () + (for ([i (in-range 10)]) + (send a + 1)))) + threads))) + ;; Give background threads time to finish sending + (brief-sleep) + (await a) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + 100) + +;;; ---- Agent holds immutable state too --------------------------- + +(test "agent holds a list" + (let ([a (agent '())]) + ;; cons takes (item list), but send passes (current-value args ...), + ;; so wrap in a lambda that flips the arg order. + (send a (lambda (lst x) (cons x lst)) 1) + (send a (lambda (lst x) (cons x lst)) 2) + (send a (lambda (lst x) (cons x lst)) 3) + (await a) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + '(3 2 1)) + +(test "agent holds a hash-map-like alist" + (let ([a (agent '())]) + (send a (lambda (m) (cons (cons 'a 1) m))) + (send a (lambda (m) (cons (cons 'b 2) m))) + (await a) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + '((b . 2) (a . 1))) + +;;; ---- Custom buffer size ----------------------------------------- + +(test "agent with explicit buffer size works" + (let ([a (agent 0 8)]) + (send a + 5) + (await a) + (let ([v (agent-value a)]) (shutdown-agent! a) v)) + 5) + +;;; ---- Summary --------------------------------------------------- +(printf "~%std/agent: ~a passed, ~a failed~%" pass fail) +(when (> fail 0) (exit 1))