osobhandClaude Opus 5.5
6185874f9c
ExtentBytes and read_exact_at/read_upto rejected short results but passed longer-than-asked ones through, and FileData forwarded them too, so a Storage that broke read_at's contract by returning extra bytes had them decoded or returned as data (a contiguous dataset read gained 37 junk bytes). gather_storage alone trimmed. - storage::exact_len (new, pub): a read of len bytes as exactly len — cut when longer, an error when short. read_exact_at, read_upto and ExtentBytes (so chunk fetches and selection gathers) go through it. - FileData cuts a backend's answer to what it asked for before laying the cache image over it. - Tests: over a storage that appends 37 junk bytes to every read, every format-crate fixture reads exactly as from the slice (overlong_reads_are_cut_to_the_range_asked_for), and every facade fixture opens and reads through File::open_storage as through File::open (overlong_storage_reads_identically). Both failed before. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
482 lines
17 KiB
Rust
482 lines
17 KiB
Rust
//! `File::open_storage` over a read_at-only storage reads every file as
|
|
//! `File::open` does (range reads, milestone M2 in
|
|
//! `docs/design/range-reads.md`).
|
|
//!
|
|
//! Each file is read end to end twice — through `File::open` (the mmap
|
|
//! fast path) and through `File::open_storage` over a
|
|
//! [`CountingStorage`], which serves the file through `read_at` only
|
|
//! (`as_contiguous()` is `None`, so no reader can fall back to a slice of
|
|
//! the whole file) — and the two transcripts must be identical: the tree
|
|
//! (every group's entries, followed by address), every object's attributes
|
|
//! (the whole map and each one by name), and every dataset's shape, types
|
|
//! and values (all bytes, as `f64`, a hyperslab of them, strings and
|
|
//! variable-length sequences).
|
|
//!
|
|
//! The storage also counts its `read_at` calls and bytes: what a remote
|
|
//! backend without a cache would be asked for. The totals and the files
|
|
//! that cost most are printed.
|
|
//!
|
|
//! - `CLAWHDF5_STORAGE_CORPUS=dir[:dir...]` adds every HDF5 file under those
|
|
//! directories (the conformance corpus is `conformance/.cache/corpus`);
|
|
//! `CLAWHDF5_STORAGE_REPORT=1` prints every file's counts.
|
|
|
|
use std::collections::{BTreeMap, HashSet, VecDeque};
|
|
use std::fmt::Write as _;
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::Arc;
|
|
|
|
use clawhdf5::{DType, File, Selection};
|
|
use clawhdf5_format::error::FormatError;
|
|
use clawhdf5_format::storage::CountingStorage;
|
|
|
|
/// Objects visited per file.
|
|
const MAX_OBJECTS: usize = 2000;
|
|
/// Datasets with more bytes than this are not read (their metadata is).
|
|
const MAX_DATA_BYTES: u64 = 64 << 20;
|
|
|
|
/// A short, stable digest of a value's `Debug` form.
|
|
fn digest<T: std::fmt::Debug>(v: &T) -> String {
|
|
let s = format!("{v:?}");
|
|
if s.len() <= 200 {
|
|
return s;
|
|
}
|
|
let mut h = 0xcbf2_9ce4_8422_2325u64;
|
|
for b in s.bytes() {
|
|
h = (h ^ u64::from(b)).wrapping_mul(0x100_0000_01b3);
|
|
}
|
|
format!("{}…[{} bytes, fnv {h:016x}]", &s[..80], s.len())
|
|
}
|
|
|
|
/// A data read's result: its value, or just `Err` — a full read goes
|
|
/// through the file's chunk cache, which lists a damaged dataset's chunks
|
|
/// in hash-map order, so which failing chunk it reports varies from one
|
|
/// `File` to the next (the format crate's harness compares these errors on
|
|
/// the uncached path).
|
|
fn value<T: std::fmt::Debug, E>(r: &Result<T, E>) -> String {
|
|
match r {
|
|
Ok(v) => digest(v),
|
|
Err(_) => "Err".into(),
|
|
}
|
|
}
|
|
|
|
fn transcript(file: &File) -> String {
|
|
let mut out = String::new();
|
|
let mut seen = HashSet::new();
|
|
let mut queue = VecDeque::from([(String::from("/"), file.superblock().root_group_address)]);
|
|
while let Some((path, addr)) = queue.pop_front() {
|
|
if seen.len() >= MAX_OBJECTS || !seen.insert(addr) {
|
|
continue;
|
|
}
|
|
let group = file.group_at(addr);
|
|
let entries = group.entries();
|
|
writeln!(out, "{path} @{addr} entries {}", digest(&entries)).unwrap();
|
|
match file.dataset_at(addr) {
|
|
Ok(ds) => dataset(&mut out, &path, &ds),
|
|
Err(e) => writeln!(out, "{path} dataset_at {e:?}").unwrap(),
|
|
}
|
|
let attrs = group.attrs_with_errors().map(|(a, e)| (sorted(a), e));
|
|
writeln!(out, "{path} attrs {}", digest(&attrs)).unwrap();
|
|
if let Ok((attrs, _)) = &attrs {
|
|
for name in attrs.keys().take(50) {
|
|
writeln!(out, "{path} attr {name:?} {}", digest(&group.attr(name))).unwrap();
|
|
}
|
|
}
|
|
if let Ok(entries) = entries {
|
|
for (name, child) in entries {
|
|
queue.push_back((format!("{}/{name}", path.trim_end_matches('/')), child));
|
|
// Name lookups (through the name index of a dense group).
|
|
if queue.len() < 64 {
|
|
writeln!(
|
|
out,
|
|
"{path} group({name:?}) {}",
|
|
digest(&group.group(&name).map(|_| ()))
|
|
)
|
|
.unwrap();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
fn sorted<V: std::fmt::Debug>(m: std::collections::HashMap<String, V>) -> BTreeMap<String, V> {
|
|
m.into_iter().collect()
|
|
}
|
|
|
|
fn dataset(out: &mut String, path: &str, ds: &clawhdf5::Dataset<'_>) {
|
|
let shape = ds.shape();
|
|
let dtype = ds.dtype();
|
|
writeln!(
|
|
out,
|
|
"{path} shape {} max {} dtype {} raw {}",
|
|
digest(&shape),
|
|
digest(&ds.max_dimensions()),
|
|
digest(&dtype),
|
|
digest(&ds.raw_datatype())
|
|
)
|
|
.unwrap();
|
|
let attrs = ds.attrs_with_errors().map(|(a, e)| (sorted(a), e));
|
|
writeln!(out, "{path} dataset attrs {}", digest(&attrs)).unwrap();
|
|
let (Ok(shape), Ok(dtype), Ok(raw_dt)) = (shape, dtype, ds.raw_datatype()) else {
|
|
return;
|
|
};
|
|
let elements = shape.iter().try_fold(1u64, |a, &d| a.checked_mul(d));
|
|
let bytes = elements.and_then(|n| n.checked_mul(u64::from(raw_dt.type_size())));
|
|
if bytes.is_none_or(|b| b > MAX_DATA_BYTES) {
|
|
writeln!(out, "{path} too large to read").unwrap();
|
|
return;
|
|
}
|
|
writeln!(
|
|
out,
|
|
"{path} all {}",
|
|
value(&ds.read_selection(&Selection::All))
|
|
)
|
|
.unwrap();
|
|
let numeric = matches!(
|
|
dtype,
|
|
DType::F32
|
|
| DType::F64
|
|
| DType::I8
|
|
| DType::I16
|
|
| DType::I32
|
|
| DType::I64
|
|
| DType::U8
|
|
| DType::U16
|
|
| DType::U32
|
|
| DType::U64
|
|
);
|
|
if numeric {
|
|
writeln!(out, "{path} f64 {}", value(&ds.read_f64())).unwrap();
|
|
writeln!(out, "{path} f32 {}", value(&ds.read_f32())).unwrap();
|
|
writeln!(out, "{path} i64 {}", value(&ds.read_i64())).unwrap();
|
|
if let Some(&d0) = shape.first() {
|
|
let rank = shape.len();
|
|
let sel = Selection::Hyperslab {
|
|
start: std::iter::once(d0 / 3)
|
|
.chain(std::iter::repeat_n(0, rank - 1))
|
|
.collect(),
|
|
stride: vec![1; rank],
|
|
count: std::iter::once(d0.div_ceil(3))
|
|
.chain(shape[1..].iter().copied())
|
|
.collect(),
|
|
block: vec![1; rank],
|
|
};
|
|
writeln!(
|
|
out,
|
|
"{path} f64 third {}",
|
|
value(&ds.read_f64_selection(&sel))
|
|
)
|
|
.unwrap();
|
|
writeln!(
|
|
out,
|
|
"{path} bytes third {}",
|
|
value(&ds.read_selection(&sel))
|
|
)
|
|
.unwrap();
|
|
}
|
|
}
|
|
match &raw_dt {
|
|
clawhdf5_format::datatype::Datatype::String { .. }
|
|
| clawhdf5_format::datatype::Datatype::VariableLength {
|
|
is_string: true, ..
|
|
} => {
|
|
writeln!(out, "{path} strings {}", value(&ds.read_string_bytes())).unwrap();
|
|
writeln!(out, "{path} string {}", value(&ds.read_string())).unwrap();
|
|
}
|
|
clawhdf5_format::datatype::Datatype::VariableLength { .. } => {
|
|
writeln!(out, "{path} vlen {}", value(&ds.read_vlen::<f64>())).unwrap();
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
/// External virtual-dataset sources as `File::open` finds them: files in
|
|
/// the same directory.
|
|
fn sibling_resolver(dir: Option<PathBuf>) -> clawhdf5::VdsResolver {
|
|
Arc::new(move |name: &str| {
|
|
let Some(dir) = dir.as_ref() else {
|
|
return Err(FormatError::ChunkedReadError(format!(
|
|
"virtual dataset source file {name:?} cannot be resolved for an in-memory file"
|
|
)));
|
|
};
|
|
let p = Path::new(name);
|
|
if name.is_empty()
|
|
|| !p.components().all(|c| {
|
|
matches!(
|
|
c,
|
|
std::path::Component::Normal(_) | std::path::Component::CurDir
|
|
)
|
|
})
|
|
{
|
|
return Err(FormatError::ChunkedReadError(format!(
|
|
"virtual dataset source file {name:?} is outside the virtual file's \
|
|
directory and is not followed"
|
|
)));
|
|
}
|
|
match std::fs::read(dir.join(p)) {
|
|
Ok(bytes) => Ok(Some(bytes)),
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
|
|
Err(e) => Err(FormatError::ChunkedReadError(format!(
|
|
"cannot read virtual dataset source file {name:?}: {e}"
|
|
))),
|
|
}
|
|
})
|
|
}
|
|
|
|
#[derive(Default)]
|
|
struct Totals {
|
|
files: usize,
|
|
opened: usize,
|
|
/// The comparison's reads (each dataset read several ways).
|
|
reads: u64,
|
|
bytes: u64,
|
|
/// One pass: open, list every group, read every attribute and every
|
|
/// dataset once (`read_selection(All)`).
|
|
pass_reads: u64,
|
|
pass_bytes: u64,
|
|
file_bytes: u64,
|
|
/// (one-pass reads, one-pass bytes, file size, name) per file.
|
|
per_file: Vec<(u64, u64, u64, String)>,
|
|
}
|
|
|
|
/// Open the file and read everything once, as a tree viewer that then
|
|
/// shows every value would.
|
|
fn one_pass(file: &File) {
|
|
let mut seen = HashSet::new();
|
|
let mut queue = VecDeque::from([file.superblock().root_group_address]);
|
|
while let Some(addr) = queue.pop_front() {
|
|
if seen.len() >= MAX_OBJECTS || !seen.insert(addr) {
|
|
continue;
|
|
}
|
|
let group = file.group_at(addr);
|
|
let _ = group.attrs();
|
|
if let Ok(ds) = file.dataset_at(addr) {
|
|
let small = ds.shape().ok().and_then(|s| {
|
|
let n = s.iter().try_fold(1u64, |a, &d| a.checked_mul(d))?;
|
|
let size = u64::from(ds.raw_datatype().ok()?.type_size());
|
|
n.checked_mul(size).filter(|&b| b <= MAX_DATA_BYTES)
|
|
});
|
|
if small.is_some() {
|
|
let _ = ds.read_selection(&Selection::All);
|
|
}
|
|
}
|
|
if let Ok(entries) = group.entries() {
|
|
queue.extend(entries.into_iter().map(|(_, a)| a));
|
|
}
|
|
}
|
|
}
|
|
|
|
fn check(path: &Path, totals: &mut Totals) {
|
|
let Ok(bytes) = std::fs::read(path) else {
|
|
return;
|
|
};
|
|
let name = path.display().to_string();
|
|
let local = File::open(path);
|
|
let storage = Arc::new(CountingStorage::new(bytes.clone()));
|
|
let resolver = sibling_resolver(path.parent().map(Path::to_path_buf));
|
|
let remote = File::open_storage(storage.clone()).map(|mut f| {
|
|
f.set_vds_resolver(resolver.clone());
|
|
f
|
|
});
|
|
totals.files += 1;
|
|
let (local, remote) = match (local, remote) {
|
|
(Ok(l), Ok(r)) => (l, r),
|
|
(l, r) => {
|
|
// Both refuse the file, with the same error.
|
|
assert_eq!(
|
|
format!("{:?}", l.map(|_| ())),
|
|
format!("{:?}", r.map(|_| ())),
|
|
"{name}: open"
|
|
);
|
|
return;
|
|
}
|
|
};
|
|
totals.opened += 1;
|
|
assert!(remote.contiguous_bytes().is_none(), "{name}");
|
|
assert_eq!(local.user_block_size(), remote.user_block_size(), "{name}");
|
|
let want = transcript(&local);
|
|
let got = transcript(&remote);
|
|
if want != got {
|
|
let first = want
|
|
.lines()
|
|
.zip(got.lines())
|
|
.find(|(w, g)| w != g)
|
|
.map(|(w, g)| format!("\n local: {w}\n storage: {g}"))
|
|
.unwrap_or_else(|| {
|
|
format!(
|
|
"\n {} vs {} lines",
|
|
want.lines().count(),
|
|
got.lines().count()
|
|
)
|
|
});
|
|
panic!("{name}: File::open_storage differs from File::open{first}");
|
|
}
|
|
totals.reads += storage.reads();
|
|
totals.bytes += storage.bytes_read();
|
|
totals.file_bytes += bytes.len() as u64;
|
|
|
|
let pass = Arc::new(CountingStorage::new(bytes.clone()));
|
|
if let Ok(mut f) = File::open_storage(pass.clone()) {
|
|
f.set_vds_resolver(resolver);
|
|
one_pass(&f);
|
|
}
|
|
let (reads, read_bytes) = (pass.reads(), pass.bytes_read());
|
|
totals.pass_reads += reads;
|
|
totals.pass_bytes += read_bytes;
|
|
if std::env::var("CLAWHDF5_STORAGE_REPORT").is_ok_and(|v| v == "1") {
|
|
eprintln!(
|
|
"{reads:>9} reads {read_bytes:>12} bytes {:>12} file {name}",
|
|
bytes.len()
|
|
);
|
|
}
|
|
totals
|
|
.per_file
|
|
.push((reads, read_bytes, bytes.len() as u64, name));
|
|
}
|
|
|
|
fn report(what: &str, totals: &mut Totals) {
|
|
eprintln!(
|
|
"{what}: {} files ({} open, {} bytes); comparison: {} read_at calls, {} bytes; \
|
|
one pass (list, attributes, every dataset once): {} read_at calls, {} bytes",
|
|
totals.files,
|
|
totals.opened,
|
|
totals.file_bytes,
|
|
totals.reads,
|
|
totals.bytes,
|
|
totals.pass_reads,
|
|
totals.pass_bytes
|
|
);
|
|
totals.per_file.sort_by_key(|a| std::cmp::Reverse(a.0));
|
|
for (reads, bytes, size, name) in totals.per_file.iter().take(10) {
|
|
eprintln!(" {reads:>9} reads {bytes:>12} bytes (file {size:>11}) {name}");
|
|
}
|
|
}
|
|
fn hdf5_files(dir: &Path, out: &mut Vec<PathBuf>) {
|
|
let Ok(entries) = std::fs::read_dir(dir) else {
|
|
return;
|
|
};
|
|
for e in entries.flatten() {
|
|
let p = e.path();
|
|
if p.is_dir() {
|
|
hdf5_files(&p, out);
|
|
} else if p
|
|
.extension()
|
|
.and_then(|x| x.to_str())
|
|
.is_some_and(|x| matches!(x, "h5" | "hdf5" | "he5" | "nc" | "h5ad" | "hdf"))
|
|
{
|
|
out.push(p);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn fixtures_read_identically_through_open_storage() {
|
|
let root = Path::new(env!("CARGO_MANIFEST_DIR"));
|
|
let mut files = Vec::new();
|
|
hdf5_files(&root.join("tests/fixtures"), &mut files);
|
|
hdf5_files(&root.join("../clawhdf5-format/tests/fixtures"), &mut files);
|
|
files.sort();
|
|
assert!(files.len() >= 45, "{} fixtures", files.len());
|
|
let mut totals = Totals::default();
|
|
for f in &files {
|
|
check(f, &mut totals);
|
|
}
|
|
report("fixtures", &mut totals);
|
|
assert!(totals.opened >= 40, "{}", totals.opened);
|
|
assert!(totals.reads > 0);
|
|
}
|
|
|
|
#[test]
|
|
fn corpus_reads_identically_through_open_storage() {
|
|
let Ok(dirs) = std::env::var("CLAWHDF5_STORAGE_CORPUS") else {
|
|
eprintln!("CLAWHDF5_STORAGE_CORPUS not set; skipping the corpus");
|
|
return;
|
|
};
|
|
let mut files = Vec::new();
|
|
for d in std::env::split_paths(&dirs) {
|
|
hdf5_files(&d, &mut files);
|
|
}
|
|
files.sort();
|
|
let mut totals = Totals::default();
|
|
for f in &files {
|
|
check(f, &mut totals);
|
|
}
|
|
report("corpus", &mut totals);
|
|
assert!(totals.files > 0);
|
|
}
|
|
|
|
/// A user block, a metadata cache image and a Storage that is itself in
|
|
/// memory: the in-memory view of a storage that has one is used as is.
|
|
#[test]
|
|
fn storage_backed_files_keep_their_zero_copy_views_only_in_memory() {
|
|
let root = Path::new(env!("CARGO_MANIFEST_DIR"));
|
|
let path = root.join("tests/fixtures/h5clear_mdc_image.h5");
|
|
let bytes = std::fs::read(&path).unwrap();
|
|
let local = File::open(&path).unwrap();
|
|
// A Vec<u8> is a Storage with a contiguous view; the image still has
|
|
// to be laid over it, so the view is not used.
|
|
let in_memory = File::open_storage(Arc::new(bytes.clone())).unwrap();
|
|
assert!(in_memory.contiguous_bytes().is_none());
|
|
assert_eq!(transcript(&local), transcript(&in_memory));
|
|
|
|
let plain = root.join("../clawhdf5-format/tests/fixtures/chunked_2d.h5");
|
|
let bytes = std::fs::read(&plain).unwrap();
|
|
let in_memory = File::open_storage(Arc::new(bytes.clone())).unwrap();
|
|
assert_eq!(in_memory.contiguous_bytes(), Some(&bytes[..]));
|
|
let counting = File::open_storage(Arc::new(CountingStorage::new(bytes))).unwrap();
|
|
assert!(counting.contiguous_bytes().is_none());
|
|
let result =
|
|
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| counting.as_bytes().len()));
|
|
assert!(
|
|
result.is_err(),
|
|
"as_bytes over a range storage must not answer"
|
|
);
|
|
}
|
|
|
|
/// A storage that returns 37 junk bytes more than every read asked for.
|
|
struct Overlong(Vec<u8>);
|
|
|
|
impl clawhdf5::Storage for Overlong {
|
|
fn read_at(&self, offset: u64, len: usize) -> Result<std::borrow::Cow<'_, [u8]>, FormatError> {
|
|
let mut v = clawhdf5::Storage::read_at(self.0.as_slice(), offset, len)?.into_owned();
|
|
v.extend(std::iter::repeat_n(0xa5, 37));
|
|
Ok(std::borrow::Cow::Owned(v))
|
|
}
|
|
|
|
fn len(&self) -> u64 {
|
|
self.0.len() as u64
|
|
}
|
|
}
|
|
|
|
/// Bytes a misbehaving storage returns past the range asked for are never
|
|
/// read as the file's: every fixture reads through it as through
|
|
/// `File::open`.
|
|
#[test]
|
|
fn overlong_storage_reads_identically() {
|
|
let root = Path::new(env!("CARGO_MANIFEST_DIR"));
|
|
let mut files = Vec::new();
|
|
hdf5_files(&root.join("tests/fixtures"), &mut files);
|
|
hdf5_files(&root.join("../clawhdf5-format/tests/fixtures"), &mut files);
|
|
files.sort();
|
|
let mut compared = 0;
|
|
for path in &files {
|
|
let Ok(bytes) = std::fs::read(path) else {
|
|
continue;
|
|
};
|
|
let Ok(local) = File::open(path) else {
|
|
continue;
|
|
};
|
|
let mut remote = File::open_storage(Arc::new(Overlong(bytes)))
|
|
.unwrap_or_else(|e| panic!("{}: {e}", path.display()));
|
|
remote.set_vds_resolver(sibling_resolver(path.parent().map(Path::to_path_buf)));
|
|
assert_eq!(
|
|
transcript(&local),
|
|
transcript(&remote),
|
|
"{}",
|
|
path.display()
|
|
);
|
|
compared += 1;
|
|
}
|
|
assert!(compared >= 40, "{compared}");
|
|
}
|