From cc3f712fec92f96364066ea5a99ce45a717c26d4 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Fri, 5 Feb 2016 09:34:08 -0800 Subject: [PATCH 01/30] propagade initial --- sync/Cargo.toml | 2 +- sync/src/chain.rs | 75 +++++++++++++++++++++++++++++++++++++-- sync/src/lib.rs | 1 + sync/src/tests/chain.rs | 28 +++++++++++++++ sync/src/tests/helpers.rs | 5 +++ 5 files changed, 108 insertions(+), 3 deletions(-) diff --git a/sync/Cargo.toml b/sync/Cargo.toml index 5f098bc26..75853e0ab 100644 --- a/sync/Cargo.toml +++ b/sync/Cargo.toml @@ -14,4 +14,4 @@ clippy = "0.0.37" log = "0.3" env_logger = "0.3" time = "0.1.34" - +rand = "0.3.13" diff --git a/sync/src/chain.rs b/sync/src/chain.rs index e143f20b1..590d351df 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -46,6 +46,8 @@ const MAX_NODE_DATA_TO_SEND: usize = 1024; const MAX_RECEIPTS_TO_SEND: usize = 1024; const MAX_HEADERS_TO_REQUEST: usize = 512; const MAX_BODIES_TO_REQUEST: usize = 256; +const MIN_PEERS_PROPAGATION: usize = 4; +const MAX_PEERS_PROPAGATION: usize = 128; const STATUS_PACKET: u8 = 0x00; const NEW_BLOCK_HASHES_PACKET: u8 = 0x01; @@ -1026,13 +1028,82 @@ impl ChainSync { } } } - /// Maintain other peers. Send out any new blocks and transactions - pub fn maintain_sync(&mut self, io: &mut SyncIo) { + + fn check_resume(&mut self, io: &mut SyncIo) { if !io.chain().queue_info().full && self.state == SyncState::Waiting { self.state = SyncState::Idle; self.continue_sync(io); } } + + fn create_new_hashes_rlp(chain: &BlockChainClient, from: &H256, to: &H256) -> Option { + match chain.tree_route(from, to) { + Some(route) => { + match route.blocks.len() { + 0 => None, + _ => { + let mut rlp_stream = RlpStream::new_list(route.blocks.len()); + for hash in route.blocks { + rlp_stream.append(&hash); + } + Some(rlp_stream.out()) + } + } + }, + None => None + } + } + + fn query_peer_latest_blocks(&self) -> Vec<(usize, H256)> { + self.peers.iter().map(|peer| (peer.0.clone(), peer.1.latest.clone())).collect() + } + + fn propagade_blocks(&mut self, io: &mut SyncIo) -> usize { + let updated_peers = { + let chain = io.chain(); + let chain_info = chain.chain_info(); + let latest_hash = chain_info.best_block_hash; + + let lagging_peers = self.query_peer_latest_blocks().iter().filter(|peer| + match io.chain().block_status(&peer.1) + { + BlockStatus::InChain => peer.1 != latest_hash, + _ => false + }).cloned().collect::>(); + + let lucky_peers = match lagging_peers.len() { + 0 ... MIN_PEERS_PROPAGATION => lagging_peers, + _ => lagging_peers.iter().filter(|_| ::rand::random::() < 64u8).cloned().collect::>() + }; + + match lucky_peers.len() { + 0 ... MAX_PEERS_PROPAGATION => lucky_peers, + _ => lucky_peers.iter().take(MAX_PEERS_PROPAGATION).cloned().collect::>() + } + }; + + let mut sent = 0; + for (peer_id, peer_hash) in updated_peers { + sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &peer_hash, &io.chain().chain_info().best_block_hash) { + Some(rlp) => { + self.send_request(io, peer_id, PeerAsking::Nothing, NEW_BLOCK_HASHES_PACKET, rlp); + 1 + }, + None => 0 + } + } + sent + } + + /// Maintain other peers. Send out any new blocks and transactions + pub fn maintain_sync(&mut self, io: &mut SyncIo) { + self.check_resume(io); + + if self.state == SyncState::Idle { + let blocks_propagaded = self.propagade_blocks(io); + debug!(target: "sync", "Sent new blocks to peers: {:?}", blocks_propagaded); + } + } } #[cfg(test)] diff --git a/sync/src/lib.rs b/sync/src/lib.rs index 1523a8a9f..8847d9611 100644 --- a/sync/src/lib.rs +++ b/sync/src/lib.rs @@ -34,6 +34,7 @@ extern crate ethcore_util as util; extern crate ethcore; extern crate env_logger; extern crate time; +extern crate rand; use std::ops::*; use std::sync::*; diff --git a/sync/src/tests/chain.rs b/sync/src/tests/chain.rs index fcd9b6a7b..e328ba33d 100644 --- a/sync/src/tests/chain.rs +++ b/sync/src/tests/chain.rs @@ -88,4 +88,32 @@ fn restart() { fn status_empty() { let net = TestNet::new(2); assert_eq!(net.peer(0).sync.status().state, SyncState::NotSynced); +} + +#[test] +fn status_packet() { + let mut net = TestNet::new(2); + net.peer_mut(0).chain.add_blocks(1000, false); + net.peer_mut(1).chain.add_blocks(1, false); + + net.start(); + + net.sync_step_peer(0); + + assert_eq!(1, net.peer(0).queue.len()); + assert_eq!(0x00, net.peer(0).queue[0].packet_id); +} + +#[test] +fn propagade() { + let mut net = TestNet::new(2); + net.peer_mut(0).chain.add_blocks(100, false); + net.peer_mut(1).chain.add_blocks(100, false); + net.sync(); + + net.peer_mut(0).chain.add_blocks(10, false); + net.sync_step_peer(0); + + assert_eq!(1, net.peer(0).queue.len()); + assert_eq!(0x01, net.peer(0).queue[0].packet_id); } \ No newline at end of file diff --git a/sync/src/tests/helpers.rs b/sync/src/tests/helpers.rs index c4a4d80cb..54fcc37b0 100644 --- a/sync/src/tests/helpers.rs +++ b/sync/src/tests/helpers.rs @@ -318,6 +318,11 @@ impl TestNet { } } + pub fn sync_step_peer(&mut self, peer_num: usize) { + let mut peer = self.peer_mut(peer_num); + peer.sync.maintain_sync(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None)); + } + 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)); From 4af85b488b659d63d54f3d34cd5eb1af675feaf1 Mon Sep 17 00:00:00 2001 From: arkpar Date: Fri, 5 Feb 2016 22:54:33 +0100 Subject: [PATCH 02/30] Fixed an issue with forked counters --- util/src/journaldb.rs | 90 ++++++++++++++++++++++++++++++------------- 1 file changed, 64 insertions(+), 26 deletions(-) diff --git a/util/src/journaldb.rs b/util/src/journaldb.rs index 810b06727..2173fdeb6 100644 --- a/util/src/journaldb.rs +++ b/util/src/journaldb.rs @@ -20,7 +20,7 @@ use common::*; use rlp::*; use hashdb::*; use memorydb::*; -use rocksdb::{DB, Writable, WriteBatch, IteratorMode}; +use rocksdb::{DB, Writable, WriteBatch, IteratorMode, DBVector}; #[cfg(test)] use std::env; @@ -105,6 +105,11 @@ impl JournalDB { // for each end_era that we journaled that we are no passing by, // we remove all of its removes assuming it is canonical and all // of its inserts otherwise. + // + // we also keep track of the counters for each key inserted in the journal to handle the following cases: + // key K is removed in block A(N) and re-inserted in block B(N + C) (where C < H). K must not be deleted from the DB. + // key K is added in block A(N) and reverted in block B(N + C) (where C < H). K must be deleted + // key K is added in blocks A(N) and A'(N) and is reverted in block B(N + C ) (where C < H). K must not be deleted // record new commit's details. let batch = WriteBatch::new(); @@ -125,6 +130,7 @@ impl JournalDB { let mut r = RlpStream::new_list(3); let inserts: Vec = self.overlay.keys().iter().filter(|&(_, &c)| c > 0).map(|(key, _)| key.clone()).collect(); + // Increase counter for each insrted key no matter if the block is canonical or not. for i in &inserts { *counters.entry(i.clone()).or_insert(0) += 1; } @@ -139,6 +145,7 @@ impl JournalDB { if let Some((end_era, canon_id)) = end { let mut index = 0usize; let mut last; + let mut canon_data: Option = None; while let Some(rlp_data) = try!(self.backing.get({ let mut r = RlpStream::new_list(2); r.append(&end_era); @@ -146,35 +153,26 @@ impl JournalDB { last = r.drain(); &last })) { - let to_add; - let rlp = Rlp::new(&rlp_data); - { - to_add = rlp.val_at(1); - for i in &to_add { - let delete_counter = { - if let Some(mut cnt) = counters.get_mut(i) { - *cnt -= 1; - *cnt == 0 - } - else { false } - - }; - if delete_counter { - counters.remove(i); - } - } + let canon = { + let rlp = Rlp::new(&rlp_data); + if canon_id != rlp.val_at(0) { + let to_add: Vec = rlp.val_at(1); + JournalDB::apply_removes(&to_add, &to_add, &mut counters, &batch); + false + } else { true } + }; + if canon { + canon_data = Some(rlp_data) } - let to_remove: Vec = if canon_id == rlp.val_at(0) {rlp.val_at(2)} else {to_add}; - for i in &to_remove { - if !counters.contains_key(i) { - batch.delete(&i).expect("Low-level database error. Some issue with your hard disk?"); - } - } - try!(batch.delete(&last)); - trace!("JournalDB: delete journal for time #{}.{}, (canon was {}): {} entries", end_era, index, canon_id, to_remove.len()); index += 1; } + // Canon must be commited last to handle a case when counter reaches 0 in a sibling block + if let Some(ref c) = canon_data { + let rlp = Rlp::new(&c); + let deleted = JournalDB::apply_removes(&rlp.val_at::>(1), &rlp.val_at::>(2), &mut counters, &batch); + trace!("JournalDB: delete journal for time #{}.{}, (canon was {}): {} entries", end_era, index, canon_id, deleted); + } try!(batch.put(&LAST_ERA_KEY, &encode(&end_era))); } @@ -200,6 +198,29 @@ impl JournalDB { Ok(ret) } + fn apply_removes(added: &[H256], removed: &[H256], counters: &mut HashMap, batch: &WriteBatch) -> usize { + let mut deleted = 0usize; + // Decrease the counters first + for i in added.iter() { + let delete_counter = { + if let Some(mut cnt) = counters.get_mut(i) { + *cnt -= 1; + *cnt == 0 + } + else { false } + }; + if delete_counter { + counters.remove(i); + } + } + // Remove only if counter reached zero + for i in removed.iter().filter(|i| !counters.contains_key(i)) { + batch.delete(&i).expect("Low-level database error. Some issue with your hard disk?"); + deleted += 1; + } + deleted + } + fn payload(&self, key: &H256) -> Option { self.backing.get(&key.bytes()).expect("Low-level database error. Some issue with your hard disk?").map(|v| v.to_vec()) } @@ -387,4 +408,21 @@ mod tests { jdb.commit(3, &b"2".sha3(), Some((0, b"2".sha3()))).unwrap(); assert!(jdb.exists(&foo)); } + + #[test] + fn fork_same_key() { + // history is 1 + let mut jdb = JournalDB::new_temp(); + jdb.commit(0, &b"0".sha3(), None).unwrap(); + + let foo = jdb.insert(b"foo"); + jdb.commit(1, &b"1a".sha3(), Some((0, b"0".sha3()))).unwrap(); + + jdb.insert(b"foo"); + jdb.commit(1, &b"1b".sha3(), Some((0, b"0".sha3()))).unwrap(); + assert!(jdb.exists(&foo)); + + jdb.commit(2, &b"2a".sha3(), Some((1, b"1a".sha3()))).unwrap(); + assert!(jdb.exists(&foo)); + } } From b01f954b05756e34306f76e012148e264d86c65c Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sat, 6 Feb 2016 01:45:25 +0300 Subject: [PATCH 03/30] final tests --- sync/src/chain.rs | 90 ++++++++++++++++++++++++++++++++++----- sync/src/tests/chain.rs | 12 ++++-- sync/src/tests/helpers.rs | 34 +++++++++++++-- 3 files changed, 119 insertions(+), 17 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index ec0659a6e..e853cf4e3 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1074,18 +1074,22 @@ impl ChainSync { self.peers.iter().map(|peer| (peer.0.clone(), peer.1.latest.clone())).collect() } + fn get_lagging_peers(&self, io: &SyncIo) -> Vec<(usize, H256)> { + let chain = io.chain(); + let chain_info = chain.chain_info(); + let latest_hash = chain_info.best_block_hash; + self.query_peer_latest_blocks().iter().filter(|peer| + match io.chain().block_status(&peer.1) + { + BlockStatus::InChain => peer.1 != latest_hash, + _ => false + }).cloned().collect::>() + } + fn propagade_blocks(&mut self, io: &mut SyncIo) -> usize { let updated_peers = { - let chain = io.chain(); - let chain_info = chain.chain_info(); - let latest_hash = chain_info.best_block_hash; - let lagging_peers = self.query_peer_latest_blocks().iter().filter(|peer| - match io.chain().block_status(&peer.1) - { - BlockStatus::InChain => peer.1 != latest_hash, - _ => false - }).cloned().collect::>(); + let lagging_peers = self.get_lagging_peers(io); let lucky_peers = match lagging_peers.len() { 0 ... MIN_PEERS_PROPAGATION => lagging_peers, @@ -1117,7 +1121,7 @@ impl ChainSync { if self.state == SyncState::Idle { let blocks_propagaded = self.propagade_blocks(io); - debug!(target: "sync", "Sent new blocks to peers: {:?}", blocks_propagaded); + trace!(target: "sync", "Sent new blocks to peers: {:?}", blocks_propagaded); } } } @@ -1127,6 +1131,8 @@ mod tests { use tests::helpers::*; use super::*; use util::*; + use super::{PeerInfo, PeerAsking}; + use ethcore::header::{BlockNumber}; #[test] fn return_receipts_empty() { @@ -1195,4 +1201,68 @@ mod tests { sync.on_packet(&mut io, 1usize, super::GET_NODE_DATA_PACKET, &node_request); assert_eq!(1, io.queue.len()); } + + fn dummy_sync_with_peer(peer_latest_hash: H256) -> ChainSync { + let mut sync = ChainSync::new(); + sync.peers.insert(0, + PeerInfo { + protocol_version: 0, + genesis: H256::zero(), + network_id: U256::zero(), + latest: peer_latest_hash, + difficulty: U256::zero(), + asking: PeerAsking::Nothing, + asking_blocks: Vec::::new(), + ask_time: 0f64, + }); + sync + } + + #[test] + fn finds_lagging_peers() { + let mut client = TestBlockChainClient::new(); + client.add_blocks(100, false); + let mut queue = VecDeque::new(); + let sync = dummy_sync_with_peer(client.block_hash_delta_minus(10)); + let io = TestIo::new(&mut client, &mut queue, None); + + let lagging_peers = sync.get_lagging_peers(&io); + + assert_eq!(1, lagging_peers.len()) + } + + #[test] + fn calculates_tree_for_lagging_peer() { + let mut client = TestBlockChainClient::new(); + client.add_blocks(15, false); + + let start = client.block_hash_delta_minus(4); + let end = client.block_hash_delta_minus(2); + + // wrong way end -> start, should be None + let rlp = ChainSync::create_new_hashes_rlp(&client, &end, &start); + assert!(rlp.is_none()); + + let rlp = ChainSync::create_new_hashes_rlp(&client, &start, &end).unwrap(); + // size of three rlp encoded hash + assert_eq!(101, rlp.len()); + } + + #[test] + fn sends_packet_to_lagging_peer() { + let mut client = TestBlockChainClient::new(); + client.add_blocks(20, false); + let mut queue = VecDeque::new(); + let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); + let mut io = TestIo::new(&mut client, &mut queue, None); + + let block_count = sync.propagade_blocks(&mut io); + + // 1 message should be send + assert_eq!(1, io.queue.len()); + // 1 peer should be updated + assert_eq!(1, block_count); + // NEW_BLOCK_HASHES_PACKET + assert_eq!(0x01, io.queue[0].packet_id); + } } \ No newline at end of file diff --git a/sync/src/tests/chain.rs b/sync/src/tests/chain.rs index 43db4c428..a9aeb2e34 100644 --- a/sync/src/tests/chain.rs +++ b/sync/src/tests/chain.rs @@ -122,14 +122,18 @@ fn status_packet() { #[test] fn propagade() { - let mut net = TestNet::new(2); - net.peer_mut(0).chain.add_blocks(100, false); - net.peer_mut(1).chain.add_blocks(100, false); + 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); net.peer_mut(0).chain.add_blocks(10, false); net.sync_step_peer(0); - assert_eq!(1, net.peer(0).queue.len()); + // 2 peers to sync + assert_eq!(2, net.peer(0).queue.len()); + // NEW_BLOCK_HASHES_PACKET assert_eq!(0x01, net.peer(0).queue[0].packet_id); } \ No newline at end of file diff --git a/sync/src/tests/helpers.rs b/sync/src/tests/helpers.rs index 1e9e70c2f..2be501ebb 100644 --- a/sync/src/tests/helpers.rs +++ b/sync/src/tests/helpers.rs @@ -69,6 +69,12 @@ impl TestBlockChainClient { self.import_block(rlp.as_raw().to_vec()).unwrap(); } } + + pub fn block_hash_delta_minus(&mut self, delta: usize) -> H256 { + let blocks_read = self.numbers.read().unwrap(); + let index = blocks_read.len() - delta; + blocks_read[&index].clone() + } } impl BlockChainClient for TestBlockChainClient { @@ -125,11 +131,33 @@ impl BlockChainClient for TestBlockChainClient { } } - fn tree_route(&self, _from: &H256, _to: &H256) -> Option { + // works only if blocks are one after another 1 -> 2 -> 3 + fn tree_route(&self, from: &H256, to: &H256) -> Option { Some(TreeRoute { - blocks: Vec::new(), ancestor: H256::new(), - index: 0 + index: 0, + blocks: { + let numbers_read = self.numbers.read().unwrap(); + let mut adding = false; + + let mut blocks = Vec::new(); + for (_, hash) in numbers_read.iter().sort_by(|tuple1, tuple2| tuple1.0.cmp(tuple2.0)) { + if hash == to { + if adding { + blocks.push(hash.clone()); + } + adding = false; + break; + } + if hash == from { + adding = true; + } + if adding { + blocks.push(hash.clone()); + } + } + if adding { Vec::new() } else { blocks } + } }) } From b606df451e464a00f088b5d88f33d0b6c2253424 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sat, 6 Feb 2016 20:56:21 +0300 Subject: [PATCH 04/30] many fixes --- sync/src/chain.rs | 92 ++++++++++++++++++++++++----------------- sync/src/tests/chain.rs | 8 ++++ 2 files changed, 63 insertions(+), 37 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index e853cf4e3..b9d0ccddf 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -64,6 +64,7 @@ const MAX_HEADERS_TO_REQUEST: usize = 512; const MAX_BODIES_TO_REQUEST: usize = 256; const MIN_PEERS_PROPAGATION: usize = 4; const MAX_PEERS_PROPAGATION: usize = 128; +const MAX_PEER_LAG_PROPAGATION: BlockNumber = 20; const STATUS_PACKET: u8 = 0x00; const NEW_BLOCK_HASHES_PACKET: u8 = 0x01; @@ -136,7 +137,7 @@ pub struct SyncStatus { pub num_active_peers: usize, } -#[derive(PartialEq, Eq, Debug)] +#[derive(PartialEq, Eq, Debug, Clone)] /// Peer data type requested enum PeerAsking { Nothing, @@ -144,6 +145,7 @@ enum PeerAsking { BlockBodies, } +#[derive(Clone)] /// Syncing peer information struct PeerInfo { /// eth protocol version @@ -162,6 +164,8 @@ struct PeerInfo { asking_blocks: Vec, /// Request timestamp ask_time: f64, + /// Latest block number + latest_number: BlockNumber } /// Blockchain sync handler. @@ -267,7 +271,7 @@ impl ChainSync { /// Called by peer to report status fn on_peer_status(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> { - let peer = PeerInfo { + let mut peer = PeerInfo { protocol_version: try!(r.val_at(0)), network_id: try!(r.val_at(1)), difficulty: try!(r.val_at(2)), @@ -276,8 +280,13 @@ impl ChainSync { asking: PeerAsking::Nothing, asking_blocks: Vec::new(), ask_time: 0f64, + latest_number: 0, }; + if io.chain().block_status(&peer.latest) == BlockStatus::InChain { + peer.latest_number = HeaderView::new(&io.chain().block_header(&peer.latest).unwrap()).number(); + } + 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(); @@ -441,6 +450,8 @@ impl ChainSync { match io.chain().import_block(block_rlp.as_raw().to_vec()) { Err(ImportError::AlreadyInChain) => { trace!(target: "sync", "New block already in chain {:?}", h); + let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer"); + peer.latest_number = max(peer.latest_number, header_view.number()); }, Err(ImportError::AlreadyQueued) => { trace!(target: "sync", "New block already queued {:?}", h); @@ -471,6 +482,7 @@ impl ChainSync { { let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer"); peer.latest = header_view.sha3(); + peer.latest_number = header_view.number(); } self.sync_peer(io, peer_id, true); } @@ -638,6 +650,7 @@ impl ChainSync { if start == 0 { self.have_common_block = true; //reached genesis self.last_imported_hash = Some(chain_info.genesis_hash); + self.last_imported_block = Some(0); } } if self.have_common_block { @@ -1032,10 +1045,6 @@ 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 { @@ -1070,41 +1079,39 @@ impl ChainSync { } } - fn query_peer_latest_blocks(&self) -> Vec<(usize, H256)> { - self.peers.iter().map(|peer| (peer.0.clone(), peer.1.latest.clone())).collect() - } - - fn get_lagging_peers(&self, io: &SyncIo) -> Vec<(usize, H256)> { + fn get_lagging_peers(&self, io: &SyncIo) -> Vec { let chain = io.chain(); let chain_info = chain.chain_info(); let latest_hash = chain_info.best_block_hash; - self.query_peer_latest_blocks().iter().filter(|peer| - match io.chain().block_status(&peer.1) + let latest_number = chain_info.best_block_number; + self.peers.iter().filter(|&(peer_id, peer_info)| + match io.chain().block_status(&peer_info.latest) { - BlockStatus::InChain => peer.1 != latest_hash, + BlockStatus::InChain => peer_info.latest != latest_hash && latest_number - peer_info.latest_number < MAX_PEER_LAG_PROPAGATION, _ => false - }).cloned().collect::>() + }) + .map(|(peer_id, peer_info)| peer_id) + .cloned().collect::>() } fn propagade_blocks(&mut self, io: &mut SyncIo) -> usize { let updated_peers = { - let lagging_peers = self.get_lagging_peers(io); - let lucky_peers = match lagging_peers.len() { + // sqrt(x)/x scaled to max u32 + let fraction = (self.peers.len() as f64).powf(-0.5).mul(u32::max_value() as f64).round() as u32; + let mut lucky_peers = match lagging_peers.len() { 0 ... MIN_PEERS_PROPAGATION => lagging_peers, - _ => lagging_peers.iter().filter(|_| ::rand::random::() < 64u8).cloned().collect::>() + _ => lagging_peers.iter().filter(|_| ::rand::random::() < fraction).cloned().collect::>() }; - match lucky_peers.len() { - 0 ... MAX_PEERS_PROPAGATION => lucky_peers, - _ => lucky_peers.iter().take(MAX_PEERS_PROPAGATION).cloned().collect::>() - } + // taking at max of MAX_PEERS_PROPAGATION + lucky_peers.iter().take(min(lucky_peers.len(), MAX_PEERS_PROPAGATION)).cloned().collect::>() }; let mut sent = 0; - for (peer_id, peer_hash) in updated_peers { - sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &peer_hash, &io.chain().chain_info().best_block_hash) { + for peer_id in updated_peers { + sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &self.peers[&peer_id].latest, &io.chain().chain_info().best_block_hash) { Some(rlp) => { self.send_request(io, peer_id, PeerAsking::Nothing, NEW_BLOCK_HASHES_PACKET, rlp); 1 @@ -1124,6 +1131,16 @@ impl ChainSync { trace!(target: "sync", "Sent new blocks to peers: {:?}", blocks_propagaded); } } + + #[cfg(test)] + pub fn get_peer_latet(&self, peer_id: usize) -> H256 { + self.peers[&peer_id].latest.clone() + } + + #[cfg(test)] + pub fn get_peer_latest_number(&self, peer_id: usize) -> BlockNumber { + self.peers[&peer_id].latest_number + } } #[cfg(test)] @@ -1205,16 +1222,17 @@ mod tests { fn dummy_sync_with_peer(peer_latest_hash: H256) -> ChainSync { let mut sync = ChainSync::new(); sync.peers.insert(0, - PeerInfo { - protocol_version: 0, - genesis: H256::zero(), - network_id: U256::zero(), - latest: peer_latest_hash, - difficulty: U256::zero(), - asking: PeerAsking::Nothing, - asking_blocks: Vec::::new(), - ask_time: 0f64, - }); + PeerInfo { + protocol_version: 0, + genesis: H256::zero(), + network_id: U256::zero(), + latest: peer_latest_hash, + latest_number: 90, + difficulty: U256::zero(), + asking: PeerAsking::Nothing, + asking_blocks: Vec::::new(), + ask_time: 0f64, + }); sync } @@ -1251,17 +1269,17 @@ mod tests { #[test] fn sends_packet_to_lagging_peer() { let mut client = TestBlockChainClient::new(); - client.add_blocks(20, false); + client.add_blocks(100, false); let mut queue = VecDeque::new(); let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); let mut io = TestIo::new(&mut client, &mut queue, None); - let block_count = sync.propagade_blocks(&mut io); + let peer_count = sync.propagade_blocks(&mut io); // 1 message should be send assert_eq!(1, io.queue.len()); // 1 peer should be updated - assert_eq!(1, block_count); + assert_eq!(1, peer_count); // NEW_BLOCK_HASHES_PACKET assert_eq!(0x01, io.queue[0].packet_id); } diff --git a/sync/src/tests/chain.rs b/sync/src/tests/chain.rs index a9aeb2e34..22c677aa0 100644 --- a/sync/src/tests/chain.rs +++ b/sync/src/tests/chain.rs @@ -126,10 +126,18 @@ fn propagade() { 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); net.peer_mut(0).chain.add_blocks(10, false); + assert_eq!(1010, net.peer(0).chain.chain_info().best_block_number); + assert_eq!(1000, net.peer(1).chain.chain_info().best_block_number); + assert_eq!(1000, net.peer(2).chain.chain_info().best_block_number); + + assert_eq!(net.peer(0).sync.get_peer_latest_number(1), 1000); + net.sync_step_peer(0); // 2 peers to sync From 49e61b87a057e271be567836176ad92d6486f060 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sat, 6 Feb 2016 22:16:59 +0300 Subject: [PATCH 05/30] calculating peer highest number on fly --- sync/src/chain.rs | 21 +++++++-------------- sync/src/tests/chain.rs | 2 -- sync/src/tests/helpers.rs | 1 - 3 files changed, 7 insertions(+), 17 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index b9d0ccddf..5b4e97dc5 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1084,13 +1084,16 @@ impl ChainSync { let chain_info = chain.chain_info(); let latest_hash = chain_info.best_block_hash; let latest_number = chain_info.best_block_number; - self.peers.iter().filter(|&(peer_id, peer_info)| + self.peers.iter().filter(|&(_, peer_info)| match io.chain().block_status(&peer_info.latest) { - BlockStatus::InChain => peer_info.latest != latest_hash && latest_number - peer_info.latest_number < MAX_PEER_LAG_PROPAGATION, + BlockStatus::InChain => { + let peer_number = HeaderView::new(&io.chain().block_header(&peer_info.latest).unwrap()).number(); + peer_info.latest != latest_hash && latest_number > peer_number && latest_number - peer_number < MAX_PEER_LAG_PROPAGATION + }, _ => false }) - .map(|(peer_id, peer_info)| peer_id) + .map(|(peer_id, _)| peer_id) .cloned().collect::>() } @@ -1100,7 +1103,7 @@ impl ChainSync { // sqrt(x)/x scaled to max u32 let fraction = (self.peers.len() as f64).powf(-0.5).mul(u32::max_value() as f64).round() as u32; - let mut lucky_peers = match lagging_peers.len() { + let lucky_peers = match lagging_peers.len() { 0 ... MIN_PEERS_PROPAGATION => lagging_peers, _ => lagging_peers.iter().filter(|_| ::rand::random::() < fraction).cloned().collect::>() }; @@ -1131,16 +1134,6 @@ impl ChainSync { trace!(target: "sync", "Sent new blocks to peers: {:?}", blocks_propagaded); } } - - #[cfg(test)] - pub fn get_peer_latet(&self, peer_id: usize) -> H256 { - self.peers[&peer_id].latest.clone() - } - - #[cfg(test)] - pub fn get_peer_latest_number(&self, peer_id: usize) -> BlockNumber { - self.peers[&peer_id].latest_number - } } #[cfg(test)] diff --git a/sync/src/tests/chain.rs b/sync/src/tests/chain.rs index 22c677aa0..a78af0ca6 100644 --- a/sync/src/tests/chain.rs +++ b/sync/src/tests/chain.rs @@ -136,8 +136,6 @@ fn propagade() { assert_eq!(1000, net.peer(1).chain.chain_info().best_block_number); assert_eq!(1000, net.peer(2).chain.chain_info().best_block_number); - assert_eq!(net.peer(0).sync.get_peer_latest_number(1), 1000); - net.sync_step_peer(0); // 2 peers to sync diff --git a/sync/src/tests/helpers.rs b/sync/src/tests/helpers.rs index 2be501ebb..8fc9d8cf0 100644 --- a/sync/src/tests/helpers.rs +++ b/sync/src/tests/helpers.rs @@ -84,7 +84,6 @@ impl BlockChainClient for TestBlockChainClient { fn block_header(&self, h: &H256) -> Option { self.blocks.read().unwrap().get(h).map(|r| Rlp::new(r).at(0).as_raw().to_vec()) - } fn block_body(&self, h: &H256) -> Option { From 0905372f7027a9d0dbce1a552428b3018c16dde3 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sat, 6 Feb 2016 22:23:25 +0300 Subject: [PATCH 06/30] updating peer best hash when sync --- sync/src/chain.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 5b4e97dc5..89c75feff 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1113,9 +1113,16 @@ impl ChainSync { }; let mut sent = 0; + let local_best = io.chain().chain_info().best_block_hash; for peer_id in updated_peers { - sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &self.peers[&peer_id].latest, &io.chain().chain_info().best_block_hash) { + sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), + &self.peers.get(&peer_id).expect("ChainSync: unknown peer").latest, + &local_best) { Some(rlp) => { + { + let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer"); + peer.latest = local_best.clone(); + } self.send_request(io, peer_id, PeerAsking::Nothing, NEW_BLOCK_HASHES_PACKET, rlp); 1 }, From 6b02b6eddb31fc7abb8779335b509a5125142a84 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sat, 6 Feb 2016 22:25:20 +0300 Subject: [PATCH 07/30] using rlp::encode --- sync/src/chain.rs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 89c75feff..d347cafa9 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1067,11 +1067,7 @@ impl ChainSync { match route.blocks.len() { 0 => None, _ => { - let mut rlp_stream = RlpStream::new_list(route.blocks.len()); - for hash in route.blocks { - rlp_stream.append(&hash); - } - Some(rlp_stream.out()) + Some(rlp::encode(&route.blocks).to_vec()) } } }, From 9727f27854788aef98dbec98105e02de50610d53 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sat, 6 Feb 2016 23:00:52 +0300 Subject: [PATCH 08/30] blocks + hashes --- sync/src/chain.rs | 48 ++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 43 insertions(+), 5 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index d347cafa9..63965c676 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1075,6 +1075,10 @@ impl ChainSync { } } + fn create_latest_block_rlp(chain: &BlockChainClient) -> Bytes { + chain.block(&chain.chain_info().best_block_hash).unwrap() + } + fn get_lagging_peers(&self, io: &SyncIo) -> Vec { let chain = io.chain(); let chain_info = chain.chain_info(); @@ -1108,6 +1112,19 @@ impl ChainSync { lucky_peers.iter().take(min(lucky_peers.len(), MAX_PEERS_PROPAGATION)).cloned().collect::>() }; + let mut sent = 0; + let local_best = io.chain().chain_info().best_block_hash; + for peer_id in updated_peers { + let rlp = ChainSync::create_latest_block_rlp(io.chain()); + self.send_request(io, peer_id, PeerAsking::Nothing, NEW_BLOCK_PACKET, rlp); + self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer").latest = local_best.clone(); + sent = sent + 1; + } + sent + } + + fn propagade_new_hashes(&mut self, io: &mut SyncIo) -> usize { + let updated_peers = self.get_lagging_peers(io); let mut sent = 0; let local_best = io.chain().chain_info().best_block_hash; for peer_id in updated_peers { @@ -1133,8 +1150,11 @@ impl ChainSync { self.check_resume(io); if self.state == SyncState::Idle { - let blocks_propagaded = self.propagade_blocks(io); - trace!(target: "sync", "Sent new blocks to peers: {:?}", blocks_propagaded); + let peers = self.propagade_new_hashes(io); + trace!(target: "sync", "Sent new hashes to peers: {:?}", peers); + + let peers = self.propagade_blocks(io); + trace!(target: "sync", "Sent latest block to peers: {:?}", peers); } } } @@ -1263,7 +1283,25 @@ mod tests { } #[test] - fn sends_packet_to_lagging_peer() { + fn sends_new_hashes_to_lagging_peer() { + let mut client = TestBlockChainClient::new(); + client.add_blocks(100, false); + let mut queue = VecDeque::new(); + let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); + let mut io = TestIo::new(&mut client, &mut queue, None); + + let peer_count = sync.propagade_new_hashes(&mut io); + + // 1 message should be send + assert_eq!(1, io.queue.len()); + // 1 peer should be updated + assert_eq!(1, peer_count); + // NEW_BLOCK_HASHES_PACKET + assert_eq!(0x01, io.queue[0].packet_id); + } + + #[test] + fn sends_latest_block_to_lagging_peer() { let mut client = TestBlockChainClient::new(); client.add_blocks(100, false); let mut queue = VecDeque::new(); @@ -1276,7 +1314,7 @@ mod tests { assert_eq!(1, io.queue.len()); // 1 peer should be updated assert_eq!(1, peer_count); - // NEW_BLOCK_HASHES_PACKET - assert_eq!(0x01, io.queue[0].packet_id); + // NEW_BLOCK_PACKET + assert_eq!(0x07, io.queue[0].packet_id); } } \ No newline at end of file From 391ef7e6644c8b5ebbaeccc23180990254be3cc4 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sat, 6 Feb 2016 23:03:26 +0300 Subject: [PATCH 09/30] actually should be this way --- sync/src/chain.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 63965c676..2acc86fc0 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1150,11 +1150,11 @@ impl ChainSync { self.check_resume(io); if self.state == SyncState::Idle { - let peers = self.propagade_new_hashes(io); - trace!(target: "sync", "Sent new hashes to peers: {:?}", peers); - let peers = self.propagade_blocks(io); trace!(target: "sync", "Sent latest block to peers: {:?}", peers); + + let peers = self.propagade_new_hashes(io); + trace!(target: "sync", "Sent new hashes to peers: {:?}", peers); } } } From 8cd55276c347c11feec6b3330016b08960b773fe Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sat, 6 Feb 2016 23:04:58 +0300 Subject: [PATCH 10/30] ... and test as well --- sync/src/tests/chain.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sync/src/tests/chain.rs b/sync/src/tests/chain.rs index a78af0ca6..0e12764bd 100644 --- a/sync/src/tests/chain.rs +++ b/sync/src/tests/chain.rs @@ -140,6 +140,6 @@ fn propagade() { // 2 peers to sync assert_eq!(2, net.peer(0).queue.len()); - // NEW_BLOCK_HASHES_PACKET - assert_eq!(0x01, net.peer(0).queue[0].packet_id); + // NEW_BLOCK_PACKET + assert_eq!(0x07, net.peer(0).queue[0].packet_id); } \ No newline at end of file From 74c97ea36db4792b69f344de6e6a2ff7894039db Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sat, 6 Feb 2016 23:08:20 +0300 Subject: [PATCH 11/30] removed unused latest_number --- sync/src/chain.rs | 13 +------------ 1 file changed, 1 insertion(+), 12 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 2acc86fc0..64e425392 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -164,8 +164,6 @@ struct PeerInfo { asking_blocks: Vec, /// Request timestamp ask_time: f64, - /// Latest block number - latest_number: BlockNumber } /// Blockchain sync handler. @@ -271,7 +269,7 @@ impl ChainSync { /// Called by peer to report status fn on_peer_status(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> { - let mut peer = PeerInfo { + let peer = PeerInfo { protocol_version: try!(r.val_at(0)), network_id: try!(r.val_at(1)), difficulty: try!(r.val_at(2)), @@ -280,13 +278,8 @@ impl ChainSync { asking: PeerAsking::Nothing, asking_blocks: Vec::new(), ask_time: 0f64, - latest_number: 0, }; - if io.chain().block_status(&peer.latest) == BlockStatus::InChain { - peer.latest_number = HeaderView::new(&io.chain().block_header(&peer.latest).unwrap()).number(); - } - 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(); @@ -450,8 +443,6 @@ impl ChainSync { match io.chain().import_block(block_rlp.as_raw().to_vec()) { Err(ImportError::AlreadyInChain) => { trace!(target: "sync", "New block already in chain {:?}", h); - let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer"); - peer.latest_number = max(peer.latest_number, header_view.number()); }, Err(ImportError::AlreadyQueued) => { trace!(target: "sync", "New block already queued {:?}", h); @@ -482,7 +473,6 @@ impl ChainSync { { let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer"); peer.latest = header_view.sha3(); - peer.latest_number = header_view.number(); } self.sync_peer(io, peer_id, true); } @@ -1243,7 +1233,6 @@ mod tests { genesis: H256::zero(), network_id: U256::zero(), latest: peer_latest_hash, - latest_number: 90, difficulty: U256::zero(), asking: PeerAsking::Nothing, asking_blocks: Vec::::new(), From 3e84691cecb5960194be1d77e588f8663424416b Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sat, 6 Feb 2016 23:31:37 +0300 Subject: [PATCH 12/30] adding expect --- sync/src/chain.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 64e425392..fe4ec30cd 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1066,7 +1066,7 @@ impl ChainSync { } fn create_latest_block_rlp(chain: &BlockChainClient) -> Bytes { - chain.block(&chain.chain_info().best_block_hash).unwrap() + chain.block(&chain.chain_info().best_block_hash).expect("Creating latest block when there is none") } fn get_lagging_peers(&self, io: &SyncIo) -> Vec { From 67c5e376b8824a2d6b99fa55651008b5c8cd4347 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sat, 6 Feb 2016 23:40:41 +0300 Subject: [PATCH 13/30] review fixes --- sync/src/chain.rs | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index fe4ec30cd..91df61145 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1069,7 +1069,7 @@ impl ChainSync { chain.block(&chain.chain_info().best_block_hash).expect("Creating latest block when there is none") } - fn get_lagging_peers(&self, io: &SyncIo) -> Vec { + fn get_lagging_peers(&self, io: &SyncIo) -> Vec { let chain = io.chain(); let chain_info = chain.chain_info(); let latest_hash = chain_info.best_block_hash; @@ -1084,7 +1084,7 @@ impl ChainSync { _ => false }) .map(|(peer_id, _)| peer_id) - .cloned().collect::>() + .cloned().collect::>() } fn propagade_blocks(&mut self, io: &mut SyncIo) -> usize { @@ -1095,11 +1095,11 @@ impl ChainSync { 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() { 0 ... MIN_PEERS_PROPAGATION => lagging_peers, - _ => lagging_peers.iter().filter(|_| ::rand::random::() < fraction).cloned().collect::>() + _ => lagging_peers.iter().filter(|_| ::rand::random::() < fraction).cloned().collect::>() }; // taking at max of MAX_PEERS_PROPAGATION - lucky_peers.iter().take(min(lucky_peers.len(), MAX_PEERS_PROPAGATION)).cloned().collect::>() + lucky_peers.iter().take(min(lucky_peers.len(), MAX_PEERS_PROPAGATION)).cloned().collect::>() }; let mut sent = 0; @@ -1118,9 +1118,7 @@ impl ChainSync { let mut sent = 0; let local_best = io.chain().chain_info().best_block_hash; for peer_id 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) { + sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &self.peers.get(&peer_id).expect("ChainSync: unknown peer").latest, &local_best) { Some(rlp) => { { let peer = self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer"); From 0e0f1fea696a68d50ea191a84ebe7b68dc12ee97 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sun, 7 Feb 2016 01:15:53 +0300 Subject: [PATCH 14/30] tests --- ethcore/src/block_queue.rs | 14 +++++++++++++- ethcore/src/client.rs | 8 ++++++-- 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/ethcore/src/block_queue.rs b/ethcore/src/block_queue.rs index 2e3728aee..48b55fc0d 100644 --- a/ethcore/src/block_queue.rs +++ b/ethcore/src/block_queue.rs @@ -38,6 +38,8 @@ pub struct BlockQueueInfo { pub verified_queue_size: usize, /// Number of blocks being verified pub verifying_queue_size: usize, + /// Indicates queue is empty + pub empty: bool } impl BlockQueueInfo { @@ -285,7 +287,6 @@ impl BlockQueue { for h in hashes { processing.remove(&h); } - //TODO: reward peers } /// Removes up to `max` verified blocks from the queue @@ -312,6 +313,7 @@ impl BlockQueue { verified_queue_size: verification.verified.len(), unverified_queue_size: verification.unverified.len(), verifying_queue_size: verification.verifying.len(), + empty: verification.verified.is_empty() && verification.unverified.is_empty() && verification.verifying.is_empty(), } } } @@ -393,4 +395,14 @@ mod tests { panic!("error importing block that has already been drained ({:?})", e); } } + + #[test] + fn returns_empty_once_finished() { + let mut queue = get_test_queue(); + queue.import_block(get_good_dummy_block()).expect("error importing block that is valid by definition"); + queue.flush(); + queue.drain(1); + + assert!(queue.queue_info().empty); + } } diff --git a/ethcore/src/client.rs b/ethcore/src/client.rs index 3a0309c1c..3b5627504 100644 --- a/ethcore/src/client.rs +++ b/ethcore/src/client.rs @@ -27,7 +27,7 @@ use spec::Spec; use engine::Engine; use views::HeaderView; use block_queue::{BlockQueue, BlockQueueInfo}; -use service::NetSyncMessage; +use service::{NetSyncMessage, SyncMessage}; use env_info::LastHashes; use verification::*; use block::*; @@ -223,7 +223,7 @@ impl Client { } /// This is triggered by a message coming from a block queue when the block is ready for insertion - pub fn import_verified_blocks(&self, _io: &IoChannel) -> usize { + pub fn import_verified_blocks(&self, io: &IoChannel) -> usize { let mut ret = 0; let mut bad = HashSet::new(); let _import_lock = self.import_lock.lock(); @@ -295,6 +295,10 @@ impl Client { self.report.write().unwrap().accrue_block(&block); trace!(target: "client", "Imported #{} ({})", header.number(), header.hash()); ret += 1; + + if self.block_queue.read().unwrap().queue_info().empty { + io.send(NetworkIoMessage::User(SyncMessage::BlockVerified)).unwrap(); + } } self.block_queue.write().unwrap().mark_as_good(&good_blocks); ret From d40d4ef87c54fad5f4b643d84f52db567c386982 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sun, 7 Feb 2016 01:43:44 +0300 Subject: [PATCH 15/30] fix tests --- sync/src/tests/helpers.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/sync/src/tests/helpers.rs b/sync/src/tests/helpers.rs index 8fc9d8cf0..262037e0e 100644 --- a/sync/src/tests/helpers.rs +++ b/sync/src/tests/helpers.rs @@ -233,6 +233,7 @@ impl BlockChainClient for TestBlockChainClient { verified_queue_size: 0, unverified_queue_size: 0, verifying_queue_size: 0, + empty: false, } } From efef36b5e8bf3c6d8872d5aa56b9a05833722e00 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sun, 7 Feb 2016 03:00:43 +0300 Subject: [PATCH 16/30] handling sync message --- sync/src/chain.rs | 12 ++++++------ sync/src/lib.rs | 6 ++++++ sync/src/tests/chain.rs | 24 +++++++++++++++--------- sync/src/tests/helpers.rs | 5 +++++ 4 files changed, 32 insertions(+), 15 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 91df61145..dc9caad9a 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1137,13 +1137,13 @@ impl ChainSync { pub fn maintain_sync(&mut self, io: &mut SyncIo) { self.check_resume(io); - if self.state == SyncState::Idle { - let peers = self.propagade_blocks(io); - trace!(target: "sync", "Sent latest block to peers: {:?}", peers); + let peers = self.propagade_new_hashes(io); + trace!(target: "sync", "Sent new hashes to peers: {:?}", peers); + } - let peers = self.propagade_new_hashes(io); - trace!(target: "sync", "Sent new hashes to peers: {:?}", peers); - } + pub fn chain_blocks_verified(&mut self, io: &mut SyncIo) { + let peers = self.propagade_blocks(io); + trace!(target: "sync", "Sent latest block to peers: {:?}", peers); } } diff --git a/sync/src/lib.rs b/sync/src/lib.rs index 520b4bcc7..b2d1fc29f 100644 --- a/sync/src/lib.rs +++ b/sync/src/lib.rs @@ -126,4 +126,10 @@ impl NetworkProtocolHandler for EthSync { self.sync.write().unwrap().maintain_peers(&mut NetSyncIo::new(io, self.chain.deref())); self.sync.write().unwrap().maintain_sync(&mut NetSyncIo::new(io, self.chain.deref())); } + + fn message(&self, io: &NetworkContext, message: &SyncMessage) { + if let SyncMessage::BlockVerified = *message { + self.sync.write().unwrap().chain_blocks_verified(&mut NetSyncIo::new(io, self.chain.deref())); + } + } } \ No newline at end of file diff --git a/sync/src/tests/chain.rs b/sync/src/tests/chain.rs index 0e12764bd..34f94f7e2 100644 --- a/sync/src/tests/chain.rs +++ b/sync/src/tests/chain.rs @@ -121,25 +121,31 @@ fn status_packet() { } #[test] -fn propagade() { +fn propagade_hashes() { 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); - net.peer_mut(0).chain.add_blocks(10, false); - assert_eq!(1010, net.peer(0).chain.chain_info().best_block_number); - assert_eq!(1000, net.peer(1).chain.chain_info().best_block_number); - assert_eq!(1000, net.peer(2).chain.chain_info().best_block_number); - net.sync_step_peer(0); // 2 peers to sync assert_eq!(2, net.peer(0).queue.len()); + // NEW_BLOCK_HASHES_PACKET + assert_eq!(0x01, net.peer(0).queue[0].packet_id); +} + +#[test] +fn propagade_blocks() { + let mut net = TestNet::new(10); + net.peer_mut(1).chain.add_blocks(10, false); + net.sync(); + + net.peer_mut(0).chain.add_blocks(10, false); + net.trigger_block_verified(0); + + assert!(!net.peer(0).queue.is_empty()); // NEW_BLOCK_PACKET assert_eq!(0x07, net.peer(0).queue[0].packet_id); } \ No newline at end of file diff --git a/sync/src/tests/helpers.rs b/sync/src/tests/helpers.rs index 262037e0e..da82363dd 100644 --- a/sync/src/tests/helpers.rs +++ b/sync/src/tests/helpers.rs @@ -395,4 +395,9 @@ impl TestNet { pub fn done(&self) -> bool { self.peers.iter().all(|p| p.queue.is_empty()) } + + pub fn trigger_block_verified(&mut self, peer_id: usize) { + let mut peer = self.peer_mut(peer_id); + peer.sync.chain_blocks_verified(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None)); + } } From 3f17acca1d89f3cfcab663f88886a317ac63527d Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Sun, 7 Feb 2016 23:01:09 +0300 Subject: [PATCH 17/30] empty new block test --- sync/src/chain.rs | 16 ++++++++++++++++ sync/src/tests/chain.rs | 2 +- 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index dc9caad9a..778da490c 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1304,4 +1304,20 @@ mod tests { // NEW_BLOCK_PACKET assert_eq!(0x07, io.queue[0].packet_id); } + + #[test] + fn handles_empty_peer_new_block() { + let mut client = TestBlockChainClient::new(); + client.add_blocks(10, false); + let mut queue = VecDeque::new(); + let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); + let mut io = TestIo::new(&mut client, &mut queue, None); + + let empty_data = vec![]; + let block = UntrustedRlp::new(&empty_data); + + let result = sync.on_peer_new_block(&mut io, 0, &block); + + assert!(result.is_err()); + } } \ No newline at end of file diff --git a/sync/src/tests/chain.rs b/sync/src/tests/chain.rs index 34f94f7e2..6526d8500 100644 --- a/sync/src/tests/chain.rs +++ b/sync/src/tests/chain.rs @@ -138,7 +138,7 @@ fn propagade_hashes() { #[test] fn propagade_blocks() { - let mut net = TestNet::new(10); + let mut net = TestNet::new(2); net.peer_mut(1).chain.add_blocks(10, false); net.sync(); From 4b1d67ef49923cc67d21eebc9e0df38efb569020 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Mon, 8 Feb 2016 00:08:15 +0300 Subject: [PATCH 18/30] bunch of tests for new block packet --- sync/src/chain.rs | 72 +++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 66 insertions(+), 6 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 778da490c..31f03fd9a 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -434,12 +434,11 @@ impl ChainSync { let block_rlp = try!(r.at(0)); let header_rlp = try!(block_rlp.at(0)); let h = header_rlp.as_raw().sha3(); - trace!(target: "sync", "{} -> NewBlock ({})", peer_id, h); - let header_view = HeaderView::new(header_rlp.as_raw()); + let header: BlockHeader = try!(header_rlp.as_val()); 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) { + if header.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); @@ -472,7 +471,7 @@ impl ChainSync { 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_view.sha3(); + peer.latest = header.hash(); } self.sync_peer(io, peer_id, true); } @@ -1153,7 +1152,32 @@ mod tests { use super::*; use util::*; use super::{PeerInfo, PeerAsking}; - use ethcore::header::{BlockNumber}; + use ethcore::header::*; + use ethcore::client::*; + + fn get_dummy_block(order: u32, parent_hash: H256) -> Bytes { + let mut header = Header::new(); + header.gas_limit = x!(0); + header.difficulty = x!(order * 100); + header.timestamp = (order * 10) as u64; + header.number = order as u64; + header.parent_hash = parent_hash; + header.state_root = H256::zero(); + + let mut rlp = RlpStream::new_list(3); + rlp.append(&header); + rlp.append_raw(&rlp::EMPTY_LIST_RLP, 1); + rlp.append_raw(&rlp::EMPTY_LIST_RLP, 1); + rlp.out() + } + + fn get_dummy_blocks(order: u32, parent_hash: H256) -> Bytes { + let mut rlp = RlpStream::new_list(1); + rlp.append_raw(&get_dummy_block(order, parent_hash), 1); + let difficulty: U256 = x!(100 * order); + rlp.append(&difficulty); + rlp.out() + } #[test] fn return_receipts_empty() { @@ -1306,7 +1330,43 @@ mod tests { } #[test] - fn handles_empty_peer_new_block() { + fn handles_peer_new_block_mallformed() { + let mut client = TestBlockChainClient::new(); + client.add_blocks(10, false); + + let block_data = get_dummy_block(11, client.chain_info().best_block_hash); + + let mut queue = VecDeque::new(); + let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); + let mut io = TestIo::new(&mut client, &mut queue, None); + + let block = UntrustedRlp::new(&block_data); + + let result = sync.on_peer_new_block(&mut io, 0, &block); + + assert!(result.is_err()); + } + + #[test] + fn handles_peer_new_block() { + let mut client = TestBlockChainClient::new(); + client.add_blocks(10, false); + + let block_data = get_dummy_blocks(11, client.chain_info().best_block_hash); + + let mut queue = VecDeque::new(); + let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); + let mut io = TestIo::new(&mut client, &mut queue, None); + + let block = UntrustedRlp::new(&block_data); + + let result = sync.on_peer_new_block(&mut io, 0, &block); + + assert!(result.is_ok()); + } + + #[test] + fn handles_peer_new_block_empty() { let mut client = TestBlockChainClient::new(); client.add_blocks(10, false); let mut queue = VecDeque::new(); From e9af2dfd9669548a15efdc1404574a0a4b38ae29 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Mon, 8 Feb 2016 00:20:59 +0300 Subject: [PATCH 19/30] new hashes tests --- sync/src/chain.rs | 47 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 31f03fd9a..571e73226 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1179,6 +1179,21 @@ mod tests { rlp.out() } + fn get_dummy_hashes() -> Bytes { + let mut rlp = RlpStream::new_list(5); + for _ in 0..5 { + let mut hash_d_rlp = RlpStream::new_list(2); + let hash: H256 = H256::from(0u64); + let diff: U256 = U256::from(1u64); + hash_d_rlp.append(&hash); + hash_d_rlp.append(&diff); + + rlp.append_raw(&hash_d_rlp.out(), 1); + } + + rlp.out() + } + #[test] fn return_receipts_empty() { let mut client = TestBlockChainClient::new(); @@ -1380,4 +1395,36 @@ mod tests { assert!(result.is_err()); } + + #[test] + fn handles_peer_new_hashes() { + let mut client = TestBlockChainClient::new(); + client.add_blocks(10, false); + let mut queue = VecDeque::new(); + let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); + let mut io = TestIo::new(&mut client, &mut queue, None); + + let hashes_data = get_dummy_hashes(); + let hashes_rlp = UntrustedRlp::new(&hashes_data); + + let result = sync.on_peer_new_hashes(&mut io, 0, &hashes_rlp); + + assert!(result.is_ok()); + } + + #[test] + fn handles_peer_new_hashes_empty() { + let mut client = TestBlockChainClient::new(); + client.add_blocks(10, false); + let mut queue = VecDeque::new(); + let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); + let mut io = TestIo::new(&mut client, &mut queue, None); + + let empty_hashes_data = vec![]; + let hashes_rlp = UntrustedRlp::new(&empty_hashes_data); + + let result = sync.on_peer_new_hashes(&mut io, 0, &hashes_rlp); + + assert!(result.is_ok()); + } } \ No newline at end of file From 69a4349ee21793a0b5d3769947faab1cee853cba Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Mon, 8 Feb 2016 00:52:56 +0300 Subject: [PATCH 20/30] documentation --- sync/src/chain.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index dc9caad9a..9470e9c80 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1051,6 +1051,7 @@ impl ChainSync { } } + /// creates rlp to send for the tree defined by 'from' and 'to' hashes fn create_new_hashes_rlp(chain: &BlockChainClient, from: &H256, to: &H256) -> Option { match chain.tree_route(from, to) { Some(route) => { @@ -1065,10 +1066,12 @@ impl ChainSync { } } + /// creates latest block rlp for the given client fn create_latest_block_rlp(chain: &BlockChainClient) -> Bytes { chain.block(&chain.chain_info().best_block_hash).expect("Creating latest block when there is none") } + /// returns peer ids that have less blocks than our chain fn get_lagging_peers(&self, io: &SyncIo) -> Vec { let chain = io.chain(); let chain_info = chain.chain_info(); @@ -1087,6 +1090,7 @@ impl ChainSync { .cloned().collect::>() } + /// propagades latest block to lagging peers fn propagade_blocks(&mut self, io: &mut SyncIo) -> usize { let updated_peers = { let lagging_peers = self.get_lagging_peers(io); @@ -1113,6 +1117,7 @@ impl ChainSync { sent } + /// propagades new known hashes to all peers fn propagade_new_hashes(&mut self, io: &mut SyncIo) -> usize { let updated_peers = self.get_lagging_peers(io); let mut sent = 0; @@ -1141,6 +1146,7 @@ impl ChainSync { trace!(target: "sync", "Sent new hashes to peers: {:?}", peers); } + /// should be called once chain has new block, triggers the latest block propagation pub fn chain_blocks_verified(&mut self, io: &mut SyncIo) { let peers = self.propagade_blocks(io); trace!(target: "sync", "Sent latest block to peers: {:?}", peers); From 871b7113ecf59bf02758400dee8cbf542e2469e6 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Mon, 8 Feb 2016 01:39:02 +0300 Subject: [PATCH 21/30] fixes for valid rlp --- sync/src/chain.rs | 49 +++++++++++++++++++++++++++++++++++---- sync/src/tests/helpers.rs | 2 +- 2 files changed, 46 insertions(+), 5 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 571e73226..0c3c9f4bc 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -1056,7 +1056,15 @@ impl ChainSync { match route.blocks.len() { 0 => None, _ => { - Some(rlp::encode(&route.blocks).to_vec()) + let mut rlp_stream = RlpStream::new_list(route.blocks.len()); + for block_hash in route.blocks { + let mut hash_rlp = RlpStream::new_list(2); + let difficulty = chain.block_total_difficulty(&block_hash).expect("Mallformed block without a difficulty on the chain!"); + hash_rlp.append(&block_hash); + hash_rlp.append(&difficulty); + rlp_stream.append_raw(&hash_rlp.out(), 1); + } + Some(rlp_stream.out()) } } }, @@ -1065,7 +1073,10 @@ impl ChainSync { } fn create_latest_block_rlp(chain: &BlockChainClient) -> Bytes { - chain.block(&chain.chain_info().best_block_hash).expect("Creating latest block when there is none") + let mut rlp_stream = RlpStream::new_list(2); + rlp_stream.append_raw(&chain.block(&chain.chain_info().best_block_hash).expect("Creating latest block when there is none"), 1); + rlp_stream.append(&chain.chain_info().total_difficulty); + rlp_stream.out() } fn get_lagging_peers(&self, io: &SyncIo) -> Vec { @@ -1304,8 +1315,8 @@ mod tests { assert!(rlp.is_none()); let rlp = ChainSync::create_new_hashes_rlp(&client, &start, &end).unwrap(); - // size of three rlp encoded hash - assert_eq!(101, rlp.len()); + // size of three rlp encoded hash-difficulty + assert_eq!(107, rlp.len()); } #[test] @@ -1427,4 +1438,34 @@ mod tests { assert!(result.is_ok()); } + + #[test] + fn hashes_rlp_mutually_acceptable() { + let mut client = TestBlockChainClient::new(); + client.add_blocks(100, false); + let mut queue = VecDeque::new(); + let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); + let mut io = TestIo::new(&mut client, &mut queue, None); + + sync.propagade_new_hashes(&mut io); + + let data = &io.queue[0].data.clone(); + let result = sync.on_peer_new_hashes(&mut io, 0, &UntrustedRlp::new(&data)); + assert!(result.is_ok()); + } + + #[test] + fn block_rlp_mutually_acceptable() { + let mut client = TestBlockChainClient::new(); + client.add_blocks(100, false); + let mut queue = VecDeque::new(); + let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5)); + let mut io = TestIo::new(&mut client, &mut queue, None); + + sync.propagade_blocks(&mut io); + + let data = &io.queue[0].data.clone(); + let result = sync.on_peer_new_block(&mut io, 0, &UntrustedRlp::new(&data)); + assert!(result.is_ok()); + } } \ No newline at end of file diff --git a/sync/src/tests/helpers.rs b/sync/src/tests/helpers.rs index da82363dd..d155fee6b 100644 --- a/sync/src/tests/helpers.rs +++ b/sync/src/tests/helpers.rs @@ -79,7 +79,7 @@ impl TestBlockChainClient { impl BlockChainClient for TestBlockChainClient { fn block_total_difficulty(&self, _h: &H256) -> Option { - unimplemented!(); + Some(U256::zero()) } fn block_header(&self, h: &H256) -> Option { From deffb271bc791c4eae572900f8e2a4bb428f1933 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Mon, 8 Feb 2016 03:14:39 -0800 Subject: [PATCH 22/30] refactoring of report functions, some comments --- sync/src/chain.rs | 8 ++++++-- sync/src/tests/helpers.rs | 2 -- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/sync/src/chain.rs b/sync/src/chain.rs index f9dfbf310..63dc47024 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -596,7 +596,7 @@ impl ChainSync { fn request_blocks(&mut self, io: &mut SyncIo, peer_id: PeerId) { self.clear_peer_download(peer_id); - if io.chain().queue_info().full { + if io.chain().queue_info().is_full() { self.pause_sync(); return; } @@ -1044,7 +1044,7 @@ impl ChainSync { } fn check_resume(&mut self, io: &mut SyncIo) { - if !io.chain().queue_info().full && self.state == SyncState::Waiting { + if !io.chain().queue_info().is_full() && self.state == SyncState::Waiting { self.state = SyncState::Idle; self.continue_sync(io); } @@ -1445,6 +1445,8 @@ mod tests { assert!(result.is_ok()); } + // idea is that what we produce when propagading latest hashes should be accepted in + // on_peer_new_hashes in our code as well #[test] fn hashes_rlp_mutually_acceptable() { let mut client = TestBlockChainClient::new(); @@ -1460,6 +1462,8 @@ mod tests { assert!(result.is_ok()); } + // idea is that what we produce when propagading latest block should be accepted in + // on_peer_new_block in our code as well #[test] fn block_rlp_mutually_acceptable() { let mut client = TestBlockChainClient::new(); diff --git a/sync/src/tests/helpers.rs b/sync/src/tests/helpers.rs index d155fee6b..f70c4d1f4 100644 --- a/sync/src/tests/helpers.rs +++ b/sync/src/tests/helpers.rs @@ -229,11 +229,9 @@ impl BlockChainClient for TestBlockChainClient { fn queue_info(&self) -> BlockQueueInfo { BlockQueueInfo { - full: false, verified_queue_size: 0, unverified_queue_size: 0, verifying_queue_size: 0, - empty: false, } } From 3dd220b62fe1b43614de0e9c6820d53ed4ee6a4b Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Mon, 8 Feb 2016 03:14:48 -0800 Subject: [PATCH 23/30] refactoring of report functions, some comments --- ethcore/src/block_queue.rs | 16 ++++++++++------ ethcore/src/client.rs | 2 +- 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/ethcore/src/block_queue.rs b/ethcore/src/block_queue.rs index e4569a21b..ed3201ee2 100644 --- a/ethcore/src/block_queue.rs +++ b/ethcore/src/block_queue.rs @@ -30,16 +30,12 @@ use client::BlockStatus; /// Block queue status #[derive(Debug)] pub struct BlockQueueInfo { - /// Indicates that queue is full - pub full: bool, /// Number of queued blocks pending verification pub unverified_queue_size: usize, /// Number of verified queued blocks pending import pub verified_queue_size: usize, /// Number of blocks being verified pub verifying_queue_size: usize, - /// Indicates queue is empty - pub empty: bool } impl BlockQueueInfo { @@ -48,6 +44,16 @@ impl BlockQueueInfo { /// The size of the unverified and verifying queues. pub fn incomplete_queue_size(&self) -> usize { self.unverified_queue_size + self.verifying_queue_size } + + /// Indicates that queue is full + pub fn is_full(&self) -> bool { + self.unverified_queue_size + self.verified_queue_size + self.verifying_queue_size > MAX_UNVERIFIED_QUEUE_SIZE + } + + /// Indicates that queue is empty + pub fn is_empty(&self) -> bool { + self.unverified_queue_size + self.verified_queue_size + self.verifying_queue_size == 0 + } } /// A queue of blocks. Sits between network or other I/O and the BlockChain. @@ -311,11 +317,9 @@ impl BlockQueue { pub fn queue_info(&self) -> BlockQueueInfo { let verification = self.verification.lock().unwrap(); BlockQueueInfo { - full: verification.unverified.len() + verification.verifying.len() + verification.verified.len() >= MAX_UNVERIFIED_QUEUE_SIZE, verified_queue_size: verification.verified.len(), unverified_queue_size: verification.unverified.len(), verifying_queue_size: verification.verifying.len(), - empty: verification.verified.is_empty() && verification.unverified.is_empty() && verification.verifying.is_empty(), } } } diff --git a/ethcore/src/client.rs b/ethcore/src/client.rs index 3b5627504..cf43395e7 100644 --- a/ethcore/src/client.rs +++ b/ethcore/src/client.rs @@ -296,7 +296,7 @@ impl Client { trace!(target: "client", "Imported #{} ({})", header.number(), header.hash()); ret += 1; - if self.block_queue.read().unwrap().queue_info().empty { + if self.block_queue.read().unwrap().queue_info().is_empty() { io.send(NetworkIoMessage::User(SyncMessage::BlockVerified)).unwrap(); } } From 11103b083a3b8c30a65823ba42ddc2e44c8a3fb8 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Mon, 8 Feb 2016 03:35:51 -0800 Subject: [PATCH 24/30] fixed test --- ethcore/src/block_queue.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ethcore/src/block_queue.rs b/ethcore/src/block_queue.rs index ed3201ee2..fb735c973 100644 --- a/ethcore/src/block_queue.rs +++ b/ethcore/src/block_queue.rs @@ -409,6 +409,6 @@ mod tests { queue.flush(); queue.drain(1); - assert!(queue.queue_info().empty); + assert!(queue.queue_info().is_empty()); } } From b411a3d55dd0447c1cc90d5a60772275d9169e77 Mon Sep 17 00:00:00 2001 From: arkpar Date: Mon, 8 Feb 2016 15:03:44 +0100 Subject: [PATCH 25/30] Check for handshake expiration before attempting replace --- util/src/network/host.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/util/src/network/host.rs b/util/src/network/host.rs index 24c3460db..50cf294bc 100644 --- a/util/src/network/host.rs +++ b/util/src/network/host.rs @@ -599,6 +599,9 @@ impl Host where Message: Send + Sync + Clone { fn start_session(&self, token: StreamToken, io: &IoContext>) { let mut connections = self.connections.write().unwrap(); + if connections.get(token).is_none() { + return; // handshake expired + } connections.replace_with(token, |c| { match Arc::try_unwrap(c).ok().unwrap().into_inner().unwrap() { ConnectionEntry::Handshake(h) => { From f5b218ba89c0744acc0eae9c548f737d2d323cb3 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Mon, 8 Feb 2016 07:25:41 -0800 Subject: [PATCH 26/30] making local coverage identical to CI --- cov.sh | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/cov.sh b/cov.sh index eeb4d21f6..ddd1b58ea 100755 --- a/cov.sh +++ b/cov.sh @@ -15,7 +15,11 @@ if ! type kcov > /dev/null; then exit 1 fi -cargo test --features ethcore/json-tests -p ethcore --no-run || exit $? +cargo test --features ethcore/json-tests -p ethash -p ethcore-util -p ethcore -p ethsync -p ethcore-rpc -p parity --no-run || exit $? mkdir -p target/coverage -kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethcore* +kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethcore-* +kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethash-* +kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethcore_util-* +kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethsync-* +kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethcore_rpc-* xdg-open target/coverage/index.html From c60c7021700c6654d0154c490bb8a12d34b4a024 Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Mon, 8 Feb 2016 20:20:45 +0300 Subject: [PATCH 27/30] making it EXACTLY the same --- cov.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cov.sh b/cov.sh index ddd1b58ea..66aa8c762 100755 --- a/cov.sh +++ b/cov.sh @@ -15,7 +15,7 @@ if ! type kcov > /dev/null; then exit 1 fi -cargo test --features ethcore/json-tests -p ethash -p ethcore-util -p ethcore -p ethsync -p ethcore-rpc -p parity --no-run || exit $? +cargo test -p ethash -p ethcore-util -p ethcore -p ethsync -p ethcore-rpc -p parity --no-run || exit $? mkdir -p target/coverage kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethcore-* kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethash-* From 8fe5b4e2ba1cbffad997e89e9d0a8f9133d38b6f Mon Sep 17 00:00:00 2001 From: Nikolay Volf Date: Mon, 8 Feb 2016 21:12:12 +0300 Subject: [PATCH 28/30] removing tests from coverage --- cov.sh | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/cov.sh b/cov.sh index 66aa8c762..c63687acf 100755 --- a/cov.sh +++ b/cov.sh @@ -16,10 +16,11 @@ if ! type kcov > /dev/null; then fi cargo test -p ethash -p ethcore-util -p ethcore -p ethsync -p ethcore-rpc -p parity --no-run || exit $? +rm -rf target/coverage mkdir -p target/coverage -kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethcore-* -kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethash-* -kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethcore_util-* -kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethsync-* -kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethcore_rpc-* +kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1,src/tests --include-pattern src --verify target/coverage target/debug/deps/ethcore-* +kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1,src/tests --include-pattern src --verify target/coverage target/debug/deps/ethash-* +kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1,src/tests --include-pattern src --verify target/coverage target/debug/deps/ethcore_util-* +kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1,src/tests --include-pattern src --verify target/coverage target/debug/deps/ethsync-* +kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1,src/tests --include-pattern src --verify target/coverage target/debug/deps/ethcore_rpc-* xdg-open target/coverage/index.html From 3fed6a2f1c98a80b960a67983983db66b2a5b2ea Mon Sep 17 00:00:00 2001 From: KKudryavtsev Date: Mon, 8 Feb 2016 19:23:42 +0100 Subject: [PATCH 29/30] added gcc dependency to Dockerfiles --- docker/ubuntu-dev/Dockerfile | 1 + docker/ubuntu-jit/Dockerfile | 1 + docker/ubuntu/Dockerfile | 1 + 3 files changed, 3 insertions(+) diff --git a/docker/ubuntu-dev/Dockerfile b/docker/ubuntu-dev/Dockerfile index 299596a58..492f5d734 100644 --- a/docker/ubuntu-dev/Dockerfile +++ b/docker/ubuntu-dev/Dockerfile @@ -8,6 +8,7 @@ RUN apt-get update && \ # add-apt-repository software-properties-common \ curl \ + gcc \ wget \ git \ # evmjit dependencies diff --git a/docker/ubuntu-jit/Dockerfile b/docker/ubuntu-jit/Dockerfile index 6229c1524..27844548b 100644 --- a/docker/ubuntu-jit/Dockerfile +++ b/docker/ubuntu-jit/Dockerfile @@ -10,6 +10,7 @@ RUN apt-get update && \ curl \ wget \ git \ + gcc \ # evmjit dependencies zlib1g-dev \ libedit-dev diff --git a/docker/ubuntu/Dockerfile b/docker/ubuntu/Dockerfile index a6b05f38a..865f1f254 100644 --- a/docker/ubuntu/Dockerfile +++ b/docker/ubuntu/Dockerfile @@ -3,6 +3,7 @@ FROM ubuntu:14.04 # install tools and dependencies RUN apt-get update && \ apt-get install -y \ + gcc \ curl \ git \ # add-apt-repository From 6e89f5ef22b60d4f53593fbacfa8fad58acedba6 Mon Sep 17 00:00:00 2001 From: KKudryavtsev Date: Mon, 8 Feb 2016 19:58:19 +0100 Subject: [PATCH 30/30] tabified --- docker/ubuntu-dev/Dockerfile | 2 +- docker/ubuntu-jit/Dockerfile | 2 +- docker/ubuntu/Dockerfile | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/docker/ubuntu-dev/Dockerfile b/docker/ubuntu-dev/Dockerfile index 492f5d734..8b016e6fd 100644 --- a/docker/ubuntu-dev/Dockerfile +++ b/docker/ubuntu-dev/Dockerfile @@ -8,7 +8,7 @@ RUN apt-get update && \ # add-apt-repository software-properties-common \ curl \ - gcc \ + gcc \ wget \ git \ # evmjit dependencies diff --git a/docker/ubuntu-jit/Dockerfile b/docker/ubuntu-jit/Dockerfile index 27844548b..90ce531be 100644 --- a/docker/ubuntu-jit/Dockerfile +++ b/docker/ubuntu-jit/Dockerfile @@ -10,7 +10,7 @@ RUN apt-get update && \ curl \ wget \ git \ - gcc \ + gcc \ # evmjit dependencies zlib1g-dev \ libedit-dev diff --git a/docker/ubuntu/Dockerfile b/docker/ubuntu/Dockerfile index 865f1f254..812e66e9e 100644 --- a/docker/ubuntu/Dockerfile +++ b/docker/ubuntu/Dockerfile @@ -3,7 +3,7 @@ FROM ubuntu:14.04 # install tools and dependencies RUN apt-get update && \ apt-get install -y \ - gcc \ + gcc \ curl \ git \ # add-apt-repository