From db2554dd8119da91828b6ec8813c2c47dbc1e720 Mon Sep 17 00:00:00 2001 From: osobh Date: Sat, 26 Sep 2026 17:13:28 -0500 Subject: [PATCH] clawhdf5-remote: block cache and HTTP range reads (open_url) 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) --- CLAUDE.md | 15 +- Cargo.toml | 1 + crates/clawhdf5-remote/Cargo.toml | 27 + crates/clawhdf5-remote/README.md | 71 ++ .../clawhdf5-remote/examples/range_server.rs | 56 ++ crates/clawhdf5-remote/examples/read_url.rs | 45 + crates/clawhdf5-remote/src/cache.rs | 825 ++++++++++++++++++ crates/clawhdf5-remote/src/error.rs | 125 +++ crates/clawhdf5-remote/src/http.rs | 565 ++++++++++++ crates/clawhdf5-remote/src/lib.rs | 127 +++ crates/clawhdf5-remote/tests/common/mod.rs | 302 +++++++ crates/clawhdf5-remote/tests/common/server.rs | 331 +++++++ crates/clawhdf5-remote/tests/http.rs | 517 +++++++++++ scripts/ci-test.sh | 23 +- 14 files changed, 3024 insertions(+), 6 deletions(-) create mode 100644 crates/clawhdf5-remote/Cargo.toml create mode 100644 crates/clawhdf5-remote/README.md create mode 100644 crates/clawhdf5-remote/examples/range_server.rs create mode 100644 crates/clawhdf5-remote/examples/read_url.rs create mode 100644 crates/clawhdf5-remote/src/cache.rs create mode 100644 crates/clawhdf5-remote/src/error.rs create mode 100644 crates/clawhdf5-remote/src/http.rs create mode 100644 crates/clawhdf5-remote/src/lib.rs create mode 100644 crates/clawhdf5-remote/tests/common/mod.rs create mode 100644 crates/clawhdf5-remote/tests/common/server.rs create mode 100644 crates/clawhdf5-remote/tests/http.rs diff --git a/CLAUDE.md b/CLAUDE.md index f493066..57a5fa4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -5,7 +5,7 @@ Pure-Rust HDF5 format implementation with HNSW vector search, WAL-backed persist ## Architecture -Cargo workspace with 18 crates under `crates/` (plus `libaec-sys`, an internal FFI bindings crate for the optional `szip` feature): +Cargo workspace with 19 crates under `crates/` (plus `libaec-sys`, an internal FFI bindings crate for the optional `szip` feature): | Crate | Role | |-------|------| @@ -26,6 +26,7 @@ Cargo workspace with 18 crates under `crates/` (plus `libaec-sys`, an internal F | `clawhdf5-napi` | Node.js native addon bindings | | `clawhdf5-py` | PyO3 Python bindings | | `clawhdf5-wasm` | WebAssembly (wasm-bindgen) reader for the browser; demo in `examples/wasm-viewer/` | +| `clawhdf5-remote` | Remote files: `open_url` over HTTP(S) range requests through a mandatory block cache (`BlockCache`) | | `clawhdf5-bench` | Benchmark suite | ## Key Features @@ -157,6 +158,18 @@ Cargo workspace with 18 crates under `crates/` (plus `libaec-sys`, an internal F `docs/known-issues.md`). Test changes with `cargo test -p clawhdf5-tools --test edit_interop` (h5py, h5dump, `h5rs check`). +- Remote files (`clawhdf5-remote`, range-read milestone M3 of + `docs/design/range-reads.md`): `open_url("http://…")` gives a + `clawhdf5::File` over `File::open_storage`, read through `BlockCache` + (1 MiB blocks, LRU byte budget, per-block in-flight dedup across threads, + runs coalesced into parallel requests). `HttpStorage` pins the file by + ETag/Last-Modified and length (a change is `RemoteError::FileChanged`), + refuses servers that ignore `Range` unless a full download is allowed, + and retries transient failures. Default build is plain HTTP with no C; + `https` (rustls + ring) is opt-in. Tests run a std-only HTTP server + (`tests/common/server.rs`, also the `range_server` example); + `CLAWHDF5_REMOTE_CORPUS=conformance/.cache/corpus` compares every corpus + file over HTTP with `File::open`. - GPU-accelerated vector distance computation (`clawhdf5-gpu`, wgpu); HDF5 I/O itself is CPU-only - Browser: `clawhdf5-wasm` (wasm-bindgen, read-only, file held in memory; no Zstd/SZIP since they link C) and the `examples/wasm-viewer/` page. diff --git a/Cargo.toml b/Cargo.toml index 0801b3d..dbb2f5c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,7 @@ members = [ "crates/clawhdf5-bench", "crates/clawhdf5-tools", "crates/clawhdf5-wasm", + "crates/clawhdf5-remote", "crates/libaec-sys", ] resolver = "2" diff --git a/crates/clawhdf5-remote/Cargo.toml b/crates/clawhdf5-remote/Cargo.toml new file mode 100644 index 0000000..6e68a73 --- /dev/null +++ b/crates/clawhdf5-remote/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "clawhdf5-remote" +version = "2.7.0" +edition = "2024" +rust-version.workspace = true +description = "Read HDF5 files over HTTP(S) range requests with clawhdf5, through a block cache" +license = "MIT" +repository = "https://git.redclaw.dev/quantumclaw/clawhdf5" +readme = "README.md" +keywords = ["hdf5", "http", "s3", "range-requests", "science"] +categories = ["science", "network-programming"] + +[features] +# Plain-HTTP range reads and the block cache: pure Rust, no TLS stack, no C. +default = ["http"] +http = ["dep:ureq"] +# HTTPS through rustls (ring provider, Mozilla roots). ring compiles C and +# assembly, so this is not part of the default build. +https = ["http", "ureq/rustls"] + +[dependencies] +clawhdf5 = { path = "../clawhdf5", version = "2.7.0" } +clawhdf5-format = { path = "../clawhdf5-format", version = "2.7.0" } +ureq = { version = "3.4", optional = true, default-features = false } + +[dev-dependencies] +tempfile = { workspace = true } diff --git a/crates/clawhdf5-remote/README.md b/crates/clawhdf5-remote/README.md new file mode 100644 index 0000000..bb20f71 --- /dev/null +++ b/crates/clawhdf5-remote/README.md @@ -0,0 +1,71 @@ +# clawhdf5-remote + +Read HDF5 files where they live — on an HTTP(S) server — with +[clawhdf5](../../README.md), without downloading them first. + +```rust +let file = clawhdf5_remote::open_url("http://127.0.0.1:8000/data.h5")?; +let temps = file.dataset("/grid/temperature")?.read_f64()?; +``` + +The result is an ordinary `clawhdf5::File`: groups, datasets, attributes, +selections, variable-length data. Only the bytes an operation needs are +fetched, by `Range` requests, through a block cache. + +## How it reads + +- **Block cache** (`BlockCache`, mandatory for remote files; see + `docs/design/range-reads.md` §2): aligned blocks of 1 MiB by default, LRU + with a byte budget (64 MiB by default), the missing blocks of one read + fetched as runs of consecutive blocks (a gap of one block is fetched to + merge two runs), each request at most 8 MiB, the requests of one read in + parallel. A read that misses more than half the budget is not kept, so a + big dataset read does not evict the metadata. Readers on several threads + share one cache; a block is never fetched twice at once — a second reader + waits for the first one's request. +- **Opening costs one request**: a `GET` of the first block, whose + `Content-Range` gives the file's length. HDF5 files keep the superblock + and usually the root group's metadata there, so listing a small file + often needs nothing more. +- **Pinned to one version of the file**: a strong `ETag` is sent back as + `If-Match` (else `Last-Modified` as `If-Unmodified-Since`) and checked on + every response, as is the length. A file replaced while open is an error + (`RemoteError::FileChanged`), never a mix of old and new bytes. A server + with neither validator can only be checked by length; + `HttpOptions::require_validator` refuses it. +- **Servers that ignore `Range`** (answer `200` with the whole file) are + refused (`RemoteError::RangeNotSupported`) without reading the body, + unless `HttpOptions::allow_full_download` is set; then the file is + downloaded once and read from memory. +- **Retries**: connection failures, timeouts, `408`/`429`/`5xx` and bodies + that end early are retried with exponential backoff (3 retries, from + 200 ms). Bodies are requested with `Accept-Encoding: identity`; an encoded + body is refused. + +The zero-copy methods of `clawhdf5` (`read_raw_ref`, `read_*_zerocopy`, +`File::as_bytes`) borrow the whole file from memory, so they are errors +(`as_bytes` a panic; use `File::contiguous_bytes`) on a remote file. + +## Features + +| Feature | What | C code | +|---|---|---| +| `http` (default) | `http://` through `ureq`, no TLS | none | +| `https` | `https://` through rustls, ring provider, Mozilla roots | ring (C and assembly) | + +The default build compiles no C (`scripts/ci-test.sh` checks it). + +## Counting requests + +```rust +use clawhdf5_remote::{storage_for_url, Options}; +let storage = storage_for_url(url, &Options::default())?; +let file = clawhdf5::File::open_storage(storage.clone())?; +// ... read ... +let s = storage.stats(); // requests, bytes_fetched, hits, misses, cached_bytes, ... +``` + +`cargo run -p clawhdf5-remote --example range_server -- DIR` serves a +directory with range support (the server the tests use), and +`cargo run -p clawhdf5-remote --example read_url -- URL [DATASET]` lists a +file and prints what it cost. diff --git a/crates/clawhdf5-remote/examples/range_server.rs b/crates/clawhdf5-remote/examples/range_server.rs new file mode 100644 index 0000000..609693f --- /dev/null +++ b/crates/clawhdf5-remote/examples/range_server.rs @@ -0,0 +1,56 @@ +//! Serve the HDF5 files of a directory over HTTP with range support — the +//! server the tests use — to try `clawhdf5_remote` and `h5rs` on URLs. +//! +//! ```text +//! cargo run -p clawhdf5-remote --example range_server -- DIR [127.0.0.1:8000] +//! ``` +//! +//! Every file under `DIR` is served at its path relative to `DIR`. The +//! files are read into memory at start. Requests are logged to stderr. + +#[path = "../tests/common/server.rs"] +mod server; + +fn main() { + let mut args = std::env::args().skip(1); + let Some(dir) = args.next() else { + eprintln!("usage: range_server DIR [ADDR]"); + std::process::exit(2); + }; + let addr = args.next().unwrap_or_else(|| "127.0.0.1:8000".into()); + let root = std::path::PathBuf::from(&dir); + let mut files = Vec::new(); + let mut stack = vec![root.clone()]; + while let Some(d) = stack.pop() { + let Ok(entries) = std::fs::read_dir(&d) else { + continue; + }; + for e in entries.flatten() { + let p = e.path(); + if p.is_dir() { + stack.push(p); + } else if let (Ok(rel), Ok(bytes)) = (p.strip_prefix(&root), std::fs::read(&p)) { + let url = format!("/{}", rel.to_string_lossy().replace('\\', "/")); + files.push((url, bytes)); + } + } + } + files.sort(); + let server = server::Server::bind(&addr, files.clone()); + for (path, bytes) in &files { + eprintln!("{} ({} bytes)", server.url(path), bytes.len()); + } + eprintln!("serving {} files on http://{}", files.len(), server.addr); + let mut logged = 0; + loop { + std::thread::sleep(std::time::Duration::from_millis(200)); + let log = server.log(); + for (path, range) in &log[logged.min(log.len())..] { + match range { + Some((a, b)) => eprintln!("GET {path} bytes={a}-{b}"), + None => eprintln!("GET {path} (whole file)"), + } + } + logged = log.len(); + } +} diff --git a/crates/clawhdf5-remote/examples/read_url.rs b/crates/clawhdf5-remote/examples/read_url.rs new file mode 100644 index 0000000..145b4ce --- /dev/null +++ b/crates/clawhdf5-remote/examples/read_url.rs @@ -0,0 +1,45 @@ +//! List a remote HDF5 file and read one dataset, then print what it cost. +//! +//! ```text +//! cargo run -p clawhdf5-remote --example read_url -- URL [DATASET] +//! ``` + +use clawhdf5::File; +use clawhdf5_remote::{Options, storage_for_url}; + +fn main() -> Result<(), Box> { + let mut args = std::env::args().skip(1); + let url = args.next().ok_or("usage: read_url URL [DATASET]")?; + let dataset = args.next(); + + let storage = storage_for_url(&url, &Options::default())?; + let file = File::open_storage(storage.clone())?; + + // Walk the tree. + let mut stack = vec![("/".to_string(), file.root())]; + while let Some((path, group)) = stack.pop() { + for (name, addr) in group.entries()? { + let child = format!("{}/{name}", path.trim_end_matches('/')); + match file.dataset_at(addr) { + Ok(ds) => println!("{child} dataset {:?} {:?}", ds.shape()?, ds.dtype()?), + Err(_) => { + println!("{child} group"); + stack.push((child, file.group_at(addr))); + } + } + } + } + + if let Some(name) = dataset { + let values = file.dataset(&name)?.read_f64()?; + let shown: Vec<_> = values.iter().take(8).collect(); + println!("{name}: {} values, first {shown:?}", values.len()); + } + + let s = storage.stats(); + println!( + "{} range requests (the one at open included), {} bytes fetched, {} bytes cached", + s.requests, s.bytes_fetched, s.cached_bytes + ); + Ok(()) +} diff --git a/crates/clawhdf5-remote/src/cache.rs b/crates/clawhdf5-remote/src/cache.rs new file mode 100644 index 0000000..233938c --- /dev/null +++ b/crates/clawhdf5-remote/src/cache.rs @@ -0,0 +1,825 @@ +//! The block cache every remote backend is read through. +//! +//! `docs/design/range-reads.md` §2 measured why it is required: the parsers +//! revisit the same structures many times (410 reads to list the 21 objects +//! of a 7.7 MB file), so mapping reads one-to-one onto requests is hopeless, +//! while a cache of 1 MiB blocks lists the same file in 2 requests. +//! +//! [`BlockCache`] wraps any [`Storage`] (the backend that does the fetching) +//! and serves reads from fixed-size, aligned blocks: +//! +//! - **LRU with a byte budget.** Blocks are evicted least-recently-used +//! first once the cached bytes exceed [`CacheConfig::capacity`]. Reads +//! hold their blocks by reference count, so eviction never invalidates +//! data a read is still using. +//! - **Coalescing.** All the blocks one `read_at`/`read_ranges` call misses +//! are fetched with one `read_ranges` call on the backend, as runs of +//! consecutive blocks (a gap of up to [`CacheConfig::coalesce_gap`] bytes +//! of uncached blocks is fetched too, to merge two runs), each run at +//! most [`CacheConfig::max_request`] bytes. A backend fetches the runs of +//! one call in parallel. +//! - **Concurrency.** The cache's lock is held only to look blocks up and +//! to insert them, never across a fetch. A block being fetched is marked +//! in flight: a second reader that needs it waits for that fetch instead +//! of issuing its own, so concurrent readers never fetch a block twice. +//! A failed fetch fails every reader waiting for it, and is not cached. +//! - **Large reads do not flush the cache.** A call whose missing blocks +//! add up to more than half the budget is served without keeping them +//! (a big chunked read would otherwise evict all the metadata). +//! - **Local storages pass through.** A backend that holds the whole file +//! in memory ([`Storage::as_contiguous`]) is read directly. + +use std::borrow::Cow; +use std::collections::{BTreeMap, BTreeSet, HashMap}; +use std::ops::Range; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Condvar, Mutex, MutexGuard}; + +use clawhdf5_format::error::FormatError; +use clawhdf5_format::storage::Storage; + +/// Default block size: 1 MiB, the size `docs/design/range-reads.md` §2 +/// measured (metadata of the test files in 1–15 blocks; the chunk index of a +/// netCDF file spread over 14 blocks of a 48 MB file). +pub const DEFAULT_BLOCK_SIZE: u64 = 1 << 20; + +/// Settings of a [`BlockCache`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CacheConfig { + /// Size of a block, in bytes; every fetch is whole, aligned blocks + /// (the last block of the file is shorter). + pub block_size: u64, + /// Byte budget of cached blocks. + pub capacity: u64, + /// Two runs of missing blocks separated by at most this many bytes of + /// uncached blocks are fetched as one request (the gap included). + pub coalesce_gap: u64, + /// Largest single request, in bytes (rounded down to whole blocks, at + /// least one block). Longer runs are split, so a backend can fetch the + /// pieces in parallel. + pub max_request: u64, +} + +impl Default for CacheConfig { + fn default() -> Self { + CacheConfig { + block_size: DEFAULT_BLOCK_SIZE, + capacity: 64 << 20, + coalesce_gap: DEFAULT_BLOCK_SIZE, + max_request: 8 << 20, + } + } +} + +/// What a [`BlockCache`] has done so far. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct CacheStats { + /// Reads served (`read_at` calls plus ranges of `read_ranges` calls). + pub reads: u64, + /// Block lookups that found the block cached. + pub hits: u64, + /// Block lookups that had to fetch the block (or wait for it). + pub misses: u64, + /// Block lookups that waited for another reader's fetch of the block. + pub waits: u64, + /// Ranges requested from the backend: for HTTP, one request each. + pub requests: u64, + /// `read_ranges` calls made on the backend. + pub fetch_calls: u64, + /// Bytes fetched from the backend. + pub bytes_fetched: u64, + /// Blocks evicted to stay within the budget. + pub evictions: u64, + /// Bytes currently cached. + pub cached_bytes: u64, +} + +type Block = Arc<[u8]>; + +/// A fetch in progress: the readers waiting for a block wait on this. +struct Flight { + result: Mutex>>, + done: Condvar, +} + +impl Flight { + fn new() -> Arc { + Arc::new(Flight { + result: Mutex::new(None), + done: Condvar::new(), + }) + } + + fn finish(&self, r: Result) { + let mut slot = lock(&self.result); + if slot.is_none() { + *slot = Some(r); + } + self.done.notify_all(); + } + + fn wait(&self) -> Result { + let mut slot = lock(&self.result); + loop { + if let Some(r) = slot.as_ref() { + return r.clone(); + } + slot = self + .done + .wait(slot) + .unwrap_or_else(std::sync::PoisonError::into_inner); + } + } +} + +enum Slot { + Ready { data: Block, tick: u64 }, + Pending(Arc), +} + +#[derive(Default)] +struct State { + blocks: HashMap, + /// tick -> block index, oldest first. + lru: BTreeMap, + tick: u64, + bytes: u64, +} + +fn lock(m: &Mutex) -> MutexGuard<'_, T> { + m.lock().unwrap_or_else(std::sync::PoisonError::into_inner) +} + +#[derive(Default)] +struct Counters { + reads: AtomicU64, + hits: AtomicU64, + misses: AtomicU64, + waits: AtomicU64, + requests: AtomicU64, + fetch_calls: AtomicU64, + bytes_fetched: AtomicU64, + evictions: AtomicU64, +} + +/// A [`Storage`] that serves reads of `inner` from an LRU cache of +/// fixed-size blocks. See the [module documentation](self). +pub struct BlockCache { + inner: S, + config: CacheConfig, + len: u64, + state: Mutex, + counters: Counters, +} + +/// Fails every flight a fetch claimed and did not complete (an error or a +/// panic in the backend), so no reader waits forever. +struct FlightGuard<'a, S> { + cache: &'a BlockCache, + flights: Vec<(u64, Arc)>, +} + +impl Drop for FlightGuard<'_, S> { + fn drop(&mut self) { + if self.flights.is_empty() { + return; + } + let mut st = lock(&self.cache.state); + for (i, f) in &self.flights { + if matches!(st.blocks.get(i), Some(Slot::Pending(p)) if Arc::ptr_eq(p, f)) { + st.blocks.remove(i); + } + } + drop(st); + for (_, f) in &self.flights { + f.finish(Err("the fetch of this block failed".into())); + } + } +} + +impl BlockCache { + /// Cache `inner` with the given settings. The length of the file is + /// taken from `inner` once, here: a remote file is pinned at open. + pub fn new(inner: S, mut config: CacheConfig) -> Self { + config.block_size = config.block_size.max(512); + config.capacity = config.capacity.max(config.block_size); + config.max_request = (config.max_request / config.block_size).max(1) * config.block_size; + let len = inner.len(); + BlockCache { + inner, + config, + len, + state: Mutex::new(State::default()), + counters: Counters::default(), + } + } + + /// The backend. + pub fn inner(&self) -> &S { + &self.inner + } + + /// The settings in use (after rounding). + pub fn config(&self) -> &CacheConfig { + &self.config + } + + /// Counters since the cache was made (or last [`reset_stats`](Self::reset_stats)). + pub fn stats(&self) -> CacheStats { + let c = &self.counters; + CacheStats { + reads: c.reads.load(Ordering::Relaxed), + hits: c.hits.load(Ordering::Relaxed), + misses: c.misses.load(Ordering::Relaxed), + waits: c.waits.load(Ordering::Relaxed), + requests: c.requests.load(Ordering::Relaxed), + fetch_calls: c.fetch_calls.load(Ordering::Relaxed), + bytes_fetched: c.bytes_fetched.load(Ordering::Relaxed), + evictions: c.evictions.load(Ordering::Relaxed), + cached_bytes: lock(&self.state).bytes, + } + } + + /// Zero the counters (the cached blocks stay). + pub fn reset_stats(&self) { + let c = &self.counters; + for a in [ + &c.reads, + &c.hits, + &c.misses, + &c.waits, + &c.requests, + &c.fetch_calls, + &c.bytes_fetched, + &c.evictions, + ] { + a.store(0, Ordering::Relaxed); + } + } + + /// Drop every cached block. + pub fn clear(&self) { + let mut st = lock(&self.state); + st.blocks.retain(|_, s| matches!(s, Slot::Pending(_))); + st.lru.clear(); + st.bytes = 0; + } + + /// Fetch the blocks covering `[offset, offset + len)` now (readahead), + /// keeping them cached. + pub fn prefetch(&self, offset: u64, len: u64) -> Result<(), FormatError> { + if self.inner.as_contiguous().is_some() { + return Ok(()); + } + let Some(blocks) = self.block_span(offset, len) else { + return Ok(()); + }; + self.blocks(&blocks.collect::>(), true).map(|_| ()) + } + + /// Put bytes already fetched (such as the first block, which a backend + /// may get while it probes the file's length) into the cache: `bytes` + /// are the file's bytes at `offset`. Only whole blocks (or the file's + /// last, short block) are kept; a block already cached is left alone. + /// The bytes count as one request in [`CacheStats`]. + pub fn insert(&self, offset: u64, bytes: &[u8]) { + self.counters.requests.fetch_add(1, Ordering::Relaxed); + self.counters + .bytes_fetched + .fetch_add(bytes.len() as u64, Ordering::Relaxed); + let bs = self.config.block_size; + let end = offset.saturating_add(bytes.len() as u64).min(self.len); + let mut i = offset.div_ceil(bs); + let mut st = lock(&self.state); + while i * bs < end { + let start = i * bs; + let block_end = (start + bs).min(self.len); + if block_end > end { + break; + } + if !st.blocks.contains_key(&i) { + let rel = (start - offset) as usize..(block_end - offset) as usize; + self.keep(&mut st, i, Arc::from(&bytes[rel])); + } + i += 1; + } + self.evict(&mut st); + } + + /// Block indices covering `[offset, offset + len)`, clamped to the file. + fn block_span(&self, offset: u64, len: u64) -> Option> { + let end = offset.saturating_add(len).min(self.len); + if offset >= end { + return None; + } + let bs = self.config.block_size; + Some(offset / bs..(end - 1) / bs + 1) + } + + fn block_len(&self, i: u64) -> u64 { + let start = i * self.config.block_size; + (start + self.config.block_size).min(self.len) - start + } + + fn keep(&self, st: &mut State, i: u64, data: Block) { + st.tick += 1; + let tick = st.tick; + st.bytes += <[u8]>::len(&data) as u64; + st.lru.insert(tick, i); + st.blocks.insert(i, Slot::Ready { data, tick }); + } + + fn evict(&self, st: &mut State) { + while st.bytes > self.config.capacity { + let Some((_, i)) = st.lru.pop_first() else { + break; + }; + if let Some(Slot::Ready { data, .. }) = st.blocks.remove(&i) { + st.bytes -= <[u8]>::len(&data) as u64; + self.counters.evictions.fetch_add(1, Ordering::Relaxed); + } + } + } + + /// The blocks `wanted` (sorted, distinct), fetching the missing ones. + fn blocks(&self, wanted: &[u64], may_keep: bool) -> Result, FormatError> { + let mut have = HashMap::with_capacity(wanted.len()); + let mut waits = Vec::new(); + let mut guard = FlightGuard { + cache: self, + flights: Vec::new(), + }; + { + let mut st = lock(&self.state); + for &i in wanted { + match st.blocks.get(&i) { + Some(Slot::Ready { data, tick }) => { + let (data, old) = (data.clone(), *tick); + st.tick += 1; + let tick = st.tick; + st.lru.remove(&old); + st.lru.insert(tick, i); + if let Some(Slot::Ready { tick: t, .. }) = st.blocks.get_mut(&i) { + *t = tick; + } + self.counters.hits.fetch_add(1, Ordering::Relaxed); + have.insert(i, data); + } + Some(Slot::Pending(f)) => { + self.counters.misses.fetch_add(1, Ordering::Relaxed); + self.counters.waits.fetch_add(1, Ordering::Relaxed); + waits.push((i, f.clone())); + } + None => { + self.counters.misses.fetch_add(1, Ordering::Relaxed); + let f = Flight::new(); + st.blocks.insert(i, Slot::Pending(f.clone())); + guard.flights.push((i, f)); + } + } + } + // Fill small gaps between runs with blocks nobody has, so the + // runs merge into one request. + if self.config.coalesce_gap > 0 && guard.flights.len() > 1 { + let gap_blocks = self.config.coalesce_gap / self.config.block_size; + let mut extra = Vec::new(); + for w in guard.flights.windows(2) { + let (a, b) = (w[0].0, w[1].0); + let gap = b - a - 1; + if gap == 0 || gap > gap_blocks { + continue; + } + if (a + 1..b).all(|j| !st.blocks.contains_key(&j)) { + extra.extend(a + 1..b); + } + } + for j in extra { + let f = Flight::new(); + st.blocks.insert(j, Slot::Pending(f.clone())); + guard.flights.push((j, f)); + } + guard.flights.sort_by_key(|(i, _)| *i); + } + } + + if !guard.flights.is_empty() { + let runs = self.runs(&guard.flights); + let missing_bytes: u64 = runs.iter().map(|r| r.end - r.start).sum(); + let keep = may_keep && missing_bytes <= self.config.capacity / 2; + self.counters.fetch_calls.fetch_add(1, Ordering::Relaxed); + self.counters + .requests + .fetch_add(runs.len() as u64, Ordering::Relaxed); + let fetched = self.inner.read_ranges(&runs)?; + if fetched.len() != runs.len() { + return Err(FormatError::Storage(format!( + "backend returned {} ranges for {} requested", + fetched.len(), + runs.len() + ))); + } + let mut got: Vec<(u64, Block)> = Vec::with_capacity(guard.flights.len()); + for (run, bytes) in runs.iter().zip(&fetched) { + self.counters + .bytes_fetched + .fetch_add(bytes.len() as u64, Ordering::Relaxed); + if bytes.len() as u64 != run.end - run.start { + return Err(FormatError::Storage(format!( + "short read from the backend: {} of {} bytes at offset {}", + bytes.len(), + run.end - run.start, + run.start + ))); + } + let bs = self.config.block_size; + let mut i = run.start / bs; + let mut pos = 0usize; + while pos < bytes.len() { + let n = self.block_len(i) as usize; + got.push((i, Arc::from(&bytes[pos..pos + n]))); + pos += n; + i += 1; + } + } + drop(fetched); + let flights = std::mem::take(&mut guard.flights); + let mut st = lock(&self.state); + let mut by_index: HashMap = got.into_iter().collect(); + for (i, f) in &flights { + let Some(data) = by_index.remove(i) else { + continue; + }; + let ours = matches!(st.blocks.get(i), Some(Slot::Pending(p)) if Arc::ptr_eq(p, f)); + if ours { + if keep { + self.keep(&mut st, *i, data.clone()); + } else { + st.blocks.remove(i); + } + } + f.finish(Ok(data.clone())); + have.insert(*i, data); + } + self.evict(&mut st); + drop(st); + // Any flight without data (cannot happen: every run is split + // into its blocks) is failed rather than left waiting. + guard.flights = flights + .into_iter() + .filter(|(i, _)| !have.contains_key(i)) + .collect(); + if !guard.flights.is_empty() { + return Err(FormatError::Storage( + "backend did not return every block".into(), + )); + } + } + + for (i, f) in waits { + match f.wait() { + Ok(data) => { + have.insert(i, data); + } + Err(e) => return Err(FormatError::Storage(e)), + } + } + Ok(have) + } + + /// Byte ranges of the runs of consecutive blocks in `flights` (sorted), + /// each at most `max_request` long. + fn runs(&self, flights: &[(u64, Arc)]) -> Vec> { + let bs = self.config.block_size; + let per_request = self.config.max_request / bs; + let mut runs: Vec<(u64, u64)> = Vec::new(); + for &(i, _) in flights { + match runs.last_mut() { + Some((first, last)) if *last + 1 == i && i - *first < per_request => *last = i, + _ => runs.push((i, i)), + } + } + runs.into_iter() + .map(|(a, b)| a * bs..(b * bs + self.block_len(b))) + .collect() + } + + /// Copy `[offset, end)` out of `blocks`. + fn assemble(&self, offset: u64, end: u64, blocks: &HashMap) -> Vec { + let bs = self.config.block_size; + let mut out = Vec::with_capacity((end - offset) as usize); + let mut pos = offset; + while pos < end { + let i = pos / bs; + let block = &blocks[&i]; + let from = (pos - i * bs) as usize; + let to = ((end - i * bs) as usize).min(<[u8]>::len(block)); + out.extend_from_slice(&block[from..to]); + pos = i * bs + to as u64; + } + out + } +} + +impl Storage for BlockCache { + fn read_at(&self, offset: u64, len: usize) -> Result, FormatError> { + if let Some(all) = self.inner.as_contiguous() { + return all.read_at(offset, len); + } + self.counters.reads.fetch_add(1, Ordering::Relaxed); + let Some(span) = self.block_span(offset, len as u64) else { + return Ok(Cow::Owned(Vec::new())); + }; + let end = offset.saturating_add(len as u64).min(self.len); + let blocks = self.blocks(&span.collect::>(), true)?; + Ok(Cow::Owned(self.assemble(offset, end, &blocks))) + } + + fn len(&self) -> u64 { + self.len + } + + fn read_ranges(&self, ranges: &[Range]) -> Result>, FormatError> { + if let Some(all) = self.inner.as_contiguous() { + return all.read_ranges(ranges); + } + self.counters + .reads + .fetch_add(ranges.len() as u64, Ordering::Relaxed); + let mut wanted = BTreeSet::new(); + for r in ranges { + if r.end < r.start { + return Err(FormatError::Storage( + "read range ends before it starts".into(), + )); + } + if let Some(span) = self.block_span(r.start, r.end - r.start) { + wanted.extend(span); + } + } + let wanted: Vec = wanted.into_iter().collect(); + let blocks = self.blocks(&wanted, true)?; + Ok(ranges + .iter() + .map(|r| { + let end = r.end.min(self.len); + if r.start >= end { + Cow::Owned(Vec::new()) + } else { + Cow::Owned(self.assemble(r.start, end, &blocks)) + } + }) + .collect()) + } + + fn as_contiguous(&self) -> Option<&[u8]> { + self.inner.as_contiguous() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use clawhdf5_format::storage::CountingStorage; + + fn data(n: usize) -> Vec { + (0..n).map(|i| (i * 7 + i / 251) as u8).collect() + } + + fn cache(n: usize, config: CacheConfig) -> (Vec, BlockCache) { + let d = data(n); + (d.clone(), BlockCache::new(CountingStorage::new(d), config)) + } + + fn small() -> CacheConfig { + CacheConfig { + block_size: 1024, + capacity: 8 * 1024, + coalesce_gap: 0, + max_request: 4 * 1024, + } + } + + #[test] + fn reads_match_the_data_and_hit_the_cache() { + let mut cfg = small(); + cfg.capacity = 64 * 1024; + let (d, c) = cache(10_000, cfg); + for (off, len) in [ + (0, 10), + (1000, 100), + (1020, 10), + (9990, 100), + (20_000, 5), + (0, 10_000), + ] { + let got = c.read_at(off, len).unwrap(); + let s = (off as usize).min(d.len()); + let e = (off as usize + len).min(d.len()); + assert_eq!(&*got, &d[s..e], "{off} {len}"); + } + let before = c.inner().reads(); + assert_eq!(&*c.read_at(5000, 3000).unwrap(), &d[5000..8000]); + assert_eq!(c.inner().reads(), before, "all cached"); + assert!(c.stats().hits > 0); + } + + #[test] + fn read_ranges_coalesces_consecutive_missing_blocks() { + let (d, c) = cache(20_000, small()); + let ranges = [100..200, 1500..1600, 2100..3000, 9000..9100]; + let got = c.read_ranges(&ranges).unwrap(); + for (r, g) in ranges.iter().zip(&got) { + assert_eq!(&**g, &d[r.start as usize..r.end as usize]); + } + // Blocks 0-2 are one run, block 8 another: one backend call, two ranges. + assert_eq!(c.stats().fetch_calls, 1); + assert_eq!(c.stats().requests, 2); + assert_eq!(c.inner().reads(), 2); + } + + #[test] + fn a_small_gap_is_fetched_to_merge_runs() { + let mut cfg = small(); + cfg.coalesce_gap = 1024; + let (d, c) = cache(20_000, cfg); + let ranges = [0..10, 2048..2058, 5000..5010]; + let got = c.read_ranges(&ranges).unwrap(); + for (r, g) in ranges.iter().zip(&got) { + assert_eq!(&**g, &d[r.start as usize..r.end as usize]); + } + // Blocks 0 and 2 merge over block 1; block 4 is two blocks away. + assert_eq!(c.stats().requests, 2); + } + + #[test] + fn long_runs_are_split_at_max_request() { + let (d, c) = cache(20_000, small()); + assert_eq!(&*c.read_at(0, 10_000).unwrap(), &d[..10_000]); + // 10 blocks, 4 per request: 3 requests in one call. + assert_eq!(c.stats().requests, 3); + assert_eq!(c.stats().fetch_calls, 1); + } + + #[test] + fn lru_stays_within_budget_and_evicts_oldest() { + let (d, c) = cache(40_000, small()); + for i in 0..8u64 { + c.read_at(i * 1024, 1).unwrap(); + } + c.read_at(0, 1).unwrap(); // block 0 is now the newest + c.read_at(8 * 1024, 1).unwrap(); // evicts block 1 + let s = c.stats(); + assert!(s.cached_bytes <= 8 * 1024); + assert_eq!(s.evictions, 1); + let before = c.inner().reads(); + c.read_at(0, 1).unwrap(); + assert_eq!(c.inner().reads(), before, "block 0 kept"); + assert_eq!(&*c.read_at(1024, 5).unwrap(), &d[1024..1029]); + assert_eq!(c.inner().reads(), before + 1, "block 1 refetched"); + } + + #[test] + fn large_reads_do_not_flush_the_cache() { + let (d, c) = cache(40_000, small()); + c.read_at(0, 1).unwrap(); + assert_eq!(&*c.read_at(10_000, 20_000).unwrap(), &d[10_000..30_000]); + assert_eq!(c.stats().cached_bytes, 1024, "only block 0 kept"); + } + + #[test] + fn insert_keeps_whole_blocks_only() { + let (d, c) = cache(5000, small()); + c.insert(0, &d[..2500]); + assert_eq!(c.stats().cached_bytes, 2048); + c.insert(4096, &d[4096..]); + assert_eq!(c.stats().cached_bytes, 2048 + 904); + assert_eq!(&*c.read_at(0, 2048).unwrap(), &d[..2048]); + assert_eq!(&*c.read_at(4500, 500).unwrap(), &d[4500..]); + assert_eq!(c.inner().reads(), 0); + } + + /// A backend that fails its first `n` fetches. + struct Flaky { + data: Vec, + fail: AtomicU64, + } + + impl Storage for Flaky { + fn read_at(&self, offset: u64, len: usize) -> Result, FormatError> { + if self + .fail + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1)) + .is_ok() + { + return Err(FormatError::Storage("boom".into())); + } + self.data + .as_slice() + .read_at(offset, len) + .map(|c| Cow::Owned(c.into_owned())) + } + fn len(&self) -> u64 { + self.data.len() as u64 + } + } + + #[test] + fn a_failed_fetch_is_an_error_and_is_not_cached() { + let d = data(5000); + let c = BlockCache::new( + Flaky { + data: d.clone(), + fail: AtomicU64::new(1), + }, + small(), + ); + assert!(c.read_at(0, 10).is_err()); + assert_eq!(&*c.read_at(0, 10).unwrap(), &d[..10]); + } + + /// A backend that returns fewer bytes than asked inside the file. + struct Short(Vec); + + impl Storage for Short { + fn read_at(&self, offset: u64, len: usize) -> Result, FormatError> { + let got = self.0.as_slice().read_at(offset, len)?; + Ok(Cow::Owned(got[..got.len() / 2].to_vec())) + } + fn len(&self) -> u64 { + self.0.len() as u64 + } + } + + #[test] + fn a_short_backend_read_is_an_error() { + let c = BlockCache::new(Short(data(5000)), small()); + assert!(c.read_at(0, 10).is_err()); + assert!(c.read_at(0, 10).is_err(), "not cached"); + } + + #[test] + fn contiguous_backends_pass_through() { + let d = data(5000); + let c = BlockCache::new(d.clone(), small()); + assert_eq!(c.as_contiguous(), Some(&d[..])); + assert!(matches!(c.read_at(0, 10).unwrap(), Cow::Borrowed(_))); + assert_eq!(c.stats().requests, 0); + } + + /// A slow backend: concurrent readers of the same blocks share fetches. + struct Slow { + data: Vec, + fetched: Mutex>>, + } + + impl Storage for Slow { + fn read_at(&self, offset: u64, len: usize) -> Result, FormatError> { + std::thread::sleep(std::time::Duration::from_millis(30)); + lock(&self.fetched).push(offset..offset + len as u64); + self.data + .as_slice() + .read_at(offset, len) + .map(|c| Cow::Owned(c.into_owned())) + } + fn len(&self) -> u64 { + self.data.len() as u64 + } + } + + #[test] + fn concurrent_readers_never_fetch_a_block_twice() { + let d = data(64 * 1024); + let cfg = CacheConfig { + block_size: 1024, + capacity: 1 << 20, + coalesce_gap: 0, + max_request: 1024, + }; + let c = BlockCache::new( + Slow { + data: d.clone(), + fetched: Mutex::new(Vec::new()), + }, + cfg, + ); + std::thread::scope(|s| { + for t in 0..8u64 { + let (c, d) = (&c, &d); + s.spawn(move || { + for k in 0..16u64 { + let off = ((k * 3 + t) % 32) * 1024 + 100; + let got = c.read_at(off, 2000).unwrap(); + assert_eq!(&*got, &d[off as usize..off as usize + 2000]); + } + }); + } + }); + let fetched = lock(&c.inner().fetched).clone(); + let mut blocks: Vec = fetched.iter().map(|r| r.start / 1024).collect(); + let n = blocks.len(); + blocks.sort_unstable(); + blocks.dedup(); + assert_eq!(n, blocks.len(), "a block was fetched twice: {fetched:?}"); + assert!(c.stats().waits > 0, "readers should have shared fetches"); + } +} diff --git a/crates/clawhdf5-remote/src/error.rs b/crates/clawhdf5-remote/src/error.rs new file mode 100644 index 0000000..d5d1feb --- /dev/null +++ b/crates/clawhdf5-remote/src/error.rs @@ -0,0 +1,125 @@ +//! Errors of the remote backends. + +use clawhdf5_format::error::FormatError; + +/// Why a remote file could not be opened or read. +/// +/// Inside a [`clawhdf5::File`] read these arrive as +/// `clawhdf5::Error::Format(FormatError::Storage(message))`, the message +/// being this error's `Display`. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub enum RemoteError { + /// The URL is malformed. + InvalidUrl(String), + /// The URL's scheme is not supported by this build (for example + /// `s3://` without the `s3` feature, or `https://` without `https`). + UnsupportedScheme(String), + /// The server answered a range request with the whole file (status + /// 200), i.e. it does not support ranges, and a full download was not + /// allowed ([`HttpOptions::allow_full_download`](crate::HttpOptions)). + RangeNotSupported(String), + /// The file changed since it was opened (a different ETag, + /// Last-Modified or length, or a failed `If-Match` precondition). + /// Nothing read after the change is returned. + FileChanged(String), + /// The server answered with an unexpected status. + Status { + /// The HTTP status code. + code: u16, + /// What was being requested. + what: String, + }, + /// A response did not carry what was asked for (a wrong + /// `Content-Range`, a body shorter or longer than announced), after + /// every retry. + BadResponse(String), + /// A network failure (connection, timeout, reset) after every retry. + Transport(String), + /// An error from the object store. + ObjectStore(String), + /// Called in a way the backend cannot serve (for example a blocking + /// read from inside an async runtime). + Usage(String), +} + +impl RemoteError { + /// Whether retrying the same request may succeed. + #[cfg_attr(not(feature = "http"), allow(dead_code))] + pub(crate) fn is_transient(&self) -> bool { + match self { + RemoteError::Transport(_) | RemoteError::BadResponse(_) => true, + RemoteError::Status { code, .. } => { + matches!(code, 408 | 429 | 500 | 502 | 503 | 504) + } + _ => false, + } + } +} + +impl std::fmt::Display for RemoteError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + RemoteError::InvalidUrl(s) => write!(f, "invalid URL: {s}"), + RemoteError::UnsupportedScheme(s) => write!(f, "unsupported URL: {s}"), + RemoteError::RangeNotSupported(s) => { + write!(f, "the server does not support range requests: {s}") + } + RemoteError::FileChanged(s) => write!(f, "the remote file changed while open: {s}"), + RemoteError::Status { code, what } => write!(f, "HTTP status {code} for {what}"), + RemoteError::BadResponse(s) => write!(f, "bad response: {s}"), + RemoteError::Transport(s) => write!(f, "network error: {s}"), + RemoteError::ObjectStore(s) => write!(f, "object store: {s}"), + RemoteError::Usage(s) => write!(f, "{s}"), + } + } +} + +impl std::error::Error for RemoteError {} + +impl From for FormatError { + fn from(e: RemoteError) -> Self { + FormatError::Storage(e.to_string()) + } +} + +/// An error of [`open_url`](crate::open_url): the remote side, or the file +/// itself. +#[derive(Debug)] +#[non_exhaustive] +pub enum Error { + /// Reaching or reading the remote file failed. + Remote(RemoteError), + /// The bytes were read, but they are not an HDF5 file clawhdf5 can open. + Hdf5(clawhdf5::Error), +} + +impl std::fmt::Display for Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Error::Remote(e) => e.fmt(f), + Error::Hdf5(e) => e.fmt(f), + } + } +} + +impl std::error::Error for Error { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Error::Remote(e) => Some(e), + Error::Hdf5(e) => Some(e), + } + } +} + +impl From for Error { + fn from(e: RemoteError) -> Self { + Error::Remote(e) + } +} + +impl From for Error { + fn from(e: clawhdf5::Error) -> Self { + Error::Hdf5(e) + } +} diff --git a/crates/clawhdf5-remote/src/http.rs b/crates/clawhdf5-remote/src/http.rs new file mode 100644 index 0000000..ce3d68f --- /dev/null +++ b/crates/clawhdf5-remote/src/http.rs @@ -0,0 +1,565 @@ +//! HTTP(S) range requests: [`HttpStorage`]. +//! +//! Every read is a `GET` with a `Range: bytes=a-b` header, answered `206 +//! Partial Content`. The file is pinned when it is opened: +//! +//! - its length comes from the `Content-Range` of the first request (which +//! also fetches the first block, so opening costs one request); +//! - a strong `ETag` is sent back as `If-Match` on every later request, and +//! compared with the `ETag` of every response; without one, `Last-Modified` +//! is sent as `If-Unmodified-Since` and compared; the length in every +//! `Content-Range` must stay the same. A file that changes while it is +//! open is [`RemoteError::FileChanged`], never a mix of old and new bytes. +//! (A server that sends neither validator cannot be checked beyond the +//! length; [`HttpOptions::require_validator`] refuses such servers.) +//! - a server that ignores `Range` and answers `200` with the whole file is +//! refused with [`RemoteError::RangeNotSupported`], unless +//! [`HttpOptions::allow_full_download`] is set: then the file is +//! downloaded once, at open, and read from memory. +//! +//! Transient failures — connection errors, timeouts, `408`/`429`/`5xx`, and +//! a body shorter or longer than its `Content-Range` — are retried with +//! exponential backoff. Responses are requested with +//! `Accept-Encoding: identity`, since a compressed body cannot be a byte +//! range of the file. +//! +//! `HttpStorage` itself does not cache: each `read_at` is one request. Read +//! it through [`BlockCache`](crate::BlockCache) (which [`open_url`](crate::open_url) +//! does); its `read_ranges` fetches the ranges of one call in parallel. + +use std::borrow::Cow; +use std::io::Read; +use std::ops::Range; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::time::Duration; + +use clawhdf5_format::error::FormatError; +use clawhdf5_format::storage::Storage; + +use crate::error::RemoteError; + +/// Settings of an [`HttpStorage`]. +#[derive(Debug, Clone)] +pub struct HttpOptions { + /// Retries of a request that failed transiently (so up to `retries + 1` + /// attempts). + pub retries: u32, + /// Delay before the first retry; doubled for each further one. + pub backoff: Duration, + /// Timeout of one request, from connecting to the end of the body. + pub timeout: Duration, + /// Requests of one `read_ranges` call in flight at once. + pub max_parallel: usize, + /// Bytes fetched by the first request, from offset 0 (the superblock and + /// usually the root group's metadata); at least 1. + pub first_request: u64, + /// When the server ignores `Range` (answers `200`), download the whole + /// file once and read it from memory, instead of failing. + pub allow_full_download: bool, + /// Largest file [`allow_full_download`](Self::allow_full_download) will + /// download. + pub max_full_download: u64, + /// Refuse a server that sends neither a strong `ETag` nor + /// `Last-Modified`, since a change of the file could then go unnoticed + /// (only its length is checked). + pub require_validator: bool, + /// Extra headers sent with every request (for example + /// `Authorization`). + pub headers: Vec<(String, String)>, +} + +impl Default for HttpOptions { + fn default() -> Self { + HttpOptions { + retries: 3, + backoff: Duration::from_millis(200), + timeout: Duration::from_secs(60), + max_parallel: 8, + first_request: crate::cache::DEFAULT_BLOCK_SIZE, + allow_full_download: false, + max_full_download: 1 << 30, + require_validator: false, + headers: Vec::new(), + } + } +} + +/// Requests and bytes an [`HttpStorage`] has used. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct HttpStats { + /// HTTP requests sent (retries included). + pub requests: u64, + /// Requests that were retries. + pub retries: u64, + /// Response body bytes received. + pub bytes: u64, +} + +/// How the file is pinned. +#[derive(Debug, Clone, PartialEq, Eq)] +enum Validator { + ETag(String), + LastModified(String), + None, +} + +/// An HTTP(S) file read by range requests. See the [module +/// documentation](self). +pub struct HttpStorage { + agent: ureq::Agent, + url: String, + len: u64, + validator: Validator, + options: HttpOptions, + /// The whole file, when the server ignores ranges and a full download + /// was allowed. + full: Option>, + requests: AtomicU64, + retries: AtomicU64, + bytes: AtomicU64, +} + +impl std::fmt::Debug for HttpStorage { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("HttpStorage") + .field("url", &self.url) + .field("len", &self.len) + .field("validator", &self.validator) + .field("full_download", &self.full.is_some()) + .finish() + } +} + +/// A parsed `Content-Range: bytes a-b/total`. +fn content_range(v: &str) -> Option<(u64, u64, Option)> { + let rest = v.trim().strip_prefix("bytes")?.trim_start(); + let (span, total) = rest.split_once('/')?; + let (a, b) = span.trim().split_once('-')?; + let a: u64 = a.trim().parse().ok()?; + let b: u64 = b.trim().parse().ok()?; + if b < a { + return None; + } + let total = match total.trim() { + "*" => None, + t => Some(t.parse().ok()?), + }; + Some((a, b, total)) +} + +fn header<'a>(resp: &'a ureq::http::Response, name: &str) -> Option<&'a str> { + resp.headers().get(name).and_then(|v| v.to_str().ok()) +} + +/// A body with a `Content-Encoding` is not a byte range of the file. +fn check_identity(url: &str, resp: &ureq::http::Response) -> Result<(), RemoteError> { + match header(resp, "content-encoding") { + Some(enc) if !enc.trim().eq_ignore_ascii_case("identity") => { + Err(RemoteError::Usage(format!( + "{url}: the server sent a {enc}-encoded body despite Accept-Encoding: identity" + ))) + } + _ => Ok(()), + } +} + +fn transport(e: ureq::Error) -> RemoteError { + match e { + ureq::Error::StatusCode(code) => RemoteError::Status { + code, + what: "request".into(), + }, + ureq::Error::BadUri(s) => RemoteError::InvalidUrl(s), + other => RemoteError::Transport(other.to_string()), + } +} + +impl HttpStorage { + /// Open `url` (`http://`, or `https://` with the `https` feature): + /// one ranged `GET` of the first [`HttpOptions::first_request`] bytes, + /// which gives the file's length and validators. Returns the storage + /// and the bytes that request fetched (the file's start), for a + /// [`BlockCache`](crate::BlockCache) to keep. + pub fn open(url: &str, options: HttpOptions) -> Result<(HttpStorage, Vec), RemoteError> { + let lower = url.to_ascii_lowercase(); + if lower.starts_with("https://") { + if !cfg!(feature = "https") { + return Err(RemoteError::UnsupportedScheme(format!( + "{url}: https:// needs the `https` feature of clawhdf5-remote" + ))); + } + } else if !lower.starts_with("http://") { + return Err(RemoteError::UnsupportedScheme(url.to_string())); + } + let config = ureq::Agent::config_builder() + .http_status_as_error(false) + .timeout_global(Some(options.timeout)) + .build(); + let mut storage = HttpStorage { + agent: ureq::Agent::new_with_config(config), + url: url.to_string(), + len: 0, + validator: Validator::None, + options, + full: None, + requests: AtomicU64::new(0), + retries: AtomicU64::new(0), + bytes: AtomicU64::new(0), + }; + let first = storage.with_retries(|| storage.probe())?; + let (len, validator, bytes, full) = first; + storage.len = len; + storage.validator = validator; + if full { + storage.full = Some(bytes); + return Ok((storage, Vec::new())); + } + if storage.options.require_validator && storage.validator == Validator::None { + return Err(RemoteError::Usage(format!( + "{url}: the server sends neither a strong ETag nor Last-Modified, so a change \ + of the file could not be detected (HttpOptions::require_validator)" + ))); + } + Ok((storage, bytes)) + } + + /// The URL. + pub fn url(&self) -> &str { + &self.url + } + + /// Whether the file was downloaded whole, because the server does not + /// support ranges and [`HttpOptions::allow_full_download`] was set. + pub fn is_full_download(&self) -> bool { + self.full.is_some() + } + + /// The `ETag` the file is pinned to, if the server sent a strong one. + pub fn etag(&self) -> Option<&str> { + match &self.validator { + Validator::ETag(e) => Some(e), + _ => None, + } + } + + /// Requests and bytes so far. + pub fn stats(&self) -> HttpStats { + HttpStats { + requests: self.requests.load(Ordering::Relaxed), + retries: self.retries.load(Ordering::Relaxed), + bytes: self.bytes.load(Ordering::Relaxed), + } + } + + fn with_retries( + &self, + mut attempt: impl FnMut() -> Result, + ) -> Result { + let mut delay = self.options.backoff; + let mut n = 0; + loop { + match attempt() { + Ok(v) => return Ok(v), + Err(e) if e.is_transient() && n < self.options.retries => { + n += 1; + self.retries.fetch_add(1, Ordering::Relaxed); + std::thread::sleep(delay); + delay = delay.saturating_mul(2); + } + Err(e) => return Err(e), + } + } + } + + fn request( + &self, + range: Option<(u64, u64)>, + ) -> ureq::RequestBuilder { + let mut req = self + .agent + .get(&self.url) + .header("Accept-Encoding", "identity"); + if let Some((a, b)) = range { + req = req.header("Range", format!("bytes={a}-{b}")); + } + match &self.validator { + Validator::ETag(e) => req = req.header("If-Match", e), + Validator::LastModified(t) => req = req.header("If-Unmodified-Since", t), + Validator::None => {} + } + for (k, v) in &self.options.headers { + req = req.header(k, v); + } + req + } + + /// Read a body of exactly `want` bytes (or up to `limit` when `want` is + /// unknown). + fn body( + &self, + resp: ureq::http::Response, + want: Option, + limit: u64, + ) -> Result, RemoteError> { + let cap = want.unwrap_or(limit); + let mut buf = Vec::with_capacity(usize::try_from(cap.min(64 << 20)).unwrap_or(0)); + let reader = resp.into_body().into_reader(); + let got = reader + .take(cap.saturating_add(1)) + .read_to_end(&mut buf) + .map_err(|e| RemoteError::Transport(format!("{}: reading the body: {e}", self.url))); + self.bytes.fetch_add(buf.len() as u64, Ordering::Relaxed); + got?; + match want { + Some(n) if buf.len() as u64 != n => Err(RemoteError::BadResponse(format!( + "{}: body of {} bytes, expected {n}", + self.url, + buf.len() + ))), + None if buf.len() as u64 > limit => Err(RemoteError::Usage(format!( + "{}: the file is larger than HttpOptions::max_full_download ({limit} bytes)", + self.url + ))), + _ => Ok(buf), + } + } + + /// The first request: length, validators and the file's first bytes. + /// The last field is true when the server ignored the range and sent + /// the whole file (only kept when a full download is allowed). + fn probe(&self) -> Result<(u64, Validator, Vec, bool), RemoteError> { + let n = self.options.first_request.max(1); + self.requests.fetch_add(1, Ordering::Relaxed); + let resp = self.request(Some((0, n - 1))).call().map_err(transport)?; + let status = resp.status().as_u16(); + check_identity(&self.url, &resp)?; + let validator = match (header(&resp, "etag"), header(&resp, "last-modified")) { + (Some(e), _) if !e.starts_with("W/") => Validator::ETag(e.to_string()), + (_, Some(t)) => Validator::LastModified(t.to_string()), + _ => Validator::None, + }; + match status { + 206 => { + let cr = header(&resp, "content-range").ok_or_else(|| { + RemoteError::BadResponse(format!("{}: 206 without Content-Range", self.url)) + })?; + let (a, b, total) = content_range(cr).ok_or_else(|| { + RemoteError::BadResponse(format!("{}: bad Content-Range {cr:?}", self.url)) + })?; + let total = total.ok_or_else(|| { + RemoteError::BadResponse(format!( + "{}: the server does not report the file's length (Content-Range {cr:?})", + self.url + )) + })?; + if a != 0 || b >= total || b > n - 1 { + return Err(RemoteError::BadResponse(format!( + "{}: asked for bytes 0-{}, got Content-Range {cr:?}", + self.url, + n - 1 + ))); + } + let bytes = self.body(resp, Some(b - a + 1), 0)?; + Ok((total, validator, bytes, false)) + } + 200 => { + if !self.options.allow_full_download { + return Err(RemoteError::RangeNotSupported(format!( + "{} answered a range request with the whole file (status 200); set \ + HttpOptions::allow_full_download to download it", + self.url + ))); + } + let want = header(&resp, "content-length").and_then(|v| v.trim().parse().ok()); + if want.is_some_and(|w: u64| w > self.options.max_full_download) { + return Err(RemoteError::Usage(format!( + "{}: the file is larger than HttpOptions::max_full_download ({} bytes)", + self.url, self.options.max_full_download + ))); + } + let bytes = self.body(resp, want, self.options.max_full_download)?; + Ok((bytes.len() as u64, validator, bytes, true)) + } + 416 => Err(RemoteError::Usage(format!( + "{}: status 416 for the first bytes (an empty file?)", + self.url + ))), + code => Err(RemoteError::Status { + code, + what: self.url.clone(), + }), + } + } + + /// One request for `[start, end)` (inside the file, non-empty). + fn fetch_once(&self, start: u64, end: u64) -> Result, RemoteError> { + self.requests.fetch_add(1, Ordering::Relaxed); + let resp = self + .request(Some((start, end - 1))) + .call() + .map_err(transport)?; + let changed = |why: String| RemoteError::FileChanged(format!("{}: {why}", self.url)); + check_identity(&self.url, &resp)?; + match resp.status().as_u16() { + 206 => {} + 200 => { + return Err(RemoteError::RangeNotSupported(format!( + "{} answered a range request with the whole file (status 200)", + self.url + ))); + } + 412 => { + return Err(changed( + "If-Match/If-Unmodified-Since failed (status 412)".into(), + )); + } + 416 => return Err(changed("range no longer satisfiable (status 416)".into())), + code => { + return Err(RemoteError::Status { + code, + what: format!("{} bytes {start}-{}", self.url, end - 1), + }); + } + } + match &self.validator { + Validator::ETag(e) => { + if let Some(got) = header(&resp, "etag") + && got != e + { + return Err(changed(format!("ETag {got} instead of {e}"))); + } + } + Validator::LastModified(t) => { + if let Some(got) = header(&resp, "last-modified") + && got != t + { + return Err(changed(format!("Last-Modified {got} instead of {t}"))); + } + } + Validator::None => {} + } + let cr = header(&resp, "content-range").ok_or_else(|| { + RemoteError::BadResponse(format!("{}: 206 without Content-Range", self.url)) + })?; + let (a, b, total) = content_range(cr).ok_or_else(|| { + RemoteError::BadResponse(format!("{}: bad Content-Range {cr:?}", self.url)) + })?; + if let Some(total) = total + && total != self.len + { + return Err(changed(format!("length {total} instead of {}", self.len))); + } + if a != start || b != end - 1 { + return Err(RemoteError::BadResponse(format!( + "{}: asked for bytes {start}-{}, got Content-Range {cr:?}", + self.url, + end - 1 + ))); + } + self.body(resp, Some(end - start), 0) + } + + fn fetch(&self, start: u64, end: u64) -> Result, RemoteError> { + self.with_retries(|| self.fetch_once(start, end)) + } + + /// `[offset, offset + len)` clamped to the file, or `None` if empty. + fn clamp(&self, offset: u64, len: u64) -> Option<(u64, u64)> { + let end = offset.saturating_add(len).min(self.len); + (offset < end).then_some((offset, end)) + } +} + +impl Storage for HttpStorage { + fn read_at(&self, offset: u64, len: usize) -> Result, FormatError> { + if let Some(all) = &self.full { + return all.as_slice().read_at(offset, len); + } + match self.clamp(offset, len as u64) { + None => Ok(Cow::Owned(Vec::new())), + Some((a, b)) => Ok(Cow::Owned(self.fetch(a, b)?)), + } + } + + fn len(&self) -> u64 { + self.len + } + + fn read_ranges(&self, ranges: &[Range]) -> Result>, FormatError> { + if let Some(all) = &self.full { + return all.as_slice().read_ranges(ranges); + } + let parallel = self.options.max_parallel.clamp(1, ranges.len().max(1)); + if parallel <= 1 { + return ranges + .iter() + .map(|r| self.read_at(r.start, (r.end.saturating_sub(r.start)) as usize)) + .collect(); + } + let next = AtomicUsize::new(0); + let failed = std::sync::atomic::AtomicBool::new(false); + type Slot = Option, RemoteError>>; + let results: std::sync::Mutex> = + std::sync::Mutex::new((0..ranges.len()).map(|_| None).collect()); + std::thread::scope(|s| { + for _ in 0..parallel { + s.spawn(|| { + loop { + let i = next.fetch_add(1, Ordering::Relaxed); + if i >= ranges.len() || failed.load(Ordering::Relaxed) { + break; + } + let r = &ranges[i]; + let got = match self.clamp(r.start, r.end.saturating_sub(r.start)) { + None => Ok(Vec::new()), + Some((a, b)) => self.fetch(a, b), + }; + if got.is_err() { + failed.store(true, Ordering::Relaxed); + } + results + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner)[i] = Some(got); + } + }); + } + }); + let results = results + .into_inner() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut out = Vec::with_capacity(ranges.len()); + for r in results { + match r { + Some(Ok(v)) => out.push(Cow::Owned(v)), + Some(Err(e)) => return Err(e.into()), + // Not fetched because another range failed first. + None => continue, + } + } + if out.len() != ranges.len() { + return Err(FormatError::Storage(format!( + "{}: a parallel range read failed", + self.url + ))); + } + Ok(out) + } + + fn as_contiguous(&self) -> Option<&[u8]> { + self.full.as_deref() + } +} + +#[cfg(test)] +mod tests { + use super::content_range; + + #[test] + fn content_range_parses() { + assert_eq!(content_range("bytes 0-99/1000"), Some((0, 99, Some(1000)))); + assert_eq!(content_range("bytes 5-5/*"), Some((5, 5, None))); + assert_eq!(content_range("bytes 9-5/10"), None); + assert_eq!(content_range("items 0-1/2"), None); + assert_eq!(content_range("bytes */1000"), None); + } +} diff --git a/crates/clawhdf5-remote/src/lib.rs b/crates/clawhdf5-remote/src/lib.rs new file mode 100644 index 0000000..597fb63 --- /dev/null +++ b/crates/clawhdf5-remote/src/lib.rs @@ -0,0 +1,127 @@ +//! Read HDF5 files where they are — on an HTTP(S) server or in an object +//! store — without downloading them first. +//! +//! This is milestone M3 of `docs/design/range-reads.md`: remote backends for +//! [`clawhdf5::File::open_storage`], each read through a mandatory +//! [`BlockCache`]. +//! +//! ```no_run +//! let file = clawhdf5_remote::open_url("http://127.0.0.1:8000/data.h5")?; +//! let temperature = file.dataset("/grid/temperature")?.read_f64()?; +//! # Ok::<(), Box>(()) +//! ``` +//! +//! - [`open_url`] / [`open_url_with`]: `http://` (default feature `http`), +//! `https://` (feature `https`) → a [`clawhdf5::File`] with the whole +//! read API. +//! - [`storage_for_url`] gives the cached storage itself, to open with +//! [`clawhdf5::File::open_storage`] and to read its [`CacheStats`]. +//! - [`HttpStorage`] (range `GET`s, pinned by ETag/Last-Modified, retried +//! with backoff), and [`BlockCache`] over any +//! [`Storage`](clawhdf5_format::storage::Storage). +//! +//! What costs what: opening costs one request (it also fetches the first +//! block, 1 MiB by default); listing a file whose metadata sits in its +//! first blocks costs nothing more; reading a chunked dataset costs one +//! parallel batch of requests for the blocks holding its chunk index, then +//! one for its chunks. The zero-copy methods of `clawhdf5` +//! (`read_raw_ref`, `read_*_zerocopy`, `File::as_bytes`) need the file in +//! memory and are errors (`as_bytes` a panic) on a remote file. + +#![warn(missing_docs)] + +pub mod cache; +pub mod error; +#[cfg(feature = "http")] +pub mod http; + +use std::sync::Arc; + +use clawhdf5::File; +use clawhdf5_format::storage::Storage; + +pub use cache::{BlockCache, CacheConfig, CacheStats}; +pub use error::{Error, RemoteError}; +#[cfg(feature = "http")] +pub use http::{HttpOptions, HttpStats, HttpStorage}; + +/// A backend a [`BlockCache`] can read through. +pub type Backend = Box; + +/// The storage [`storage_for_url`] returns: a block cache over the URL's +/// backend. +pub type RemoteStorage = BlockCache; + +/// Settings of [`open_url_with`] and [`storage_for_url`]. +#[derive(Debug, Clone, Default)] +pub struct Options { + /// The block cache. + pub cache: CacheConfig, + /// HTTP(S) requests. + #[cfg(feature = "http")] + pub http: HttpOptions, +} + +/// Open the HDF5 file at `url` with default [`Options`]. +/// +/// `http://…` needs the (default) `http` feature, `https://…` the `https` +/// feature. +pub fn open_url(url: &str) -> Result { + open_url_with(url, &Options::default()) +} + +/// [`open_url`] with explicit [`Options`]. +pub fn open_url_with(url: &str, options: &Options) -> Result { + let storage = storage_for_url(url, options)?; + Ok(File::open_storage(storage)?) +} + +/// The cached storage for `url`, with the first block already fetched: +/// open it with [`clawhdf5::File::open_storage`] (a clone of the `Arc`), +/// and read its [`BlockCache::stats`] as you go. +pub fn storage_for_url(url: &str, options: &Options) -> Result, Error> { + let scheme = url + .split_once("://") + .map(|(s, _)| s.to_ascii_lowercase()) + .ok_or_else(|| RemoteError::InvalidUrl(format!("{url}: no scheme")))?; + match scheme.as_str() { + "http" | "https" => http_storage(url, options), + "s3" | "s3a" | "gs" | "az" | "azure" | "abfs" | "abfss" | "adl" => { + cloud_storage(url, &scheme, options) + } + _ => Err(RemoteError::UnsupportedScheme(url.to_string()).into()), + } +} + +#[cfg(feature = "http")] +fn http_storage(url: &str, options: &Options) -> Result, Error> { + let mut http = options.http.clone(); + http.first_request = http.first_request.max(options.cache.block_size.max(1)); + let (storage, first) = HttpStorage::open(url, http)?; + let cache = BlockCache::new(Box::new(storage) as Backend, options.cache.clone()); + cache.insert(0, &first); + Ok(Arc::new(cache)) +} + +#[cfg(not(feature = "http"))] +fn http_storage(url: &str, _options: &Options) -> Result, Error> { + Err(RemoteError::UnsupportedScheme(format!( + "{url}: http(s):// needs the `http` feature of clawhdf5-remote" + )) + .into()) +} + +fn cloud_storage(url: &str, scheme: &str, _options: &Options) -> Result, Error> { + Err(RemoteError::UnsupportedScheme(format!("{url}: {scheme}:// is not supported yet")).into()) +} + +/// A [`BlockCache`] over `backend` with its first block fetched (readahead +/// of the superblock and the metadata usually written next to it). +pub fn cached(backend: Backend, options: &Options) -> Result { + let cache = BlockCache::new(backend, options.cache.clone()); + let first = cache.config().block_size; + cache + .prefetch(0, first) + .map_err(|e| Error::Hdf5(clawhdf5::Error::Format(e)))?; + Ok(cache) +} diff --git a/crates/clawhdf5-remote/tests/common/mod.rs b/crates/clawhdf5-remote/tests/common/mod.rs new file mode 100644 index 0000000..498b603 --- /dev/null +++ b/crates/clawhdf5-remote/tests/common/mod.rs @@ -0,0 +1,302 @@ +//! 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(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(r: &Result) -> String { + match r { + Ok(v) => digest(v), + Err(_) => "Err".into(), + } +} + +fn sorted(m: std::collections::HashMap) -> BTreeMap { + 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::())).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) { + 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 { + 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> { + 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 { + let mut b = clawhdf5::FileBuilder::new(); + b.set_attr("title", clawhdf5::AttrValue::String("remote test".into())); + let big: Vec = (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 = (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() +} diff --git a/crates/clawhdf5-remote/tests/common/server.rs b/crates/clawhdf5-remote/tests/common/server.rs new file mode 100644 index 0000000..e556d7f --- /dev/null +++ b/crates/clawhdf5-remote/tests/common/server.rs @@ -0,0 +1,331 @@ +//! A small HTTP/1.1 file server on 127.0.0.1 for tests and the example: +//! `Range: bytes=a-b` requests (206, 416), `HEAD`, keep-alive, strong ETags +//! and Last-Modified with `If-Match` / `If-Unmodified-Since` (412), and +//! switches to misbehave — ignore ranges (200 with the whole file), cut +//! bodies short, answer 503, respond slowly, send no validators. It counts +//! requests and body bytes, and logs every range asked for. + +#![allow(dead_code)] + +use std::collections::HashMap; +use std::io::{BufRead, BufReader, Write}; +use std::net::{TcpListener, TcpStream}; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, RwLock}; +use std::time::Duration; + +/// A logged GET: the path and the range asked for (`None`: whole file). +pub type LogEntry = (String, Option<(u64, u64)>); + +struct Resource { + data: Arc>, + etag: String, + last_modified: String, +} + +/// Switches and counters shared with the connection threads. +#[derive(Default)] +pub struct Shared { + files: RwLock>, + version: AtomicU64, + /// Answer every request with 200 and the whole file. + pub ignore_range: AtomicBool, + /// Send no ETag and no Last-Modified. + pub no_validators: AtomicBool, + /// Send a weak ETag only (no Last-Modified). + pub weak_etag: AtomicBool, + /// Send Last-Modified but no ETag. + pub no_etag: AtomicBool, + /// Label bodies `Content-Encoding: gzip` (they are not). + pub gzip_label: AtomicBool, + /// Cut the body of the next N ranged responses in half (then close the + /// connection). + pub truncate_next: AtomicU32, + /// Answer the next N requests with 503. + pub fail_next: AtomicU32, + /// Sleep this long before answering each request. + pub delay_ms: AtomicU64, + /// Requests served (every status). + pub requests: AtomicU64, + /// Body bytes sent. + pub bytes: AtomicU64, + /// (path, range) of every GET. + pub log: Mutex>, + stop: AtomicBool, +} + +/// A running server; stops accepting when dropped. +pub struct Server { + pub addr: std::net::SocketAddr, + pub shared: Arc, +} + +impl Server { + /// Serve `files` (URL path such as `/a.h5` → bytes) on `127.0.0.1`, on + /// a free port. + pub fn start(files: Vec<(String, Vec)>) -> Server { + Server::bind("127.0.0.1:0", files) + } + + /// Serve `files` on `addr`. + pub fn bind(addr: &str, files: Vec<(String, Vec)>) -> Server { + let listener = TcpListener::bind(addr).expect("bind"); + let addr = listener.local_addr().unwrap(); + let shared = Arc::new(Shared::default()); + for (path, data) in files { + shared.put(&path, data); + } + let s = shared.clone(); + std::thread::spawn(move || { + for conn in listener.incoming() { + if s.stop.load(Ordering::SeqCst) { + break; + } + let Ok(conn) = conn else { continue }; + let s = s.clone(); + std::thread::spawn(move || { + let _ = serve(conn, &s); + }); + } + }); + Server { addr, shared } + } + + /// The URL of `path` (which starts with `/`). + pub fn url(&self, path: &str) -> String { + format!("http://{}{path}", self.addr) + } + + /// Requests served so far. + pub fn requests(&self) -> u64 { + self.shared.requests.load(Ordering::SeqCst) + } + + /// Body bytes sent so far. + pub fn bytes(&self) -> u64 { + self.shared.bytes.load(Ordering::SeqCst) + } + + /// Zero the counters and the log. + pub fn reset(&self) { + self.shared.requests.store(0, Ordering::SeqCst); + self.shared.bytes.store(0, Ordering::SeqCst); + self.shared.log.lock().unwrap().clear(); + } + + /// The ranges asked for so far. + pub fn log(&self) -> Vec { + self.shared.log.lock().unwrap().clone() + } +} + +impl Drop for Server { + fn drop(&mut self) { + self.shared.stop.store(true, Ordering::SeqCst); + let _ = TcpStream::connect(self.addr); + } +} + +impl Shared { + /// Add or replace a file (a replacement gets a new ETag and + /// Last-Modified). + pub fn put(&self, path: &str, data: Vec) { + let v = self.version.fetch_add(1, Ordering::SeqCst) + 1; + let secs = 1_700_000_000 + v; + self.files.write().unwrap().insert( + path.to_string(), + Resource { + data: Arc::new(data), + etag: format!("\"v{v}-{path}\""), + last_modified: http_date(secs), + }, + ); + } +} + +/// An IMF-fixdate for `secs` since the epoch (enough for distinct values). +fn http_date(secs: u64) -> String { + const DAYS: [&str; 7] = ["Thu", "Fri", "Sat", "Sun", "Mon", "Tue", "Wed"]; + const MONTHS: [&str; 12] = [ + "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec", + ]; + let days = secs / 86_400; + let rem = secs % 86_400; + // Civil-from-days (Howard Hinnant). + let z = days as i64 + 719_468; + let era = z.div_euclid(146_097); + let doe = z - era * 146_097; + let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; + let y = yoe + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = doy - (153 * mp + 2) / 5 + 1; + let m = if mp < 10 { mp + 3 } else { mp - 9 }; + let y = if m <= 2 { y + 1 } else { y }; + format!( + "{}, {:02} {} {} {:02}:{:02}:{:02} GMT", + DAYS[(days % 7) as usize], + d, + MONTHS[(m - 1) as usize], + y, + rem / 3600, + rem % 3600 / 60, + rem % 60 + ) +} + +fn parse_range(v: &str, len: u64) -> Option> { + let spec = v.trim().strip_prefix("bytes=")?; + if spec.contains(',') { + return None; // multiple ranges: serve the whole file + } + let (a, b) = spec.split_once('-')?; + let a: u64 = a.trim().parse().ok()?; + let b: u64 = match b.trim() { + "" => u64::MAX, + b => b.parse().ok()?, + }; + if b < a { + return None; + } + if a >= len { + return Some(Err(())); + } + Some(Ok((a, b.min(len - 1)))) +} + +fn serve(conn: TcpStream, s: &Shared) -> std::io::Result<()> { + conn.set_read_timeout(Some(Duration::from_secs(30)))?; + // One write per response and no Nagle delay: otherwise every request + // waits for a delayed ACK (~40 ms). + conn.set_nodelay(true)?; + let mut reader = BufReader::new(conn.try_clone()?); + let mut out = conn; + loop { + let mut line = String::new(); + if reader.read_line(&mut line)? == 0 { + return Ok(()); + } + let mut parts = line.split_whitespace(); + let method = parts.next().unwrap_or("").to_string(); + let path = parts.next().unwrap_or("").to_string(); + let mut headers = HashMap::new(); + loop { + let mut h = String::new(); + if reader.read_line(&mut h)? == 0 { + return Ok(()); + } + let h = h.trim_end(); + if h.is_empty() { + break; + } + if let Some((k, v)) = h.split_once(':') { + headers.insert(k.trim().to_ascii_lowercase(), v.trim().to_string()); + } + } + s.requests.fetch_add(1, Ordering::SeqCst); + let delay = s.delay_ms.load(Ordering::SeqCst); + if delay > 0 { + std::thread::sleep(Duration::from_millis(delay)); + } + let close = headers + .get("connection") + .is_some_and(|v| v.eq_ignore_ascii_case("close")); + if s.fail_next + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1)) + .is_ok() + { + write!( + out, + "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\n\r\n" + )?; + continue; + } + let res = { + let files = s.files.read().unwrap(); + files + .get(&path) + .map(|r| (r.data.clone(), r.etag.clone(), r.last_modified.clone())) + }; + let Some((data, etag, lm)) = res else { + write!(out, "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n")?; + continue; + }; + let len = data.len() as u64; + let mut validators = String::new(); + if s.weak_etag.load(Ordering::SeqCst) { + validators = format!("ETag: W/{etag}\r\n"); + } else if s.no_etag.load(Ordering::SeqCst) { + validators = format!("Last-Modified: {lm}\r\n"); + } else if !s.no_validators.load(Ordering::SeqCst) { + validators = format!("ETag: {etag}\r\nLast-Modified: {lm}\r\n"); + } + let precondition_failed = headers.get("if-match").is_some_and(|v| v != &etag) + || headers.get("if-unmodified-since").is_some_and(|v| v != &lm); + if precondition_failed { + write!( + out, + "HTTP/1.1 412 Precondition Failed\r\n{validators}Content-Length: 0\r\n\r\n" + )?; + continue; + } + let range = if s.ignore_range.load(Ordering::SeqCst) { + None + } else { + headers.get("range").and_then(|v| parse_range(v, len)) + }; + if method == "GET" { + s.log + .lock() + .unwrap() + .push((path.clone(), range.and_then(Result::ok))); + } + let (status, body, extra) = match range { + Some(Err(())) => { + write!( + out, + "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Range: bytes */{len}\r\n\ + {validators}Content-Length: 0\r\n\r\n" + )?; + continue; + } + Some(Ok((a, b))) => ( + "206 Partial Content", + &data[a as usize..=b as usize], + format!("Content-Range: bytes {a}-{b}/{len}\r\n"), + ), + None => ("200 OK", &data[..], String::new()), + }; + let extra = if s.gzip_label.load(Ordering::SeqCst) { + format!("{extra}Content-Encoding: gzip\r\n") + } else { + extra + }; + let head = format!( + "HTTP/1.1 {status}\r\nContent-Length: {}\r\nAccept-Ranges: bytes\r\n{extra}{validators}\r\n", + body.len() + ); + if method == "HEAD" { + out.write_all(head.as_bytes())?; + continue; + } + let truncate = status.starts_with("206") + && s.truncate_next + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1)) + .is_ok(); + let body = if truncate { + &body[..body.len() / 2] + } else { + body + }; + let mut response = head.into_bytes(); + response.extend_from_slice(body); + out.write_all(&response)?; + out.flush()?; + s.bytes.fetch_add(body.len() as u64, Ordering::SeqCst); + if truncate || close { + let _ = out.shutdown(std::net::Shutdown::Both); + return Ok(()); + } + } +} diff --git a/crates/clawhdf5-remote/tests/http.rs b/crates/clawhdf5-remote/tests/http.rs new file mode 100644 index 0000000..e7faed8 --- /dev/null +++ b/crates/clawhdf5-remote/tests/http.rs @@ -0,0 +1,517 @@ +//! HTTP range reads against a local server (no internet): every value read +//! over HTTP equals `File::open`'s, and the server's misbehaviour — no +//! range support, a file replaced mid-read, cut-off bodies, errors, slow +//! answers under concurrent readers — is an error or the right data, never +//! wrong data. +//! +//! `CLAWHDF5_REMOTE_CORPUS=dir[:dir...]` adds every HDF5 file under those +//! directories (the conformance corpus is `conformance/.cache/corpus`), and +//! `CLAWHDF5_REMOTE_REPORT=1` prints the per-file request counts. + +#![cfg(feature = "http")] + +mod common; + +use std::sync::Arc; +use std::sync::atomic::Ordering; +use std::time::Duration; + +use clawhdf5::File; +use clawhdf5_remote::{ + CacheConfig, Error, HttpOptions, HttpStorage, Options, RemoteError, open_url, open_url_with, + storage_for_url, +}; +use common::server::Server; +use common::{list_and_read_one, multi_block_file, transcript}; + +/// Options for tests: fast retries. +fn quick() -> Options { + Options { + http: HttpOptions { + backoff: Duration::from_millis(5), + ..HttpOptions::default() + }, + ..Options::default() + } +} + +fn changed(e: &str) -> bool { + e.contains("changed while open") +} + +/// Serve `files`, open each by URL and through `File::open`, and require +/// identical transcripts. Returns (files compared, files both refused). +fn compare(files: &[std::path::PathBuf], report: bool) -> (usize, usize) { + let served: Vec<(String, Vec)> = files + .iter() + .enumerate() + .filter_map(|(i, p)| { + let bytes = std::fs::read(p).ok()?; + let name = p.file_name()?.to_str()?.replace(' ', "_"); + Some((format!("/f{i}/{name}"), bytes)) + }) + .collect(); + let server = Server::start(served.clone()); + let (mut same, mut refused) = (0, 0); + let mut totals = [0u64; 5]; + for (i, p) in files.iter().enumerate() { + let Some((url_path, bytes)) = served + .iter() + .find(|(u, _)| u.starts_with(&format!("/f{i}/"))) + else { + continue; + }; + let url = server.url(url_path); + let local = File::open(p); + let remote = storage_for_url(&url, &quick()) + .map_err(|e| e.to_string()) + .and_then(|s| { + File::open_storage(s.clone()) + .map(|mut f| { + f.set_vds_resolver(common::sibling_resolver(p.parent().unwrap().into())); + (f, s) + }) + .map_err(|e| e.to_string()) + }); + let (local, (remote, storage)) = match (local, remote) { + (Ok(l), Ok(r)) => (l, r), + (Err(l), Err(r)) => { + assert!( + !r.contains("HTTP") && !r.contains("network"), + "{url}: {r} ({l})" + ); + refused += 1; + continue; + } + (l, r) => panic!( + "{}: File::open {:?}, open_url {:?}", + p.display(), + l.map(|_| ()), + r.map(|_| ()) + ), + }; + 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 remote: {g}")) + .unwrap_or_default(); + panic!("{}: open_url differs from File::open{first}", p.display()); + } + assert!(storage.stats().reads > 0, "read through the cache"); + same += 1; + + // Cost of "list + read one dataset": with the block cache, and + // with none (every read a request). + let with = storage_for_url(&url, &quick()).unwrap(); + server.reset(); + if let Ok(f) = File::open_storage(with.clone()) { + list_and_read_one(&f); + } + let (cached_requests, cached_bytes) = (server.requests(), server.bytes()); + let (bare, _) = HttpStorage::open(&url, quick().http).unwrap(); + let bare = Arc::new(bare); + server.reset(); + if let Ok(f) = File::open_storage(bare.clone()) { + list_and_read_one(&f); + } + let uncached_requests = server.requests(); + totals[0] += 1 + cached_requests; // + the open probe + totals[1] += cached_bytes + with.config().block_size.min(bytes.len() as u64); + totals[2] += 1 + uncached_requests; + totals[3] += bytes.len() as u64; + totals[4] += 1; + if report { + eprintln!( + "requests cached {:>6} uncached {:>8} bytes {:>12} of {:>12} {}", + 1 + cached_requests, + 1 + uncached_requests, + cached_bytes + with.config().block_size.min(bytes.len() as u64), + bytes.len(), + p.display() + ); + } + // Files within one block: opening fetched everything. + if bytes.len() as u64 <= with.config().block_size { + assert_eq!(cached_requests, 0, "{}", p.display()); + } + } + eprintln!( + "list + read one dataset over {} files: {} requests with the 1 MiB block cache \ + ({} bytes transferred, files {} bytes), {} requests without a cache", + totals[4], totals[0], totals[1], totals[3], totals[2] + ); + (same, refused) +} + +#[test] +fn fixtures_read_identically_over_http() { + let files = common::fixtures(); + assert!(files.len() >= 45, "{} fixtures", files.len()); + let report = std::env::var("CLAWHDF5_REMOTE_REPORT").is_ok_and(|v| v == "1"); + let (same, _) = compare(&files, report); + assert!(same >= 40, "{same}"); +} + +#[test] +fn corpus_reads_identically_over_http() { + let Some(files) = common::corpus() else { + eprintln!("CLAWHDF5_REMOTE_CORPUS not set; skipping the corpus"); + return; + }; + let report = std::env::var("CLAWHDF5_REMOTE_REPORT").is_ok_and(|v| v == "1"); + let (same, refused) = compare(&files, report); + eprintln!("corpus: {same} files identical, {refused} refused by both"); + assert!(same > 0); +} + +#[test] +fn multi_block_file_values_and_request_budget() { + let bytes = multi_block_file(); + assert!(bytes.len() > 3 << 20, "{}", bytes.len()); + let server = Server::start(vec![("/m.h5".into(), bytes.clone())]); + let url = server.url("/m.h5"); + let local = File::from_bytes(bytes.clone()).unwrap(); + let storage = storage_for_url(&url, &quick()).unwrap(); + assert_eq!(server.requests(), 1, "opening is one request"); + let remote = File::open_storage(storage.clone()).unwrap(); + assert!(remote.contiguous_bytes().is_none()); + let mut names = remote.root().datasets().unwrap(); + names.sort(); + assert_eq!(names, ["big", "flat"]); + assert_eq!(remote.root().groups().unwrap(), ["grp"]); + assert_eq!( + remote.dataset("grp/small").unwrap().read_f64().unwrap(), + [1.0, 2.0, 3.0] + ); + let big = remote.dataset("big").unwrap().read_f64().unwrap(); + assert_eq!(big, local.dataset("big").unwrap().read_f64().unwrap()); + let flat = remote.dataset("flat").unwrap().read_f64().unwrap(); + assert_eq!(flat.len(), 300_000); + assert_eq!(flat[299_999], 299_999.0); + // Every byte was fetched at most once, in whole 1 MiB blocks. + let log = server.log(); + let mut blocks: Vec = Vec::new(); + for (_, r) in &log { + let (a, b) = r.expect("every request is ranged"); + assert_eq!(a % (1 << 20), 0, "block-aligned"); + blocks.extend(a >> 20..=b >> 20); + } + let n = blocks.len(); + blocks.sort_unstable(); + blocks.dedup(); + assert_eq!(n, blocks.len(), "a block was fetched twice: {log:?}"); + assert!( + server.requests() <= (bytes.len() as u64 >> 20) + 2, + "{} requests for a {} byte file", + server.requests(), + bytes.len() + ); + // Reading again costs nothing. + let before = server.requests(); + assert_eq!(transcript(&remote), transcript(&local)); + assert_eq!(server.requests(), before); +} + +#[test] +fn h5py_written_file_over_http() { + if !common::have_h5py() { + return; + } + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("py.h5"); + common::run_python( + r#" +import sys, h5py, numpy as np +with h5py.File(sys.argv[1], "w", libver="latest") as f: + f.attrs["title"] = "h5py remote test" + g = f.create_group("sensors") + for i in range(40): + d = g.create_dataset(f"s{i:02d}", data=np.arange(1000, dtype="().unwrap()); + assert_eq!(vl[1], it.next().unwrap()); + assert_eq!( + remote.dataset("link").unwrap().read_i32().unwrap(), + local.dataset("sensors/s03").unwrap().read_i32().unwrap() + ); +} + +#[test] +fn a_server_that_ignores_range_is_refused_or_downloaded_when_allowed() { + let bytes = multi_block_file(); + let server = Server::start(vec![("/m.h5".into(), bytes.clone())]); + server.shared.ignore_range.store(true, Ordering::SeqCst); + let url = server.url("/m.h5"); + let err = open_url_with(&url, &quick()).unwrap_err(); + assert!( + matches!(err, Error::Remote(RemoteError::RangeNotSupported(_))), + "{err}" + ); + assert!( + server.bytes() < bytes.len() as u64, + "refusing must not download the file ({} bytes read)", + server.bytes() + ); + let mut opts = quick(); + opts.http.allow_full_download = true; + let storage = storage_for_url(&url, &opts).unwrap(); + let f = File::open_storage(storage.clone()).unwrap(); + assert_eq!(f.contiguous_bytes(), Some(&bytes[..]), "read from memory"); + assert_eq!(server.requests(), 2); + let local = File::from_bytes(bytes).unwrap(); + assert_eq!(transcript(&f), transcript(&local)); + assert_eq!(server.requests(), 2, "no further requests"); + // Too large for the download limit. + opts.http.max_full_download = 1000; + assert!(open_url_with(&url, &opts).is_err()); +} + +#[test] +fn a_file_replaced_mid_read_is_an_error_not_mixed_data() { + for mode in ["etag", "last-modified", "none"] { + let bytes = multi_block_file(); + let server = Server::start(vec![("/m.h5".into(), bytes.clone())]); + match mode { + "last-modified" => server.shared.no_etag.store(true, Ordering::SeqCst), + "none" => server.shared.no_validators.store(true, Ordering::SeqCst), + _ => {} + } + let url = server.url("/m.h5"); + let f = open_url_with(&url, &quick()).unwrap(); + assert_eq!(f.root().groups().unwrap(), ["grp"], "{mode}"); + // Replace the file: same layout, other values (and, for "none", + // one byte longer, which is all that can be checked). + let mut other = bytes.clone(); + let n = other.len(); + for b in &mut other[n / 2..n / 2 + 1000] { + *b ^= 0xff; + } + if mode == "none" { + other.push(0); + } + server.shared.put("/m.h5", other); + let err = f + .dataset("big") + .unwrap() + .read_f64() + .unwrap_err() + .to_string(); + assert!(changed(&err), "{mode}: {err}"); + } +} + +#[test] +fn a_weak_etag_or_no_validator_is_refused_when_required() { + let server = Server::start(vec![("/m.h5".into(), multi_block_file())]); + server.shared.weak_etag.store(true, Ordering::SeqCst); + let mut opts = quick(); + opts.http.require_validator = true; + assert!(open_url_with(&server.url("/m.h5"), &opts).is_err()); + opts.http.require_validator = false; + assert!(open_url_with(&server.url("/m.h5"), &opts).is_ok()); +} + +#[test] +fn truncated_bodies_are_retried_then_an_error() { + let bytes = multi_block_file(); + let server = Server::start(vec![("/m.h5".into(), bytes.clone())]); + let url = server.url("/m.h5"); + let local = File::from_bytes(bytes).unwrap(); + let want = local.dataset("big").unwrap().read_f64().unwrap(); + + // One cut-off body: retried, right values. + let (http, first) = HttpStorage::open(&url, quick().http).unwrap(); + assert_eq!(first.len(), 1 << 20); + let storage = Arc::new(clawhdf5_remote::BlockCache::new( + http, + CacheConfig::default(), + )); + let f = File::open_storage(storage.clone()).unwrap(); + server.shared.truncate_next.store(1, Ordering::SeqCst); + assert_eq!(f.dataset("big").unwrap().read_f64().unwrap(), want); + assert_eq!(storage.inner().stats().retries, 1); + + // Every body cut off: an error, never partial data. + let f = open_url_with(&url, &quick()).unwrap(); + server.shared.truncate_next.store(1000, Ordering::SeqCst); + let err = f + .dataset("big") + .unwrap() + .read_f64() + .unwrap_err() + .to_string(); + assert!( + err.contains("network error") || err.contains("bad response"), + "{err}" + ); + server.shared.truncate_next.store(0, Ordering::SeqCst); + // The failure was not cached: the next read succeeds. + assert_eq!(f.dataset("big").unwrap().read_f64().unwrap(), want); +} + +#[test] +fn server_errors_are_retried_with_backoff() { + let bytes = multi_block_file(); + let server = Server::start(vec![("/m.h5".into(), bytes.clone())]); + server.shared.fail_next.store(2, Ordering::SeqCst); + let url = server.url("/m.h5"); + let f = open_url_with(&url, &quick()).unwrap(); + let local = File::from_bytes(bytes).unwrap(); + server.shared.fail_next.store(3, Ordering::SeqCst); + assert_eq!( + f.dataset("flat").unwrap().read_f64().unwrap(), + local.dataset("flat").unwrap().read_f64().unwrap() + ); + // More failures than retries: an error. + let f = open_url_with(&url, &quick()).unwrap(); + server.shared.fail_next.store(100, Ordering::SeqCst); + let err = f + .dataset("big") + .unwrap() + .read_f64() + .unwrap_err() + .to_string(); + assert!(err.contains("503"), "{err}"); +} + +#[test] +fn slow_server_concurrent_readers_fetch_each_block_once() { + let bytes = multi_block_file(); + let server = Server::start(vec![("/m.h5".into(), bytes.clone())]); + let url = server.url("/m.h5"); + let opts = Options { + cache: CacheConfig { + block_size: 256 << 10, + max_request: 256 << 10, + ..CacheConfig::default() + }, + ..quick() + }; + let storage = storage_for_url(&url, &opts).unwrap(); + let file = File::open_storage(storage.clone()).unwrap(); + let local = File::from_bytes(bytes).unwrap(); + let want_big = local.dataset("big").unwrap().read_f64().unwrap(); + let want_flat = local.dataset("flat").unwrap().read_f64().unwrap(); + server.shared.delay_ms.store(40, Ordering::SeqCst); + server.reset(); + std::thread::scope(|s| { + for t in 0..8 { + let (file, want_big, want_flat) = (&file, &want_big, &want_flat); + s.spawn(move || { + if t % 2 == 0 { + assert_eq!(&file.dataset("big").unwrap().read_f64().unwrap(), want_big); + } else { + assert_eq!( + &file.dataset("flat").unwrap().read_f64().unwrap(), + want_flat + ); + } + }); + } + }); + let log = server.log(); + let mut starts: Vec = log.iter().map(|(_, r)| r.unwrap().0).collect(); + let n = starts.len(); + starts.sort_unstable(); + starts.dedup(); + assert_eq!(n, starts.len(), "a block was fetched twice"); + assert!(storage.stats().waits > 0, "readers shared fetches"); +} + +#[test] +fn bad_urls_and_statuses_are_clean_errors() { + let server = Server::start(vec![("/m.h5".into(), multi_block_file())]); + let err = open_url_with(&server.url("/missing.h5"), &quick()).unwrap_err(); + assert!( + matches!(err, Error::Remote(RemoteError::Status { code: 404, .. })), + "{err}" + ); + assert!(matches!( + open_url("ftp://example.com/a.h5"), + Err(Error::Remote(RemoteError::UnsupportedScheme(_))) + )); + assert!(matches!( + open_url("no-scheme"), + Err(Error::Remote(RemoteError::InvalidUrl(_))) + )); + assert!(matches!( + open_url("s3://bucket/key.h5"), + Err(Error::Remote(RemoteError::UnsupportedScheme(_))) + )); + #[cfg(not(feature = "https"))] + { + let e = open_url("https://example.com/a.h5") + .unwrap_err() + .to_string(); + assert!(e.contains("`https` feature"), "{e}"); + } + // A content-encoded body is not a byte range. + let server = Server::start(vec![("/m.h5".into(), multi_block_file())]); + server.shared.gzip_label.store(true, Ordering::SeqCst); + let e = open_url(&server.url("/m.h5")).unwrap_err().to_string(); + assert!(e.contains("gzip-encoded"), "{e}"); + // Not HDF5. + let server = Server::start(vec![("/x.h5".into(), vec![7u8; 5000])]); + assert!(matches!( + open_url(&server.url("/x.h5")), + Err(Error::Hdf5(_)) + )); + // A server that closes every connection unanswered: a network error + // after the retries. (Not a closed port: another test's server could + // take it meanwhile.) + let port = { + let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = l.local_addr().unwrap().port(); + std::thread::spawn(move || { + for c in l.incoming() { + drop(c); + } + }); + port + }; + let err = open_url_with(&format!("http://127.0.0.1:{port}/a.h5"), &quick()).unwrap_err(); + assert!( + matches!(err, Error::Remote(RemoteError::Transport(_))), + "{err}" + ); +} diff --git a/scripts/ci-test.sh b/scripts/ci-test.sh index ad9af8b..2fce9c1 100755 --- a/scripts/ci-test.sh +++ b/scripts/ci-test.sh @@ -109,23 +109,36 @@ run_step "cargo clippy (fast-deflate / zlib-ng)" cargo clippy \ --features clawhdf5-format/fast-deflate,clawhdf5-filters/fast-deflate \ -- -D warnings +# clawhdf5-remote's HTTPS backend (rustls). +run_step "cargo clippy (remote, https)" cargo clippy \ + -p clawhdf5-remote \ + --all-targets \ + --features https \ + -- -D warnings + # The README promises that the core crates build no C by default. Hold it to # that: fail if a crate that compiles C (a *-sys crate, cc or cmake) enters the # default dependency tree of any of them. clawhdf5-migrate (bundled SQLite), # clawhdf5-napi (Node) and clawhdf5-gpu (graphics drivers) are exempt. # js-sys (clawhdf5-wasm's bindings to JavaScript) builds no C. +# clawhdf5-remote is checked by default (plain HTTP); its https feature +# (ring) builds C and is opt-in. no_c_in_default_build() { - local crate found=0 - for crate in clawhdf5-format clawhdf5-io clawhdf5-filters clawhdf5 \ + local entry crate features found=0 + for entry in clawhdf5-format clawhdf5-io clawhdf5-filters clawhdf5 \ clawhdf5-agent clawhdf5-ann clawhdf5-accel clawhdf5-netcdf4 clawhdf5-cli \ clawhdf5-tools \ - clawhdf5-wasm; do + clawhdf5-wasm \ + clawhdf5-remote; do + crate=${entry%%:*} + features=() + [ "$entry" != "$crate" ] && features=(--features "${entry#*:}") local c_deps - c_deps=$(cargo tree -q -p "$crate" -e normal,build --prefix none \ + c_deps=$(cargo tree -q -p "$crate" "${features[@]}" -e normal,build --prefix none \ | grep -E '^([a-z0-9_-]+-sys|cc|cmake) v' \ | grep -v '^js-sys v' | sort -u) if [ -n "$c_deps" ]; then - echo "$crate pulls in C by default:" + echo "$entry pulls in C by default:" echo "$c_deps" | sed 's/^/ /' found=1 fi