diff --git a/crates/maestro-daemon/src/proc.rs b/crates/maestro-daemon/src/proc.rs index 5f7942c..28cba42 100644 --- a/crates/maestro-daemon/src/proc.rs +++ b/crates/maestro-daemon/src/proc.rs @@ -1,8 +1,8 @@ //! Best-effort `/proc` inspection for the activity watcher (NAP-340, Linux). //! //! Decides whether a session is **blocked waiting on the user**: ask the kernel -//! which process group owns the terminal, then check that process is parked -//! waiting on terminal input. +//! which process group owns the terminal, then check whether any live member of +//! that group is parked on the terminal. //! //! `/proc//syscall` would say so exactly, but it's ptrace-gated and //! unreadable for our own sessions on a default kernel, so we combine three @@ -16,97 +16,81 @@ //! Every step degrades to `false` on a read failure, so a non-Linux host or a //! denied `/proc` read falls back to `Idle`. +use std::collections::HashMap; + +/// One walk of `/proc`, shared by every session in an activity pass. +/// +/// The foreground process group is the unit that matters, and its members are +/// not reliably reachable from the pane: the group leader can exit while others +/// run on, and a stage that forks a terminal-reading child and exits leaves +/// that child reparented to init. Walking the pane's process tree misses both. +/// +/// Snapshotting once per pass costs one `/proc` walk however many sessions the +/// daemon owns — cheaper than the per-session `tmux list-panes` fork that +/// already runs on the same tick. +#[derive(Debug, Default)] +pub struct ProcGroups { + stats: HashMap, + by_pgid: HashMap>, +} + +impl ProcGroups { + pub fn snapshot() -> Self { + let mut stats = HashMap::new(); + let mut by_pgid: HashMap> = HashMap::new(); + let Ok(entries) = std::fs::read_dir("/proc") else { + return Self::default(); + }; + for pid in entries + .flatten() + .filter_map(|e| e.file_name().to_str()?.parse::().ok()) + { + // A process can exit mid-walk; skipping it is the correct answer. + if let Some(stat) = read_stat(pid) { + by_pgid.entry(stat.pgrp).or_default().push(pid); + stats.insert(pid, stat); + } + } + Self { stats, by_pgid } + } + + /// The process group the terminal considers foreground, per `tpgid`. + fn foreground_pgid(&self, pane_pid: i32) -> Option { + let tpgid = self.stats.get(&pane_pid)?.tpgid; + // -1 when the process has no controlling terminal. + (tpgid > 0).then_some(tpgid) + } + + fn members(&self, pgid: i32) -> &[i32] { + self.by_pgid.get(&pgid).map_or(&[], Vec::as_slice) + } + + fn state(&self, pid: i32) -> Option { + self.stats.get(&pid).map(|s| s.state) + } +} + /// Whether the session on `tty` (e.g. `/dev/pts/3`) is blocked waiting on the /// user — the signal for `AwaitingInput`. -pub fn awaiting_input(pane_pid: i32, tty: &str) -> bool { - let Some(pgid) = foreground_pgid(pane_pid) else { +/// +/// True when *any* member of the terminal's foreground group is parked on it: a +/// group can hold a busy or timer-sleeping stage alongside the one actually +/// reading, and either can be the reparented one. +pub fn awaiting_input(pane_pid: i32, tty: &str, groups: &ProcGroups) -> bool { + let Some(pgid) = groups.foreground_pgid(pane_pid) else { return false; }; - foreground_members(pane_pid, pgid) - .into_iter() - .any(|pid| parked_on_tty(pid, tty)) -} - -/// The process group the terminal considers foreground. -/// -/// `tpgid` is the kernel's own answer, which is why this replaced walking -/// `children` to the deepest descendant: that guessed the newest child was the -/// foreground one, so an agent with a background helper looked idle. -fn foreground_pgid(pane_pid: i32) -> Option { - let tpgid = read_stat(pane_pid)?.tpgid; - // -1 when the process has no controlling terminal. - (tpgid > 0).then_some(tpgid) -} - -/// Cap on processes examined per pass. A session's tree is a shell plus a -/// handful of children; this only bounds a pathological one. -const MAX_TREE_NODES: usize = 256; - -/// Live members of process group `pgid`. -/// -/// `pgid` names a *group*, not a process, so `/proc/` may not exist: the -/// leader can exit while another member still holds the terminal. -/// -/// The pane's subtree answers this in the normal case and costs a handful of -/// reads. A member can still end up outside it — a stage that forks a -/// terminal-reading child and exits leaves that child reparented to init while -/// it keeps the foreground `pgid` — so an empty subtree result falls back to -/// scanning `/proc`. That scan only runs when the cheap path found nothing. -fn foreground_members(pane_pid: i32, pgid: i32) -> Vec { - let in_subtree = subtree_members(pane_pid, pgid); - if !in_subtree.is_empty() { - return in_subtree; - } - scan_for_pgid(pgid) -} - -/// Members of `pgid` among `pane_pid` and its descendants. -fn subtree_members(pane_pid: i32, pgid: i32) -> Vec { - let mut found = Vec::new(); - let mut queue = vec![pane_pid]; - let mut visited = 0usize; - while let Some(pid) = queue.pop() { - visited += 1; - if visited > MAX_TREE_NODES { - break; - } - if read_stat(pid).is_some_and(|s| s.pgrp == pgid) { - found.push(pid); - } - queue.extend(children(pid)); - } - found -} - -/// Every live process in `pgid`, by walking `/proc`. The fallback for a member -/// that has been reparented out of the pane's subtree. -fn scan_for_pgid(pgid: i32) -> Vec { - let Ok(entries) = std::fs::read_dir("/proc") else { - return Vec::new(); - }; - entries - .flatten() - .filter_map(|e| e.file_name().to_str()?.parse::().ok()) - .filter(|pid| read_stat(*pid).is_some_and(|s| s.pgrp == pgid)) - .collect() -} - -/// Direct children of `pid`, via `/proc//task//children`. -fn children(pid: i32) -> Vec { - std::fs::read_to_string(format!("/proc/{pid}/task/{pid}/children")) - .map(|s| { - s.split_whitespace() - .filter_map(|x| x.parse().ok()) - .collect() - }) - .unwrap_or_default() + groups + .members(pgid) + .iter() + .any(|pid| parked_on_tty(*pid, tty, groups)) } /// Whether `pid` is parked on `tty` rather than busy or sleeping on a timer. -fn parked_on_tty(pid: i32, tty: &str) -> bool { +fn parked_on_tty(pid: i32, tty: &str, groups: &ProcGroups) -> bool { // Interruptible sleep is where a blocking read parks; anything else // (running, uninterruptible) is busy, not waiting on us. - if read_stat(pid).map(|s| s.state) != Some('S') { + if groups.state(pid) != Some('S') { return false; } // A pure timer sleep (`sleep`, `usleep`) also sits in 'S' with the terminal @@ -234,31 +218,46 @@ mod tests { assert!(!is_timer_wchan("0")); } - /// The fallback used when no member of the foreground group is in the - /// pane's subtree. Exercised against this process's own group, which is - /// guaranteed to have at least one live member. + /// The snapshot indexes by process group, not by parentage — that is the + /// whole point, since a group member need not be under the pane. Checked + /// against this process's own group, which certainly has a live member. #[test] - fn scan_for_pgid_finds_this_process() { + fn snapshot_indexes_this_process_under_its_group() { // SAFETY: argument-less query of the caller's own process group. let own = unsafe { libc::getpgid(0) }; assert!(own > 0, "expected a real process group, got {own}"); - let members = scan_for_pgid(own); + let groups = ProcGroups::snapshot(); let me = std::process::id() as i32; assert!( - members.contains(&me), - "scan missed this process ({me}) in group {own}: {members:?}" + groups.members(own).contains(&me), + "snapshot missed this process ({me}) in group {own}: {:?}", + groups.members(own) + ); + assert!( + groups.state(me).is_some(), + "a process in the snapshot must carry its state" ); } #[test] - fn scan_for_pgid_is_empty_for_a_group_that_cannot_exist() { - assert!(scan_for_pgid(i32::MAX).is_empty()); + fn snapshot_has_no_members_for_a_group_that_cannot_exist() { + assert!(ProcGroups::snapshot().members(i32::MAX).is_empty()); + } + + #[test] + fn a_pid_with_no_controlling_terminal_has_no_foreground_group() { + let groups = ProcGroups::snapshot(); + assert_eq!(groups.foreground_pgid(i32::MAX), None); } #[test] fn awaiting_input_is_false_for_nonexistent_pid() { - // A pid that can't exist → every read fails → false (fall back to Idle). - assert!(!awaiting_input(i32::MAX, "/dev/pts/0")); + // A pid that can't exist → not in the snapshot → false (falls back to Idle). + assert!(!awaiting_input( + i32::MAX, + "/dev/pts/0", + &ProcGroups::snapshot() + )); } } diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 1019dcb..2372aef 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -595,6 +595,11 @@ async fn run_activity_pass(inner: &Arc) { return; } + // One `/proc` walk for the whole pass, rather than one per session. Reading + // it is pure filesystem work, so it goes on the blocking pool alongside the + // tmux calls below. + let groups = tmux_blocking(crate::proc::ProcGroups::snapshot).await; + for id in ids { // The tmux pane lookup is a subprocess — keep it off the runtime worker // and outside the sessions lock. @@ -611,7 +616,7 @@ async fn run_activity_pass(inner: &Arc) { } }; // `/proc` inspection is cheap, but still done off the async lock. - let awaiting = crate::proc::awaiting_input(pane.pane_pid, &pane.tty); + let awaiting = crate::proc::awaiting_input(pane.pane_pid, &pane.tty, &groups); let updated = { let mut sessions = inner.sessions.lock().await; diff --git a/crates/maestro-daemon/tests/activity.rs b/crates/maestro-daemon/tests/activity.rs index b3f0994..d7fe201 100644 --- a/crates/maestro-daemon/tests/activity.rs +++ b/crates/maestro-daemon/tests/activity.rs @@ -10,7 +10,7 @@ mod common; use std::time::Duration; use common::TmuxFixture; -use maestro_daemon::proc::awaiting_input; +use maestro_daemon::proc::{awaiting_input, ProcGroups}; use tempfile::TempDir; use uuid::Uuid; @@ -22,7 +22,8 @@ fn wait_for_awaiting(fx: &TmuxFixture, id: Uuid, want: bool, label: &str) { let mut last = None; while std::time::Instant::now() < deadline { if let Ok(Some(pane)) = fx.tmux.pane_info(id) { - let got = awaiting_input(pane.pane_pid, &pane.tty); + // Fresh each poll: the tree is still settling. + let got = awaiting_input(pane.pane_pid, &pane.tty, &ProcGroups::snapshot()); if got == want { return; } @@ -70,6 +71,15 @@ fn a_foreground_reader_with_a_background_helper_is_awaiting_input() { ) .expect("new_session"); + // Wait for the helper itself. Asserting before it exists would pass + // against a deepest-descendant walk too, because until then the newest + // process in the tree *is* the one blocked on read. + let pane = fx.tmux.pane_info(id).expect("pane_info").expect("pane"); + wait_until( + || subtree_has_comm(pane.pane_pid, "sleep"), + "the background helper to be spawned", + ); + wait_for_awaiting( &fx, id, @@ -92,6 +102,39 @@ fn a_foreground_sleep_is_not_awaiting_input() { wait_for_awaiting(&fx, id, false, "a timer sleep is not user input"); } +/// Whether any process under `root` (inclusive) has this `comm`. +fn subtree_has_comm(root: i32, comm: &str) -> bool { + let mut queue = vec![root]; + while let Some(pid) = queue.pop() { + if std::fs::read_to_string(format!("/proc/{pid}/comm")).is_ok_and(|c| c.trim() == comm) { + return true; + } + queue.extend( + std::fs::read_to_string(format!("/proc/{pid}/task/{pid}/children")) + .map(|s| { + s.split_whitespace() + .filter_map(|x| x.parse::().ok()) + .collect::>() + }) + .unwrap_or_default(), + ); + } + false +} + +/// Block until `cond` holds, so a test asserts on the state it means to +/// exercise rather than on whatever the tree looked like first. +fn wait_until(cond: impl Fn() -> bool, what: &str) { + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while std::time::Instant::now() < deadline { + if cond() { + return; + } + std::thread::sleep(Duration::from_millis(20)); + } + panic!("timed out waiting for {what}"); +} + /// Read `pgrp` and `tpgid` for a pid straight from `/proc`, so the test can /// wait for the exact condition it means to exercise rather than a proxy. fn pgrp_and_tpgid(pid: i32) -> Option<(i32, i32)> { @@ -147,7 +190,7 @@ fn a_foreground_pipeline_whose_leader_exited_is_still_awaiting_input() { ); assert!( - awaiting_input(pane.pane_pid, &pane.tty), + awaiting_input(pane.pane_pid, &pane.tty, &ProcGroups::snapshot()), "a surviving pipeline member still holds the tty" ); }