diff --git a/safekeeper/src/broker.rs b/safekeeper/src/broker.rs index ea16ce450f..2518a77e25 100644 --- a/safekeeper/src/broker.rs +++ b/safekeeper/src/broker.rs @@ -46,6 +46,8 @@ async fn push_loop(conf: SafeKeeperConf) -> anyhow::Result<()> { return Ok(()); } + let timelines_set = GlobalTimelines::get_global_broker_active_set(); + let mut client = storage_broker::connect(conf.broker_endpoint.clone(), conf.broker_keepalive_interval)?; let push_interval = Duration::from_millis(PUSH_INTERVAL_MSEC); @@ -57,15 +59,9 @@ 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 = GlobalTimelines::get_all(); + let all_tlis = timelines_set.get_all(); let mut n_pushed_tlis = 0; for tli in &all_tlis { - // filtering alternative futures::stream::iter(all_tlis) - // .filter(|tli| {let tli = tli.clone(); async move { tli.is_active().await}}).collect::>().await; - // doesn't look better, and I'm not sure how to do that without collect. - if !tli.is_active().await { - continue; - } let sk_info = tli.get_safekeeper_info(&conf).await; yield sk_info; BROKER_PUSHED_UPDATES.inc(); diff --git a/safekeeper/src/lib.rs b/safekeeper/src/lib.rs index 4429f94fc2..8d8d2cf23e 100644 --- a/safekeeper/src/lib.rs +++ b/safekeeper/src/lib.rs @@ -32,6 +32,7 @@ pub mod send_wal; pub mod state; pub mod timeline; pub mod timeline_manager; +pub mod timelines_set; pub mod wal_backup; pub mod wal_backup_partial; pub mod wal_service; diff --git a/safekeeper/src/metrics.rs b/safekeeper/src/metrics.rs index 28ae042bb3..12b6f6a7fd 100644 --- a/safekeeper/src/metrics.rs +++ b/safekeeper/src/metrics.rs @@ -614,8 +614,7 @@ impl Collector for TimelineCollector { self.written_wal_seconds.reset(); self.flushed_wal_seconds.reset(); - let timelines = GlobalTimelines::get_all(); - let timelines_count = timelines.len(); + let timelines_count = GlobalTimelines::get_all().len(); let mut active_timelines_count = 0; // Prometheus Collector is sync, and data is stored under async lock. To @@ -746,7 +745,7 @@ impl Collector for TimelineCollector { async fn collect_timeline_metrics() -> Vec { let mut res = vec![]; - let timelines = GlobalTimelines::get_all(); + let timelines = GlobalTimelines::get_global_broker_active_set().get_all(); for tli in timelines { if let Some(info) = tli.info_for_metrics().await { diff --git a/safekeeper/src/receive_wal.rs b/safekeeper/src/receive_wal.rs index 0f213abb4d..e76ee6eeb5 100644 --- a/safekeeper/src/receive_wal.rs +++ b/safekeeper/src/receive_wal.rs @@ -486,14 +486,7 @@ impl WalAcceptor { /// The main loop. Returns Ok(()) if either msg_rx or reply_tx got closed; /// it must mean that network thread terminated. async fn run(&mut self) -> anyhow::Result<()> { - // Register the connection and defer unregister. - // Order of the next two lines is important: we want first to remove our entry and then - // update status which depends on registered connections. - let _compute_conn_guard = ComputeConnectionGuard { - timeline: Arc::clone(&self.tli), - }; let walreceiver_guard = self.tli.get_walreceivers().register(self.conn_id); - self.tli.update_status_notify().await?; // After this timestamp we will stop processing AppendRequests and send a response // to the walproposer. walproposer sends at least one AppendRequest per second, @@ -559,19 +552,3 @@ impl WalAcceptor { } } } - -/// Calls update_status_notify in drop to update timeline status. -struct ComputeConnectionGuard { - timeline: Arc, -} - -impl Drop for ComputeConnectionGuard { - fn drop(&mut self) { - let tli = self.timeline.clone(); - tokio::spawn(async move { - if let Err(e) = tli.update_status_notify().await { - error!("failed to update timeline status: {}", e); - } - }); - } -} diff --git a/safekeeper/src/remove_wal.rs b/safekeeper/src/remove_wal.rs index 03f035da0a..98ce671182 100644 --- a/safekeeper/src/remove_wal.rs +++ b/safekeeper/src/remove_wal.rs @@ -7,23 +7,12 @@ use tracing::*; use crate::{GlobalTimelines, SafeKeeperConf}; -const ALLOW_INACTIVE_TIMELINES: bool = true; - pub async fn task_main(_conf: SafeKeeperConf) -> anyhow::Result<()> { let wal_removal_interval = Duration::from_millis(5000); loop { let now = tokio::time::Instant::now(); - let mut active_timelines = 0; - let tlis = GlobalTimelines::get_all(); for tli in &tlis { - let is_active = tli.is_active().await; - if is_active { - active_timelines += 1; - } - if !ALLOW_INACTIVE_TIMELINES && !is_active { - continue; - } let ttid = tli.ttid; async { if let Err(e) = tli.maybe_persist_control_file().await { @@ -42,8 +31,8 @@ pub async fn task_main(_conf: SafeKeeperConf) -> anyhow::Result<()> { if elapsed > wal_removal_interval { info!( - "WAL removal is too long, processed {} active timelines ({} total) in {:?}", - active_timelines, total_timelines, elapsed + "WAL removal is too long, processed {} timelines in {:?}", + total_timelines, elapsed ); } diff --git a/safekeeper/src/timeline.rs b/safekeeper/src/timeline.rs index 7102038494..1b22f1747e 100644 --- a/safekeeper/src/timeline.rs +++ b/safekeeper/src/timeline.rs @@ -9,6 +9,7 @@ use tokio::fs; use std::cmp::max; use std::ops::{Deref, DerefMut}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Duration; use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard}; @@ -31,6 +32,7 @@ use crate::safekeeper::{ }; use crate::send_wal::WalSenders; use crate::state::{TimelineMemState, TimelinePersistentState}; +use crate::timelines_set::TimelinesSet; use crate::wal_backup::{self}; use crate::{control_file, safekeeper::UNKNOWN_SERVER_VERSION}; @@ -157,16 +159,6 @@ pub struct SharedState { pub(crate) sk: SafeKeeper, /// In memory list containing state of peers sent in latest messages from them. pub(crate) peers_info: PeersInfo, - /// True whenever there is at least some pending activity on timeline: live - /// compute connection, pageserver is not caughtup (it must have latest WAL - /// for new compute start) or WAL backuping is not finished. Practically it - /// means safekeepers broadcast info to peers about the timeline, old WAL is - /// trimmed. - /// FIXME: this is not true if WAL backuping is not finished - /// - /// TODO: it might be better to remove tli completely from GlobalTimelines - /// when tli is inactive instead of having this flag. - pub(crate) active: bool, pub(crate) last_removed_segno: XLogSegNo, } @@ -204,7 +196,6 @@ impl SharedState { Ok(Self { sk, peers_info: PeersInfo(vec![]), - active: false, last_removed_segno: 0, }) } @@ -222,39 +213,10 @@ impl SharedState { Ok(Self { sk: SafeKeeper::new(control_store, wal_store, conf.my_id)?, peers_info: PeersInfo(vec![]), - active: false, last_removed_segno: 0, }) } - fn is_active(&self) -> bool { - // FIXME: add tracking of relevant pageservers and check them here individually, - // otherwise migration won't work (we suspend too early). - true || self.sk.state.inmem.remote_consistent_lsn < self.sk.state.inmem.commit_lsn - } - - /// Mark timeline active/inactive. - /// If timeline is deactivated, control file is persisted - /// as maintenance task does that only for active timelines. - async fn update_status(&mut self, ttid: TenantTimelineId) { - let is_active = self.is_active(); - if self.active != is_active { - info!( - "timeline {} active={} now, remote_consistent_lsn={}, commit_lsn={}", - ttid, - is_active, - self.sk.state.inmem.remote_consistent_lsn, - self.sk.state.inmem.commit_lsn - ); - if !is_active { - if let Err(e) = self.sk.state.flush().await { - warn!("control file save in update_status failed: {:?}", e); - } - } - } - self.active = is_active; - } - fn get_wal_seg_size(&self) -> usize { self.sk.state.server.wal_seg_size as usize } @@ -391,6 +353,10 @@ pub struct Timeline { /// with different speed. // TODO: add `Arc` here instead of adding each field separately. walsenders_keep_horizon: bool, + + // timeline_manager controlled state + pub(crate) broker_active: AtomicBool, + pub(crate) wal_backup_active: AtomicBool, } impl Timeline { @@ -424,6 +390,8 @@ impl Timeline { cancellation_tx, timeline_dir: conf.timeline_dir(&ttid), walsenders_keep_horizon: conf.walsenders_keep_horizon, + broker_active: AtomicBool::new(false), + wal_backup_active: AtomicBool::new(false), }) } @@ -460,6 +428,8 @@ impl Timeline { cancellation_tx, timeline_dir: conf.timeline_dir(&ttid), walsenders_keep_horizon: conf.walsenders_keep_horizon, + broker_active: AtomicBool::new(false), + wal_backup_active: AtomicBool::new(false), }) } @@ -472,6 +442,7 @@ impl Timeline { self: &Arc, shared_state: &mut WriteGuardSharedState<'_>, conf: &SafeKeeperConf, + broker_active_set: Arc, ) -> Result<()> { match fs::metadata(&self.timeline_dir).await { Ok(_) => { @@ -502,15 +473,23 @@ impl Timeline { return Err(e); } - self.bootstrap(conf); + self.bootstrap(conf, broker_active_set); Ok(()) } /// Bootstrap new or existing timeline starting background stasks. - pub fn bootstrap(self: &Arc, conf: &SafeKeeperConf) { + pub fn bootstrap( + self: &Arc, + conf: &SafeKeeperConf, + broker_active_set: Arc, + ) { // Start manager task which will monitor timeline state and update // background tasks. - tokio::spawn(timeline_manager::main_task(self.clone(), conf.clone())); + tokio::spawn(timeline_manager::main_task( + self.clone(), + conf.clone(), + broker_active_set, + )); // Start recovery task which always runs on the timeline. if conf.peer_recovery_enabled { @@ -531,8 +510,7 @@ impl Timeline { &self, shared_state: &mut WriteGuardSharedState<'_>, only_local: bool, - ) -> Result<(bool, bool)> { - let was_active = shared_state.active; + ) -> Result { self.cancel(shared_state); // TODO: It's better to wait for s3 offloader termination before @@ -546,7 +524,7 @@ impl Timeline { wal_backup::delete_timeline(&self.ttid).await?; } let dir_existed = delete_dir(&self.timeline_dir).await?; - Ok((dir_existed, was_active)) + Ok(dir_existed) } /// Cancel timeline to prevent further usage. Background tasks will stop @@ -583,22 +561,6 @@ impl Timeline { self.mutex.read().await } - async fn update_status(&self, shared_state: &mut SharedState) { - shared_state.update_status(self.ttid).await - } - - /// Update timeline status. - pub async fn update_status_notify(self: &Arc) -> Result<()> { - if self.is_cancelled() { - bail!(TimelineError::Cancelled(self.ttid)); - } - { - let mut shared_state = self.write_shared_state().await; - self.update_status(&mut shared_state).await; - } - Ok(()) - } - /// Returns true if walsender should stop sending WAL to pageserver. We /// terminate it if remote_consistent_lsn reached commit_lsn and there is no /// computes. While there might be nothing to stream already, we learn about @@ -671,15 +633,6 @@ impl Timeline { self.read_shared_state().await.get_wal_seg_size() } - /// Returns true only if the timeline is loaded and active. - pub async fn is_active(&self) -> bool { - if self.is_cancelled() { - return false; - } - - self.read_shared_state().await.active - } - /// Returns state of the timeline. pub async fn get_state(&self) -> (TimelineMemState, TimelinePersistentState) { let state = self.read_shared_state().await; @@ -877,8 +830,7 @@ impl Timeline { .await } - /// Gather timeline data for metrics. If the timeline is not active, returns - /// None, we do not collect these. + /// Gather timeline data for metrics. pub async fn info_for_metrics(&self) -> Option { if self.is_cancelled() { return None; @@ -886,24 +838,20 @@ impl Timeline { let (ps_feedback_count, last_ps_feedback) = self.walsenders.get_ps_feedback_stats(); let state = self.read_shared_state().await; - if state.active { - Some(FullTimelineInfo { - ttid: self.ttid, - ps_feedback_count, - last_ps_feedback, - wal_backup_active: false, // TODO: fill this - timeline_is_active: state.active, - num_computes: self.walreceivers.get_num() as u32, - last_removed_segno: state.last_removed_segno, - epoch_start_lsn: state.sk.epoch_start_lsn, - mem_state: state.sk.state.inmem.clone(), - persisted_state: state.sk.state.clone(), - flush_lsn: state.sk.wal_store.flush_lsn(), - wal_storage: state.sk.wal_store.get_metrics(), - }) - } else { - None - } + Some(FullTimelineInfo { + ttid: self.ttid, + ps_feedback_count, + last_ps_feedback, + wal_backup_active: self.wal_backup_active.load(Ordering::Relaxed), + timeline_is_active: self.broker_active.load(Ordering::Relaxed), + num_computes: self.walreceivers.get_num() as u32, + last_removed_segno: state.last_removed_segno, + epoch_start_lsn: state.sk.epoch_start_lsn, + mem_state: state.sk.state.inmem.clone(), + persisted_state: state.sk.state.clone(), + flush_lsn: state.sk.wal_store.flush_lsn(), + wal_storage: state.sk.wal_store.get_metrics(), + }) } /// Returns in-memory timeline state to build a full debug dump. @@ -917,8 +865,8 @@ impl Timeline { is_cancelled: self.is_cancelled(), peers_info_len: state.peers_info.0.len(), walsenders: self.walsenders.get_all(), - wal_backup_active: false, // TODO: fill this - active: state.active, + wal_backup_active: self.wal_backup_active.load(Ordering::Relaxed), + active: self.broker_active.load(Ordering::Relaxed), num_computes: self.walreceivers.get_num() as u32, last_removed_segno: state.last_removed_segno, epoch_start_lsn: state.sk.epoch_start_lsn, diff --git a/safekeeper/src/timeline_manager.rs b/safekeeper/src/timeline_manager.rs index 43bb7b3b07..698c42e8d7 100644 --- a/safekeeper/src/timeline_manager.rs +++ b/safekeeper/src/timeline_manager.rs @@ -5,6 +5,7 @@ use utils::lsn::Lsn; use crate::{ timeline::{PeerInfo, ReadGuardSharedState, Timeline}, + timelines_set::TimelinesSet, wal_backup::{self, WalBackupTaskHandle}, SafeKeeperConf, }; @@ -12,6 +13,7 @@ use crate::{ pub struct StateSnapshot { pub commit_lsn: Lsn, pub backup_lsn: Lsn, + pub remote_consistent_lsn: Lsn, pub peers: Vec, } @@ -20,6 +22,7 @@ impl StateSnapshot { Self { commit_lsn: read_guard.sk.state.inmem.commit_lsn, backup_lsn: read_guard.sk.state.inmem.backup_lsn, + remote_consistent_lsn: read_guard.sk.state.inmem.remote_consistent_lsn, peers: read_guard.get_peers(heartbeat_timeout), } } @@ -32,7 +35,11 @@ const REFRESH_INTERVAL: Duration = Duration::from_millis(300); /// This task gets spawned alongside each timeline and is responsible for managing the timeline's /// background tasks. #[instrument(name = "manager", skip_all, fields(ttid = %tli.ttid))] -pub async fn main_task(tli: Arc, conf: SafeKeeperConf) { +pub async fn main_task( + tli: Arc, + conf: SafeKeeperConf, + broker_active_set: Arc, +) { let mut cancellation_rx = match tli.get_cancellation_rx() { Ok(rx) => rx, Err(_) => { @@ -49,6 +56,9 @@ pub async fn main_task(tli: Arc, conf: SafeKeeperConf) { } }; + // sets whether timeline is active for broker pushes or not + let mut tli_broker_active = broker_active_set.guard(tli.clone()); + let ttid = tli.ttid; let wal_seg_size = tli.get_wal_seg_size().await; let heartbeat_timeout = conf.heartbeat_timeout; @@ -65,19 +75,48 @@ pub async fn main_task(tli: Arc, conf: SafeKeeperConf) { let state_snapshot = StateSnapshot::new(tli.read_shared_state().await, heartbeat_timeout); let num_computes = *num_computes_rx.borrow(); + let is_wal_backup_required = + wal_backup::is_wal_backup_required(wal_seg_size, num_computes, &state_snapshot); + if conf.is_wal_backup_enabled() { wal_backup::update_task( &conf, ttid, - wal_seg_size, - num_computes, + is_wal_backup_required, &state_snapshot, &mut backup_task, ) .await; } - // sleeping to make the loop less busy + // FIXME: add tracking of relevant pageservers and check them here individually, + // otherwise migration won't work (we suspend too early). + let is_active = is_wal_backup_required + || state_snapshot.remote_consistent_lsn < state_snapshot.commit_lsn; + + // update the broker map + if tli_broker_active.set(is_active) { + // write log if state has changed + info!( + "timeline {} active={} now, remote_consistent_lsn={}, commit_lsn={}", + ttid, is_active, state_snapshot.remote_consistent_lsn, state_snapshot.commit_lsn, + ); + + if !is_active { + // TODO: maybe use tokio::spawn? + if let Err(e) = tli.maybe_persist_control_file().await { + warn!("control file save in update_status failed: {:?}", e); + } + } + } + + // update the state in Arc + tli.wal_backup_active + .store(is_wal_backup_required, std::sync::atomic::Ordering::SeqCst); + tli.broker_active + .store(is_active, std::sync::atomic::Ordering::SeqCst); + + // sleep to make the loop less busy tokio::time::sleep(REFRESH_INTERVAL).await; // wait until something changes diff --git a/safekeeper/src/timelines_global_map.rs b/safekeeper/src/timelines_global_map.rs index bb2a1e9575..6c9a2bae6b 100644 --- a/safekeeper/src/timelines_global_map.rs +++ b/safekeeper/src/timelines_global_map.rs @@ -4,6 +4,7 @@ use crate::safekeeper::ServerInfo; use crate::timeline::{Timeline, TimelineError}; +use crate::timelines_set::TimelinesSet; use crate::SafeKeeperConf; use anyhow::{bail, Context, Result}; use camino::Utf8PathBuf; @@ -19,6 +20,7 @@ use utils::lsn::Lsn; struct GlobalTimelinesState { timelines: HashMap>, conf: Option, + broker_active_set: Arc, load_lock: Arc>, } @@ -34,8 +36,8 @@ impl GlobalTimelinesState { } /// Get dependencies for a timeline constructor. - fn get_dependencies(&self) -> SafeKeeperConf { - self.get_conf().clone() + fn get_dependencies(&self) -> (SafeKeeperConf, Arc) { + (self.get_conf().clone(), self.broker_active_set.clone()) } /// Insert timeline into the map. Returns error if timeline with the same id already exists. @@ -61,6 +63,7 @@ static TIMELINES_STATE: Lazy> = Lazy::new(|| { Mutex::new(GlobalTimelinesState { timelines: HashMap::new(), conf: None, + broker_active_set: Arc::new(TimelinesSet::default()), load_lock: Arc::new(tokio::sync::Mutex::new(TimelineLoadLock)), }) }); @@ -118,9 +121,9 @@ impl GlobalTimelines { /// this function is called during init when nothing else is running, so /// this is fine. async fn load_tenant_timelines(tenant_id: TenantId) -> Result<()> { - let conf = { + let (conf, broker_active_set) = { let state = TIMELINES_STATE.lock().unwrap(); - state.get_conf().clone() + state.get_dependencies() }; let timelines_dir = conf.tenant_dir(&tenant_id); @@ -141,8 +144,7 @@ impl GlobalTimelines { .unwrap() .timelines .insert(ttid, tli.clone()); - tli.bootstrap(&conf); - tli.update_status_notify().await.unwrap(); + tli.bootstrap(&conf, broker_active_set.clone()); } // If we can't load a timeline, it's most likely because of a corrupted // directory. We will log an error and won't allow to delete/recreate @@ -175,7 +177,7 @@ impl GlobalTimelines { _guard: &tokio::sync::MutexGuard<'a, TimelineLoadLock>, ttid: TenantTimelineId, ) -> Result> { - let conf = TIMELINES_STATE.lock().unwrap().get_dependencies(); + let (conf, broker_active_set) = TIMELINES_STATE.lock().unwrap().get_dependencies(); match Timeline::load_timeline(&conf, ttid) { Ok(timeline) => { @@ -188,7 +190,7 @@ impl GlobalTimelines { .timelines .insert(ttid, tli.clone()); - tli.bootstrap(&conf); + tli.bootstrap(&conf, broker_active_set); Ok(tli) } @@ -207,6 +209,10 @@ impl GlobalTimelines { TIMELINES_STATE.lock().unwrap().get_conf().clone() } + pub fn get_global_broker_active_set() -> Arc { + TIMELINES_STATE.lock().unwrap().broker_active_set.clone() + } + /// Create a new timeline with the given id. If the timeline already exists, returns /// an existing timeline. pub async fn create( @@ -215,7 +221,7 @@ impl GlobalTimelines { commit_lsn: Lsn, local_start_lsn: Lsn, ) -> Result> { - let conf = { + let (conf, broker_active_set) = { let state = TIMELINES_STATE.lock().unwrap(); if let Ok(timeline) = state.get(&ttid) { // Timeline already exists, return it. @@ -249,7 +255,10 @@ impl GlobalTimelines { // Write the new timeline to the disk and start background workers. // Bootstrap is transactional, so if it fails, the timeline will be deleted, // and the state on disk should remain unchanged. - if let Err(e) = timeline.init_new(&mut shared_state, &conf).await { + if let Err(e) = timeline + .init_new(&mut shared_state, &conf, broker_active_set) + .await + { // Note: the most likely reason for init failure is that the timeline // directory already exists on disk. This happens when timeline is corrupted // and wasn't loaded from disk on startup because of that. We want to preserve @@ -266,7 +275,6 @@ impl GlobalTimelines { // We are done with bootstrap, release the lock, return the timeline. // {} block forces release before .await } - timeline.update_status_notify().await?; Ok(timeline) } @@ -323,8 +331,7 @@ impl GlobalTimelines { let mut shared_state = timeline.write_shared_state().await; info!("deleting timeline {}, only_local={}", ttid, only_local); - let (dir_existed, was_active) = - timeline.delete(&mut shared_state, only_local).await?; + let dir_existed = timeline.delete(&mut shared_state, only_local).await?; // Remove timeline from the map. // FIXME: re-enable it once we fix the issue with recreation of deleted timelines @@ -333,7 +340,7 @@ impl GlobalTimelines { Ok(TimelineDeleteForceResult { dir_existed, - was_active, + was_active: true, // TODO: we probably should remove this field }) } Err(_) => { diff --git a/safekeeper/src/timelines_set.rs b/safekeeper/src/timelines_set.rs new file mode 100644 index 0000000000..8012bc6a65 --- /dev/null +++ b/safekeeper/src/timelines_set.rs @@ -0,0 +1,78 @@ +use std::{collections::HashMap, sync::Arc}; + +use utils::id::TenantTimelineId; + +use crate::timeline::Timeline; + +/// Set of timelines, used to keep subset of timelines used for some tasks (like WAL removal). +pub struct TimelinesSet { + timelines: std::sync::Mutex>>, +} + +impl Default for TimelinesSet { + fn default() -> Self { + Self { + timelines: std::sync::Mutex::new(HashMap::new()), + } + } +} + +impl TimelinesSet { + pub fn insert(&self, tli: Arc) { + self.timelines.lock().unwrap().insert(tli.ttid, tli); + } + + pub fn delete(&self, ttid: &TenantTimelineId) { + self.timelines.lock().unwrap().remove(ttid); + } + + pub fn set_present(&self, tli: Arc, present: bool) { + if present { + self.insert(tli); + } else { + self.delete(&tli.ttid); + } + } + + pub fn is_present(&self, ttid: &TenantTimelineId) -> bool { + self.timelines.lock().unwrap().contains_key(ttid) + } + + pub fn get_all(&self) -> Vec> { + self.timelines.lock().unwrap().values().cloned().collect() + } + + pub fn guard(self: &Arc, tli: Arc) -> TimelineSetGuard { + let is_present = self.is_present(&tli.ttid); + TimelineSetGuard { + timelines_set: self.clone(), + tli, + is_present, + } + } +} + +pub struct TimelineSetGuard { + timelines_set: Arc, + tli: Arc, + is_present: bool, +} + +impl TimelineSetGuard { + /// Returns true if the state was changed. + pub fn set(&mut self, present: bool) -> bool { + if present == self.is_present { + return false; + } + self.is_present = present; + self.timelines_set.set_present(self.tli.clone(), present); + true + } +} + +impl Drop for TimelineSetGuard { + fn drop(&mut self) { + // remove timeline from the map on drop + self.timelines_set.delete(&self.tli.ttid); + } +} diff --git a/safekeeper/src/wal_backup.rs b/safekeeper/src/wal_backup.rs index fd88ff0088..b8116201d4 100644 --- a/safekeeper/src/wal_backup.rs +++ b/safekeeper/src/wal_backup.rs @@ -48,7 +48,11 @@ pub struct WalBackupTaskHandle { } /// Do we have anything to upload to S3, i.e. should we run the backup task? -fn is_wal_backup_required(wal_seg_size: usize, num_computes: usize, state: &StateSnapshot) -> bool { +pub fn is_wal_backup_required( + wal_seg_size: usize, + num_computes: usize, + state: &StateSnapshot, +) -> bool { num_computes > 0 || // Currently only the whole segment is offloaded, so compare segment numbers. (state.commit_lsn.segment_number(wal_seg_size) > state.backup_lsn.segment_number(wal_seg_size)) @@ -60,15 +64,14 @@ fn is_wal_backup_required(wal_seg_size: usize, num_computes: usize, state: &Stat pub async fn update_task( conf: &SafeKeeperConf, ttid: TenantTimelineId, - wal_seg_size: usize, - num_computes: usize, + need_backup: bool, 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 = is_wal_backup_required(wal_seg_size, num_computes, state) && elected_me; + let should_task_run = need_backup && elected_me; if should_task_run != (entry.is_some()) { if should_task_run {