Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_multicore / _build / build_controller.py: 0%

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 shared resources and package framework variants with selected SHMEM backends.""" 

16 

17from __future__ import annotations 

18import argparse 

19from concurrent.futures import ThreadPoolExecutor, as_completed 

20from copy import copy 

21from queue import Queue, Empty 

22import re 

23import json 

24import os 

25from pathlib import Path 

26import platform 

27import shutil 

28import shlex 

29import signal 

30import sys 

31import threading 

32import time 

33from . import artifacts as HELPERS, framework_environments as FRAMEWORK 

34from . import vendor_cache as VENDOR 

35from . import command_runner as CONSOLE, build_workspace as WORKSPACE 

36from ._bootstrap import load_module 

37from .command_runner import message 

38from .build_utils import atomic_json as write_json, measured, version_label 

39from .build_state import BuildContext, BuildResult, PreparedFramework 

40from .artifacts import _payload_hashes, _source_identity 

41from .soc_selection import ARCHITECTURES, DEFAULT_SOC_LIST, resolve_soc_selection 

42from .build_publication import SourceGuard, publish_wheel, publish_payload, validate_elf, validate_wheel 

43 

44ROOT = Path(__file__).resolve().parents[2] 

45NATIVE = ROOT / "build/native" 

46 

47MANIFEST = load_module("native_manifest") 

48 

49def probe_identity(python: Path, socs: str, run: BuildContext, *, source=None, expected=None) -> dict: 

50 """Collect framework identity and installed versions in the target process.""" 

51 code = ( 

52 "import json,sys; from pathlib import Path; sys.path.insert(0,str(Path(sys.argv[1]).parent)); " 

53 "from _hyper_multicore_bootstrap import load_module; h=load_module('artifacts'); timings={}; " 

54 "info=h.build_identity(Path(sys.argv[2]),sys.argv[3],source=json.loads(sys.argv[4])," 

55 "timings=timings,expected_versions=json.loads(sys.argv[5])); " 

56 "print(json.dumps({'identity':info,'timings':timings,'packages':h.installed_packages()}))" 

57 ) 

58 result = run.commands( 

59 [ 

60 str(python), 

61 "-c", 

62 code, 

63 str(ROOT / "hyper_multicore/_build/artifacts.py"), 

64 str(ROOT), 

65 socs, 

66 json.dumps(source), 

67 json.dumps(expected or {}), 

68 ], 

69 label=f"probe-{python.parent.parent.name}", 

70 env=FRAMEWORK.build_environment(python, isolated=run.isolated), 

71 capture=True, 

72 ) 

73 data = json.loads(result) 

74 run.frameworks[str(python)] = PreparedFramework(python, data["identity"], data["packages"]) 

75 run.events[-1]["identity_timings"] = data["timings"] 

76 write_json(run.directory / "stages.json", run.events) 

77 identity = data["identity"] 

78 identity["torch_release"] = MANIFEST.release(identity["torch_version"]) 

79 identity["pta_release"] = MANIFEST.release(identity["pta_version"]) 

80 return identity 

81 

82 

83def adapter_environment(python: Path, *, isolated=False) -> dict: 

84 """Allow newer framework compilers without changing the CANN vendor toolchain.""" 

85 env = FRAMEWORK.build_environment(python, isolated=isolated) 

86 if "HYPER_MULTICORE_FRAMEWORK_CXX" in env: 

87 env["CXX"] = env["HYPER_MULTICORE_FRAMEWORK_CXX"] 

88 return env 

89 

90 

91def build_framework(python: Path, identity: dict, common: Path, sdks: dict[str, dict], 

92 args: argparse.Namespace, run: BuildContext) -> tuple[Path, dict]: 

93 """Build one shared Torch adapter and the selected SHMEM bindings in one workspace.""" 

94 directory = NATIVE / "variants" / WORKSPACE.framework_slot(identity) 

95 WORKSPACE.reset_workspace(directory, ROOT) 

96 identifier = ("pt" + version_label(identity["torch_version"]).removesuffix(".cpu") 

97 + "-pta" + version_label(identity["pta_version"]) + "-" + identity["build_fingerprint"][:12]) 

98 paths = {"torch_so": f"adapters/{identifier}/libhyper_multicore_torch.so", 

99 "shmem_so": {name: f"adapters/{identifier}/shmem/{name}/hyper_multicore_shmem_torch.so" 

100 for name in sdks}} 

101 payload = directory / "payload" 

102 payload.mkdir() 

103 env = adapter_environment(python, isolated=run.isolated) 

104 compile_adapter(python, directory / "torch", payload, "torch", 

105 [f"-DHP_MULTICORE_VENDOR_ROOT={common / 'lib/vendors/hyper_multicore_nn'}"], 

106 args, run, env, identifier) 

107 validate_elf(payload / "libhyper_multicore_torch.so", identity["host_arch"], run.commands, env) 

108 for backend, sdk in sdks.items(): 

109 selected_env = shmem_environment(env, common, sdk) 

110 output = payload / "shmem" / backend 

111 compile_adapter(python, directory / "shmem" / backend, output, "shmem", 

112 [f"-DSHMEM_OPS_LIB={common / sdk['shmem_root'] / 'libhyper_multicore_shmem_ops.so'}", 

113 f"-DHP_SHMEM_SOC_TYPE={sdk['sdk_soc_type']}"], 

114 args, run, selected_env, identifier + "-" + backend) 

115 with run.commands.scope("Load"): 

116 load_backend(python, payload, common, sdk, backend, run, selected_env, identifier) 

117 validate_elf(output / "hyper_multicore_shmem_torch.so", identity["host_arch"], run.commands, selected_env) 

118 record = {**identity, **paths, "id": identifier, 

119 "native_build_identity": {k: identity[k] for k in ("build_commit", "source_sha256", "build_fingerprint")}} 

120 write_json(directory / "identity.json", record) 

121 run.commands.add_event({"stage": "adapter", "pta_version": identity["pta_version"], 

122 "architectures": list(sdks), "compiled": True, "compiler_jobs": args.jobs}) 

123 return payload, record 

124 

125 

126def shmem_environment(env: dict[str, str], common: Path, sdk: dict) -> dict[str, str]: 

127 """Use one backend's exact SDK for compilation and ELF dependency checks.""" 

