crossbeam_epoch/
collector.rs

1/// Epoch-based garbage collector.
2///
3/// # Examples
4///
5/// ```
6/// use crossbeam_epoch::Collector;
7///
8/// let collector = Collector::new();
9///
10/// let handle = collector.register();
11/// drop(collector); // `handle` still works after dropping `collector`
12///
13/// handle.pin().flush();
14/// ```
15use core::fmt;
16
17use crate::guard::Guard;
18use crate::internal::{Global, Local};
19use crate::primitive::sync::Arc;
20
21/// An epoch-based garbage collector.
22pub struct Collector {
23    pub(crate) global: Arc<Global>,
24}
25
26unsafe impl Send for Collector {}
27unsafe impl Sync for Collector {}
28
29impl Default for Collector {
30    fn default() -> Self {
31        Self {
32            global: Arc::new(Global::new()),
33        }
34    }
35}
36
37impl Collector {
38    /// Creates a new collector.
39    pub fn new() -> Self {
40        Self::default()
41    }
42
43    /// Registers a new handle for the collector.
44    pub fn register(&self) -> LocalHandle {
45        Local::register(self)
46    }
47}
48
49impl Clone for Collector {
50    /// Creates another reference to the same garbage collector.
51    fn clone(&self) -> Self {
52        Collector {
53            global: self.global.clone(),
54        }
55    }
56}
57
58impl fmt::Debug for Collector {
59    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
60        f.pad("Collector { .. }")
61    }
62}
63
64impl PartialEq for Collector {
65    /// Checks if both handles point to the same collector.
66    fn eq(&self, rhs: &Collector) -> bool {
67        Arc::ptr_eq(&self.global, &rhs.global)
68    }
69}
70impl Eq for Collector {}
71
72/// A handle to a garbage collector.
73pub struct LocalHandle {
74    pub(crate) local: *const Local,
75}
76
77impl LocalHandle {
78    /// Pins the handle.
79    #[inline]
80    pub fn pin(&self) -> Guard {
81        unsafe { (*self.local).pin() }
82    }
83
84    /// Returns `true` if the handle is pinned.
85    #[inline]
86    pub fn is_pinned(&self) -> bool {
87        unsafe { (*self.local).is_pinned() }
88    }
89
90    /// Returns the `Collector` associated with this handle.
91    #[inline]
92    pub fn collector(&self) -> &Collector {
93        unsafe { (*self.local).collector() }
94    }
95}
96
97impl Drop for LocalHandle {
98    #[inline]
99    fn drop(&mut self) {
100        unsafe {
101            Local::release_handle(&*self.local);
102        }
103    }
104}
105
106impl fmt::Debug for LocalHandle {
107    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
108        f.pad("LocalHandle { .. }")
109    }
110}
111
112#[cfg(all(test, not(crossbeam_loom)))]
113mod tests {
114    use std::mem::ManuallyDrop;
115    use std::sync::atomic::{AtomicUsize, Ordering};
116    use std::vec::Vec;
117
118    use crossbeam_utils::thread;
119
120    use crate::{Collector, Owned};
121
122    const NUM_THREADS: usize = 8;
123
124    #[test]
125    fn pin_reentrant() {
126        let collector = Collector::new();
127        let handle = collector.register();
128        drop(collector);
129
130        assert!(!handle.is_pinned());
131        {
132            let _guard = &handle.pin();
133            assert!(handle.is_pinned());
134            {
135                let _guard = &handle.pin();
136                assert!(handle.is_pinned());
137            }
138            assert!(handle.is_pinned());
139        }
140        assert!(!handle.is_pinned());
141    }
142
143    #[test]
144    fn flush_local_bag() {
145        let collector = Collector::new();
146        let handle = collector.register();
147        drop(collector);
148
149        for _ in 0..100 {
150            let guard = &handle.pin();
151            unsafe {
152                let a = Owned::new(7).into_shared(guard);
153                guard.defer_destroy(a);
154
155                assert!(!(*guard.local).bag.with(|b| (*b).is_empty()));
156
157                while !(*guard.local).bag.with(|b| (*b).is_empty()) {
158                    guard.flush();
159                }
160            }
161        }
162    }
163
164    #[test]
165    fn garbage_buffering() {
166        let collector = Collector::new();
167        let handle = collector.register();
168        drop(collector);
169
170        let guard = &handle.pin();
171        unsafe {
172            for _ in 0..10 {
173                let a = Owned::new(7).into_shared(guard);
174                guard.defer_destroy(a);
175            }
176            assert!(!(*guard.local).bag.with(|b| (*b).is_empty()));
177        }
178    }
179
180    #[test]
181    fn pin_holds_advance() {
182        #[cfg(miri)]
183        const N: usize = 500;
184        #[cfg(not(miri))]
185        const N: usize = 500_000;
186
187        let collector = Collector::new();
188
189        thread::scope(|scope| {
190            for _ in 0..NUM_THREADS {
191                scope.spawn(|_| {
192                    let handle = collector.register();
193                    for _ in 0..N {
194                        let guard = &handle.pin();
195
196                        let before = collector.global.epoch.load(Ordering::Relaxed);
197                        collector.global.collect(guard);
198                        let after = collector.global.epoch.load(Ordering::Relaxed);
199
200                        assert!(after.wrapping_sub(before) <= 2);
201                    }
202                });
203            }
204        })
205        .unwrap();
206    }
207
208    #[cfg(not(crossbeam_sanitize))] // TODO: assertions failed due to `cfg(crossbeam_sanitize)` reduce `internal::MAX_OBJECTS`
209    #[test]
210    fn incremental() {
211        #[cfg(miri)]
212        const COUNT: usize = 500;
213        #[cfg(not(miri))]
214        const COUNT: usize = 100_000;
215        static DESTROYS: AtomicUsize = AtomicUsize::new(0);
216
217        let collector = Collector::new();
218        let handle = collector.register();
219
220        unsafe {
221            let guard = &handle.pin();
222            for _ in 0..COUNT {
223                let a = Owned::new(7i32).into_shared(guard);
224                guard.defer_unchecked(move || {
225                    drop(a.into_owned());
226                    DESTROYS.fetch_add(1, Ordering::Relaxed);
227                });
228            }
229            guard.flush();
230        }
231
232        let mut last = 0;
233
234        while last < COUNT {
235            let curr = DESTROYS.load(Ordering::Relaxed);
236            assert!(curr - last <= 1024);
237            last = curr;
238
239            let guard = &handle.pin();
240            collector.global.collect(guard);
241        }
242        assert!(DESTROYS.load(Ordering::Relaxed) == COUNT);
243    }
244
245    #[test]
246    fn buffering() {
247        const COUNT: usize = 10;
248        #[cfg(miri)]
249        const N: usize = 500;
250        #[cfg(not(miri))]
251        const N: usize = 100_000;
252        static DESTROYS: AtomicUsize = AtomicUsize::new(0);
253
254        let collector = Collector::new();
255        let handle = collector.register();
256
257        unsafe {
258            let guard = &handle.pin();
259            for _ in 0..COUNT {
260                let a = Owned::new(7i32).into_shared(guard);
261                guard.defer_unchecked(move || {
262                    drop(a.into_owned());
263                    DESTROYS.fetch_add(1, Ordering::Relaxed);
264                });
265            }
266        }
267
268        for _ in 0..N {
269            collector.global.collect(&handle.pin());
270        }
271        assert!(DESTROYS.load(Ordering::Relaxed) < COUNT);
272
273        handle.pin().flush();
274
275        while DESTROYS.load(Ordering::Relaxed) < COUNT {
276            let guard = &handle.pin();
277            collector.global.collect(guard);
278        }
279        assert_eq!(DESTROYS.load(Ordering::Relaxed), COUNT);
280    }
281
282    #[test]
283    fn count_drops() {
284        #[cfg(miri)]
285        const COUNT: usize = 500;
286        #[cfg(not(miri))]
287        const COUNT: usize = 100_000;
288        static DROPS: AtomicUsize = AtomicUsize::new(0);
289
290        struct Elem(#[allow(dead_code)] i32);
291
292        impl Drop for Elem {
293            fn drop(&mut self) {
294                DROPS.fetch_add(1, Ordering::Relaxed);
295            }
296        }
297
298        let collector = Collector::new();
299        let handle = collector.register();
300
301        unsafe {
302            let guard = &handle.pin();
303
304            for _ in 0..COUNT {
305                let a = Owned::new(Elem(7i32)).into_shared(guard);
306                guard.defer_destroy(a);
307            }
308            guard.flush();
309        }
310
311        while DROPS.load(Ordering::Relaxed) < COUNT {
312            let guard = &handle.pin();
313            collector.global.collect(guard);
314        }
315        assert_eq!(DROPS.load(Ordering::Relaxed), COUNT);
316    }
317
318    #[test]
319    fn count_destroy() {
320        #[cfg(miri)]
321        const COUNT: usize = 500;
322        #[cfg(not(miri))]
323        const COUNT: usize = 100_000;
324        static DESTROYS: AtomicUsize = AtomicUsize::new(0);
325
326        let collector = Collector::new();
327        let handle = collector.register();
328
329        unsafe {
330            let guard = &handle.pin();
331
332            for _ in 0..COUNT {
333                let a = Owned::new(7i32).into_shared(guard);
334                guard.defer_unchecked(move || {
335                    drop(a.into_owned());
336                    DESTROYS.fetch_add(1, Ordering::Relaxed);
337                });
338            }
339            guard.flush();
340        }
341
342        while DESTROYS.load(Ordering::Relaxed) < COUNT {
343            let guard = &handle.pin();
344            collector.global.collect(guard);
345        }
346        assert_eq!(DESTROYS.load(Ordering::Relaxed), COUNT);
347    }
348
349    #[test]
350    fn drop_array() {
351        const COUNT: usize = 700;
352        static DROPS: AtomicUsize = AtomicUsize::new(0);
353
354        struct Elem(#[allow(dead_code)] i32);
355
356        impl Drop for Elem {
357            fn drop(&mut self) {
358                DROPS.fetch_add(1, Ordering::Relaxed);
359            }
360        }
361
362        let collector = Collector::new();
363        let handle = collector.register();
364
365        let mut guard = handle.pin();
366
367        let mut v = Vec::with_capacity(COUNT);
368        for i in 0..COUNT {
369            v.push(Elem(i as i32));
370        }
371
372        {
373            let a = Owned::new(v).into_shared(&guard);
374            unsafe {
375                guard.defer_destroy(a);
376            }
377            guard.flush();
378        }
379
380        while DROPS.load(Ordering::Relaxed) < COUNT {
381            guard.repin();
382            collector.global.collect(&guard);
383        }
384        assert_eq!(DROPS.load(Ordering::Relaxed), COUNT);
385    }
386
387    #[test]
388    fn destroy_array() {
389        #[cfg(miri)]
390        const COUNT: usize = 500;
391        #[cfg(not(miri))]
392        const COUNT: usize = 100_000;
393        static DESTROYS: AtomicUsize = AtomicUsize::new(0);
394
395        let collector = Collector::new();
396        let handle = collector.register();
397
398        unsafe {
399            let guard = &handle.pin();
400
401            let mut v = Vec::with_capacity(COUNT);
402            for i in 0..COUNT {
403                v.push(i as i32);
404            }
405
406            let len = v.len();
407            let cap = v.capacity();
408            let ptr = ManuallyDrop::new(v).as_mut_ptr();
409            guard.defer_unchecked(move || {
410                drop(Vec::from_raw_parts(ptr, len, cap));
411                DESTROYS.fetch_add(len, Ordering::Relaxed);
412            });
413            guard.flush();
414        }
415
416        while DESTROYS.load(Ordering::Relaxed) < COUNT {
417            let guard = &handle.pin();
418            collector.global.collect(guard);
419        }
420        assert_eq!(DESTROYS.load(Ordering::Relaxed), COUNT);
421    }
422
423    #[test]
424    fn stress() {
425        const THREADS: usize = 8;
426        #[cfg(miri)]
427        const COUNT: usize = 500;
428        #[cfg(not(miri))]
429        const COUNT: usize = 100_000;
430        static DROPS: AtomicUsize = AtomicUsize::new(0);
431
432        struct Elem(#[allow(dead_code)] i32);
433
434        impl Drop for Elem {
435            fn drop(&mut self) {
436                DROPS.fetch_add(1, Ordering::Relaxed);
437            }
438        }
439
440        let collector = Collector::new();
441
442        thread::scope(|scope| {
443            for _ in 0..THREADS {
444                scope.spawn(|_| {
445                    let handle = collector.register();
446                    for _ in 0..COUNT {
447                        let guard = &handle.pin();
448                        unsafe {
449                            let a = Owned::new(Elem(7i32)).into_shared(guard);
450                            guard.defer_destroy(a);
451                        }
452                    }
453                });
454            }
455        })
456        .unwrap();
457
458        let handle = collector.register();
459        while DROPS.load(Ordering::Relaxed) < COUNT * THREADS {
460            let guard = &handle.pin();
461            collector.global.collect(guard);
462        }
463        assert_eq!(DROPS.load(Ordering::Relaxed), COUNT * THREADS);
464    }
465}
crossbeam_epoch/collector.rs

crossbeam_epoch/
collector.rs
