Compute correct accumulated effective block gas fees with module publ…

…ishing txn(s) in the block (#11582)

* Compute correct accumulated effective block gas fees with module publishing txn(s) in the block

Start computing the accumulated block gas fee normally until you find a
module publishing txn. Then recompute the accumulated block gas fee by setting
conflict_multiplier == conflict_penalty_window for all preceeding txns, and
continue accumulating with the same conflict_multiplier for subsequent txns in the block
subsequent txns as well

* Address review comments

* Set 'use_module_publishing_block_conflict' flag and rebase with main

* Address review comments; modify 'append_and_check_module_rw_conflict()' logic

* Set 'use_module_publishing_block_conflict' in default_for_genesis() to true

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

@@ -1229,6 +1229,9 @@ where
             num_txns,
         );
 
+        let last_input_output: TxnLastInputOutput<T, E::Output, E::Error> =
+            TxnLastInputOutput::new(num_txns as TxnIndex);
+
         for (idx, txn) in signature_verified_block.iter().enumerate() {
             let latest_view = LatestView::<T, S, X>::new(
                 base_view,
@@ -1241,7 +1244,6 @@ where
                 idx as TxnIndex,
             );
             let res = executor.execute_transaction(&latest_view, txn, idx as TxnIndex);
-
             let must_skip = matches!(res, ExecutionStatus::SkipRest(_));
             match res {
                 ExecutionStatus::Success(output) | ExecutionStatus::SkipRest(output) => {
@@ -1267,17 +1269,26 @@ where
                                 } as u64
                         });
 
+                    let sequential_reads = latest_view.take_sequential_reads();
                     let read_write_summary = self
                         .config
                         .onchain
                         .block_gas_limit_type
                         .conflict_penalty_window()
                         .map(|_| {
                             ReadWriteSummary::new(
-                                latest_view.take_sequential_reads().get_read_summary(),
+                                sequential_reads.get_read_summary(),
                                 output.get_write_summary(),
                             )
                         });
+
+                    if last_input_output.check_and_append_module_rw_conflict(
+                        sequential_reads.module_reads.iter(),
+                        output.module_write_set().keys(),
+                    ) {
+                        block_limit_processor.process_module_rw_conflict();
+                    }
+
                     block_limit_processor.accumulate_fee_statement(
                         fee_statement,
                         read_write_summary,
@@ -1417,7 +1428,7 @@ where
                         msg,
                     )));
                 },
-            }
+            };
             // When the txn is a SkipRest txn, halt sequential execution.
             if must_skip {
                 break;

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

@@ -17,6 +17,7 @@ pub struct BlockGasLimitProcessor<T: Transaction> {
     txn_fee_statements: Vec<FeeStatement>,
     txn_read_write_summaries: Vec<ReadWriteSummary<T>>,
     block_limit_reached: bool,
+    module_rw_conflict: bool,
 }
 
 impl<T: Transaction> BlockGasLimitProcessor<T> {
@@ -29,6 +30,7 @@ impl<T: Transaction> BlockGasLimitProcessor<T> {
             txn_fee_statements: Vec::with_capacity(init_size),
             txn_read_write_summaries: Vec::with_capacity(init_size),
             block_limit_reached: false,
+            module_rw_conflict: false,
         }
     }
 
@@ -58,7 +60,11 @@ impl<T: Transaction> BlockGasLimitProcessor<T> {
                     txn_read_write_summary.collapse_resource_group_conflicts()
                 },
             );
-            self.compute_conflict_multiplier(conflict_overlap_length as usize)
+            if self.module_rw_conflict {
+                conflict_overlap_length as u64
+            } else {
+                self.compute_conflict_multiplier(conflict_overlap_length as usize)
+            }
         } else {
             assert_none!(txn_read_write_summary);
             1
@@ -83,6 +89,33 @@ impl<T: Transaction> BlockGasLimitProcessor<T> {
         }
     }
 
