asc_cache/threads.rs
1//! How many worker threads a search takes when its caller fixed none: the
2//! CPUs of this machine split between this process and the user's other
3//! busy registered ones ([`crate::ProcessRegistry`]). One policy for the DEX
4//! and the HarmonyOS engine, so a laptop running both is not oversubscribed.
5
6use crate::{LockExt, ProcessRegistry};
7use std::sync::Mutex;
8use std::time::{Duration, Instant};
9
10/// Busy for [`auto_threads`], for a registry entry that does not say (its
11/// process predates the busy flag): one updated this recently. Memory
12/// caches republish at most every few seconds while they load entries, and
13/// not at all when idle (an MCP server waiting for an agent's next call).
14const BUSY_WINDOW: Duration = Duration::from_secs(5);
15
16/// How often [`auto_threads`] re-reads the registry: a burst of scans
17/// started together by several agents lasts about a second. A read checks
18/// every entry's process: about 1 ms for 31 servers on macOS, which callers
19/// wait for (letting them go on with the last count made 16 servers' cold
20/// sweep 10 % slower).
21const THREADS_EVERY: Duration = Duration::from_millis(100);
22
23/// Workers the busy processes may start together, per CPU. An exact split
24/// (1) leaves CPUs idle while the processes holding the largest APKs finish
25/// alone: 8 servers scanning 312 APKs on 16 CPUs took 223 ms with 2 workers
26/// each, 180 ms with 8 and 176 ms with 16, for 13 % more memory at 8.
27const OVERSUBSCRIBE: usize = 4;
28
29/// How often a worker [`Threads::wait_turn`] holds back looks again.
30const TURN_POLL: Duration = Duration::from_millis(10);
31
32/// Worker threads of a search whose caller fixed none: this machine's CPUs
33/// split between this process and the user's other busy registered ones
34/// ([`auto_threads_for`]). Re-read at most every 100 ms, so
35/// a sweep follows the others starting and finishing their scans
36/// ([`Threads::wait_turn`]). A process that does not register itself (a
37/// one-shot command) still gives way to the resident hosts.
38pub fn auto_threads() -> usize {
39 static LAST: Mutex<Option<(Instant, usize)>> = Mutex::new(None);
40 let mut last = LAST.lock_or_default();
41 if let Some((at, threads)) = *last
42 && at.elapsed() < THREADS_EVERY
43 {
44 return threads;
45 }
46 let others = ProcessRegistry::default_dir().map(|dir| ProcessRegistry::others_in(&dir)).unwrap_or_default();
47 let busy = others.iter().filter(|other| other.busy.unwrap_or(other.idle < BUSY_WINDOW)).count();
48 let threads = auto_threads_for(cpus(), busy);
49 *last = Some((Instant::now(), threads));
50 threads
51}
52
53/// This machine's CPUs.
54pub fn cpus() -> usize {
55 std::thread::available_parallelism().map_or(8, |n| n.get())
56}
57
58/// 4 x `cpus` workers split evenly between this process and
59/// `busy_others`, at most `cpus` and at least 2 (where there are 2): a scan
60/// on one thread gives up its overlap of inflating one entry while searching
61/// another. Up to 4 busy processes each keep every CPU.
62pub fn auto_threads_for(cpus: usize, busy_others: usize) -> usize {
63 (cpus * OVERSUBSCRIBE / (busy_others + 1)).clamp(cpus.min(2), cpus)
64}
65
66/// Worker threads of a search.
67#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
68pub enum Threads {
69 /// [`auto_threads`], read again by every search.
70 #[default]
71 Auto,
72 /// Exactly this many workers (0 counts as 1).
73 Fixed(usize),
74}
75
76impl Threads {
77 /// The thread count for a search starting now (at least 1).
78 pub fn count(self) -> usize {
79 match self {
80 Threads::Auto => auto_threads(),
81 Threads::Fixed(n) => n.max(1),
82 }
83 }
84
85 /// The most workers [`Threads::wait_turn`] ever lets run at once.
86 pub fn most(self) -> usize {
87 match self {
88 Threads::Auto => cpus(),
89 Threads::Fixed(n) => n.max(1),
90 }
91 }
92
93 /// For a long search that starts [`Threads::most`] workers: hold worker
94 /// `k` back (between units of work) while [`auto_threads`] allows fewer
95 /// than `k + 1`, or until `stop()`. The count shrinks as other processes
96 /// start scans and grows back as they finish, which a count read once at
97 /// the start cannot follow: every process of a burst would see the others
98 /// idle and take every CPU.
99 pub fn wait_turn(self, k: usize, stop: impl Fn() -> bool) {
100 if self == Threads::Auto {
101 while k >= auto_threads() && !stop() {
102 std::thread::sleep(TURN_POLL);
103 }
104 }
105 }
106}
107
108impl From<usize> for Threads {
109 fn from(n: usize) -> Threads {
110 Threads::Fixed(n)
111 }
112}
113
114/// `None` (no `--threads`) is [`Threads::Auto`].
115impl From<Option<usize>> for Threads {
116 fn from(n: Option<usize>) -> Threads {
117 n.map_or(Threads::Auto, Threads::Fixed)
118 }
119}
120
121#[cfg(test)]
122mod tests {
123 use super::*;
124 use std::sync::atomic::{AtomicUsize, Ordering};
125
126 #[test]
127 fn threads_split_between_busy_processes() {
128 assert_eq!(auto_threads_for(32, 0), 32);
129 assert_eq!(auto_threads_for(32, 3), 32, "4 busy processes keep every CPU");
130 assert_eq!(auto_threads_for(32, 7), 16);
131 assert_eq!(auto_threads_for(32, 15), 8);
132 assert_eq!(auto_threads_for(32, 255), 2, "never below 2");
133 assert_eq!(auto_threads_for(1, 3), 1, "nor above the CPUs");
134 assert!(auto_threads() >= 1);
135 }
136
137 #[test]
138 fn workers_wait_for_their_turn() {
139 Threads::Auto.wait_turn(0, || panic!("the first worker always runs"));
140 Threads::Fixed(1).wait_turn(5, || panic!("fixed counts never hold a worker back"));
141 let asked = AtomicUsize::new(0);
142 // Past every CPU a worker waits until the search tells it to stop.
143 Threads::Auto.wait_turn(Threads::Auto.most(), || asked.fetch_add(1, Ordering::Relaxed) >= 2);
144 assert_eq!(asked.load(Ordering::Relaxed), 3);
145 assert_eq!((Threads::from(Some(3)), Threads::from(None)), (Threads::Fixed(3), Threads::Auto));
146 }
147}