docs: tools to measure how a range reader would read HDF5
inventory.py counts the functions and call sites that take the whole file as &[u8]; range-trace (standalone crate, x86-64 Linux) records every load clawhdf5 makes from a file by mprotect + single-step, unchanged library code; libhdf5_reads.py counts libhdf5's reads through h5py's fileobj driver and prints a dataset's chunk extents. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
osobh
@@ -0,0 +1,17 @@
|
||||
[package]
|
||||
name = "range-trace"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
rust-version = "1.92"
|
||||
publish = false
|
||||
description = "Records which byte ranges of an HDF5 file the clawhdf5 facade reads (docs/design/range-reads.md)"
|
||||
|
||||
# Outside the main workspace on purpose: a measurement tool for a design
|
||||
# document, never built by `cargo test --workspace`. x86-64 Linux only.
|
||||
[workspace]
|
||||
|
||||
[dependencies]
|
||||
# Default features minus nothing: `parallel` is off by default, and the tracer
|
||||
# relies on every read happening on the main thread.
|
||||
clawhdf5 = { path = "../../../../crates/clawhdf5" }
|
||||
libc = "0.2"
|
||||
@@ -0,0 +1,375 @@
|
||||
//! range-trace: which bytes of an HDF5 file does clawhdf5 read?
|
||||
//!
|
||||
//! Measurement tool for `docs/design/range-reads.md` (x86-64 Linux only).
|
||||
//!
|
||||
//! The file is loaded into a page-aligned buffer and handed to
|
||||
//! `clawhdf5::File::from_bytes`, so the library parses it exactly as it does
|
||||
//! today. The buffer is then `mprotect`ed to `PROT_NONE`. Every load from it
|
||||
//! faults; the SIGSEGV handler logs the exact faulting address, makes that
|
||||
//! page readable and sets the x86 trap flag, so the CPU single-steps the one
|
||||
//! instruction and the SIGTRAP handler re-protects the page. Every load
|
||||
//! instruction that touches the file is therefore recorded (with its first
|
||||
//! byte; widths are not decoded, see `ACCESS_WIDTH`).
|
||||
//!
|
||||
//! Bulk copies (raw data) would fault once per load; a page that faults more
|
||||
//! than `BULK_THRESHOLD` times in one phase is left readable for the rest of
|
||||
//! that phase and counted as wholly read ("bulk page").
|
||||
//!
|
||||
//! Phases: `open` (superblock), `list` (walk every group; for every dataset
|
||||
//! its shape and dtype, i.e. what `h5ls -r -v` or a tree view needs), and
|
||||
//! `read` (read the named dataset in full through `read_selection(All)`).
|
||||
//!
|
||||
//! Usage: range-trace FILE DATASET_PATH [RAW_EXTENTS]
|
||||
//!
|
||||
//! RAW_EXTENTS (optional) lists the dataset's stored data as `offset size`
|
||||
//! lines (absolute file offsets; `libhdf5_reads.py --extents` writes it from
|
||||
//! h5py). With it the `read` phase is split into `read (metadata)`, the
|
||||
//! loads outside those extents, and `read (raw data)`, the loads inside.
|
||||
//!
|
||||
//! Every read must happen on this thread: build without the facade's
|
||||
//! `parallel` feature (off by default).
|
||||
|
||||
use std::alloc::{Layout, alloc_zeroed};
|
||||
use std::collections::BTreeSet;
|
||||
use std::sync::atomic::{AtomicPtr, AtomicUsize, Ordering};
|
||||
|
||||
use clawhdf5::{File, Group, Selection};
|
||||
|
||||
const PAGE: usize = 4096;
|
||||
const BULK_THRESHOLD: u32 = 4096;
|
||||
/// Bytes assumed read by one logged load (the widest scalar load the parsers
|
||||
/// issue; SIMD copies are bulk anyway).
|
||||
const ACCESS_WIDTH: u64 = 8;
|
||||
/// Two accesses closer than this belong to the same structure, i.e. would be
|
||||
/// one range request.
|
||||
const MERGE_GAP: u64 = 64;
|
||||
const LOG_CAP: usize = 64 << 20;
|
||||
|
||||
static BUF_START: AtomicUsize = AtomicUsize::new(0);
|
||||
static BUF_LEN: AtomicUsize = AtomicUsize::new(0);
|
||||
static LOG: AtomicPtr<u64> = AtomicPtr::new(std::ptr::null_mut());
|
||||
static LOG_LEN: AtomicUsize = AtomicUsize::new(0);
|
||||
static HITS: AtomicPtr<u32> = AtomicPtr::new(std::ptr::null_mut());
|
||||
static BULK: AtomicPtr<u8> = AtomicPtr::new(std::ptr::null_mut());
|
||||
static PENDING: [AtomicUsize; 8] = [const { AtomicUsize::new(0) }; 8];
|
||||
static PENDING_N: AtomicUsize = AtomicUsize::new(0);
|
||||
|
||||
const TF: i64 = 0x100;
|
||||
|
||||
extern "C" fn on_segv(_sig: libc::c_int, info: *mut libc::siginfo_t, ctx: *mut libc::c_void) {
|
||||
unsafe {
|
||||
let addr = (*info).si_addr() as usize;
|
||||
let start = BUF_START.load(Ordering::Relaxed);
|
||||
let len = BUF_LEN.load(Ordering::Relaxed);
|
||||
if addr < start || addr >= start + len {
|
||||
// A genuine crash: restore the default action and re-fault.
|
||||
libc::signal(libc::SIGSEGV, libc::SIG_DFL);
|
||||
return;
|
||||
}
|
||||
let off = addr - start;
|
||||
let n = LOG_LEN.load(Ordering::Relaxed);
|
||||
if n < LOG_CAP {
|
||||
*LOG.load(Ordering::Relaxed).add(n) = off as u64;
|
||||
LOG_LEN.store(n + 1, Ordering::Relaxed);
|
||||
}
|
||||
let page = off / PAGE;
|
||||
let hits = HITS.load(Ordering::Relaxed).add(page);
|
||||
*hits += 1;
|
||||
libc::mprotect(
|
||||
(start + page * PAGE) as *mut libc::c_void,
|
||||
PAGE,
|
||||
libc::PROT_READ | libc::PROT_WRITE,
|
||||
);
|
||||
if *hits >= BULK_THRESHOLD {
|
||||
*BULK.load(Ordering::Relaxed).add(page) = 1;
|
||||
} else {
|
||||
let p = PENDING_N.load(Ordering::Relaxed);
|
||||
if p < PENDING.len() {
|
||||
PENDING[p].store(page, Ordering::Relaxed);
|
||||
PENDING_N.store(p + 1, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
let uc = ctx as *mut libc::ucontext_t;
|
||||
(*uc).uc_mcontext.gregs[libc::REG_EFL as usize] |= TF;
|
||||
}
|
||||
}
|
||||
|
||||
extern "C" fn on_trap(_sig: libc::c_int, _info: *mut libc::siginfo_t, ctx: *mut libc::c_void) {
|
||||
unsafe {
|
||||
let start = BUF_START.load(Ordering::Relaxed);
|
||||
let n = PENDING_N.load(Ordering::Relaxed);
|
||||
for p in PENDING.iter().take(n) {
|
||||
let page = p.load(Ordering::Relaxed);
|
||||
libc::mprotect(
|
||||
(start + page * PAGE) as *mut libc::c_void,
|
||||
PAGE,
|
||||
libc::PROT_NONE,
|
||||
);
|
||||
}
|
||||
PENDING_N.store(0, Ordering::Relaxed);
|
||||
let uc = ctx as *mut libc::ucontext_t;
|
||||
(*uc).uc_mcontext.gregs[libc::REG_EFL as usize] &= !TF;
|
||||
}
|
||||
}
|
||||
|
||||
fn install(
|
||||
sig: libc::c_int,
|
||||
h: extern "C" fn(libc::c_int, *mut libc::siginfo_t, *mut libc::c_void),
|
||||
) {
|
||||
unsafe {
|
||||
let mut sa: libc::sigaction = std::mem::zeroed();
|
||||
sa.sa_sigaction = h as usize;
|
||||
sa.sa_flags = libc::SA_SIGINFO | libc::SA_NODEFER;
|
||||
libc::sigemptyset(&mut sa.sa_mask);
|
||||
assert_eq!(libc::sigaction(sig, &sa, std::ptr::null_mut()), 0);
|
||||
}
|
||||
}
|
||||
|
||||
fn protect(prot: libc::c_int) {
|
||||
let start = BUF_START.load(Ordering::Relaxed);
|
||||
let len = BUF_LEN.load(Ordering::Relaxed);
|
||||
unsafe {
|
||||
assert_eq!(libc::mprotect(start as *mut libc::c_void, len, prot), 0);
|
||||
}
|
||||
}
|
||||
|
||||
struct Phase {
|
||||
name: &'static str,
|
||||
log: Vec<u64>,
|
||||
bulk_pages: Vec<u64>,
|
||||
}
|
||||
|
||||
/// Start a phase: clear the per-page state and protect the buffer.
|
||||
fn begin() {
|
||||
let pages = BUF_LEN.load(Ordering::Relaxed) / PAGE;
|
||||
unsafe {
|
||||
std::ptr::write_bytes(HITS.load(Ordering::Relaxed), 0, pages);
|
||||
std::ptr::write_bytes(BULK.load(Ordering::Relaxed), 0, pages);
|
||||
}
|
||||
LOG_LEN.store(0, Ordering::Relaxed);
|
||||
protect(libc::PROT_NONE);
|
||||
}
|
||||
|
||||
fn end(name: &'static str) -> Phase {
|
||||
protect(libc::PROT_READ | libc::PROT_WRITE);
|
||||
let n = LOG_LEN.load(Ordering::Relaxed);
|
||||
let log = unsafe { std::slice::from_raw_parts(LOG.load(Ordering::Relaxed), n) }.to_vec();
|
||||
let pages = BUF_LEN.load(Ordering::Relaxed) / PAGE;
|
||||
let bulk = unsafe { std::slice::from_raw_parts(BULK.load(Ordering::Relaxed), pages) };
|
||||
let bulk_pages = (0..pages)
|
||||
.filter(|&p| bulk[p] != 0)
|
||||
.map(|p| p as u64)
|
||||
.collect();
|
||||
if n == LOG_CAP {
|
||||
eprintln!("warning: access log full in phase {name}");
|
||||
}
|
||||
Phase {
|
||||
name,
|
||||
log,
|
||||
bulk_pages,
|
||||
}
|
||||
}
|
||||
|
||||
/// Byte intervals [lo, hi) read in a phase, merged when closer than `gap`.
|
||||
fn ranges(ph: &[&Phase], gap: u64) -> Vec<(u64, u64)> {
|
||||
let mut iv: Vec<(u64, u64)> = Vec::new();
|
||||
for p in ph {
|
||||
iv.extend(p.log.iter().map(|&o| (o, o + ACCESS_WIDTH)));
|
||||
iv.extend(
|
||||
p.bulk_pages
|
||||
.iter()
|
||||
.map(|&pg| (pg * PAGE as u64, (pg + 1) * PAGE as u64)),
|
||||
);
|
||||
}
|
||||
iv.sort_unstable();
|
||||
let mut out: Vec<(u64, u64)> = Vec::new();
|
||||
for (lo, hi) in iv {
|
||||
match out.last_mut() {
|
||||
Some(last) if lo <= last.1 + gap => last.1 = last.1.max(hi),
|
||||
_ => out.push((lo, hi)),
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Requests a reader with no cache at all would make: a new request each
|
||||
/// time the access stream leaves the neighbourhood of the current run.
|
||||
fn uncached_requests(p: &Phase) -> usize {
|
||||
let mut n = 0;
|
||||
let (mut lo, mut hi) = (u64::MAX, 0u64);
|
||||
for &o in &p.log {
|
||||
if lo != u64::MAX && o + MERGE_GAP >= lo && o <= hi + MERGE_GAP {
|
||||
lo = lo.min(o);
|
||||
hi = hi.max(o + ACCESS_WIDTH);
|
||||
} else {
|
||||
n += 1;
|
||||
lo = o;
|
||||
hi = o + ACCESS_WIDTH;
|
||||
}
|
||||
}
|
||||
n + p.bulk_pages.len()
|
||||
}
|
||||
|
||||
fn blocks(ph: &[&Phase], block: u64) -> usize {
|
||||
let mut set = BTreeSet::new();
|
||||
for (lo, hi) in ranges(ph, 0) {
|
||||
for b in lo / block..=(hi - 1) / block {
|
||||
set.insert(b);
|
||||
}
|
||||
}
|
||||
set.len()
|
||||
}
|
||||
|
||||
fn read_extents(path: &str) -> Vec<(u64, u64)> {
|
||||
let text = std::fs::read_to_string(path).expect("read extents");
|
||||
let mut v: Vec<(u64, u64)> = text
|
||||
.lines()
|
||||
.filter_map(|l| {
|
||||
let mut it = l.split_whitespace().map(|t| t.parse::<u64>());
|
||||
match (it.next(), it.next()) {
|
||||
(Some(Ok(o)), Some(Ok(n))) => Some((o, o + n)),
|
||||
_ => None,
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
v.sort_unstable();
|
||||
v
|
||||
}
|
||||
|
||||
fn in_extents(ext: &[(u64, u64)], lo: u64, hi: u64) -> bool {
|
||||
let i = ext.partition_point(|e| e.1 <= lo);
|
||||
i < ext.len() && ext[i].0 < hi
|
||||
}
|
||||
|
||||
/// Split a phase into loads outside and inside the raw-data extents. A bulk
|
||||
/// page counts as raw data when it overlaps an extent.
|
||||
fn split_raw(p: &Phase, ext: &[(u64, u64)]) -> (Phase, Phase) {
|
||||
let (mut m, mut r) = (Vec::new(), Vec::new());
|
||||
for &o in &p.log {
|
||||
if in_extents(ext, o, o + 1) {
|
||||
r.push(o)
|
||||
} else {
|
||||
m.push(o)
|
||||
}
|
||||
}
|
||||
let pg = PAGE as u64;
|
||||
let (bm, br): (Vec<u64>, Vec<u64>) = p
|
||||
.bulk_pages
|
||||
.iter()
|
||||
.partition(|&&b| !in_extents(ext, b * pg, (b + 1) * pg));
|
||||
(
|
||||
Phase {
|
||||
name: "read (metadata)",
|
||||
log: m,
|
||||
bulk_pages: bm,
|
||||
},
|
||||
Phase {
|
||||
name: "read (raw data)",
|
||||
log: r,
|
||||
bulk_pages: br,
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
fn walk(g: &Group<'_>, path: &str, objs: &mut usize) {
|
||||
for name in g.datasets().unwrap_or_default() {
|
||||
*objs += 1;
|
||||
if let Ok(ds) = g.dataset(&name) {
|
||||
let _ = ds.shape();
|
||||
let _ = ds.dtype();
|
||||
}
|
||||
}
|
||||
for name in g.groups().unwrap_or_default() {
|
||||
*objs += 1;
|
||||
if let Ok(sub) = g.group(&name) {
|
||||
walk(&sub, &format!("{path}/{name}"), objs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn main() {
|
||||
let args: Vec<String> = std::env::args().collect();
|
||||
if args.len() != 3 && args.len() != 4 {
|
||||
eprintln!("usage: range-trace FILE DATASET_PATH [RAW_EXTENTS]");
|
||||
std::process::exit(2);
|
||||
}
|
||||
let bytes = std::fs::read(&args[1]).expect("read file");
|
||||
let len = bytes.len();
|
||||
let cap = len.div_ceil(PAGE).max(1) * PAGE;
|
||||
let pages = cap / PAGE;
|
||||
// Page-aligned buffer so that protection covers exactly the file. The
|
||||
// Vec is never dropped (its layout differs from Vec's own), see the end.
|
||||
let ptr = unsafe { alloc_zeroed(Layout::from_size_align(cap, PAGE).unwrap()) };
|
||||
unsafe { std::ptr::copy_nonoverlapping(bytes.as_ptr(), ptr, len) };
|
||||
drop(bytes);
|
||||
let buf = unsafe { Vec::from_raw_parts(ptr, len, cap) };
|
||||
BUF_START.store(ptr as usize, Ordering::Relaxed);
|
||||
BUF_LEN.store(cap, Ordering::Relaxed);
|
||||
let mut log = vec![0u64; LOG_CAP];
|
||||
LOG.store(log.as_mut_ptr(), Ordering::Relaxed);
|
||||
let mut hits = vec![0u32; pages];
|
||||
HITS.store(hits.as_mut_ptr(), Ordering::Relaxed);
|
||||
let mut bulk = vec![0u8; pages];
|
||||
BULK.store(bulk.as_mut_ptr(), Ordering::Relaxed);
|
||||
install(libc::SIGSEGV, on_segv);
|
||||
install(libc::SIGTRAP, on_trap);
|
||||
|
||||
begin();
|
||||
let file = File::from_bytes(buf).expect("open");
|
||||
let p_open = end("open");
|
||||
|
||||
begin();
|
||||
let mut objs = 0;
|
||||
walk(&file.root(), "", &mut objs);
|
||||
let p_list = end("list");
|
||||
|
||||
begin();
|
||||
let out = file
|
||||
.dataset(&args[2])
|
||||
.and_then(|d| d.read_selection(&Selection::All))
|
||||
.expect("read dataset");
|
||||
let p_read = end("read");
|
||||
|
||||
println!("file: {} ({} bytes), objects listed: {objs}", args[1], len);
|
||||
println!("dataset: {} ({} bytes decoded)", args[2], out.len());
|
||||
println!(
|
||||
"| phase | loads logged | bulk 4K pages | uncached requests | distinct ranges (gap<{MERGE_GAP}B) | bytes in ranges | 4 KiB blocks | 64 KiB blocks | 1 MiB blocks |"
|
||||
);
|
||||
println!("|---|---:|---:|---:|---:|---:|---:|---:|---:|");
|
||||
let row = |label: String, ph: &[&Phase], uncached: usize| {
|
||||
let r = ranges(ph, MERGE_GAP);
|
||||
let bytes: u64 = r.iter().map(|(a, b)| b - a).sum();
|
||||
let loads: usize = ph.iter().map(|p| p.log.len()).sum();
|
||||
let bulk: usize = ph.iter().map(|p| p.bulk_pages.len()).sum();
|
||||
println!(
|
||||
"| {label} | {loads} | {bulk} | {uncached} | {} | {bytes} | {} | {} | {} |",
|
||||
r.len(),
|
||||
blocks(ph, 4 << 10),
|
||||
blocks(ph, 64 << 10),
|
||||
blocks(ph, 1 << 20)
|
||||
);
|
||||
};
|
||||
for p in [&p_open, &p_list, &p_read] {
|
||||
row(p.name.to_string(), &[p], uncached_requests(p));
|
||||
}
|
||||
if let Some(path) = args.get(3) {
|
||||
let (meta, raw) = split_raw(&p_read, &read_extents(path));
|
||||
for p in [&meta, &raw] {
|
||||
row(p.name.to_string(), &[p], uncached_requests(p));
|
||||
}
|
||||
let all = [&p_open, &p_list, &meta];
|
||||
let unc: usize = all.iter().map(|p| uncached_requests(p)).sum();
|
||||
row("all metadata".into(), &all, unc);
|
||||
}
|
||||
let all = [&p_open, &p_list, &p_read];
|
||||
let unc: usize = all.iter().map(|p| uncached_requests(p)).sum();
|
||||
row("open+list+read".into(), &all, unc);
|
||||
let meta = [&p_open, &p_list];
|
||||
let unc: usize = meta.iter().map(|p| uncached_requests(p)).sum();
|
||||
row("open+list".into(), &meta, unc);
|
||||
// The File owns a buffer whose layout Vec does not know; never drop it.
|
||||
std::mem::forget(file);
|
||||
std::mem::forget(log);
|
||||
std::mem::forget(hits);
|
||||
std::mem::forget(bulk);
|
||||
}
|
||||
Reference in New Issue
Block a user