Merge remote-tracking branch 'parity/master' into bft

Conflicts:
	ethcore/src/client/client.rs

sync/Cargo.toml

@@ -16,6 +16,7 @@ ethcore-util = { path = "../util" }
 ethcore-network = { path = "../util/network" }
 ethcore-io = { path = "../util/io" }
 ethcore = { path = "../ethcore" }
+rlp = { path = "../util/rlp" }
 clippy = { version = "0.0.85", optional = true}
 log = "0.3"
 env_logger = "0.3"

sync/src/api.rs

@@ -17,7 +17,7 @@
 use std::sync::Arc;
 use network::{NetworkProtocolHandler, NetworkService, NetworkContext, PeerId,
 	NetworkConfiguration as BasicNetworkConfiguration, NonReservedPeerMode, NetworkError};
-use util::{U256, H256, Populatable};
+use util::{U256, H256};
 use io::{TimerToken};
 use ethcore::client::{BlockChainClient, ChainNotify};
 use ethcore::header::BlockNumber;
@@ -35,7 +35,7 @@ pub const ETH_PROTOCOL: &'static str = "eth";
 pub const INF_PROTOCOL: &'static str = "inf";
 
 /// Sync configuration
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Copy)]
 pub struct SyncConfig {
 	/// Max blocks to download ahead
 	pub max_download_ahead_blocks: usize,
@@ -126,6 +126,7 @@ impl NetworkProtocolHandler for SyncProtocolHandler {
 	fn timeout(&self, io: &NetworkContext, _timer: TimerToken) {
 		self.sync.write().maintain_peers(&mut NetSyncIo::new(io, &*self.chain));
 		self.sync.write().maintain_sync(&mut NetSyncIo::new(io, &*self.chain));
+		self.sync.write().propagate_new_transactions(&mut NetSyncIo::new(io, &*self.chain));
 	}
 }
 

sync/src/blocks.rs

@@ -15,6 +15,7 @@
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
 use util::*;
+use rlp::*;
 use network::NetworkError;
 use ethcore::header::{ Header as BlockHeader};
 
@@ -283,11 +284,11 @@ impl BlockCollection {
 			transactions_root: info.transactions_root().clone(),
 			uncles: info.uncles_hash().clone(),
 		};
-		if header_id.transactions_root == rlp::SHA3_NULL_RLP && header_id.uncles == rlp::SHA3_EMPTY_LIST_RLP {
+		if header_id.transactions_root == sha3::SHA3_NULL_RLP && header_id.uncles == sha3::SHA3_EMPTY_LIST_RLP {
 			// empty body, just mark as downloaded
 			let mut body_stream = RlpStream::new_list(2);
-			body_stream.append_raw(&rlp::EMPTY_LIST_RLP, 1);
-			body_stream.append_raw(&rlp::EMPTY_LIST_RLP, 1);
+			body_stream.append_raw(&::rlp::EMPTY_LIST_RLP, 1);
+			body_stream.append_raw(&::rlp::EMPTY_LIST_RLP, 1);
 			block.body = Some(body_stream.out());
 		}
 		else {
@@ -337,6 +338,7 @@ mod test {
 	use ethcore::views::HeaderView;
 	use ethcore::header::BlockNumber;
 	use util::*;
+	use rlp::*;
 
 	fn is_empty(bc: &BlockCollection) -> bool {
 		bc.heads.is_empty() &&

sync/src/chain.rs

@@ -88,6 +88,7 @@
 ///
 
 use util::*;
+use rlp::*;
 use network::*;
 use std::mem::{replace};
 use ethcore::views::{HeaderView, BlockView};
@@ -156,7 +157,7 @@ pub enum SyncState {
 }
 
 /// Syncing status and statistics
-#[derive(Clone)]
+#[derive(Clone, Copy)]
 pub struct SyncStatus {
 	/// State
 	pub state: SyncState,
@@ -241,7 +242,9 @@ struct PeerInfo {
 	asking_hash: Option<H256>,
 	/// Request timestamp
 	ask_time: f64,
-	/// Pending request is expird and result should be ignored
+	/// Holds a set of transactions recently sent to this peer to avoid spamming.
+	last_sent_transactions: HashSet<H256>,
+	/// Pending request is expired and result should be ignored
 	expired: bool,
 	/// Peer fork confirmation status
 	confirmation: ForkConfirmation,
@@ -406,6 +409,7 @@ impl ChainSync {
 			asking_blocks: Vec::new(),
 			asking_hash: None,
 			ask_time: 0f64,
+			last_sent_transactions: HashSet::new(),
 			expired: false,
 			confirmation: if self.fork_block.is_none() { ForkConfirmation::Confirmed } else { ForkConfirmation::Unconfirmed },
 		};
@@ -1447,42 +1451,67 @@ impl ChainSync {
 	}
 
 	/// propagates new transactions to all peers
-	fn propagate_new_transactions(&mut self, io: &mut SyncIo) -> usize {
+	pub fn propagate_new_transactions(&mut self, io: &mut SyncIo) -> usize {
 
 		// Early out of nobody to send to.
 		if self.peers.is_empty() {
 			return 0;
 		}
 
-		let mut transactions = io.chain().pending_transactions();
+		let transactions = io.chain().pending_transactions();
 		if transactions.is_empty() {
 			return 0;
 		}
 
-		let mut packet = RlpStream::new_list(transactions.len());
-		let tx_count = transactions.len();
-		for tx in transactions.drain(..) {
-			packet.append(&tx);
-		}
-		let rlp = packet.out();
-
-		let lucky_peers = {
-			// sqrt(x)/x scaled to max u32
-			let fraction = (self.peers.len() as f64).powf(-0.5).mul(u32::max_value() as f64).round() as u32;
-			let small = self.peers.len() < MIN_PEERS_PROPAGATION;
-			let lucky_peers = self.peers.iter()
-				.filter_map(|(&p, _)| if small || ::rand::random::<u32>() < fraction { Some(p.clone()) } else { None })
-				.collect::<Vec<_>>();
-
-			// taking at max of MAX_PEERS_PROPAGATION
-			lucky_peers.iter().cloned().take(min(lucky_peers.len(), MAX_PEERS_PROPAGATION)).collect::<Vec<PeerId>>()
+		let all_transactions_hashes = transactions.iter().map(|ref tx| tx.hash()).collect::<HashSet<H256>>();
+		let all_transactions_rlp = {
+			let mut packet = RlpStream::new_list(transactions.len());
+			for tx in &transactions { packet.append(tx); }
+			packet.out()
 		};
 
+		// sqrt(x)/x scaled to max u32
+		let fraction = (self.peers.len() as f64).powf(-0.5).mul(u32::max_value() as f64).round() as u32;
+		let small = self.peers.len() < MIN_PEERS_PROPAGATION;
+
+		let lucky_peers = self.peers.iter_mut()
+			.filter(|_| small || ::rand::random::<u32>() < fraction)
+			.take(MAX_PEERS_PROPAGATION)
+			.filter_map(|(peer_id, mut peer_info)| {
+				// Send all transactions
+				if peer_info.last_sent_transactions.is_empty() {
+					peer_info.last_sent_transactions = all_transactions_hashes.clone();
+					return Some((*peer_id, all_transactions_rlp.clone()));
+				}
+
+				// Get hashes of all transactions to send to this peer
+				let to_send = all_transactions_hashes.difference(&peer_info.last_sent_transactions).cloned().collect::<HashSet<_>>();
+				if to_send.is_empty() {
+					return None;
+				}
+
+				// Construct RLP
+				let mut packet = RlpStream::new_list(to_send.len());
+				for tx in &transactions {
+					if to_send.contains(&tx.hash()) {
+						packet.append(tx);
+					}
+				}
+
+				peer_info.last_sent_transactions = all_transactions_hashes.clone();
+				Some((*peer_id, packet.out()))
+			})
+			.collect::<Vec<_>>();
+
+		// Send RLPs
 		let sent = lucky_peers.len();
-		for peer_id in lucky_peers {
-			self.send_packet(io, peer_id, TRANSACTIONS_PACKET, rlp.clone());
+		if sent > 0 {
+			for (peer_id, rlp) in lucky_peers.into_iter() {
+				self.send_packet(io, peer_id, TRANSACTIONS_PACKET, rlp);
+			}
+
+			trace!(target: "sync", "Sent up to {} transactions to {} peers.", transactions.len(), sent);
 		}
-		trace!(target: "sync", "Sent {} transactions to {} peers.", tx_count, sent);
 		sent
 	}
 
@@ -1512,16 +1541,18 @@ impl ChainSync {
 		self.check_resume(io);
 	}
 
-	/// called when block is imported to chain, updates transactions queue and propagates the blocks
+	/// called when block is imported to chain - propagates the blocks and updates transactions sent to peers
 	pub fn chain_new_blocks(&mut self, io: &mut SyncIo, _imported: &[H256], invalid: &[H256], _enacted: &[H256], _retracted: &[H256], sealed: &[H256]) {
 		if io.is_chain_queue_empty() {
-			// Propagate latests blocks
 			self.propagate_latest_blocks(io, sealed);
 		}
 		if !invalid.is_empty() {
 			trace!(target: "sync", "Bad blocks in the queue, restarting");
 			self.restart_on_bad_block(io);
 		}
+		for peer_info in self.peers.values_mut() {
+			peer_info.last_sent_transactions.clear();
+		}
 	}
 }
 
@@ -1531,6 +1562,7 @@ mod tests {
 	use super::*;
 	use ::SyncConfig;
 	use util::*;
+	use rlp::*;
 	use super::{PeerInfo, PeerAsking};
 	use ethcore::views::BlockView;
 	use ethcore::header::*;
@@ -1548,8 +1580,8 @@ mod tests {
 
 		let mut rlp = RlpStream::new_list(3);
 		rlp.append(&header);
-		rlp.append_raw(&rlp::EMPTY_LIST_RLP, 1);
-		rlp.append_raw(&rlp::EMPTY_LIST_RLP, 1);
+		rlp.append_raw(&::rlp::EMPTY_LIST_RLP, 1);
+		rlp.append_raw(&::rlp::EMPTY_LIST_RLP, 1);
 		rlp.out()
 	}
 
@@ -1723,6 +1755,7 @@ mod tests {
 				asking_blocks: Vec::new(),
 				asking_hash: None,
 				ask_time: 0f64,
+				last_sent_transactions: HashSet::new(),
 				expired: false,
 				confirmation: super::ForkConfirmation::Confirmed,
 			});
@@ -1819,6 +1852,83 @@ mod tests {
 		assert_eq!(0x07, io.queue[0].packet_id);
 	}
 
+	#[test]
+	fn propagates_transactions() {
+		let mut client = TestBlockChainClient::new();
+		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 mut io = TestIo::new(&mut client, &mut 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);
+
+		// 1 message should be send
+		assert_eq!(1, io.queue.len());
+		// 1 peer should be updated but only once
+		assert_eq!(1, peer_count);
+		assert_eq!(0, peer_count2);
+		// TRANSACTIONS_PACKET
+		assert_eq!(0x02, io.queue[0].packet_id);
+	}
+
+	#[test]
+	fn propagates_transactions_again_after_new_block() {
+		let mut client = TestBlockChainClient::new();
+		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 mut io = TestIo::new(&mut client, &mut queue, None);
+		let peer_count = sync.propagate_new_transactions(&mut io);
+		sync.chain_new_blocks(&mut io, &[], &[], &[], &[], &[]);
+		// Try to propagate same transactions for the second time
+		let peer_count2 = sync.propagate_new_transactions(&mut io);
+
+		// 2 message should be send
+		assert_eq!(2, io.queue.len());
+		// 1 peer should be updated twice
+		assert_eq!(1, peer_count);
+		assert_eq!(1, peer_count2);
+		// TRANSACTIONS_PACKET
+		assert_eq!(0x02, io.queue[0].packet_id);
+		assert_eq!(0x02, io.queue[1].packet_id);
+	}
+
+	#[test]
+	fn propagates_transactions_without_alternating() {
+		let mut client = TestBlockChainClient::new();
+		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();
+		// should sent some
+		{
+
+			let mut io = TestIo::new(&mut client, &mut queue, None);
+			let peer_count = sync.propagate_new_transactions(&mut io);
+			assert_eq!(1, io.queue.len());
+			assert_eq!(1, peer_count);
+		}
+		// Insert some more
+		client.insert_transaction_to_queue();
+		let mut io = TestIo::new(&mut client, &mut queue, None);
+		// Propagate new transactions
+		let peer_count2 = sync.propagate_new_transactions(&mut io);
+		// And now the peer should have all transactions
+		let peer_count3 = sync.propagate_new_transactions(&mut io);
+
+		// 2 message should be send (in total)
+		assert_eq!(2, io.queue.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, io.queue[1].packet_id);
+	}
+
 	#[test]
 	fn handles_peer_new_block_malformed() {
 		let mut client = TestBlockChainClient::new();

sync/src/lib.rs

@@ -34,7 +34,6 @@
 //! extern crate ethcore;
 //! extern crate ethsync;
 //! use std::env;
-//! use std::sync::Arc;
 //! use io::IoChannel;
 //! use ethcore::client::{Client, ClientConfig};
 //! use ethsync::{EthSync, SyncConfig, ManageNetwork, NetworkConfiguration};
@@ -62,22 +61,24 @@
 //! }
 //! ```
 
-#[macro_use]
-extern crate log;
-#[macro_use]
-extern crate ethcore_util as util;
 extern crate ethcore_network as network;
 extern crate ethcore_io as io;
 extern crate ethcore;
 extern crate env_logger;
 extern crate time;
 extern crate rand;
+extern crate semver;
+extern crate parking_lot;
+extern crate rlp;
+
+#[macro_use]
+extern crate log;
+#[macro_use]
+extern crate ethcore_util as util;
 #[macro_use]
 extern crate heapsize;
 #[macro_use]
 extern crate ethcore_ipc as ipc;
-extern crate semver;
-extern crate parking_lot;
 
 mod chain;
 mod blocks;