fix(maestro): snapshot /proc once per pass and check the whole group
The subtree search returned early whenever it found *any* group member, so an in-tree stage that was busy or sleeping on a timer masked a reparented member genuinely parked on the tty. Checking every member instead would have meant a /proc scan per session per tick whenever nothing was parked — the common case. ProcGroups snapshots /proc once per activity pass and indexes by process group, so the cost is one walk however many sessions the daemon owns, and the subtree walk disappears: the index is a superset of it. Also fixes the background-helper test, which had stopped discriminating. It polled until awaiting_input was true and caught a moment before the helper spawned, so it passed against a deepest-descendant walk too. It now waits for the helper to exist first. Both regression tests were re-checked by patching each old implementation back in. Reported in review on #51. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit was merged in pull request #51.
This commit is contained in:
co-authored by
Claude Opus 5
parent
cd7e76d911
commit
9d777e2a22
@@ -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/<pid>/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`.
|
||||
|
||||
/// 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 {
|
||||
return false;
|
||||
};
|
||||
foreground_members(pane_pid, pgid)
|
||||
.into_iter()
|
||||
.any(|pid| parked_on_tty(pid, tty))
|
||||
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<i32, Stat>,
|
||||
by_pgid: HashMap<i32, Vec<i32>>,
|
||||
}
|
||||
|
||||
/// 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<i32> {
|
||||
let tpgid = read_stat(pane_pid)?.tpgid;
|
||||
impl ProcGroups {
|
||||
pub fn snapshot() -> Self {
|
||||
let mut stats = HashMap::new();
|
||||
let mut by_pgid: HashMap<i32, Vec<i32>> = 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::<i32>().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<i32> {
|
||||
let tpgid = self.stats.get(&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;
|
||||
fn members(&self, pgid: i32) -> &[i32] {
|
||||
self.by_pgid.get(&pgid).map_or(&[], Vec::as_slice)
|
||||
}
|
||||
|
||||
/// Live members of process group `pgid`.
|
||||
fn state(&self, pid: i32) -> Option<char> {
|
||||
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`.
|
||||
///
|
||||
/// `pgid` names a *group*, not a process, so `/proc/<pgid>` 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<i32> {
|
||||
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<i32> {
|
||||
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<i32> {
|
||||
let Ok(entries) = std::fs::read_dir("/proc") else {
|
||||
return Vec::new();
|
||||
/// 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;
|
||||
};
|
||||
entries
|
||||
.flatten()
|
||||
.filter_map(|e| e.file_name().to_str()?.parse::<i32>().ok())
|
||||
.filter(|pid| read_stat(*pid).is_some_and(|s| s.pgrp == pgid))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Direct children of `pid`, via `/proc/<pid>/task/<pid>/children`.
|
||||
fn children(pid: i32) -> Vec<i32> {
|
||||
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()
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -595,6 +595,11 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
|
||||
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<Inner>) {
|
||||
}
|
||||
};
|
||||
// `/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;
|
||||
|
||||
@@ -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::<i32>().ok())
|
||||
.collect::<Vec<_>>()
|
||||
})
|
||||
.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"
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user