atom: add-watch! / remove-watch! + volatile! family
ober
24218d8343657d30fe00936e560a4a20ec469c59
--- a/docs/clojure-remaining.md +++ b/docs/clojure-remaining.md @@ -1335,6 +1335,20 @@ defprotocol extend-type extend-protocol satisfies? ### 4.7 Atom watches + volatiles +**[landed]** Added to `(std misc atom)` and re-exported from the +prelude and `(std clojure)`. `add-watch!` registers a callback under +a key and fires it on every successful `reset!`/`swap!`/`atom-update!` +and on successful `compare-and-set!` (a failed CAS does NOT fire +watches). Callbacks run OUTSIDE the atom's mutex so they can call +back into the atom without deadlock; raised exceptions are swallowed +so a broken watch can't corrupt the atom. Same-key re-add replaces +the previous callback (matches Clojure), and `remove-watch!` is +idempotent. Both return the atom so calls chain. Volatiles are a +separate lightweight record (`volatile!` / `volatile?` / `vreset!` / +`vswap!` / `vderef`) with no mutex, no watches, and no CAS — meant +for single-threaded transient accumulators inside transducers. +Covered by `tests/test-atom.ss` (25 tests). + **Watches — the gap.** Clojure atoms support `add-watch` and `remove-watch`: register a callback that fires after every successful swap with the key, the atom, the old value, and the new value. Used for @@ -1806,8 +1820,8 @@ in this doc. **[deferred]** items are non-goals. | Metadata (`with-meta`/`meta`) | [gap] | §4.4 | | `defmulti`/`defmethod` value-dispatch | [gap] | §4.5 | | `defprotocol`/`extend-type` | [gap] | §4.6 | -| Atom watches | [gap] | §4.7 | -| Volatiles | [gap] | §4.7 | +| Atom watches | [current] `(std misc atom)` | §4.7 landed | +| Volatiles | [current] `(std misc atom)` | §4.7 landed | | Agents | [gap] | §4.8 | | Record-as-map | [gap] | §4.10 | | `#!clojure-reader` literal switch | [gap] (risky) | §4.9 | --- a/lib/jerboa/prelude.sls +++ b/lib/jerboa/prelude.sls @@ -214,6 +214,10 @@ atom atom? atom-deref atom-reset! atom-swap! atom-update! ;; Clojure-style aliases (familiar to clojure users) deref reset! swap! compare-and-set! + ;; Watches (fires on every successful swap/reset/update/CAS) + add-watch! remove-watch! + ;; Volatiles (single-threaded transient cells for transducers) + volatile! volatile? vreset! vswap! vderef ;; ---- std/misc/shared (atomic cell with CAS) ---- make-shared shared? shared-ref shared-set! --- a/lib/std/clojure.sls +++ b/lib/std/clojure.sls @@ -78,6 +78,8 @@ ;; ---- Re-exports from (std misc atom) ---- atom atom? deref reset! swap! compare-and-set! + add-watch! remove-watch! + volatile! volatile? vreset! vswap! vderef ;; ---- Re-exports from (std misc nested) ---- get-in assoc-in update-in --- a/lib/std/misc/atom.sls +++ b/lib/std/misc/atom.sls @@ -17,6 +17,32 @@ ;;; (swap! counter + 1) ;; (apply + @counter 1) → 43, variadic ;;; (swap! counter inc) ;; → 44 ;;; (compare-and-set! counter 44 100) ;; CAS → #t/#f +;;; +;;; Watches (Clojure parity, §4.7) +;;; ------------------------------ +;;; `(add-watch! atom key fn)` registers `fn` to run after every +;;; successful reset!/swap!/update!/CAS, with the signature +;;; `(fn key atom old-val new-val)`. Same-key re-add replaces the +;;; previous callback. Callbacks run OUTSIDE the atom's lock, so +;;; they can call back into the atom without deadlock; raised +;;; exceptions are swallowed (a broken watch can't corrupt the atom). +;;; +;;; `(remove-watch! atom key)` removes the watch for `key`. Both +;;; calls return the atom, so they chain. +;;; +;;; Volatiles (Clojure parity, §4.7) +;;; -------------------------------- +;;; `volatile!` is a lightweight, SINGLE-THREADED mutable cell for +;;; transient accumulators inside transducers (the `partition-by` +;;; pattern). It has no mutex, no watches, and no CAS. Use when you +;;; know the cell is captured in a closure that cannot be touched +;;; from multiple threads. If you need thread-safety, use `atom`. +;;; +;;; (define v (volatile! 0)) +;;; (vderef v) ;; → 0 +;;; (vreset! v 10) ;; → 10 +;;; (vswap! v + 5) ;; → 15 +;;; (library (std misc atom) (export @@ -25,18 +51,23 @@ ;; ---- Clojure-style aliases ---- ;; Note: no `atom?` alias — Clojure doesn't expose one either. ;; Use atom? (the existing predicate) or shared? from (std misc shared). - deref reset! swap! compare-and-set!) + deref reset! swap! compare-and-set! + ;; ---- Watches (§4.7) ---- + add-watch! remove-watch! + ;; ---- Volatiles (§4.7) ---- + volatile! volatile? vreset! vswap! vderef) (import (except (chezscheme) atom?)) (define-record-type atom-rec (fields (mutable val) - (immutable mtx)) + (immutable mtx) + (mutable watches)) ;; alist of (key . (lambda (k atom old new) ...)) (sealed #t)) (define (atom initial-value) - (make-atom-rec initial-value (make-mutex))) + (make-atom-rec initial-value (make-mutex) '())) (define (atom? x) (atom-rec? x)) @@ -44,24 +75,46 @@ (with-mutex (atom-rec-mtx a) (atom-rec-val a))) + ;; Fire watches OUTSIDE the atom's mutex so a watch can call back + ;; into the atom without deadlock. Exceptions are swallowed so a + ;; broken watch cannot corrupt the atom's state. + (define (%fire-watches! a old new watches) + (for-each + (lambda (w) + (guard (_ [else (void)]) + ((cdr w) (car w) a old new))) + watches)) + (define (atom-reset! a new-val) - (with-mutex (atom-rec-mtx a) - (atom-rec-val-set! a new-val) + (let-values ([(old watches) + (with-mutex (atom-rec-mtx a) + (let ([o (atom-rec-val a)]) + (atom-rec-val-set! a new-val) + (values o (atom-rec-watches a))))]) + (%fire-watches! a old new-val watches) new-val)) (define (atom-swap! a fn) ;; Atomically apply fn to current value, store and return result. - (with-mutex (atom-rec-mtx a) - (let ([new-val (fn (atom-rec-val a))]) - (atom-rec-val-set! a new-val) - new-val))) + (let-values ([(old new watches) + (with-mutex (atom-rec-mtx a) + (let* ([o (atom-rec-val a)] + [n (fn o)]) + (atom-rec-val-set! a n) + (values o n (atom-rec-watches a))))]) + (%fire-watches! a old new watches) + new)) (define (atom-update! a fn . args) ;; Atomically apply (fn current-val args ...), store and return result. - (with-mutex (atom-rec-mtx a) - (let ([new-val (apply fn (atom-rec-val a) args)]) - (atom-rec-val-set! a new-val) - new-val))) + (let-values ([(old new watches) + (with-mutex (atom-rec-mtx a) + (let* ([o (atom-rec-val a)] + [n (apply fn o args)]) + (atom-rec-val-set! a n) + (values o n (atom-rec-watches a))))]) + (%fire-watches! a old new watches) + new)) ;; ========================================================================= ;; Clojure-style aliases @@ -83,12 +136,63 @@ (define (compare-and-set! a expected new-val) ;; Atomically: if current value is equal? to expected, replace - ;; with new-val and return #t. Otherwise return #f. + ;; with new-val and return #t. Otherwise return #f. Fires watches + ;; ONLY on successful swap. + (let-values ([(swapped? old watches) + (with-mutex (atom-rec-mtx a) + (cond + [(equal? (atom-rec-val a) expected) + (let ([o (atom-rec-val a)]) + (atom-rec-val-set! a new-val) + (values #t o (atom-rec-watches a)))] + [else (values #f #f '())]))]) + (when swapped? + (%fire-watches! a old new-val watches)) + swapped?)) + + ;; ========================================================================= + ;; Watches (§4.7) + ;; ========================================================================= + + ;; Register `fn` under `key`. Same-key re-add replaces the previous + ;; callback (matches Clojure). Returns the atom so calls chain. + (define (add-watch! a key fn) (with-mutex (atom-rec-mtx a) - (if (equal? (atom-rec-val a) expected) - (begin - (atom-rec-val-set! a new-val) - #t) - #f))) + (atom-rec-watches-set! a + (cons (cons key fn) + (remp (lambda (w) (equal? (car w) key)) + (atom-rec-watches a))))) + a) + + ;; Remove the watch registered under `key`. Idempotent: removing a + ;; key that isn't present is a no-op. Returns the atom. + (define (remove-watch! a key) + (with-mutex (atom-rec-mtx a) + (atom-rec-watches-set! a + (remp (lambda (w) (equal? (car w) key)) (atom-rec-watches a)))) + a) + + ;; ========================================================================= + ;; Volatiles (§4.7) — single-threaded transient cells for transducers + ;; ========================================================================= + + (define-record-type volatile-rec + (fields (mutable val)) + (sealed #t)) + + (define (volatile! v) (make-volatile-rec v)) + + (define (volatile? x) (volatile-rec? x)) + + (define (vderef vol) (volatile-rec-val vol)) + + (define (vreset! vol v) + (volatile-rec-val-set! vol v) + v) + + (define (vswap! vol fn . args) + (let ([new (apply fn (volatile-rec-val vol) args)]) + (volatile-rec-val-set! vol new) + new)) ) ;; end library new file mode 100644 --- /dev/null +++ b/tests/test-atom.ss @@ -0,0 +1,208 @@ +#!chezscheme +;;; Tests for (std misc atom) — atom, watches, volatiles. + +(import (jerboa prelude)) + +(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)))))])) + +(printf "--- atom / watches / volatiles ---~%~%") + +;;; ---- Core atom ops (regression safety) ---- + +(test "atom deref initial" + (deref (atom 42)) + 42) + +(test "reset! returns new value" + (let ([a (atom 0)]) + (list (reset! a 10) (deref a))) + '(10 10)) + +(test "swap! variadic" + (let ([a (atom 10)]) + (swap! a + 3 4) + (deref a)) + 17) + +(test "compare-and-set! success" + (let ([a (atom 5)]) + (list (compare-and-set! a 5 100) (deref a))) + '(#t 100)) + +(test "compare-and-set! failure" + (let ([a (atom 5)]) + (list (compare-and-set! a 999 100) (deref a))) + '(#f 5)) + +;;; ---- Watches ---- + +(test "add-watch! fires on reset!" + (let ([a (atom 0)] + [seen (atom '())]) + (add-watch! a 'w + (lambda (k atm old new) + (swap! seen (lambda (xs) (cons (list k old new) xs))))) + (reset! a 42) + (reverse (deref seen))) + '((w 0 42))) + +(test "add-watch! fires on swap!" + (let ([a (atom 1)] + [calls (atom '())]) + (add-watch! a 'tracker + (lambda (k atm old new) + (swap! calls (lambda (xs) (cons (cons old new) xs))))) + (swap! a + 2) + (swap! a * 5) + (reverse (deref calls))) + '((1 . 3) (3 . 15))) + +(test "add-watch! gets atom reference" + (let ([a (atom 0)] + [seen (atom #f)]) + (add-watch! a 'who + (lambda (k atm old new) + (reset! seen (eq? atm a)))) + (reset! a 1) + (deref seen)) + #t) + +(test "add-watch! receives key" + (let ([a (atom 0)] + [seen-key (atom #f)]) + (add-watch! a 'my-key + (lambda (k atm old new) (reset! seen-key k))) + (reset! a 1) + (deref seen-key)) + 'my-key) + +(test "multiple watches all fire" + (let ([a (atom 0)] + [w1 (atom 0)] + [w2 (atom 0)]) + (add-watch! a 'one (lambda (k atm old new) (swap! w1 + 1))) + (add-watch! a 'two (lambda (k atm old new) (swap! w2 + 1))) + (reset! a 1) + (reset! a 2) + (list (deref w1) (deref w2))) + '(2 2)) + +(test "same-key re-add replaces previous callback" + (let ([a (atom 0)] + [seen (atom #f)]) + (add-watch! a 'k (lambda (k atm old new) (reset! seen 'first))) + (add-watch! a 'k (lambda (k atm old new) (reset! seen 'second))) + (reset! a 1) + (deref seen)) + 'second) + +(test "remove-watch! stops firing" + (let ([a (atom 0)] + [count (atom 0)]) + (add-watch! a 'w (lambda (k atm old new) (swap! count + 1))) + (reset! a 1) + (reset! a 2) + (remove-watch! a 'w) + (reset! a 3) + (deref count)) + 2) + +(test "remove-watch! idempotent on missing key" + (let ([a (atom 0)]) + (remove-watch! a 'never-registered) + (deref a)) + 0) + +(test "compare-and-set! success fires watches" + (let ([a (atom 10)] + [count (atom 0)]) + (add-watch! a 'w (lambda (k atm old new) (swap! count + 1))) + (compare-and-set! a 10 20) + (deref count)) + 1) + +(test "compare-and-set! failure does NOT fire watches" + (let ([a (atom 10)] + [count (atom 0)]) + (add-watch! a 'w (lambda (k atm old new) (swap! count + 1))) + (compare-and-set! a 999 20) + (deref count)) + 0) + +(test "watch exception does not break the atom" + (let ([a (atom 0)]) + (add-watch! a 'bad (lambda (k atm old new) (error 'bad "boom"))) + (reset! a 1) + (reset! a 2) + (deref a)) + 2) + +(test "watch can reenter atom (runs outside lock)" + (let ([a (atom 0)] + [log (atom '())]) + (add-watch! a 'w + (lambda (k atm old new) + ;; Reading deref here would deadlock if watches ran inside the mutex. + (swap! log (lambda (xs) (cons (deref atm) xs))))) + (reset! a 1) + (reset! a 2) + (reverse (deref log))) + '(1 2)) + +(test "add-watch! / remove-watch! return atom for chaining" + (let ([a (atom 0)]) + (eq? (remove-watch! (add-watch! a 'w (lambda args #f)) 'w) + a)) + #t) + +;;; ---- Volatiles ---- + +(test "volatile! constructs" + (volatile? (volatile! 42)) + #t) + +(test "volatile? rejects non-volatiles" + (list (volatile? 42) (volatile? (atom 0)) (volatile? '())) + '(#f #f #f)) + +(test "vderef reads" + (vderef (volatile! 'hi)) + 'hi) + +(test "vreset! sets and returns new" + (let ([v (volatile! 0)]) + (list (vreset! v 10) (vderef v))) + '(10 10)) + +(test "vswap! applies function" + (let ([v (volatile! 10)]) + (list (vswap! v + 5) (vderef v))) + '(15 15)) + +(test "vswap! variadic" + (let ([v (volatile! 2)]) + (vswap! v * 3 4) + (vderef v)) + 24) + +(test "volatile has no watches" + ;; Sanity: volatiles aren't atoms, no watch API applies. + (not (atom? (volatile! 0))) + #t) + +;;; ---- Summary ---- +(printf "~%atom: ~a passed, ~a failed~%" pass fail) +(when (> fail 0) (exit 1))