videre_core/
semaphore.rs

1use std::sync::{Condvar, Mutex};
2
3/// A tiny counting semaphore for bounding how many callers run a section of
4/// code concurrently. Used to cap concurrent `qlmanage` subprocess launches:
5/// QuickLook's thumbnail-generation agent is a shared per-user service that
6/// doesn't parallelize well, so letting many callers (e.g. a UI rendering
7/// hundreds of HEIC thumbnails at once) spawn `qlmanage` unbounded causes the
8/// agent, and the source drive's I/O, to queue up and occasionally exceed
9/// even a generous per-call timeout.
10pub struct Semaphore {
11    state: Mutex<usize>,
12    cond: Condvar,
13    max: usize,
14}
15
16pub struct SemaphorePermit<'a> {
17    sem: &'a Semaphore,
18}
19
20impl Drop for SemaphorePermit<'_> {
21    fn drop(&mut self) {
22        let mut count = self.sem.state.lock().unwrap();
23        *count -= 1;
24        self.sem.cond.notify_one();
25    }
26}
27
28impl Semaphore {
29    pub fn new(max: usize) -> Self {
30        Semaphore { state: Mutex::new(0), cond: Condvar::new(), max }
31    }
32
33    /// Blocks until fewer than `max` permits are held, then takes one.
34    /// Released automatically when the returned guard drops.
35    pub fn acquire(&self) -> SemaphorePermit<'_> {
36        let mut count = self.state.lock().unwrap();
37        while *count >= self.max {
38            count = self.cond.wait(count).unwrap();
39        }
40        *count += 1;
41        SemaphorePermit { sem: self }
42    }
43}
44
45#[cfg(test)]
46mod tests {
47    use super::*;
48    use std::sync::atomic::{AtomicUsize, Ordering};
49    use std::sync::Arc;
50    use std::thread;
51    use std::time::Duration;
52
53    #[test]
54    fn never_exceeds_max_concurrent_holders() {
55        let sem = Arc::new(Semaphore::new(2));
56        let current = Arc::new(AtomicUsize::new(0));
57        let max_observed = Arc::new(AtomicUsize::new(0));
58
59        let handles: Vec<_> = (0..8)
60            .map(|_| {
61                let sem = Arc::clone(&sem);
62                let current = Arc::clone(&current);
63                let max_observed = Arc::clone(&max_observed);
64                thread::spawn(move || {
65                    let _permit = sem.acquire();
66                    let now = current.fetch_add(1, Ordering::SeqCst) + 1;
67                    max_observed.fetch_max(now, Ordering::SeqCst);
68                    thread::sleep(Duration::from_millis(20));
69                    current.fetch_sub(1, Ordering::SeqCst);
70                })
71            })
72            .collect();
73        for h in handles {
74            h.join().unwrap();
75        }
76
77        assert!(max_observed.load(Ordering::SeqCst) <= 2);
78    }
79}
videre_core/semaphore.rs

videre_core/
semaphore.rs
