crates/mimir-driver/src/bin/mimir-super.rs
//! `mimir-super <job.json>` — **the super build** (dv2-b1-super): every repo of
//! a workspace compiled at once over ONE target dir, the Mímir driver dumping
//! as each unit compiles.
//!
//! * one `cargo-mimir-dump` per repo, running together, at most
//! `parallel_repos` at a time; a repo starts once the units an EARLIER repo
//! (the planner's order, `repos` as listed) builds THE SAME WAY are printed
//! by that repo (`waits_on`) — cargo decides freshness when it plans, so
//! started earlier it would compile them again, or deadlock on their locks;
//! * cargo's `-Zfine-grain-locking`: the processes share the target dir unit
//! by unit — a unit two repos need (same cargo unit hash) is compiled ONCE,
//! the second waits for it and finds it fresh. Cargo's own unit identity
//! stays the truth; nothing is unified across repos;
//! * ONE jobserver (`jobs` tokens) handed to every cargo: the whole graph
//! shares the cores, a repo's tail overlaps the next repo's start.
//!
//! Per repo it writes `<out>/.super/{cargo.json,cargo.err,exit,ms}` (stdout,
//! stderr, exit code, wall ms) — what the consumer reads instead of running
//! the repo's own compile — and prints `<repo>\t<line>` for each stdout line
//! (progress), `@done\t<repo>\t<exit>\t<ms>` as each repo ends.
//!
//! **Deadlock watchdog** (measured 2026-10-16 on dwarf: korp-collectors' cargo
//! held a READ lock on `tracing`'s unit and waited to WRITE `uuid`, `nix`,
//! `gix-sec`, `log`, while znippy-zoomies' cargo held READ on those and waited
//! to WRITE `tracing` — 16 minutes, no process used a CPU tick, the super
//! build never ended): when no repo printed a line and no process of a
//! running repo used CPU for `MIMIR_SUPER_STALL_SECS` (default 180) — or for
//! `MIMIR_SUPER_LOCK_STALL_SECS` (default 30) while every running repo has a
//! process waiting on a file lock — every running repo but the
//! earliest-started is stopped, in one round (its whole process group: cargo,
//! rustc, build scripts; one repo alone is stopped itself) and compiled again
//! once nothing else runs, with a fresh jobserver (tokens its stopped
//! processes held are gone). Stopping the latest only left a cycle of three
//! two-deadlocked for a second round (loki 2026-10-17, 6 minutes). A stopped
//! repo's record is the second run's. At most two stops per repo.
use std::collections::HashMap;
use std::io::{BufRead, BufReader, Write};
use std::os::unix::process::CommandExt;
use std::path::PathBuf;
use std::process::{Command, ExitCode, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, mpsc};
use std::time::{Duration, Instant};
#[derive(serde::Deserialize)]
struct Job {
cargo_mimir_dump: PathBuf,
jobs: usize,
#[serde(default)]
parallel_repos: usize,
repos: Vec<RepoJob>,
}
#[derive(serde::Deserialize)]
struct RepoJob {
repo: String,
tree: PathBuf,
out: PathBuf,
envs: Vec<(String, String)>,
args: Vec<String>,
/// Units an earlier repo builds the same way: this repo starts once that
/// repo printed them (cargo plans freshness at its start — started earlier
/// it would compile them again, or deadlock on their unit locks).
#[serde(default)]
waits_on: Vec<WaitOn>,
}
/// The earlier repo `after` (index in the job) has printed `count`
/// `compiler-artifact` lines of `key` (`package_id|kind:name|features`).
#[derive(serde::Deserialize)]
struct WaitOn {
after: usize,
key: String,
count: u32,
}
/// A repo may start: every repo it waits on printed the units it waits for,
/// or has ended (a unit that never builds holds nobody up). Lines count per
/// printing repo — another repo printing another variant of the key is no
/// sign the waited-for unit is built.
fn waits_met(waits: &[WaitOn], printed: &HashMap<(usize, String), u32>, done: &[bool]) -> bool {
waits.iter().all(|w| done.get(w.after).copied().unwrap_or(true) || printed.get(&(w.after, w.key.clone())).copied().unwrap_or(0) >= w.count)
}
/// The key of a compiled unit in a cargo JSON line (`compiler-artifact`).
fn artifact_key(line: &str) -> Option<String> {
let v: serde_json::Value = serde_json::from_str(line).ok()?;
if v["reason"].as_str() != Some("compiler-artifact") {
return None;
}
let mut f: Vec<&str> = v["features"].as_array()?.iter().filter_map(|x| x.as_str()).collect();
f.sort_unstable();
Some(format!(
"{}|{}:{}|{}",
v["package_id"].as_str()?,
v["target"]["kind"].as_array()?.first()?.as_str()?,
v["target"]["name"].as_str()?,
f.join(",")
))
}
enum Ev {
/// A repo's tool runs, as process group `pgid`.
Started(usize, i32),
Line(usize, String, String),
/// (index, exit code, wall ms, stopped by the watchdog)
Done(usize, i32, u128, bool),
}
/// Start one repo's compile on a thread of the scope, as its own process
/// group; its end is an [`Ev::Done`]. A run the watchdog stops (`stopped`)
/// leaves no record: the next run writes it.
fn start<'scope, 'env>(s: &'scope std::thread::Scope<'scope, 'env>, i: usize, r: &'env RepoJob, tool: &'env std::path::Path, client: jobserver::Client, stopped: Arc<AtomicBool>, tx: mpsc::Sender<Ev>) {
let dir = r.out.join(".super");
s.spawn(move || {
let t = Instant::now();
let code = (|| -> Result<i32, String> {
std::fs::create_dir_all(&dir).map_err(|e| format!("{}: {e}", dir.display()))?;
let err = std::fs::File::create(dir.join("cargo.err")).map_err(|e| format!("{}: {e}", dir.display()))?;
let mut cmd = Command::new(tool);
cmd.args(&r.args).current_dir(&r.tree).stdin(Stdio::null()).stdout(Stdio::piped()).stderr(err).process_group(0);
for (k, v) in &r.envs {
cmd.env(k, v);
}
// repos share the target dir unit by unit, and the cores through one jobserver
cmd.env("CARGO_UNSTABLE_FINE_GRAIN_LOCKING", "true");
client.configure(&mut cmd);
let mut child = cmd.spawn().map_err(|e| format!("cannot start {}: {e}", tool.display()))?;
let _ = tx.send(Ev::Started(i, child.id() as i32));
let mut file = std::fs::File::create(dir.join("cargo.json")).map_err(|e| format!("{}: {e}", dir.display()))?;
if let Some(out) = child.stdout.take() {
for line in BufReader::new(out).lines().map_while(Result::ok) {
let _ = writeln!(file, "{line}");
let _ = tx.send(Ev::Line(i, r.repo.clone(), line));
}
}
Ok(child.wait().ok().and_then(|st| st.code()).unwrap_or(255))
})()
.unwrap_or_else(|e| {
eprintln!("mimir-super: {}: {e}", r.repo);
127
});
let ms = t.elapsed().as_millis();
let was_stopped = stopped.load(Ordering::SeqCst);
if !was_stopped {
let _ = std::fs::write(dir.join("exit"), code.to_string());
let _ = std::fs::write(dir.join("ms"), ms.to_string());
}
let _ = tx.send(Ev::Done(i, code, ms, was_stopped));
});
}
/// CPU ticks (user + system) of every process in the process groups `pgids`,
/// and the groups with a thread waiting on a file lock (`flock`, cargo's unit
/// locks), read from `/proc`. `None` when `/proc` cannot be read.
fn group_ticks(pgids: &[i32]) -> Option<(u64, std::collections::HashSet<i32>)> {
let mut total = 0u64;
let mut waiting = std::collections::HashSet::new();
for e in std::fs::read_dir("/proc").ok()?.flatten() {
let name = e.file_name();
let Some(pid) = name.to_str().filter(|n| n.bytes().all(|b| b.is_ascii_digit())) else { continue };
let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else { continue };
if let Some((pgrp, ticks)) = stat_pgrp_ticks(&stat)
&& pgids.contains(&pgrp)
{
total += ticks;
if !waiting.contains(&pgrp) && lock_waiting(pid) {
waiting.insert(pgrp);
}
}
}
Some((total, waiting))
}
/// A thread of process `pid` sleeps in the kernel's file-lock wait.
fn lock_waiting(pid: &str) -> bool {
let Ok(tasks) = std::fs::read_dir(format!("/proc/{pid}/task")) else { return false };
tasks.flatten().any(|t| std::fs::read_to_string(t.path().join("wchan")).is_ok_and(|w| w.trim() == "locks_lock_inode_wait"))
}
/// `(pgrp, utime + stime)` of one `/proc/<pid>/stat` line (the comm field may
/// hold spaces and parentheses: the fields after its LAST `)` are counted).
fn stat_pgrp_ticks(stat: &str) -> Option<(i32, u64)> {
let rest = &stat[stat.rfind(')')? + 1..];
let f: Vec<&str> = rest.split_whitespace().collect();
// after comm: state(0) ppid(1) pgrp(2) … utime(11) stime(12)
Some((f.get(2)?.parse().ok()?, f.get(11)?.parse::<u64>().ok()? + f.get(12)?.parse::<u64>().ok()?))
}
/// The watchdog's rule: the running repos are deadlocked when nothing was
/// printed for `stall` and their processes used next to no CPU over the last
/// `stall` — at most 1% of one core (`used_ticks` of `clk_tck` per second;
/// `None` when `/proc` cannot say, or the samples do not cover the window
/// yet: then never). Not NO cpu: a cargo stuck on a unit lock still ticks
/// its timer threads now and then (dwarf 2026-10-16: one tick in 15 s, and
/// five cargos waited 30 minutes for a rule of zero).
fn stalled(quiet_for: Duration, used_ticks: Option<u64>, stall: Duration, clk_tck: u64) -> bool {
quiet_for >= stall && used_ticks.is_some_and(|t| t * 100 <= stall.as_secs().max(1) * clk_tck)
}
/// CPU ticks used over the last `window`, from `(when, ticks)` samples (oldest
/// first): `None` until a sample at least `window` old is there.
fn used_over(samples: &std::collections::VecDeque<(Instant, u64)>, now: (Instant, u64), window: Duration) -> Option<u64> {
let base = samples.iter().rev().find(|(t, _)| now.0.duration_since(*t) >= window)?;
Some(now.1.saturating_sub(base.1))
}
/// The repos the watchdog stops: every running one but the earliest-started
/// (it then holds no lock another stuck repo waits for); one alone is itself.
fn victims(running: &[(usize, Instant)]) -> Vec<usize> {
let mut r = running.to_vec();
r.sort_by_key(|(i, t)| (*t, *i));
if r.len() > 1 { r[1..].iter().map(|(i, _)| *i).collect() } else { r.iter().map(|(i, _)| *i).collect() }
}
fn main() -> ExitCode {
let Some(path) = std::env::args().nth(1) else {
eprintln!("usage: mimir-super <job.json>");
return ExitCode::from(2);
};
let job: Job = match std::fs::read(&path).map_err(|e| e.to_string()).and_then(|b| serde_json::from_slice(&b).map_err(|e| e.to_string())) {
Ok(j) => j,
Err(e) => {
eprintln!("mimir-super: {path}: {e}");
return ExitCode::from(2);
}
};
let jobs = job.jobs.max(1);
let client = match jobserver::Client::new(jobs) {
Ok(c) => c,
Err(e) => {
eprintln!("mimir-super: no jobserver: {e}");
return ExitCode::from(2);
}
};
let stall = Duration::from_secs(std::env::var("MIMIR_SUPER_STALL_SECS").ok().and_then(|v| v.parse().ok()).unwrap_or(180));
let lock_stall = Duration::from_secs(std::env::var("MIMIR_SUPER_LOCK_STALL_SECS").ok().and_then(|v| v.parse().ok()).unwrap_or(30));
let window = stall.max(lock_stall);
let tick = (stall.min(lock_stall) / 4).clamp(Duration::from_millis(250), Duration::from_secs(30));
let parallel = if job.parallel_repos == 0 { job.repos.len() } else { job.parallel_repos };
let (tx, rx) = mpsc::channel::<Ev>();
let t0 = Instant::now();
let mut failed = 0usize;
std::thread::scope(|s| {
let n = job.repos.len();
let mut started = vec![false; n];
let mut done = vec![false; n];
// stopped by the watchdog, compiled again once nothing else runs
let mut retry = vec![false; n];
let mut stops = vec![0u32; n];
let mut flags: Vec<Arc<AtomicBool>> = (0..n).map(|_| Arc::new(AtomicBool::new(false))).collect();
let mut groups: HashMap<usize, (i32, Instant)> = HashMap::new();
let mut running = 0usize;
let mut printed: HashMap<(usize, String), u32> = HashMap::new();
let stdout = std::io::stdout();
let mut last_line = Instant::now();
// the running repos' CPU ticks, sampled once per tick (not per line:
// a /proc scan per cargo line cost 274 s of system time in one run)
let clk_tck = (unsafe { libc::sysconf(libc::_SC_CLK_TCK) }).max(1) as u64;
let mut samples: std::collections::VecDeque<(Instant, u64)> = std::collections::VecDeque::new();
let mut last_scan = Instant::now();
loop {
// start every repo that may start: its shared units are built, or
// every repo before it has ended (a unit that never builds holds
// nobody up); a stopped one only when nothing else runs
for i in 0..n {
if started[i] || running >= parallel {
continue;
}
let ready = if retry[i] { running == 0 } else { waits_met(&job.repos[i].waits_on, &printed, &done) || (0..i).all(|j| done[j]) };
if ready {
started[i] = true;
running += 1;
let _ = writeln!(stdout.lock(), "@start\t{}", job.repos[i].repo);
// a wait met only by the waited-on repo's END is a unit
// it never printed: the plan and the compile disagree
let by_end = job.repos[i].waits_on.iter().filter(|w| printed.get(&(w.after, w.key.clone())).copied().unwrap_or(0) < w.count).count();
if by_end > 0 && !retry[i] {
eprintln!("mimir-super: {} starts with {by_end} of {} waits met by the end of the repo waited on, not by its lines", job.repos[i].repo, job.repos[i].waits_on.len());
}
// a run after a stop gets a fresh jobserver: tokens the
// stopped processes held are gone with them
let c = if retry[i] { jobserver::Client::new(jobs).unwrap_or_else(|_| client.clone()) } else { client.clone() };
flags[i] = Arc::new(AtomicBool::new(false));
start(s, i, &job.repos[i], &job.cargo_mimir_dump, c, flags[i].clone(), tx.clone());
last_line = Instant::now();
}
}
if running == 0 {
break;
}
match rx.recv_timeout(tick) {
Ok(Ev::Started(i, pgid)) => {
groups.insert(i, (pgid, Instant::now()));
}
Ok(Ev::Line(i, repo, line)) => {
last_line = Instant::now();
if let Some(k) = artifact_key(&line) {
*printed.entry((i, k)).or_default() += 1;
}
let _ = writeln!(stdout.lock(), "{repo}\t{line}");
}
Ok(Ev::Done(i, code, ms, was_stopped)) => {
running -= 1;
groups.remove(&i);
last_line = Instant::now();
if was_stopped {
started[i] = false;
retry[i] = true;
continue;
}
done[i] = true;
if code != 0 {
failed += 1;
}
let _ = writeln!(stdout.lock(), "@done\t{}\t{code}\t{ms}", job.repos[i].repo);
}
Err(mpsc::RecvTimeoutError::Timeout) => {}
Err(mpsc::RecvTimeoutError::Disconnected) => break,
}
// the watchdog: no line and no CPU in any running repo for `stall`
if last_scan.elapsed() < tick {
continue;
}
last_scan = Instant::now();
let pgids: Vec<i32> = groups.values().map(|(g, _)| *g).collect();
let Some((ticks, waiting)) = group_ticks(&pgids) else {
samples.clear();
continue;
};
let now = (Instant::now(), ticks);
let used = used_over(&samples, now, stall);
let used_short = used_over(&samples, now, lock_stall);
samples.push_back(now);
while samples.len() > 2 && samples.get(1).is_some_and(|(t, _)| now.0.duration_since(*t) >= window) {
samples.pop_front();
}
// every running repo stuck on a file lock: the short window is enough
let on_locks = !pgids.is_empty() && pgids.iter().all(|g| waiting.contains(g));
let quiet = last_line.elapsed();
if stalled(quiet, used, stall, clk_tck) || (on_locks && stalled(quiet, used_short, lock_stall, clk_tck)) {
let cands: Vec<(usize, Instant)> = groups.iter().filter(|(i, _)| stops[**i] < 2).map(|(i, (_, t))| (*i, *t)).collect();
let vs = victims(&cands);
if !vs.is_empty() {
let mut stuck: Vec<(usize, Instant)> = groups.iter().map(|(i, (_, t))| (*i, *t)).collect();
stuck.sort_by_key(|(i, t)| (*t, *i));
let names = |is: &mut dyn Iterator<Item = usize>| is.map(|i| job.repos[i].repo.as_str()).collect::<Vec<_>>().join(", ");
eprintln!(
"mimir-super: no line and no CPU for {} s in {}{} — cargo's unit locks deadlocked; {} {} stopped and compiled again after the others",
quiet.as_secs(),
names(&mut stuck.iter().map(|(i, _)| *i)),
if on_locks { " (each waiting on a file lock)" } else { "" },
names(&mut vs.iter().copied()),
if vs.len() == 1 { "is" } else { "are" },
);
for v in vs {
stops[v] += 1;
flags[v].store(true, Ordering::SeqCst);
if let Some((pgid, _)) = groups.get(&v) {
unsafe { libc::kill(-*pgid, libc::SIGKILL) };
}
}
last_line = Instant::now();
// the stopped groups' ticks leave the sum: start the window again
samples.clear();
}
}
}
});
eprintln!("mimir-super: {} repo(s) in {} ms, {failed} exited non-zero", job.repos.len(), t0.elapsed().as_millis());
ExitCode::SUCCESS
}
#[cfg(test)]
mod tests {
use super::*;
/// znippy-zoomies (2) waits for skade (1) to print nix's lib, as skade
/// builds the variant both need. nornir (0) printing ITS nix lib — same
/// key, another variant — does not start znippy-zoomies (loki
/// 2026-10-17: it did, and three repos deadlocked on the unit locks).
#[test]
fn a_wait_is_met_by_the_repo_it_waits_on_not_by_the_key() {
let key = "registry+https://github.com/rust-lang/crates.io-index#nix@0.31.3|lib:nix|default,mman".to_string();
let waits = vec![WaitOn { after: 1, key: key.clone(), count: 2 }];
let mut printed = HashMap::new();
let done = [false, false, false];
printed.insert((0, key.clone()), 3);
assert!(!waits_met(&waits, &printed, &done), "another repo's lines of the key");
printed.insert((1, key.clone()), 1);
assert!(!waits_met(&waits, &printed, &done), "one of skade's two units of the key");
printed.insert((1, key.clone()), 2);
assert!(waits_met(&waits, &printed, &done));
printed.clear();
assert!(waits_met(&waits, &printed, &[false, true, false]), "skade ended without printing them");
}
/// A `/proc/<pid>/stat` line: the process group and its CPU ticks, also
/// when the command name holds spaces and parentheses.
#[test]
fn a_stat_line_gives_the_group_and_the_ticks() {
let stat = "4141863 (cargo (x) y) S 4141743 4141700 4141700 0 -1 4194560 100 0 0 0 1234 56 0 0 20 0 7 0 1 2 3";
assert_eq!(stat_pgrp_ticks(stat), Some((4141700, 1290)));
assert_eq!(stat_pgrp_ticks("garbage"), None);
}
/// Deadlocked = no line AND next to no CPU (≤ 1% of a core) over the
/// whole stall; a quiet compile (one big crate, no line for minutes) burns
/// CPU and is not.
#[test]
fn only_silence_with_next_to_no_cpu_is_a_deadlock() {
let (s, m) = (Duration::from_secs(180), |x: u64| Duration::from_secs(x));
assert!(stalled(m(200), Some(0), s, 100));
assert!(stalled(m(200), Some(12), s, 100), "a timer thread ticking now and then");
assert!(!stalled(m(200), Some(18_000), s, 100), "a long compile prints nothing but burns a core");
assert!(!stalled(m(200), Some(181), s, 100), "over 1% of a core");
assert!(!stalled(m(10), Some(0), s, 100), "a line came");
assert!(!stalled(m(500), None, s, 100), "no /proc, or the window not covered: never");
}
/// The ticks used over the window come from the newest sample at least a
/// window old; none that old yet: not known.
#[test]
fn ticks_over_the_window_need_a_sample_that_old() {
let t0 = Instant::now();
let w = Duration::from_secs(10);
let mut s = std::collections::VecDeque::new();
assert_eq!(used_over(&s, (t0, 5), w), None);
s.push_back((t0, 100));
s.push_back((t0 + Duration::from_secs(5), 104));
assert_eq!(used_over(&s, (t0 + Duration::from_secs(9), 106), w), None);
assert_eq!(used_over(&s, (t0 + Duration::from_secs(12), 107), w), Some(7));
assert_eq!(used_over(&s, (t0 + Duration::from_secs(16), 109), w), Some(5));
}
/// The latest-started running repo is the one stopped.
#[test]
fn every_repo_but_the_earliest_started_is_stopped() {
let t = Instant::now();
let mut v = victims(&[(3, t), (5, t + Duration::from_secs(9)), (1, t + Duration::from_secs(2))]);
v.sort();
assert_eq!(v, vec![1, 5]);
assert_eq!(victims(&[(4, t)]), vec![4], "one stuck alone is stopped itself");
assert!(victims(&[]).is_empty());
}
}