diff --git a/src/prune.rs b/src/prune.rs index 200b16b..ea794bb 100644 --- a/src/prune.rs +++ b/src/prune.rs @@ -135,42 +135,89 @@ pub fn format_bytes(bytes: u64) -> String { } } -/// Total size of `path` in bytes: a file's length, or the recursive walk -/// of a directory. Unreadable entries count as zero — prune removes them -/// all the same. The walk never crosses a mount point (`du -x`): residual -/// chroots hold bind-mounted `/proc` and overlay filesystems whose virtual -/// files (`/proc/kcore` alone reports 128 TiB) would poison the total. -fn path_bytes(path: &Path) -> u64 { +/// Total size of every artifact in bytes, measured concurrently: the +/// records hold hundreds of thousands of files (a single kernel build +/// tree is ~100 GB of objects), and a serial walk costs tens of seconds. +/// +/// Directory walks never cross a mount point (`du -x`): residual chroots +/// hold bind-mounted `/proc` and overlay filesystems whose virtual files +/// (`/proc/kcore` alone reports 128 TiB) would poison the totals. +/// Unreadable entries count as zero — prune removes them all the same. +fn size_artifacts(artifacts: &[&Artifact]) -> Vec { + use std::collections::VecDeque; use std::os::unix::fs::MetadataExt; + use std::sync::Mutex; + use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; - let Ok(meta) = fs::symlink_metadata(path) else { - return 0; - }; - if !meta.is_dir() { - return meta.len(); - } - let root_dev = meta.dev(); - let mut total = 0; - let mut stack = vec![path.to_path_buf()]; - while let Some(dir) = stack.pop() { - let Ok(entries) = fs::read_dir(&dir) else { - continue; - }; - for entry in entries.flatten() { - let path = entry.path(); - let Ok(meta) = fs::symlink_metadata(&path) else { - continue; - }; - if meta.is_dir() { - if meta.dev() == root_dev { - stack.push(path); + // The device each artifact lives on: the walk descends only into + // directories on the same device. + let root_devs: Vec = artifacts + .iter() + .map(|a| fs::symlink_metadata(a.path()).map_or(0, |meta| meta.dev())) + .collect(); + + let totals: Vec = (0..artifacts.len()).map(|_| AtomicU64::new(0)).collect(); + let queue: Mutex> = Mutex::new( + artifacts + .iter() + .enumerate() + .map(|(idx, artifact)| (idx, artifact.path().to_path_buf())) + .collect(), + ); + // Every queued-but-unfinished item: the workers stop when the queue + // is empty and nothing is in flight. + let pending = AtomicUsize::new(artifacts.len()); + + let workers = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(4) + .min(16); + std::thread::scope(|scope| { + for _ in 0..workers { + scope.spawn(|| { + loop { + let item = queue.lock().unwrap().pop_front(); + let Some((idx, path)) = item else { + if pending.load(Ordering::Acquire) == 0 { + return; + } + std::thread::yield_now(); + continue; + }; + + let Ok(meta) = fs::symlink_metadata(&path) else { + pending.fetch_sub(1, Ordering::AcqRel); + continue; + }; + if !meta.is_dir() { + totals[idx].fetch_add(meta.len(), Ordering::Relaxed); + pending.fetch_sub(1, Ordering::AcqRel); + continue; + } + let Ok(entries) = fs::read_dir(&path) else { + pending.fetch_sub(1, Ordering::AcqRel); + continue; + }; + for entry in entries.flatten() { + let child = entry.path(); + let Ok(meta) = fs::symlink_metadata(&child) else { + continue; + }; + let same_device = meta.dev() == root_devs[idx]; + if meta.is_dir() && same_device { + pending.fetch_add(1, Ordering::AcqRel); + queue.lock().unwrap().push_back((idx, child)); + } else if !meta.is_dir() && same_device { + totals[idx].fetch_add(meta.len(), Ordering::Relaxed); + } + } + pending.fetch_sub(1, Ordering::AcqRel); } - } else { - total += meta.len(); - } + }); } - } - total + }); + + totals.into_iter().map(|total| total.into_inner()).collect() } /// Number of newest build logs kept when `--all` is not passed. @@ -759,7 +806,7 @@ pub fn prune_in_roots( // The actionable subset with sizes measured BEFORE anything is // removed: a removed tree has no size left to report. - let mut actionable: Vec<(&Artifact, u64)> = Vec::new(); + let mut filtered: Vec<&Artifact> = Vec::new(); for artifact in &artifacts { // Cached tarballs are only removed when --all is requested: they are // expensive to re-download. @@ -774,12 +821,14 @@ pub fn prune_in_roots( continue; } - actionable.push((artifact, path_bytes(artifact.path()))); + filtered.push(artifact); } + let sizes = size_artifacts(&filtered); + let actionable: Vec<(&Artifact, u64)> = filtered.iter().copied().zip(sizes).collect(); for (artifact, _) in &actionable { let path = artifact.path(); - log::info!("{}: {}", artifact.kind(), path.display()); + log::debug!("{}: {}", artifact.kind(), path.display()); if options.dry_run { report.removed.push(path.to_path_buf());