adjust default timeout value for auto sync

This commit is contained in:
boqiu 2024-08-21 11:29:28 +08:00
parent 4aae302df5
commit eb521fc24b
7 changed files with 54 additions and 46 deletions

View File

@ -1,7 +1,7 @@
use crate::{controllers::SyncState, Config, SyncRequest, SyncResponse, SyncSender}; use crate::{controllers::SyncState, SyncRequest, SyncResponse, SyncSender};
use anyhow::{bail, Result}; use anyhow::{bail, Result};
use serde::{Deserialize, Serialize}; 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 storage_async::Store;
use tokio::sync::RwLock; use tokio::sync::RwLock;
@ -15,18 +15,23 @@ pub enum SyncResult {
/// Supports to sync files concurrently. /// Supports to sync files concurrently.
#[derive(Clone)] #[derive(Clone)]
pub struct Batcher { pub struct Batcher {
pub(crate) config: Config,
capacity: usize, capacity: usize,
find_peer_timeout: Duration,
tasks: Arc<RwLock<HashSet<u64>>>, // files to sync tasks: Arc<RwLock<HashSet<u64>>>, // files to sync
store: Store, store: Store,
sync_send: SyncSender, sync_send: SyncSender,
} }
impl Batcher { 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 { Self {
config,
capacity, capacity,
find_peer_timeout,
tasks: Default::default(), tasks: Default::default(),
store, store,
sync_send, sync_send,
@ -128,7 +133,7 @@ impl Batcher {
// finding peers timeout // finding peers timeout
Some(SyncState::FindingPeers { origin, .. }) 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"); debug!(%tx_seq, "Terminate file sync due to finding peers timeout");
self.terminate_file_sync(tx_seq, false).await; self.terminate_file_sync(tx_seq, false).await;
@ -137,7 +142,7 @@ impl Batcher {
// connecting peers timeout // connecting peers timeout
Some(SyncState::ConnectingPeers { origin, .. }) 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"); debug!(%tx_seq, "Terminate file sync due to connecting peers timeout");
self.terminate_file_sync(tx_seq, false).await; self.terminate_file_sync(tx_seq, false).await;

View File

@ -22,6 +22,7 @@ pub struct RandomBatcherState {
#[derive(Clone)] #[derive(Clone)]
pub struct RandomBatcher { pub struct RandomBatcher {
config: Config,
batcher: Batcher, batcher: Batcher,
sync_store: Arc<SyncStore>, sync_store: Arc<SyncStore>,
} }
@ -34,7 +35,13 @@ impl RandomBatcher {
sync_store: Arc<SyncStore>, sync_store: Arc<SyncStore>,
) -> Self { ) -> Self {
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, sync_store,
} }
} }
@ -56,7 +63,7 @@ impl RandomBatcher {
// disable file sync until catched up // disable file sync until catched up
if !catched_up.load(Ordering::Relaxed) { if !catched_up.load(Ordering::Relaxed) {
trace!("Cannot sync file in catch-up phase"); 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; continue;
} }
@ -73,11 +80,11 @@ impl RandomBatcher {
"File sync still in progress or idle, state = {:?}", "File sync still in progress or idle, state = {:?}",
self.get_state().await self.get_state().await
); );
sleep(self.batcher.config.auto_sync_idle_interval).await; sleep(self.config.auto_sync_idle_interval).await;
} }
Err(err) => { Err(err) => {
warn!(%err, "Failed to sync file once, state = {:?}", self.get_state().await); 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;
} }
} }
} }

View File

@ -23,6 +23,7 @@ use tokio::{
/// Supports to sync files in sequence concurrently. /// Supports to sync files in sequence concurrently.
#[derive(Clone)] #[derive(Clone)]
pub struct SerialBatcher { pub struct SerialBatcher {
config: Config,
batcher: Batcher, batcher: Batcher,
/// Next tx seq to sync. /// Next tx seq to sync.
@ -80,13 +81,17 @@ impl SerialBatcher {
sync_send: SyncSender, sync_send: SyncSender,
sync_store: Arc<SyncStore>, sync_store: Arc<SyncStore>,
) -> Result<Self> { ) -> Result<Self> {
let capacity = config.max_sequential_workers;
// continue file sync from break point in db // continue file sync from break point in db
let (next_tx_seq, max_tx_seq) = sync_store.get_tx_seq_range().await?; let (next_tx_seq, max_tx_seq) = sync_store.get_tx_seq_range().await?;
Ok(Self { 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))), 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))), max_tx_seq: Arc::new(AtomicU64::new(max_tx_seq.unwrap_or(u64::MAX))),
pending_completed_txs: Default::default(), pending_completed_txs: Default::default(),
@ -136,7 +141,7 @@ impl SerialBatcher {
// disable file sync until catched up // disable file sync until catched up
if !catched_up.load(Ordering::Relaxed) { if !catched_up.load(Ordering::Relaxed) {
trace!("Cannot sync file in catch-up phase"); 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; continue;
} }
@ -157,11 +162,11 @@ impl SerialBatcher {
"File sync still in progress or idle, state = {:?}", "File sync still in progress or idle, state = {:?}",
self.get_state().await self.get_state().await
); );
sleep(self.batcher.config.auto_sync_idle_interval).await; sleep(self.config.auto_sync_idle_interval).await;
} }
Err(err) => { Err(err) => {
warn!(%err, "Failed to sync file once, state = {:?}", self.get_state().await); 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;
} }
} }
} }

