diff --git a/pageserver/src/basebackup.rs b/pageserver/src/basebackup.rs index c18c78c3b3..a6087920fd 100644 --- a/pageserver/src/basebackup.rs +++ b/pageserver/src/basebackup.rs @@ -124,6 +124,13 @@ where full_backup, replica, ctx, + io_concurrency: IoConcurrency::spawn_from_conf( + timeline.conf, + timeline + .gate + .enter() + .map_err(|e| BasebackupError::Server(e.into()))?, + ), }; basebackup .send_tarball() @@ -145,6 +152,7 @@ where full_backup: bool, replica: bool, ctx: &'a RequestContext, + io_concurrency: IoConcurrency, } /// A sink that accepts SLRU blocks ordered by key and forwards @@ -301,17 +309,10 @@ where let mut slru_builder = SlruSegmentsBuilder::new(&mut self.ar); - let io_concurrency = IoConcurrency::spawn_from_conf( - self.timeline.conf, - self.timeline - .gate - .enter() - .map_err(|e| BasebackupError::Server(e.into()))?, - ); for part in slru_partitions.parts { let blocks = self .timeline - .get_vectored(part, self.lsn, io_concurrency.clone(), self.ctx) + .get_vectored(part, self.lsn, self.io_concurrency.clone(), self.ctx) .await .map_err(|e| BasebackupError::Server(e.into()))?; @@ -366,7 +367,7 @@ where let start_time = Instant::now(); let aux_files = self .timeline - .list_aux_files(self.lsn, self.ctx) + .list_aux_files(self.lsn, self.ctx, self.io_concurrency.clone()) .await .map_err(|e| BasebackupError::Server(e.into()))?; let aux_scan_time = start_time.elapsed(); @@ -430,7 +431,7 @@ where } let repl_origins = self .timeline - .get_replorigins(self.lsn, self.ctx) + .get_replorigins(self.lsn, self.ctx, self.io_concurrency.clone()) .await .map_err(|e| BasebackupError::Server(e.into()))?; let n_origins = repl_origins.len(); @@ -497,7 +498,13 @@ where for blknum in startblk..endblk { let img = self .timeline - .get_rel_page_at_lsn(src, blknum, Version::Lsn(self.lsn), self.ctx) + .get_rel_page_at_lsn( + src, + blknum, + Version::Lsn(self.lsn), + self.ctx, + self.io_concurrency.clone(), + ) .await .map_err(|e| BasebackupError::Server(e.into()))?; segment_data.extend_from_slice(&img[..]); diff --git a/pageserver/src/http/routes.rs b/pageserver/src/http/routes.rs index 33b2d04588..5452719bcd 100644 --- a/pageserver/src/http/routes.rs +++ b/pageserver/src/http/routes.rs @@ -84,6 +84,7 @@ use crate::tenant::remote_timeline_client::list_remote_tenant_shards; use crate::tenant::remote_timeline_client::list_remote_timelines; use crate::tenant::secondary::SecondaryController; use crate::tenant::size::ModelInputs; +use crate::tenant::storage_layer::IoConcurrency; use crate::tenant::storage_layer::LayerAccessStatsReset; use crate::tenant::storage_layer::LayerName; use crate::tenant::timeline::import_pgdata; @@ -2938,8 +2939,15 @@ async fn list_aux_files( active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id) .await?; + let io_concurrency = IoConcurrency::spawn_from_conf( + state.conf, + timeline.gate.enter().map_err(|_| ApiError::Cancelled)?, + ); + let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download); - let files = timeline.list_aux_files(body.lsn, &ctx).await?; + let files = timeline + .list_aux_files(body.lsn, &ctx, io_concurrency) + .await?; json_response(StatusCode::OK, files) } diff --git a/pageserver/src/pgdatadir_mapping.rs b/pageserver/src/pgdatadir_mapping.rs index 781ef68943..c9a83e6ea6 100644 --- a/pageserver/src/pgdatadir_mapping.rs +++ b/pageserver/src/pgdatadir_mapping.rs @@ -201,6 +201,7 @@ impl Timeline { blknum: BlockNumber, version: Version<'_>, ctx: &RequestContext, + io_concurrency: IoConcurrency, ) -> Result { match version { Version::Lsn(effective_lsn) => { @@ -209,7 +210,7 @@ impl Timeline { .get_rel_page_at_lsn_batched( pages.iter().map(|(tag, blknum)| (tag, blknum)), effective_lsn, - IoConcurrency::sequential(), + io_concurrency.clone(), ctx, ) .await; @@ -895,9 +896,15 @@ impl Timeline { &self, lsn: Lsn, ctx: &RequestContext, + io_concurrency: IoConcurrency, ) -> Result, PageReconstructError> { let kv = self - .scan(KeySpace::single(Key::metadata_aux_key_range()), lsn, ctx) + .scan( + KeySpace::single(Key::metadata_aux_key_range()), + lsn, + ctx, + io_concurrency, + ) .await?; let mut result = HashMap::new(); let mut sz = 0; @@ -921,7 +928,8 @@ impl Timeline { lsn: Lsn, ctx: &RequestContext, ) -> Result<(), PageReconstructError> { - self.list_aux_files_v2(lsn, ctx).await?; + self.list_aux_files_v2(lsn, ctx, IoConcurrency::sequential()) + .await?; Ok(()) } @@ -929,17 +937,24 @@ impl Timeline { &self, lsn: Lsn, ctx: &RequestContext, + io_concurrency: IoConcurrency, ) -> Result, PageReconstructError> { - self.list_aux_files_v2(lsn, ctx).await + self.list_aux_files_v2(lsn, ctx, io_concurrency).await } pub(crate) async fn get_replorigins( &self, lsn: Lsn, ctx: &RequestContext, + io_concurrency: IoConcurrency, ) -> Result, PageReconstructError> { let kv = self - .scan(KeySpace::single(repl_origin_key_range()), lsn, ctx) + .scan( + KeySpace::single(repl_origin_key_range()), + lsn, + ctx, + io_concurrency, + ) .await?; let mut result = HashMap::new(); for (k, v) in kv { @@ -2438,7 +2453,11 @@ mod tests { ("foo/bar2".to_string(), Bytes::from_static(b"content2")), ]); - let readback = tline.list_aux_files(Lsn(0x1008), &ctx).await?; + let io_concurrency = IoConcurrency::spawn_for_test(); + + let readback = tline + .list_aux_files(Lsn(0x1008), &ctx, io_concurrency.clone()) + .await?; assert_eq!(readback, expect_1008); // Second modification: update one key, remove the other @@ -2450,11 +2469,15 @@ mod tests { let expect_2008 = HashMap::from([("foo/bar1".to_string(), Bytes::from_static(b"content3"))]); - let readback = tline.list_aux_files(Lsn(0x2008), &ctx).await?; + let readback = tline + .list_aux_files(Lsn(0x2008), &ctx, io_concurrency.clone()) + .await?; assert_eq!(readback, expect_2008); // Reading back in time works - let readback = tline.list_aux_files(Lsn(0x1008), &ctx).await?; + let readback = tline + .list_aux_files(Lsn(0x1008), &ctx, io_concurrency.clone()) + .await?; assert_eq!(readback, expect_1008); Ok(()) diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs index 3e84530a8d..01b51a7585 100644 --- a/pageserver/src/tenant.rs +++ b/pageserver/src/tenant.rs @@ -6617,6 +6617,7 @@ mod tests { let harness = TenantHarness::create("test_get_vectored_aux_files").await?; let (tenant, ctx) = harness.load().await; + let io_concurrency = IoConcurrency::spawn_for_test(); let tline = tenant .create_empty_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx) .await?; @@ -6651,7 +6652,7 @@ mod tests { .get_vectored_impl( aux_keyspace.clone(), read_lsn, - &mut ValuesReconstructState::new(IoConcurrency::sequential()), + &mut ValuesReconstructState::new(io_concurrency.clone()), &ctx, ) .await; @@ -6699,6 +6700,7 @@ mod tests { ) .await?; let (tenant, ctx) = harness.load().await; + let io_concurrency = IoConcurrency::spawn_for_test(); let mut current_key = Key::from_hex("010000000033333333444444445500000000").unwrap(); let gap_at_key = current_key.add(100); @@ -6799,7 +6801,7 @@ mod tests { .get_vectored_impl( read.clone(), current_lsn, - &mut ValuesReconstructState::new(IoConcurrency::sequential()), + &mut ValuesReconstructState::new(io_concurrency.clone()), &ctx, ) .await?; @@ -6842,6 +6844,7 @@ mod tests { async fn test_get_vectored_ancestor_descent() -> anyhow::Result<()> { let harness = TenantHarness::create("test_get_vectored_on_lsn_axis").await?; let (tenant, ctx) = harness.load().await; + let io_concurrency = IoConcurrency::spawn_for_test(); let start_key = Key::from_hex("010000000033333333444444445500000000").unwrap(); let end_key = start_key.add(1000); @@ -6934,7 +6937,7 @@ mod tests { ranges: vec![child_gap_at_key..child_gap_at_key.next()], }, query_lsn, - &mut ValuesReconstructState::new(IoConcurrency::sequential()), + &mut ValuesReconstructState::new(io_concurrency.clone()), &ctx, ) .await; @@ -7380,6 +7383,7 @@ mod tests { async fn test_metadata_scan() -> anyhow::Result<()> { let harness = TenantHarness::create("test_metadata_scan").await?; let (tenant, ctx) = harness.load().await; + let io_concurrency = IoConcurrency::spawn_for_test(); let tline = tenant .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx) .await?; @@ -7433,7 +7437,7 @@ mod tests { .get_vectored_impl( keyspace.clone(), lsn, - &mut ValuesReconstructState::new(IoConcurrency::sequential()), + &mut ValuesReconstructState::new(io_concurrency.clone()), &ctx, ) .await? @@ -7548,6 +7552,7 @@ mod tests { let harness = TenantHarness::create("test_aux_file_e2e").await.unwrap(); let (tenant, ctx) = harness.load().await; + let io_concurrency = IoConcurrency::spawn_for_test(); let mut lsn = Lsn(0x08); @@ -7567,7 +7572,10 @@ mod tests { } // we can read everything from the storage - let files = tline.list_aux_files(lsn, &ctx).await.unwrap(); + let files = tline + .list_aux_files(lsn, &ctx, io_concurrency.clone()) + .await + .unwrap(); assert_eq!( files.get("pg_logical/mappings/test1"), Some(&bytes::Bytes::from_static(b"first")) @@ -7583,7 +7591,10 @@ mod tests { modification.commit(&ctx).await.unwrap(); } - let files = tline.list_aux_files(lsn, &ctx).await.unwrap(); + let files = tline + .list_aux_files(lsn, &ctx, io_concurrency.clone()) + .await + .unwrap(); assert_eq!( files.get("pg_logical/mappings/test2"), Some(&bytes::Bytes::from_static(b"second")) @@ -7594,7 +7605,10 @@ mod tests { .await .unwrap(); - let files = child.list_aux_files(lsn, &ctx).await.unwrap(); + let files = child + .list_aux_files(lsn, &ctx, io_concurrency.clone()) + .await + .unwrap(); assert_eq!(files.get("pg_logical/mappings/test1"), None); assert_eq!(files.get("pg_logical/mappings/test2"), None); } @@ -7603,6 +7617,7 @@ mod tests { async fn test_metadata_image_creation() -> anyhow::Result<()> { let harness = TenantHarness::create("test_metadata_image_creation").await?; let (tenant, ctx) = harness.load().await; + let io_concurrency = IoConcurrency::spawn_for_test(); let tline = tenant .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx) .await?; @@ -7622,8 +7637,9 @@ mod tests { keyspace: &KeySpace, lsn: Lsn, ctx: &RequestContext, + io_concurrency: IoConcurrency, ) -> anyhow::Result<(BTreeMap>, usize)> { - let mut reconstruct_state = ValuesReconstructState::new(IoConcurrency::sequential()); + let mut reconstruct_state = ValuesReconstructState::new(io_concurrency); let res = tline .get_vectored_impl(keyspace.clone(), lsn, &mut reconstruct_state, ctx) .await?; @@ -7671,7 +7687,8 @@ mod tests { if iter % 5 == 0 { let (_, before_delta_file_accessed) = - scan_with_statistics(&tline, &keyspace, lsn, &ctx).await?; + scan_with_statistics(&tline, &keyspace, lsn, &ctx, io_concurrency.clone()) + .await?; tline .compact( &cancel, @@ -7685,7 +7702,8 @@ mod tests { ) .await?; let (_, after_delta_file_accessed) = - scan_with_statistics(&tline, &keyspace, lsn, &ctx).await?; + scan_with_statistics(&tline, &keyspace, lsn, &ctx, io_concurrency.clone()) + .await?; assert!(after_delta_file_accessed < before_delta_file_accessed, "after_delta_file_accessed={after_delta_file_accessed}, before_delta_file_accessed={before_delta_file_accessed}"); // Given that we already produced an image layer, there should be no delta layer needed for the scan, but still setting a low threshold there for unforeseen circumstances. assert!( @@ -7774,6 +7792,7 @@ mod tests { async fn test_vectored_missing_metadata_key_reads() -> anyhow::Result<()> { let harness = TenantHarness::create("test_vectored_missing_metadata_key_reads").await?; let (tenant, ctx) = harness.load().await; + let io_concurrency = IoConcurrency::spawn_for_test(); let base_key = Key::from_hex("620000000033333333444444445500000000").unwrap(); let base_key_child = Key::from_hex("620000000033333333444444445500000001").unwrap(); @@ -7912,7 +7931,7 @@ mod tests { ); // test vectored scan on parent timeline - let mut reconstruct_state = ValuesReconstructState::new(IoConcurrency::sequential()); + let mut reconstruct_state = ValuesReconstructState::new(io_concurrency.clone()); let res = tline .get_vectored_impl( KeySpace::single(Key::metadata_key_range()), @@ -7938,7 +7957,7 @@ mod tests { ); // test vectored scan on child timeline - let mut reconstruct_state = ValuesReconstructState::new(IoConcurrency::sequential()); + let mut reconstruct_state = ValuesReconstructState::new(io_concurrency.clone()); let res = child .get_vectored_impl( KeySpace::single(Key::metadata_key_range()), @@ -7976,7 +7995,9 @@ mod tests { lsn: Lsn, ctx: &RequestContext, ) -> Result, GetVectoredError> { - let mut reconstruct_state = ValuesReconstructState::new(IoConcurrency::sequential()); + let io_concurrency = + IoConcurrency::spawn_from_conf(tline.conf, tline.gate.enter().unwrap()); + let mut reconstruct_state = ValuesReconstructState::new(io_concurrency); let mut res = tline .get_vectored_impl( KeySpace::single(key..key.next()), diff --git a/pageserver/src/tenant/storage_layer.rs b/pageserver/src/tenant/storage_layer.rs index 0fc7ddf675..5cfe8c115a 100644 --- a/pageserver/src/tenant/storage_layer.rs +++ b/pageserver/src/tenant/storage_layer.rs @@ -243,6 +243,7 @@ impl IoConcurrency { pub(crate) fn sequential() -> Self { Self::spawn(SelectedIoConcurrency::Sequential) } + pub(crate) fn spawn_from_conf( conf: &'static PageServerConf, gate_guard: GateGuard, @@ -444,6 +445,38 @@ impl IoConcurrency { } } } + + #[cfg(test)] + pub(crate) fn spawn_for_test() -> impl std::ops::DerefMut { + use std::ops::{Deref, DerefMut}; + use utils::sync::gate::Gate; + + // Spawn needs a Gate, give it one. + struct Wrapper { + inner: IoConcurrency, + #[allow(dead_code)] + gate: Box, + } + impl Deref for Wrapper { + type Target = IoConcurrency; + + fn deref(&self) -> &Self::Target { + &self.inner + } + } + impl DerefMut for Wrapper { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.inner + } + } + let gate = Box::new(Gate::default()); + Wrapper { + inner: Self::spawn(SelectedIoConcurrency::SidecarTask( + gate.enter().expect("just created it"), + )), + gate, + } + } } impl ValuesReconstructState { diff --git a/pageserver/src/tenant/storage_layer/layer/tests.rs b/pageserver/src/tenant/storage_layer/layer/tests.rs index 864509cd09..743631b30b 100644 --- a/pageserver/src/tenant/storage_layer/layer/tests.rs +++ b/pageserver/src/tenant/storage_layer/layer/tests.rs @@ -34,6 +34,7 @@ async fn smoke_test() { let span = h.span(); let download_span = span.in_scope(|| tracing::info_span!("downloading", timeline_id = 1)); let (tenant, _) = h.load().await; + let io_concurrency = IoConcurrency::spawn_for_test(); let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Download); @@ -92,7 +93,7 @@ async fn smoke_test() { }; let img_before = { - let mut data = ValuesReconstructState::new(IoConcurrency::sequential()); + let mut data = ValuesReconstructState::new(io_concurrency.clone()); layer .get_values_reconstruct_data( controlfile_keyspace.clone(), diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index 606e0fe6f7..100fb9d93a 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -1112,6 +1112,7 @@ impl Timeline { keyspace: KeySpace, lsn: Lsn, ctx: &RequestContext, + io_concurrency: super::storage_layer::IoConcurrency, ) -> Result>, GetVectoredError> { if !lsn.is_valid() { return Err(GetVectoredError::InvalidLsn(lsn)); @@ -1143,7 +1144,7 @@ impl Timeline { .get_vectored_impl( keyspace.clone(), lsn, - &mut ValuesReconstructState::new(IoConcurrency::sequential()), + &mut ValuesReconstructState::new(io_concurrency), ctx, ) .await; diff --git a/pageserver/src/walingest.rs b/pageserver/src/walingest.rs index ad7bcc0714..e0283d99e0 100644 --- a/pageserver/src/walingest.rs +++ b/pageserver/src/walingest.rs @@ -499,7 +499,13 @@ impl WalIngest { let content = modification .tline - .get_rel_page_at_lsn(src_rel, blknum, Version::Modified(modification), ctx) + .get_rel_page_at_lsn( + src_rel, + blknum, + Version::Modified(modification), + ctx, + crate::tenant::storage_layer::IoConcurrency::sequential(), + ) .await?; modification.put_rel_page_image(dst_rel, blknum, content)?; num_blocks_copied += 1; @@ -1489,6 +1495,7 @@ mod tests { use super::*; use crate::tenant::harness::*; use crate::tenant::remote_timeline_client::{remote_initdb_archive_path, INITDB_PATH}; + use crate::tenant::storage_layer::IoConcurrency; use postgres_ffi::RELSEG_SIZE; use crate::DEFAULT_PG_VERSION; @@ -1532,6 +1539,7 @@ mod tests { #[tokio::test] async fn test_relsize() -> Result<()> { let (tenant, ctx) = TenantHarness::create("test_relsize").await?.load().await; + let io_concurrency = IoConcurrency::spawn_for_test(); let tline = tenant .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx) .await?; @@ -1599,7 +1607,13 @@ mod tests { // Check page contents at each LSN assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 0, Version::Lsn(Lsn(0x20)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 0, + Version::Lsn(Lsn(0x20)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img("foo blk 0 at 2") @@ -1607,7 +1621,13 @@ mod tests { assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 0, Version::Lsn(Lsn(0x30)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 0, + Version::Lsn(Lsn(0x30)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img("foo blk 0 at 3") @@ -1615,14 +1635,26 @@ mod tests { assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 0, Version::Lsn(Lsn(0x40)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 0, + Version::Lsn(Lsn(0x40)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img("foo blk 0 at 3") ); assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 1, Version::Lsn(Lsn(0x40)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 1, + Version::Lsn(Lsn(0x40)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img("foo blk 1 at 4") @@ -1630,21 +1662,39 @@ mod tests { assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 0, Version::Lsn(Lsn(0x50)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 0, + Version::Lsn(Lsn(0x50)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img("foo blk 0 at 3") ); assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 1, Version::Lsn(Lsn(0x50)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 1, + Version::Lsn(Lsn(0x50)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img("foo blk 1 at 4") ); assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 2, Version::Lsn(Lsn(0x50)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 2, + Version::Lsn(Lsn(0x50)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img("foo blk 2 at 5") @@ -1667,14 +1717,26 @@ mod tests { ); assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 0, Version::Lsn(Lsn(0x60)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 0, + Version::Lsn(Lsn(0x60)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img("foo blk 0 at 3") ); assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 1, Version::Lsn(Lsn(0x60)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 1, + Version::Lsn(Lsn(0x60)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img("foo blk 1 at 4") @@ -1689,7 +1751,13 @@ mod tests { ); assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 2, Version::Lsn(Lsn(0x50)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 2, + Version::Lsn(Lsn(0x50)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img("foo blk 2 at 5") @@ -1722,14 +1790,26 @@ mod tests { ); assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 0, Version::Lsn(Lsn(0x70)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 0, + Version::Lsn(Lsn(0x70)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, ZERO_PAGE ); assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 1, Version::Lsn(Lsn(0x70)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 1, + Version::Lsn(Lsn(0x70)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img("foo blk 1") @@ -1750,7 +1830,13 @@ mod tests { for blk in 2..1500 { assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, blk, Version::Lsn(Lsn(0x80)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + blk, + Version::Lsn(Lsn(0x80)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, ZERO_PAGE @@ -1758,7 +1844,13 @@ mod tests { } assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, 1500, Version::Lsn(Lsn(0x80)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + 1500, + Version::Lsn(Lsn(0x80)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img("foo blk 1500") @@ -1851,6 +1943,7 @@ mod tests { .await? .load() .await; + let io_concurrency = IoConcurrency::spawn_for_test(); let tline = tenant .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx) .await?; @@ -1903,7 +1996,13 @@ mod tests { let data = format!("foo blk {} at {}", blkno, lsn); assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, blkno, Version::Lsn(lsn), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + blkno, + Version::Lsn(lsn), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img(&data) @@ -1931,7 +2030,13 @@ mod tests { let data = format!("foo blk {} at {}", blkno, lsn); assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, blkno, Version::Lsn(Lsn(0x60)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + blkno, + Version::Lsn(Lsn(0x60)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img(&data) @@ -1950,7 +2055,13 @@ mod tests { let data = format!("foo blk {} at {}", blkno, lsn); assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, blkno, Version::Lsn(Lsn(0x50)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + blkno, + Version::Lsn(Lsn(0x50)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img(&data) @@ -1987,7 +2098,13 @@ mod tests { let data = format!("foo blk {} at {}", blkno, lsn); assert_eq!( tline - .get_rel_page_at_lsn(TESTREL_A, blkno, Version::Lsn(Lsn(0x80)), &ctx) + .get_rel_page_at_lsn( + TESTREL_A, + blkno, + Version::Lsn(Lsn(0x80)), + &ctx, + io_concurrency.clone() + ) .instrument(test_span.clone()) .await?, test_img(&data)