aft/inspect/
dispatch.rs

1use std::sync::atomic::{AtomicUsize, Ordering};
2use std::sync::{Arc, LazyLock};
3use std::thread;
4use std::time::Instant;
5
6use crossbeam_channel::{unbounded, Receiver, Sender};
7
8use super::job::{InspectCategory, InspectJob, InspectResult};
9
10pub type InspectWorker = Arc<dyn Fn(InspectJob) -> InspectResult + Send + Sync + 'static>;
11
12#[derive(Clone)]
13pub struct DispatchHandles {
14    pub request_tx: Sender<InspectJob>,
15    pub result_rx: Receiver<InspectResult>,
16    pub pool: Arc<rayon::ThreadPool>,
17}
18
19/// Number of live workers in the process-wide inspect pool. The pool's start
20/// and exit handlers maintain this counter for platform-independent diagnostics.
21static INSPECT_THREAD_COUNT: AtomicUsize = AtomicUsize::new(0);
22
23/// The inspect pool is process-wide because roots are independently evictable
24/// while their scans still share the daemon process. Inspect mixes file IO and
25/// parsing; `min(8, available_parallelism)` is the measured point where more
26/// workers stop improving the supported repositories while leaving cores for
27/// interactive lanes.
28///
29/// `ColdBuildLimiter` remains separate: it admits concurrent heavy operations
30/// (including Tier-2 scans), while this pool bounds the threads those admitted
31/// operations can use. The pool is intentionally process-lifetime; dropping a
32/// root's `InspectManager` only releases that root's handle to this pool.
33static INSPECT_POOL: LazyLock<Arc<rayon::ThreadPool>> = LazyLock::new(|| {
34    Arc::new(
35        rayon::ThreadPoolBuilder::new()
36            .num_threads(default_pool_size())
37            .thread_name(|index| format!("aft-inspect-{index}"))
38            // Rayon defaults workers to ~2MB stacks (vs the main thread's 8MB).
39            // The duplicates scanner walks the AST recursively, and deep trees
40            // (minified bundles, generated code, long chains) previously
41            // overflowed a 2MB worker stack and SIGABRT'd the whole bridge.
42            // Match the main thread's 8MB so the bounded recursion in
43            // collect_fragments (MAX_FRAGMENT_DEPTH) has comfortable headroom.
44            .stack_size(8 * 1024 * 1024)
45            .start_handler(|_| {
46                INSPECT_THREAD_COUNT.fetch_add(1, Ordering::SeqCst);
47            })
48            .exit_handler(|_| {
49                INSPECT_THREAD_COUNT.fetch_sub(1, Ordering::SeqCst);
50            })
51            .build()
52            .expect("inspect worker pool must build"),
53    )
54});
55
56pub fn start_dispatch_loop(worker: InspectWorker) -> DispatchHandles {
57    let (request_tx, request_rx) = unbounded::<InspectJob>();
58    let (result_tx, result_rx) = unbounded::<InspectResult>();
59    let pool = inspect_pool();
60
61    let loop_pool = Arc::clone(&pool);
62    thread::spawn(move || dispatch_loop(request_rx, result_tx, loop_pool, worker));
63
64    DispatchHandles {
65        request_tx,
66        result_rx,
67        pool,
68    }
69}
70
71fn inspect_pool() -> Arc<rayon::ThreadPool> {
72    Arc::clone(&INSPECT_POOL)
73}
74
75#[doc(hidden)]
76pub fn inspect_pool_size_for_test() -> usize {
77    inspect_pool().current_num_threads()
78}
79
80#[doc(hidden)]
81pub fn inspect_pool_thread_count_for_test() -> usize {
82    INSPECT_THREAD_COUNT.load(Ordering::SeqCst)
83}
84
85pub fn default_worker() -> InspectWorker {
86    Arc::new(dispatch_category)
87}
88
89fn dispatch_loop(
90    request_rx: Receiver<InspectJob>,
91    result_tx: Sender<InspectResult>,
92    pool: Arc<rayon::ThreadPool>,
93    worker: InspectWorker,
94) {
95    while let Ok(job) = request_rx.recv() {
96        let tx = result_tx.clone();
97        let worker = Arc::clone(&worker);
98        pool.spawn_fifo(move || {
99            let result = worker(job);
100            let _ = tx.send(result);
101        });
102    }
103}
104
105fn dispatch_category(job: InspectJob) -> InspectResult {
106    use crate::inspect::scanners;
107
108    match job.category {
109        InspectCategory::Todos => scanners::todos::run_todos_scan(&job),
110        InspectCategory::Metrics => scanners::metrics::run_metrics_scan(&job),
111        InspectCategory::DeadCode => scanners::dead_code::run_dead_code_scan(&job),
112        InspectCategory::UnusedExports => scanners::unused_exports::run_unused_exports_scan(&job),
113        InspectCategory::Duplicates => scanners::duplicates::run_duplicates_scan(&job),
114        InspectCategory::Cycles => scanners::cycles::run_cycles_scan(&job),
115        InspectCategory::Complexity => scanners::complexity::run_complexity_scan(&job),
116        InspectCategory::Diagnostics => {
117            // Diagnostics are backed by the AppContext LSP manager and run via
118            // the serial LSP/status lane in `handle_inspect` — never through
119            // this rayon worker pool. Reaching this arm means a caller routed
120            // Diagnostics into the worker path incorrectly; surface that as a
121            // routing bug instead of a misleading "pending" status.
122            let started = Instant::now();
123            InspectResult::failed(
124                &job,
125                "diagnostics must run on the main thread (run_diagnostics_category), \
126                 not the rayon inspect worker pool",
127                started.elapsed(),
128            )
129        }
130        other => {
131            let started = Instant::now();
132            InspectResult::failed(
133                &job,
134                format!("inspect category '{other}' is not active in v0.33"),
135                started.elapsed(),
136            )
137        }
138    }
139}
140
141fn default_pool_size() -> usize {
142    // Dev-only override for reproducing thread-regime-dependent behaviour
143    // (glibc allocates arenas per contending thread, so pool width changes
144    // fragmentation). Not a tuning surface and deliberately undocumented.
145    resolve_pool_size(std::env::var("AFT_INSPECT_POOL_THREADS").ok().as_deref())
146}
147
148/// Split from the env read so the parsing and clamping are testable without
149/// mutating process-global state: `INSPECT_POOL` is a `LazyLock`, and any
150/// concurrent test that builds an `InspectManager` would otherwise capture
151/// whatever width the env happened to hold.
152fn resolve_pool_size(override_value: Option<&str>) -> usize {
153    if let Some(threads) = override_value.and_then(|value| value.parse::<usize>().ok()) {
154        return threads.clamp(1, 512);
155    }
156
157    std::thread::available_parallelism()
158        .map(|parallelism| parallelism.get())
159        .unwrap_or(1)
160        .min(8)
161}
162
163#[cfg(test)]
164mod tests {
165    use super::resolve_pool_size;
166
167    #[test]
168    fn pool_thread_override_wins_and_clamps() {
169        assert_eq!(resolve_pool_size(Some("17")), 17);
170        assert_eq!(resolve_pool_size(Some("0")), 1);
171        assert_eq!(resolve_pool_size(Some("100000")), 512);
172    }
173
174    #[test]
175    fn pool_thread_override_ignores_absent_and_unparseable_values() {
176        let derived = resolve_pool_size(None);
177
178        assert_eq!(resolve_pool_size(Some("wide")), derived);
179        assert_eq!(resolve_pool_size(Some("-4")), derived);
180        assert_eq!(resolve_pool_size(Some("")), derived);
181    }
182}
aft/inspect/dispatch.rs

aft/inspect/
dispatch.rs
