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

Merge branch 'master' into lightserv

Browse Source
This commit is contained in:
Robert Habermeier
2016-12-09 21:37:55 +01:00
parent 415fccfffb 2226324495
commit 503b126c93
88 changed files with 1606 additions and 736 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
sync/Cargo.toml
View File

@@ -27,6 +27,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

@@ -39,6 +39,9 @@ extern crate rlp;
extern crate ethcore_light as light;
#[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,6 +17,7 @@
pub mod helpers;
pub mod snapshot;
mod chain;
mod consensus;
#[cfg(feature = "ipc")]
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());