diff --git a/safekeeper/src/broker.rs b/safekeeper/src/broker.rs index 2518a77e25..35ae746994 100644 --- a/safekeeper/src/broker.rs +++ b/safekeeper/src/broker.rs @@ -46,7 +46,7 @@ async fn push_loop(conf: SafeKeeperConf) -> anyhow::Result<()> { return Ok(()); } - let timelines_set = GlobalTimelines::get_global_broker_active_set(); + let active_timelines_set = GlobalTimelines::get_global_broker_active_set(); let mut client = storage_broker::connect(conf.broker_endpoint.clone(), conf.broker_keepalive_interval)?; @@ -59,7 +59,7 @@ async fn push_loop(conf: SafeKeeperConf) -> anyhow::Result<()> { // sensitive and there is no risk of deadlock as we don't await while // lock is held. let now = Instant::now(); - let all_tlis = timelines_set.get_all(); + let all_tlis = active_timelines_set.get_all(); let mut n_pushed_tlis = 0; for tli in &all_tlis { let sk_info = tli.get_safekeeper_info(&conf).await; diff --git a/safekeeper/src/metrics.rs b/safekeeper/src/metrics.rs index 12b6f6a7fd..044386d0a9 100644 --- a/safekeeper/src/metrics.rs +++ b/safekeeper/src/metrics.rs @@ -745,9 +745,9 @@ impl Collector for TimelineCollector { async fn collect_timeline_metrics() -> Vec { let mut res = vec![]; - let timelines = GlobalTimelines::get_global_broker_active_set().get_all(); + let active_timelines = GlobalTimelines::get_global_broker_active_set().get_all(); - for tli in timelines { + for tli in active_timelines { if let Some(info) = tli.info_for_metrics().await { res.push(info); } diff --git a/safekeeper/src/receive_wal.rs b/safekeeper/src/receive_wal.rs index e76ee6eeb5..03cfa882c4 100644 --- a/safekeeper/src/receive_wal.rs +++ b/safekeeper/src/receive_wal.rs @@ -113,7 +113,7 @@ impl WalReceivers { self.mutex.lock().get_num() } - /// Get number of walreceivers (compute connections). + /// Get channel for number of walreceivers. pub fn get_num_rx(self: &Arc) -> tokio::sync::watch::Receiver { self.num_computes_rx.clone() } diff --git a/safekeeper/src/timelines_global_map.rs b/safekeeper/src/timelines_global_map.rs index 6c9a2bae6b..8d37bd6371 100644 --- a/safekeeper/src/timelines_global_map.rs +++ b/safekeeper/src/timelines_global_map.rs @@ -12,6 +12,7 @@ use once_cell::sync::Lazy; use serde::Serialize; use std::collections::HashMap; use std::str::FromStr; +use std::sync::atomic::Ordering; use std::sync::{Arc, Mutex}; use tracing::*; use utils::id::{TenantId, TenantTimelineId, TimelineId}; @@ -327,6 +328,8 @@ impl GlobalTimelines { let tli_res = TIMELINES_STATE.lock().unwrap().get(ttid); match tli_res { Ok(timeline) => { + let was_active = timeline.broker_active.load(Ordering::Relaxed); + // Take a lock and finish the deletion holding this mutex. let mut shared_state = timeline.write_shared_state().await; @@ -340,7 +343,7 @@ impl GlobalTimelines { Ok(TimelineDeleteForceResult { dir_existed, - was_active: true, // TODO: we probably should remove this field + was_active, // TODO: we probably should remove this field }) } Err(_) => { diff --git a/safekeeper/src/timelines_set.rs b/safekeeper/src/timelines_set.rs index 8012bc6a65..9a48773447 100644 --- a/safekeeper/src/timelines_set.rs +++ b/safekeeper/src/timelines_set.rs @@ -4,7 +4,12 @@ use utils::id::TenantTimelineId; use crate::timeline::Timeline; -/// Set of timelines, used to keep subset of timelines used for some tasks (like WAL removal). +/// Set of timelines, supports operations: +/// - add timeline +/// - remove timeline +/// - clone the set +/// +/// Used for keeping subset of active (for which broker push is required) timelines. pub struct TimelinesSet { timelines: std::sync::Mutex>>, } @@ -26,6 +31,7 @@ impl TimelinesSet { self.timelines.lock().unwrap().remove(ttid); } + /// If present is true, adds timeline to the set, otherwise removes it. pub fn set_present(&self, tli: Arc, present: bool) { if present { self.insert(tli); @@ -38,10 +44,12 @@ impl TimelinesSet { self.timelines.lock().unwrap().contains_key(ttid) } + /// Returns all timelines in the set. pub fn get_all(&self) -> Vec> { self.timelines.lock().unwrap().values().cloned().collect() } + /// Returns a timeline guard for easy presence control. pub fn guard(self: &Arc, tli: Arc) -> TimelineSetGuard { let is_present = self.is_present(&tli.ttid); TimelineSetGuard { @@ -52,6 +60,10 @@ impl TimelinesSet { } } +/// Guard is used to add or remove timeline from the set. +/// If the timeline present in set, it will be removed from it on drop. +/// Note: do not use more than one guard for the same timeline, it caches the presence state. +/// It is designed to be used in the manager task only. pub struct TimelineSetGuard { timelines_set: Arc, tli: Arc, diff --git a/safekeeper/src/wal_backup.rs b/safekeeper/src/wal_backup.rs index f71c46d083..72e7c78326 100644 --- a/safekeeper/src/wal_backup.rs +++ b/safekeeper/src/wal_backup.rs @@ -47,7 +47,7 @@ pub struct WalBackupTaskHandle { handle: JoinHandle<()>, } -/// Do we have anything to upload to S3, i.e. should we run the backup task? +/// Do we have anything to upload to S3, i.e. should safekeepers run backup activity? pub fn is_wal_backup_required( wal_seg_size: usize, num_computes: usize, @@ -68,11 +68,16 @@ pub async fn update_task( state: &StateSnapshot, entry: &mut Option, ) { - let (offloader, election_dbg_str) = - determine_offloader(&state.peers, state.backup_lsn, ttid, conf); - let elected_me = Some(conf.my_id) == offloader; - let should_task_run = need_backup && elected_me; + let (should_task_run, election_dbg_str) = if need_backup { + let (offloader, election_dbg_str) = + determine_offloader(&state.peers, state.backup_lsn, ttid, conf); + let elected_me = Some(conf.my_id) == offloader; + (elected_me, election_dbg_str) + } else { + (false, String::new()) + }; + // start or stop the task if should_task_run != (entry.is_some()) { if should_task_run { info!("elected for backup: {}", election_dbg_str);