clawhdf5-remote, h5rs: never allocate a length the server only claims
h5rs check URL read the whole file with one read_at(0, len), len being whatever Content-Range said. BlockCache listed every block index of the span and preallocated len bytes: a server claiming 2^62 bytes for a 10 KB file made h5rs abort (memory allocation of 35184372088832 bytes failed). - BlockCache: a read spanning more than the budget (or eight max_requests) is fetched piece by piece and not kept, its output growing only as data arrives; read_ranges falls back to that per range; prefetch is clamped to the budget. - New clawhdf5_remote::download(storage, max_bytes): refuses a claimed length above the limit (RemoteError::TooLarge) before any request, then reads in 64 MiB steps. New RemoteError::Backend for read errors. - h5rs check downloads through it, with --max-download N (default 1 GiB). Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
@@ -25,7 +25,10 @@
|
||||
//! 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).
|
||||
//! (a big chunked read would otherwise evict all the metadata). A read
|
||||
//! spanning more than the budget (or eight `max_request`s, if more) is
|
||||
//! fetched piece by piece, its output growing only as data arrives:
|
||||
//! nothing is allocated for a length a server merely claims.
|
||||
//! - **Local storages pass through.** A backend that holds the whole file
|
||||
//! in memory ([`Storage::as_contiguous`]) is read directly.
|
||||
|
||||
@@ -285,12 +288,46 @@ impl<S: Storage> BlockCache<S> {
|
||||
if self.inner.as_contiguous().is_some() {
|
||||
return Ok(());
|
||||
}
|
||||
let Some(blocks) = self.block_span(offset, len) else {
|
||||
// Readahead beyond the budget would only evict itself.
|
||||
let Some(blocks) = self.block_span(offset, len.min(self.config.capacity)) else {
|
||||
return Ok(());
|
||||
};
|
||||
self.blocks(&blocks.collect::<Vec<_>>(), true).map(|_| ())
|
||||
}
|
||||
|
||||
/// Blocks fetched per step of a read spanning more than this: the
|
||||
/// larger of the budget and a full parallel batch of requests (eight
|
||||
/// `max_request`s), 64 MiB by default.
|
||||
fn piece_blocks(&self) -> u64 {
|
||||
let bytes = self
|
||||
.config
|
||||
.capacity
|
||||
.max(self.config.max_request.saturating_mul(8));
|
||||
(bytes / self.config.block_size).max(1)
|
||||
}
|
||||
|
||||
/// A read of more blocks than [`piece_blocks`](Self::piece_blocks):
|
||||
/// fetched and copied one piece at a time, nothing kept. The block list
|
||||
/// is never materialised and the output grows only as data arrives, so
|
||||
/// a length a server merely claims (up to `u64::MAX`) costs nothing
|
||||
/// until bytes actually come back.
|
||||
fn read_large(&self, offset: u64, end: u64, span: Range<u64>) -> Result<Vec<u8>, FormatError> {
|
||||
let bs = self.config.block_size;
|
||||
let piece = self.piece_blocks();
|
||||
let mut out = Vec::new();
|
||||
let mut first = span.start;
|
||||
while first < span.end {
|
||||
let last = first.saturating_add(piece).min(span.end);
|
||||
let wanted: Vec<u64> = (first..last).collect();
|
||||
let blocks = self.blocks(&wanted, false)?;
|
||||
let a = offset.max(first.saturating_mul(bs));
|
||||
let b = end.min(last.saturating_mul(bs));
|
||||
self.assemble_into(&mut out, a, b, &blocks);
|
||||
first = last;
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// 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
|
||||
@@ -534,8 +571,20 @@ impl<S: Storage> BlockCache<S> {
|
||||
|
||||
/// Copy `[offset, end)` out of `blocks`.
|
||||
fn assemble(&self, offset: u64, end: u64, blocks: &HashMap<u64, Block>) -> Vec<u8> {
|
||||
let mut out = Vec::with_capacity(usize::try_from(end - offset).unwrap_or(0));
|
||||
self.assemble_into(&mut out, offset, end, blocks);
|
||||
out
|
||||
}
|
||||
|
||||
/// Append `[offset, end)` out of `blocks` to `out`.
|
||||
fn assemble_into(
|
||||
&self,
|
||||
out: &mut Vec<u8>,
|
||||
offset: u64,
|
||||
end: u64,
|
||||
blocks: &HashMap<u64, Block>,
|
||||
) {
|
||||
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;
|
||||
@@ -545,7 +594,6 @@ impl<S: Storage> BlockCache<S> {
|
||||
out.extend_from_slice(&block[from..to]);
|
||||
pos = i * bs + to as u64;
|
||||
}
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
@@ -559,6 +607,9 @@ impl<S: Storage> Storage for BlockCache<S> {
|
||||
return Ok(Cow::Owned(Vec::new()));
|
||||
};
|
||||
let end = offset.saturating_add(len as u64).min(self.len);
|
||||
if span.end - span.start > self.piece_blocks() {
|
||||
return Ok(Cow::Owned(self.read_large(offset, end, span)?));
|
||||
}
|
||||
let blocks = self.blocks(&span.collect::<Vec<_>>(), true)?;
|
||||
Ok(Cow::Owned(self.assemble(offset, end, &blocks)))
|
||||
}
|
||||
@@ -574,7 +625,8 @@ impl<S: Storage> Storage for BlockCache<S> {
|
||||
self.counters
|
||||
.reads
|
||||
.fetch_add(ranges.len() as u64, Ordering::Relaxed);
|
||||
let mut wanted = BTreeSet::new();
|
||||
let mut spans = Vec::with_capacity(ranges.len());
|
||||
let mut total = 0u64;
|
||||
for r in ranges {
|
||||
if r.end < r.start {
|
||||
return Err(FormatError::Storage(
|
||||
@@ -582,9 +634,30 @@ impl<S: Storage> Storage for BlockCache<S> {
|
||||
));
|
||||
}
|
||||
if let Some(span) = self.block_span(r.start, r.end - r.start) {
|
||||
wanted.extend(span);
|
||||
total = total.saturating_add(span.end - span.start);
|
||||
spans.push(span);
|
||||
}
|
||||
}
|
||||
if total > self.piece_blocks() {
|
||||
// Too many blocks for one batch: range by range, each in pieces.
|
||||
return ranges
|
||||
.iter()
|
||||
.map(|r| {
|
||||
let end = r.end.min(self.len);
|
||||
match self.block_span(r.start, r.end - r.start) {
|
||||
None => Ok(Cow::Owned(Vec::new())),
|
||||
Some(span) if span.end - span.start > self.piece_blocks() => {
|
||||
Ok(Cow::Owned(self.read_large(r.start, end, span)?))
|
||||
}
|
||||
Some(span) => {
|
||||
let blocks = self.blocks(&span.collect::<Vec<_>>(), true)?;
|
||||
Ok(Cow::Owned(self.assemble(r.start, end, &blocks)))
|
||||
}
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
}
|
||||
let wanted: BTreeSet<u64> = spans.into_iter().flatten().collect();
|
||||
let wanted: Vec<u64> = wanted.into_iter().collect();
|
||||
let blocks = self.blocks(&wanted, true)?;
|
||||
Ok(ranges
|
||||
@@ -917,4 +990,34 @@ mod tests {
|
||||
c.prefetch(len - 1, 10).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
/// A backend claiming 2^62 bytes that holds only a few: a read of
|
||||
/// "the whole file" fails at the first short piece instead of
|
||||
/// allocating (or listing the blocks of) the claimed length.
|
||||
struct Claims(Vec<u8>);
|
||||
|
||||
impl Storage for Claims {
|
||||
fn read_at(&self, offset: u64, len: usize) -> Result<Cow<'_, [u8]>, FormatError> {
|
||||
self.0
|
||||
.as_slice()
|
||||
.read_at(offset, len)
|
||||
.map(|c| Cow::Owned(c.into_owned()))
|
||||
}
|
||||
fn len(&self) -> u64 {
|
||||
1 << 62
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_huge_claimed_length_is_not_allocated() {
|
||||
let c = BlockCache::new(Claims(data(5000)), CacheConfig::default());
|
||||
let e = c.read_at(0, usize::MAX).unwrap_err().to_string();
|
||||
assert!(e.contains("short read"), "{e}");
|
||||
let e = c
|
||||
.read_ranges(&[0..u64::MAX, 10..20])
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(e.contains("short read"), "{e}");
|
||||
c.prefetch(0, u64::MAX).unwrap_err();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -38,9 +38,19 @@ pub enum RemoteError {
|
||||
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).
|
||||
/// Called in a way the backend cannot serve.
|
||||
Usage(String),
|
||||
/// The file is larger than a download was allowed to be
|
||||
/// ([`download`](crate::download)).
|
||||
TooLarge {
|
||||
/// The file's length, as the server reports it.
|
||||
len: u64,
|
||||
/// The limit.
|
||||
limit: u64,
|
||||
},
|
||||
/// A read through a backend failed; its error, as text (the form a
|
||||
/// [`clawhdf5::File`] read reports it in).
|
||||
Backend(String),
|
||||
}
|
||||
|
||||
impl RemoteError {
|
||||
@@ -71,6 +81,11 @@ impl std::fmt::Display for RemoteError {
|
||||
RemoteError::Transport(s) => write!(f, "network error: {s}"),
|
||||
RemoteError::ObjectStore(s) => write!(f, "object store: {s}"),
|
||||
RemoteError::Usage(s) => write!(f, "{s}"),
|
||||
RemoteError::TooLarge { len, limit } => write!(
|
||||
f,
|
||||
"the remote file is {len} bytes, more than the download limit of {limit} bytes"
|
||||
),
|
||||
RemoteError::Backend(s) => write!(f, "{s}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
//! `gcs`, `azure`) → 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`].
|
||||
//! - [`download`] reads a whole remote file into memory, up to a limit.
|
||||
//! - [`HttpStorage`] (range `GET`s, pinned by ETag/Last-Modified, retried
|
||||
//! with backoff), [`ObjectStoreStorage`] (any `object_store` store,
|
||||
//! feature `object-store`), and [`BlockCache`] over any
|
||||
@@ -151,6 +152,47 @@ pub fn cached(backend: Backend, options: &Options) -> Result<RemoteStorage, Erro
|
||||
Ok(cache)
|
||||
}
|
||||
|
||||
/// Default limit of [`download`]: 1 GiB.
|
||||
pub const DEFAULT_MAX_DOWNLOAD: u64 = 1 << 30;
|
||||
|
||||
/// The whole file behind `storage`, read into memory — at most
|
||||
/// `max_bytes` of it (for example [`DEFAULT_MAX_DOWNLOAD`]).
|
||||
///
|
||||
/// The length is only what the server claims, so it is never used to
|
||||
/// allocate: a file longer than `max_bytes` is refused with
|
||||
/// [`RemoteError::TooLarge`] before anything is read, and the buffer grows
|
||||
/// only as bytes arrive (64 MiB per step, fetched as parallel requests by a
|
||||
/// [`BlockCache`]). A read that comes back short is an error.
|
||||
pub fn download(storage: &dyn Storage, max_bytes: u64) -> Result<Vec<u8>, Error> {
|
||||
let len = storage.len();
|
||||
if len > max_bytes {
|
||||
return Err(RemoteError::TooLarge {
|
||||
len,
|
||||
limit: max_bytes,
|
||||
}
|
||||
.into());
|
||||
}
|
||||
const STEP: u64 = 64 << 20;
|
||||
let mut out = Vec::new();
|
||||
let mut pos = 0u64;
|
||||
while pos < len {
|
||||
let want = (len - pos).min(STEP);
|
||||
let got = storage
|
||||
.read_at(pos, want as usize)
|
||||
.map_err(|e| RemoteError::Backend(e.to_string()))?;
|
||||
if got.len() as u64 != want {
|
||||
return Err(RemoteError::BadResponse(format!(
|
||||
"{} bytes at offset {pos} instead of {want}",
|
||||
got.len()
|
||||
))
|
||||
.into());
|
||||
}
|
||||
out.extend_from_slice(&got);
|
||||
pos += want;
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// Open the object at `path` of any `object_store` store (in memory, local
|
||||
/// files, or a cloud store you configured) through a block cache.
|
||||
#[cfg(feature = "object-store")]
|
||||
|
||||
Reference in New Issue
Block a user