Coverage for cuda / pathfinder / _dynamic_libs / search_steps.py: 99.05%

105 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-04-29 01:27 +0000

1# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. 

2# SPDX-License-Identifier: Apache-2.0 

3 

4"""Composable search steps for locating NVIDIA libraries. 

5 

6Each find step is a callable with signature:: 

7 

8 (SearchContext) -> FindResult | None 

9 

10Find steps locate a library file on disk without loading it. The 

11orchestrator in :mod:`load_nvidia_dynamic_lib` handles loading, the 

12already-loaded check, and dependency resolution. 

13 

14Step sequences are defined per search strategy so that adding a new 

15step or strategy only requires adding a function and a tuple entry. 

16 

17This module is intentionally platform-agnostic: it does not branch on the 

18current operating system. Platform differences are routed through the 

19:data:`~cuda.pathfinder._dynamic_libs.search_platform.PLATFORM` instance. 

20""" 

21 

22import glob 

23import os 

24from collections.abc import Callable 

25from dataclasses import dataclass, field 

26from typing import NoReturn, cast 

27 

28from cuda.pathfinder._dynamic_libs.lib_descriptor import LibDescriptor 

29from cuda.pathfinder._dynamic_libs.load_dl_common import DynamicLibNotFoundError 

30from cuda.pathfinder._dynamic_libs.search_platform import PLATFORM, SearchPlatform 

31from cuda.pathfinder._utils.env_vars import get_cuda_path_or_home 

32 

33# --------------------------------------------------------------------------- 

34# Data types 

35# --------------------------------------------------------------------------- 

36 

37 

38@dataclass 

39class FindResult: 

40 """A library file located on disk (not yet loaded).""" 

41 

42 abs_path: str 

43 found_via: str 

44 

45 

46@dataclass 

47class SearchContext: 

48 """Mutable state accumulated during the search cascade.""" 

49 

50 desc: LibDescriptor 

51 platform: SearchPlatform = PLATFORM 

52 error_messages: list[str] = field(default_factory=list) 

53 attachments: list[str] = field(default_factory=list) 

54 

55 @property 

56 def libname(self) -> str: 

57 return self.desc.name # type: ignore[no-any-return] # mypy can't resolve new sibling module 1acbeursqgUhnijklmopEFGHIJ!#%VW

58 

59 @property 

60 def lib_searched_for(self) -> str: 

61 return cast(str, self.platform.lib_searched_for(self.libname)) 1acbeursqgUhnijklmopEFGHIJ!#VW

62 

63 def raise_not_found(self) -> NoReturn: 

64 err = ", ".join(self.error_messages) 1uUVW

65 att = "\n".join(self.attachments) 1uUVW

66 raise DynamicLibNotFoundError(f'Failure finding "{self.lib_searched_for}": {err}\n{att}') 1uUVW

67 

68 

69#: Type alias for a find step callable. 

70FindStep = Callable[[SearchContext], FindResult | None] 

71 

72 

73def _find_lib_dir_using_anchor(desc: LibDescriptor, platform: SearchPlatform, anchor_point: str) -> str | None: 

74 """Find the library directory under *anchor_point* using the descriptor's relative paths.""" 

75 rel_dirs = platform.anchor_rel_dirs(desc) 1acbdersvghZXYnijklmop

76 for rel_path in rel_dirs: 1acbdersvghZXYnijklmop

77 for dirname in sorted(glob.glob(os.path.join(anchor_point, rel_path))): 1acbdersvghZXYnijklmop

78 if os.path.isdir(dirname): 1acbersghXYnijklmop

79 return os.path.normpath(dirname) 1acbersghXYnijklmop

80 return None 1dvZ

81 

82 

83def _find_using_lib_dir(ctx: SearchContext, lib_dir: str | None) -> str | None: 

84 """Find a library file in a resolved lib directory.""" 

85 if lib_dir is None: 1acbdersvghnijklmop

86 return None 1dv

87 return cast( 1acbersghnijklmop

88 str | None, 

89 ctx.platform.find_in_lib_dir( 

90 lib_dir, 

91 ctx.libname, 

92 ctx.lib_searched_for, 

93 ctx.error_messages, 

94 ctx.attachments, 

95 ), 

96 ) 

97 

98 

99def _derive_ctk_root_linux(resolved_lib_path: str) -> str | None: 

100 """Derive CTK root from Linux canary path. 

101 

102 Supports: 

103 - ``$CTK_ROOT/lib64/libfoo.so.*`` 

104 - ``$CTK_ROOT/lib/libfoo.so.*`` 

105 - ``$CTK_ROOT/targets/<triple>/lib64/libfoo.so.*`` 

106 - ``$CTK_ROOT/targets/<triple>/lib/libfoo.so.*`` 

107 """ 

108 lib_dir = os.path.dirname(resolved_lib_path) 1cbtdKLMNO0wxyzABCDf

109 basename = os.path.basename(lib_dir) 1cbtdKLMNO0wxyzABCDf

110 if basename in ("lib64", "lib"): 1cbtdKLMNO0wxyzABCDf

111 parent = os.path.dirname(lib_dir) 1cbdKLMNOwxyzABCDf

112 grandparent = os.path.dirname(parent) 1cbdKLMNOwxyzABCDf

113 if os.path.basename(grandparent) == "targets": 1cbdKLMNOwxyzABCDf

114 return os.path.dirname(grandparent) 1NO

115 return parent 1cbdKLMwxyzABCDf

116 return None 1cbtd0f

117 

118 

119def _derive_ctk_root_windows(resolved_lib_path: str) -> str | None: 