128 selected = env.copy() 

129 root = common / sdk["shmem_root"] 

130 selected.update(SHMEM_HOME_PATH=sdk["install_root"], SHMEM_SOURCE_DIR=str(Path(sdk["install_root"]) / "shmem")) 

131 selected["LD_LIBRARY_PATH"] = os.pathsep.join( 

132 [str(root), str(root / "shmem"), selected.get("LD_LIBRARY_PATH", "")]) 

133 return selected 

134 

135 

136def compile_adapter(python: Path, work: Path, output: Path, kind: str, extra: list[str], 

137 args: argparse.Namespace, run: BuildContext, env: dict[str, str], identifier: str) -> None: 

138 """Compile, test and install one real framework adapter.""" 

139 source = "hyper_multicore/shmem/ccsrc" if kind == "shmem" else "hyper_multicore/torch" 

140 with run.commands.scope(kind.title()): 

141 run.commands(["cmake", "-S", str(ROOT / source), "-B", str(work), 

142 "-DCMAKE_BUILD_TYPE=Release", "-DCMAKE_EXPORT_COMPILE_COMMANDS=ON", 

143 f"-DPython3_EXECUTABLE={python}", f"-DCMAKE_INSTALL_PREFIX={output}", *extra], 

144 label=f"configure-{identifier}-{kind}", env=env) 

145 run.commands(["cmake", "--build", str(work), "--parallel", str(args.jobs), "--verbose"], 

146 label=f"compile-{identifier}-{kind}", env=env) 

147 if kind == "shmem": 

148 run.commands(["ctest", "--test-dir", str(work), "--output-on-failure"], 

149 label=f"ctest-{identifier}", env=env) 

150 run.commands(["cmake", "--install", str(work)], label=f"install-adapter-{identifier}-{kind}", env=env) 

151 

152 

153def load_backend(python: Path, payload: Path, common: Path, sdk: dict, backend: str, 

154 run: BuildContext, env: dict[str, str], identifier: str) -> None: 

155 """Validate each dependency chain in a separate process without device initialization.""" 

