diff --git a/cov.sh b/cov.sh index eeb4d21f6..c63687acf 100755 --- a/cov.sh +++ b/cov.sh @@ -15,7 +15,12 @@ if ! type kcov > /dev/null; then exit 1 fi -cargo test --features ethcore/json-tests -p ethcore --no-run || exit $? +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,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 diff --git a/docker/ubuntu-dev/Dockerfile b/docker/ubuntu-dev/Dockerfile index 299596a58..8b016e6fd 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..90ce531be 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..812e66e9e 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 diff --git a/ethcore/src/block_queue.rs b/ethcore/src/block_queue.rs index 808982b99..fb735c973 100644 --- a/ethcore/src/block_queue.rs +++ b/ethcore/src/block_queue.rs @@ -30,8 +30,6 @@ 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 @@ -46,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. @@ -287,7 +295,6 @@ impl BlockQueue { for h in hashes { processing.remove(&h); } - //TODO: reward peers } /// Removes up to `max` verified blocks from the queue @@ -310,7 +317,6 @@ 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(), @@ -395,4 +401,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().is_empty()); + } } diff --git a/ethcore/src/client.rs b/ethcore/src/client.rs index d6b91acc9..46b53e7b9 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::*; @@ -220,7 +220,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(); @@ -292,6 +292,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().is_empty() { + io.send(NetworkIoMessage::User(SyncMessage::BlockVerified)).unwrap(); + } } self.block_queue.write().unwrap().mark_as_good(&good_blocks); ret 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 3d5ff91e9..63dc47024 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -62,6 +62,9 @@ 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 MAX_PEER_LAG_PROPAGATION: BlockNumber = 20; const STATUS_PACKET: u8 = 0x00; const NEW_BLOCK_HASHES_PACKET: u8 = 0x01; @@ -134,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, @@ -142,6 +145,7 @@ enum PeerAsking { BlockBodies, } +#[derive(Clone)] /// Syncing peer information struct PeerInfo { /// eth protocol version @@ -430,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); @@ -468,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); } @@ -593,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; } @@ -636,6 +639,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 { @@ -1030,10 +1034,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 { @@ -1042,13 +1042,125 @@ impl ChainSync { } } } - /// Maintain other peers. Send out any new blocks and transactions - pub fn maintain_sync(&mut self, io: &mut SyncIo) { - if !io.chain().queue_info().full && self.state == SyncState::Waiting { + + fn check_resume(&mut self, io: &mut SyncIo) { + if !io.chain().queue_info().is_full() && self.state == SyncState::Waiting { self.state = SyncState::Idle; self.continue_sync(io); } } + + /// 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) => { + match route.blocks.len() { + 0 => None, + _ => { + 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()) + } + } + }, + None => None + } + } + + /// creates latest block rlp for the given client + fn create_latest_block_rlp(chain: &BlockChainClient) -> Bytes { + 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() + } + + /// 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(); + let latest_hash = chain_info.best_block_hash; + let latest_number = chain_info.best_block_number; + self.peers.iter().filter(|&(_, peer_info)| + match io.chain().block_status(&peer_info.latest) + { + 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_id) + .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); + + // 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 lucky_peers = match lagging_peers.len() { + 0 ... MIN_PEERS_PROPAGATION => lagging_peers, + _ => 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::>() + }; + + 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 + } + + /// 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; + 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) { + 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 + }, + 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); + + 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 + 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); + } } #[cfg(test)] @@ -1056,6 +1168,48 @@ mod tests { use tests::helpers::*; use super::*; use util::*; + use super::{PeerInfo, PeerAsking}; + 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() + } + + 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() { @@ -1124,4 +1278,204 @@ 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-difficulty + assert_eq!(107, rlp.len()); + } + + #[test] + 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(); + 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_blocks(&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_PACKET + assert_eq!(0x07, io.queue[0].packet_id); + } + + #[test] + 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(); + 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()); + } + + #[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()); + } + + // 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(); + 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()); + } + + // 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(); + 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/lib.rs b/sync/src/lib.rs index f8ccf0635..b2d1fc29f 100644 --- a/sync/src/lib.rs +++ b/sync/src/lib.rs @@ -50,6 +50,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::*; @@ -125,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 21922851c..6526d8500 100644 --- a/sync/src/tests/chain.rs +++ b/sync/src/tests/chain.rs @@ -104,4 +104,48 @@ 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_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(); + + net.peer_mut(0).chain.add_blocks(10, false); + 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(2); + 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 a5a60d62d..f70c4d1f4 100644 --- a/sync/src/tests/helpers.rs +++ b/sync/src/tests/helpers.rs @@ -69,16 +69,21 @@ 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 { fn block_total_difficulty(&self, _h: &H256) -> Option { - unimplemented!(); + Some(U256::zero()) } 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 { @@ -125,11 +130,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 } + } }) } @@ -202,7 +229,6 @@ impl BlockChainClient for TestBlockChainClient { fn queue_info(&self) -> BlockQueueInfo { BlockQueueInfo { - full: false, verified_queue_size: 0, unverified_queue_size: 0, verifying_queue_size: 0, @@ -334,6 +360,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)); @@ -362,4 +393,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)); + } } 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) => {