diff --git a/src/log-store/src/object_store_wal/catalog.rs b/src/log-store/src/object_store_wal/catalog.rs index 61b4d2c09cc..49567db3060 100644 --- a/src/log-store/src/object_store_wal/catalog.rs +++ b/src/log-store/src/object_store_wal/catalog.rs @@ -15,7 +15,7 @@ //! In-memory index over the footers of the objects of one WAL prefix. use std::collections::{BTreeMap, HashMap}; -use std::ops::Bound::{Excluded, Unbounded}; +use std::ops::Bound::{Excluded, Included, Unbounded}; use snafu::{OptionExt, ensure}; use store_api::logstore::EntryId; @@ -28,14 +28,24 @@ use crate::object_store_wal::batch::{OBJECT_SEQ_LIMIT, sequence_floor}; use crate::object_store_wal::format::{FooterEntry, RELEASE_REGION_ID}; /// Indexes objects by sequence and, per region, the objects that hold entries -/// of that region. +/// of that region, and counts the live segments of every object under the +/// obsolete watermarks. #[derive(Debug, Default)] pub(crate) struct ObjectCatalog { objects: BTreeMap>, - regions: BTreeMap>, + /// The objects that hold a segment of the region, as the sequence and + /// the footer entry of the segment, keyed by the largest entry id of the + /// segment. The entry ranges of a region increase strictly with the + /// object sequence, so this is also sequence order. + regions: BTreeMap>, /// Largest entry id ever indexed per region, kept after the object that /// held it was removed. max_entry_ids: HashMap, + /// The obsolete watermark per region, see [`obsolete`](Self::obsolete). + watermarks: HashMap, + /// The number of live segments of every indexed object that holds one, + /// see [`reclaim_boundary`](Self::reclaim_boundary). + live_segments: BTreeMap, } impl ObjectCatalog { @@ -84,49 +94,46 @@ impl ObjectCatalog { ); // Validate every region before mutating either index so insertion is atomic. + // While the indexed ranges of a region increase with the sequence, a + // segment can only conflict with the segments right below and right + // above its largest entry id: each must lie on the side of it that + // its sequence is on. for entry in &footer { let Some(region_objects) = self.regions.get(&entry.region_id) else { continue; }; - if let Some((&previous_seq, previous)) = region_objects.range(..object_seq).next_back() - { + let below = region_objects.range(..entry.max_entry_id).next_back(); + let above = region_objects.range(entry.max_entry_id..).next(); + for (_, (neighbor_seq, neighbor)) in below.into_iter().chain(above) { + let ((lower_seq, lower), (upper_seq, upper)) = if *neighbor_seq < object_seq { + ((*neighbor_seq, neighbor), (object_seq, entry)) + } else { + ((object_seq, entry), (*neighbor_seq, neighbor)) + }; ensure!( - previous.max_entry_id < entry.min_entry_id, + lower.max_entry_id < upper.min_entry_id, CorruptedWalObjectSnafu { reason: out_of_order_reason( entry.region_id, - previous_seq, - previous.max_entry_id, - object_seq, - entry.min_entry_id - ), - } - ); - } - if let Some((&next_seq, next)) = region_objects - .range((Excluded(object_seq), Unbounded)) - .next() - { - ensure!( - entry.max_entry_id < next.min_entry_id, - CorruptedWalObjectSnafu { - reason: out_of_order_reason( - entry.region_id, - object_seq, - entry.max_entry_id, - next_seq, - next.min_entry_id + lower_seq, + lower.max_entry_id, + upper_seq, + upper.min_entry_id ), } ); } } + let live = footer.iter().filter(|entry| self.is_live(entry)).count(); + if live > 0 { + self.live_segments.insert(object_seq, live); + } for entry in &footer { self.regions .entry(entry.region_id) .or_default() - .insert(object_seq, entry.clone()); + .insert(entry.max_entry_id, (object_seq, entry.clone())); self.max_entry_ids .entry(entry.region_id) .and_modify(|max| *max = (*max).max(entry.max_entry_id)) @@ -141,9 +148,10 @@ impl ObjectCatalog { /// largest entry id of every region it held is kept. pub(crate) fn remove_object(&mut self, object_seq: u64) -> Option> { let footer = self.objects.remove(&object_seq)?; + self.live_segments.remove(&object_seq); for entry in &footer { if let Some(objects) = self.regions.get_mut(&entry.region_id) { - objects.remove(&object_seq); + objects.remove(&entry.max_entry_id); if objects.is_empty() { self.regions.remove(&entry.region_id); } @@ -157,34 +165,54 @@ impl ObjectCatalog { self.objects.contains_key(&object_seq) } - /// Returns the reclaim boundary the indexed objects and the obsolete - /// watermarks allow, given that every object below `from` holds no live - /// segment: the first indexed object at or above `from` that holds a - /// live segment, or the sequence after the last indexed object when none - /// does, and never less than `from`. - /// - /// A segment is live when its maximum entry id is above the watermark of - /// its region, or its region has no watermark; a region whose watermark - /// is [`EntryId::MAX`] holds no live segment, and neither does + /// Moves the obsolete watermark of `region_id` up to `entry_id`; it never + /// moves down. The segments of the region whose largest entry id the + /// watermark reaches stop being live, so a move costs a lookup per + /// segment it reaches, however many segments their objects hold. + pub(crate) fn obsolete(&mut self, region_id: RegionId, entry_id: EntryId) { + let previous = self.watermarks.get(®ion_id).copied(); + if previous.is_some_and(|previous| previous >= entry_id) { + return; + } + self.watermarks.insert(region_id, entry_id); + if region_id == RELEASE_REGION_ID { + return; + } + let Some(objects) = self.regions.get(®ion_id) else { + return; + }; + let reached = (previous.map_or(Unbounded, Excluded), Included(entry_id)); + for (object_seq, _) in objects.range(reached).map(|(_, object)| object) { + if let Some(live) = self.live_segments.get_mut(object_seq) { + *live -= 1; + if *live == 0 { + self.live_segments.remove(object_seq); + } + } + } + } + + /// Returns whether a segment is live: its region is not /// [`RELEASE_REGION_ID`], whose records recovery reads only from the - /// objects it indexes. Watermarks never move down - /// and a region first appears in an object above the boundary, so the - /// scan starts at the previous boundary and never reads an object twice - /// once the boundary has passed it. - pub(crate) fn reclaim_boundary( - &self, - from: u64, - obsolete_entry_ids: &HashMap, - ) -> u64 { - let live = self.objects.range(from..).find(|(_, footer)| { - footer.iter().any(|entry| { - entry.region_id != RELEASE_REGION_ID - && obsolete_entry_ids - .get(&entry.region_id) - .is_none_or(|obsolete| entry.max_entry_id > *obsolete) - }) - }); - match live { + /// objects it indexes, and either has no watermark or a watermark below + /// the largest entry id of the segment. A region whose watermark is + /// [`EntryId::MAX`] holds no live segment. + fn is_live(&self, entry: &FooterEntry) -> bool { + entry.region_id != RELEASE_REGION_ID + && self + .watermarks + .get(&entry.region_id) + .is_none_or(|watermark| entry.max_entry_id > *watermark) + } + + /// Returns the reclaim boundary the indexed objects and the obsolete + /// watermarks allow: the first indexed object at or above `from` that + /// holds a live segment, see [`is_live`](Self::is_live), or the sequence + /// after the last indexed object when none does, and never less than + /// `from`. The objects holding a live segment are tracked as objects are + /// indexed and watermarks move, so the boundary is a lookup. + pub(crate) fn reclaim_boundary(&self, from: u64) -> u64 { + match self.live_segments.range(from..).next() { Some((&object_seq, _)) => object_seq, None => self .objects @@ -196,7 +224,11 @@ impl ObjectCatalog { } /// Returns the objects that hold entries of `region_id` overlapping - /// `start_entry_id..=end_entry_id`, ordered by object sequence. + /// `start_entry_id..=end_entry_id`, ordered by object sequence. The + /// segments of a region are ordered by entry id, so the lookup seeks to + /// the first segment that ends at or after `start_entry_id` and stops + /// before the first that starts after `end_entry_id`, whatever the number + /// of objects the region holds. pub(crate) fn objects_for_entry_range( &self, region_id: RegionId, @@ -216,11 +248,9 @@ impl ObjectCatalog { return Ok(Vec::new()); }; Ok(objects - .iter() - .filter(|(_, entry)| { - entry.max_entry_id >= start_entry_id && entry.min_entry_id <= end_entry_id - }) - .map(|(&object_seq, entry)| (object_seq, entry)) + .range(start_entry_id..) + .map(|(_, (object_seq, entry))| (*object_seq, entry)) + .take_while(|(_, entry)| entry.min_entry_id <= end_entry_id) .collect()) } @@ -286,6 +316,10 @@ fn out_of_order_reason( #[cfg(test)] mod tests { + use rand::rngs::StdRng; + use rand::seq::SliceRandom; + use rand::{Rng, SeedableRng}; + use super::*; use crate::error::Error; use crate::object_store_wal::batch::entry_id; @@ -467,6 +501,35 @@ mod tests { "are not strictly increasing", ); assert_eq!(1, catalog.objects_in_order().count()); + + // A segment that fits between two others by its entry ids must also + // fit between them by its sequence. + catalog + .insert_object(8, vec![footer_entry(region_id, 50, 60)]) + .unwrap(); + assert_corrupted( + catalog.insert_object(1, vec![footer_entry(region_id, 30, 40)]), + "object 1 ends at 40, object 2 starts at 10", + ); + assert_corrupted( + catalog.insert_object(9, vec![footer_entry(region_id, 30, 40)]), + "object 8 ends at 60, object 9 starts at 30", + ); + assert_corrupted( + catalog.insert_object(5, vec![footer_entry(region_id, 61, 70)]), + "object 5 ends at 70, object 8 starts at 50", + ); + catalog + .insert_object(5, vec![footer_entry(region_id, 30, 40)]) + .unwrap(); + assert_eq!( + vec![2, 5, 8], + object_seqs( + &catalog + .objects_for_entry_range(region_id, 0, EntryId::MAX) + .unwrap() + ) + ); } #[test] @@ -500,6 +563,7 @@ mod tests { fn test_catalog_reclaim_boundary_follows_the_watermarks() { let region_one = RegionId::new(1, 1); let region_two = RegionId::new(2, 1); + let region_three = RegionId::new(3, 1); let mut catalog = ObjectCatalog::default(); catalog.insert_object(0, Vec::new()).unwrap(); catalog @@ -526,49 +590,47 @@ mod tests { vec![footer_entry(region_two, entry_id(3, 1), entry_id(3, 1))], ) .unwrap(); - let boundary = |catalog: &ObjectCatalog, obsolete: &[(RegionId, EntryId)]| { - catalog.reclaim_boundary(0, &obsolete.iter().copied().collect()) - }; // A region without a watermark holds the boundary at its first object. - assert_eq!(1, boundary(&catalog, &[])); - assert_eq!(1, boundary(&catalog, &[(region_one, entry_id(2, 1))])); - // A watermark inside an object keeps that object. - assert_eq!( - 1, - boundary( - &catalog, - &[(region_one, entry_id(1, 1)), (region_two, EntryId::MAX)] - ) - ); - assert_eq!( - 2, - boundary( - &catalog, - &[(region_one, entry_id(1, 2)), (region_two, entry_id(1, 1))] - ) - ); + assert_eq!(1, catalog.reclaim_boundary(0)); + catalog.obsolete(region_two, entry_id(1, 1)); + assert_eq!(1, catalog.reclaim_boundary(0)); + // A watermark inside a segment keeps its object. + catalog.obsolete(region_one, entry_id(1, 1)); + assert_eq!(1, catalog.reclaim_boundary(0)); + catalog.obsolete(region_one, entry_id(1, 2)); + assert_eq!(2, catalog.reclaim_boundary(0)); // The segment of release records never holds the boundary. - assert_eq!( - 3, - boundary( - &catalog, - &[(region_one, entry_id(2, 1)), (region_two, entry_id(1, 1))] - ) - ); - // A region of `obsolete_all` contributes nothing, and when no region - // does, the boundary is past the last object. - assert_eq!( - 4, - boundary( - &catalog, - &[(region_one, entry_id(2, 1)), (region_two, EntryId::MAX)] - ) - ); + catalog.obsolete(region_one, entry_id(2, 1)); + assert_eq!(3, catalog.reclaim_boundary(0)); + // A region whose watermark is `EntryId::MAX` holds nothing, and when + // no region does, the boundary is past the last object. + catalog.obsolete(region_two, EntryId::MAX); + assert_eq!(4, catalog.reclaim_boundary(0)); + // A watermark never moves down, and the boundary is never below + // `from`. + catalog.obsolete(region_one, entry_id(1, 1)); + assert_eq!(4, catalog.reclaim_boundary(0)); + assert_eq!(5, catalog.reclaim_boundary(5)); - // Objects below the previous boundary are not scanned again. - assert_eq!(3, catalog.reclaim_boundary(3, &HashMap::new())); - assert_eq!(5, catalog.reclaim_boundary(5, &HashMap::new())); + // A segment indexed at or below the watermark of its region is not + // live, one of a region without a watermark is. + catalog + .insert_object( + 4, + vec![footer_entry(region_two, entry_id(4, 1), entry_id(4, 1))], + ) + .unwrap(); + assert_eq!(5, catalog.reclaim_boundary(0)); + catalog + .insert_object( + 5, + vec![footer_entry(region_three, entry_id(5, 1), entry_id(5, 1))], + ) + .unwrap(); + assert_eq!(5, catalog.reclaim_boundary(0)); + catalog.obsolete(region_three, entry_id(5, 1)); + assert_eq!(6, catalog.reclaim_boundary(0)); // Removing objects keeps the largest entry id of every region, and a // region without indexed objects contributes nothing. @@ -588,8 +650,194 @@ mod tests { .unwrap() .is_empty() ); - assert_eq!(3, boundary(&catalog, &[])); - assert_eq!(4, catalog.next_object_seq().unwrap()); + assert_eq!(6, catalog.reclaim_boundary(0)); + assert_eq!(6, catalog.next_object_seq().unwrap()); + } + + /// Random appends, watermark moves, releases, drops, collections and + /// recoveries give the reclaim boundary that a scan of the footers from + /// `from` gives, whatever `from` is, and the objects of a region that a + /// scan finds for a range of entry ids. + #[test] + fn test_catalog_matches_a_scan_of_the_footers() { + let regions = (1..=6) + .map(|number| RegionId::new(1, number)) + .collect::>(); + for seed in 0..256 { + let mut rng = StdRng::seed_from_u64(seed); + let mut catalog = ObjectCatalog::default(); + let mut watermarks = HashMap::new(); + // The largest entry id written per region, which a later segment + // of the region is above. + let mut written = HashMap::new(); + let mut next_object_seq = 0; + let mut boundary = 0; + for step in 0..128 { + let mut releases = rng.random_bool(0.2); + match rng.random_range(0..8) { + 0..=2 => {} + // A watermark move, which may be below the watermark or + // above every id written, of a region or, rarely, of the + // reserved region, which holds nothing live either way. + 3 | 4 => { + let region_id = if rng.random_bool(0.1) { + RELEASE_REGION_ID + } else { + regions[rng.random_range(0..regions.len())] + }; + let ceiling = written.get(®ion_id).copied().unwrap_or(0) + 2; + let entry_id = rng.random_range(0..=ceiling); + obsolete(&mut catalog, &mut watermarks, region_id, entry_id); + } + // A release through the largest id written, or a drop, + // seals an object with a release record. + 5 => { + let region_id = regions[rng.random_range(0..regions.len())]; + let entry_id = if rng.random_bool(0.5) { + written.get(®ion_id).copied().unwrap_or(0) + } else { + EntryId::MAX + }; + obsolete(&mut catalog, &mut watermarks, region_id, entry_id); + releases = true; + } + // A collection deletes objects below the boundary, in any + // order and not always all of them. Removing an object + // above it, which may hold live segments, is followed too. + 6 => { + let collected = catalog + .objects_in_order() + .map(|(object_seq, _)| object_seq) + .filter(|object_seq| { + (*object_seq < boundary || rng.random_bool(0.1)) + && rng.random_bool(0.7) + }) + .collect::>(); + for object_seq in collected { + assert!(catalog.remove_object(object_seq).is_some()); + } + } + // A recovery indexes the objects in any order, then the + // watermarks. + _ => { + let mut objects = catalog + .objects_in_order() + .map(|(object_seq, footer)| (object_seq, footer.to_vec())) + .collect::>(); + objects.shuffle(&mut rng); + catalog = ObjectCatalog::default(); + for (object_seq, footer) in objects { + catalog.insert_object(object_seq, footer).unwrap(); + } + for (region_id, entry_id) in &watermarks { + catalog.obsolete(*region_id, *entry_id); + } + } + } + if rng.random_bool(0.5) || releases { + let segments = regions + .iter() + .copied() + .filter(|_| rng.random_bool(0.5)) + .chain(releases.then_some(RELEASE_REGION_ID)) + .collect::>(); + let mut footer = Vec::new(); + for region_id in segments { + let min_entry_id = written.get(®ion_id).copied().unwrap_or(0) + + 1 + + rng.random_range(0..3); + let max_entry_id = min_entry_id + rng.random_range(0..3); + written.insert(region_id, max_entry_id); + footer.push(footer_entry(region_id, min_entry_id, max_entry_id)); + } + catalog.insert_object(next_object_seq, footer).unwrap(); + next_object_seq += rng.random_range(1..=2); + } + + let expected = scanned_reclaim_boundary(&catalog, boundary, &watermarks); + assert_eq!( + expected, + catalog.reclaim_boundary(boundary), + "seed {seed}, step {step}" + ); + let from = rng.random_range(0..=next_object_seq + 1); + assert_eq!( + scanned_reclaim_boundary(&catalog, from, &watermarks), + catalog.reclaim_boundary(from), + "seed {seed}, step {step}, from {from}" + ); + boundary = expected; + + for region_id in regions.iter().chain([&RELEASE_REGION_ID]) { + let ceiling = written.get(region_id).copied().unwrap_or(0) + 2; + let start = rng.random_range(0..=ceiling); + let end = rng.random_range(start..=ceiling); + let scanned = catalog + .objects_in_order() + .filter(|(_, footer)| { + footer.iter().any(|entry| { + entry.region_id == *region_id + && entry.max_entry_id >= start + && entry.min_entry_id <= end + }) + }) + .map(|(object_seq, _)| object_seq) + .collect::>(); + assert_eq!( + scanned, + object_seqs( + &catalog + .objects_for_entry_range(*region_id, start, end) + .unwrap() + ), + "seed {seed}, step {step}, region {region_id}, {start}..={end}" + ); + } + } + } + } + + fn obsolete( + catalog: &mut ObjectCatalog, + watermarks: &mut HashMap, + region_id: RegionId, + entry_id: EntryId, + ) { + catalog.obsolete(region_id, entry_id); + watermarks + .entry(region_id) + .and_modify(|watermark: &mut EntryId| *watermark = (*watermark).max(entry_id)) + .or_insert(entry_id); + } + + /// The first object at or above `from` holding a segment that is live + /// under `watermarks`, found by reading every footer, or the sequence + /// after the last object when none does, and never less than `from`. + fn scanned_reclaim_boundary( + catalog: &ObjectCatalog, + from: u64, + watermarks: &HashMap, + ) -> u64 { + let live = catalog + .objects_in_order() + .filter(|(object_seq, _)| *object_seq >= from) + .find(|(_, footer)| { + footer.iter().any(|entry| { + entry.region_id != RELEASE_REGION_ID + && watermarks + .get(&entry.region_id) + .is_none_or(|watermark| entry.max_entry_id > *watermark) + }) + }); + match live { + Some((object_seq, _)) => object_seq, + None => catalog + .objects_in_order() + .last() + .map_or(from, |(object_seq, _)| { + object_seq.saturating_add(1).max(from) + }), + } } fn footer_entry(region_id: RegionId, min_entry_id: u64, max_entry_id: u64) -> FooterEntry { diff --git a/src/log-store/src/object_store_wal/store.rs b/src/log-store/src/object_store_wal/store.rs index df037b43771..d5096c073eb 100644 --- a/src/log-store/src/object_store_wal/store.rs +++ b/src/log-store/src/object_store_wal/store.rs @@ -394,8 +394,10 @@ impl ObjectStoreLogStore { METRIC_OBJECT_STORE_WAL_INDEXED_OBJECTS.inc(); METRIC_OBJECT_STORE_WAL_INDEXED_BYTES.add(object_len(&[]) as i64); // The release records are the only watermarks recorded yet. - let writer_reclaim_boundary = - catalog.reclaim_boundary(reclaim_boundary, &released_entry_ids); + for (®ion_id, &entry_id) in &released_entry_ids { + catalog.obsolete(region_id, entry_id); + } + let writer_reclaim_boundary = catalog.reclaim_boundary(reclaim_boundary); let catalog = Arc::new(RwLock::new(catalog)); let obsolete_entry_ids = Arc::new(Mutex::new(released_entry_ids)); let wal_holes = WalHoles::default(); @@ -710,14 +712,24 @@ fn injected_create_failure(io: &dyn WalObjectIo, object_seq: u64) -> Result, + region_id: RegionId, + entry_id: EntryId, +) { obsolete_entry_ids .lock() .unwrap_or_else(PoisonError::into_inner) .entry(region_id) .and_modify(|current| *current = (*current).max(entry_id)) .or_insert(entry_id); + catalog + .write() + .unwrap_or_else(PoisonError::into_inner) + .obsolete(region_id, entry_id); } fn positive_bytes(bytes: u64, name: &str) -> Result { @@ -1130,7 +1142,12 @@ impl LogStore for ObjectStoreLogStore { // still queued behind the stop: nothing is assigned an id any // more, so the watermark alone is consistent. None => { - record_obsolete(&self.obsolete_entry_ids, region_id, watermark); + record_obsolete( + &self.obsolete_entry_ids, + &self.catalog, + region_id, + watermark, + ); Ok(()) } } @@ -2412,7 +2429,12 @@ impl Actor { response: oneshot::Sender>, ) { let result = self.raise_sequence_floor(region_id, entry_id).map(|_| { - record_obsolete(&self.obsolete_entry_ids, region_id, watermark); + record_obsolete( + &self.obsolete_entry_ids, + &self.catalog, + region_id, + watermark, + ); }); if result.is_ok() { self.update_reclaim_boundary(); @@ -2451,7 +2473,7 @@ impl Actor { } else { self.issued_entry_ids.get(®ion_id).copied().unwrap_or(0) }; - record_obsolete(&self.obsolete_entry_ids, region_id, entry_id); + record_obsolete(&self.obsolete_entry_ids, &self.catalog, region_id, entry_id); self.update_reclaim_boundary(); self.collect(); let release = Release { @@ -2481,17 +2503,11 @@ impl Actor { /// Raises the reclaim boundary to what the catalog and the obsolete /// watermarks allow. fn update_reclaim_boundary(&mut self) { - let boundary = { - let obsolete_entry_ids = self - .obsolete_entry_ids - .lock() - .unwrap_or_else(PoisonError::into_inner); - self.catalog - .read() - .unwrap_or_else(PoisonError::into_inner) - .reclaim_boundary(self.reclaim_boundary, &obsolete_entry_ids) - }; - self.reclaim_boundary = boundary; + self.reclaim_boundary = self + .catalog + .read() + .unwrap_or_else(PoisonError::into_inner) + .reclaim_boundary(self.reclaim_boundary); } /// Starts a collection, which also attempts the deletes that failed since @@ -9926,14 +9942,157 @@ mod tests { } } - /// Measurements of costs that grow with the prefix, taken through the - /// store. They are ignored by default and print what they measure; run - /// them with `--run-ignored only` and `--no-capture`, preferably with - /// `--cargo-profile release`. + /// Measurements of costs that grow with the catalog or the prefix, + /// taken through the store. They are ignored by default and print what + /// they measure; run them with `--run-ignored only` and `--no-capture`, + /// preferably with `--cargo-profile release`. #[allow(clippy::print_stdout)] mod measurements { use super::*; + /// Times every watermark update while one object holds a segment of + /// each region and the regions are flushed in region id order, so + /// each update leaves the next region as the first live segment of + /// that object. An update is timed as the caller of `obsolete` sees + /// it, which includes the round trip of its command to the actor. + #[tokio::test] + #[ignore] + async fn measure_watermark_updates_over_one_shared_object() { + let _serialized = SERIALIZED.lock().await; + for regions in [1_000, 5_000, 10_000, 20_000] { + let store = open_with_one_shared_object(regions).await; + let mut pauses = Vec::with_capacity(regions as usize); + for number in 1..=regions { + let region_id = region(number); + let started_at = Instant::now(); + store + .obsolete(&provider(region_id), region_id, id(1, 1)) + .await + .unwrap(); + pauses.push(started_at.elapsed()); + } + // The next object carries the boundary past every flushed + // object. + append(&store, region(regions + 1), "b").await.unwrap(); + assert_eq!(vec![2], collected_object_seqs(&store).await); + print_pauses("watermark updates", regions, pauses); + store.stop().await.unwrap(); + } + } + + /// Times the steps the actor takes for each of the same watermark + /// updates that depend on the catalog, recording the watermark and + /// recomputing the reclaim boundary, without the round trip: the + /// store is stopped first and the steps run on its catalog. + #[tokio::test] + #[ignore] + async fn measure_actor_watermark_steps_over_one_shared_object() { + let _serialized = SERIALIZED.lock().await; + for regions in [1_000, 5_000, 10_000, 20_000] { + let store = open_with_one_shared_object(regions).await; + store.stop().await.unwrap(); + let mut boundary = 0; + let mut pauses = Vec::with_capacity(regions as usize); + for number in 1..=regions { + let started_at = Instant::now(); + record_obsolete( + &store.obsolete_entry_ids, + &store.catalog, + region(number), + id(1, 1), + ); + boundary = store.catalog.read().unwrap().reclaim_boundary(boundary); + pauses.push(started_at.elapsed()); + } + assert_eq!(2, boundary); + print_pauses("actor watermark steps", regions, pauses); + } + } + + /// Opens a store whose object 1 holds one segment of each of + /// `regions` regions. + async fn open_with_one_shared_object(regions: u32) -> Arc { + let store = open(memory_store(), &eager()).await; + let entries = (1..=regions) + .map(|number| entry(&store, region(number), "a")) + .collect(); + store.append_batch(entries).await.unwrap(); + store + } + + fn print_pauses(label: &str, regions: u32, mut pauses: Vec) { + let total = pauses.iter().sum::(); + pauses.sort_unstable(); + let micros = |pause: Duration| pause.as_secs_f64() * 1e6; + println!( + "{label}: regions={regions} total_ms={:.3} mean_us={:.3} p50_us={:.3} p99_us={:.3} max_us={:.3}", + total.as_secs_f64() * 1e3, + micros(total) / pauses.len() as f64, + micros(pauses[pauses.len() / 2]), + micros(pauses[pauses.len() * 99 / 100]), + micros(pauses[pauses.len() - 1]), + ); + } + + /// Times 100 reads of the last ten objects of a region that holds a + /// segment in every retained object, and 100 lookups of those objects + /// in the catalog alone. + #[tokio::test] + #[ignore] + async fn measure_tail_reads_over_retained_objects() { + let _serialized = SERIALIZED.lock().await; + let region_id = region(1); + for objects in [10_000, 100_000, 1_000_000] { + let object_store = memory_store(); + let io = ObjectStoreIo::new(object_store.clone(), PREFIX).unwrap(); + futures::stream::iter(0..objects) + .for_each_concurrent(RECOVERY_CONCURRENCY, |object_seq| { + let io = &io; + async move { + let header = Header { + object_seq, + epoch: 1, + prev: object_seq.checked_sub(1).map(|prev| ChainLink { + object_seq: prev, + epoch: 1, + }), + reclaim_boundary: 0, + }; + let record = Record { + region_id, + entry_id: id(object_seq, 1), + payload: Bytes::from_static(b"a"), + }; + put_object_with_header(io, header, &[record]).await; + } + }) + .await; + let store = open(object_store, &eager()).await; + let start = id(objects - 10, 1); + + let started_at = Instant::now(); + for _ in 0..100 { + assert_eq!(10, read_entries(&store, region_id, start).await.len()); + } + let reads = started_at.elapsed(); + let started_at = Instant::now(); + for _ in 0..100 { + let catalog = store.catalog.read().unwrap(); + let found = catalog + .objects_for_entry_range(region_id, start, EntryId::MAX) + .unwrap(); + assert_eq!(10, found.len()); + } + let lookups = started_at.elapsed(); + println!( + "tail reads: objects={objects} read_100_ms={:.3} lookup_100_ms={:.3}", + reads.as_secs_f64() * 1e3, + lookups.as_secs_f64() * 1e3, + ); + store.stop().await.unwrap(); + } + } + /// Object access that counts the reads of whole objects and of byte /// ranges. struct CountingIo {