Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_multicore / _loader.py: 71%

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"""Locate and load the component-owned multicore payload.""" 

16 

17from __future__ import annotations 

18 

19import ctypes 

20import json 

21from functools import lru_cache 

22import importlib 

23import os 

24from pathlib import Path 

25import platform 

26import sys 

27from types import ModuleType 

28from contextlib import contextmanager 

29from threading import RLock 

30 

31from hyper_multicore._native_manifest import ( 

32 framework_package, check_integrity, read_manifest, resource_path, select_variant, select_shmem_backend, 

33) 

34 

35_VENDOR_NAME = "hyper_multicore_nn" 

36_NATIVE_LOCK = RLock() 

37_selection = None 

38_native_failure = None 

39_loaded_kinds = set() 

40_ACL_DEV_ATTR_NPU_ARCH = 601 

41 

42 

43class NativeComponentUnavailableError(RuntimeError): 

44 """Report an unavailable or incorrectly activated multicore component.""" 

45 

46 

47def diagnostic_context() -> str: 

48 """Read best-effort environment/build metadata without importing either framework.""" 

49 actual = {"python": sys.executable, "python_abi": f"cp{sys.version_info.major}{sys.version_info.minor}", 

50 "host_arch": platform.machine(), "cann_version": "unknown"} 

51 for package in ("torch", "torch-npu"): 

52 try: 

53 actual[package] = framework_package(package.replace("-", "_"))["version"] 

54 # Diagnostics must not replace the original import/load failure. 

55 except Exception: # pylint: disable=broad-exception-caught 

56 actual[package] = "unavailable" 

57 try: 

58 for line in (Path(os.environ["ASCEND_HOME_PATH"]) / "opp/version.info").read_text().splitlines(): 

59 key, separator, value = line.partition("=") 

60 if separator and key.strip() == "Version": 

61 actual["cann_version"] = value.strip() 

62 break 

63 except Exception: # pylint: disable=broad-exception-caught 

64 pass 

65 try: 

66 info = json.loads((_component_root().parent / "build_info.json").read_text()) 

67 matrix = [{key: variant.get(key) for key in ("id", "torch_version", "pta_version", "torch_cxx11_abi")} 

68 for variant in info.get("variants", [])] 

69 except Exception: # pylint: disable=broad-exception-caught 

70 matrix = "unavailable" 

71 return f"runtime={actual} packaged_matrix={matrix}" 

72 

73 

74@lru_cache(maxsize=1) 

75def load_torch_npu() -> ModuleType: 

76 """Register the optional NPU backend only when an operation needs it. 

77 

78 Import errors propagate so an unusable backend cannot masquerade as successful 

79 native initialization. Failed imports are not cached and may be retried. 

80 """ 

81 try: 

82 framework_package("torch_npu") 

83 except ModuleNotFoundError as error: 

84 raise NativeComponentUnavailableError( 

85 f"[HYPER-MULTICORE-PTA-NOT-INSTALLED] Python={sys.executable}; install torch-npu " 

86 "for your Torch/CANN environment. Inspect hyper_multicore_build_info for packaged branches. " 

87 f"{diagnostic_context()}" 

88 ) from error 

89 try: 

90 return importlib.import_module("torch_npu") 

91 except (ImportError, OSError, RuntimeError) as error: 

92 raise NativeComponentUnavailableError( 

93 f"[HYPER-MULTICORE-PTA-IMPORT-FAILED] {diagnostic_context()}: {error}" 

94 ) from error 

95 

96 

97def require_multicore_environment() -> Path: 

98 """Require the packaged custom OPP environment prepared by ``set_env.bash``.""" 

99 component_root = _component_root() 

100 vendor_root = component_root / "vendors" / _VENDOR_NAME 

101 if not vendor_root.is_dir(): 

102 raise NativeComponentUnavailableError( 

103 f"[HP-NATIVE-PAYLOAD-MISSING] component=multicore vendor={vendor_root}. " 

104 "Inspect the current build log; for source/PYTHONPATH development, run " 

105 "./build.sh." 

106 ) 

107 vendor_root = vendor_root.resolve() 

108 op_api_root = (vendor_root / "op_api" / "lib").resolve() 

