diff --git a/Cargo.lock b/Cargo.lock index 8678104f5d4..d84446150d6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8138,6 +8138,7 @@ dependencies = [ "snafu 0.8.6", "store-api", "tokio", + "tokio-util", "uuid", ] @@ -15489,6 +15490,7 @@ dependencies = [ "futures-core", "futures-io", "futures-sink", + "futures-util", "pin-project-lite", "slab", "tokio", diff --git a/src/log-store/Cargo.toml b/src/log-store/Cargo.toml index 7e4da1557e1..5870110be67 100644 --- a/src/log-store/Cargo.toml +++ b/src/log-store/Cargo.toml @@ -46,6 +46,7 @@ serde_json.workspace = true snafu.workspace = true store-api.workspace = true tokio.workspace = true +tokio-util = { workspace = true, features = ["rt"] } [dev-dependencies] common-meta = { workspace = true, features = ["testing"] } diff --git a/src/log-store/src/error.rs b/src/log-store/src/error.rs index 85000dbea3c..64d918d5d78 100644 --- a/src/log-store/src/error.rs +++ b/src/log-store/src/error.rs @@ -427,6 +427,15 @@ pub enum Error { location: Location, }, + #[snafu(display("Failed to fetch a segment of WAL object, path: {}", path))] + WalSegmentFetch { + path: String, + #[snafu(source)] + error: tokio::task::JoinError, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Invalid WAL object, path: {}", path))] InvalidWalObject { path: String, @@ -509,7 +518,8 @@ impl ErrorExt for Error { | WaitProduceResultReceiver { .. } | WaitDumpIndex { .. } | MetaLengthExceededLimit { .. } - | ObjectStoreWalStopped { .. } => StatusCode::Internal, + | ObjectStoreWalStopped { .. } + | WalSegmentFetch { .. } => StatusCode::Internal, CorruptedWalObject { .. } | WalObjectConflict { .. } diff --git a/src/log-store/src/object_store_wal/store.rs b/src/log-store/src/object_store_wal/store.rs index 98b0c8cad6b..1d75739350a 100644 --- a/src/log-store/src/object_store_wal/store.rs +++ b/src/log-store/src/object_store_wal/store.rs @@ -35,15 +35,16 @@ use store_api::logstore::entry::{Entry, NaiveEntry}; use store_api::logstore::provider::{ObjectStoreProvider, Provider}; use store_api::logstore::{AppendBatchResponse, EntryId, LogStore, SendableEntryStream, WalIndex}; use store_api::storage::RegionId; -use tokio::sync::{mpsc, oneshot, watch}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc, oneshot, watch}; use tokio::time::MissedTickBehavior; +use tokio_util::task::AbortOnDropHandle; use crate::error::{ CorruptedWalObjectSnafu, Error, IncompleteWalEntrySnafu, InvalidProviderSnafu, InvalidWalObjectSnafu, InvalidWalObjectStoreSnafu, MismatchedWalPrefixSnafu, MismatchedWalRegionSnafu, ObjectStoreWalSnafu, ObjectStoreWalStoppedSnafu, Result, StaleWalObjectSnafu, UnconfirmedWalEpochStartSnafu, WalObjectSequenceExhaustedSnafu, - WalObjectSequenceUnsettledSnafu, + WalObjectSequenceUnsettledSnafu, WalSegmentFetchSnafu, }; use crate::metrics::{ METRIC_OBJECT_STORE_WAL_APPEND_ACK_SECONDS, METRIC_OBJECT_STORE_WAL_CREATE_CONFLICTS_TOTAL, @@ -81,8 +82,31 @@ const CREATE_RETRY_DELAY: Duration = Duration::from_millis(100); /// append from its channel only while two more fit: a stalled object store /// holds callers back instead of growing the backlog. const MAX_SEALED_BATCHES: usize = 2 * MAX_IN_FLIGHT_CREATES; -/// Number of objects whose footers recovery fetches at a time. -const RECOVERY_CONCURRENCY: usize = 8; +/// Number of objects whose footers recovery fetches at a time. On S3 an +/// object costs two reads, its header and its tail, tens of milliseconds per +/// lane however many vCPUs the node has, and recovery time scales with objects +/// divided by concurrency up to 128 lanes without throttling: at 64 about +/// 10,000 objects recover in about 10 s instead of about 100 s at 8, while the +/// reads use about a third of S3's request rate per prefix, leaving room for +/// the data file reads of start-up. Each lane holds one +/// [`RECOVERY_TAIL_WINDOW`], 4 MiB in all. +const RECOVERY_CONCURRENCY: usize = 64; +/// Number of segments a read fetches ahead of the one whose entries it +/// yields. On S3 a replay that fetched each segment after the previous one +/// spent about 30 ms per object; with eight fetches in flight a segment +/// arrives about every 4 ms. The segments fetched ahead also take their bytes +/// from [`READ_PREFETCH_BYTES`]. +const READ_PREFETCH_DEPTH: usize = 8; +/// Bytes of decoded segments that all reads of a store together fetch ahead; a +/// segment counts as its length, which bounds its payloads, plus a [`Record`] +/// per entry. A read always fetches the segment it yields next, as a read that +/// fetched one segment at a time did, and a later segment only while it fits +/// in what is left, so regions replayed in parallel hold one segment each, up +/// to about `max_batch_bytes` or more for one oversized append, plus at most +/// these bytes, however many regions open at once. While a fetch copies the +/// payloads out of the downloaded segment it briefly holds the segment's +/// length once more. +const READ_PREFETCH_BYTES: usize = 64 * 1024 * 1024; /// Bytes recovery reads from the end of an object in one request. The window /// holds the trailer and the footer of an object with up to 1364 regions of /// 48 bytes each, so a second request for the footer is rare. @@ -113,6 +137,8 @@ pub struct ObjectStoreLogStore { wal_holes: WalHoles, /// Objects with a delete in flight, which a read consults. deleting: Arc, + /// Bytes the reads may fetch ahead, see [`READ_PREFETCH_BYTES`]. + read_prefetch_budget: Arc, terminal_error: TerminalError, stopped: Arc, command_tx: mpsc::Sender, @@ -393,6 +419,7 @@ impl ObjectStoreLogStore { obsolete_entry_ids, wal_holes, deleting, + read_prefetch_budget: Arc::new(Semaphore::new(READ_PREFETCH_BYTES)), terminal_error, stopped, command_tx, @@ -721,7 +748,9 @@ impl LogStore for ObjectStoreLogStore { /// An object collected after the read listed it is skipped. Otherwise a /// segment that does not decode fails the read or is skipped and recorded /// as a hole, as `on_corrupted_segment` says; a failed fetch always fails - /// the read. + /// the read. Up to [`READ_PREFETCH_DEPTH`] segments are fetched ahead + /// while [`READ_PREFETCH_BYTES`] allows, and each outcome is handled in + /// sequence order. async fn read( &self, provider: &Provider, @@ -760,10 +789,39 @@ impl LogStore for ObjectStoreLogStore { let provider = provider.clone(); let on_corrupted_segment = self.on_corrupted_segment; let wal_holes = self.wal_holes.clone(); + let prefetch_budget = self.read_prefetch_budget.clone(); Ok(Box::pin(try_stream! { let started_at = Instant::now(); - for (object_seq, footer_entry) in objects { - let records = match fetch_segment(io.as_ref(), object_seq, &footer_entry).await { + let mut objects = objects.into_iter().peekable(); + let mut fetches = VecDeque::new(); + loop { + // The next segment is always fetched, a later one only with + // its bytes reserved from the budget. + while fetches.len() < READ_PREFETCH_DEPTH { + let Some((_, entry)) = objects.peek() else { + break; + }; + let permit = if fetches.is_empty() { + None + } else { + match try_reserve_prefetch_bytes(&prefetch_budget, entry) { + Some(permit) => Some(permit), + None => break, + } + }; + let Some((object_seq, footer_entry)) = objects.next() else { + break; + }; + let fetch = + spawn_fetch_segment(io.clone(), object_seq, footer_entry.clone(), permit); + fetches.push_back((object_seq, footer_entry, fetch)); + } + let Some((object_seq, footer_entry, fetch)) = fetches.pop_front() else { + break; + }; + // The permit is held until the entries of the segment are yielded. + let (fetched, _permit) = fetch.await; + let records = match fetched { Ok(records) => records, Err(error) if is_collected(&catalog, &deleting, object_seq).await => { debug!( @@ -2135,6 +2193,47 @@ async fn fetch_segment( }) } +/// Takes from `budget` the bytes the segment `entry` describes holds once +/// decoded, if they are free: its length plus a [`Record`] per entry. +fn try_reserve_prefetch_bytes( + budget: &Arc, + entry: &FooterEntry, +) -> Option { + let records = u64::from(entry.entry_count) * std::mem::size_of::() as u64; + let bytes = u32::try_from(entry.segment_len.checked_add(records)?).ok()?; + budget.clone().try_acquire_many_owned(bytes).ok() +} + +/// Runs [`fetch_segment`] on a task of its own, so the fetch proceeds while +/// the read is busy with earlier segments, and returns the outcome with +/// `permit`. The task holds the permit, so the reserved bytes are released +/// only with the buffers of the fetch. Dropping the returned future, as a +/// dropped read does, aborts the task; a task that panics fails the fetch with +/// [`Error::WalSegmentFetch`]. +fn spawn_fetch_segment( + io: Arc, + object_seq: u64, + entry: FooterEntry, + permit: Option, +) -> impl Future>, Option)> + Send + 'static { + let task_io = io.clone(); + let task = AbortOnDropHandle::new(common_runtime::spawn_global(async move { + let fetched = fetch_segment(task_io.as_ref(), object_seq, &entry).await; + (fetched, permit) + })); + async move { + match task.await { + Ok(fetched) => fetched, + Err(error) => ( + Err(error).context(WalSegmentFetchSnafu { + path: io.object_path(object_seq), + }), + None, + ), + } + } +} + /// Records the segment `entry` describes in the object `object_seq` as a hole /// of its region, once per object, counts the skip and warns about it. An /// object collected meanwhile is skipped without a hole: the catalog is held @@ -3547,6 +3646,7 @@ mod tests { obsolete_entry_ids: ObsoleteEntryIds::default(), wal_holes: WalHoles::default(), deleting: Arc::default(), + read_prefetch_budget: Arc::new(Semaphore::new(READ_PREFETCH_BYTES)), terminal_error: Arc::default(), stopped: Arc::new(AtomicBool::new(false)), command_tx, @@ -3629,7 +3729,8 @@ mod tests { async fn test_store_recovery_bounds_concurrency_and_orders_by_sequence() { let _serialized = SERIALIZED.lock().await; let object_store = memory_store(); - populate(&object_store, 16, 3).await; + let depth = RECOVERY_CONCURRENCY as u64; + populate(&object_store, 2 * RECOVERY_CONCURRENCY, 3).await; let (io, mut parked) = ParkedIo::over(object_store); let objects = io.list().await.unwrap(); @@ -3650,12 +3751,12 @@ mod tests { wave.push(timeout(WAIT, parked.recv()).await.unwrap().unwrap()); } assert_eq!( - (0..8).collect::>(), + (0..depth).collect::>(), wave.iter() .map(|(object_seq, _)| *object_seq) .collect::>() ); - // No ninth fetch can start while eight are parked. + // No further fetch can start while a full wave is parked. for _ in 0..16 { tokio::task::yield_now().await; } @@ -3663,7 +3764,7 @@ mod tests { // Complete the initial wave in reverse order, parking each replacement. let mut replacements = Vec::new(); - for (expected_next, (_, release)) in (8..16).zip(wave.into_iter().rev()) { + for (expected_next, (_, release)) in (depth..2 * depth).zip(wave.into_iter().rev()) { release.send(true).unwrap(); let (object_seq, release) = timeout(WAIT, parked.recv()).await.unwrap().unwrap(); assert_eq!(expected_next, object_seq); @@ -3674,17 +3775,21 @@ mod tests { release.send(true).unwrap(); } assert_eq!( - (0..16).collect::>(), + (0..2 * depth).collect::>(), timeout(WAIT, &mut fetch).await.unwrap().unwrap() ); - assert_eq!(8, io.max_in_flight.load(Ordering::SeqCst)); + assert_eq!( + RECOVERY_CONCURRENCY, + io.max_in_flight.load(Ordering::SeqCst) + ); } #[tokio::test] async fn test_store_recovery_abandons_pending_fetches_on_failure() { let _serialized = SERIALIZED.lock().await; let object_store = memory_store(); - populate(&object_store, 16, 3).await; + let objects = 2 * RECOVERY_CONCURRENCY as u64; + populate(&object_store, objects as usize, 3).await; let (io, mut parked) = ParkedIo::over(object_store.clone()); let fetch = { let io = io.clone(); @@ -3709,9 +3814,13 @@ mod tests { } assert!(parked.try_recv().is_err()); let store = open(object_store, &eager()).await; - assert_eq!(id(15, 1), latest(&store, region(1))); - // Sequence 16 holds the object that started the epoch of the store. - assert_eq!(17, store.catalog.read().unwrap().next_object_seq().unwrap()); + assert_eq!(id(objects - 1, 1), latest(&store, region(1))); + // The sequence after the populated objects holds the object that + // started the epoch of the store. + assert_eq!( + objects + 1, + store.catalog.read().unwrap().next_object_seq().unwrap() + ); store.stop().await.unwrap(); } @@ -3846,7 +3955,10 @@ mod tests { }) .collect::>(); assert_eq!(expected, entries); - assert_eq!(expected_ranges[®ion(number)], *reads.lock().unwrap()); + // Segments are fetched concurrently, so in no particular order. + let mut reads = reads.lock().unwrap().clone(); + reads.sort_unstable(); + assert_eq!(expected_ranges[®ion(number)], reads); } reads.lock().unwrap().clear(); assert_eq!( @@ -3950,6 +4062,18 @@ mod tests { store.stop().await.unwrap(); } + /// The range reads of object `object_seq` among `reads`, in the order + /// they were issued. + fn range_reads_of(reads: &RangeReads, object_seq: u64) -> Vec<(u64, u64, u64)> { + reads + .lock() + .unwrap() + .iter() + .filter(|read| read.0 == object_seq) + .copied() + .collect() + } + /// Writes object 0 with one entry of region 1 and one of region 2 and /// object 1 with an entry of region 2, then flips a byte in the segment of /// region 2 in object 0. Returns the key of object 0 and its footer. @@ -4018,8 +4142,7 @@ mod tests { read_entries(&store, region(2), 0).await ); let corrupt = (0, footer[1].segment_offset, footer[1].segment_len); - assert_eq!(corrupt, reads.lock().unwrap()[0]); - assert_eq!(corrupt, reads.lock().unwrap()[1]); + assert_eq!(vec![corrupt, corrupt], range_reads_of(&reads, 0)); let hole = WalHole { path, object_seq: 0, @@ -4087,7 +4210,7 @@ mod tests { &format!("segment of region {} checksum mismatch", region(2)), ); let corrupt = (0, footer[1].segment_offset, footer[1].segment_len); - assert_eq!(vec![corrupt, corrupt], *reads.lock().unwrap()); + assert_eq!(vec![corrupt, corrupt], range_reads_of(&reads, 0)); assert!(store.wal_holes(&provider(region(2))).unwrap().is_empty()); assert_eq!( skipped, @@ -4158,6 +4281,379 @@ mod tests { } } + /// The entries of region 1 that [`put_object_per_entry`] wrote to the + /// objects `object_seqs`. + fn object_entries(object_seqs: impl IntoIterator) -> Vec { + object_seqs + .into_iter() + .map(|object_seq| { + let entry_id = object_seq + 1; + Entry::Naive(NaiveEntry { + provider: provider(region(1)), + region_id: region(1), + entry_id, + data: format!("e{entry_id}").into_bytes(), + }) + }) + .collect() + } + + /// Writes objects 0 to `objects - 1`, object `n` with entry `n + 1` of + /// region 1. + async fn put_object_per_entry(object_store: &ObjectStore, objects: u64) { + for object_seq in 0..objects { + put_object(object_store, object_seq, region(1), &[object_seq + 1]).await; + } + } + + /// Opens a store over [`put_object_per_entry`] objects, releasing every + /// read of recovery; range reads park from then on. + async fn open_parking_range_reads( + objects: u64, + ) -> ( + Arc, + Arc, + mpsc::UnboundedReceiver<(u64, oneshot::Sender)>, + ) { + let object_store = memory_store(); + put_object_per_entry(&object_store, objects).await; + let (io, mut parked) = ParkedIo::over(object_store); + let mut opening = { + let io = io.clone(); + tokio::spawn(async move { + ObjectStoreLogStore::open(io, &eager(), PREFIX.to_string()).await + }) + }; + let store = loop { + tokio::select! { + opened = &mut opening => break opened.unwrap().unwrap(), + Some((_, release)) = parked.recv() => release.send(true).unwrap(), + } + }; + io.range_reads_parked.store(true, Ordering::SeqCst); + io.max_in_flight.store(0, Ordering::SeqCst); + (store, io, parked) + } + + /// Polls until `settled` holds. + async fn wait_until(settled: impl Fn() -> bool) { + let deadline = Instant::now() + WAIT; + while !settled() { + assert!(Instant::now() < deadline, "the condition never held"); + tokio::time::sleep(Duration::from_millis(1)).await; + } + } + + async fn assert_nothing_parks( + parked: &mut mpsc::UnboundedReceiver<(u64, oneshot::Sender)>, + ) { + assert!( + timeout(Duration::from_millis(100), parked.recv()) + .await + .is_err() + ); + } + + async fn next_batch(stream: &mut SendableEntryStream<'static, Entry, Error>) -> Vec { + timeout(WAIT, stream.next()) + .await + .unwrap() + .unwrap() + .unwrap() + } + + #[tokio::test] + async fn test_store_read_prefetches_up_to_the_depth_in_order() { + let _serialized = SERIALIZED.lock().await; + let depth = READ_PREFETCH_DEPTH as u64; + let objects = depth + 4; + let (store, io, mut parked) = open_parking_range_reads(objects).await; + let reads = METRIC_OBJECT_STORE_WAL_READ_SECONDS.get_sample_count(); + let mut stream = store.read(&provider(region(1)), 0, None).await.unwrap(); + + // The first poll starts the fetches of the first `depth` objects. + assert!(stream.next().now_or_never().is_none()); + let mut wave = parked_operations(&mut parked, READ_PREFETCH_DEPTH).await; + assert_eq!( + (0..depth).collect::>(), + wave.keys().copied().collect() + ); + assert_nothing_parks(&mut parked).await; + + // Later objects complete first: they wait for object 0 and still + // count toward the depth. + for object_seq in (1..depth).rev() { + wave.remove(&object_seq).unwrap().send(true).unwrap(); + } + wait_until(|| io.in_flight.load(Ordering::SeqCst) == 1).await; + assert!(stream.next().now_or_never().is_none()); + assert_nothing_parks(&mut parked).await; + wave.remove(&0).unwrap().send(true).unwrap(); + assert_eq!(object_entries([0]), next_batch(&mut stream).await); + + // No fetch starts while the caller holds the entries of object 0; + // each later poll starts the fetch of the next object. + assert_nothing_parks(&mut parked).await; + let rest = tokio::spawn(stream.try_collect::>()); + let refills = parked_operations(&mut parked, (objects - depth) as usize).await; + assert_eq!( + (depth..objects).collect::>(), + refills.keys().copied().collect() + ); + for release in refills.into_values() { + release.send(true).unwrap(); + } + let rest = timeout(WAIT, rest).await.unwrap().unwrap().unwrap(); + assert_eq!( + object_entries(1..objects), + rest.into_iter().flatten().collect::>() + ); + assert_eq!(READ_PREFETCH_DEPTH, io.max_in_flight.load(Ordering::SeqCst)); + assert_eq!( + reads + 1, + METRIC_OBJECT_STORE_WAL_READ_SECONDS.get_sample_count() + ); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_read_fails_at_a_prefetched_object_in_order() { + let _serialized = SERIALIZED.lock().await; + let depth = READ_PREFETCH_DEPTH as u64; + let objects = depth + 4; + let failed = 5; + let (store, io, mut parked) = open_parking_range_reads(objects).await; + let reads = METRIC_OBJECT_STORE_WAL_READ_SECONDS.get_sample_count(); + let mut stream = store.read(&provider(region(1)), 0, None).await.unwrap(); + assert!(stream.next().now_or_never().is_none()); + let mut wave = parked_operations(&mut parked, READ_PREFETCH_DEPTH).await; + + // The fetch of object 5 fails before any earlier fetch completes. + wave.remove(&failed).unwrap().send(false).unwrap(); + wait_until(|| io.in_flight.load(Ordering::SeqCst) == READ_PREFETCH_DEPTH - 1).await; + for release in wave.into_values() { + release.send(true).unwrap(); + } + let mut refills = Vec::new(); + for object_seq in 0..failed { + assert_eq!(object_entries([object_seq]), next_batch(&mut stream).await); + if object_seq > 0 { + refills.push(next_parked_operation(&mut parked).await); + } + } + let error = timeout(WAIT, stream.next()) + .await + .unwrap() + .unwrap() + .unwrap_err(); + assert!( + matches!(&error, Error::WalObjectStore { operation: "read", path, .. } + if *path == io.object_path(failed)), + "unexpected error: {error:?}" + ); + assert!(timeout(WAIT, stream.next()).await.unwrap().is_none()); + + // The fetches started meanwhile are abandoned. + assert_eq!( + (depth..objects).collect::>(), + refills + .iter() + .map(|(object_seq, _)| *object_seq) + .collect::>() + ); + for (_, mut release) in refills { + timeout(WAIT, release.closed()).await.unwrap(); + } + assert_nothing_parks(&mut parked).await; + assert_eq!( + reads, + METRIC_OBJECT_STORE_WAL_READ_SECONDS.get_sample_count() + ); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_read_skips_prefetched_collected_and_corrupted_objects() { + let _serialized = SERIALIZED.lock().await; + let objects = READ_PREFETCH_DEPTH as u64 + 4; + let (corrupted, collected) = (3, 9); + let object_store = memory_store(); + put_object_per_entry(&object_store, objects).await; + let path = object_path(&object_store, corrupted); + let bytes = object_store.read(&path).await.unwrap().to_vec(); + let segment = footer_of(&bytes).1.remove(0); + corrupt_object(&object_store, &path, |bytes| { + bytes[(segment.segment_offset + segment.segment_len - 1) as usize] ^= 1; + }) + .await; + let store = open(object_store.clone(), &eager()).await; + assert_eq!(CorruptedSegmentAction::Skip, store.on_corrupted_segment); + let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get(); + + // A collection removes object 9 after the read listed it. + let stream = store.read(&provider(region(1)), 0, None).await.unwrap(); + store.catalog.write().unwrap().remove_object(collected); + object_store + .delete(&object_path(&object_store, collected)) + .await + .unwrap(); + let entries = timeout(WAIT, stream.try_collect::>()) + .await + .unwrap() + .unwrap() + .into_iter() + .flatten() + .collect::>(); + assert_eq!( + object_entries((0..objects).filter(|seq| ![corrupted, collected].contains(seq))), + entries + ); + assert_eq!( + vec![WalHole { + path, + object_seq: corrupted, + min_entry_id: corrupted + 1, + max_entry_id: corrupted + 1, + }], + store.wal_holes(&provider(region(1))).unwrap() + ); + assert_eq!( + skipped + 1, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() + ); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_dropped_read_abandons_its_prefetches() { + let _serialized = SERIALIZED.lock().await; + let objects = READ_PREFETCH_DEPTH as u64 + 4; + let (store, _io, mut parked) = open_parking_range_reads(objects).await; + let reads = METRIC_OBJECT_STORE_WAL_READ_SECONDS.get_sample_count(); + let mut stream = store.read(&provider(region(1)), 0, None).await.unwrap(); + assert!(stream.next().now_or_never().is_none()); + let mut wave = parked_operations(&mut parked, READ_PREFETCH_DEPTH).await; + wave.remove(&0).unwrap().send(true).unwrap(); + assert_eq!(object_entries([0]), next_batch(&mut stream).await); + + drop(stream); + for (_, mut release) in wave { + timeout(WAIT, release.closed()).await.unwrap(); + } + assert_nothing_parks(&mut parked).await; + assert_eq!( + reads, + METRIC_OBJECT_STORE_WAL_READ_SECONDS.get_sample_count() + ); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_read_fails_at_a_panicking_fetch() { + let _serialized = SERIALIZED.lock().await; + let object_store = memory_store(); + put_object_per_entry(&object_store, 3).await; + let (io, _) = RecordingIo::over(object_store); + let store = open_over(io.clone(), &eager()).await; + *io.panic_range_read_of.lock().unwrap() = Some(1); + let mut stream = store.read(&provider(region(1)), 0, None).await.unwrap(); + assert_eq!(object_entries([0]), next_batch(&mut stream).await); + let error = timeout(WAIT, stream.next()) + .await + .unwrap() + .unwrap() + .unwrap_err(); + assert!( + matches!(&error, Error::WalSegmentFetch { path, .. } if *path == io.object_path(1)), + "unexpected error: {error:?}" + ); + assert!(timeout(WAIT, stream.next()).await.unwrap().is_none()); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_reads_fetch_ahead_within_a_shared_budget() { + let _serialized = SERIALIZED.lock().await; + let depth = READ_PREFETCH_DEPTH as u64; + let objects = depth + 4; + let (store, _io, mut parked) = open_parking_range_reads(objects).await; + + // While other reads hold the whole budget, a read fetches only the + // segment it yields next. + let held = store + .read_prefetch_budget + .clone() + .try_acquire_many_owned(READ_PREFETCH_BYTES as u32) + .unwrap(); + let mut stream = store.read(&provider(region(1)), 0, None).await.unwrap(); + assert!(stream.next().now_or_never().is_none()); + let (object_seq, release) = next_parked_operation(&mut parked).await; + assert_eq!(0, object_seq); + assert_nothing_parks(&mut parked).await; + release.send(true).unwrap(); + assert_eq!(object_entries([0]), next_batch(&mut stream).await); + + // Once the budget is free, the next poll fetches ahead up to the depth. + drop(held); + assert!(stream.next().now_or_never().is_none()); + let wave = parked_operations(&mut parked, READ_PREFETCH_DEPTH).await; + assert_eq!( + (1..=depth).collect::>(), + wave.keys().copied().collect() + ); + assert_nothing_parks(&mut parked).await; + for release in wave.into_values() { + release.send(true).unwrap(); + } + let rest = tokio::spawn(stream.try_collect::>()); + for release in parked_operations(&mut parked, (objects - depth - 1) as usize) + .await + .into_values() + { + release.send(true).unwrap(); + } + let rest = timeout(WAIT, rest).await.unwrap().unwrap().unwrap(); + assert_eq!( + object_entries(1..objects), + rest.into_iter().flatten().collect::>() + ); + assert_eq!( + READ_PREFETCH_BYTES, + store.read_prefetch_budget.available_permits() + ); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_dropped_read_keeps_reservations_until_its_fetches_end() { + let _serialized = SERIALIZED.lock().await; + let object_store = memory_store(); + put_object_per_entry(&object_store, 2).await; + let (io, _) = RecordingIo::over(object_store); + let store = open_over(io.clone(), &eager()).await; + let budget = store.read_prefetch_budget.clone(); + let (began_tx, began_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + *io.block_range_read_of.lock().unwrap() = Some((1, began_tx, release_rx)); + + // The first poll fetches object 0, and object 1 with its bytes + // reserved; the read of object 1 blocks its thread. + let mut stream = store.read(&provider(region(1)), 0, None).await.unwrap(); + let _ = stream.next().now_or_never(); + began_rx.recv_timeout(WAIT).unwrap(); + assert!(budget.available_permits() < READ_PREFETCH_BYTES); + + // The aborted fetch keeps its bytes until it ends. + drop(stream); + for _ in 0..16 { + tokio::task::yield_now().await; + } + assert!(budget.available_permits() < READ_PREFETCH_BYTES); + release_tx.send(()).unwrap(); + wait_until(|| budget.available_permits() == READ_PREFETCH_BYTES).await; + store.stop().await.unwrap(); + } + #[tokio::test] async fn test_store_assigns_ids_from_the_object_sequence() { let _serialized = SERIALIZED.lock().await; @@ -7009,7 +7505,9 @@ mod tests { /// after it wrote the object, fails it with a given error before it /// writes, stores the same content before it writes, or damages the bytes /// the next range read returns; it also fails the next listing, or the - /// next range read of an object at an offset, with a transient error. + /// next range read of an object at an offset, with a transient error, + /// panics in the next range read of a given object, and blocks its thread + /// in the next range read of a given object until the test releases it. struct RecordingIo { inner: ObjectStoreIo, reads: RangeReads, @@ -7019,8 +7517,18 @@ mod tests { duplicate_next_put: AtomicBool, fail_next_list: AtomicBool, fail_range_read_at: Mutex>, + panic_range_read_of: Mutex>, + block_range_read_of: Mutex>, } + /// The object whose range read blocks, the sender that reports the read + /// began and the receiver that releases it. + type BlockedRangeRead = ( + u64, + std::sync::mpsc::Sender<()>, + std::sync::mpsc::Receiver<()>, + ); + impl RecordingIo { fn over(object_store: ObjectStore) -> (Arc, RangeReads) { let reads = RangeReads::default(); @@ -7033,6 +7541,8 @@ mod tests { duplicate_next_put: AtomicBool::new(false), fail_next_list: AtomicBool::new(false), fail_range_read_at: Mutex::new(None), + panic_range_read_of: Mutex::new(None), + block_range_read_of: Mutex::new(None), }; (Arc::new(io), reads) } @@ -7091,6 +7601,23 @@ mod tests { { return injected_failure("read", self.object_path(object_seq)); } + let panics = self + .panic_range_read_of + .lock() + .unwrap() + .take_if(|of| *of == object_seq) + .is_some(); + assert!(!panics, "injected panic"); + let blocked = self + .block_range_read_of + .lock() + .unwrap() + .take_if(|(of, ..)| *of == object_seq); + if let Some((_, began, release)) = blocked { + began.send(()).unwrap(); + // Blocking the thread keeps an abort from cancelling the read. + let _ = release.recv(); + } let bytes = self.inner.get_range(object_seq, offset, len).await?; if self.damage_next_range_read.swap(false, Ordering::Relaxed) { let mut damaged = bytes.to_vec(); @@ -7389,15 +7916,6 @@ mod tests { ); } - /// Polls until `settled` holds. - async fn wait_until(settled: impl Fn() -> bool) { - let deadline = Instant::now() + WAIT; - while !settled() { - assert!(Instant::now() < deadline, "the actor never settled"); - tokio::time::sleep(Duration::from_millis(1)).await; - } - } - #[tokio::test] async fn test_store_fault_matrix() { let _serialized = SERIALIZED.lock().await;