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) <[email protected]>
This commit is contained in:
osobh
@@ -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<Option<Result<Block, String>>>,
|
||||
done: Condvar,
|
||||
}
|
||||
|
||||
impl Flight {
|
||||
fn new() -> Arc<Flight> {
|
||||
Arc::new(Flight {
|
||||
result: Mutex::new(None),
|
||||
done: Condvar::new(),
|
||||
})
|
||||
}
|
||||
|
||||
fn finish(&self, r: Result<Block, String>) {
|
||||
let mut slot = lock(&self.result);
|
||||
if slot.is_none() {
|
||||
*slot = Some(r);
|
||||
}
|
||||
self.done.notify_all();
|
||||
}
|
||||
|
||||
fn wait(&self) -> Result<Block, String> {
|
||||
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<Flight>),
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct State {
|
||||
blocks: HashMap<u64, Slot>,
|
||||
/// tick -> block index, oldest first.
|
||||
lru: BTreeMap<u64, u64>,
|
||||
tick: u64,
|
||||
bytes: u64,
|
||||
}
|
||||
|
||||
fn lock<T>(m: &Mutex<T>) -> 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<S> {
|
||||
inner: S,
|
||||
config: CacheConfig,
|
||||
len: u64,
|
||||
state: Mutex<State>,
|
||||
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<S>,
|
||||
flights: Vec<(u64, Arc<Flight>)>,
|
||||
}
|
||||
|
||||
impl<S> 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<S: Storage> BlockCache<S> {
|
||||
/// 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::<Vec<_>>(), 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<Range<u64>> {
|
||||
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<HashMap<u64, Block>, 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<u64, Block> = 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<Flight>)]) -> Vec<Range<u64>> {
|
||||
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<u64, Block>) -> Vec<u8> {
|
||||
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<S: Storage> Storage for BlockCache<S> {
|
||||
fn read_at(&self, offset: u64, len: usize) -> Result<Cow<'_, [u8]>, 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::<Vec<_>>(), true)?;
|
||||
Ok(Cow::Owned(self.assemble(offset, end, &blocks)))
|
||||
}
|
||||
|
||||
fn len(&self) -> u64 {
|
||||
self.len
|
||||
}
|
||||
|
||||
fn read_ranges(&self, ranges: &[Range<u64>]) -> Result<Vec<Cow<'_, [u8]>>, 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<u64> = 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<u8> {
|
||||
(0..n).map(|i| (i * 7 + i / 251) as u8).collect()
|
||||
}
|
||||
|
||||
fn cache(n: usize, config: CacheConfig) -> (Vec<u8>, BlockCache<CountingStorage>) {
|
||||
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<u8>,
|
||||
fail: AtomicU64,
|
||||
}
|
||||
|
||||
impl Storage for Flaky {
|
||||
fn read_at(&self, offset: u64, len: usize) -> Result<Cow<'_, [u8]>, 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<u8>);
|
||||
|
||||
impl Storage for Short {
|
||||
fn read_at(&self, offset: u64, len: usize) -> Result<Cow<'_, [u8]>, 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<u8>,
|
||||
fetched: Mutex<Vec<Range<u64>>>,
|
||||
}
|
||||
|
||||
impl Storage for Slow {
|
||||
fn read_at(&self, offset: u64, len: usize) -> Result<Cow<'_, [u8]>, 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<u64> = 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");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user