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

initial mocknet with chain notify

Browse Source
This commit is contained in:
keorn 2016-09-19 10:43:01 +02:00
parent fdcda41280
commit 49b8e144fd
5 changed files with 319 additions and 459 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

212
sync/src/tests/consensus.rs
View File

@ -15,211 +15,23 @@
// along with Parity. If not, see <http://www.gnu.org/licenses/>. // along with Parity. If not, see <http://www.gnu.org/licenses/>.
use util::*; use util::*;
use ethcore::client::{Client, BlockChainClient, BlockID, EachBlockWith}; use ethcore::spec::Spec;
use chain::{SyncState}; use super::mocknet::*;
use super::test_net::*; use std::thread::sleep;
use std::time::Duration;
use ethcore::client::BlockChainClient;
#[test] #[test]
fn two_peers() { fn issue_tx() {
::env_logger::init().ok(); ::env_logger::init().ok();
let mut net = TestNet::new(3); let mut net = MockNet::new_with_spec(2, vec!["1".sha3()], &Spec::new_test_round);
net.peer_mut(1).chain.add_blocks(1000, EachBlockWith::Uncle); net.peer(1).issue_rand_tx();
net.peer_mut(2).chain.add_blocks(1000, EachBlockWith::Uncle); sleep(Duration::from_secs(1));
net.sync(); net.sync();
assert!(net.peer(0).chain.block(BlockID::Number(1000)).is_some()); sleep(Duration::from_secs(1));
assert_eq!(*net.peer(0).chain.blocks.read(), *net.peer(1).chain.blocks.read());
}
#[test]
fn long_chain() {
::env_logger::init().ok();
let mut net = TestNet::new(2);
net.peer_mut(1).chain.add_blocks(50000, EachBlockWith::Nothing);
net.sync(); 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());
}
#[test]
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.sync(); net.sync();
let status = net.peer(0).sync.read().status();
assert_eq!(status.state, SyncState::Idle);
}
#[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);
let total_steps = net.sync();
assert!(total_steps < 20);
}
#[test]
fn empty_blocks() {
::env_logger::init().ok();
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.sync(); net.sync();
assert!(net.peer(0).chain.block(BlockID::Number(1000)).is_some()); println!("{:?}", net.peer(0).client.chain_info());
assert_eq!(*net.peer(0).chain.blocks.read(), *net.peer(1).chain.blocks.read()); println!("{:?}", net.peer(1).client.chain_info());
} }
#[test]
fn forked() {
::env_logger::init().ok();
let mut net = TestNet::new(3);
net.peer_mut(0).chain.add_blocks(300, EachBlockWith::Uncle);
net.peer_mut(1).chain.add_blocks(300, EachBlockWith::Uncle);
net.peer_mut(2).chain.add_blocks(300, EachBlockWith::Uncle);
net.peer_mut(0).chain.add_blocks(100, EachBlockWith::Nothing); //fork
net.peer_mut(1).chain.add_blocks(200, EachBlockWith::Uncle);
net.peer_mut(2).chain.add_blocks(200, EachBlockWith::Uncle);
net.peer_mut(1).chain.add_blocks(100, EachBlockWith::Uncle); //fork between 1 and 2
net.peer_mut(2).chain.add_blocks(10, EachBlockWith::Nothing);
// peer 1 has the best chain of 601 blocks
let peer1_chain = net.peer(1).chain.numbers.read().clone();
net.sync();
assert_eq!(*net.peer(0).chain.difficulty.read(), *net.peer(1).chain.difficulty.read());
assert_eq!(&*net.peer(0).chain.numbers.read(), &peer1_chain);
assert_eq!(&*net.peer(1).chain.numbers.read(), &peer1_chain);
assert_eq!(&*net.peer(2).chain.numbers.read(), &peer1_chain);
}
#[test]
fn net_hard_fork() {
::env_logger::init().ok();
let ref_client = TestBlockChainClient::new();
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.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.sync();
assert_eq!(net.peer(1).chain.chain_info().best_block_number, 0);
}
}
#[test]
fn restart() {
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.sync_steps(8);
// make sure that sync has actually happened
assert!(net.peer(0).chain.chain_info().best_block_number > 100);
net.restart_peer(0);
let status = net.peer(0).sync.read().status();
assert_eq!(status.state, SyncState::Idle);
}
#[test]
fn status_empty() {
let net = TestNet::new(2);
assert_eq!(net.peer(0).sync.read().status().state, SyncState::Idle);
}
#[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.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 propagate_hashes() {
let mut net = TestNet::new(6);
net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle);
net.sync();
net.peer_mut(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());
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 {
hashes += 1;
}
if net.peer(0).queue[i].packet_id == 0x7 {
blocks += 1;
}
}
assert_eq!(blocks, 4);
assert_eq!(hashes, 5);
}
#[test]
fn propagate_blocks() {
let mut net = TestNet::new(20);
net.peer_mut(1).chain.add_blocks(10, EachBlockWith::Uncle);
net.sync();
net.peer_mut(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());
// NEW_BLOCK_PACKET
let blocks = net.peer(0).queue.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.sync_steps(20);
assert_eq!(net.peer(0).chain.chain_info().best_block_number, 5);
}
#[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.sync_steps(20);
assert_eq!(net.peer(0).chain.chain_info().best_block_number, 5);
}
#[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.sync_steps(20);
assert_eq!(net.peer(0).chain.chain_info().best_block_number, 5);
}

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

