Add ref watches and validators (#627)

* Add ref watches and validators

* Bust CircleCI cache

* Atom validator and watch tests

* Test CAS with validators too

* Ignore Pip safety check

* Var ROOT tests

* change the log change the world

* Thread-local var tests

Author: chrisrink10

.circleci/config.yml

@@ -15,7 +15,7 @@ commands:
       - run: sudo chown -R circleci:circleci /usr/local/bin
       - run: sudo chown -R circleci:circleci /usr/local/lib/python<< parameters.python_version >>/site-packages
       - restore_cache:
-          key: << parameters.cache_key_prefix >>1-{{ checksum "tox.ini" }}-{{ checksum "pyproject.toml" }}
+          key: << parameters.cache_key_prefix >>2-{{ checksum "tox.ini" }}-{{ checksum "pyproject.toml" }}
       - run:
           name: Install Poetry and Tox
           shell: /bin/bash -leo pipefail
@@ -35,7 +35,7 @@ commands:
         type: string
     steps:
       - save_cache:
-          key: << parameters.cache_key_prefix >>1-{{ checksum "tox.ini" }}-{{ checksum "pyproject.toml" }}
+          key: << parameters.cache_key_prefix >>2-{{ checksum "tox.ini" }}-{{ checksum "pyproject.toml" }}
           paths:
             - "/home/circleci/project/.tox"
             - "/usr/local/bin"
@@ -85,7 +85,7 @@ commands:
     steps:
       - checkout
       - restore_cache:
-          key: pypy-deps2-{{ checksum "tox.ini" }}-{{ checksum "pyproject.toml" }}
+          key: pypy-deps3-{{ checksum "tox.ini" }}-{{ checksum "pyproject.toml" }}
       - run:
           name: Install Poetry and Tox
           command: |
@@ -103,7 +103,7 @@ commands:
             tox -e pypy3 -- $CCI_NODE_TESTS
           no_output_timeout: 30m
       - save_cache:
-          key: pypy-deps2-{{ checksum "tox.ini" }}-{{ checksum "pyproject.toml" }}
+          key: pypy-deps3-{{ checksum "tox.ini" }}-{{ checksum "pyproject.toml" }}
           paths:
             - "/root/project/.tox"
             - "/usr/local/bin"

CHANGELOG.md

@@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 ### Added
  * Added a bootstrapping function for easily bootstrapping Basilisp projects from Python (#620)
+ * Added support for watchers and validator functions on Atoms and Vars (#627)
 
 ### Changed
  * PyTest is now an optional extra dependency, rather than a required dependency (#622)

src/basilisp/core.lpy

@@ -877,8 +877,10 @@
   may be returned by deref'ing it. The value of an atom may be reset using
   reset! and may be swapped using swap!. All operations on an atom occur
   atomically."
-  [v]
-  (basilisp.lang.atom/Atom v))
+  ([v]
+   (basilisp.lang.atom/Atom v))
+  ([v & kwargs]
+   (apply-kw basilisp.lang.atom/Atom v (apply hash-map kwargs))))
 
 (defn realized?
   "Return true if the delay, future, lazy sequence, or promise has been
@@ -4029,9 +4031,9 @@
                                                    ctx
                                                    namespace))))
 
-;;;;;;;;;;;;;;;;;;;
-;; Ref Utilities ;;
-;;;;;;;;;;;;;;;;;;;
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+;; Reference (Atom/Namespace/Var) Utilities ;;
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
 (defn alter-meta!
   "Atomically swap the metadata on reference o to the result of (apply f m args) where
@@ -4046,6 +4048,56 @@
   [o meta]
   (.reset-meta o meta))
 
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+;; Ref (Atom/Var) Utilities ;;
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+
+(defn add-watch
+  "Add a watch function `wf` identified by the key `k` to the ref (Atom or Var).
+
+  Watch functions should be functions of four arguments: the key associated with the
+  watch function, the ref object, the old value, and the new value.
+
+  Watch functions will be called synchronously. Note that the value of a ref may have
+  changed by the time a watch function is called, so watch functions should use the
+  old and new state arguments rather than attempting to `deref` the ref.
+
+  Watch functions cannot prevent state changes to a ref, regardless of the result of
+  the watch function.
+
+  Watch function keys are used to identify watches and may be used to remove a watch
+  from a ref (using `remove-watch`), but are otherwise unused by the watch feature.
+
+  Watches for a Var are only notified for Var root changes, not thread-local bindings."
+  [ref k wf]
+  (.add-watch ref k wf))
+
+(defn remove-watch
+  "Remove the watch function identified by `k` from the ref (Atom or Var), if it exists.
+  Returns the ref."
+  [ref k]
+  (.remove-watch ref k))
+
+(defn get-validator
+  "Return the validator defined for the ref (Atom or Var), or `nil` if no validator is
+  defined."
+  [ref]
+  (.get-validator ref))
+
+(defn set-validator!
+  "Set the validator function for the ref (Atom or Var).
+
+  Validator functions should be side-effect free functions of one argument: the proposed
+  new value of the ref. The validator should either return false or throw an error if
+  the ref value is invalid.
+
+  `nil` may be passed to remove the validator for a ref.
+
+  If the existing ref value is not valid according to `vf`, an exception will be thrown
+  and the new validator function will not be applied."
+  [ref vf]
+  (.set-validator ref vf))
+
 ;;;;;;;;;;;;;;;;;;;
 ;; Var Utilities ;;
 ;;;;;;;;;;;;;;;;;;;

