prune: size the artifacts concurrently, drop the per-path info lines
The dry run still spammed one info line per artifact before the grouped digest, and the size walk was serial: tens of seconds of pure readdir/stat over the records' hundreds of thousands of files. The per-path lines drop to debug (the grouped report is the user-facing output), and the size pass now fans out over a shared work queue with scoped threads — directory-level parallelism, same du -x mount-crossing rule. 16s to 6.4s wall on a 440 GiB sessions tree, totals unchanged.
This commit is contained in:
+84
-35
@@ -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<u64> {
|
||||
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<u64> = artifacts
|
||||
.iter()
|
||||
.map(|a| fs::symlink_metadata(a.path()).map_or(0, |meta| meta.dev()))
|
||||
.collect();
|
||||
|
||||
let totals: Vec<AtomicU64> = (0..artifacts.len()).map(|_| AtomicU64::new(0)).collect();
|
||||
let queue: Mutex<VecDeque<(usize, PathBuf)>> = 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());
|
||||
|
||||
Reference in New Issue
Block a user