Generate transactions in multiple threads. (#9842)

execution/executor-benchmark/Cargo.toml

@@ -38,6 +38,7 @@ async-trait = { workspace = true }
 bcs = { workspace = true }
 chrono = { workspace = true }
 clap = { workspace = true }
+derivative = { workspace = true }
 indicatif = { workspace = true }
 itertools = { workspace = true }
 move-core-types = { workspace = true }

execution/executor-benchmark/src/account_generator.rs

@@ -1,7 +1,8 @@
 // Copyright © Aptos Foundation
 // SPDX-License-Identifier: Apache-2.0
 
-use aptos_sdk::{move_types::account_address::AccountAddress, types::LocalAccount};
+use crate::transaction_generator::get_progress_bar;
+use aptos_sdk::types::LocalAccount;
 use rand::{rngs::StdRng, RngCore, SeedableRng};
 use std::{collections::VecDeque, sync::mpsc};
 
@@ -60,18 +61,25 @@ impl AccountGenerator {
 }
 
 pub struct AccountCache {
-    generator: AccountGenerator,
     pub accounts: VecDeque<LocalAccount>,
     pub rng: StdRng,
 }
 
 impl AccountCache {
     const SEED: Seed = [1; 32];
 
-    pub fn new(generator: AccountGenerator) -> Self {
+    pub fn new(mut generator: AccountGenerator, num_accounts: usize) -> Self {
+        let bar = get_progress_bar(num_accounts);
+        let accounts = (0..num_accounts)
+            .map(|_| {
+                let account = generator.generate();
+                bar.inc(1);
+                account
+            })
+            .collect();
+        bar.finish();
         Self {
-            generator,
-            accounts: VecDeque::new(),
+            accounts,
             rng: StdRng::from_seed(Self::SEED),
         }
     }
@@ -89,30 +97,19 @@ impl AccountCache {
         &self.accounts
     }
 
-    pub fn grow(&mut self, n: usize) {
-        let accounts: Vec<_> = (0..n).map(|_| self.generator.generate()).collect();
-        self.accounts.extend(accounts);
-    }
-
     pub fn get_random(&mut self) -> &mut LocalAccount {
-        let indices = rand::seq::index::sample(&mut self.rng, self.accounts.len(), 1);
-        let index = indices.index(0);
-
+        let index = self.get_random_index();
         &mut self.accounts[index]
     }
 
-    pub fn get_random_transfer_batch(
-        &mut self,
-        batch_size: usize,
-    ) -> (&LocalAccount, Vec<AccountAddress>) {
+    pub fn get_random_index(&mut self) -> usize {
+        rand::seq::index::sample(&mut self.rng, self.accounts.len(), 1).index(0)
+    }
+
+    pub fn get_random_transfer_batch(&mut self, batch_size: usize) -> (usize, Vec<usize>) {
         let indices = rand::seq::index::sample(&mut self.rng, self.accounts.len(), batch_size + 1);
         let sender_idx = indices.index(0);
-        let receivers = indices
-            .iter()
-            .skip(1)
-            .map(|i| self.accounts[i].address())
-            .collect();
-        let sender = &self.accounts[sender_idx];
-        (sender, receivers)
+        let receivers = indices.iter().skip(1).collect();
+        (sender_idx, receivers)
     }
 }

execution/executor-benchmark/src/lib.rs

@@ -20,7 +20,6 @@ use crate::{
     transaction_executor::TransactionExecutor, transaction_generator::TransactionGenerator,
 };
 use aptos_block_executor::counters as block_executor_counters;
-use aptos_block_partitioner::PartitionerConfig;
 use aptos_config::config::{NodeConfig, PrunerConfig};
 use aptos_db::AptosDB;
 use aptos_executor::{
@@ -166,17 +165,7 @@ pub fn run_benchmark<V>(
             db.clone(),
             // Initialization pipeline is temporary, so needs to be fully committed.
             // No discards/aborts allowed during initialization, even if they are allowed later.
-            PipelineConfig {
-                delay_execution_start: false,
-                split_stages: false,
-                skip_commit: false,
-                allow_discards: false,
-                allow_aborts: false,
-                num_executor_shards: 1,
-                async_partitioning: false,
-                use_global_executor: false,
-                partitioner_config: PartitionerConfig::default(),
-            },
+            PipelineConfig::default(),
         )
     });
 
@@ -207,8 +196,8 @@ pub fn run_benchmark<V>(
         genesis_key,
         block_sender,
         source_dir,
-        version,
         Some(num_accounts_to_load),
+        pipeline_config.num_generator_workers,
     );
 
     let mut start_time = Instant::now();
@@ -446,11 +435,11 @@ fn add_accounts_impl<V>(
     config.storage.rocksdb_configs.skip_index_and_usage = skip_index_and_usage;
     let (db, executor) = init_db_and_executor::<V>(&config);
 
-    let version = db.reader.get_latest_version().unwrap();
+    let start_version = db.reader.get_latest_version().unwrap();
 
     let (pipeline, block_sender) = Pipeline::new(
         executor,
-        version,
+        start_version,
         pipeline_config,
         Some(1 + num_new_accounts / block_size * 101 / 100),
     );
@@ -460,8 +449,8 @@ fn add_accounts_impl<V>(
         genesis_key,
         block_sender,
         &source_dir,
-        version,
         None,
+        pipeline_config.num_generator_workers,
     );
 
     let start_time = Instant::now();
@@ -477,7 +466,8 @@ fn add_accounts_impl<V>(
     pipeline.join();
 
     let elapsed = start_time.elapsed().as_secs_f32();
-    let delta_v = db.reader.get_latest_version().unwrap() - version;
+    let now_version = db.reader.get_latest_version().unwrap();
+    let delta_v = now_version - start_version;
     info!(
         "Overall TPS: account creation: {} txn/s",
         delta_v as f32 / elapsed,
@@ -492,7 +482,7 @@ fn add_accounts_impl<V>(
     println!(
         "Created {} new accounts. Now at version {}, total # of accounts {}.",
         num_new_accounts,
-        generator.version(),
+        now_version,
         generator.num_existing_accounts() + num_new_accounts,
     );
 
@@ -587,17 +577,7 @@ mod tests {
             false,
             false,
             false,
-            PipelineConfig {
-                delay_execution_start: false,
-                split_stages: false,
-                skip_commit: false,
-                allow_discards: false,
-                allow_aborts: false,
-                num_executor_shards: 1,
-                async_partitioning: false,
-                use_global_executor: false,
-                partitioner_config: Default::default(),
-            },
+            PipelineConfig::default(),
         );
 
         println!("run_benchmark");
@@ -619,17 +599,7 @@ mod tests {
             false,
             false,
             false,
-            PipelineConfig {
-                delay_execution_start: false,
-                split_stages: true,
-                skip_commit: false,
-                allow_discards: false,
-                allow_aborts: false,
-                num_executor_shards: 1,
-                async_partitioning: false,
-                use_global_executor: false,
-                partitioner_config: Default::default(),
-            },
+            PipelineConfig::default(),
         );
     }
 

execution/executor-benchmark/src/main.rs

@@ -105,6 +105,8 @@ pub struct PipelineOpt {
     #[clap(long)]
     use_global_executor: bool,
     #[clap(long, default_value = "4")]
+    num_generator_workers: usize,
+    #[clap(long, default_value = "4")]
     max_partitioning_rounds: usize,
     #[clap(long, default_value = "0.90")]
     partitioner_cross_shard_dep_avoid_threshold: f32,
@@ -127,6 +129,7 @@ impl PipelineOpt {
             num_executor_shards: self.num_executor_shards,
             async_partitioning: self.async_partitioning,
             use_global_executor: self.use_global_executor,
+            num_generator_workers: self.num_generator_workers,
             partitioner_config: self.partitioner_config(),
         }
     }

execution/executor-benchmark/src/pipeline.rs

@@ -14,6 +14,7 @@ use aptos_types::{
     block_executor::partitioner::ExecutableBlock,
     transaction::{Transaction, Version},
 };
+use derivative::Derivative;
 use std::{
     marker::PhantomData,
     sync::{
@@ -24,16 +25,20 @@ use std::{
     time::{Duration, Instant},
 };
 
-#[derive(Clone, Copy, Debug)]
+#[derive(Clone, Copy, Debug, Derivative)]
+#[derivative(Default)]
 pub struct PipelineConfig {
     pub delay_execution_start: bool,
     pub split_stages: bool,
     pub skip_commit: bool,
     pub allow_discards: bool,
     pub allow_aborts: bool,
+    #[derivative(Default(value = "1"))]
     pub num_executor_shards: usize,
     pub async_partitioning: bool,
     pub use_global_executor: bool,
+    #[derivative(Default(value = "4"))]
+    pub num_generator_workers: usize,
     pub partitioner_config: PartitionerConfig,
 }
 
@@ -63,7 +68,7 @@ where
             if config.delay_execution_start {
                 (num_blocks.unwrap() + 1).max(50)
             } else {
-                50
+                10
             }, /* bound */
         );