running_process_platform_internal/
platform_linux_descendants.rs

1//! Linux implementation of launched-tree descendant monitoring.
2
3use std::collections::HashMap;
4use std::sync::Arc;
5use std::time::Duration;
6
7use crate::platform::process::{DescendantEvent, DescendantMonitorStop};
8
9const POLL_INTERVAL: Duration = Duration::from_millis(20);
10
11#[derive(Clone, Copy, Debug, Eq, PartialEq)]
12struct ProcessIdentity {
13    start_ticks: u64,
14}
15
16pub fn start_descendant_monitor(
17    root_pid: u32,
18    stop: Arc<DescendantMonitorStop>,
19    emit: Box<dyn Fn(DescendantEvent) + Send>,
20) -> std::io::Result<()> {
21    let Some(root_identity) = process_identity(root_pid) else {
22        emit(DescendantEvent::Completed);
23        return Ok(());
24    };
25    enable_subreaper();
26    std::thread::Builder::new()
27        .name("rp-linux-descpump".to_string())
28        .spawn(move || pump_loop(root_pid, root_identity, stop, emit))
29        .map(|_| ())
30        .map_err(|error| std::io::Error::other(format!("spawn descendant monitor: {error}")))
31}
32
33fn enable_subreaper() {
34    // Failure is intentionally best-effort: the monitor can still track a
35    // descendant while its immediate parent remains alive.
36    let _ = unsafe { libc::prctl(libc::PR_SET_CHILD_SUBREAPER, 1, 0, 0, 0) };
37}
38
39fn parse_start_ticks(stat: &str) -> Option<u64> {
40    let suffix = stat.get(stat.rfind(')')? + 1..)?;
41    suffix
42        .split_ascii_whitespace()
43        .nth(19)
44        .and_then(|field| field.parse().ok())
45}
46
47fn process_identity(pid: u32) -> Option<ProcessIdentity> {
48    let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
49    Some(ProcessIdentity {
50        start_ticks: parse_start_ticks(&stat)?,
51    })
52}
53
54/// Map of live descendant pid -> immediate parent pid. The parent is a
55/// free by-product of the `children`-file walk: the pid whose children
56/// file listed the entry.
57fn descendant_pids(root_pid: u32) -> HashMap<u32, u32> {
58    let mut result = HashMap::new();
59    let mut stack = vec![root_pid];
60    while let Some(pid) = stack.pop() {
61        let path = format!("/proc/{pid}/task/{pid}/children");
62        let Ok(contents) = std::fs::read_to_string(path) else {
63            continue;
64        };
65        for token in contents.split_ascii_whitespace() {
66            if let Ok(child) = token.parse::<u32>() {
67                if result.insert(child, pid).is_none() {
68                    stack.push(child);
69                }
70            }
71        }
72    }
73    result
74}
75
76fn snapshot(root_pid: u32, expected: ProcessIdentity) -> Option<HashMap<u32, u32>> {
77    let before = process_identity(root_pid);
78    let descendants = descendant_pids(root_pid);
79    verified_snapshot(
80        expected,
81        before,
82        descendants,
83        process_identity(root_pid),
84    )
85}
86
87fn verified_snapshot(
88    expected: ProcessIdentity,
89    before: Option<ProcessIdentity>,
90    descendants: HashMap<u32, u32>,
91    after: Option<ProcessIdentity>,
92) -> Option<HashMap<u32, u32>> {
93    (before == Some(expected) && after == Some(expected)).then_some(descendants)
94}
95
96fn emit_diff(
97    previous: &HashMap<u32, u32>,
98    current: &HashMap<u32, u32>,
99    emit: &dyn Fn(DescendantEvent),
100) {
101    for (&pid, &parent_pid) in current {
102        if !previous.contains_key(&pid) {
103            emit(DescendantEvent::Started {
104                pid,
105                parent_pid: Some(parent_pid),
106            });
107        }
108    }
109    for &pid in previous.keys() {
110        if !current.contains_key(&pid) {
111            emit(DescendantEvent::Exited(pid));
112        }
113    }
114}
115
116fn pump_loop_with(
117    stop: &DescendantMonitorStop,
118    mut take_snapshot: impl FnMut() -> Option<HashMap<u32, u32>>,
119    emit: &dyn Fn(DescendantEvent),
120    mut wait: impl FnMut() -> bool,
121) {
122    let mut known = HashMap::new();
123    loop {
124        if stop.is_stopped() {
125            emit(DescendantEvent::Completed);
126            return;
127        }
128        let Some(current) = take_snapshot() else {
129            for pid in known.into_keys() {
130                emit(DescendantEvent::Exited(pid));
131            }
132            emit(DescendantEvent::Completed);
133            return;
134        };
135        emit_diff(&known, &current, emit);
136        known = current;
137        if wait() {
138            emit(DescendantEvent::Completed);
139            return;
140        }
141    }
142}
143
144fn pump_loop(
145    root_pid: u32,
146    root_identity: ProcessIdentity,
147    stop: Arc<DescendantMonitorStop>,
148    emit: Box<dyn Fn(DescendantEvent) + Send>,
149) {
150    pump_loop_with(
151        &stop,
152        || snapshot(root_pid, root_identity),
153        emit.as_ref(),
154        || stop.wait_timeout(POLL_INTERVAL),
155    );
156}
157
158#[cfg(test)]
159mod tests {
160    use super::*;
161    use std::collections::HashSet;
162    use std::sync::mpsc;
163
164    fn collect_diff(
165        previous: &HashMap<u32, u32>,
166        current: &HashMap<u32, u32>,
167    ) -> Vec<DescendantEvent> {
168        let (tx, rx) = mpsc::channel();
169        emit_diff(previous, current, &|event| tx.send(event).unwrap());
170        drop(tx);
171        rx.iter().collect()
172    }
173
174    #[test]
175    fn descendant_sampling_cadence_is_twenty_milliseconds() {
176        assert_eq!(POLL_INTERVAL, Duration::from_millis(20));
177    }
178
179    #[test]
180    fn emit_diff_fires_one_started_per_new_pid_with_its_parent() {
181        let previous = [(10, 1), (20, 10)].into_iter().collect();
182        let current = [(10, 1), (20, 10), (30, 20), (40, 10)].into_iter().collect();
183        let events = collect_diff(&previous, &current);
184        let started: HashMap<_, _> = events
185            .into_iter()
186            .filter_map(|event| match event {
187                DescendantEvent::Started { pid, parent_pid } => Some((pid, parent_pid)),
188                DescendantEvent::Exited(_) | DescendantEvent::Completed => None,
189            })
190            .collect();
191        assert_eq!(
192            started,
193            [(30, Some(20)), (40, Some(10))].into_iter().collect()
194        );
195    }
196
197    #[test]
198    fn emit_diff_fires_one_exited_per_gone_pid() {
199        let previous = [(10, 1), (20, 10), (30, 20)].into_iter().collect();
200        let current = [(10, 1)].into_iter().collect();
201        let events = collect_diff(&previous, &current);
202        let exited: HashSet<_> = events
203            .into_iter()
204            .filter_map(|event| match event {
205                DescendantEvent::Exited(pid) => Some(pid),
206                DescendantEvent::Started { .. } | DescendantEvent::Completed => None,
207            })
208            .collect();
209        assert_eq!(exited, [20, 30].into_iter().collect());
210    }
211
212    #[test]
213    fn emit_diff_no_events_when_steady_state() {
214        let current = [(10, 1), (20, 10)].into_iter().collect();
215        assert!(collect_diff(&current, &current).is_empty());
216    }
217
218    #[test]
219    fn descendant_pids_for_nonexistent_root_returns_empty() {
220        assert!(descendant_pids(0x7fff_fffe).is_empty());
221    }
222
223    #[test]
224    fn descendant_pids_for_self_includes_no_phantom_entries() {
225        assert!(descendant_pids(std::process::id())
226            .into_iter()
227            .all(|(pid, parent_pid)| pid > 1 && parent_pid > 0));
228    }
229
230    #[test]
231    fn parse_start_ticks_handles_spaces_and_parentheses_in_comm() {
232        let mut fields = vec!["S".to_string()];
233        fields.extend((4..=21).map(|number| number.to_string()));
234        fields.push("424242".to_string());
235        let stat = format!("123 (odd ) process name) {}", fields.join(" "));
236        assert_eq!(parse_start_ticks(&stat), Some(424242));
237    }
238
239    #[test]
240    fn identity_mismatch_terminates_pump_without_tracking_reused_pid() {
241        let stop = DescendantMonitorStop::new();
242        let (tx, rx) = mpsc::channel();
243        let mut polls = 0;
244        pump_loop_with(
245            &stop,
246            || {
247                polls += 1;
248                None
249            },
250            &|event| tx.send(event).unwrap(),
251            || panic!("terminated pump must not wait"),
252        );
253        assert_eq!(polls, 1);
254        assert_eq!(rx.try_recv(), Ok(DescendantEvent::Completed));
255        assert!(rx.try_recv().is_err());
256    }
257
258    #[test]
259    fn reused_pid_with_different_start_ticks_rejects_descendant_snapshot() {
260        let expected = ProcessIdentity { start_ticks: 100 };
261        let recycled = ProcessIdentity { start_ticks: 200 };
262        assert_eq!(
263            verified_snapshot(
264                expected,
265                Some(recycled),
266                [(42, 7)].into_iter().collect(),
267                Some(recycled),
268            ),
269            None
270        );
271    }
272
273    #[test]
274    fn scripted_normal_pump_emits_descendant_start_and_exit() {
275        let stop = DescendantMonitorStop::new();
276        let (tx, rx) = mpsc::channel();
277        let mut snapshots = [
278            Some([(42, 7)].into_iter().collect()),
279            Some(HashMap::new()),
280            None,
281        ]
282        .into_iter();
283        pump_loop_with(
284            &stop,
285            || snapshots.next().flatten(),
286            &|event| tx.send(event).unwrap(),
287            || false,
288        );
289        drop(tx);
290        assert_eq!(
291            rx.iter().collect::<Vec<_>>(),
292            [
293                DescendantEvent::Started {
294                    pid: 42,
295                    parent_pid: Some(7),
296                },
297                DescendantEvent::Exited(42),
298                DescendantEvent::Completed,
299            ]
300        );
301    }
302
303    #[test]
304    fn stop_wakes_waiting_pump_without_polling() {
305        let stop = Arc::new(DescendantMonitorStop::new());
306        let pump_stop = Arc::clone(&stop);
307        let (waiting_tx, waiting_rx) = mpsc::channel();
308        let (done_tx, done_rx) = mpsc::channel();
309        let (event_tx, event_rx) = mpsc::channel();
310        let pump = std::thread::spawn(move || {
311            let mut announced = false;
312            pump_loop_with(
313                &pump_stop,
314                || {
315                    if !announced {
316                        announced = true;
317                        waiting_tx.send(()).unwrap();
318                    }
319                    Some(HashMap::new())
320                },
321                &|event| event_tx.send(event).unwrap(),
322                || pump_stop.wait_timeout(Duration::from_secs(30)),
323            );
324            done_tx.send(()).unwrap();
325        });
326        waiting_rx.recv_timeout(Duration::from_secs(1)).unwrap();
327        stop.stop();
328        done_rx.recv_timeout(Duration::from_millis(250)).unwrap();
329        pump.join().unwrap();
330        assert_eq!(event_rx.try_recv(), Ok(DescendantEvent::Completed));
331        assert!(event_rx.try_recv().is_err());
332    }
333}
running_process_platform_internal/platform_linux_descendants.rs

running_process_platform_internal/
platform_linux_descendants.rs
