Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_multicore / _build / artifacts.py: 0%
209 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"""Build-time artifact identity and payload validation; no runtime package import."""
17from __future__ import annotations
18import argparse
19import hashlib
20from importlib import import_module, metadata
21import json
22import os
23from pathlib import Path
24import platform
25import re
26import subprocess
27import sys
28import time
29from contextlib import contextmanager
30from packaging.version import Version
31from packaging.utils import canonicalize_name
32from ._bootstrap import load_module
33from .build_utils import version_label
34from .soc_selection import DEFAULT_SOC_LIST, SOC_ARCHITECTURES, SUPPORTED_SOCS
36COMPONENT_VERSION = "0.1.0.dev0"
37_MANIFEST = load_module("native_manifest")
38REQUIRED_LIBRARIES = (
39 "lib/vendors/hyper_multicore_nn/op_api/lib/libcust_opapi.so",
40)
41_VENDOR_OPERATORS = (
42 "hyper_mega_moe",
43 "hyper_mega_moe_grad",
44 "hyper_mega_mhc",
45 "hyper_mega_mhc_grad",
46 "hyper_mega_gate_route",
47 "hyper_mega_gate_route_grad",
48)
51def artifact_version(pta_version: str) -> str:
52 """Return the component version tagged with the actual PTA distribution version.
54 Args:
55 pta_version: Full installed torch-npu version, preserving post releases.
57 Returns:
58 Component version with a safe local label; the original PTA version remains in metadata.
59 """
60 return str(Version(f"{COMPONENT_VERSION}+pt.{version_label(pta_version)}"))
63def torch_op_plugin_version(torch_root: Path) -> str:
64 """Read the compiled Torch interface version independently of its distribution label."""
65 header = torch_root / "include/torch/csrc/api/include/torch/version.h"
66 contents = header.read_text()
67 if re.search(r"^#include <torch/headeronly/version.h>", contents, re.MULTILINE):
68 header = torch_root / "include/torch/headeronly/version.h"
69 contents = header.read_text()
70 numbers = []
71 for component in ("MAJOR", "MINOR"):
72 values = re.findall(rf"^#define TORCH_VERSION_{component} ([0-9]+)", contents, re.MULTILINE)
73 if len(values) != 1:
74 raise RuntimeError(f"Missing or ambiguous Torch {component} definition in {header}")
75 numbers.append(values[0])
76 return f"V{numbers[0]}R{numbers[1]}"
79def _sha256(path: Path) -> str:
80 digest = hashlib.sha256()
81 with path.open("rb") as source:
82 for chunk in iter(lambda: source.read(1024 * 1024), b""):
83 digest.update(chunk)
84 return digest.hexdigest()
87def _shmem_cpp_test_sources(root: Path) -> list[Path]:
88 """Return SHMEM C++ test sources in a stable order."""
89 return sorted(path for path in (root / "tests/ut/shmem/cpp").glob("*.cpp") if path.is_file())
92def _source_identity(root: Path) -> dict:
93 """Hash build inputs and record the checkout commit."""
94 inputs = {}
95 for directory in ("hyper_multicore", "scripts"):
96 for path in sorted((root / directory).rglob("*")):
97 if path.is_file() and (
98 path.name == "hyper_multicore_build_info"
99 or path.suffix
100 in {".py", ".cpp", ".cc", ".h", ".hpp", ".sh", ".bash", ".patch", ".json", ".ini", ".cmake", ".txt"}
101 ):
102 if any(part in {"lib", "__pycache__"} for part in path.relative_to(root).parts):
103 continue
104 inputs[path.relative_to(root).as_posix()] = _sha256(path)
105 for name in ("setup.py", "MANIFEST.in", "build.sh", "requirements.txt"):
106 inputs[name] = _sha256(root / name)
107 for path in _shmem_cpp_test_sources(root):
108 inputs[path.relative_to(root).as_posix()] = _sha256(path)
109 build_commit = None
110 if (root / ".git").exists():
111 result = subprocess.run(["git", "-C", str(root), "rev-parse", "HEAD"], check=False, capture_output=True)
112 if result.returncode == 0:
113 build_commit = result.stdout.decode().strip()
114 return {
115 "build_commit": build_commit,
116 "source_sha256": hashlib.sha256(json.dumps(inputs, sort_keys=True).encode()).hexdigest(),
117 }
120@contextmanager
121def identity_phase(timings, name):
122 """Diagnostic timings never enter artifact fingerprints."""
123 started = time.monotonic()
124 try:
125 yield
126 finally:
127 timings[name] = timings.get(name, 0.0) + time.monotonic() - started
130def verify_package_versions(expected: dict, timings: dict) -> None:
131 """Check recorded matrix packages without imposing local release requirements."""
132 with identity_phase(timings, "package_versions"):
133 actual = {name: metadata.version(name) for name in expected}
134 if actual != expected:
135 raise RuntimeError(f"Framework environment changed: {actual!r} != {expected!r}")
138def build_identity(root: Path, soc_list: str, *, source=None, timings=None, expected_versions=None) -> dict:
139 """Record the installed framework and actual build inputs without cache certification."""
140 timings = {} if timings is None else timings
141 if sys.version_info[:2] < (3, 10):
142 raise RuntimeError("Native builds require Python >=3.10")
143 verify_package_versions(expected_versions or {}, timings)
144 pta = _MANIFEST.framework_package("torch_npu")
145 with identity_phase(timings, "framework_import"):
146 torch = import_module("torch")
147 with identity_phase(timings, "source_identity"):
148 source = _source_identity(root) if source is None else source
149 info = identity_metadata(root, soc_list, torch, str(getattr(torch, "__version__", "unknown")),
150 pta["version"], source)
151 info["framework_locations"] = {"torch": str(Path(torch.__file__).resolve().parent), "torch_npu": pta["root"]}
152 info["build_fingerprint"] = hashlib.sha256(json.dumps(info, sort_keys=True).encode()).hexdigest()
153 return info
156def cann_identity(cann_root: Path) -> dict:
157 """Record local toolkit facts without requiring a particular version or version file."""
158 cann_version = cann_root / "opp/version.info"
159 contents = cann_version.read_text() if cann_version.is_file() else ""
160 versions = [line.partition("=")[2].strip() for line in contents.splitlines()
161 if line.partition("=")[0].strip() == "Version"]
162 return {
163 "cann_root": str(cann_root.resolve()),
164 "cann_version": ", ".join(versions) or "unknown",
165 "cann_version_info": contents,
166 "cann_version_sha256": _sha256(cann_version) if cann_version.is_file() else None,
167 }
170def identity_metadata(root, soc_list, torch, torch_version, pta_version, source):
171 return {
172 "schema_version": 1,
173 "component_version": COMPONENT_VERSION,
174 "artifact_version": artifact_version(pta_version),
175 "pta_version": pta_version,
176 "torch_version": torch_version,
177 "op_plugin_version": torch_op_plugin_version(Path(torch.__file__).resolve().parent),
178 "torch_commit": getattr(torch.version, "git_version", None),
179 "torch_cxx11_abi": torch.compiled_with_cxx11_abi(),
180 "python_abi": f"cp{sys.version_info.major}{sys.version_info.minor}",
181 "host_arch": platform.machine(),
182 "soc_list": soc_list.split(","),
183 **cann_identity(Path(os.environ["ASCEND_HOME_PATH"])),
184 "dependency_lock_sha256": _sha256(root / "hyper_multicore/_build/dependencies.lock.json"),
185 **source,
186 }
189def _payload_hashes(root: Path) -> dict[str, str]:
190 """Hash every packaged resource except the identity manifest itself."""
191 return {
192 path.relative_to(root).as_posix(): _sha256(path)
193 for path in sorted(root.rglob("*"))
194 if path.is_file() and path != root / "build_info.json"
195 }
198def _kernel_references(value):
199 """Read CANN binary and per-operator index references, including nested binInfo."""
200 if isinstance(value, dict):
201 for key, item in value.items():
202 if key in {"binPath", "jsonPath", "jsonFilePath"}:
203 if not isinstance(item, str) or not item:
204 raise RuntimeError(f"Invalid kernel index reference: {key}={item!r}")
205 yield item
206 else:
207 yield from _kernel_references(item)
208 elif isinstance(value, list):
209 for item in value:
210 yield from _kernel_references(item)
213def _kernel_index_resources(kernel_root: Path, index: Path, device_arch: str) -> list[Path]:
214 """Require every indexed kernel variant and its metadata/object pair."""
215 if not index.is_file():
216 raise RuntimeError(f"Missing native payload resource: {index}")
217 try:
218 contents = json.loads(index.read_text(encoding="utf-8"))
219 except (ValueError, UnicodeError) as error:
220 raise RuntimeError(f"Invalid kernel index: {index}") from error
221 references = list(_kernel_references(contents))
222 if not references:
223 raise RuntimeError(f"Missing kernel index references: {index}")
224 required = []
225 for relative in references:
226 path = kernel_root / relative
227 if Path(relative).parts[0] != device_arch or not path.resolve().is_relative_to(kernel_root.resolve()):
228 raise RuntimeError(f"Kernel index reference outside payload: {relative}")
229 required.append(path)
230 if path.suffix == ".json":
231 required.append(path.with_suffix(".o"))
232 return required
235def _validate_vendor_resources(root: Path, soc_list: list[str], host_arch: str) -> None:
236 """Apply the vendor build's required runtime-resource checks at packaging time."""
237 if (
238 not isinstance(soc_list, list)
239 or not soc_list
240 or any(soc not in SUPPORTED_SOCS for soc in soc_list)
241 ):
242 raise RuntimeError(f"Missing or unsupported payload SoC list: {soc_list}")
243 if host_arch not in {"aarch64", "x86_64"}:
244 raise RuntimeError(f"Missing or unsupported payload Host architecture: {host_arch}")
245 vendor = root / "lib/vendors/hyper_multicore_nn/op_impl/ai_core/tbe"
246 host_libraries = [
247 vendor / "op_tiling/liboptiling.so",
248 vendor / f"op_tiling/lib/linux/{host_arch}/libcust_opmaster_rt2.0.so",
249 root / f"lib/vendors/hyper_multicore_nn/op_proto/lib/linux/{host_arch}/libcust_opsproto_rt2.0.so",
250 ]
251 for soc in soc_list:
252 indexes = [vendor / f"kernel/config/{soc}/binary_info_config.json"]
253 required = [*host_libraries, vendor / f"config/{soc}/aic-{soc}-ops-info.json", *indexes]
254 for operator in _VENDOR_OPERATORS:
255 index = vendor / f"kernel/config/{soc}/{operator}.json"
256 required.append(index)
257 indexes.append(index)
258 for suffix in (".o", ".json"):
259 matches = sorted((vendor / f"kernel/{SOC_ARCHITECTURES[soc]}/{operator}").glob(f"*{suffix}"))
260 if not matches:
261 raise RuntimeError(f"Missing native payload resource: {soc}/{operator}/*{suffix}")
262 required.extend(matches)
263 for index in indexes:
264 required.extend(_kernel_index_resources(vendor / "kernel", index, SOC_ARCHITECTURES[soc]))
265 for path in required:
266 if not path.is_file() or path.stat().st_size == 0:
267 raise RuntimeError(f"Missing or empty native payload resource: {path.relative_to(root)}")
270def validate_payload(root: Path, info: dict) -> None:
271 """Require the complete native payload and its recorded file hashes.
273 Args:
274 root: Prepared hyper_multicore payload directory.
275 info: Its generated build_info object.
277 Raises:
278 RuntimeError: If identity, required files or native checksums are invalid.
279 """
280 for relative in (*REQUIRED_LIBRARIES, "lib/set_env.bash"):
281 path = root / relative
282 if not path.is_file() or path.stat().st_size == 0:
283 raise RuntimeError(f"Missing native payload file: {relative}")
284 if (root / "lib/framework").exists() or (root / "shmem/lib/framework").exists():
285 raise RuntimeError("Expected only the shared Torch adapter in each variant; stale framework layout")
286 _validate_vendor_resources(root, info.get("soc_list"), info.get("host_arch"))
287 try:
288 manifest = _MANIFEST.read_manifest(root)
289 _MANIFEST.check_integrity(root, info, manifest)
290 except (OSError, ValueError, KeyError) as error:
291 raise RuntimeError(str(error)) from error
292 expected_version = (
293 COMPONENT_VERSION if info.get("build_mode") == "all" else artifact_version(info["variants"][0]["pta_version"])
294 )
295 if info["artifact_version"] != expected_version:
296 raise RuntimeError("Payload version does not match its framework matrix")
297 if info.get("build_mode") not in ("current", "all"):
298 raise RuntimeError("Unknown payload build mode")
299 if info["build_mode"] == "current" and len(manifest["variants"]) != 1:
300 raise RuntimeError("Current-environment payload must contain one framework variant")
301 _validate_payload_hashes(root, info)
304def _validate_payload_hashes(root: Path, info: dict):
305 """Verify the complete file set and contents independently of layout validation."""
306 recorded = info.get("payload_sha256")
307 if not isinstance(recorded, dict) or not recorded:
308 raise RuntimeError("Missing complete payload checksum manifest: payload_sha256")
309 actual = _payload_hashes(root)
310 missing = sorted(recorded.keys() - actual.keys())
311 unrecorded = sorted(actual.keys() - recorded.keys())
312 if missing or unrecorded:
313 raise RuntimeError(f"Payload file set mismatch: missing={missing}, unrecorded={unrecorded}")
314 for relative, digest in actual.items():
315 if recorded[relative] != digest:
316 raise RuntimeError(f"Changed payload resource: {relative}")
319def main() -> None:
320 """Prepare identity before compilation, finalize its resources, or validate a payload."""
321 parser = argparse.ArgumentParser(description=__doc__)
322 parser.add_argument("--output", required=True, type=Path)
323 parser.add_argument("--soc-list", default=DEFAULT_SOC_LIST)
324 action = parser.add_mutually_exclusive_group(required=True)
325 action.add_argument("--prepare", action="store_true")
326 action.add_argument("--finalize", type=Path, metavar="IDENTITY_JSON")
327 action.add_argument("--validate", action="store_true")
328 args = parser.parse_args()
329 if args.prepare:
330 info = build_identity(Path(__file__).resolve().parents[2], args.soc_list)
331 args.output.parent.mkdir(parents=True, exist_ok=True)
332 args.output.write_text(json.dumps(info, indent=2) + "\n", encoding="utf-8")
333 print(info["build_fingerprint"])
334 elif args.finalize:
335 info = json.loads(args.finalize.read_text(encoding="utf-8"))
336 info["native_sha256"] = {p.relative_to(args.output).as_posix(): _sha256(p) for p in args.output.rglob("*.so")}
337 info["payload_sha256"] = _payload_hashes(args.output)
338 validate_payload(args.output, info)
339 (args.output / "build_info.json").write_text(json.dumps(info, indent=2) + "\n", encoding="utf-8")
340 else:
341 validate_payload(args.output, json.loads((args.output / "build_info.json").read_text(encoding="utf-8")))
344def installed_packages():
345 """Observe actual distributions for provenance, without enforcing a dependency lock."""
346 return sorted([{"name": canonicalize_name(d.metadata["Name"]), "version": d.version}
347 for d in metadata.distributions() if d.metadata["Name"]], key=lambda p: (p["name"], p["version"]))
350if __name__ == "__main__":
351 main()