Quickly skip invalid transactions during block authorship.

bin/node/bench/src/construct.rs

@@ -30,8 +30,8 @@ use std::{borrow::Cow, collections::HashMap, pin::Pin, sync::Arc};
 use node_primitives::Block;
 use node_testing::bench::{BenchDb, BlockType, DatabaseType, KeyTypes, Profile};
 use sc_transaction_pool_api::{
- ImportNotificationStream, PoolFuture, PoolStatus, TransactionFor, TransactionSource,
- TransactionStatusStreamFor, TxHash,
+ ImportNotificationStream, PoolFuture, PoolStatus, ReadyTransactions, TransactionFor,
+ TransactionSource, TransactionStatusStreamFor, TxHash,
 };
 use sp_consensus::{Environment, Proposer};
 use sp_inherents::InherentDataProvider;
@@ -216,6 +216,19 @@ impl sc_transaction_pool_api::InPoolTransaction for PoolTransaction {
 
 #[derive(Clone, Debug)]
 pub struct Transactions(Vec<Arc<PoolTransaction>>);
+pub struct TransactionsIterator(std::vec::IntoIter<Arc<PoolTransaction>>);
+
+impl Iterator for TransactionsIterator {
+ type Item = Arc<PoolTransaction>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ self.0.next()
+ }
+}
+
+impl ReadyTransactions for TransactionsIterator {
+ fn report_invalid(&mut self) {}
+}
 
 impl sc_transaction_pool_api::TransactionPool for Transactions {
  type Block = Block;
@@ -257,16 +270,17 @@ impl sc_transaction_pool_api::TransactionPool for Transactions {
  _at: NumberFor<Self::Block>,
  ) -> Pin<
  Box<
- dyn Future<Output = Box<dyn Iterator<Item = Arc<Self::InPoolTransaction>> + Send>>
- + Send,
+ dyn Future<
+ Output = Box<dyn ReadyTransactions<Item = Arc<Self::InPoolTransaction>> + Send>,
+ > + Send,
  >,
  > {
- let iter: Box<dyn Iterator<Item = Arc<PoolTransaction>> + Send> =
- Box::new(self.0.clone().into_iter());
+ let iter: Box<dyn ReadyTransactions<Item = Arc<PoolTransaction>> + Send> =
+ Box::new(TransactionsIterator(self.0.clone().into_iter()));
  Box::pin(futures::future::ready(iter))
  }
 
- fn ready(&self) -> Box<dyn Iterator<Item = Arc<Self::InPoolTransaction>> + Send> {
+ fn ready(&self) -> Box<dyn ReadyTransactions<Item = Arc<Self::InPoolTransaction>> + Send> {
  unimplemented!()
  }
 

client/basic-authorship/src/basic_authorship.rs

@@ -344,7 +344,7 @@ where
  let mut t2 =
  futures_timer::Delay::new(deadline.saturating_duration_since((self.now)()) / 8).fuse();
 
- let pending_iterator = select! {
+ let mut pending_iterator = select! {
  res = t1 => res,
  _ = t2 => {
  log::warn!(
@@ -363,7 +363,7 @@ where
  let mut transaction_pushed = false;
  let mut hit_block_size_limit = false;
 
- for pending_tx in pending_iterator {
+ while let Some(pending_tx) = pending_iterator.next() {
  if (self.now)() > deadline {
  debug!(
  "Consensus deadline reached when pushing block transactions, \
@@ -378,6 +378,7 @@ where
  let block_size =
  block_builder.estimate_block_size(self.include_proof_in_block_size_estimation);
  if block_size + pending_tx_data.encoded_size() > block_size_limit {
+ pending_iterator.report_invalid();
  if skipped < MAX_SKIPPED_TRANSACTIONS {
  skipped += 1;
  debug!(
@@ -400,6 +401,7 @@ where
  debug!("[{:?}] Pushed to the block.", pending_tx_hash);
  },
  Err(ApplyExtrinsicFailed(Validity(e))) if e.exhausted_resources() => {
+ pending_iterator.report_invalid();
  if skipped < MAX_SKIPPED_TRANSACTIONS {
  skipped += 1;
  debug!(
@@ -412,13 +414,15 @@ where
  }
  },
  Err(e) if skipped > 0 => {
+ pending_iterator.report_invalid();
  trace!(
  "[{:?}] Ignoring invalid transaction when skipping: {}",
  pending_tx_hash,
  e
  );
  },
  Err(e) => {
+ pending_iterator.report_invalid();
  debug!("[{:?}] Invalid transaction: {}", pending_tx_hash, e);
  unqueue_invalid.push(pending_tx_hash);
  },

client/transaction-pool/api/src/lib.rs

@@ -223,13 +223,14 @@ pub trait TransactionPool: Send + Sync {
  at: NumberFor<Self::Block>,
  ) -> Pin<
  Box<
- dyn Future<Output = Box<dyn Iterator<Item = Arc<Self::InPoolTransaction>> + Send>>
- + Send,
+ dyn Future<
+ Output = Box<dyn ReadyTransactions<Item = Arc<Self::InPoolTransaction>> + Send>,
+ > + Send,
  >,
  >;
 
  /// Get an iterator for ready transactions ordered by priority.
- fn ready(&self) -> Box<dyn Iterator<Item = Arc<Self::InPoolTransaction>> + Send>;
+ fn ready(&self) -> Box<dyn ReadyTransactions<Item = Arc<Self::InPoolTransaction>> + Send>;
 
  // *** Block production
  /// Remove transactions identified by given hashes (and dependent transactions) from the pool.
@@ -254,6 +255,27 @@ pub trait TransactionPool: Send + Sync {
  fn ready_transaction(&self, hash: &TxHash<Self>) -> Option<Arc<Self::InPoolTransaction>>;
 }
 
+/// An iterator of ready transactions.
+///
+/// The trait extends regular [`std::iter::Iterator`] trait and allows reporting
+/// last-returned element as invalid.
+///
+/// The implementation is then allowed, for performance reasons, to change the elements
+/// returned next, by e.g. skipping elements that are known to depend on the reported
+/// transaction, which yields them invalid as well.
+pub trait ReadyTransactions: Iterator {
+ /// Report last-returned transaction as invalid.
+ ///
+ /// This might affect subsequent elements returned by the iterator, so dependent transactions
+ /// are skipped for performance reasons.
+ fn report_invalid(&mut self);
+}
+
+/// A no-op implementation for an empty iterator.
+impl<T> ReadyTransactions for std::iter::Empty<T> {
+ fn report_invalid(&mut self) {}
+}
+
 /// Events that the transaction pool listens for.
 pub enum ChainEvent<B: BlockT> {
  /// New best block have been added to the chain

client/transaction-pool/src/graph/base_pool.rs

@@ -36,7 +36,7 @@ use sp_runtime::{
 
 use super::{
  future::{FutureTransactions, WaitingTransaction},
- ready::ReadyTransactions,
+ ready::{BestIterator, ReadyTransactions},
 };
 
 /// Successful import result.
@@ -355,7 +355,7 @@ impl<Hash: hash::Hash + Member + Serialize, Ex: std::fmt::Debug> BasePool<Hash,
  }
 
  /// Returns an iterator over ready transactions in the pool.
- pub fn ready(&self) -> impl Iterator<Item = Arc<Transaction<Hash, Ex>>> {
+ pub fn ready(&self) -> BestIterator<Hash, Ex> {
  self.ready.get()
  }