diff --git a/lopper/assists/lopper_lib.py b/lopper/assists/lopper_lib.py index 58a3747d..d0f6a9ee 100644 --- a/lopper/assists/lopper_lib.py +++ b/lopper/assists/lopper_lib.py @@ -18,6 +18,7 @@ import shutil import ast from dataclasses import dataclass +from enum import Enum from pathlib import Path from pathlib import PurePath from io import StringIO @@ -38,6 +39,30 @@ lopper.log._init(__name__) + +class CpuSelectionSource(Enum): + """How a domain CPU selection was resolved.""" + + CLUSTER_RELATIVE = "cluster-relative" + LEGACY_CLUSTER_CPU = "legacy-cluster-cpu" + LEGACY_CORE_INDEX = "legacy-core-index" + UNRESOLVED = "unresolved" + + +@dataclass +class CpuSelection: + """Result of resolving a domain CPU mask against an SDT cluster.""" + + cluster: object = None + mask: object = None + cpus: list = None + source: CpuSelectionSource = CpuSelectionSource.UNRESOLVED + diagnostic: str = "" + + def __post_init__(self): + if self.cpus is None: + self.cpus = [] + # utility function to return true or false if a number # is 32 bit, or not. def check_32_bit(n): @@ -63,6 +88,155 @@ def chunks(l, n): yield l[i:i+n] +def cpu_nodes_from_mask(cluster, mask): + """Return CPU children selected by a cluster-relative CPU mask.""" + if cluster is None or mask is None: + return [] + + try: + mask = int(mask) + except (TypeError, ValueError): + return [] + + cpus = [node for node in cluster.subnodes(children_only=True) + if re.match(r"cpu@.*", node.name)] + return [cpu for index, cpu in enumerate(cpus) + if check_bit_set(mask, index)] + + +def _cpu_matches_legacy_label(tree, cpu, legacy_cpu_label): + """Return whether a CPU is named by an SDT symbol or node label.""" + if legacy_cpu_label in (cpu.label, cpu.name): + return True + + # Parsed DTS nodes do not necessarily retain their source label on the + # LopperNode. In that case, resolve the label through /__symbols__, just + # as baremetalconfig_xlnx.get_cpu_node() does for processor arguments. + try: + symbol_path = tree["/__symbols__"].propval(legacy_cpu_label) + except (KeyError, TypeError, AttributeError): + return False + if isinstance(symbol_path, list): + symbol_path = symbol_path[0] if symbol_path else None + return bool(symbol_path and symbol_path == getattr(cpu, "abs_path", None)) + + +def resolve_domain_cpus(tree, domain, legacy_cpu_label=None): + """Resolve a domain CPU tuple while accepting unambiguous legacy masks. + + CPU masks are defined relative to the CPU children of their referenced + cluster. Existing domain files may instead encode a system-wide core + number, such as mask 0x2 for the sole ``cpu@1`` child of ``cpus-r5@1``. + + Resolution follows these rules in order: + + 1. Resolve the cluster and interpret the mask as cluster-relative. + 2. Use that result only when the mask contains no out-of-range bits. + 3. If it is invalid or empty, accept ``legacy_cpu_label`` only when that + label names a direct CPU child of the referenced cluster. + 4. Without a label, accept a legacy mask for a one-CPU cluster only when + it is the one-hot bit corresponding to that child's small ``reg`` ID. + 5. Otherwise report the selection as unresolved; never guess a CPU. + + A valid cluster-relative mask is authoritative. The legacy label is not + allowed to override it when the two disagree. + """ + if isinstance(legacy_cpu_label, list): + legacy_cpu_label = ( + legacy_cpu_label[0] if legacy_cpu_label else None) + if not legacy_cpu_label: + legacy_cpu_label = None + + cpus_property = domain.propval("cpus") + if cpus_property == [""] or not cpus_property: + return CpuSelection(diagnostic="domain has no cpus property") + + cluster = tree.pnode(cpus_property[0]) + if cluster is None: + return CpuSelection(diagnostic="cpus references an unknown cluster") + + mask = None + mask_diagnostic = "cpus property has no mask" + if len(cpus_property) >= 2: + try: + mask = int(cpus_property[1]) + mask_diagnostic = f"cpumask {mask:#x} selects no CPU" + except (TypeError, ValueError): + mask = cpus_property[1] + mask_diagnostic = "cpus mask is not an integer" + + cpu_nodes = [ + node for node in cluster.subnodes(children_only=True) + if re.match(r"cpu@.*", node.name) + ] + valid_mask = (1 << len(cpu_nodes)) - 1 + selected = cpu_nodes_from_mask(cluster, mask) + if (isinstance(mask, int) and mask > 0 + and not (mask & ~valid_mask) and selected): + selected_names = { + name for cpu in selected for name in (cpu.label, cpu.name) if name + } + diagnostic = "" + if legacy_cpu_label and legacy_cpu_label not in selected_names: + diagnostic = ( + f"cluster_cpu {legacy_cpu_label} disagrees with cpumask " + f"{mask:#x}; using the cluster-relative mask" + ) + return CpuSelection( + cluster=cluster, + mask=mask, + cpus=selected, + source=CpuSelectionSource.CLUSTER_RELATIVE, + diagnostic=diagnostic, + ) + + if legacy_cpu_label: + legacy_cpu = next( + (cpu for cpu in cpu_nodes + if _cpu_matches_legacy_label(tree, cpu, legacy_cpu_label)), + None, + ) + if legacy_cpu is not None: + return CpuSelection( + cluster=cluster, + mask=mask, + cpus=[legacy_cpu], + source=CpuSelectionSource.LEGACY_CLUSTER_CPU, + diagnostic=( + f"{mask_diagnostic} relative to " + f"{cluster.label or cluster.name}; using legacy " + f"cluster_cpu {legacy_cpu_label}" + ), + ) + + if len(cpu_nodes) == 1 and isinstance(mask, int) and mask > 0: + reg = cpu_nodes[0].propval("reg") + try: + core_id = int(reg[0]) + except (IndexError, TypeError, ValueError): + core_id = -1 + if 0 <= core_id < 32 and mask == (1 << core_id): + return CpuSelection( + cluster=cluster, + mask=mask, + cpus=cpu_nodes, + source=CpuSelectionSource.LEGACY_CORE_INDEX, + diagnostic=( + f"cpumask {mask:#x} uses the legacy core index for " + f"{cpu_nodes[0].label or cpu_nodes[0].name}; use 0x1" + ), + ) + + return CpuSelection( + cluster=cluster, + mask=mask, + diagnostic=( + f"{mask_diagnostic} in " + f"{cluster.label or cluster.name}" + ), + ) + + def json_expand( node ): lopper.log._debug( f"========> json expanding node: {node.name}", level=lopper.log.TRACE ) for p in node: @@ -942,20 +1116,34 @@ def cpu_refs( tree, cpu_prop, verbose = 0 ): lopper.log._info( f"cpu node: {cpu_node}" ) lopper.log._info( f"sub cpus: {sub_cpus}" ) - # we'll now walk from 0 -> 31. Checking the mask to see if access is - # allowed. If it is allowed, we'll check to see if there's a sub-cpu at - # the same offset. If so, we refcount it AND the parent. For sub-cpus - # that are available, but have no access, we log them to be delete later - # (we don't delete them now, since it will shift node numbers. - for idx in range( 0, 32 ): - if check_bit_set( cpu_mask, idx ): - try: - sub_cpu_node = sub_cpus[idx] - # refcount it AND the parent - tree.ref_all( sub_cpu_node, True ) - refd_cpus.append( sub_cpu_node ) - except: - pass + selected_cpus = cpu_nodes_from_mask(cpu_node, cpu_mask) + + # OpenAMP domains created by older YAML may use an absolute RPU bit + # with a split, one-CPU cluster. Resolve that single tuple with the + # same compatibility rules used by OpenAMP relation matching. + if not selected_cpus and len(cpu_prop_list) == 1 and cpu_prop.node: + domain = cpu_prop.node + dtd = next( + (node for node in domain.subnodes(children_only=True) + if node.name == "domain-to-domain"), + None, + ) + legacy_cpu = dtd.propval("cluster_cpu") if dtd else None + if legacy_cpu == [""]: + legacy_cpu = None + selection = resolve_domain_cpus(tree, domain, legacy_cpu) + selected_cpus = selection.cpus + if (selection.diagnostic + and selection.source != CpuSelectionSource.UNRESOLVED): + lopper.log._warning( + f"{domain.abs_path}: {selection.diagnostic}; migrate the " + "domain to a cluster-relative mask" + ) + + for sub_cpu_node in selected_cpus: + # refcount it AND the parent + tree.ref_all(sub_cpu_node, True) + refd_cpus.append(sub_cpu_node) unrefd_cpus = [] for s in sub_cpus_all: diff --git a/lopper/assists/openamp_xlnx.py b/lopper/assists/openamp_xlnx.py index 13950d73..c65b1b15 100644 --- a/lopper/assists/openamp_xlnx.py +++ b/lopper/assists/openamp_xlnx.py @@ -33,6 +33,10 @@ sys.path.append(os.path.dirname(__file__)) from openamp_xlnx_common import * +from openamp_xlnx_common import ( + _openamp_domain_processor, + _openamp_domain_selects_cpu, +) from baremetalconfig_xlnx import get_cpu_node from string import ascii_lowercase as alc @@ -130,11 +134,7 @@ def xlnx_openamp_update_relation_timers(sdt, target_os, machine): if domain is None: continue - domain_cpus = domain.propval("cpus") - domain_cluster = ( - tree.pnode(domain_cpus[0]) if domain_cpus != [''] else None - ) - if match_cpu.parent == domain_cluster: + if _openamp_domain_selects_cpu(tree, domain, match_cpu): relation_groups.append(node) if not relation_groups: @@ -216,7 +216,8 @@ def xlnx_handle_relations(sdt, machine, find_only = True, os = None): continue # ensure target domain matches - if match_cpunode.parent == sdt.tree.pnode(n.parent.parent.propval("cpus")[0]): + if _openamp_domain_selects_cpu( + tree, n.parent.parent, match_cpunode): if find_only: return n else: # do processing on found nodes @@ -345,7 +346,8 @@ def xlnx_openamp_get_ddr_elf_load(machine, sdt): continue # ensure target domain matches - if match_cpunode.parent == sdt.tree.pnode(n.parent.parent.propval("cpus")[0]): + if _openamp_domain_selects_cpu( + sdt.tree, n.parent.parent, match_cpunode): target_node = n break @@ -1362,8 +1364,11 @@ def openamp_nontree_outputs_handler(sdt, output_file_name, openamp_args, verbose match_cpunode = get_cpu_node(sdt, {'args':[machine]}) if os != "linux_dt" else None if not match_cpunode and os != "linux_dt": - print("openamp_nontree_outputs_handler: unable to find machine: ", machine) - return False + _error( + "openamp_xlnx: cannot generate '%s': processor '%s' was not " + "found in the system device tree" % (output_file_name, machine), + 1, + ) domains = sdt.tree['/domains'] relation_node = None @@ -1396,7 +1401,9 @@ def openamp_nontree_outputs_handler(sdt, output_file_name, openamp_args, verbose (domain_node.name, domain_os, domain_processor)) # ensure target domain matches - if os != "linux_dt" and match_cpunode.parent != sdt.tree.pnode(domain_node.propval("cpus")[0]): + if (os != "linux_dt" + and not _openamp_domain_selects_cpu( + sdt.tree, domain_node, match_cpunode)): continue # filter based on name @@ -1409,10 +1416,12 @@ def openamp_nontree_outputs_handler(sdt, output_file_name, openamp_args, verbose if relation_node is None: compatible = openamp_args['compatible_string'] or "any" targets = ", ".join(supported_targets) if supported_targets else "none" - _error("openamp_xlnx: no %s relation found for processor '%s' and OS " - "'%s'; supported targets: %s" % - (compatible, machine, os, targets)) - return False + _error( + "openamp_xlnx: cannot generate '%s': no %s relation found for " + "processor '%s' and OS '%s'; supported targets: %s" % + (output_file_name, compatible, machine, os, targets), + 1, + ) carveouts = None ipi_node = None diff --git a/lopper/assists/openamp_xlnx_common.py b/lopper/assists/openamp_xlnx_common.py index 0921857b..27aaac75 100644 --- a/lopper/assists/openamp_xlnx_common.py +++ b/lopper/assists/openamp_xlnx_common.py @@ -15,6 +15,12 @@ from pathlib import Path from baremetalconfig_xlnx import get_cpu_node +from lopper.log import _warning +from lopper.assists.lopper_lib import ( + CpuSelectionSource, + _cpu_matches_legacy_label, + resolve_domain_cpus, +) IPI_MAILBOX_COMPATIBLES = { "xlnx,versal-ipi-mailbox", @@ -82,13 +88,58 @@ def _openamp_ipi_controllers(tree): return controllers -def _openamp_domain_processor(tree, domain): +def _openamp_domain_cpu_assignment(tree, domain): + """Resolve a domain's CPUs using canonical and legacy metadata.""" + legacy_cpu = _openamp_legacy_domain_processor(domain) + selection = resolve_domain_cpus(tree, domain, legacy_cpu) + + if selection.diagnostic: + domain_name = getattr(domain, "abs_path", domain.name) + if selection.source == CpuSelectionSource.UNRESOLVED: + _warning(f"{domain_name}: {selection.diagnostic}") + else: + _warning(f"{domain_name}: {selection.diagnostic}; migrate " + "the domain to a cluster-relative mask") + + return selection.cluster, selection.mask, selection.cpus + + +def _openamp_legacy_domain_processor(domain): + """Read the historical OpenAMP core label, when present.""" dtd = next((n for n in domain.subnodes(children_only=True) if n.name == "domain-to-domain"), None) - if dtd and dtd.propval("cluster_cpu") != [""]: - return dtd.propval("cluster_cpu")[0] - cpus = domain.propval("cpus") - cluster = tree.pnode(cpus[0]) if cpus != [""] else None + value = dtd.propval("cluster_cpu") if dtd else [""] + return value[0] if value != [""] and value else None + + +def _openamp_domain_selects_cpu(tree, domain, cpu): + """Return whether the domain's resolved CPU set contains ``cpu``.""" + cluster, _, selected = _openamp_domain_cpu_assignment(tree, domain) + if cpu in selected: + return True + + # Some released Versal Net domains use the unprefixed historical value + # ``cortexr52_N`` for cluster_cpu while their SDT CPU symbol is + # ``psx_cortexr52_N``. Match either spelling, but only for the named CPU + # in the referenced cluster so an unresolved mask cannot widen selection. + legacy_cpu = _openamp_legacy_domain_processor(domain) + return bool( + not selected + and legacy_cpu + and cpu is not None + and cpu.parent == cluster + and (_cpu_matches_legacy_label(tree, cpu, legacy_cpu) + or _cpu_matches_legacy_label(tree, cpu, f"psx_{legacy_cpu}")) + ) + + +def _openamp_domain_processor(tree, domain): + cluster, _, selected = _openamp_domain_cpu_assignment(tree, domain) + if selected: + return ", ".join(cpu.label or cpu.name for cpu in selected) + legacy = _openamp_legacy_domain_processor(domain) + if legacy: + return legacy return (cluster.label or cluster.name) if cluster else "unspecified" @@ -105,7 +156,7 @@ def _openamp_configured_relations(tree): continue dtd = next((n for n in domain.subnodes(children_only=True) if n.name == "domain-to-domain"), None) - if not dtd or dtd.propval("cluster_cpu") == [""]: + if not dtd: continue for relation in dtd.subnodes(children_only=True): compatible = relation.propval("compatible") diff --git a/lopper/assists/yaml_to_dts_expansion.py b/lopper/assists/yaml_to_dts_expansion.py index 3259d60e..424fddf7 100755 --- a/lopper/assists/yaml_to_dts_expansion.py +++ b/lopper/assists/yaml_to_dts_expansion.py @@ -32,7 +32,10 @@ import lopper import json -from .lopper_lib import check_bit_set, clear_bit, chunks, property_set, set_bit, expand_start_size_to_reg +from .lopper_lib import (check_bit_set, clear_bit, chunks, + CpuSelectionSource, property_set, + resolve_domain_cpus, set_bit, + expand_start_size_to_reg) from lopper.log import _init, _warning, _info, _error, _debug from .zephyr_memory import ( LINKER_SCALAR_PROPERTIES, @@ -1174,21 +1177,38 @@ def openamp_remote_cpu_expand( tree, subnode, cluster_cpu, cluster_node, verbose Returns: None """ - if cluster_cpu == None: - return - + # Preserve the legacy property for old consumers while deriving metadata + # from the canonical CPU assignment below. for n in subnode.subnodes(): - if n.name == "domain-to-domain": + if cluster_cpu is not None and n.name == "domain-to-domain": n + LopperProp(name="cluster_cpu", value=cluster_cpu) - if cluster_node is not None: - pd_prop_node = [ n for n in cluster_node.subnodes() if n.propval("power-domains") != [''] ] - if len(pd_prop_node) == 1: - subnode + LopperProp(name="rpu_pd_val", value=pd_prop_node[0].propval("power-domains")) - - if cluster_node != None and "r5" in cluster_node.name: + # Resolve the standard cluster-relative mask first. Older domain YAML + # used cluster_cpu, or encoded a split RPU's global core number in the + # mask. The shared resolver accepts only validated legacy forms so YAML + # expansion and later OpenAMP matching cannot disagree about the CPU. + selection = resolve_domain_cpus(tree, subnode, cluster_cpu) + selected_cores = selection.cpus + if selection.diagnostic: + if selection.source == CpuSelectionSource.CLUSTER_RELATIVE: + _warning(f"{subnode.abs_path}: {selection.diagnostic}") + elif selection.source != CpuSelectionSource.UNRESOLVED: + _warning(f"{subnode.abs_path}: {selection.diagnostic}; " + "migrate the domain to a cluster-relative mask") + + if len(selected_cores) == 1: + power_domains = selected_cores[0].propval("power-domains") + if power_domains != ['']: + subnode + LopperProp(name="rpu_pd_val", value=power_domains) + + if cluster_node is not None and "r5" in cluster_node.name: subnode + LopperProp(name="cpu_config_str", value="lockstep" if check_bit_set(subnode.propval("cpus")[2], 30) else "split") - subnode + LopperProp(name="core_num", value=cluster_node.name[-1]) + if len(selected_cores) == 1: + subnode + LopperProp( + name="core_num", value=selected_cores[0].propval("reg")[0]) + elif cluster_cpu is not None: + # Preserve legacy behavior when no CPU node can be resolved. + subnode + LopperProp(name="core_num", value=cluster_node.name[-1]) def _cpu_mask_value(mask): diff --git a/tests/test_libmetal_zynqmp.py b/tests/test_libmetal_zynqmp.py index 96845be2..5ba865bd 100644 --- a/tests/test_libmetal_zynqmp.py +++ b/tests/test_libmetal_zynqmp.py @@ -62,15 +62,37 @@ def _assert_node_compatible(dts, unit_name, compatible): assert f'compatible = "{compatible}";' in match.group(1) -def test_zynqmp_libmetal_linux_and_baremetal_outputs(tmp_path): - """Generate domain slices and Libmetal CMake data for both endpoints.""" +@pytest.mark.parametrize( + "r5_mask, include_legacy_cpu", + [("0x1", False), ("0x2", True)], +) +def test_zynqmp_libmetal_linux_and_baremetal_outputs( + tmp_path, r5_mask, include_legacy_cpu): + """Generate both endpoints from canonical and legacy R5 masks.""" + domain_yaml = tmp_path / "libmetal-overlay-zynqmp.yaml" + yaml_text = LIBMETAL_YAML.read_text() + mask_pattern = r"(cluster: cpus_r5_1\n\s+cpumask:) 0x[12]" + yaml_text, replacements = re.subn( + mask_pattern, + rf"\g<1> {r5_mask}", + yaml_text, + count=1, + ) + assert replacements == 1 + if not include_legacy_cpu: + yaml_text = yaml_text.replace( + " cluster_cpu: psu_cortexr5_1\n", "", 1) + domain_yaml.write_text(yaml_text) + expanded = tmp_path / "libmetal-zynqmp-expanded.dts" - _run([ + output = _run([ "-f", "--permissive", "--enhanced", "--auto", - "-i", LIBMETAL_YAML, + "-i", domain_yaml, "-i", DOMAIN_ACCESS_YAML, SDT, expanded, ]) + if r5_mask == "0x2": + assert "using legacy cluster_cpu psu_cortexr5_1" in output expanded_text = expanded.read_text() assert "__lopper-overlays__" in expanded_text assert 'lopper,activate = "linux";' in expanded_text diff --git a/tests/test_openamp.py b/tests/test_openamp.py index b14c5a4c..a3480886 100644 --- a/tests/test_openamp.py +++ b/tests/test_openamp.py @@ -17,7 +17,12 @@ import os import pytest -from lopper.assists import openamp_xlnx +from lopper.assists import ( + lopper_lib, + openamp_xlnx, + openamp_xlnx_common, + yaml_to_dts_expansion, +) from lopper.tree import LopperNode, LopperTree @@ -69,19 +74,177 @@ def subnodes(self, children_only=False): class _FakeTree: - def __init__(self, domains, phandles): + def __init__(self, domains, phandles, symbols=None): self._domains = domains self._phandles = phandles + self._symbols = symbols def __getitem__(self, path): if path == "/domains": return self._domains + if path == "/__symbols__" and self._symbols is not None: + return self._symbols raise KeyError(path) def pnode(self, phandle): return self._phandles.get(phandle) +def _cpu_selection_fixture(mask, cpu_count=1, first_reg=0): + cpus = [ + _FakeNode( + f"cpu@{first_reg + index:x}", + {"reg": [first_reg + index]}, + label=f"cpu_{first_reg + index}", + ) + for index in range(cpu_count) + ] + cluster = _FakeNode("cpus-r5@0", label="cpus_r5", children=cpus) + for cpu in cpus: + cpu.parent = cluster + domain = _FakeNode("RPU", {"cpus": [1, mask, 0]}) + tree = _FakeTree(_FakeNode("domains"), {1: cluster}) + return tree, domain, cpus + + +@pytest.mark.parametrize( + "mask, expected", + [(0x1, [0]), (0x2, [1]), (0x3, [0, 1])], +) +def test_domain_cpu_resolver_uses_cluster_relative_masks(mask, expected): + tree, domain, cpus = _cpu_selection_fixture(mask, cpu_count=2) + + selection = lopper_lib.resolve_domain_cpus(tree, domain) + + assert selection.source == lopper_lib.CpuSelectionSource.CLUSTER_RELATIVE + assert selection.cpus == [cpus[index] for index in expected] + + +def test_domain_cpu_resolver_accepts_legacy_label_for_split_cluster(): + tree, domain, cpus = _cpu_selection_fixture(0x2, first_reg=1) + + selection = lopper_lib.resolve_domain_cpus(tree, domain, "cpu_1") + + assert selection.source == lopper_lib.CpuSelectionSource.LEGACY_CLUSTER_CPU + assert selection.cpus == cpus + assert "use 0x1" not in selection.diagnostic + + +def test_domain_cpu_resolver_accepts_unambiguous_legacy_core_mask(): + tree, domain, cpus = _cpu_selection_fixture(0x2, first_reg=1) + + selection = lopper_lib.resolve_domain_cpus(tree, domain) + + assert selection.source == lopper_lib.CpuSelectionSource.LEGACY_CORE_INDEX + assert selection.cpus == cpus + assert "use 0x1" in selection.diagnostic + + +@pytest.mark.parametrize("mask", [0, 0x4]) +def test_domain_cpu_resolver_rejects_ambiguous_masks(mask): + tree, domain, _ = _cpu_selection_fixture(mask, first_reg=1) + + selection = lopper_lib.resolve_domain_cpus(tree, domain) + + assert selection.source == lopper_lib.CpuSelectionSource.UNRESOLVED + assert selection.cpus == [] + + +def test_domain_cpu_resolver_uses_legacy_label_for_zero_mask(): + tree, domain, cpus = _cpu_selection_fixture(0, first_reg=1) + + selection = lopper_lib.resolve_domain_cpus(tree, domain, "cpu_1") + + assert selection.source == lopper_lib.CpuSelectionSource.LEGACY_CLUSTER_CPU + assert selection.cpus == cpus + + +def test_domain_cpu_resolver_rejects_missing_legacy_label_sentinel(): + tree, domain, cpus = _cpu_selection_fixture(0, first_reg=1) + cpus[0].label = "" + + selection = lopper_lib.resolve_domain_cpus(tree, domain, [""]) + + assert selection.source == lopper_lib.CpuSelectionSource.UNRESOLVED + assert selection.cpus == [] + + +def test_domain_cpu_resolver_resolves_legacy_symbol_for_zero_mask(): + tree, domain, cpus = _cpu_selection_fixture(0, first_reg=1) + cpus[0].label = None + cpus[0].abs_path = "/cpus-r5@0/cpu@1" + tree._symbols = _FakeNode( + "__symbols__", {"psu_cortexr5_1": [cpus[0].abs_path]}) + + selection = lopper_lib.resolve_domain_cpus( + tree, domain, "psu_cortexr5_1") + + assert selection.source == lopper_lib.CpuSelectionSource.LEGACY_CLUSTER_CPU + assert selection.cpus == cpus + + +def test_openamp_legacy_processor_falls_back_to_referenced_cluster(): + tree, domain, cpus = _cpu_selection_fixture(0, first_reg=1) + cpus[0].label = "psx_cortexr52_1" + dtd = _FakeNode( + "domain-to-domain", {"cluster_cpu": ["cortexr52_1"]}) + domain._children = [dtd] + + assert openamp_xlnx_common._openamp_domain_selects_cpu( + tree, domain, cpus[0]) + + other_cpu = _FakeNode("cpu@1", parent=_FakeNode("cpus-r52@0")) + assert not openamp_xlnx_common._openamp_domain_selects_cpu( + tree, domain, other_cpu) + + +def test_openamp_legacy_processor_does_not_widen_within_cluster(): + tree, domain, cpus = _cpu_selection_fixture(0x4, cpu_count=2) + cpus[0].label = "psx_cortexr52_0" + cpus[1].label = "psx_cortexr52_1" + dtd = _FakeNode( + "domain-to-domain", {"cluster_cpu": ["cortexr52_1"]}) + domain._children = [dtd] + + assert not openamp_xlnx_common._openamp_domain_selects_cpu( + tree, domain, cpus[0]) + assert openamp_xlnx_common._openamp_domain_selects_cpu( + tree, domain, cpus[1]) + + +def test_domain_cpu_resolver_prefers_valid_mask_over_legacy_label(): + tree, domain, cpus = _cpu_selection_fixture(0x2, cpu_count=2) + + selection = lopper_lib.resolve_domain_cpus(tree, domain, "cpu_0") + + assert selection.source == lopper_lib.CpuSelectionSource.CLUSTER_RELATIVE + assert selection.cpus == [cpus[1]] + assert "disagrees" in selection.diagnostic + + +def test_domain_cpu_resolver_rejects_unknown_cluster(): + domain = _FakeNode("RPU", {"cpus": [99, 0x1, 0]}) + tree = _FakeTree(_FakeNode("domains"), {}) + + selection = lopper_lib.resolve_domain_cpus(tree, domain) + + assert selection.source == lopper_lib.CpuSelectionSource.UNRESOLVED + assert selection.cpus == [] + assert "unknown cluster" in selection.diagnostic + + +def test_domain_cpu_resolver_rejects_cluster_without_cpus(): + cluster = _FakeNode("cpus-r5@1", label="cpus_r5_1") + domain = _FakeNode("RPU", {"cpus": [1, 0x1, 0]}) + tree = _FakeTree(_FakeNode("domains"), {1: cluster}) + + selection = lopper_lib.resolve_domain_cpus(tree, domain) + + assert selection.source == lopper_lib.CpuSelectionSource.UNRESOLVED + assert selection.cpus == [] + assert "selects no CPU" in selection.diagnostic + + def test_legacy_zephyr_memories_remain_compatible(caplog): """Legacy memory lists mark every bank and select the first bank.""" tree = LopperTree() @@ -320,19 +483,22 @@ def test_libmetal_missing_processor_lists_supported_targets(monkeypatch, caplog) monkeypatch.setattr(openamp_xlnx, "get_cpu_node", lambda sdt, options: requested_cpu) - result = openamp_xlnx.openamp_nontree_outputs_handler( - sdt, - "unused.cmake", - { - "machine": "psu_cortexr5_0", - "dt_type": "baremetal_dt", - "relation_parent": None, - "relation": None, - "compatible_string": "libmetal,ipc-v1", - }, - ) - - assert result is False - assert "no libmetal,ipc-v1 relation found for processor 'psu_cortexr5_0'" in caplog.text + with pytest.raises(SystemExit) as error: + openamp_xlnx.openamp_nontree_outputs_handler( + sdt, + "unused.cmake", + { + "machine": "psu_cortexr5_0", + "dt_type": "baremetal_dt", + "relation_parent": None, + "relation": None, + "compatible_string": "libmetal,ipc-v1", + }, + ) + + assert error.value.code == 1 + assert "cannot generate 'unused.cmake'" in caplog.text + assert "no libmetal,ipc-v1 relation found" in caplog.text + assert "processor 'psu_cortexr5_0'" in caplog.text assert "APU_Linux (os=linux, processor=cpus_a53)" in caplog.text assert "R5_1_BAREMETAL (os=baremetal, processor=psu_cortexr5_1)" in caplog.text