Merge pull request #352 from ethcore/sync-tests-i3

test efforts, receipt requests
This commit is contained in:
Gav Wood 2016-02-05 13:04:22 +01:00
commit c3850ff880
6 changed files with 153 additions and 41 deletions

View File

@ -83,6 +83,7 @@ pub mod header;
pub mod service; pub mod service;
pub mod spec; pub mod spec;
pub mod views; pub mod views;
pub mod receipt;
mod common; mod common;
mod basic_types; mod basic_types;
@ -98,7 +99,6 @@ mod state;
mod account; mod account;
mod action_params; mod action_params;
mod transaction; mod transaction;
mod receipt;
mod null_engine; mod null_engine;
mod builtin; mod builtin;
mod extras; mod extras;

View File

@ -1,3 +1,5 @@
//! Receipt
use util::*; use util::*;
use basic_types::LogBloom; use basic_types::LogBloom;
use log_entry::LogEntry; use log_entry::LogEntry;

View File

@ -5,5 +5,5 @@ fi
cargo test --no-run || exit $? cargo test --no-run || exit $?
mkdir -p target/coverage mkdir -p target/coverage
kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern sync/src --verify target/coverage target/debug/ethsync* kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1,sync/src/tests --include-pattern sync/src --verify target/coverage target/debug/ethsync*
xdg-open target/coverage/index.html xdg-open target/coverage/index.html

View File

