[BlockSTM] Per-block Gas Limit

aptos-move/aptos-transaction-benchmarks/src/main.rs

@@ -47,6 +47,9 @@ struct ParamSweepOpt {
 
     #[clap(long, default_value = "10")]
     pub num_runs: usize,
+
+    #[clap(long)]
+    pub maybe_gas_limit: Option<u64>,
 }
 
 #[derive(Debug, Parser)]
@@ -71,6 +74,9 @@ struct ExecuteOpt {
 
     #[clap(long, default_value = "true")]
     pub no_conflict_txns: bool,
+
+    #[clap(long)]
+    pub maybe_gas_limit: Option<u64>,
 }
 
 fn param_sweep(opt: ParamSweepOpt) {
@@ -85,6 +91,8 @@ fn param_sweep(opt: ParamSweepOpt) {
     let run_parallel = !opt.skip_parallel;
     let run_sequential = !opt.skip_sequential;
 
+    let maybe_gas_limit = opt.maybe_gas_limit;
+
     assert!(
         run_sequential || run_parallel,
         "Must run at least one of parallel or sequential"
@@ -102,6 +110,7 @@ fn param_sweep(opt: ParamSweepOpt) {
                 1,
                 concurrency_level,
                 false,
+                maybe_gas_limit,
             );
             par_tps.sort();
             seq_tps.sort();
@@ -162,6 +171,7 @@ fn execute(opt: ExecuteOpt) {
         opt.num_executor_shards,
         opt.concurrency_level_per_shard,
         opt.no_conflict_txns,
+        opt.maybe_gas_limit,
     );
 
     let sum: usize = par_tps.iter().sum();

aptos-move/aptos-transaction-benchmarks/src/transactions.rs

@@ -73,6 +73,7 @@ where
                     self.num_transactions,
                     1,
                     AccountPickStyle::Unlimited,
+                    None,
                 )
             },
             |state| state.execute_sequential(),
@@ -91,6 +92,7 @@ where
                     self.num_transactions,
                     1,
                     AccountPickStyle::Unlimited,
+                    None,
                 )
             },
             |state| state.execute_parallel(),
@@ -111,6 +113,7 @@ where
         num_executor_shards: usize,
         concurrency_level_per_shard: usize,
         no_conflict_txn: bool,
+        maybe_gas_limit: Option<u64>,
     ) -> (Vec<usize>, Vec<usize>) {
         let mut par_tps = Vec::new();
         let mut seq_tps = Vec::new();
@@ -135,6 +138,7 @@ where
             num_txn,
             num_executor_shards,
             account_pick_style,
+            maybe_gas_limit,
         );
 
         for i in 0..total_runs {
@@ -184,12 +188,14 @@ where
         num_transactions: usize,
         num_executor_shards: usize,
         account_pick_style: AccountPickStyle,
+        maybe_gas_limit: Option<u64>,
     ) -> Self {
         Self::with_universe(
             strategy,
             universe_strategy(num_accounts, num_transactions, account_pick_style),
             num_transactions,
             num_executor_shards,
+            maybe_gas_limit,
         )
     }
 
@@ -200,6 +206,7 @@ where
         universe_strategy: impl Strategy<Value = AccountUniverseGen>,
         num_transactions: usize,
         num_executor_shards: usize,
