Merge branch 'master' into nvolf

util/src/network/connection.rs

@@ -216,6 +216,12 @@ pub struct EncryptedConnection {
 }
 
 impl EncryptedConnection {
+	
+	/// Get socket token 
+	pub fn token(&self) -> StreamToken {
+		self.connection.token
+	}
+
 	/// 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));
@@ -607,4 +613,4 @@ mod tests {
 		assert!(!status.is_ok());
 		assert_eq!(1, connection.send_queue.len());
 	}
-}
+}

util/src/network/error.rs

@@ -5,17 +5,17 @@ use rlp::*;
 pub enum DisconnectReason
 {
 	DisconnectRequested,
-	//TCPError,
-	//BadProtocol,
+	_TCPError,
+	_BadProtocol,
 	UselessPeer,
-	//TooManyPeers,
-	//DuplicatePeer,
-	//IncompatibleProtocol,
-	//NullIdentity,
-	//ClientQuit,
-	//UnexpectedIdentity,
-	//LocalIdentity,
-	//PingTimeout,
+	_TooManyPeers,
+	_DuplicatePeer,
+	_IncompatibleProtocol,
+	_NullIdentity,
+	_ClientQuit,
+	_UnexpectedIdentity,
+	_LocalIdentity,
+	PingTimeout,
 }
 
 #[derive(Debug)]

util/src/network/host.rs

@@ -19,6 +19,7 @@ use io::*;
 use network::NetworkProtocolHandler;
 use network::node::*;
 use network::stats::NetworkStats;
+use network::error::DisconnectReason;
 
 type Slab<T> = ::slab::Slab<T, usize>;
 
@@ -108,6 +109,8 @@ pub enum NetworkIoMessage<Message> where Message: Send + Sync + Clone {
 		/// Timer delay in milliseconds.
 		delay: u64,
 	},
+	/// Disconnect a peer
+	Disconnect(PeerId),
 	/// User message
 	User(Message),
 }
@@ -181,8 +184,14 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
 	}
 
 	/// Disable current protocol capability for given peer. If no capabilities left peer gets disconnected.
-	pub fn disable_peer(&self, _peer: PeerId) {
+	pub fn disable_peer(&self, peer: PeerId) {
 		//TODO: remove capability, disconnect if no capabilities left
+		self.disconnect_peer(peer);
+	}
+
+	/// Disconnect peer. Reconnect can be attempted later.
+	pub fn disconnect_peer(&self, peer: PeerId) {
+		self.io.message(NetworkIoMessage::Disconnect(peer));
 	}
 
 	/// Register a new IO timer. 'IoHandler::timeout' will be called with the token.
@@ -332,6 +341,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 	}
 
 	fn maintain_network(&self, io: &IoContext<NetworkIoMessage<Message>>) {
+		self.keep_alive(io);
 		self.connect_peers(io);
 	}
 
@@ -343,6 +353,21 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 		self.connections.read().unwrap().iter().any(|e| match *e.lock().unwrap().deref() { ConnectionEntry::Handshake(ref h) => h.id.eq(&id), _ => false  })
 	}
 
