Coverage for cuda / pathfinder / _dynamic_libs / search_steps.py: 98.10%
105 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-25 01:07 +0000
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-25 01:07 +0000
1# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2# SPDX-License-Identifier: Apache-2.0
4"""Composable search steps for locating NVIDIA libraries.
6Each find step is a callable with signature::
8 (SearchContext) -> FindResult | None
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.
14Step sequences are defined per search strategy so that adding a new
15step or strategy only requires adding a function and a tuple entry.
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"""
22import glob
23import os
24from collections.abc import Callable
25from dataclasses import dataclass, field
26from typing import NoReturn, cast
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
33# ---------------------------------------------------------------------------
34# Data types
35# ---------------------------------------------------------------------------
38@dataclass
39class FindResult:
40 """A library file located on disk (not yet loaded)."""
42 abs_path: str
43 found_via: str
46@dataclass
47class SearchContext:
48 """Mutable state accumulated during the search cascade."""
50 desc: LibDescriptor
51 platform: SearchPlatform = PLATFORM
52 error_messages: list[str] = field(default_factory=list)
53 attachments: list[str] = field(default_factory=list)
55 @property
56 def libname(self) -> str:
57 return self.desc.name # type: ignore[no-any-return] # mypy can't resolve new sibling module 1acbgtqrpiPjdemklnoCDE568QR
59 @property
60 def lib_searched_for(self) -> str:
61 return cast(str, self.platform.lib_searched_for(self.libname)) 1acbgtqrpiPjdemklnoCDE56QR
63 def raise_not_found(self) -> NoReturn:
64 err = ", ".join(self.error_messages) 1tPQR
65 att = "\n".join(self.attachments) 1tPQR
66 raise DynamicLibNotFoundError(f'Failure finding "{self.lib_searched_for}": {err}\n{att}') 1tPQR
69#: Type alias for a find step callable.
70FindStep = Callable[[SearchContext], FindResult | None]
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) 1acbfgqruijdeUSTmklno
76 for rel_path in rel_dirs: 1acbfgqruijdeUSTmklno
77 for dirname in sorted(glob.glob(os.path.join(anchor_point, rel_path))): 1acbfgqruijdeUSTmklno
78 if os.path.isdir(dirname): 1acbgqrijdeSTmklno
79 return os.path.normpath(dirname) 1acbgqrijdeSTmklno
80 return None 1fuU
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: 1acbfgqruijdemklno
86 return None 1fu
87 return cast( 1acbgqrijdemklno
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 )
99def _derive_ctk_root_linux(resolved_lib_path: str) -> str | None:
100 """Derive CTK root from Linux canary path.
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) 1cbsfFGHIJVvwxyzABh
109 basename = os.path.basename(lib_dir) 1cbsfFGHIJVvwxyzABh
110 if basename in ("lib64", "lib"): 1cbsfFGHIJVvwxyzABh
111 parent = os.path.dirname(lib_dir) 1cbfFGHIJvwxyzABh
112 grandparent = os.path.dirname(parent) 1cbfFGHIJvwxyzABh
113 if os.path.basename(grandparent) == "targets": 1cbfFGHIJvwxyzABh
114 return os.path.dirname(grandparent) 1IJ
115 return parent 1cbfFGHvwxyzABh
116 return None 1cbsfVh
119def _derive_ctk_root_windows(resolved_lib_path: str) -> str | None:
120 """Derive CTK root from Windows canary path.
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 1cbsfMKNLOh
128 lib_dir = ntpath.dirname(resolved_lib_path) 1cbsfMKNLOh
129 basename = ntpath.basename(lib_dir).lower() 1cbsfMKNLOh
130 if basename == "x64": 1cbsfMKNLOh
131 parent = ntpath.dirname(lib_dir) 1KLh
132 if ntpath.basename(parent).lower() == "bin": 1KLh
133 return ntpath.dirname(parent) 1KLh
134 elif basename == "bin": 1cbsfMNO
135 return ntpath.dirname(lib_dir) 1cbfMN
136 return None 1sO
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) 1cbsfZ0vwxyzABh
142 if ctk_root is not None: 1cbsfZ0vwxyzABh
143 return ctk_root 1cbfZvwxyzABh
144 return _derive_ctk_root_windows(resolved_lib_path) 1cbsf0h
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) 1cbfqru
150 abs_path = _find_using_lib_dir(ctx, lib_dir) 1cbfqru
151 if abs_path is None: 1cbfqru
152 return None 1fu
153 return FindResult(abs_path, "system-ctk-root") 1cbqr
156# ---------------------------------------------------------------------------
157# Find steps
158# ---------------------------------------------------------------------------
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) 1apdeCDE1
164 if not rel_dirs: 1apdeCDE1
165 return None 11
166 abs_path = ctx.platform.find_in_site_packages(rel_dirs, ctx.lib_searched_for, ctx.error_messages, ctx.attachments) 1apdeCDE
167 if abs_path is not None: 1apdeCDE
168 return FindResult(abs_path, "site-packages") 1CD
169 return None 1apdeE
172def find_in_conda(ctx: SearchContext) -> FindResult | None:
173 """Search ``$CONDA_PREFIX``."""
174 conda_prefix = os.environ.get("CONDA_PREFIX") 1apijdekl23
175 if not conda_prefix: 1apijdekl23
176 return None 1apde23
177 anchor = ctx.platform.conda_anchor_point(conda_prefix) 1ijkl
178 lib_dir = _find_lib_dir_using_anchor(ctx.desc, ctx.platform, anchor) 1ijkl
179 abs_path = _find_using_lib_dir(ctx, lib_dir) 1ijkl
180 if abs_path is not None: 1ijkl
181 return FindResult(abs_path, "conda") 1ijkl
182 return None
185def find_in_cuda_path(ctx: SearchContext) -> FindResult | None:
186 """Search ``$CUDA_PATH`` / ``$CUDA_HOME``.
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.
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() 1abgtdemno4
197 if cuda_home is None: 1abgtdemno4
198 return None 1bt4
199 lib_dir = _find_lib_dir_using_anchor(ctx.desc, ctx.platform, cuda_home) 1agdemno
200 abs_path = _find_using_lib_dir(ctx, lib_dir) 1agdemno
201 if abs_path is not None: 1agdemno
202 return FindResult(abs_path, "CUDA_PATH") 1agdemno
203 return None
206# ---------------------------------------------------------------------------
207# Step sequences per strategy
208# ---------------------------------------------------------------------------
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)
213#: Find steps that run after system search fails.
214LATE_FIND_STEPS: tuple[FindStep, ...] = (find_in_cuda_path,)
217# ---------------------------------------------------------------------------
218# Cascade runner
219# ---------------------------------------------------------------------------
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: 1abgtpde7WXY
225 result = step(ctx) 1abgtpdeWXY
226 if result is not None: 1abgtpdeWXY
227 return result 1agdeWY
228 return None 1abtpde7X