+    pub(crate) fn process_module_rw_conflict(&mut self) {
+        if self.module_rw_conflict
+            || !self
+                .block_gas_limit_type
+                .use_module_publishing_block_conflict()
+        {
+            return;
+        }
+
+        let conflict_multiplier = if let Some(conflict_overlap_length) =
+            self.block_gas_limit_type.conflict_penalty_window()
+        {
+            conflict_overlap_length
+        } else {
+            return;
+        };
+
+        self.accumulated_effective_block_gas = conflict_multiplier as u64
+            * (self.accumulated_fee_statement.execution_gas_used()
+                * self
+                    .block_gas_limit_type
+                    .execution_gas_effective_multiplier()
+                + self.accumulated_fee_statement.io_gas_used()
+                    * self.block_gas_limit_type.io_gas_effective_multiplier());
+        self.module_rw_conflict = true;
+    }
+
     fn should_end_block(&mut self, mode: &str) -> bool {
         if let Some(per_block_gas_limit) = self.block_gas_limit_type.block_gas_limit() {
             // When the accumulated block gas of the committed txns exceeds
@@ -427,4 +460,59 @@ mod test {
         assert_eq!(processor.accumulated_effective_block_gas, 20);
         assert!(!processor.should_end_block_parallel());
     }
+
+    #[test]
+    fn test_module_publishing_txn_conflict() {
+        let conflict_penalty_window = 4;
+        let block_gas_limit = BlockGasLimitType::ComplexLimitV1 {
+            effective_block_gas_limit: 1000,
+            execution_gas_effective_multiplier: 1,
+            io_gas_effective_multiplier: 1,
+            conflict_penalty_window,
+            use_module_publishing_block_conflict: true,
+            block_output_limit: None,
+            include_user_txn_size_in_block_output: true,
+            add_block_limit_outcome_onchain: false,
+            use_granular_resource_group_conflicts: true,
+        };
+
+        let mut processor = BlockGasLimitProcessor::<TestTxn>::new(block_gas_limit, 10);
+        processor.accumulate_fee_statement(
+            execution_fee(10),
+            Some(ReadWriteSummary::new(
+                to_map(&[InputOutputKey::Group(2, 2)]),
+                to_map(&[InputOutputKey::Group(2, 2)]),
+            )),
+            None,
+        );
+        processor.accumulate_fee_statement(
+            execution_fee(20),
+            Some(ReadWriteSummary::new(
+                to_map(&[InputOutputKey::Group(1, 1)]),
+                to_map(&[InputOutputKey::Group(1, 1)]),
+            )),
+            None,
+        );
+        assert_eq!(1, processor.compute_conflict_multiplier(8));
+        assert_eq!(processor.accumulated_effective_block_gas, 30);
+
+        processor.process_module_rw_conflict();
+        assert_eq!(
+            processor.accumulated_effective_block_gas,
+            30 * conflict_penalty_window as u64
+        );
+
+        processor.accumulate_fee_statement(
+            execution_fee(25),
+            Some(ReadWriteSummary::new(
+                to_map(&[InputOutputKey::Group(1, 1)]),
+                to_map(&[InputOutputKey::Group(1, 1)]),
+            )),
+            None,
+        );
+        assert_eq!(
+            processor.accumulated_effective_block_gas,
+            55 * conflict_penalty_window as u64
+        );
+    }
 }

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

@@ -138,21 +138,10 @@ impl<T: Transaction, O: TransactionOutput<Txn = T>, E: Debug + Send + Clone>
             | ExecutionStatus::DelayedFieldsCodeInvariantError(_) => BTreeMap::new(),
         };
 
