Merge pull request #440 from ethcore/discovery

UDP Discovery

Cargo.lock

@@ -217,6 +217,7 @@ dependencies = [
  "itertools 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)",
  "json-tests 0.1.0",
  "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)",
  "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)",

ethcore/src/block_queue.rs

@@ -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>) {
-		while !deleting.load(AtomicOrdering::Relaxed) {
+		while !deleting.load(AtomicOrdering::Acquire) {
 			{
 				let mut lock = verification.lock().unwrap();
 
@@ -161,11 +161,11 @@ impl BlockQueue {
 					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();
 				}
 
-				if deleting.load(AtomicOrdering::Relaxed) {
+				if deleting.load(AtomicOrdering::Acquire) {
 					return;
 				}
 			}
@@ -347,7 +347,7 @@ impl MayPanic for BlockQueue {
 impl Drop for BlockQueue {
 	fn drop(&mut self) {
 		self.clear();
-		self.deleting.store(true, AtomicOrdering::Relaxed);
+		self.deleting.store(true, AtomicOrdering::Release);
 		self.more_to_verify.notify_all();
 		for t in self.verifiers.drain(..) {
 			t.join().unwrap();

ethcore/src/client.rs

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

ethcore/src/service.rs

@@ -124,8 +124,6 @@ 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)]
 mod tests {
 	use super::*;
@@ -136,8 +134,7 @@ mod tests {
 	fn it_can_be_started() {
 		let spec = get_test_spec();
 		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());
 	}
 }
-*/

parity/main.rs

@@ -37,6 +37,7 @@ extern crate ethcore_rpc as rpc;
 
 use std::net::{SocketAddr};
 use std::env;
+use std::path::PathBuf;
 use rlog::{LogLevelFilter};
 use env_logger::LogBuilder;
 use ctrlc::CtrlC;
@@ -69,8 +70,10 @@ Options:
 
   --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].
-  --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.
+  --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.
   --node-key KEY           Specify node secret key as hex string.
 
@@ -95,8 +98,10 @@ struct Args {
 	flag_keys_path: String,
 	flag_no_bootstrap: bool,
 	flag_listen_address: String,
-	flag_public_address: String,
+	flag_public_address: Option<String>,
 	flag_address: Option<String>,
+	flag_peers: u32,
+	flag_no_discovery: bool,
 	flag_upnp: bool,
 	flag_node_key: Option<String>,
 	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())
 	}
 
-	fn keys_path(&self) -> String {
+	fn _keys_path(&self) -> String {
 		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");
+			listen_address = public_address;
-				public_address = SocketAddr::from_str(self.args.flag_public_address.as_ref()).expect("Invalid public address given with --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");
 			}
-			Some(ref a) => {
+			public_address = Some(SocketAddr::from_str(a.as_ref()).expect("Invalid listen address given with --public-address"));
-				public_address = SocketAddr::from_str(a.as_ref()).expect("Invalid listen/public address given with --address");
+		}
-				listen_address = public_address;
-			}
-		};
-
 		(listen_address, public_address)
 	}
 
@@ -236,6 +243,11 @@ impl Configuration {
 		net_settings.listen_address = listen;
 		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.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
 		let mut service = ClientService::start(spec, net_settings, &Path::new(&self.path())).unwrap();

sync/src/chain.rs

@@ -82,7 +82,7 @@ const RECEIPTS_PACKET: u8 = 0x10;
 
 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 {
 	/// Header data
@@ -314,7 +314,7 @@ impl ChainSync {
 		}
 
 		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);
 		Ok(())
 	}
@@ -545,7 +545,7 @@ impl ChainSync {
 	pub fn on_peer_aborting(&mut self, io: &mut SyncIo, peer: PeerId) {
 		trace!(target: "sync", "== Disconnecting {}", peer);
 		if self.peers.contains_key(&peer) {
-			info!(target: "sync", "Disconnected {}", peer);
+			debug!(target: "sync", "Disconnected {}", peer);
 			self.clear_peer_download(peer);
 			self.peers.remove(&peer);
 			self.continue_sync(io);
@@ -1179,7 +1179,7 @@ impl ChainSync {
 		for (peer_id, peer_number) in updated_peers {
 			let mut peer_best = self.peers.get(&peer_id).unwrap().latest_hash.clone();
 			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();
 			}
 			sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &peer_best, &local_best) {

util/Cargo.toml

@@ -30,6 +30,7 @@ clippy = { version = "0.0.42", optional = true }
 json-tests = { path = "json-tests" }
 target_info = "0.1.0"
 igd = "0.4.2"
+libc = "0.2.7"
 
 [features]
 default = []

util/fdlimit/src/raise_fd_limit.rs

@@ -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() {}

util/src/bytes.rs

@@ -170,32 +170,8 @@ pub trait BytesConvertable {
 	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]

util/src/hash.rs

@@ -77,12 +77,6 @@ macro_rules! impl_hash {
 		/// Unformatted binary data of fixed length.
 		pub struct $from (pub [u8; $size]);
 
-		impl BytesConvertable for $from {
-			fn bytes(&self) -> &[u8] {
-				&self.0
-			}
-		}
-
 		impl Deref for $from {
 			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 {
 			#[inline]
 			fn deref_mut(&mut self) -> &mut [u8] {

util/src/io/service.rs

@@ -256,6 +256,11 @@ impl<Message> Handler for IoManager<Message> where Message: Send + Clone + Sync
 			IoMessage::DeregisterStream { handler_id, token } => {
 				let handler = self.handlers.get(handler_id).expect("Unknown handler id").clone();
 				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 } => {
 				let handler = self.handlers.get(handler_id).expect("Unknown handler id").clone();

util/src/io/worker.rs

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

util/src/lib.rs

@@ -106,6 +106,7 @@ extern crate serde;
 #[macro_use]
 extern crate log as rlog;
 extern crate igd;
+extern crate libc;
 
 pub mod standard;
 #[macro_use]

util/src/network/connection.rs

@@ -16,6 +16,7 @@
 
 use std::sync::Arc;
 use std::collections::VecDeque;
+use std::net::SocketAddr;
 use mio::{Handler, Token, EventSet, EventLoop, PollOpt, TryRead, TryWrite};
 use mio::tcp::*;
 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.
 	pub fn register_socket<Host: Handler>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> io::Result<()> {
 		trace!(target: "net", "connection register; token={:?}", reg);
@@ -238,6 +254,16 @@ impl EncryptedConnection {
 		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.
 	pub fn new(mut handshake: Handshake) -> Result<EncryptedConnection, UtilError> {
 		let shared = try!(crypto::ecdh::agree(handshake.ecdhe.secret(), &handshake.remote_public));

util/src/network/discovery.rs

@@ -14,115 +14,180 @@
 // You should have received a copy of the GNU General Public License
 // 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::udp::*;
+use sha3::*;
+use time;
 use hash::*;
-use sha3::Hashable;
 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 {
-	distance: u32,
+	nodes: VecDeque<BucketEntry>, //sorted by last active
-	nodes: Vec<NodeId>
 }
 
 impl NodeBucket {
-	fn new(distance: u32) -> NodeBucket {
+	fn new() -> NodeBucket {
 		NodeBucket {
-			distance: distance,
+			nodes: VecDeque::new()
-			nodes: Vec::new()
 		}
 	}
 }
 
-struct Discovery {
+struct Datagramm {
+	payload: Bytes,
+	address: SocketAddr,
+}
+
+pub struct Discovery {
 	id: NodeId,
+	secret: Secret,
+	public_endpoint: NodeEndpoint,
+	udp_socket: UdpSocket,
+	token: StreamToken,
 	discovery_round: u16,
 	discovery_id: NodeId,
 	discovery_nodes: HashSet<NodeId>,
 	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 {
-	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 {
-			id: id.clone(),
+			id: key.public().clone(),
+			secret: key.secret().clone(),
+			public_endpoint: public,
+			token: token,
 			discovery_round: 0,
 			discovery_id: NodeId::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_id.randomize();
+		self.discovery_id.randomize(); //TODO: use cryptographic nonce
 		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;
 		}
+		trace!(target: "discovery", "Starting round {:?}", self.discovery_round);
 		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 {
-				//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;
+				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;
 		}
 		self.discovery_round += 1;
-		//event_loop.timeout_ms(Token(NODETABLE_DISCOVERY), 1200).unwrap();
 	}
 
 	fn distance(a: &NodeId, b: &NodeId) -> u32 {
@@ -138,86 +203,353 @@ impl Discovery {
 		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;
 
-		// 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();
-						else {
+						last.is_empty()
-							break;
+					};
-						}
+					if remove {
-				}
+						found.remove(&distance);
-				if count < BUCKET_SIZE && tail != 0 {
-					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;
-						}
 					}
 				}
-
+				else {
-				head += 1;
+					count += 1;
-				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();
+		let mut ret:Vec<NodeEntry> = Vec::new();
 		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
 	}
+
+	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);
+	}
 }

util/src/network/error.rs

@@ -15,23 +15,45 @@
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
 use io::IoError;
+use crypto::CryptoError;
 use rlp::*;
 
-#[derive(Debug, Copy, Clone)]
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum DisconnectReason
 {
 	DisconnectRequested,
-	_TCPError,
+	TCPError,
-	_BadProtocol,
+	BadProtocol,
 	UselessPeer,
-	_TooManyPeers,
+	TooManyPeers,
-	_DuplicatePeer,
+	DuplicatePeer,
-	_IncompatibleProtocol,
+	IncompatibleProtocol,
-	_NullIdentity,
+	NullIdentity,
-	_ClientQuit,
+	ClientQuit,
-	_UnexpectedIdentity,
+	UnexpectedIdentity,
-	_LocalIdentity,
+	LocalIdentity,
 	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)]
@@ -41,6 +63,8 @@ pub enum NetworkError {
 	Auth,
 	/// Unrecognised protocol.
 	BadProtocol,
+	/// Message expired.
+	Expired,
 	/// Peer not found.
 	PeerNotFound,
 	/// 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"),
+	}
+}

util/src/network/handshake.rs

@@ -24,7 +24,7 @@ use crypto::*;
 use crypto;
 use network::connection::{Connection};
 use network::host::{HostInfo};
-use network::node::NodeId;
+use network::node_table::NodeId;
 use error::*;
 use network::error::NetworkError;
 use network::stats::NetworkStats;
@@ -68,7 +68,7 @@ pub struct Handshake {
 
 const AUTH_PACKET_SIZE: usize = 307;
 const ACK_PACKET_SIZE: usize = 210;
-const HANDSHAKE_TIMEOUT: u64 = 30000;
+const HANDSHAKE_TIMEOUT: u64 = 5000;
 
 impl Handshake {
 	/// 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
 	pub fn start<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo, originated: bool) -> Result<(), UtilError> where Message: Send + Clone{
 		self.originated = originated;

util/src/network/ip_utils.rs

@@ -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);
+}

util/src/network/mod.rs

@@ -56,7 +56,7 @@
 //! }
 //!
 //! 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]);
 //!
 //! 	// Wait for quit condition
@@ -71,8 +71,9 @@ mod session;
 mod discovery;
 mod service;
 mod error;
-mod node;
+mod node_table;
 mod stats;
+mod ip_utils;
 
 #[cfg(test)]
 mod tests;
@@ -88,6 +89,9 @@ pub use network::host::NetworkConfiguration;
 pub use network::stats::NetworkStats;
 
 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.
 /// All the handler function are called from within IO event loop.

util/src/network/node.rs

@@ -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);
-	}
-}

util/src/network/node_table.rs

@@ -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 tests::helpers::*;
+
+	#[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[..]);
+		}
+	}
+}

util/src/network/service.rs

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

util/src/network/session.rs

@@ -14,6 +14,8 @@
 // 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;
+use std::io;
 use mio::*;
 use hash::*;
 use rlp::*;
@@ -23,7 +25,7 @@ use error::*;
 use io::{IoContext, StreamToken};
 use network::error::{NetworkError, DisconnectReason};
 use network::host::*;
-use network::node::NodeId;
+use network::node_table::NodeId;
 use time;
 
 const PING_TIMEOUT_SEC: u64 = 30;
@@ -89,7 +91,7 @@ impl Decodable for PeerCapabilityInfo {
 	}
 }
 
-#[derive(Debug, PartialEq, Eq)]
+#[derive(Debug)]
 struct SessionCapabilityInfo {
 	pub protocol: &'static str,
 	pub version: u8,
@@ -108,7 +110,7 @@ const PACKET_LAST: u8 = 0x7f;
 
 impl Session {
 	/// 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 connection = try!(EncryptedConnection::new(h));
 		let mut session = Session {
@@ -129,11 +131,26 @@ impl 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
 	pub fn is_ready(&self) -> bool {
 		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.
 	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)) {
@@ -214,7 +231,11 @@ impl Session {
 				try!(self.read_hello(&rlp, host));
 				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 => {
 				try!(self.send_pong());
 				Ok(SessionData::None)
@@ -301,7 +322,12 @@ impl Session {
 		trace!(target: "net", "Hello: {} v{} {} {:?}", client_version, protocol, id, caps);
 		self.info.client_version = client_version;
 		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 {
+			trace!("Peer protocol version mismatch: {}", protocol);
 			return Err(From::from(self.disconnect(DisconnectReason::UselessPeer)));
 		}
 		self.had_hello = true;

util/src/network/tests.rs

@@ -23,17 +23,10 @@ use io::TimerToken;
 use crypto::KeyPair;
 
 pub struct TestProtocol {
+	drop_session: bool,
 	pub packet: Mutex<Bytes>,
 	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)]
@@ -42,9 +35,17 @@ pub struct TestProtocolMessage {
 }
 
 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
-	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");
 		handler
 	}
@@ -56,6 +57,10 @@ impl TestProtocol {
 	pub fn got_timeout(&self) -> bool {
 		self.got_timeout.load(AtomicOrdering::Relaxed)
 	}
+
+	pub fn got_disconnect(&self) -> bool {
+		self.got_disconnect.load(AtomicOrdering::Relaxed)
+	}
 }
 
 impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol {
@@ -68,15 +73,22 @@ impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol {
 		self.packet.lock().unwrap().extend(data);
 	}
 
-	fn connected(&self, io: &NetworkContext<TestProtocolMessage>, _peer: &PeerId) {
+	fn connected(&self, io: &NetworkContext<TestProtocolMessage>, peer: &PeerId) {
-		io.respond(33, "hello".to_owned().into_bytes()).unwrap();
+		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();
+		}
 	}
 
 	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`.
-	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);
 		self.got_timeout.store(true, AtomicOrdering::Relaxed);
 	}
@@ -85,34 +97,57 @@ impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol {
 
 #[test]
 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]
 fn net_connect() {
 	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.nat_enabled = false;
 	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 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));
 	}
 	assert!(service1.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]
 fn net_timeout() {
 	let config = NetworkConfiguration::new_with_port(30346);
 	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() {
 		thread::sleep(Duration::from_millis(50));
 	}