156 code = ( 

157 "import ctypes,importlib.util,sys,torch,torch_npu; " 

158 "handles=[ctypes.CDLL(path,mode=ctypes.RTLD_LOCAL) for path in sys.argv[1:4]]; " 

159 "torch.ops.load_library(sys.argv[4]); " 

160 "spec=importlib.util.spec_from_file_location('hyper_multicore_shmem_torch',sys.argv[5]); " 

161 "module=importlib.util.module_from_spec(spec); spec.loader.exec_module(module); " 

162 "assert module._debug_state()['backend_soc_type']==sys.argv[6], 'SHMEM backend mismatch'" 

163 ) 

164 root = common / sdk["shmem_root"] 

165 run.commands([str(python), "-c", code, *[str(root / path) for path in MANIFEST.SHMEM_DEPENDENCIES], 

166 str(payload / "libhyper_multicore_torch.so"), 

167 str(payload / "shmem" / backend / "hyper_multicore_shmem_torch.so"), sdk["sdk_soc_type"]], 

168 label=f"load-{identifier}-{backend}", env=env) 

169 

170 

171def validate_toolchain() -> None: 

172 """Reject unavailable common build inputs before preparing any framework environment.""" 

173 cann = os.environ.get("ASCEND_HOME_PATH", "") 

174 if not cann or not Path(cann).is_dir(): 

175 raise ValueError("ASCEND_HOME_PATH must identify a toolkit directory; " 

176 "source the CANN set_env.sh before building") 

177 identity = HELPERS.cann_identity(Path(cann)) 

178 message(f"CANN: {identity['cann_version']}; toolkit {identity['cann_root']}") 

179 required = ("cmake", "ctest", "gcc", "g++", "make", "git", "bisheng", "asc_opc", "readelf", "ldd", "ld") 

180 missing = [name for name in required if shutil.which(name) is None] 

181 if missing: 

182 raise ValueError(f"Missing native build tools: {missing}") 

183 

184 

185def build(args, publication) -> Path: 

186 """Publish only unchanged source builds; maintenance cannot undo a commit.""" 

187 run = BuildContext(ROOT, NATIVE) 

188 transaction = SourceGuard(ROOT) 

189 result = _build(args, run, transaction) 

190 transaction.check_source() 

191 publication.committed = True 

192 finish_maintenance(args, run, result) 

193 return result.payload 

194 

195 

196def _build(args, run: BuildContext, transaction: SourceGuard) -> BuildResult: 

197 """Compile shared resources and each framework variant from fresh workspaces.""" 

198 started = time.monotonic() 

199 if args.tests: 

200 checks = ROOT / "tests/ut/build/check_build.py" 

201 if not checks.is_file(): 

202 raise FileNotFoundError(f"Build source checks are missing: {checks}; use a complete source checkout") 

203 with measured(run.events, "build-source-checks"): 

204 run.commands([sys.executable, str(checks)], label="build-source-checks") 

205 validate_toolchain() 

206 run_id = run.run_id 

207 with measured(run.events, "framework-preparation"): 

208 _, pythons, identities = prepare_frameworks(args, run) 

209 with measured(run.events, "common-compilation"): 

210 common, sdk = build_common(args, run, Path(sys.executable).absolute()) 

211 transaction.check_source() 

212 with measured(run.events, "adapter-phase"): 

213 outputs, records = build_adapters(args, run, pythons, identities, common, sdk, transaction) 

214 with measured(run.events, "payload-assembly-validation"): 

215 payload, info = assemble_payload(args, run_id, identities, outputs, records, common, sdk) 

216 if run.vendor_requests: 

217 info["vendor_sources"] = [{k: r[k] for k in ("soc", "origin", "reused")} for r in run.vendor_requests] 

218 write_json(payload / "build_info.json", info) 

219 wheel = None if args.native_only else package_wheel(pythons[0], run, payload, info, transaction) 

220 transaction.check_source() 

221 if wheel is not None: 

222 destination = ROOT / "dist" / wheel.name 

223 publish_wheel(wheel, destination) 

224 message(f"Wheel: {destination}") 

225 message(f"SHA256: {MANIFEST.file_sha256(destination)}") 

226 selected = publish_build_payload(payload, run_id, transaction) 

227 transaction.check_source() 

228 message(f"Published payload: {selected}; PTA adapters: {', '.join(r['pta_version'] for r in records)}") 

