Merge branch 'master' into geth-import

Cargo.lock

@@ -15,7 +15,6 @@ dependencies = [
  "fdlimit 0.1.0",
  "log 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustc-serialize 0.3.18 (registry+https://github.com/rust-lang/crates.io-index)",
- "target_info 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -53,6 +52,11 @@ name = "bitflags"
 version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
+[[package]]
+name = "blastfig"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
 [[package]]
 name = "bytes"
 version = "0.3.0"
@@ -208,7 +212,6 @@ dependencies = [
  "serde_codegen 0.6.14 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_json 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "syntex 0.29.0 (registry+https://github.com/rust-lang/crates.io-index)",
- "target_info 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -234,12 +237,13 @@ dependencies = [
  "rocksdb 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "rust-crypto 0.2.34 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustc-serialize 0.3.18 (registry+https://github.com/rust-lang/crates.io-index)",
+ "rustc_version 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde 0.6.14 (registry+https://github.com/rust-lang/crates.io-index)",
  "sha3 0.1.0",
  "slab 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
- "target_info 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "time 0.1.34 (registry+https://github.com/rust-lang/crates.io-index)",
  "tiny-keccak 1.0.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "vergen 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -688,11 +692,6 @@ dependencies = [
  "unicode-xid 0.0.3 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
-[[package]]
-name = "target_info"
-version = "0.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-
 [[package]]
 name = "term"
 version = "0.2.14"
@@ -789,6 +788,15 @@ dependencies = [
  "rustc-serialize 0.3.18 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "vergen"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "blastfig 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "time 0.1.34 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "winapi"
 version = "0.2.5"

Cargo.toml

@@ -17,7 +17,6 @@ ethcore = { path = "ethcore" }
 ethsync = { path = "sync" }
 ethcore-rpc = { path = "rpc", optional = true }
 fdlimit = { path = "util/fdlimit" }
-target_info = "0.1"
 daemonize = "0.2"
 ethcore-devtools = { path = "devtools" }
 

install-deps.sh

@@ -425,7 +425,7 @@ function run_installer()
 			depFound=$((depFound+1))
 			check "multirust"
 			isMultirust=true
-			if [[ $(multirust show-default 2>/dev/null | grep beta | wc -l) == 4 ]]; then
+			if [[ $(multirust show-default 2>/dev/null | grep beta | wc -l) == 3 ]]; then
 				depFound=$((depFound+1))
 				check "rust beta"
 				isMultirustBeta=true

parity/main.rs

@@ -29,7 +29,6 @@ extern crate log as rlog;
 extern crate env_logger;
 extern crate ctrlc;
 extern crate fdlimit;
-extern crate target_info;
 extern crate daemonize;
 
 #[cfg(feature = "rpc")]
@@ -51,7 +50,6 @@ use ethcore::ethereum;
 use ethcore::blockchain::CacheSize;
 use ethsync::EthSync;
 use docopt::Docopt;
-use target_info::Target;
 use daemonize::Daemonize;
 
 const USAGE: &'static str = "
@@ -146,14 +144,15 @@ fn setup_rpc_server(_client: Arc<Client>, _sync: Arc<EthSync>, _url: &str) {
 
 fn print_version() {
 	println!("\
-Parity version {} ({}-{}-{})
+Parity
+  version {}
 Copyright 2015, 2016 Ethcore (UK) Limited
 License GPLv3+: GNU GPL version 3 or later <http://gnu.org/licenses/gpl.html>.
 This is free software: you are free to change and redistribute it.
 There is NO WARRANTY, to the extent permitted by law.
 
 By Wood/Paronyan/Kotewicz/Drwięga/Volf.\
-", env!("CARGO_PKG_VERSION"), Target::arch(), Target::env(), Target::os());
+", version());
 }
 
 fn die_with_message(msg: &str) -> ! {

rpc/Cargo.toml

@@ -17,7 +17,6 @@ ethcore-util = { path = "../util" }
 ethcore = { path = "../ethcore" }
 ethsync = { path = "../sync" }
 clippy = { version = "0.0.42", optional = true }
-target_info = "0.1.0"
 rustc-serialize = "0.3"
 serde_macros = { version = "0.6.13", optional = true }
 

rpc/src/lib.rs

@@ -20,7 +20,6 @@
 #![cfg_attr(nightly, plugin(serde_macros, clippy))]
 
 extern crate rustc_serialize;
-extern crate target_info;
 extern crate serde;
 extern crate serde_json;
 extern crate jsonrpc_core;

rpc/src/v1/impls/web3.rs

@@ -15,8 +15,8 @@
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
 //! Web3 rpc implementation.
-use target_info::Target;
 use jsonrpc_core::*;
+use util::version;
 use v1::traits::Web3;
 
 /// Web3 rpc implementation.
@@ -30,7 +30,9 @@ impl Web3Client {
 impl Web3 for Web3Client {
 	fn client_version(&self, params: Params) -> Result<Value, Error> {
 		match params {
-			Params::None => Ok(Value::String(format!("Parity/-/{}/{}-{}-{}/rust1.8-nightly", env!("CARGO_PKG_VERSION"), Target::arch(), Target::env(), Target::os()))),
+			Params::None => {
+				Ok(Value::String(version())),
+			}
 			_ => Err(Error::invalid_params())
 		}
 	}

util/Cargo.toml

@@ -5,6 +5,7 @@ license = "GPL-3.0"
 name = "ethcore-util"
 version = "0.9.99"
 authors = ["Ethcore <admin@ethcore.io>"]
+build = "build.rs"
 
 [dependencies]
 log = "0.3"
@@ -28,11 +29,15 @@ sha3 = { path = "sha3" }
 serde = "0.6.7"
 clippy = { version = "0.0.42", optional = true }
 json-tests = { path = "json-tests" }
-target_info = "0.1.0"
+rustc_version = "0.1.0"
 igd = "0.4.2"
 ethcore-devtools = { path = "../devtools" }
 libc = "0.2.7"
+vergen = "0.1"
 
 [features]
 default = []
 dev = ["clippy"]
+
+[build-dependencies]
+vergen = "*"

util/build.rs

@@ -0,0 +1,6 @@
+extern crate vergen;
+use vergen::*;
+
+fn main() {
+	vergen(OutputFns::all()).unwrap();
+}

util/src/lib.rs

@@ -82,7 +82,6 @@
 //!   cargo build --release
 //!   ```
 
-extern crate target_info;
 extern crate slab;
 extern crate rustc_serialize;
 extern crate mio;
@@ -108,6 +107,8 @@ extern crate log as rlog;
 extern crate igd;
 extern crate ethcore_devtools as devtools;
 extern crate libc;
+extern crate rustc_version;
+extern crate vergen;
 
 pub mod standard;
 #[macro_use]

util/src/misc.rs

@@ -18,6 +18,9 @@
 
 use std::fs::File;
 use common::*;
+use rustc_version;
+
+include!(concat!(env!("OUT_DIR"), "/version.rs"));
 
 #[derive(Debug,Clone,PartialEq,Eq)]
 /// Diff type for specifying a change (or not).
@@ -62,3 +65,8 @@ pub fn contents(name: &str) -> Result<Bytes, UtilError> {
 	try!(file.read_to_end(&mut ret));
 	Ok(ret)
 }
+
+/// Get the standard version string for this software.
+pub fn version() -> String {
+	format!("Parity/{}/{}-{}/{}/rustc{}", env!("CARGO_PKG_VERSION"), short_sha(), commit_date(), target(), rustc_version::version())
+}

util/src/network/connection.rs

@@ -175,13 +175,26 @@ impl Connection {
 		self.socket.peer_addr()
 	}
 
+	/// Clone this connection. Clears the receiving buffer of the returned connection.
+	pub fn try_clone(&self) -> io::Result<Self> {
+		Ok(Connection {
+			token: self.token,
+			socket: try!(self.socket.try_clone()),
+			rec_buf: Vec::new(),
+			rec_size: 0,
+			send_queue: self.send_queue.clone(),
+			interest: EventSet::hup() | EventSet::readable(),
+			stats: self.stats.clone(),
+		})
+	}
+
 	/// 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);
-		event_loop.register(&self.socket, reg, self.interest, PollOpt::edge() | PollOpt::oneshot()).or_else(|e| {
+		if let Err(e) = event_loop.register(&self.socket, reg, self.interest, PollOpt::edge() | PollOpt::oneshot()) {
 			debug!("Failed to register {:?}, {:?}", reg, e);
-			Ok(())
+		}
-		})
+		Ok(())
 	}
 
 	/// Update connection registration. Should be called at the end of the IO handler.
@@ -265,7 +278,7 @@ impl EncryptedConnection {
 	}
 
 	/// Create an encrypted connection out of the handshake. Consumes a handshake object.
-	pub fn new(mut handshake: Handshake) -> Result<EncryptedConnection, UtilError> {
+	pub fn new(handshake: &mut Handshake) -> Result<EncryptedConnection, UtilError> {
 		let shared = try!(crypto::ecdh::agree(handshake.ecdhe.secret(), &handshake.remote_public));
 		let mut nonce_material = H512::new();
 		if handshake.originated {
@@ -300,9 +313,8 @@ impl EncryptedConnection {
 		ingress_mac.update(&mac_material);
 		ingress_mac.update(if handshake.originated { &handshake.ack_cipher } else { &handshake.auth_cipher });
 
-		handshake.connection.expect(ENCRYPTED_HEADER_LEN);
+		let mut enc = EncryptedConnection {
-		Ok(EncryptedConnection {
+			connection: try!(handshake.connection.try_clone()),
-			connection: handshake.connection,
 			encoder: encoder,
 			decoder: decoder,
 			mac_encoder: mac_encoder,
@@ -311,7 +323,9 @@ impl EncryptedConnection {
 			read_state: EncryptedConnectionState::Header,
 			protocol_id: 0,
 			payload_len: 0
-		})
+		};
+		enc.connection.expect(ENCRYPTED_HEADER_LEN);
+		Ok(enc)
 	}
 
 	/// Send a packet
@@ -440,6 +454,12 @@ impl EncryptedConnection {
 		Ok(())
 	}
 
+	/// Register socket with the event lpop. This should be called at the end of the event loop.
+	pub fn register_socket<Host:Handler>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
+		try!(self.connection.register_socket(reg, event_loop));
+		Ok(())
+	}
+
 	/// Update connection registration. This should be called at the end of the event loop.
 	pub fn update_socket<Host:Handler>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
 		try!(self.connection.update_socket(reg, event_loop));

util/src/network/handshake.rs

@@ -63,7 +63,9 @@ pub struct Handshake {
 	/// A copy of received encryped auth packet 
 	pub auth_cipher: Bytes,
 	/// A copy of received encryped ack packet 
-	pub ack_cipher: Bytes
+	pub ack_cipher: Bytes,
+	/// This Handshake is marked for deleteion flag
+	pub expired: bool,
 }
 
 const AUTH_PACKET_SIZE: usize = 307;
@@ -84,6 +86,7 @@ impl Handshake {
 			remote_nonce: H256::new(),
 			auth_cipher: Bytes::new(),
 			ack_cipher: Bytes::new(),
+			expired: false,
 		})
 	}
 
@@ -97,6 +100,16 @@ impl Handshake {
 		self.connection.token()
 	}
 
+	/// Mark this handshake as inactive to be deleted lated.
+	pub fn set_expired(&mut self) {
+		self.expired = true;
+	}
+
+	/// Check if this handshake is expired.
+	pub fn expired(&self) -> bool {
+		self.expired
+	}
+
 	/// 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;
@@ -118,47 +131,56 @@ impl Handshake {
 
 	/// Readable IO handler. Drives the state change.
 	pub fn readable<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo) -> Result<(), UtilError> where Message: Send + Clone {
-		io.clear_timer(self.connection.token).unwrap();
+		if !self.expired() {
-		match self.state {
+			io.clear_timer(self.connection.token).unwrap();
-			HandshakeState::ReadingAuth => {
+			match self.state {
-				if let Some(data) = try!(self.connection.readable()) {
+				HandshakeState::ReadingAuth => {
-					try!(self.read_auth(host, &data));
+					if let Some(data) = try!(self.connection.readable()) {
-					try!(self.write_ack());
+						try!(self.read_auth(host, &data));
-				};
+						try!(self.write_ack());
-			},
+					};
-			HandshakeState::ReadingAck => {
+				},
-				if let Some(data) = try!(self.connection.readable()) {
+				HandshakeState::ReadingAck => {
-					try!(self.read_ack(host, &data));
+					if let Some(data) = try!(self.connection.readable()) {
-					self.state = HandshakeState::StartSession;
+						try!(self.read_ack(host, &data));
-				};
+						self.state = HandshakeState::StartSession;
-			},
+					};
-			HandshakeState::StartSession => {},
+				},
-			_ => { panic!("Unexpected state"); }
+				HandshakeState::StartSession => {},
-		}
+				_ => { panic!("Unexpected state"); }
-		if self.state != HandshakeState::StartSession {
+			}
-			try!(io.update_registration(self.connection.token));
+			if self.state != HandshakeState::StartSession {
+				try!(io.update_registration(self.connection.token));
+			}
 		}
 		Ok(())
 	}
 
 	/// Writabe IO handler.
 	pub fn writable<Message>(&mut self, io: &IoContext<Message>, _host: &HostInfo) -> Result<(), UtilError> where Message: Send + Clone {
-		io.clear_timer(self.connection.token).unwrap();
+		if !self.expired() {
-		try!(self.connection.writable());
+			io.clear_timer(self.connection.token).unwrap();
-		if self.state != HandshakeState::StartSession {
+			try!(self.connection.writable());
-			io.update_registration(self.connection.token).unwrap();
+			if self.state != HandshakeState::StartSession {
+				io.update_registration(self.connection.token).unwrap();
+			}
 		}
 		Ok(())
 	}
 
 	/// Register the socket with the event loop
 	pub fn register_socket<Host:Handler<Timeout=Token>>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
-		try!(self.connection.register_socket(reg, event_loop));
+		if !self.expired() {
+			try!(self.connection.register_socket(reg, event_loop));
+		}
 		Ok(())
 	}
 
+	/// Update socket registration with the event loop.
 	pub fn update_socket<Host:Handler<Timeout=Token>>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
-		try!(self.connection.update_socket(reg, event_loop));
+		if !self.expired() {
+			try!(self.connection.update_socket(reg, event_loop));
+		}
 		Ok(())
 	}
 

util/src/network/host.rs

@@ -26,8 +26,8 @@ use std::io::{Read, Write};
 use std::fs;
 use mio::*;
 use mio::tcp::*;
-use target_info::Target;
 use hash::*;
+use misc::version;
 use crypto::*;
 use sha3::Hashable;
 use rlp::*;
@@ -190,11 +190,11 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
 		let session = { self.sessions.read().unwrap().get(peer).cloned() };
 		if let Some(session) = session {
 			session.lock().unwrap().deref_mut().send_packet(self.protocol, packet_id as u8, &data).unwrap_or_else(|e| {
-						warn!(target: "net", "Send error: {:?}", e);
+						warn!(target: "network", "Send error: {:?}", e);
 					}); //TODO: don't copy vector data
 			try!(self.io.update_registration(peer));
 		} else  {
-			trace!(target: "net", "Send: Peer no longer exist")
+			trace!(target: "network", "Send: Peer no longer exist")
 		}
 		Ok(())
 	}
@@ -360,7 +360,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 				config: config,
 				nonce: H256::random(),
 				protocol_version: PROTOCOL_VERSION,
-				client_version: format!("Parity/{}/{}-{}-{}", env!("CARGO_PKG_VERSION"), Target::arch(), Target::env(), Target::os()),
+				client_version: version(),
 				listen_port: 0,
 				capabilities: Vec::new(),
 			}),
@@ -470,18 +470,18 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 			.take(min(MAX_HANDSHAKES_PER_ROUND, handshake_limit - handshake_count)) {
 			self.connect_peer(&id, io);
 		}
-		debug!(target: "net", "Connecting peers: {} sessions, {} pending", self.session_count(), self.handshake_count());
+		debug!(target: "network", "Connecting peers: {} sessions, {} pending", self.session_count(), self.handshake_count());
 	}
 
 	#[cfg_attr(feature="dev", allow(single_match))]
 	fn connect_peer(&self, id: &NodeId, io: &IoContext<NetworkIoMessage<Message>>) {
 		if self.have_session(id)
 		{
-			trace!("Aborted connect. Node already connected.");
+			trace!(target: "network", "Aborted connect. Node already connected.");
 			return;
 		}
 		if self.connecting_to(id) {
-			trace!("Aborted connect. Node already connecting.");
+			trace!(target: "network", "Aborted connect. Node already connecting.");
 			return;
 		}
 
@@ -493,7 +493,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 					node.endpoint.address
 				}
 				else {
-					debug!("Connection to expired node aborted");
+					debug!(target: "network", "Connection to expired node aborted");
 					return;
 				}
 			};
@@ -515,16 +515,16 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 		if handshakes.insert_with(|token| {
 			let mut handshake = Handshake::new(token, id, socket, &nonce, self.stats.clone()).expect("Can't create handshake");
 			handshake.start(io, &self.info.read().unwrap(), id.is_some()).and_then(|_| io.register_stream(token)).unwrap_or_else (|e| {
-				debug!(target: "net", "Handshake create error: {:?}", e);
+				debug!(target: "network", "Handshake create error: {:?}", e);
 			});
 			Arc::new(Mutex::new(handshake))
 		}).is_none() {
-			debug!("Max handshakes reached");
+			debug!(target: "network", "Max handshakes reached");
 		}
 	}
 
 	fn accept(&self, io: &IoContext<NetworkIoMessage<Message>>) {
-		trace!(target: "net", "accept");
+		trace!(target: "network", "Accepting incoming connection");
 		loop {
 			let socket = match self.tcp_listener.lock().unwrap().accept() {
 				Ok(None) => break,
@@ -544,7 +544,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 		if let Some(handshake) = handshake {
 			let mut h = handshake.lock().unwrap();
 			if let Err(e) = h.writable(io, &self.info.read().unwrap()) {
-				debug!(target: "net", "Handshake write error: {}:{:?}", token, e);
+				trace!(target: "network", "Handshake write error: {}: {:?}", token, e);
 			}
 		}
 	}
@@ -554,9 +554,9 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 		if let Some(session) = session {
 			let mut s = session.lock().unwrap();
 			if let Err(e) = s.writable(io, &self.info.read().unwrap()) {
-				debug!(target: "net", "Session write error: {}:{:?}", token, e);
+				trace!(target: "network", "Session write error: {}: {:?}", token, e);
 			}
-			io.update_registration(token).unwrap_or_else(|e| debug!(target: "net", "Session registration error: {:?}", e));
+			io.update_registration(token).unwrap_or_else(|e| debug!(target: "network", "Session registration error: {:?}", e));
 		}
 	}
 
@@ -571,7 +571,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 		if let Some(handshake) = handshake {
 			let mut h = handshake.lock().unwrap();
 			if let Err(e) = h.readable(io, &self.info.read().unwrap()) {
-				debug!(target: "net", "Handshake read error: {}:{:?}", token, e);
+				debug!(target: "network", "Handshake read error: {}: {:?}", token, e);
 				kill = true;
 			}
 			if h.done() {
@@ -583,9 +583,9 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 			return;
 		} else if create_session {
 			self.start_session(token, io);
-			io.update_registration(token).unwrap_or_else(|e| debug!(target: "net", "Session registration error: {:?}", e));
+			return;
 		}
-		io.update_registration(token).unwrap_or_else(|e| debug!(target: "net", "Token registration error: {:?}", e));
+		io.update_registration(token).unwrap_or_else(|e| debug!(target: "network", "Token registration error: {:?}", e));
 	}
 
 	fn session_readable(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) {
@@ -597,7 +597,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 			let mut s = session.lock().unwrap();
 			match s.readable(io, &self.info.read().unwrap()) {
 				Err(e) => {
-					debug!(target: "net", "Session read error: {}:{:?}", token, e);
+					debug!(target: "network", "Session read error: {}: {:?}", token, e);
 					kill = true;
 				},
 				Ok(SessionData::Ready) => {
@@ -613,7 +613,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 					packet_id,
 				}) => {
 					match self.handlers.read().unwrap().get(protocol) {
-						None => { warn!(target: "net", "No handler found for protocol: {:?}", protocol) },
+						None => { warn!(target: "network", "No handler found for protocol: {:?}", protocol) },
 						Some(_) => packet_data = Some((protocol, packet_id, data)),
 					}
 				},
@@ -631,7 +631,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 			let h = self.handlers.read().unwrap().get(p).unwrap().clone();
 			h.read(&NetworkContext::new(io, p, Some(token), self.sessions.clone()), &token, packet_id, &data[1..]);
 		}
-		io.update_registration(token).unwrap_or_else(|e| debug!(target: "net", "Token registration error: {:?}", e));
+		io.update_registration(token).unwrap_or_else(|e| debug!(target: "network", "Token registration error: {:?}", e));
 	}
 
 	fn start_session(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) {
@@ -642,26 +642,25 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 
 		// turn a handshake into a session
 		let mut sessions = self.sessions.write().unwrap();
-		let mut h = handshakes.remove(token).unwrap();
+		let mut h = handshakes.get_mut(token).unwrap().lock().unwrap();
-		// wait for other threads to stop using it
+		if h.expired {
-		{
+			return;
-			while Arc::get_mut(&mut h).is_none() {
-				h.lock().ok();
-			}
 		}
-		let h = Arc::try_unwrap(h).ok().unwrap().into_inner().unwrap();
 		let originated = h.originated;
-		let mut session = match Session::new(h, &self.info.read().unwrap()) {
+		let mut session = match Session::new(&mut h, &self.info.read().unwrap()) {
 			Ok(s) => s,
 			Err(e) => {
-				debug!("Session creation error: {:?}", e);
+				debug!(target: "network", "Session creation error: {:?}", e);
 				return;
 			}
 		};
 		let result = sessions.insert_with(move |session_token| {
 			session.set_token(session_token);
-			io.update_registration(session_token).expect("Error updating session registration");
+			io.deregister_stream(token).expect("Error deleting handshake registration");
+			h.set_expired();
+			io.register_stream(session_token).expect("Error creating session registration");
 			self.stats.inc_sessions();
+			trace!(target: "network", "Creating session {} -> {}", token, session_token);
 			if !originated {
 				// Add it no node table
 				if let Ok(address) = session.remote_addr() {
@@ -690,26 +689,34 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
 			FIRST_HANDSHAKE ... LAST_HANDSHAKE => {
 				let handshakes = self.handshakes.write().unwrap();
 				if let Some(handshake) = handshakes.get(token).cloned() {
-					failure_id = Some(handshake.lock().unwrap().id().clone());
+					let mut handshake = handshake.lock().unwrap();
+					if !handshake.expired() {
+						handshake.set_expired();
+						failure_id = Some(handshake.id().clone());
+						io.deregister_stream(token).expect("Error deregistering stream");
+					}
 				}
 			},
 			FIRST_SESSION ... LAST_SESSION => {
 				let sessions = self.sessions.write().unwrap();
 				if let Some(session) = sessions.get(token).cloned() {
-					let s = session.lock().unwrap();
+					let mut s = session.lock().unwrap();
-					if s.is_ready() {
+					if !s.expired() {
-						for (p, _) in self.handlers.read().unwrap().iter() {
+						if s.is_ready() {
-							if s.have_capability(p)  {
+							for (p, _) in self.handlers.read().unwrap().iter() {
-								to_disconnect.push(p);
+								if s.have_capability(p)  {
+									to_disconnect.push(p);
+								}
 							}
 						}
+						s.set_expired();
+						failure_id = Some(s.id().clone());
+						io.deregister_stream(token).expect("Error deregistering stream");
 					}
-					failure_id = Some(s.id().clone());
 				}
 			},
 			_ => {},
 		}
-		io.deregister_stream(token).expect("Error deregistering stream");
 		if let Some(id) = failure_id {
 			if remote {
 				self.nodes.write().unwrap().note_failure(&id);
@@ -764,11 +771,11 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
 	}
 
 	fn stream_hup(&self, io: &IoContext<NetworkIoMessage<Message>>, stream: StreamToken) {
-		trace!(target: "net", "Hup: {}", stream);
+		trace!(target: "network", "Hup: {}", stream);
 		match stream {
 			FIRST_SESSION ... LAST_SESSION => self.connection_closed(stream, io),
 			FIRST_HANDSHAKE ... LAST_HANDSHAKE => self.connection_closed(stream, io),
-			_ => warn!(target: "net", "Unexpected hup"),
+			_ => warn!(target: "network", "Unexpected hup"),
 		};
 	}
 
@@ -777,7 +784,8 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
 			FIRST_SESSION ... LAST_SESSION => self.session_readable(stream, io),
 			FIRST_HANDSHAKE ... LAST_HANDSHAKE => self.handshake_readable(stream, io),
 			DISCOVERY => {
-				if let Some(node_changes) = self.discovery.as_ref().unwrap().lock().unwrap().readable() {
+				let node_changes = { self.discovery.as_ref().unwrap().lock().unwrap().readable() };
+				if let Some(node_changes) = node_changes {
 					self.update_nodes(io, node_changes);
 				}
 				io.update_registration(DISCOVERY).expect("Error updating discovery registration");
@@ -809,14 +817,15 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
 				io.update_registration(DISCOVERY).expect("Error updating discovery registration");
 			},
 			DISCOVERY_ROUND => {
-				if let Some(node_changes) = self.discovery.as_ref().unwrap().lock().unwrap().round() {
+				let node_changes = { self.discovery.as_ref().unwrap().lock().unwrap().round() };
+				if let Some(node_changes) = node_changes {
 					self.update_nodes(io, node_changes);
 				}
 				io.update_registration(DISCOVERY).expect("Error updating discovery registration");
 			},
 			_ => match self.timers.read().unwrap().get(&token).cloned() {
 				Some(timer) => match self.handlers.read().unwrap().get(timer.protocol).cloned() {
-						None => { warn!(target: "net", "No handler found for protocol: {:?}", timer.protocol) },
+						None => { warn!(target: "network", "No handler found for protocol: {:?}", timer.protocol) },
 						Some(h) => { h.timeout(&NetworkContext::new(io, timer.protocol, None, self.sessions.clone()), timer.token); }
 				},
 				None => { warn!("Unknown timer token: {}", token); } // timer is not registerd through us
@@ -872,7 +881,10 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
 	fn register_stream(&self, stream: StreamToken, reg: Token, event_loop: &mut EventLoop<IoManager<NetworkIoMessage<Message>>>) {
 		match stream {
 			FIRST_SESSION ... LAST_SESSION => {
-				warn!("Unexpected session stream registration");
+				let session = { self.sessions.read().unwrap().get(stream).cloned() };
+				if let Some(session) = session {
+					session.lock().unwrap().register_socket(reg, event_loop).expect("Error registering socket");
+				}
 			}
 			FIRST_HANDSHAKE ... LAST_HANDSHAKE => {
 				let connection = { self.handshakes.read().unwrap().get(stream).cloned() };

util/src/network/session.rs

@@ -41,6 +41,8 @@ pub struct Session {
 	connection: EncryptedConnection,
 	/// Session ready flag. Set after successfull Hello packet exchange
 	had_hello: bool,
+	/// Session is no longer active flag.
+	expired: bool,
 	ping_time_ns: u64,
 	pong_time_ns: Option<u64>,
 }
@@ -109,8 +111,9 @@ const PACKET_USER: u8 = 0x10;
 const PACKET_LAST: u8 = 0x7f;
 
 impl Session {
-	/// Create a new session out of comepleted handshake. Consumes handshake object.
+	/// Create a new session out of comepleted handshake. This clones the handshake connection object 
-	pub fn new(h: Handshake, host: &HostInfo) -> Result<Session, UtilError> {
+	/// and leaves the handhsake in limbo to be deregistered from the event loop.
+	pub fn new(h: &mut Handshake, host: &HostInfo) -> Result<Session, UtilError> {
 		let id = h.id.clone();
 		let connection = try!(EncryptedConnection::new(h));
 		let mut session = Session {
@@ -125,6 +128,7 @@ impl Session {
 			},
 			ping_time_ns: 0,
 			pong_time_ns: None,
+			expired: false,
 		};
 		try!(session.write_hello(host));
 		try!(session.send_ping());
@@ -141,6 +145,16 @@ impl Session {
 		self.had_hello
 	}
 
+	/// Mark this session as inactive to be deleted lated.
+	pub fn set_expired(&mut self) {
+		self.expired = true;
+	}
+
+	/// Check if this session is expired.
+	pub fn expired(&self) -> bool {
+		self.expired
+	}
+
 	/// Replace socket token 
 	pub fn set_token(&mut self, token: StreamToken) {
 		self.connection.set_token(token);
@@ -153,6 +167,9 @@ impl Session {
 
 	/// 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 {
+		if self.expired() {
+			return Ok(SessionData::None) 
+		}
 		match try!(self.connection.readable(io)) {
 			Some(data) => Ok(try!(self.read_packet(data, host))),
 			None => Ok(SessionData::None)
@@ -161,6 +178,9 @@ impl Session {
 
 	/// Writable IO handler. Sends pending packets.
 	pub fn writable<Message>(&mut self, io: &IoContext<Message>, _host: &HostInfo) -> Result<(), UtilError> where Message: Send + Sync + Clone {
+		if self.expired() {
+			return Ok(()) 
+		}
 		self.connection.writable(io)
 	}
 
@@ -169,8 +189,20 @@ impl Session {
 		self.info.capabilities.iter().any(|c| c.protocol == protocol)
 	}
 
+	/// Register the session socket with the event loop
+	pub fn register_socket<Host:Handler<Timeout = Token>>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
+		if self.expired() {
+			return Ok(());
+		}
+		try!(self.connection.register_socket(reg, event_loop));
+		Ok(())
+	}
+
 	/// Update registration with the event loop. Should be called at the end of the IO handler.
 	pub fn update_socket<Host:Handler>(&self, reg:Token, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
+		if self.expired() {
+			return Ok(());
+		}
 		self.connection.update_socket(reg, event_loop)
 	}
 

util/src/network/tests.rs

@@ -74,7 +74,7 @@ impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol {
 	}
 
 	fn connected(&self, io: &NetworkContext<TestProtocolMessage>, peer: &PeerId) {
-		assert!(io.peer_info(*peer).contains("parity"));
+		assert!(io.peer_info(*peer).contains("Parity"));
 		if self.drop_session {
 			io.disconnect_peer(*peer)
 		} else {