@ -19,7 +19,6 @@ use network::*;
use tests::snapshot::*; use tests::snapshot::*;
use ethcore::client::{TestBlockChainClient, BlockChainClient}; use ethcore::client::{TestBlockChainClient, BlockChainClient};
use ethcore::header::BlockNumber; use ethcore::header::BlockNumber;
use ethcore::spec::Spec;
use ethcore::snapshot::SnapshotService; use ethcore::snapshot::SnapshotService;
use sync_io::SyncIo; use sync_io::SyncIo;
use chain::ChainSync; use chain::ChainSync;

306
sync/src/tests/mocknet.rs Normal file
View File

@ -0,0 +1,306 @@
// 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 network::*;
use tests::snapshot::*;
use ethcore::client::{Client, BlockChainClient, ChainNotify};
use ethcore::spec::Spec;
use ethcore::snapshot::SnapshotService;
use ethcore::transaction::{Transaction, SignedTransaction, Action};
use ethcore::account_provider::AccountProvider;
use ethkey::{Random, Generator};
use sync_io::SyncIo;
use chain::ChainSync;
use ::SyncConfig;
use devtools::RandomTempPath;
use ethcore::miner::Miner;
use ethcore::service::ClientService;
use std::time::Duration;
use std::thread::sleep;
pub struct TestIo<'p> {
pub client: Arc<Client>,
pub snapshot_service: &'p TestSnapshotService,
pub queue: &'p mut VecDeque<TestPacket>,
pub sender: Option<PeerId>,
}
impl<'p> TestIo<'p> {
pub fn new(client: Arc<Client>, ss: &'p TestSnapshotService, queue: &'p mut VecDeque<TestPacket>, sender: Option<PeerId>) -> TestIo<'p> {
TestIo {
client: client,
snapshot_service: ss,
queue: queue,
sender: sender
}
}
}
impl<'p> SyncIo for TestIo<'p> {
fn disable_peer(&mut self, _peer_id: PeerId) {
}
fn disconnect_peer(&mut self, _peer_id: PeerId) {
}
fn is_expired(&self) -> bool {
false
}
fn respond(&mut self, packet_id: PacketId, data: Vec<u8>) -> Result<(), NetworkError> {
self.queue.push_back(TestPacket {
data: data,
packet_id: packet_id,
recipient: self.sender.unwrap()
});
Ok(())
}
fn send(&mut self, peer_id: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), NetworkError> {
self.queue.push_back(TestPacket {
data: data,
packet_id: packet_id,
recipient: peer_id,
});
Ok(())
}
fn chain(&self) -> &BlockChainClient {
self.client.as_ref()
}
fn snapshot_service(&self) -> &SnapshotService {
self.snapshot_service
}
fn eth_protocol_version(&self, _peer: PeerId) -> u8 {
64
}
}
pub struct TestPacket {
pub data: Bytes,
pub packet_id: PacketId,
pub recipient: PeerId,
}
fn transaction() -> Transaction {
Transaction {
action: Action::Create,
value: U256::zero(),
data: "3331600055".from_hex().unwrap(),
gas: U256::from(100_000),
gas_price: U256::zero(),
nonce: U256::zero(),
}
}
pub fn random_transaction() -> SignedTransaction {
let keypair = Random.generate().unwrap();
transaction().sign(&keypair.secret())
}
pub struct MockPeer {
pub client: Arc<Client>,
pub snapshot_service: Arc<TestSnapshotService>,
pub sync: RwLock<ChainSync>,
pub queue: RwLock<VecDeque<TestPacket>>,
_service: ClientService,
_paths: Vec<RandomTempPath>,
}
impl ChainNotify for MockPeer {
fn new_blocks(&self,
imported: Vec<H256>,
invalid: Vec<H256>,
enacted: Vec<H256>,
retracted: Vec<H256>,
sealed: Vec<H256>,
_duration: u64) {
println!("New sync blocks");
let ref mut q = *self.queue.write();
let mut sync_io = TestIo::new(
self.client.clone(),
&self.snapshot_service,
q,
None);
self.sync.write().chain_new_blocks(
&mut sync_io,
&imported,
&invalid,
&enacted,
&retracted,
&sealed);
}
}
impl MockPeer {
pub fn new_with_spec<S>(get_spec: &S, author_secret: Option<Secret>) -> Arc<MockPeer> where S: Fn()->Spec {
let (accounts, address) = if let Some(secret) = author_secret {
let tap = AccountProvider::transient_provider();
let addr = tap.insert_account(secret, "").unwrap();
tap.unlock_account_permanently(addr, "".into()).unwrap();
(Some(Arc::new(tap)), Some(addr))
} else {
(None, None)
};
let client_path = RandomTempPath::new();
let snapshot_path = RandomTempPath::new();
let ipc_path = RandomTempPath::new();
let spec = get_spec();
let service = ClientService::start(
Default::default(),
&spec,
client_path.as_path(),
snapshot_path.as_path(),
ipc_path.as_path(),
Arc::new(Miner::with_spec_and_accounts(&spec, accounts.clone())),
).unwrap();
let client = service.client();
if let Some(addr) = address { client.set_author(addr) }
let sync = ChainSync::new(SyncConfig::default(), &*client);
let peer = Arc::new(MockPeer {
sync: RwLock::new(sync),
snapshot_service: Arc::new(TestSnapshotService::new()),
client: client,
queue: RwLock::new(VecDeque::new()),
_service: service,
_paths: vec![client_path, snapshot_path, ipc_path]
});
peer.client.add_notify(peer.clone());
peer
}
pub fn issue_tx(&self, transaction: SignedTransaction) {
self.client.import_own_transaction(transaction).unwrap();
}
pub fn issue_rand_tx(&self) {
self.issue_tx(random_transaction())
}
}
pub struct MockNet {
pub peers: Vec<Arc<MockPeer>>,
pub started: bool,
}
impl MockNet {
pub fn new_with_spec<S>(nodes: usize, author_secrets: Vec<H256>, get_spec: &S) -> MockNet where S: Fn()->Spec {
let mut net = MockNet {
peers: Vec::new(),
started: false,
};
for secret in author_secrets {
net.peers.push(MockPeer::new_with_spec(get_spec, Some(secret)));
}
for _ in net.peers.len()..nodes {
net.peers.push(MockPeer::new_with_spec(get_spec, None));
}
net
}
pub fn peer(&self, i: usize) -> Arc<MockPeer> {
self.peers.get(i).unwrap().clone()
}
pub fn peer_mut(&mut self, i: usize) -> &mut MockPeer {
Arc::get_mut(self.peers.get_mut(i).unwrap()).unwrap()
}
pub fn start(&mut self) {
for peer in 0..self.peers.len() {
for client in 0..self.peers.len() {
if peer != client {
let p = self.peers.get_mut(peer).unwrap();
let mut q = p.queue.write();
p.sync.write().on_peer_connected(&mut TestIo::new(p.client.clone(),
&p.snapshot_service,
&mut *q,
Some(client as PeerId)),
client as PeerId);
}
}
}
}
pub fn sync_step(&mut self) {
for (i, peer0) in self.peers.iter().enumerate() {
let mut q0 = peer0.queue.write();
if let Some(packet) = q0.pop_front() {
let p = self.peers.get(packet.recipient).unwrap();
let mut q1 = p.queue.write();
trace!(target: "mocknet", "--- {} -> {} ---", i, packet.recipient);
ChainSync::dispatch_packet(&p.sync,
&mut TestIo::new(p.client.clone(),
&p.snapshot_service,
&mut *q1,
Some(i as PeerId)),
i as PeerId,
packet.packet_id,
&packet.data);
trace!(target: "mocknet", "----------------");
}
let p = self.peers.get(i).unwrap();
peer0.client.flush_queue();
let mut io = TestIo::new(peer0.client.clone(), &peer0.snapshot_service, &mut *q0, None);
p.sync.write().maintain_sync(&mut io);
p.sync.write().propagate_new_transactions(&mut io);
sleep(Duration::from_secs(2));
}
}
pub fn sync_step_peer(&mut self, peer_num: usize) {
let mut peer = self.peer_mut(peer_num);
let ref mut q = *peer.queue.write();
peer.sync.write().maintain_sync(&mut TestIo::new(peer.client.clone(), &peer.snapshot_service, q, None));
}
pub fn restart_peer(&mut self, i: usize) {
let peer = self.peer_mut(i);
let ref mut q = *peer.queue.write();
peer.sync.write().restart(&mut TestIo::new(peer.client.clone(), &peer.snapshot_service, q, None));
}
pub fn sync(&mut self) -> u32 {
self.start();
let mut total_steps = 0;
while !self.done() {
self.sync_step();
total_steps += 1;
}
total_steps
}
pub fn sync_steps(&mut self, count: usize) {
if !self.started {
self.start();
self.started = true;
}
for _ in 0..count {
self.sync_step();
}
}
pub fn done(&self) -> bool {
self.peers.iter().all(|p| p.queue.try_read().unwrap().is_empty())
}
}

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

