prototype integration in gas meter

aptos-move/aptos-gas-meter/src/algebra.rs

@@ -5,8 +5,9 @@ use crate::traits::GasAlgebra;
 use aptos_gas_algebra::{Fee, FeePerGasUnit, Gas, GasExpression, NumBytes, NumModules, Octa};
 use aptos_gas_schedule::{gas_feature_versions, VMGasParameters};
 use aptos_logger::error;
-use aptos_vm_types::storage::{
-    io_pricing::IoPricing, space_pricing::DiskSpacePricing, StorageGasParameters,
+use aptos_vm_types::{
+    resolver::BlockSynchronizationView,
+    storage::{io_pricing::IoPricing, space_pricing::DiskSpacePricing, StorageGasParameters},
 };
 use move_binary_format::errors::{PartialVMError, PartialVMResult};
 use move_core_types::{
@@ -18,7 +19,7 @@ use std::fmt::Debug;
 /// Base gas algebra implementation that tracks the gas usage using its internal counters.
 ///
 /// Abstract gas amounts are always evaluated to concrete values at the spot.
-pub struct StandardGasAlgebra {
+pub struct StandardGasAlgebra<'a> {
     feature_version: u64,
     vm_gas_params: VMGasParameters,
     storage_gas_params: StorageGasParameters,
@@ -40,15 +41,18 @@ pub struct StandardGasAlgebra {
 
     num_dependencies: NumModules,
     total_dependency_size: NumBytes,
+
+    maybe_block_synchronization_view: Option<&'a dyn BlockSynchronizationView>,
 }
 
-impl StandardGasAlgebra {
+impl<'a> StandardGasAlgebra<'a> {
     pub fn new(
         gas_feature_version: u64,
         vm_gas_params: VMGasParameters,
         storage_gas_params: StorageGasParameters,
         is_approved_gov_script: bool,
         balance: impl Into<Gas>,
+        maybe_block_synchronization_view: Option<&'a dyn BlockSynchronizationView>,
     ) -> Self {
         let balance = balance.into().to_unit_with_params(&vm_gas_params.txn);
 
@@ -83,11 +87,12 @@ impl StandardGasAlgebra {
             storage_fee_used: 0.into(),
             num_dependencies: 0.into(),
             total_dependency_size: 0.into(),
+            maybe_block_synchronization_view,
         }
     }
 }
 
-impl StandardGasAlgebra {
+impl StandardGasAlgebra<'_> {
     fn charge(&mut self, amount: InternalGas) -> (InternalGas, PartialVMResult<()>) {
         match self.balance.checked_sub(amount) {
             Some(new_balance) => {
@@ -106,7 +111,7 @@ impl StandardGasAlgebra {
     }
 }
 
-impl GasAlgebra for StandardGasAlgebra {
+impl GasAlgebra for StandardGasAlgebra<'_> {
     fn feature_version(&self) -> u64 {
         self.feature_version
     }
@@ -165,6 +170,16 @@ impl GasAlgebra for StandardGasAlgebra {
         &mut self,
         abstract_amount: impl GasExpression<VMGasParameters, Unit = InternalGasUnit> + Debug,
     ) -> PartialVMResult<()> {
+        if self
+            .maybe_block_synchronization_view
+            .is_some_and(|view| view.early_abort_execution())
+        {
+            return Err(
+                PartialVMError::new(StatusCode::SPECULATIVE_EXECUTION_ABORT_ERROR)
+                    .with_message("Interrupted from block synchronization view".to_string()),
+            );
+        }
+
         let amount = abstract_amount.evaluate(self.feature_version, &self.vm_gas_params);
 
         let (actual, res) = self.charge(amount);
@@ -187,6 +202,16 @@ impl GasAlgebra for StandardGasAlgebra {
         &mut self,
         abstract_amount: impl GasExpression<VMGasParameters, Unit = InternalGasUnit>,
     ) -> PartialVMResult<()> {
+        if self
+            .maybe_block_synchronization_view
+            .is_some_and(|view| view.early_abort_execution())
+        {
+            return Err(
+                PartialVMError::new(StatusCode::SPECULATIVE_EXECUTION_ABORT_ERROR)
+                    .with_message("Interrupted from block synchronization view".to_string()),
+            );
+        }
+
         let amount = abstract_amount.evaluate(self.feature_version, &self.vm_gas_params);
 
         let (actual, res) = self.charge(amount);

aptos-move/aptos-vm-types/src/resolver.rs

@@ -19,6 +19,15 @@ use move_core_types::{language_storage::StructTag, value::MoveTypeLayout, vm_sta
 use move_vm_types::delayed_values::delayed_field_id::DelayedFieldID;
 use std::collections::{BTreeMap, HashMap};
 
+/// Gives a global (from an executing txn perspective) view into the synchronization aspects of the
+/// encompassing block execution. The goal is to improve the efficiency of overall block execution.
+/// For example, if the block execution has halted due to gas limit (committing only a prefix),
+/// ongoing speculative executions can return early. Similarly, if a read-write dependency is
+/// preficted between txns, it can allow the dependent txn to wait and avoid costly re-executions.
+pub trait BlockSynchronizationView {
+    fn early_abort_execution(&self) -> bool;
+}
+
 /// Allows to query resources from the state.
 pub trait TResourceView {
     type Key;
@@ -204,7 +213,8 @@ pub trait StateStorageView {
 /// resolve AggregatorV2 via the state-view based default implementation, as it
 /// doesn't provide a value exchange functionality).
 pub trait TExecutorView<K, T, L, I, V>:
-    TResourceView<Key = K, Layout = L>
+    BlockSynchronizationView
+    + TResourceView<Key = K, Layout = L>
     + TModuleView<Key = K>
     + TAggregatorV1View<Identifier = K>
     + TDelayedFieldView<Identifier = I, ResourceKey = K, ResourceGroupTag = T>
@@ -213,7 +223,8 @@ pub trait TExecutorView<K, T, L, I, V>:
 }
 
 impl<A, K, T, L, I, V> TExecutorView<K, T, L, I, V> for A where
-    A: TResourceView<Key = K, Layout = L>
+    A: BlockSynchronizationView
+        + TResourceView<Key = K, Layout = L>
         + TModuleView<Key = K>
         + TAggregatorV1View<Identifier = K>
         + TDelayedFieldView<Identifier = I, ResourceKey = K, ResourceGroupTag = T>
@@ -299,6 +310,15 @@ where
     }
 }
 
+impl<S> BlockSynchronizationView for S
+where
+    S: StateView,
+{
+    fn early_abort_execution(&self) -> bool {
+        false
+    }
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum ResourceGroupSize {
     Concrete(u64),

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

@@ -89,7 +89,7 @@ use aptos_vm_types::{
     },
     module_write_set::ModuleWriteSet,
     output::VMOutput,
-    resolver::{ExecutorView, ResourceGroupView},
+    resolver::{BlockSynchronizationView, ExecutorView, ResourceGroupView},
     storage::{change_set_configs::ChangeSetConfigs, StorageGasParameters},
 };
 use ark_bn254::Bn254;
@@ -2033,17 +2033,24 @@ impl AptosVM {
 
     /// Main entrypoint for executing a user transaction that also allows the customization of the
     /// gas meter to be used.
-    pub fn execute_user_transaction_with_custom_gas_meter<G, F>(
+    pub fn execute_user_transaction_with_custom_gas_meter<'a, G, F>(
         &self,
-        resolver: &impl AptosMoveResolver,
+        resolver: &'a impl AptosMoveResolver,
         code_storage: &impl AptosCodeStorage,
         txn: &SignedTransaction,
         log_context: &AdapterLogSchema,
         make_gas_meter: F,
     ) -> Result<(VMStatus, VMOutput, G), VMStatus>
     where
         G: AptosGasMeter,
-        F: FnOnce(u64, VMGasParameters, StorageGasParameters, bool, Gas) -> G,
+        F: FnOnce(
+            u64,
+            VMGasParameters,
+            StorageGasParameters,
+            bool,
+            Gas,
+            Option<&'a dyn BlockSynchronizationView>,
+        ) -> G,
     {
         let txn_metadata = TransactionMetadata::new(txn);
 
@@ -2056,6 +2063,7 @@ impl AptosVM {
             self.storage_gas_params(log_context)?.clone(),
             is_approved_gov_script,
             balance,
+            Some(resolver.as_block_synchronization_view()),
         );
         let (status, output) = self.execute_user_transaction_impl(
             resolver,
@@ -2075,16 +2083,16 @@ impl AptosVM {
     ///
     /// This can be useful for off-chain applications that wants to perform additional
     /// measurements or analysis while preserving the production gas behavior.
-    pub fn execute_user_transaction_with_modified_gas_meter<G, F>(
+    pub fn execute_user_transaction_with_modified_gas_meter<'a, G, F>(
         &self,
-        resolver: &impl AptosMoveResolver,
+        resolver: &'a impl AptosMoveResolver,
         code_storage: &impl AptosCodeStorage,
         txn: &SignedTransaction,
         log_context: &AdapterLogSchema,
         modify_gas_meter: F,
     ) -> Result<(VMStatus, VMOutput, G), VMStatus>
     where
-        F: FnOnce(ProdGasMeter) -> G,
+        F: FnOnce(ProdGasMeter<'a>) -> G,
         G: AptosGasMeter,
     {
         self.execute_user_transaction_with_custom_gas_meter(
@@ -2096,13 +2104,15 @@ impl AptosVM {
              vm_gas_params,
              storage_gas_params,
              is_approved_gov_script,
-             meter_balance| {
+             meter_balance,
+             _maybe_block_synchronization_view| {
                 modify_gas_meter(make_prod_gas_meter(
                     gas_feature_version,
                     vm_gas_params,
                     storage_gas_params,
                     is_approved_gov_script,
                     meter_balance,
+                    None, // No block synchronization view
                 ))
             },
         )
@@ -2473,6 +2483,7 @@ impl AptosVM {
             storage_gas_params,
             /* is_approved_gov_script */ false,
             max_gas_amount.into(),
+            None, // No block synchronization view
         );
 
         let resolver = state_view.as_move_resolver();

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

@@ -29,7 +29,8 @@ use aptos_vm_environment::{
 };
 use aptos_vm_types::{
     resolver::{
-        ExecutorView, ResourceGroupSize, ResourceGroupView, StateStorageView, TResourceGroupView,
+        BlockSynchronizationView, ExecutorView, ResourceGroupSize, ResourceGroupView,
+        StateStorageView, TResourceGroupView,
     },
     resource_group_adapter::ResourceGroupAdapter,
 };
@@ -332,6 +333,10 @@ impl<'e, E: ExecutorView> AsExecutorView for StorageAdapter<'e, E> {
     fn as_executor_view(&self) -> &dyn ExecutorView {
         self.executor_view
     }
+
+    fn as_block_synchronization_view(&self) -> &dyn BlockSynchronizationView {
+        self.executor_view
+    }
 }
 
 // Allows to extract the view from `StorageAdapter`.

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

@@ -14,6 +14,7 @@ use aptos_types::on_chain_config::Features;
 use aptos_vm_logging::{log_schema::AdapterLogSchema, speculative_log, speculative_warn};
 use aptos_vm_types::{
     module_and_script_storage::module_storage::AptosModuleStorage,
+    resolver::BlockSynchronizationView,
     storage::{space_pricing::DiskSpacePricing, StorageGasParameters},
 };
 use move_core_types::vm_status::{StatusCode, VMStatus};
@@ -22,7 +23,7 @@ use move_core_types::vm_status::{StatusCode, VMStatus};
 const MAXIMUM_APPROVED_TRANSACTION_SIZE_LEGACY: u64 = 1024 * 1024;
 
 /// Gas meter used in the production (validator) setup.
-pub type ProdGasMeter = MemoryTrackedGasMeter<StandardGasMeter<StandardGasAlgebra>>;
+pub type ProdGasMeter<'a> = MemoryTrackedGasMeter<StandardGasMeter<StandardGasAlgebra<'a>>>;
 
 /// Creates a gas meter intended for executing transactions in the production.
 ///
@@ -33,13 +34,15 @@ pub fn make_prod_gas_meter(
     storage_gas_params: StorageGasParameters,
     is_approved_gov_script: bool,
     meter_balance: Gas,
+    maybe_block_synchronization_view: Option<&dyn BlockSynchronizationView>,
 ) -> ProdGasMeter {
     MemoryTrackedGasMeter::new(StandardGasMeter::new(StandardGasAlgebra::new(
         gas_feature_version,
         vm_gas_params,
         storage_gas_params,
         is_approved_gov_script,
         meter_balance,
+        maybe_block_synchronization_view,
     )))
 }
 

aptos-move/aptos-vm/src/move_vm_ext/resolver.rs

@@ -5,7 +5,7 @@ use aptos_aggregator::resolver::{AggregatorV1Resolver, DelayedFieldResolver};
 use aptos_table_natives::TableResolver;
 use aptos_types::{on_chain_config::ConfigStorage, state_store::state_key::StateKey};
 use aptos_vm_types::resolver::{
-    ExecutorView, ResourceGroupSize, ResourceGroupView, StateStorageView,
+    BlockSynchronizationView, ExecutorView, ResourceGroupSize, ResourceGroupView, StateStorageView,
 };
 use bytes::Bytes;
 use move_binary_format::errors::PartialVMResult;
@@ -51,6 +51,9 @@ pub trait ResourceGroupResolver {
 
 pub trait AsExecutorView {
     fn as_executor_view(&self) -> &dyn ExecutorView;
+
+    // TODO: remove once trait upcasting coercion is stabilized (rust issue #65991).
+    fn as_block_synchronization_view(&self) -> &dyn BlockSynchronizationView;
 }
 
 pub trait AsResourceGroupView {

aptos-move/aptos-vm/src/move_vm_ext/session/view_with_change_set.rs

@@ -23,8 +23,8 @@ use aptos_vm_types::{
     abstract_write_op::{AbstractResourceWriteOp, WriteWithDelayedFieldsOp},
     change_set::{randomly_check_layout_matches, VMChangeSet},
     resolver::{
-        ExecutorView, ResourceGroupSize, ResourceGroupView, StateStorageView, TModuleView,
-        TResourceGroupView, TResourceView,
+        BlockSynchronizationView, ExecutorView, ResourceGroupSize, ResourceGroupView,
+        StateStorageView, TModuleView, TResourceGroupView, TResourceView,
     },
 };
 use bytes::Bytes;
@@ -176,6 +176,12 @@ impl<'r> TDelayedFieldView for ExecutorViewWithChangeSet<'r> {
     }
 }
 
+impl<'r> BlockSynchronizationView for ExecutorViewWithChangeSet<'r> {
+    fn early_abort_execution(&self) -> bool {
+        self.base_executor_view.early_abort_execution()
+    }
+}
+
 impl<'r> TResourceView for ExecutorViewWithChangeSet<'r> {
     type Key = StateKey;
     type Layout = MoveTypeLayout;

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

@@ -94,6 +94,7 @@ impl AptosVM {
             storage_gas_params,
             false,
             gas_meter_balance.into(),
+            None,
         );
 
         let change_set_configs = &self

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

@@ -652,6 +652,10 @@ impl Scheduler {
 
         !self.has_halted.swap(true, Ordering::SeqCst)
     }
+
+    pub(crate) fn has_halted(&self) -> bool {
+        self.has_halted.load(Ordering::Relaxed)
+    }
 }
 
 impl TWaitForDependency for Scheduler {

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

@@ -48,7 +48,8 @@ use aptos_types::{
 };
 use aptos_vm_logging::{log_schema::AdapterLogSchema, prelude::*};
 use aptos_vm_types::resolver::{
-    ResourceGroupSize, StateStorageView, TModuleView, TResourceGroupView, TResourceView,
+    BlockSynchronizationView, ResourceGroupSize, StateStorageView, TModuleView, TResourceGroupView,
+    TResourceView,
 };
 use bytes::Bytes;
 use claims::assert_ok;
@@ -1433,6 +1434,17 @@ impl<'a, T: Transaction, S: TStateView<Key = T::Key>, X: Executable> LatestView<
     }
 }
 
+impl<'a, T: Transaction, S: TStateView<Key = T::Key>, X: Executable> BlockSynchronizationView
+    for LatestView<'a, T, S, X>
+{
+    fn early_abort_execution(&self) -> bool {
+        match &self.latest_view {
+            ViewState::Sync(state) => state.scheduler.has_halted(),
+            ViewState::Unsync(_) => false,
+        }
+    }
+}
+
 impl<'a, T: Transaction, S: TStateView<Key = T::Key>, X: Executable> TResourceView
     for LatestView<'a, T, S, X>
 {

aptos-move/e2e-tests/src/executor.rs

@@ -1061,6 +1061,7 @@ impl FakeExecutor {
                         env.storage_gas_params().as_ref().unwrap().clone(),
                         false,
                         1_000_000_000_000_000.into(),
+                        None,
                     )),
                     None,
                 ),
@@ -1171,6 +1172,7 @@ impl FakeExecutor {
                         env.storage_gas_params().as_ref().unwrap().clone(),
                         false,
                         10_000_000_000_000,
+                        None,
                     ),
                     shared_buffer: Arc::clone(&a1),
                 }),