109 missing_variables = [] 

110 if not _environment_contains_path("ASCEND_CUSTOM_OPP_PATH", vendor_root): 

111 missing_variables.append("ASCEND_CUSTOM_OPP_PATH") 

112 if not _environment_contains_path("LD_LIBRARY_PATH", op_api_root): 

113 missing_variables.append("LD_LIBRARY_PATH") 

114 if missing_variables: 

115 set_env_script = component_root / "set_env.bash" 

116 missing_text = ",".join(missing_variables) 

117 if any(module in sys.modules for module in ("torch", "torch_npu")): 

118 raise NativeComponentUnavailableError( 

119 "[HP-NATIVE-OPP-ACTIVATION-TOO-LATE] component=multicore " 

120 f"missing={missing_text}. Torch/torch_npu has already been imported; " 

121 f"exit the current Python process, run source {set_env_script}, and start a new process." 

122 ) 

123 raise NativeComponentUnavailableError( 

124 "[HP-NATIVE-OPP-NOT-ACTIVATED] component=multicore " 

125 f"missing={missing_text}. Run source {set_env_script} before starting Python with multicore." 

126 ) 

127 return vendor_root.resolve() 

128 

129 

130def get_multicore_paths() -> tuple[Path, Path]: 

131 """Return the vendor root and shared Torch adapter after ABI validation.""" 

132 vendor_root = require_multicore_environment() 

133 selected = validate_native_compatibility() 

134 return vendor_root, resource_path(selected["root"], selected["variant"]["torch_so"]) 

135 

136 

137def _runtime_identity() -> dict: 

138 """Read branch and ABI selectors without comparing external binary fingerprints.""" 

139 try: 

140 pta = framework_package("torch_npu")["version"] 

141 except ModuleNotFoundError as error: 

142 raise NativeComponentUnavailableError( 

143 f"[HYPER-MULTICORE-PTA-NOT-INSTALLED] Python={sys.executable}; " 

144 "install matching Torch/PTA/CANN; inspect hyper_multicore_build_info." 

145 ) from error 

146 torch = importlib.import_module("torch") 

147 

148 actual = {"torch_version": str(getattr(torch, "__version__", "unknown")), "pta_version": pta, 

149 "python_abi": f"cp{sys.version_info.major}{sys.version_info.minor}", 

150 "host_arch": platform.machine(), "torch_cxx11_abi": torch.compiled_with_cxx11_abi(), 

151 "cann_version": "unknown"} 

152 try: 

153 version_file = Path(os.environ["ASCEND_HOME_PATH"]) / "opp/version.info" 

154 for line in version_file.read_text().splitlines(): 

155 name, separator, value = line.partition("=") 

156 if separator and name.strip() == "Version": 

157 actual["cann_version"] = value.strip() 

158 break 

159 except (OSError, KeyError): 

160 pass 

161 return actual 

162 

163 

164def validate_native_compatibility() -> dict: 

165 """Choose an internally validated framework variant; external post/CANN values do not gate it.""" 

166 global _selection 

167 actual = _runtime_identity() # Import the framework outside our native-registration lock. 

168 with _NATIVE_LOCK: 

169 if _selection is not None: 

170 return _selection 

171 root = _component_root().parent 

172 try: 

173 info = json.loads((root / "build_info.json").read_text(encoding="utf-8")) 

174 manifest = read_manifest(root) 

175 check_integrity(root, info, manifest) 

176 variant = select_variant(manifest, actual) 

177 except (OSError, ValueError, KeyError, RuntimeError) as error: 

178 raise NativeComponentUnavailableError( 

179 f"[HYPER-MULTICORE-ABI-MISMATCH] root={root} runtime={actual}: {error}" 

180 ) from error 

181 _selection = {"root": root, "variant": variant, "actual": actual, "build_info": info, "manifest": manifest} 

182 return _selection 

183 

184 

185def query_npu_arch() -> int: 

186 """Read visible logical device 0's architecture without creating a device context.""" 

187 acl = ctypes.CDLL("libascendcl.so", mode=ctypes.RTLD_LOCAL) 

188 query = acl.aclrtGetDeviceInfo 

