From edd41ac574d07ae94519a432a9b7a2edcf25060c Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Wed, 3 Feb 2016 23:05:04 +0300 Subject: [PATCH 1/4] getting old one --- sync/src/chain.rs | 138 +++++++++++++++++++--------------------------- 1 file changed, 56 insertions(+), 82 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 9752a5013..15068a47e 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -15,14 +15,13 @@ use util::*; use std::mem::{replace}; -use ethcore::views::{HeaderView}; -use ethcore::header::{BlockNumber, Header as BlockHeader}; -use ethcore::client::{BlockChainClient, BlockStatus}; -use range_collection::{RangeCollection, ToUsize, FromUsize}; -use ethcore::error::*; -use ethcore::block::Block; -use io::SyncIo; -use time; +use views::{HeaderView}; +use header::{BlockNumber, Header as BlockHeader}; +use client::{BlockChainClient, BlockStatus}; +use sync::range_collection::{RangeCollection, ToUsize, FromUsize}; +use error::*; +use sync::io::SyncIo; +use std::option::Option; impl ToUsize for BlockNumber { fn to_usize(&self) -> usize { @@ -62,8 +61,6 @@ const RECEIPTS_PACKET: u8 = 0x10; const NETWORK_ID: U256 = ONE_U256; //TODO: get this from parent -const CONNECTION_TIMEOUT_SEC: f64 = 30f64; - struct Header { /// Header data data: Bytes, @@ -81,7 +78,7 @@ struct HeaderId { } #[derive(Copy, Clone, Eq, PartialEq, Debug)] -/// Sync state +/// Sync state pub enum SyncState { /// Initial chain sync has not started yet NotSynced, @@ -103,14 +100,14 @@ pub struct SyncStatus { pub protocol_version: u8, /// BlockChain height for the moment the sync started. pub start_block_number: BlockNumber, - /// Last fully downloaded and imported block number. - pub last_imported_block_number: BlockNumber, - /// Highest block number in the download queue. - pub highest_block_number: BlockNumber, + /// Last fully downloaded and imported block number (if any). + pub last_imported_block_number: Option, + /// Highest block number in the download queue (if any). + pub highest_block_number: Option, /// Total number of blocks for the sync process. - pub blocks_total: usize, + pub blocks_total: BlockNumber, /// Number of blocks downloaded so far. - pub blocks_received: usize, + pub blocks_received: BlockNumber, /// Total number of connected peers pub num_peers: usize, /// Total number of active peers @@ -131,7 +128,7 @@ struct PeerInfo { protocol_version: u32, /// Peer chain genesis hash genesis: H256, - /// Peer network id + /// Peer network id network_id: U256, /// Peer best block hash latest: H256, @@ -141,8 +138,6 @@ struct PeerInfo { asking: PeerAsking, /// A set of block numbers being requested asking_blocks: Vec, - /// Request timestamp - ask_time: f64, } /// Blockchain sync handler. @@ -153,7 +148,7 @@ pub struct ChainSync { /// Last block number for the start of sync starting_block: BlockNumber, /// Highest block number seen - highest_block: BlockNumber, + highest_block: Option, /// Set of block header numbers being downloaded downloading_headers: HashSet, /// Set of block body numbers being downloaded @@ -167,9 +162,9 @@ pub struct ChainSync { /// Used to map body to header header_ids: HashMap, /// Last impoted block number - last_imported_block: BlockNumber, + last_imported_block: Option, /// Last impoted block hash - last_imported_hash: H256, + last_imported_hash: Option, /// Syncing total difficulty syncing_difficulty: U256, /// True if common block for our and remote chain has been found @@ -183,15 +178,15 @@ impl ChainSync { ChainSync { state: SyncState::NotSynced, starting_block: 0, - highest_block: 0, + highest_block: None, downloading_headers: HashSet::new(), downloading_bodies: HashSet::new(), headers: Vec::new(), bodies: Vec::new(), peers: HashMap::new(), header_ids: HashMap::new(), - last_imported_block: 0, - last_imported_hash: H256::new(), + last_imported_block: None, + last_imported_hash: None, syncing_difficulty: U256::from(0u64), have_common_block: false, } @@ -205,8 +200,8 @@ impl ChainSync { start_block_number: self.starting_block, last_imported_block_number: self.last_imported_block, highest_block_number: self.highest_block, - blocks_received: (self.last_imported_block - self.starting_block) as usize, - blocks_total: (self.highest_block - self.starting_block) as usize, + blocks_received: match self.last_imported_block { None => 0, Some(x) => x - self.starting_block }, + blocks_total: match self.highest_block { None => 0, Some(x) => x - self.starting_block }, num_peers: self.peers.len(), num_active_peers: self.peers.values().filter(|p| p.asking != PeerAsking::Nothing).count(), } @@ -235,10 +230,10 @@ impl ChainSync { /// Restart sync pub fn restart(&mut self, io: &mut SyncIo) { self.reset(); - self.last_imported_block = 0; - self.last_imported_hash = H256::new(); + self.last_imported_block = None; + self.last_imported_hash = None; self.starting_block = 0; - self.highest_block = 0; + self.highest_block = None; self.have_common_block = false; io.chain().clear_queue(); self.starting_block = io.chain().chain_info().best_block_number; @@ -255,11 +250,10 @@ impl ChainSync { genesis: try!(r.val_at(4)), asking: PeerAsking::Nothing, asking_blocks: Vec::new(), - 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); - + let chain_info = io.chain().chain_info(); if peer.genesis != chain_info.genesis_hash { io.disable_peer(peer_id); @@ -300,25 +294,27 @@ impl ChainSync { for i in 0..item_count { let info: BlockHeader = try!(r.val_at(i)); let number = BlockNumber::from(info.number); - if number <= self.last_imported_block || self.headers.have_item(&number) { + if number <= self.current_base_block() || self.headers.have_item(&number) { trace!(target: "sync", "Skipping existing block header"); continue; } - if number > self.highest_block { - self.highest_block = number; + + if self.highest_block == None || number > self.highest_block.unwrap() { + self.highest_block = Some(number); } let hash = info.hash(); match io.chain().block_status(&hash) { BlockStatus::InChain => { self.have_common_block = true; - self.last_imported_block = number; - self.last_imported_hash = hash.clone(); + self.last_imported_block = Some(number); + self.last_imported_hash = Some(hash.clone()); trace!(target: "sync", "Found common header {} ({})", number, hash); }, _ => { if self.have_common_block { //validate chain - if self.have_common_block && number == self.last_imported_block + 1 && info.parent_hash != self.last_imported_hash { + let base_hash = self.last_imported_hash.clone().unwrap(); + if self.have_common_block && number == self.current_base_block() + 1 && info.parent_hash != base_hash { // TODO: lower peer rating debug!(target: "sync", "Mismatched block header {} {}", number, hash); continue; @@ -414,8 +410,7 @@ impl ChainSync { trace!(target: "sync", "{} -> NewBlock ({})", peer_id, h); let header_view = HeaderView::new(header_rlp.as_raw()); // TODO: Decompose block and add to self.headers and self.bodies instead - let mut unknown = false; - if header_view.number() == From::from(self.last_imported_block + 1) { + if header_view.number() == From::from(self.current_base_block() + 1) { match io.chain().import_block(block_rlp.as_raw().to_vec()) { Err(ImportError::AlreadyInChain) => { trace!(target: "sync", "New block already in chain {:?}", h); @@ -423,10 +418,6 @@ impl ChainSync { Err(ImportError::AlreadyQueued) => { trace!(target: "sync", "New block already queued {:?}", h); }, - Err(ImportError::UnknownParent) => { - unknown = true; - trace!(target: "sync", "New block with unknown parent {:?}", h); - }, Ok(_) => { trace!(target: "sync", "New block queued {:?}", h); }, @@ -435,11 +426,8 @@ impl ChainSync { io.disable_peer(peer_id); } }; - } - else { - unknown = true; } - if unknown { + else { trace!(target: "sync", "New block unknown {:?}", h); //TODO: handle too many unknown blocks let difficulty: U256 = try!(r.val_at(1)); @@ -456,7 +444,7 @@ impl ChainSync { Ok(()) } - /// 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> { if self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer").asking != PeerAsking::Nothing { trace!(target: "sync", "Ignoring new hashes since we're already downloading."); @@ -565,6 +553,10 @@ impl ChainSync { } } + fn current_base_block(&self) -> BlockNumber { + match self.last_imported_block { None => 0, Some(x) => x } + } + /// Find some headers or blocks to download for a peer. fn request_blocks(&mut self, io: &mut SyncIo, peer_id: PeerId) { self.clear_peer_download(peer_id); @@ -578,7 +570,7 @@ impl ChainSync { let mut needed_bodies: Vec = Vec::new(); let mut needed_numbers: Vec = Vec::new(); - if self.have_common_block && !self.headers.is_empty() && self.headers.range_iter().next().unwrap().0 == self.last_imported_block + 1 { + if self.have_common_block && !self.headers.is_empty() && self.headers.range_iter().next().unwrap().0 == self.current_base_block() + 1 { for (start, ref items) in self.headers.range_iter() { if needed_bodies.len() > MAX_BODIES_TO_REQUEST { break; @@ -611,12 +603,12 @@ impl ChainSync { } if start == 0 { self.have_common_block = true; //reached genesis - self.last_imported_hash = chain_info.genesis_hash; + self.last_imported_hash = Some(chain_info.genesis_hash); } } if self.have_common_block { let mut headers: Vec = Vec::new(); - let mut prev = self.last_imported_block + 1; + let mut prev = self.current_base_block() + 1; for (next, ref items) in self.headers.range_iter() { if !headers.is_empty() { break; @@ -671,7 +663,7 @@ impl ChainSync { { let headers = self.headers.range_iter().next().unwrap(); let bodies = self.bodies.range_iter().next().unwrap(); - if headers.0 != bodies.0 || headers.0 != self.last_imported_block + 1 { + if headers.0 != bodies.0 || headers.0 != self.current_base_block() + 1 { return; } @@ -684,27 +676,21 @@ impl ChainSync { block_rlp.append_raw(body.at(0).as_raw(), 1); block_rlp.append_raw(body.at(1).as_raw(), 1); let h = &headers.1[i].hash; - // Perform basic block verification - if !Block::is_good(block_rlp.as_raw()) { - debug!(target: "sync", "Bad block rlp {:?} : {:?}", h, block_rlp.as_raw()); - restart = true; - break; - } match io.chain().import_block(block_rlp.out()) { Err(ImportError::AlreadyInChain) => { trace!(target: "sync", "Block already in chain {:?}", h); - self.last_imported_block = headers.0 + i as BlockNumber; - self.last_imported_hash = h.clone(); + self.last_imported_block = Some(headers.0 + i as BlockNumber); + self.last_imported_hash = Some(h.clone()); }, Err(ImportError::AlreadyQueued) => { trace!(target: "sync", "Block already queued {:?}", h); - self.last_imported_block = headers.0 + i as BlockNumber; - self.last_imported_hash = h.clone(); + self.last_imported_block = Some(headers.0 + i as BlockNumber); + self.last_imported_hash = Some(h.clone()); }, Ok(_) => { trace!(target: "sync", "Block queued {:?}", h); - self.last_imported_block = headers.0 + i as BlockNumber; - self.last_imported_hash = h.clone(); + self.last_imported_block = Some(headers.0 + i as BlockNumber); + self.last_imported_hash = Some(h.clone()); imported += 1; }, Err(e) => { @@ -721,8 +707,8 @@ impl ChainSync { return; } - self.headers.remove_head(&(self.last_imported_block + 1)); - self.bodies.remove_head(&(self.last_imported_block + 1)); + self.headers.remove_head(&(self.last_imported_block.unwrap() + 1)); + self.bodies.remove_head(&(self.last_imported_block.unwrap() + 1)); if self.headers.is_empty() { assert!(self.bodies.is_empty()); @@ -730,7 +716,7 @@ impl ChainSync { } } - /// Remove downloaded bocks/headers starting from specified number. + /// Remove downloaded bocks/headers starting from specified number. /// Used to recover from an error and re-download parts of the chain detected as bad. fn remove_downloaded_blocks(&mut self, start: BlockNumber) { for n in self.headers.get_tail(&start) { @@ -809,7 +795,6 @@ impl ChainSync { Ok(_) => { let mut peer = self.peers.get_mut(&peer_id).unwrap(); peer.asking = asking; - peer.ask_time = time::precise_time_s(); } } } @@ -974,7 +959,7 @@ impl ChainSync { NEW_BLOCK_HASHES_PACKET => self.on_peer_new_hashes(io, peer, &rlp), GET_NODE_DATA_PACKET => self.return_node_data(io, &rlp), GET_RECEIPTS_PACKET => self.return_receipts(io, &rlp), - _ => { + _ => { debug!(target: "sync", "Unknown packet {}", packet_id); Ok(()) } @@ -984,18 +969,7 @@ impl ChainSync { }) } - /// Handle peer timeouts - pub fn maintain_peers(&self, io: &mut SyncIo) { - let tick = time::precise_time_s(); - for (peer_id, peer) in &self.peers { - if peer.asking != PeerAsking::Nothing && (tick - peer.ask_time) > CONNECTION_TIMEOUT_SEC { - io.disconnect_peer(*peer_id); - } - } - } - /// Maintain other peers. Send out any new blocks and transactions pub fn _maintain_sync(&mut self, _io: &mut SyncIo) { } } - From d398b848025fbedf05238f831ee65a071b99c7c3 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Wed, 3 Feb 2016 23:42:30 +0300 Subject: [PATCH 2/4] remerge of 264 --- sync/src/chain.rs | 47 ++++++++++++++++++++++++----- sync/src/tests.rs | 77 ++++++++++++++++++++++++++++++++++++++++++----- 2 files changed, 109 insertions(+), 15 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 15068a47e..e351ad837 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -15,12 +15,14 @@ use util::*; use std::mem::{replace}; -use views::{HeaderView}; -use header::{BlockNumber, Header as BlockHeader}; -use client::{BlockChainClient, BlockStatus}; -use sync::range_collection::{RangeCollection, ToUsize, FromUsize}; -use error::*; -use sync::io::SyncIo; +use ethcore::views::{HeaderView}; +use ethcore::header::{BlockNumber, Header as BlockHeader}; +use ethcore::client::{BlockChainClient, BlockStatus}; +use range_collection::{RangeCollection, ToUsize, FromUsize}; +use ethcore::error::*; +use ethcore::block::Block; +use io::SyncIo; +use time; use std::option::Option; impl ToUsize for BlockNumber { @@ -61,6 +63,8 @@ const RECEIPTS_PACKET: u8 = 0x10; const NETWORK_ID: U256 = ONE_U256; //TODO: get this from parent +const CONNECTION_TIMEOUT_SEC: f64 = 30f64; + struct Header { /// Header data data: Bytes, @@ -138,6 +142,8 @@ struct PeerInfo { asking: PeerAsking, /// A set of block numbers being requested asking_blocks: Vec, + /// Request timestamp + ask_time: f64, } /// Blockchain sync handler. @@ -250,6 +256,7 @@ impl ChainSync { genesis: try!(r.val_at(4)), asking: PeerAsking::Nothing, asking_blocks: Vec::new(), + 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); @@ -409,6 +416,7 @@ impl ChainSync { trace!(target: "sync", "{} -> NewBlock ({})", peer_id, h); let header_view = HeaderView::new(header_rlp.as_raw()); + let mut unknown = false; // TODO: Decompose block and add to self.headers and self.bodies instead if header_view.number() == From::from(self.current_base_block() + 1) { match io.chain().import_block(block_rlp.as_raw().to_vec()) { @@ -421,13 +429,20 @@ impl ChainSync { Ok(_) => { trace!(target: "sync", "New block queued {:?}", h); }, + Err(ImportError::UnknownParent) => { + unknown = true; + trace!(target: "sync", "New block with unknown parent {:?}", h); + }, Err(e) => { debug!(target: "sync", "Bad new block {:?} : {:?}", h, e); io.disable_peer(peer_id); } }; } - else { + else { + unknown = true; + } + if unknown { trace!(target: "sync", "New block unknown {:?}", h); //TODO: handle too many unknown blocks let difficulty: U256 = try!(r.val_at(1)); @@ -676,6 +691,14 @@ impl ChainSync { block_rlp.append_raw(body.at(0).as_raw(), 1); block_rlp.append_raw(body.at(1).as_raw(), 1); let h = &headers.1[i].hash; + + // Perform basic block verification + if !Block::is_good(block_rlp.as_raw()) { + debug!(target: "sync", "Bad block rlp {:?} : {:?}", h, block_rlp.as_raw()); + restart = true; + break; + } + match io.chain().import_block(block_rlp.out()) { Err(ImportError::AlreadyInChain) => { trace!(target: "sync", "Block already in chain {:?}", h); @@ -795,6 +818,7 @@ impl ChainSync { Ok(_) => { let mut peer = self.peers.get_mut(&peer_id).unwrap(); peer.asking = asking; + peer.ask_time = time::precise_time_s(); } } } @@ -972,4 +996,13 @@ impl ChainSync { /// Maintain other peers. Send out any new blocks and transactions pub fn _maintain_sync(&mut self, _io: &mut SyncIo) { } + + pub fn maintain_peers(&self, io: &mut SyncIo) { + let tick = time::precise_time_s(); + for (peer_id, peer) in &self.peers { + if peer.asking != PeerAsking::Nothing && (tick - peer.ask_time) > CONNECTION_TIMEOUT_SEC { + io.disconnect_peer(*peer_id); + } + } + } } diff --git a/sync/src/tests.rs b/sync/src/tests.rs index 5b796e6f1..ccf9cacf1 100644 --- a/sync/src/tests.rs +++ b/sync/src/tests.rs @@ -4,7 +4,7 @@ use ethcore::block_queue::BlockQueueInfo; use ethcore::header::{Header as BlockHeader, BlockNumber}; use ethcore::error::*; use io::SyncIo; -use chain::ChainSync; +use chain::{ChainSync, SyncState}; struct TestBlockChainClient { blocks: RwLock>, @@ -248,13 +248,15 @@ struct TestPeer { } struct TestNet { - peers: Vec + peers: Vec, + started: bool, } impl TestNet { pub fn new(n: usize) -> TestNet { let mut net = TestNet { peers: Vec::new(), + started: false, }; for _ in 0..n { net.peers.push(TestPeer { @@ -298,10 +300,28 @@ impl TestNet { } } - pub fn sync(&mut self) { + pub fn restart_peer(&mut self, i: usize) { + let peer = self.peer_mut(i); + peer.sync.restart(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None)); + } + + pub fn sync(&mut self) -> u32 { self.start(); + let mut total_steps = 0; while !self.done() { - self.sync_step() + self.sync_step(); + total_steps = total_steps + 1; + } + total_steps + } + + pub fn sync_steps(&mut self, count: usize) { + if !self.started { + self.start(); + self.started = true; + } + for _ in 0..count { + self.sync_step(); } } @@ -310,9 +330,8 @@ impl TestNet { } } - #[test] -fn full_sync_two_peers() { +fn chain_two_peers() { ::env_logger::init().ok(); let mut net = TestNet::new(3); net.peer_mut(1).chain.add_blocks(1000, false); @@ -323,7 +342,27 @@ fn full_sync_two_peers() { } #[test] -fn full_sync_empty_blocks() { +fn chain_status_after_sync() { + ::env_logger::init().ok(); + let mut net = TestNet::new(3); + net.peer_mut(1).chain.add_blocks(1000, false); + net.peer_mut(2).chain.add_blocks(1000, false); + net.sync(); + let status = net.peer(0).sync.status(); + assert_eq!(status.state, SyncState::Idle); +} + +#[test] +fn chain_takes_few_steps() { + let mut net = TestNet::new(3); + net.peer_mut(1).chain.add_blocks(100, false); + net.peer_mut(2).chain.add_blocks(100, false); + let total_steps = net.sync(); + assert!(total_steps < 7); +} + +#[test] +fn chain_empty_blocks() { ::env_logger::init().ok(); let mut net = TestNet::new(3); for n in 0..200 { @@ -336,7 +375,7 @@ fn full_sync_empty_blocks() { } #[test] -fn forked_sync() { +fn chain_forged() { ::env_logger::init().ok(); let mut net = TestNet::new(3); net.peer_mut(0).chain.add_blocks(300, false); @@ -354,3 +393,25 @@ fn forked_sync() { assert_eq!(net.peer(1).chain.numbers.read().unwrap().deref(), &peer1_chain); assert_eq!(net.peer(2).chain.numbers.read().unwrap().deref(), &peer1_chain); } + +#[test] +fn chain_restart() { + let mut net = TestNet::new(3); + net.peer_mut(1).chain.add_blocks(1000, false); + net.peer_mut(2).chain.add_blocks(1000, false); + + net.sync_steps(8); + + // make sure that sync has actually happened + assert!(net.peer(0).chain.chain_info().best_block_number > 100); + net.restart_peer(0); + + let status = net.peer(0).sync.status(); + assert_eq!(status.state, SyncState::NotSynced); +} + +#[test] +fn chain_status_empty() { + let net = TestNet::new(2); + assert_eq!(net.peer(0).sync.status().state, SyncState::NotSynced); +} \ No newline at end of file From 52f0ee175406147c982f70598ad48968eb196973 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Thu, 4 Feb 2016 04:38:40 +0300 Subject: [PATCH 3/4] resurrect proper name --- sync/src/tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sync/src/tests.rs b/sync/src/tests.rs index ccf9cacf1..a0850c062 100644 --- a/sync/src/tests.rs +++ b/sync/src/tests.rs @@ -375,7 +375,7 @@ fn chain_empty_blocks() { } #[test] -fn chain_forged() { +fn chain_forked() { ::env_logger::init().ok(); let mut net = TestNet::new(3); net.peer_mut(0).chain.add_blocks(300, false); From 583a1261c56b70d00405e3262408015f7b82512a Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Thu, 4 Feb 2016 05:10:05 +0300 Subject: [PATCH 4/4] ethsync tests fix --- sync/src/tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sync/src/tests.rs b/sync/src/tests.rs index a0850c062..88f19a8b6 100644 --- a/sync/src/tests.rs +++ b/sync/src/tests.rs @@ -296,7 +296,7 @@ impl TestNet { trace!("----------------"); } let mut p = self.peers.get_mut(peer).unwrap(); - p.sync._maintain_sync(&mut TestIo::new(&mut p.chain, &mut p.queue, None)); + p.sync.maintain_sync(&mut TestIo::new(&mut p.chain, &mut p.queue, None)); } }