diff --git a/CHANGELOG.md b/CHANGELOG.md index 09a36f0..43037a6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,19 @@ and the versions follow [semver](https://semver.org). ## [Unreleased] +### Added + +- `taskguard top`: the Queue view is a tree. The jobs of one task-runner + run sit under one row, named by the command that started it, with their + counts, CPU and memory. The taskguard calls running inside a job, such as a + dev stack's start-up builds, show under that job with their own CPU and + memory. `←` folds a row, `→` unfolds it, and `Enter` on a run's row does + both. Running jobs keep their place, + oldest first. +- `--wraps`: a command that runs other taskguard jobs runs straight through + from its first run, and the jobs inside it queue on their own. A queued run + that had jobs inside it prints advice to pass it. + ### Changed - A command that wraps other taskguard jobs, such as `taskguard bun run diff --git a/README.md b/README.md index edf6811..84406b6 100644 --- a/README.md +++ b/README.md @@ -322,7 +322,9 @@ taskguard jobs inside them runs straight through, with one line saying so. The jobs inside it then queue on their own, each by its own need. A long-lived job, a job in a pool and a job with a minimum stay queued: their claim is on purpose. `taskguard prune 'KEY' --apply` forgets the history, so the command -queues again. +queues again. When a queued run had jobs inside it, taskguard prints advice to +pass `--wraps`: `taskguard --wraps bun run ci:local` runs straight through +from the first run, with nothing to learn first. Long-running commands are never queued: watch modes, dev servers, `--version` and `--help` run straight through (the list is in `src/builtin.rs`). Add your @@ -420,7 +422,7 @@ run, and the minimum to pin when a job starves again and again. | View | What it shows | | --- | --- | | 1 Overview | Machine CPU and memory over time. The load of jobs taskguard started is stacked in colour per namespace. The load it did not start is split into groups: agents (with everything they started), browsers, editors, dev services such as databases, containers, and chat and other apps. The legend names the biggest programs in each group, so you can see what to close to make room. A strip shows how many jobs waited, coloured by reason, and `!` marks a starved run. | -| 2 Queue | Running jobs, then waiting ones. The bars show in red the load that running jobs have above their estimates. The Why panel checks every rule for the selected job, with numbers, and says what will unblock it and when. `g` starts a waiting job now, and `e` sets a job's needs for this run only (for example `2 4G`), so it can fit into room you know it will not outgrow. | +| 2 Queue | Running jobs, then waiting ones, as a tree: the jobs of one task-runner run sit under one row named by the command that started it (`bun run ci:local`), and the taskguard calls running inside a job (a dev stack's start-up builds) under that job, with their CPU and memory. `←` folds a row, `→` unfolds it, and `Enter` on a run's row does both. The bars show in red the load that running jobs have above their estimates. The Why panel checks every rule for the selected job, with numbers, and says what will unblock it and when. `g` starts a waiting job now, and `e` sets a job's needs for this run only (for example `2 4G`), so it can fit into room you know it will not outgrow. | | 3 Runs | Past runs, sortable and filterable, with the reasons each one waited. | | 4 Tasks | Every task taskguard knows, with its namespace, runs, last run, and what it learned. `n` (namespace), `/` (text) and `a` (not run for 1, 7, 30 or 90 days) filter the list. `x` prunes every run of the selected task; `X` prunes every task the filters leave, and needs a filter. Both ask first. Pruned runs move aside, as with `taskguard prune`, and `taskguard prune --undo --apply` puts them back. In a job's view, `x` prunes the selected run. | | 5 Trends | Commands whose memory, CPU or duration grows: the last 10 runs against the 10 before. | diff --git a/src/config.rs b/src/config.rs index 8932157..849a69c 100644 --- a/src/config.rs +++ b/src/config.rs @@ -265,7 +265,7 @@ pub enum SettingKind { Size(u64, u64, u64), } -pub const SETTINGS: [Setting; 13] = [ +pub const SETTINGS: [Setting; 14] = [ Setting { key: "enabled", what: "off: every command runs straight through, unqueued and unmeasured (as TASKGUARD_DISABLE=1)", @@ -283,9 +283,14 @@ pub const SETTINGS: [Setting; 13] = [ }, Setting { key: "pressure_max", - what: "Linux: no new job while memory pressure (PSI) is above this", + what: "no new job at this memory pressure (Linux: PSI share; macOS: 50 warning, 100 critical)", kind: SettingKind::Number(5.0, 5.0, 90.0, "%"), }, + Setting { + key: "warn_admit", + what: "macOS warning level: jobs still start while memory fits and no first run still grows", + kind: SettingKind::Bool, + }, Setting { key: "auto_pause", what: "pause the newest running job when memory reaches pause_at; resume it under resume_at", @@ -335,6 +340,7 @@ impl Config { "enabled" => (if self.enabled { "on" } else { "off" }).into(), "hints" => (if self.hints { "on" } else { "off" }).into(), "auto_pause" => (if self.auto_pause { "on" } else { "off" }).into(), + "warn_admit" => (if self.warn_admit { "on" } else { "off" }).into(), "pause_at" => format!("{:.0}%", self.pause_at), "resume_at" => format!("{:.0}%", self.resume_at), "max_bypass" => format!("{}s", self.max_bypass), @@ -351,6 +357,7 @@ impl Config { "enabled" => self.enabled, "hints" => self.hints, "auto_pause" => self.auto_pause, + "warn_admit" => self.warn_admit, _ => false, } } diff --git a/src/help.rs b/src/help.rs index fa32fa7..f469e89 100644 --- a/src/help.rs +++ b/src/help.rs @@ -13,6 +13,8 @@ options (sem style; the default is to run in the foreground): --min-mem SIZE never start with less than SIZE free (6G, 512M) --priority N higher starts first (default 0; negative is allowed) --now skip the queue, but still measure and learn + --wraps COMMAND runs other taskguard jobs (turbo, a package + script): run it straight through and let them queue --bg wait for room, then return and let the job run on --fg run in the foreground (the default) --pipe with --bg: pass stdin to the command @@ -51,8 +53,12 @@ taskguard top [--view NAME] [--range RANGE] [--job KEY] [--print [--width N] [-- The dashboard: the machine, the queue, past runs, every known task, trends and warnings. Inside it, ? lists the keys of the view you are in. In the -Tasks view (tab 4), filter tasks and prune their history: x one task, X every -task shown. +Queue view (tab 2), the jobs of one task-runner run (a `turbo run`) sit under +one row named by the command that started it, and the taskguard calls running +inside a job sit under that job: ← folds a row, → unfolds it, and Enter on a +run's row folds or unfolds it. In the Tasks +view (tab 4), filter tasks and prune their history: x one task, X every task +shown. options: --view NAME open on a view: overview, queue, runs, tasks, trends, diff --git a/src/main.rs b/src/main.rs index 345f49b..7c75e49 100644 --- a/src/main.rs +++ b/src/main.rs @@ -122,6 +122,7 @@ pub fn parse_opts(args: &[String]) -> Result { o.priority = Some(v.parse().map_err(|_| anyhow::anyhow!("--priority needs a whole number"))?); } "--now" => o.now = true, + "--wraps" => o.wraps = true, "--bg" => o.bg = true, "--fg" => o.bg = false, "--pipe" => o.pipe = true, diff --git a/src/queue.rs b/src/queue.rs index 2210703..1e35be3 100644 --- a/src/queue.rs +++ b/src/queue.rs @@ -12,6 +12,15 @@ use serde::{Deserialize, Serialize}; use std::fs::{self, File, OpenOptions}; use std::path::{Path, PathBuf}; +/// A taskguard call that runs inside a running job, under its claim. +#[derive(Debug, Clone, PartialEq)] +pub struct Nested { + pub outer: i32, + pub pid: i32, + pub since: f64, + pub key: String, +} + /// One waiting or running job, as a JSON file in `wait/` or `run/`. /// /// Every taskguard version on the machine reads these files, and DALP pins a @@ -287,9 +296,29 @@ impl Queue { /// A taskguard call nested inside the running job of `outer` (a pid) /// leaves `nested/.`, so that job learns that it wraps other - /// taskguard jobs. - pub fn note_nested(&self, outer: i32, pid: i32) { - let _ = fs::write(self.dir.join("nested").join(format!("{outer}.{pid}")), ""); + /// taskguard jobs. The note holds the call's job key, for the tree in + /// `taskguard top`; when it was written is when the call started. + pub fn note_nested(&self, outer: i32, pid: i32, key: &str) { + let _ = fs::write(self.dir.join("nested").join(format!("{outer}.{pid}")), key); + } + + /// Every nested call with a note: the outer job's pid, the call's pid, + /// when it started, and its key (empty from a note without one). + pub fn nested(&self) -> Vec { + let Ok(rd) = fs::read_dir(self.dir.join("nested")) else { return Vec::new() }; + rd.filter_map(|e| e.ok()) + .filter_map(|e| { + let name = e.file_name().into_string().ok()?; + let (outer, pid) = name.split_once('.')?; + let since = e.metadata().ok()?.modified().ok()?.duration_since(std::time::UNIX_EPOCH).ok()?.as_secs_f64(); + Some(Nested { + outer: outer.parse().ok()?, + pid: pid.parse().ok()?, + since, + key: fs::read_to_string(e.path()).unwrap_or_default(), + }) + }) + .collect() } /// How many nested calls left a note for the job of `owner`. The notes @@ -2288,9 +2317,13 @@ mod tests { fn nested_calls_leave_notes_for_their_outer_job() { let tmp = tempfile::tempdir().unwrap(); let q = Queue::open(tmp.path()).unwrap(); - q.note_nested(100, 201); - q.note_nested(100, 202); - q.note_nested(1000, 203); + q.note_nested(100, 201, "a:tsc"); + q.note_nested(100, 202, "b:tsc"); + q.note_nested(1000, 203, "c:lint"); + let mut seen: Vec<(i32, i32, String)> = q.nested().into_iter().map(|n| (n.outer, n.pid, n.key)).collect(); + seen.sort(); + assert_eq!(seen[0], (100, 201, "a:tsc".to_string())); + assert_eq!(seen.len(), 3); assert_eq!(q.take_nested(100), 2, "only its own notes, not those of pid 1000"); assert_eq!(q.take_nested(100), 0, "taken once"); assert_eq!(q.take_nested(1000), 1); diff --git a/src/runner.rs b/src/runner.rs index b628717..30a196b 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -34,6 +34,9 @@ pub struct Opts { pub min_cpu: Option, pub min_mem_kb: Option, pub now: bool, + /// The command wraps other taskguard jobs: it runs straight through, and + /// the jobs inside it queue on their own. + pub wraps: bool, pub bg: bool, pub pipe: bool, pub quiet: bool, @@ -99,14 +102,34 @@ fn held_by_ancestor(q: &Queue) -> bool { /// Tell the running job this nested call runs inside that it wraps other /// taskguard jobs: see `queue::Queue::note_nested`. -fn note_nested(q: &Queue) { +fn note_nested(q: &Queue, o: &Opts) { let procs = sys::list_procs(); let me = std::process::id() as i32; if let Some(outer) = sys::ancestors(me, &procs).into_iter().skip(1).find(|p| q.run_path(*p).exists()) { - q.note_nested(outer, me); + let key = match (&o.key, std::env::current_dir()) { + (Some(k), _) => k.clone(), + (None, Ok(cwd)) => key::run_key(&cwd, &o.cmd), + (None, Err(_)) => String::new(), + }; + q.note_nested(outer, me, &key); } } +/// A command as one line to paste into a shell: words with spaces or +/// shell characters in single quotes. +fn shell_words(cmd: &[String]) -> String { + cmd.iter() + .map(|w| { + if !w.is_empty() && w.chars().all(|c| c.is_ascii_alphanumeric() || "-_./:=@%+,".contains(c)) { + w.clone() + } else { + format!("'{}'", w.replace('\'', "'\\''")) + } + }) + .collect::>() + .join(" ") +} + /// Task runners whose run groups the jobs it starts. A job belongs to the /// nearest one above it. const LAUNCHERS: &[&str] = &["turbo", "nx", "make", "gmake", "moon", "lage", "just"]; @@ -196,7 +219,7 @@ pub fn run(mut o: Opts) -> Result { exec(&o.cmd); } if held_by_ancestor(&q) { - note_nested(&q); + note_nested(&q, &o); exec(&o.cmd); } @@ -208,6 +231,12 @@ pub fn run(mut o: Opts) -> Result { if !cfg.enabled || cls.passthrough { exec(&o.cmd); } + if o.wraps { + if !o.quiet { + say(&format!("straight through {jkey} - --wraps: the taskguard jobs inside it queue on their own")); + } + exec(&o.cmd); + } if o.bg { return run_bg(&o); } @@ -268,7 +297,7 @@ pub fn run(mut o: Opts) -> Result { { if !o.quiet { say(&format!( - "straight through {jkey} - its recent runs started up to {n} taskguard jobs inside it, and those queue on their own (taskguard prune '{jkey}' --apply queues it again)" + "straight through {jkey} - its recent runs started up to {n} taskguard jobs inside it, and those queue on their own; pass --wraps to say so up front (taskguard prune '{jkey}' --apply queues it again)" )); } exec(&o.cmd); @@ -829,6 +858,13 @@ pub fn run(mut o: Opts) -> Result { } else { None }; + let nested_jobs = q.take_nested(me_pid); + if nested_jobs > 0 && startup.is_none() && pool.is_none() && min_mem.is_none() && min_cpu.is_none() { + lines.advice(&[format!( + "{jkey} started {nested_jobs} taskguard jobs inside it, which ran under its claim instead of queuing on their own. Next time it runs straight through by itself; to say so up front, run: taskguard --wraps {}", + shell_words(&o.cmd) + )]); + } if let Some(d) = &database { let _ = d.finish_run( run_id, @@ -847,7 +883,7 @@ pub fn run(mut o: Opts) -> Result { measured, paused_s: me.paused_s, steady: tracker.steady(ended, me.paused_s - paused_before_steady), - nested_jobs: q.take_nested(me_pid), + nested_jobs, }, ); } diff --git a/src/sys/linux.rs b/src/sys/linux.rs index 0d937f3..da1fa6f 100644 --- a/src/sys/linux.rs +++ b/src/sys/linux.rs @@ -139,6 +139,13 @@ pub fn proc_name(pid: i32) -> String { fs::read_to_string(format!("/proc/{pid}/comm")).map(|s| s.trim().to_string()).unwrap_or_else(|_| format!("pid {pid}")) } +/// The command line of a process. None when it is gone or a kernel thread. +pub fn proc_args(pid: i32) -> Option> { + let raw = fs::read(format!("/proc/{pid}/cmdline")).ok()?; + let args: Vec = raw.split(|b| *b == 0).filter(|a| !a.is_empty()).map(|a| String::from_utf8_lossy(a).into_owned()).collect(); + (!args.is_empty()).then_some(args) +} + /// The program's path, for grouping processes that share a name. pub fn proc_path(pid: i32) -> Option { fs::read_link(format!("/proc/{pid}/exe")).ok().map(|p| p.display().to_string()) diff --git a/src/sys/macos.rs b/src/sys/macos.rs index 12b460d..466a70d 100644 --- a/src/sys/macos.rs +++ b/src/sys/macos.rs @@ -174,6 +174,33 @@ pub fn proc_name(pid: i32) -> String { String::from_utf8_lossy(&buf[..n as usize]).into_owned() } +/// The command line of a process, from KERN_PROCARGS2: the argument count, +/// the executable path, padding, then the arguments, each ending in a NUL. +/// None when the process is gone or belongs to another user. +pub fn proc_args(pid: i32) -> Option> { + let mut argmax: libc::c_int = 0; + let mut len = size_of::(); + let mut mib = [libc::CTL_KERN, libc::KERN_ARGMAX]; + let r = unsafe { libc::sysctl(mib.as_mut_ptr(), 2, &mut argmax as *mut _ as *mut c_void, &mut len, std::ptr::null_mut(), 0) }; + if r != 0 || argmax <= 0 { + return None; + } + let mut buf = vec![0u8; argmax as usize]; + let mut len = buf.len(); + let mut mib = [libc::CTL_KERN, libc::KERN_PROCARGS2, pid]; + let r = unsafe { libc::sysctl(mib.as_mut_ptr(), 3, buf.as_mut_ptr() as *mut c_void, &mut len, std::ptr::null_mut(), 0) }; + if r != 0 || len < size_of::() { + return None; + } + let argc = i32::from_ne_bytes(buf[..4].try_into().ok()?) as usize; + let rest = &buf[4..len]; + // Skip the executable path and the NULs after it. + let path_end = rest.iter().position(|b| *b == 0)?; + let start = path_end + rest[path_end..].iter().position(|b| *b != 0)?; + let args: Vec = rest[start..].split(|b| *b == 0).take(argc).map(|a| String::from_utf8_lossy(a).into_owned()).collect(); + (!args.is_empty()).then_some(args) +} + /// The program's path. Some programs have a name that says little: Claude /// Code's process is named after its version, for example "2.1.282". pub fn proc_path(pid: i32) -> Option { diff --git a/src/sys/mod.rs b/src/sys/mod.rs index b0306b5..8efdfe9 100644 --- a/src/sys/mod.rs +++ b/src/sys/mod.rs @@ -156,6 +156,24 @@ pub fn ancestors(pid: i32, procs: &[ProcInfo]) -> Vec { mod tests { use super::*; + #[test] + fn proc_args_reads_a_command_line() { + let mut child = std::process::Command::new("sleep").arg("30").spawn().unwrap(); + // Right after the fork the child still has the parent's command line, + // until it execs sleep. + let read = || proc_args(child.id() as i32).filter(|a| a.last().is_some_and(|l| l == "30")); + let t0 = std::time::Instant::now(); + let mut args = read(); + while args.is_none() && t0.elapsed() < std::time::Duration::from_secs(2) { + std::thread::sleep(std::time::Duration::from_millis(20)); + args = read(); + } + child.kill().unwrap(); + child.wait().unwrap(); + assert_eq!(args.map(|a| a[1..].to_vec()), Some(vec!["30".to_string()])); + assert_eq!(proc_args(-1), None); + } + #[test] fn descendants_walks_the_whole_tree() { let procs = [ diff --git a/src/top/mod.rs b/src/top/mod.rs index 4627571..36a173c 100644 --- a/src/top/mod.rs +++ b/src/top/mod.rs @@ -1,6 +1,7 @@ //! `taskguard top`: the dashboard. mod chart; +mod tree; mod views; use crate::commands; @@ -13,7 +14,7 @@ use crate::report::Snapshot; use crate::runner; use anyhow::Result; use crossterm::event::{self, Event, KeyCode, KeyEvent, KeyEventKind, KeyModifiers, MouseButton, MouseEventKind}; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::path::PathBuf; use std::time::{Duration, Instant}; @@ -132,6 +133,10 @@ pub struct Data { pub cursor_runs: Vec, pub job: Option, pub run_spans: Vec<(f64, String, String)>, + /// The Queue view's tree: the taskguard calls running inside each job, + /// by the job's pid, and the command each task-runner run stands for. + pub subs: HashMap>, + pub labels: HashMap, } pub struct JobData { @@ -206,6 +211,10 @@ pub struct App { /// The pool whose match patterns the Config view lists, and the row of /// that pool to return to. pub pool_edit: Option<(String, usize)>, + /// Folded Queue rows: task-runner runs by id, jobs as "pid:". + pub folded: HashSet, + /// CPU time of each nested call at the last load, for its cores. + sub_cpu: HashMap, /// Kept open between loads. db: Option, history: Option, @@ -271,6 +280,8 @@ impl App { checkout: PathBuf::new(), save_to: None, pool_edit: None, + folded: HashSet::new(), + sub_cpu: HashMap::new(), db: None, history: None, } @@ -408,11 +419,24 @@ impl App { }; let mut namespaces = dash::namespaces(db, live_to - span, self.cfg.sample_every)?; sort_ns(&mut namespaces, self.sort, self.reverse); + let snap = commands::queue_snapshot(&self.dir, &self.cfg, Some(db)).ok(); + // The tree reads the process table, so only while the Queue shows. + let (subs, labels) = match (&snap, self.view) { + (Some(s), View::Queue) => { + let procs = crate::sys::list_procs(); + let subs = tree::subtasks(&Queue { dir: self.dir.clone() }, &s.running, &procs, now, &mut self.sub_cpu); + let ids: HashSet<&String> = + s.running.iter().chain(s.waiting.iter().map(|w| &w.entry)).filter_map(|e| e.pipeline.as_ref()).collect(); + let labels = ids.into_iter().filter_map(|id| Some((id.clone(), tree::label(id, &procs)?))).collect(); + (subs, labels) + } + _ => Default::default(), + }; self.data = Data { now, from, to, - snap: commands::queue_snapshot(&self.dir, &self.cfg, Some(db)).ok(), + snap, machine, ncpu, mem_total_kb: mem_total, @@ -433,6 +457,8 @@ impl App { cursor_runs, job, run_spans, + subs, + labels, }; self.pick = self.pick.map(|p| p.min(self.data.cursor_runs.len().saturating_sub(1))).filter(|_| !self.data.cursor_runs.is_empty()); Ok(()) @@ -440,7 +466,7 @@ impl App { fn list_len(&self) -> usize { match self.view { - View::Queue => self.data.snap.as_ref().map(|s| s.waiting.len() + s.running.len()).unwrap_or(0), + View::Queue => self.queue_tree().rows.len(), View::Runs => self.data.runs.len(), View::Trends => self.data.trends.len(), View::Tasks => self.data.tasks.len(), @@ -569,7 +595,12 @@ impl App { View::Trends => self.data.trends.get(self.sel).map(|t| t.key.clone()), View::Tasks => self.data.tasks.get(self.sel).map(|t| t.key.clone()), View::Warnings => self.data.warnings.get(self.sel).map(|w| w.key.clone()), - View::Queue => self.queue_row(self.sel).map(|(e, _)| e.key.clone()), + View::Queue => match self.queue_tree().rows.get(self.sel) { + Some(tree::Row::Sub { outer, i, .. }) => { + self.data.subs.get(&self.data.snap.as_ref()?.running.get(*outer)?.pid)?.get(*i).map(|s| s.key.clone()) + } + _ => self.queue_row(self.sel).map(|(e, _)| e.key.clone()), + }, View::Namespaces => { let ns = self.data.namespaces.get(self.sel)?; ns.top_keys.first().map(|k| k.0.clone()) @@ -578,17 +609,67 @@ impl App { } } - /// Queue view rows: running jobs first, then waiting ones in queue - /// order. The bool is true for a waiting job. + /// The Queue view's rows: see `tree::build`. + pub fn queue_tree(&self) -> tree::Tree { + let Some(s) = self.data.snap.as_ref() else { return tree::Tree::default() }; + let waiting: Vec<&crate::queue::Entry> = s.waiting.iter().map(|w| &w.entry).collect(); + tree::build(&s.running, &waiting, &self.data.subs, &self.data.labels, &self.folded) + } + + /// The job on Queue row `i`; the bool is true for a waiting job. None + /// on a group header or a nested call. pub fn queue_row(&self, i: usize) -> Option<(&crate::queue::Entry, bool)> { let s = self.data.snap.as_ref()?; - if i < s.running.len() { Some((&s.running[i], false)) } else { s.waiting.get(i - s.running.len()).map(|w| (&w.entry, true)) } + match self.queue_tree().rows.get(i)? { + tree::Row::Job { waiting: false, i, .. } => s.running.get(*i).map(|e| (e, false)), + tree::Row::Job { waiting: true, i, .. } => s.waiting.get(*i).map(|w| (&w.entry, true)), + _ => None, + } } /// The waiting job on the selected Queue row, with its decision. pub fn queue_waiting(&self) -> Option<&crate::report::Waiting> { let s = self.data.snap.as_ref()?; - self.sel.checked_sub(s.running.len()).and_then(|i| s.waiting.get(i)) + match self.queue_tree().rows.get(self.sel)? { + tree::Row::Job { waiting: true, i, .. } => s.waiting.get(*i), + _ => None, + } + } + + /// ← folds the selected group or job, or, on a row inside one, folds + /// that and selects it. → unfolds. + fn fold(&mut self, fold: bool) { + let Some(s) = self.data.snap.as_ref() else { return }; + let t = self.queue_tree(); + let id_at = |r: usize| match t.rows.get(r) { + Some(tree::Row::Group(g)) => t.groups.get(*g).map(|g| g.id.clone()), + Some(tree::Row::Job { waiting: false, i, .. }) => { + s.running.get(*i).filter(|e| self.data.subs.contains_key(&e.pid)).map(tree::fold_id_of_job) + } + _ => None, + }; + if !fold { + if let Some(id) = id_at(self.sel) { + self.folded.remove(&id); + } + return; + } + if let Some(id) = id_at(self.sel).filter(|id| !self.folded.contains(id)) { + self.folded.insert(id); + return; + } + // Inside a group or a job: fold the nearest row above that holds it. + let depth = |r: &tree::Row| match r { + tree::Row::Group(_) => 0, + tree::Row::Job { depth, .. } | tree::Row::Sub { depth, .. } => *depth, + }; + let Some(here) = t.rows.get(self.sel).map(depth).filter(|d| *d > 0) else { return }; + if let Some(up) = (0..self.sel).rev().find(|&r| depth(&t.rows[r]) < here && id_at(r).is_some()) { + if let Some(id) = id_at(up) { + self.folded.insert(id); + } + self.sel = up; + } } /// A job from an older taskguard does not read nudges. @@ -996,6 +1077,7 @@ impl App { KeyCode::Char('s') => self.sort += 1, KeyCode::Char('r') => self.reverse = !self.reverse, KeyCode::Char(' ') => self.paused = !self.paused, + KeyCode::Left | KeyCode::Right if self.view == View::Queue => self.fold(k.code == KeyCode::Left), KeyCode::PageUp | KeyCode::PageDown => self.page_list(k.code == KeyCode::PageDown), KeyCode::Up => self.sel = self.sel.saturating_sub(1), KeyCode::Down => { @@ -1006,7 +1088,14 @@ impl App { KeyCode::Home => self.sel = 0, KeyCode::End => self.sel = self.list_len().saturating_sub(1), KeyCode::Enter => { - if let Some(k) = self.selected_key() { + if let (View::Queue, Some(tree::Row::Group(g))) = (self.view, self.queue_tree().rows.get(self.sel).copied()) { + // A run's row has no details of its own: Enter folds or unfolds it. + let t = self.queue_tree(); + let id = t.groups[g].id.clone(); + if !self.folded.remove(&id) { + self.folded.insert(id); + } + } else if let Some(k) = self.selected_key() { self.open_job(k); } } diff --git a/src/top/tree.rs b/src/top/tree.rs new file mode 100644 index 0000000..66e657f --- /dev/null +++ b/src/top/tree.rs @@ -0,0 +1,322 @@ +//! The Queue view as a tree: the jobs of one task-runner run under one +//! header, and the taskguard calls that run inside a job under that job. +//! +//! Jobs of one `turbo run` used to show as loose rows, and since a wrapper +//! such as `bun run ci:local` runs straight through, nothing showed that one +//! command started them. Calls nested inside a job (a dev stack's start-up +//! builds) had no row at all: they run under the job's claim and leave only +//! a note in `nested/`. + +use crate::queue::{Entry, Nested, Queue}; +use crate::sys; +use std::collections::{HashMap, HashSet}; + +/// One row of the Queue view. +#[derive(Debug, Clone, Copy, PartialEq)] +pub enum Row { + /// The header of a task-runner run: an index into `Tree::groups`. + Group(usize), + /// A running (`waiting` false) or waiting job: an index into the + /// snapshot's running or waiting list. `depth` 1 is inside a group. + Job { waiting: bool, i: usize, depth: u8 }, + /// A call nested inside the running job `outer` (an index into the + /// running list): an index into that job's subtasks. + Sub { outer: usize, i: usize, depth: u8 }, +} + +/// A task-runner run with more than one job in the queue. +#[derive(Debug, Clone, PartialEq)] +pub struct Group { + /// The run's id, as "turbo:4242". + pub id: String, + /// The command that started it, or the id when it cannot be read. + pub label: String, + pub running: usize, + pub waiting: usize, + /// What its running jobs use now, and the memory they need. + pub cpu: f64, + pub mem_kb: u64, + pub need_mem_kb: u64, + /// When the run queued its first job. + pub since: f64, + pub folded: bool, +} + +/// A taskguard call running inside a job, measured from its process tree. +#[derive(Debug, Clone, PartialEq)] +pub struct Sub { + pub key: String, + pub pid: i32, + pub since: f64, + pub cpu: f64, + pub mem_kb: u64, +} + +#[derive(Debug, Default)] +pub struct Tree { + pub rows: Vec, + pub groups: Vec, + /// The tree lines in front of each row's name: "├ ", "└ ", and "│ " for + /// a row above that still has rows below it. + pub lead: Vec, +} + +impl Row { + pub fn depth(&self) -> u8 { + match self { + Row::Group(_) => 0, + Row::Job { depth, .. } | Row::Sub { depth, .. } => *depth, + } + } +} + +/// The tree lines for rows at these depths. A row is the last of its +/// parent's when no later row at its depth comes before one above it. +fn leads(depths: &[u8]) -> Vec { + let last = |r: usize| depths[r + 1..].iter().find(|&&d| d <= depths[r]).is_none_or(|&d| d < depths[r]); + // For each depth: whether the row that opened it was the last one. + let mut open: Vec = Vec::new(); + (0..depths.len()) + .map(|r| { + let d = depths[r] as usize; + open.truncate(d); + let mut lead = String::new(); + if d > 0 { + for &was_last in &open[1..] { + lead.push_str(if was_last { " " } else { "│ " }); + } + lead.push_str(if last(r) { "└ " } else { "├ " }); + } + open.resize(d, false); + open.push(last(r)); + lead + }) + .collect() +} + +/// What folds a row: a group by its run id (`Group::id`), a job by its pid. +pub fn fold_id_of_job(e: &Entry) -> String { + format!("pid:{}", e.pid) +} + +/// Lay out the rows. Jobs keep their order: running first, then waiting in +/// queue order. A run with two jobs or more becomes a group where its first +/// job would be, with its running and then its waiting jobs below it. Calls +/// nested in a running job follow that job. +pub fn build( + running: &[Entry], + waiting: &[&Entry], + subs: &HashMap>, + labels: &HashMap, + folded: &HashSet, +) -> Tree { + let mut count: HashMap<&str, usize> = HashMap::new(); + for e in running.iter().chain(waiting.iter().copied()) { + if let Some(p) = &e.pipeline { + *count.entry(p).or_default() += 1; + } + } + + let mut tree = Tree::default(); + let job = |tree: &mut Tree, waiting: bool, i: usize, depth: u8| { + tree.rows.push(Row::Job { waiting, i, depth }); + if waiting { + return; + } + let e = &running[i]; + if folded.contains(&fold_id_of_job(e)) { + return; + } + for k in 0..subs.get(&e.pid).map(Vec::len).unwrap_or(0) { + tree.rows.push(Row::Sub { outer: i, i: k, depth: depth + 1 }); + } + }; + let mut done: HashSet<&str> = HashSet::new(); + // Running jobs come in directory order, which may change between two + // loads; the oldest first keeps the rows in place. + let mut by_start: Vec = (0..running.len()).collect(); + by_start.sort_by(|&a, &b| { + let start = |i: usize| running[i].started_at.unwrap_or(f64::MAX); + start(a).total_cmp(&start(b)).then(running[a].pid.cmp(&running[b].pid)) + }); + let order = by_start.iter().map(|&i| (false, i)).chain((0..waiting.len()).map(|i| (true, i))); + for (is_waiting, i) in order { + let e = if is_waiting { waiting[i] } else { &running[i] }; + let Some(p) = e.pipeline.as_deref().filter(|p| count.get(p).copied().unwrap_or(0) >= 2) else { + job(&mut tree, is_waiting, i, 0); + continue; + }; + if !done.insert(p) { + continue; + } + let mine_running: Vec = by_start.iter().copied().filter(|&j| running[j].pipeline.as_deref() == Some(p)).collect(); + let mine_waiting: Vec = (0..waiting.len()).filter(|&j| waiting[j].pipeline.as_deref() == Some(p)).collect(); + let all = mine_running.iter().map(|&j| &running[j]).chain(mine_waiting.iter().map(|&j| waiting[j])); + let since = all.filter_map(|e| e.pipeline_since).fold(f64::MAX, f64::min); + let g = Group { + id: p.to_string(), + label: labels.get(p).cloned().unwrap_or_else(|| p.to_string()), + running: mine_running.len(), + waiting: mine_waiting.len(), + cpu: mine_running.iter().map(|&j| running[j].live_cpu).sum(), + mem_kb: mine_running.iter().map(|&j| running[j].live_mem_kb).sum(), + need_mem_kb: mine_running.iter().map(|&j| running[j].need_mem_kb).sum(), + since: if since == f64::MAX { e.queued_at } else { since }, + folded: folded.contains(p), + }; + let folded = g.folded; + tree.rows.push(Row::Group(tree.groups.len())); + tree.groups.push(g); + if folded { + continue; + } + for j in mine_running { + job(&mut tree, false, j, 1); + } + for j in mine_waiting { + job(&mut tree, true, j, 1); + } + } + tree.lead = leads(&tree.rows.iter().map(Row::depth).collect::>()); + tree +} + +/// The command a task-runner run stands for, as people typed it: the first +/// package script above the runner (`bun run ci:local` around `turbo run +/// ...`), else the runner's own command line. A `node .../turbo` shim +/// between them is skipped. +pub fn label(id: &str, procs: &[sys::ProcInfo]) -> Option { + let pid: i32 = id.split_once(':')?.1.parse().ok()?; + let ups = std::iter::once(pid).chain(sys::ancestors(pid, procs).into_iter().skip(1)).take(6); + let script = ups.filter(|&p| p > 1).filter_map(sys::proc_args).find(|a| is_script_run(a)); + let args = script.or_else(|| sys::proc_args(pid))?; + Some(command_text(&args)) +} + +/// `bun run ci:local`, `npm run test`, `pnpm run lint`, `yarn run x`. +fn is_script_run(args: &[String]) -> bool { + let tool = |a: &String| a.rsplit('/').next().unwrap_or(a).to_string(); + args.len() >= 3 && matches!(tool(&args[0]).as_str(), "bun" | "npm" | "pnpm" | "yarn") && args[1] == "run" +} + +/// A command line with the program's directory left out. +fn command_text(args: &[String]) -> String { + let mut words = args.iter(); + let first = words.next().map(|a| a.rsplit('/').next().unwrap_or(a)).unwrap_or_default(); + std::iter::once(first).chain(words.map(String::as_str)).collect::>().join(" ") +} + +/// The calls nested in each running job, by the job's pid, measured now. +/// `prev` holds each call's CPU time at the last reading, for its cores. +pub fn subtasks( + q: &Queue, + running: &[Entry], + procs: &[sys::ProcInfo], + now: f64, + prev: &mut HashMap, +) -> HashMap> { + let outers: HashSet = running.iter().map(|e| e.pid).collect(); + let children = sys::children_map(procs); + let mut out: HashMap> = HashMap::new(); + let mut seen = HashMap::new(); + let mut notes: Vec = q.nested().into_iter().filter(|n| outers.contains(&n.outer)).collect(); + notes.sort_by(|a, b| a.since.total_cmp(&b.since)); + for n in notes { + // A call that has ended leaves its note until its job ends. + let tree = sys::descendants(n.pid, &children); + let samples: Vec = tree.iter().filter_map(|&p| sys::proc_sample(p)).collect(); + if samples.is_empty() { + continue; + } + let cpu_ns: u64 = samples.iter().map(|s| s.cpu_ns).sum(); + let cpu = match prev.get(&n.pid) { + Some((t, ns)) if now > *t => cpu_ns.saturating_sub(*ns) as f64 / 1e9 / (now - t), + _ => 0.0, + }; + seen.insert(n.pid, (now, cpu_ns)); + let key = if n.key.is_empty() { sys::proc_name(n.pid) } else { n.key }; + out.entry(n.outer).or_default().push(Sub { + key, + pid: n.pid, + since: n.since, + cpu, + mem_kb: samples.iter().map(|s| s.footprint_kb).sum(), + }); + } + *prev = seen; + out +} + +#[cfg(test)] +mod tests { + use super::*; + + fn job(pid: i32, pipeline: Option<&str>) -> Entry { + Entry { + pid, + key: format!("j{pid}"), + pipeline: pipeline.map(str::to_string), + pipeline_since: pipeline.map(|_| 100.0), + ..Default::default() + } + } + + fn sub(key: &str) -> Sub { + Sub { key: key.into(), pid: 0, since: 0.0, cpu: 0.0, mem_kb: 0 } + } + + #[test] + fn a_run_with_two_jobs_or_more_becomes_one_group() { + let running = + [job(1, Some("turbo:9")), job(2, None), job(3, Some("turbo:9"))].map(|e| Entry { started_at: Some(e.pid as f64), ..e }); + let w = [job(4, Some("turbo:9")), job(5, Some("turbo:7"))]; + let waiting: Vec<&Entry> = w.iter().collect(); + let labels = HashMap::from([("turbo:9".to_string(), "bun run ci:local".to_string())]); + let t = build(&running, &waiting, &HashMap::new(), &labels, &HashSet::new()); + assert_eq!( + t.rows, + vec![ + Row::Group(0), + Row::Job { waiting: false, i: 0, depth: 1 }, + Row::Job { waiting: false, i: 2, depth: 1 }, + Row::Job { waiting: true, i: 0, depth: 1 }, + Row::Job { waiting: false, i: 1, depth: 0 }, + // A run with one job in the queue stays a plain row. + Row::Job { waiting: true, i: 1, depth: 0 }, + ] + ); + let g = &t.groups[0]; + assert_eq!((g.label.as_str(), g.running, g.waiting, g.since), ("bun run ci:local", 2, 1, 100.0)); + } + + #[test] + fn rows_get_tree_lines() { + let l = leads(&[0, 1, 2, 2, 1, 2, 0, 1]); + assert_eq!(l, ["", "├ ", "│ ├ ", "│ └ ", "└ ", " └ ", "", "└ "]); + } + + #[test] + fn a_run_is_named_by_the_script_above_it() { + let a = |s: &str| s.split(' ').map(str::to_string).collect::>(); + assert!(is_script_run(&a("/opt/bin/bun run ci:local"))); + assert!(!is_script_run(&a("node /w/node_modules/turbo/bin/turbo run lint")), "the shim is skipped"); + assert_eq!(command_text(&a("/opt/bin/bun run ci:local")), "bun run ci:local"); + } + + #[test] + fn nested_calls_follow_their_job_and_rows_fold() { + let running = [job(1, None), job(2, Some("turbo:9")), job(3, Some("turbo:9"))]; + let subs = HashMap::from([(1, vec![sub("dalp:prepare"), sub("dalp:sdk")])]); + let t = build(&running, &[], &subs, &HashMap::new(), &HashSet::new()); + assert_eq!( + t.rows[..3], + [Row::Job { waiting: false, i: 0, depth: 0 }, Row::Sub { outer: 0, i: 0, depth: 1 }, Row::Sub { outer: 0, i: 1, depth: 1 }] + ); + assert_eq!(t.groups[0].label, "turbo:9", "without a label, the run's id"); + + let folded = HashSet::from(["pid:1".to_string(), "turbo:9".to_string()]); + let t = build(&running, &[], &subs, &HashMap::new(), &folded); + assert_eq!(t.rows, vec![Row::Job { waiting: false, i: 0, depth: 0 }, Row::Group(0)]); + assert!(t.groups[0].folded); + } +} diff --git a/src/top/views.rs b/src/top/views.rs index 9006ddb..b13ab0c 100644 --- a/src/top/views.rs +++ b/src/top/views.rs @@ -174,11 +174,13 @@ fn view_keys(app: &App) -> Vec { } View::Queue => vec![ ("↑/↓".into(), "select", None), - ("Enter".into(), "details", Some(KeyCode::Enter)), + ("Enter".into(), "details; on a run: fold", Some(KeyCode::Enter)), ("g".into(), "start now", Some(KeyCode::Char('g'))), ("e".into(), "edit needs", Some(KeyCode::Char('e'))), ("p".into(), "pause/resume", Some(KeyCode::Char('p'))), ("k".into(), "stop", Some(KeyCode::Char('k'))), + ("←".into(), "fold", Some(KeyCode::Left)), + ("→".into(), "unfold", Some(KeyCode::Right)), ], View::Runs => vec![ ("↑/↓".into(), "select", None), @@ -645,7 +647,7 @@ fn queue(f: &mut Frame, app: &mut App, area: Rect) { Span::styled(" limit", DIM), ]), ]; - let n_rows = s.waiting.len() + s.running.len(); + let n_rows = app.queue_tree().rows.len(); let [top, list, why] = Layout::vertical([Constraint::Length(4), Constraint::Length((n_rows as u16 + 2).clamp(3, 14)), Constraint::Min(6)]).areas(area); f.render_widget(Paragraph::new(bars), top); @@ -659,40 +661,83 @@ fn queue(f: &mut Frame, app: &mut App, area: Rect) { " est" } }; + let t = app.queue_tree(); let mut rows: Vec = Vec::new(); - for e in &s.running { - let over = - e.live_cpu > e.start_need_cpu.unwrap_or(e.need_cpu) + 0.05 || e.live_mem_kb > e.start_need_mem_kb.unwrap_or(e.need_mem_kb); - rows.push(Row::new(vec![ - match (e.paused_since, e.now) { - (Some(_), _) => Cell::from("PAUSE").style(Style::default().fg(Color::LightMagenta)), - (None, true) => Cell::from("NOW").style(Style::default().fg(Color::Green)), - (None, false) => Cell::from("RUN").style(Style::default().fg(Color::Green)), - }, - Cell::from(report::job_name(e, 44)), - Cell::from(e.ns.clone()), - Cell::from(dur(s.now - e.started_at.unwrap_or(s.now))), - Cell::from(format!("{:.1}c {} now", e.live_cpu, gb(e.live_mem_kb))).style(if over { - Style::default().fg(Color::Red) - } else { - Style::default() - }), - Cell::from(format!("needs {:.1}c {}{}", e.need_cpu, gb(e.need_mem_kb), est(e))), - ])); - } - for w in &s.waiting { - let e = &w.entry; - let blocked = report::main_blocker(&w.decision).map(|b| b.name().to_uppercase()).unwrap_or_else(|| "starting".into()) - + &w.eta_s.map(|t| format!(" ~{}", dur(t.max(1.0)))).unwrap_or_default(); - rows.push(Row::new(vec![ - Cell::from("WAIT").style(Style::default().fg(Color::Yellow)), - Cell::from(report::job_name(e, 44)), - Cell::from(e.ns.clone()), - Cell::from(dur(s.now - e.queued_at)), - Cell::from(format!("{:.1}c {}{}", e.need_cpu, gb(e.need_mem_kb), est(e))), - Cell::from(blocked) - .style(Style::default().fg(chart::blocker_color(report::main_blocker(&w.decision).map(|b| b.name()).unwrap_or("")))), - ])); + for (n, r) in t.rows.iter().enumerate() { + let lead = t.lead[n].as_str(); + let room = 44usize.saturating_sub(lead.chars().count()); + match *r { + super::tree::Row::Group(g) => { + let g = &t.groups[g]; + rows.push( + Row::new(vec![ + Cell::from(""), + Cell::from(format!("{} {}", if g.folded { "▸" } else { "▾" }, trunc(&g.label, 42))), + Cell::from(g.id.clone()), + Cell::from(dur(s.now - g.since)), + Cell::from(if g.running > 0 { format!("{:.1}c {} now", g.cpu.max(0.0), gb(g.mem_kb)) } else { String::new() }), + Cell::from(format!("{} run {} wait", g.running, g.waiting)), + ]) + .style(BOLD), + ); + } + super::tree::Row::Job { waiting: false, i, .. } => { + let e = &s.running[i]; + let over = e.live_cpu > e.start_need_cpu.unwrap_or(e.need_cpu) + 0.05 + || e.live_mem_kb > e.start_need_mem_kb.unwrap_or(e.need_mem_kb); + let fold = match app.data.subs.get(&e.pid) { + Some(_) if app.folded.contains(&super::tree::fold_id_of_job(e)) => "▸ ", + Some(_) => "▾ ", + None => "", + }; + rows.push(Row::new(vec![ + match (e.paused_since, e.now) { + (Some(_), _) => Cell::from("PAUSE").style(Style::default().fg(Color::LightMagenta)), + (None, true) => Cell::from("NOW").style(Style::default().fg(Color::Green)), + (None, false) => Cell::from("RUN").style(Style::default().fg(Color::Green)), + }, + Cell::from(format!("{lead}{fold}{}", report::job_name(e, room.saturating_sub(fold.chars().count())))), + Cell::from(e.ns.clone()), + Cell::from(dur(s.now - e.started_at.unwrap_or(s.now))), + Cell::from(format!("{:.1}c {} now", e.live_cpu, gb(e.live_mem_kb))).style(if over { + Style::default().fg(Color::Red) + } else { + Style::default() + }), + Cell::from(format!("needs {:.1}c {}{}", e.need_cpu, gb(e.need_mem_kb), est(e))), + ])); + } + super::tree::Row::Job { waiting: true, i, .. } => { + let w = &s.waiting[i]; + let e = &w.entry; + let blocked = report::main_blocker(&w.decision).map(|b| b.name().to_uppercase()).unwrap_or_else(|| "starting".into()) + + &w.eta_s.map(|t| format!(" ~{}", dur(t.max(1.0)))).unwrap_or_default(); + rows.push(Row::new(vec![ + Cell::from("WAIT").style(Style::default().fg(Color::Yellow)), + Cell::from(format!("{lead}{}", report::job_name(e, room))), + Cell::from(e.ns.clone()), + Cell::from(dur(s.now - e.queued_at)), + Cell::from(format!("{:.1}c {}{}", e.need_cpu, gb(e.need_mem_kb), est(e))), + Cell::from(blocked).style( + Style::default().fg(chart::blocker_color(report::main_blocker(&w.decision).map(|b| b.name()).unwrap_or(""))), + ), + ])); + } + super::tree::Row::Sub { outer, i, .. } => { + let Some(sub) = app.data.subs.get(&s.running[outer].pid).and_then(|v| v.get(i)) else { continue }; + rows.push( + Row::new(vec![ + Cell::from(""), + Cell::from(format!("{lead}{}", trunc(&sub.key, room))), + Cell::from(""), + Cell::from(dur(s.now - sub.since)), + Cell::from(format!("{:.1}c {} now", sub.cpu, gb(sub.mem_kb))), + Cell::from("inside, not queued"), + ]) + .style(DIM), + ); + } + } } let table = Table::new( rows, @@ -707,7 +752,16 @@ fn queue(f: &mut Frame, app: &mut App, area: Rect) { ) .header(Row::new(vec!["", "job", "namespace", "time", "needs / now", "blocked by"]).style(BOLD)) .row_highlight_style(Style::default().bg(Color::DarkGray)) - .block(Block::new().borders(Borders::TOP).title(format!(" {} running, {} waiting ", s.running.len(), s.waiting.len()))); + .block(Block::new().borders(Borders::TOP).title(format!( + " {} running, {} waiting{} ←/→ fold ", + s.running.len(), + s.waiting.len(), + match t.groups.len() { + 0 => String::new(), + 1 => ", in 1 task-runner run".into(), + n => format!(", in {n} task-runner runs"), + } + ))); render_list(f, app, table, list, 2, n_rows); // The Why panel. @@ -767,6 +821,47 @@ fn queue(f: &mut Frame, app: &mut App, area: Rect) { )); } lines.push(Line::styled("g start it now e change its needs for this run Enter details", DIM)); + } else if let Some(super::tree::Row::Group(g)) = t.rows.get(app.sel) { + let g = &t.groups[*g]; + lines.push(Line::styled(format!("{} ({})", g.label, g.id), BOLD)); + lines.push(Line::raw(format!( + "One task-runner run: {} of its jobs run and {} wait. It queued its first job {} ago.", + g.running, + g.waiting, + dur(s.now - g.since) + ))); + lines.push(Line::raw(format!( + "Its running jobs use {:.1} cores and {} now, and need {} in all.", + g.cpu, + gb(g.mem_kb), + gb(g.need_mem_kb) + ))); + let mut why: std::collections::BTreeMap<&str, usize> = Default::default(); + for w in s.waiting.iter().filter(|w| w.entry.pipeline.as_deref() == Some(g.id.as_str())) { + *why.entry(report::main_blocker(&w.decision).map(|b| b.name()).unwrap_or("starting")).or_default() += 1; + } + if !why.is_empty() { + let parts: Vec = why.iter().map(|(k, n)| format!("{n} for {k}")).collect(); + lines.push(Line::raw(format!("Its waiting jobs wait: {}.", parts.join(", ")))); + } + lines.push(Line::styled("Each job queues on its own, by its own needs. Enter fold or unfold ← fold → unfold", DIM)); + } else if let Some(super::tree::Row::Sub { outer, i, .. }) = t.rows.get(app.sel) + && let Some(sub) = app.data.subs.get(&s.running[*outer].pid).and_then(|v| v.get(*i)) + { + let outer = &s.running[*outer]; + lines.push(Line::styled(format!("{} runs inside {}", sub.key, outer.key), BOLD)); + lines.push(Line::raw(format!( + "now {:.1} cores, {}; started {} ago (pid {})", + sub.cpu, + gb(sub.mem_kb), + dur(s.now - sub.since), + sub.pid + ))); + lines.push(Line::raw(format!( + "A taskguard call nested in a running job is not queued: it runs under {}'s claim, and its use counts as that job's.", + outer.key + ))); + lines.push(Line::styled("Enter details of this key ← fold", DIM)); } else if let Some((e, _)) = app.queue_row(app.sel) { lines.push(Line::styled(format!("{} is running", e.key), BOLD)); lines.push(Line::raw(format!( @@ -1405,7 +1500,8 @@ fn help(f: &mut Frame, area: Rect) { let text = "\ VIEWS 1 Overview machine load over time, with taskguard's jobs stacked per namespace - 2 Queue what runs and what waits, and a Why panel for the selected job + 2 Queue what runs and what waits, and a Why panel for the selected job; the jobs of one + task-runner run under one row, and calls nested in a job under that job 3 Runs past runs; the panel below shows why the selected run waited 4 Tasks every task taskguard knows, per namespace: filter them, and prune their history 5 Trends commands whose memory, CPU or duration grows @@ -1424,6 +1520,8 @@ KEYS (? lists the keys of the view you are in) Overview: PgDn / PgUp move the chart back / forward in time; l (or End) goes back to live Enter picks a job at the time cursor, ↑/↓ select it, Enter opens that run, Esc back to the chart Queue: g start a waiting job now e change a job's needs for this run k stop a job + ← fold a task-runner run or a job's nested calls (on a row inside one, folds that) → unfold + Enter on a task-runner run folds or unfolds it p pause or resume a running job by hand (or: taskguard pause JOB / taskguard resume JOB) PAUSE: a paused job. Paused for memory (auto_pause, in Config): nothing new starts before it resumes. Paused by hand, or with kill -STOP: it stays paused until resumed by hand @@ -1681,6 +1779,77 @@ mod tests { assert!(s.contains("Why is packages/small:tsc waiting?"), "{s}"); } + #[test] + fn queue_shows_task_runner_runs_and_nested_calls_as_a_tree() { + let (_t, mut app) = fixture(); + let sleep = |s: &str| std::process::Command::new("sleep").arg(s).spawn().unwrap(); + // A stand-in for turbo, a call nested in the stack, and the owner of a turbo job. + let mut kids = [sleep("101"), sleep("102"), sleep("103")]; + let (turbo, nested, owner) = (kids[0].id() as i32, kids[1].id() as i32, kids[2].id() as i32); + let q = crate::queue::Queue::open(&app.dir).unwrap(); + let me = std::process::id() as i32; + let base = crate::queue::Entry { ns: "dalp".into(), known: true, version: Some("0.9.0".into()), ..Default::default() }; + let run = format!("turbo:{turbo}"); + let stack = crate::queue::Entry { pid: me, key: "dev-env:stack".into(), started_at: Some(now() - 60.0), ..base.clone() }; + let lint = crate::queue::Entry { + pid: owner, + key: "web/dapp:oxlint".into(), + started_at: Some(now() - 20.0), + pipeline: Some(run.clone()), + pipeline_since: Some(now() - 30.0), + ..base.clone() + }; + let check = crate::queue::Entry { + pid: me, + ticket: 2, + key: "web/dapp:bun_check".into(), + queued_at: now() - 10.0, + pipeline: Some(run), + pipeline_since: Some(now() - 30.0), + ..base + }; + q.write(&q.run_path(me), &stack).unwrap(); + q.write(&q.run_path(owner), &lint).unwrap(); + q.write(&q.wait_path(&check), &check).unwrap(); + q.note_nested(me, nested, "dalp:prepare"); + app.view = View::Queue; + app.load().unwrap(); + let s = screen(&mut app, 150, 30); + let at = |t: &str| s.find(t).unwrap_or_else(|| panic!("{t} missing:\n{s}")); + assert!(at("dev-env:stack") < at("└ dalp:prepare"), "the nested call follows its job:\n{s}"); + assert!(at("├ web/dapp:oxlint") < at("└ web/dapp:bun_check"), "jobs of a run get tree lines:\n{s}"); + assert!(at("sleep 101") < at("web/dapp:oxlint"), "the run is named by its command:\n{s}"); + assert!(at("web/dapp:oxlint") < at("web/dapp:bun_check"), "{s}"); + assert!(s.contains("in 1 task-runner run"), "{s}"); + + let press = |app: &mut App, code| { + app.key(crossterm::event::KeyEvent::new(code, crossterm::event::KeyModifiers::NONE)); + }; + // Rows: the stack, its nested call, the run, its two jobs. ← on a job inside the run folds the run. + app.sel = 3; + press(&mut app, KeyCode::Left); + assert_eq!(app.sel, 2, "the fold selects the run"); + let s = screen(&mut app, 150, 30); + assert!(!s.contains("web/dapp:oxlint") && s.contains("▸"), "{s}"); + assert!(s.contains("1 run 1 wait"), "{s}"); + press(&mut app, KeyCode::Enter); + assert!(screen(&mut app, 150, 30).contains("web/dapp:oxlint"), "Enter unfolds the run"); + press(&mut app, KeyCode::Enter); + assert!(!screen(&mut app, 150, 30).contains("web/dapp:oxlint"), "and folds it again"); + assert_eq!(app.view, View::Queue, "a run's row opens no details"); + app.sel = 0; + press(&mut app, KeyCode::Left); + let s = screen(&mut app, 150, 30); + assert!(!s.contains("dalp:prepare"), "{s}"); + press(&mut app, KeyCode::Right); + let s = screen(&mut app, 150, 30); + assert!(s.contains("dalp:prepare"), "→ unfolds:\n{s}"); + for k in &mut kids { + let _ = k.kill(); + let _ = k.wait(); + } + } + #[test] fn enter_opens_a_job_and_esc_returns_to_the_same_row() { let (_t, mut app) = fixture(); diff --git a/tests/cli.rs b/tests/cli.rs index d033557..533c4d8 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -296,19 +296,33 @@ fn a_command_that_wraps_taskguard_jobs_runs_straight_through_next_time() { let inner = format!("{} -- sh -c 'echo inner'", env!("CARGO_BIN_EXE_taskguard")); let first = e.run(&["--", "sh", "-c", &inner]); assert!(first.status.success(), "{}", stderr(&first)); - assert!(!stderr(&first).contains("straight through"), "{}", stderr(&first)); + assert!(!stderr(&first).contains("] straight through"), "{}", stderr(&first)); + assert!(stderr(&first).contains("taskguard --wraps sh -c"), "it advises the flag: {}", stderr(&first)); let nested: i64 = e.db().query_row("SELECT nested_jobs FROM runs", [], |r| r.get(0)).unwrap(); assert_eq!(nested, 1); let second = e.run(&["--", "sh", "-c", &inner]); assert!(second.status.success(), "{}", stderr(&second)); - assert!(stderr(&second).contains("straight through"), "{}", stderr(&second)); + assert!(stderr(&second).contains("] straight through"), "{}", stderr(&second)); let keys: Vec = e.db().prepare("SELECT key FROM runs ORDER BY id").unwrap().query_map([], |r| r.get(0)).unwrap().map(Result::unwrap).collect(); assert_eq!(keys.len(), 2, "the wrapper once, then the job inside it: {keys:?}"); assert_ne!(keys[0], keys[1]); } +#[test] +fn wraps_runs_straight_through_from_the_first_run() { + let e = Env::new(""); + let inner = format!("{} -- sh -c 'echo inner'", env!("CARGO_BIN_EXE_taskguard")); + let o = e.run(&["--wraps", "--", "sh", "-c", &inner]); + assert!(o.status.success(), "{}", stderr(&o)); + assert!(stderr(&o).contains("straight through") && stderr(&o).contains("--wraps"), "{}", stderr(&o)); + let keys: Vec = + e.db().prepare("SELECT key FROM runs").unwrap().query_map([], |r| r.get(0)).unwrap().map(Result::unwrap).collect(); + assert_eq!(keys.len(), 1, "only the job inside it is recorded: {keys:?}"); + assert!(!keys[0].contains("taskguard"), "{keys:?}"); +} + /// Wait until a waiting entry exists for `pid`, so a nudge reaches a job that waits. fn wait_until_queued(e: &Env, pid: u32) { let t = Instant::now();