189 query.argtypes = [ctypes.c_uint32, ctypes.c_int32, ctypes.POINTER(ctypes.c_int64)] 

190 query.restype = ctypes.c_int32 

191 arch = ctypes.c_int64() 

192 status = query(0, _ACL_DEV_ATTR_NPU_ARCH, ctypes.byref(arch)) 

193 if status != 0: 

194 raise NativeComponentUnavailableError(f"aclrtGetDeviceInfo(NPU_ARCH) failed for visible device 0: {status}") 

195 return arch.value 

196 

197 

198def select_native_shmem(selected: dict) -> tuple[str, dict]: 

199 """Select a packaged backend before any project SHMEM library is loaded.""" 

200 try: 

201 return select_shmem_backend(selected["manifest"], query_npu_arch()) 

202 except (OSError, AttributeError, ValueError) as error: 

203 raise NativeComponentUnavailableError(f"[HP-NATIVE-SHMEM-BACKEND-UNAVAILABLE] {error}") from error 

204 

205 

206def native_ready(kind: str) -> bool: 

207 """Check terminal state even when a previously loaded adapter takes its fast path.""" 

208 with _NATIVE_LOCK: 

209 if _native_failure is not None: 

210 raise NativeComponentUnavailableError( 

211 f"[HYPER-MULTICORE-INITIALIZATION-FAILED] Restart the process: {_native_failure}" 

212 ) from _native_failure 

213 return kind in _loaded_kinds 

214 

215 

216@contextmanager 

217def native_loading(kind: str): 

218 """Serialize initial registration and make partial native initialization failures terminal.""" 

219 global _native_failure 

220 with _NATIVE_LOCK: 

221 if _native_failure is not None: 

222 raise NativeComponentUnavailableError( 

223 f"[HYPER-MULTICORE-INITIALIZATION-FAILED] Restart the process: {_native_failure}" 

224 ) from _native_failure 

225 if kind in _loaded_kinds: 

226 yield False 

227 return 

228 try: 

229 yield True 

230 except BaseException as error: 

231 _native_failure = error 

232 raise 

233 _loaded_kinds.add(kind) 

234 

235 

236def native_context() -> str: 

237 """Describe the fixed adapter and runtime without hiding the original loader failure.""" 

238 if _selection is None: 

239 return diagnostic_context() 

240 record = next(v for v in _selection["build_info"]["variants"] if v["id"] == _selection["variant"]["id"]) 

241 return f"runtime={_selection['actual']} adapter={_selection['variant']} built={record}" 

242 

243 

244def preload_vendor_library(vendor_root: Path) -> None: 

245 """Load the exact ACLNN library without exposing its dependency symbols globally.""" 

246 library_path = vendor_root / "op_api" / "lib" / "libcust_opapi.so" 

247 try: 

248 # CANN libraries can export identically named globals. Global promotion 

249 # can bind their independent destructors to the same object at exit. 

250 # ACLNN entry points are resolved through the vendor library handle. 

251 ctypes.CDLL(str(library_path), mode=ctypes.RTLD_LOCAL) 

252 except OSError as error: 

253 raise NativeComponentUnavailableError( 

254 f"[HP-NATIVE-VENDOR-LOAD-FAILED] library={library_path} {native_context()} error={error}." 

255 ) from error 

256 

257 

258def _component_root() -> Path: 

259 """Return the installed or source-build multicore payload root.""" 

260 packaged = Path(__file__).resolve().parent / "lib" 

261 if packaged.is_dir(): 

262 return packaged 

263 repository_root = Path(__file__).resolve().parents[1] 

264 if (repository_root / "setup.py").is_file(): 

265 source_build = ( 

266 repository_root / "build" / "native" / "payload" / "hyper_multicore" / "lib" 

267 ) 

268 if source_build.is_dir(): 

269 return source_build 

270 return packaged 

271 

272 

273def _environment_contains_path(variable: str, expected: Path) -> bool: 

274 """Return whether an environment path list contains ``expected``.""" 

275 configured = [value for value in os.environ.get(variable, "").split(os.pathsep) if value] 

276 return str(expected.resolve()) in {str(Path(value).resolve()) for value in configured}