diff --git a/crates/clawhdf5-remote/src/cache.rs b/crates/clawhdf5-remote/src/cache.rs index 233938c..a434814 100644 --- a/crates/clawhdf5-remote/src/cache.rs +++ b/crates/clawhdf5-remote/src/cache.rs @@ -98,7 +98,7 @@ type Block = Arc<[u8]>; /// A fetch in progress: the readers waiting for a block wait on this. struct Flight { - result: Mutex>>, + result: Mutex>>, done: Condvar, } @@ -110,7 +110,7 @@ impl Flight { }) } - fn finish(&self, r: Result) { + fn finish(&self, r: Result) { let mut slot = lock(&self.result); if slot.is_none() { *slot = Some(r); @@ -118,7 +118,7 @@ impl Flight { self.done.notify_all(); } - fn wait(&self) -> Result { + fn wait(&self) -> Result { let mut slot = lock(&self.result); loop { if let Some(r) = slot.as_ref() { @@ -173,10 +173,20 @@ pub struct BlockCache { } /// Fails every flight a fetch claimed and did not complete (an error or a -/// panic in the backend), so no reader waits forever. +/// panic in the backend), so no reader waits forever. The waiting readers +/// get the fetch's error. struct FlightGuard<'a, S> { cache: &'a BlockCache, flights: Vec<(u64, Arc)>, + error: Option, +} + +impl FlightGuard<'_, S> { + /// Record `e` for the readers waiting on this fetch; returns it. + fn fail(&mut self, e: FormatError) -> FormatError { + self.error = Some(e.clone()); + e + } } impl Drop for FlightGuard<'_, S> { @@ -191,8 +201,12 @@ impl Drop for FlightGuard<'_, S> { } } drop(st); + let error = self + .error + .take() + .unwrap_or_else(|| FormatError::Storage("the fetch of this block failed".into())); for (_, f) in &self.flights { - f.finish(Err("the fetch of this block failed".into())); + f.finish(Err(error.clone())); } } } @@ -348,6 +362,7 @@ impl BlockCache { let mut guard = FlightGuard { cache: self, flights: Vec::new(), + error: None, }; { let mut st = lock(&self.state); @@ -410,13 +425,16 @@ impl BlockCache { self.counters .requests .fetch_add(runs.len() as u64, Ordering::Relaxed); - let fetched = self.inner.read_ranges(&runs)?; + let fetched = match self.inner.read_ranges(&runs) { + Ok(f) => f, + Err(e) => return Err(guard.fail(e)), + }; if fetched.len() != runs.len() { - return Err(FormatError::Storage(format!( + return Err(guard.fail(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) { @@ -424,12 +442,12 @@ impl BlockCache { .bytes_fetched .fetch_add(bytes.len() as u64, Ordering::Relaxed); if bytes.len() as u64 != run.end - run.start { - return Err(FormatError::Storage(format!( + return Err(guard.fail(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; @@ -476,12 +494,7 @@ impl BlockCache { } for (i, f) in waits { - match f.wait() { - Ok(data) => { - have.insert(i, data); - } - Err(e) => return Err(FormatError::Storage(e)), - } + have.insert(i, f.wait()?); } Ok(have) } @@ -786,6 +799,34 @@ mod tests { } } + /// A slow backend whose fetches all fail with a telling error. + struct SlowFailing; + + impl Storage for SlowFailing { + fn read_at(&self, _: u64, _: usize) -> Result, FormatError> { + std::thread::sleep(std::time::Duration::from_millis(50)); + Err(FormatError::Storage("the file changed while open".into())) + } + fn len(&self) -> u64 { + 1 << 20 + } + } + + #[test] + fn readers_waiting_on_a_failed_fetch_get_its_error() { + let c = BlockCache::new(SlowFailing, small()); + std::thread::scope(|s| { + let hs: Vec<_> = (0..4) + .map(|_| s.spawn(|| c.read_at(0, 10).map(|b| b.len()))) + .collect(); + for h in hs { + let e = h.join().unwrap().unwrap_err().to_string(); + assert!(e.contains("changed while open"), "{e}"); + } + }); + assert!(c.stats().waits > 0); + } + #[test] fn concurrent_readers_never_fetch_a_block_twice() { let d = data(64 * 1024); diff --git a/crates/clawhdf5-remote/tests/http.rs b/crates/clawhdf5-remote/tests/http.rs index b0a0e34..5a68072 100644 --- a/crates/clawhdf5-remote/tests/http.rs +++ b/crates/clawhdf5-remote/tests/http.rs @@ -301,11 +301,7 @@ fn a_server_that_ignores_range_is_refused_or_downloaded_when_allowed() { matches!(err, Error::Remote(RemoteError::RangeNotSupported(_))), "{err}" ); - assert!( - server.bytes() < bytes.len() as u64, - "refusing must not download the file ({} bytes read)", - server.bytes() - ); + assert_eq!(server.requests(), 1, "refused at the first response"); let mut opts = quick(); opts.http.allow_full_download = true; let storage = storage_for_url(&url, &opts).unwrap();