Skip to content
This repository has been archived by the owner on Nov 15, 2023. It is now read-only.

Commit

Permalink
Actually fix major sync detection (#12114)
Browse files Browse the repository at this point in the history
* Actually fix major sync detection

* Introduce `SyncState::Importing` state

* Add target to SyncState enum variants and add `is_major_syncing` method on it

* Remove unnecessary duplicated `best_seen_block` from `SyncState` struct

* Revert "Remove unnecessary duplicated `best_seen_block` from `SyncState` struct"

This reverts commit bb8abd4.

* Add missing `websocket` feature to `libp2p`

Co-authored-by: parity-processbot <>
  • Loading branch information
nazar-pc authored Oct 21, 2022
1 parent ac17629 commit e458835
Show file tree
Hide file tree
Showing 6 changed files with 71 additions and 62 deletions.
63 changes: 29 additions & 34 deletions client/informant/src/display.rs
Original file line number Diff line number Diff line change
Expand Up @@ -93,42 +93,37 @@ impl<B: BlockT> InformantDisplay<B> {
(diff_bytes_inbound, diff_bytes_outbound)
};

let (level, status, target) = match (
net_status.sync_state,
net_status.best_seen_block,
net_status.state_sync,
net_status.warp_sync,
) {
(
_,
_,
_,
Some(WarpSyncProgress { phase: WarpSyncPhase::DownloadingBlocks(n), .. }),
) => ("⏩", "Block history".into(), format!(", #{}", n)),
(_, _, _, Some(warp)) => (
"⏩",
"Warping".into(),
format!(
", {}, {:.2} Mib",
warp.phase,
(warp.total_bytes as f32) / (1024f32 * 1024f32)
let (level, status, target) =
match (net_status.sync_state, net_status.state_sync, net_status.warp_sync) {
(
_,
_,
Some(WarpSyncProgress { phase: WarpSyncPhase::DownloadingBlocks(n), .. }),
) => ("⏩", "Block history".into(), format!(", #{}", n)),
(_, _, Some(warp)) => (
"⏩",
"Warping".into(),
format!(
", {}, {:.2} Mib",
warp.phase,
(warp.total_bytes as f32) / (1024f32 * 1024f32)
),
),
),
(_, _, Some(state), _) => (
"⚙️ ",
"Downloading state".into(),
format!(
", {}%, {:.2} Mib",
state.percentage,
(state.size as f32) / (1024f32 * 1024f32)
(_, Some(state), _) => (
"⚙️ ",
"Downloading state".into(),
format!(
", {}%, {:.2} Mib",
state.percentage,
(state.size as f32) / (1024f32 * 1024f32)
),
),
),
(SyncState::Idle, _, _, _) => ("💤", "Idle".into(), "".into()),
(SyncState::Downloading, None, _, _) =>
("⚙️ ", format!("Preparing{}", speed), "".into()),
(SyncState::Downloading, Some(n), None, _) =>
("⚙️ ", format!("Syncing{}", speed), format!(", target=#{}", n)),
};
(SyncState::Idle, _, _) => ("💤", "Idle".into(), "".into()),
(SyncState::Downloading { target }, _, _) =>
("⚙️ ", format!("Syncing{}", speed), format!(", target=#{target}")),
(SyncState::Importing { target }, _, _) =>
("⚙️ ", format!("Preparing{}", speed), format!(", target=#{target}")),
};

if self.format.enable_color {
info!(
Expand Down
2 changes: 1 addition & 1 deletion client/network/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ fnv = "1.0.6"
futures = "0.3.21"
futures-timer = "3.0.2"
ip_network = "0.4.1"
libp2p = { version = "0.49.0", features = ["async-std", "dns", "identify", "kad", "mdns-async-io", "mplex", "noise", "ping", "tcp", "yamux"] }
libp2p = { version = "0.49.0", features = ["async-std", "dns", "identify", "kad", "mdns-async-io", "mplex", "noise", "ping", "tcp", "yamux", "websocket"] }
linked_hash_set = "0.1.3"
linked-hash-map = "0.5.4"
log = "0.4.17"
Expand Down
2 changes: 1 addition & 1 deletion client/network/common/src/service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ where
#[derive(Clone)]
pub struct NetworkStatus<B: BlockT> {
/// Current global sync state.
pub sync_state: SyncState,
pub sync_state: SyncState<NumberFor<B>>,
/// Target sync block number.
pub best_seen_block: Option<NumberFor<B>>,
/// Number of peers participating in syncing.
Expand Down
15 changes: 12 additions & 3 deletions client/network/common/src/sync.rs
Original file line number Diff line number Diff line change
Expand Up @@ -44,11 +44,20 @@ pub struct PeerInfo<Block: BlockT> {

/// Reported sync state.
#[derive(Clone, Eq, PartialEq, Debug)]
pub enum SyncState {
pub enum SyncState<BlockNumber> {
/// Initial sync is complete, keep-up sync is active.
Idle,
/// Actively catching up with the chain.
Downloading,
Downloading { target: BlockNumber },
/// All blocks are downloaded and are being imported.
Importing { target: BlockNumber },
}

impl<BlockNumber> SyncState<BlockNumber> {
/// Are we actively catching up with the chain?
pub fn is_major_syncing(&self) -> bool {
!matches!(self, SyncState::Idle)
}
}

/// Reported state download progress.
Expand All @@ -64,7 +73,7 @@ pub struct StateDownloadProgress {
#[derive(Clone)]
pub struct SyncStatus<Block: BlockT> {
/// Current global sync state.
pub state: SyncState,
pub state: SyncState<NumberFor<Block>>,
/// Target sync block number.
pub best_seen_block: Option<NumberFor<Block>>,
/// Number of peers participating in syncing.
Expand Down
14 changes: 8 additions & 6 deletions client/network/src/service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ use sc_network_common::{
NotificationSender as NotificationSenderT, NotificationSenderError,
NotificationSenderReady as NotificationSenderReadyT, Signature, SigningError,
},
sync::{SyncState, SyncStatus},
sync::SyncStatus,
ExHashT,
};
use sc_peerset::PeersetHandle;
Expand Down Expand Up @@ -1997,11 +1997,13 @@ where
*this.external_addresses.lock() = external_addresses;
}

let is_major_syncing =
match this.network_service.behaviour_mut().user_protocol_mut().sync_state().state {
SyncState::Idle => false,
SyncState::Downloading => true,
};
let is_major_syncing = this
.network_service
.behaviour_mut()
.user_protocol_mut()
.sync_state()
.state
.is_major_syncing();

this.is_major_syncing.store(is_major_syncing, Ordering::Relaxed);

Expand Down
37 changes: 20 additions & 17 deletions client/network/sync/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -410,13 +410,21 @@ where

/// Returns the current sync status.
fn status(&self) -> SyncStatus<B> {
let best_seen = self.best_seen();
let sync_state = if let Some(n) = best_seen {
let median_seen = self.median_seen();
let best_seen_block =
median_seen.and_then(|median| (median > self.best_queued_number).then_some(median));
let sync_state = if let Some(target) = median_seen {
// A chain is classified as downloading if the provided best block is
// more than `MAJOR_SYNC_BLOCKS` behind the best block.
// more than `MAJOR_SYNC_BLOCKS` behind the best block or as importing
// if the same can be said about queued blocks.
let best_block = self.client.info().best_number;
if n > best_block && n - best_block > MAJOR_SYNC_BLOCKS.into() {
SyncState::Downloading
if target > best_block && target - best_block > MAJOR_SYNC_BLOCKS.into() {
// If target is not queued, we're downloading, otherwise importing.
if target > self.best_queued_number {
SyncState::Downloading { target }
} else {
SyncState::Importing { target }
}
} else {
SyncState::Idle
}
Expand All @@ -437,7 +445,7 @@ where

SyncStatus {
state: sync_state,
best_seen_block: best_seen,
best_seen_block,
num_peers: self.peers.len() as u32,
queued_blocks: self.queue_blocks.len() as u32,
state_sync: self.state_sync.as_ref().map(|s| s.progress()),
Expand Down Expand Up @@ -693,7 +701,7 @@ where
trace!(target: "sync", "Too many blocks in the queue.");
return Box::new(std::iter::empty())
}
let major_sync = self.status().state == SyncState::Downloading;
let is_major_syncing = self.status().state.is_major_syncing();
let attrs = self.required_block_attributes();
let blocks = &mut self.blocks;
let fork_targets = &mut self.fork_targets;
Expand All @@ -703,7 +711,7 @@ where
let client = &self.client;
let queue = &self.queue_blocks;
let allowed_requests = self.allowed_requests.take();
let max_parallel = if major_sync { 1 } else { self.max_parallel_downloads };
let max_parallel = if is_major_syncing { 1 } else { self.max_parallel_downloads };
let gap_sync = &mut self.gap_sync;
let iter = self.peers.iter_mut().filter_map(move |(&id, peer)| {
if !peer.state.is_available() || !allowed_requests.contains(&id) {
Expand Down Expand Up @@ -1797,8 +1805,8 @@ where
Ok((sync, Box::new(ChainSyncInterfaceHandle::new(tx))))
}

/// Returns the best seen block number if we don't have that block yet, `None` otherwise.
fn best_seen(&self) -> Option<NumberFor<B>> {
/// Returns the median seen block number.
fn median_seen(&self) -> Option<NumberFor<B>> {
let mut best_seens = self.peers.values().map(|p| p.best_number).collect::<Vec<_>>();

if best_seens.is_empty() {
Expand All @@ -1807,12 +1815,7 @@ where
let middle = best_seens.len() / 2;

// Not the "perfect median" when we have an even number of peers.
let median = *best_seens.select_nth_unstable(middle).1;
if median > self.best_queued_number {
Some(median)
} else {
None
}
Some(*best_seens.select_nth_unstable(middle).1)
}
}

Expand Down Expand Up @@ -1854,7 +1857,7 @@ where
);
}

let origin = if !gap && self.status().state != SyncState::Downloading {
let origin = if !gap && !self.status().state.is_major_syncing() {
BlockOrigin::NetworkBroadcast
} else {
BlockOrigin::NetworkInitialSync
Expand Down

0 comments on commit e458835

Please sign in to comment.