-        if !self.module_read_write_intersection.load(Ordering::Relaxed) {
-            // Check if adding new read & write modules leads to intersections.
-            if Self::append_and_check(
-                input.module_reads.iter(),
-                &self.module_reads,
-                &self.module_writes,
-            ) || Self::append_and_check(
-                written_modules.keys(),
-                &self.module_writes,
-                &self.module_reads,
-            ) {
-                self.module_read_write_intersection
-                    .store(true, Ordering::Release);
-                return false;
-            }
+        if self
+            .check_and_append_module_rw_conflict(input.module_reads.iter(), written_modules.keys())
+        {
+            return false;
         }
 
         self.inputs[txn_idx as usize].store(Some(Arc::new(input)));
@@ -161,6 +150,26 @@ impl<T: Transaction, O: TransactionOutput<Txn = T>, E: Debug + Send + Clone>
         true
     }
 
+    pub(crate) fn check_and_append_module_rw_conflict<'a>(
+        &self,
+        module_reads_keys: impl Iterator<Item = &'a T::Key>,
+        module_writes_keys: impl Iterator<Item = &'a T::Key>,
+    ) -> bool {
+        if self.module_read_write_intersection.load(Ordering::Relaxed) {
+            return true;
+        }
+
+        // Check if adding new read & write modules leads to intersections.
+        if Self::append_and_check(module_reads_keys, &self.module_reads, &self.module_writes)
+            || Self::append_and_check(module_writes_keys, &self.module_writes, &self.module_reads)
+        {
+            self.module_read_write_intersection
+                .store(true, Ordering::Release);
+            return true;
+        }
+        false
+    }
+
     pub(crate) fn read_set(&self, txn_idx: TxnIndex) -> Option<Arc<CapturedReads<T>>> {
         self.inputs[txn_idx as usize].load_full()
     }

execution/executor-benchmark/src/lib.rs

@@ -596,9 +596,11 @@ impl OverallMeasuring {
         );
         info!("{} GPS: {} gas/s", prefix, delta_gas.gas / elapsed);
         info!(
-            "{} effectiveGPS: {} gas/s",
+            "{} effectiveGPS: {} gas/s ({} effective block gas, in {} s)",
             prefix,
-            delta_gas.effective_block_gas / elapsed
+            delta_gas.effective_block_gas / elapsed,
+            delta_gas.effective_block_gas,
+            elapsed
         );
         info!("{} ioGPS: {} gas/s", prefix, delta_gas.io_gas / elapsed);
         info!(

types/src/block_executor/config.rs

@@ -40,7 +40,7 @@ impl BlockExecutorConfigFromOnchain {
                     io_gas_effective_multiplier: 1,
                     block_output_limit: Some(1_000_000_000_000),
                     conflict_penalty_window: 8,
-                    use_module_publishing_block_conflict: false,
+                    use_module_publishing_block_conflict: true,
                     include_user_txn_size_in_block_output: true,
                     add_block_limit_outcome_onchain: false,
                     use_granular_resource_group_conflicts: false,

types/src/on_chain_config/execution_config.rs

@@ -77,7 +77,7 @@ impl OnChainExecutionConfig {
                 io_gas_effective_multiplier: 1,
                 conflict_penalty_window: 6,
                 use_granular_resource_group_conflicts: false,
-                use_module_publishing_block_conflict: false,
+                use_module_publishing_block_conflict: true,
                 block_output_limit: Some(3 * 1024 * 1024),
                 include_user_txn_size_in_block_output: true,
                 add_block_limit_outcome_onchain: false,
@@ -255,6 +255,17 @@ impl BlockGasLimitType {
         }
     }
 
+    pub fn use_module_publishing_block_conflict(&self) -> bool {
+        match self {
+            BlockGasLimitType::NoLimit => false,
+            BlockGasLimitType::Limit(_) => false,
+            BlockGasLimitType::ComplexLimitV1 {
+                use_module_publishing_block_conflict,
+                ..
+            } => *use_module_publishing_block_conflict,
+        }
+    }
+
     pub fn include_user_txn_size_in_block_output(&self) -> bool {
         match self {
             BlockGasLimitType::NoLimit => false,