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

Merge branch 'master' into temp-path

Browse Source 
Conflicts:
	util/Cargo.toml
	util/src/lib.rs
This commit is contained in:
Nikolay Volf 2016-02-19 19:11:24 +03:00
commit 84e7ce2758
25 changed files with 1914 additions and 693 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.lock generated
View File

@ -174,6 +174,7 @@ dependencies = [
"crossbeam 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", "crossbeam 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
"env_logger 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)", "env_logger 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)",
"ethash 0.9.99", "ethash 0.9.99",
"ethcore-devtools 0.9.99",
"ethcore-util 0.9.99", "ethcore-util 0.9.99",
"heapsize 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)", "heapsize 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)",
"lazy_static 0.1.15 (registry+https://github.com/rust-lang/crates.io-index)", "lazy_static 0.1.15 (registry+https://github.com/rust-lang/crates.io-index)",
@ -226,6 +227,7 @@ dependencies = [
"itertools 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", "itertools 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)",
"json-tests 0.1.0", "json-tests 0.1.0",
"lazy_static 0.1.15 (registry+https://github.com/rust-lang/crates.io-index)", "lazy_static 0.1.15 (registry+https://github.com/rust-lang/crates.io-index)",
"libc 0.2.7 (registry+https://github.com/rust-lang/crates.io-index)",
"log 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
"mio 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)", "mio 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)",
"rand 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)", "rand 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)",

8
ethcore/src/block_queue.rs
View File

@ -153,7 +153,7 @@ impl BlockQueue {
} }
fn verify(verification: Arc<Mutex<Verification>>, engine: Arc<Box<Engine>>, wait: Arc<Condvar>, ready: Arc<QueueSignal>, deleting: Arc<AtomicBool>, empty: Arc<Condvar>) { fn verify(verification: Arc<Mutex<Verification>>, engine: Arc<Box<Engine>>, wait: Arc<Condvar>, ready: Arc<QueueSignal>, deleting: Arc<AtomicBool>, empty: Arc<Condvar>) {
while !deleting.load(AtomicOrdering::Relaxed) { while !deleting.load(AtomicOrdering::Acquire) {
{ {
let mut lock = verification.lock().unwrap(); let mut lock = verification.lock().unwrap();
@ -161,11 +161,11 @@ impl BlockQueue {
empty.notify_all(); empty.notify_all();
} }
while lock.unverified.is_empty() && !deleting.load(AtomicOrdering::Relaxed) { while lock.unverified.is_empty() && !deleting.load(AtomicOrdering::Acquire) {
lock = wait.wait(lock).unwrap(); lock = wait.wait(lock).unwrap();
} }
if deleting.load(AtomicOrdering::Relaxed) { if deleting.load(AtomicOrdering::Acquire) {
return; return;
} }
} }
@ -347,7 +347,7 @@ impl MayPanic for BlockQueue {
impl Drop for BlockQueue { impl Drop for BlockQueue {
fn drop(&mut self) { fn drop(&mut self) {
self.clear(); self.clear();
self.deleting.store(true, AtomicOrdering::Relaxed); self.deleting.store(true, AtomicOrdering::Release);
self.more_to_verify.notify_all(); self.more_to_verify.notify_all();
for t in self.verifiers.drain(..) { for t in self.verifiers.drain(..) {
t.join().unwrap(); t.join().unwrap();

4
ethcore/src/client.rs
View File

@ -419,11 +419,11 @@ impl BlockChainClient for Client {
} }
fn state_data(&self, _hash: &H256) -> Option<Bytes> { fn state_data(&self, _hash: &H256) -> Option<Bytes> {
unimplemented!(); None
} }
fn block_receipts(&self, _hash: &H256) -> Option<Bytes> { fn block_receipts(&self, _hash: &H256) -> Option<Bytes> {
unimplemented!(); None
} }
fn import_block(&self, bytes: Bytes) -> ImportResult { fn import_block(&self, bytes: Bytes) -> ImportResult {

6
ethcore/src/service.rs
View File

@ -124,20 +124,18 @@ impl IoHandler<NetSyncMessage> for ClientIoHandler {
} }
} }
// TODO: rewrite into something that doesn't dependent on the testing environment having a particular port ready for use.
/*
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use tests::helpers::*; use tests::helpers::*;
use util::network::*; use util::network::*;
use devtools::*;
#[test] #[test]
fn it_can_be_started() { fn it_can_be_started() {
let spec = get_test_spec(); let spec = get_test_spec();
let temp_path = RandomTempPath::new(); let temp_path = RandomTempPath::new();
let service = ClientService::start(spec, NetworkConfiguration::new(), &temp_path.as_path()); let service = ClientService::start(spec, NetworkConfiguration::new_with_port(40456), &temp_path.as_path());
assert!(service.is_ok()); assert!(service.is_ok());
} }
} }
*/

42
parity/main.rs
View File

@ -37,6 +37,7 @@ extern crate ethcore_rpc as rpc;
use std::net::{SocketAddr}; use std::net::{SocketAddr};
use std::env; use std::env;
use std::path::PathBuf;
use rlog::{LogLevelFilter}; use rlog::{LogLevelFilter};
use env_logger::LogBuilder; use env_logger::LogBuilder;
use ctrlc::CtrlC; use ctrlc::CtrlC;
@ -69,8 +70,10 @@ Options:
--no-bootstrap Don't bother trying to connect to any nodes initially. --no-bootstrap Don't bother trying to connect to any nodes initially.
--listen-address URL Specify the IP/port on which to listen for peers [default: 0.0.0.0:30304]. --listen-address URL Specify the IP/port on which to listen for peers [default: 0.0.0.0:30304].
--public-address URL Specify the IP/port on which peers may connect [default: 0.0.0.0:30304]. --public-address URL Specify the IP/port on which peers may connect.
--address URL Equivalent to --listen-address URL --public-address URL. --address URL Equivalent to --listen-address URL --public-address URL.
--peers NUM Try to manintain that many peers [default: 25].
--no-discovery Disable new peer discovery.
--upnp Use UPnP to try to figure out the correct network settings. --upnp Use UPnP to try to figure out the correct network settings.
--node-key KEY Specify node secret key as hex string. --node-key KEY Specify node secret key as hex string.
@ -95,8 +98,10 @@ struct Args {
flag_keys_path: String, flag_keys_path: String,
flag_no_bootstrap: bool, flag_no_bootstrap: bool,
flag_listen_address: String, flag_listen_address: String,
flag_public_address: String, flag_public_address: Option<String>,
flag_address: Option<String>, flag_address: Option<String>,
flag_peers: u32,
flag_no_discovery: bool,
flag_upnp: bool, flag_upnp: bool,
flag_node_key: Option<String>, flag_node_key: Option<String>,
flag_cache_pref_size: usize, flag_cache_pref_size: usize,
@ -165,7 +170,7 @@ impl Configuration {
self.args.flag_db_path.replace("$HOME", env::home_dir().unwrap().to_str().unwrap()) self.args.flag_db_path.replace("$HOME", env::home_dir().unwrap().to_str().unwrap())
} }
fn keys_path(&self) -> String { fn _keys_path(&self) -> String {
self.args.flag_keys_path.replace("$HOME", env::home_dir().unwrap().to_str().unwrap()) self.args.flag_keys_path.replace("$HOME", env::home_dir().unwrap().to_str().unwrap())
} }
@ -187,21 +192,23 @@ impl Configuration {
} }
} }
fn net_addresses(&self) -> (SocketAddr, SocketAddr) { fn net_addresses(&self) -> (Option<SocketAddr>, Option<SocketAddr>) {
let listen_address; let mut listen_address = None;
let public_address; let mut public_address = None;
match self.args.flag_address { if let Some(ref a) = self.args.flag_address {
None => { public_address = Some(SocketAddr::from_str(a.as_ref()).expect("Invalid listen/public address given with --address"));
listen_address = SocketAddr::from_str(self.args.flag_listen_address.as_ref()).expect("Invalid listen address given with --listen-address");
public_address = SocketAddr::from_str(self.args.flag_public_address.as_ref()).expect("Invalid public address given with --public-address");
}
Some(ref a) => {
public_address = SocketAddr::from_str(a.as_ref()).expect("Invalid listen/public address given with --address");
listen_address = public_address; listen_address = public_address;
} }
}; if listen_address.is_none() {
listen_address = Some(SocketAddr::from_str(self.args.flag_listen_address.as_ref()).expect("Invalid listen address given with --listen-address"));
}
if let Some(ref a) = self.args.flag_public_address {
if public_address.is_some() {
panic!("Conflicting flags: --address and --public-address");
}
public_address = Some(SocketAddr::from_str(a.as_ref()).expect("Invalid listen address given with --public-address"));
}
(listen_address, public_address) (listen_address, public_address)
} }
@ -236,6 +243,11 @@ impl Configuration {
net_settings.listen_address = listen; net_settings.listen_address = listen;
net_settings.public_address = public; net_settings.public_address = public;
net_settings.use_secret = self.args.flag_node_key.as_ref().map(|s| Secret::from_str(&s).expect("Invalid key string")); net_settings.use_secret = self.args.flag_node_key.as_ref().map(|s| Secret::from_str(&s).expect("Invalid key string"));
net_settings.discovery_enabled = !self.args.flag_no_discovery;
net_settings.ideal_peers = self.args.flag_peers;
let mut net_path = PathBuf::from(&self.path());
net_path.push("network");
net_settings.config_path = Some(net_path.to_str().unwrap().to_owned());
// Build client // Build client
let mut service = ClientService::start(spec, net_settings, &Path::new(&self.path())).unwrap(); let mut service = ClientService::start(spec, net_settings, &Path::new(&self.path())).unwrap();

8
sync/src/chain.rs
View File

@ -82,7 +82,7 @@ const RECEIPTS_PACKET: u8 = 0x10;
const NETWORK_ID: U256 = ONE_U256; //TODO: get this from parent const NETWORK_ID: U256 = ONE_U256; //TODO: get this from parent
const CONNECTION_TIMEOUT_SEC: f64 = 30f64; const CONNECTION_TIMEOUT_SEC: f64 = 10f64;
struct Header { struct Header {
/// Header data /// Header data
@ -314,7 +314,7 @@ impl ChainSync {
} }
self.peers.insert(peer_id.clone(), peer); self.peers.insert(peer_id.clone(), peer);
info!(target: "sync", "Connected {}:{}", peer_id, io.peer_info(peer_id)); debug!(target: "sync", "Connected {}:{}", peer_id, io.peer_info(peer_id));
self.sync_peer(io, peer_id, false); self.sync_peer(io, peer_id, false);
Ok(()) Ok(())
} }
@ -545,7 +545,7 @@ impl ChainSync {
pub fn on_peer_aborting(&mut self, io: &mut SyncIo, peer: PeerId) { pub fn on_peer_aborting(&mut self, io: &mut SyncIo, peer: PeerId) {
trace!(target: "sync", "== Disconnecting {}", peer); trace!(target: "sync", "== Disconnecting {}", peer);
if self.peers.contains_key(&peer) { if self.peers.contains_key(&peer) {
info!(target: "sync", "Disconnected {}", peer); debug!(target: "sync", "Disconnected {}", peer);
self.clear_peer_download(peer); self.clear_peer_download(peer);
self.peers.remove(&peer); self.peers.remove(&peer);
self.continue_sync(io); self.continue_sync(io);
@ -1179,7 +1179,7 @@ impl ChainSync {
for (peer_id, peer_number) in updated_peers { for (peer_id, peer_number) in updated_peers {
let mut peer_best = self.peers.get(&peer_id).unwrap().latest_hash.clone(); let mut peer_best = self.peers.get(&peer_id).unwrap().latest_hash.clone();
if best_number - peer_number > MAX_PEERS_PROPAGATION as BlockNumber { if best_number - peer_number > MAX_PEERS_PROPAGATION as BlockNumber {
// If we think peer is too far behind just end one latest hash // If we think peer is too far behind just send one latest hash
peer_best = last_parent.clone(); peer_best = last_parent.clone();
} }
sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &peer_best, &local_best) { sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &peer_best, &local_best) {

2
util/Cargo.toml
View File

@ -31,6 +31,8 @@ json-tests = { path = "json-tests" }
target_info = "0.1.0" target_info = "0.1.0"
igd = "0.4.2" igd = "0.4.2"
ethcore-devtools = { path = "../devtools" } ethcore-devtools = { path = "../devtools" }
libc = "0.2.7"
[features] [features]
default = [] default = []
dev = ["clippy"] dev = ["clippy"]

25
util/fdlimit/src/raise_fd_limit.rs
View File

@ -57,5 +57,28 @@ pub unsafe fn raise_fd_limit() {
} }
} }
#[cfg(not(any(target_os = "macos", target_os = "ios")))] #[cfg(any(target_os = "linux"))]
#[allow(non_camel_case_types)]
pub unsafe fn raise_fd_limit() {
use libc;
use std::io;
// Fetch the current resource limits
let mut rlim = libc::rlimit{rlim_cur: 0, rlim_max: 0};
if libc::getrlimit(libc::RLIMIT_NOFILE, &mut rlim) != 0 {
let err = io::Error::last_os_error();
panic!("raise_fd_limit: error calling getrlimit: {}", err);
}
// Set soft limit to hard imit
rlim.rlim_cur = rlim.rlim_max;
// Set our newly-increased resource limit
if libc::setrlimit(libc::RLIMIT_NOFILE, &rlim) != 0 {
let err = io::Error::last_os_error();
panic!("raise_fd_limit: error calling setrlimit: {}", err);
}
}
#[cfg(not(any(target_os = "macos", target_os = "ios", target_os = "linux")))]
pub unsafe fn raise_fd_limit() {} pub unsafe fn raise_fd_limit() {}

28
util/src/bytes.rs
View File

@ -170,32 +170,8 @@ pub trait BytesConvertable {
fn to_bytes(&self) -> Bytes { self.as_slice().to_vec() } fn to_bytes(&self) -> Bytes { self.as_slice().to_vec() }
} }
impl<'a> BytesConvertable for &'a [u8] { impl<T> BytesConvertable for T where T: AsRef<[u8]> {
fn bytes(&self) -> &[u8] { self } fn bytes(&self) -> &[u8] { self.as_ref() }
}
impl BytesConvertable for Vec<u8> {
fn bytes(&self) -> &[u8] { self }
}
impl BytesConvertable for String {
fn bytes(&self) -> &[u8] { &self.as_bytes() }
}
macro_rules! impl_bytes_convertable_for_array {
($zero: expr) => ();
($len: expr, $($idx: expr),*) => {
impl BytesConvertable for [u8; $len] {
fn bytes(&self) -> &[u8] { self }
}
impl_bytes_convertable_for_array! { $($idx),* }
}
}
// -1 at the end is not expanded
impl_bytes_convertable_for_array! {
32, 31, 30, 29, 28, 27, 26, 25, 24, 23, 22, 21, 20, 19, 18, 17, 16,
15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0, -1
} }
#[test] #[test]

13
util/src/hash.rs
View File

@ -77,12 +77,6 @@ macro_rules! impl_hash {
/// Unformatted binary data of fixed length. /// Unformatted binary data of fixed length.
pub struct $from (pub [u8; $size]); pub struct $from (pub [u8; $size]);
impl BytesConvertable for $from {
fn bytes(&self) -> &[u8] {
&self.0
}
}
impl Deref for $from { impl Deref for $from {
type Target = [u8]; type Target = [u8];
@ -92,6 +86,13 @@ macro_rules! impl_hash {
} }
} }
impl AsRef<[u8]> for $from {
#[inline]
fn as_ref(&self) -> &[u8] {
&self.0
}
}
impl DerefMut for $from { impl DerefMut for $from {
#[inline] #[inline]
fn deref_mut(&mut self) -> &mut [u8] { fn deref_mut(&mut self) -> &mut [u8] {

5
util/src/io/service.rs
View File

@ -256,6 +256,11 @@ impl<Message> Handler for IoManager<Message> where Message: Send + Clone + Sync
IoMessage::DeregisterStream { handler_id, token } => { IoMessage::DeregisterStream { handler_id, token } => {
let handler = self.handlers.get(handler_id).expect("Unknown handler id").clone(); let handler = self.handlers.get(handler_id).expect("Unknown handler id").clone();
handler.deregister_stream(token, event_loop); handler.deregister_stream(token, event_loop);
// unregister a timer associated with the token (if any)
let timer_id = token + handler_id * TOKENS_PER_HANDLER;
if let Some(timer) = self.timers.write().unwrap().remove(&timer_id) {
event_loop.clear_timeout(timer.timeout);
}
}, },
IoMessage::UpdateStreamRegistration { handler_id, token } => { IoMessage::UpdateStreamRegistration { handler_id, token } => {
let handler = self.handlers.get(handler_id).expect("Unknown handler id").clone(); let handler = self.handlers.get(handler_id).expect("Unknown handler id").clone();

15
util/src/io/worker.rs
View File

@ -44,6 +44,7 @@ pub struct Worker {
thread: Option<JoinHandle<()>>, thread: Option<JoinHandle<()>>,
wait: Arc<Condvar>, wait: Arc<Condvar>,
deleting: Arc<AtomicBool>, deleting: Arc<AtomicBool>,
wait_mutex: Arc<Mutex<()>>,
} }
impl Worker { impl Worker {
@ -61,6 +62,7 @@ impl Worker {
thread: None, thread: None,
wait: wait.clone(), wait: wait.clone(),
deleting: deleting.clone(), deleting: deleting.clone(),
wait_mutex: wait_mutex.clone(),
}; };
worker.thread = Some(thread::Builder::new().name(format!("IO Worker #{}", index)).spawn( worker.thread = Some(thread::Builder::new().name(format!("IO Worker #{}", index)).spawn(
move || { move || {
@ -77,13 +79,17 @@ impl Worker {
wait_mutex: Arc<Mutex<()>>, wait_mutex: Arc<Mutex<()>>,
deleting: Arc<AtomicBool>) deleting: Arc<AtomicBool>)
where Message: Send + Sync + Clone + 'static { where Message: Send + Sync + Clone + 'static {
while !deleting.load(AtomicOrdering::Relaxed) { loop {
{ {
let lock = wait_mutex.lock().unwrap(); let lock = wait_mutex.lock().unwrap();
let _ = wait.wait(lock).unwrap(); if deleting.load(AtomicOrdering::Acquire) {
if deleting.load(AtomicOrdering::Relaxed) {
return; return;
} }
let _ = wait.wait(lock).unwrap();
}
if deleting.load(AtomicOrdering::Acquire) {
return;
} }
while let chase_lev::Steal::Data(work) = stealer.steal() { while let chase_lev::Steal::Data(work) = stealer.steal() {
Worker::do_work(work, channel.clone()); Worker::do_work(work, channel.clone());
@ -114,7 +120,8 @@ impl Worker {
impl Drop for Worker { impl Drop for Worker {
fn drop(&mut self) { fn drop(&mut self) {
self.deleting.store(true, AtomicOrdering::Relaxed); let _ = self.wait_mutex.lock();
self.deleting.store(true, AtomicOrdering::Release);
self.wait.notify_all(); self.wait.notify_all();
let thread = mem::replace(&mut self.thread, None).unwrap(); let thread = mem::replace(&mut self.thread, None).unwrap();
thread.join().ok(); thread.join().ok();

2
util/src/lib.rs
View File

@ -106,8 +106,8 @@ extern crate serde;
#[macro_use] #[macro_use]
extern crate log as rlog; extern crate log as rlog;
extern crate igd; extern crate igd;
#[cfg(test)]
extern crate ethcore_devtools as devtools; extern crate ethcore_devtools as devtools;
extern crate libc;
pub mod standard; pub mod standard;
#[macro_use] #[macro_use]

26
util/src/network/connection.rs
View File

@ -16,6 +16,7 @@
use std::sync::Arc; use std::sync::Arc;
use std::collections::VecDeque; use std::collections::VecDeque;
use std::net::SocketAddr;
use mio::{Handler, Token, EventSet, EventLoop, PollOpt, TryRead, TryWrite}; use mio::{Handler, Token, EventSet, EventLoop, PollOpt, TryRead, TryWrite};
use mio::tcp::*; use mio::tcp::*;
use hash::*; use hash::*;
@ -159,6 +160,21 @@ impl Connection {
} }
} }
/// Get socket token
pub fn token(&self) -> StreamToken {
self.token
}
/// Replace socket token
pub fn set_token(&mut self, token: StreamToken) {
self.token = token;
}
/// Get remote peer address
pub fn remote_addr(&self) -> io::Result<SocketAddr> {
self.socket.peer_addr()
}
/// Register this connection with the IO event loop. /// Register this connection with the IO event loop.
pub fn register_socket<Host: Handler>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> io::Result<()> { pub fn register_socket<Host: Handler>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> io::Result<()> {
trace!(target: "net", "connection register; token={:?}", reg); trace!(target: "net", "connection register; token={:?}", reg);
@ -238,6 +254,16 @@ impl EncryptedConnection {
self.connection.token self.connection.token
} }
/// Replace socket token
pub fn set_token(&mut self, token: StreamToken) {
self.connection.set_token(token);
}
/// Get remote peer address
pub fn remote_addr(&self) -> io::Result<SocketAddr> {
self.connection.remote_addr()
}
/// Create an encrypted connection out of the handshake. Consumes a handshake object. /// Create an encrypted connection out of the handshake. Consumes a handshake object.
pub fn new(mut handshake: Handshake) -> Result<EncryptedConnection, UtilError> { pub fn new(mut handshake: Handshake) -> Result<EncryptedConnection, UtilError> {
let shared = try!(crypto::ecdh::agree(handshake.ecdhe.secret(), &handshake.remote_public)); let shared = try!(crypto::ecdh::agree(handshake.ecdhe.secret(), &handshake.remote_public));

586
util/src/network/discovery.rs
View File

@ -14,115 +14,180 @@
// You should have received a copy of the GNU General Public License // You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>. // along with Parity. If not, see <http://www.gnu.org/licenses/>.
// This module is a work in progress use bytes::Bytes;
use std::net::SocketAddr;
#![allow(dead_code)] //TODO: remove this after everything is done use std::collections::{HashSet, HashMap, BTreeMap, VecDeque};
use std::mem;
use std::collections::{HashSet, BTreeMap}; use std::cmp;
use std::cell::{RefCell};
use std::ops::{DerefMut};
use mio::*; use mio::*;
use mio::udp::*; use mio::udp::*;
use sha3::*;
use time;
use hash::*; use hash::*;
use sha3::Hashable;
use crypto::*; use crypto::*;
use network::node::*; use rlp::*;
use network::node_table::*;
use network::error::NetworkError;
use io::StreamToken;
const ADDRESS_BYTES_SIZE: u32 = 32; ///< Size of address type in bytes. use network::PROTOCOL_VERSION;
const ADDRESS_BITS: u32 = 8 * ADDRESS_BYTES_SIZE; ///< Denoted by n in [Kademlia].
const NODE_BINS: u32 = ADDRESS_BITS - 1; ///< Size of m_state (excludes root, which is us). const ADDRESS_BYTES_SIZE: u32 = 32; // Size of address type in bytes.
const DISCOVERY_MAX_STEPS: u16 = 8; ///< Max iterations of discovery. (discover) const ADDRESS_BITS: u32 = 8 * ADDRESS_BYTES_SIZE; // Denoted by n in [Kademlia].
const BUCKET_SIZE: u32 = 16; ///< Denoted by k in [Kademlia]. Number of nodes stored in each bucket. const NODE_BINS: u32 = ADDRESS_BITS - 1; // Size of m_state (excludes root, which is us).
const ALPHA: usize = 3; ///< Denoted by \alpha in [Kademlia]. Number of concurrent FindNode requests. const DISCOVERY_MAX_STEPS: u16 = 8; // Max iterations of discovery. (discover)
const BUCKET_SIZE: usize = 16; // Denoted by k in [Kademlia]. Number of nodes stored in each bucket.
const ALPHA: usize = 3; // Denoted by \alpha in [Kademlia]. Number of concurrent FindNode requests.
const MAX_DATAGRAM_SIZE: usize = 1280;
const PACKET_PING: u8 = 1;
const PACKET_PONG: u8 = 2;
const PACKET_FIND_NODE: u8 = 3;
const PACKET_NEIGHBOURS: u8 = 4;
const PING_TIMEOUT_MS: u64 = 300;
#[derive(Clone, Debug)]
pub struct NodeEntry {
pub id: NodeId,
pub endpoint: NodeEndpoint,
}
pub struct BucketEntry {
pub address: NodeEntry,
pub timeout: Option<u64>,
}
struct NodeBucket { struct NodeBucket {
distance: u32, nodes: VecDeque<BucketEntry>, //sorted by last active
nodes: Vec<NodeId>
} }
impl NodeBucket { impl NodeBucket {
fn new(distance: u32) -> NodeBucket { fn new() -> NodeBucket {
NodeBucket { NodeBucket {
distance: distance, nodes: VecDeque::new()
nodes: Vec::new()
} }
} }
} }
struct Discovery { struct Datagramm {
payload: Bytes,
address: SocketAddr,
}
pub struct Discovery {
id: NodeId, id: NodeId,
secret: Secret,
public_endpoint: NodeEndpoint,
udp_socket: UdpSocket,
token: StreamToken,
discovery_round: u16, discovery_round: u16,
discovery_id: NodeId, discovery_id: NodeId,
discovery_nodes: HashSet<NodeId>, discovery_nodes: HashSet<NodeId>,
node_buckets: Vec<NodeBucket>, node_buckets: Vec<NodeBucket>,
send_queue: VecDeque<Datagramm>
} }
struct FindNodePacket; pub struct TableUpdates {
pub added: HashMap<NodeId, NodeEntry>,
impl FindNodePacket { pub removed: HashSet<NodeId>,
fn new(_endpoint: &NodeEndpoint, _id: &NodeId) -> FindNodePacket {
FindNodePacket
}
fn sign(&mut self, _secret: &Secret) {
}
fn send(& self, _socket: &mut UdpSocket) {
}
} }
impl Discovery { impl Discovery {
pub fn new(id: &NodeId) -> Discovery { pub fn new(key: &KeyPair, listen: SocketAddr, public: NodeEndpoint, token: StreamToken) -> Discovery {
let socket = UdpSocket::bound(&listen).expect("Error binding UDP socket");
Discovery { Discovery {
id: id.clone(), id: key.public().clone(),
secret: key.secret().clone(),
public_endpoint: public,
token: token,
discovery_round: 0, discovery_round: 0,
discovery_id: NodeId::new(), discovery_id: NodeId::new(),
discovery_nodes: HashSet::new(), discovery_nodes: HashSet::new(),
node_buckets: (0..NODE_BINS).map(NodeBucket::new).collect(), node_buckets: (0..NODE_BINS).map(|_| NodeBucket::new()).collect(),
udp_socket: socket,
send_queue: VecDeque::new(),
} }
} }
pub fn add_node(&mut self, id: &NodeId) { /// Add a new node to discovery table. Pings the node.
self.node_buckets[Discovery::distance(&self.id, &id) as usize].nodes.push(id.clone()); pub fn add_node(&mut self, e: NodeEntry) {
let endpoint = e.endpoint.clone();
self.update_node(e);
self.ping(&endpoint);
} }
fn start_node_discovery<Host:Handler>(&mut self, event_loop: &mut EventLoop<Host>) { /// Add a list of known nodes to the table.
pub fn init_node_list(&mut self, mut nodes: Vec<NodeEntry>) {
for n in nodes.drain(..) {
self.update_node(n);
}
}
fn update_node(&mut self, e: NodeEntry) {
trace!(target: "discovery", "Inserting {:?}", &e);
let ping = {
let mut bucket = self.node_buckets.get_mut(Discovery::distance(&self.id, &e.id) as usize).unwrap();
let updated = if let Some(node) = bucket.nodes.iter_mut().find(|n| n.address.id == e.id) {
node.address = e.clone();
node.timeout = None;
true
} else { false };
if !updated {
bucket.nodes.push_front(BucketEntry { address: e, timeout: None });
}
if bucket.nodes.len() > BUCKET_SIZE {
//ping least active node
bucket.nodes.back_mut().unwrap().timeout = Some(time::precise_time_ns());
Some(bucket.nodes.back().unwrap().address.endpoint.clone())
} else { None }
};
if let Some(endpoint) = ping {
self.ping(&endpoint);
}
}
fn clear_ping(&mut self, id: &NodeId) {
let mut bucket = self.node_buckets.get_mut(Discovery::distance(&self.id, &id) as usize).unwrap();
if let Some(node) = bucket.nodes.iter_mut().find(|n| &n.address.id == id) {
node.timeout = None;
}
}
fn start(&mut self) {
trace!(target: "discovery", "Starting discovery");
self.discovery_round = 0; self.discovery_round = 0;
self.discovery_id.randomize(); self.discovery_id.randomize(); //TODO: use cryptographic nonce
self.discovery_nodes.clear(); self.discovery_nodes.clear();
self.discover(event_loop);
} }
fn discover<Host:Handler>(&mut self, event_loop: &mut EventLoop<Host>) { fn discover(&mut self) {
if self.discovery_round == DISCOVERY_MAX_STEPS if self.discovery_round == DISCOVERY_MAX_STEPS {
{
debug!("Restarting discovery");
self.start_node_discovery(event_loop);
return; return;
} }
trace!(target: "discovery", "Starting round {:?}", self.discovery_round);
let mut tried_count = 0; let mut tried_count = 0;
{ {
let nearest = Discovery::nearest_node_entries(&self.id, &self.discovery_id, &self.node_buckets).into_iter(); let nearest = Discovery::nearest_node_entries(&self.discovery_id, &self.node_buckets).into_iter();
let nodes = RefCell::new(&mut self.discovery_nodes); let nearest = nearest.filter(|x| !self.discovery_nodes.contains(&x.id)).take(ALPHA).collect::<Vec<_>>();
let nearest = nearest.filter(|x| nodes.borrow().contains(&x)).take(ALPHA);
for r in nearest { for r in nearest {
//let mut p = FindNodePacket::new(&r.endpoint, &self.discovery_id); let rlp = encode(&(&[self.discovery_id.clone()][..]));
//p.sign(&self.secret); self.send_packet(PACKET_FIND_NODE, &r.endpoint.udp_address(), &rlp);
//p.send(&mut self.udp_socket); self.discovery_nodes.insert(r.id.clone());
let mut borrowed = nodes.borrow_mut();
borrowed.deref_mut().insert(r.clone());
tried_count += 1; tried_count += 1;
trace!(target: "discovery", "Sent FindNode to {:?}", &r.endpoint);
} }
} }
if tried_count == 0 if tried_count == 0 {
{ trace!(target: "discovery", "Completing discovery");
debug!("Restarting discovery"); self.discovery_round = DISCOVERY_MAX_STEPS;
self.start_node_discovery(event_loop); self.discovery_nodes.clear();
return; return;
} }
self.discovery_round += 1; self.discovery_round += 1;
//event_loop.timeout_ms(Token(NODETABLE_DISCOVERY), 1200).unwrap();
} }
fn distance(a: &NodeId, b: &NodeId) -> u32 { fn distance(a: &NodeId, b: &NodeId) -> u32 {
@ -138,86 +203,353 @@ impl Discovery {
ret ret
} }
#[cfg_attr(feature="dev", allow(cyclomatic_complexity))] fn ping(&mut self, node: &NodeEndpoint) {
fn nearest_node_entries<'b>(source: &NodeId, target: &NodeId, buckets: &'b [NodeBucket]) -> Vec<&'b NodeId> let mut rlp = RlpStream::new_list(3);
{ rlp.append(&PROTOCOL_VERSION);
// send ALPHA FindNode packets to nodes we know, closest to target self.public_endpoint.to_rlp_list(&mut rlp);
const LAST_BIN: u32 = NODE_BINS - 1; node.to_rlp_list(&mut rlp);
let mut head = Discovery::distance(source, target); trace!(target: "discovery", "Sent Ping to {:?}", &node);
let mut tail = if head == 0 { LAST_BIN } else { (head - 1) % NODE_BINS }; self.send_packet(PACKET_PING, &node.udp_address(), &rlp.drain());
}
let mut found: BTreeMap<u32, Vec<&'b NodeId>> = BTreeMap::new(); fn send_packet(&mut self, packet_id: u8, address: &SocketAddr, payload: &[u8]) {
let mut rlp = RlpStream::new();
rlp.append_raw(&[packet_id], 1);
let source = Rlp::new(payload);
rlp.begin_list(source.item_count() + 1);
for i in 0 .. source.item_count() {
rlp.append_raw(source.at(i).as_raw(), 1);
}
let timestamp = time::get_time().sec as u32 + 60;
rlp.append(&timestamp);
let bytes = rlp.drain();
let hash = bytes.as_ref().sha3();
let signature = match ec::sign(&self.secret, &hash) {
Ok(s) => s,
Err(_) => {
warn!("Error signing UDP packet");
return;
}
};
let mut packet = Bytes::with_capacity(bytes.len() + 32 + 65);
packet.extend(hash.iter());
packet.extend(signature.iter());
packet.extend(bytes.iter());
let signed_hash = (&packet[32..]).sha3();
packet[0..32].clone_from_slice(&signed_hash);
self.send_to(packet, address.clone());
}
#[cfg_attr(feature="dev", allow(map_clone))]
fn nearest_node_entries(target: &NodeId, buckets: &[NodeBucket]) -> Vec<NodeEntry> {
let mut found: BTreeMap<u32, Vec<&NodeEntry>> = BTreeMap::new();
let mut count = 0; let mut count = 0;
// if d is 0, then we roll look forward, if last, we reverse, else, spread from d // Sort nodes by distance to target
if head > 1 && tail != LAST_BIN { for bucket in buckets {
while head != tail && head < NODE_BINS && count < BUCKET_SIZE for node in &bucket.nodes {
{ let distance = Discovery::distance(target, &node.address.id);
for n in &buckets[head as usize].nodes found.entry(distance).or_insert_with(Vec::new).push(&node.address);
{ if count == BUCKET_SIZE {
if count < BUCKET_SIZE { // delete the most distant element
count += 1; let remove = {
found.entry(Discovery::distance(target, &n)).or_insert_with(Vec::new).push(n); let (_, last) = found.iter_mut().next_back().unwrap();
last.pop();
last.is_empty()
};
if remove {
found.remove(&distance);
}
} }
else { else {
break;
}
}
if count < BUCKET_SIZE && tail != 0 {
for n in &buckets[tail as usize].nodes {
if count < BUCKET_SIZE {
count += 1; count += 1;
found.entry(Discovery::distance(target, &n)).or_insert_with(Vec::new).push(n);
}
else {
break;
} }
} }
} }
head += 1; let mut ret:Vec<NodeEntry> = Vec::new();
if tail > 0 {
tail -= 1;
}
}
}
else if head < 2 {
while head < NODE_BINS && count < BUCKET_SIZE {
for n in &buckets[head as usize].nodes {
if count < BUCKET_SIZE {
count += 1;
found.entry(Discovery::distance(target, &n)).or_insert_with(Vec::new).push(n);
}
else {
break;
}
}
head += 1;
}
}
else {
while tail > 0 && count < BUCKET_SIZE {
for n in &buckets[tail as usize].nodes {
if count < BUCKET_SIZE {
count += 1;
found.entry(Discovery::distance(target, &n)).or_insert_with(Vec::new).push(n);
}
else {
break;
}
}
tail -= 1;
}
}
let mut ret:Vec<&NodeId> = Vec::new();
for nodes in found.values() { for nodes in found.values() {
for n in nodes { ret.extend(nodes.iter().map(|&n| n.clone()));
if ret.len() < BUCKET_SIZE as usize /* && n->endpoint && n->endpoint.isAllowed() */ {
ret.push(n);
}
}
} }
ret ret
} }
pub fn writable(&mut self) {
if self.send_queue.is_empty() {
return;
}
while !self.send_queue.is_empty() {
let data = self.send_queue.pop_front().unwrap();
match self.udp_socket.send_to(&data.payload, &data.address) {
Ok(Some(size)) if size == data.payload.len() => {
},
Ok(Some(_)) => {
warn!("UDP sent incomplete datagramm");
},
Ok(None) => {
self.send_queue.push_front(data);
return;
}
Err(e) => {
warn!("UDP send error: {:?}, address: {:?}", e, &data.address);
return;
}
}
}
}
fn send_to(&mut self, payload: Bytes, address: SocketAddr) {
self.send_queue.push_back(Datagramm { payload: payload, address: address });
}
pub fn readable(&mut self) -> Option<TableUpdates> {
let mut buf: [u8; MAX_DATAGRAM_SIZE] = unsafe { mem::uninitialized() };
match self.udp_socket.recv_from(&mut buf) {
Ok(Some((len, address))) => self.on_packet(&buf[0..len], address).unwrap_or_else(|e| {
debug!("Error processing UDP packet: {:?}", e);
None
}),
Ok(_) => None,
Err(e) => {
warn!("Error reading UPD socket: {:?}", e);
None
}
}
}
fn on_packet(&mut self, packet: &[u8], from: SocketAddr) -> Result<Option<TableUpdates>, NetworkError> {
// validate packet
if packet.len() < 32 + 65 + 4 + 1 {
return Err(NetworkError::BadProtocol);
}
let hash_signed = (&packet[32..]).sha3();
if hash_signed[..] != packet[0..32] {
return Err(NetworkError::BadProtocol);
}
let signed = &packet[(32 + 65)..];
let signature = Signature::from_slice(&packet[32..(32 + 65)]);
let node_id = try!(ec::recover(&signature, &signed.sha3()));
let packet_id = signed[0];
let rlp = UntrustedRlp::new(&signed[1..]);
match packet_id {
PACKET_PING => self.on_ping(&rlp, &node_id, &from),
PACKET_PONG => self.on_pong(&rlp, &node_id, &from),
PACKET_FIND_NODE => self.on_find_node(&rlp, &node_id, &from),
PACKET_NEIGHBOURS => self.on_neighbours(&rlp, &node_id, &from),
_ => {
debug!("Unknown UDP packet: {}", packet_id);
Ok(None)
}
}
}
fn on_ping(&mut self, rlp: &UntrustedRlp, node: &NodeId, from: &SocketAddr) -> Result<Option<TableUpdates>, NetworkError> {
trace!(target: "discovery", "Got Ping from {:?}", &from);
let version: u32 = try!(rlp.val_at(0));
if version != PROTOCOL_VERSION {
debug!(target: "discovery", "Unexpected protocol version: {}", version);
return Err(NetworkError::BadProtocol);
}
let source = try!(NodeEndpoint::from_rlp(&try!(rlp.at(1))));
let dest = try!(NodeEndpoint::from_rlp(&try!(rlp.at(2))));
let timestamp: u64 = try!(rlp.val_at(3));
if timestamp < time::get_time().sec as u64{
debug!(target: "discovery", "Expired ping");
return Err(NetworkError::Expired);
}
let mut added_map = HashMap::new();
let entry = NodeEntry { id: node.clone(), endpoint: source.clone() };
if !entry.endpoint.is_valid() || !entry.endpoint.is_global() {
debug!(target: "discovery", "Got bad address: {:?}", entry);
}
else {
self.update_node(entry.clone());
added_map.insert(node.clone(), entry);
}
let hash = rlp.as_raw().sha3();
let mut response = RlpStream::new_list(2);
dest.to_rlp_list(&mut response);
response.append(&hash);
self.send_packet(PACKET_PONG, from, &response.drain());
Ok(Some(TableUpdates { added: added_map, removed: HashSet::new() }))
}
fn on_pong(&mut self, rlp: &UntrustedRlp, node: &NodeId, from: &SocketAddr) -> Result<Option<TableUpdates>, NetworkError> {
trace!(target: "discovery", "Got Pong from {:?}", &from);
// TODO: validate pong packet
let dest = try!(NodeEndpoint::from_rlp(&try!(rlp.at(0))));
let timestamp: u64 = try!(rlp.val_at(2));
if timestamp < time::get_time().sec as u64 {
return Err(NetworkError::Expired);
}
let mut entry = NodeEntry { id: node.clone(), endpoint: dest };
if !entry.endpoint.is_valid() {
debug!(target: "discovery", "Bad address: {:?}", entry);
entry.endpoint.address = from.clone();
}
self.clear_ping(node);
let mut added_map = HashMap::new();
added_map.insert(node.clone(), entry);
Ok(None)
}
fn on_find_node(&mut self, rlp: &UntrustedRlp, _node: &NodeId, from: &SocketAddr) -> Result<Option<TableUpdates>, NetworkError> {
trace!(target: "discovery", "Got FindNode from {:?}", &from);
let target: NodeId = try!(rlp.val_at(0));
let timestamp: u64 = try!(rlp.val_at(1));
if timestamp < time::get_time().sec as u64 {
return Err(NetworkError::Expired);
}
let limit = (MAX_DATAGRAM_SIZE - 109) / 90;
let nearest = Discovery::nearest_node_entries(&target, &self.node_buckets);
if nearest.is_empty() {
return Ok(None);
}
let mut rlp = RlpStream::new_list(1);
rlp.begin_list(cmp::min(limit, nearest.len()));
for n in 0 .. nearest.len() {
rlp.begin_list(4);
nearest[n].endpoint.to_rlp(&mut rlp);
rlp.append(&nearest[n].id);
if (n + 1) % limit == 0 || n == nearest.len() - 1 {
self.send_packet(PACKET_NEIGHBOURS, &from, &rlp.drain());
trace!(target: "discovery", "Sent {} Neighbours to {:?}", n, &from);
rlp = RlpStream::new_list(1);
rlp.begin_list(cmp::min(limit, nearest.len() - n));
}
}
Ok(None)
}
fn on_neighbours(&mut self, rlp: &UntrustedRlp, _node: &NodeId, from: &SocketAddr) -> Result<Option<TableUpdates>, NetworkError> {
// TODO: validate packet
let mut added = HashMap::new();
trace!(target: "discovery", "Got {} Neighbours from {:?}", try!(rlp.at(0)).item_count(), &from);
for r in try!(rlp.at(0)).iter() {
let endpoint = try!(NodeEndpoint::from_rlp(&r));
if !endpoint.is_valid() {
debug!(target: "discovery", "Bad address: {:?}", endpoint);
continue;
}
let node_id: NodeId = try!(r.val_at(3));
if node_id == self.id {
continue;
}
let entry = NodeEntry { id: node_id.clone(), endpoint: endpoint };
added.insert(node_id, entry.clone());
self.ping(&entry.endpoint);
self.update_node(entry);
}
Ok(Some(TableUpdates { added: added, removed: HashSet::new() }))
}
fn check_expired(&mut self, force: bool) -> HashSet<NodeId> {
let now = time::precise_time_ns();
let mut removed: HashSet<NodeId> = HashSet::new();
for bucket in &mut self.node_buckets {
bucket.nodes.retain(|node| {
if let Some(timeout) = node.timeout {
if !force && now - timeout < PING_TIMEOUT_MS * 1000_0000 {
true
}
else {
trace!(target: "discovery", "Removed expired node {:?}", &node.address);
removed.insert(node.address.id.clone());
false
}
} else { true }
});
}
removed
}
pub fn round(&mut self) -> Option<TableUpdates> {
let removed = self.check_expired(false);
self.discover();
if !removed.is_empty() {
Some(TableUpdates { added: HashMap::new(), removed: removed })
} else { None }
}
pub fn refresh(&mut self) {
self.start();
}
pub fn register_socket<Host:Handler>(&self, event_loop: &mut EventLoop<Host>) -> Result<(), NetworkError> {
event_loop.register(&self.udp_socket, Token(self.token), EventSet::all(), PollOpt::edge()).expect("Error registering UDP socket");
Ok(())
}
pub fn update_registration<Host:Handler>(&self, event_loop: &mut EventLoop<Host>) -> Result<(), NetworkError> {
let mut registration = EventSet::readable();
if !self.send_queue.is_empty() {
registration = registration | EventSet::writable();
}
event_loop.reregister(&self.udp_socket, Token(self.token), registration, PollOpt::edge()).expect("Error reregistering UDP socket");
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use hash::*;
use std::net::*;
use network::node_table::*;
use crypto::KeyPair;
use std::str::FromStr;
#[test]
fn discovery() {
let key1 = KeyPair::create().unwrap();
let key2 = KeyPair::create().unwrap();
let ep1 = NodeEndpoint { address: SocketAddr::from_str("127.0.0.1:40444").unwrap(), udp_port: 40444 };
let ep2 = NodeEndpoint { address: SocketAddr::from_str("127.0.0.1:40445").unwrap(), udp_port: 40445 };
let mut discovery1 = Discovery::new(&key1, ep1.address.clone(), ep1.clone(), 0);
let mut discovery2 = Discovery::new(&key2, ep2.address.clone(), ep2.clone(), 0);
let node1 = Node::from_str("enode://a979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c@127.0.0.1:7770").unwrap();
let node2 = Node::from_str("enode://b979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c@127.0.0.1:7771").unwrap();
discovery1.add_node(NodeEntry { id: node1.id.clone(), endpoint: node1.endpoint.clone() });
discovery1.add_node(NodeEntry { id: node2.id.clone(), endpoint: node2.endpoint.clone() });
discovery2.add_node(NodeEntry { id: key1.public().clone(), endpoint: ep1.clone() });
discovery2.refresh();
for _ in 0 .. 10 {
while !discovery1.send_queue.is_empty() {
let datagramm = discovery1.send_queue.pop_front().unwrap();
if datagramm.address == ep2.address {
discovery2.on_packet(&datagramm.payload, ep1.address.clone()).ok();
}
}
while !discovery2.send_queue.is_empty() {
let datagramm = discovery2.send_queue.pop_front().unwrap();
if datagramm.address == ep1.address {
discovery1.on_packet(&datagramm.payload, ep2.address.clone()).ok();
}
}
discovery2.round();
}
assert_eq!(Discovery::nearest_node_entries(&NodeId::new(), &discovery2.node_buckets).len(), 3)
}
#[test]
fn removes_expired() {
let key = KeyPair::create().unwrap();
let ep = NodeEndpoint { address: SocketAddr::from_str("127.0.0.1:40446").unwrap(), udp_port: 40444 };
let mut discovery = Discovery::new(&key, ep.address.clone(), ep.clone(), 0);
for _ in 0..1200 {
discovery.add_node(NodeEntry { id: NodeId::random(), endpoint: ep.clone() });
}
assert!(Discovery::nearest_node_entries(&NodeId::new(), &discovery.node_buckets).len() <= 16);
let removed = discovery.check_expired(true).len();
assert!(removed > 0);
}
} }

69
util/src/network/error.rs
View File

@ -15,23 +15,45 @@
// along with Parity. If not, see <http://www.gnu.org/licenses/>. // along with Parity. If not, see <http://www.gnu.org/licenses/>.
use io::IoError; use io::IoError;
use crypto::CryptoError;
use rlp::*; use rlp::*;
#[derive(Debug, Copy, Clone)] #[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum DisconnectReason pub enum DisconnectReason
{ {
DisconnectRequested, DisconnectRequested,
_TCPError, TCPError,
_BadProtocol, BadProtocol,
UselessPeer, UselessPeer,
_TooManyPeers, TooManyPeers,
_DuplicatePeer, DuplicatePeer,
_IncompatibleProtocol, IncompatibleProtocol,
_NullIdentity, NullIdentity,
_ClientQuit, ClientQuit,
_UnexpectedIdentity, UnexpectedIdentity,
_LocalIdentity, LocalIdentity,
PingTimeout, PingTimeout,
Unknown,
}
impl DisconnectReason {
pub fn from_u8(n: u8) -> DisconnectReason {
match n {
0 => DisconnectReason::DisconnectRequested,
1 => DisconnectReason::TCPError,
2 => DisconnectReason::BadProtocol,
3 => DisconnectReason::UselessPeer,
4 => DisconnectReason::TooManyPeers,
5 => DisconnectReason::DuplicatePeer,
6 => DisconnectReason::IncompatibleProtocol,
7 => DisconnectReason::NullIdentity,
8 => DisconnectReason::ClientQuit,
9 => DisconnectReason::UnexpectedIdentity,
10 => DisconnectReason::LocalIdentity,
11 => DisconnectReason::PingTimeout,
_ => DisconnectReason::Unknown,
}
}
} }
#[derive(Debug)] #[derive(Debug)]
@ -41,6 +63,8 @@ pub enum NetworkError {
Auth, Auth,
/// Unrecognised protocol. /// Unrecognised protocol.
BadProtocol, BadProtocol,
/// Message expired.
Expired,
/// Peer not found. /// Peer not found.
PeerNotFound, PeerNotFound,
/// Peer is diconnected. /// Peer is diconnected.
@ -61,3 +85,28 @@ impl From<IoError> for NetworkError {
} }
} }
impl From<CryptoError> for NetworkError {
fn from(_err: CryptoError) -> NetworkError {
NetworkError::Auth
}
}
#[test]
fn test_errors() {
assert_eq!(DisconnectReason::ClientQuit, DisconnectReason::from_u8(8));
let mut r = DisconnectReason::DisconnectRequested;
for i in 0 .. 20 {
r = DisconnectReason::from_u8(i);
}
assert_eq!(DisconnectReason::Unknown, r);
match <NetworkError as From<DecoderError>>::from(DecoderError::RlpIsTooBig) {
NetworkError::Auth => {},
_ => panic!("Unexpeceted error"),
}
match <NetworkError as From<CryptoError>>::from(CryptoError::InvalidSecret) {
NetworkError::Auth => {},
_ => panic!("Unexpeceted error"),
}
}

14
util/src/network/handshake.rs
View File

@ -24,7 +24,7 @@ use crypto::*;
use crypto; use crypto;
use network::connection::{Connection}; use network::connection::{Connection};
use network::host::{HostInfo}; use network::host::{HostInfo};
use network::node::NodeId; use network::node_table::NodeId;
use error::*; use error::*;
use network::error::NetworkError; use network::error::NetworkError;
use network::stats::NetworkStats; use network::stats::NetworkStats;
@ -68,7 +68,7 @@ pub struct Handshake {
const AUTH_PACKET_SIZE: usize = 307; const AUTH_PACKET_SIZE: usize = 307;
const ACK_PACKET_SIZE: usize = 210; const ACK_PACKET_SIZE: usize = 210;
const HANDSHAKE_TIMEOUT: u64 = 30000; const HANDSHAKE_TIMEOUT: u64 = 5000;
impl Handshake { impl Handshake {
/// Create a new handshake object /// Create a new handshake object
@ -87,6 +87,16 @@ impl Handshake {
}) })
} }
/// Get id of the remote node if known
pub fn id(&self) -> &NodeId {
&self.id
}
/// Get stream token id
pub fn token(&self) -> StreamToken {
self.connection.token()
}
/// Start a handhsake /// Start a handhsake
pub fn start<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo, originated: bool) -> Result<(), UtilError> where Message: Send + Clone{ pub fn start<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo, originated: bool) -> Result<(), UtilError> where Message: Send + Clone{
self.originated = originated; self.originated = originated;

713
util/src/network/host.rs
View File

File diff suppressed because it is too large Load Diff

268
util/src/network/ip_utils.rs Normal file
View File

@ -0,0 +1,268 @@
// 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/>.
// Based on original work by David Levy https://raw.githubusercontent.com/dlevy47/rust-interfaces
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6};
use std::io;
use igd::{PortMappingProtocol, search_gateway_from_timeout};
use std::time::Duration;
use network::node_table::{NodeEndpoint};
pub enum IpAddr{
V4(Ipv4Addr),
V6(Ipv6Addr),
}
/// Socket address extension for rustc beta. To be replaces with now unstable API
pub trait SocketAddrExt {
/// Returns true for the special 'unspecified' address 0.0.0.0.
fn is_unspecified_s(&self) -> bool;
/// Returns true if the address appears to be globally routable.
fn is_global_s(&self) -> bool;
}
impl SocketAddrExt for Ipv4Addr {
fn is_unspecified_s(&self) -> bool {
self.octets() == [0, 0, 0, 0]
}
fn is_global_s(&self) -> bool {
!self.is_private() && !self.is_loopback() && !self.is_link_local() &&
!self.is_broadcast() && !self.is_documentation()
}
}
impl SocketAddrExt for Ipv6Addr {
fn is_unspecified_s(&self) -> bool {
self.segments() == [0, 0, 0, 0, 0, 0, 0, 0]
}
fn is_global_s(&self) -> bool {
if self.is_multicast() {
self.segments()[0] & 0x000f == 14
} else {
!self.is_loopback() && !((self.segments()[0] & 0xffc0) == 0xfe80) &&
!((self.segments()[0] & 0xffc0) == 0xfec0) && !((self.segments()[0] & 0xfe00) == 0xfc00)
}
}
}
#[cfg(not(windows))]
mod getinterfaces {
use std::{mem, io, ptr};
use libc::{AF_INET, AF_INET6};
use libc::{getifaddrs, freeifaddrs, ifaddrs, sockaddr, sockaddr_in, sockaddr_in6};
use std::net::{Ipv4Addr, Ipv6Addr};
use super::IpAddr;
fn convert_sockaddr (sa: *mut sockaddr) -> Option<IpAddr> {
if sa == ptr::null_mut() { return None; }
let (addr, _) = match unsafe { *sa }.sa_family as i32 {
AF_INET => {
let sa: *const sockaddr_in = unsafe { mem::transmute(sa) };
let sa = & unsafe { *sa };
let (addr, port) = (sa.sin_addr.s_addr, sa.sin_port);
(IpAddr::V4(Ipv4Addr::new(
(addr & 0x000000FF) as u8,
((addr & 0x0000FF00) >> 8) as u8,
((addr & 0x00FF0000) >> 16) as u8,
((addr & 0xFF000000) >> 24) as u8)),
port)
},
AF_INET6 => {
let sa: *const sockaddr_in6 = unsafe { mem::transmute(sa) };
let sa = & unsafe { *sa };
let (addr, port) = (sa.sin6_addr.s6_addr, sa.sin6_port);
let addr: [u16; 8] = unsafe { mem::transmute(addr) };
(IpAddr::V6(Ipv6Addr::new(
addr[0],
addr[1],
addr[2],
addr[3],
addr[4],
addr[5],
addr[6],
addr[7])),
port)
},
_ => return None,
};
Some(addr)
}
fn convert_ifaddrs (ifa: *mut ifaddrs) -> Option<IpAddr> {
let ifa = unsafe { &mut *ifa };
convert_sockaddr(ifa.ifa_addr)
}
pub fn get_all() -> io::Result<Vec<IpAddr>> {
let mut ifap: *mut ifaddrs = unsafe { mem::zeroed() };
if unsafe { getifaddrs(&mut ifap as *mut _) } != 0 {
return Err(io::Error::last_os_error());
}
let mut ret = Vec::new();
let mut cur: *mut ifaddrs = ifap;
while cur != ptr::null_mut() {
if let Some(ip_addr) = convert_ifaddrs(cur) {
ret.push(ip_addr);
}
//TODO: do something else maybe?
cur = unsafe { (*cur).ifa_next };
}
unsafe { freeifaddrs(ifap) };
Ok(ret)
}
}
#[cfg(not(windows))]
fn get_if_addrs() -> io::Result<Vec<IpAddr>> {
getinterfaces::get_all()
}
#[cfg(windows)]
fn get_if_addrs() -> io::Result<Vec<IpAddr>> {
Ok(Vec::new())
}
/// Select the best available public address
pub fn select_public_address(port: u16) -> SocketAddr {
match get_if_addrs() {
Ok(list) => {
//prefer IPV4 bindings
for addr in &list { //TODO: use better criteria than just the first in the list
match *addr {
IpAddr::V4(a) if !a.is_unspecified_s() && !a.is_loopback() && !a.is_link_local() => {
return SocketAddr::V4(SocketAddrV4::new(a, port));
},
_ => {},
}
}
for addr in list {
match addr {
IpAddr::V6(a) if !a.is_unspecified_s() && !a.is_loopback() => {
return SocketAddr::V6(SocketAddrV6::new(a, port, 0, 0));
},
_ => {},
}
}
},
Err(e) => debug!("Error listing public interfaces: {:?}", e)
}
SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), port))
}
pub fn map_external_address(local: &NodeEndpoint) -> Option<NodeEndpoint> {
if let SocketAddr::V4(ref local_addr) = local.address {
match search_gateway_from_timeout(local_addr.ip().clone(), Duration::new(5, 0)) {
Err(ref err) => debug!("Gateway search error: {}", err),
Ok(gateway) => {
match gateway.get_external_ip() {
Err(ref err) => {
debug!("IP request error: {}", err);
},
Ok(external_addr) => {
match gateway.add_any_port(PortMappingProtocol::TCP, SocketAddrV4::new(local_addr.ip().clone(), local_addr.port()), 0, "Parity Node/TCP") {
Err(ref err) => {
debug!("Port mapping error: {}", err);
},
Ok(tcp_port) => {
match gateway.add_any_port(PortMappingProtocol::UDP, SocketAddrV4::new(local_addr.ip().clone(), local.udp_port), 0, "Parity Node/UDP") {
Err(ref err) => {
debug!("Port mapping error: {}", err);
},
Ok(udp_port) => {
return Some(NodeEndpoint { address: SocketAddr::V4(SocketAddrV4::new(external_addr, tcp_port)), udp_port: udp_port });
},
}
},
}
},
}
},
}
}
None
}
#[test]
fn can_select_public_address() {
let pub_address = select_public_address(40477);
assert!(pub_address.port() == 40477);
}
#[test]
fn can_map_external_address_or_fail() {
let pub_address = select_public_address(40478);
let _ = map_external_address(&NodeEndpoint { address: pub_address, udp_port: 40478 });
}
#[test]
fn ipv4_properties() {
fn check(octets: &[u8; 4], unspec: bool, loopback: bool,
private: bool, link_local: bool, global: bool,
multicast: bool, broadcast: bool, documentation: bool) {
let ip = Ipv4Addr::new(octets[0], octets[1], octets[2], octets[3]);
assert_eq!(octets, &ip.octets());
assert_eq!(ip.is_unspecified_s(), unspec);
assert_eq!(ip.is_loopback(), loopback);
assert_eq!(ip.is_private(), private);
assert_eq!(ip.is_link_local(), link_local);
assert_eq!(ip.is_global_s(), global);
assert_eq!(ip.is_multicast(), multicast);
assert_eq!(ip.is_broadcast(), broadcast);
assert_eq!(ip.is_documentation(), documentation);
}
// address unspec loopbk privt linloc global multicast brdcast doc
check(&[0, 0, 0, 0], true, false, false, false, true, false, false, false);
check(&[0, 0, 0, 1], false, false, false, false, true, false, false, false);
check(&[1, 0, 0, 0], false, false, false, false, true, false, false, false);
check(&[10, 9, 8, 7], false, false, true, false, false, false, false, false);
check(&[127, 1, 2, 3], false, true, false, false, false, false, false, false);
check(&[172, 31, 254, 253], false, false, true, false, false, false, false, false);
check(&[169, 254, 253, 242], false, false, false, true, false, false, false, false);
check(&[192, 0, 2, 183], false, false, false, false, false, false, false, true);
check(&[192, 1, 2, 183], false, false, false, false, true, false, false, false);
check(&[192, 168, 254, 253], false, false, true, false, false, false, false, false);
check(&[198, 51, 100, 0], false, false, false, false, false, false, false, true);
check(&[203, 0, 113, 0], false, false, false, false, false, false, false, true);
check(&[203, 2, 113, 0], false, false, false, false, true, false, false, false);
check(&[224, 0, 0, 0], false, false, false, false, true, true, false, false);
check(&[239, 255, 255, 255], false, false, false, false, true, true, false, false);
check(&[255, 255, 255, 255], false, false, false, false, false, false, true, false);
}
#[test]
fn ipv6_properties() {
fn check(str_addr: &str, unspec: bool, loopback: bool, global: bool) {
let ip: Ipv6Addr = str_addr.parse().unwrap();
assert_eq!(str_addr, ip.to_string());
assert_eq!(ip.is_unspecified_s(), unspec);
assert_eq!(ip.is_loopback(), loopback);
assert_eq!(ip.is_global_s(), global);
}
// unspec loopbk global
check("::", true, false, true);
check("::1", false, true, false);
}

8
util/src/network/mod.rs
View File

@ -56,7 +56,7 @@
//! } //! }
//! //!
//! fn main () { //! fn main () {
//! let mut service = NetworkService::<MyMessage>::start(NetworkConfiguration::new()).expect("Error creating network service"); //! let mut service = NetworkService::<MyMessage>::start(NetworkConfiguration::new_with_port(40412)).expect("Error creating network service");
//! service.register_protocol(Arc::new(MyHandler), "myproto", &[1u8]); //! service.register_protocol(Arc::new(MyHandler), "myproto", &[1u8]);
//! //!
//! // Wait for quit condition //! // Wait for quit condition
@ -71,8 +71,9 @@ mod session;
mod discovery; mod discovery;
mod service; mod service;
mod error; mod error;
mod node; mod node_table;
mod stats; mod stats;
mod ip_utils;
#[cfg(test)] #[cfg(test)]
mod tests; mod tests;
@ -88,6 +89,9 @@ pub use network::host::NetworkConfiguration;
pub use network::stats::NetworkStats; pub use network::stats::NetworkStats;
use io::TimerToken; use io::TimerToken;
pub use network::node_table::is_valid_node_url;
const PROTOCOL_VERSION: u32 = 4;
/// Network IO protocol handler. This needs to be implemented for each new subprotocol. /// Network IO protocol handler. This needs to be implemented for each new subprotocol.
/// All the handler function are called from within IO event loop. /// All the handler function are called from within IO event loop.

134
util/src/network/node.rs
View File

@ -1,134 +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 std::net::{SocketAddr, ToSocketAddrs};
use std::hash::{Hash, Hasher};
use std::str::{FromStr};
use hash::*;
use rlp::*;
use time::Tm;
use error::*;
/// Node public key
pub type NodeId = H512;
#[derive(Debug)]
/// Noe address info
pub struct NodeEndpoint {
/// IP(V4 or V6) address
pub address: SocketAddr,
/// Address as string (can be host name).
pub address_str: String,
/// Conneciton port.
pub udp_port: u16
}
impl FromStr for NodeEndpoint {
type Err = UtilError;
/// Create endpoint from string. Performs name resolution if given a host name.
fn from_str(s: &str) -> Result<NodeEndpoint, UtilError> {
let address = s.to_socket_addrs().map(|mut i| i.next());
match address {
Ok(Some(a)) => Ok(NodeEndpoint {
address: a,
address_str: s.to_owned(),
udp_port: a.port()
}),
Ok(_) => Err(UtilError::AddressResolve(None)),
Err(e) => Err(UtilError::AddressResolve(Some(e)))
}
}
}
#[derive(PartialEq, Eq, Copy, Clone)]
pub enum PeerType {
Required,
Optional
}
pub struct Node {
pub id: NodeId,
pub endpoint: NodeEndpoint,
pub peer_type: PeerType,
pub last_attempted: Option<Tm>,
}
impl FromStr for Node {
type Err = UtilError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let (id, endpoint) = if &s[0..8] == "enode://" && s.len() > 136 && &s[136..137] == "@" {
(try!(NodeId::from_str(&s[8..136])), try!(NodeEndpoint::from_str(&s[137..])))
}
else {
(NodeId::new(), try!(NodeEndpoint::from_str(s)))
};
Ok(Node {
id: id,
endpoint: endpoint,
peer_type: PeerType::Optional,
last_attempted: None,
})
}
}
impl PartialEq for Node {
fn eq(&self, other: &Self) -> bool {
self.id == other.id
}
}
impl Eq for Node { }
impl Hash for Node {
fn hash<H>(&self, state: &mut H) where H: Hasher {
self.id.hash(state)
}
}
#[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);
}
}

420
util/src/network/node_table.rs Normal file
View File

@ -0,0 +1,420 @@
// 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 std::mem;
use std::slice::from_raw_parts;
use std::net::{SocketAddr, ToSocketAddrs, SocketAddrV4, SocketAddrV6, Ipv4Addr, Ipv6Addr};
use std::hash::{Hash, Hasher};
use std::str::{FromStr};
use std::collections::HashMap;
use std::fmt::{Display, Formatter};
use std::path::{PathBuf};
use std::fmt;
use std::fs;
use std::io::{Read, Write};
use hash::*;
use rlp::*;
use time::Tm;
use error::*;
use network::discovery::{TableUpdates, NodeEntry};
use network::ip_utils::*;
pub use rustc_serialize::json::Json;
/// Node public key
pub type NodeId = H512;
#[derive(Debug, Clone)]
/// Node address info
pub struct NodeEndpoint {
/// IP(V4 or V6) address
pub address: SocketAddr,
/// Conneciton port.
pub udp_port: u16
}
impl NodeEndpoint {
pub fn udp_address(&self) -> SocketAddr {
match self.address {
SocketAddr::V4(a) => SocketAddr::V4(SocketAddrV4::new(a.ip().clone(), self.udp_port)),
SocketAddr::V6(a) => SocketAddr::V6(SocketAddrV6::new(a.ip().clone(), self.udp_port, a.flowinfo(), a.scope_id())),
}
}
}
impl NodeEndpoint {
pub fn from_rlp(rlp: &UntrustedRlp) -> Result<Self, DecoderError> {
let tcp_port = try!(rlp.val_at::<u16>(2));
let udp_port = try!(rlp.val_at::<u16>(1));
let addr_bytes = try!(try!(rlp.at(0)).data());
let address = try!(match addr_bytes.len() {
4 => Ok(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(addr_bytes[0], addr_bytes[1], addr_bytes[2], addr_bytes[3]), tcp_port))),
16 => unsafe {
let o: *const u16 = mem::transmute(addr_bytes.as_ptr());
let o = from_raw_parts(o, 8);
Ok(SocketAddr::V6(SocketAddrV6::new(Ipv6Addr::new(o[0], o[1], o[2], o[3], o[4], o[5], o[6], o[7]), tcp_port, 0, 0)))
},
_ => Err(DecoderError::RlpInconsistentLengthAndData)
});
Ok(NodeEndpoint { address: address, udp_port: udp_port })
}
pub fn to_rlp(&self, rlp: &mut RlpStream) {
match self.address {
SocketAddr::V4(a) => {
rlp.append(&(&a.ip().octets()[..]));
}
SocketAddr::V6(a) => unsafe {
let o: *const u8 = mem::transmute(a.ip().segments().as_ptr());
rlp.append(&from_raw_parts(o, 16));
}
};
rlp.append(&self.udp_port);
rlp.append(&self.address.port());
}
pub fn to_rlp_list(&self, rlp: &mut RlpStream) {
rlp.begin_list(3);
self.to_rlp(rlp);
}
pub fn is_valid(&self) -> bool {
self.udp_port != 0 && self.address.port() != 0 &&
match self.address {
SocketAddr::V4(a) => !a.ip().is_unspecified_s(),
SocketAddr::V6(a) => !a.ip().is_unspecified_s()
}
}
pub fn is_global(&self) -> bool {
match self.address {
SocketAddr::V4(a) => a.ip().is_global_s(),
SocketAddr::V6(a) => a.ip().is_global_s()
}
}
}
impl FromStr for NodeEndpoint {
type Err = UtilError;
/// Create endpoint from string. Performs name resolution if given a host name.
fn from_str(s: &str) -> Result<NodeEndpoint, UtilError> {
let address = s.to_socket_addrs().map(|mut i| i.next());
match address {
Ok(Some(a)) => Ok(NodeEndpoint {
address: a,
udp_port: a.port()
}),
Ok(_) => Err(UtilError::AddressResolve(None)),
Err(e) => Err(UtilError::AddressResolve(Some(e)))
}
}
}
#[derive(PartialEq, Eq, Copy, Clone)]
pub enum PeerType {
_Required,
Optional
}
pub struct Node {
pub id: NodeId,
pub endpoint: NodeEndpoint,
pub peer_type: PeerType,
pub failures: u32,
pub last_attempted: Option<Tm>,
}
impl Node {
pub fn new(id: NodeId, endpoint: NodeEndpoint) -> Node {
Node {
id: id,
endpoint: endpoint,
peer_type: PeerType::Optional,
failures: 0,
last_attempted: None,
}
}
}
impl Display for Node {
fn fmt(&self, f: &mut Formatter) -> fmt::Result {
if self.endpoint.udp_port != self.endpoint.address.port() {
try!(write!(f, "enode://{}@{}+{}", self.id.hex(), self.endpoint.address, self.endpoint.udp_port));
} else {
try!(write!(f, "enode://{}@{}", self.id.hex(), self.endpoint.address));
}
Ok(())
}
}
impl FromStr for Node {
type Err = UtilError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let (id, endpoint) = if &s[0..8] == "enode://" && s.len() > 136 && &s[136..137] == "@" {
(try!(NodeId::from_str(&s[8..136])), try!(NodeEndpoint::from_str(&s[137..])))
}
else {
(NodeId::new(), try!(NodeEndpoint::from_str(s)))
};
Ok(Node {
id: id,
endpoint: endpoint,
peer_type: PeerType::Optional,
last_attempted: None,
failures: 0,
})
}
}
impl PartialEq for Node {
fn eq(&self, other: &Self) -> bool {
self.id == other.id
}
}
impl Eq for Node {}
impl Hash for Node {
fn hash<H>(&self, state: &mut H) where H: Hasher {
self.id.hash(state)
}
}
/// Node table backed by disk file.
pub struct NodeTable {
nodes: HashMap<NodeId, Node>,
path: Option<String>,
}
impl NodeTable {
pub fn new(path: Option<String>) -> NodeTable {
NodeTable {
path: path.clone(),
nodes: NodeTable::load(path),
}
}
/// Add a node to table
pub fn add_node(&mut self, mut node: Node) {
// preserve failure counter
let failures = self.nodes.get(&node.id).map_or(0, |n| n.failures);
node.failures = failures;
self.nodes.insert(node.id.clone(), node);
}
/// Returns node ids sorted by number of failures
pub fn nodes(&self) -> Vec<NodeId> {
let mut refs: Vec<&Node> = self.nodes.values().collect();
refs.sort_by(|a, b| a.failures.cmp(&b.failures));
refs.iter().map(|n| n.id.clone()).collect()
}
/// Unordered list of all entries
pub fn unordered_entries(&self) -> Vec<NodeEntry> {
// preserve failure counter
self.nodes.values().map(|n| NodeEntry { endpoint: n.endpoint.clone(), id: n.id.clone() }).collect()
}
/// Get particular node
pub fn get_mut(&mut self, id: &NodeId) -> Option<&mut Node> {
self.nodes.get_mut(id)
}
/// Apply table changes coming from discovery
pub fn update(&mut self, mut update: TableUpdates) {
for (_, node) in update.added.drain() {
let mut entry = self.nodes.entry(node.id.clone()).or_insert_with(|| Node::new(node.id.clone(), node.endpoint.clone()));
entry.endpoint = node.endpoint;
}
for r in update.removed {
self.nodes.remove(&r);
}
}
/// Increase failure counte for a node
pub fn note_failure(&mut self, id: &NodeId) {
if let Some(node) = self.nodes.get_mut(id) {
node.failures += 1;
}
}
fn save(&self) {
if let Some(ref path) = self.path {
let mut path_buf = PathBuf::from(path);
if let Err(e) = fs::create_dir_all(path_buf.as_path()) {
warn!("Error creating node table directory: {:?}", e);
return;
};
path_buf.push("nodes.json");
let mut json = String::new();
json.push_str("{\n");
json.push_str("\"nodes\": [\n");
let node_ids = self.nodes();
for i in 0 .. node_ids.len() {
let node = self.nodes.get(&node_ids[i]).unwrap();
json.push_str(&format!("\t{{ \"url\": \"{}\", \"failures\": {} }}{}\n", node, node.failures, if i == node_ids.len() - 1 {""} else {","}))
}
json.push_str("]\n");
json.push_str("}");
let mut file = match fs::File::create(path_buf.as_path()) {
Ok(file) => file,
Err(e) => {
warn!("Error creating node table file: {:?}", e);
return;
}
};
if let Err(e) = file.write(&json.into_bytes()) {
warn!("Error writing node table file: {:?}", e);
}
}
}
fn load(path: Option<String>) -> HashMap<NodeId, Node> {
let mut nodes: HashMap<NodeId, Node> = HashMap::new();
if let Some(path) = path {
let mut path_buf = PathBuf::from(path);
path_buf.push("nodes.json");
let mut file = match fs::File::open(path_buf.as_path()) {
Ok(file) => file,
Err(e) => {
debug!("Error opening node table file: {:?}", e);
return nodes;
}
};
let mut buf = String::new();
match file.read_to_string(&mut buf) {
Ok(_) => {},
Err(e) => {
warn!("Error reading node table file: {:?}", e);
return nodes;
}
}
let json = match Json::from_str(&buf) {
Ok(json) => json,
Err(e) => {
warn!("Error parsing node table file: {:?}", e);
return nodes;
}
};
if let Some(list) = json.as_object().and_then(|o| o.get("nodes")).and_then(|n| n.as_array()) {
for n in list.iter().filter_map(|n| n.as_object()) {
if let Some(url) = n.get("url").and_then(|u| u.as_string()) {
if let Ok(mut node) = Node::from_str(url) {
if let Some(failures) = n.get("failures").and_then(|f| f.as_u64()) {
node.failures = failures as u32;
}
nodes.insert(node.id.clone(), node);
}
}
}
}
}
nodes
}
}
impl Drop for NodeTable {
fn drop(&mut self) {
self.save();
}
}
/// Check if node url is valid
pub fn is_valid_node_url(url: &str) -> bool {
use std::str::FromStr;
Node::from_str(url).is_ok()
}
#[cfg(test)]
mod tests {
use super::*;
use std::str::FromStr;
use std::net::*;
use hash::*;
use devtools::*;
#[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() {
assert!(is_valid_node_url("enode://a979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c@22.99.55.44:7770"));
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);
}
#[test]
fn table_failure_order() {
let node1 = Node::from_str("enode://a979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c@22.99.55.44:7770").unwrap();
let node2 = Node::from_str("enode://b979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c@22.99.55.44:7770").unwrap();
let node3 = Node::from_str("enode://c979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c@22.99.55.44:7770").unwrap();
let id1 = H512::from_str("a979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c").unwrap();
let id2 = H512::from_str("b979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c").unwrap();
let id3 = H512::from_str("c979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c").unwrap();
let mut table = NodeTable::new(None);
table.add_node(node3);
table.add_node(node1);
table.add_node(node2);
table.note_failure(&id1);
table.note_failure(&id1);
table.note_failure(&id2);
let r = table.nodes();
assert_eq!(r[0][..], id3[..]);
assert_eq!(r[1][..], id2[..]);
assert_eq!(r[2][..], id1[..]);
}
#[test]
fn table_save_load() {
let temp_path = RandomTempPath::create_dir();
let node1 = Node::from_str("enode://a979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c@22.99.55.44:7770").unwrap();
let node2 = Node::from_str("enode://b979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c@22.99.55.44:7770").unwrap();
let id1 = H512::from_str("a979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c").unwrap();
let id2 = H512::from_str("b979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c").unwrap();
{
let mut table = NodeTable::new(Some(temp_path.as_path().to_str().unwrap().to_owned()));
table.add_node(node1);
table.add_node(node2);
table.note_failure(&id2);
}
{
let table = NodeTable::new(Some(temp_path.as_path().to_str().unwrap().to_owned()));
let r = table.nodes();
assert_eq!(r[0][..], id1[..]);
assert_eq!(r[1][..], id2[..]);
}
}
}

2
util/src/network/service.rs
View File

@ -42,7 +42,7 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
let host = Arc::new(Host::new(config)); let host = Arc::new(Host::new(config));
let stats = host.stats().clone(); let stats = host.stats().clone();
let host_info = host.client_version(); let host_info = host.client_version();
info!("Host ID={:?}", host.client_id()); info!("Node URL: {}", host.client_url());
try!(io_service.register_handler(host)); try!(io_service.register_handler(host));
Ok(NetworkService { Ok(NetworkService {
io_service: io_service, io_service: io_service,

34
util/src/network/session.rs
View File

@ -14,6 +14,8 @@
// You should have received a copy of the GNU General Public License // You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>. // along with Parity. If not, see <http://www.gnu.org/licenses/>.
use std::net::SocketAddr;
use std::io;
use mio::*; use mio::*;
use hash::*; use hash::*;
use rlp::*; use rlp::*;
@ -23,7 +25,7 @@ use error::*;
use io::{IoContext, StreamToken}; use io::{IoContext, StreamToken};
use network::error::{NetworkError, DisconnectReason}; use network::error::{NetworkError, DisconnectReason};
use network::host::*; use network::host::*;
use network::node::NodeId; use network::node_table::NodeId;
use time; use time;
const PING_TIMEOUT_SEC: u64 = 30; const PING_TIMEOUT_SEC: u64 = 30;
@ -89,7 +91,7 @@ impl Decodable for PeerCapabilityInfo {
} }
} }
#[derive(Debug, PartialEq, Eq)] #[derive(Debug)]
struct SessionCapabilityInfo { struct SessionCapabilityInfo {
pub protocol: &'static str, pub protocol: &'static str,
pub version: u8, pub version: u8,
@ -108,7 +110,7 @@ const PACKET_LAST: u8 = 0x7f;
impl Session { impl Session {
/// Create a new session out of comepleted handshake. Consumes handshake object. /// Create a new session out of comepleted handshake. Consumes handshake object.
pub fn new<Message>(h: Handshake, _io: &IoContext<Message>, host: &HostInfo) -> Result<Session, UtilError> where Message: Send + Sync + Clone { pub fn new(h: Handshake, host: &HostInfo) -> Result<Session, UtilError> {
let id = h.id.clone(); let id = h.id.clone();
let connection = try!(EncryptedConnection::new(h)); let connection = try!(EncryptedConnection::new(h));
let mut session = Session { let mut session = Session {
@ -129,11 +131,26 @@ impl Session {
Ok(session) Ok(session)
} }
/// Get id of the remote peer
pub fn id(&self) -> &NodeId {
&self.info.id
}
/// Check if session is ready to send/receive data /// Check if session is ready to send/receive data
pub fn is_ready(&self) -> bool { pub fn is_ready(&self) -> bool {
self.had_hello self.had_hello
} }
/// Replace socket token
pub fn set_token(&mut self, token: StreamToken) {
self.connection.set_token(token);
}
/// Get remote peer address
pub fn remote_addr(&self) -> io::Result<SocketAddr> {
self.connection.remote_addr()
}
/// Readable IO handler. Returns packet data if available. /// Readable IO handler. Returns packet data if available.
pub fn readable<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo) -> Result<SessionData, UtilError> where Message: Send + Sync + Clone { pub fn readable<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo) -> Result<SessionData, UtilError> where Message: Send + Sync + Clone {
match try!(self.connection.readable(io)) { match try!(self.connection.readable(io)) {
@ -214,7 +231,11 @@ impl Session {
try!(self.read_hello(&rlp, host)); try!(self.read_hello(&rlp, host));
Ok(SessionData::Ready) Ok(SessionData::Ready)
}, },
PACKET_DISCONNECT => Err(From::from(NetworkError::Disconnect(DisconnectReason::DisconnectRequested))), PACKET_DISCONNECT => {
let rlp = UntrustedRlp::new(&packet.data[1..]);
let reason: u8 = try!(rlp.val_at(0));
Err(From::from(NetworkError::Disconnect(DisconnectReason::from_u8(reason))))
}
PACKET_PING => { PACKET_PING => {
try!(self.send_pong()); try!(self.send_pong());
Ok(SessionData::None) Ok(SessionData::None)
@ -301,7 +322,12 @@ impl Session {
trace!(target: "net", "Hello: {} v{} {} {:?}", client_version, protocol, id, caps); trace!(target: "net", "Hello: {} v{} {} {:?}", client_version, protocol, id, caps);
self.info.client_version = client_version; self.info.client_version = client_version;
self.info.capabilities = caps; self.info.capabilities = caps;
if self.info.capabilities.is_empty() {
trace!("No common capabilities with peer.");
return Err(From::from(self.disconnect(DisconnectReason::UselessPeer)));
}
if protocol != host.protocol_version { if protocol != host.protocol_version {
trace!("Peer protocol version mismatch: {}", protocol);
return Err(From::from(self.disconnect(DisconnectReason::UselessPeer))); return Err(From::from(self.disconnect(DisconnectReason::UselessPeer)));
} }
self.had_hello = true; self.had_hello = true;

79
util/src/network/tests.rs
View File

@ -23,17 +23,10 @@ use io::TimerToken;
use crypto::KeyPair; use crypto::KeyPair;
pub struct TestProtocol { pub struct TestProtocol {
drop_session: bool,
pub packet: Mutex<Bytes>, pub packet: Mutex<Bytes>,
pub got_timeout: AtomicBool, pub got_timeout: AtomicBool,
} pub got_disconnect: AtomicBool,
impl Default for TestProtocol {
fn default() -> Self {
TestProtocol {
packet: Mutex::new(Vec::new()),
got_timeout: AtomicBool::new(false),
}
}
} }
#[derive(Clone)] #[derive(Clone)]
@ -42,9 +35,17 @@ pub struct TestProtocolMessage {
} }
impl TestProtocol { impl TestProtocol {
pub fn new(drop_session: bool) -> Self {
TestProtocol {
packet: Mutex::new(Vec::new()),
got_timeout: AtomicBool::new(false),
got_disconnect: AtomicBool::new(false),
drop_session: drop_session,
}
}
/// Creates and register protocol with the network service /// Creates and register protocol with the network service
pub fn register(service: &mut NetworkService<TestProtocolMessage>) -> Arc<TestProtocol> { pub fn register(service: &mut NetworkService<TestProtocolMessage>, drop_session: bool) -> Arc<TestProtocol> {
let handler = Arc::new(TestProtocol::default()); let handler = Arc::new(TestProtocol::new(drop_session));
service.register_protocol(handler.clone(), "test", &[42u8, 43u8]).expect("Error registering test protocol handler"); service.register_protocol(handler.clone(), "test", &[42u8, 43u8]).expect("Error registering test protocol handler");
handler handler
} }
@ -56,6 +57,10 @@ impl TestProtocol {
pub fn got_timeout(&self) -> bool { pub fn got_timeout(&self) -> bool {
self.got_timeout.load(AtomicOrdering::Relaxed) self.got_timeout.load(AtomicOrdering::Relaxed)
} }
pub fn got_disconnect(&self) -> bool {
self.got_disconnect.load(AtomicOrdering::Relaxed)
}
} }
impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol { impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol {
@ -68,15 +73,22 @@ impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol {
self.packet.lock().unwrap().extend(data); self.packet.lock().unwrap().extend(data);
} }
fn connected(&self, io: &NetworkContext<TestProtocolMessage>, _peer: &PeerId) { fn connected(&self, io: &NetworkContext<TestProtocolMessage>, peer: &PeerId) {
assert!(io.peer_info(*peer).contains("parity"));
if self.drop_session {
io.disconnect_peer(*peer)
} else {
io.respond(33, "hello".to_owned().into_bytes()).unwrap(); io.respond(33, "hello".to_owned().into_bytes()).unwrap();
} }
}
fn disconnected(&self, _io: &NetworkContext<TestProtocolMessage>, _peer: &PeerId) { fn disconnected(&self, _io: &NetworkContext<TestProtocolMessage>, _peer: &PeerId) {
self.got_disconnect.store(true, AtomicOrdering::Relaxed);
} }
/// Timer function called after a timeout created with `NetworkContext::timeout`. /// Timer function called after a timeout created with `NetworkContext::timeout`.
fn timeout(&self, _io: &NetworkContext<TestProtocolMessage>, timer: TimerToken) { fn timeout(&self, io: &NetworkContext<TestProtocolMessage>, timer: TimerToken) {
io.message(TestProtocolMessage { payload: 22 });
assert_eq!(timer, 0); assert_eq!(timer, 0);
self.got_timeout.store(true, AtomicOrdering::Relaxed); self.got_timeout.store(true, AtomicOrdering::Relaxed);
} }
@ -85,34 +97,57 @@ impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol {
#[test] #[test]
fn net_service() { fn net_service() {
let mut service = NetworkService::<TestProtocolMessage>::start(NetworkConfiguration::new()).expect("Error creating network service"); let mut service = NetworkService::<TestProtocolMessage>::start(NetworkConfiguration::new_with_port(40414)).expect("Error creating network service");
service.register_protocol(Arc::new(TestProtocol::default()), "myproto", &[1u8]).unwrap(); service.register_protocol(Arc::new(TestProtocol::new(false)), "myproto", &[1u8]).unwrap();
} }
#[test] #[test]
fn net_connect() { fn net_connect() {
let key1 = KeyPair::create().unwrap(); let key1 = KeyPair::create().unwrap();
let mut config1 = NetworkConfiguration::new_with_port(30344); let mut config1 = NetworkConfiguration::new_with_port(30354);
config1.use_secret = Some(key1.secret().clone()); config1.use_secret = Some(key1.secret().clone());
config1.nat_enabled = false;
config1.boot_nodes = vec![ ]; config1.boot_nodes = vec![ ];
let mut config2 = NetworkConfiguration::new_with_port(30345); let mut config2 = NetworkConfiguration::new_with_port(30355);
config2.boot_nodes = vec![ format!("enode://{}@127.0.0.1:30344", key1.public().hex()) ]; config2.boot_nodes = vec![ format!("enode://{}@127.0.0.1:30354", key1.public().hex()) ];
config2.nat_enabled = false;
let mut service1 = NetworkService::<TestProtocolMessage>::start(config1).unwrap(); let mut service1 = NetworkService::<TestProtocolMessage>::start(config1).unwrap();
let mut service2 = NetworkService::<TestProtocolMessage>::start(config2).unwrap(); let mut service2 = NetworkService::<TestProtocolMessage>::start(config2).unwrap();
let handler1 = TestProtocol::register(&mut service1); let handler1 = TestProtocol::register(&mut service1, false);
let handler2 = TestProtocol::register(&mut service2); let handler2 = TestProtocol::register(&mut service2, false);
while !handler1.got_packet() && !handler2.got_packet() { while !handler1.got_packet() && !handler2.got_packet() && (service1.stats().sessions() == 0 || service2.stats().sessions() == 0) {
thread::sleep(Duration::from_millis(50)); thread::sleep(Duration::from_millis(50));
} }
assert!(service1.stats().sessions() >= 1); assert!(service1.stats().sessions() >= 1);
assert!(service2.stats().sessions() >= 1); assert!(service2.stats().sessions() >= 1);
} }
#[test]
fn net_disconnect() {
let key1 = KeyPair::create().unwrap();
let mut config1 = NetworkConfiguration::new_with_port(30364);
config1.use_secret = Some(key1.secret().clone());
config1.nat_enabled = false;
config1.boot_nodes = vec![ ];
let mut config2 = NetworkConfiguration::new_with_port(30365);
config2.boot_nodes = vec![ format!("enode://{}@127.0.0.1:30364", key1.public().hex()) ];
config2.nat_enabled = false;
let mut service1 = NetworkService::<TestProtocolMessage>::start(config1).unwrap();
let mut service2 = NetworkService::<TestProtocolMessage>::start(config2).unwrap();
let handler1 = TestProtocol::register(&mut service1, false);
let handler2 = TestProtocol::register(&mut service2, true);
while !(handler1.got_disconnect() && handler2.got_disconnect()) {
thread::sleep(Duration::from_millis(50));
}
assert!(handler1.got_disconnect());
assert!(handler2.got_disconnect());
}
#[test] #[test]
fn net_timeout() { fn net_timeout() {
let config = NetworkConfiguration::new_with_port(30346); let config = NetworkConfiguration::new_with_port(30346);
let mut service = NetworkService::<TestProtocolMessage>::start(config).unwrap(); let mut service = NetworkService::<TestProtocolMessage>::start(config).unwrap();
let handler = TestProtocol::register(&mut service); let handler = TestProtocol::register(&mut service, false);
while !handler.got_timeout() { while !handler.got_timeout() {
thread::sleep(Duration::from_millis(50)); thread::sleep(Duration::from_millis(50));
} }