Limit the number of transactions in pending set (#8777)

* Unordered iterator.

* Use unordered and limited set if full not required.

* Split timeout work into smaller timers.

* Avoid collecting all pending transactions when mining

* Remove println.

* Use priority ordering in eth-filter.

* Fix ethcore-miner tests and tx propagation.

* Review grumbles addressed.

* Add test for unordered not populating the cache.

* Fix ethcore tests.

* Fix light tests.

* Fix ethcore-sync tests.

* Fix RPC tests.

miner/src/pool/queue.rs

@@ -26,7 +26,10 @@ use parking_lot::RwLock;
 use transaction;
 use txpool::{self, Verifier};
 
-use pool::{self, scoring, verifier, client, ready, listener, PrioritizationStrategy};
+use pool::{
+	self, scoring, verifier, client, ready, listener,
+	PrioritizationStrategy, PendingOrdering, PendingSettings,
+};
 use pool::local_transactions::LocalTransactionsList;
 
 type Listener = (LocalTransactionsList, (listener::Notifier, listener::Logger));
@@ -74,6 +77,7 @@ struct CachedPending {
 	nonce_cap: Option<U256>,
 	has_local_pending: bool,
 	pending: Option<Vec<Arc<pool::VerifiedTransaction>>>,
+	max_len: usize,
 }
 
 impl CachedPending {
@@ -85,6 +89,7 @@ impl CachedPending {
 			has_local_pending: false,
 			pending: None,
 			nonce_cap: None,
+			max_len: 0,
 		}
 	}
 
@@ -99,6 +104,7 @@ impl CachedPending {
 		block_number: u64,
 		current_timestamp: u64,
 		nonce_cap: Option<&U256>,
+		max_len: usize,
 	) -> Option<Vec<Arc<pool::VerifiedTransaction>>> {
 		// First check if we have anything in cache.
 		let pending = self.pending.as_ref()?;
@@ -123,7 +129,12 @@ impl CachedPending {
 			return None;
 		}
 
-		Some(pending.clone())
+		// It's fine to just take a smaller subset, but not other way around.
+		if max_len > self.max_len {
+			return None;
+		}
+
+		Some(pending.iter().take(max_len).cloned().collect())
 	}
 }
 
@@ -173,7 +184,7 @@ impl TransactionQueue {
 		transactions: Vec<verifier::Transaction>,
 	) -> Vec<Result<(), transaction::Error>> {
 		// Run verification
-		let _timer = ::trace_time::PerfTimer::new("pool::verify_and_import");
+		trace_time!("pool::verify_and_import");
 		let options = self.options.read().clone();
 
 		let verifier = verifier::Verifier::new(client, options, self.insertion_id.clone());
@@ -203,13 +214,13 @@ impl TransactionQueue {
 		results
 	}
 
-	/// Returns all transactions in the queue ordered by priority.
+	/// Returns all transactions in the queue without explicit ordering.
 	pub fn all_transactions(&self) -> Vec<Arc<pool::VerifiedTransaction>> {
 		let ready = |_tx: &pool::VerifiedTransaction| txpool::Readiness::Ready;
-		self.pool.read().pending(ready).collect()
+		self.pool.read().unordered_pending(ready).collect()
 	}
 
-	/// Returns current pneding transactions.
+	/// Returns current pending transactions ordered by priority.
 	///
 	/// NOTE: This may return a cached version of pending transaction set.
 	/// Re-computing the pending set is possible with `#collect_pending` method,
@@ -217,24 +228,31 @@ impl TransactionQueue {
 	pub fn pending<C>(
 		&self,
 		client: C,
-		block_number: u64,
-		current_timestamp: u64,
-		nonce_cap: Option<U256>,
+		settings: PendingSettings,
 	) -> Vec<Arc<pool::VerifiedTransaction>> where
 		C: client::NonceClient,
 	{
-
-		if let Some(pending) = self.cached_pending.read().pending(block_number, current_timestamp, nonce_cap.as_ref()) {
+		let PendingSettings { block_number, current_timestamp, nonce_cap, max_len, ordering } = settings;
+		if let Some(pending) = self.cached_pending.read().pending(block_number, current_timestamp, nonce_cap.as_ref(), max_len) {
 			return pending;
 		}
 
 		// Double check after acquiring write lock
 		let mut cached_pending = self.cached_pending.write();
-		if let Some(pending) = cached_pending.pending(block_number, current_timestamp, nonce_cap.as_ref()) {
+		if let Some(pending) = cached_pending.pending(block_number, current_timestamp, nonce_cap.as_ref(), max_len) {
 			return pending;
 		}
 
-		let pending: Vec<_> = self.collect_pending(client, block_number, current_timestamp, nonce_cap, |i| i.collect());
+		// In case we don't have a cached set, but we don't care about order
+		// just return the unordered set.
+		if let PendingOrdering::Unordered = ordering {
+			let ready = Self::ready(client, block_number, current_timestamp, nonce_cap);
+			return self.pool.read().unordered_pending(ready).take(max_len).collect();
+		}
+
+		let pending: Vec<_> = self.collect_pending(client, block_number, current_timestamp, nonce_cap, |i| {
+			i.take(max_len).collect()
+		});
 
 		*cached_pending = CachedPending {
 			block_number,
@@ -242,6 +260,7 @@ impl TransactionQueue {
 			nonce_cap,
 			has_local_pending: self.has_local_pending_transactions(),
 			pending: Some(pending.clone()),
+			max_len,
 		};
 
 		pending
@@ -266,15 +285,27 @@ impl TransactionQueue {
 			scoring::NonceAndGasPrice,
 			Listener,
 		>) -> T,
+	{
+		debug!(target: "txqueue", "Re-computing pending set for block: {}", block_number);
+		trace_time!("pool::collect_pending");
+		let ready = Self::ready(client, block_number, current_timestamp, nonce_cap);
+		collect(self.pool.read().pending(ready))
+	}
+
+	fn ready<C>(
+		client: C,
+		block_number: u64,
+		current_timestamp: u64,
+		nonce_cap: Option<U256>,
+	) -> (ready::Condition, ready::State<C>) where
+		C: client::NonceClient,
 	{
 		let pending_readiness = ready::Condition::new(block_number, current_timestamp);
 		// don't mark any transactions as stale at this point.
 		let stale_id = None;
 		let state_readiness = ready::State::new(client, stale_id, nonce_cap);
 
-		let ready = (pending_readiness, state_readiness);
-
-		collect(self.pool.read().pending(ready))
+		(pending_readiness, state_readiness)
 	}
 
 	/// Culls all stalled transactions from the pool.
@@ -415,6 +446,12 @@ impl TransactionQueue {
 		let mut pool = self.pool.write();
 		(pool.listener_mut().1).0.add(f);
 	}
+
+	/// Check if pending set is cached.
+	#[cfg(test)]
+	pub fn is_pending_cached(&self) -> bool {
+		self.cached_pending.read().pending.is_some()
+	}
 }
 
 fn convert_error(err: txpool::Error) -> transaction::Error {
@@ -440,7 +477,7 @@ mod tests {
 	fn should_get_pending_transactions() {
 		let queue = TransactionQueue::new(txpool::Options::default(), verifier::Options::default(), PrioritizationStrategy::GasPriceOnly);
 
-		let pending: Vec<_> = queue.pending(TestClient::default(), 0, 0, None);
+		let pending: Vec<_> = queue.pending(TestClient::default(), PendingSettings::all_prioritized(0, 0));
 
 		for tx in pending {
 			assert!(tx.signed().nonce > 0.into());