From 5ce22ab020135e919ec42c56204a29bbccc62e65 Mon Sep 17 00:00:00 2001 From: boqiu <82121246@qq.com> Date: Wed, 4 Sep 2024 15:24:35 +0800 Subject: [PATCH] Supports to disable sequential auto sync --- node/sync/src/auto_sync/batcher_random.rs | 13 +- node/sync/src/auto_sync/batcher_serial.rs | 23 ++- node/sync/src/auto_sync/sync_store.rs | 206 ++++++++++++---------- 3 files changed, 134 insertions(+), 108 deletions(-) diff --git a/node/sync/src/auto_sync/batcher_random.rs b/node/sync/src/auto_sync/batcher_random.rs index a070eb1..16cd5f2 100644 --- a/node/sync/src/auto_sync/batcher_random.rs +++ b/node/sync/src/auto_sync/batcher_random.rs @@ -1,6 +1,6 @@ use super::{batcher::Batcher, sync_store::SyncStore}; use crate::{ - auto_sync::{batcher::SyncResult, metrics}, + auto_sync::{batcher::SyncResult, metrics, sync_store::Queue}, Config, SyncSender, }; use anyhow::Result; @@ -108,16 +108,17 @@ impl RandomBatcher { SyncResult::Timeout => metrics::RANDOM_SYNC_RESULT_TIMEOUT.inc(1), } - match sync_result { - SyncResult::Completed => self.sync_store.remove_tx(tx_seq).await?, - _ => self.sync_store.downgrade_tx_to_pending(tx_seq).await?, - }; + if matches!(sync_result, SyncResult::Completed) { + self.sync_store.remove(tx_seq).await?; + } else { + self.sync_store.insert(tx_seq, Queue::Pending).await?; + } Ok(true) } async fn schedule(&mut self) -> Result { - let tx_seq = match self.sync_store.random_tx().await? { + let tx_seq = match self.sync_store.random().await? { Some(v) => v, None => return Ok(false), }; diff --git a/node/sync/src/auto_sync/batcher_serial.rs b/node/sync/src/auto_sync/batcher_serial.rs index e4a3986..0709592 100644 --- a/node/sync/src/auto_sync/batcher_serial.rs +++ b/node/sync/src/auto_sync/batcher_serial.rs @@ -2,7 +2,10 @@ use super::{ batcher::{Batcher, SyncResult}, sync_store::SyncStore, }; -use crate::{auto_sync::metrics, Config, SyncSender}; +use crate::{ + auto_sync::{metrics, sync_store::Queue}, + Config, SyncSender, +}; use anyhow::Result; use log_entry_sync::LogSyncEvent; use serde::{Deserialize, Serialize}; @@ -200,7 +203,7 @@ impl SerialBatcher { } // otherwise, mark tx as ready for sync - if let Err(err) = self.sync_store.upgrade_tx_to_ready(announced_tx_seq).await { + if let Err(err) = self.sync_store.upgrade(announced_tx_seq).await { error!(%err, %announced_tx_seq, "Failed to promote announced tx to ready, state = {:?}", self.get_state().await); } @@ -280,11 +283,23 @@ impl SerialBatcher { /// Schedule file sync in sequence. async fn schedule_next(&mut self) -> Result { - let next_tx_seq = self.next_tx_seq.load(Ordering::Relaxed); + let mut next_tx_seq = self.next_tx_seq.load(Ordering::Relaxed); if next_tx_seq > self.max_tx_seq.load(Ordering::Relaxed) { return Ok(false); } + // If sequential sync disabled, delegate to random sync. + if self.config.max_sequential_workers == 0 { + self.sync_store.insert(next_tx_seq, Queue::Ready).await?; + + next_tx_seq += 1; + self.sync_store.set_next_tx_seq(next_tx_seq).await?; + self.next_tx_seq.store(next_tx_seq, Ordering::Relaxed); + self.next_tx_seq_in_db.store(next_tx_seq, Ordering::Relaxed); + + return Ok(true); + } + if !self.batcher.add(next_tx_seq).await? { return Ok(false); } @@ -309,7 +324,7 @@ impl SerialBatcher { // downgrade to random sync if file sync failed or timeout if matches!(sync_result, SyncResult::Failed | SyncResult::Timeout) { - self.sync_store.add_pending_tx(current).await?; + self.sync_store.insert(current, Queue::Pending).await?; } // always move forward in db diff --git a/node/sync/src/auto_sync/sync_store.rs b/node/sync/src/auto_sync/sync_store.rs index 8c59448..796dc1a 100644 --- a/node/sync/src/auto_sync/sync_store.rs +++ b/node/sync/src/auto_sync/sync_store.rs @@ -8,6 +8,20 @@ use tokio::sync::RwLock; const KEY_NEXT_TX_SEQ: &str = "sync.manager.next_tx_seq"; const KEY_MAX_TX_SEQ: &str = "sync.manager.max_tx_seq"; +#[derive(Debug, PartialEq, Eq)] +pub enum Queue { + Ready, + Pending, +} + +#[derive(Debug, PartialEq, Eq)] +pub enum InsertResult { + NewAdded, // new added in target queue + AlreadyExists, // already exists in target queue + Upgraded, // upgraded from pending queue to ready queue + Downgraded, // downgraged from ready queue to pending queue +} + pub struct SyncStore { store: Arc>, @@ -64,31 +78,69 @@ impl SyncStore { store.set_config_encoded(&KEY_MAX_TX_SEQ, &tx_seq) } - pub async fn add_pending_tx(&self, tx_seq: u64) -> Result { - let async_store = self.store.write().await; + pub async fn contains(&self, tx_seq: u64) -> Result> { + let async_store = self.store.read().await; let store = async_store.get_store(); - // already in ready queue if self.ready_txs.has(store, tx_seq)? { - return Ok(false); + return Ok(Some(Queue::Ready)); } - // always add in pending queue - self.pending_txs.add(store, None, tx_seq) + if self.pending_txs.has(store, tx_seq)? { + return Ok(Some(Queue::Pending)); + } + + Ok(None) } - pub async fn upgrade_tx_to_ready(&self, tx_seq: u64) -> Result { + pub async fn insert(&self, tx_seq: u64, queue: Queue) -> Result { + let async_store = self.store.write().await; + let store = async_store.get_store(); + + let mut tx = ConfigTx::default(); + + match queue { + Queue::Ready => { + if !self.ready_txs.add(store, Some(&mut tx), tx_seq)? { + return Ok(InsertResult::AlreadyExists); + } + + let removed = self.pending_txs.remove(store, Some(&mut tx), tx_seq)?; + store.exec_configs(tx)?; + + if removed { + Ok(InsertResult::Upgraded) + } else { + Ok(InsertResult::NewAdded) + } + } + Queue::Pending => { + if !self.pending_txs.add(store, Some(&mut tx), tx_seq)? { + return Ok(InsertResult::AlreadyExists); + } + + let removed = self.ready_txs.remove(store, Some(&mut tx), tx_seq)?; + store.exec_configs(tx)?; + + if removed { + Ok(InsertResult::Downgraded) + } else { + Ok(InsertResult::NewAdded) + } + } + } + } + + pub async fn upgrade(&self, tx_seq: u64) -> Result { let async_store = self.store.write().await; let store = async_store.get_store(); let mut tx = ConfigTx::default(); - // not in pending queue if !self.pending_txs.remove(store, Some(&mut tx), tx_seq)? { return Ok(false); } - // move from pending to ready queue let added = self.ready_txs.add(store, Some(&mut tx), tx_seq)?; store.exec_configs(tx)?; @@ -96,26 +148,7 @@ impl SyncStore { Ok(added) } - pub async fn downgrade_tx_to_pending(&self, tx_seq: u64) -> Result { - let async_store = self.store.write().await; - let store = async_store.get_store(); - - let mut tx = ConfigTx::default(); - - // not in ready queue - if !self.ready_txs.remove(store, Some(&mut tx), tx_seq)? { - return Ok(false); - } - - // move from ready to pending queue - let added = self.pending_txs.add(store, Some(&mut tx), tx_seq)?; - - store.exec_configs(tx)?; - - Ok(added) - } - - pub async fn random_tx(&self) -> Result> { + pub async fn random(&self) -> Result> { let async_store = self.store.read().await; let store = async_store.get_store(); @@ -128,17 +161,21 @@ impl SyncStore { self.pending_txs.random(store) } - pub async fn remove_tx(&self, tx_seq: u64) -> Result { + pub async fn remove(&self, tx_seq: u64) -> Result> { let async_store = self.store.write().await; let store = async_store.get_store(); // removed in ready queue if self.ready_txs.remove(store, None, tx_seq)? { - return Ok(true); + return Ok(Some(Queue::Ready)); } - // otherwise, try to remove in pending queue - self.pending_txs.remove(store, None, tx_seq) + // removed in pending queue + if self.pending_txs.remove(store, None, tx_seq)? { + return Ok(Some(Queue::Pending)); + } + + Ok(None) } } @@ -146,7 +183,7 @@ impl SyncStore { mod tests { use crate::test_util::tests::TestStoreRuntime; - use super::SyncStore; + use super::{InsertResult::*, Queue::*, SyncStore}; #[tokio::test] async fn test_tx_seq_range() { @@ -165,106 +202,79 @@ mod tests { } #[tokio::test] - async fn test_add_pending_tx() { + async fn test_insert() { let runtime = TestStoreRuntime::default(); let store = SyncStore::new(runtime.store.clone()); - // add pending tx 3 - assert!(store.add_pending_tx(3).await.unwrap()); + assert_eq!(store.contains(1).await.unwrap(), None); + assert_eq!(store.insert(1, Pending).await.unwrap(), NewAdded); + assert_eq!(store.contains(1).await.unwrap(), Some(Pending)); + assert_eq!(store.insert(1, Pending).await.unwrap(), AlreadyExists); + assert_eq!(store.insert(1, Ready).await.unwrap(), Upgraded); + assert_eq!(store.contains(1).await.unwrap(), Some(Ready)); - // cannot add pending tx 3 again - assert!(!store.add_pending_tx(3).await.unwrap()); + assert_eq!(store.insert(2, Ready).await.unwrap(), NewAdded); + assert_eq!(store.contains(2).await.unwrap(), Some(Ready)); + assert_eq!(store.insert(2, Ready).await.unwrap(), AlreadyExists); + assert_eq!(store.insert(2, Pending).await.unwrap(), Downgraded); + assert_eq!(store.contains(2).await.unwrap(), Some(Pending)); } #[tokio::test] - async fn test_upgrade_tx() { + async fn test_upgrade() { let runtime = TestStoreRuntime::default(); let store = SyncStore::new(runtime.store.clone()); // cannot upgrade by default - assert!(!store.upgrade_tx_to_ready(3).await.unwrap()); + assert!(!store.upgrade(3).await.unwrap()); // add pending tx 3 - assert!(store.add_pending_tx(3).await.unwrap()); + assert_eq!(store.insert(3, Pending).await.unwrap(), NewAdded); // can upgrade to ready - assert!(store.upgrade_tx_to_ready(3).await.unwrap()); - - // cannot add pending tx 3 again event upgraded to ready - assert!(!store.add_pending_tx(3).await.unwrap()); + assert!(store.upgrade(3).await.unwrap()); + assert_eq!(store.contains(3).await.unwrap(), Some(Ready)); // cannot upgrade again - assert!(!store.upgrade_tx_to_ready(3).await.unwrap()); + assert!(!store.upgrade(3).await.unwrap()); } #[tokio::test] - async fn test_downgrade_tx() { - let runtime = TestStoreRuntime::default(); - let store = SyncStore::new(runtime.store.clone()); - - // cannot downgrade by default - assert!(!store.downgrade_tx_to_pending(3).await.unwrap()); - - // add pending tx 3 - assert!(store.add_pending_tx(3).await.unwrap()); - - // cannot downgrade for non-ready tx - assert!(!store.downgrade_tx_to_pending(3).await.unwrap()); - - // upgrade tx 3 to ready - assert!(store.upgrade_tx_to_ready(3).await.unwrap()); - - // can downgrade now - assert!(store.downgrade_tx_to_pending(3).await.unwrap()); - - // cannot downgrade now - assert!(!store.downgrade_tx_to_pending(3).await.unwrap()); - } - - #[tokio::test] - async fn test_random_tx() { + async fn test_random() { let runtime = TestStoreRuntime::default(); let store = SyncStore::new(runtime.store.clone()); // no tx by default - assert_eq!(store.random_tx().await.unwrap(), None); + assert_eq!(store.random().await.unwrap(), None); // add pending txs 1, 2, 3 - assert!(store.add_pending_tx(1).await.unwrap()); - assert!(store.add_pending_tx(2).await.unwrap()); - assert!(store.add_pending_tx(3).await.unwrap()); - let tx = store.random_tx().await.unwrap().unwrap(); + assert_eq!(store.insert(1, Pending).await.unwrap(), NewAdded); + assert_eq!(store.insert(2, Pending).await.unwrap(), NewAdded); + assert_eq!(store.insert(3, Pending).await.unwrap(), NewAdded); + let tx = store.random().await.unwrap().unwrap(); assert!((1..=3).contains(&tx)); - // upgrade tx 1 to ready - assert!(store.upgrade_tx_to_ready(2).await.unwrap()); - assert_eq!(store.random_tx().await.unwrap(), Some(2)); + // upgrade tx 2 to ready + assert_eq!(store.insert(2, Ready).await.unwrap(), Upgraded); + assert_eq!(store.random().await.unwrap(), Some(2)); } #[tokio::test] - async fn test_remove_tx() { + async fn test_remove() { let runtime = TestStoreRuntime::default(); let store = SyncStore::new(runtime.store.clone()); // cannot remove by default - assert!(!store.remove_tx(1).await.unwrap()); + assert_eq!(store.remove(1).await.unwrap(), None); - // add pending tx 1, 2 - assert!(store.add_pending_tx(1).await.unwrap()); - assert!(store.add_pending_tx(2).await.unwrap()); + // add tx 1, 2 + assert_eq!(store.insert(1, Pending).await.unwrap(), NewAdded); + assert_eq!(store.insert(2, Ready).await.unwrap(), NewAdded); - // upgrade tx 1 to ready - assert!(store.upgrade_tx_to_ready(1).await.unwrap()); - assert_eq!(store.random_tx().await.unwrap(), Some(1)); - - // remove tx 1 - assert!(store.remove_tx(1).await.unwrap()); - assert_eq!(store.random_tx().await.unwrap(), Some(2)); - assert!(!store.remove_tx(1).await.unwrap()); - - // remove tx 2 - assert!(store.remove_tx(2).await.unwrap()); - assert_eq!(store.random_tx().await.unwrap(), None); - assert!(!store.remove_tx(2).await.unwrap()); + // remove txs + assert_eq!(store.remove(1).await.unwrap(), Some(Pending)); + assert_eq!(store.remove(1).await.unwrap(), None); + assert_eq!(store.remove(2).await.unwrap(), Some(Ready)); + assert_eq!(store.remove(2).await.unwrap(), None); } }