229 return BuildResult(selected, pythons, len(records), started) 

230 

231 

232def finish_maintenance(args, run: BuildContext, result: BuildResult): 

233 """The artifact transaction committed; reclamation can no longer invalidate its result.""" 

234 pythons, frameworks, started = result.pythons, result.frameworks, result.started 

235 maintenance_started = time.monotonic() 

236 try: 

237 VENDOR.publish(ROOT, run.vendor_requests) 

238 run.events.extend({"stage": "vendor-cache-publication", "soc": r["soc"], 

239 "elapsed_seconds": r["publish_seconds"]} 

240 for r in run.vendor_requests if "publish_seconds" in r) 

241 if args.all: 

242 removed = FRAMEWORK.prune_environments(NATIVE, pythons) 

243 run.events.append({"stage": "prune-framework-environments", "removed": removed}) 

244 WORKSPACE.clear_temporary_outputs(NATIVE) 

245 except (OSError, ValueError, RuntimeError, KeyboardInterrupt) as error: 

246 detail = str(error) or type(error).__name__ 

247 run.events.append({"stage": "build-maintenance", "status": "warning", "error": detail}) 

248 message(f"Build committed; maintenance stopped: {detail}. The new artifact is valid and retained.", "WARN") 

249 run.events.append({"stage": "maintenance", "elapsed_seconds": time.monotonic() - maintenance_started}) 

250 try: 

251 write_json(run.directory / "stages.json", run.events) 

252 except (OSError, ValueError, RuntimeError, KeyboardInterrupt) as error: 

253 message(f"Build committed; unable to finish maintenance records: {error}", "WARN") 

254 build_progress_summary(args, run, frameworks, time.monotonic() - started) 

255 

256 

257def prepare_frameworks(args, run: BuildContext): 

258 """Only matrix mode selects exact pairs; local builds use the caller's environment.""" 

259 run.isolated = args.all 

260 config = ROOT / "hyper_multicore/_build/frameworks.lock.json" 

261 targets = FRAMEWORK.load_targets(config) if args.all else [] 

262 base = Path(sys.executable).absolute() 

263 mode = f"full framework matrix: {len(targets)} targets" if args.all else "locally installed framework" 

264 message(f"Building {mode}; Python cp{sys.version_info.major}{sys.version_info.minor}, " 

265 f"{platform.machine()}, SoC {args.soc_list}, jobs {args.jobs}") 

266 source = _source_identity(ROOT) 

267 if args.all: 

268 reference = json.loads(config.read_text()).get("reference_cann") 

269 report_cann_reference(reference, HELPERS.cann_identity(Path(os.environ["ASCEND_HOME_PATH"]))["cann_version"]) 

270 pythons, identities = prepare_matrix_environments(targets, run, base, args.soc_list, source) 

271 for python, identity in zip(pythons, identities): 

272 identity["reference_cann"] = reference 

273 environment = json.loads((python.parent.parent / "ready.json").read_text()) 

274 identity["framework_environment"] = { 

275 "request": environment["request"], 

276 "installed_packages": run.frameworks[str(python)].packages, 

277 "installation": environment.get("installation"), 

278 } 

279 else: 

280 report_local_framework(base) 

281 pythons = [base] 

282 identities = [probe_identity(base, args.soc_list, run, source=source)] 

283 return targets, pythons, identities 

284 

285 

286def report_cann_reference(reference, actual): 

287 """The matrix reference is informational, never an environment admission gate.""" 

288 if reference: 

289 message(f"CANN reference: {reference}; actual: {actual}") 

290 if actual != reference: 

291 message("CANN differs from the matrix reference or is unknown; " 

292 "continuing with the installed toolkit", "WARN") 

293 else: 

294 message(f"CANN actual: {actual}") 

295 

296 

297def report_local_framework(python: Path): 

298 """Show installed framework facts before capability probes, without importing them.""" 

299 message("Using locally installed framework; no environment creation or installation") 

300 message(f" {'Python':<14}: {python}") 

301 for label, name in (("Torch", "torch"), ("torch_npu", "torch_npu")): 

302 package = MANIFEST.framework_package(name) 

303 message(f" {label:<14}: {package['version']}") 

304 message(f" {label + ' path':<14}: {package['root']}") 

305 

306 