src/basilisp/lang/atom.py

@@ -2,48 +2,71 @@
 
 from readerwriterlock.rwlock import RWLockFair
 
-from basilisp.lang.interfaces import IDeref, IPersistentMap
-from basilisp.lang.reference import ReferenceBase
+from basilisp.lang.interfaces import IPersistentMap, RefValidator
+from basilisp.lang.map import PersistentMap
+from basilisp.lang.reference import RefBase
 
 T = TypeVar("T")
 
 
-class Atom(IDeref[T], ReferenceBase, Generic[T]):
-    __slots__ = ("_meta", "_state", "_rlock", "_wlock")
+class Atom(RefBase[T], Generic[T]):
+    __slots__ = ("_meta", "_state", "_rlock", "_wlock", "_watches", "_validator")
 
-    # pylint: disable=assigning-non-slot
-    def __init__(self, state: T) -> None:
-        self._meta: Optional[IPersistentMap] = None
+    def __init__(
+        self,
+        state: T,
+        meta: Optional[IPersistentMap] = None,
+        validator: Optional[RefValidator] = None,
+    ) -> None:
+        self._meta: Optional[IPersistentMap] = meta
         self._state = state
         lock = RWLockFair()
         self._rlock = lock.gen_rlock()
         self._wlock = lock.gen_wlock()
+        self._watches = PersistentMap.empty()
+        self._validator = validator
 
-    def compare_and_set(self, old: T, new: T) -> bool:
-        """Compare the current state of the Atom to `old`. If the value is the same,
-        atomically set the value of the state of Atom to `new`. Return True if the
-        value was swapped. Return False otherwise."""
+        if validator is not None:
+            self._validate(state)
+
+    def _compare_and_set(self, old: T, new: T) -> bool:
         with self._wlock:
             if self._state != old:
                 return False
             self._state = new
             return True
 
+    def compare_and_set(self, old: T, new: T) -> bool:
+        """Compare the current state of the Atom to `old`. If the value is the same,
+        atomically set the value of the state of Atom to `new`. Return True if the
+        value was swapped. Return False otherwise."""
+        self._validate(new)
+        if self._compare_and_set(old, new):
+            self._notify_watches(old, new)
+            return True
+        return False
+
     def deref(self) -> T:
         """Return the state stored within the Atom."""
         with self._rlock:
             return self._state
 
     def reset(self, v: T) -> T:
         """Reset the state of the Atom to `v` without regard to the current value."""
-        with self._wlock:
-            self._state = v
-            return v
+        while True:
+            oldval = self._state
+            self._validate(v)
+            if self._compare_and_set(oldval, v):
+                self._notify_watches(oldval, v)
+                return v
 
     def swap(self, f: Callable[..., T], *args, **kwargs) -> T:
         """Atomically swap the state of the Atom to the return value of
         `f(old, *args, **kwargs)`, returning the new value."""
-        with self._wlock:
-            newval = f(self._state, *args, **kwargs)
-            self._state = newval
-            return newval
+        while True:
+            oldval = self._state
+            newval = f(oldval, *args, **kwargs)
+            self._validate(newval)
+            if self._compare_and_set(oldval, newval):
+                self._notify_watches(oldval, newval)
+                return newval

src/basilisp/lang/interfaces.py

