access_cell/
lib.rs

1//! A cell that allows mutable access with re-entrancy support.
2//!
3//! When you call [`access`](AccessCell::access) with a closure that itself calls `access`
4//! on the same cell, those nested calls are queued and run after the current closure
5//! finishes, avoiding deadlock.
6//!
7//! This type is thread-safe: it implements [`Send`] and [`Sync`] when `T: Send`.
8
9use std::{
10    collections::VecDeque,
11    sync::{
12        atomic::{AtomicBool, Ordering},
13        Mutex, RwLock,
14    },
15};
16
17/// A cell holding a value of type `T` with re-entrant mutable access.
18///
19/// The first caller of [`access`](AccessCell::access) runs immediately; any further
20/// calls to `access` from within that closure (or its queue) are enqueued and
21/// executed in order after the current work completes.
22///
23/// Thread-safe: safe to share across threads and use from multiple threads.
24pub struct AccessCell<T> {
25    value: RwLock<T>,
26    running: AtomicBool,
27    queue: Mutex<VecDeque<Box<dyn FnOnce(&mut T) + Send>>>,
28}
29
30// Safe: only one thread holds the write guard at a time (enforced by `running` + queue).
31unsafe impl<T: Send> Send for AccessCell<T> {}
32unsafe impl<T: Send> Sync for AccessCell<T> {}
33
34impl<T> AccessCell<T> {
35    /// Creates a new `AccessCell` wrapping `value`.
36    pub fn new(value: T) -> Self {
37        Self {
38            value: RwLock::new(value),
39            running: AtomicBool::new(false),
40            queue: Mutex::new(VecDeque::new()),
41        }
42    }
43
44    /// Runs `f` with exclusive mutable access to the inner value.
45    ///
46    /// If called while another `access` closure is running (re-entrant or from
47    /// another thread), `f` is queued and run after the current closure and
48    /// any other queued closures finish.
49    ///
50    /// The closure must be `Send` so it can be moved into the queue when called
51    /// from another thread.
52    pub fn access(&self, f: impl FnOnce(&mut T) + Send + 'static) {
53        // If already running (this thread re-entrant or another thread), enqueue.
54        if self.running.swap(true, Ordering::Acquire) {
55            self.queue.lock().unwrap().push_back(Box::new(f));
56            return;
57        }
58
59        // We are the executor: hold the write lock for the whole run + drain.
60        let mut guard = self.value.write().unwrap();
61        f(&mut *guard);
62
63        while let Some(job) = self.queue.lock().unwrap().pop_front() {
64            job(&mut *guard);
65        }
66
67        drop(guard);
68        self.running.store(false, Ordering::Release);
69    }
70
71    /// Returns an immutable reference to the inner value.
72    ///
73    /// Blocks if an `access` closure is currently running. Safe to call from
74    /// any thread.
75    pub fn access_ref(&self) -> std::sync::RwLockReadGuard<'_, T> {
76        self.value.read().unwrap()
77    }
78}
79
80#[cfg(test)]
81mod tests {
82    use std::sync::Arc;
83    use std::thread;
84
85    #[test]
86    fn test_access() {
87        let value = Arc::new(crate::AccessCell::new(0));
88
89        value.access({
90            let value = value.clone();
91
92            move |_| {
93                // normally this would cause a deadlock
94                value.access(|v| {
95                    *v = 10;
96                })
97            }
98        });
99
100        assert_eq!(*value.access_ref(), 10);
101    }
102
103    #[test]
104    fn test_access_multithreaded() {
105        let cell = Arc::new(crate::AccessCell::new(0i32));
106        let num_threads = 8;
107        let increments_per_thread = 100;
108
109        let handles: Vec<_> = (0..num_threads)
110            .map(|_| {
111                let cell = Arc::clone(&cell);
112                thread::spawn(move || {
113                    for _ in 0..increments_per_thread {
114                        cell.access(|v| *v += 1);
115                    }
116                })
117            })
118            .collect();
119
120        for h in handles {
121            h.join().unwrap();
122        }
123
124        let expected = num_threads * increments_per_thread;
125        assert_eq!(*cell.access_ref(), expected);
126    }
127}
access_cell/lib.rs

access_cell/
lib.rs