307def prepare_matrix_environments(targets, run, base, socs, source): 

308 """Prepare and verify each target once, including its actual installed package versions.""" 

309 message(f"Framework environment storage: {FRAMEWORK.environment_directory(NATIVE)}") 

310 pythons, identities = [], [] 

311 

312 def collect(python, expected): 

313 identities.append(probe_identity(python, socs, run, source=source, expected=expected)) 

314 return run.frameworks[str(python)].packages 

315 

316 for index, target in enumerate(targets, 1): 

317 message( 

318 f"Environment {index}/{len(targets)}: Torch {target['torch_version']}, PTA {target['pta_build_version']}" 

319 ) 

320 pythons.append(FRAMEWORK.prepare_environment(target, NATIVE, run.commands, base, identity_probe=collect)) 

321 return pythons, identities 

322 

323 

324def build_progress_summary(args, run, frameworks: int, elapsed: float): 

325 """Report actual work instead of cross-build cache decisions.""" 

326 message(f"Build succeeded in {elapsed:.1f}s; built common resources and {frameworks} framework variants") 

327 if run.vendor_requests: 

328 reused = sum(r["reused"] for r in run.vendor_requests) 

329 message(f"Vendor snapshots: reused {reused}, compiled {len(run.vendor_requests) - reused}; SDK/ops rebuilt") 

330 if args.all: 

331 created = sum(event.get("stage", "").startswith("venv-") for event in run.events) 

332 message(f"Framework environments: created {created}, reused {frameworks - created}") 

333 message(f"Build log: {run.directory / 'build.log'}") 

334 

335 

336def build_common(args, run: BuildContext, python: Path): 

337 """Use caller CANN tools for SDK/ops and cached vendor resources in either mode.""" 

338 directory = NATIVE / "common" / WORKSPACE.safe_component(platform.machine()) 

339 WORKSPACE.reset_workspace(directory, ROOT) 

340 common = directory / "payload" 

341 env = FRAMEWORK.cann_tool_environment(python, isolated=False) 

342 message(f"Common CANN tools: Python {python}; using caller environment") 

343 env.update(PYTHON_BIN=str(python), HP_NATIVE_STAGE="common", HP_COMPONENT_ROOT=str(common), 

344 HP_NATIVE_WORK_ROOT=str(directory / "work"), HP_SHMEM_WORK_ROOT=str(NATIVE / "sdk" / platform.machine()), 

345 HP_SHMEM_HANDOFF=str(directory / "sdk.json")) 

346 origin = _source_identity(ROOT) 

347 env.update(HP_VENDOR_CACHE_ROOT=str(NATIVE), HP_VENDOR_ORIGIN=json.dumps(origin)) 

348 if getattr(args, "clean", False): 

349 env["HP_VENDOR_CACHE_CLEAN"] = "1" 

350 else: 

351 env.pop("HP_VENDOR_CACHE_CLEAN", None) 

352 command = ["bash", str(ROOT / "hyper_multicore/build.sh"), "--stage-only", 

353 "--soc-list", args.soc_list] 

354 prepared = json.loads(run.commands(command + ["--phase", "prepare", "--jobs", str(args.jobs)], 

355 label="common-prepare", env=env, capture=True)) 

356 handoff = json.loads((directory / "sdk.json").read_text()) 

357 sdks = handoff["architectures"] 

358 if set(sdks) != set(args.architectures): 

359 raise ValueError("SHMEM SDK backends do not match the requested SoCs") 

360 for name, socs in args.architectures.items(): 

361 sdk = sdks[name] 

362 if sdk["sdk_soc_type"] != ARCHITECTURES[name]["sdk_soc_type"] or sdk["soc_list"] != socs: 

363 raise ValueError(f"SHMEM SDK backend {name} does not match {socs}") 

364 sdk.update(**ARCHITECTURES[name], shmem_root=f"shmem/lib/{name}", 

365 kernel_root=f"lib/vendors/hyper_multicore_nn/op_impl/ai_core/tbe/kernel/{name}") 

366 env.update(HP_OPS_NN_SOURCE=prepared["ops_nn_source"], 

367 HP_OPS_TRANSFORMER_SOURCE=prepared["ops_transformer_source"], 

368 HP_VENDOR_HOST_IDENTITY=prepared["host_input_identity"]) 

