Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 17 additions & 6 deletions src/fpy/state.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@
FpyType,
FpyValue,
TypeKind,
U64,
)
from fpy.bytecode.directives import update_configurable_types_from_dict

Expand Down Expand Up @@ -348,9 +349,9 @@ def _validate_and_replace_type(
def _update_seq_args_from_dict(dict_type_name_dict: dict[str, FpyType]) -> None:
"""Update the canonical SEQ_ARGS singleton from the dictionary's Svc.SeqArgs.

The dictionary's `Svc.SeqArgs` defines the actual buffer capacity for this
deployment. The compiler adopts that length onto the canonical singleton's
buffer type so codegen and semantics use the correct size.
The dictionary's `Svc.SeqArgs` defines the size member type and buffer
capacity. Adopt both onto the canonical singleton used by codegen and
semantics.

A dictionary that omits `Svc.SeqArgs` (no sequencer in it takes arguments)
gets the canonical default capacity. This fallback is temporary: sequence
Expand All @@ -362,6 +363,7 @@ def _update_seq_args_from_dict(dict_type_name_dict: dict[str, FpyType]) -> None:
# the buffer capacity may differ from the previous dictionary's.
SEQ_ARGS.member_defaults = None
if "Svc.SeqArgs" not in dict_type_name_dict:
canonical_size_member.type = U64
canonical_buffer_type.length = DEFAULT_SEQ_ARGS_BUFFER_SIZE
canonical_buffer_type.name = f"Array_U8_{DEFAULT_SEQ_ARGS_BUFFER_SIZE}"
SEQ_ARGS.json_default = {
Expand All @@ -378,7 +380,7 @@ def _update_seq_args_from_dict(dict_type_name_dict: dict[str, FpyType]) -> None:
f"but Fpy expects a STRUCT type.",
)
expected = (
f"({canonical_size_member.name}: {canonical_size_member.type}, "
f"({canonical_size_member.name}: unsigned integer, "
f"{canonical_buffer_member.name}: [N] {canonical_buffer_type.elem_type})"
)
members_match = (
Expand All @@ -388,7 +390,7 @@ def _update_seq_args_from_dict(dict_type_name_dict: dict[str, FpyType]) -> None:
size_member, buffer_member = dict_seq_args.members
members_match = (
size_member.name == canonical_size_member.name
and size_member.type == canonical_size_member.type
and size_member.type.is_unsigned
and buffer_member.name == canonical_buffer_member.name
and buffer_member.type.kind == TypeKind.ARRAY
and buffer_member.type.elem_type == canonical_buffer_type.elem_type
Expand All @@ -407,8 +409,17 @@ def _update_seq_args_from_dict(dict_type_name_dict: dict[str, FpyType]) -> None:
f"Its {buffer_member.name} member must have a positive length, "
f"but the dictionary gives it length {dict_buffer_type.length}.",
)
maximum_size = size_member.type.value_range()[1]
if dict_buffer_type.length > maximum_size:
raise DictionaryError(
"Svc.SeqArgs",
f"Its {buffer_member.name} length {dict_buffer_type.length} exceeds "
f"the range of its {size_member.name} type {size_member.type.name} "
f"({maximum_size}).",
)

# Adopt the dictionary's buffer length onto the canonical buffer singleton.
# Adopt the dictionary's size field and buffer length onto the singletons.
canonical_size_member.type = size_member.type
canonical_buffer_type.length = dict_buffer_type.length
canonical_buffer_type.name = f"Array_U8_{dict_buffer_type.length}"

Expand Down
174 changes: 174 additions & 0 deletions test/fpy/test_seq_args_size_type.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,174 @@
"""Regression tests for dictionary-configured Svc.SeqArgs sizes."""

import json
from pathlib import Path

import pytest

from fpy.compiler import (
analysis_to_fpybc_directives,
analysis_to_llvm_module,
analyze_ast,
text_to_ast,
)
from fpy.dictionary import load_dictionary
from fpy.error import CompileError, DictionaryError
from fpy.state import _build_global_scopes, get_base_compile_state
from fpy.types import DEFAULT_SEQ_ARGS_BUFFER_SIZE, FpyValue, SEQ_ARGS, U64

DEFAULT_DICTIONARY = Path(__file__).with_name("RefTopologyDictionary.json")


@pytest.fixture(autouse=True)
def isolate_dictionary_configurations():
"""Rebuild scopes and SeqArgs defaults independently for each test."""
load_dictionary.cache_clear()
_build_global_scopes.cache_clear()
SEQ_ARGS.member_defaults = None
yield
load_dictionary.cache_clear()
_build_global_scopes.cache_clear()
SEQ_ARGS.member_defaults = None
get_base_compile_state(str(DEFAULT_DICTIONARY))


def _dictionary(tmp_path, *, width=32, capacity=8, default_size=0, signed=False):
"""Copy the reference dictionary with a configured size field and buffer."""
data = json.loads(DEFAULT_DICTIONARY.read_text(encoding="utf-8"))
for definition in data["typeDefinitions"]:
if definition["qualifiedName"] == "FwSizeType":
definition["underlyingType"] = {
"name": f"{'I' if signed else 'U'}{width}",
"kind": "integer",
"size": width,
"signed": signed,
}
elif definition["qualifiedName"] == "Svc.SeqArgs":
definition["members"]["buffer"]["size"] = capacity
definition["default"] = {
"size": default_size,
"buffer": [0] * capacity,
}
path = tmp_path / "dictionary.json"
path.write_text(json.dumps(data), encoding="utf-8")
return str(path)


@pytest.mark.parametrize("width", [8, 16, 32, 64])
def test_size_field_and_exact_serialized_bytes(tmp_path, width):
get_base_compile_state(_dictionary(tmp_path, width=width))
size, buffer = SEQ_ARGS.members
assert size.type.name == f"U{width}"
assert buffer.type.length == 8
assert SEQ_ARGS.max_size == width // 8 + 8
value = FpyValue(SEQ_ARGS, {"size": 3, "buffer": list(range(8))})
assert value.serialize() == (3).to_bytes(width // 8, "big") + bytes(range(8))


@pytest.mark.parametrize("backend", ["fpybc", "llvm"])
@pytest.mark.parametrize("width", [32, 64])
def test_constructor_default_and_codegen(tmp_path, backend, width):
state = get_base_compile_state(_dictionary(tmp_path, width=width, default_size=3))
state = analyze_ast(
text_to_ast("args: Svc.SeqArgs = Svc.SeqArgs()\nassert args.size == 3\n"), state
)
if backend == "fpybc":
directives, _ = analysis_to_fpybc_directives(state)
expected = (3).to_bytes(width // 8, "big") + bytes(8)
assert any(getattr(d, "val", None) == expected for d in directives)
else:
import llvmlite.binding as llvm

module, _ = analysis_to_llvm_module(state)
llvm.parse_assembly(str(module)).verify()

assert SEQ_ARGS.member_defaults["size"].type.name == f"U{width}"
assert SEQ_ARGS.member_defaults["size"].val == 3
ctor = state.type_ctors[SEQ_ARGS]
assert ctor.args[0][1].name == f"U{width}"
assert ctor.args[0][2].type.name == f"U{width}"
assert ctor.args[0][2].val == 3
assert len(ctor.args[1][2].val) == 8


def test_dictionary_without_seq_args_uses_builtin_fallback(tmp_path):
data = json.loads(DEFAULT_DICTIONARY.read_text(encoding="utf-8"))
data["typeDefinitions"] = [
definition
for definition in data["typeDefinitions"]
if definition["qualifiedName"] != "Svc.SeqArgs"
]
data["commands"] = [
command
for command in data["commands"]
if not any(
arg["type"].get("name") == "Svc.SeqArgs"
for arg in command["formalParams"]
)
]
path = tmp_path / "no-seq-args.json"
path.write_text(json.dumps(data), encoding="utf-8")
state = get_base_compile_state(str(path))
assert SEQ_ARGS.members[0].type is U64
assert SEQ_ARGS.members[1].type.length == DEFAULT_SEQ_ARGS_BUFFER_SIZE
assert state.type_ctors[SEQ_ARGS].args[0][2].type is U64


def test_signed_size_member_is_rejected(tmp_path):
with pytest.raises(DictionaryError, match="Svc.SeqArgs"):
get_base_compile_state(_dictionary(tmp_path, signed=True))


@pytest.mark.parametrize(
"capacity, message",
[(0, "positive length"), (256, "length 256 exceeds.*U8")],
)
def test_invalid_buffer_capacities_are_rejected(tmp_path, capacity, message):
with pytest.raises(DictionaryError, match=message):
get_base_compile_state(_dictionary(tmp_path, width=8, capacity=capacity))


def test_u8_capacity_limit_is_accepted(tmp_path):
get_base_compile_state(_dictionary(tmp_path, width=8, capacity=255))
assert SEQ_ARGS.members[1].type.length == 255
assert SEQ_ARGS.max_size == 256


@pytest.mark.parametrize("backend", ["fpybc", "llvm"])
@pytest.mark.parametrize("width", [32, 64])
@pytest.mark.parametrize("dynamic", [False, True])
def test_seq_run_exact_capacity_compiles(tmp_path, backend, width, dynamic):
(tmp_path / "child.fpy").write_text(
"sequence(x: U32, y: U32)\nassert x == 42\nassert y == 7\n"
)
source = (
'x: U32 = 42\nRef.seqDisp.RUN_ARGS("child.bin", Svc.BlockState.BLOCK, x, 7)\n'
if dynamic
else 'Ref.seqDisp.RUN_ARGS("child.bin", Svc.BlockState.BLOCK, 42, 7)\n'
)
state = get_base_compile_state(
_dictionary(tmp_path, width=width), seq_maps=[("", str(tmp_path) + "/")]
)
state = analyze_ast(text_to_ast(source), state)
if backend == "fpybc":
analysis_to_fpybc_directives(state)
else:
import llvmlite.binding as llvm

module, _ = analysis_to_llvm_module(state)
llvm.parse_assembly(str(module)).verify()


@pytest.mark.parametrize("width", [32, 64])
def test_seq_run_over_capacity_is_a_source_diagnostic(tmp_path, width):
(tmp_path / "child.fpy").write_text("sequence(x: U32, y: U32, z: U8)\n")
state = get_base_compile_state(
_dictionary(tmp_path, width=width), seq_maps=[("", str(tmp_path) + "/")]
)
with pytest.raises(CompileError, match="SeqArgs buffer capacity"):
analyze_ast(
text_to_ast(
'Ref.seqDisp.RUN_ARGS("child.bin", Svc.BlockState.BLOCK, 1, 2, 3)\n'
),
state,
)
Loading