grassrootseconomics/openethereum
8
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request #420 from ethcore/net

Browse Source 
More sync and propagation fixes
This commit is contained in:
Gav Wood 2016-02-15 14:06:36 +01:00
commit 9f03640a06
4 changed files with 169 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

191
sync/src/chain.rs
View File

@ -155,13 +155,17 @@ struct PeerInfo {
/// Peer network id /// Peer network id
network_id: U256, network_id: U256,
/// Peer best block hash /// Peer best block hash
latest: H256, latest_hash: H256,
/// Peer best block number if known
latest_number: Option<BlockNumber>,
/// Peer total difficulty /// Peer total difficulty
difficulty: U256, difficulty: U256,
/// Type of data currenty being requested from peer. /// Type of data currenty being requested from peer.
asking: PeerAsking, asking: PeerAsking,
/// A set of block numbers being requested /// A set of block numbers being requested
asking_blocks: Vec<BlockNumber>, asking_blocks: Vec<BlockNumber>,
/// Holds requested header hash if currently requesting block header by hash
asking_hash: Option<H256>,
/// Request timestamp /// Request timestamp
ask_time: f64, ask_time: f64,
} }
@ -179,6 +183,8 @@ pub struct ChainSync {
downloading_headers: HashSet<BlockNumber>, downloading_headers: HashSet<BlockNumber>,
/// Set of block body numbers being downloaded /// Set of block body numbers being downloaded
downloading_bodies: HashSet<BlockNumber>, downloading_bodies: HashSet<BlockNumber>,
/// Set of block headers being downloaded by hash
downloading_hashes: HashSet<H256>,
/// Downloaded headers. /// Downloaded headers.
headers: Vec<(BlockNumber, Vec<Header>)>, //TODO: use BTreeMap once range API is sable. For now it is a vector sorted in descending order headers: Vec<(BlockNumber, Vec<Header>)>, //TODO: use BTreeMap once range API is sable. For now it is a vector sorted in descending order
/// Downloaded bodies /// Downloaded bodies
@ -195,6 +201,8 @@ pub struct ChainSync {
syncing_difficulty: U256, syncing_difficulty: U256,
/// True if common block for our and remote chain has been found /// True if common block for our and remote chain has been found
have_common_block: bool, have_common_block: bool,
/// Last propagated block number
last_send_block_number: BlockNumber,
} }
type RlpResponseResult = Result<Option<(PacketId, RlpStream)>, PacketDecodeError>; type RlpResponseResult = Result<Option<(PacketId, RlpStream)>, PacketDecodeError>;
@ -208,6 +216,7 @@ impl ChainSync {
highest_block: None, highest_block: None,
downloading_headers: HashSet::new(), downloading_headers: HashSet::new(),
downloading_bodies: HashSet::new(), downloading_bodies: HashSet::new(),
downloading_hashes: HashSet::new(),
headers: Vec::new(), headers: Vec::new(),
bodies: Vec::new(), bodies: Vec::new(),
peers: HashMap::new(), peers: HashMap::new(),
@ -216,6 +225,7 @@ impl ChainSync {
last_imported_hash: None, last_imported_hash: None,
syncing_difficulty: U256::from(0u64), syncing_difficulty: U256::from(0u64),
have_common_block: false, have_common_block: false,
last_send_block_number: 0,
} }
} }
@ -250,6 +260,7 @@ impl ChainSync {
self.bodies.clear(); self.bodies.clear();
for (_, ref mut p) in &mut self.peers { for (_, ref mut p) in &mut self.peers {
p.asking_blocks.clear(); p.asking_blocks.clear();
p.asking_hash = None;
} }
self.header_ids.clear(); self.header_ids.clear();
self.syncing_difficulty = From::from(0u64); self.syncing_difficulty = From::from(0u64);
@ -275,15 +286,21 @@ impl ChainSync {
protocol_version: try!(r.val_at(0)), protocol_version: try!(r.val_at(0)),
network_id: try!(r.val_at(1)), network_id: try!(r.val_at(1)),
difficulty: try!(r.val_at(2)), difficulty: try!(r.val_at(2)),
latest: try!(r.val_at(3)), latest_hash: try!(r.val_at(3)),
latest_number: None,
genesis: try!(r.val_at(4)), genesis: try!(r.val_at(4)),
asking: PeerAsking::Nothing, asking: PeerAsking::Nothing,
asking_blocks: Vec::new(), asking_blocks: Vec::new(),
asking_hash: None,
ask_time: 0f64, ask_time: 0f64,
}; };
trace!(target: "sync", "New peer {} (protocol: {}, network: {:?}, difficulty: {:?}, latest:{}, genesis:{})", peer_id, peer.protocol_version, peer.network_id, peer.difficulty, peer.latest, peer.genesis); trace!(target: "sync", "New peer {} (protocol: {}, network: {:?}, difficulty: {:?}, latest:{}, genesis:{})", peer_id, peer.protocol_version, peer.network_id, peer.difficulty, peer.latest_hash, peer.genesis);
if self.peers.contains_key(&peer_id) {
warn!("Unexpected status packet from {}:{}", peer_id, io.peer_info(peer_id));
return Ok(());
}
let chain_info = io.chain().chain_info(); let chain_info = io.chain().chain_info();
if peer.genesis != chain_info.genesis_hash { if peer.genesis != chain_info.genesis_hash {
io.disable_peer(peer_id); io.disable_peer(peer_id);
@ -296,10 +313,7 @@ impl ChainSync {
return Ok(()); return Ok(());
} }
let old = self.peers.insert(peer_id.clone(), peer); self.peers.insert(peer_id.clone(), peer);
if old.is_some() {
panic!("ChainSync: new peer already exists");
}
info!(target: "sync", "Connected {}:{}", peer_id, io.peer_info(peer_id)); info!(target: "sync", "Connected {}:{}", peer_id, io.peer_info(peer_id));
self.sync_peer(io, peer_id, false); self.sync_peer(io, peer_id, false);
Ok(()) Ok(())
@ -439,6 +453,11 @@ impl ChainSync {
trace!(target: "sync", "{} -> NewBlock ({})", peer_id, h); trace!(target: "sync", "{} -> NewBlock ({})", peer_id, h);
let header: BlockHeader = try!(header_rlp.as_val()); let header: BlockHeader = try!(header_rlp.as_val());
let mut unknown = false; let mut unknown = false;
{
let peer = self.peers.get_mut(&peer_id).unwrap();
peer.latest_hash = header.hash();
peer.latest_number = Some(header.number());
}
// TODO: Decompose block and add to self.headers and self.bodies instead // TODO: Decompose block and add to self.headers and self.bodies instead
if header.number == From::from(self.current_base_block() + 1) { if header.number == From::from(self.current_base_block() + 1) {
match io.chain().import_block(block_rlp.as_raw().to_vec()) { match io.chain().import_block(block_rlp.as_raw().to_vec()) {
@ -449,6 +468,7 @@ impl ChainSync {
trace!(target: "sync", "New block already queued {:?}", h); trace!(target: "sync", "New block already queued {:?}", h);
}, },
Ok(_) => { Ok(_) => {
self.last_imported_block = Some(header.number);
trace!(target: "sync", "New block queued {:?}", h); trace!(target: "sync", "New block queued {:?}", h);
}, },
Err(ImportError::UnknownParent) => { Err(ImportError::UnknownParent) => {
@ -468,13 +488,9 @@ impl ChainSync {
trace!(target: "sync", "New block unknown {:?}", h); trace!(target: "sync", "New block unknown {:?}", h);
//TODO: handle too many unknown blocks //TODO: handle too many unknown blocks
let difficulty: U256 = try!(r.val_at(1)); let difficulty: U256 = try!(r.val_at(1));
let peer_difficulty = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer").difficulty; let peer_difficulty = self.peers.get_mut(&peer_id).unwrap().difficulty;
if difficulty > peer_difficulty { if difficulty > peer_difficulty {
trace!(target: "sync", "Received block {:?} with no known parent. Peer needs syncing...", h); trace!(target: "sync", "Received block {:?} with no known parent. Peer needs syncing...", h);
{
let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer");
peer.latest = header.hash();
}
self.sync_peer(io, peer_id, true); self.sync_peer(io, peer_id, true);
} }
} }
@ -483,16 +499,19 @@ impl ChainSync {
/// Handles NewHashes packet. Initiates headers download for any unknown hashes. /// Handles NewHashes packet. Initiates headers download for any unknown hashes.
fn on_peer_new_hashes(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> { fn on_peer_new_hashes(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
if self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer").asking != PeerAsking::Nothing { if self.peers.get_mut(&peer_id).unwrap().asking != PeerAsking::Nothing {
trace!(target: "sync", "Ignoring new hashes since we're already downloading."); trace!(target: "sync", "Ignoring new hashes since we're already downloading.");
return Ok(()); return Ok(());
} }
trace!(target: "sync", "{} -> NewHashes ({} entries)", peer_id, r.item_count()); trace!(target: "sync", "{} -> NewHashes ({} entries)", peer_id, r.item_count());
let hashes = r.iter().map(|item| (item.val_at::<H256>(0), item.val_at::<U256>(1))); let hashes = r.iter().map(|item| (item.val_at::<H256>(0), item.val_at::<BlockNumber>(1)));
let mut max_height: U256 = From::from(0); let mut max_height: BlockNumber = 0;
for (rh, rd) in hashes { for (rh, rd) in hashes {
let h = try!(rh); let h = try!(rh);
let d = try!(rd); let d = try!(rd);
if self.downloading_hashes.contains(&h) {
continue;
}
match io.chain().block_status(BlockId::Hash(h.clone())) { match io.chain().block_status(BlockId::Hash(h.clone())) {
BlockStatus::InChain => { BlockStatus::InChain => {
trace!(target: "sync", "New block hash already in chain {:?}", h); trace!(target: "sync", "New block hash already in chain {:?}", h);
@ -501,10 +520,11 @@ impl ChainSync {
trace!(target: "sync", "New hash block already queued {:?}", h); trace!(target: "sync", "New hash block already queued {:?}", h);
}, },
BlockStatus::Unknown => { BlockStatus::Unknown => {
trace!(target: "sync", "New unknown block hash {:?}", h);
if d > max_height { if d > max_height {
let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer"); trace!(target: "sync", "New unknown block hash {:?}", h);
peer.latest = h.clone(); let peer = self.peers.get_mut(&peer_id).unwrap();
peer.latest_hash = h.clone();
peer.latest_number = Some(d);
max_height = d; max_height = d;
} }
}, },
@ -515,7 +535,7 @@ impl ChainSync {
} }
} }
}; };
if max_height != x!(0) { if max_height != 0 {
self.sync_peer(io, peer_id, true); self.sync_peer(io, peer_id, true);
} }
Ok(()) Ok(())
@ -525,7 +545,7 @@ impl ChainSync {
pub fn on_peer_aborting(&mut self, io: &mut SyncIo, peer: PeerId) { pub fn on_peer_aborting(&mut self, io: &mut SyncIo, peer: PeerId) {
trace!(target: "sync", "== Disconnecting {}", peer); trace!(target: "sync", "== Disconnecting {}", peer);
if self.peers.contains_key(&peer) { if self.peers.contains_key(&peer) {
info!(target: "sync", "Disconnected {}:{}", peer, io.peer_info(peer)); info!(target: "sync", "Disconnected {}", peer);
self.clear_peer_download(peer); self.clear_peer_download(peer);
self.peers.remove(&peer); self.peers.remove(&peer);
self.continue_sync(io); self.continue_sync(io);
@ -563,7 +583,7 @@ impl ChainSync {
/// Find something to do for a peer. Called for a new peer or when a peer is done with it's task. /// Find something to do for a peer. Called for a new peer or when a peer is done with it's task.
fn sync_peer(&mut self, io: &mut SyncIo, peer_id: PeerId, force: bool) { fn sync_peer(&mut self, io: &mut SyncIo, peer_id: PeerId, force: bool) {
let (peer_latest, peer_difficulty) = { let (peer_latest, peer_difficulty) = {
let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer"); let peer = self.peers.get_mut(&peer_id).unwrap();
if peer.asking != PeerAsking::Nothing { if peer.asking != PeerAsking::Nothing {
return; return;
} }
@ -571,7 +591,7 @@ impl ChainSync {
trace!(target: "sync", "Waiting for block queue"); trace!(target: "sync", "Waiting for block queue");
return; return;
} }
(peer.latest.clone(), peer.difficulty.clone()) (peer.latest_hash.clone(), peer.difficulty.clone())
}; };
let td = io.chain().chain_info().pending_total_difficulty; let td = io.chain().chain_info().pending_total_difficulty;
@ -583,6 +603,8 @@ impl ChainSync {
self.state = SyncState::Blocks; self.state = SyncState::Blocks;
} }
trace!(target: "sync", "Starting sync with better chain"); trace!(target: "sync", "Starting sync with better chain");
self.peers.get_mut(&peer_id).unwrap().asking_hash = Some(peer_latest.clone());
self.downloading_hashes.insert(peer_latest.clone());
self.request_headers_by_hash(io, peer_id, &peer_latest, 1, 0, false); self.request_headers_by_hash(io, peer_id, &peer_latest, 1, 0, false);
} }
else if self.state == SyncState::Blocks && io.chain().block_status(BlockId::Hash(peer_latest)) == BlockStatus::Unknown { else if self.state == SyncState::Blocks && io.chain().block_status(BlockId::Hash(peer_latest)) == BlockStatus::Unknown {
@ -675,6 +697,8 @@ impl ChainSync {
} }
} }
else { else {
// continue search for common block
self.downloading_headers.insert(start as BlockNumber);
self.request_headers_by_number(io, peer_id, start as BlockNumber, 1, 0, false); self.request_headers_by_number(io, peer_id, start as BlockNumber, 1, 0, false);
} }
} }
@ -682,7 +706,10 @@ impl ChainSync {
/// Clear all blocks/headers marked as being downloaded by a peer. /// Clear all blocks/headers marked as being downloaded by a peer.
fn clear_peer_download(&mut self, peer_id: PeerId) { fn clear_peer_download(&mut self, peer_id: PeerId) {
let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer"); let peer = self.peers.get_mut(&peer_id).unwrap();
if let Some(hash) = peer.asking_hash.take() {
self.downloading_hashes.remove(&hash);
}
for b in &peer.asking_blocks { for b in &peer.asking_blocks {
self.downloading_headers.remove(&b); self.downloading_headers.remove(&b);
self.downloading_bodies.remove(&b); self.downloading_bodies.remove(&b);
@ -815,7 +842,7 @@ impl ChainSync {
/// Reset peer status after request is complete. /// Reset peer status after request is complete.
fn reset_peer_asking(&mut self, peer_id: PeerId, asking: PeerAsking) { fn reset_peer_asking(&mut self, peer_id: PeerId, asking: PeerAsking) {
let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer"); let peer = self.peers.get_mut(&peer_id).unwrap();
if peer.asking != asking { if peer.asking != asking {
warn!(target:"sync", "Asking {:?} while expected {:?}", peer.asking, asking); warn!(target:"sync", "Asking {:?} while expected {:?}", peer.asking, asking);
} }
@ -827,9 +854,9 @@ impl ChainSync {
/// Generic request sender /// Generic request sender
fn send_request(&mut self, sync: &mut SyncIo, peer_id: PeerId, asking: PeerAsking, packet_id: PacketId, packet: Bytes) { fn send_request(&mut self, sync: &mut SyncIo, peer_id: PeerId, asking: PeerAsking, packet_id: PacketId, packet: Bytes) {
{ {
let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer"); let peer = self.peers.get_mut(&peer_id).unwrap();
if peer.asking != PeerAsking::Nothing { if peer.asking != PeerAsking::Nothing {
warn!(target:"sync", "Asking {:?} while requesting {:?}", asking, peer.asking); warn!(target:"sync", "Asking {:?} while requesting {:?}", peer.asking, asking);
} }
} }
match sync.send(peer_id, packet_id, packet) { match sync.send(peer_id, packet_id, packet) {
@ -846,6 +873,14 @@ impl ChainSync {
} }
} }
/// Generic packet sender
fn send_packet(&mut self, sync: &mut SyncIo, peer_id: PeerId, packet_id: PacketId, packet: Bytes) {
if let Err(e) = sync.send(peer_id, packet_id, packet) {
warn!(target:"sync", "Error sending packet: {:?}", e);
sync.disable_peer(peer_id);
self.on_peer_aborting(sync, peer_id);
}
}
/// Called when peer sends us new transactions /// Called when peer sends us new transactions
fn on_peer_transactions(&mut self, _io: &mut SyncIo, _peer_id: PeerId, _r: &UntrustedRlp) -> Result<(), PacketDecodeError> { fn on_peer_transactions(&mut self, _io: &mut SyncIo, _peer_id: PeerId, _r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
Ok(()) Ok(())
@ -1002,6 +1037,11 @@ impl ChainSync {
/// Dispatch incoming requests and responses /// Dispatch incoming requests and responses
pub fn on_packet(&mut self, io: &mut SyncIo, peer: PeerId, packet_id: u8, data: &[u8]) { pub fn on_packet(&mut self, io: &mut SyncIo, peer: PeerId, packet_id: u8, data: &[u8]) {
let rlp = UntrustedRlp::new(data); let rlp = UntrustedRlp::new(data);
if packet_id != STATUS_PACKET && !self.peers.contains_key(&peer) {
warn!(target:"sync", "Unexpected packet from unregistered peer: {}:{}", peer, io.peer_info(peer));
return;
}
let result = match packet_id { let result = match packet_id {
STATUS_PACKET => self.on_peer_status(io, peer, &rlp), STATUS_PACKET => self.on_peer_status(io, peer, &rlp),
TRANSACTIONS_PACKET => self.on_peer_transactions(io, peer, &rlp), TRANSACTIONS_PACKET => self.on_peer_transactions(io, peer, &rlp),
@ -1063,7 +1103,6 @@ impl ChainSync {
for block_hash in route.blocks { for block_hash in route.blocks {
let mut hash_rlp = RlpStream::new_list(2); let mut hash_rlp = RlpStream::new_list(2);
let difficulty = chain.block_total_difficulty(BlockId::Hash(block_hash.clone())).expect("Mallformed block without a difficulty on the chain!"); let difficulty = chain.block_total_difficulty(BlockId::Hash(block_hash.clone())).expect("Mallformed block without a difficulty on the chain!");
hash_rlp.append(&block_hash); hash_rlp.append(&block_hash);
hash_rlp.append(&difficulty); hash_rlp.append(&difficulty);
rlp_stream.append_raw(&hash_rlp.out(), 1); rlp_stream.append_raw(&hash_rlp.out(), 1);
@ -1079,32 +1118,34 @@ impl ChainSync {
/// creates latest block rlp for the given client /// creates latest block rlp for the given client
fn create_latest_block_rlp(chain: &BlockChainClient) -> Bytes { fn create_latest_block_rlp(chain: &BlockChainClient) -> Bytes {
let mut rlp_stream = RlpStream::new_list(2); let mut rlp_stream = RlpStream::new_list(2);
rlp_stream.append_raw(&chain.block(BlockId::Hash(chain.chain_info().best_block_hash)).expect("Creating latest block when there is none"), 1); rlp_stream.append_raw(&chain.block(BlockId::Hash(chain.chain_info().best_block_hash)).unwrap(), 1);
rlp_stream.append(&chain.chain_info().total_difficulty); rlp_stream.append(&chain.chain_info().total_difficulty);
rlp_stream.out() rlp_stream.out()
} }
/// returns peer ids that have less blocks than our chain /// returns peer ids that have less blocks than our chain
fn get_lagging_peers(&self, io: &SyncIo) -> Vec<PeerId> { fn get_lagging_peers(&mut self, io: &SyncIo) -> Vec<(PeerId, BlockNumber)> {
let chain = io.chain(); let chain = io.chain();
let chain_info = chain.chain_info(); let chain_info = chain.chain_info();
let latest_hash = chain_info.best_block_hash; let latest_hash = chain_info.best_block_hash;
let latest_number = chain_info.best_block_number; let latest_number = chain_info.best_block_number;
self.peers.iter().filter(|&(_, peer_info)| self.peers.iter_mut().filter_map(|(&id, ref mut peer_info)|
match io.chain().block_status(BlockId::Hash(peer_info.latest.clone())) match io.chain().block_status(BlockId::Hash(peer_info.latest_hash.clone())) {
{
BlockStatus::InChain => { BlockStatus::InChain => {
let peer_number = HeaderView::new(&io.chain().block_header(BlockId::Hash(peer_info.latest.clone())).unwrap()).number(); if peer_info.latest_number.is_none() {
peer_info.latest != latest_hash && latest_number > peer_number && latest_number - peer_number < MAX_PEER_LAG_PROPAGATION peer_info.latest_number = Some(HeaderView::new(&io.chain().block_header(BlockId::Hash(peer_info.latest_hash.clone())).unwrap()).number());
}
if peer_info.latest_hash != latest_hash && latest_number > peer_info.latest_number.unwrap() {
Some((id, peer_info.latest_number.unwrap()))
} else { None }
}, },
_ => false _ => None
}) })
.map(|(peer_id, _)| peer_id) .collect::<Vec<_>>()
.cloned().collect::<Vec<PeerId>>()
} }
/// propagades latest block to lagging peers /// propagades latest block to lagging peers
fn propagade_blocks(&mut self, io: &mut SyncIo) -> usize { fn propagade_blocks(&mut self, local_best: &H256, best_number: BlockNumber, io: &mut SyncIo) -> usize {
let updated_peers = { let updated_peers = {
let lagging_peers = self.get_lagging_peers(io); let lagging_peers = self.get_lagging_peers(io);
@ -1112,37 +1153,43 @@ impl ChainSync {
let fraction = (self.peers.len() as f64).powf(-0.5).mul(u32::max_value() as f64).round() as u32; let fraction = (self.peers.len() as f64).powf(-0.5).mul(u32::max_value() as f64).round() as u32;
let lucky_peers = match lagging_peers.len() { let lucky_peers = match lagging_peers.len() {
0 ... MIN_PEERS_PROPAGATION => lagging_peers, 0 ... MIN_PEERS_PROPAGATION => lagging_peers,
_ => lagging_peers.iter().filter(|_| ::rand::random::<u32>() < fraction).cloned().collect::<Vec<PeerId>>() _ => lagging_peers.into_iter().filter(|_| ::rand::random::<u32>() < fraction).collect::<Vec<_>>()
}; };
// taking at max of MAX_PEERS_PROPAGATION // taking at max of MAX_PEERS_PROPAGATION
lucky_peers.iter().take(min(lucky_peers.len(), MAX_PEERS_PROPAGATION)).cloned().collect::<Vec<PeerId>>() lucky_peers.iter().map(|&(id, _)| id.clone()).take(min(lucky_peers.len(), MAX_PEERS_PROPAGATION)).collect::<Vec<PeerId>>()
}; };
let mut sent = 0; let mut sent = 0;
let local_best = io.chain().chain_info().best_block_hash;
for peer_id in updated_peers { for peer_id in updated_peers {
let rlp = ChainSync::create_latest_block_rlp(io.chain()); let rlp = ChainSync::create_latest_block_rlp(io.chain());
self.send_request(io, peer_id, PeerAsking::Nothing, NEW_BLOCK_PACKET, rlp); self.send_packet(io, peer_id, NEW_BLOCK_PACKET, rlp);
self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer").latest = local_best.clone(); self.peers.get_mut(&peer_id).unwrap().latest_hash = local_best.clone();
self.peers.get_mut(&peer_id).unwrap().latest_number = Some(best_number);
sent = sent + 1; sent = sent + 1;
} }
sent sent
} }
/// propagades new known hashes to all peers /// propagades new known hashes to all peers
fn propagade_new_hashes(&mut self, io: &mut SyncIo) -> usize { fn propagade_new_hashes(&mut self, local_best: &H256, best_number: BlockNumber, io: &mut SyncIo) -> usize {
let updated_peers = self.get_lagging_peers(io); let updated_peers = self.get_lagging_peers(io);
let mut sent = 0; let mut sent = 0;
let local_best = io.chain().chain_info().best_block_hash; let last_parent = HeaderView::new(&io.chain().block_header(BlockId::Hash(local_best.clone())).unwrap()).parent_hash();
for peer_id in updated_peers { for (peer_id, peer_number) in updated_peers {
sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &self.peers.get(&peer_id).expect("ChainSync: unknown peer").latest, &local_best) { let mut peer_best = self.peers.get(&peer_id).unwrap().latest_hash.clone();
if best_number - peer_number > MAX_PEERS_PROPAGATION as BlockNumber {
// If we think peer is too far behind just end one latest hash
peer_best = last_parent.clone();
}
sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &peer_best, &local_best) {
Some(rlp) => { Some(rlp) => {
{ {
let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer"); let peer = self.peers.get_mut(&peer_id).unwrap();
peer.latest = local_best.clone(); peer.latest_hash = local_best.clone();
peer.latest_number = Some(best_number);
} }
self.send_request(io, peer_id, PeerAsking::Nothing, NEW_BLOCK_HASHES_PACKET, rlp); self.send_packet(io, peer_id, NEW_BLOCK_HASHES_PACKET, rlp);
1 1
}, },
None => 0 None => 0
@ -1154,15 +1201,19 @@ impl ChainSync {
/// Maintain other peers. Send out any new blocks and transactions /// Maintain other peers. Send out any new blocks and transactions
pub fn maintain_sync(&mut self, io: &mut SyncIo) { pub fn maintain_sync(&mut self, io: &mut SyncIo) {
self.check_resume(io); self.check_resume(io);
let peers = self.propagade_new_hashes(io);
trace!(target: "sync", "Sent new hashes to peers: {:?}", peers);
} }
/// should be called once chain has new block, triggers the latest block propagation /// should be called once chain has new block, triggers the latest block propagation
pub fn chain_blocks_verified(&mut self, io: &mut SyncIo) { pub fn chain_blocks_verified(&mut self, io: &mut SyncIo) {
let peers = self.propagade_blocks(io); let chain = io.chain().chain_info();
trace!(target: "sync", "Sent latest block to peers: {:?}", peers); if (((chain.best_block_number as i64) - (self.last_send_block_number as i64)).abs() as BlockNumber) < MAX_PEER_LAG_PROPAGATION {
let blocks = self.propagade_blocks(&chain.best_block_hash, chain.best_block_number, io);
let hashes = self.propagade_new_hashes(&chain.best_block_hash, chain.best_block_number, io);
if blocks != 0 || hashes != 0 {
trace!(target: "sync", "Sent latest {} blocks and {} hashes to peers.", blocks, hashes);
}
}
self.last_send_block_number = chain.best_block_number;
} }
} }
@ -1248,9 +1299,9 @@ mod tests {
// the length of two rlp-encoded receipts // the length of two rlp-encoded receipts
assert_eq!(597, rlp_result.unwrap().1.out().len()); assert_eq!(597, rlp_result.unwrap().1.out().len());
let mut sync = ChainSync::new(); let mut sync = dummy_sync_with_peer(H256::new());
io.sender = Some(2usize); io.sender = Some(2usize);
sync.on_packet(&mut io, 1usize, super::GET_RECEIPTS_PACKET, &receipts_request); sync.on_packet(&mut io, 0usize, super::GET_RECEIPTS_PACKET, &receipts_request);
assert_eq!(1, io.queue.len()); assert_eq!(1, io.queue.len());
} }
@ -1276,9 +1327,9 @@ mod tests {
// the length of one rlp-encoded hashe // the length of one rlp-encoded hashe
assert_eq!(34, rlp_result.unwrap().1.out().len()); assert_eq!(34, rlp_result.unwrap().1.out().len());
let mut sync = ChainSync::new(); let mut sync = dummy_sync_with_peer(H256::new());
io.sender = Some(2usize); io.sender = Some(2usize);
sync.on_packet(&mut io, 1usize, super::GET_NODE_DATA_PACKET, &node_request); sync.on_packet(&mut io, 0usize, super::GET_NODE_DATA_PACKET, &node_request);
assert_eq!(1, io.queue.len()); assert_eq!(1, io.queue.len());
} }
@ -1289,10 +1340,12 @@ mod tests {
protocol_version: 0, protocol_version: 0,
genesis: H256::zero(), genesis: H256::zero(),
network_id: U256::zero(), network_id: U256::zero(),
latest: peer_latest_hash, latest_hash: peer_latest_hash,
latest_number: None,
difficulty: U256::zero(), difficulty: U256::zero(),
asking: PeerAsking::Nothing, asking: PeerAsking::Nothing,
asking_blocks: Vec::<BlockNumber>::new(), asking_blocks: Vec::<BlockNumber>::new(),
asking_hash: None,
ask_time: 0f64, ask_time: 0f64,
}); });
sync sync
@ -1303,7 +1356,7 @@ mod tests {
let mut client = TestBlockChainClient::new(); let mut client = TestBlockChainClient::new();
client.add_blocks(100, false); client.add_blocks(100, false);
let mut queue = VecDeque::new(); let mut queue = VecDeque::new();
let sync = dummy_sync_with_peer(client.block_hash_delta_minus(10)); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(10));
let io = TestIo::new(&mut client, &mut queue, None); let io = TestIo::new(&mut client, &mut queue, None);
let lagging_peers = sync.get_lagging_peers(&io); let lagging_peers = sync.get_lagging_peers(&io);
@ -1334,9 +1387,11 @@ mod tests {
client.add_blocks(100, false); client.add_blocks(100, false);
let mut queue = VecDeque::new(); let mut queue = VecDeque::new();
let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5));
let best_hash = client.chain_info().best_block_hash.clone();
let best_number = client.chain_info().best_block_number;
let mut io = TestIo::new(&mut client, &mut queue, None); let mut io = TestIo::new(&mut client, &mut queue, None);
let peer_count = sync.propagade_new_hashes(&mut io); let peer_count = sync.propagade_new_hashes(&best_hash, best_number, &mut io);
// 1 message should be send // 1 message should be send
assert_eq!(1, io.queue.len()); assert_eq!(1, io.queue.len());
@ -1352,9 +1407,11 @@ mod tests {
client.add_blocks(100, false); client.add_blocks(100, false);
let mut queue = VecDeque::new(); let mut queue = VecDeque::new();
let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5));
let best_hash = client.chain_info().best_block_hash.clone();
let best_number = client.chain_info().best_block_number;
let mut io = TestIo::new(&mut client, &mut queue, None); let mut io = TestIo::new(&mut client, &mut queue, None);
let peer_count = sync.propagade_blocks(&mut io); let peer_count = sync.propagade_blocks(&best_hash, best_number, &mut io);
// 1 message should be send // 1 message should be send
assert_eq!(1, io.queue.len()); assert_eq!(1, io.queue.len());
@ -1456,9 +1513,11 @@ mod tests {
client.add_blocks(100, false); client.add_blocks(100, false);
let mut queue = VecDeque::new(); let mut queue = VecDeque::new();
let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5));
let best_hash = client.chain_info().best_block_hash.clone();
let best_number = client.chain_info().best_block_number;
let mut io = TestIo::new(&mut client, &mut queue, None); let mut io = TestIo::new(&mut client, &mut queue, None);
sync.propagade_new_hashes(&mut io); sync.propagade_new_hashes(&best_hash, best_number, &mut io);
let data = &io.queue[0].data.clone(); let data = &io.queue[0].data.clone();
let result = sync.on_peer_new_hashes(&mut io, 0, &UntrustedRlp::new(&data)); let result = sync.on_peer_new_hashes(&mut io, 0, &UntrustedRlp::new(&data));
@ -1473,9 +1532,11 @@ mod tests {
client.add_blocks(100, false); client.add_blocks(100, false);
let mut queue = VecDeque::new(); let mut queue = VecDeque::new();
let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5));
let best_hash = client.chain_info().best_block_hash.clone();
let best_number = client.chain_info().best_block_number;
let mut io = TestIo::new(&mut client, &mut queue, None); let mut io = TestIo::new(&mut client, &mut queue, None);
sync.propagade_blocks(&mut io); sync.propagade_blocks(&best_hash, best_number, &mut io);
let data = &io.queue[0].data.clone(); let data = &io.queue[0].data.clone();
let result = sync.on_peer_new_block(&mut io, 0, &UntrustedRlp::new(&data)); let result = sync.on_peer_new_block(&mut io, 0, &UntrustedRlp::new(&data));

40
sync/src/tests/chain.rs
View File

@ -109,7 +109,7 @@ fn status_empty() {
#[test] #[test]
fn status_packet() { fn status_packet() {
let mut net = TestNet::new(2); let mut net = TestNet::new(2);
net.peer_mut(0).chain.add_blocks(1000, false); net.peer_mut(0).chain.add_blocks(100, false);
net.peer_mut(1).chain.add_blocks(1, false); net.peer_mut(1).chain.add_blocks(1, false);
net.start(); net.start();
@ -122,18 +122,28 @@ fn status_packet() {
#[test] #[test]
fn propagade_hashes() { fn propagade_hashes() {
let mut net = TestNet::new(3); let mut net = TestNet::new(6);
net.peer_mut(1).chain.add_blocks(1000, false); net.peer_mut(1).chain.add_blocks(10, false);
net.peer_mut(2).chain.add_blocks(1000, false);
net.sync(); net.sync();
net.peer_mut(0).chain.add_blocks(10, false); net.peer_mut(0).chain.add_blocks(10, false);
net.sync_step_peer(0); net.sync();
net.trigger_block_verified(0); //first event just sets the marker
net.trigger_block_verified(0);
// 2 peers to sync // 5 peers to sync
assert_eq!(2, net.peer(0).queue.len()); assert_eq!(5, net.peer(0).queue.len());
// NEW_BLOCK_HASHES_PACKET let mut hashes = 0;
assert_eq!(0x01, net.peer(0).queue[0].packet_id); let mut blocks = 0;
for i in 0..5 {
if net.peer(0).queue[i].packet_id == 0x1 {
hashes += 1;
}
if net.peer(0).queue[i].packet_id == 0x7 {
blocks += 1;
}
}
assert!(blocks + hashes == 5);
} }
#[test] #[test]
@ -143,9 +153,21 @@ fn propagade_blocks() {
net.sync(); net.sync();
net.peer_mut(0).chain.add_blocks(10, false); net.peer_mut(0).chain.add_blocks(10, false);
net.trigger_block_verified(0); //first event just sets the marker
net.trigger_block_verified(0); net.trigger_block_verified(0);
assert!(!net.peer(0).queue.is_empty()); assert!(!net.peer(0).queue.is_empty());
// NEW_BLOCK_PACKET // NEW_BLOCK_PACKET
assert_eq!(0x07, net.peer(0).queue[0].packet_id); assert_eq!(0x07, net.peer(0).queue[0].packet_id);
} }
#[test]
fn restart_on_malformed_block() {
let mut net = TestNet::new(2);
net.peer_mut(1).chain.add_blocks(10, false);
net.peer_mut(1).chain.corrupt_block(6);
net.sync_steps(10);
assert_eq!(net.peer(0).chain.chain_info().best_block_number, 4);
}

11
sync/src/tests/helpers.rs
View File

@ -71,6 +71,17 @@ impl TestBlockChainClient {
} }
} }
pub fn corrupt_block(&mut self, n: BlockNumber) {
let hash = self.block_hash(BlockId::Number(n)).unwrap();
let mut header: BlockHeader = decode(&self.block_header(BlockId::Number(n)).unwrap());
header.parent_hash = H256::new();
let mut rlp = RlpStream::new_list(3);
rlp.append(&header);
rlp.append_raw(&rlp::NULL_RLP, 1);
rlp.append_raw(&rlp::NULL_RLP, 1);
self.blocks.write().unwrap().insert(hash, rlp.out());
}
pub fn block_hash_delta_minus(&mut self, delta: usize) -> H256 { pub fn block_hash_delta_minus(&mut self, delta: usize) -> H256 {
let blocks_read = self.numbers.read().unwrap(); let blocks_read = self.numbers.read().unwrap();
let index = blocks_read.len() - delta; let index = blocks_read.len() - delta;

1
util/src/network/host.rs
View File

@ -217,6 +217,7 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
s.send_packet(self.protocol, packet_id as u8, &data).unwrap_or_else(|e| { s.send_packet(self.protocol, packet_id as u8, &data).unwrap_or_else(|e| {
warn!(target: "net", "Send error: {:?}", e); warn!(target: "net", "Send error: {:?}", e);
}); //TODO: don't copy vector data }); //TODO: don't copy vector data
try!(self.io.update_registration(peer));
}, },
_ => warn!(target: "net", "Send: Peer is not connected yet") _ => warn!(target: "net", "Send: Peer is not connected yet")
} }