369 build_vendors(command, prepared["pending"], args.jobs, run, env) 

370 run.commands(command + ["--phase", "merge"], label="common-merge", env=env) 

371 run.vendor_requests = [json.loads(p.read_text()) 

372 for p in sorted((directory / "work").glob("vendor-request-*.json"))] 

373 for request in run.vendor_requests: 

374 run.events.append({"stage": "vendor", **{k: request[k] for k in 

375 ("soc", "reused", "lookup_seconds", "compile_seconds") if k in request}}) 

376 run.events.append({"stage": "common", "compiled": True, "socs": args.soc_list}) 

377 return common, sdks 

378 

379 

380def compiler_budgets(count: int, jobs: int) -> list[int]: 

381 """Use serial, two-slot or all-target scheduling without exceeding the compile budget.""" 

382 workers = count if jobs >= 2 * count else (2 if jobs >= 4 else 1) 

383 workers = min(workers, count) 

384 quotient, remainder = divmod(jobs, workers) 

385 return [quotient + (slot < remainder) for slot in range(workers)] 

386 

387 

388def build_vendors(command: list[str], pending: list[str], jobs: int, 

389 run: BuildContext, env: dict[str, str]) -> None: 

390 """Compile cache misses within one job budget; merge only after every task succeeds.""" 

391 if not pending: 

392 return 

393 budgets = compiler_budgets(len(pending), jobs) 

394 run.commands.add_event({"stage": "vendor-schedule", "architectures": pending, 

395 "worker_jobs": budgets, "total_jobs": jobs}) 

396 

397 def compile_slot(architectures: list[str], budget: int) -> None: 

398 """Execute assigned architectures with an independent environment and fixed quota.""" 

399 for arch in architectures: 

400 with run.commands.scope(f"{arch} / Vendor"): 

401 run.commands(command + ["--phase", "vendor", "--arch", arch, "--jobs", str(budget)], 

402 label=f"vendor-{arch}", env=env.copy()) 

403 

404 if len(budgets) == 1: 

405 compile_slot(pending, budgets[0]) 

406 return 

407 pool = ThreadPoolExecutor(max_workers=len(budgets), thread_name_prefix="native-vendor") 

408 try: 

409 futures = [pool.submit(compile_slot, pending[slot::len(budgets)], budget) 

410 for slot, budget in enumerate(budgets)] 

411 for future in as_completed(futures): 

412 future.result() 

413 except BaseException as error: 

414 run.commands.cancel(error) 

415 raise run.commands.failure or error 

416 finally: 

417 pool.shutdown(wait=True, cancel_futures=True) 

418 

419 

420def build_adapter(index, count, python, identity, jobs, args, run, common, sdk, transaction): 

421 """Build one framework variant with private arguments and thread-local diagnostic labels.""" 

422 selected = copy(args) 

423 selected.jobs = jobs 

424 started, wall_started = time.monotonic(), time.time() 

425 error = None 

426 try: 

427 with run.commands.scope(f"PTA {identity['pta_version']}"): 

428 message(f"Adapter {index + 1}/{count}: Torch {identity['torch_version']}, " 

429 f"PTA {identity['pta_version']}; compiler jobs {jobs}") 

430 transaction.check_source() 

431 try: 

432 return build_framework(python, identity, common, sdk, selected, run) 

433 except CONSOLE.CommandCancelled: 

434 raise 

435 except RuntimeError as cause: 

436 raise RuntimeError(framework_failure(identity, args, python, cause)) from cause 

437 except BaseException as caught: 

438 error = caught 

439 raise 

440 finally: 

441 run.commands.add_event({"stage": f"adapter-compilation-{identity['pta_version']}", 

442 "elapsed_seconds": time.monotonic() - started, 

443 "started_at": wall_started, "ended_at": time.time(), 

444 "status": CONSOLE.command_status(0, error), "compiler_jobs": jobs}) 

445 

446 

447def parallel_adapters(args, run, pythons, identities, common, sdk, transaction, budgets): 

448 """Bind quotas to fixed workers so out-of-order completion cannot oversubscribe jobs.""" 

449 targets = Queue() 

450 for index, (python, identity) in enumerate(zip(pythons, identities)): 

451 targets.put((index, python, identity)) 

