Coverage for cuda / pathfinder / _dynamic_libs / search_platform.py: 90.38%

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

2# SPDX-License-Identifier: Apache-2.0 

3 

4"""Platform abstraction for filesystem search steps. 

5 

6The goal is to keep :mod:`search_steps` platform-agnostic: it should not branch 

7on OS flags like ``IS_WINDOWS``. Instead, it calls through the single 

8``PLATFORM`` instance exported here. 

9""" 

10 

11from __future__ import annotations (empty)

12 

13import glob (empty)

14import os (empty)

15from collections.abc import Sequence (empty)

16from dataclasses import dataclass (empty)

17from typing import Protocol, cast (empty)

18 

19from cuda.pathfinder._dynamic_libs.lib_descriptor import LibDescriptor (empty)

20from cuda.pathfinder._dynamic_libs.supported_nvidia_libs import is_suppressed_dll_file (empty)

21from cuda.pathfinder._utils.find_sub_dirs import find_sub_dirs_all_sitepackages (empty)

22from cuda.pathfinder._utils.platform_aware import IS_WINDOWS (empty)

23 

24 

25def _no_such_file_in_sub_dirs( (empty)

26 sub_dirs: Sequence[str], file_wild: str, error_messages: list[str], attachments: list[str] 

27) -> None: 

28 error_messages.append(f"No such file: {file_wild}") 4 ctx1abcd

29 for sub_dir in find_sub_dirs_all_sitepackages(sub_dirs): 4 ctx1abcd

30 attachments.append(f' listdir("{sub_dir}"):') 2 ctx1cd

31 for node in sorted(os.listdir(sub_dir)): 2 ctx1cd

32 attachments.append(f" {node}") 1 ctx1c

33 

34 

35def _find_so_in_rel_dirs( (empty)

36 rel_dirs: tuple[str, ...], 

37 so_basename: str, 

38 error_messages: list[str], 

39 attachments: list[str], 

40) -> str | None: 

41 sub_dirs_searched: list[tuple[str, ...]] = [] 7 ctx1abnklcd

42 file_wild = so_basename + "*" 7 ctx1abnklcd

43 for rel_dir in rel_dirs: 7 ctx1abnklcd

44 sub_dir = tuple(rel_dir.split(os.path.sep)) 7 ctx1abnklcd

45 for abs_dir in find_sub_dirs_all_sitepackages(sub_dir): 7 ctx1abnklcd

46 # Exact unversioned match first; fall back to versioned names because some 

47 # distros only ship lib<name>.so.<major> (e.g. conda libcupti). Only one match 

48 # is expected in practice. Sort in reverse so the newest-sorting name wins if 

49 # multiple coexist, matching the newest-first bias elsewhere in pathfinder 

50 # (see LinuxSearchPlatform.find_in_lib_dir and load_dl_linux._candidate_sonames). 

51 # Issue #1732 tracks the deferred question of raising on true ambiguity. 

52 so_name = os.path.join(abs_dir, so_basename) 5 ctx1nklcd

53 if os.path.isfile(so_name): 5 ctx1nklcd

54 return so_name 1 ctx1n

55 for so_name in sorted(glob.glob(os.path.join(abs_dir, file_wild)), reverse=True): 4 ctx1klcd

56 if os.path.isfile(so_name): 2 ctx1kl

57 return so_name 2 ctx1kl

58 sub_dirs_searched.append(sub_dir) 4 ctx1abcd

59 for sub_dir in sub_dirs_searched: 4 ctx1abcd

60 _no_such_file_in_sub_dirs(sub_dir, file_wild, error_messages, attachments) 4 ctx1abcd

61 return None 4 ctx1abcd

62 

63 

64def _find_dll_under_dir(dirpath: str, file_wild: str) -> str | None: (empty)

65 for path in sorted(glob.glob(os.path.join(dirpath, file_wild))): 9 ctx1aefghiopj

66 if not os.path.isfile(path): 9 ctx1aefghiopj

67 continue 

68 if not is_suppressed_dll_file(os.path.basename(path)): 9 ctx1aefghiopj

69 return path 9 ctx1aefghiopj

70 return None 

71 

72 

73def _find_dll_in_rel_dirs( (empty)

74 rel_dirs: tuple[str, ...], 

75 lib_searched_for: str, 

76 error_messages: list[str], 

77 attachments: list[str], 

78) -> str | None: 

79 sub_dirs_searched: list[tuple[str, ...]] = [] 3 ctx1abj

80 for rel_dir in rel_dirs: 3 ctx1abj

81 sub_dir = tuple(rel_dir.split(os.path.sep)) 3 ctx1abj

82 for abs_dir in find_sub_dirs_all_sitepackages(sub_dir): 3 ctx1abj

83 dll_name = _find_dll_under_dir(abs_dir, lib_searched_for) 1 ctx1j

84 if dll_name is not None: 1 ctx1j

85 return dll_name 1 ctx1j

86 sub_dirs_searched.append(sub_dir) 2 ctx1ab

87 for sub_dir in sub_dirs_searched: 2 ctx1ab

88 _no_such_file_in_sub_dirs(sub_dir, lib_searched_for, error_messages, attachments) 2 ctx1ab

89 return None 2 ctx1ab

90 

91 

92class SearchPlatform(Protocol): (empty)

93 def lib_searched_for(self, libname: str) -> str: ... (empty)

94 

95 def site_packages_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: ... (empty)

96 

97 def conda_anchor_point(self, conda_prefix: str) -> str: ... (empty)

98 

99 def anchor_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: ... (empty)

100 

