test_state_root_computation()

Documentation for tests/benchmark/compute/eip7928_block_level_access_lists/test_block_access_list.py::test_state_root_computation@892e6d1e.

Generate fixtures for these test cases for Amsterdam with:

fill -v tests/benchmark/compute/eip7928_block_level_access_lists/test_block_access_list.py::test_state_root_computation --gas-benchmark-values 1

Benchmark state root computation with disjoint storage writes.

Deploy contracts with pre-populated storage. Each execution transaction writes to a non-overlapping range of sequential cold storage slots via a gas-check loop, so all transactions are genuinely independent.

The contract_per_tx parameter controls whether each transaction targets a unique contract (maximizing account trie width) or all transactions share a single contract (maximizing storage trie depth).

Source code in tests/benchmark/compute/eip7928_block_level_access_lists/test_block_access_list.py

@pytest.mark.parametrize(
    "contract_per_tx",
    [
        pytest.param(False, id="single_contract"),
        pytest.param(True, id="contract_per_tx"),
    ],
)
@pytest.mark.parametrize("tx_density", TX_DENSITY_PARAMS)
def test_state_root_computation(
    benchmark_test: BenchmarkTestFiller,
    pre: Alloc,
    fork: Fork,
    gas_benchmark_value: int,
    tx_gas_limit: int,
    tx_density: TxDensity,
    contract_per_tx: bool,
) -> None:
    """
    Benchmark state root computation with disjoint storage writes.

    Deploy contracts with pre-populated storage. Each execution
    transaction writes to a non-overlapping range of sequential cold
    storage slots via a gas-check loop, so all transactions are
    genuinely independent.

    The ``contract_per_tx`` parameter controls whether each
    transaction targets a unique contract (maximizing account trie
    width) or all transactions share a single contract (maximizing
    storage trie depth).
    """
    intrinsic_gas_calculator = fork.transaction_intrinsic_cost_calculator()
    # Worst-case calldata: 32 nonzero bytes for start_slot.
    intrinsic_gas = intrinsic_gas_calculator(calldata=b"\xff" * 32)

    # Reconstruct body bytecode to extract gas components;
    # _build_sequential_sstore_code only returns assembled code.
    setup = Op.MSTORE(
        0,
        Op.CALLDATALOAD(0),
        old_memory_size=0,
        new_memory_size=32,
    )

    sstore_body = Op.SSTORE(
        Op.MLOAD(0),
        2**256 - 1,
        key_warm=False,
        original_value=1,
        current_value=1,
        new_value=2**256 - 1,
    ) + Op.MSTORE(
        0,
        Op.SUB(Op.MLOAD(0), 1),
        old_memory_size=32,
        new_memory_size=32,
    )

    setup_gas = setup.gas_cost(fork)
    per_iter_gas, exit_overhead = _derive_loop_gas(sstore_body, fork)
    cleanup_gas = Op.STOP.gas_cost(fork)
    reserve_gas = per_iter_gas + exit_overhead + cleanup_gas

    runtime_code = _build_sequential_sstore_code(reserve_gas)
    min_per_tx_gas = intrinsic_gas + setup_gas + reserve_gas

    tx_gas_schedule = _derive_tx_schedule(
        gas_benchmark_value, min_per_tx_gas, tx_gas_limit, tx_density
    )
    num_exec_txs = len(tx_gas_schedule)

    available_gas = tx_gas_schedule[0] - intrinsic_gas - setup_gas
    estimated_slots_per_tx = max(1, available_gas // per_iter_gas)

    num_contracts = num_exec_txs if contract_per_tx else 1
    txs_per_contract = math.ceil(num_exec_txs / num_contracts)
    slots_per_contract = (estimated_slots_per_tx + 1) * txs_per_contract

    # Pre-populate storage counting down from near-max uint256.
    # High slot keys + 32-byte stored values maximize RLP weight
    # per trie leaf for state root computation.
    high_start = 2**256 - 1
    contracts = []
    for _ in range(num_contracts):
        storage: Storage.StorageDictType = {
            high_start - i: 1 for i in range(slots_per_contract)
        }
        addr = pre.deploy_contract(
            code=runtime_code,
            storage=storage,
        )
        contracts.append(addr)

    blocks: list[Block] = []
    contract_tx_counts = [0] * num_contracts

    with TestPhaseManager.execution():
        exec_txs = []
        for tx_idx in range(num_exec_txs):
            c_idx = tx_idx % num_contracts
            start_slot = (
                high_start - contract_tx_counts[c_idx] * estimated_slots_per_tx
            )
            contract_tx_counts[c_idx] += 1
            exec_txs.append(
                Transaction(
                    to=contracts[c_idx],
                    gas_limit=tx_gas_schedule[tx_idx],
                    data=Hash(start_slot),
                    sender=pre.fund_eoa(),
                )
            )
        blocks.append(Block(txs=exec_txs))

    benchmark_test(blocks=blocks, skip_gas_used_validation=True)

Parametrized Test Cases

This test generates 6 parametrized test cases across 1 fork.