utils/size,prune: share the concurrent tree walker and byte formatting
The incremental fork's staged-tree copy wants to report byte progress against the same measurements the prune report is built from: move the concurrent size walk and the human-readable byte formatting out of prune into utils::size and rewire prune to them. Behavior is unchanged: same-device walks, unreadable entries count as zero.
This commit is contained in:
+5
-98
@@ -122,107 +122,14 @@ impl PruneGroup {
|
|||||||
/// Human-readable size for the prune report ("512 B", "3.5 MiB",
|
/// Human-readable size for the prune report ("512 B", "3.5 MiB",
|
||||||
/// "1.2 GiB", "1.3 TiB").
|
/// "1.2 GiB", "1.3 TiB").
|
||||||
pub fn format_bytes(bytes: u64) -> String {
|
pub fn format_bytes(bytes: u64) -> String {
|
||||||
const KIB: f64 = 1024.0;
|
crate::utils::size::format_bytes(bytes)
|
||||||
const MIB: f64 = 1024.0 * KIB;
|
|
||||||
const GIB: f64 = 1024.0 * MIB;
|
|
||||||
const TIB: f64 = 1024.0 * GIB;
|
|
||||||
let size = bytes as f64;
|
|
||||||
if size >= TIB {
|
|
||||||
format!("{:.1} TiB", size / TIB)
|
|
||||||
} else if size >= GIB {
|
|
||||||
format!("{:.1} GiB", size / GIB)
|
|
||||||
} else if size >= MIB {
|
|
||||||
format!("{:.1} MiB", size / MIB)
|
|
||||||
} else if size >= KIB {
|
|
||||||
format!("{:.1} KiB", size / KIB)
|
|
||||||
} else {
|
|
||||||
format!("{bytes} B")
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Total size of every artifact in bytes, measured concurrently: the
|
/// Total size of every artifact in bytes, measured concurrently (the
|
||||||
/// records hold hundreds of thousands of files (a single kernel build
|
/// walker of [`crate::utils::size`]).
|
||||||
/// 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> {
|
fn size_artifacts(artifacts: &[&Artifact]) -> Vec<u64> {
|
||||||
use std::collections::VecDeque;
|
let paths: Vec<&Path> = artifacts.iter().map(|a| a.path()).collect();
|
||||||
use std::os::unix::fs::MetadataExt;
|
crate::utils::size::tree_sizes(&paths)
|
||||||
use std::sync::Mutex;
|
|
||||||
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
|
|
||||||
|
|
||||||
// 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);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
totals.into_iter().map(|total| total.into_inner()).collect()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Number of newest build logs kept when `--all` is not passed.
|
/// Number of newest build logs kept when `--all` is not passed.
|
||||||
|
|||||||
@@ -1,2 +1,4 @@
|
|||||||
pub mod gpg;
|
pub mod gpg;
|
||||||
pub mod root;
|
pub mod root;
|
||||||
|
/// Directory-tree sizing (concurrent walker) and byte formatting
|
||||||
|
pub mod size;
|
||||||
|
|||||||
@@ -0,0 +1,162 @@
|
|||||||
|
//! Directory-tree sizing: the concurrent walker shared by the prune
|
||||||
|
//! report and the staged-tree copy progress, plus the byte formatting
|
||||||
|
//! both display.
|
||||||
|
|
||||||
|
use std::fs;
|
||||||
|
use std::path::{Path, PathBuf};
|
||||||
|
|
||||||
|
/// Total size of every tree in `paths`, in bytes, measured concurrently:
|
||||||
|
/// the sized trees 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.
|
||||||
|
pub(crate) fn tree_sizes(paths: &[&Path]) -> Vec<u64> {
|
||||||
|
use std::collections::VecDeque;
|
||||||
|
use std::os::unix::fs::MetadataExt;
|
||||||
|
use std::sync::Mutex;
|
||||||
|
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
|
||||||
|
|
||||||
|
// The device each tree lives on: the walk descends only into
|
||||||
|
// directories on the same device.
|
||||||
|
let root_devs: Vec<u64> = paths
|
||||||
|
.iter()
|
||||||
|
.map(|path| fs::symlink_metadata(path).map_or(0, |meta| meta.dev()))
|
||||||
|
.collect();
|
||||||
|
|
||||||
|
let totals: Vec<AtomicU64> = (0..paths.len()).map(|_| AtomicU64::new(0)).collect();
|
||||||
|
let queue: Mutex<VecDeque<(usize, PathBuf)>> = Mutex::new(
|
||||||
|
paths
|
||||||
|
.iter()
|
||||||
|
.enumerate()
|
||||||
|
.map(|(idx, path)| (idx, 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(paths.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);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
totals.into_iter().map(|total| total.into_inner()).collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Human-readable size ("512 B", "3.5 MiB", "1.2 GiB", "1.3 TiB").
|
||||||
|
pub(crate) fn format_bytes(bytes: u64) -> String {
|
||||||
|
const KIB: f64 = 1024.0;
|
||||||
|
const MIB: f64 = 1024.0 * KIB;
|
||||||
|
const GIB: f64 = 1024.0 * MIB;
|
||||||
|
const TIB: f64 = 1024.0 * GIB;
|
||||||
|
let size = bytes as f64;
|
||||||
|
if size >= TIB {
|
||||||
|
format!("{:.1} TiB", size / TIB)
|
||||||
|
} else if size >= GIB {
|
||||||
|
format!("{:.1} GiB", size / GIB)
|
||||||
|
} else if size >= MIB {
|
||||||
|
format!("{:.1} MiB", size / MIB)
|
||||||
|
} else if size >= KIB {
|
||||||
|
format!("{:.1} KiB", size / KIB)
|
||||||
|
} else {
|
||||||
|
format!("{bytes} B")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use std::fs;
|
||||||
|
use tempfile::tempdir;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn sizes_files_directories_and_missing_roots() {
|
||||||
|
let temp = tempdir().unwrap();
|
||||||
|
let tree = temp.path().join("tree");
|
||||||
|
fs::create_dir_all(tree.join("sub")).unwrap();
|
||||||
|
fs::write(tree.join("sub/obj.o"), vec![0u8; 3000]).unwrap();
|
||||||
|
fs::write(tree.join("top"), vec![0u8; 10]).unwrap();
|
||||||
|
let file = temp.path().join("file");
|
||||||
|
fs::write(&file, vec![0u8; 5]).unwrap();
|
||||||
|
let missing = temp.path().join("missing");
|
||||||
|
|
||||||
|
let sizes = tree_sizes(&[tree.as_path(), file.as_path(), missing.as_path()]);
|
||||||
|
assert_eq!(sizes, vec![3010, 5, 0]);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn symlinks_count_as_themselves_never_their_target() {
|
||||||
|
let temp = tempdir().unwrap();
|
||||||
|
let target = temp.path().join("target");
|
||||||
|
fs::write(&target, vec![0u8; 100]).unwrap();
|
||||||
|
let link = temp.path().join("link");
|
||||||
|
std::os::unix::fs::symlink(&target, &link).unwrap();
|
||||||
|
|
||||||
|
// symlink_metadata: the link's own size (the stored target path),
|
||||||
|
// never the target's content.
|
||||||
|
assert_eq!(
|
||||||
|
tree_sizes(&[link.as_path()]),
|
||||||
|
vec![target.as_os_str().len() as u64]
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn formats_bytes_like_the_reports_display_them() {
|
||||||
|
assert_eq!(format_bytes(0), "0 B");
|
||||||
|
assert_eq!(format_bytes(512), "512 B");
|
||||||
|
assert_eq!(format_bytes(2048), "2.0 KiB");
|
||||||
|
assert_eq!(format_bytes(5 * 1024 * 1024), "5.0 MiB");
|
||||||
|
assert_eq!(format_bytes(3 * 1024 * 1024 * 1024), "3.0 GiB");
|
||||||
|
// /proc/kcore inside a bind-mounted chroot /proc reports exactly
|
||||||
|
// this size; the mount-crossing walk keeps it out of the totals.
|
||||||
|
assert_eq!(format_bytes(140_737_477_959_680), "128.0 TiB");
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user