Improved block gas limit - output size limit, conflict aware, io/compute multipliers (aptos-labs#10943)

throughput changes based on whether there are conflicts between transactions (and they need to be executed sequentially), or if they are parallel. But the gas charged stays constant, as gas charging is independent for each transaction.

In order to limit to the total amount of time block can be executing, we introduce conflict-aware block gas limit - where we multiply gas used for individual transactions with the coefficient of how many of the recent 8 transaction it had conflicts with (including self, which we assume it had conflict with - to have the number in the [1, 8] range)

Author: igor-aptos

Cargo.lock

@@ -1029,8 +1029,14 @@ impl Context {
                     } else if let Some(full_block_gas_used) =
                         block_config.block_gas_limit_type.block_gas_limit()
                     {
-                        prices_and_used.iter().map(|(_, used)| *used).sum::<u64>()
-                            >= full_block_gas_used
+                        // be pessimistic for conflicts, as such information is not onchain
+                        let gas_used = prices_and_used.iter().map(|(_, used)| *used).sum::<u64>();
+                        let max_conflict_multiplier = block_config
+                            .block_gas_limit_type
+                            .conflict_penalty_window()
+                            .unwrap_or(1)
+                            as u64;
+                        gas_used * max_conflict_multiplier >= full_block_gas_used
                     } else {
                         false
                     };

api/src/context.rs

@@ -460,7 +460,6 @@ impl VMChangeSet {
             .collect::<anyhow::Result<BTreeMap<StateKey, WriteOp>, VMStatus>>()?;
         self.aggregator_v1_write_set
             .extend(materialized_aggregator_delta_set);
-
         Ok(())
     }
 

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

@@ -15,7 +15,7 @@ use aptos_aggregator::{
 use aptos_block_executor::{
     errors::Error, executor::BlockExecutor,
     task::TransactionOutput as BlockExecutorTransactionOutput,
-    txn_commit_hook::TransactionCommitHook,
+    txn_commit_hook::TransactionCommitHook, types::InputOutputKey,
 };
 use aptos_infallible::Mutex;
 use aptos_state_view::{StateView, StateViewId};
@@ -36,7 +36,10 @@ use aptos_vm_types::{abstract_write_op::AbstractResourceWriteOp, output::VMOutpu
 use move_core_types::{language_storage::StructTag, value::MoveTypeLayout, vm_status::VMStatus};
 use once_cell::sync::OnceCell;
 use rayon::ThreadPool;
-use std::{collections::BTreeMap, sync::Arc};
+use std::{
+    collections::{BTreeMap, HashSet},
+    sync::Arc,
+};
 
 /// Output type wrapper used by block executor. VM output is stored first, then
 /// transformed into TransactionOutput type that is returned.
@@ -316,6 +319,61 @@ impl BlockExecutorTransactionOutput for AptosTransactionOutput {
             .expect("Output to be set to get fee statement")
             .fee_statement()
     }
+
+    fn output_approx_size(&self) -> u64 {
+        let vm_output = self.vm_output.lock();
+        let change_set = vm_output
+            .as_ref()
+            .expect("Output to be set to get write summary")
+            .change_set();
+
+        let mut size = 0;
+        for (state_key, write_size) in change_set.write_set_size_iter() {
+            size += state_key.size() as u64 + write_size.write_len().unwrap_or(0);
+        }
+
+        for (event, _) in change_set.events() {
+            size += event.size() as u64;
+        }
+
+        size
+    }
+
+    fn get_write_summary(&self) -> HashSet<InputOutputKey<StateKey, StructTag, DelayedFieldID>> {
+        let vm_output = self.vm_output.lock();
+        let change_set = vm_output
+            .as_ref()
+            .expect("Output to be set to get write summary")
+            .change_set();
+
+        let mut writes = HashSet::new();
+
+        for (state_key, write) in change_set.resource_write_set() {
+            match write {
+                AbstractResourceWriteOp::Write(_)
+                | AbstractResourceWriteOp::WriteWithDelayedFields(_) => {
+                    writes.insert(InputOutputKey::Resource(state_key.clone()));
+                },
+                AbstractResourceWriteOp::WriteResourceGroup(write) => {
+                    for tag in write.inner_ops().keys() {
+                        writes.insert(InputOutputKey::Group(state_key.clone(), tag.clone()));
+                    }
+                },
+                AbstractResourceWriteOp::InPlaceDelayedFieldChange(_)
+                | AbstractResourceWriteOp::ResourceGroupInPlaceDelayedFieldChange(_) => {
+                    // No conflicts on resources from in-place delayed field changes.
+                    // Delayed fields conflicts themselves are handled via
+                    // delayed_field_change_set below.
+                },
+            }
+        }
+
+        for identifier in change_set.delayed_field_change_set().keys() {
+            writes.insert(InputOutputKey::DelayedField(*identifier));
+        }
+
+        writes
+    }
 }
 
 pub struct BlockAptosVM();

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

@@ -1,6 +1,7 @@
 // Copyright © Aptos Foundation
 // SPDX-License-Identifier: Apache-2.0
 
+use crate::types::InputOutputKey;
 use anyhow::bail;
 use aptos_aggregator::{
     delta_math::DeltaHistory,
@@ -335,8 +336,6 @@ impl<T: Transaction> CapturedReads<T> {
         &'a self,
         skip: &'a HashSet<T::Key>,
     ) -> impl Iterator<Item = (&T::Key, &GroupRead<T>)> {
-        // TODO[agg_v2](optimize) - We could potentially filter out inner_reads
-        // to only contain those that have Some(layout)
         self.group_reads.iter().filter(|(key, group_read)| {
             !skip.contains(key)
                 && group_read
@@ -662,6 +661,37 @@ impl<T: Transaction> CapturedReads<T> {
         Ok(true)
     }
 
+    pub(crate) fn get_read_summary(
+        &self,
+    ) -> HashSet<InputOutputKey<T::Key, T::Tag, T::Identifier>> {
+        let mut ret = HashSet::new();
+        for (key, read) in &self.data_reads {
+            if let DataRead::Versioned(_, _, _) = read {
+                ret.insert(InputOutputKey::Resource(key.clone()));
+            }
+        }
+
+        for (key, group_reads) in &self.group_reads {
+            for (tag, read) in &group_reads.inner_reads {
+                if let DataRead::Versioned(_, _, _) = read {
+                    ret.insert(InputOutputKey::Group(key.clone(), tag.clone()));
+                }
+            }
+        }
+
+        for key in &self.module_reads {
+            ret.insert(InputOutputKey::Resource(key.clone()));
+        }
+
+        for (key, read) in &self.delayed_field_reads {
+            if let DelayedFieldRead::Value { .. } = read {
+                ret.insert(InputOutputKey::DelayedField(*key));
+            }
+        }
+
+        ret
+    }
+
     pub(crate) fn mark_failure(&mut self) {
         self.speculative_failure = true;
     }
@@ -671,6 +701,42 @@ impl<T: Transaction> CapturedReads<T> {
     }
 }
 
+#[derive(Derivative)]
+#[derivative(Default(bound = "", new = "true"))]
+pub(crate) struct UnsyncReadSet<T: Transaction> {
+    pub(crate) resource_reads: HashSet<T::Key>,
+    pub(crate) module_reads: HashSet<T::Key>,
+    pub(crate) group_reads: HashMap<T::Key, HashSet<T::Tag>>,
+    pub(crate) delayed_field_reads: HashSet<T::Identifier>,
+}
+
+impl<T: Transaction> UnsyncReadSet<T> {
+    pub(crate) fn get_read_summary(
+        &self,
+    ) -> HashSet<InputOutputKey<T::Key, T::Tag, T::Identifier>> {
+        let mut ret = HashSet::new();
+        for key in &self.resource_reads {
+            ret.insert(InputOutputKey::Resource(key.clone()));
+        }
+
+        for (key, group_reads) in &self.group_reads {
+            for tag in group_reads {
+                ret.insert(InputOutputKey::Group(key.clone(), tag.clone()));
+            }
+        }
+
+        for key in &self.module_reads {
+            ret.insert(InputOutputKey::Resource(key.clone()));
+        }
+
+        for key in &self.delayed_field_reads {
+            ret.insert(InputOutputKey::DelayedField(*key));
+        }
+
+        ret
+    }
+}
+
 #[cfg(test)]
 mod test {
     use super::*;