Enhance Mempool performance

consensus/core/src/api/mod.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 
 use crate::{
  acceptance_data::AcceptanceData,
- block::{Block, BlockTemplate},
+ block::{Block, BlockTemplate, TemplateBuildMode, TemplateTransactionSelector},
  block_count::BlockCount,
  blockstatus::BlockStatus,
  coinbase::MinerData,
@@ -27,7 +27,12 @@ pub type BlockValidationFuture = BoxFuture<'static, BlockProcessResult<BlockStat
 /// Abstracts the consensus external API
 #[allow(unused_variables)]
 pub trait ConsensusApi: Send + Sync {
- fn build_block_template(&self, miner_data: MinerData, txs: Vec<Transaction>) -> Result<BlockTemplate, RuleError> {
+ fn build_block_template(
+ &self,
+ miner_data: MinerData,
+ tx_selector: Box<dyn TemplateTransactionSelector>,
+ build_mode: TemplateBuildMode,
+ ) -> Result<BlockTemplate, RuleError> {
  unimplemented!()
  }
 
@@ -40,8 +45,24 @@ pub trait ConsensusApi: Send + Sync {
  }
 
  /// Populates the mempool transaction with maximally found UTXO entry data and proceeds to full transaction
- /// validation if all are found. If validation is successful, also [`transaction.calculated_fee`] is expected to be populated
- fn validate_mempool_transaction_and_populate(&self, transaction: &mut MutableTransaction) -> TxResult<()> {
+ /// validation if all are found. If validation is successful, also [`transaction.calculated_fee`] is expected to be populated.
+ fn validate_mempool_transaction(&self, transaction: &mut MutableTransaction) -> TxResult<()> {
+ unimplemented!()
+ }
+
+ /// Populates the mempool transactions with maximally found UTXO entry data and proceeds to full transactions
+ /// validation if all are found. If validation is successful, also [`transaction.calculated_fee`] is expected to be populated.
+ fn validate_mempool_transactions_in_parallel(&self, transactions: &mut [MutableTransaction]) -> Vec<TxResult<()>> {
+ unimplemented!()
+ }
+
+ /// Populates the mempool transaction with maximally found UTXO entry data.
+ fn populate_mempool_transaction(&self, transaction: &mut MutableTransaction) -> TxResult<()> {
+ unimplemented!()
+ }
+
+ /// Populates the mempool transactions with maximally found UTXO entry data.
+ fn populate_mempool_transactions_in_parallel(&self, transactions: &mut [MutableTransaction]) -> Vec<TxResult<()>> {
  unimplemented!()
  }
 

consensus/core/src/block.rs

@@ -1,6 +1,10 @@
 use std::sync::Arc;
 
-use crate::{coinbase::MinerData, header::Header, tx::Transaction};
+use crate::{
+ coinbase::MinerData,
+ header::Header,
+ tx::{Transaction, TransactionId},
+};
 use kaspa_hashes::Hash;
 
 /// A mutable block structure where header and transactions within can still be mutated.
@@ -64,6 +68,34 @@ impl Block {
  }
 }
 
+/// An abstraction for a recallable transaction selector with persistent state
+pub trait TemplateTransactionSelector {
+ /// Expected to return a batch of transactions which were not previously selected.
+ /// The batch will typically contain sufficient transactions to fill the block
+ /// mass (along with the previously unrejected txs), or will drain the selector 
+ fn select_transactions(&mut self) -> Vec<Transaction>;
+
+ /// Should be used to report invalid transactions obtained from the *most recent*
+ /// `select_transactions` call. Implementors should use this call to internally
+ /// track the selection state and discard the rejected tx from internal occupation calculations
+ fn reject_selection(&mut self, tx_id: TransactionId);
+
+ /// Determine whether this was an overall successful selection episode
+ fn is_successful(&self) -> bool;
+}
+
+/// Block template build mode
+#[derive(Clone, Copy, Debug)]
+pub enum TemplateBuildMode {
+ /// Block template build can possibly fail if `TemplateTransactionSelector::is_successful` deems the operation unsuccessful.
+ ///
+ /// In such a case, the build fails with `BlockRuleError::InvalidTransactionsInNewBlock`.
+ Standard,
+
+ /// Block template build always succeeds. The built block contains only the validated transactions.
+ Infallible,
+}
+
 /// A block template for miners.
 #[derive(Debug, Clone)]
 pub struct BlockTemplate {

consensus/core/src/config/mod.rs

@@ -59,6 +59,8 @@ pub struct Config {
 
  pub externalip: Option<IpAddress>,
 
+ pub block_template_cache_lifetime: Option<u64>,
+
  #[cfg(feature = "devnet-prealloc")]
  pub initial_utxo_set: Arc<UtxoCollection>,
 }
@@ -82,6 +84,7 @@ impl Config {
  user_agent_comments: Default::default(),
  externalip: None,
  p2p_listen_address: ContextualNetAddress::unspecified(),
+ block_template_cache_lifetime: None,
 
  #[cfg(feature = "devnet-prealloc")]
  initial_utxo_set: Default::default(),

consensus/core/src/errors/block.rs

@@ -1,4 +1,4 @@
-use std::fmt::Display;
+use std::{collections::HashMap, fmt::Display};
 
 use crate::{
  constants,
@@ -140,7 +140,7 @@ pub enum RuleError {
  InvalidTransactionsInUtxoContext(usize, usize),
 
  #[error("invalid transactions in new block template")]
- InvalidTransactionsInNewBlock(Vec<(TransactionId, TxRuleError)>),
+ InvalidTransactionsInNewBlock(HashMap<TransactionId, TxRuleError>),
 
  #[error("DAA window data has only {0} entries")]
  InsufficientDaaWindowSize(usize),

consensus/core/src/tx.rs

@@ -340,6 +340,12 @@ impl<T: AsRef<Transaction>> MutableTransaction<T> {
  }
 }
 
+impl<T: AsRef<Transaction>> AsRef<Transaction> for MutableTransaction<T> {
+ fn as_ref(&self) -> &Transaction {
+ self.tx.as_ref()
+ }
+}
+
 /// Private struct used to wrap a [`MutableTransaction`] as a [`VerifiableTransaction`]
 struct MutableTransactionVerifiableWrapper<'a, T: AsRef<Transaction>> {
  inner: &'a MutableTransaction<T>,

consensus/src/consensus/factory.rs

@@ -11,6 +11,7 @@ use kaspa_database::{
  registry::DatabaseStorePrefixes,
 };
 
+use kaspa_txscript::caches::TxScriptCacheCounters;
 use parking_lot::RwLock;
 use rocksdb::WriteBatch;
 use serde::{Deserialize, Serialize};
@@ -153,6 +154,7 @@ pub struct Factory {
  db_parallelism: usize,
  notification_root: Arc<ConsensusNotificationRoot>,
  counters: Arc<ProcessingCounters>,
+ tx_script_cache_counters: Arc<TxScriptCacheCounters>,
 }
 
 impl Factory {
@@ -163,6 +165,7 @@ impl Factory {
  db_parallelism: usize,
  notification_root: Arc<ConsensusNotificationRoot>,
  counters: Arc<ProcessingCounters>,
+ tx_script_cache_counters: Arc<TxScriptCacheCounters>,
  ) -> Self {
  let mut config = config.clone();
  #[cfg(feature = "devnet-prealloc")]
@@ -175,6 +178,7 @@ impl Factory {
  db_parallelism,
  notification_root,
  counters,
+ tx_script_cache_counters,
  }
  }
 }
@@ -208,6 +212,7 @@ impl ConsensusFactory for Factory {
  session_lock.clone(),
  self.notification_root.clone(),
  self.counters.clone(),
+ self.tx_script_cache_counters.clone(),
  entry.creation_timestamp,
  ));
 
@@ -236,6 +241,7 @@ impl ConsensusFactory for Factory {
  session_lock.clone(),
  self.notification_root.clone(),
  self.counters.clone(),
+ self.tx_script_cache_counters.clone(),
  entry.creation_timestamp,
  ));
 

consensus/src/consensus/mod.rs

@@ -41,7 +41,7 @@ use crate::{
 use kaspa_consensus_core::{
  acceptance_data::AcceptanceData,
  api::{BlockValidationFuture, ConsensusApi},
- block::{Block, BlockTemplate},
+ block::{Block, BlockTemplate, TemplateBuildMode, TemplateTransactionSelector},
  block_count::BlockCount,
  blockhash::BlockHashExtensions,
  blockstatus::BlockStatus,
@@ -70,6 +70,7 @@ use kaspa_consensusmanager::{SessionLock, SessionReadGuard};
 use kaspa_database::prelude::StoreResultExtensions;
 use kaspa_hashes::Hash;
 use kaspa_muhash::MuHash;
+use kaspa_txscript::caches::TxScriptCacheCounters;
 
 use std::thread::{self, JoinHandle};
 use std::{
@@ -132,6 +133,7 @@ impl Consensus {
  pruning_lock: SessionLock,
  notification_root: Arc<ConsensusNotificationRoot>,
  counters: Arc<ProcessingCounters>,
+ tx_script_cache_counters: Arc<TxScriptCacheCounters>,
  creation_timestamp: u64,
  ) -> Self {
  let params = &config.params;
@@ -147,7 +149,7 @@ impl Consensus {
  // Services and managers
  //
 
- let services = ConsensusServices::new(db.clone(), storage.clone(), config.clone());
+ let services = ConsensusServices::new(db.clone(), storage.clone(), config.clone(), tx_script_cache_counters);
 
  //
  // Processor channels
@@ -353,8 +355,13 @@ impl Consensus {
 }
 
 impl ConsensusApi for Consensus {
- fn build_block_template(&self, miner_data: MinerData, txs: Vec<Transaction>) -> Result<BlockTemplate, RuleError> {
- self.virtual_processor.build_block_template(miner_data, txs)
+ fn build_block_template(
+ &self,
+ miner_data: MinerData,
+ tx_selector: Box<dyn TemplateTransactionSelector>,
+ build_mode: TemplateBuildMode,
+ ) -> Result<BlockTemplate, RuleError> {
+ self.virtual_processor.build_block_template(miner_data, tx_selector, build_mode)
  }
 
  fn validate_and_insert_block(&self, block: Block) -> BlockValidationFuture {
@@ -367,11 +374,24 @@ impl ConsensusApi for Consensus {
  Box::pin(result)
  }
 
- fn validate_mempool_transaction_and_populate(&self, transaction: &mut MutableTransaction) -> TxResult<()> {
- self.virtual_processor.validate_mempool_transaction_and_populate(transaction)?;
+ fn validate_mempool_transaction(&self, transaction: &mut MutableTransaction) -> TxResult<()> {
+ self.virtual_processor.validate_mempool_transaction(transaction)?;
+ Ok(())
+ }
+
+ fn validate_mempool_transactions_in_parallel(&self, transactions: &mut [MutableTransaction]) -> Vec<TxResult<()>> {
+ self.virtual_processor.validate_mempool_transactions_in_parallel(transactions)
+ }
+
+ fn populate_mempool_transaction(&self, transaction: &mut MutableTransaction) -> TxResult<()> {
+ self.virtual_processor.populate_mempool_transaction(transaction)?;
  Ok(())
  }
 
+ fn populate_mempool_transactions_in_parallel(&self, transactions: &mut [MutableTransaction]) -> Vec<TxResult<()>> {
+ self.virtual_processor.populate_mempool_transactions_in_parallel(transactions)
+ }
+
  fn calculate_transaction_mass(&self, transaction: &Transaction) -> u64 {
  self.services.mass_calculator.calc_tx_mass(transaction)
  }

consensus/src/consensus/services.rs

@@ -18,6 +18,7 @@ use crate::{
 };
 
 use itertools::Itertools;
+use kaspa_txscript::caches::TxScriptCacheCounters;
 use std::sync::Arc;
 
 pub type DbGhostdagManager =
@@ -65,7 +66,12 @@ pub struct ConsensusServices {
 }
 
 impl ConsensusServices {
- pub fn new(db: Arc<DB>, storage: Arc<ConsensusStorage>, config: Arc<Config>) -> Arc<Self> {
+ pub fn new(
+ db: Arc<DB>,
+ storage: Arc<ConsensusStorage>,
+ config: Arc<Config>,
+ tx_script_cache_counters: Arc<TxScriptCacheCounters>,
+ ) -> Arc<Self> {
  let params = &config.params;
 
  let statuses_service = MTStatusesService::new(storage.statuses_store.clone());
@@ -144,6 +150,7 @@ impl ConsensusServices {
  params.ghostdag_k,
  params.coinbase_payload_script_public_key_max_len,
  params.coinbase_maturity,
+ tx_script_cache_counters,
  );
 
  let pruning_point_manager = PruningPointManager::new(

consensus/src/consensus/test_consensus.rs

@@ -49,8 +49,17 @@ impl TestConsensus {
  /// Creates a test consensus instance based on `config` with the provided `db` and `notification_sender`
  pub fn with_db(db: Arc<DB>, config: &Config, notification_sender: Sender<Notification>) -> Self {
  let notification_root = Arc::new(ConsensusNotificationRoot::new(notification_sender));
- let counters = Arc::new(ProcessingCounters::default());
- let consensus = Arc::new(Consensus::new(db, Arc::new(config.clone()), Default::default(), notification_root, counters, 0));
+ let counters = Default::default();
+ let tx_script_cache_counters = Default::default();
+ let consensus = Arc::new(Consensus::new(
+ db,
+ Arc::new(config.clone()),
+ Default::default(),
+ notification_root,
+ counters,
+ tx_script_cache_counters,
+ 0,
+ ));
  let block_builder = TestBlockBuilder::new(consensus.virtual_processor.clone());
 
  Self { params: config.params.clone(), consensus, block_builder, db_lifetime: Default::default() }
@@ -60,8 +69,17 @@ impl TestConsensus {
  pub fn with_notifier(config: &Config, notification_sender: Sender<Notification>) -> Self {
  let (db_lifetime, db) = create_temp_db!(ConnBuilder::default());
  let notification_root = Arc::new(ConsensusNotificationRoot::new(notification_sender));
- let counters = Arc::new(ProcessingCounters::default());
- let consensus = Arc::new(Consensus::new(db, Arc::new(config.clone()), Default::default(), notification_root, counters, 0));
+ let counters = Default::default();
+ let tx_script_cache_counters = Default::default();
+ let consensus = Arc::new(Consensus::new(
+ db,
+ Arc::new(config.clone()),
+ Default::default(),
+ notification_root,
+ counters,
+ tx_script_cache_counters,
+ 0,
+ ));
  let block_builder = TestBlockBuilder::new(consensus.virtual_processor.clone());
 
  Self { consensus, block_builder, params: config.params.clone(), db_lifetime }
@@ -72,8 +90,17 @@ impl TestConsensus {
  let (db_lifetime, db) = create_temp_db!(ConnBuilder::default());
  let (dummy_notification_sender, _) = async_channel::unbounded();
  let notification_root = Arc::new(ConsensusNotificationRoot::new(dummy_notification_sender));
- let counters = Arc::new(ProcessingCounters::default());
- let consensus = Arc::new(Consensus::new(db, Arc::new(config.clone()), Default::default(), notification_root, counters, 0));
+ let counters = Default::default();
+ let tx_script_cache_counters = Default::default();
+ let consensus = Arc::new(Consensus::new(
+ db,
+ Arc::new(config.clone()),
+ Default::default(),
+ notification_root,
+ counters,
+ tx_script_cache_counters,
+ 0,
+ ));
  let block_builder = TestBlockBuilder::new(consensus.virtual_processor.clone());
 
  Self { consensus, block_builder, params: config.params.clone(), db_lifetime }