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

perf(log-store): fetch object store WAL replay segments concurrently (#9555)

Browse Source 
* perf(log-store): fetch object store WAL replay segments concurrently

A region read fetched each segment only after the previous object's
entries were yielded, so a replay paid the full request latency once per
object. Fetch up to READ_PREFETCH_DEPTH (8) segments ahead, each on its
own task, and handle their outcomes in object sequence order: entries,
collected-object skips, corrupted-segment holes and errors surface
exactly as before, and dropping the stream aborts the fetches in flight.
A read holds at most eight decoded segments besides the one it yields.

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

* perf(log-store): raise object store WAL recovery concurrency to 64

On S3, recovery reads the header and tail of every object, about 57 to
91 ms per object per lane on 4, 8 and 16 vCPUs alike, and its time
scales with objects divided by concurrency up to 128 lanes without
throttling: about 9,500 objects recovered in 99 s at 8 and 10.5 s at
64. At 64 recovery issues about 1,800 reads per second, a third of
S3's request rate per prefix, and holds one 64 KiB tail window per
lane, 4 MiB in all.

The recovery concurrency tests now scale with the constant.

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

* perf(log-store): share one prefetch byte budget among object store WAL reads

Each read fetched up to eight segments ahead on its own, so regions
opened in parallel (16 by default) could together hold 128 segments of
up to max_batch_bytes each. A read now always fetches the segment it
yields next, as before, and fetches a later one only while its length
fits in a 64 MiB budget shared by every read of the store. Regions
replayed in parallel thus hold one segment each plus at most 64 MiB,
however many open at once. Nothing waits on the budget: a read that
cannot reserve stops fetching ahead until its next poll.

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

* fix(log-store): fail an object store WAL read whose fetch task panics

A fetch task that panicked was re-raised in the reader with
resume_unwind, which unwinds through the caller, such as Mito's region
opener task, past its cleanup. Return it as a WalSegmentFetch error
instead, so the read ends at that object like any other failed fetch.

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

* fix(log-store): bound replay prefetch by decoded size held by fetch tasks

The prefetch reservation sat in the read's queue, so dropping a read
released it at once while an aborted fetch could still hold its
buffers until it next yielded. The fetch task now holds the reservation
and returns it with its outcome, so it is released with the buffers.

A reservation also counted only the encoded segment length, while a
fetched segment is held decoded. It now counts the length plus one
Record per entry, and the comment states the whole bound.

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

* docs(log-store): round the measured figures in the recovery concurrency comment

Keep the magnitudes in the comment; the exact S3 figures belong to the
measurement run rather than the code.

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

* refactor(log-store): abort replay fetch tasks with tokio-util's AbortOnDropHandle

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

* refactor(log-store): rename the replay prefetch reservation to try_reserve_prefetch_bytes

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

* test(log-store): share the bounded polling helper across the object store WAL tests

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

---------

Signed-off-by: jeremyhi <fengjiachun@gmail.com>
This commit is contained in:
jeremyhi
2026-10-13 12:26:16 +00:00
committed by GitHub
parent 2a70194930
commit 55d753ca58
4 changed files with 563 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -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",
src/log-store/Cargo.toml
+1
View File
@@ -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"] }
src/log-store/src/error.rs
+11 -1
View File
@@ -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 { .. }
src/log-store/src/object_store_wal/store.rs
+549 -31
View File
@@ -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<DeletingObjects>,
/// Bytes the reads may fetch ahead, see [`READ_PREFETCH_BYTES`].
read_prefetch_budget: Arc<Semaphore>,
terminal_error: TerminalError,
stopped: Arc<AtomicBool>,
command_tx: mpsc::Sender<Command>,
@@ -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<Semaphore>,
entry: &FooterEntry,
) -> Option<OwnedSemaphorePermit> {
let records = u64::from(entry.entry_count) * std::mem::size_of::<Record>() 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<dyn WalObjectIo>,
object_seq: u64,
entry: FooterEntry,
permit: Option<OwnedSemaphorePermit>,
) -> impl Future<Output = (Result<Vec<Record>>, Option<OwnedSemaphorePermit>)> + 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::<Vec<_>>(),
(0..depth).collect::<Vec<_>>(),
wave.iter()
.map(|(object_seq, _)| *object_seq)
.collect::<Vec<_>>()
);
// 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::<Vec<u64>>(),
(0..2 * depth).collect::<Vec<u64>>(),
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::<Vec<_>>();
assert_eq!(expected, entries);
assert_eq!(expected_ranges[&region(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[&region(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<Item = u64>) -> Vec<Entry> {
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<ObjectStoreLogStore>,
Arc<ParkedIo>,
mpsc::UnboundedReceiver<(u64, oneshot::Sender<bool>)>,
) {
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<bool>)>,
) {
assert!(
timeout(Duration::from_millis(100), parked.recv())
.await
.is_err()
);
}
async fn next_batch(stream: &mut SendableEntryStream<'static, Entry, Error>) -> Vec<Entry> {
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::<BTreeSet<_>>(),
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::<Vec<_>>());
let refills = parked_operations(&mut parked, (objects - depth) as usize).await;
assert_eq!(
(depth..objects).collect::<BTreeSet<_>>(),
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::<Vec<_>>()
);
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::<Vec<_>>(),
refills
.iter()
.map(|(object_seq, _)| *object_seq)
.collect::<Vec<_>>()
);
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::<Vec<_>>())
.await
.unwrap()
.unwrap()
.into_iter()
.flatten()
.collect::<Vec<_>>();
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::<BTreeSet<_>>(),
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::<Vec<_>>());
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::<Vec<_>>()
);
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<Option<(u64, u64)>>,
panic_range_read_of: Mutex<Option<u64>>,
block_range_read_of: Mutex<Option<BlockedRangeRead>>,
}
/// 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<Self>, 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;