From 6f14f4c998ea25f19dddc145b7f3186c392571cf Mon Sep 17 00:00:00 2001 From: Ignotus Peverell Date: Tue, 30 Oct 2018 14:09:24 -0700 Subject: [PATCH] Cleanup and simplify sync (locator, body, etc.) (#1860) * Cleanup syncer and sync header (locator) * Simplify body sync * Remove duplicate head in locator, add greater case in close_enough * Various sync small fixes and tuning after testing * More close_enough tests and related minor fixes --- servers/src/grin/sync/body_sync.rs | 129 +++++-------- servers/src/grin/sync/header_sync.rs | 261 ++++++++++++++++----------- servers/src/grin/sync/syncer.rs | 74 ++++---- 3 files changed, 234 insertions(+), 230 deletions(-) diff --git a/servers/src/grin/sync/body_sync.rs b/servers/src/grin/sync/body_sync.rs index 284fff8b..ac3fe1d7 100644 --- a/servers/src/grin/sync/body_sync.rs +++ b/servers/src/grin/sync/body_sync.rs @@ -19,7 +19,7 @@ use std::sync::Arc; use chain; use common::types::{SyncState, SyncStatus}; -use core::core::hash::{Hash, Hashed, ZERO_HASH}; +use core::core::hash::Hashed; use core::global; use p2p; @@ -28,12 +28,10 @@ pub struct BodySync { peers: Arc, sync_state: Arc, - prev_body_sync: (DateTime, u64), - sync_start_ts: DateTime, - body_sync_hashes: Vec, - prev_body_received: Option>, - prev_tip: chain::Tip, - prev_orphans_len: usize, + blocks_requested: u64, + + receive_timeout: DateTime, + prev_blocks_received: u64, } impl BodySync { @@ -46,19 +44,16 @@ impl BodySync { sync_state, peers, chain, - prev_body_sync: (Utc::now(), 0), - sync_start_ts: Utc::now(), - body_sync_hashes: vec![], - prev_body_received: None, - prev_tip: chain::Tip::new(ZERO_HASH), - prev_orphans_len: 0, + blocks_requested: 0, + receive_timeout: Utc::now(), + prev_blocks_received: 0, } } /// Check whether a body sync is needed and run it if so pub fn check_run(&mut self, head: &chain::Tip, highest_height: u64) -> bool { // if fast_sync disabled or not needed, run the body_sync every 5s - if self.body_sync_due(head) { + if self.body_sync_due() { self.body_sync(); self.sync_state.update(SyncStatus::BodySync { @@ -70,28 +65,12 @@ impl BodySync { false } - fn body_sync_due(&mut self, head: &chain::Tip) -> bool { - let now = Utc::now(); - let (prev_ts, prev_height) = self.prev_body_sync; - - if head.height >= prev_height + 96 - || now - prev_ts > Duration::seconds(5) - || self.block_batch_received() - { - self.prev_body_sync = (now, head.height); - return true; - } - false - } - fn body_sync(&mut self) { let horizon = global::cut_through_horizon() as u64; let body_head = self.chain.head().unwrap(); let header_head = self.chain.header_head().unwrap(); let sync_head = self.chain.get_sync_head().unwrap(); - self.reset(); - debug!( "body_sync: body_head - {}, {}, header_head - {}, {}, sync_head - {}, {}", body_head.last_block_h, @@ -154,75 +133,63 @@ impl BodySync { peers.len(), ); - let mut peers_iter = peers.iter().cycle(); + // reinitialize download tracking state + self.blocks_requested = 0; + self.receive_timeout = Utc::now() + Duration::seconds(6); + let mut peers_iter = peers.iter().cycle(); for hash in hashes_to_get.clone() { if let Some(peer) = peers_iter.next() { if let Err(e) = peer.send_block_request(*hash) { debug!("Skipped request to {}: {:?}", peer.info.addr, e); } else { - self.body_sync_hashes.push(hash.clone()); + self.blocks_requested += 1; } } } } - - self.reset_start(); } - fn reset(&mut self) { - self.body_sync_hashes.clear(); - self.prev_body_received = None; - } + // Should we run block body sync and ask for more full blocks? + fn body_sync_due(&mut self) -> bool { + let blocks_received = self.blocks_received(); - fn reset_start(&mut self) { - self.prev_tip = self.chain.head().unwrap(); - self.prev_orphans_len = self.chain.orphans_len() + self.chain.orphans_evicted_len(); - self.sync_start_ts = Utc::now(); - } - - fn block_batch_received(&mut self) -> bool { - let tip = self.chain.head().unwrap(); - - match self.prev_body_received { - Some(prev_ts) => { - if tip.last_block_h == self.prev_tip.last_block_h - && self.chain.orphans_len() + self.chain.orphans_evicted_len() - == self.prev_orphans_len - && Utc::now() - prev_ts > Duration::milliseconds(200) - { - let hashes_not_get = self - .body_sync_hashes - .iter() - .filter(|x| !self.chain.get_block(*x).is_ok() && !self.chain.is_orphan(*x)) - .collect::>(); - debug!( - "body_sync: {}/{} blocks received, and no more in 200ms", - self.body_sync_hashes.len() - hashes_not_get.len(), - self.body_sync_hashes.len(), - ); - return true; - } - } - None => { - if Utc::now() - self.sync_start_ts > Duration::seconds(5) { - debug!( - "body_sync: 0/{} blocks received in 5s", - self.body_sync_hashes.len(), - ); - return true; - } + // some blocks have been requested + if self.blocks_requested > 0 { + // but none received since timeout, ask again + let timeout = Utc::now() > self.receive_timeout; + if timeout && blocks_received <= self.prev_blocks_received { + debug!( + "body_sync: expecting {} more blocks and none received for a while", + self.blocks_requested, + ); + return true; } } - if tip.last_block_h != self.prev_tip.last_block_h - || self.chain.orphans_len() + self.chain.orphans_evicted_len() != self.prev_orphans_len - { - self.prev_tip = tip; - self.prev_body_received = Some(Utc::now()); - self.prev_orphans_len = self.chain.orphans_len() + self.chain.orphans_evicted_len(); + if blocks_received > self.prev_blocks_received { + // some received, update for next check + self.receive_timeout = Utc::now() + Duration::seconds(1); + self.blocks_requested = self + .blocks_requested + .saturating_sub(blocks_received - self.prev_blocks_received); + self.prev_blocks_received = blocks_received; + } + + // off by one to account for broadcast adding a couple orphans + if self.blocks_requested < 2 { + // no pending block requests, ask more + debug!("body_sync: no pending block request, asking more"); + return true; } return false; } + + // Total numbers received on this chain, including the head and orphans + fn blocks_received(&self) -> u64 { + self.chain.head().unwrap().height + + self.chain.orphans_len() as u64 + + self.chain.orphans_evicted_len() as u64 + } } diff --git a/servers/src/grin/sync/header_sync.rs b/servers/src/grin/sync/header_sync.rs index 1989a9e9..e3f4d528 100644 --- a/servers/src/grin/sync/header_sync.rs +++ b/servers/src/grin/sync/header_sync.rs @@ -26,7 +26,7 @@ pub struct HeaderSync { peers: Arc, chain: Arc, - history_locators: Vec<(u64, Hash)>, + history_locator: Vec<(u64, Hash)>, prev_header_sync: (DateTime, u64, u64), } @@ -40,7 +40,7 @@ impl HeaderSync { sync_state, peers, chain, - history_locators: vec![], + history_locator: vec![], prev_header_sync: (Utc::now(), 0, 0), } } @@ -50,9 +50,7 @@ impl HeaderSync { return false; } - let status = self.sync_state.status(); - - let enable_header_sync = match status { + let enable_header_sync = match self.sync_state.status() { SyncStatus::BodySync { .. } | SyncStatus::HeaderSync { .. } => true, SyncStatus::NoSync | SyncStatus::Initial | SyncStatus::AwaitingPeers(_) => { // Reset sync_head to header_head on transition to HeaderSync, @@ -72,7 +70,7 @@ impl HeaderSync { // Rebuild the sync MMR to match our updates sync_head. self.chain.rebuild_sync_mmr(&header_head).unwrap(); - self.history_locators.clear(); + self.history_locator.clear(); true } _ => false, @@ -151,123 +149,67 @@ impl HeaderSync { /// Even if sync_head is significantly out of date we will "reset" it once we /// start getting headers back from a peer. fn get_locator(&mut self) -> Result, Error> { - let mut this_height = 0; - let tip = self.chain.get_sync_head()?; let heights = get_locator_heights(tip.height); - let mut new_heights: Vec = vec![]; - // for security, clear history_locators[] in any case of header chain rollback, + // for security, clear history_locator[] in any case of header chain rollback, // the easiest way is to check whether the sync head and the header head are identical. - if self.history_locators.len() > 0 && tip.hash() != self.chain.header_head()?.hash() { - self.history_locators.clear(); + if self.history_locator.len() > 0 && tip.hash() != self.chain.header_head()?.hash() { + self.history_locator.clear(); } - debug!("sync: locator heights : {:?}", heights); - - let mut locator: Vec = vec![]; - let mut current = self.chain.get_block_header(&tip.last_block_h); - while let Ok(header) = current { - if heights.contains(&header.height) { - locator.push(header.hash()); - new_heights.push(header.height); - if self.history_locators.len() > 0 - && tip.height - header.height + 1 >= p2p::MAX_BLOCK_HEADERS as u64 - 1 - { - this_height = header.height; - break; + // for each height we need, we either check if something is close enough from + // last locator, or go to the db + let mut locator: Vec<(u64, Hash)> = vec![(tip.height, tip.last_block_h)]; + for h in heights { + if let Some(l) = close_enough(&self.history_locator, h) { + locator.push(l); + } else { + // start at last known hash and go backward + let last_loc = locator.last().unwrap().clone(); + let mut header_cursor = self.chain.get_block_header(&last_loc.1); + while let Ok(header) = header_cursor { + if header.height == h && header.height != last_loc.0 { + locator.push((header.height, header.hash())); + break; + } + header_cursor = self.chain.get_block_header(&header.previous); } } - current = self.chain.get_block_header(&header.previous); - } - - // update history locators - { - let mut tmp: Vec<(u64, Hash)> = vec![]; - *&mut tmp = new_heights - .clone() - .into_iter() - .zip(locator.clone().into_iter()) - .collect(); - tmp.reverse(); - if self.history_locators.len() > 0 && tmp[0].0 == 0 { - tmp = tmp[1..].to_vec(); - } - self.history_locators.append(&mut tmp); - } - - // reuse remaining part of locator from history - if this_height > 0 { - let this_height_index = heights.iter().position(|&r| r == this_height).unwrap(); - let next_height = heights[this_height_index + 1]; - - let reuse_index = self - .history_locators - .iter() - .position(|&r| r.0 >= next_height) - .unwrap(); - let mut tmp = self.history_locators[..reuse_index + 1].to_vec(); - tmp.reverse(); - for (height, hash) in &mut tmp { - if *height == 0 { - break; - } - - // check the locator to make sure the gap >= 2^n, where n = index of heights Vec - if this_height >= *height + 2u64.pow(locator.len() as u32) { - locator.push(hash.clone()); - this_height = *height; - new_heights.push(this_height); - } - if locator.len() >= (p2p::MAX_LOCATORS as usize) - 1 { - break; - } - } - - // push height 0 if it's not there - if new_heights[new_heights.len() - 1] != 0 { - locator.push( - self.history_locators[self.history_locators.len() - 1] - .1 - .clone(), - ); - new_heights.push(0); - } } + locator.dedup_by(|a, b| a.0 == b.0); + debug!("sync: locator : {:?}", locator.clone()); + self.history_locator = locator.clone(); - debug!("sync: locator heights': {:?}", new_heights); - - // shrink history_locators properly - if heights.len() > 1 { - let shrink_height = heights[heights.len() - 2]; - let mut shrunk_size = 0; - let shrink_index = self - .history_locators - .iter() - .position(|&r| r.0 > shrink_height); - if let Some(shrink_index) = shrink_index { - if shrink_index > 100 { - // shrink but avoid trivial shrinking - let mut shrunk = self.history_locators[shrink_index..].to_vec(); - shrunk_size = shrink_index; - self.history_locators.clear(); - self.history_locators.push((0, locator[locator.len() - 1])); - self.history_locators.append(&mut shrunk); - } - } - debug!( - "sync: history locators: len={}, shrunk={}", - self.history_locators.len(), - shrunk_size - ); - } - - debug!("sync: locator: {:?}", locator); - - Ok(locator) + Ok(locator.iter().map(|l| l.1).collect()) } } +// Whether we have a value close enough to the provided height in the locator +fn close_enough(locator: &Vec<(u64, Hash)>, height: u64) -> Option<(u64, Hash)> { + if locator.len() == 0 { + return None; + } + // bounds, lower that last is last + if locator.last().unwrap().0 >= height { + return locator.last().map(|l| l.clone()); + } + // higher than first is first if within an acceptable gap + if locator[0].0 < height && height.saturating_sub(127) < locator[0].0 { + return Some(locator[0]); + } + for hh in locator.windows(2) { + if height <= hh[0].0 && height > hh[1].0 { + if hh[0].0 - height < height - hh[1].0 { + return Some(hh[0].clone()); + } else { + return Some(hh[1].clone()); + } + } + } + None +} + // current height back to 0 decreasing in powers of 2 fn get_locator_heights(height: u64) -> Vec { let mut current = height; @@ -287,6 +229,7 @@ fn get_locator_heights(height: u64) -> Vec { #[cfg(test)] mod test { use super::*; + use core::core::hash; #[test] fn test_get_locator_heights() { @@ -307,4 +250,102 @@ mod test { vec![10000, 9998, 9994, 9986, 9970, 9938, 9874, 9746, 9490, 8978, 7954, 5906, 1810, 0,] ); } + + #[test] + fn test_close_enough() { + let zh = hash::ZERO_HASH; + + // empty check + assert_eq!(close_enough(&vec![], 0), None); + + // just 1 locator in history + let heights: Vec = vec![64, 62, 58, 50, 34, 2, 0]; + let history_locator: Vec<(u64, Hash)> = vec![ + (0, zh.clone()), + ]; + let mut locator: Vec<(u64, Hash)> = vec![]; + for h in heights { + if let Some(l) = close_enough(&history_locator, h) { + locator.push(l); + } + } + assert_eq!(locator, vec![(0, zh.clone())]); + + // simple dummy example + let locator = vec![ + (1000, zh.clone()), + (500, zh.clone()), + (250, zh.clone()), + (125, zh.clone()), + ]; + assert_eq!(close_enough(&locator, 2000), None); + assert_eq!(close_enough(&locator, 1050), Some((1000, zh))); + assert_eq!(close_enough(&locator, 900), Some((1000, zh))); + assert_eq!(close_enough(&locator, 270), Some((250, zh))); + assert_eq!(close_enough(&locator, 20), Some((125, zh))); + assert_eq!(close_enough(&locator, 125), Some((125, zh))); + assert_eq!(close_enough(&locator, 500), Some((500, zh))); + + // more realistic test with 11 history + let heights: Vec = vec![ + 2554, 2552, 2548, 2540, 2524, 2492, 2428, 2300, 2044, 1532, 508, 0 + ]; + let history_locator: Vec<(u64, Hash)> = vec![ + (2043, zh.clone()), + (2041, zh.clone()), + (2037, zh.clone()), + (2029, zh.clone()), + (2013, zh.clone()), + (1981, zh.clone()), + (1917, zh.clone()), + (1789, zh.clone()), + (1532, zh.clone()), + (1021, zh.clone()), + (0, zh.clone()), + ]; + let mut locator: Vec<(u64, Hash)> = vec![]; + for h in heights { + if let Some(l) = close_enough(&history_locator, h) { + locator.push(l); + } + } + locator.dedup_by(|a, b| a.0 == b.0); + assert_eq!(locator, vec![ + (2043, zh.clone()), + (1532, zh.clone()), + (0, zh.clone()), + ]); + + // more realistic test with 12 history + let heights: Vec = vec![ + 4598, 4596, 4592, 4584, 4568, 4536, 4472, 4344, 4088, 3576, 2552, 504, 0 + ]; + let history_locator: Vec<(u64, Hash)> = vec![ + (4087, zh.clone()), + (4085, zh.clone()), + (4081, zh.clone()), + (4073, zh.clone()), + (4057, zh.clone()), + (4025, zh.clone()), + (3961, zh.clone()), + (3833, zh.clone()), + (3576, zh.clone()), + (3065, zh.clone()), + (1532, zh.clone()), + (0, zh.clone()), + ]; + let mut locator: Vec<(u64, Hash)> = vec![]; + for h in heights { + if let Some(l) = close_enough(&history_locator, h) { + locator.push(l); + } + } + locator.dedup_by(|a, b| a.0 == b.0); + assert_eq!(locator, vec![ + (4087, zh.clone()), + (3576, zh.clone()), + (3065, zh.clone()), + (0, zh.clone()), + ]); + } } diff --git a/servers/src/grin/sync/syncer.rs b/servers/src/grin/sync/syncer.rs index 562e802a..77cda73a 100644 --- a/servers/src/grin/sync/syncer.rs +++ b/servers/src/grin/sync/syncer.rs @@ -156,55 +156,51 @@ impl SyncRunner { /// just receiving blocks through gossip. fn needs_syncing(&self) -> (bool, u64) { let local_diff = self.chain.head().unwrap().total_difficulty; + let mut is_syncing = self.sync_state.is_syncing(); let peer = self.peers.most_work_peer(); - let is_syncing = self.sync_state.is_syncing(); - let mut most_work_height = 0; + + let peer_info = if let Some(p) = peer { + p.info.clone() + } else { + warn!("sync: no peers available, disabling sync"); + return (false, 0); + }; // if we're already syncing, we're caught up if no peer has a higher // difficulty than us if is_syncing { - if let Some(peer) = peer { - most_work_height = peer.info.height(); - if peer.info.total_difficulty() <= local_diff { - let ch = self.chain.head().unwrap(); - info!( - "synchronized at {} @ {} [{}]", - local_diff.to_num(), - ch.height, - ch.last_block_h - ); + if peer_info.total_difficulty() <= local_diff { + let ch = self.chain.head().unwrap(); + info!( + "synchronized at {} @ {} [{}]", + local_diff.to_num(), + ch.height, + ch.last_block_h + ); - let _ = self.chain.reset_head(); - return (false, most_work_height); - } - } else { - warn!("sync: no peers available, disabling sync"); - return (false, 0); + let _ = self.chain.reset_head(); + is_syncing = false; } } else { - if let Some(peer) = peer { - most_work_height = peer.info.height(); + // sum the last 5 difficulties to give us the threshold + let threshold = self + .chain + .difficulty_iter() + .map(|x| x.difficulty) + .take(5) + .fold(Difficulty::zero(), |sum, val| sum + val); - // sum the last 5 difficulties to give us the threshold - let threshold = self - .chain - .difficulty_iter() - .map(|x| x.difficulty) - .take(5) - .fold(Difficulty::zero(), |sum, val| sum + val); - - let peer_diff = peer.info.total_difficulty(); - if peer_diff > local_diff.clone() + threshold.clone() { - info!( - "sync: total_difficulty {}, peer_difficulty {}, threshold {} (last 5 blocks), enabling sync", - local_diff, - peer_diff, - threshold, - ); - return (true, most_work_height); - } + let peer_diff = peer_info.total_difficulty(); + if peer_diff > local_diff.clone() + threshold.clone() { + info!( + "sync: total_difficulty {}, peer_difficulty {}, threshold {} (last 5 blocks), enabling sync", + local_diff, + peer_diff, + threshold, + ); + is_syncing = true; } } - (is_syncing, most_work_height) + (is_syncing, peer_info.height()) } }