diff --git a/Cargo.lock b/Cargo.lock index f603410f0..bedcfebe4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -660,12 +660,14 @@ dependencies = [ "clippy 0.0.103 (registry+https://github.com/rust-lang/crates.io-index)", "env_logger 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)", "ethcore 1.5.0", + "ethcore-devtools 1.5.0", "ethcore-io 1.5.0", "ethcore-ipc 1.5.0", "ethcore-ipc-codegen 1.5.0", "ethcore-ipc-nano 1.5.0", "ethcore-network 1.5.0", "ethcore-util 1.5.0", + "ethkey 0.2.0", "heapsize 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)", "parking_lot 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)", diff --git a/ethcore/res/authority_round.json b/ethcore/res/authority_round.json index a0f88b85b..85beb51b4 100644 --- a/ethcore/res/authority_round.json +++ b/ethcore/res/authority_round.json @@ -4,7 +4,8 @@ "AuthorityRound": { "params": { "gasLimitBoundDivisor": "0x0400", - "stepDuration": "1", + "stepDuration": 1, + "startStep": 2, "authorities" : [ "0x7d577a597b2742b498cb5cf0c26cdcd726d39e6e", "0x82a978b3f5962a5b0957d9ee9eef472ee55b42f1" diff --git a/ethcore/src/client/client.rs b/ethcore/src/client/client.rs index 17d741b0f..bad09794a 100644 --- a/ethcore/src/client/client.rs +++ b/ethcore/src/client/client.rs @@ -255,6 +255,11 @@ impl Client { self.notify.write().push(Arc::downgrade(&target)); } + /// Returns engine reference. + pub fn engine(&self) -> &Engine { + &*self.engine + } + fn notify(&self, f: F) where F: Fn(&ChainNotify) { for np in self.notify.read().iter() { if let Some(n) = np.upgrade() { @@ -570,6 +575,11 @@ impl Client { results.len() } + /// Get shared miner reference. + pub fn miner(&self) -> Arc { + self.miner.clone() + } + /// Handle messages from the IO queue pub fn handle_queued_message(&self, message: &Bytes) { if let Err(e) = self.engine.handle_message(UntrustedRlp::new(message)) { diff --git a/ethcore/src/client/test_client.rs b/ethcore/src/client/test_client.rs index c8f8cdbf7..01a91fa66 100644 --- a/ethcore/src/client/test_client.rs +++ b/ethcore/src/client/test_client.rs @@ -255,7 +255,7 @@ impl TestBlockChainClient { } /// Make a bad block by setting invalid extra data. - pub fn corrupt_block(&mut self, n: BlockNumber) { + pub fn corrupt_block(&self, n: BlockNumber) { let hash = self.block_hash(BlockID::Number(n)).unwrap(); let mut header: BlockHeader = decode(&self.block_header(BlockID::Number(n)).unwrap()); header.set_extra_data(b"This extra data is way too long to be considered valid".to_vec()); @@ -267,7 +267,7 @@ impl TestBlockChainClient { } /// Make a bad block by setting invalid parent hash. - pub fn corrupt_block_parent(&mut self, n: BlockNumber) { + pub fn corrupt_block_parent(&self, n: BlockNumber) { let hash = self.block_hash(BlockID::Number(n)).unwrap(); let mut header: BlockHeader = decode(&self.block_header(BlockID::Number(n)).unwrap()); header.set_parent_hash(H256::from(42)); diff --git a/ethcore/src/engines/authority_round.rs b/ethcore/src/engines/authority_round.rs index 807e31c9a..7df000659 100644 --- a/ethcore/src/engines/authority_round.rs +++ b/ethcore/src/engines/authority_round.rs @@ -49,6 +49,8 @@ pub struct AuthorityRoundParams { pub authorities: Vec
, /// Number of authorities. pub authority_n: usize, + /// Starting step, + pub start_step: Option, } impl From for AuthorityRoundParams { @@ -58,6 +60,7 @@ impl From for AuthorityRoundParams { step_duration: Duration::from_secs(p.step_duration.into()), authority_n: p.authorities.len(), authorities: p.authorities.into_iter().map(Into::into).collect::>(), + start_step: p.start_step.map(Into::into), } } } @@ -97,7 +100,7 @@ impl AsMillis for Duration { impl AuthorityRound { /// Create a new instance of AuthorityRound engine. pub fn new(params: CommonParams, our_params: AuthorityRoundParams, builtins: BTreeMap) -> Result, Error> { - let initial_step = (unix_now().as_secs() / our_params.step_duration.as_secs()) as usize; + let initial_step = our_params.start_step.unwrap_or_else(|| (unix_now().as_secs() / our_params.step_duration.as_secs())) as usize; let engine = Arc::new( AuthorityRound { params: params, @@ -160,14 +163,7 @@ impl IoHandler<()> for TransitionHandler { fn timeout(&self, io: &IoContext<()>, timer: TimerToken) { if timer == ENGINE_TIMEOUT_TOKEN { if let Some(engine) = self.engine.upgrade() { - engine.step.fetch_add(1, AtomicOrdering::SeqCst); - engine.proposed.store(false, AtomicOrdering::SeqCst); - if let Some(ref channel) = *engine.message_channel.lock() { - match channel.send(ClientIoMessage::UpdateSealing) { - Ok(_) => trace!(target: "poa", "timeout: UpdateSealing message sent for step {}.", engine.step.load(AtomicOrdering::Relaxed)), - Err(err) => trace!(target: "poa", "timeout: Could not send a sealing message {} for step {}.", err, engine.step.load(AtomicOrdering::Relaxed)), - } - } + engine.step(); io.register_timer_once(ENGINE_TIMEOUT_TOKEN, engine.remaining_step_duration().as_millis()) .unwrap_or_else(|e| warn!(target: "poa", "Failed to restart consensus step timer: {}.", e)) } @@ -184,6 +180,17 @@ impl Engine for AuthorityRound { fn params(&self) -> &CommonParams { &self.params } fn builtins(&self) -> &BTreeMap { &self.builtins } + fn step(&self) { + self.step.fetch_add(1, AtomicOrdering::SeqCst); + self.proposed.store(false, AtomicOrdering::SeqCst); + if let Some(ref channel) = *self.message_channel.lock() { + match channel.send(ClientIoMessage::UpdateSealing) { + Ok(_) => trace!(target: "poa", "timeout: UpdateSealing message sent for step {}.", self.step.load(AtomicOrdering::Relaxed)), + Err(err) => trace!(target: "poa", "timeout: Could not send a sealing message {} for step {}.", err, self.step.load(AtomicOrdering::Relaxed)), + } + } + } + /// Additional engine-specific information for the user/developer concerning `header`. fn extra_info(&self, header: &Header) -> BTreeMap { map![ @@ -236,6 +243,8 @@ impl Engine for AuthorityRound { } else { warn!(target: "poa", "generate_seal: FAIL: Accounts not provided."); } + } else { + trace!(target: "poa", "generate_seal: Not a proposer for step {}.", step); } Seal::None } diff --git a/ethcore/src/engines/mod.rs b/ethcore/src/engines/mod.rs index db53b551d..7793d636d 100644 --- a/ethcore/src/engines/mod.rs +++ b/ethcore/src/engines/mod.rs @@ -215,4 +215,6 @@ pub trait Engine : Sync + Send { /// Add an account provider useful for Engines that sign stuff. fn register_account_provider(&self, _account_provider: Arc) {} + /// Trigger next step of the consensus engine. + fn step(&self) {} } diff --git a/ethcore/src/spec/spec.rs b/ethcore/src/spec/spec.rs index cd501cbdf..79314948f 100644 --- a/ethcore/src/spec/spec.rs +++ b/ethcore/src/spec/spec.rs @@ -274,7 +274,7 @@ impl Spec { pub fn new_instant() -> Spec { load_bundled!("instant_seal") } /// Create a new Spec with AuthorityRound consensus which does internal sealing (not requiring work). - /// Accounts with secrets "1".sha3() and "2".sha3() are the authorities. + /// Accounts with secrets "0".sha3() and "1".sha3() are the authorities. pub fn new_test_round() -> Self { load_bundled!("authority_round") } /// Create a new Spec with Tendermint consensus which does internal sealing (not requiring work). diff --git a/json/src/spec/authority_round.rs b/json/src/spec/authority_round.rs index 3d73ef1ef..bae17bb24 100644 --- a/json/src/spec/authority_round.rs +++ b/json/src/spec/authority_round.rs @@ -30,6 +30,10 @@ pub struct AuthorityRoundParams { pub step_duration: Uint, /// Valid authorities pub authorities: Vec
, + /// Starting step. Determined automatically if not specified. + /// To be used for testing only. + #[serde(rename="startStep")] + pub start_step: Option, } /// Authority engine deserialization. @@ -50,7 +54,8 @@ mod tests { "params": { "gasLimitBoundDivisor": "0x0400", "stepDuration": "0x02", - "authorities" : ["0xc6d9d2cd449a754c494264e1809c50e34d64562b"] + "authorities" : ["0xc6d9d2cd449a754c494264e1809c50e34d64562b"], + "startStep" : 24 } }"#; diff --git a/sync/Cargo.toml b/sync/Cargo.toml index 738f5f55c..d7980f0d9 100644 --- a/sync/Cargo.toml +++ b/sync/Cargo.toml @@ -26,6 +26,8 @@ heapsize = "0.3" ethcore-ipc = { path = "../ipc/rpc" } semver = "0.2" ethcore-ipc-nano = { path = "../ipc/nano" } +ethcore-devtools = { path = "../devtools" } +ethkey = { path = "../ethkey" } parking_lot = "0.3" [features] diff --git a/sync/src/lib.rs b/sync/src/lib.rs index 2061e4e3a..ced4c3f52 100644 --- a/sync/src/lib.rs +++ b/sync/src/lib.rs @@ -37,6 +37,9 @@ extern crate semver; extern crate parking_lot; extern crate rlp; +#[cfg(test)] extern crate ethcore_devtools as devtools; +#[cfg(test)] extern crate ethkey; + #[macro_use] extern crate log; #[macro_use] diff --git a/sync/src/tests/chain.rs b/sync/src/tests/chain.rs index 7705215f5..361d53e29 100644 --- a/sync/src/tests/chain.rs +++ b/sync/src/tests/chain.rs @@ -24,8 +24,8 @@ use SyncConfig; fn two_peers() { ::env_logger::init().ok(); let mut net = TestNet::new(3); - net.peer_mut(1).chain.add_blocks(1000, EachBlockWith::Uncle); - net.peer_mut(2).chain.add_blocks(1000, EachBlockWith::Uncle); + net.peer(1).chain.add_blocks(1000, EachBlockWith::Uncle); + net.peer(2).chain.add_blocks(1000, EachBlockWith::Uncle); net.sync(); assert!(net.peer(0).chain.block(BlockID::Number(1000)).is_some()); assert_eq!(*net.peer(0).chain.blocks.read(), *net.peer(1).chain.blocks.read()); @@ -35,7 +35,7 @@ fn two_peers() { fn long_chain() { ::env_logger::init().ok(); let mut net = TestNet::new(2); - net.peer_mut(1).chain.add_blocks(50000, EachBlockWith::Nothing); + net.peer(1).chain.add_blocks(50000, EachBlockWith::Nothing); net.sync(); assert!(net.peer(0).chain.block(BlockID::Number(50000)).is_some()); assert_eq!(*net.peer(0).chain.blocks.read(), *net.peer(1).chain.blocks.read()); @@ -45,8 +45,8 @@ fn long_chain() { fn status_after_sync() { ::env_logger::init().ok(); let mut net = TestNet::new(3); - net.peer_mut(1).chain.add_blocks(1000, EachBlockWith::Uncle); - net.peer_mut(2).chain.add_blocks(1000, EachBlockWith::Uncle); + net.peer(1).chain.add_blocks(1000, EachBlockWith::Uncle); + net.peer(2).chain.add_blocks(1000, EachBlockWith::Uncle); net.sync(); let status = net.peer(0).sync.read().status(); assert_eq!(status.state, SyncState::Idle); @@ -55,8 +55,8 @@ fn status_after_sync() { #[test] fn takes_few_steps() { let mut net = TestNet::new(3); - net.peer_mut(1).chain.add_blocks(100, EachBlockWith::Uncle); - net.peer_mut(2).chain.add_blocks(100, EachBlockWith::Uncle); + net.peer(1).chain.add_blocks(100, EachBlockWith::Uncle); + net.peer(2).chain.add_blocks(100, EachBlockWith::Uncle); let total_steps = net.sync(); assert!(total_steps < 20); } @@ -67,8 +67,8 @@ fn empty_blocks() { let mut net = TestNet::new(3); for n in 0..200 { let with = if n % 2 == 0 { EachBlockWith::Nothing } else { EachBlockWith::Uncle }; - net.peer_mut(1).chain.add_blocks(5, with.clone()); - net.peer_mut(2).chain.add_blocks(5, with); + net.peer(1).chain.add_blocks(5, with.clone()); + net.peer(2).chain.add_blocks(5, with); } net.sync(); assert!(net.peer(0).chain.block(BlockID::Number(1000)).is_some()); @@ -79,14 +79,14 @@ fn empty_blocks() { fn forked() { ::env_logger::init().ok(); let mut net = TestNet::new(3); - net.peer_mut(0).chain.add_blocks(30, EachBlockWith::Uncle); - net.peer_mut(1).chain.add_blocks(30, EachBlockWith::Uncle); - net.peer_mut(2).chain.add_blocks(30, EachBlockWith::Uncle); - net.peer_mut(0).chain.add_blocks(10, EachBlockWith::Nothing); //fork - net.peer_mut(1).chain.add_blocks(20, EachBlockWith::Uncle); - net.peer_mut(2).chain.add_blocks(20, EachBlockWith::Uncle); - net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle); //fork between 1 and 2 - net.peer_mut(2).chain.add_blocks(1, EachBlockWith::Nothing); + net.peer(0).chain.add_blocks(30, EachBlockWith::Uncle); + net.peer(1).chain.add_blocks(30, EachBlockWith::Uncle); + net.peer(2).chain.add_blocks(30, EachBlockWith::Uncle); + net.peer(0).chain.add_blocks(10, EachBlockWith::Nothing); //fork + net.peer(1).chain.add_blocks(20, EachBlockWith::Uncle); + net.peer(2).chain.add_blocks(20, EachBlockWith::Uncle); + net.peer(1).chain.add_blocks(10, EachBlockWith::Uncle); //fork between 1 and 2 + net.peer(2).chain.add_blocks(1, EachBlockWith::Nothing); // peer 1 has the best chain of 601 blocks let peer1_chain = net.peer(1).chain.numbers.read().clone(); net.sync(); @@ -102,12 +102,12 @@ fn forked_with_misbehaving_peer() { let mut net = TestNet::new(3); // peer 0 is on a totally different chain with higher total difficulty net.peer_mut(0).chain = TestBlockChainClient::new_with_extra_data(b"fork".to_vec()); - net.peer_mut(0).chain.add_blocks(50, EachBlockWith::Nothing); - net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Nothing); - net.peer_mut(2).chain.add_blocks(10, EachBlockWith::Nothing); + net.peer(0).chain.add_blocks(50, EachBlockWith::Nothing); + net.peer(1).chain.add_blocks(10, EachBlockWith::Nothing); + net.peer(2).chain.add_blocks(10, EachBlockWith::Nothing); - net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Nothing); - net.peer_mut(2).chain.add_blocks(20, EachBlockWith::Uncle); + net.peer(1).chain.add_blocks(10, EachBlockWith::Nothing); + net.peer(2).chain.add_blocks(20, EachBlockWith::Uncle); // peer 1 should sync to peer 2, others should not change let peer0_chain = net.peer(0).chain.numbers.read().clone(); let peer2_chain = net.peer(2).chain.numbers.read().clone(); @@ -124,13 +124,13 @@ fn net_hard_fork() { ref_client.add_blocks(50, EachBlockWith::Uncle); { let mut net = TestNet::new_with_fork(2, Some((50, ref_client.block_hash(BlockID::Number(50)).unwrap()))); - net.peer_mut(0).chain.add_blocks(100, EachBlockWith::Uncle); + net.peer(0).chain.add_blocks(100, EachBlockWith::Uncle); net.sync(); assert_eq!(net.peer(1).chain.chain_info().best_block_number, 100); } { let mut net = TestNet::new_with_fork(2, Some((50, ref_client.block_hash(BlockID::Number(50)).unwrap()))); - net.peer_mut(0).chain.add_blocks(100, EachBlockWith::Nothing); + net.peer(0).chain.add_blocks(100, EachBlockWith::Nothing); net.sync(); assert_eq!(net.peer(1).chain.chain_info().best_block_number, 0); } @@ -140,8 +140,8 @@ fn net_hard_fork() { fn restart() { ::env_logger::init().ok(); let mut net = TestNet::new(3); - net.peer_mut(1).chain.add_blocks(1000, EachBlockWith::Uncle); - net.peer_mut(2).chain.add_blocks(1000, EachBlockWith::Uncle); + net.peer(1).chain.add_blocks(1000, EachBlockWith::Uncle); + net.peer(2).chain.add_blocks(1000, EachBlockWith::Uncle); net.sync(); @@ -166,37 +166,37 @@ fn status_empty() { #[test] fn status_packet() { let mut net = TestNet::new(2); - net.peer_mut(0).chain.add_blocks(100, EachBlockWith::Uncle); - net.peer_mut(1).chain.add_blocks(1, EachBlockWith::Uncle); + net.peer(0).chain.add_blocks(100, EachBlockWith::Uncle); + net.peer(1).chain.add_blocks(1, EachBlockWith::Uncle); 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); + assert_eq!(1, net.peer(0).queue.read().len()); + assert_eq!(0x00, net.peer(0).queue.read()[0].packet_id); } #[test] fn propagate_hashes() { let mut net = TestNet::new(6); - net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle); + net.peer(1).chain.add_blocks(10, EachBlockWith::Uncle); net.sync(); - net.peer_mut(0).chain.add_blocks(10, EachBlockWith::Uncle); + net.peer(0).chain.add_blocks(10, EachBlockWith::Uncle); net.sync(); net.trigger_chain_new_blocks(0); //first event just sets the marker net.trigger_chain_new_blocks(0); // 5 peers with NewHahses, 4 with blocks - assert_eq!(9, net.peer(0).queue.len()); + assert_eq!(9, net.peer(0).queue.read().len()); let mut hashes = 0; let mut blocks = 0; - for i in 0..net.peer(0).queue.len() { - if net.peer(0).queue[i].packet_id == 0x1 { + for i in 0..net.peer(0).queue.read().len() { + if net.peer(0).queue.read()[i].packet_id == 0x1 { hashes += 1; } - if net.peer(0).queue[i].packet_id == 0x7 { + if net.peer(0).queue.read()[i].packet_id == 0x7 { blocks += 1; } } @@ -207,24 +207,24 @@ fn propagate_hashes() { #[test] fn propagate_blocks() { let mut net = TestNet::new(20); - net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle); + net.peer(1).chain.add_blocks(10, EachBlockWith::Uncle); net.sync(); - net.peer_mut(0).chain.add_blocks(10, EachBlockWith::Uncle); + net.peer(0).chain.add_blocks(10, EachBlockWith::Uncle); net.trigger_chain_new_blocks(0); //first event just sets the marker net.trigger_chain_new_blocks(0); - assert!(!net.peer(0).queue.is_empty()); + assert!(!net.peer(0).queue.read().is_empty()); // NEW_BLOCK_PACKET - let blocks = net.peer(0).queue.iter().filter(|p| p.packet_id == 0x7).count(); + let blocks = net.peer(0).queue.read().iter().filter(|p| p.packet_id == 0x7).count(); assert!(blocks > 0); } #[test] fn restart_on_malformed_block() { let mut net = TestNet::new(2); - net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle); - net.peer_mut(1).chain.corrupt_block(6); + net.peer(1).chain.add_blocks(10, EachBlockWith::Uncle); + net.peer(1).chain.corrupt_block(6); net.sync_steps(20); assert_eq!(net.peer(0).chain.chain_info().best_block_number, 5); @@ -233,8 +233,8 @@ fn restart_on_malformed_block() { #[test] fn restart_on_broken_chain() { let mut net = TestNet::new(2); - net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle); - net.peer_mut(1).chain.corrupt_block_parent(6); + net.peer(1).chain.add_blocks(10, EachBlockWith::Uncle); + net.peer(1).chain.corrupt_block_parent(6); net.sync_steps(20); assert_eq!(net.peer(0).chain.chain_info().best_block_number, 5); @@ -243,8 +243,8 @@ fn restart_on_broken_chain() { #[test] fn high_td_attach() { let mut net = TestNet::new(2); - net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle); - net.peer_mut(1).chain.corrupt_block_parent(6); + net.peer(1).chain.add_blocks(10, EachBlockWith::Uncle); + net.peer(1).chain.corrupt_block_parent(6); net.sync_steps(20); assert_eq!(net.peer(0).chain.chain_info().best_block_number, 5); @@ -255,8 +255,8 @@ fn high_td_attach() { fn disconnect_on_unrelated_chain() { ::env_logger::init().ok(); let mut net = TestNet::new(2); - net.peer_mut(0).chain.add_blocks(200, EachBlockWith::Uncle); - net.peer_mut(1).chain.add_blocks(100, EachBlockWith::Nothing); + net.peer(0).chain.add_blocks(200, EachBlockWith::Uncle); + net.peer(1).chain.add_blocks(100, EachBlockWith::Nothing); net.sync(); assert_eq!(net.disconnect_events, vec![(0, 0)]); } diff --git a/sync/src/tests/consensus.rs b/sync/src/tests/consensus.rs new file mode 100644 index 000000000..00a036a54 --- /dev/null +++ b/sync/src/tests/consensus.rs @@ -0,0 +1,78 @@ +// Copyright 2015, 2016 Ethcore (UK) Ltd. +// This file is part of Parity. + +// Parity is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// Parity is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with Parity. If not, see . + +use util::*; +use ethcore::client::BlockChainClient; +use ethcore::spec::Spec; +use ethcore::miner::MinerService; +use ethcore::transaction::*; +use ethcore::account_provider::AccountProvider; +use ethkey::KeyPair; +use super::helpers::*; +use SyncConfig; + +#[test] +fn test_authority_round() { + ::env_logger::init().ok(); + + let s1 = KeyPair::from_secret("1".sha3()).unwrap(); + let s2 = KeyPair::from_secret("0".sha3()).unwrap(); + let spec_factory = || { + let spec = Spec::new_test_round(); + let account_provider = AccountProvider::transient_provider(); + account_provider.insert_account(s1.secret().clone(), "").unwrap(); + account_provider.insert_account(s2.secret().clone(), "").unwrap(); + spec.engine.register_account_provider(Arc::new(account_provider)); + spec + }; + let mut net = TestNet::new_with_spec(2, SyncConfig::default(), spec_factory); + let mut net = &mut *net; + // Push transaction to both clients. Only one of them gets lucky to mine a block. + net.peer(0).chain.miner().set_author(s1.address()); + net.peer(0).chain.engine().set_signer(s1.address(), "".to_owned()); + net.peer(1).chain.miner().set_author(s2.address()); + net.peer(1).chain.engine().set_signer(s2.address(), "".to_owned()); + let tx1 = Transaction { + nonce: 0.into(), + gas_price: 0.into(), + gas: 21000.into(), + action: Action::Call(Address::default()), + value: 0.into(), + data: Vec::new(), + }.sign(s1.secret(), None); + // exhange statuses + net.sync_steps(5); + net.peer(0).chain.miner().import_own_transaction(&net.peer(0).chain, tx1).unwrap(); + net.sync(); + assert_eq!(net.peer(0).chain.chain_info().best_block_number, 1); + assert_eq!(net.peer(1).chain.chain_info().best_block_number, 1); + + let tx2 = Transaction { + nonce: 0.into(), + gas_price: 0.into(), + gas: 21000.into(), + action: Action::Call(Address::default()), + value: 0.into(), + data: Vec::new(), + }.sign(s2.secret(), None); + net.peer(1).chain.miner().import_own_transaction(&net.peer(1).chain, tx2).unwrap(); + net.peer(1).chain.engine().step(); + net.peer(1).chain.miner().update_sealing(&net.peer(1).chain); + net.sync(); + assert_eq!(net.peer(0).chain.chain_info().best_block_number, 2); + assert_eq!(net.peer(1).chain.chain_info().best_block_number, 2); +} + diff --git a/sync/src/tests/helpers.rs b/sync/src/tests/helpers.rs index f3254fbad..3211f27af 100644 --- a/sync/src/tests/helpers.rs +++ b/sync/src/tests/helpers.rs @@ -17,16 +17,33 @@ use util::*; use network::*; use tests::snapshot::*; -use ethcore::client::{TestBlockChainClient, BlockChainClient}; +use ethcore::client::{TestBlockChainClient, BlockChainClient, Client as EthcoreClient, ClientConfig, ChainNotify}; use ethcore::header::BlockNumber; use ethcore::snapshot::SnapshotService; +use ethcore::spec::Spec; +use ethcore::miner::Miner; +use ethcore::db::NUM_COLUMNS; use sync_io::SyncIo; +use io::IoChannel; use api::WARP_SYNC_PROTOCOL_ID; use chain::ChainSync; use ::SyncConfig; +use devtools::{self, GuardedTempResult}; -pub struct TestIo<'p> { - pub chain: &'p mut TestBlockChainClient, +pub trait FlushingBlockChainClient: BlockChainClient { + fn flush(&self) {} +} + +impl FlushingBlockChainClient for EthcoreClient { + fn flush(&self) { + self.flush_queue(); + } +} + +impl FlushingBlockChainClient for TestBlockChainClient {} + +pub struct TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p { + pub chain: &'p C, pub snapshot_service: &'p TestSnapshotService, pub queue: &'p mut VecDeque, pub sender: Option, @@ -34,8 +51,8 @@ pub struct TestIo<'p> { overlay: RwLock>, } -impl<'p> TestIo<'p> { - pub fn new(chain: &'p mut TestBlockChainClient, ss: &'p TestSnapshotService, queue: &'p mut VecDeque, sender: Option) -> TestIo<'p> { +impl<'p, C> TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p { + pub fn new(chain: &'p C, ss: &'p TestSnapshotService, queue: &'p mut VecDeque, sender: Option) -> TestIo<'p, C> { TestIo { chain: chain, snapshot_service: ss, @@ -47,7 +64,7 @@ impl<'p> TestIo<'p> { } } -impl<'p> SyncIo for TestIo<'p> { +impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p { fn disable_peer(&mut self, peer_id: PeerId) { self.disconnect_peer(peer_id); } @@ -99,7 +116,7 @@ impl<'p> SyncIo for TestIo<'p> { } fn protocol_version(&self, protocol: &ProtocolId, peer_id: PeerId) -> u8 { - if protocol == &WARP_SYNC_PROTOCOL_ID { 1 } else { self.eth_protocol_version(peer_id) } + if protocol == &WARP_SYNC_PROTOCOL_ID { 2 } else { self.eth_protocol_version(peer_id) } } fn chain_overlay(&self) -> &RwLock> { @@ -113,31 +130,31 @@ pub struct TestPacket { pub recipient: PeerId, } -pub struct TestPeer { - pub chain: TestBlockChainClient, +pub struct TestPeer where C: FlushingBlockChainClient { + pub chain: C, pub snapshot_service: Arc, pub sync: RwLock, - pub queue: VecDeque, + pub queue: RwLock>, } -pub struct TestNet { - pub peers: Vec, +pub struct TestNet where C: FlushingBlockChainClient { + pub peers: Vec>>, pub started: bool, pub disconnect_events: Vec<(PeerId, PeerId)>, //disconnected (initiated by, to) } -impl TestNet { - pub fn new(n: usize) -> TestNet { +impl TestNet { + pub fn new(n: usize) -> TestNet { Self::new_with_config(n, SyncConfig::default()) } - pub fn new_with_fork(n: usize, fork: Option<(BlockNumber, H256)>) -> TestNet { + pub fn new_with_fork(n: usize, fork: Option<(BlockNumber, H256)>) -> TestNet { let mut config = SyncConfig::default(); config.fork_block = fork; Self::new_with_config(n, config) } - pub fn new_with_config(n: usize, config: SyncConfig) -> TestNet { + pub fn new_with_config(n: usize, config: SyncConfig) -> TestNet { let mut net = TestNet { peers: Vec::new(), started: false, @@ -147,31 +164,77 @@ impl TestNet { let chain = TestBlockChainClient::new(); let ss = Arc::new(TestSnapshotService::new()); let sync = ChainSync::new(config.clone(), &chain); - net.peers.push(TestPeer { + net.peers.push(Arc::new(TestPeer { sync: RwLock::new(sync), snapshot_service: ss, chain: chain, - queue: VecDeque::new(), - }); + queue: RwLock::new(VecDeque::new()), + })); } net } +} - pub fn peer(&self, i: usize) -> &TestPeer { +impl TestNet { + pub fn new_with_spec(n: usize, config: SyncConfig, spec_factory: F) -> GuardedTempResult> + where F: Fn() -> Spec + { + let mut net = TestNet { + peers: Vec::new(), + started: false, + disconnect_events: Vec::new(), + }; + let dir = devtools::RandomTempPath::new(); + for _ in 0..n { + let mut client_dir = dir.as_path().clone(); + client_dir.push(devtools::random_filename()); + + let db_config = DatabaseConfig::with_columns(NUM_COLUMNS); + + let spec = spec_factory(); + let client = Arc::try_unwrap(EthcoreClient::new( + ClientConfig::default(), + &spec, + client_dir.as_path(), + Arc::new(Miner::with_spec(&spec)), + IoChannel::disconnected(), + &db_config + ).unwrap()).ok().unwrap(); + + let ss = Arc::new(TestSnapshotService::new()); + let sync = ChainSync::new(config.clone(), &client); + let peer = Arc::new(TestPeer { + sync: RwLock::new(sync), + snapshot_service: ss, + chain: client, + queue: RwLock::new(VecDeque::new()), + }); + peer.chain.add_notify(peer.clone()); + net.peers.push(peer); + } + GuardedTempResult::> { + _temp: dir, + result: Some(net) + } + } +} + +impl TestNet where C: FlushingBlockChainClient { + pub fn peer(&self, i: usize) -> &TestPeer { &self.peers[i] } - pub fn peer_mut(&mut self, i: usize) -> &mut TestPeer { - &mut self.peers[i] + pub fn peer_mut(&mut self, i: usize) -> &mut TestPeer { + Arc::get_mut(&mut self.peers[i]).unwrap() } pub fn start(&mut self) { for peer in 0..self.peers.len() { for client in 0..self.peers.len() { if peer != client { - let mut p = &mut self.peers[peer]; + let p = &self.peers[peer]; p.sync.write().update_targets(&p.chain); - p.sync.write().on_peer_connected(&mut TestIo::new(&mut p.chain, &p.snapshot_service, &mut p.queue, Some(client as PeerId)), client as PeerId); + p.sync.write().on_peer_connected(&mut TestIo::new(&p.chain, &p.snapshot_service, &mut p.queue.write(), Some(client as PeerId)), client as PeerId); } } } @@ -179,18 +242,20 @@ impl TestNet { pub fn sync_step(&mut self) { for peer in 0..self.peers.len() { - if let Some(packet) = self.peers[peer].queue.pop_front() { + let packet = self.peers[peer].queue.write().pop_front(); + if let Some(packet) = packet { let disconnecting = { - let mut p = &mut self.peers[packet.recipient]; + let p = &self.peers[packet.recipient]; + let mut queue = p.queue.write(); trace!("--- {} -> {} ---", peer, packet.recipient); let to_disconnect = { - let mut io = TestIo::new(&mut p.chain, &p.snapshot_service, &mut p.queue, Some(peer as PeerId)); + let mut io = TestIo::new(&p.chain, &p.snapshot_service, &mut queue, Some(peer as PeerId)); ChainSync::dispatch_packet(&p.sync, &mut io, peer as PeerId, packet.packet_id, &packet.data); io.to_disconnect }; for d in &to_disconnect { // notify this that disconnecting peers are disconnecting - let mut io = TestIo::new(&mut p.chain, &p.snapshot_service, &mut p.queue, Some(*d)); + let mut io = TestIo::new(&p.chain, &p.snapshot_service, &mut queue, Some(*d)); p.sync.write().on_peer_aborting(&mut io, *d); self.disconnect_events.push((peer, *d)); } @@ -198,8 +263,9 @@ impl TestNet { }; for d in &disconnecting { // notify other peers that this peer is disconnecting - let mut p = &mut self.peers[*d]; - let mut io = TestIo::new(&mut p.chain, &p.snapshot_service, &mut p.queue, Some(peer as PeerId)); + let p = &self.peers[*d]; + let mut queue = p.queue.write(); + let mut io = TestIo::new(&p.chain, &p.snapshot_service, &mut queue, Some(peer as PeerId)); p.sync.write().on_peer_aborting(&mut io, peer as PeerId); } } @@ -209,13 +275,17 @@ impl TestNet { } pub fn sync_step_peer(&mut self, peer_num: usize) { - let mut peer = self.peer_mut(peer_num); - peer.sync.write().maintain_sync(&mut TestIo::new(&mut peer.chain, &peer.snapshot_service, &mut peer.queue, None)); + let peer = self.peer(peer_num); + peer.chain.flush(); + let mut queue = peer.queue.write(); + peer.sync.write().maintain_peers(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None)); + peer.sync.write().maintain_sync(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None)); + peer.sync.write().propagate_new_transactions(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None)); } pub fn restart_peer(&mut self, i: usize) { - let peer = self.peer_mut(i); - peer.sync.write().restart(&mut TestIo::new(&mut peer.chain, &peer.snapshot_service, &mut peer.queue, None)); + let peer = self.peer(i); + peer.sync.write().restart(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut peer.queue.write(), None)); } pub fn sync(&mut self) -> u32 { @@ -239,11 +309,46 @@ impl TestNet { } pub fn done(&self) -> bool { - self.peers.iter().all(|p| p.queue.is_empty()) + self.peers.iter().all(|p| p.queue.read().is_empty()) } pub fn trigger_chain_new_blocks(&mut self, peer_id: usize) { - let mut peer = self.peer_mut(peer_id); - peer.sync.write().chain_new_blocks(&mut TestIo::new(&mut peer.chain, &peer.snapshot_service, &mut peer.queue, None), &[], &[], &[], &[], &[], &[]); + let peer = self.peer(peer_id); + let mut queue = peer.queue.write(); + peer.sync.write().chain_new_blocks(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None), &[], &[], &[], &[], &[], &[]); } } + +impl ChainNotify for TestPeer { + fn new_blocks(&self, + imported: Vec, + invalid: Vec, + enacted: Vec, + retracted: Vec, + sealed: Vec, + proposed: Vec, + _duration: u64) + { + let mut queue = self.queue.write(); + let mut io = TestIo::new(&self.chain, &self.snapshot_service, &mut queue, None); + self.sync.write().chain_new_blocks( + &mut io, + &imported, + &invalid, + &enacted, + &retracted, + &sealed, + &proposed); + } + + fn start(&self) {} + + fn stop(&self) {} + + fn broadcast(&self, message: Vec) { + let mut queue = self.queue.write(); + let mut io = TestIo::new(&self.chain, &self.snapshot_service, &mut queue, None); + self.sync.write().propagate_consensus_packet(&mut io, message.clone()); + } +} + diff --git a/sync/src/tests/mod.rs b/sync/src/tests/mod.rs index bdb4ae4f9..f805f6c24 100644 --- a/sync/src/tests/mod.rs +++ b/sync/src/tests/mod.rs @@ -17,4 +17,5 @@ pub mod helpers; pub mod snapshot; mod chain; +mod consensus; mod rpc; diff --git a/sync/src/tests/snapshot.rs b/sync/src/tests/snapshot.rs index 5d0b21b47..283d59ee3 100644 --- a/sync/src/tests/snapshot.rs +++ b/sync/src/tests/snapshot.rs @@ -129,7 +129,7 @@ fn snapshot_sync() { let snapshot_service = Arc::new(TestSnapshotService::new_with_snapshot(16, H256::new(), 500000)); for i in 0..4 { net.peer_mut(i).snapshot_service = snapshot_service.clone(); - net.peer_mut(i).chain.add_blocks(1, EachBlockWith::Nothing); + net.peer(i).chain.add_blocks(1, EachBlockWith::Nothing); } net.sync_steps(50); assert_eq!(net.peer(4).snapshot_service.state_restoration_chunks.lock().len(), net.peer(0).snapshot_service.manifest.as_ref().unwrap().state_hashes.len());