diff --git a/pageserver/src/consumption_metrics.rs b/pageserver/src/consumption_metrics.rs index 23cfc55b65..9d1c252f00 100644 --- a/pageserver/src/consumption_metrics.rs +++ b/pageserver/src/consumption_metrics.rs @@ -150,7 +150,7 @@ pub async fn collect_metrics_iteration( let mut tenant_resident_size = 0; // iterate through list of timelines in tenant - for timeline in tenant.list_timelines().iter() { + for timeline in tenant.list_timelines().await.iter() { // collect per-timeline metrics only for active timelines if timeline.is_active() { let timeline_written_size = u64::from(timeline.get_last_record_lsn()); diff --git a/pageserver/src/disk_usage_eviction_task.rs b/pageserver/src/disk_usage_eviction_task.rs index f4a0f3f18e..7f8691d81e 100644 --- a/pageserver/src/disk_usage_eviction_task.rs +++ b/pageserver/src/disk_usage_eviction_task.rs @@ -508,7 +508,7 @@ async fn collect_eviction_candidates( // a little unfair to tenants during shutdown in such a situation is tolerable. let mut tenant_candidates = Vec::new(); let mut max_layer_size = 0; - for tl in tenant.list_timelines() { + for tl in tenant.list_timelines().await { if !tl.is_active() { continue; } diff --git a/pageserver/src/http/routes.rs b/pageserver/src/http/routes.rs index 3c87e6f79d..e7192f5589 100644 --- a/pageserver/src/http/routes.rs +++ b/pageserver/src/http/routes.rs @@ -300,7 +300,7 @@ async fn timeline_list_handler(request: Request) -> Result, let response_data = async { let tenant = mgr::get_tenant(tenant_id, true).await?; - let timelines = tenant.list_timelines(); + let timelines = tenant.list_timelines().await; let mut response_data = Vec::with_capacity(timelines.len()); for timeline in timelines { @@ -339,6 +339,7 @@ async fn timeline_detail_handler(request: Request) -> Result) -> Result, ApiErro // Calculate total physical size of all timelines let mut current_physical_size = 0; - for timeline in tenant.list_timelines().iter() { + for timeline in tenant.list_timelines().await.iter() { current_physical_size += timeline.layer_size_sum(); } @@ -932,6 +933,7 @@ async fn active_timeline_of_active_tenant( let tenant = mgr::get_tenant(tenant_id, true).await?; tenant .get_timeline(timeline_id, true) + .await .map_err(ApiError::NotFound) } diff --git a/pageserver/src/page_service.rs b/pageserver/src/page_service.rs index bd3ece2dfc..020df17108 100644 --- a/pageserver/src/page_service.rs +++ b/pageserver/src/page_service.rs @@ -376,7 +376,7 @@ impl PageServerHandler { }; // Check that the timeline exists - let timeline = tenant.get_timeline(timeline_id, true)?; + let timeline = tenant.get_timeline(timeline_id, true).await?; // switch client to COPYBOTH pgb.write_message_noflush(&BeMessage::CopyBothResponse)?; @@ -475,7 +475,9 @@ impl PageServerHandler { // Create empty timeline info!("creating new timeline"); let tenant = get_active_tenant_with_timeout(tenant_id, &ctx).await?; - let timeline = tenant.create_empty_timeline(timeline_id, base_lsn, pg_version, &ctx)?; + let timeline = tenant + .create_empty_timeline(timeline_id, base_lsn, pg_version, &ctx) + .await?; // TODO mark timeline as not ready until it reaches end_lsn. // We might have some wal to import as well, and we should prevent compute @@ -1184,6 +1186,6 @@ async fn get_active_tenant_timeline( ctx: &RequestContext, ) -> Result, GetActiveTenantError> { let tenant = get_active_tenant_with_timeout(tenant_id, ctx).await?; - let timeline = tenant.get_timeline(timeline_id, true)?; + let timeline = tenant.get_timeline(timeline_id, true).await?; Ok(timeline) } diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs index 6080aaa39b..07206cfc7f 100644 --- a/pageserver/src/tenant.rs +++ b/pageserver/src/tenant.rs @@ -41,8 +41,7 @@ use std::process::Stdio; use std::sync::atomic::AtomicU64; use std::sync::atomic::Ordering; use std::sync::Arc; -use std::sync::MutexGuard; -use std::sync::{Mutex, RwLock}; +use std::sync::RwLock; use std::time::{Duration, Instant}; use self::config::TenantConf; @@ -136,7 +135,7 @@ pub struct Tenant { tenant_conf: Arc>, tenant_id: TenantId, - timelines: Mutex>>, + timelines: tokio::sync::Mutex>>, // This mutex prevents creation of new timelines during GC. // Adding yet another mutex (in addition to `timelines`) is needed because holding // `timelines` mutex during all GC iteration @@ -186,8 +185,8 @@ impl UninitializedTimeline<'_> { /// /// The new timeline is initialized in Active state, and its background jobs are /// started - pub fn initialize(self, ctx: &RequestContext) -> anyhow::Result> { - let mut timelines = self.owning_tenant.timelines.lock().unwrap(); + pub async fn initialize(self, ctx: &RequestContext) -> anyhow::Result> { + let mut timelines = self.owning_tenant.timelines.lock().await; self.initialize_with_lock(ctx, &mut timelines, true, true) } @@ -278,7 +277,7 @@ impl UninitializedTimeline<'_> { // Initialize without loading the layer map. We started with an empty layer map, and already // updated it for the layers that we created during the import. - let mut timelines = self.owning_tenant.timelines.lock().unwrap(); + let mut timelines = self.owning_tenant.timelines.lock().await; self.initialize_with_lock(ctx, &mut timelines, false, true) } @@ -497,7 +496,7 @@ impl Tenant { let timeline = { // avoiding holding it across awaits - let mut timelines_accessor = self.timelines.lock().unwrap(); + let mut timelines_accessor = self.timelines.lock().await; if timelines_accessor.contains_key(&timeline_id) { anyhow::bail!( "Timeline {tenant_id}/{timeline_id} already exists in the tenant map" @@ -802,7 +801,7 @@ impl Tenant { // Start background operations and open the tenant for business. // The loops will shut themselves down when they notice that the tenant is inactive. - self.activate(&ctx)?; + self.activate(&ctx).await?; info!("Done"); @@ -816,7 +815,7 @@ impl Tenant { pub async fn get_remote_size(&self) -> anyhow::Result { let mut size = 0; - for timeline in self.list_timelines().iter() { + for timeline in self.list_timelines().await.iter() { if let Some(remote_client) = &timeline.remote_client { size += remote_client.get_remote_physical_size(); } @@ -840,7 +839,7 @@ impl Tenant { .context("Failed to create new timeline directory")?; let ancestor = if let Some(ancestor_id) = remote_metadata.ancestor_timeline() { - let timelines = self.timelines.lock().unwrap(); + let timelines = self.timelines.lock().await; Some(Arc::clone(timelines.get(&ancestor_id).ok_or_else( || { anyhow::anyhow!( @@ -1081,7 +1080,7 @@ impl Tenant { // Start background operations and open the tenant for business. // The loops will shut themselves down when they notice that the tenant is inactive. - self.activate(ctx)?; + self.activate(ctx).await?; info!("Done"); @@ -1149,7 +1148,7 @@ impl Tenant { }; let ancestor = if let Some(ancestor_timeline_id) = local_metadata.ancestor_timeline() { - let ancestor_timeline = self.get_timeline(ancestor_timeline_id, false) + let ancestor_timeline = self.get_timeline(ancestor_timeline_id, false).await .with_context(|| anyhow::anyhow!("cannot find ancestor timeline {ancestor_timeline_id} for timeline {timeline_id}"))?; Some(ancestor_timeline) } else { @@ -1174,12 +1173,12 @@ impl Tenant { /// Get Timeline handle for given Neon timeline ID. /// This function is idempotent. It doesn't change internal state in any way. - pub fn get_timeline( + pub async fn get_timeline( &self, timeline_id: TimelineId, active_only: bool, ) -> anyhow::Result> { - let timelines_accessor = self.timelines.lock().unwrap(); + let timelines_accessor = self.timelines.lock().await; let timeline = timelines_accessor.get(&timeline_id).with_context(|| { format!("Timeline {}/{} was not found", self.tenant_id, timeline_id) })?; @@ -1198,10 +1197,10 @@ impl Tenant { /// Lists timelines the tenant contains. /// Up to tenant's implementation to omit certain timelines that ar not considered ready for use. - pub fn list_timelines(&self) -> Vec> { + pub async fn list_timelines(&self) -> Vec> { self.timelines .lock() - .unwrap() + .await .values() .map(Arc::clone) .collect() @@ -1210,7 +1209,7 @@ impl Tenant { /// This is used to create the initial 'main' timeline during bootstrapping, /// or when importing a new base backup. The caller is expected to load an /// initial image of the datadir to the new timeline after this. - pub fn create_empty_timeline( + pub async fn create_empty_timeline( &self, new_timeline_id: TimelineId, initdb_lsn: Lsn, @@ -1222,7 +1221,7 @@ impl Tenant { "Cannot create empty timelines on inactive tenant" ); - let timelines = self.timelines.lock().unwrap(); + let timelines = self.timelines.lock().await; let timeline_uninit_mark = self.create_timeline_uninit_mark(new_timeline_id, &timelines)?; drop(timelines); @@ -1264,7 +1263,7 @@ impl Tenant { "Cannot create timelines on inactive tenant" ); - if self.get_timeline(new_timeline_id, false).is_ok() { + if self.get_timeline(new_timeline_id, false).await.is_ok() { debug!("timeline {new_timeline_id} already exists"); return Ok(None); } @@ -1273,6 +1272,7 @@ impl Tenant { Some(ancestor_timeline_id) => { let ancestor_timeline = self .get_timeline(ancestor_timeline_id, false) + .await .context("Cannot branch off the timeline that's not present in pageserver")?; if let Some(lsn) = ancestor_start_lsn.as_mut() { @@ -1354,7 +1354,7 @@ impl Tenant { // compactions. We don't want to block everything else while the // compaction runs. let timelines_to_compact = { - let timelines = self.timelines.lock().unwrap(); + let timelines = self.timelines.lock().await; let timelines_to_compact = timelines .iter() .map(|(timeline_id, timeline)| (*timeline_id, timeline.clone())) @@ -1383,7 +1383,7 @@ impl Tenant { // flushing. We don't want to block everything else while the // flushing is performed. let timelines_to_flush = { - let timelines = self.timelines.lock().unwrap(); + let timelines = self.timelines.lock().await; timelines .iter() .map(|(_id, timeline)| Arc::clone(timeline)) @@ -1408,7 +1408,7 @@ impl Tenant { // Transition the timeline into TimelineState::Stopping. // This should prevent new operations from starting. let timeline = { - let mut timelines = self.timelines.lock().unwrap(); + let mut timelines = self.timelines.lock().await; // Ensure that there are no child timelines **attached to that pageserver**, // because detach removes files, which will break child branches @@ -1555,7 +1555,7 @@ impl Tenant { }); // Remove the timeline from the map. - let mut timelines = self.timelines.lock().unwrap(); + let mut timelines = self.timelines.lock().await; let children_exist = timelines .iter() .any(|(_, entry)| entry.get_ancestor_timeline_id() == Some(timeline_id)); @@ -1599,7 +1599,7 @@ impl Tenant { } /// Changes tenant status to active, unless shutdown was already requested. - fn activate(&self, ctx: &RequestContext) -> anyhow::Result<()> { + pub async fn activate(&self, ctx: &RequestContext) -> anyhow::Result<()> { let mut result = Ok(()); Self::state_send_modify_async(&self.state, |current_state| { match &*current_state { @@ -1627,7 +1627,7 @@ impl Tenant { Either::Right( async move { debug!(tenant_id = %self.tenant_id, "Activating tenant"); - let timelines_accessor = self.timelines.lock().unwrap(); + let timelines_accessor = self.timelines.lock().await; let not_broken_timelines = timelines_accessor.values().filter(|timeline| { timeline.current_state() != TimelineState::Broken @@ -1700,7 +1700,7 @@ impl Tenant { // might be created after this. That's harmless, as the Timelines // won't be accessible to anyone, when the Tenant is in Stopping // state. - let timelines_accessor = self.timelines.lock().unwrap(); + let timelines_accessor = self.timelines.lock().await; let not_broken_timelines = timelines_accessor .values() .filter(|timeline| timeline.current_state() != TimelineState::Broken); @@ -1948,12 +1948,12 @@ impl Tenant { .or(self.conf.default_tenant_conf.min_resident_size_override) } - pub fn set_new_tenant_config(&self, new_tenant_conf: TenantConfOpt) { + pub async fn set_new_tenant_config(&self, new_tenant_conf: TenantConfOpt) { *self.tenant_conf.write().unwrap() = new_tenant_conf; // Don't hold self.timelines.lock() during the notifies. // There's no risk of deadlock right now, but there could be if we consolidate // mutexes in struct Timeline in the future. - let timelines = self.list_timelines(); + let timelines = self.list_timelines().await; for timeline in timelines { timeline.tenant_conf_updated(); } @@ -2031,7 +2031,7 @@ impl Tenant { // activation times. loading_started_at: Instant::now(), tenant_conf: Arc::new(RwLock::new(tenant_conf)), - timelines: Mutex::new(HashMap::new()), + timelines: tokio::sync::Mutex::new(HashMap::new()), gc_cs: tokio::sync::Mutex::new(()), walredo_mgr, remote_storage, @@ -2259,7 +2259,7 @@ impl Tenant { // Scan all timelines. For each timeline, remember the timeline ID and // the branch point where it was created. let (all_branchpoints, timeline_ids): (BTreeSet<(TimelineId, Lsn)>, _) = { - let timelines = self.timelines.lock().unwrap(); + let timelines = self.timelines.lock().await; let mut all_branchpoints = BTreeSet::new(); let timeline_ids = { if let Some(target_timeline_id) = target_timeline_id.as_ref() { @@ -2306,6 +2306,7 @@ impl Tenant { // Timeline is known to be local and loaded. let timeline = self .get_timeline(timeline_id, false) + .await .with_context(|| format!("Timeline {timeline_id} was not found"))?; // If target_timeline is specified, ignore all other timelines @@ -2359,7 +2360,7 @@ impl Tenant { // Create a placeholder for the new branch. This will error // out if the new timeline ID is already in use. let timeline_uninit_mark = { - let timelines = self.timelines.lock().unwrap(); + let timelines = self.timelines.lock().await; self.create_timeline_uninit_mark(dst_id, &timelines)? }; @@ -2424,7 +2425,7 @@ impl Tenant { src_timeline.initdb_lsn, src_timeline.pg_version, ); - let mut timelines = self.timelines.lock().unwrap(); + let mut timelines = self.timelines.lock().await; let new_timeline = self .prepare_timeline( dst_id, @@ -2461,7 +2462,7 @@ impl Tenant { ctx: &RequestContext, ) -> anyhow::Result> { let timeline_uninit_mark = { - let timelines = self.timelines.lock().unwrap(); + let timelines = self.timelines.lock().await; self.create_timeline_uninit_mark(timeline_id, &timelines)? }; // create a `tenant/{tenant_id}/timelines/basebackup-{timeline_id}.{TEMP_FILE_SUFFIX}/` @@ -2547,7 +2548,7 @@ impl Tenant { // Initialize the timeline without loading the layer map, because we already updated the layer // map above, when we imported the datadir. let timeline = { - let mut timelines = self.timelines.lock().unwrap(); + let mut timelines = self.timelines.lock().await; raw_timeline.initialize_with_lock(ctx, &mut timelines, false, true)? }; @@ -2650,7 +2651,7 @@ impl Tenant { fn create_timeline_uninit_mark( &self, timeline_id: TimelineId, - timelines: &MutexGuard>>, + timelines: &tokio::sync::MutexGuard>>, ) -> anyhow::Result { let tenant_id = self.tenant_id; diff --git a/pageserver/src/tenant/mgr.rs b/pageserver/src/tenant/mgr.rs index 8191880bb5..0ac8c9193f 100644 --- a/pageserver/src/tenant/mgr.rs +++ b/pageserver/src/tenant/mgr.rs @@ -310,7 +310,7 @@ pub async fn set_new_tenant_config( new_tenant_conf, false, )?; - tenant.set_new_tenant_config(new_tenant_conf); + tenant.set_new_tenant_config(new_tenant_conf).await; Ok(()) } @@ -658,6 +658,7 @@ pub async fn immediate_compact( let timeline = tenant .get_timeline(timeline_id, true) + .await .map_err(ApiError::NotFound)?; // Run in task_mgr to avoid race with tenant_detach operation diff --git a/pageserver/src/tenant/size.rs b/pageserver/src/tenant/size.rs index 9b2e5dc4f1..d05c103573 100644 --- a/pageserver/src/tenant/size.rs +++ b/pageserver/src/tenant/size.rs @@ -135,7 +135,7 @@ pub(super) async fn gather_inputs( .context("Failed to refresh gc_info before gathering inputs")?; // Collect information about all the timelines - let mut timelines = tenant.list_timelines(); + let mut timelines = tenant.list_timelines().await; if timelines.is_empty() { // perhaps the tenant has just been created, and as such doesn't have any data yet