cljrs_runtime/env/
monitor.rs

1//! Reentrant object monitors. A waiter holds no OS lock while parked.
2use cljrs_value::parking::Event;
3use cljrs_value::types::Realizer;
4use cljrs_value::{MapValue, SetValue, Value, ValueError, ValueResult};
5use std::collections::HashMap;
6use std::sync::{Arc, Mutex, OnceLock, Weak};
7
8#[derive(Default)]
9struct Monitor {
10    owner: Mutex<Option<(Realizer, usize)>>,
11    changed: Event,
12}
13static MONITORS: OnceLock<Mutex<HashMap<usize, Weak<Monitor>>>> = OnceLock::new();
14
15// The caller roots the object throughout ownership, so addresses cannot be
16// reused while a live monitor has an owner or a waiter.
17fn key(value: &Value) -> ValueResult<usize> {
18    macro_rules! ptr {
19        ($p:expr) => {
20            $p.get() as *const _ as usize
21        };
22    }
23    Ok(match value {
24        Value::BoundFn(p) => ptr!(p),
25        Value::ObjectArray(p) => ptr!(p),
26        Value::Error(p) => ptr!(p),
27        Value::Uuid(p) => ptr!(p),
28        Value::BigInt(p) => ptr!(p),
29        Value::BigDecimal(p) => ptr!(p),
30        Value::Ratio(p) => ptr!(p),
31        Value::Str(p) => ptr!(p),
32        Value::Symbol(p) => ptr!(p),
33        Value::Keyword(p) => ptr!(p),
34        Value::List(p) => ptr!(p),
35        Value::Vector(p) => ptr!(p),
36        Value::Queue(p) => ptr!(p),
37        Value::NativeFunction(p) => ptr!(p),
38        Value::Fn(p) => ptr!(p),
39        Value::Macro(p) => ptr!(p),
40        Value::Var(p) => ptr!(p),
41        Value::Atom(p) => ptr!(p),
42        Value::Namespace(p) => ptr!(p),
43        Value::LazySeq(p) => ptr!(p),
44        Value::Cons(p) => ptr!(p),
45        Value::Protocol(p) => ptr!(p),
46        Value::ProtocolFn(p) => ptr!(p),
47        Value::MultiFn(p) => ptr!(p),
48        Value::Volatile(p) => ptr!(p),
49        Value::Delay(p) => ptr!(p),
50        Value::Promise(p) => ptr!(p),
51        Value::Future(p) => ptr!(p),
52        Value::Agent(p) => ptr!(p),
53        Value::TypeInstance(p) => ptr!(p),
54        Value::NativeObject(p) => ptr!(p),
55        Value::Map(MapValue::Array(p)) => ptr!(p),
56        Value::Map(MapValue::Hash(p)) => ptr!(p),
57        Value::Map(MapValue::Sorted(p)) => ptr!(p),
58        Value::Set(SetValue::Hash(p)) => ptr!(p),
59        Value::Set(SetValue::Sorted(p)) => ptr!(p),
60        Value::Resource(p) => Arc::as_ptr(&p.0) as *const () as usize,
61        Value::SharedAtom(p) => Arc::as_ptr(p) as usize,
62        _ => return Err(ValueError::Other("locking requires an object".into())),
63    })
64}
65
66// Every owner/waiter keeps a registry reference. The final reference removes
67// only its own entry under the table lock, without scanning unrelated objects.
68struct MonitorRef {
69    key: usize,
70    monitor: Option<Arc<Monitor>>,
71}
72impl MonitorRef {
73    fn monitor(&self) -> &Monitor {
74        self.monitor.as_deref().unwrap()
75    }
76}
77impl Drop for MonitorRef {
78    fn drop(&mut self) {
79        let monitor = self.monitor.take().unwrap();
80        let mut table = MONITORS.get().unwrap().lock().unwrap();
81        if Arc::strong_count(&monitor) == 1
82            && table
83                .get(&self.key)
84                .is_some_and(|entry| entry.as_ptr() == Arc::as_ptr(&monitor))
85        {
86            table.remove(&self.key);
87        }
88        // Serialize the decrement with lookups AND other reference drops.
89        // Otherwise concurrent final drops could both observe count > 1.
90        drop(monitor);
91    }
92}
93
94pub struct MonitorGuard(MonitorRef, std::marker::PhantomData<std::rc::Rc<()>>);
95pub fn enter(value: &Value) -> ValueResult<MonitorGuard> {
96    let key = key(value)?;
97    let monitor = {
98        let mut table = MONITORS.get_or_init(Default::default).lock().unwrap();
99        match table.get(&key).and_then(Weak::upgrade) {
100            Some(monitor) => monitor,
101            None => {
102                let monitor = Arc::new(Monitor::default());
103                table.insert(key, Arc::downgrade(&monitor));
104                monitor
105            }
106        }
107    };
108    let reference = MonitorRef {
109        key,
110        monitor: Some(monitor),
111    };
112    let monitor = reference.monitor();
113    let owner = Realizer::current();
114    loop {
115        let listener = monitor.changed.listen();
116        {
117            let mut state = monitor.owner.lock().unwrap();
118            match &mut *state {
119                None => *state = Some((owner, 1)),
120                Some((active, depth)) if *active == owner => *depth += 1,
121                Some((active, _)) => {
122                    active.ensure_can_wait().map_err(ValueError::Other)?;
123                    drop(state);
124                    listener.wait(None);
125                    continue;
126                }
127            }
128        }
129        drop(listener);
130        return Ok(MonitorGuard(reference, std::marker::PhantomData));
131    }
132}
133impl Drop for MonitorGuard {
134    fn drop(&mut self) {
135        let mut state = self.0.monitor().owner.lock().unwrap();
136        let (owner, depth) = state.as_mut().expect("unowned monitor");
137        assert!(*owner == Realizer::current());
138        *depth -= 1;
139        if *depth == 0 {
140            *state = None;
141        }
142        drop(state);
143        self.0.monitor().changed.notify_all();
144    }
145}
146
147struct HeldMonitor {
148    guard: MonitorGuard,
149    value: Value,
150    #[cfg(feature = "no-gc")]
151    _lease: cljrs_gc::region::RegionLease,
152}
153#[derive(Default)]
154struct MonitorState {
155    held: Vec<HeldMonitor>,
156    legacy_async: bool,
157}
158thread_local! {
159    static HELD_MONITORS: std::cell::RefCell<MonitorState> = const { std::cell::RefCell::new(MonitorState { held: Vec::new(), legacy_async: false }) };
160}
161cljrs_gc::mutator_slot!("cljrs-runtime/HELD_MONITORS", refcell HELD_MONITORS: MonitorState, trace trace_held);
162fn trace_held(state: &MonitorState, visitor: &mut cljrs_gc::MarkVisitor) {
163    use cljrs_gc::Trace;
164    for entry in &state.held {
165        entry.value.trace(visitor);
166    }
167}
168#[cfg(not(feature = "no-gc"))]
169pub(crate) fn trace_current(visitor: &mut cljrs_gc::MarkVisitor) {
170    HELD_MONITORS.with(|held| trace_held(&held.borrow(), visitor));
171}
172
173// Legacy async tasks share the root mutator and cannot retain explicit
174// ownership across a yield. Fibers have their own monitor state and can.
175fn ensure_explicit_owner() -> ValueResult<()> {
176    if cljrs_value::parking::current().is_none()
177        && (HELD_MONITORS.with(|state| state.borrow().legacy_async)
178            || crate::env::callback::current_is_async())
179    {
180        return Err(ValueError::Other(
181            "explicit monitors require a synchronous context or a fiber".into(),
182        ));
183    }
184    Ok(())
185}
186
187/// Mark one poll of a legacy async task. Explicit monitors are rejected even
188/// when that task calls a synchronous helper. Drop restores the caller's state.
189#[must_use]
190pub struct AsyncMonitorScope {
191    previous: bool,
192    _not_send: std::marker::PhantomData<std::rc::Rc<()>>,
193}
194impl AsyncMonitorScope {
195    pub fn enter() -> Self {
196        Self {
197            previous: HELD_MONITORS
198                .with(|state| std::mem::replace(&mut state.borrow_mut().legacy_async, true)),
199            _not_send: std::marker::PhantomData,
200        }
201    }
202}
203impl Drop for AsyncMonitorScope {
204    fn drop(&mut self) {
205        HELD_MONITORS.with(|state| state.borrow_mut().legacy_async = self.previous);
206    }
207}
208
209/// Acquire an explicit monitor for this computation until `monitor_exit`.
210pub fn monitor_enter(value: &Value) -> ValueResult<()> {
211    ensure_explicit_owner()?;
212    #[cfg(not(feature = "no-gc"))]
213    cljrs_gc::region::poison_active_regions();
214    let guard = enter(value)?;
215    let entry = HeldMonitor {
216        guard,
217        value: value.clone(),
218        #[cfg(feature = "no-gc")]
219        _lease: cljrs_gc::region::RegionLease::for_value(value),
220    };
221    HELD_MONITORS.with(|held| held.borrow_mut().held.push(entry));
222    Ok(())
223}
224
225/// Release one explicit acquisition. Reject objects this computation does not own.
226pub fn monitor_exit(value: &Value) -> ValueResult<()> {
227    ensure_explicit_owner()?;
228    let key = key(value)?;
229    let entry = HELD_MONITORS.with(|held| {
230        let mut state = held.borrow_mut();
231        let held = &mut state.held;
232        let index = held
233            .iter()
234            .rposition(|entry| entry.guard.0.key == key)
235            .ok_or_else(|| {
236                ValueError::Other("monitor-exit: this computation does not own the monitor".into())
237            })?;
238        Ok(held.remove(index))
239    });
240    let entry: HeldMonitor = entry?;
241    drop(entry);
242    Ok(())
243}
244
245/// Release unmatched explicit acquisitions on fiber completion or cancellation.
246pub struct MonitorCleanup;
247impl Drop for MonitorCleanup {
248    fn drop(&mut self) {
249        let held = HELD_MONITORS.with(|held| std::mem::take(&mut held.borrow_mut().held));
250        drop(held);
251    }
252}
253
254#[cfg(test)]
255mod tests {
256    use super::*;
257    #[test]
258    fn final_reference_removes_only_its_own_monitor_entry() {
259        let locks: Vec<_> = (0..1000)
260            .map(|_| Value::Atom(cljrs_gc::GcPtr::new(cljrs_value::Atom::new(Value::Nil))))
261            .collect();
262        let keys: Vec<_> = locks.iter().map(|value| key(value).unwrap()).collect();
263        let guards: Vec<_> = locks.iter().map(|value| enter(value).unwrap()).collect();
264        {
265            let table = MONITORS.get().unwrap().lock().unwrap();
266            assert!(keys.iter().all(|key| table.contains_key(key)));
267        }
268        // Reentrant release must leave the first owner's registration live.
269        drop(enter(&locks[0]).unwrap());
270        assert!(
271            MONITORS
272                .get()
273                .unwrap()
274                .lock()
275                .unwrap()
276                .contains_key(&keys[0])
277        );
278        drop(guards);
279        let table = MONITORS.get().unwrap().lock().unwrap();
280        assert!(keys.iter().all(|key| !table.contains_key(key)));
281    }
282}
cljrs_runtime/env/monitor.rs

cljrs_runtime/env/
monitor.rs