+	fn keep_alive(&self, io: &IoContext<NetworkIoMessage<Message>>) {
+		let mut to_kill = Vec::new();
+		for e in self.connections.write().unwrap().iter_mut() {
+			if let ConnectionEntry::Session(ref mut s) = *e.lock().unwrap().deref_mut() {
+				if !s.keep_alive() {
+					s.disconnect(DisconnectReason::PingTimeout);
+					to_kill.push(s.token());
+				}
+			}
+		}
+		for p in to_kill {
+			self.kill_connection(p, io);
+		}
+	}
+
 	fn connect_peers(&self, io: &IoContext<NetworkIoMessage<Message>>) {
 		struct NodeInfo {
 			id: NodeId,
@@ -684,6 +709,15 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
 				self.timers.write().unwrap().insert(handler_token, ProtocolTimer { protocol: protocol, token: *token });
 				io.register_timer(handler_token, *delay).expect("Error registering timer");
 			},
+			NetworkIoMessage::Disconnect(ref peer) => {
+				if let Some(connection) = self.connections.read().unwrap().get(*peer).cloned() {
+					match *connection.lock().unwrap().deref_mut() {
+						ConnectionEntry::Handshake(_) => {},
+						ConnectionEntry::Session(ref mut s) => { s.disconnect(DisconnectReason::DisconnectRequested); } 
+					}
+				} 
+				self.kill_connection(*peer, io);
+			},
 			NetworkIoMessage::User(ref message) => {
 				for (p, h) in self.handlers.read().unwrap().iter() {
 					h.message(&NetworkContext::new(io, p, None, self.connections.clone()), &message);

util/src/network/service.rs

@@ -21,7 +21,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!("NetworkService::start(): id={:?}", host.client_id());
+		info!("Host ID={:?}", host.client_id());
 		try!(io_service.register_handler(host));
 		Ok(NetworkService {
 			io_service: io_service,

util/src/network/session.rs

@@ -4,10 +4,14 @@ use rlp::*;
 use network::connection::{EncryptedConnection, Packet};
 use network::handshake::Handshake;
 use error::*;
-use io::{IoContext};
+use io::{IoContext, StreamToken};
 use network::error::{NetworkError, DisconnectReason};
 use network::host::*;
 use network::node::NodeId;
+use time;
+
+const PING_TIMEOUT_SEC: u64 = 30;
+const PING_INTERVAL_SEC: u64 = 30;
 
 /// Peer session over encrypted connection.
 /// When created waits for Hello packet exchange and signals ready state.
@@ -19,6 +23,8 @@ pub struct Session {
 	connection: EncryptedConnection,
 	/// Session ready flag. Set after successfull Hello packet exchange
 	had_hello: bool,
+	ping_time_ns: u64,
+	pong_time_ns: Option<u64>,
 }
 
 /// Structure used to report various session events.
@@ -47,6 +53,8 @@ pub struct SessionInfo {
 	pub protocol_version: u32,
 	/// Peer protocol capabilities
 	capabilities: Vec<SessionCapabilityInfo>,
+	/// Peer ping delay in milliseconds
+	pub ping_ms: Option<u64>,
 }
 
 #[derive(Debug, PartialEq, Eq)]
@@ -95,10 +103,13 @@ impl Session {
 				client_version: String::new(),
 				protocol_version: 0,
 				capabilities: Vec::new(),
+				ping_ms: None,
 			},
+			ping_time_ns: 0,
+			pong_time_ns: None,
 		};
 		try!(session.write_hello(host));
-		try!(session.write_ping());
+		try!(session.send_ping());
 		Ok(session)
 	}
 
@@ -141,7 +152,7 @@ impl Session {
 		while protocol != self.info.capabilities[i].protocol {
 			i += 1;
 			if i == self.info.capabilities.len() {
-				debug!(target: "net", "Unkown protocol: {:?}", protocol);
+				debug!(target: "net", "Unknown protocol: {:?}", protocol);
 				return Ok(())
 			}
 		}
@@ -152,6 +163,26 @@ impl Session {
 		self.connection.send_packet(&rlp.out())
 	}
 
+	/// Keep this session alive. Returns false if ping timeout happened
+	pub fn keep_alive(&mut self) -> bool {
+		let timed_out = if let Some(pong) = self.pong_time_ns {
+			pong - self.ping_time_ns > PING_TIMEOUT_SEC * 1000_000_000
+		} else {
+			time::precise_time_ns() - self.ping_time_ns > PING_TIMEOUT_SEC * 1000_000_000
+		};
+
+		if !timed_out && time::precise_time_ns() - self.ping_time_ns > PING_INTERVAL_SEC * 1000_000_000 {
+			if let Err(e) = self.send_ping() {
+				debug!("Error sending ping message: {:?}", e);
+			}
+		}
+		!timed_out
+	}
+
+	pub fn token(&self) -> StreamToken {
+		self.connection.token()
+	}
+
 	fn read_packet(&mut self, packet: Packet, host: &HostInfo) -> Result<SessionData, UtilError> {
 		if packet.data.len() < 2 {
 			return Err(From::from(NetworkError::BadProtocol));
@@ -168,7 +199,12 @@ impl Session {
 			},
 			PACKET_DISCONNECT => Err(From::from(NetworkError::Disconnect(DisconnectReason::DisconnectRequested))),
 			PACKET_PING => {
-				try!(self.write_pong());
+				try!(self.send_pong());
+				Ok(SessionData::None)
+			},
+			PACKET_PONG => {
+				self.pong_time_ns = Some(time::precise_time_ns());
+				self.info.ping_ms = Some((self.pong_time_ns.unwrap() - self.ping_time_ns) / 1000_000);
 				Ok(SessionData::None)
 			},
 			PACKET_GET_PEERS => Ok(SessionData::None), //TODO;
@@ -178,7 +214,7 @@ impl Session {
 				while packet_id < self.info.capabilities[i].id_offset {
 					i += 1;
 					if i == self.info.capabilities.len() {
-						debug!(target: "net", "Unkown packet: {:?}", packet_id);
+						debug!(target: "net", "Unknown packet: {:?}", packet_id);
 						return Ok(SessionData::None)
 					}
 				}
@@ -189,7 +225,7 @@ impl Session {
 				Ok(SessionData::Packet { data: packet.data, protocol: protocol, packet_id: pid } )
 			},
 			_ => {
-				debug!(target: "net", "Unkown packet: {:?}", packet_id);
+				debug!(target: "net", "Unknown packet: {:?}", packet_id);
 				Ok(SessionData::None)
 			}
 		}
@@ -255,15 +291,20 @@ impl Session {
 		Ok(())
 	}
 
-	fn write_ping(&mut self) -> Result<(), UtilError>  {
-		self.send(try!(Session::prepare(PACKET_PING)))
+	/// Senf ping packet
+	pub fn send_ping(&mut self) -> Result<(), UtilError>  {
+		try!(self.send(try!(Session::prepare(PACKET_PING))));
+		self.ping_time_ns = time::precise_time_ns();
+		self.pong_time_ns = None;
+		Ok(())
 	}
 
-	fn write_pong(&mut self) -> Result<(), UtilError>  {
+	fn send_pong(&mut self) -> Result<(), UtilError>  {
 		self.send(try!(Session::prepare(PACKET_PONG)))
 	}
 
-	fn disconnect(&mut self, reason: DisconnectReason) -> NetworkError {
+	/// Disconnect this session
+	pub fn disconnect(&mut self, reason: DisconnectReason) -> NetworkError {
 		let mut rlp = RlpStream::new();
 		rlp.append(&(PACKET_DISCONNECT as u32));
 		rlp.begin_list(1);