doom_fish_utils/
callback_context.rs

1use std::ffi::c_void;
2use std::fmt;
3use std::sync::atomic::{AtomicBool, Ordering};
4use std::sync::Arc;
5
6use crate::panic_safe::{catch_user_panic, catch_user_panic_result};
7
8struct Inner<T> {
9    active: AtomicBool,
10    value: T,
11}
12
13pub struct CallbackContext<T: Send + Sync + 'static> {
14    inner: Arc<Inner<T>>,
15}
16
17impl<T: Send + Sync + 'static> CallbackContext<T> {
18    pub const RETAIN: unsafe extern "C" fn(*mut c_void) = Self::retain;
19    pub const RELEASE: unsafe extern "C" fn(*mut c_void) = Self::release;
20
21    #[must_use]
22    pub fn new(value: T) -> Self {
23        Self {
24            inner: Arc::new(Inner {
25                active: AtomicBool::new(true),
26                value,
27            }),
28        }
29    }
30
31    #[must_use]
32    pub fn as_ptr(&self) -> *mut c_void {
33        Arc::as_ptr(&self.inner).cast_mut().cast()
34    }
35
36    #[must_use]
37    pub fn retained_ptr(&self) -> *mut c_void {
38        Arc::into_raw(Arc::clone(&self.inner)).cast_mut().cast()
39    }
40
41    pub fn deactivate(&self) {
42        self.inner.active.store(false, Ordering::Release);
43    }
44
45    #[must_use]
46    pub fn is_active(&self) -> bool {
47        self.inner.active.load(Ordering::Acquire)
48    }
49
50    #[must_use]
51    pub fn get(&self) -> &T {
52        &self.inner.value
53    }
54
55    #[allow(clippy::missing_safety_doc)]
56    pub unsafe fn with<R>(ptr: *mut c_void, site: &str, f: impl FnOnce(&T) -> R) -> Option<R> {
57        if ptr.is_null() {
58            return None;
59        }
60        let inner = unsafe { &*ptr.cast::<Inner<T>>() };
61        if !inner.active.load(Ordering::Acquire) {
62            return None;
63        }
64        catch_user_panic_result(site, || f(&inner.value))
65    }
66
67    unsafe extern "C" fn retain(ptr: *mut c_void) {
68        if !ptr.is_null() {
69            unsafe { Arc::increment_strong_count(ptr.cast::<Inner<T>>()) };
70        }
71    }
72
73    unsafe extern "C" fn release(ptr: *mut c_void) {
74        if !ptr.is_null() {
75            catch_user_panic("CallbackContext::RELEASE", || unsafe {
76                Arc::decrement_strong_count(ptr.cast::<Inner<T>>());
77            });
78        }
79    }
80}
81
82impl<T: Send + Sync + 'static> Drop for CallbackContext<T> {
83    fn drop(&mut self) {
84        self.deactivate();
85    }
86}
87
88impl<T: Send + Sync + 'static> fmt::Debug for CallbackContext<T> {
89    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
90        f.debug_struct("CallbackContext")
91            .field("active", &self.is_active())
92            .finish_non_exhaustive()
93    }
94}
95
96#[cfg(test)]
97mod tests {
98    use std::ffi::c_void;
99    use std::ptr;
100    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
101    use std::sync::{Arc, Barrier, Mutex};
102    use std::thread;
103
104    use super::{CallbackContext, Inner};
105
106    struct Counted {
107        drops: Arc<AtomicUsize>,
108        hits: AtomicUsize,
109    }
110
111    impl Drop for Counted {
112        fn drop(&mut self) {
113            self.drops.fetch_add(1, Ordering::SeqCst);
114        }
115    }
116
117    type CountedContext = CallbackContext<Counted>;
118
119    fn counted() -> (CountedContext, Arc<AtomicUsize>) {
120        let drops = Arc::new(AtomicUsize::new(0));
121        let context = CallbackContext::new(Counted {
122            drops: Arc::clone(&drops),
123            hits: AtomicUsize::new(0),
124        });
125        (context, drops)
126    }
127
128    fn hit(ptr: *mut c_void) -> Option<usize> {
129        unsafe {
130            CountedContext::with(ptr, "hit", |counted| {
131                counted.hits.fetch_add(1, Ordering::SeqCst) + 1
132            })
133        }
134    }
135
136    #[test]
137    fn retain_and_release_are_balanced() {
138        let (context, drops) = counted();
139        let ptr = context.as_ptr();
140
141        for _ in 0..8 {
142            unsafe { (CountedContext::RETAIN)(ptr) };
143        }
144        for _ in 0..8 {
145            unsafe { (CountedContext::RELEASE)(ptr) };
146        }
147
148        assert_eq!(drops.load(Ordering::SeqCst), 0);
149        assert!(context.is_active());
150        assert_eq!(hit(ptr), Some(1));
151
152        drop(context);
153        assert_eq!(drops.load(Ordering::SeqCst), 1);
154    }
155
156    #[test]
157    fn value_survives_rust_handle_drop_while_retained() {
158        let (context, drops) = counted();
159        let ptr = context.retained_ptr();
160        assert_eq!(ptr, context.as_ptr());
161        assert_eq!(hit(ptr), Some(1));
162        assert_eq!(hit(ptr), Some(2));
163
164        drop(context);
165
166        assert_eq!(drops.load(Ordering::SeqCst), 0);
167        let inner = unsafe { &*ptr.cast::<Inner<Counted>>() };
168        assert!(!inner.active.load(Ordering::SeqCst));
169        assert_eq!(inner.value.hits.load(Ordering::SeqCst), 2);
170        assert_eq!(hit(ptr), None);
171
172        unsafe { (CountedContext::RELEASE)(ptr) };
173        assert_eq!(drops.load(Ordering::SeqCst), 1);
174    }
175
176    #[test]
177    fn release_frees_the_value_at_zero() {
178        let (context, drops) = counted();
179        let first = context.retained_ptr();
180        let second = context.as_ptr();
181        unsafe { (CountedContext::RETAIN)(second) };
182
183        drop(context);
184        assert_eq!(drops.load(Ordering::SeqCst), 0);
185
186        unsafe { (CountedContext::RELEASE)(first) };
187        assert_eq!(drops.load(Ordering::SeqCst), 0);
188
189        unsafe { (CountedContext::RELEASE)(second) };
190        assert_eq!(drops.load(Ordering::SeqCst), 1);
191    }
192
193    #[test]
194    fn with_skips_null_and_deactivated_contexts() {
195        let (context, drops) = counted();
196        let called = AtomicBool::new(false);
197
198        let null = unsafe {
199            CountedContext::with(ptr::null_mut(), "null", |_| {
200                called.store(true, Ordering::SeqCst);
201            })
202        };
203        assert_eq!(null, None);
204        unsafe {
205            (CountedContext::RETAIN)(ptr::null_mut());
206            (CountedContext::RELEASE)(ptr::null_mut());
207        }
208
209        assert_eq!(hit(context.as_ptr()), Some(1));
210
211        context.deactivate();
212        assert!(!context.is_active());
213        let inactive = unsafe {
214            CountedContext::with(context.as_ptr(), "inactive", |_| {
215                called.store(true, Ordering::SeqCst);
216            })
217        };
218        assert_eq!(inactive, None);
219        assert!(!called.load(Ordering::SeqCst));
220        assert_eq!(context.get().hits.load(Ordering::SeqCst), 1);
221
222        drop(context);
223        assert_eq!(drops.load(Ordering::SeqCst), 1);
224    }
225
226    #[test]
227    fn panic_inside_callback_is_contained() {
228        let (context, _drops) = counted();
229
230        let result = unsafe {
231            CountedContext::with(context.as_ptr(), "panicking callback", |_| -> usize {
232                panic!("callback panic");
233            })
234        };
235
236        assert_eq!(result, None);
237        assert!(context.is_active());
238        assert_eq!(hit(context.as_ptr()), Some(1));
239    }
240
241    struct PanicOnDrop;
242
243    impl Drop for PanicOnDrop {
244        fn drop(&mut self) {
245            panic!("context value destructor panic");
246        }
247    }
248
249    #[test]
250    fn release_contains_a_panicking_destructor() {
251        let context = CallbackContext::new(PanicOnDrop);
252        let ptr = context.retained_ptr();
253        drop(context);
254
255        unsafe { (CallbackContext::<PanicOnDrop>::RELEASE)(ptr) };
256    }
257
258    #[test]
259    fn concurrent_with_calls_from_several_threads() {
260        const THREADS: usize = 8;
261        const CALLS: usize = 2_000;
262
263        let (context, drops) = counted();
264        let barrier = Barrier::new(THREADS);
265
266        thread::scope(|scope| {
267            for _ in 0..THREADS {
268                scope.spawn(|| {
269                    let ptr = context.retained_ptr();
270                    barrier.wait();
271                    for _ in 0..CALLS {
272                        assert!(hit(ptr).is_some());
273                    }
274                    unsafe { (CountedContext::RELEASE)(ptr) };
275                });
276            }
277        });
278
279        assert_eq!(context.get().hits.load(Ordering::SeqCst), THREADS * CALLS);
280        assert_eq!(drops.load(Ordering::SeqCst), 0);
281        drop(context);
282        assert_eq!(drops.load(Ordering::SeqCst), 1);
283    }
284
285    type Handler = Mutex<Box<dyn FnMut(u32) + Send>>;
286
287    struct ForeignOwner {
288        context: *mut c_void,
289        release: unsafe extern "C" fn(*mut c_void),
290    }
291
292    impl ForeignOwner {
293        unsafe fn register(
294            context: *mut c_void,
295            retain: unsafe extern "C" fn(*mut c_void),
296            release: unsafe extern "C" fn(*mut c_void),
297        ) -> Self {
298            unsafe { retain(context) };
299            Self { context, release }
300        }
301
302        fn deliver(&self, value: u32) {
303            unsafe { trampoline(self.context, value) };
304        }
305    }
306
307    impl Drop for ForeignOwner {
308        fn drop(&mut self) {
309            unsafe { (self.release)(self.context) };
310        }
311    }
312
313    unsafe extern "C" fn trampoline(context: *mut c_void, value: u32) {
314        let _ = unsafe {
315            CallbackContext::<Handler>::with(context, "trampoline", |handler| {
316                let mut handler = handler
317                    .lock()
318                    .unwrap_or_else(std::sync::PoisonError::into_inner);
319                handler(value);
320            })
321        };
322    }
323
324    #[test]
325    fn closure_context_follows_the_foreign_owner_lifecycle() {
326        let received = Arc::new(Mutex::new(Vec::new()));
327        let sink = Arc::clone(&received);
328        let handler: Handler = Mutex::new(Box::new(move |value| {
329            sink.lock().unwrap().push(value);
330        }));
331        let context = CallbackContext::new(handler);
332        let owner = unsafe {
333            ForeignOwner::register(
334                context.as_ptr(),
335                CallbackContext::<Handler>::RETAIN,
336                CallbackContext::<Handler>::RELEASE,
337            )
338        };
339
340        owner.deliver(1);
341        owner.deliver(2);
342        drop(context);
343        owner.deliver(3);
344        assert_eq!(Arc::strong_count(&received), 2);
345
346        drop(owner);
347        assert_eq!(Arc::strong_count(&received), 1);
348        assert_eq!(*received.lock().unwrap(), vec![1, 2]);
349    }
350}
doom_fish_utils/callback_context.rs

doom_fish_utils/
callback_context.rs
