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
147#[derive(Default)]
148struct MonitorState {
149    held: Vec<(MonitorGuard, Value)>,
150    legacy_async: bool,
151}
152thread_local! {
153    static HELD_MONITORS: std::cell::RefCell<MonitorState> = const { std::cell::RefCell::new(MonitorState { held: Vec::new(), legacy_async: false }) };
154}
155cljrs_gc::mutator_slot!("cljrs-runtime/HELD_MONITORS", refcell HELD_MONITORS: MonitorState, trace trace_held);
156fn trace_held(state: &MonitorState, visitor: &mut cljrs_gc::MarkVisitor) {
157    use cljrs_gc::Trace;
158    for (_, value) in &state.held {
159        value.trace(visitor);
160    }
161}
162#[cfg(not(feature = "no-gc"))]
163pub(crate) fn trace_current(visitor: &mut cljrs_gc::MarkVisitor) {
164    HELD_MONITORS.with(|held| trace_held(&held.borrow(), visitor));
165}
166
167// Legacy async tasks share the root mutator and cannot retain explicit
168// ownership across a yield. Fibers have their own monitor state and can.
169fn ensure_explicit_owner() -> ValueResult<()> {
170    if cljrs_value::parking::current().is_none()
171        && (HELD_MONITORS.with(|state| state.borrow().legacy_async)
172            || crate::env::callback::current_is_async())
173    {
174        return Err(ValueError::Other(
175            "explicit monitors require a synchronous context or a fiber".into(),
176        ));
177    }
178    Ok(())
179}
180
181/// Mark one poll of a legacy async task. Explicit monitors are rejected even
182/// when that task calls a synchronous helper. Drop restores the caller's state.
183#[must_use]
184pub struct AsyncMonitorScope {
185    previous: bool,
186    _not_send: std::marker::PhantomData<std::rc::Rc<()>>,
187}
188impl AsyncMonitorScope {
189    pub fn enter() -> Self {
190        Self {
191            previous: HELD_MONITORS
192                .with(|state| std::mem::replace(&mut state.borrow_mut().legacy_async, true)),
193            _not_send: std::marker::PhantomData,
194        }
195    }
196}
197impl Drop for AsyncMonitorScope {
198    fn drop(&mut self) {
199        HELD_MONITORS.with(|state| state.borrow_mut().legacy_async = self.previous);
200    }
201}
202
203/// Acquire an explicit monitor for this computation until `monitor_exit`.
204pub fn monitor_enter(value: &Value) -> ValueResult<()> {
205    ensure_explicit_owner()?;
206    cljrs_gc::region::poison_active_regions();
207    let guard = enter(value)?;
208    HELD_MONITORS.with(|held| held.borrow_mut().held.push((guard, value.clone())));
209    Ok(())
210}
211
212/// Release one explicit acquisition. Reject objects this computation does not own.
213pub fn monitor_exit(value: &Value) -> ValueResult<()> {
214    ensure_explicit_owner()?;
215    let key = key(value)?;
216    let entry = HELD_MONITORS.with(|held| {
217        let mut state = held.borrow_mut();
218        let held = &mut state.held;
219        let index = held
220            .iter()
221            .rposition(|(guard, _)| guard.0.key == key)
222            .ok_or_else(|| {
223                ValueError::Other("monitor-exit: this computation does not own the monitor".into())
224            })?;
225        Ok(held.remove(index))
226    });
227    let entry: (MonitorGuard, Value) = entry?;
228    drop(entry);
229    Ok(())
230}
231
232/// Release unmatched explicit acquisitions on fiber completion or cancellation.
233pub struct MonitorCleanup;
234impl Drop for MonitorCleanup {
235    fn drop(&mut self) {
236        let held = HELD_MONITORS.with(|held| std::mem::take(&mut held.borrow_mut().held));
237        drop(held);
238    }
239}
240
241#[cfg(test)]
242mod tests {
243    use super::*;
244    #[test]
245    fn final_reference_removes_only_its_own_monitor_entry() {
246        let locks: Vec<_> = (0..1000)
247            .map(|_| Value::Atom(cljrs_gc::GcPtr::new(cljrs_value::Atom::new(Value::Nil))))
248            .collect();
249        let keys: Vec<_> = locks.iter().map(|value| key(value).unwrap()).collect();
250        let guards: Vec<_> = locks.iter().map(|value| enter(value).unwrap()).collect();
251        {
252            let table = MONITORS.get().unwrap().lock().unwrap();
253            assert!(keys.iter().all(|key| table.contains_key(key)));
254        }
255        // Reentrant release must leave the first owner's registration live.
256        drop(enter(&locks[0]).unwrap());
257        assert!(
258            MONITORS
259                .get()
260                .unwrap()
261                .lock()
262                .unwrap()
263                .contains_key(&keys[0])
264        );
265        drop(guards);
266        let table = MONITORS.get().unwrap().lock().unwrap();
267        assert!(keys.iter().all(|key| !table.contains_key(key)));
268    }
269}
cljrs_runtime/env/monitor.rs

cljrs_runtime/env/
monitor.rs