452 results = [None] * len(identities) 

453 progress = threading.Lock() 

454 completed = 0 

455 

456 def slot(jobs): 

457 nonlocal completed 

458 while not run.commands.cancelled.is_set(): 

459 try: 

460 index, python, identity = targets.get_nowait() 

461 except Empty: 

462 return 

463 try: 

464 result = build_adapter(index, len(identities), python, identity, jobs, 

465 args, run, common, sdk, transaction) 

466 with progress: 

467 results[index] = result 

468 completed += 1 

469 message(f"Adapters completed {completed}/{len(identities)}; PTA {identity['pta_version']}") 

470 except BaseException as error: 

471 run.commands.cancel(error) 

472 raise 

473 finally: 

474 targets.task_done() 

475 

476 pool = ThreadPoolExecutor(max_workers=len(budgets), thread_name_prefix="native-adapter") 

477 try: 

478 futures = [pool.submit(slot, jobs) for jobs in budgets] 

479 for future in as_completed(futures): 

480 future.result() 

481 except BaseException as caught: 

482 run.commands.cancel(caught) 

483 raise run.commands.failure or caught 

484 finally: 

485 pool.shutdown(wait=True, cancel_futures=True) 

486 return results 

487 

488 

489def build_adapters(args, run: BuildContext, pythons, identities, common, sdk, transaction: SourceGuard): 

490 """Compile framework variants within the shared job budget and preserve matrix order.""" 

491 count = len(identities) 

492 budgets = compiler_budgets(count, args.jobs) if args.all else [args.jobs] 

493 message(f"Adapter schedule: {len(budgets)} concurrent framework variants; " 

494 f"compiler jobs per slot {', '.join(map(str, budgets))}; total budget {args.jobs}") 

495 run.commands.add_event({"stage": "adapter-schedule", "workers": len(budgets), 

496 "worker_jobs": budgets, "total_jobs": args.jobs, "targets": count}) 

497 if len(budgets) == 1: 

498 results = [build_adapter(index, count, python, identity, args.jobs, args, run, common, sdk, transaction) 

499 for index, (python, identity) in enumerate(zip(pythons, identities))] 

500 else: 

501 results = parallel_adapters(args, run, pythons, identities, common, sdk, transaction, budgets) 

502 outputs, records = zip(*results) 

503 return list(outputs), list(records) 

504 

505 

506def assemble_payload(args, run_id, identities, outputs, records, common, architectures): 

507 """Assemble and validate one selected payload without changing runtime registration policy.""" 

508 staging = NATIVE / "staging" / run_id 

509 payload = staging / "hyper_multicore" 

510 shutil.copytree(common, payload) 

511 for output, record in zip(outputs, records): 

512 shutil.copytree(output, payload / "adapters" / record["id"]) 

513 manifest = { 

514 "schema_version": 2, 

515 "architectures": { 

516 name: {k: sdk[k] for k in ("shmem_root", "kernel_root", "npu_arch", "sdk_soc_type", "soc_list")} 

517 for name, sdk in architectures.items()}, 

518 "python_abi": identities[0]["python_abi"], 

519 "host_arch": identities[0]["host_arch"], 

520 "variants": [ 

521 {k: record[k] for k in ("id", "torch_release", "pta_release", "torch_cxx11_abi", "torch_so", "shmem_so")} 

522 for record in records 

523 ], 

524 "validated_runtime_pairs": [], 

525 "known_runtime_issues": [], 

526 } 

527 write_json(payload / "adapters/manifest.json", manifest) 

528 info = { 

529 **identities[0], 

530 "schema_version": 3, 

531 "architectures": architectures, 

532 "build_mode": "all" if args.all else "current", 

533 "variants": records, 

534 "artifact_version": ( 

535 HELPERS.COMPONENT_VERSION if args.all else HELPERS.artifact_version(identities[0]["pta_version"]) 

536 ), 

537 "requested_target_ids": [v["id"] for v in records], 

538 "reference_cann": identities[0].get("reference_cann"), 

539 "framework_lock_sha256": ( 

540 MANIFEST.file_sha256(ROOT / "hyper_multicore/_build/frameworks.lock.json") if args.all else None 

541 ), 

542 } 