@ -16,7 +16,7 @@
pub mod helpers; pub mod helpers;
pub mod snapshot; pub mod snapshot;
pub mod test_net; pub mod mocknet;
mod chain; mod chain;
mod rpc; mod rpc;
mod consensus; mod consensus;

257
sync/src/tests/test_net.rs
View File

@ -1,257 +0,0 @@
// 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 network::*;
use tests::snapshot::*;
use ethcore::client::{Client, BlockChainClient};
use ethcore::tests::helpers::*;
use ethcore::header::BlockNumber;
use ethcore::spec::Spec;
use ethcore::snapshot::SnapshotService;
use ethcore::transaction::{Transaction, SignedTransaction, Action};
use ethkey::{Random, Generator};
use sync_io::SyncIo;
use chain::ChainSync;
use ::SyncConfig;
pub struct TestIo<'p> {
pub chain: &'p mut Client,
pub snapshot_service: &'p TestSnapshotService,
pub queue: &'p mut VecDeque<TestPacket>,
pub sender: Option<PeerId>,
}
impl<'p> TestIo<'p> {
pub fn new(chain: &'p mut Arc<Client>, ss: &'p TestSnapshotService, queue: &'p mut VecDeque<TestPacket>, sender: Option<PeerId>) -> TestIo<'p> {
TestIo {
chain: Arc::get_mut(chain).unwrap(),
snapshot_service: ss,
queue: queue,
sender: sender
}
}
}
impl<'p> SyncIo for TestIo<'p> {
fn disable_peer(&mut self, _peer_id: PeerId) {
}
fn disconnect_peer(&mut self, _peer_id: PeerId) {
}
fn is_expired(&self) -> bool {
false
}
fn respond(&mut self, packet_id: PacketId, data: Vec<u8>) -> Result<(), NetworkError> {
self.queue.push_back(TestPacket {
data: data,
packet_id: packet_id,
recipient: self.sender.unwrap()
});
Ok(())
}
fn send(&mut self, peer_id: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), NetworkError> {
self.queue.push_back(TestPacket {
data: data,
packet_id: packet_id,
recipient: peer_id,
});
Ok(())
}
fn chain(&self) -> &BlockChainClient {
self.chain
}
fn snapshot_service(&self) -> &SnapshotService {
self.snapshot_service
}
fn eth_protocol_version(&self, _peer: PeerId) -> u8 {
64
}
}
pub struct TestPacket {
pub data: Bytes,
pub packet_id: PacketId,
pub recipient: PeerId,
}
pub fn random_transaction() -> SignedTransaction {
let keypair = Random.generate().unwrap();
Transaction {
action: Action::Create,
value: U256::zero(),
data: "3331600055".from_hex().unwrap(),
gas: U256::from(100_000),
gas_price: U256::zero(),
nonce: U256::zero(),
}.sign(keypair.secret())
}
pub struct TestPeer {
pub chain: Arc<Client>,
pub snapshot_service: Arc<TestSnapshotService>,
pub sync: RwLock<ChainSync>,
pub queue: VecDeque<TestPacket>,
}
impl TestPeer {
pub fn issue_tx(&self, transaction: SignedTransaction) {
self.chain.import_own_transaction(transaction);
}
pub fn issue_rand_tx(&self) {
self.issue_tx(random_transaction())
}
}
pub struct TestNet {
pub peers: Vec<TestPeer>,
pub started: bool,
}
impl TestNet {
pub fn new(n: usize) -> TestNet {
Self::new_with_fork(n, None)
}
pub fn new_with_fork(n: usize, fork: Option<(BlockNumber, H256)>) -> TestNet {
let mut net = TestNet {
peers: Vec::new(),
started: false,
};
for _ in 0..n {
let mut chain = generate_dummy_client(1);
let mut config = SyncConfig::default();
config.fork_block = fork;
let ss = Arc::new(TestSnapshotService::new());
let sync = ChainSync::new(config, chain.reference().as_ref());
net.peers.push(TestPeer {
sync: RwLock::new(sync),
snapshot_service: ss,
chain: chain.take(),
queue: VecDeque::new(),
});
}
net
}
pub fn new_with_spec<S>(n: usize, get_spec: &S) -> TestNet where S: Fn()->Spec {
let mut net = TestNet {
peers: Vec::new(),
started: false,
};
for _ in 0..n {
let mut chain = generate_dummy_client_with_spec(get_spec, 1);
let sync = ChainSync::new(SyncConfig::default(), chain.reference().as_ref());
net.peers.push(TestPeer {
sync: RwLock::new(sync),
snapshot_service: Arc::new(TestSnapshotService::new()),
chain: chain.take(),
queue: VecDeque::new(),
});
}
net
}
pub fn peer(&self, i: usize) -> &TestPeer {
self.peers.get(i).unwrap()
}
pub fn peer_mut(&mut self, i: usize) -> &mut TestPeer {
self.peers.get_mut(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 = self.peers.get_mut(peer).unwrap();
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);
}
}
}
}
pub fn sync_step(&mut self) {
for peer in 0..self.peers.len() {
if let Some(packet) = self.peers[peer].queue.pop_front() {
let mut p = self.peers.get_mut(packet.recipient).unwrap();
trace!("--- {} -> {} ---", peer, packet.recipient);
ChainSync::dispatch_packet(&p.sync,
&mut TestIo::new(&mut p.chain,
&p.snapshot_service,
&mut p.queue,
Some(peer as PeerId)),
peer as PeerId,
packet.packet_id,
&packet.data);
trace!("----------------");
}
let mut p = self.peers.get_mut(peer).unwrap();
p.sync.write().maintain_sync(&mut TestIo::new(&mut p.chain, &p.snapshot_service, &mut p.queue, None));
}
}
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));
}
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));
}
pub fn sync(&mut self) -> u32 {
self.start();
let mut total_steps = 0;
while !self.done() {
self.sync_step();
total_steps += 1;
}
total_steps
}
pub fn sync_steps(&mut self, count: usize) {
if !self.started {
self.start();
self.started = true;
}
for _ in 0..count {
self.sync_step();
}
}
pub fn done(&self) -> bool {
self.peers.iter().all(|p| p.queue.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), &[], &[], &[], &[], &[]);
}
}