rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-22 11:52:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(log-store): keep the object store WAL reclaim boundary independent of the region count (#9644)

Browse Source 
* perf(log-store): keep the object store WAL reclaim boundary independent of the region count

The WAL actor recomputed the reclaim boundary after every watermark
update, release and indexed object by scanning the footers of the objects
from the previous boundary, checking each segment against the obsolete
watermarks. Each scan started at the first segment of the boundary object,
so when one object holds segments of many regions whose watermarks advance
one region at a time, every update inspected the segments already known to
be obsolete again, about R^2/2 footer entries in total for R regions.

The catalog now keeps the obsolete watermarks itself and counts the live
segments of every object. A watermark move visits only the segments of its
region that it newly reaches, through the per-region index, and the
boundary is the first object at or above the floor that still holds a live
segment. The boundary is unchanged: segments of release records never hold
it, a region without a watermark does, a watermark of EntryId::MAX releases
every segment of its region, and it never moves below the floor. The store
records each watermark in the catalog together with the one reads use.

The per-region index is now keyed by the largest entry id of each segment,
which is also object sequence order because the entry ranges of a region
increase strictly with the sequence. Indexing an object already enforced
that; it now checks the neighbours by entry id instead of by sequence,
which accepts and rejects the same objects.

A property test checks the boundary against a scan of the footers over
random appends, watermark moves, releases, drops, collections and
recoveries, and ignored measurement tests time watermark updates and tail
reads through the store.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* perf(log-store): seek the object store WAL objects of a region by entry id

A read found the objects of its region by filtering every object of the
region that the catalog retains, so a read of the last few objects cost
time in proportion to the whole retained history of the region, under the
catalog read lock.

The per-region index is keyed by the largest entry id of each segment, and
the entry ranges of a region increase strictly with the object sequence,
which the catalog enforces when it indexes an object, whatever entry ids
the objects carry. So the lookup seeks to the first segment that ends at
or after the start of the range and stops at the first that starts after
its end, and the result is still in sequence order.

The property test of the catalog now also checks range lookups against a
scan of the footers.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

---------

Signed-off-by: jeremyhi <fengjiachun@gmail.com>
This commit is contained in:
jeremyhi
2026-10-16 14:27:04 +00:00
committed by GitHub
parent 095c288f16
commit baa5630119
2 changed files with 532 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/log-store/src/object_store_wal/catalog.rs
+351 -103
View File
@@ -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<u64, Vec<FooterEntry>>,
regions: BTreeMap<RegionId, BTreeMap<u64, FooterEntry>>,
/// 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<RegionId, BTreeMap<EntryId, (u64, FooterEntry)>>,
/// Largest entry id ever indexed per region, kept after the object that
/// held it was removed.
max_entry_ids: HashMap<RegionId, EntryId>,
/// The obsolete watermark per region, see [`obsolete`](Self::obsolete).
watermarks: HashMap<RegionId, EntryId>,
/// The number of live segments of every indexed object that holds one,
/// see [`reclaim_boundary`](Self::reclaim_boundary).
live_segments: BTreeMap<u64, usize>,
}
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<Vec<FooterEntry>> {
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(&region_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(&region_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<RegionId, EntryId>,
) -> 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::<Vec<_>>();
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(&region_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(&region_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::<Vec<_>>();
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::<Vec<_>>();
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::<Vec<_>>();
let mut footer = Vec::new();
for region_id in segments {
let min_entry_id = written.get(&region_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::<Vec<_>>();
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<RegionId, EntryId>,
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<RegionId, EntryId>,
) -> 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 {
src/log-store/src/object_store_wal/store.rs
+181 -22
View File
@@ -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 (&region_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<PutR
})
}
/// Records the obsolete watermark of `region_id`, which never moves down.
fn record_obsolete(obsolete_entry_ids: &ObsoleteEntryIds, region_id: RegionId, entry_id: EntryId) {
/// Records the obsolete watermark of `region_id`, which never moves down, for
/// reads and in the catalog for the reclaim boundary.
fn record_obsolete(
obsolete_entry_ids: &ObsoleteEntryIds,
catalog: &RwLock<ObjectCatalog>,
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<usize> {
@@ -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<Result<()>>,
) {
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(&region_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<ObjectStoreLogStore> {
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<Duration>) {
let total = pauses.iter().sum::<Duration>();
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 {