Sync channel for consensus test

ethcore/src/client/client.rs

@@ -580,6 +580,12 @@ impl Client {
 		self.miner.clone()
 	}
 
+
+	/// Replace io channel. Useful for testing.
+	pub fn set_io_channel(&self, io_channel: IoChannel<ClientIoMessage>) {
+		*self.io_channel.lock() = io_channel;
+	}
+
 	/// Attempt to get a copy of a specific block's final state.
 	///
 	/// This will not fail if given BlockId::Latest.
@@ -1289,7 +1295,8 @@ impl BlockChainClient for Client {
 	}
 
 	fn queue_consensus_message(&self, message: Bytes) {
-		if let Err(e) = self.io_channel.lock().send(ClientIoMessage::NewMessage(message)) {
+		let channel = self.io_channel.lock().clone();
+		if let Err(e) = channel.send(ClientIoMessage::NewMessage(message)) {
 			debug!("Ignoring the message, error queueing: {}", e);
 		}
 	}

ethcore/src/client/test_client.rs

@@ -678,7 +678,7 @@ impl BlockChainClient for TestBlockChainClient {
 	}
 
 	fn queue_consensus_message(&self, message: Bytes) {
-		self.spec.engine.handle_message(UntrustedRlp::new(&message)).unwrap();
+		self.spec.engine.handle_message(&message).unwrap();
 	}
 
 	fn pending_transactions(&self) -> Vec<SignedTransaction> {

ethcore/src/engines/mod.rs

@@ -28,7 +28,6 @@ pub use self::basic_authority::BasicAuthority;
 pub use self::authority_round::AuthorityRound;
 pub use self::tendermint::Tendermint;
 
-use rlp::UntrustedRlp;
 use util::*;
 use account_provider::AccountProvider;
 use block::ExecutedBlock;
@@ -178,7 +177,7 @@ pub trait Engine : Sync + Send {
 
 	/// Handle any potential consensus messages;
 	/// updating consensus state and potentially issuing a new one.
-	fn handle_message(&self, _message: UntrustedRlp) -> Result<(), Error> { Err(EngineError::UnexpectedMessage.into()) }
+	fn handle_message(&self, _message: &[u8]) -> Result<(), Error> { Err(EngineError::UnexpectedMessage.into()) }
 
 	// TODO: builtin contract routing - to do this properly, it will require removing the built-in configuration-reading logic
 	// from Spec into here and removing the Spec::builtins field.

ethcore/src/engines/tendermint/mod.rs

@@ -146,7 +146,8 @@ impl Tendermint {
 	}
 
 	fn broadcast_message(&self, message: Bytes) {
-		if let Some(ref channel) = *self.message_channel.lock() {
+		let channel = self.message_channel.lock().clone();
+		if let Some(ref channel) = channel {
 			match channel.send(ClientIoMessage::BroadcastMessage(message)) {
 				Ok(_) => trace!(target: "poa", "broadcast_message: BroadcastMessage message sent."),
 				Err(err) => warn!(target: "poa", "broadcast_message: Could not send a sealing message {}.", err),
@@ -449,7 +450,8 @@ impl Engine for Tendermint {
 		}
 	}
 
-	fn handle_message(&self, rlp: UntrustedRlp) -> Result<(), Error> {
+	fn handle_message(&self, rlp: &[u8]) -> Result<(), Error> {
+		let rlp = UntrustedRlp::new(rlp);
 		let message: ConsensusMessage = try!(rlp.as_val());
 		if !self.votes.is_old_or_known(&message) {
 			let sender = public_to_address(&try!(recover(&message.signature.into(), &try!(rlp.at(1)).as_raw().sha3())));

ethcore/src/service.rs

@@ -17,7 +17,6 @@
 //! Creates and registers client and network services.
 
 use util::*;
-use rlp::{UntrustedRlp, View};
 use io::*;
 use spec::Spec;
 use error::*;
@@ -228,7 +227,7 @@ impl IoHandler<ClientIoMessage> for ClientIoHandler {
 			ClientIoMessage::UpdateSealing => self.client.update_sealing(),
 			ClientIoMessage::SubmitSeal(ref hash, ref seal) => self.client.submit_seal(*hash, seal.clone()),
 			ClientIoMessage::BroadcastMessage(ref message) => self.client.broadcast_message(message.clone()),
-			ClientIoMessage::NewMessage(ref message) => if let Err(e) = self.client.engine().handle_message(UntrustedRlp::new(message)) {
+			ClientIoMessage::NewMessage(ref message) => if let Err(e) = self.client.engine().handle_message(message) {
 				trace!(target: "poa", "Invalid message received: {}", e);
 			},
 			_ => {} // ignore other messages

parity/informant.rs

@@ -183,7 +183,7 @@ impl ChainNotify for Informant {
 		let ripe = Instant::now() > *last_import + Duration::from_secs(1) && !importing;
 		let txs_imported = imported.iter()
 			.take(imported.len().saturating_sub(if ripe { 1 } else { 0 }))
-			.filter_map(|h| self.client.block(BlockID::Hash(*h)))
+			.filter_map(|h| self.client.block(BlockId::Hash(*h)))
 			.map(|b| BlockView::new(&b).transactions_count())
 			.sum();
 

sync/src/chain.rs

@@ -2041,7 +2041,8 @@ impl ChainSync {
 	}
 
 	/// Called when peer sends us new consensus packet
-	fn on_consensus_packet(io: &mut SyncIo, _peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
+	fn on_consensus_packet(io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
+		trace!(target: "sync", "Received consensus packet from {:?}", peer_id);
 		io.chain().queue_consensus_message(r.as_raw().to_vec());
 		Ok(())
 	}
@@ -2116,9 +2117,9 @@ mod tests {
 	#[test]
 	fn return_receipts_empty() {
 		let mut client = TestBlockChainClient::new();
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let ss = TestSnapshotService::new();
-		let io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let result = ChainSync::return_receipts(&io, &UntrustedRlp::new(&[0xc0]), 0);
 
@@ -2128,10 +2129,10 @@ mod tests {
 	#[test]
 	fn return_receipts() {
 		let mut client = TestBlockChainClient::new();
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let sync = dummy_sync_with_peer(H256::new(), &client);
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let mut receipt_list = RlpStream::new_list(4);
 		receipt_list.append(&H256::from("0000000000000000000000000000000000000000000000005555555555555555"));
@@ -2152,7 +2153,7 @@ mod tests {
 
 		io.sender = Some(2usize);
 		ChainSync::dispatch_packet(&RwLock::new(sync), &mut io, 0usize, super::GET_RECEIPTS_PACKET, &receipts_request);
-		assert_eq!(1, io.queue.len());
+		assert_eq!(1, io.packets.len());
 	}
 
 	#[test]
@@ -2185,9 +2186,9 @@ mod tests {
 		let headers: Vec<_> = blocks.iter().map(|b| Rlp::new(b).at(0).as_raw().to_vec()).collect();
 		let hashes: Vec<_> = headers.iter().map(|h| HeaderView::new(h).sha3()).collect();
 
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let ss = TestSnapshotService::new();
-		let io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let unknown: H256 = H256::new();
 		let result = ChainSync::return_block_headers(&io, &UntrustedRlp::new(&make_hash_req(&unknown, 1, 0, false)), 0);
@@ -2223,10 +2224,10 @@ mod tests {
 	#[test]
 	fn return_nodes() {
 		let mut client = TestBlockChainClient::new();
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let sync = dummy_sync_with_peer(H256::new(), &client);
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let mut node_list = RlpStream::new_list(3);
 		node_list.append(&H256::from("0000000000000000000000000000000000000000000000005555555555555555"));
@@ -2249,7 +2250,7 @@ mod tests {
 		io.sender = Some(2usize);
 
 		ChainSync::dispatch_packet(&RwLock::new(sync), &mut io, 0usize, super::GET_NODE_DATA_PACKET, &node_request);
-		assert_eq!(1, io.queue.len());
+		assert_eq!(1, io.packets.len());
 	}
 
 	fn dummy_sync_with_peer(peer_latest_hash: H256, client: &BlockChainClient) -> ChainSync {
@@ -2280,11 +2281,11 @@ mod tests {
 	fn finds_lagging_peers() {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(100, EachBlockWith::Uncle);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(10), &client);
 		let chain_info = client.chain_info();
 		let ss = TestSnapshotService::new();
-		let io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let lagging_peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging);
 		let current_peers = sync.get_peers(&chain_info, &io, PeerStatus::Current);
@@ -2297,11 +2298,11 @@ mod tests {
 	fn finds_current_peers() {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(100, EachBlockWith::Uncle);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let mut sync = dummy_sync_with_peer(client.block_hash(BlockId::Latest).unwrap(), &client);
 		let chain_info = client.chain_info();
 		let ss = TestSnapshotService::new();
-		let io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let current_peers = sync.get_peers(&chain_info, &io, PeerStatus::Current);
 		let lagging_peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging);
@@ -2331,79 +2332,79 @@ mod tests {
 	fn sends_new_hashes_to_lagging_peer() {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(100, EachBlockWith::Uncle);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client);
 		let chain_info = client.chain_info();
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging);
 		let peer_count = sync.propagate_new_hashes(&chain_info, &mut io, &peers);
 
 		// 1 message should be send
-		assert_eq!(1, io.queue.len());
+		assert_eq!(1, io.packets.len());
 		// 1 peer should be updated
 		assert_eq!(1, peer_count);
 		// NEW_BLOCK_HASHES_PACKET
-		assert_eq!(0x01, io.queue[0].packet_id);
+		assert_eq!(0x01, io.packets[0].packet_id);
 	}
 
 	#[test]
 	fn sends_latest_block_to_lagging_peer() {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(100, EachBlockWith::Uncle);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client);
 		let chain_info = client.chain_info();
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 		let peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging);
 		let peer_count = sync.propagate_blocks(&chain_info, &mut io, &[], &peers);
 
 		// 1 message should be send
-		assert_eq!(1, io.queue.len());
+		assert_eq!(1, io.packets.len());
 		// 1 peer should be updated
 		assert_eq!(1, peer_count);
 		// NEW_BLOCK_PACKET
-		assert_eq!(0x07, io.queue[0].packet_id);
+		assert_eq!(0x07, io.packets[0].packet_id);
 	}
 
 	#[test]
 	fn sends_sealed_block() {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(100, EachBlockWith::Uncle);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let hash = client.block_hash(BlockId::Number(99)).unwrap();
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client);
 		let chain_info = client.chain_info();
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 		let peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging);
 		let peer_count = sync.propagate_blocks(&chain_info, &mut io, &[hash.clone()], &peers);
 
 		// 1 message should be send
-		assert_eq!(1, io.queue.len());
+		assert_eq!(1, io.packets.len());
 		// 1 peer should be updated
 		assert_eq!(1, peer_count);
 		// NEW_BLOCK_PACKET
-		assert_eq!(0x07, io.queue[0].packet_id);
+		assert_eq!(0x07, io.packets[0].packet_id);
 	}
 
 	#[test]
 	fn sends_proposed_block() {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(100, EachBlockWith::Uncle);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let block = client.block(BlockId::Latest).unwrap();
 		let mut sync = dummy_sync_with_peer(client.block_hash(BlockId::Latest).unwrap(), &client);
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 		sync.propagate_proposed_blocks(&mut io, &[block]);
 
 		// 1 message should be send
-		assert_eq!(1, io.queue.len());
+		assert_eq!(1, io.packets.len());
 		// NEW_BLOCK_PACKET
-		assert_eq!(0x07, io.queue[0].packet_id);
+		assert_eq!(0x07, io.packets[0].packet_id);
 	}
 
 	#[test]
@@ -2412,9 +2413,9 @@ mod tests {
 		client.add_blocks(100, EachBlockWith::Uncle);
 		client.insert_transaction_to_queue();
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(1), &client);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 		let peer_count = sync.propagate_new_transactions(&mut io);
 		// Try to propagate same transactions for the second time
 		let peer_count2 = sync.propagate_new_transactions(&mut io);
@@ -2424,13 +2425,13 @@ mod tests {
 		let peer_count3 = sync.propagate_new_transactions(&mut io);
 
 		// 1 message should be send
-		assert_eq!(1, io.queue.len());
+		assert_eq!(1, io.packets.len());
 		// 1 peer should be updated but only once
 		assert_eq!(1, peer_count);
 		assert_eq!(0, peer_count2);
 		assert_eq!(0, peer_count3);
 		// TRANSACTIONS_PACKET
-		assert_eq!(0x02, io.queue[0].packet_id);
+		assert_eq!(0x02, io.packets[0].packet_id);
 	}
 
 	#[test]
@@ -2439,21 +2440,21 @@ mod tests {
 		client.add_blocks(100, EachBlockWith::Uncle);
 		client.insert_transaction_to_queue();
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(1), &client);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 		let peer_count = sync.propagate_new_transactions(&mut io);
 		io.chain.insert_transaction_to_queue();
 		// New block import should trigger propagation.
 		sync.chain_new_blocks(&mut io, &[], &[], &[], &[], &[], &[]);
 
 		// 2 message should be send
-		assert_eq!(2, io.queue.len());
+		assert_eq!(2, io.packets.len());
 		// 1 peer should receive the message
 		assert_eq!(1, peer_count);
 		// TRANSACTIONS_PACKET
-		assert_eq!(0x02, io.queue[0].packet_id);
+		assert_eq!(0x02, io.packets[0].packet_id);
-		assert_eq!(0x02, io.queue[1].packet_id);
+		assert_eq!(0x02, io.packets[1].packet_id);
 	}
 
 	#[test]
@@ -2462,31 +2463,34 @@ mod tests {
 		client.add_blocks(100, EachBlockWith::Uncle);
 		client.insert_transaction_to_queue();
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(1), &client);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let ss = TestSnapshotService::new();
 		// should sent some
 		{
-			let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+			let mut io = TestIo::new(&mut client, &ss, &queue, None);
 			let peer_count = sync.propagate_new_transactions(&mut io);
-			assert_eq!(1, io.queue.len());
+			assert_eq!(1, io.packets.len());
 			assert_eq!(1, peer_count);
 		}
 		// Insert some more
 		client.insert_transaction_to_queue();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let (peer_count2, peer_count3) = {
-		// Propagate new transactions
+			let mut io = TestIo::new(&mut client, &ss, &queue, None);
-		let peer_count2 = sync.propagate_new_transactions(&mut io);
+			// Propagate new transactions
-		// And now the peer should have all transactions
+			let peer_count2 = sync.propagate_new_transactions(&mut io);
-		let peer_count3 = sync.propagate_new_transactions(&mut io);
+			// And now the peer should have all transactions
+			let peer_count3 = sync.propagate_new_transactions(&mut io);
+			(peer_count2, peer_count3)
+		};
 
 		// 2 message should be send (in total)
-		assert_eq!(2, io.queue.len());
+		assert_eq!(2, queue.read().len());
 		// 1 peer should be updated but only once after inserting new transaction
 		assert_eq!(1, peer_count2);
 		assert_eq!(0, peer_count3);
 		// TRANSACTIONS_PACKET
-		assert_eq!(0x02, io.queue[0].packet_id);
+		assert_eq!(0x02, queue.read()[0].packet_id);
-		assert_eq!(0x02, io.queue[1].packet_id);
+		assert_eq!(0x02, queue.read()[1].packet_id);
 	}
 
 	#[test]
@@ -2495,9 +2499,9 @@ mod tests {
 		client.add_blocks(100, EachBlockWith::Uncle);
 		client.insert_transaction_to_queue();
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(1), &client);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 		sync.propagate_new_transactions(&mut io);
 
 		let stats = sync.transactions_stats();
@@ -2511,11 +2515,11 @@ mod tests {
 
 		let block_data = get_dummy_block(11, client.chain_info().best_block_hash);
 
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client);
 		//sync.have_common_block = true;
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let block = UntrustedRlp::new(&block_data);
 
@@ -2531,10 +2535,10 @@ mod tests {
 
 		let block_data = get_dummy_blocks(11, client.chain_info().best_block_hash);
 
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client);
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let block = UntrustedRlp::new(&block_data);
 
@@ -2547,10 +2551,10 @@ mod tests {
 	fn handles_peer_new_block_empty() {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(10, EachBlockWith::Uncle);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client);
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let empty_data = vec![];
 		let block = UntrustedRlp::new(&empty_data);
@@ -2564,10 +2568,10 @@ mod tests {
 	fn handles_peer_new_hashes() {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(10, EachBlockWith::Uncle);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client);
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let hashes_data = get_dummy_hashes();
 		let hashes_rlp = UntrustedRlp::new(&hashes_data);
@@ -2581,10 +2585,10 @@ mod tests {
 	fn handles_peer_new_hashes_empty() {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(10, EachBlockWith::Uncle);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client);
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let empty_hashes_data = vec![];
 		let hashes_rlp = UntrustedRlp::new(&empty_hashes_data);
@@ -2600,16 +2604,16 @@ mod tests {
 	fn hashes_rlp_mutually_acceptable() {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(100, EachBlockWith::Uncle);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client);
 		let chain_info = client.chain_info();
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging);
 		sync.propagate_new_hashes(&chain_info, &mut io, &peers);
 
-		let data = &io.queue[0].data.clone();
+		let data = &io.packets[0].data.clone();
 		let result = sync.on_peer_new_hashes(&mut io, 0, &UntrustedRlp::new(data));
 		assert!(result.is_ok());
 	}
@@ -2620,16 +2624,16 @@ mod tests {
 	fn block_rlp_mutually_acceptable() {
 		let mut client = TestBlockChainClient::new();
 		client.add_blocks(100, EachBlockWith::Uncle);
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5), &client);
 		let chain_info = client.chain_info();
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 
 		let peers = sync.get_peers(&chain_info, &io, PeerStatus::Lagging);
 		sync.propagate_blocks(&chain_info, &mut io, &[], &peers);
 
-		let data = &io.queue[0].data.clone();
+		let data = &io.packets[0].data.clone();
 		let result = sync.on_peer_new_block(&mut io, 0, &UntrustedRlp::new(data));
 		assert!(result.is_ok());
 	}
@@ -2657,9 +2661,9 @@ mod tests {
 
 		// when
 		{
-			let mut queue = VecDeque::new();
+			let queue = RwLock::new(VecDeque::new());
 			let ss = TestSnapshotService::new();
-			let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+			let mut io = TestIo::new(&mut client, &ss, &queue, None);
 			io.chain.miner.chain_new_blocks(io.chain, &[], &[], &[], &good_blocks);
 			sync.chain_new_blocks(&mut io, &[], &[], &[], &good_blocks, &[], &[]);
 			assert_eq!(io.chain.miner.status().transactions_in_future_queue, 0);
@@ -2672,9 +2676,9 @@ mod tests {
 			client.set_nonce(view.transactions()[0].sender().unwrap(), U256::from(1));
 		}
 		{
-			let mut queue = VecDeque::new();
+			let queue = RwLock::new(VecDeque::new());
 			let ss = TestSnapshotService::new();
-			let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+			let mut io = TestIo::new(&client, &ss, &queue, None);
 			io.chain.miner.chain_new_blocks(io.chain, &[], &[], &good_blocks, &retracted_blocks);
 			sync.chain_new_blocks(&mut io, &[], &[], &good_blocks, &retracted_blocks, &[], &[]);
 		}
@@ -2697,9 +2701,9 @@ mod tests {
 		let good_blocks = vec![client.block_hash_delta_minus(2)];
 		let retracted_blocks = vec![client.block_hash_delta_minus(1)];
 
-		let mut queue = VecDeque::new();
+		let queue = RwLock::new(VecDeque::new());
 		let ss = TestSnapshotService::new();
-		let mut io = TestIo::new(&mut client, &ss, &mut queue, None);
+		let mut io = TestIo::new(&mut client, &ss, &queue, None);
 
 		// when
 		sync.chain_new_blocks(&mut io, &[], &[], &[], &good_blocks, &[], &[]);

sync/src/tests/chain.rs

@@ -101,7 +101,7 @@ fn forked_with_misbehaving_peer() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(3);
 	// peer 0 is on a totally different chain with higher total difficulty
-	net.peer_mut(0).chain = TestBlockChainClient::new_with_extra_data(b"fork".to_vec());
+	net.peer_mut(0).chain = Arc::new(TestBlockChainClient::new_with_extra_data(b"fork".to_vec()));
 	net.peer(0).chain.add_blocks(50, EachBlockWith::Nothing);
 	net.peer(1).chain.add_blocks(10, EachBlockWith::Nothing);
 	net.peer(2).chain.add_blocks(10, EachBlockWith::Nothing);

sync/src/tests/consensus.rs

@@ -15,7 +15,9 @@
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
 use util::*;
-use ethcore::client::BlockChainClient;
+use io::{IoHandler, IoContext, IoChannel};
+use ethcore::client::{BlockChainClient, Client, MiningBlockChainClient};
+use ethcore::service::{ClientIoMessage};
 use ethcore::spec::Spec;
 use ethcore::miner::MinerService;
 use ethcore::transaction::*;
@@ -24,23 +26,62 @@ use ethkey::KeyPair;
 use super::helpers::*;
 use SyncConfig;
 
+struct TestIoHandler {
+	client: Arc<Client>,
+}
+
+impl IoHandler<ClientIoMessage> for TestIoHandler {
+	fn message(&self, _io: &IoContext<ClientIoMessage>, net_message: &ClientIoMessage) {
+		match *net_message {
+			ClientIoMessage::UpdateSealing => self.client.update_sealing(),
+			ClientIoMessage::SubmitSeal(ref hash, ref seal) => self.client.submit_seal(*hash, seal.clone()),
+			ClientIoMessage::BroadcastMessage(ref message) => self.client.broadcast_message(message.clone()),
+			ClientIoMessage::NewMessage(ref message) => if let Err(e) = self.client.engine().handle_message(message) {
+				panic!("Invalid message received: {}", e);
+			},
+			_ => {} // ignore other messages
+		}
+	}
+}
+
 #[test]
-fn test_authority_round() {
+fn authority_round() {
-	let s1 = KeyPair::from_secret("1".sha3()).unwrap();
+	let s0 = KeyPair::from_secret("1".sha3()).unwrap();
-	let s2 = KeyPair::from_secret("0".sha3()).unwrap();
+	let s1 = KeyPair::from_secret("0".sha3()).unwrap();
 	let spec_factory = || {
 		let spec = Spec::new_test_round();
 		let account_provider = AccountProvider::transient_provider();
+		account_provider.insert_account(s0.secret().clone(), "").unwrap();
 		account_provider.insert_account(s1.secret().clone(), "").unwrap();
-		account_provider.insert_account(s2.secret().clone(), "").unwrap();
 		spec.engine.register_account_provider(Arc::new(account_provider));
 		spec
 	};
-	let mut net = TestNet::new_with_spec(2, SyncConfig::default(), spec_factory);
+	let mut net = TestNet::with_spec(2, SyncConfig::default(), spec_factory);
 	let mut net = &mut *net;
-	// Push transaction to both clients. Only one of them gets lucky to mine a block.
+	let io_handler0: Arc<IoHandler<ClientIoMessage>> = Arc::new(TestIoHandler { client: net.peer(0).chain.clone() });
-	net.peer(0).chain.miner().set_engine_signer(s1.address(), "".to_owned()).unwrap();
+	let io_handler1: Arc<IoHandler<ClientIoMessage>> = Arc::new(TestIoHandler { client: net.peer(1).chain.clone() });
-	net.peer(1).chain.miner().set_engine_signer(s2.address(), "".to_owned()).unwrap();
+	// Push transaction to both clients. Only one of them gets lucky to produce a block.
+	net.peer(0).chain.miner().set_engine_signer(s0.address(), "".to_owned()).unwrap();
+	net.peer(1).chain.miner().set_engine_signer(s1.address(), "".to_owned()).unwrap();
+	net.peer(0).chain.engine().register_message_channel(IoChannel::to_handler(Arc::downgrade(&io_handler0)));
+	net.peer(1).chain.engine().register_message_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1)));
+	net.peer(0).chain.set_io_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1)));
+	net.peer(1).chain.set_io_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1)));
+	let tx0 = Transaction {
+		nonce: 0.into(),
+		gas_price: 0.into(),
+		gas: 21000.into(),
+		action: Action::Call(Address::default()),
+		value: 0.into(),
+		data: Vec::new(),
+	}.sign(s0.secret(), None);
+	// exchange statuses
+	net.sync();
+	net.peer(0).chain.miner().import_own_transaction(&*net.peer(0).chain, tx0).unwrap();
+	net.sync();
+	assert_eq!(net.peer(0).chain.chain_info().best_block_number, 1);
+	assert_eq!(net.peer(1).chain.chain_info().best_block_number, 1);
+
 	let tx1 = Transaction {
 		nonce: 0.into(),
 		gas_price: 0.into(),
@@ -49,59 +90,49 @@ fn test_authority_round() {
 		value: 0.into(),
 		data: Vec::new(),
 	}.sign(s1.secret(), None);
-	// exhange statuses
+	net.peer(1).chain.miner().import_own_transaction(&*net.peer(1).chain, tx1).unwrap();
-	net.sync_steps(5);
-	net.peer(0).chain.miner().import_own_transaction(&net.peer(0).chain, tx1).unwrap();
-	net.sync();
-	assert_eq!(net.peer(0).chain.chain_info().best_block_number, 1);
-	assert_eq!(net.peer(1).chain.chain_info().best_block_number, 1);
-
-	let tx2 = Transaction {
-		nonce: 0.into(),
-		gas_price: 0.into(),
-		gas: 21000.into(),
-		action: Action::Call(Address::default()),
-		value: 0.into(),
-		data: Vec::new(),
-	}.sign(s2.secret(), None);
-	net.peer(1).chain.miner().import_own_transaction(&net.peer(1).chain, tx2).unwrap();
 	net.peer(1).chain.engine().step();
-	net.peer(1).chain.miner().update_sealing(&net.peer(1).chain);
 	net.sync();
 	assert_eq!(net.peer(0).chain.chain_info().best_block_number, 2);
 	assert_eq!(net.peer(1).chain.chain_info().best_block_number, 2);
 }
 
 #[test]
-fn test_tendermint() {
+fn tendermint() {
 	::env_logger::init().ok();
 
-	let s1 = KeyPair::from_secret("1".sha3()).unwrap();
+	let s0 = KeyPair::from_secret("1".sha3()).unwrap();
-	let s2 = KeyPair::from_secret("0".sha3()).unwrap();
+	let s1 = KeyPair::from_secret("0".sha3()).unwrap();
 	let spec_factory = || {
 		let spec = Spec::new_test_tendermint();
 		let account_provider = AccountProvider::transient_provider();
+		account_provider.insert_account(s0.secret().clone(), "").unwrap();
 		account_provider.insert_account(s1.secret().clone(), "").unwrap();
-		account_provider.insert_account(s2.secret().clone(), "").unwrap();
 		spec.engine.register_account_provider(Arc::new(account_provider));
 		spec
 	};
-	let mut net = TestNet::new_with_spec(2, SyncConfig::default(), spec_factory);
+	let mut net = TestNet::with_spec(2, SyncConfig::default(), spec_factory);
 	let mut net = &mut *net;
+	let io_handler0: Arc<IoHandler<ClientIoMessage>> = Arc::new(TestIoHandler { client: net.peer(0).chain.clone() });
+	let io_handler1: Arc<IoHandler<ClientIoMessage>> = Arc::new(TestIoHandler { client: net.peer(1).chain.clone() });
 	// Push transaction to both clients. Only one of them issues a proposal.
-	net.peer(0).chain.miner().set_engine_signer(s1.address(), "".to_owned()).unwrap();
+	net.peer(0).chain.miner().set_engine_signer(s0.address(), "".to_owned()).unwrap();
-	net.peer(1).chain.miner().set_engine_signer(s2.address(), "".to_owned()).unwrap();
+	net.peer(1).chain.miner().set_engine_signer(s1.address(), "".to_owned()).unwrap();
-	let tx1 = Transaction {
+	net.peer(0).chain.engine().register_message_channel(IoChannel::to_handler(Arc::downgrade(&io_handler0)));
+	net.peer(1).chain.engine().register_message_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1)));
+	net.peer(0).chain.set_io_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1)));
+	net.peer(1).chain.set_io_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1)));
+	let tx0 = Transaction {
 		nonce: 0.into(),
 		gas_price: 0.into(),
 		gas: 21000.into(),
 		action: Action::Call(Address::default()),
 		value: 0.into(),
 		data: Vec::new(),
-	}.sign(s1.secret(), None);
+	}.sign(s0.secret(), None);
 	// exhange statuses
 	net.sync_steps(5);
-	net.peer(0).chain.miner().import_own_transaction(&net.peer(0).chain, tx1).unwrap();
+	net.peer(0).chain.miner().import_own_transaction(&*net.peer(0).chain, tx0).unwrap();
 	// Propose
 	net.sync();
 	// Propose timeout
@@ -110,23 +141,19 @@ fn test_tendermint() {
 	// Precommit
 	net.sync();
 
-	net.sync_steps(5);
-	net.sync();
-	::std::thread::sleep(::std::time::Duration::from_millis(100));
 	assert_eq!(net.peer(0).chain.chain_info().best_block_number, 1);
 	assert_eq!(net.peer(1).chain.chain_info().best_block_number, 1);
 
-	let tx2 = Transaction {
+	let tx1 = Transaction {
 		nonce: 0.into(),
 		gas_price: 0.into(),
 		gas: 21000.into(),
 		action: Action::Call(Address::default()),
 		value: 0.into(),
 		data: Vec::new(),
-	}.sign(s2.secret(), None);
+	}.sign(s1.secret(), None);
-	net.peer(1).chain.miner().import_own_transaction(&net.peer(1).chain, tx2).unwrap();
+	net.peer(1).chain.miner().import_own_transaction(&*net.peer(1).chain, tx1).unwrap();
 	net.peer(1).chain.engine().step();
-	net.peer(1).chain.miner().update_sealing(&net.peer(1).chain);
 	net.sync();
 	assert_eq!(net.peer(0).chain.chain_info().best_block_number, 2);
 	assert_eq!(net.peer(1).chain.chain_info().best_block_number, 2);

sync/src/tests/helpers.rs

@@ -45,14 +45,15 @@ impl FlushingBlockChainClient for TestBlockChainClient {}
 pub struct TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
 	pub chain: &'p C,
 	pub snapshot_service: &'p TestSnapshotService,
-	pub queue: &'p mut VecDeque<TestPacket>,
+	pub queue: &'p RwLock<VecDeque<TestPacket>>,
 	pub sender: Option<PeerId>,
 	pub to_disconnect: HashSet<PeerId>,
+	pub packets: Vec<TestPacket>,
 	overlay: RwLock<HashMap<BlockNumber, Bytes>>,
 }
 
 impl<'p, C> TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
-	pub fn new(chain: &'p C, ss: &'p TestSnapshotService, queue: &'p mut VecDeque<TestPacket>, sender: Option<PeerId>) -> TestIo<'p, C> {
+	pub fn new(chain: &'p C, ss: &'p TestSnapshotService, queue: &'p RwLock<VecDeque<TestPacket>>, sender: Option<PeerId>) -> TestIo<'p, C> {
 		TestIo {
 			chain: chain,
 			snapshot_service: ss,
@@ -60,10 +61,17 @@ impl<'p, C> TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
 			sender: sender,
 			to_disconnect: HashSet::new(),
 			overlay: RwLock::new(HashMap::new()),
+			packets: Vec::new(),
 		}
 	}
 }
 
+impl<'p, C> Drop for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
+	fn drop(&mut self) {
+		self.queue.write().extend(self.packets.drain(..));
+	}
+}
+
 impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
 	fn disable_peer(&mut self, peer_id: PeerId) {
 		self.disconnect_peer(peer_id);
@@ -78,7 +86,7 @@ impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
 	}
 
 	fn respond(&mut self, packet_id: PacketId, data: Vec<u8>) -> Result<(), NetworkError> {
-		self.queue.push_back(TestPacket {
+		self.packets.push(TestPacket {
 			data: data,
 			packet_id: packet_id,
 			recipient: self.sender.unwrap()
@@ -87,7 +95,7 @@ impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
 	}
 
 	fn send(&mut self, peer_id: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), NetworkError> {
-		self.queue.push_back(TestPacket {
+		self.packets.push(TestPacket {
 			data: data,
 			packet_id: packet_id,
 			recipient: peer_id,
@@ -100,7 +108,7 @@ impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
 	}
 
 	fn chain(&self) -> &BlockChainClient {
-		self.chain
+		&*self.chain
 	}
 
 	fn snapshot_service(&self) -> &SnapshotService {
@@ -131,7 +139,7 @@ pub struct TestPacket {
 }
 
 pub struct TestPeer<C> where C: FlushingBlockChainClient {
-	pub chain: C,
+	pub chain: Arc<C>,
 	pub snapshot_service: Arc<TestSnapshotService>,
 	pub sync: RwLock<ChainSync>,
 	pub queue: RwLock<VecDeque<TestPacket>>,
@@ -167,7 +175,7 @@ impl TestNet<TestBlockChainClient> {
 			net.peers.push(Arc::new(TestPeer {
 				sync: RwLock::new(sync),
 				snapshot_service: ss,
-				chain: chain,
+				chain: Arc::new(chain),
 				queue: RwLock::new(VecDeque::new()),
 			}));
 		}
@@ -176,7 +184,7 @@ impl TestNet<TestBlockChainClient> {
 }
 
 impl TestNet<EthcoreClient> {
-	pub fn new_with_spec<F>(n: usize, config: SyncConfig, spec_factory: F) -> GuardedTempResult<TestNet<EthcoreClient>>
+	pub fn with_spec<F>(n: usize, config: SyncConfig, spec_factory: F) -> GuardedTempResult<TestNet<EthcoreClient>>
 		where F: Fn() -> Spec
 	{
 		let mut net = TestNet {
@@ -192,17 +200,17 @@ impl TestNet<EthcoreClient> {
 			let db_config = DatabaseConfig::with_columns(NUM_COLUMNS);
 
 			let spec = spec_factory();
-			let client = Arc::try_unwrap(EthcoreClient::new(
+			let client = EthcoreClient::new(
 				ClientConfig::default(),
 				&spec,
 				client_dir.as_path(),
 				Arc::new(Miner::with_spec(&spec)),
 				IoChannel::disconnected(),
 				&db_config
-			).unwrap()).ok().unwrap();
+			).unwrap();
 
 			let ss = Arc::new(TestSnapshotService::new());
-			let sync = ChainSync::new(config.clone(), &client);
+			let sync = ChainSync::new(config.clone(), &*client);
 			let peer = Arc::new(TestPeer {
 				sync: RwLock::new(sync),
 				snapshot_service: ss,
@@ -233,8 +241,8 @@ impl<C> TestNet<C> where C: FlushingBlockChainClient {
 			for client in 0..self.peers.len() {
 				if peer != client {
 					let p = &self.peers[peer];
-					p.sync.write().update_targets(&p.chain);
+					p.sync.write().update_targets(&*p.chain);
-					p.sync.write().on_peer_connected(&mut TestIo::new(&p.chain, &p.snapshot_service, &mut p.queue.write(), Some(client as PeerId)), client as PeerId);
+					p.sync.write().on_peer_connected(&mut TestIo::new(&*p.chain, &p.snapshot_service, &p.queue, Some(client as PeerId)), client as PeerId);
 				}
 			}
 		}
@@ -246,16 +254,15 @@ impl<C> TestNet<C> where C: FlushingBlockChainClient {
 			if let Some(packet) = packet {
 				let disconnecting = {
 					let p = &self.peers[packet.recipient];
-					let mut queue = p.queue.write();
 					trace!("--- {} -> {} ---", peer, packet.recipient);
 					let to_disconnect = {
-						let mut io = TestIo::new(&p.chain, &p.snapshot_service, &mut queue, Some(peer as PeerId));
+						let mut io = TestIo::new(&*p.chain, &p.snapshot_service, &p.queue, Some(peer as PeerId));
 						ChainSync::dispatch_packet(&p.sync, &mut io, peer as PeerId, packet.packet_id, &packet.data);
-						io.to_disconnect
+						io.to_disconnect.clone()
 					};
 					for d in &to_disconnect {
 						// notify this that disconnecting peers are disconnecting
-						let mut io = TestIo::new(&p.chain, &p.snapshot_service, &mut queue, Some(*d));
+						let mut io = TestIo::new(&*p.chain, &p.snapshot_service, &p.queue, Some(*d));
 						p.sync.write().on_peer_aborting(&mut io, *d);
 						self.disconnect_events.push((peer, *d));
 					}
@@ -264,8 +271,7 @@ impl<C> TestNet<C> where C: FlushingBlockChainClient {
 				for d in &disconnecting {
 					// notify other peers that this peer is disconnecting
 					let p = &self.peers[*d];
-					let mut queue = p.queue.write();
+					let mut io = TestIo::new(&*p.chain, &p.snapshot_service, &p.queue, Some(peer as PeerId));
-					let mut io = TestIo::new(&p.chain, &p.snapshot_service, &mut queue, Some(peer as PeerId));
 					p.sync.write().on_peer_aborting(&mut io, peer as PeerId);
 				}
 			}
@@ -277,15 +283,14 @@ impl<C> TestNet<C> where C: FlushingBlockChainClient {
 	pub fn sync_step_peer(&mut self, peer_num: usize) {
 		let peer = self.peer(peer_num);
 		peer.chain.flush();
-		let mut queue = peer.queue.write();
+		peer.sync.write().maintain_peers(&mut TestIo::new(&*peer.chain, &peer.snapshot_service, &peer.queue, None));
-		peer.sync.write().maintain_peers(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None));
+		peer.sync.write().maintain_sync(&mut TestIo::new(&*peer.chain, &peer.snapshot_service, &peer.queue, None));
-		peer.sync.write().maintain_sync(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None));
+		peer.sync.write().propagate_new_transactions(&mut TestIo::new(&*peer.chain, &peer.snapshot_service, &peer.queue, None));
-		peer.sync.write().propagate_new_transactions(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None));
 	}
 
 	pub fn restart_peer(&mut self, i: usize) {
 		let peer = self.peer(i);
-		peer.sync.write().restart(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut peer.queue.write(), None));
+		peer.sync.write().restart(&mut TestIo::new(&*peer.chain, &peer.snapshot_service, &peer.queue, None));
 	}
 
 	pub fn sync(&mut self) -> u32 {
@@ -314,8 +319,7 @@ impl<C> TestNet<C> where C: FlushingBlockChainClient {
 
 	pub fn trigger_chain_new_blocks(&mut self, peer_id: usize) {
 		let peer = self.peer(peer_id);
-		let mut queue = peer.queue.write();
+		peer.sync.write().chain_new_blocks(&mut TestIo::new(&*peer.chain, &peer.snapshot_service, &peer.queue, None), &[], &[], &[], &[], &[], &[]);
-		peer.sync.write().chain_new_blocks(&mut TestIo::new(&peer.chain, &peer.snapshot_service, &mut queue, None), &[], &[], &[], &[], &[], &[]);
 	}
 }
 
@@ -329,8 +333,7 @@ impl ChainNotify for TestPeer<EthcoreClient> {
 		proposed: Vec<Bytes>,
 		_duration: u64)
 	{
-		let mut queue = self.queue.write();
+		let mut io = TestIo::new(&*self.chain, &self.snapshot_service, &self.queue, None);
-		let mut io = TestIo::new(&self.chain, &self.snapshot_service, &mut queue, None);
 		self.sync.write().chain_new_blocks(
 			&mut io,
 			&imported,
@@ -346,8 +349,7 @@ impl ChainNotify for TestPeer<EthcoreClient> {
 	fn stop(&self) {}
 
 	fn broadcast(&self, message: Vec<u8>) {
-		let mut queue = self.queue.write();
+		let mut io = TestIo::new(&*self.chain, &self.snapshot_service, &self.queue, None);
-		let mut io = TestIo::new(&self.chain, &self.snapshot_service, &mut queue, None);
 		self.sync.write().propagate_consensus_packet(&mut io, message.clone());
 	}
 }

util/io/src/service.rs

@@ -329,11 +329,18 @@ impl<Message> Handler for IoManager<Message> where Message: Send + Clone + Sync
 	}
 }
 
+#[derive(Clone)]
+enum Handlers<Message> where Message: Send + Clone {
+	SharedCollection(Weak<RwLock<Slab<Arc<IoHandler<Message>>, HandlerId>>>),
+	Single(Weak<IoHandler<Message>>),
+}
+
 /// Allows sending messages into the event loop. All the IO handlers will get the message
 /// in the `message` callback.
 pub struct IoChannel<Message> where Message: Send + Clone{
 	channel: Option<Sender<IoMessage<Message>>>,
-	handlers: Weak<RwLock<Slab<Arc<IoHandler<Message>>, HandlerId>>>,
+	handlers: Handlers<Message>,
+
 }
 
 impl<Message> Clone for IoChannel<Message> where Message: Send + Clone + Sync + 'static {
@@ -348,19 +355,29 @@ impl<Message> Clone for IoChannel<Message> where Message: Send + Clone + Sync +
 impl<Message> IoChannel<Message> where Message: Send + Clone + Sync + 'static {
 	/// Send a message through the channel
 	pub fn send(&self, message: Message) -> Result<(), IoError> {
-		if let Some(ref channel) = self.channel {
+		match self.channel {
-			try!(channel.send(IoMessage::UserMessage(message)));
+			Some(ref channel) => try!(channel.send(IoMessage::UserMessage(message))),
+			None => try!(self.send_sync(message))
 		}
 		Ok(())
 	}
 
 	/// Send a message through the channel and handle it synchronously
 	pub fn send_sync(&self, message: Message) -> Result<(), IoError> {
-		if let Some(handlers) = self.handlers.upgrade() {
+		match self.handlers {
-			for id in 0 .. MAX_HANDLERS {
+			Handlers::SharedCollection(ref handlers) => {
-				if let Some(h) = handlers.read().get(id) {
+				if let Some(handlers) = handlers.upgrade() {
-					let handler = h.clone();
+					for id in 0 .. MAX_HANDLERS {
-					handler.message(&IoContext::new(self.clone(), id), &message);
+						if let Some(h) = handlers.read().get(id) {
+							let handler = h.clone();
+							handler.message(&IoContext::new(self.clone(), id), &message);
+						}
+					}
+				}
+			},
+			Handlers::Single(ref handler) => {
+				if let Some(handler) = handler.upgrade() {
+					handler.message(&IoContext::new(self.clone(), 0), &message);
 				}
 			}
 		}
@@ -378,14 +395,21 @@ impl<Message> IoChannel<Message> where Message: Send + Clone + Sync + 'static {
 	pub fn disconnected() -> IoChannel<Message> {
 		IoChannel {
 			channel: None,
-			handlers: Weak::default(),
+			handlers: Handlers::SharedCollection(Weak::default()),
 		}
 	}
 
+	/// Create a new synchronous channel to a given handler.
+	pub fn to_handler(handler: Weak<IoHandler<Message>>) -> IoChannel<Message> {
+		IoChannel {
+			channel: None,
+			handlers: Handlers::Single(handler),
+		}
+	}
 	fn new(channel: Sender<IoMessage<Message>>, handlers: Weak<RwLock<Slab<Arc<IoHandler<Message>>, HandlerId>>>) -> IoChannel<Message> {
 		IoChannel {
 			channel: Some(channel),
-			handlers: handlers,
+			handlers: Handlers::SharedCollection(handlers),
 		}
 	}
 }