101 def find_in_site_packages( (empty)

102 self, 

103 rel_dirs: tuple[str, ...], 

104 lib_searched_for: str, 

105 error_messages: list[str], 

106 attachments: list[str], 

107 ) -> str | None: ... 

108 

109 def find_in_lib_dir( (empty)

110 self, 

111 lib_dir: str, 

112 libname: str, 

113 lib_searched_for: str, 

114 error_messages: list[str], 

115 attachments: list[str], 

116 ) -> str | None: ... 

117 

118 

119@dataclass(frozen=True, slots=True) (empty)

120class LinuxSearchPlatform: (empty)

121 def lib_searched_for(self, libname: str) -> str: (empty)

122 return f"lib{libname}.so" 25 ctx1aefgxhibqArvustmwnklcdBCD

123 

124 def site_packages_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: (empty)

125 return cast(tuple[str, ...], desc.site_packages_linux) 8 ctx1abnklcdE

126 

127 def conda_anchor_point(self, conda_prefix: str) -> str: (empty)

128 return conda_prefix 6 ctx1qrustm

129 

130 def anchor_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: (empty)

131 return cast(tuple[str, ...], desc.anchor_rel_dirs_linux) 18 ctx1aefyghizqrFGvustmw

132 

133 def find_in_site_packages( (empty)

134 self, 

135 rel_dirs: tuple[str, ...], 

136 lib_searched_for: str, 

137 error_messages: list[str], 

138 attachments: list[str], 

139 ) -> str | None: 

140 return _find_so_in_rel_dirs(rel_dirs, lib_searched_for, error_messages, attachments) 7 ctx1abnklcd

141 

142 def find_in_lib_dir( (empty)

143 self, 

144 lib_dir: str, 

145 _libname: str, 

146 lib_searched_for: str, 

147 error_messages: list[str], 

148 attachments: list[str], 

149 ) -> str | None: 

150 # Most libraries have both unversioned and versioned files/symlinks (exact match first) 

151 so_name = os.path.join(lib_dir, lib_searched_for) 14 ctx1aefghiqrvustmw

152 if os.path.isfile(so_name): 14 ctx1aefghiqrvustmw

153 return so_name 9 ctx1aefghivuw

154 # Some libraries only exist as versioned files (e.g., libcupti.so.13 in conda), 

155 # so the glob fallback is needed 

156 file_wild = lib_searched_for + "*" 5 ctx1qrstm

157 # Only one match is expected, but to ensure deterministic behavior in unexpected 

158 # situations, and to be internally consistent, we sort in reverse order with the 

159 # intent to return the newest version first. Issue #1732 tracks the deferred 

160 # question of raising on true ambiguity. 

161 for so_name in sorted(glob.glob(os.path.join(lib_dir, file_wild)), reverse=True): 5 ctx1qrstm

162 if os.path.isfile(so_name): 4 ctx1qrst

163 return so_name 4 ctx1qrst

164 error_messages.append(f"No such file: {file_wild}") 1 ctx1m

165 attachments.append(f' listdir("{lib_dir}"):') 1 ctx1m

166 if not os.path.isdir(lib_dir): 1 ctx1m

167 attachments.append(" DIRECTORY DOES NOT EXIST") 

168 else: 

169 for node in sorted(os.listdir(lib_dir)): 1 ctx1m

170 attachments.append(f" {node}") 1 ctx1m

171 return None 1 ctx1m

172 

173 

174@dataclass(frozen=True, slots=True) (empty)

175class WindowsSearchPlatform: (empty)

176 def lib_searched_for(self, libname: str) -> str: (empty)

177 return f"{libname}*.dll" 12 ctx1aefgxhibopjH

178 

179 def site_packages_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: (empty)

180 return cast(tuple[str, ...], desc.site_packages_windows) 3 ctx1abj

181 

182 def conda_anchor_point(self, conda_prefix: str) -> str: (empty)

183 return os.path.join(conda_prefix, "Library") 1 ctx1o

184 

185 def anchor_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: (empty)

186 return cast(tuple[str, ...], desc.anchor_rel_dirs_windows) 11 ctx1aefyghizIop

187 

188 def find_in_site_packages( (empty)

189 self, 

190 rel_dirs: tuple[str, ...], 

191 lib_searched_for: str, 

192 error_messages: list[str], 

193 attachments: list[str], 

194 ) -> str | None: 

195 return _find_dll_in_rel_dirs(rel_dirs, lib_searched_for, error_messages, attachments) 3 ctx1abj

196 

197 def find_in_lib_dir( (empty)

198 self, 

199 lib_dir: str, 

200 libname: str, 

201 _lib_searched_for: str, 

202 error_messages: list[str], 

203 attachments: list[str], 

204 ) -> str | None: 

205 file_wild = libname + "*.dll" 8 ctx1aefghiop

206 dll_name = _find_dll_under_dir(lib_dir, file_wild) 8 ctx1aefghiop

207 if dll_name is not None: 8 ctx1aefghiop

208 return dll_name 8 ctx1aefghiop

209 error_messages.append(f"No such file: {file_wild}") 

210 attachments.append(f' listdir("{lib_dir}"):') 

211 if not os.path.isdir(lib_dir): 

212 attachments.append(" DIRECTORY DOES NOT EXIST") 

213 else: 

214 for node in sorted(os.listdir(lib_dir)): 

215 attachments.append(f" {node}") 

216 return None 

217 

218 

219PLATFORM: SearchPlatform = WindowsSearchPlatform() if IS_WINDOWS else LinuxSearchPlatform() (empty)