GRIN/grin-node
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cleanup and simplify sync (locator, body, etc.) (#1860)

Browse Source 
* 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
This commit is contained in:
Ignotus Peverell
2018-10-30 14:09:24 -07:00
committed by GitHub
parent d3b491674a
commit 6f14f4c998
3 changed files with 234 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files
servers/src/grin/sync/body_sync.rs
+48 -81
View File
@@ -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<p2p::Peers>,
sync_state: Arc<SyncState>,
prev_body_sync: (DateTime<Utc>, u64),
sync_start_ts: DateTime<Utc>,
body_sync_hashes: Vec<Hash>,
prev_body_received: Option<DateTime<Utc>>,
prev_tip: chain::Tip,
prev_orphans_len: usize,
blocks_requested: u64,
receive_timeout: DateTime<Utc>,
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::<Vec<_>>();
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
}
}
servers/src/grin/sync/header_sync.rs
+151 -110
View File
@@ -26,7 +26,7 @@ pub struct HeaderSync {
peers: Arc<p2p::Peers>,
chain: Arc<chain::Chain>,
history_locators: Vec<(u64, Hash)>,
history_locator: Vec<(u64, Hash)>,
prev_header_sync: (DateTime<Utc>, 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<Vec<Hash>, 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<u64> = 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<Hash> = 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<u64> {
let mut current = height;
@@ -287,6 +229,7 @@ fn get_locator_heights(height: u64) -> Vec<u64> {
#[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<u64> = 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<u64> = 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<u64> = 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()),
]);
}
}
servers/src/grin/sync/syncer.rs
+35 -39
View File
@@ -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())
}
}