[BlockSTM] Per-block Gas Limit

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

@@ -81,6 +81,11 @@ impl BlockExecutorTransactionOutput for AptosTransactionOutput {
             TransactionStatus::Retry,
         )))
     }
+
+    /// Return the amount of gas consumed by the transaction.
+    fn gas_used(&self) -> u64 {
+        self.0.txn_output().gas_used()
+    }
 }
 
 pub struct BlockAptosVM();

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

@@ -10,7 +10,7 @@ use crate::{
     },
     errors::*,
     output_delta_resolver::OutputDeltaResolver,
-    scheduler::{Scheduler, SchedulerTask, Wave},
+    scheduler::{DependencyStatus, Scheduler, SchedulerTask, Wave},
     task::{ExecutionStatus, ExecutorTask, Transaction, TransactionOutput},
     txn_last_input_output::TxnLastInputOutput,
     view::{LatestView, MVHashMapView},
@@ -34,6 +34,9 @@ use std::{
     sync::atomic::{AtomicBool, Ordering},
 };
 
+// todo: add this number to config and think about the number.
+const PER_BLOCK_GAS_LIMIT: u64 = 100000;
+
 pub static RAYON_EXEC_POOL: Lazy<rayon::ThreadPool> = Lazy::new(|| {
     rayon::ThreadPoolBuilder::new()
         .num_threads(num_cpus::get())
@@ -141,7 +144,15 @@ where
             versioned_cache.delete(&k, idx_to_execute);
         }
 
-        last_input_output.record(idx_to_execute, speculative_view.take_reads(), result);
+        if last_input_output
+            .record(idx_to_execute, speculative_view.take_reads(), result)
+            .is_err()
+        {
+            // When there is module publishing r/w intersection, can early halt BlockSTM to
+            // fallback to sequential execution.
+            scheduler.halt();
+            return SchedulerTask::NoTask;
+        }
         scheduler.finish_execution(idx_to_execute, incarnation, updates_outside)
     }
 
@@ -160,7 +171,7 @@ where
         let (idx_to_validate, incarnation) = version_to_validate;
         let read_set = last_input_output
             .read_set(idx_to_validate)
-            .expect("Prior read-set must be recorded");
+            .expect("[BlockSTM]: Prior read-set must be recorded");
 
         let valid = read_set.iter().all(|r| {
             match versioned_cache.fetch_data(r.path(), idx_to_validate) {
@@ -216,16 +227,48 @@ where
 
         let _timer = WORK_WITH_TASK_SECONDS.start_timer();
         let mut scheduler_task = SchedulerTask::NoTask;
+        let mut accumulated_gas = 0;
         loop {
             // Only one thread try_commit to avoid contention.
             if committing {
                 // Keep committing txns until there is no more that can be committed now.
                 while let Some(txn_idx) = scheduler.try_commit() {
+                    // Committed the last transaction, BlockSTM finishes execution.
                     if txn_idx as usize + 1 == block.len() {
-                        // Committed the last transaction / everything.
                         scheduler_task = SchedulerTask::Done;
                         break;
                     }
+
+                    // When there is a txn with Abort or SkipRest as the execution output, can early halt BlockSTM.
+                    if let ExecutionStatus::Abort(_) | ExecutionStatus::SkipRest(_) =
+                        last_input_output.write_set(txn_idx).as_ref()
+                    {
+                        scheduler.halt();
+                        break;
+                    }
+
+                    // Calculating the accumulated gas of the committed txns.
+                    let txn_gas = match last_input_output.write_set(txn_idx).as_ref() {
+                        ExecutionStatus::Success(output) => output.gas_used(),
+                        _ => unreachable!(),
+                    };
+                    accumulated_gas += txn_gas;
+
+                    // When the accumulated gas of the committed txns exceeds PER_BLOCK_GAS_LIMIT, early halt BlockSTM.
+                    if accumulated_gas >= PER_BLOCK_GAS_LIMIT {
+                        // Set the execution output status to be SkipRest, to skip the rest of the txns.
+                        last_input_output.update_to_skip_rest(txn_idx);
+                        scheduler.halt();
+                        break;
+                    }
+
+                    // Remark: When early halting the BlockSTM, we have to make sure the current / new tasks
+                    // will be properly handled by the threads. For instance, it is possible that the committing
+                    // thread holds an execution task from the last iteration, and then early halts the BlockSTM
+                    // due to a txn execution abort. In this case, we cannot reset the scheduler_task of the
+                    // committing thread (to be Done), otherwise some other pending thread waiting for the execution
+                    // will be pending on read forever (since the halt logic let the execution task to wake up such
+                    // pending task).
                 }
             }
             scheduler_task = match scheduler_task {
@@ -248,7 +291,7 @@ where
                 SchedulerTask::ExecutionTask(_, Some(condvar)) => {
                     let (lock, cvar) = &*condvar;
                     // Mark dependency resolved.
-                    *lock.lock() = true;
+                    *lock.lock() = DependencyStatus::Resolved;
                     // Wake up the process waiting for dependency.
                     cvar.notify_one();
 
@@ -358,6 +401,7 @@ where
         let mut data_map = BTreeMap::new();
 
         let mut ret = Vec::with_capacity(num_txns);
+        let mut accumulated_gas = 0;
         for (idx, txn) in signature_verified_block.iter().enumerate() {
             let res = executor.execute_transaction(
                 &LatestView::<T, S>::new_btree_view(base_view, &data_map, idx as TxnIndex),
@@ -367,7 +411,7 @@ where
             );
 
             let must_skip = matches!(res, ExecutionStatus::SkipRest(_));
-
+            let txn_gas;
             match res {
                 ExecutionStatus::Success(output) | ExecutionStatus::SkipRest(output) => {
                     assert_eq!(
@@ -379,6 +423,7 @@ where
                     for (ap, write_op) in output.get_writes().into_iter() {
                         data_map.insert(ap, write_op);
                     }
+                    txn_gas = output.gas_used();
                     ret.push(output);
                 },
                 ExecutionStatus::Abort(err) => {
@@ -390,6 +435,14 @@ where
             if must_skip {
                 break;
             }
+
+            // Calculating the accumulated gas of the committed txns.
+            accumulated_gas += txn_gas;
+
+            // When the accumulated gas of the committed txns exceeds PER_BLOCK_GAS_LIMIT, halt sequential execution.
+            if accumulated_gas >= PER_BLOCK_GAS_LIMIT {
+                break;
+            }
         }
 
         ret.resize_with(num_txns, E::Output::skip_output);

aptos-move/block-executor/src/proptest_types/types.rs

@@ -449,7 +449,10 @@ where
                 let mut reads_result = vec![];
                 for k in reads[read_idx].iter() {
                     // TODO: later test errors as well? (by fixing state_view behavior).
-                    reads_result.push(view.get_state_value_bytes(k).unwrap());
+                    match view.get_state_value_bytes(k) {
+                        Ok(v) => reads_result.push(v),
+                        Err(_) => reads_result.push(None),
+                    }
                 }
                 ExecutionStatus::Success(Output(
                     writes_and_deltas[write_idx].0.clone(),
@@ -484,6 +487,10 @@ where
     fn skip_output() -> Self {
         Self(vec![], vec![], vec![])
     }
+
+    fn gas_used(&self) -> u64 {
+        0
+    }
 }
 
 ///////////////////////////////////////////////////////////////////////////