543 info["native_sha256"] = {p.relative_to(payload).as_posix(): MANIFEST.file_sha256(p) for p in payload.rglob("*.so")} 

544 info["payload_sha256"] = _payload_hashes(payload) 

545 HELPERS.validate_payload(payload, info) 

546 write_json(payload / "build_info.json", info) 

547 return payload, info 

548 

549 

550def package_wheel(python: Path, run: BuildContext, payload: Path, info: dict, transaction: SourceGuard): 

551 """Validate the complete wheel and source state before atomically publishing its path.""" 

552 env = FRAMEWORK.build_environment(python, isolated=run.isolated) 

553 env["HYPER_MULTICORE_NATIVE_OUTPUT_ROOT"] = str(payload) 

554 wheel_dir = payload.parent / "wheel" 

555 run.commands( 

556 [ 

557 str(python), 

558 "setup.py", 

559 "-q", 

560 "build", 

561 "--build-base", 

562 str(payload.parent / "wheel-build"), 

563 "bdist_wheel", 

564 "--bdist-dir", 

565 str(payload.parent / "bdist"), 

566 "--dist-dir", 

567 str(wheel_dir), 

568 ], 

569 label="package-wheel", 

570 env=env, 

571 ) 

572 wheels = list(wheel_dir.glob("*.whl")) 

573 if len(wheels) != 1: 

574 raise RuntimeError("Expected exactly one native wheel") 

575 with measured(run.events, "wheel-integrity-validation"): 

576 validate_wheel(wheels[0], payload, info) 

577 transaction.check_source() 

578 return wheels[0] 

579 

580 

581def publish_build_payload(payload: Path, run_id: str, transaction: SourceGuard) -> Path: 

582 """Expose a complete selected payload after the final source check.""" 

583 selected = NATIVE / "payload/hyper_multicore" 

584 selected.parent.mkdir(parents=True, exist_ok=True) 

585 candidate = selected.with_name(f".payload-{run_id}") 

586 shutil.copytree(payload, candidate) 

587 transaction.check_source() 

588 publish_payload(candidate, selected) 

589 return selected 

590 

591 

592def parse_arguments(argv=None): 

593 """Resolve source-check defaults without importing frameworks or reading the matrix.""" 

594 parser = argparse.ArgumentParser(description=__doc__) 

595 parser.add_argument("--all", action="store_true", 

596 help="Build every configured framework target; adapter concurrency follows --jobs") 

597 parser.add_argument("--soc-list", default=DEFAULT_SOC_LIST) 

598 parser.add_argument("--jobs", type=int, default=os.cpu_count(), 

599 help="Compiler job budget, shared by concurrent adapters") 

600 parser.add_argument("--clean", action="store_true") 

601 parser.add_argument("--tests", choices=("on", "off"), 

602 help="Run lightweight build source checks (default: on with --all, otherwise off)") 

603 parser.add_argument("--native-only", action="store_true", help=argparse.SUPPRESS) 

604 args = parser.parse_args(argv) 

605 args.tests = args.all if args.tests is None else args.tests == "on" 

606 if args.jobs < 1: 

607 parser.error("--jobs must be positive") 

608 try: 

609 socs, args.architectures = resolve_soc_selection(args.soc_list) 

610 except ValueError as error: 

611 parser.error(str(error)) 

612 args.soc_list = ",".join(socs) 

613 if platform.system() != "Linux": 

614 parser.error("Native builds require Linux/CANN/PTA") 

615 return args 

616 

617 

618def main() -> None: 

619 """Expose default local-framework builds and the explicit complete release matrix.""" 

620 args = parse_arguments() 

621 NATIVE.mkdir(parents=True, exist_ok=True) 

622 signal.signal(signal.SIGTERM, lambda *_: (_ for _ in ()).throw(KeyboardInterrupt())) 

623 acquired = False 

624 publication = WORKSPACE.BuildOutputs(ROOT / "dist", NATIVE / "payload/hyper_multicore") 

625 try: 

626 with WORKSPACE.build_lock(NATIVE, "all" if args.all else "current"): 

627 acquired = True 

628 run_locked_build(args, publication=publication) 

629 except KeyboardInterrupt: 

630 if publication.committed: 

631 message("Build committed; final lock maintenance cancelled. The new artifact is retained.", "WARN")