+        maybe_gas_limit: Option<u64>,
     ) -> Self {
         let mut runner = TestRunner::default();
         let universe_gen = universe_strategy
@@ -214,9 +221,13 @@ where
         let universe = universe_gen.setup_gas_cost_stability(&mut executor);
 
         let state_view = Arc::new(executor.get_state_view().clone());
-        let parallel_block_executor =
-            Arc::new(ShardedBlockExecutor::new(num_executor_shards, None));
-        let sequential_block_executor = Arc::new(ShardedBlockExecutor::new(1, Some(1)));
+        let parallel_block_executor = Arc::new(ShardedBlockExecutor::new(
+            num_executor_shards,
+            None,
+            maybe_gas_limit,
+        ));
+        let sequential_block_executor =
+            Arc::new(ShardedBlockExecutor::new(1, Some(1), maybe_gas_limit));
 
         let validator_set = ValidatorSet::fetch_config(
             &FakeExecutor::from_head_genesis()

aptos-move/aptos-vm-logging/src/lib.rs

@@ -76,12 +76,12 @@ pub fn speculative_log(level: Level, context: &AdapterLogSchema, message: String
     };
 }
 
-/// Flushes the currently stored logs, and swaps the speculative log / event storage with None.
+/// Flushes the first num_to_flush logs in the currently stored logs, and swaps the speculative log / event storage with None.
 /// Must be called after block execution is complete (removes the storage from Arc).
-pub fn flush_speculative_logs() {
+pub fn flush_speculative_logs(num_to_flush: usize) {
     if let Some(log_events_ptr) = BUFFERED_LOG_EVENTS.swap(None) {
         match Arc::try_unwrap(log_events_ptr) {
-            Ok(log_events) => log_events.flush(),
+            Ok(log_events) => log_events.flush(num_to_flush),
             Err(_) => {
                 alert!("Speculative log storage must be uniquely owned to flush");
             },

aptos-move/aptos-vm/src/aptos_vm.rs

@@ -1483,6 +1483,41 @@ impl VMExecutor for AptosVM {
             transactions,
             state_view,
             Self::get_concurrency_level(),
+            None,
+        );
+        if ret.is_ok() {
+            // Record the histogram count for transactions per block.
+            BLOCK_TRANSACTION_COUNT.observe(count as f64);
+        }
+        ret
+    }
+
+    fn execute_block_with_gas_limit(
+        transactions: Vec<Transaction>,
+        state_view: &(impl StateView + Sync),
+        maybe_gas_limit: Option<u64>,
+    ) -> std::result::Result<Vec<TransactionOutput>, VMStatus> {
+        fail_point!("move_adapter::execute_block", |_| {
+            Err(VMStatus::Error(
+                StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
+                None,
+            ))
+        });
+
+        let log_context = AdapterLogSchema::new(state_view.id(), 0);
+        info!(
+            log_context,
+            "Executing block, transaction count: {}",
+            transactions.len()
+        );
+
+        let count = transactions.len();
+        let ret = BlockAptosVM::execute_block(
+            Arc::clone(&RAYON_EXEC_POOL),
+            transactions,
+            state_view,
+            Self::get_concurrency_level(),
+            maybe_gas_limit,
         );
         if ret.is_ok() {
             // Record the histogram count for transactions per block.

aptos-move/aptos-vm/src/block_executor/mod.rs

@@ -124,6 +124,13 @@ impl BlockExecutorTransactionOutput for AptosTransactionOutput {
             "Could not combine TransactionOutputExt with deltas"
         );
     }
+
+    /// Return the amount of gas consumed by the transaction.
+    fn gas_used(&self) -> u64 {
+        self.committed_output
+            .get()
+            .map_or(0, |output| output.gas_used())
+    }
 }
 
 pub struct BlockAptosVM();
@@ -134,6 +141,7 @@ impl BlockAptosVM {
         transactions: Vec<Transaction>,
         state_view: &S,
         concurrency_level: usize,
+        maybe_gas_limit: Option<u64>,
     ) -> Result<Vec<TransactionOutput>, VMStatus> {
         let _timer = BLOCK_EXECUTOR_EXECUTE_BLOCK_SECONDS.start_timer();
         // Verify the signatures of all the transactions in parallel.
@@ -151,23 +159,32 @@ impl BlockAptosVM {
             });
         drop(signature_verification_timer);
 
-        init_speculative_logs(signature_verified_block.len());
+        let num_txns = signature_verified_block.len();
+        init_speculative_logs(num_txns);
 
         BLOCK_EXECUTOR_CONCURRENCY.set(concurrency_level as i64);
         let executor = BlockExecutor::<PreprocessedTransaction, AptosExecutorTask<S>, S>::new(
             concurrency_level,
             executor_thread_pool,
+            maybe_gas_limit,
         );
 
         let ret = executor.execute_block(state_view, signature_verified_block, state_view);
 
-        flush_speculative_logs();
-
         match ret {
-            Ok(outputs) => Ok(outputs
-                .into_iter()
-                .map(|output| output.take_output())
-                .collect()),
+            Ok(outputs) => {
+                let output_vec: Vec<TransactionOutput> = outputs
+                    .into_iter()
+                    .map(|output| output.take_output())
+                    .collect();
+
+                // Flush the speculative logs of the committed transactions.
+                let pos = output_vec.partition_point(|o| !o.status().is_retry());
+
+                flush_speculative_logs(pos);
+
+                Ok(output_vec)
+            },
             Err(Error::ModulePathReadWrite) => {
                 unreachable!("[Execution]: Must be handled by sequential fallback")
             },

aptos-move/aptos-vm/src/lib.rs

@@ -155,6 +155,14 @@ pub trait VMExecutor: Send + Sync {
         state_view: &(impl StateView + Sync),
     ) -> Result<Vec<TransactionOutput>, VMStatus>;
 
+    /// Executes a block of transactions with per_block_gas_limit
+    /// and returns output for each one of them.
+    fn execute_block_with_gas_limit(
+        transactions: Vec<Transaction>,
+        state_view: &(impl StateView + Sync),
+        maybe_gas_limit: Option<u64>,
+    ) -> Result<Vec<TransactionOutput>, VMStatus>;
+
     /// Executes a block of transactions using a sharded block executor and returns the results.
     fn execute_block_sharded<S: StateView + Sync + Send + 'static>(
         sharded_block_executor: &ShardedBlockExecutor<S>,

aptos-move/aptos-vm/src/sharded_block_executor/executor_shard.rs

@@ -18,6 +18,7 @@ pub struct ExecutorShard<S: StateView + Sync + Send + 'static> {
     executor_thread_pool: Arc<rayon::ThreadPool>,
     command_rx: Receiver<ExecutorShardCommand<S>>,
     result_tx: Sender<Result<Vec<TransactionOutput>, VMStatus>>,
+    maybe_gas_limit: Option<u64>,
 }
 
 impl<S: StateView + Sync + Send + 'static> ExecutorShard<S> {
@@ -26,6 +27,7 @@ impl<S: StateView + Sync + Send + 'static> ExecutorShard<S> {
         num_executor_threads: usize,
         command_rx: Receiver<ExecutorShardCommand<S>>,
         result_tx: Sender<Result<Vec<TransactionOutput>, VMStatus>>,
+        maybe_gas_limit: Option<u64>,
     ) -> Self {
         let executor_thread_pool = Arc::new(
             rayon::ThreadPoolBuilder::new()
@@ -38,6 +40,7 @@ impl<S: StateView + Sync + Send + 'static> ExecutorShard<S> {
             executor_thread_pool,
             command_rx,
             result_tx,
+            maybe_gas_limit,
         }
     }
 
@@ -60,6 +63,7 @@ impl<S: StateView + Sync + Send + 'static> ExecutorShard<S> {
                         transactions,
                         state_view.as_ref(),
                         concurrency_level_per_shard,
+                        self.maybe_gas_limit,
                     );
                     drop(state_view);
                     self.result_tx.send(ret).unwrap();

aptos-move/aptos-vm/src/sharded_block_executor/mod.rs

@@ -38,7 +38,11 @@ pub enum ExecutorShardCommand<S: StateView + Sync + Send + 'static> {
 }
 
 impl<S: StateView + Sync + Send + 'static> ShardedBlockExecutor<S> {
-    pub fn new(num_executor_shards: usize, executor_threads_per_shard: Option<usize>) -> Self {
+    pub fn new(
+        num_executor_shards: usize,
+        executor_threads_per_shard: Option<usize>,
+        maybe_gas_limit: Option<u64>,
+    ) -> Self {
         assert!(num_executor_shards > 0, "num_executor_shards must be > 0");
         let executor_threads_per_shard = executor_threads_per_shard.unwrap_or_else(|| {
             (num_cpus::get() as f64 / num_executor_shards as f64).ceil() as usize
@@ -56,6 +60,7 @@ impl<S: StateView + Sync + Send + 'static> ShardedBlockExecutor<S> {
                 executor_threads_per_shard,
                 transactions_rx,
                 result_tx,
+                maybe_gas_limit,
             ));
         }
         info!(
@@ -128,13 +133,19 @@ fn spawn_executor_shard<S: StateView + Sync + Send + 'static>(
     concurrency_level: usize,
     command_rx: Receiver<ExecutorShardCommand<S>>,
     result_tx: Sender<Result<Vec<TransactionOutput>, VMStatus>>,
+    maybe_gas_limit: Option<u64>,
 ) -> thread::JoinHandle<()> {
     // create and start a new executor shard in a separate thread
     thread::Builder::new()
         .name(format!("executor-shard-{}", shard_id))
         .spawn(move || {
-            let executor_shard =
-                ExecutorShard::new(shard_id, concurrency_level, command_rx, result_tx);
+            let executor_shard = ExecutorShard::new(
+                shard_id,
+                concurrency_level,
+                command_rx,
+                result_tx,
+                maybe_gas_limit,
+            );
             executor_shard.start();
         })
         .unwrap()

aptos-move/block-executor/src/counters.rs

@@ -24,6 +24,24 @@ pub static SPECULATIVE_ABORT_COUNT: Lazy<IntCounter> = Lazy::new(|| {
     .unwrap()
 });
 
+/// Count of times the BlockSTM is early halted due to exceeding the per-block gas limit.
+pub static PARALLEL_EXCEED_PER_BLOCK_GAS_LIMIT_COUNT: Lazy<IntCounter> = Lazy::new(|| {
+    register_int_counter!(
+        "aptos_execution_par_gas_limit_count",
+        "Count of times the BlockSTM is early halted due to exceeding the per-block gas limit"
+    )
+    .unwrap()
+});
+
+/// Count of times the sequential execution is early halted due to exceeding the per-block gas limit.
+pub static SEQUENTIAL_EXCEED_PER_BLOCK_GAS_LIMIT_COUNT: Lazy<IntCounter> = Lazy::new(|| {
+    register_int_counter!(
+        "aptos_execution_seq_gas_limit_count",
+        "Count of times the sequential execution is early halted due to exceeding the per-block gas limit"
+    )
+    .unwrap()
+});
+
 pub static PARALLEL_EXECUTION_SECONDS: Lazy<Histogram> = Lazy::new(|| {
     register_histogram!(
         // metric name
@@ -109,3 +127,39 @@ pub static DEPENDENCY_WAIT_SECONDS: Lazy<Histogram> = Lazy::new(|| {
     )
     .unwrap()
 });
+
+pub static PARALLEL_PER_BLOCK_GAS: Lazy<Histogram> = Lazy::new(|| {
+    register_histogram!(
+        "aptos_execution_par_per_block_gas",
+        "The per-block consumed gas in parallel execution (Block STM)",
+        exponential_buckets(/*start=*/ 1.0, /*factor=*/ 2.0, /*count=*/ 30).unwrap(),
+    )
+    .unwrap()
+});
+
+pub static SEQUENTIAL_PER_BLOCK_GAS: Lazy<Histogram> = Lazy::new(|| {
+    register_histogram!(
+        "aptos_execution_seq_per_block_gas",
+        "The per-block consumed gas in sequential execution",
+        exponential_buckets(/*start=*/ 1.0, /*factor=*/ 2.0, /*count=*/ 30).unwrap(),
+    )
+    .unwrap()
+});
+
+pub static PARALLEL_PER_TXN_GAS: Lazy<Histogram> = Lazy::new(|| {
+    register_histogram!(
+        "aptos_execution_par_per_txn_gas",
+        "The per-txn consumed gas in parallel execution (Block STM)",
+        exponential_buckets(/*start=*/ 1.0, /*factor=*/ 1.5, /*count=*/ 30).unwrap(),
+    )
+    .unwrap()
+});
+
+pub static SEQUENTIAL_PER_TXN_GAS: Lazy<Histogram> = Lazy::new(|| {
+    register_histogram!(
+        "aptos_execution_seq_per_txn_gas",
+        "The per-txn consumed gas in sequential execution",
+        exponential_buckets(/*start=*/ 1.0, /*factor=*/ 1.5, /*count=*/ 30).unwrap(),
+    )
+    .unwrap()
+});