View File

@ -52,7 +52,9 @@ pub struct Config {
pub max_sequential_workers: usize, pub max_sequential_workers: usize,
pub max_random_workers: usize, pub max_random_workers: usize,
#[serde(deserialize_with = "deserialize_duration")] #[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 { impl Default for Config {
@ -61,7 +63,7 @@ impl Default for Config {
// sync service config // sync service config
heartbeat_interval: Duration::from_secs(5), heartbeat_interval: Duration::from_secs(5),
auto_sync_enabled: false, auto_sync_enabled: false,
max_sync_files: 16, max_sync_files: 32,
sync_file_by_rpc_enabled: true, sync_file_by_rpc_enabled: true,
sync_file_on_announcement_enabled: false, sync_file_on_announcement_enabled: false,
@ -78,9 +80,10 @@ impl Default for Config {
// auto sync config // auto sync config
auto_sync_idle_interval: Duration::from_secs(3), auto_sync_idle_interval: Duration::from_secs(3),
auto_sync_error_interval: Duration::from_secs(10), auto_sync_error_interval: Duration::from_secs(10),
max_sequential_workers: 8, max_sequential_workers: 24,
max_random_workers: 4, max_random_workers: 8,
find_peer_timeout: Duration::from_secs(10), sequential_find_peer_timeout: Duration::from_secs(60),
random_find_peer_timeout: Duration::from_secs(500),
} }
} }
} }

View File

@ -228,11 +228,7 @@ reward_contract_address = "0x0496D0817BD8519e0de4894Dc379D35c35275609"
auto_sync_enabled = true auto_sync_enabled = true
# Maximum number of files in sync from other peers simultaneously. # Maximum number of files in sync from other peers simultaneously.
max_sync_files = 32 # 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"
# Enable to start a file sync via RPC (e.g. `admin_startSyncFile`). # Enable to start a file sync via RPC (e.g. `admin_startSyncFile`).
# sync_file_by_rpc_enabled = true # sync_file_by_rpc_enabled = true
@ -241,10 +237,10 @@ max_sync_files = 32
# sync_file_on_announcement_enabled = false # sync_file_on_announcement_enabled = false
# Maximum threads to sync files in sequence. # Maximum threads to sync files in sequence.
max_sequential_workers = 24 # max_sequential_workers = 24
# Maximum threads to sync files randomly. # Maximum threads to sync files randomly.
# max_random_workers = 4 # max_random_workers = 8
####################################################################### #######################################################################
### File Location Cache Options ### ### File Location Cache Options ###
@ -265,4 +261,4 @@ max_sequential_workers = 24
# Validity period of location information. # 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. # 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 # entry_expiration_time_secs = 3600

View File

@ -228,11 +228,7 @@ reward_contract_address = "0x51998C4d486F406a788B766d93510980ae1f9360"
auto_sync_enabled = true auto_sync_enabled = true
# Maximum number of files in sync from other peers simultaneously. # Maximum number of files in sync from other peers simultaneously.
max_sync_files = 32 # 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"
# Enable to start a file sync via RPC (e.g. `admin_startSyncFile`). # Enable to start a file sync via RPC (e.g. `admin_startSyncFile`).
# sync_file_by_rpc_enabled = true # sync_file_by_rpc_enabled = true
@ -241,10 +237,10 @@ max_sync_files = 32
# sync_file_on_announcement_enabled = false # sync_file_on_announcement_enabled = false
# Maximum threads to sync files in sequence. # Maximum threads to sync files in sequence.
max_sequential_workers = 24 # max_sequential_workers = 24
# Maximum threads to sync files randomly. # Maximum threads to sync files randomly.
# max_random_workers = 4 # max_random_workers = 8
####################################################################### #######################################################################
### File Location Cache Options ### ### File Location Cache Options ###
@ -265,4 +261,4 @@ max_sequential_workers = 24
# Validity period of location information. # 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. # 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 # entry_expiration_time_secs = 3600

View File

@ -227,11 +227,7 @@
# auto_sync_enabled = false # auto_sync_enabled = false
# Maximum number of files in sync from other peers simultaneously. # Maximum number of files in sync from other peers simultaneously.
# max_sync_files = 16 # 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"
# Enable to start a file sync via RPC (e.g. `admin_startSyncFile`). # Enable to start a file sync via RPC (e.g. `admin_startSyncFile`).
# sync_file_by_rpc_enabled = true # sync_file_by_rpc_enabled = true
@ -240,10 +236,10 @@
# sync_file_on_announcement_enabled = false # sync_file_on_announcement_enabled = false
# Maximum threads to sync files in sequence. # Maximum threads to sync files in sequence.
# max_sequential_workers = 8 # max_sequential_workers = 24
# Maximum threads to sync files randomly. # Maximum threads to sync files randomly.
# max_random_workers = 4 # max_random_workers = 8
####################################################################### #######################################################################
### File Location Cache Options ### ### File Location Cache Options ###
@ -264,4 +260,4 @@
# Validity period of location information. # 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. # 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 # entry_expiration_time_secs = 3600