@@ -2,8 +2,10 @@
 from abc import ABC, abstractmethod
 from typing import (
     AbstractSet,
+    Any,
     Callable,
     Generic,
+    Hashable,
     Iterable,
     Iterator,
     Mapping,
@@ -122,6 +124,31 @@ def reset_meta(self, meta: "IPersistentMap") -> "IPersistentMap":
         raise NotImplementedError()
 
 
+RefValidator = Callable[[Any], Any]
+RefWatchKey = Hashable
+RefWatcher = Callable[[RefWatchKey, "IRef", Any, Any], None]
+
+
+class IRef(IReference):
+    __slots__ = ()
+
+    @abstractmethod
+    def add_watch(self, k: RefWatchKey, wf: RefWatcher) -> "IReference":
+        raise NotImplementedError()
+
+    @abstractmethod
+    def remove_watch(self, k: RefWatchKey) -> "IReference":
+        raise NotImplementedError()
+
+    @abstractmethod
+    def get_validator(self) -> Optional[RefValidator]:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def set_validator(self, vf: Optional[RefValidator] = None) -> None:
+        raise NotImplementedError()
+
+
 class IReversible(Generic[T]):
     """IReversible types can produce a sequences of their elements in reverse in
     constant time.

src/basilisp/lang/reference.py

@@ -1,8 +1,19 @@
-from typing import Callable, Optional
+from typing import Any, Callable, Optional, TypeVar
 
 from readerwriterlock.rwlock import Lockable
 
-from basilisp.lang.interfaces import IPersistentMap, IReference
+from basilisp.lang import keyword as kw
+from basilisp.lang import map as lmap
+from basilisp.lang.exception import ExceptionInfo
+from basilisp.lang.interfaces import (
+    IDeref,
+    IPersistentMap,
+    IRef,
+    IReference,
+    RefValidator,
+    RefWatcher,
+    RefWatchKey,
+)
 
 try:
     from typing import Protocol
@@ -20,7 +31,10 @@ def __call__(
 class ReferenceBase(IReference):
     """Mixin for IReference classes to define the full IReference interface.
 
-    Consumers must have a `_lock` and `_meta` property defined."""
+    `basilisp.lang.runtime.Namespace` objects are the only objects which are
+    `IReference` objects without also being `IRef` objects.
+
+    Implementers must have the `_rlock`, `_wlock`, and `_meta` properties defined."""
 
     _rlock: Lockable
     _wlock: Lockable
@@ -40,3 +54,60 @@ def reset_meta(self, meta: IPersistentMap) -> IPersistentMap:
         with self._wlock:
             self._meta = meta
             return meta
+
+
+T = TypeVar("T")
+
+
+class RefBase(IDeref[T], IRef, ReferenceBase):
+    """
+    Mixin for IRef classes to define the full IRef interface.
+
+    `IRef` objects are generally shared, mutable state objects such as Atoms and
+    Vars.
+
+    Implementers must have the `_validators` and `_watches` properties defined.
+    """
+
+    _validator: Optional[RefValidator]
+    _watches: IPersistentMap
+
+    def add_watch(self, k: RefWatchKey, wf: RefWatcher) -> "RefBase[T]":
+        with self._wlock:
+            self._watches = self._watches.assoc(k, wf)
+            return self
+
+    def _notify_watches(self, old: Any, new: Any):
+        for k, wf in self._watches.items():
+            wf(k, self, old, new)
+
+    def remove_watch(self, k: RefWatchKey) -> "RefBase[T]":
+        with self._wlock:
+            self._watches = self._watches.dissoc(k)
+            return self
+
+    def get_validator(self) -> Optional[RefValidator]:
+        return self._validator
+
+    def set_validator(self, vf: Optional[RefValidator] = None) -> None:
+        # We cannot use a write lock here since we're calling `self.deref()` which
+        # attempts to acquire the read lock for the Ref and will deadlock if the
+        # lock is not reentrant.
+        #
+        # There are no guarantees that the Ref lock is reentrant and the default
+        # locks for Atoms and Vars are not).
+        #
+        # This is probably ok for most cases since we expect contention is low or
+        # non-existent while setting a validator function.
+        if vf is not None:
+            self._validate(self.deref(), vf=vf)
+        self._validator = vf
+
+    def _validate(self, val: Any, vf: Optional[RefValidator] = None):
+        vf = vf or self._validator
+        if vf is not None:
+            if not vf(val):
+                raise ExceptionInfo(
+                    "Invalid reference state",
+                    lmap.map({kw.keyword("data"): val, kw.keyword("validator"): vf}),
+                )