test_sstore_bloated()

Documentation for tests/benchmark/stateful/bloatnet/test_single_opcode.py::test_sstore_bloated@b314d18e.

Generate fixtures for these test cases for Amsterdam with:

fill -v tests/benchmark/stateful/bloatnet/test_single_opcode.py::test_sstore_bloated --gas-benchmark-values 1

Benchmark SSTORE opcodes targeting an EOA with storage bloated.

Source code in tests/benchmark/stateful/bloatnet/test_single_opcode.py

@pytest.mark.repricing
@pytest.mark.stub_parametrize("token_name", "bloated_eoa_")
@pytest.mark.parametrize("write_new_value", [False, True])
@pytest.mark.parametrize("existing_slots", [True, False])
@pytest.mark.parametrize("cache_strategy", [CacheStrategy.NO_CACHE])
def test_sstore_bloated(
    benchmark_test: BenchmarkTestFiller,
    pre: Alloc,
    fork: Fork,
    gas_benchmark_value: int,
    tx_gas_limit: int,
    token_name: str,
    write_new_value: bool,
    existing_slots: bool,
    cache_strategy: CacheStrategy,
) -> None:
    """
    Benchmark SSTORE opcodes targeting an EOA with storage bloated.
    """
    sstore_metadata: dict[str, Any] = {}
    # If CACHE_TX, there would be one cold SLOAD before SSTORE
    sstore_metadata["key_warm"] = cache_strategy == CacheStrategy.CACHE_TX

    # SSTORE metadata matrix:
    #
    # existing_slots | write_new_value | original | current | new
    # ---------------+-----------------+----------+---------+-----
    # True           | True            | 1        | 1       | 2
    # True           | False           | 1        | 1       | 1
    # False          | True            | 0        | 0       | 1
    # False          | False           | 0        | 0       | 0

    initial_value = int(existing_slots)

    # When existing_slots is False, the initial value is always 0
    # Otherwise, the initial value starts at 1 instead.
    sstore_metadata["original_value"] = initial_value
    sstore_metadata["current_value"] = initial_value

    # If not writing a new value, the new value is the same as the current one
    # If writing a new value, the new value is current value + 1
    sstore_metadata["new_value"] = (
        initial_value if not write_new_value else initial_value + 1
    )

    setup = (
        Op.CALLDATALOAD(32)  # [end_slot]
        + Op.CALLDATALOAD(0)  # [counter, end_slot]
    )

    # stack element: [counter, end_slot]

    loop = Bytecode()
    loop += Op.JUMPDEST  # jump target

    # If CACHE_TX, warm the slot with a cold SLOAD before the SSTORE loop
    if cache_strategy == CacheStrategy.CACHE_TX:
        loop += Op.POP(Op.SLOAD(Op.DUP1, key_warm=False))

    sstore_op: Bytecode = Bytecode()
    if write_new_value:
        # s[counter] = counter + 1
        sstore_op = (
            Op.DUP1  # [counter, counter, end_slot]
            + Op.DUP1  # [counter, counter, counter, end_slot]
            + Op.PUSH1(1)  # [1, counter, counter, counter, end_slot]
            + Op.ADD  # [counter+1, counter, counter, end_slot]
            + Op.SWAP1  # [counter, counter+1, counter, end_slot]
            + Op.SSTORE(**sstore_metadata)  # [counter, end_slot]
        )
    else:
        # s[counter] = counter (existing slot) or 0 (non existing slot)
        push_value = Op.DUP1 if existing_slots else Op.PUSH1(0)
        sstore_op = (
            push_value  # [value, counter, end_slot]
            + Op.DUP2  # [counter, value, counter, end_slot]
            + Op.SSTORE(**sstore_metadata)  # [counter, end_slot]
        )

    loop += sstore_op

    # stack element: [counter, end_slot]

    loop += (
        Op.PUSH1(1)  # [1, counter, end_slot]
        + Op.ADD  # [counter+1, end_slot]
        + Op.DUP2  # [end_slot, counter+1, end_slot]
        + Op.DUP2  # [counter+1, end_slot, counter+1, end_slot]
        + Op.LT  # [counter+1<end_slot, counter+1, end_slot]
        + Op.PUSH1(len(setup))  # [dest, condition, counter+1, end_slot]
        + Op.JUMPI  # [counter+1, end_slot]
    )

    runtime_code = IteratingBytecode(
        setup=setup,
        iterating=loop,
        cleanup=Op.STOP,
        iterating_state_gas=loop.state_cost(fork),
    )

    authority = pre.stub_eoa(token_name)
    start_slot = 1 if existing_slots else START_SLOT

    def calldata_gen(iteration_count: int, start_iteration: int) -> bytes:
        return Hash(start_iteration) + Hash(start_iteration + iteration_count)

    def tx_generator(sender: EOA) -> list[Transaction]:
        return list(
            runtime_code.transactions_by_gas_limit(
                fork=fork,
                gas_limit=gas_benchmark_value,
                sender=sender,
                to=authority,
                start_iteration=start_slot,
                calldata=calldata_gen,
            )
        )

    run_bloated_eoa_benchmark(
        benchmark_test=benchmark_test,
        pre=pre,
        fork=fork,
        gas_benchmark_value=gas_benchmark_value,
        tx_gas_limit=tx_gas_limit,
        authority=authority,
        existing_slots=existing_slots,
        runtime_code=runtime_code,
        cache_strategy=cache_strategy,
        tx_generator=tx_generator,
    )

Parametrized Test Cases

This test generates 4 parametrized test cases across 3 forks.