@ -177,6 +177,7 @@ pub struct ChainSync {
have_common_block: bool, have_common_block: bool,
} }
type RlpResponseResult = Result<Option<(PacketId, RlpStream)>, PacketDecodeError>;
impl ChainSync { impl ChainSync {
/// Create a new instance of syncing strategy. /// Create a new instance of syncing strategy.
@ -845,7 +846,7 @@ impl ChainSync {
} }
/// Respond to GetBlockHeaders request /// Respond to GetBlockHeaders request
fn return_block_headers(&self, io: &mut SyncIo, r: &UntrustedRlp) -> Result<(), PacketDecodeError> { fn return_block_headers(io: &SyncIo, r: &UntrustedRlp) -> RlpResponseResult {
// Packet layout: // Packet layout:
// [ block: { P , B_32 }, maxHeaders: P, skip: P, reverse: P in { 0 , 1 } ] // [ block: { P , B_32 }, maxHeaders: P, skip: P, reverse: P in { 0 , 1 } ]
let max_headers: usize = try!(r.val_at(1)); let max_headers: usize = try!(r.val_at(1));
@ -892,18 +893,16 @@ impl ChainSync {
} }
let mut rlp = RlpStream::new_list(count as usize); let mut rlp = RlpStream::new_list(count as usize);
rlp.append_raw(&data, count as usize); rlp.append_raw(&data, count as usize);
io.respond(BLOCK_HEADERS_PACKET, rlp.out()).unwrap_or_else(|e|
debug!(target: "sync", "Error sending headers: {:?}", e));
trace!(target: "sync", "-> GetBlockHeaders: returned {} entries", count); trace!(target: "sync", "-> GetBlockHeaders: returned {} entries", count);
Ok(()) Ok(Some((BLOCK_HEADERS_PACKET, rlp)))
} }
/// Respond to GetBlockBodies request /// Respond to GetBlockBodies request
fn return_block_bodies(&self, io: &mut SyncIo, r: &UntrustedRlp) -> Result<(), PacketDecodeError> { fn return_block_bodies(io: &SyncIo, r: &UntrustedRlp) -> RlpResponseResult {
let mut count = r.item_count(); let mut count = r.item_count();
if count == 0 { if count == 0 {
debug!(target: "sync", "Empty GetBlockBodies request, ignoring."); debug!(target: "sync", "Empty GetBlockBodies request, ignoring.");
return Ok(()); return Ok(None);
} }
trace!(target: "sync", "-> GetBlockBodies: {} entries", count); trace!(target: "sync", "-> GetBlockBodies: {} entries", count);
count = min(count, MAX_BODIES_TO_SEND); count = min(count, MAX_BODIES_TO_SEND);
@ -917,18 +916,16 @@ impl ChainSync {
} }
let mut rlp = RlpStream::new_list(added); let mut rlp = RlpStream::new_list(added);
rlp.append_raw(&data, added); rlp.append_raw(&data, added);
io.respond(BLOCK_BODIES_PACKET, rlp.out()).unwrap_or_else(|e|
debug!(target: "sync", "Error sending headers: {:?}", e));
trace!(target: "sync", "-> GetBlockBodies: returned {} entries", added); trace!(target: "sync", "-> GetBlockBodies: returned {} entries", added);
Ok(()) Ok(Some((BLOCK_BODIES_PACKET, rlp)))
} }
/// Respond to GetNodeData request /// Respond to GetNodeData request
fn return_node_data(&self, io: &mut SyncIo, r: &UntrustedRlp) -> Result<(), PacketDecodeError> { fn return_node_data(io: &SyncIo, r: &UntrustedRlp) -> RlpResponseResult {
let mut count = r.item_count(); let mut count = r.item_count();
if count == 0 { if count == 0 {
debug!(target: "sync", "Empty GetNodeData request, ignoring."); debug!(target: "sync", "Empty GetNodeData request, ignoring.");
return Ok(()); return Ok(None);
} }
count = min(count, MAX_NODE_DATA_TO_SEND); count = min(count, MAX_NODE_DATA_TO_SEND);
let mut added = 0usize; let mut added = 0usize;
@ -941,33 +938,44 @@ impl ChainSync {
} }
let mut rlp = RlpStream::new_list(added); let mut rlp = RlpStream::new_list(added);
rlp.append_raw(&data, added); rlp.append_raw(&data, added);
io.respond(NODE_DATA_PACKET, rlp.out()).unwrap_or_else(|e| Ok(Some((NODE_DATA_PACKET, rlp)))
debug!(target: "sync", "Error sending headers: {:?}", e));
Ok(())
} }
/// Respond to GetReceipts request fn return_receipts(io: &SyncIo, rlp: &UntrustedRlp) -> RlpResponseResult {
fn return_receipts(&self, io: &mut SyncIo, r: &UntrustedRlp) -> Result<(), PacketDecodeError> { let mut count = rlp.item_count();
let mut count = r.item_count();
if count == 0 { if count == 0 {
debug!(target: "sync", "Empty GetReceipts request, ignoring."); debug!(target: "sync", "Empty GetReceipts request, ignoring.");
return Ok(()); return Ok(None);
} }
count = min(count, MAX_RECEIPTS_TO_SEND); count = min(count, MAX_RECEIPTS_TO_SEND);
let mut added = 0usize; let mut added = 0usize;
let mut data = Bytes::new(); let mut data = Bytes::new();
for i in 0..count { for i in 0..count {
if let Some(mut hdr) = io.chain().block_receipts(&try!(r.val_at::<H256>(i))) { if let Some(mut hdr) = io.chain().block_receipts(&try!(rlp.val_at::<H256>(i))) {
data.append(&mut hdr); data.append(&mut hdr);
added += 1; added += 1;
} }
} }
let mut rlp = RlpStream::new_list(added); let mut rlp_result = RlpStream::new_list(added);
rlp.append_raw(&data, added); rlp_result.append_raw(&data, added);
io.respond(RECEIPTS_PACKET, rlp.out()).unwrap_or_else(|e| Ok(Some((RECEIPTS_PACKET, rlp_result)))
debug!(target: "sync", "Error sending headers: {:?}", e)); }
fn return_rlp<FRlp, FError>(&self, io: &mut SyncIo, rlp: &UntrustedRlp, rlp_func: FRlp, error_func: FError) -> Result<(), PacketDecodeError>
where FRlp : Fn(&SyncIo, &UntrustedRlp) -> RlpResponseResult,
FError : FnOnce(UtilError) -> String
{
let response = rlp_func(io, rlp);
match response {
Err(e) => Err(e),
Ok(Some((packet_id, rlp_stream))) => {
io.respond(packet_id, rlp_stream.out()).unwrap_or_else(
|e| debug!(target: "sync", "{:?}", error_func(e)));
Ok(()) Ok(())
} }
_ => Ok(())
}
}
/// Dispatch incoming requests and responses /// Dispatch incoming requests and responses
pub fn on_packet(&mut self, io: &mut SyncIo, peer: PeerId, packet_id: u8, data: &[u8]) { pub fn on_packet(&mut self, io: &mut SyncIo, peer: PeerId, packet_id: u8, data: &[u8]) {
@ -975,14 +983,27 @@ impl ChainSync {
let result = match packet_id { let result = match packet_id {
STATUS_PACKET => self.on_peer_status(io, peer, &rlp), STATUS_PACKET => self.on_peer_status(io, peer, &rlp),
TRANSACTIONS_PACKET => self.on_peer_transactions(io, peer, &rlp), TRANSACTIONS_PACKET => self.on_peer_transactions(io, peer, &rlp),
GET_BLOCK_HEADERS_PACKET => self.return_block_headers(io, &rlp),
BLOCK_HEADERS_PACKET => self.on_peer_block_headers(io, peer, &rlp), BLOCK_HEADERS_PACKET => self.on_peer_block_headers(io, peer, &rlp),
GET_BLOCK_BODIES_PACKET => self.return_block_bodies(io, &rlp),
BLOCK_BODIES_PACKET => self.on_peer_block_bodies(io, peer, &rlp), BLOCK_BODIES_PACKET => self.on_peer_block_bodies(io, peer, &rlp),
NEW_BLOCK_PACKET => self.on_peer_new_block(io, peer, &rlp), NEW_BLOCK_PACKET => self.on_peer_new_block(io, peer, &rlp),
NEW_BLOCK_HASHES_PACKET => self.on_peer_new_hashes(io, peer, &rlp), NEW_BLOCK_HASHES_PACKET => self.on_peer_new_hashes(io, peer, &rlp),
GET_NODE_DATA_PACKET => self.return_node_data(io, &rlp),
GET_RECEIPTS_PACKET => self.return_receipts(io, &rlp), GET_BLOCK_BODIES_PACKET => self.return_rlp(io, &rlp,
ChainSync::return_block_bodies,
|e| format!("Error sending block bodies: {:?}", e)),
GET_BLOCK_HEADERS_PACKET => self.return_rlp(io, &rlp,
ChainSync::return_block_headers,
|e| format!("Error sending block headers: {:?}", e)),
GET_RECEIPTS_PACKET => self.return_rlp(io, &rlp,
ChainSync::return_receipts,
|e| format!("Error sending receipts: {:?}", e)),
GET_NODE_DATA_PACKET => self.return_rlp(io, &rlp,
ChainSync::return_node_data,
|e| format!("Error sending nodes: {:?}", e)),
_ => { _ => {
debug!(target: "sync", "Unknown packet {}", packet_id); debug!(target: "sync", "Unknown packet {}", packet_id);
Ok(()) Ok(())
@ -1016,4 +1037,75 @@ impl ChainSync {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use tests::helpers::*;
use super::*;
use util::*;
#[test]
fn return_receipts_empty() {
let mut client = TestBlockChainClient::new();
let mut queue = VecDeque::new();
let io = TestIo::new(&mut client, &mut queue, None);
let result = ChainSync::return_receipts(&io, &UntrustedRlp::new(&[0xc0]));
assert!(result.is_ok());
}
#[test]
fn return_receipts() {
let mut client = TestBlockChainClient::new();
let mut queue = VecDeque::new();
let mut io = TestIo::new(&mut client, &mut queue, None);
let mut receipt_list = RlpStream::new_list(4);
receipt_list.append(&H256::from("0000000000000000000000000000000000000000000000005555555555555555"));
receipt_list.append(&H256::from("ff00000000000000000000000000000000000000000000000000000000000000"));
receipt_list.append(&H256::from("fff0000000000000000000000000000000000000000000000000000000000000"));
receipt_list.append(&H256::from("aff0000000000000000000000000000000000000000000000000000000000000"));
let receipts_request = receipt_list.out();
// it returns rlp ONLY for hashes started with "f"
let result = ChainSync::return_receipts(&io, &UntrustedRlp::new(&receipts_request.clone()));
assert!(result.is_ok());
let rlp_result = result.unwrap();
assert!(rlp_result.is_some());
// the length of two rlp-encoded receipts
assert_eq!(597, rlp_result.unwrap().1.out().len());
let mut sync = ChainSync::new();
io.sender = Some(2usize);
sync.on_packet(&mut io, 1usize, super::GET_RECEIPTS_PACKET, &receipts_request);
assert_eq!(1, io.queue.len());
}
#[test]
fn return_nodes() {
let mut client = TestBlockChainClient::new();
let mut queue = VecDeque::new();
let mut io = TestIo::new(&mut client, &mut queue, None);
let mut node_list = RlpStream::new_list(3);
node_list.append(&H256::from("0000000000000000000000000000000000000000000000005555555555555555"));
node_list.append(&H256::from("ffffffffffffffffffffffffffffffffffffffffffffaaaaaaaaaaaaaaaaaaaa"));
node_list.append(&H256::from("aff0000000000000000000000000000000000000000000000000000000000000"));
let node_request = node_list.out();
// it returns rlp ONLY for hashes started with "f"
let result = ChainSync::return_node_data(&io, &UntrustedRlp::new(&node_request.clone()));
assert!(result.is_ok());
let rlp_result = result.unwrap();
assert!(rlp_result.is_some());
// the length of one rlp-encoded hashe
assert_eq!(34, rlp_result.unwrap().1.out().len());
let mut sync = ChainSync::new();
io.sender = Some(2usize);
sync.on_packet(&mut io, 1usize, super::GET_NODE_DATA_PACKET, &node_request);
assert_eq!(1, io.queue.len());
}
} }

View File

@ -5,7 +5,7 @@ use chain::{SyncState};
use super::helpers::*; use super::helpers::*;
#[test] #[test]
fn chain_two_peers() { fn two_peers() {
::env_logger::init().ok(); ::env_logger::init().ok();
let mut net = TestNet::new(3); let mut net = TestNet::new(3);
net.peer_mut(1).chain.add_blocks(1000, false); net.peer_mut(1).chain.add_blocks(1000, false);
@ -16,7 +16,7 @@ fn chain_two_peers() {
} }
#[test] #[test]
fn chain_status_after_sync() { fn status_after_sync() {
::env_logger::init().ok(); ::env_logger::init().ok();
let mut net = TestNet::new(3); let mut net = TestNet::new(3);
net.peer_mut(1).chain.add_blocks(1000, false); net.peer_mut(1).chain.add_blocks(1000, false);
@ -27,7 +27,7 @@ fn chain_status_after_sync() {
} }
#[test] #[test]
fn chain_takes_few_steps() { fn takes_few_steps() {
let mut net = TestNet::new(3); let mut net = TestNet::new(3);
net.peer_mut(1).chain.add_blocks(100, false); net.peer_mut(1).chain.add_blocks(100, false);
net.peer_mut(2).chain.add_blocks(100, false); net.peer_mut(2).chain.add_blocks(100, false);
@ -36,7 +36,7 @@ fn chain_takes_few_steps() {
} }
#[test] #[test]
fn chain_empty_blocks() { fn empty_blocks() {
::env_logger::init().ok(); ::env_logger::init().ok();
let mut net = TestNet::new(3); let mut net = TestNet::new(3);
for n in 0..200 { for n in 0..200 {
@ -49,7 +49,7 @@ fn chain_empty_blocks() {
} }
#[test] #[test]
fn chain_forked() { fn forked() {
::env_logger::init().ok(); ::env_logger::init().ok();
let mut net = TestNet::new(3); let mut net = TestNet::new(3);
net.peer_mut(0).chain.add_blocks(300, false); net.peer_mut(0).chain.add_blocks(300, false);
@ -69,7 +69,7 @@ fn chain_forked() {
} }
#[test] #[test]
fn chain_restart() { fn restart() {
let mut net = TestNet::new(3); let mut net = TestNet::new(3);
net.peer_mut(1).chain.add_blocks(1000, false); net.peer_mut(1).chain.add_blocks(1000, false);
net.peer_mut(2).chain.add_blocks(1000, false); net.peer_mut(2).chain.add_blocks(1000, false);
@ -85,7 +85,7 @@ fn chain_restart() {
} }
#[test] #[test]
fn chain_status_empty() { fn status_empty() {
let net = TestNet::new(2); let net = TestNet::new(2);
assert_eq!(net.peer(0).sync.status().state, SyncState::NotSynced); assert_eq!(net.peer(0).sync.status().state, SyncState::NotSynced);
} }

View File

@ -5,6 +5,7 @@ use ethcore::header::{Header as BlockHeader, BlockNumber};
use ethcore::error::*; use ethcore::error::*;
use io::SyncIo; use io::SyncIo;
use chain::{ChainSync}; use chain::{ChainSync};
use ethcore::receipt::Receipt;
pub struct TestBlockChainClient { pub struct TestBlockChainClient {
pub blocks: RwLock<HashMap<H256, Bytes>>, pub blocks: RwLock<HashMap<H256, Bytes>>,
@ -15,7 +16,7 @@ pub struct TestBlockChainClient {
} }
impl TestBlockChainClient { impl TestBlockChainClient {
fn new() -> TestBlockChainClient { pub fn new() -> TestBlockChainClient {
let mut client = TestBlockChainClient { let mut client = TestBlockChainClient {
blocks: RwLock::new(HashMap::new()), blocks: RwLock::new(HashMap::new()),
@ -116,11 +117,28 @@ impl BlockChainClient for TestBlockChainClient {
}) })
} }
fn state_data(&self, _h: &H256) -> Option<Bytes> { // TODO: returns just hashes instead of node state rlp(?)
fn state_data(&self, hash: &H256) -> Option<Bytes> {
// starts with 'f' ?
if *hash > H256::from("f000000000000000000000000000000000000000000000000000000000000000") {
let mut rlp = RlpStream::new();
rlp.append(&hash.clone());
return Some(rlp.out());
}
None None
} }
fn block_receipts(&self, _h: &H256) -> Option<Bytes> { fn block_receipts(&self, hash: &H256) -> Option<Bytes> {
// starts with 'f' ?
if *hash > H256::from("f000000000000000000000000000000000000000000000000000000000000000") {
let receipt = Receipt::new(
H256::zero(),
U256::zero(),
vec![]);
let mut rlp = RlpStream::new();
rlp.append(&receipt);
return Some(rlp.out());
}
None None
} }
@ -196,7 +214,7 @@ pub struct TestIo<'p> {
} }
impl<'p> TestIo<'p> { impl<'p> TestIo<'p> {
fn new(chain: &'p mut TestBlockChainClient, queue: &'p mut VecDeque<TestPacket>, sender: Option<PeerId>) -> TestIo<'p> { pub fn new(chain: &'p mut TestBlockChainClient, queue: &'p mut VecDeque<TestPacket>, sender: Option<PeerId>) -> TestIo<'p> {
TestIo { TestIo {
chain: chain, chain: chain,
queue: queue, queue: queue,