120 """Derive CTK root from Windows canary path. 

121 

122 Supports: 

123 - ``$CTK_ROOT/bin/x64/foo.dll`` (CTK 13 style) 

124 - ``$CTK_ROOT/bin/foo.dll`` (CTK 12 style) 

125 """ 

126 import ntpath 1cbtdRPSQTf

127 

128 lib_dir = ntpath.dirname(resolved_lib_path) 1cbtdRPSQTf

129 basename = ntpath.basename(lib_dir).lower() 1cbtdRPSQTf

130 if basename == "x64": 1cbtdRPSQTf

131 parent = ntpath.dirname(lib_dir) 1PQf

132 if ntpath.basename(parent).lower() == "bin": 1PQf

133 return ntpath.dirname(parent) 1PQf

134 elif basename == "bin": 1cbtdRST

135 return ntpath.dirname(lib_dir) 1cbdRS

136 return None 1tT

137 

138 

139def derive_ctk_root(resolved_lib_path: str) -> str | None: 

140 """Derive CTK root from a resolved canary library path.""" 

141 ctk_root = _derive_ctk_root_linux(resolved_lib_path) 1cbtd45wxyzABCDf

142 if ctk_root is not None: 1cbtd45wxyzABCDf

143 return ctk_root 1cbd4wxyzABCDf

144 return _derive_ctk_root_windows(resolved_lib_path) 1cbtd5f

145 

146 

147def find_via_ctk_root(ctx: SearchContext, ctk_root: str) -> FindResult | None: 

148 """Find a library under a previously derived CTK root.""" 

149 lib_dir = _find_lib_dir_using_anchor(ctx.desc, ctx.platform, ctk_root) 1cbdrsv

150 abs_path = _find_using_lib_dir(ctx, lib_dir) 1cbdrsv

151 if abs_path is None: 1cbdrsv

152 return None 1dv

153 return FindResult(abs_path, "system-ctk-root") 1cbrs

154 

155 

156# --------------------------------------------------------------------------- 

157# Find steps 

158# --------------------------------------------------------------------------- 

159 

160 

161def find_in_site_packages(ctx: SearchContext) -> FindResult | None: 

162 """Search pip wheel install locations.""" 

163 rel_dirs = ctx.platform.site_packages_rel_dirs(ctx.desc) 1aqEFGHIJ6

164 if not rel_dirs: 1aqEFGHIJ6

165 return None 16

166 abs_path = ctx.platform.find_in_site_packages(rel_dirs, ctx.lib_searched_for, ctx.error_messages, ctx.attachments) 1aqEFGHIJ

167 if abs_path is not None: 1aqEFGHIJ

168 return FindResult(abs_path, "site-packages") 1EFGH

169 return None 1aqIJ

170 

171 

172def find_in_conda(ctx: SearchContext) -> FindResult | None: 

173 """Search ``$CONDA_PREFIX``.""" 

174 conda_prefix = os.environ.get("CONDA_PREFIX") 1aqghijklm78

175 if not conda_prefix: 1aqghijklm78

176 return None 1aq78

177 anchor = ctx.platform.conda_anchor_point(conda_prefix) 1ghijklm

178 lib_dir = _find_lib_dir_using_anchor(ctx.desc, ctx.platform, anchor) 1ghijklm

179 abs_path = _find_using_lib_dir(ctx, lib_dir) 1ghijklm

180 if abs_path is not None: 1ghijklm

181 return FindResult(abs_path, "conda") 1ghijkl

182 return None 1m

183 

184 

185def find_in_cuda_path(ctx: SearchContext) -> FindResult | None: 

186 """Search ``$CUDA_PATH`` / ``$CUDA_HOME``. 

187 

188 On Windows, this is the normal fallback for system-installed CTK DLLs when 

189 they are not already discoverable via the native ``LoadLibraryExW(..., 0)`` 

190 path used by :func:`cuda.pathfinder._dynamic_libs.load_dl_windows.load_with_system_search`. 

191 Python 3.8+ does not include ``PATH`` in that native DLL search. 

192 

193 The returned ``found_via`` is always ``"CUDA_PATH"`` regardless of which 

194 environment variable actually provided the value. 

195 """ 

196 cuda_home = get_cuda_path_or_home() 1abeunop9

197 if cuda_home is None: 1abeunop9

198 return None 1bu9

199 lib_dir = _find_lib_dir_using_anchor(ctx.desc, ctx.platform, cuda_home) 1aenop

200 abs_path = _find_using_lib_dir(ctx, lib_dir) 1aenop

201 if abs_path is not None: 1aenop

202 return FindResult(abs_path, "CUDA_PATH") 1aenop

203 return None 

204 

205 

206# --------------------------------------------------------------------------- 

207# Step sequences per strategy 

208# --------------------------------------------------------------------------- 

209 

210#: Find steps that run before the already-loaded check and system search. 

211EARLY_FIND_STEPS: tuple[FindStep, ...] = (find_in_site_packages, find_in_conda) 

212 

213#: Find steps that run after system search fails. 

214LATE_FIND_STEPS: tuple[FindStep, ...] = (find_in_cuda_path,) 

215 

216 

217# --------------------------------------------------------------------------- 

218# Cascade runner 

219# --------------------------------------------------------------------------- 

220 

221 

222def run_find_steps(ctx: SearchContext, steps: tuple[FindStep, ...]) -> FindResult | None: 

223 """Run find steps in order, returning the first hit.""" 

224 for step in steps: 1abeuq$123

225 result = step(ctx) 1abeuq123

226 if result is not None: 1abeuq123

227 return result 1ae13

228 return None 1abuq$2