cljrs_value/
shared.rs

1//! Cross-isolate shared mutable state: `SharedValue`, `SharedAtom` (Phase B3).
2//!
3//! The isolate model is share-nothing for GC-heap data.  But some state
4//! genuinely needs to be visible across isolates — global configuration, shared
5//! counters, published results.  Phase B3 adds an *explicit*, *honest* escape
6//! hatch: `shared-atom`.
7//!
8//! ## Two-tier mutable references (per the ADR)
9//!
10//! | Primitive | Backing | `Send`? | Use case |
11//! |-----------|---------|---------|----------|
12//! | `atom`      | `GcPtr<Atom>` (`Mutex<Value>`) | `!Send` | isolate-local, fast |
13//! | `shared-atom` | `Arc<SharedAtom>` (`ArcSwap<SharedValue>`) | ✓ | cross-isolate, lock-free CAS |
14//!
15//! A value stored in a `shared-atom` must be **promotable** — it must be
16//! representable as a `SharedValue`.  The promotion cost is paid once on
17//! publish; reads (`deref`) are an atomic load.
18//!
19//! ## `SharedValue`
20//!
21//! Covers only the "plain data" subset of `Value`:
22//! - Scalars (stored inline, no allocation)
23//! - Strings (`Arc<str>`, immutable, refcounted)
24//! - Keywords / symbols (`StaticGcPtr<T>`, interned, program-lifetime)
25//! - Large byte buffers (`Arc<[u8]>`, the BEAM off-heap-binary trick)
26//!
27//! Closures, native resources, and isolate-bound GC objects are not
28//! promotable.  This restriction is enforced at publish time via `promote`.
29
30use std::sync::{Arc, Mutex};
31
32use arc_swap::ArcSwap;
33use cljrs_gc::{GcPtr, StaticGcPtr};
34
35use crate::intern::{intern_keyword, intern_symbol};
36use crate::keyword::Keyword;
37use crate::symbol::Symbol;
38use crate::value::Value;
39
40// ── PromoteError ─────────────────────────────────────────────────────────────
41
42/// Returned when a `Value` cannot be promoted to [`SharedValue`].
43#[derive(Debug, Clone)]
44pub struct PromoteError {
45    pub type_name: &'static str,
46}
47
48impl PromoteError {
49    pub fn not_promotable(type_name: &'static str) -> Self {
50        Self { type_name }
51    }
52}
53
54impl std::fmt::Display for PromoteError {
55    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
56        write!(
57            f,
58            "value of type '{}' cannot be promoted to a shared-atom: \
59             only scalars, strings, keywords, symbols, and byte-blobs are supported",
60            self.type_name
61        )
62    }
63}
64
65impl std::error::Error for PromoteError {}
66
67// ── SharedValue ───────────────────────────────────────────────────────────────
68
69/// A `Send + Sync` value representation for cross-isolate sharing.
70///
71/// Covers a carefully chosen subset of `Value`: scalars (stored inline),
72/// immutable strings and byte buffers (refcounted via `Arc`), and interned
73/// keywords/symbols (program-lifetime `StaticGcPtr`).  Anything that holds
74/// isolate-local `GcPtr`s is excluded.
75///
76/// Cycles are impossible — `SharedValue` contains no `SharedValue` references
77/// — so plain `Arc` refcounting is sufficient and cycle-free.
78#[derive(Debug, Clone)]
79pub enum SharedValue {
80    Nil,
81    Bool(bool),
82    Long(i64),
83    Double(f64),
84    Char(char),
85    Uuid(u128),
86    /// Immutable interned string slice.
87    Str(Arc<str>),
88    /// Keyword interned into the global static arena.
89    Keyword(StaticGcPtr<Keyword>),
90    /// Symbol interned into the global static arena.
91    Symbol(StaticGcPtr<Symbol>),
92    /// Large refcounted byte buffer (the BEAM off-heap-binary trick).
93    /// Shared without copy across the isolate boundary; freed when the last
94    /// reference is dropped.
95    ByteBlob(Arc<[u8]>),
96}
97
98// SAFETY: all variants are either value types or `Arc`/`StaticGcPtr` wrappers
99// that are themselves `Send + Sync`.
100unsafe impl Send for SharedValue {}
101unsafe impl Sync for SharedValue {}
102
103impl SharedValue {
104    pub fn type_name(&self) -> &'static str {
105        match self {
106            SharedValue::Nil => "nil",
107            SharedValue::Bool(_) => "boolean",
108            SharedValue::Long(_) => "long",
109            SharedValue::Double(_) => "double",
110            SharedValue::Char(_) => "char",
111            SharedValue::Uuid(_) => "uuid",
112            SharedValue::Str(_) => "string",
113            SharedValue::Keyword(_) => "keyword",
114            SharedValue::Symbol(_) => "symbol",
115            SharedValue::ByteBlob(_) => "byte-blob",
116        }
117    }
118}
119
120// ── promote / demote ──────────────────────────────────────────────────────────
121
122/// Promote a `Value` to a [`SharedValue`] for cross-isolate publishing.
123///
124/// The promotion cost is paid once (at `reset!`/`swap!` time); reads of the
125/// resulting `SharedAtom` are lock-free atomic loads.
126///
127/// Returns [`PromoteError`] for values that hold isolate-local state.
128pub fn promote(value: &Value) -> Result<SharedValue, PromoteError> {
129    match value {
130        Value::Nil => Ok(SharedValue::Nil),
131        Value::Bool(b) => Ok(SharedValue::Bool(*b)),
132        Value::Long(n) => Ok(SharedValue::Long(*n)),
133        Value::Double(d) => Ok(SharedValue::Double(*d)),
134        Value::Char(c) => Ok(SharedValue::Char(*c)),
135        Value::Uuid(u) => Ok(SharedValue::Uuid(*u)),
136        Value::Str(s) => Ok(SharedValue::Str(Arc::from(s.get().as_str()))),
137        Value::Keyword(kw) => {
138            let kw = kw.get();
139            let ptr = intern_keyword(kw.namespace.as_deref(), &kw.name);
140            Ok(SharedValue::Keyword(ptr))
141        }
142        Value::Symbol(sym) => {
143            let sym = sym.get();
144            let ptr = intern_symbol(sym.namespace.as_deref(), &sym.name, sym.version.as_deref());
145            Ok(SharedValue::Symbol(ptr))
146        }
147        Value::ByteArray(arr) => {
148            let bytes = arr.get().lock().unwrap();
149            let blob: Arc<[u8]> = bytes.iter().map(|&b| b as u8).collect::<Vec<_>>().into();
150            Ok(SharedValue::ByteBlob(blob))
151        }
152        Value::ByteBlob(blob) => Ok(SharedValue::ByteBlob(blob.clone())),
153        other => Err(PromoteError::not_promotable(other.type_name())),
154    }
155}
156
157/// Demote a [`SharedValue`] back into an isolate-local `Value`.
158///
159/// Always succeeds.  Keywords and symbols are cloned from their interned
160/// static-arena allocation into a fresh `GcPtr` on the calling isolate's heap.
161/// Map lookups compare by namespace/name content (not pointer identity) so
162/// equality is preserved across the round-trip.
163pub fn demote(sv: &SharedValue) -> Value {
164    match sv {
165        SharedValue::Nil => Value::Nil,
166        SharedValue::Bool(b) => Value::Bool(*b),
167        SharedValue::Long(n) => Value::Long(*n),
168        SharedValue::Double(d) => Value::Double(*d),
169        SharedValue::Char(c) => Value::Char(*c),
170        SharedValue::Uuid(u) => Value::Uuid(*u),
171        SharedValue::Str(s) => Value::Str(GcPtr::new(s.as_ref().to_owned())),
172        SharedValue::Keyword(kw) => Value::Keyword(GcPtr::new(kw.get().clone())),
173        SharedValue::Symbol(sym) => Value::Symbol(GcPtr::new(sym.get().clone())),
174        SharedValue::ByteBlob(blob) => Value::ByteBlob(blob.clone()),
175    }
176}
177
178// ── SharedAtom ────────────────────────────────────────────────────────────────
179
180/// A cross-isolate mutable reference backed by a lock-free `ArcSwap`.
181///
182/// `reset!` and `swap!` promote the new value and perform an atomic pointer
183/// swap — no locking, O(1) for simple resets, O(retries) for CAS-retry
184/// `swap!` under write contention.  `deref` is a single atomic load plus an
185/// `Arc` reference-count bump.
186///
187/// Wrapped in `Value::SharedAtom(Arc<SharedAtom>)`.  The `Arc` gives the atom
188/// identity (pointer equality) and allows any number of isolates to hold a
189/// reference.
190#[derive(Debug)]
191pub struct SharedAtom {
192    pub cell: Arc<ArcSwap<SharedValue>>,
193    pub meta: Mutex<Option<SharedValue>>,
194}
195
196impl SharedAtom {
197    pub fn new(val: SharedValue) -> Self {
198        Self {
199            cell: Arc::new(ArcSwap::new(Arc::new(val))),
200            meta: Mutex::new(None),
201        }
202    }
203
204    /// Load the current value.  Lock-free atomic load + refcount bump.
205    pub fn deref_val(&self) -> Arc<SharedValue> {
206        self.cell.load_full()
207    }
208
209    /// Atomically replace the value and return the new `Arc`.
210    pub fn reset(&self, val: SharedValue) -> Arc<SharedValue> {
211        let arc = Arc::new(val);
212        self.cell.store(arc.clone());
213        arc
214    }
215
216    /// CAS-retry swap: apply `f` to the current value and store the result.
217    /// Returns the new value.  Retries automatically if another writer races.
218    pub fn swap<F>(&self, mut f: F) -> Arc<SharedValue>
219    where
220        F: FnMut(&SharedValue) -> SharedValue,
221    {
222        self.cell.rcu(|old| Arc::new(f(old)))
223    }
224
225    /// Single lock-free compare-and-set.  Atomically stores `new` iff the cell
226    /// still holds `current` (by `Arc` identity).  Returns `true` on success.
227    ///
228    /// This is the primitive used by Clojure-level `compare-and-set!` and by the
229    /// retry loop behind `swap!`: a caller that needs to run arbitrary
230    /// (interpreter) code between the load and the store cannot use the closure
231    /// form of [`swap`], so it loads via [`deref_val`](Self::deref_val), computes
232    /// the next value, then commits with this method, retrying on contention.
233    pub fn compare_and_set(&self, current: &Arc<SharedValue>, new: SharedValue) -> bool {
234        let prev = self.cell.compare_and_swap(current, Arc::new(new));
235        // The swap committed iff the value we replaced is the one we expected.
236        std::ptr::eq(Arc::as_ptr(current), Arc::as_ptr(&prev))
237    }
238}
239
240// ── Tests ─────────────────────────────────────────────────────────────────────
241
242#[cfg(test)]
243mod tests {
244    use std::sync::Arc;
245
246    use crate::keyword::Keyword;
247    use crate::value::Value;
248    use cljrs_gc::GcPtr;
249
250    use super::*;
251
252    fn kw(name: &str) -> Value {
253        Value::Keyword(GcPtr::new(Keyword::simple(name)))
254    }
255
256    #[test]
257    fn promote_scalars() {
258        assert!(matches!(promote(&Value::Nil), Ok(SharedValue::Nil)));
259        assert!(matches!(
260            promote(&Value::Bool(true)),
261            Ok(SharedValue::Bool(true))
262        ));
263        assert!(matches!(
264            promote(&Value::Long(42)),
265            Ok(SharedValue::Long(42))
266        ));
267    }
268
269    #[test]
270    fn promote_keyword_interns() {
271        let v = kw("foo");
272        let sv = promote(&v).unwrap();
273        let sv2 = promote(&kw("foo")).unwrap();
274        if let (SharedValue::Keyword(a), SharedValue::Keyword(b)) = (sv, sv2) {
275            assert!(
276                cljrs_gc::StaticGcPtr::ptr_eq(&a, &b),
277                "same keyword should intern to same StaticGcPtr"
278            );
279        } else {
280            panic!("expected SharedValue::Keyword");
281        }
282    }
283
284    #[test]
285    fn demote_roundtrip_long() {
286        let sv = SharedValue::Long(99);
287        assert!(matches!(demote(&sv), Value::Long(99)));
288    }
289
290    #[test]
291    fn demote_roundtrip_keyword() {
292        let sv = promote(&kw("test")).unwrap();
293        let v = demote(&sv);
294        if let Value::Keyword(kw_ptr) = v {
295            assert_eq!(kw_ptr.get().name.as_ref(), "test");
296        } else {
297            panic!("expected Value::Keyword");
298        }
299    }
300
301    #[test]
302    fn promote_non_promotable_returns_err() {
303        let atom = Value::Atom(GcPtr::new(crate::types::Atom::new(Value::Nil)));
304        assert!(promote(&atom).is_err());
305    }
306
307    #[test]
308    fn promote_byte_blob() {
309        let arr: Vec<i8> = vec![1, 2, 3];
310        let v = Value::ByteArray(GcPtr::new(std::sync::Mutex::new(arr)));
311        let sv = promote(&v).unwrap();
312        assert!(matches!(sv, SharedValue::ByteBlob(_)));
313    }
314
315    #[test]
316    fn shared_atom_reset_and_deref() {
317        let atom = SharedAtom::new(SharedValue::Long(0));
318        atom.reset(SharedValue::Long(42));
319        let val = atom.deref_val();
320        assert!(matches!(val.as_ref(), SharedValue::Long(42)));
321    }
322
323    #[test]
324    fn shared_atom_swap() {
325        let atom = SharedAtom::new(SharedValue::Long(1));
326        atom.swap(|old| {
327            if let SharedValue::Long(n) = old {
328                SharedValue::Long(n + 1)
329            } else {
330                SharedValue::Long(0)
331            }
332        });
333        let val = atom.deref_val();
334        assert!(matches!(val.as_ref(), SharedValue::Long(2)));
335    }
336
337    #[test]
338    fn shared_atom_compare_and_set() {
339        let atom = SharedAtom::new(SharedValue::Long(1));
340        let cur = atom.deref_val();
341        // Stale expectation succeeds while no one races us.
342        assert!(atom.compare_and_set(&cur, SharedValue::Long(2)));
343        assert!(matches!(atom.deref_val().as_ref(), SharedValue::Long(2)));
344        // `cur` is now stale: a second CAS against it must fail and not write.
345        assert!(!atom.compare_and_set(&cur, SharedValue::Long(99)));
346        assert!(matches!(atom.deref_val().as_ref(), SharedValue::Long(2)));
347    }
348
349    #[test]
350    fn shared_atom_is_send_sync() {
351        fn assert_send_sync<T: Send + Sync>() {}
352        assert_send_sync::<SharedAtom>();
353        assert_send_sync::<Arc<SharedAtom>>();
354    }
355
356    #[test]
357    fn byte_blob_shared_across_clone() {
358        let blob: Arc<[u8]> = vec![10u8, 20, 30].into();
359        let v1 = Value::ByteBlob(blob.clone());
360        let v2 = Value::ByteBlob(blob.clone());
361        // Same underlying buffer
362        if let (Value::ByteBlob(a), Value::ByteBlob(b)) = (&v1, &v2) {
363            assert!(Arc::ptr_eq(a, b));
364        }
365    }
366}
cljrs_value/shared.rs

cljrs_value/
shared.rs
