Range-read milestone M3, first half: a new crate with the block cache the
design makes mandatory for remote files and an HTTP backend, so
open_url("http://...") gives a clawhdf5::File over File::open_storage.
BlockCache wraps any Storage: aligned blocks (1 MiB by default, the size
docs/design/range-reads.md section 2 measured), LRU with a byte budget,
the missing blocks of one read_at/read_ranges fetched with one backend
read_ranges call as runs of consecutive blocks (a one-block gap filled to
merge runs, each request at most 8 MiB), and reads that miss more than
half the budget not kept. Thread-safe without holding the lock across a
fetch: a block being fetched is in flight, a second reader waits for it
instead of fetching it again, and a failed fetch fails its waiters and is
not cached. A backend holding the file in memory passes through.
HttpStorage (ureq, no TLS by default; `https` adds rustls with ring):
opening is one ranged GET of the first block, whose Content-Range gives
the length (the cache keeps the bytes). The file is pinned by a strong
ETag (If-Match), else Last-Modified (If-Unmodified-Since), and its length,
checked on every response: a change is RemoteError::FileChanged, never
mixed data. A server that ignores Range is refused without reading the
body unless a full download is allowed. Connection errors, timeouts,
408/429/5xx and short bodies are retried with exponential backoff;
Accept-Encoding: identity, and an encoded body is refused. read_ranges
fetches its ranges in parallel.
Tests (a std-only HTTP/1.1 server in tests/common/server.rs, also the
range_server example): every fixture read over HTTP gives File::open's
transcript (CLAWHDF5_REMOTE_CORPUS adds the conformance corpus), with
request counts per file with and without the cache; an h5py-written file
against libhdf5's values; a multi-block file fetched in whole blocks, each
once; a server ignoring Range; a file replaced mid-read (ETag,
Last-Modified, length only); truncated bodies and 503s (retried, then an
error, never cached); a slow server with 8 concurrent readers (no block
fetched twice); bad URLs, 404, encoded bodies, non-HDF5 data. The cache
has unit tests for coalescing, splitting, LRU order, large reads,
failures and concurrent in-flight dedup.
ci-test.sh: clawhdf5-remote joins the no-C default-build check, and its
https feature is linted.
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
303 lines
9.8 KiB
Rust
303 lines
9.8 KiB
Rust
//! Shared by the integration tests: the test server, a transcript of a
|
|
//! file (tree, attributes, values) to compare two ways of reading it, and
|
|
//! the test files.
|
|
|
|
#![allow(dead_code)]
|
|
|
|
pub mod server;
|
|
|
|
use std::collections::{BTreeMap, HashSet, VecDeque};
|
|
use std::fmt::Write as _;
|
|
use std::path::{Path, PathBuf};
|
|
use std::process::Command;
|
|
use std::sync::Arc;
|
|
|
|
use clawhdf5::{DType, File, Selection};
|
|
use clawhdf5_format::error::FormatError;
|
|
|
|
/// Objects visited per file.
|
|
const MAX_OBJECTS: usize = 2000;
|
|
/// Datasets with more bytes than this are not read (their metadata is).
|
|
pub 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 value, or `Err` (which chunk a damaged dataset reports
|
|
/// can vary between two `File`s: the chunk cache lists in hash order).
|
|
fn value<T: std::fmt::Debug, E>(r: &Result<T, E>) -> String {
|
|
match r {
|
|
Ok(v) => digest(v),
|
|
Err(_) => "Err".into(),
|
|
}
|
|
}
|
|
|
|
fn sorted<V: std::fmt::Debug>(m: std::collections::HashMap<String, V>) -> BTreeMap<String, V> {
|
|
m.into_iter().collect()
|
|
}
|
|
|
|
/// Everything a reader sees in `file`: every group's entries, every
|
|
/// object's attributes, and every dataset's shape, types and values.
|
|
pub 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();
|
|
if let Ok(ds) = file.dataset_at(addr) {
|
|
dataset(&mut out, &path, &ds);
|
|
}
|
|
let attrs = group.attrs_with_errors().map(|(a, e)| (sorted(a), e));
|
|
writeln!(out, "{path} attrs {}", digest(&attrs)).unwrap();
|
|
if let Ok(entries) = entries {
|
|
for (name, child) in entries {
|
|
queue.push_back((format!("{}/{name}", path.trim_end_matches('/')), child));
|
|
}
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
fn dataset(out: &mut String, path: &str, ds: &clawhdf5::Dataset<'_>) {
|
|
let shape = ds.shape();
|
|
let dtype = ds.dtype();
|
|
writeln!(
|
|
out,
|
|
"{path} shape {} dtype {} raw {}",
|
|
digest(&shape),
|
|
digest(&dtype),
|
|
digest(&ds.raw_datatype())
|
|
)
|
|
.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();
|
|
if matches!(
|
|
dtype,
|
|
DType::F32
|
|
| DType::F64
|
|
| DType::I8
|
|
| DType::I16
|
|
| DType::I32
|
|
| DType::I64
|
|
| DType::U8
|
|
| DType::U16
|
|
| DType::U32
|
|
| DType::U64
|
|
) {
|
|
writeln!(out, "{path} f64 {}", value(&ds.read_f64())).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();
|
|
}
|
|
}
|
|
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();
|
|
}
|
|
clawhdf5_format::datatype::Datatype::VariableLength { .. } => {
|
|
writeln!(out, "{path} vlen {}", value(&ds.read_vlen::<f64>())).unwrap();
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
/// Open, list every group, and read the first dataset found whose data is
|
|
/// at most `MAX_DATA_BYTES` (a tree view plus one plot).
|
|
pub fn list_and_read_one(file: &File) {
|
|
let mut seen = HashSet::new();
|
|
let mut read_one = false;
|
|
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);
|
|
if let Ok(ds) = file.dataset_at(addr) {
|
|
let _ = (ds.shape(), ds.dtype());
|
|
if !read_one {
|
|
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);
|
|
read_one = true;
|
|
}
|
|
}
|
|
}
|
|
if let Ok(entries) = group.entries() {
|
|
queue.extend(entries.into_iter().map(|(_, a)| a));
|
|
}
|
|
}
|
|
}
|
|
|
|
/// External virtual-dataset sources read from `dir`, as `File::open` finds
|
|
/// them next to the file.
|
|
pub fn sibling_resolver(dir: PathBuf) -> clawhdf5::VdsResolver {
|
|
Arc::new(move |name: &str| {
|
|
let p = Path::new(name);
|
|
if name.is_empty()
|
|
|| !p
|
|
.components()
|
|
.all(|c| matches!(c, std::path::Component::Normal(_)))
|
|
{
|
|
return Err(FormatError::Storage(format!("{name:?} 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::Storage(e.to_string())),
|
|
}
|
|
})
|
|
}
|
|
|
|
/// HDF5 files under `dir`, recursively.
|
|
pub 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);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The repository's HDF5 test fixtures.
|
|
pub fn fixtures() -> Vec<PathBuf> {
|
|
let root = Path::new(env!("CARGO_MANIFEST_DIR"));
|
|
let mut files = Vec::new();
|
|
hdf5_files(&root.join("../clawhdf5/tests/fixtures"), &mut files);
|
|
hdf5_files(&root.join("../clawhdf5-format/tests/fixtures"), &mut files);
|
|
files.sort();
|
|
files
|
|
}
|
|
|
|
/// Files of `CLAWHDF5_REMOTE_CORPUS` (directories separated like `PATH`).
|
|
pub fn corpus() -> Option<Vec<PathBuf>> {
|
|
let dirs = std::env::var("CLAWHDF5_REMOTE_CORPUS").ok()?;
|
|
let mut files = Vec::new();
|
|
for d in std::env::split_paths(&dirs) {
|
|
hdf5_files(&d, &mut files);
|
|
}
|
|
files.sort();
|
|
Some(files)
|
|
}
|
|
|
|
pub fn python() -> String {
|
|
std::env::var("CLAWHDF5_PYTHON").unwrap_or_else(|_| "python3".to_string())
|
|
}
|
|
|
|
pub fn interop_required() -> bool {
|
|
std::env::var("CLAWHDF5_REQUIRE_INTEROP").is_ok_and(|v| v == "1")
|
|
}
|
|
|
|
/// Whether python3 with h5py and numpy runs; panics when interop is
|
|
/// required and it does not.
|
|
pub fn have_h5py() -> bool {
|
|
let ok = Command::new(python())
|
|
.args(["-c", "import h5py, numpy"])
|
|
.output()
|
|
.map(|o| o.status.success())
|
|
.unwrap_or(false);
|
|
assert!(
|
|
ok || !interop_required(),
|
|
"CLAWHDF5_REQUIRE_INTEROP=1 but python3 with h5py is not available"
|
|
);
|
|
if !ok {
|
|
eprintln!("SKIP: python3 with h5py not available");
|
|
}
|
|
ok
|
|
}
|
|
|
|
/// Run a Python script; its stdout.
|
|
pub fn run_python(script: &str, args: &[&str]) -> String {
|
|
let out = Command::new(python())
|
|
.arg("-c")
|
|
.arg(script)
|
|
.args(args)
|
|
.output()
|
|
.expect("failed to run python");
|
|
assert!(
|
|
out.status.success(),
|
|
"python failed:\n{}",
|
|
String::from_utf8_lossy(&out.stderr)
|
|
);
|
|
String::from_utf8(out.stdout).unwrap()
|
|
}
|
|
|
|
/// A clawhdf5-written file with a multi-block chunked dataset (`/big`,
|
|
/// 1 000 000 f64 in chunks of 10 000, deflated), a contiguous one and a
|
|
/// group — several blocks of 1 MiB, with no Python needed.
|
|
pub fn multi_block_file() -> Vec<u8> {
|
|
let mut b = clawhdf5::FileBuilder::new();
|
|
b.set_attr("title", clawhdf5::AttrValue::String("remote test".into()));
|
|
let big: Vec<f64> = (0..1_000_000u64)
|
|
.map(|i| ((i * 2_654_435_761) % 1_000_003) as f64 * 0.5)
|
|
.collect();
|
|
b.create_dataset("big")
|
|
.with_f64_data(&big)
|
|
.with_chunks(&[10_000])
|
|
.with_deflate(1);
|
|
let flat: Vec<f64> = (0..300_000u64).map(|i| i as f64).collect();
|
|
b.create_dataset("flat").with_f64_data(&flat);
|
|
let mut g = b.create_group("grp");
|
|
g.create_dataset("small").with_f64_data(&[1.0, 2.0, 3.0]);
|
|
b.add_group(g.finish());
|
|
b.finish().unwrap()
|
|
}
|