test_create()¶

@pytest.mark.repricing(max_code_size_ratio=0)
@pytest.mark.parametrize(
    "opcode",
    [
        Op.CREATE,
        Op.CREATE2,
    ],
)
@pytest.mark.parametrize(
    "max_code_size_ratio, non_zero_data, value",
    [
        # To avoid a blowup of combinations, the value dimension is only
        # explored for the non-zero data case, so isn't affected by code size
        # influence.
        pytest.param(0, False, 0, id="0 bytes without value"),
        pytest.param(0, False, 1, id="0 bytes with value"),
        pytest.param(
            0.25, True, 0, id="0.25x max code size with non-zero data"
        ),
        pytest.param(0.25, False, 0, id="0.25x max code size with zero data"),
        pytest.param(
            0.50, True, 0, id="0.50x max code size with non-zero data"
        ),
        pytest.param(0.50, False, 0, id="0.50x max code size with zero data"),
        pytest.param(
            0.75, True, 0, id="0.75x max code size with non-zero data"
        ),
        pytest.param(0.75, False, 0, id="0.75x max code size with zero data"),
        pytest.param(1.00, True, 0, id="max code size with non-zero data"),
        pytest.param(1.00, False, 0, id="max code size with zero data"),
    ],
)
def test_create(
    benchmark_test: BenchmarkTestFiller,
    pre: Alloc,
    fork: Fork,
    opcode: Op,
    max_code_size_ratio: float,
    non_zero_data: bool,
    value: int,
    gas_benchmark_value: int,
    fixed_opcode_count: float | None,
) -> None:
    """Benchmark CREATE and CREATE2 instructions."""
    max_code_size = fork.max_code_size()

    code_size = int(max_code_size * max_code_size_ratio)

    copy = (
        Op.CODECOPY(
            dest_offset=0,
            offset=0,
            size=code_size,
            # gas accounting
            data_size=code_size,
            new_memory_size=code_size,
        )
        if non_zero_data
        else Bytecode()
    )

    initcode_body = copy + Op.RETURN(
        0,
        code_size,
        # gas accounting
        code_deposit_size=code_size,
        new_memory_size=0 if non_zero_data else code_size,
    )

    initcode = initcode_body

    if non_zero_data:  # Pad to code_size so CODECOPY has code_size bytes.
        initcode += bytes(
            [i % 256 for i in range(code_size - len(initcode_body))]
        )

    initcode_template_contract = pre.deploy_contract(code=initcode)

    # CALLDATA[0:32] = start index
    # CALLDATA[32:64] = end index
    setup = (
        Op.EXTCODECOPY(
            address=initcode_template_contract,
            dest_offset=0,
            offset=0,
            size=len(initcode),
            # gas accounting
            data_size=len(initcode),
            new_memory_size=len(initcode),
        )
        + Op.ADD(1, Op.CALLDATALOAD(32))  # [end+1 = limit]
        + Op.CALLDATALOAD(0)  # [start = index, limit]
    )

    # CREATE2 takes the loop index (stack top) as its salt;
    salt_kwarg: dict[str, Any] = {}
    if opcode == Op.CREATE2:
        salt_kwarg = {"salt": Op.DUP1}

    create_op = opcode(
        value=value,
        offset=0,
        size=len(initcode),
        init_code_size=len(initcode),
        **salt_kwarg,
    )

    loop = While(
        body=Op.POP(create_op),  # [index, limit]
        condition=Op.PUSH1(1)  # [1, index, limit]
        + Op.ADD  # [index+1, limit]
        + Op.DUP1  # [index+1, index+1, limit]
        + Op.DUP3  # [limit, index+1, index+1, limit]
        + Op.GT,  # [limit > index+1, index+1, limit]
    )

    code = IteratingBytecode(
        setup=setup,
        iterating=loop,
        iterating_subcall=initcode_body,
        cleanup=Op.STOP,
    )

    contract_address = pre.deploy_contract(
        code=code,
        balance=10**9 if value > 0 else 0,
    )

    def calldata(iteration_count: int, start_iteration: int) -> bytes:
        index_end = iteration_count + start_iteration - 1
        return Hash(start_iteration) + Hash(index_end)

    num_contracts = (
        sum(
            code.tx_iterations_by_gas_limit(
                fork=fork,
                gas_limit=gas_benchmark_value,
                calldata=calldata,
            )
        )
        if fixed_opcode_count is None
        else int(fixed_opcode_count * 1000)
    )

    if num_contracts == 0:
        pytest.skip(
            "Benchmark gas value cannot cover a single contract creation."
        )

    with TestPhaseManager.execution():
        sender = pre.fund_eoa()
        if fixed_opcode_count is not None:
            exec_txs = list(
                code.transactions_by_total_iteration_count(
                    fork=fork,
                    total_iterations=num_contracts,
                    sender=sender,
                    to=contract_address,
                    calldata=calldata,
                )
            )
        else:
            exec_txs = list(
                code.transactions_by_gas_limit(
                    fork=fork,
                    gas_limit=gas_benchmark_value,
                    sender=sender,
                    to=contract_address,
                    calldata=calldata,
                )
            )
        total_gas_cost = sum(tx.gas_cost for tx in exec_txs)

    post = {
        compute_create_address(
            address=contract_address,
            nonce=1 + i,
            salt=i,
            initcode=initcode,
            opcode=opcode,
        ): Account(nonce=1)
        for i in range(num_contracts)
    }

    benchmark_test(
        post=post,
        target_opcode=opcode,
        blocks=[Block(txs=exec_txs)],
        expected_benchmark_gas_used=total_gas_cost,
    )