Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_multicore / _native_manifest.py: 82%
185 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-11-15 05:03 +0800
1# Copyright 2026 Huawei Technologies Co., Ltd
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14# ============================================================================
15"""Standard-library contracts for native adapter selection and payload integrity."""
17from __future__ import annotations
19import ast
20import hashlib
21from importlib import metadata, util
22import json
23from pathlib import Path
24import re
26SHMEM_DEPENDENCIES = (
27 "shmem/libhyper_multicore_shmem_utils.so",
28 "shmem/libhyper_multicore_shmem.so",
29 "libhyper_multicore_shmem_ops.so",
30)
31SHMEM_RESOURCES = (*SHMEM_DEPENDENCIES,
32 "shmem/aclshmem_bootstrap_uid.so", "shmem/aclshmem_bootstrap_config_store.so")
35def framework_package(package: str) -> dict:
36 """Describe the package selected by Python, including source trees without dist-info."""
37 spec = util.find_spec(package)
38 if spec is None or not spec.submodule_search_locations:
39 raise ModuleNotFoundError(f"Missing {package} package")
40 root = Path(next(iter(spec.submodule_search_locations))).resolve()
41 version = source_package_version(root)
42 if version == "unknown":
43 try:
44 distribution = metadata.distribution(package.replace("_", "-"))
45 if Path(distribution.locate_file(package)).resolve() == root:
46 version = distribution.version
47 except metadata.PackageNotFoundError:
48 pass
49 return {"version": version, "root": str(root)}
52def source_package_version(root: Path) -> str:
53 """Read literal version declarations without importing a backend or executing source."""
54 version = "unknown"
55 for filename in ("version.py", "__init__.py"):
56 path = root / filename
57 if not path.is_file():
58 continue
59 try:
60 tree = ast.parse(path.read_text(encoding="utf-8"))
61 except (SyntaxError, UnicodeError):
62 continue
63 for node in tree.body:
64 if isinstance(node, ast.Assign) and any(
65 isinstance(t, ast.Name) and t.id == "__version__" for t in node.targets):
66 if isinstance(node.value, ast.Constant) and isinstance(node.value.value, str):
67 version = node.value.value
68 return version
71def release(version: str) -> str:
72 """Extract a numeric branch, or retain an opaque label without version admission rules."""
73 match = re.match(r"[vV]?(\d+\.\d+\.\d+)", version)
74 return match[1] if match else version
77def file_sha256(path: Path) -> str:
78 """Hash a file without reading its entire contents into memory."""
79 digest = hashlib.sha256()
80 with path.open("rb") as source:
81 for block in iter(lambda: source.read(1024 * 1024), b""):
82 digest.update(block)
83 return digest.hexdigest()
86def resource_path(root: Path, relative: str) -> Path:
87 """Resolve an owned relative path, rejecting traversal and escaped symlinks."""
88 if not isinstance(relative, str) or not relative or "\\" in relative:
89 raise ValueError(f"Invalid payload path: {relative!r}")
90 path = Path(relative)
91 if path.is_absolute() or any(part in (".", "..") for part in relative.split("/")):
92 raise ValueError(f"Unsafe payload path: {relative!r}")
93 resolved = (root / path).resolve()
94 if not resolved.is_relative_to(root.resolve()):
95 raise ValueError(f"Escaped payload path: {relative!r}")
96 return resolved
99def variant_selection_key(variant: dict):
100 """Validate the unique adapter ID and stable branch/ABI selector."""
101 identifier = variant.get("id")
102 if not isinstance(identifier, str) or not re.fullmatch(r"[a-zA-Z0-9._-]+", identifier):
103 raise ValueError(f"Invalid adapter ID: {identifier!r}")
104 if not isinstance(variant.get("torch_cxx11_abi"), bool):
105 raise ValueError("Adapter C++ ABI must be boolean")
106 key = (variant.get("torch_release"), variant.get("pta_release"), variant["torch_cxx11_abi"])
107 if not all(isinstance(v, str) and release(v) == v for v in key[:2]):
108 raise ValueError("Adapter selection requires branch or opaque framework labels")
109 return identifier, key
112def validate_adapter_paths(root: Path, variant: dict, backends: dict):
113 """Require the shared Torch adapter and one binding for each packaged backend."""
114 expected_torch = f"adapters/{variant['id']}/libhyper_multicore_torch.so"
115 expected_shmem = {name: f"adapters/{variant['id']}/shmem/{name}/hyper_multicore_shmem_torch.so"
116 for name in backends}
117 if variant.get("torch_so") != expected_torch or variant.get("shmem_so") != expected_shmem:
118 raise ValueError(f"Invalid adapter paths for {variant['id']}")
119 for expected in (expected_torch, *expected_shmem.values()):
120 path = resource_path(root, expected)
121 if not path.is_file() or not path.stat().st_size:
122 raise ValueError(f"Missing adapter resource: {expected}")
125def validate_architecture(root: Path, name: str, backend: dict) -> None:
126 """Validate one backend's canonical paths, architecture and SoC membership."""
127 if not re.fullmatch(r"arch[0-9]+", name) or backend.get("shmem_root") != f"shmem/lib/{name}":
128 raise ValueError(f"Invalid SHMEM backend root: {name}")
129 kernel_root = f"lib/vendors/hyper_multicore_nn/op_impl/ai_core/tbe/kernel/{name}"
130 if backend.get("kernel_root") != kernel_root or not isinstance(backend.get("sdk_soc_type"), str):
131 raise ValueError(f"Invalid architecture resources: {name}")
132 arch = backend.get("npu_arch")
133 selected = backend.get("soc_list")
134 if not isinstance(arch, int) or isinstance(arch, bool) or arch <= 0:
135 raise ValueError(f"Invalid SHMEM architecture: {name}")
136 if not isinstance(selected, list) or not selected or any(not isinstance(s, str) for s in selected):
137 raise ValueError(f"Missing SHMEM backend SoCs: {name}")
138 for relative in SHMEM_RESOURCES:
139 path = resource_path(root, f"{backend['shmem_root']}/{relative}")
140 if not path.is_file() or not path.stat().st_size:
141 raise ValueError(f"Missing SHMEM resource: {path}")
144def validate_architectures(root: Path, backends: dict) -> None:
145 """Validate SDK resources and reject overlapping backend identities."""
146 if not isinstance(backends, dict) or not backends:
147 raise ValueError("Missing SHMEM backends")
148 for name, backend in backends.items():
149 validate_architecture(root, name, backend)
150 arches = [backend["npu_arch"] for backend in backends.values()]
151 socs = [soc for backend in backends.values() for soc in backend["soc_list"]]
152 if len(set(arches)) != len(arches) or len(set(socs)) != len(socs):
153 raise ValueError("Duplicate SHMEM backend architecture or SoC")
156def read_manifest(root: Path) -> dict:
157 """Read and validate framework and backend resources before loading any SO."""
158 manifest = json.loads((root / "adapters/manifest.json").read_text(encoding="utf-8"))
159 if manifest.get("schema_version") != 2:
160 raise ValueError("Unsupported native adapter manifest schema")
161 if not manifest.get("python_abi") or not manifest.get("host_arch"):
162 raise ValueError("Missing manifest Python ABI or host architecture")
163 backends = manifest.get("architectures")
164 validate_architectures(root, backends)
165 variants = manifest.get("variants")
166 if not isinstance(variants, list) or not variants:
167 raise ValueError("Adapter manifest must contain at least one variant")
168 ids, keys = set(), set()
169 for variant in variants:
170 identifier, key = variant_selection_key(variant)
171 if identifier in ids or key in keys:
172 raise ValueError("Duplicate adapter ID or ambiguous selection key")
173 ids.add(identifier)
174 keys.add(key)
175 validate_adapter_paths(root, variant, backends)
176 return manifest
179def select_variant(manifest: dict, actual: dict) -> dict:
180 """Select by branch and ABI; external CANN and post versions are diagnostic only."""
181 for key in ("python_abi", "host_arch"):
182 if actual[key] != manifest[key]:
183 raise ValueError(f"Native {key} mismatch: built={manifest[key]}, actual={actual[key]}")
184 key = (release(actual["torch_version"]), release(actual["pta_version"]), actual["torch_cxx11_abi"])
185 matches = [v for v in manifest["variants"]
186 if (v["torch_release"], v["pta_release"], v["torch_cxx11_abi"]) == key]
187 if len(matches) != 1:
188 available = [(v["torch_release"], v["pta_release"], v["torch_cxx11_abi"]) for v in manifest["variants"]]
189 raise ValueError(f"Expected one adapter for {key}; found {len(matches)}; available={available}")
190 return matches[0]
193def select_shmem_backend(manifest: dict, npu_arch: int) -> tuple[str, dict]:
194 """Match a device architecture to exactly one packaged SHMEM backend."""
195 matches = [(name, entry) for name, entry in manifest["architectures"].items()
196 if entry["npu_arch"] == npu_arch]
197 if len(matches) != 1:
198 raise ValueError(f"No packaged SHMEM backend for npu_arch={npu_arch}; "
199 f"available={list(manifest['architectures'])}")
200 return matches[0]
203def validate_build_records(info: dict, manifest: dict):
204 """Match platform and requested adapter IDs to actual build provenance."""
205 if info.get("schema_version") != 3:
206 raise ValueError("Native payload requires schema 3; rebuild with bash build.sh")
207 if any(info.get(k) != manifest[k] for k in ("python_abi", "host_arch")):
208 raise ValueError("Build identity and manifest platform disagree")
209 validate_backend_provenance(info, manifest)
210 records = info.get("variants")
211 if not isinstance(records, list):
212 raise ValueError("Missing adapter build identities")
213 if sorted(v["id"] for v in records) != sorted(v["id"] for v in manifest["variants"]):
214 raise ValueError("Build identity and adapter manifest disagree")
215 if sorted(info.get("requested_target_ids", [])) != sorted(v["id"] for v in records):
216 raise ValueError("Missing requested adapter targets")
217 return records
220def validate_backend_provenance(info: dict, manifest: dict) -> None:
221 """Match the selected SoCs and SDK identities to the runtime index."""
222 backends = manifest["architectures"]
223 built = info.get("architectures", {})
224 if set(built) != set(backends) or any(
225 any(built[name].get(key) != value for key, value in entry.items()) for name, entry in backends.items()):
226 raise ValueError("Build identity and SHMEM backends disagree")
227 socs = [soc for entry in backends.values() for soc in entry["soc_list"]]
228 if sorted(socs) != sorted(info.get("soc_list", [])):
229 raise ValueError("Build identity and backend SoCs disagree")
232def validate_variant_provenance(record: dict, variant: dict):
233 """Match each adapter selector and library path to its actual framework build."""
234 if (release(record["torch_version"]) != variant["torch_release"]
235 or release(record["pta_version"]) != variant["pta_release"]):
236 raise ValueError(f"Adapter branch disagrees with actual build: {variant['id']}")
237 if any(record.get(k) != variant.get(k)
238 for k in ("torch_release", "pta_release", "torch_cxx11_abi", "torch_so", "shmem_so")):
239 raise ValueError(f"Adapter provenance mismatch: {variant['id']}")
242def payload_members(root: Path, info: dict):
243 """Require an exact native-resource set, including files not selected for loading."""
244 recorded = info.get("payload_sha256")
245 if not isinstance(recorded, dict) or not recorded:
246 raise ValueError("Missing complete payload checksum manifest: payload_sha256")
247 actual = {p.relative_to(root).as_posix() for prefix in ("lib", "shmem/lib", "adapters")
248 for p in (root / prefix).rglob("*") if p.is_file()}
249 if actual != set(recorded):
250 raise ValueError(f"Payload file set mismatch: missing={sorted(set(recorded)-actual)}, "
251 f"unrecorded={sorted(actual-set(recorded))}")
252 return recorded, actual
255def validate_resource_hashes(root: Path, info: dict, recorded: dict, actual: set):
256 """Check every recorded resource and require separate native-library digests."""
257 for relative, digest in recorded.items():
258 if file_sha256(resource_path(root, relative)) != digest:
259 raise ValueError(f"Changed payload resource: {relative}")
260 for relative in actual:
261 if relative.endswith(".so") and info.get("native_sha256", {}).get(relative) != recorded[relative]:
262 raise ValueError(f"Missing native checksum: {relative}")
265def check_integrity(root: Path, info: dict, manifest: dict) -> None:
266 """Validate native membership, framework provenance and all recorded resource hashes."""
267 records = validate_build_records(info, manifest)
268 for variant in manifest["variants"]:
269 record = next(v for v in records if v["id"] == variant["id"])
270 validate_variant_provenance(record, variant)
271 recorded, actual = payload_members(root, info)
272 validate_resource_hashes(root, info, recorded, actual)