cull before repropagating; repropagate on timer

polkadot/cli/src/lib.rs

@@ -82,17 +82,15 @@ impl substrate_rpc::author::AuthorApi for RpcTransactionPool {
 		use codec::Slicable;
 
 		info!("Extrinsic submitted: {}", HexDisplay::from(&xt.0));
-		let (decoded, encoded_bytes) = xt.using_encoded(|s| {
-				UncheckedExtrinsic::decode(&mut &s[..]).map(|d| (d, s.to_vec()))
-			})
+		let decoded = xt.using_encoded(|ref mut s| UncheckedExtrinsic::decode(s))
 			.ok_or(substrate_rpc::author::error::ErrorKind::InvalidFormat)?;
+
 		info!("Correctly formatted: {:?}", decoded);
 
-		let hash = self.inner.lock().import(decoded)
-			.map(|v| v.hash().0.into())
+		self.inner.lock().import(decoded)
 			.map_err(|_| substrate_rpc::author::error::ErrorKind::PoolError)?;
 
-		self.network.on_new_transactions(&[(hash, encoded_bytes)]);
+		self.network.trigger_repropagate();
 		Ok(())
 	}
 }

substrate/network/src/protocol.rs

@@ -446,11 +446,14 @@ impl Protocol {
 	}
 
 	/// Called when peer sends us new transactions
-	pub fn propagate_transactions(&self, io: &mut SyncIo, transactions: &[(ExtrinsicHash, Vec<u8>)]) {
+	pub fn propagate_transactions(&self, io: &mut SyncIo) {
 		// Accept transactions only when fully synced
 		if self.sync.read().status().state != SyncState::Idle {
 			return;
 		}
+
+		let transactions = self.transaction_pool.transactions();
+
 		let mut peers = self.peers.write();
 		for (peer_id, ref mut peer) in peers.iter_mut() {
 			let to_send: Vec<_> = transactions.iter().filter_map(|&(hash, ref t)|

substrate/network/src/service.rs

@@ -17,6 +17,7 @@
 use std::sync::Arc;
 use std::collections::{BTreeMap};
 use std::io;
+use std::time::Duration;
 use futures::sync::{oneshot, mpsc};
 use network::{NetworkProtocolHandler, NetworkContext, HostInfo, PeerId, ProtocolId,
 NetworkConfiguration , NonReservedPeerMode, ErrorKind};
@@ -41,6 +42,12 @@ pub type StatementStream = mpsc::UnboundedReceiver<Statement>;
 /// Type that represents bft messages stream.
 pub type BftMessageStream = mpsc::UnboundedReceiver<LocalizedBftMessage>;
 
+const TICK_TOKEN: TimerToken = 0;
+const TICK_TIMEOUT: Duration = Duration::from_millis(1000);
+
+const PROPAGATE_TOKEN: TimerToken = 1;
+const PROPAGATE_TIMEOUT: Duration = Duration::from_millis(5000);
+
 bitflags! {
 	/// Node roles bitmask.
 	pub struct Role: u32 {
@@ -162,9 +169,9 @@ impl Service {
 	}
 
 	/// Called when new transactons are imported by the client.
-	pub fn on_new_transactions(&self, transactions: &[(ExtrinsicHash, Vec<u8>)]) {
+	pub fn trigger_repropagate(&self) {
 		self.network.with_context(DOT_PROTOCOL_ID, |context| {
-			self.handler.protocol.propagate_transactions(&mut NetSyncIo::new(context), transactions);
+			self.handler.protocol.propagate_transactions(&mut NetSyncIo::new(context));
 		});
 	}
 
@@ -268,7 +275,11 @@ impl ConsensusService for Service {
 
 impl NetworkProtocolHandler for ProtocolHandler {
 	fn initialize(&self, io: &NetworkContext, _host_info: &HostInfo) {
-		io.register_timer(0, ::std::time::Duration::from_millis(1000)).expect("Error registering sync timer");
+		io.register_timer(TICK_TOKEN, TICK_TIMEOUT)
+			.expect("Error registering sync timer");
+
+		io.register_timer(PROPAGATE_TOKEN, PROPAGATE_TIMEOUT)
+			.expect("Error registering transaction propagation timer");
 	}
 
 	fn read(&self, io: &NetworkContext, peer: &PeerId, _packet_id: u8, data: &[u8]) {
@@ -283,8 +294,12 @@ impl NetworkProtocolHandler for ProtocolHandler {
 		self.protocol.on_peer_disconnected(&mut NetSyncIo::new(io), *peer);
 	}
 
-	fn timeout(&self, io: &NetworkContext, _timer: TimerToken) {
-		self.protocol.tick(&mut NetSyncIo::new(io));
+	fn timeout(&self, io: &NetworkContext, timer: TimerToken) {
+		match timer {
+			TICK_TOKEN => self.protocol.tick(&mut NetSyncIo::new(io)),
+			PROPAGATE_TOKEN => self.protocol.propagate_transactions(&mut NetSyncIo::new(io)),
+			_ => {}
+		}
 	}
 }