diff --git a/cov.sh b/cov.sh index a78e36205..eeb4d21f6 100755 --- a/cov.sh +++ b/cov.sh @@ -15,7 +15,7 @@ if ! type kcov > /dev/null; then exit 1 fi -cargo test -p ethcore --no-run || exit $? +cargo test --features ethcore/json-tests -p ethcore --no-run || exit $? mkdir -p target/coverage kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethcore* xdg-open target/coverage/index.html diff --git a/ethcore/src/block_queue.rs b/ethcore/src/block_queue.rs index 1fe31ba19..e14f2a06a 100644 --- a/ethcore/src/block_queue.rs +++ b/ethcore/src/block_queue.rs @@ -79,6 +79,8 @@ struct Verification { bad: HashSet, } +const MAX_UNVERIFIED_QUEUE_SIZE: usize = 50000; + impl BlockQueue { /// Creates a new queue instance. pub fn new(engine: Arc>, message_channel: IoChannel) -> BlockQueue { @@ -290,7 +292,7 @@ impl BlockQueue { pub fn queue_info(&self) -> BlockQueueInfo { let verification = self.verification.lock().unwrap(); BlockQueueInfo { - full: false, + full: verification.unverified.len() + verification.verifying.len() + verification.verified.len() >= MAX_UNVERIFIED_QUEUE_SIZE, verified_queue_size: verification.verified.len(), unverified_queue_size: verification.unverified.len(), verifying_queue_size: verification.verifying.len(), diff --git a/ethcore/src/transaction.rs b/ethcore/src/transaction.rs index 00ea784d4..a5e70b0d6 100644 --- a/ethcore/src/transaction.rs +++ b/ethcore/src/transaction.rs @@ -67,6 +67,24 @@ impl Transaction { } } + /// Create a new message-call transaction. + #[allow(dead_code)] + pub fn new_call(to: Address, value: U256, data: Bytes, gas: U256, gas_price: U256, nonce: U256) -> Transaction { + Transaction { + nonce: nonce, + gas_price: gas_price, + gas: gas, + action: Action::Call(to), + value: value, + data: data, + v: 0, + r: x!(0), + s: x!(0), + hash: RefCell::new(None), + sender: RefCell::new(None), + } + } + /// Create a new contract-creation transaction. #[cfg(test)] pub fn new_create(value: U256, data: Bytes, gas: U256, gas_price: U256, nonce: U256) -> Transaction { diff --git a/install-deps.sh b/install-deps.sh index b10d373b7..15b863682 100755 --- a/install-deps.sh +++ b/install-deps.sh @@ -463,4 +463,4 @@ function run_installer() finish } -run_installer \ No newline at end of file +run_installer diff --git a/parity/main.rs b/parity/main.rs index fe567e2a0..bc492277a 100644 --- a/parity/main.rs +++ b/parity/main.rs @@ -37,10 +37,10 @@ Usage: Options: -l --logging LOGGING Specify the logging level. -j --jsonrpc Enable the JSON-RPC API sever. - --jsonrpc-url URL Specify URL for JSON-RPC API server (default: 127.0.0.1:8545). + --jsonrpc-url URL Specify URL for JSON-RPC API server [default: 127.0.0.1:8545]. - --cache-pref-size BYTES Specify the prefered size of the blockchain cache in bytes (default: 16384). - --cache-max-size BYTES Specify the maximum size of the blockchain cache in bytes (default: 262144). + --cache-pref-size BYTES Specify the prefered size of the blockchain cache in bytes [default: 16384]. + --cache-max-size BYTES Specify the maximum size of the blockchain cache in bytes [default: 262144]. -h --help Show this screen. ", flag_cache_pref_size: usize, flag_cache_max_size: usize); diff --git a/rpc/Cargo.toml b/rpc/Cargo.toml index 60b29c5d4..bea85a74f 100644 --- a/rpc/Cargo.toml +++ b/rpc/Cargo.toml @@ -1,7 +1,7 @@ [package] description = "Ethcore jsonrpc" name = "ethcore-rpc" -version = "0.1.0" +version = "0.9.0" license = "GPL-3.0" authors = ["Ethcore Result { match params { - //Params::None => Ok(Value::String("parity/0.1.0/-/rust1.7-nightly".to_owned())), - Params::None => Ok(Value::String("surprise/0.1.0/surprise/surprise".to_owned())), + Params::None => Ok(Value::String(format!("Parity/-/{}/{}-{}-{}/rust1.8-nightly", env!("CARGO_PKG_VERSION"), Target::arch(), Target::env(), Target::os()))), _ => Err(Error::invalid_params()) } } diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 9752a5013..665508d92 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -23,6 +23,7 @@ use ethcore::error::*; use ethcore::block::Block; use io::SyncIo; use time; +use std::option::Option; impl ToUsize for BlockNumber { fn to_usize(&self) -> usize { @@ -81,7 +82,7 @@ struct HeaderId { } #[derive(Copy, Clone, Eq, PartialEq, Debug)] -/// Sync state +/// Sync state pub enum SyncState { /// Initial chain sync has not started yet NotSynced, @@ -103,14 +104,14 @@ pub struct SyncStatus { pub protocol_version: u8, /// BlockChain height for the moment the sync started. pub start_block_number: BlockNumber, - /// Last fully downloaded and imported block number. - pub last_imported_block_number: BlockNumber, - /// Highest block number in the download queue. - pub highest_block_number: BlockNumber, + /// Last fully downloaded and imported block number (if any). + pub last_imported_block_number: Option, + /// Highest block number in the download queue (if any). + pub highest_block_number: Option, /// Total number of blocks for the sync process. - pub blocks_total: usize, + pub blocks_total: BlockNumber, /// Number of blocks downloaded so far. - pub blocks_received: usize, + pub blocks_received: BlockNumber, /// Total number of connected peers pub num_peers: usize, /// Total number of active peers @@ -131,7 +132,7 @@ struct PeerInfo { protocol_version: u32, /// Peer chain genesis hash genesis: H256, - /// Peer network id + /// Peer network id network_id: U256, /// Peer best block hash latest: H256, @@ -153,7 +154,7 @@ pub struct ChainSync { /// Last block number for the start of sync starting_block: BlockNumber, /// Highest block number seen - highest_block: BlockNumber, + highest_block: Option, /// Set of block header numbers being downloaded downloading_headers: HashSet, /// Set of block body numbers being downloaded @@ -167,9 +168,9 @@ pub struct ChainSync { /// Used to map body to header header_ids: HashMap, /// Last impoted block number - last_imported_block: BlockNumber, + last_imported_block: Option, /// Last impoted block hash - last_imported_hash: H256, + last_imported_hash: Option, /// Syncing total difficulty syncing_difficulty: U256, /// True if common block for our and remote chain has been found @@ -183,15 +184,15 @@ impl ChainSync { ChainSync { state: SyncState::NotSynced, starting_block: 0, - highest_block: 0, + highest_block: None, downloading_headers: HashSet::new(), downloading_bodies: HashSet::new(), headers: Vec::new(), bodies: Vec::new(), peers: HashMap::new(), header_ids: HashMap::new(), - last_imported_block: 0, - last_imported_hash: H256::new(), + last_imported_block: None, + last_imported_hash: None, syncing_difficulty: U256::from(0u64), have_common_block: false, } @@ -205,8 +206,8 @@ impl ChainSync { start_block_number: self.starting_block, last_imported_block_number: self.last_imported_block, highest_block_number: self.highest_block, - blocks_received: (self.last_imported_block - self.starting_block) as usize, - blocks_total: (self.highest_block - self.starting_block) as usize, + blocks_received: match self.last_imported_block { None => 0, Some(x) => x - self.starting_block }, + blocks_total: match self.highest_block { None => 0, Some(x) => x - self.starting_block }, num_peers: self.peers.len(), num_active_peers: self.peers.values().filter(|p| p.asking != PeerAsking::Nothing).count(), } @@ -235,10 +236,10 @@ impl ChainSync { /// Restart sync pub fn restart(&mut self, io: &mut SyncIo) { self.reset(); - self.last_imported_block = 0; - self.last_imported_hash = H256::new(); + self.last_imported_block = None; + self.last_imported_hash = None; self.starting_block = 0; - self.highest_block = 0; + self.highest_block = None; self.have_common_block = false; io.chain().clear_queue(); self.starting_block = io.chain().chain_info().best_block_number; @@ -259,7 +260,7 @@ impl ChainSync { }; trace!(target: "sync", "New peer {} (protocol: {}, network: {:?}, difficulty: {:?}, latest:{}, genesis:{})", peer_id, peer.protocol_version, peer.network_id, peer.difficulty, peer.latest, peer.genesis); - + let chain_info = io.chain().chain_info(); if peer.genesis != chain_info.genesis_hash { io.disable_peer(peer_id); @@ -300,32 +301,34 @@ impl ChainSync { for i in 0..item_count { let info: BlockHeader = try!(r.val_at(i)); let number = BlockNumber::from(info.number); - if number <= self.last_imported_block || self.headers.have_item(&number) { + if number <= self.current_base_block() || self.headers.have_item(&number) { trace!(target: "sync", "Skipping existing block header"); continue; } - if number > self.highest_block { - self.highest_block = number; + + if self.highest_block == None || number > self.highest_block.unwrap() { + self.highest_block = Some(number); } let hash = info.hash(); match io.chain().block_status(&hash) { BlockStatus::InChain => { self.have_common_block = true; - self.last_imported_block = number; - self.last_imported_hash = hash.clone(); + self.last_imported_block = Some(number); + self.last_imported_hash = Some(hash.clone()); trace!(target: "sync", "Found common header {} ({})", number, hash); }, _ => { if self.have_common_block { //validate chain - if self.have_common_block && number == self.last_imported_block + 1 && info.parent_hash != self.last_imported_hash { - // TODO: lower peer rating - debug!(target: "sync", "Mismatched block header {} {}", number, hash); - continue; + let base_hash = self.last_imported_hash.clone().unwrap(); + if self.have_common_block && number == self.current_base_block() + 1 && info.parent_hash != base_hash { + // Part of the forked chain. Restart to find common block again + debug!(target: "sync", "Mismatched block header {} {}, restarting sync", number, hash); + self.restart(io); + return Ok(()); } if self.headers.find_item(&(number - 1)).map_or(false, |p| p.hash != info.parent_hash) { // mismatching parent id, delete the previous block and don't add this one - // TODO: lower peer rating debug!(target: "sync", "Mismatched block header {} {}", number, hash); self.remove_downloaded_blocks(number - 1); continue; @@ -376,7 +379,7 @@ impl ChainSync { trace!(target: "sync", "Ignored unexpected block bodies"); return Ok(()); } - if self.state == SyncState::Waiting { + if self.state == SyncState::Waiting { trace!(target: "sync", "Ignored block bodies while waiting"); return Ok(()); } @@ -413,9 +416,9 @@ impl ChainSync { trace!(target: "sync", "{} -> NewBlock ({})", peer_id, h); let header_view = HeaderView::new(header_rlp.as_raw()); + let mut unknown = false; // TODO: Decompose block and add to self.headers and self.bodies instead - let mut unknown = false; - if header_view.number() == From::from(self.last_imported_block + 1) { + if header_view.number() == From::from(self.current_base_block() + 1) { match io.chain().import_block(block_rlp.as_raw().to_vec()) { Err(ImportError::AlreadyInChain) => { trace!(target: "sync", "New block already in chain {:?}", h); @@ -423,20 +426,20 @@ impl ChainSync { Err(ImportError::AlreadyQueued) => { trace!(target: "sync", "New block already queued {:?}", h); }, + Ok(_) => { + trace!(target: "sync", "New block queued {:?}", h); + }, Err(ImportError::UnknownParent) => { unknown = true; trace!(target: "sync", "New block with unknown parent {:?}", h); }, - Ok(_) => { - trace!(target: "sync", "New block queued {:?}", h); - }, Err(e) => { debug!(target: "sync", "Bad new block {:?} : {:?}", h, e); io.disable_peer(peer_id); } }; - } - else { + } + else { unknown = true; } if unknown { @@ -456,7 +459,7 @@ impl ChainSync { Ok(()) } - /// Handles NewHashes packet. Initiates headers download for any unknown hashes. + /// Handles NewHashes packet. Initiates headers download for any unknown hashes. fn on_peer_new_hashes(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> { if self.peers.get_mut(&peer_id).expect("ChainSync: unknown peer").asking != PeerAsking::Nothing { trace!(target: "sync", "Ignoring new hashes since we're already downloading."); @@ -565,6 +568,10 @@ impl ChainSync { } } + fn current_base_block(&self) -> BlockNumber { + match self.last_imported_block { None => 0, Some(x) => x } + } + /// Find some headers or blocks to download for a peer. fn request_blocks(&mut self, io: &mut SyncIo, peer_id: PeerId) { self.clear_peer_download(peer_id); @@ -578,7 +585,7 @@ impl ChainSync { let mut needed_bodies: Vec = Vec::new(); let mut needed_numbers: Vec = Vec::new(); - if self.have_common_block && !self.headers.is_empty() && self.headers.range_iter().next().unwrap().0 == self.last_imported_block + 1 { + if self.have_common_block && !self.headers.is_empty() && self.headers.range_iter().next().unwrap().0 == self.current_base_block() + 1 { for (start, ref items) in self.headers.range_iter() { if needed_bodies.len() > MAX_BODIES_TO_REQUEST { break; @@ -611,12 +618,12 @@ impl ChainSync { } if start == 0 { self.have_common_block = true; //reached genesis - self.last_imported_hash = chain_info.genesis_hash; + self.last_imported_hash = Some(chain_info.genesis_hash); } } if self.have_common_block { let mut headers: Vec = Vec::new(); - let mut prev = self.last_imported_block + 1; + let mut prev = self.current_base_block() + 1; for (next, ref items) in self.headers.range_iter() { if !headers.is_empty() { break; @@ -671,7 +678,7 @@ impl ChainSync { { let headers = self.headers.range_iter().next().unwrap(); let bodies = self.bodies.range_iter().next().unwrap(); - if headers.0 != bodies.0 || headers.0 != self.last_imported_block + 1 { + if headers.0 != bodies.0 || headers.0 != self.current_base_block() + 1 { return; } @@ -684,27 +691,29 @@ impl ChainSync { block_rlp.append_raw(body.at(0).as_raw(), 1); block_rlp.append_raw(body.at(1).as_raw(), 1); let h = &headers.1[i].hash; - // Perform basic block verification + + // Perform basic block verification if !Block::is_good(block_rlp.as_raw()) { debug!(target: "sync", "Bad block rlp {:?} : {:?}", h, block_rlp.as_raw()); restart = true; break; } + match io.chain().import_block(block_rlp.out()) { Err(ImportError::AlreadyInChain) => { trace!(target: "sync", "Block already in chain {:?}", h); - self.last_imported_block = headers.0 + i as BlockNumber; - self.last_imported_hash = h.clone(); + self.last_imported_block = Some(headers.0 + i as BlockNumber); + self.last_imported_hash = Some(h.clone()); }, Err(ImportError::AlreadyQueued) => { trace!(target: "sync", "Block already queued {:?}", h); - self.last_imported_block = headers.0 + i as BlockNumber; - self.last_imported_hash = h.clone(); + self.last_imported_block = Some(headers.0 + i as BlockNumber); + self.last_imported_hash = Some(h.clone()); }, Ok(_) => { trace!(target: "sync", "Block queued {:?}", h); - self.last_imported_block = headers.0 + i as BlockNumber; - self.last_imported_hash = h.clone(); + self.last_imported_block = Some(headers.0 + i as BlockNumber); + self.last_imported_hash = Some(h.clone()); imported += 1; }, Err(e) => { @@ -721,8 +730,8 @@ impl ChainSync { return; } - self.headers.remove_head(&(self.last_imported_block + 1)); - self.bodies.remove_head(&(self.last_imported_block + 1)); + self.headers.remove_head(&(self.last_imported_block.unwrap() + 1)); + self.bodies.remove_head(&(self.last_imported_block.unwrap() + 1)); if self.headers.is_empty() { assert!(self.bodies.is_empty()); @@ -730,7 +739,7 @@ impl ChainSync { } } - /// Remove downloaded bocks/headers starting from specified number. + /// Remove downloaded bocks/headers starting from specified number. /// Used to recover from an error and re-download parts of the chain detected as bad. fn remove_downloaded_blocks(&mut self, start: BlockNumber) { for n in self.headers.get_tail(&start) { @@ -974,7 +983,7 @@ impl ChainSync { 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), - _ => { + _ => { debug!(target: "sync", "Unknown packet {}", packet_id); Ok(()) } @@ -984,7 +993,10 @@ impl ChainSync { }) } - /// Handle peer timeouts + /// Maintain other peers. Send out any new blocks and transactions + pub fn _maintain_sync(&mut self, _io: &mut SyncIo) { + } + pub fn maintain_peers(&self, io: &mut SyncIo) { let tick = time::precise_time_s(); for (peer_id, peer) in &self.peers { @@ -993,9 +1005,11 @@ impl ChainSync { } } } - /// Maintain other peers. Send out any new blocks and transactions - pub fn _maintain_sync(&mut self, _io: &mut SyncIo) { + pub fn maintain_sync(&mut self, io: &mut SyncIo) { + if !io.chain().queue_info().full && self.state == SyncState::Waiting { + self.state = SyncState::Idle; + self.continue_sync(io); + } } } - diff --git a/sync/src/lib.rs b/sync/src/lib.rs index 40b67dc5b..f3b43396c 100644 --- a/sync/src/lib.rs +++ b/sync/src/lib.rs @@ -107,6 +107,7 @@ impl NetworkProtocolHandler for EthSync { fn timeout(&self, io: &NetworkContext, _timer: TimerToken) { self.sync.write().unwrap().maintain_peers(&mut NetSyncIo::new(io, self.chain.deref())); + self.sync.write().unwrap().maintain_sync(&mut NetSyncIo::new(io, self.chain.deref())); } } diff --git a/sync/src/tests.rs b/sync/src/tests.rs index 5b796e6f1..88f19a8b6 100644 --- a/sync/src/tests.rs +++ b/sync/src/tests.rs @@ -4,7 +4,7 @@ use ethcore::block_queue::BlockQueueInfo; use ethcore::header::{Header as BlockHeader, BlockNumber}; use ethcore::error::*; use io::SyncIo; -use chain::ChainSync; +use chain::{ChainSync, SyncState}; struct TestBlockChainClient { blocks: RwLock>, @@ -248,13 +248,15 @@ struct TestPeer { } struct TestNet { - peers: Vec + peers: Vec, + started: bool, } impl TestNet { pub fn new(n: usize) -> TestNet { let mut net = TestNet { peers: Vec::new(), + started: false, }; for _ in 0..n { net.peers.push(TestPeer { @@ -294,14 +296,32 @@ impl TestNet { trace!("----------------"); } let mut p = self.peers.get_mut(peer).unwrap(); - p.sync._maintain_sync(&mut TestIo::new(&mut p.chain, &mut p.queue, None)); + p.sync.maintain_sync(&mut TestIo::new(&mut p.chain, &mut p.queue, None)); } } - pub fn sync(&mut self) { + pub fn restart_peer(&mut self, i: usize) { + let peer = self.peer_mut(i); + peer.sync.restart(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None)); + } + + pub fn sync(&mut self) -> u32 { self.start(); + let mut total_steps = 0; while !self.done() { - self.sync_step() + self.sync_step(); + total_steps = 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(); } } @@ -310,9 +330,8 @@ impl TestNet { } } - #[test] -fn full_sync_two_peers() { +fn chain_two_peers() { ::env_logger::init().ok(); let mut net = TestNet::new(3); net.peer_mut(1).chain.add_blocks(1000, false); @@ -323,7 +342,27 @@ fn full_sync_two_peers() { } #[test] -fn full_sync_empty_blocks() { +fn chain_status_after_sync() { + ::env_logger::init().ok(); + let mut net = TestNet::new(3); + net.peer_mut(1).chain.add_blocks(1000, false); + net.peer_mut(2).chain.add_blocks(1000, false); + net.sync(); + let status = net.peer(0).sync.status(); + assert_eq!(status.state, SyncState::Idle); +} + +#[test] +fn chain_takes_few_steps() { + let mut net = TestNet::new(3); + net.peer_mut(1).chain.add_blocks(100, false); + net.peer_mut(2).chain.add_blocks(100, false); + let total_steps = net.sync(); + assert!(total_steps < 7); +} + +#[test] +fn chain_empty_blocks() { ::env_logger::init().ok(); let mut net = TestNet::new(3); for n in 0..200 { @@ -336,7 +375,7 @@ fn full_sync_empty_blocks() { } #[test] -fn forked_sync() { +fn chain_forked() { ::env_logger::init().ok(); let mut net = TestNet::new(3); net.peer_mut(0).chain.add_blocks(300, false); @@ -354,3 +393,25 @@ fn forked_sync() { assert_eq!(net.peer(1).chain.numbers.read().unwrap().deref(), &peer1_chain); assert_eq!(net.peer(2).chain.numbers.read().unwrap().deref(), &peer1_chain); } + +#[test] +fn chain_restart() { + let mut net = TestNet::new(3); + net.peer_mut(1).chain.add_blocks(1000, false); + net.peer_mut(2).chain.add_blocks(1000, false); + + 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.status(); + assert_eq!(status.state, SyncState::NotSynced); +} + +#[test] +fn chain_status_empty() { + let net = TestNet::new(2); + assert_eq!(net.peer(0).sync.status().state, SyncState::NotSynced); +} \ No newline at end of file diff --git a/util/cov.sh b/util/cov.sh deleted file mode 120000 index 72bb061f8..000000000 --- a/util/cov.sh +++ /dev/null @@ -1 +0,0 @@ -../cov.sh \ No newline at end of file diff --git a/util/cov.sh b/util/cov.sh new file mode 100755 index 000000000..19aa3b892 --- /dev/null +++ b/util/cov.sh @@ -0,0 +1,9 @@ +if ! type kcov > /dev/null; then + echo "Install kcov first (details inside this file). Aborting." + exit 1 +fi + +cargo test --no-run || exit $? +mkdir -p target/coverage +kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/ethcore_util* +xdg-open target/coverage/index.html diff --git a/util/src/network/connection.rs b/util/src/network/connection.rs index d7bfe5284..de1ca185a 100644 --- a/util/src/network/connection.rs +++ b/util/src/network/connection.rs @@ -6,7 +6,7 @@ use hash::*; use sha3::*; use bytes::*; use rlp::*; -use std::io::{self, Cursor, Read}; +use std::io::{self, Cursor, Read, Write}; use error::*; use io::{IoContext, StreamToken}; use network::error::NetworkError; @@ -22,12 +22,17 @@ use tiny_keccak::Keccak; const ENCRYPTED_HEADER_LEN: usize = 32; const RECIEVE_PAYLOAD_TIMEOUT: u64 = 30000; -/// Low level tcp connection -pub struct Connection { +pub trait GenericSocket : Read + Write { +} + +impl GenericSocket for TcpStream { +} + +pub struct GenericConnection { /// Connection id (token) pub token: StreamToken, /// Network socket - pub socket: TcpStream, + pub socket: Socket, /// Receive buffer rec_buf: Bytes, /// Expected size @@ -36,34 +41,11 @@ pub struct Connection { send_queue: VecDeque>, /// Event flags this connection expects interest: EventSet, - /// Shared network staistics + /// Shared network statistics stats: Arc, } -/// Connection write status. -#[derive(PartialEq, Eq)] -pub enum WriteStatus { - /// Some data is still pending for current packet - Ongoing, - /// All data sent. - Complete -} - -impl Connection { - /// Create a new connection with given id and socket. - pub fn new(token: StreamToken, socket: TcpStream, stats: Arc) -> Connection { - Connection { - token: token, - socket: socket, - send_queue: VecDeque::new(), - rec_buf: Bytes::new(), - rec_size: 0, - interest: EventSet::hup() | EventSet::readable(), - stats: stats, - } - } - - /// Put a connection into read mode. Receiving up `size` bytes of data. +impl GenericConnection { pub fn expect(&mut self, size: usize) { if self.rec_size != self.rec_buf.len() { warn!(target:"net", "Unexpected connection read start"); @@ -79,7 +61,7 @@ impl Connection { } let max = self.rec_size - self.rec_buf.len(); // resolve "multiple applicable items in scope [E0034]" error - let sock_ref = ::by_ref(&mut self.socket); + let sock_ref = ::by_ref(&mut self.socket); match sock_ref.take(max as u64).try_read_buf(&mut self.rec_buf) { Ok(Some(size)) if size != 0 => { self.stats.inc_recv(size); @@ -142,6 +124,24 @@ impl Connection { Ok(r) }) } +} + +/// Low level tcp connection +pub type Connection = GenericConnection; + +impl Connection { + /// Create a new connection with given id and socket. + pub fn new(token: StreamToken, socket: TcpStream, stats: Arc) -> Connection { + Connection { + token: token, + socket: socket, + send_queue: VecDeque::new(), + rec_buf: Bytes::new(), + rec_size: 0, + interest: EventSet::hup() | EventSet::readable(), + stats: stats, + } + } /// Register this connection with the IO event loop. pub fn register_socket(&self, reg: Token, event_loop: &mut EventLoop) -> io::Result<()> { @@ -169,6 +169,15 @@ impl Connection { } } +/// Connection write status. +#[derive(PartialEq, Eq)] +pub enum WriteStatus { + /// Some data is still pending for current packet + Ongoing, + /// All data sent. + Complete +} + /// RLPx packet pub struct Packet { pub protocol: u16, @@ -424,4 +433,230 @@ pub fn test_encryption() { assert_eq!(got, after2); } +#[cfg(test)] +mod tests { + use super::*; + use std::sync::*; + use super::super::stats::*; + use std::io::{Read, Write, Error, Cursor, ErrorKind}; + use std::cmp; + use mio::{EventSet}; + use std::collections::VecDeque; + use bytes::*; + struct TestSocket { + read_buffer: Vec, + write_buffer: Vec, + cursor: usize, + buf_size: usize, + } + + impl TestSocket { + fn new() -> TestSocket { + TestSocket { + read_buffer: vec![], + write_buffer: vec![], + cursor: 0, + buf_size: 0, + } + } + + fn new_buf(buf_size: usize) -> TestSocket { + TestSocket { + read_buffer: vec![], + write_buffer: vec![], + cursor: 0, + buf_size: buf_size, + } + } + } + + impl Read for TestSocket { + fn read(&mut self, buf: &mut [u8]) -> Result { + let end_position = cmp::min(self.read_buffer.len(), self.cursor+buf.len()); + let len = cmp::max(end_position - self.cursor, 0); + match len { + 0 => Ok(0), + _ => { + for i in self.cursor..end_position { + buf[i-self.cursor] = self.read_buffer[i]; + } + self.cursor = self.cursor + buf.len(); + Ok(len) + } + } + } + } + + impl Write for TestSocket { + fn write(&mut self, buf: &[u8]) -> Result { + if self.buf_size == 0 || buf.len() < self.buf_size { + self.write_buffer.extend(buf.iter().cloned()); + Ok(buf.len()) + } + else { + self.write_buffer.extend(buf.iter().take(self.buf_size).cloned()); + Ok(self.buf_size) + } + } + + fn flush(&mut self) -> Result<(), Error> { + unimplemented!(); + } + } + + impl GenericSocket for TestSocket {} + + struct TestBrokenSocket { + error: String + } + + impl Read for TestBrokenSocket { + fn read(&mut self, _: &mut [u8]) -> Result { + Err(Error::new(ErrorKind::Other, self.error.clone())) + } + } + + impl Write for TestBrokenSocket { + fn write(&mut self, _: &[u8]) -> Result { + Err(Error::new(ErrorKind::Other, self.error.clone())) + } + + fn flush(&mut self) -> Result<(), Error> { + unimplemented!(); + } + } + + impl GenericSocket for TestBrokenSocket {} + + type TestConnection = GenericConnection; + + impl TestConnection { + pub fn new() -> TestConnection { + TestConnection { + token: 999998888usize, + socket: TestSocket::new(), + send_queue: VecDeque::new(), + rec_buf: Bytes::new(), + rec_size: 0, + interest: EventSet::hup() | EventSet::readable(), + stats: Arc::::new(NetworkStats::new()), + } + } + } + + type TestBrokenConnection = GenericConnection; + + impl TestBrokenConnection { + pub fn new() -> TestBrokenConnection { + TestBrokenConnection { + token: 999998888usize, + socket: TestBrokenSocket { error: "test broken socket".to_owned() }, + send_queue: VecDeque::new(), + rec_buf: Bytes::new(), + rec_size: 0, + interest: EventSet::hup() | EventSet::readable(), + stats: Arc::::new(NetworkStats::new()), + } + } + } + + #[test] + fn connection_expect() { + let mut connection = TestConnection::new(); + connection.expect(1024); + assert_eq!(1024, connection.rec_size); + } + + #[test] + fn connection_write_empty() { + let mut connection = TestConnection::new(); + let status = connection.writable(); + assert!(status.is_ok()); + assert!(WriteStatus::Complete == status.unwrap()); + } + + #[test] + fn connection_write() { + let mut connection = TestConnection::new(); + let data = Cursor::new(vec![0; 10240]); + connection.send_queue.push_back(data); + + let status = connection.writable(); + assert!(status.is_ok()); + assert!(WriteStatus::Complete == status.unwrap()); + assert_eq!(10240, connection.socket.write_buffer.len()); + } + + #[test] + fn connection_write_is_buffered() { + let mut connection = TestConnection::new(); + connection.socket = TestSocket::new_buf(1024); + let data = Cursor::new(vec![0; 10240]); + connection.send_queue.push_back(data); + + let status = connection.writable(); + + assert!(status.is_ok()); + assert!(WriteStatus::Ongoing == status.unwrap()); + assert_eq!(1024, connection.socket.write_buffer.len()); + } + + #[test] + fn connection_write_to_broken() { + let mut connection = TestBrokenConnection::new(); + let data = Cursor::new(vec![0; 10240]); + connection.send_queue.push_back(data); + + let status = connection.writable(); + + assert!(!status.is_ok()); + assert_eq!(1, connection.send_queue.len()); + } + + #[test] + fn connection_read() { + let mut connection = TestConnection::new(); + connection.rec_size = 2048; + connection.rec_buf = vec![10; 1024]; + connection.socket.read_buffer = vec![99; 2048]; + + let status = connection.readable(); + + assert!(status.is_ok()); + assert_eq!(1024, connection.socket.cursor); + } + + #[test] + fn connection_read_from_broken() { + let mut connection = TestBrokenConnection::new(); + connection.rec_size = 2048; + + let status = connection.readable(); + assert!(!status.is_ok()); + assert_eq!(0, connection.rec_buf.len()); + } + + #[test] + fn connection_read_nothing() { + let mut connection = TestConnection::new(); + connection.rec_size = 2048; + + let status = connection.readable(); + + assert!(status.is_ok()); + assert_eq!(0, connection.rec_buf.len()); + } + + #[test] + fn connection_read_full() { + let mut connection = TestConnection::new(); + connection.rec_size = 1024; + connection.rec_buf = vec![76;1024]; + + let status = connection.readable(); + + assert!(status.is_ok()); + assert_eq!(0, connection.socket.cursor); + } +} diff --git a/util/src/network/node.rs b/util/src/network/node.rs index fd27e58e2..d99754cff 100644 --- a/util/src/network/node.rs +++ b/util/src/network/node.rs @@ -83,3 +83,36 @@ impl Hash for Node { } } +#[cfg(test)] +mod tests { + use super::*; + use std::str::FromStr; + use std::net::*; + use hash::*; + + #[test] + fn endpoint_parse() { + let endpoint = NodeEndpoint::from_str("123.99.55.44:7770"); + assert!(endpoint.is_ok()); + let v4 = match endpoint.unwrap().address { + SocketAddr::V4(v4address) => v4address, + _ => panic!("should ve v4 address") + }; + assert_eq!(SocketAddrV4::new(Ipv4Addr::new(123, 99, 55, 44), 7770), v4); + } + + #[test] + fn node_parse() { + let node = Node::from_str("enode://a979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c@22.99.55.44:7770"); + assert!(node.is_ok()); + let node = node.unwrap(); + let v4 = match node.endpoint.address { + SocketAddr::V4(v4address) => v4address, + _ => panic!("should ve v4 address") + }; + assert_eq!(SocketAddrV4::new(Ipv4Addr::new(22, 99, 55, 44), 7770), v4); + assert_eq!( + H512::from_str("a979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c").unwrap(), + node.id); + } +} diff --git a/util/src/network/stats.rs b/util/src/network/stats.rs index 02d904985..39ad4e079 100644 --- a/util/src/network/stats.rs +++ b/util/src/network/stats.rs @@ -48,4 +48,13 @@ impl NetworkStats { pub fn sessions(&self) -> usize { self.sessions.load(Ordering::Relaxed) } + + #[cfg(test)] + pub fn new() -> NetworkStats { + NetworkStats { + recv: AtomicUsize::new(0), + send: AtomicUsize::new(0), + sessions: AtomicUsize::new(0), + } + } } diff --git a/util/src/network/tests.rs b/util/src/network/tests.rs index 06966abb5..bccd4c27d 100644 --- a/util/src/network/tests.rs +++ b/util/src/network/tests.rs @@ -68,13 +68,13 @@ impl NetworkProtocolHandler for TestProtocol { #[test] -fn test_net_service() { +fn net_service() { let mut service = NetworkService::::start(NetworkConfiguration::new()).expect("Error creating network service"); service.register_protocol(Arc::new(TestProtocol::default()), "myproto", &[1u8]).unwrap(); } #[test] -fn test_net_connect() { +fn net_connect() { let key1 = KeyPair::create().unwrap(); let mut config1 = NetworkConfiguration::new_with_port(30344); config1.use_secret = Some(key1.secret().clone()); @@ -93,7 +93,7 @@ fn test_net_connect() { } #[test] -fn test_net_timeout() { +fn net_timeout() { let config = NetworkConfiguration::new_with_port(30346); let mut service = NetworkService::::start(config).unwrap(); let handler = TestProtocol::register(&mut service);