Additional peformance improvements and metrics for parallel execution

benches/Cargo.toml

@@ -13,6 +13,7 @@ default = [
     "fuel-core/u32-tx-count",
     "fuel-core-chain-config/u32-tx-count",
     "fuel-core-types/u32-tx-pointer",
+    "parallel-executor"
 ]
 fault-proving = [
     "fuel-core-types/fault-proving",
@@ -48,6 +49,7 @@ fuel-core = { path = "../crates/fuel-core", default-features = false, features =
 ] }
 fuel-core-chain-config = { workspace = true }
 fuel-core-database = { path = "./../crates/database" }
+fuel-core-metrics = { workspace = true }
 fuel-core-services = { path = "./../crates/services" }
 fuel-core-storage = { path = "./../crates/storage" }
 fuel-core-sync = { path = "./../crates/services/sync", features = [

benches/src/bin/tps_bench.rs

@@ -1,11 +1,15 @@
 // Define arguments
 
-use fuel_core::service::config::Trigger;
+use fuel_core::service::config::{
+    ExecutorMode,
+    Trigger,
+};
 use fuel_core_chain_config::{
     ChainConfig,
     CoinConfig,
     SnapshotMetadata,
 };
+use fuel_core_metrics::encode_metrics;
 use fuel_core_storage::transactional::AtomicView;
 use fuel_core_types::{
     blockchain::transaction::TransactionExt,
@@ -54,7 +58,7 @@ use test_helpers::builder::{
     TestSetupBuilder,
 };
 
-const PATH_SNAPSHOT: &str = "/Users/green/fuel/fuel-core-2/benches/local-testnet";
+const PATH_SNAPSHOT: &str = "./local-testnet";
 
 fn checked_parameters() -> CheckPredicateParams {
     let metadata = SnapshotMetadata::read(PATH_SNAPSHOT).unwrap();
@@ -70,6 +74,16 @@ struct Args {
     pub number_of_cores: usize,
     #[clap(short = 't', long, default_value = "150000")]
     pub number_of_transactions: u64,
+    #[clap(long, default_value = "0")]
+    pub txpool_verification_threads: usize,
+    #[clap(long, default_value = "0")]
+    pub txpool_verification_queue_size: usize,
+    #[clap(long, default_value = "0")]
+    pub txpool_p2p_sync_threads: usize,
+    #[clap(long, default_value = "0")]
+    pub txpool_p2p_sync_queue_size: usize,
+    #[clap(long, default_value = "0")]
+    pub executor_parallel_worker_count: usize,
 }
 
 fn generate_transactions(nb_txs: u64, rng: &mut StdRng) -> Vec<Transaction> {
@@ -135,6 +149,9 @@ fn generate_transactions(nb_txs: u64, rng: &mut StdRng) -> Vec<Transaction> {
 }
 
 fn main() {
+    let _ = tracing_subscriber::fmt()
+        .with_max_level(tracing::Level::WARN)
+        .try_init();
     let args = Args::parse();
     let mut rng = StdRng::seed_from_u64(2322u64);
 
@@ -200,9 +217,29 @@ fn main() {
             Some(tx.max_gas(&chain_conf.consensus_parameters).unwrap())
         })
         .sum();
-    test_builder.gas_limit = Some(gas_limit * 4);
+    test_builder.gas_limit = Some(gas_limit);
     test_builder.block_size_limit = Some(u64::MAX);
     test_builder.number_threads_pool_verif = args.number_of_cores;
+    if args.txpool_verification_threads != 0 {
+        test_builder.txpool_verification_threads = args.txpool_verification_threads;
+    }
+    if args.txpool_verification_queue_size != 0 {
+        test_builder.txpool_verification_queue_size = args.txpool_verification_queue_size;
+    }
+    if args.txpool_p2p_sync_threads != 0 {
+        test_builder.txpool_p2p_sync_threads = args.txpool_p2p_sync_threads;
+    }
+    if args.txpool_p2p_sync_queue_size != 0 {
+        test_builder.txpool_p2p_sync_queue_size = args.txpool_p2p_sync_queue_size;
+    }
+    if args.executor_parallel_worker_count != 0 {
+        test_builder.executor_parallel_worker_count = args.executor_parallel_worker_count;
+    }
+    #[cfg(feature = "parallel-executor")]
+    {
+        test_builder.executor_mode = ExecutorMode::Parallel;
+        test_builder.executor_metrics = true;
+    }
     test_builder.max_txs = transactions.len();
     // spin up node
     let rt = tokio::runtime::Builder::new_multi_thread()
@@ -241,10 +278,19 @@ fn main() {
                 .unwrap()
                 .unwrap();
             assert_eq!(block.entity.transactions().len(), transactions.len() + 1);
+            println!("transaction count: {}", block.entity.transactions().len());
             block
         }
     });
 
+    if let Ok(metrics) = encode_metrics() {
+        for line in metrics.lines().filter(|line| {
+            line.starts_with("parallel_executor_") && !line.starts_with('#')
+        }) {
+            println!("metrics: {line}");
+        }
+    }
+
     // rt.block_on(async move {
     //     test_builder.set_chain_config(chain_conf.clone());
     //     let TestContext { srv, .. } = test_builder.finalize().await;

bin/fuel-core/src/cli/run.rs

@@ -32,6 +32,7 @@ use fuel_core::{
         RelayerConsensusConfig,
         config::{
             DaCompressionMode,
+            ExecutorConfig,
             Trigger,
         },
         genesis::NotifyCancel,
@@ -65,10 +66,7 @@ use fuel_core_metrics::config::{
     DisableConfig,
     Module,
 };
-use fuel_core_types::{
-    blockchain::header::StateTransitionBytecodeVersion,
-    signer::SignMode,
-};
+use fuel_core_types::signer::SignMode;
 use pyroscope::{
     PyroscopeAgent,
     pyroscope::PyroscopeAgentRunning,
@@ -104,9 +102,6 @@ use fuel_core::state::historical_rocksdb::StateRewindPolicy;
 
 use crate::cli::run::gas_price::GasPriceArgs;
 use fuel_core::service::config::GasPriceConfig;
-#[cfg(feature = "parallel-executor")]
-use std::num::NonZeroUsize;
-
 #[cfg(feature = "p2p")]
 mod p2p;
 
@@ -117,6 +112,7 @@ mod rpc;
 mod shared_sequencer;
 
 mod consensus;
+mod executor;
 mod gas_price;
 mod graphql;
 #[cfg(feature = "p2p")]
@@ -127,6 +123,10 @@ mod relayer;
 mod tx_pool;
 mod tx_status_manager;
 
+use crate::cli::run::executor::ExecutorArgs;
+#[cfg(feature = "parallel-executor")]
+use fuel_core::service::config::ParallelExecutorConfig;
+
 /// Run the Fuel client node locally.
 #[derive(Debug, Clone, Parser)]
 pub struct Command {
@@ -229,13 +229,8 @@ pub struct Command {
     pub utxo_validation: bool,
 
     /// Overrides the version of the native executor.
-    #[arg(long = "native-executor-version", env)]
-    pub native_executor_version: Option<StateTransitionBytecodeVersion>,
-
-    /// Number of cores to use for the parallel executor.
-    #[cfg(feature = "parallel-executor")]
-    #[arg(long = "executor-number-of-cores", env, default_value = "1")]
-    pub executor_number_of_cores: NonZeroUsize,
+    #[clap(flatten)]
+    pub executor: ExecutorArgs,
 
     /// All the configurations for the gas price service.
     #[clap(flatten)]
@@ -363,9 +358,7 @@ impl Command {
             allow_syscall,
             expensive_subscriptions,
             utxo_validation,
-            native_executor_version,
-            #[cfg(feature = "parallel-executor")]
-            executor_number_of_cores,
+            executor,
             gas_price,
             consensus_key,
             #[cfg(feature = "aws-kms")]
@@ -418,6 +411,17 @@ impl Command {
             anyhow::bail!("`--allow-syscall` is only allowed in debug mode");
         }
 
+        let ExecutorArgs {
+            executor_mode,
+            native_executor_version,
+            #[cfg(feature = "parallel-executor")]
+            executor_number_of_cores,
+            #[cfg(feature = "parallel-executor")]
+            executor_metrics,
+            #[cfg(feature = "parallel-executor")]
+            executor_worker_count_policy,
+        } = executor;
+
         let enabled_metrics = metrics.list_of_enabled();
 
         if !enabled_metrics.is_empty() {
@@ -740,12 +744,19 @@ impl Command {
             debug,
             historical_execution,
             expensive_subscriptions,
-            native_executor_version,
+            executor: ExecutorConfig {
+                mode: executor_mode,
+                native_executor_version,
+                #[cfg(feature = "parallel-executor")]
+                parallel: ParallelExecutorConfig {
+                    worker_count: executor_number_of_cores,
+                    worker_count_policy: executor_worker_count_policy,
+                    metrics: executor_metrics,
+                },
+            },
             continue_on_error,
             allow_syscall,
             utxo_validation,
-            #[cfg(feature = "parallel-executor")]
-            executor_number_of_cores,
             block_production: trigger,
             leader_lock,
             predefined_blocks_path,
@@ -762,6 +773,7 @@ impl Command {
                 pending_pool_tx_ttl: tx_pending_pool_ttl.into(),
                 max_pending_pool_size_percentage: tx_pending_pool_size_percentage,
                 metrics: metrics.is_enabled(Module::TxPool),
+                eagerly_include_tx_dependency_graphs: false,
             },
             block_producer: ProducerConfig {
                 coinbase_recipient,

bin/fuel-core/src/cli/run/executor.rs

@@ -0,0 +1,44 @@
+use clap::Args;
+use fuel_core::service::config::ExecutorMode;
+#[cfg(feature = "parallel-executor")]
+use fuel_core::service::config::ParallelExecutorWorkerCountPolicy;
+use fuel_core_types::blockchain::header::StateTransitionBytecodeVersion;
+
+#[cfg(feature = "parallel-executor")]
+use std::num::NonZeroUsize;
+
+#[derive(Debug, Clone, Args)]
+pub struct ExecutorArgs {
+    /// Overrides the version of the native executor.
+    #[arg(long = "native-executor-version", env)]
+    pub native_executor_version: Option<StateTransitionBytecodeVersion>,
+
+    /// Executor mode to use.
+    #[arg(long = "executor-mode", value_enum, default_value = "normal", env)]
+    pub executor_mode: ExecutorMode,
+
+    /// Number of cores to use for the parallel executor.
+    #[cfg(feature = "parallel-executor")]
+    #[arg(
+        long = "executor-number-of-cores",
+        env,
+        default_value = "1",
+        alias = "executor-worker-count"
+    )]
+    pub executor_number_of_cores: NonZeroUsize,
+
+    /// Enable metrics for the parallel executor.
+    #[cfg(feature = "parallel-executor")]
+    #[arg(long = "executor-metrics", env)]
+    pub executor_metrics: bool,
+
+    /// Worker count policy for tx selection in parallel scheduler.
+    #[cfg(feature = "parallel-executor")]
+    #[arg(
+        long = "executor-worker-count-policy",
+        value_enum,
+        default_value = "static-max",
+        env
+    )]
+    pub executor_worker_count_policy: ParallelExecutorWorkerCountPolicy,
+}

crates/fuel-core/src/graphql_api/worker_service.rs

@@ -162,6 +162,7 @@ where
     D: ports::worker::OffChainDatabase,
 {
     fn process_block(&mut self, result: SharedImportResult) -> anyhow::Result<()> {
+        let instant = tokio::time::Instant::now();
         let block = &result.sealed_block.entity;
         let mut transaction = self.database.transaction();
         // save the status for every transaction using the finalized block id
@@ -170,12 +171,21 @@ where
             self.asset_metadata_indexation_enabled,
             &mut transaction,
         )?;
+        tracing::warn!(
+            "persist_transaction_status elapsed time: {:?}",
+            instant.elapsed()
+        );
 
         // save the associated owner for each transaction in the block
         index_tx_owners_for_block(block, &mut transaction, &self.chain_id)?;
+        tracing::warn!(
+            "index_tx_owners_for_block elapsed time: {:?}",
+            instant.elapsed()
+        );
 
         // save the transaction related information
         process_transactions(block.transactions().iter(), &mut transaction)?;
+        tracing::warn!("process_transactions elapsed time: {:?}", instant.elapsed());
 
         let height = block.header().height();
         let block_id = block.id();
@@ -186,6 +196,7 @@ where
         let total_tx_count = transaction
             .increase_tx_count(block.transactions().len() as u64)
             .unwrap_or_default();
+        tracing::warn!("increase_tx_count elapsed time: {:?}", instant.elapsed());
 
         process_executor_events(
             result.events.iter().map(Cow::Borrowed),
@@ -194,8 +205,13 @@ where
             self.coins_to_spend_indexation_enabled,
             &self.base_asset_id,
         )?;
+        tracing::warn!(
+            "process_executor_events elapsed time: {:?}",
+            instant.elapsed()
+        );
 
         transaction.commit()?;
+        tracing::warn!("transaction.commit elapsed time: {:?}", instant.elapsed());
 
         for status in result.tx_status.iter() {
             let tx_id = status.id;
@@ -205,6 +221,11 @@ where
                 .send_complete(tx_id, height, status.into());
         }
 
+        tracing::warn!(
+            "process_block elapsed time: {:?} for block: {:?}",
+            instant.elapsed(),
+            height
+        );
         // Notify subscribers and update last seen block height
         self.shared_state
             .block_height_subscription_handler