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

Merge pull request #3778 from ethcore/auth-round-test

Browse Source 
AuthorityRound network simulation test
This commit is contained in:
Gav Wood 2016-12-09 20:58:05 +01:00 committed by GitHub
commit 2226324495
15 changed files with 311 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.lock generated
View File

@ -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)",

3
ethcore/res/authority_round.json
View File

@ -4,7 +4,8 @@
"AuthorityRound": {
"params": {
"gasLimitBoundDivisor": "0x0400",
"stepDuration": "1",
"stepDuration": 1,
"startStep": 2,
"authorities" : [
"0x7d577a597b2742b498cb5cf0c26cdcd726d39e6e",
"0x82a978b3f5962a5b0957d9ee9eef472ee55b42f1"

10
ethcore/src/client/client.rs
View File

@ -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<F>(&self, f: F) where F: Fn(&ChainNotify) {
for np in self.notify.read().iter() {
if let Some(n) = np.upgrade() {
@ -563,6 +568,11 @@ impl Client {
results.len()
}
/// Get shared miner reference.
pub fn miner(&self) -> Arc<Miner> {
self.miner.clone()
}
/// Used by PoA to try sealing on period change.
pub fn update_sealing(&self) {
self.miner.update_sealing(self)

4
ethcore/src/client/test_client.rs
View File

@ -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));

27
ethcore/src/engines/authority_round.rs
View File

@ -49,6 +49,8 @@ pub struct AuthorityRoundParams {
pub authorities: Vec<Address>,
/// Number of authorities.
pub authority_n: usize,
/// Starting step,
pub start_step: Option<u64>,
}
impl From<ethjson::spec::AuthorityRoundParams> for AuthorityRoundParams {
@ -58,6 +60,7 @@ impl From<ethjson::spec::AuthorityRoundParams> 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::<Vec<_>>(),
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<Address, Builtin>) -> Result<Arc<Self>, 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<Address, Builtin> { &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<String, String> {
map![
@ -235,6 +242,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);
}
None
}

2
ethcore/src/engines/mod.rs
View File

@ -160,4 +160,6 @@ pub trait Engine : Sync + Send {
/// Add an account provider useful for Engines that sign stuff.
fn register_account_provider(&self, _account_provider: Arc<AccountProvider>) {}
/// Trigger next step of the consensus engine.
fn step(&self) {}
}

2
ethcore/src/spec/spec.rs
View File

@ -273,7 +273,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") }
}

7
json/src/spec/authority_round.rs
View File

@ -30,6 +30,10 @@ pub struct AuthorityRoundParams {
pub step_duration: Uint,
/// Valid authorities
pub authorities: Vec<Address>,
/// Starting step. Determined automatically if not specified.
/// To be used for testing only.
#[serde(rename="startStep")]
pub start_step: Option<Uint>,
}
/// Authority engine deserialization.
@ -50,7 +54,8 @@ mod tests {
"params": {
"gasLimitBoundDivisor": "0x0400",
"stepDuration": "0x02",
"authorities" : ["0xc6d9d2cd449a754c494264e1809c50e34d64562b"]
"authorities" : ["0xc6d9d2cd449a754c494264e1809c50e34d64562b"],
"startStep" : 24
}
}"#;

2
sync/Cargo.toml
View File

@ -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]

3
sync/src/lib.rs
View File

@ -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]

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

@ -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)]);
}

78
sync/src/tests/consensus.rs Normal file
View File

@ -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 <http://www.gnu.org/licenses/>.
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);
}

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

@ -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<TestPacket>,
pub sender: Option<PeerId>,
@ -34,8 +51,8 @@ pub struct TestIo<'p> {
overlay: RwLock<HashMap<BlockNumber, Bytes>>,
}
impl<'p> TestIo<'p> {
pub fn new(chain: &'p mut TestBlockChainClient, ss: &'p TestSnapshotService, queue: &'p mut VecDeque<TestPacket>, sender: Option<PeerId>) -> 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<TestPacket>, sender: Option<PeerId>) -> 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<HashMap<BlockNumber, Bytes>> {
@ -113,31 +130,31 @@ pub struct TestPacket {
pub recipient: PeerId,
}
pub struct TestPeer {
pub chain: TestBlockChainClient,
pub struct TestPeer<C> where C: FlushingBlockChainClient {
pub chain: C,
pub snapshot_service: Arc<TestSnapshotService>,
pub sync: RwLock<ChainSync>,
pub queue: VecDeque<TestPacket>,
pub queue: RwLock<VecDeque<TestPacket>>,
}
pub struct TestNet {
pub peers: Vec<TestPeer>,
pub struct TestNet<C> where C: FlushingBlockChainClient {
pub peers: Vec<Arc<TestPeer<C>>>,
pub started: bool,
pub disconnect_events: Vec<(PeerId, PeerId)>, //disconnected (initiated by, to)
}
impl TestNet {
pub fn new(n: usize) -> TestNet {
impl TestNet<TestBlockChainClient> {
pub fn new(n: usize) -> TestNet<TestBlockChainClient> {
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<TestBlockChainClient> {
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<TestBlockChainClient> {
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<EthcoreClient> {
pub fn new_with_spec<F>(n: usize, config: SyncConfig, spec_factory: F) -> GuardedTempResult<TestNet<EthcoreClient>>
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::<TestNet<EthcoreClient>> {
_temp: dir,
result: Some(net)
}
}
}
impl<C> TestNet<C> where C: FlushingBlockChainClient {
pub fn peer(&self, i: usize) -> &TestPeer<C> {
&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<C> {
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,38 @@ 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<EthcoreClient> {
fn new_blocks(&self,
imported: Vec<H256>,
invalid: Vec<H256>,
enacted: Vec<H256>,
retracted: Vec<H256>,
sealed: Vec<H256>,
_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);
}
fn start(&self) {}
fn stop(&self) {}
}

1
sync/src/tests/mod.rs
View File

@ -17,4 +17,5 @@
pub mod helpers;
pub mod snapshot;
mod chain;
mod consensus;
mod rpc;

2
sync/src/tests/snapshot.rs
View File

@ -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());