From eb521fc24b3ef6e1fdfef1dfcef970eddf76aa05 Mon Sep 17 00:00:00 2001 From: boqiu <82121246@qq.com> Date: Wed, 21 Aug 2024 11:29:28 +0800 Subject: [PATCH] adjust default timeout value for auto sync --- node/sync/src/auto_sync/batcher.rs | 19 ++++++++++++------- node/sync/src/auto_sync/batcher_random.rs | 15 +++++++++++---- node/sync/src/auto_sync/batcher_serial.rs | 17 +++++++++++------ node/sync/src/lib.rs | 13 ++++++++----- run/config-testnet-standard.toml | 12 ++++-------- run/config-testnet-turbo.toml | 12 ++++-------- run/config.toml | 12 ++++-------- 7 files changed, 54 insertions(+), 46 deletions(-) diff --git a/node/sync/src/auto_sync/batcher.rs b/node/sync/src/auto_sync/batcher.rs index 0107413..ab1f9fd 100644 --- a/node/sync/src/auto_sync/batcher.rs +++ b/node/sync/src/auto_sync/batcher.rs @@ -1,7 +1,7 @@ -use crate::{controllers::SyncState, Config, SyncRequest, SyncResponse, SyncSender}; +use crate::{controllers::SyncState, SyncRequest, SyncResponse, SyncSender}; use anyhow::{bail, Result}; use serde::{Deserialize, Serialize}; -use std::{collections::HashSet, fmt::Debug, sync::Arc}; +use std::{collections::HashSet, fmt::Debug, sync::Arc, time::Duration}; use storage_async::Store; use tokio::sync::RwLock; @@ -15,18 +15,23 @@ pub enum SyncResult { /// Supports to sync files concurrently. #[derive(Clone)] pub struct Batcher { - pub(crate) config: Config, capacity: usize, + find_peer_timeout: Duration, tasks: Arc>>, // files to sync store: Store, sync_send: SyncSender, } impl Batcher { - pub fn new(config: Config, capacity: usize, store: Store, sync_send: SyncSender) -> Self { + pub fn new( + capacity: usize, + find_peer_timeout: Duration, + store: Store, + sync_send: SyncSender, + ) -> Self { Self { - config, capacity, + find_peer_timeout, tasks: Default::default(), store, sync_send, @@ -128,7 +133,7 @@ impl Batcher { // finding peers timeout Some(SyncState::FindingPeers { origin, .. }) - if origin.elapsed() > self.config.find_peer_timeout => + if origin.elapsed() > self.find_peer_timeout => { debug!(%tx_seq, "Terminate file sync due to finding peers timeout"); self.terminate_file_sync(tx_seq, false).await; @@ -137,7 +142,7 @@ impl Batcher { // connecting peers timeout Some(SyncState::ConnectingPeers { origin, .. }) - if origin.elapsed() > self.config.find_peer_timeout => + if origin.elapsed() > self.find_peer_timeout => { debug!(%tx_seq, "Terminate file sync due to connecting peers timeout"); self.terminate_file_sync(tx_seq, false).await; diff --git a/node/sync/src/auto_sync/batcher_random.rs b/node/sync/src/auto_sync/batcher_random.rs index 496b0ab..a070eb1 100644 --- a/node/sync/src/auto_sync/batcher_random.rs +++ b/node/sync/src/auto_sync/batcher_random.rs @@ -22,6 +22,7 @@ pub struct RandomBatcherState { #[derive(Clone)] pub struct RandomBatcher { + config: Config, batcher: Batcher, sync_store: Arc, } @@ -34,7 +35,13 @@ impl RandomBatcher { sync_store: Arc, ) -> Self { Self { - batcher: Batcher::new(config, config.max_random_workers, store, sync_send), + config, + batcher: Batcher::new( + config.max_random_workers, + config.random_find_peer_timeout, + store, + sync_send, + ), sync_store, } } @@ -56,7 +63,7 @@ impl RandomBatcher { // disable file sync until catched up if !catched_up.load(Ordering::Relaxed) { trace!("Cannot sync file in catch-up phase"); - sleep(self.batcher.config.auto_sync_idle_interval).await; + sleep(self.config.auto_sync_idle_interval).await; continue; } @@ -73,11 +80,11 @@ impl RandomBatcher { "File sync still in progress or idle, state = {:?}", self.get_state().await ); - sleep(self.batcher.config.auto_sync_idle_interval).await; + sleep(self.config.auto_sync_idle_interval).await; } Err(err) => { warn!(%err, "Failed to sync file once, state = {:?}", self.get_state().await); - sleep(self.batcher.config.auto_sync_error_interval).await; + sleep(self.config.auto_sync_error_interval).await; } } } diff --git a/node/sync/src/auto_sync/batcher_serial.rs b/node/sync/src/auto_sync/batcher_serial.rs index 9322c72..e4a3986 100644 --- a/node/sync/src/auto_sync/batcher_serial.rs +++ b/node/sync/src/auto_sync/batcher_serial.rs @@ -23,6 +23,7 @@ use tokio::{ /// Supports to sync files in sequence concurrently. #[derive(Clone)] pub struct SerialBatcher { + config: Config, batcher: Batcher, /// Next tx seq to sync. @@ -80,13 +81,17 @@ impl SerialBatcher { sync_send: SyncSender, sync_store: Arc, ) -> Result { - let capacity = config.max_sequential_workers; - // continue file sync from break point in db let (next_tx_seq, max_tx_seq) = sync_store.get_tx_seq_range().await?; Ok(Self { - batcher: Batcher::new(config, capacity, store, sync_send), + config, + batcher: Batcher::new( + config.max_sequential_workers, + config.sequential_find_peer_timeout, + store, + sync_send, + ), next_tx_seq: Arc::new(AtomicU64::new(next_tx_seq.unwrap_or(0))), max_tx_seq: Arc::new(AtomicU64::new(max_tx_seq.unwrap_or(u64::MAX))), pending_completed_txs: Default::default(), @@ -136,7 +141,7 @@ impl SerialBatcher { // disable file sync until catched up if !catched_up.load(Ordering::Relaxed) { trace!("Cannot sync file in catch-up phase"); - sleep(self.batcher.config.auto_sync_idle_interval).await; + sleep(self.config.auto_sync_idle_interval).await; continue; } @@ -157,11 +162,11 @@ impl SerialBatcher { "File sync still in progress or idle, state = {:?}", self.get_state().await ); - sleep(self.batcher.config.auto_sync_idle_interval).await; + sleep(self.config.auto_sync_idle_interval).await; } Err(err) => { warn!(%err, "Failed to sync file once, state = {:?}", self.get_state().await); - sleep(self.batcher.config.auto_sync_error_interval).await; + sleep(self.config.auto_sync_error_interval).await; } } } diff --git a/node/sync/src/lib.rs b/node/sync/src/lib.rs index f2250ca..ae25d49 100644 --- a/node/sync/src/lib.rs +++ b/node/sync/src/lib.rs @@ -52,7 +52,9 @@ pub struct Config { pub max_sequential_workers: usize, pub max_random_workers: usize, #[serde(deserialize_with = "deserialize_duration")] - pub find_peer_timeout: Duration, + pub sequential_find_peer_timeout: Duration, + #[serde(deserialize_with = "deserialize_duration")] + pub random_find_peer_timeout: Duration, } impl Default for Config { @@ -61,7 +63,7 @@ impl Default for Config { // sync service config heartbeat_interval: Duration::from_secs(5), auto_sync_enabled: false, - max_sync_files: 16, + max_sync_files: 32, sync_file_by_rpc_enabled: true, sync_file_on_announcement_enabled: false, @@ -78,9 +80,10 @@ impl Default for Config { // auto sync config auto_sync_idle_interval: Duration::from_secs(3), auto_sync_error_interval: Duration::from_secs(10), - max_sequential_workers: 8, - max_random_workers: 4, - find_peer_timeout: Duration::from_secs(10), + max_sequential_workers: 24, + max_random_workers: 8, + sequential_find_peer_timeout: Duration::from_secs(60), + random_find_peer_timeout: Duration::from_secs(500), } } } diff --git a/run/config-testnet-standard.toml b/run/config-testnet-standard.toml index 57fb869..d201c3a 100644 --- a/run/config-testnet-standard.toml +++ b/run/config-testnet-standard.toml @@ -228,11 +228,7 @@ reward_contract_address = "0x0496D0817BD8519e0de4894Dc379D35c35275609" auto_sync_enabled = true # Maximum number of files in sync from other peers simultaneously. -max_sync_files = 32 - -# Timeout to terminate a file sync when automatically sync from other peers. -# If timeout, terminated file sync will be triggered later. -# find_peer_timeout = "10s" +# max_sync_files = 32 # Enable to start a file sync via RPC (e.g. `admin_startSyncFile`). # sync_file_by_rpc_enabled = true @@ -241,10 +237,10 @@ max_sync_files = 32 # sync_file_on_announcement_enabled = false # Maximum threads to sync files in sequence. -max_sequential_workers = 24 +# max_sequential_workers = 24 # Maximum threads to sync files randomly. -# max_random_workers = 4 +# max_random_workers = 8 ####################################################################### ### File Location Cache Options ### @@ -265,4 +261,4 @@ max_sequential_workers = 24 # Validity period of location information. # If the timestamp in the storage location information exceeds this duration from the current time, it will be removed from the cache. -# entry_expiration_time_secs = 3600 \ No newline at end of file +# entry_expiration_time_secs = 3600 diff --git a/run/config-testnet-turbo.toml b/run/config-testnet-turbo.toml index c3756a1..31243cf 100644 --- a/run/config-testnet-turbo.toml +++ b/run/config-testnet-turbo.toml @@ -228,11 +228,7 @@ reward_contract_address = "0x51998C4d486F406a788B766d93510980ae1f9360" auto_sync_enabled = true # Maximum number of files in sync from other peers simultaneously. -max_sync_files = 32 - -# Timeout to terminate a file sync when automatically sync from other peers. -# If timeout, terminated file sync will be triggered later. -# find_peer_timeout = "10s" +# max_sync_files = 32 # Enable to start a file sync via RPC (e.g. `admin_startSyncFile`). # sync_file_by_rpc_enabled = true @@ -241,10 +237,10 @@ max_sync_files = 32 # sync_file_on_announcement_enabled = false # Maximum threads to sync files in sequence. -max_sequential_workers = 24 +# max_sequential_workers = 24 # Maximum threads to sync files randomly. -# max_random_workers = 4 +# max_random_workers = 8 ####################################################################### ### File Location Cache Options ### @@ -265,4 +261,4 @@ max_sequential_workers = 24 # Validity period of location information. # If the timestamp in the storage location information exceeds this duration from the current time, it will be removed from the cache. -# entry_expiration_time_secs = 3600 \ No newline at end of file +# entry_expiration_time_secs = 3600 diff --git a/run/config.toml b/run/config.toml index e58f7e9..023f3c3 100644 --- a/run/config.toml +++ b/run/config.toml @@ -227,11 +227,7 @@ # auto_sync_enabled = false # Maximum number of files in sync from other peers simultaneously. -# max_sync_files = 16 - -# Timeout to terminate a file sync when automatically sync from other peers. -# If timeout, terminated file sync will be triggered later. -# find_peer_timeout = "10s" +# max_sync_files = 32 # Enable to start a file sync via RPC (e.g. `admin_startSyncFile`). # sync_file_by_rpc_enabled = true @@ -240,10 +236,10 @@ # sync_file_on_announcement_enabled = false # Maximum threads to sync files in sequence. -# max_sequential_workers = 8 +# max_sequential_workers = 24 # Maximum threads to sync files randomly. -# max_random_workers = 4 +# max_random_workers = 8 ####################################################################### ### File Location Cache Options ### @@ -264,4 +260,4 @@ # Validity period of location information. # If the timestamp in the storage location information exceeds this duration from the current time, it will be removed from the cache. -# entry_expiration_time_secs = 3600 \ No newline at end of file +# entry_expiration_time_secs = 3600