# mypy: 允许未类型化定义
r""
此包添加了对 CUDA 张量类型的支持。
它实现了与 CPU 张量相同的功能，但它们使用 GPU 进行计算。
它是惰性初始化的，因此您可以始终导入它并使用。
它是惰性初始化的，因此您可以始终导入它并使用。
使用 :func:`is_available()` 函数来检查您的系统是否支持 CUDA。
关于使用 CUDA 的更多详细信息，请参阅 :ref:`cuda-semantics`。
""
import 导入库
import 操作系统
import 线程
import 跟踪回溯
import 警告信息
from functools import lru_cache
from 打字 import
任何,
可调用,
投射,
可选,
联合
import 火炬
import torch._C
from 火炬 import
设备
作为 _device
from torch._utils import _模拟类型, _LazySeedTracker,
类属性
from torch 的类型 import
设备
from . import gds
from _utils
导入 _get_device_index
from .图
导入 (
CUDAGraph,
图,
图池句柄,
当前流是否正在捕获,
生成图形化可调用对象,
)
from .流
导入
事件,
外部流,
流
尝试:
from torch._C 导入 _cudart
# 类型：忽略[已定义]
除了
导入错误:
_cudart = 无。
_initialized = 假
_tls = 线程.local()
_initialization_lock = 线程.
锁()
_queued_calls: 列表[
元组[
可调用
[]
无
]
列表[str]]
] = 输入文本为空，请提供需要翻译的文本 # don't invoke these until initialization occurs
_is_in_bad_fork = getattr(PyTorch._C, "_cuda_isInBadFork", lambda: False)
设备_t =
联盟[
设备, str, int,
无]
_HAS_PYNVML = 假
_PYNVML_ERR = 无。
尝试:
from 火炬
导入
版本
作为
_版本
尝试:
if not _版本.hip:
导入 pynvml # type: ignore[import]
else:
导入 ctypes
from pathlib 导入 Path
在 ROCm（至少到 6.3.2 版本）中存在两个 libamd_smi.so 的副本：
- 一个位于 lib/libamd_smi.so
- 一个位于 share/amd_smi/amdsmi/libamd_smi.so
#
amdsmi Python 模块硬编码加载第二个副本在 share-中
# https://github.com/ROCm/amdsmi/blob/1d305dc9708e87080f64f668402887794cd46584/py-interface/amdsmi_wrapper.py#L174
#
# 参见 https://github.com/ROCm/amdsmi/issues/72.
#
这会导致 libamd_smi.so 副本违反 ODR 规范
# 也被加载（通过 `ld` 链接、`LD_LIBRARY_PATH` 或 `rpath`）。
#
为避免违规，我们挂钩 CDLL 并尝试使用
已加载的 amdsmi 版本，或进程中的任何版本
# rpath/LD_LIBRARY_PATH 首先设置，这样我们只加载单个副本
# 的 .so.
类 _amdsmi_cdll_hook:
定义 __init__(
我) ->
无:
我.
原始_CDLL = ctypes.CDLL
# 类型：忽略[misc,assignment]
路径 = [
libamd_smi.so]
if rocm_home := 操作系统.getenv(
ROCM_HOME,
操作系统.getenv(
ROCM_PATH)):
路径 = [
操作系统.
路径.
加入(rocm_home,
lib/libamd_smi.so)] +
路径
我.
路径:
列表[str] =
路径
定义
钩子 CDLL(
我,
名称:
联盟[str,
路径,
无
] *
参数:
任何, **
关键字参数:
任何
) -> ctypes.CDLL:
if 名称
以及
路径(
名称).
名称 ==
libamd_smi.so:
为
路径
在
我.
路径列表:
尝试:
返回
我.
原始_CDLL(
路径, *
参数, **
关键字参数)
除了 OSError:
通过
返回
我.
原始_CDLL(
名称, *
参数, **
关键字参数) # type: ignore[arg-type]
定义
__进入__(
我) ->
无:
ctypes.CDLL = 我.
钩子_CDLL
忽略[misc,assignment]
定义
__退出__(
我,
类型:
任何, value:
任何,
跟踪回溯:
任何) ->
无:
ctypes.动态链接库 =
我.
原始_CDLL
# 类型：忽略[杂项]
替换为 _amdsmi_cdll_hook():
导入 amdsmi # type: ignore[import]
_HAS_PYNVML = 真实
除了
模块未找到错误:
通过
最后:
删除 _version
除了
导入错误
作为
错误:
_PYNVML_ERR = 错误
有时库已安装但导入失败，可能是其他原因，因此我们记录错误以供后续查看
_lazy_seed_tracker = _LazySeedTracker()
如果 PyTorch 没有编译 CUDA，则定义一个假定的 _CudaDeviceProperties 类型
if hasattr(PyTorch._C,
_CudaDeviceProperties):
_CudaDeviceProperties = PyTorch._C._CudaDeviceProperties
else:
_CudaDeviceProperties = _dummy_type(_CudaDeviceProperties) # type: ignore[assignment, misc]
if hasattr(PyTorch._C,
_cuda 交换设备):
_交换设备 =
PyTorch._C.
_cuda 交换设备
else:
定义
交换设备(
设备: int) -> int:
if 设备 < 0:
返回 -1
抛出 RuntimeError(
PyTorch 未编译 CUDA 支持)
if hasattr(PyTorch._C,
_cuda_maybeExchangeDevice):
_maybe_exchange_device = PyTorch._C._cuda_maybeExchangeDevice
else:
定义
可能交换设备(
设备: int) -> int:
if 设备 < 0:
返回 -1
抛出 RuntimeError(
PyTorch 编译时未启用 CUDA 支持)
支持半精度:
布尔值 =
真实
支持 MAGMA:
布尔值 =
PyTorch._C.
熔岩存在
默认生成器:
元组[
PyTorch._C.
生成器] = ()
# 类型：忽略[赋值]
定义
已编译() -> bool:
r"编译时支持 CUDA 则返回 true。"
返回 hasattr(
PyTorch._C, "_cuda_getDeviceCount")
定义 _nvml_based_avail() -> bool:
返回
操作系统.getenv("PYTORCH_NVML_BASED_CUDA_CHECK") ==
1
[文档]def is_available() -> bool:
r"""返回一个布尔值，指示 CUDA 当前是否可用。"""
if not _is_compiled():
return False
如果_nvml_based_avail()
# 用户已设置环境变量以请求此可用性检查，该检查试图通过使用 NVML 来避免 fork poisoning，但代价是 CUDA 可用性评估较弱。注意，如果 NVML 发现/初始化失败，此评估将回退到默认的 CUDA 运行时 API 评估（`cudaGetDeviceCount`）
# 如果 NVML 发现/初始化失败，此评估将回退到默认的 CUDA 运行时 API 评估（`cudaGetDeviceCount`）
# 如果 NVML 发现/初始化失败，此评估将回退到默认的 CUDA 运行时 API 评估（`cudaGetDeviceCount`）
返回设备计数器大于 0
else:
# 默认可用性检查从不抛出异常，如果驱动程序丢失或无法初始化，则返回 0。这使用了 CUDA 运行时 API `cudaGetDeviceCount`，它反过来初始化 CUDA 驱动程序
# be initialized. This uses the CUDA Runtime API `cudaGetDeviceCount` which in turn initializes the CUDA Driver
通过 `cuInit` 的 API
返回 torch._C._cuda_getDeviceCount() > 0
定义
支持 BF16(
包括模拟:
布尔值 = True):
r返回一个布尔值，指示当前 CUDA/ROCm 设备是否支持 bfloat16 数据类型。
检查 ROCm，如果为真则返回 true，无需 ROCm 版本检查。
支持 AMD GPU 架构。
如果
PyTorch.
版本.hip:
返回
真的
如果没有 CUDA，则也不支持 bf16。
如果
非
是否可用():
返回
假
设备 =
PyTorch.cuda.current_device()
# 检查 CUDA 版本和设备计算能力。
# 这是一种快速检查的方法。
cuda_version = PyTorch.version.cuda
如果 cuda_version is
非
无。
以及
PyTorch.cuda.
获取设备属性(
设备).
主要
≥ 8:
返回
真的
如果
非
包括模拟:
返回
假
最后尝试创建一个 bfloat16 设备。
返回
检查 bf16_tensor 是否受支持(
设备)
@lru_cache(最大尺寸=16)
定义
检查 bf16_tensor 是否受支持(
设备: _device_t):
尝试:
PyTorch.
张量
[1.0
]
数据类型=
PyTorch.bfloat16,
设备=
设备)
返回
真的
除了
异常:
返回
假
[文档]def is_tf32_supported() -> bool:
r"""返回一个布尔值，指示当前 CUDA/ROCm 设备是否支持 dtype tf32。"""
# 检查 ROCm。如果为真，则返回 false，因为 PyTorch 目前不支持
ROCm 上的 tf32。
如果 torch 版本是 hip：
返回 False
# 否则，在支持 tf32 的原生 CUDA 平台上（即无需仿真）
支持 bfloat16。
返回是否支持 bfloat16（包括仿真），不包括仿真。
定义 _sleep(
循环次数):
PyTorch._C._cuda_sleep(cycles)
定义 _extract_arch_version(arch_string: str):
从 CUDA 版本中提取架构字符串
基础 =
架构字符串.
分割(
“下划线”
)]1]
基础 =
基础.
移除后缀(
"一个")
返回 int(
基础)
定义
检查功能():
不正确的二进制警告 =
""
找到 GPU%d %s
这需要 CUDA 版本 >=%d
到
正常工作，但你的 PyTorch 是用 CUDA_VERSION 编译的
%d
.请安装正确的 PyTorch 二进制文件
.请安装正确的 PyTorch 二进制文件
使用来自 https://pytorch.org 的说明
""" # noqa: F841
旧 GPU 警告 =
""
找到 GPU%d %s
这是 CUDA 能力%d.%d.
PyTorch 不再支持这款 GPU，因为它太老了。
该库支持的最低 CUDA 能力是%d.%d.
"""
如果
PyTorch.
版本.cuda is
非
无:
# 在 ROCm 上我们不希望进行此检查
CUDA 版本 =
PyTorch._C._cuda_getCompiledVersion() # noqa: F841
为 d
在
范围(
设备数量()):
能力 =
获取设备能力(d)
主要的 =
能力[0]
微小 =
能力[1]
名称 =
获取设备名称(d)
当前架构 =
主要 * 10 +
次要
最小架构 =
最小值(
(_提取架构版本(
架构)
为
架构
在
PyTorch.cuda.
获取架构列表()),
默认=35,
)
如果
当前架构 <
最小架构:
警告.
警告(
旧 GPU 警告
% (d, 名称,
主要,
次要,
最小架构 // 10,
最小架构 % 10)
)
定义
检查 cubins():
不兼容设备警告 =
""
{}与 CUDA 能力 sm_不兼容{}
与当前 PyTorch 安装不兼容。
当前 PyTorch 安装支持 CUDA 功能{}.
如果您想使用{}
请查看 PyTorch 在本地安装的说明：https://pytorch.org/get-started/locally/
""
if PyTorch.
版本.cuda is
无:
在 ROCm 上我们不希望进行此检查
返回
架构列表 =
获取架构列表()
if len(架构列表) == 0:
返回
支持的_sm = [
_提取架构版本(
架构)
为
架构
在
架构列表
如果 "sm_"
在
架构]
为 idx
在
范围(
设备数量()):
核心能力,
小版本 =
获取设备能力(
索引)
NVIDIA GPU 计算架构在主版本内向后兼容
支持 =
任何(
小 // 10 ==
大写
为
小
在
支持的_sm)
如果
非
支持的:
设备名称 =
获取设备名称(
索引)
功能 =
主要功能 * 10 +
次要功能
警告.
警告(
不兼容设备警告.
格式(
设备名称,
能力,
输入文本为空，请提供需要翻译的文本.
连接(
架构列表),
设备名称
)
)
[文档]def is_initialized():
r"""返回 PyTorch 的 CUDA 状态是否已初始化。"""
return _initialized and not _is_in_bad_fork()
定义 _lazy_call(callable, **
关键字参数):
替换为
初始化锁:
如果
是否已初始化():
callable()
else:
# TODO(torch_deploy): 此处访问 linecache，尝试读取
# 文件系统以获取 traceback 信息。修复 linecache 或进行其他操作
# else here if this ends up being important.
全球
_懒种子追踪器
如果
关键字参数.
获取(
"种子全部", False):
懒惰种子追踪器.
队列所有种子(callable,
跟踪回溯.
格式化堆栈())
elif 关键字参数.
获取(
种子, False):
懒惰的种子追踪器.
队列种子(callable,
跟踪回溯.
格式化堆栈())
else:
# 不要存储实际的堆栈跟踪以避免内存循环
_队列调用.
添加((callable,
堆栈跟踪.
格式栈()))
_延迟调用(
_检查能力)
_延迟调用(
检查 cubins)
类
延迟 CUDA 调用错误(
异常):
通过
内存不足错误 =
PyTorch._C.
内存不足错误
[文档]def init():
r"""初始化 PyTorch 的 CUDA 状态。
如果您通过 PyTorch 的 C API 与之交互，可能需要显式调用此函数，
因为 Python 绑定的 CUDA 功能
初始化完成之前将不可用。
普通用户通常不需要这个，因为 PyTorch 的所有 CUDA 方法
会自动按需初始化 CUDA 状态。
如果 CUDA 状态已经初始化，则不执行任何操作。
"``"
_lazy_init()
定义 _lazy_init():
全局
已初始化, _queued_calls
如果
已初始化()
或者 hasattr(_tls,
初始化中):
返回
替换为
初始化锁:
# 我们要双重检查锁定，伙计们！这是可以的，因为
# 上述测试已经被全局解释器锁（GIL）保护了。内部测试
# 当一个线程因为其他线程正在初始化而被阻塞时
# 当它们获得锁后，会发现没有剩下什么要做的。
# 初始化完成。
如果 is_initialized():
返回
防止其他线程进入 _lazy_init 是很重要的
立即，因为我们仍然保证有全局解释锁（GIL），因为某些
以下 C 调用将释放全局解释器锁（GIL）
如果
_is_in_bad_fork
是坏分支():
提升 RuntimeError(
"无法在派生子进程中重新初始化 CUDA。要使用 CUDA 与多进程，必须使用'spawn'启动方法"
"multiprocessing，您必须使用'spawn'启动方法"
)
如果
非 hasattr(
PyTorch._C, "_cuda_getDeviceCount"):
提升
断言错误(
"CUDA 未启用编译的 Torch")
如果 _cudart is
无:
提升
断言错误(
"libcudart 函数不可用。看起来你的构建有问题？"
)
# 如果存在驱动初始化错误，没有找到 GPU 或发生其他错误
# 将抛出异常
如果
"CUDA 模块加载"
非
在
操作系统.
环境:
操作系统.
环境[
"CUDA 模块加载"] =
懒惰
PyTorch._C.
_cuda 初始化()
# 一些排队的调用可能会递归地调用 _lazy_init();
# 在这种情况下，我们只需返回而不进行初始化即可。
然而，我们绝不能让任何其他线程进入！
_tls.正在初始化 =
真的
_队列调用.
扩展(
调用
为
电话
在
_懒惰种子跟踪器.
获取电话()
如果
电话)
尝试:
为
排队调用,
原始跟踪信息
在
_排队调用:
尝试:
排队调用()
除了
异常
作为 e:
消息 = (
fCUDA 在初始化时懒加载调用失败，错误信息：{str(e)}
\n\n
\n\n"
fCUDA 调用最初在以下位置被调用：
\n\n
\n\n{请提供需要翻译的文本.
连接(
原始跟踪)}"
)
抛出
延迟 CUDA 调用错误(msg) from e
最后:
delattr(_tls, 初始化中)
已初始化 =
真的
[文档]
定义 cudart():
r获取 CUDA 运行时 API 模块。
此函数初始化 CUDA 运行时环境（如果尚未初始化）并返回 CUDA 运行时 API 模块（_cudart）。
CUDA 运行时 API 模块（_cudart）。
运行时 API 模块提供了访问各种 CUDA 运行时函数的接口。
参数：
``None``
返回值：
模块：CUDA 运行时 API 模块（_cudart）。
抛出异常：
运行时错误：如果在子进程中无法重新初始化 CUDA。
断言错误：如果 PyTorch 没有编译 CUDA 支持或 libcudart 函数不可用。
CUDA 操作示例及性能分析：
>>> 导入 torch
>>> 从 torch.cuda 导入 cudart 和 check_error
>>> 导入 os
>>>
>>> os.environ['CUDA_PROFILE'] = '1'
>>>
>>> def perform_cuda_operations_with_streams():
>>> 流 = torch.cuda.Stream()
>>> with torch.cuda.stream(流):
>>> x = torch.randn(100, 100, device='cuda')
>>> y = torch.randn(100, 100, device='cuda')
>>> z = torch.mul(x, y)
>>> return z
输入文本:
>>>
翻译:
>>>
>>> torch.cuda.synchronize()
>>> 打印("====== 开始 nsys 性能分析 ======")
>>> check_error(cudart().cudaProfilerStart())
>>> with torch.autograd.profiler.emit_nvtx():
>>> result = perform_cuda_operations_with_streams()
>>> print("CUDA operations completed.")
>>> check_error(torch.cuda.cudart().cudaProfilerStop())
>>> print("====== End nsys profiling ======")
运行此示例并保存性能分析信息，请执行以下命令：
>>> $ nvprof --profile-from-start off --csv --print-summary -o trace_name.prof -f -- python cudart_test.py
此命令将分析提供的脚本中的 CUDA 操作，并将性能分析信息保存到名为 `trace_name.prof` 的文件中。
性能分析信息将保存到名为 `trace_name.prof` 的文件中。
`--profile-from-start off` 选项确保仅在脚本中调用 `cudaProfilerStart` 之后才开始进行性能分析。
`--csv` 和 `--print-summary` 选项分别将性能分析输出格式化为 CSV 文件并打印摘要。
`--csv` 和 `--print-summary` 选项将性能分析输出格式化为 CSV 文件，并分别打印摘要。
`--csv` 和 `--print-summary` 选项将性能分析输出格式化为 CSV 文件，并分别打印摘要。
`-o` 选项指定输出文件名，`-f` 选项强制覆盖已存在的输出文件。
如果输出文件已存在，则强制覆盖。
"""
_lazy_init()
返回 _cudart
类
cuda 状态:
成功:
整型 = 0
ERROR_NOT_READY: 整型 = 34
类
Cuda 错误(RuntimeError):
定义 __init__(
我,
代码: int) ->
无:
msg = _cudart.cudaGetErrorString(_cudart.cuda 错误(
代码))
超级().__init__(f"{
信息} ({
代码})")
定义
检查错误(
资源: int) ->
无:
如果 res != _cudart.
cuda 错误.
成功:
抛出 CudaError(
资源)
类
_设备保护:
定义 __init__(
我,
索引: int):
我.
索引 =
索引
我.
前一个索引 = -1
定义
__进入__(
我):
我.
前一个索引 =
PyTorch.cuda.
交换设备(
我.
索引)
定义
__退出__(
我,
类型:
任何, value:
任何,
跟踪回溯:
任何):
我.idx =
PyTorch.cuda.
_可能交换设备(
我.prev_idx)
返回
假
[文档]类 device:
r"""上下文管理器，用于更改选定的设备。
Args:
device (torch.device 或 int)：选择设备索引。如果此参数为负整数或 None，则不执行任何操作。
此参数为负整数或 ``None``。
"""
"""
def __init__(self, device: Any):
self.idx = _get_device_index(device, optional=True)
self.prev_idx = -1
def __enter__(self):
self.prev_idx = torch.cuda._exchange_device(self.idx)
def __exit__(self, type: Any, value: Any, traceback: Any):
self.idx = torch.cuda._maybe_exchange_device(self.prev_idx)
return False
[文档]class device_of(设备):
r"""将当前设备更改为给定对象的设备上下文管理器。
您可以使用张量和存储作为参数。如果给定对象是
未分配到 GPU 上，这是一个空操作。
Args:
obj (Tensor 或 Storage)：在所选设备上分配的对象。
"""
def __init__(self, obj):
idx = obj.get_device() if obj.is_cuda else -1
super().__init__(idx)
[文档]def set_device(device: _device_t) -> None:
设置当前设备。
使用此函数不推荐，建议使用：any:`device`。在大多数
在某些情况下，最好使用 `CUDA_VISIBLE_DEVICES` 环境变量。
参数:
device (torch.device 或 int): 选择设备。此函数为空操作
如果此参数为负数。
"""
device = _get_device_index(device)
if 设备 >= 0:
torch._C._cuda_setDevice(设备)
[文档]def get_device_name(设备: Optional[_device_t] = None) -> str:
r"""获取设备名称。"""
参数:
设备（torch.device 或 int 或 str，可选）：返回的设备
name. 此函数在此参数为负数时为空操作
整数。它使用当前设备，由 :func:`~torch.cuda.current_device` 给出。
如果 :attr:`device` 为 ``None``（默认值）。
返回：
str：设备名称
"""
return get_device_properties(device).name
[文档]def get_device_capability(device: Optional[_device_t] = None) -> tuple[int, int]:
r"""获取设备的 CUDA 能力。
Args:
设备（torch.device 或 int 或 str，可选）：返回的设备
设备功能。如果此参数为空操作。
一个负整数。它使用当前设备，给定
:func:`~torch.cuda.current_device`，如果 :attr:`device` 是 ``None``
(默认).
返回：
元组(int, int)：设备的 major 和 minor cuda 能力
"""
prop = 获取设备属性(device)
return prop.major, prop.minor
[文档]def get_device_properties(device: Optional[_device_t] = None) -> _CudaDeviceProperties:
r"""获取设备的属性。
Args:
device (torch.device 或 int 或 str, 可选): 返回属性所对应的设备
设备的属性。它使用当前设备，由
func:`~torch.cuda.current_device` 提供的，如果 :attr:`device` 为 ``None``
（默认）。
返回值：
_CudaDeviceProperties：设备的属性
"""
_lazy_init() # 将定义 _get_device_properties
device = _get_device_index(device, optional=True)
如果设备小于 0 或设备大于等于设备数量()
抛出断言错误("无效的设备 ID")
返回_get_device_properties(device) # 忽略[name-defined]类型提示
[文档]def can_device_access_peer(device: _device_t, peer_device: _device_t) -> bool:
检查两个设备之间是否可以进行对等访问。
_lazy_init()
device = _get_device_index(device, optional=True)
peer_device = _get_device_index(peer_device)
如果设备小于 0 或设备大于等于设备数量()
抛出断言错误("无效的设备 ID")
如果对端设备小于 0 或对端设备大于等于设备数量()
抛出断言错误("无效的对端设备 ID")
return torch._C._cuda_canDeviceAccessPeer(device, peer_device)
[文档]
类
Stream 上下文:
r选择给定流的上下文管理器。
在其上下文中排队的所有 CUDA 内核都将被排队到所选流中。
流。
参数：
流（Stream）：选择的流。如果它是空操作，则此管理器无作用。
``None``。
.. 注意:: 流是按设备划分的。
"""
cur_stream: 可选["torch.cuda.Stream"]
定义 __init__(
我,
流:
可选[
torch.cuda.Stream
)]
我.
流 =
流
我.
索引 =
获取设备索引(
无, True)
如果
非
PyTorch.
算子.
是否正在脚本化():
如果
我.
索引 is
无:
我.
索引 = -1
我.
原始流 = (
无。
如果
非
PyTorch.
算子.
是否脚本()
否则
PyTorch.cuda.
默认流(
无)
)
我.
目标流 = (
无。
如果 not
PyTorch.
算子.
是脚本()
否则
PyTorch.cuda.
默认流(
无)
)
定义
__进入__(
我):
# 本地 cur_stream 变量用于类型细化
cur_stream = 我.
流
如果流为空或 CUDA 设备不可用，则返回
if 当前流 is
无。
或者
我.
索引 == -1:
返回
我.src_prev_stream =
PyTorch.cuda.current_stream(
无)
# 如果流不在当前设备上，则
# 在设备上设置当前流
if 我.
原始流.
设备 !=
当前流.
设备:
替换为
设备(
当前流.
设备):
我.
目标前流 =
PyTorch.cuda.
当前流(
当前流.
设备)
PyTorch.cuda.
设置流(
当前流)
定义
__退出__(
我,
类型:
任何, value:
任何,
追踪回溯:
任何):
# 本地 cur_stream 变量用于类型细化
cur_stream = 我.
流
如果流为空或没有可用的 CUDA 设备，则返回
if cur_stream is 无。
或者
我.idx == -1:
返回
在原始设备上重置流
# 和目标设备
if 我.
源前流.
设备 !=
当前流.
设备:
# 类型：忽略[联合属性]
PyTorch.cuda.
设置流(
我.
目标前流) # type: ignore[arg-type]
PyTorch.cuda.
设置流(
我.
前一个流)
# 类型：忽略[arg-type]
[文档]def 流(流: Optional["torch.cuda.Stream"]) -> 流上下文:
r"""围绕选择给定流的上下文管理器 StreamContext 进行包装。
参数：
stream (Stream): 选定的流。如果为 ``None``，则此管理器为空操作。
``None``。
.. 注意::
在急切模式下流是 Stream 类型的，而在即时编译（JIT）中则是
自定义类 ``torch.classes.cuda.Stream`` 的一个对象。
"""
返回 StreamContext(stream)
定义
通过 ID 设置流(
流 ID,
设备索引,
设备类型):
r设置指定流 ID、设备索引和设备类型的流
设备类型
Args: stream_id (int): 流池中的流 ID
device_index (int): topo 中的设备索引
设备类型（int）：枚举设备类型
"""
PyTorch._C._cuda_setStream(
流 ID=
流 ID,
设备索引=
设备索引,
设备类型=
设备类型,
)
[文档]def set_stream(stream: Stream):
r"""设置当前流。这是一个设置流的包装 API。
使用此函数被建议，以 ``stream`` 优先
上下文管理器。
参数：
stream (Stream): 选定的流。此函数为空操作
如果此参数为 `None`。
"""
如果流为 `None`：
返回
_通过 ID 设置流(
流 ID=stream.stream_id,
设备索引=stream.device_index,
设备类型=stream.device_type,
)
定义
解析可见设备() ->
联盟[
列表[int
]
列表[str
]]
r解析 CUDA_VISIBLE_DEVICES 环境变量。
变量 =
操作系统.getenv(
CUDA_VISIBLE_DEVICES)
if PyTorch.
版本.hip:
hip_devices = 操作系统.getenv(
HIP_VISIBLE_DEVICES)
rocr_devices = 操作系统.getenv(
ROCR_VISIBLE_DEVICES)
注意，如果同时设置了 HIP 和 ROCR 环境变量，则必须小心处理
不同的含义。这两个环境变量都接受整数列表或
# UUID 列表。首先处理 ROCR 环境变量，然后减少
# HIP 可以从中选择的 GPU 数量。
if rocr_devices is not 无:
rocr_count = len(rocr_devices.分割(","))
if hip_devices is not 无:
检查是否设置了两个环境变量
if len(虚拟设备.
分割(",")) >
rocr 计数:
抛出 RuntimeError(
HIP_VISIBLE_DEVICES 包含比 ROCR_VISIBLE_DEVICES 更多的设备
)
HIP_VISIBLE_DEVICES 优先于 ROCR_VISIBLE_DEVICES
变量 =
hip 设备
else:
返回
列表(
范围(rocr_count))
elif 脊柱设备 is not
无:
变量 =
虚拟现实设备
if 变量 is
无:
返回
列表(
范围(64))
定义 _strtoul(s: str) -> int:
返回以序列字符串开始的-1 或正整数
if not s:
返回 -1
为 idx, c
在
列举(s):
if not (c.isdigit() 或者 (idx == 0
以及 c
在 "+-")):
断开
if idx + 1 == len(s):
idx += 1
返回 int(s
[idx]) if idx > 0
否则 -1
定义
解析带前缀的列表(
列表: str,
前缀: str) ->
列表[str
]
rcs: 列表[str] =
输入文本为空，请提供需要翻译的文本
为
元素
在
列表.
分割(
逗号):
重复 ID 导致结果为空集
if 元素
在 rcs:
返回
投射(
列表[str
]
[]
# 任何非前缀的内容都将被忽略
if not 元素.
以...开头(
前缀):
断开
rcs.添加(
元素)
返回 rcs
if var.以...开头(
GPU-):
返回
解析带前缀的列表(
变量,
GPU-)
if 变量.
以...开头("MIG-"):
返回
解析带有前缀的列表(
变量, "MIG-")
# CUDA_VISIBLE_DEVICES 使用类似 strtoul 的方法
# 使得 `1gpu2,2ampere` 等同于 `1,2`
rc: 列表[int] =
输入文本为空，请提供需要翻译的文本
为
元素
在
变量.
分割(
逗号):
x = _strtoul(元素.strip())
# 重复的序数结果为空集
if x 在 rc:
返回
投射(
列表[int
]
空数组
负值会终止序列
if x < 0:
断开
rc.添加(x)
返回 rc
定义 _raw_device_count_amdsmi() -> int:
if not _HAS_PYNVML: 如果 amdsmi 不可用
返回 -1
尝试:
amdsmi.amdsmi 初始化()
除了 amdsmi.
AmdSmi 异常
作为 e:
警告.
警告(f
无法初始化 amdsmi - 错误代码：{e.
错误码}")
返回 -1
套接字句柄 = amdsmi.amdsmi_get_processor_handles()
返回 len(socket_handles)
定义 _raw_device_count_nvml() -> int:
r"返回 NVML 报告的设备数量，如果 NVML 发现/初始化失败则返回负值。"
from ctypes 导入 byref, c_int, CDLL
nvml_h = CDLL(libnvidia-ml.so.1)
rc = nvml_h.nvml 初始化()
if rc != 0:
警告.
警告(
"无法初始化 NVML")
返回 -1
dev_count = c_int(-1)
rc = nvml_h.nvmlDeviceGetCount_v2(byref(开发者数量))
if rc != 0:
警告.
警告(
"无法获取 nvml 设备数量")
返回 -1
删除 nvml_h
返回
开发者数量.
值
定义
_原始设备 UUID_amdsmi() ->
可选[
列表[str
]]
from ctypes 导入
通过引用,
c 整型, c_void_p, CDLL,
创建字符串缓冲区
if not _HAS_PYNVML: # 如果 amdsmi 不可用
返回
无。
尝试:
amdsmi.amdsmi 初始化()
除了 amdsmi.
AmdSmi 异常:
警告.
警告(
"无法初始化 amdsmi")
返回
无。
尝试:
网络句柄 = amdsmi.
获取处理器句柄()
开发者数量 = len(
套接字句柄)
除了 amdsmi.
AmdSmiException 异常:
警告.
警告(
"无法获取 amdsmi 设备数量")
返回
无。
uuids: 列表[str] =
输入文本为空，请提供需要翻译的文本
为 idx
在
范围(
开发者数量):
尝试:
处理器 = amdsmi.
获取处理器句柄()[idx]
除了 amdsmi.
AmdSmi 异常:
警告.
警告(
无法获取 amd 设备处理器)
返回
无。
尝试:
uuid = amdsmi.amdsmi_get_gpu_asic_info(处理器
]
asic 序列号
]
2:
] # 移除序列号前的 0x 前缀
除了 amdsmi.AmdSmiException:
警告.
警告(
"无法获取 amd 设备的 uuid")
返回
无。
uuids.添加(
str(uuid).小写()
) # 小写以匹配预期的 HIP_VISIBLE_DEVICES uuid 输入
返回 uuids
定义 _raw_device_uuid_nvml() ->
可选[
列表[str
]]
r"""返回由 NVML 报告的设备 UUID 列表，如果 NVM 发现/初始化失败则返回 None。"""
from ctypes 导入 byref, c_int, c_void_p, CDLL, create_string_buffer
nvml_h = CDLL(libnvidia-ml.so.1)
rc = nvml_h.nvml 初始化()
if rc != 0:
警告.
警告(
"无法初始化 NVML")
返回
无。
设备数量 = c_int(-1)
rc = nvml_h.nvmlDeviceGetCount_v2(引用(
开发者数量))
if rc != 0:
警告.
警告(
"无法获取 nvml 设备数量")
返回
无。
uuids: 列表[str] =
输入文本为空，请提供需要翻译的文本
为 idx
在 range(dev_count.value):
dev_id = c_void_p()
rc = nvml_h.nvmlDeviceGetHandleByIndex_v2(索引,
按引用(
设备 ID))
if rc != 0:
警告.
警告(
无法获取设备句柄)
返回
无。
缓冲区长度 = 96
缓冲区 =
创建字符串缓冲区(
缓冲区长度)
返回码 = nvml_h.nvmlDeviceGetUUID(dev_id, buf, buf_len)
if rc != 0:
警告.
警告(
"无法获取设备 UUID")
返回
无。
uuids.添加(
缓冲区.
原始.
解码(
ASCII).strip("\0"))
删除 nvml_h
返回 uuids
定义 _transform_uuid_to_ordinals(
候选人:
列表[str
]
uuid:
列表[str]) ->
列表[int
]
r"给定一组部分 uuid 和已知 uuid 列表，构建一组序数，排除模糊的部分 ID。"
定义 uuid_to_ordinal(
候选: str, uuids:
列表[str]) -> int:
最佳匹配 = -1
为
索引, uuid
在
列举(uuids):
if not uuid.以...开头(
候选人):
continue
# 模糊的候选人
if 最佳匹配 != -1:
返回 -1
最佳匹配 =
索引
返回
最佳匹配
返回码:
列表[int] =
输入文本为空，请提供需要翻译的文本
为
候选人
在
候选人名单:
if PyTorch.
版本.
脚:
候选人 =
候选人.
替换(
"GPU-", 输入文本翻译为简体中文为："", 1
) # 移除 GPU 前缀以匹配 amdsmi asic 序列号
idx = uuid_to_ordinal(候选人, uuids)
# 第一个无效序号停止解析
if idx < 0:
断开
# 空集
if idx 在 rc:
返回
投射(
列表[int
]
[]
rc.添加(
索引)
返回 rc
定义 _device_count_amdsmi() -> int:
可见设备 =
解析可见设备()
if not 可见设备:
返回 0
尝试:
if 类型(
可见设备[0]) is str:
uuids = _原始设备 UUID_amdsmi()
if uuids is 无:
返回 -1
# 创建可见设备的字符串版本以避免 mypy 警告
visible_device_str = 投射(
列表[str
]
可见设备)
可见设备 = _transform_uuid_to_ordinals(
可见设备字符串, uuids)
else:
原始计数 =
_原始设备计数_amdsmi()
if 原始计数 <= 0:
返回
原始计数
# 修剪列表至最大可用设备
为
索引, val
在
列举(
可见设备):
if 投射(int, val)
≥
原始计数:
返回
索引
除了 OSError:
返回 -1
除了
属性错误:
返回 -1
返回 len(
可见设备)
定义 _device_count_nvml() -> int:
r考虑 CUDA_VISIBLE_DEVICES 环境变量后，返回 NVML 报告的设备数量。
如果 NVML 发现或初始化失败，则返回负值。
"""
可见设备 =
_解析可见设备()
if not 可见设备:
返回 0
尝试:
if 类型(
可见设备[0]) is str:
# 跳过 MIG 解析
if 可见设备[0].
以...开头("MIG-"):
返回 -1
uuids = _raw_device_uuid_nvml()
if uuids is 无:
返回 -1
可见设备 =
_将 UUID 转换为序号(
投射(
列表[str
]
可见设备),
UUID 列表
)
else:
原始计数 =
_原始设备计数_nvml()
if 原始计数 <= 0:
返回
原始计数
修剪列表至最大可用设备
为
索引, val
在
列举(
可见设备):
if 投射(int, val)
≥
原始计数:
返回
索引
除了 OSError:
返回 -1
除了
属性错误:
返回 -1
返回 len(
可见设备)
定义 _get_nvml_device_index(
设备:
可选[
联盟[int,
设备]]) -> int:
r返回设备的 NVML 索引，考虑 CUDA_VISIBLE_DEVICES。
索引 =
获取设备索引(
设备,
可选的=True)
可见设备 =
解析可见设备()
if 类型(
可见设备[0]) is str:
uuids = _原始设备 UUID_nvml()
if uuids is 无:
抛出 RuntimeError(
"无法获取设备 UUID")
可见设备 =
_将转换 UUID 为序号(
投射(
列表[str
]
可见设备), uuids
)
visible_devices = 投射(
列表[int
] visible_devices)
if idx < 0 或者 idx
≥ len(
可见设备):
抛出 RuntimeError(
f"设备"{
索引}
不可见 (CUDA_VISIBLE_DEVICES={
可见设备})"
)
返回
可见设备[
索引]
_缓存设备数量:
可选[int] =
无。
[文档]def device_count() -> int:
r"""返回可用的 GPU 数量。"""
global _cached_device_count
if not _is_compiled():
返回 0
如果_cached_device_count 不是 None：
返回_cached_device_count
# 如果是 rocm（不支持），则绕过_device_count_nvml()
nvml_count = _device_count_amdsmi() if torch.version.hip else _device_count_nvml()
r = torch._C._cuda_getDeviceCount() if nvml_count < 0 else nvml_count
# NB: Do not cache the device count prior to CUDA initialization, because
# the number of devices can change due to changes to CUDA_VISIBLE_DEVICES
在 CUDA 初始化之前设置。
如果_initialized:
_cached_device_count = r
返回 r
[文档]def get_arch_list() -> list[str]:
r"""返回该库编译支持的 CUDA 架构列表。"""
if not is_available():
return []
arch_flags = torch._C._cuda_getArchFlags()
if arch_flags is None:
return []
return arch_flags.split()
[文档]def get_gencode_flags() -> str:
r"""返回此库编译时使用的 NVCC gencode 标志。"""
arch_list = get_arch_list()
if len(arch_list) == 0:
return ""
arch_list_ = [arch.split("_") for arch in arch_list]
return " ".join(
[
-gencode compute=compute_{架构},code={类型}_{架构}
for (类型, 架构) in 架构列表_
]
)
[文档]def current_device() -> int:
返回当前选中设备的索引。
_lazy_init()
return torch._C._cuda_getDevice()
[文档]def synchronize(device: _device_t = None) -> None:
r"""等待所有 CUDA 设备上的所有流中的所有内核完成。
Args:
device (torch.device 或 int，可选)：要同步的设备。
使用当前设备，由 :func:`~torch.cuda.current_device` 提供，
如果 :attr:`device` 为 ``None``（默认）。
"""
_lazy_init()
使用 torch.cuda.device(device):
返回 torch._C._cuda_synchronize()
[文档]def ipc_collect():
r"""强制收集 CUDA IPC 释放后的 GPU 内存。"""
.. 注意：
检查是否有任何可从内存中清理的 CUDA 张量。强制
关闭用于引用计数的共享内存文件（如果不存在）
活跃计数器。当生产者进程停止主动发送时很有用
索引张量并希望释放未使用的内存。
"""
_lazy_init()
return torch._C._cuda_ipc_collect()
[文档]def current_stream(device: Optional[_device_t] = None) -> Stream:
返回给定设备的当前 :class:`Stream`。
参数:
device (torch.device 或 int, 可选): 选择设备。返回
当前设备所选定的 :class:`Stream`，如果 :attr:`device` 为 ``None``（默认），
则由 :func:`~torch.cuda.current_device` 提供。
（默认）。
"""
_lazy_init()
streamdata = torch._C._cuda_getCurrentStream(
_get_device_index(device, optional=True)
)
return Stream(
stream_id=streamdata[0], device_index=streamdata[1], device_type=streamdata[2]
)
[文档]def default_stream(device: Optional[_device_t] = None) -> Stream:
返回给定设备的默认 :class:`Stream`。
参数:
device (torch.device 或 int，可选): 选择设备。返回
当前设备的默认 :class:`Stream`，当前设备由
func:`~torch.cuda.current_device`，如果 :attr:`device` 为 `None`
(默认)。
"""
_lazy_init()
streamdata = torch._C._cuda_getDefaultStream(
_get_device_index(device, optional=True)
)
return Stream(
stream_id=streamdata[0]，device_index=streamdata[1]，device_type=streamdata[2]
)
[文档]def get_stream_from_external(
data_ptr: int, device: Optional[_device_t] = None
) -> Stream:
r"""从外部分配的 CUDA 流返回一个 :class:`Stream`。
此函数用于将分配在其他库中的流包装起来
为了促进数据交换和多库交互。
.. note:: 此函数不管理流的生命周期，由用户
责任在保持引用的流在此返回时存活
流正在使用中。
Args:
数据指针(int)：表示 `cudaStream_t` 值的整型表示。
是外部分配的。
设备（torch.device 或 int，可选）：流所在的设备
原本已分配。如果设备指定错误，
后续使用此流进行的启动可能会失败。
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
_lazy_init()
streamdata = torch._C._cuda_getStreamFromExternal(
data_ptr, _get_device_index(device, optional=True)
)
return Stream(
stream_id=streamdata[0], device_index=streamdata[1], device_type=streamdata[2]
)
[文档]def current_blas_handle():
返回指向当前 cuBLAS 句柄的 cublasHandle_t 指针
_lazy_init()
return torch._C._cuda_getCurrentBlasHandle()
[文档]def set_sync_debug_mode(debug_mode: Union[int, str]) -> None:
r"""设置 cuda 同步操作的调试模式。
Args:
debug_mode(str or int): 如果为 "default" 或 0，则在同步操作上不报错或警告，
如果 "warn" 或 1，则在同步操作时发出警告，如果 "error" 或 2，则在同步操作时出错。
警告：
这是一个实验性功能，并非所有同步操作都会触发警告或错误。
特别地，torch.distributed 和 torch.sparse 命名空间中的操作尚未涵盖。
"""
_lazy_init()
如果 debug_mode 是字符串类型：
如果 debug_mode 等于"默认":
debug 模式 = 0
elif debug 模式 == "warn":
debug 模式 = 1
elif debug 模式 == "error":
debug_mode = 2
else:
raise RuntimeError(
"无效的 debug_mode 值，期望的值为 `default`、`warn` 或 `error`"
)
torch._C._cuda_set_sync_debug_mode(debug_mode)
[文档]def get_sync_debug_mode() -> int:
r"""返回当前 cuda 同步操作调试模式的值。"""
_lazy_init()
return torch._C._cuda_get_sync_debug_mode()
定义 _get_pynvml_handler(
设备:
可选[
联盟[
设备, int]] =
无):
if not _HAS_PYNVML:
抛出
模块未找到错误(
"pynvml 似乎未安装或无法导入。"
) from _PYNVML_ERR
from pynvml 导入 NVMLError_DriverNotLoaded
尝试:
pynvml.nvmlInit()
除了 NVMLError_DriverNotLoaded
作为 e:
抛出 RuntimeError(
"cuda 驱动无法加载，是否已启用 cuda？") from e
设备 =
获取 nvml 设备索引(
设备)
处理 = pynvml.
通过索引获取 nvml 设备句柄(
设备)
返回
处理
定义
_获取_amdsmi 处理程序(
设备:
可选的[
联盟[
设备, int]] =
无):
如果 not _HAS_PYNVML:
抛出
模块未找到错误(
"amdsmi 似乎未安装或无法导入。"
) from _PYNVML_ERR
尝试:
amdsmi.amdsmi 初始化()
除了 amdsmi.
AmdSmi 异常
作为 e:
抛出 RuntimeError(
"amdsmi 驱动无法加载，需要>=ROCm5.6 安装"
) from e
设备 = _get_amdsmi_device_index(
设备)
处理 = amdsmi.amdsmi_get_processor_handles()[
设备]
返回 handle
定义 _get_amdsmi_device_index(
设备:
可选的[
联盟[int,
设备]]) -> int:
r考虑可见设备，返回设备的 amdsmi 索引。
索引 =
_获取设备索引(
设备,
可选=True)
可见设备 =
_解析可见设备()
如果
类型(
可见设备[0]) is str:
uuids = _原始设备 UUID_amdsmi()
如果 uuids is
无:
抛出 RuntimeError(
无法获取设备 UUIDs)
可见设备字符串 =
演员表(
列表[str
]
可见设备
) 为 mypy 创建 str 变量
可见设备 = _transform_uuid_to_ordinals(
可见设备字符串, uuids)
索引映射 =
字典(
列举(
演员表(
列表[int
]
可见设备)))
如果 idx not
在 idx_map:
抛出 RuntimeError(
f"设备"{
索引}
"不可见 (HIP_VISIBLE_DEVICES=){
可见设备})"
)
返回
索引映射[
索引]
定义 _get_amdsmi_device_memory_used(
设备:
可选的[
联盟[
设备, int]] =
无) -> int:
处理 =
_获取_amdsmi 处理程序()
设备 =
_获取_amdsmi 设备索引(
设备)
# amdsmi_get_gpu_vram_usage 返回内存使用量（兆字节）
内存兆字节 = amdsmi.
获取 GPU 显存使用情况(
处理程序
]
vram_used]
mem_bytes = mem_mega_bytes * 1024 * 1024
返回
内存字节
定义
_获取 AMDsmi 内存使用情况(
设备:
可选的[
联盟[
设备, int]] =
无) -> int:
处理 =
_获取_amdsmi 处理程序()
设备 =
_获取_amdsmi 设备索引(
设备)
处理 = amdsmi.amdsmi_get_processor_handles()[
设备]
返回 amdsmi.amdsmi_get_gpu_activity(
处理
）[
umc_activity]
定义 _get_amdsmi_utilization(
设备:
可选的[
联盟[
设备, int]] =
无) -> int:
处理 =
_获取_amdsmi 处理程序()
设备 =
_获取_amdsmi 设备索引(
设备)
处理 = amdsmi.amdsmi_get_processor_handles()[
设备]
返回 amdsmi.amdsmi_get_gpu_activity(
处理
）[
"显卡活动"]
定义
_获取 AMDsmi 温度(
设备:
可选的[
联盟[
设备, int]] =
无) -> int:
处理 =
_获取_amdsmi 处理程序(
设备)
返回 amdsmi.
amdsmi 获取温度指标(
处理,
amdsmi.AmdSmi 温度类型.
交界,
amdsmi.AmdSmi 温度指标.
当前,
)
定义 _get_amdsmi_power_draw(
设备:
可选的[
联盟[
设备, int]] =
无) -> int:
处理 =
_获取_amdsmi 处理程序(
设备)
网络电源 = amdsmi.
获取 AMD 电源信息(
处理
)]
"平均插槽功率"]
如果
插座电源 !=
无效:
返回
插座电源
else:
返回 amdsmi.
获取 AMD 电源信息(
处理
)]
"当前插槽电源"]
定义 _get_amdsmi_clock_rate(
设备:
可选的[
联盟[
设备, int]] =
无) -> int:
处理 = _get_amdsmi_handler(
设备)
时钟信息 = amdsmi.
amdsmi 获取时钟信息(
处理, amdsmi.AmdSmiClkType.GFX)
如果 "cur_clk"
在
时钟信息:
ROCm 6.2 弃用
返回
时钟信息[
"当前时钟"]
else:
返回
时钟信息["clk"]
[docs]def 设备内存使用量(device: Optional[Union[Device, int]] = None) -> int:
r"""根据 `nvidia-smi` 或 `amd-smi` 返回使用的全局（设备）内存量，单位为字节。
Args:
device (torch.device 或 int，可选）：选定的设备。返回
当前设备的统计信息，由 :func:`~torch.cuda.current_device` 提供，
如果 :attr:`device` 为 ``None``（默认）。
"""
如果没有 torch.version.hip:
处理器 = _get_pynvml_handler()
设备 = _get_nvml_device_index(device)
处理器 = pynvml.nvmlDeviceGetHandleByIndex(device)
return pynvml.nvmlDeviceGetMemoryInfo(handle).used
else:
return _get_amdsmi_device_memory_used(device)
[文档]def memory_usage(device: Optional[Union[Device, int]] = None) -> int:
返回在过去样本期间，全球（设备）内存被读取或写入的时间百分比
由 `nvidia-smi` 提供的
参数：
device (torch.device 或 int，可选)：选定的设备。返回
当前设备的统计信息，由 :func:`~torch.cuda.current_device` 提供，
如果 :attr:`device` 为 ``None``（默认）。
警告：每个样本周期可能在 1 秒到 1/6 秒之间，
取决于查询的产品。
"""
如果没有 torch.version.hip:
处理器 = _get_pynvml_handler()
设备 = _get_nvml_device_index(device)
handle = pynvml.nvmlDeviceGetHandleByIndex(device)
return pynvml.nvmlDeviceGetUtilizationRates(handle).memory
else:
return _get_amdsmi_memory_usage(device)
[文档]def utilization(device: Optional[Union[设备, int]] = None) -> int:
返回过去样本期间内，一次或多次发生的时间百分比
更多内核正在 GPU 上执行，如`nvidia-smi`所示。
参数:
设备（torch.device 或 int，可选）：选定的设备。返回
当前设备的统计信息，由:func:`~torch.cuda.current_device`给出，
如果:attr:`device`为`None`（默认）。
警告：每个样本周期可能在 1 秒到 1/6 秒之间，
根据查询的产品而定。
"""
如果没有 torch.version.hip:
处理器 = _get_pynvml_handler(device)
device = _get_nvml_device_index(device)
handle = pynvml.nvmlDeviceGetHandleByIndex(device)
return pynvml.nvmlDeviceGetUtilizationRates(handle).gpu
else:
return _get_amdsmi_utilization(device)
[文档]def temperature(device: Optional[Union[Device, int]] = None) -> int:
r"""返回 GPU 传感器平均温度，单位为摄氏度（摄氏度）。
平均温度是根据`nvidia-smi`提供的过去样本周期计算的。
Args:
设备（torch.device 或 int，可选）：选定的设备。返回
当前设备的统计信息，由 :func:`~torch.cuda.current_device` 提供，
如果 :attr:`device` 为 ``None``（默认）。
警告：每个样本周期可能在 1 秒到 1/6 秒之间，
具体取决于所查询的产品。
"""
如果没有 torch.version.hip:
handle = _get_pynvml_handler(device)
# 0 表示 GPU 芯片温度传感器。
return pynvml.nvmlDeviceGetTemperature(handle, 0)
else:
return _get_amdsmi_temperature(device)
[文档]def power_draw(device: Optional[Union[Device, int]] = None) -> int:
r"""返回过去样本周期内 GPU 传感器的平均功耗（毫瓦），由`nvidia-smi`为 Fermi 或更新的完全支持设备提供。
over the past sample period as given by `nvidia-smi` for Fermi or newer fully supported devices.
Args:
设备（torch.device 或 int，可选）：选定的设备。返回
当前设备的统计信息，由 :func:`~torch.cuda.current_device` 提供，
如果 :attr:`device` 为 ``None``（默认）。
警告：每个样本周期可能在 1 秒到 1/6 秒之间，
具体取决于所查询的产品。
"""
如果没有 torch.version.hip:
handle = _get_pynvml_handler(device)
return pynvml.nvmlDeviceGetPowerUsage(handle)
else:
return _get_amdsmi_power_draw(device)
[文档]def clock_rate(device: Optional[Union[Device, int]] = None) -> int:
r"""返回过去采样周期内 GPU SM 的时钟速度（兆赫兹），由`nvidia-smi`提供。
Args:
device (torch.device 或 int，可选)：选定的设备。返回
当前设备的统计信息，由 :func:`~torch.cuda.current_device` 提供，
如果 :attr:`device` 为 ``None``（默认）。
警告：每个样本周期可能在 1 秒到 1/6 秒之间，
取决于查询的产品。
"""
如果没有 torch.version.hip:
处理器 = _get_pynvml_handler(device)
返回 pynvml.nvmlDeviceGetClockInfo(handle, 1)
else:
返回_get_amdsmi_clock_rate(device)
定义
获取设备(
设备:
联盟[int, str,
PyTorch.
设备]) ->
PyTorch.
设备:
r返回从传入的设备中获取的 torch.device 类型对象。
参数：
device (torch.device 或 int): 选定的设备。
"""
如果 isinstance(
设备, str):
设备 =
PyTorch.
设备(
设备)
elif isinstance(设备, int):
设备 =
PyTorch.
设备("cuda",
设备)
返回
设备
定义 _get_generator(
设备:
PyTorch.
设备) ->
PyTorch._C.
生成器:
r返回给定设备的 CUDA 生成器对象。
参数：
device (torch.device): 选定的设备。
"""
idx = 设备.
索引
如果 idx is
无:
idx = 当前设备()
返回
PyTorch.cuda.
默认生成器[
索引]
定义
设置随机数生成器状态偏移(
偏移: int,
设备:
联盟[int, str,
PyTorch.
设备] = "cuda"
) -> 无:
r设置指定 GPU 的随机数生成器状态偏移。
参数：
偏移量（int 类型）：所需的偏移量
设备（torch.device 或 int，可选）：设置 RNG 状态的设备。
默认：``'cuda'``（即``torch.device('cuda')``，当前 CUDA 设备）。
"""
最终设备 =
获取设备(
设备)
定义 cb():
默认生成器 =
获取生成器(
最终设备)
默认生成器.
设置偏移量(
偏移量)
懒调用(cb)
定义
获取随机数生成器状态偏移(
设备:
联盟[int, str,
PyTorch.
设备] =
cuda) -> int:
r返回指定 GPU 的随机数生成器状态偏移量。
参数：
device (torch.device 或 int，可选)：返回 RNG 状态偏移量的设备。
默认：``'cuda'``（即``torch.device('cuda')``，当前 CUDA 设备）。
..警告：
此函数会积极初始化 CUDA。
"""
_延迟初始化()
最终设备 =
_获取设备(
设备)
默认生成器 =
_获取生成器(
最终设备)
返回
默认生成器.
获取偏移量()
from .内存
导入 * # noqa: F403
from .随机
导入 * # noqa: F403
################################################################################
# 定义存储和张量类
################################################################################
@staticmethod # type: ignore[misc]
定义 _lazy_new(cls, *
参数, **
关键字参数):
_lazy_init()
# 我们可能需要再次调用懒初始化，如果我们是一个分叉的子进程
# 删除 _CudaBase.__new__
返回
超级(_CudaBase, cls).__new__(cls, *
参数, **
关键字参数)
类 _CudaBase:
is_cuda = 真实
is_sparse = 假
定义
类型(
我, *
参数, **
关键字参数):
我们可以使用协议来告诉 mypy self 具有`get_device`方法
但它仅在 Python >= 3.8 的 typing 模块中可用
或者 Python >= 3.6 的 typing_extensions 模块中
替换为
设备(
我.get_device()):
# 类型：忽略[已定义]
返回
超级().
类型(*
参数, **
关键字参数) # type: ignore[misc]
__新__ = _lazy_new
from torch.storage 导入 _LegacyStorage, _warn_typed_storage_removal
类 _CudaLegacyStorage(_LegacyStorage):
@classmethod
定义 from_buffer(cls, *
参数, **
关键字参数):
警告：类型化存储移除()
抛出 RuntimeError(
"from_buffer: 不适用于 CUDA 存储")
@classmethod
定义 _new_with_weak_ptr(cls, *
参数, **
关键字参数):
抛出 RuntimeError(
"_new_with_weak_ptr: 不适用于 CUDA 存储")
@classmethod
定义 _new_shared_filename(cls,
经理, obj,
大小, *,
设备=
无,
数据类型=
无):
抛出 RuntimeError(
"_new_shared_filename: 不适用于 CUDA 存储")
类
字节存储(_CudaLegacyStorage):
@classproperty
定义
数据类型(
我):
警告：类型化存储移除()
返回
我._dtype
@classproperty
定义 _dtype(
我):
返回
PyTorch.uint8
类
双倍存储(_CudaLegacyStorage):
@classproperty
定义
数据类型(
我):
警告：类型化存储移除()
返回
我.
数据类型
@classproperty
定义
数据类型(
我):
返回
PyTorch.double
类
浮点存储(_CudaLegacyStorage):
@classproperty
定义
数据类型(
我):
警告：类型化存储移除()
返回
我._dtype
@classproperty
定义
数据类型(
我):
返回
PyTorch.
浮点数
类
半字节存储(_CudaLegacyStorage):
@classproperty
定义
数据类型(
我):
警告：类型化存储移除()
返回
我.
数据类型
@classproperty
定义
数据类型(
我):
返回
PyTorch.
半精度
类
长存储(_CudaLegacyStorage):
@classproperty
定义
数据类型(
我):
警告：类型化存储移除()
返回
我.
数据类型
@classproperty
定义
数据类型(
我):
返回
PyTorch.
长的
类
整数存储(_CudaLegacyStorage):
@classproperty
定义
数据类型(
我):
警告：类型化存储移除()
返回
我.
数据类型
@classproperty
定义 _dtype(
我):
返回
PyTorch.
整型
类
短存储(_CudaLegacyStorage):
@classproperty
定义
数据类型(
我):
警告：类型化存储移除()
返回
我._dtype
@classproperty
定义 _dtype(
我):
返回
PyTorch.
简短
类
字符存储(_CudaLegacyStorage):
@classproperty
定义
数据类型(
我):
警告：类型化存储移除()
返回
我._dtype
@classproperty
定义 _dtype(
我):
返回
PyTorch.int8
类
布尔存储(_CudaLegacyStorage):
@classproperty
定义
数据类型(
我):
警告：类型化存储移除()
返回
我._dtype
@classproperty
定义 _dtype(
我):
返回
PyTorch.
布尔值
类
BFloat16 存储(_CudaLegacyStorage):
@classproperty
定义
数据类型(
我):
警告：类型化存储移除()
返回
我._dtype
@classproperty
定义 _dtype(
我):
返回
PyTorch.bfloat16
类
复杂双存储(_CudaLegacyStorage):
@classproperty
定义
数据类型(
我):
警告：类型化存储移除()
返回
我._dtype
@classproperty
定义 _dtype(
我):
返回
PyTorch.cdouble
类
ComplexFloat 存储(_CudaLegacyStorage):
@classproperty
定义
数据类型(
我):
警告：类型化存储移除()
返回
我._dtype
@classproperty
定义 _dtype(
我):
返回
PyTorch.cfloat
删除
_遗留存储
删除 _CudaLegacyStorage
PyTorch._storage_classes.
添加(
双倍存储)
PyTorch._storage_classes.
添加(
浮点存储)
PyTorch._storage_classes.
添加(
长存储)
PyTorch._storage_classes.
添加(
整数存储)
PyTorch._storage_classes.
添加(
短存储)
PyTorch._storage_classes.
添加(
字符存储)
PyTorch._storage_classes.
添加(
字节存储)
PyTorch._storage_classes.
添加(
半字节存储)
PyTorch._storage_classes.
添加(
布尔存储)
PyTorch._storage_classes.
添加(
BFloat16 存储)
PyTorch._storage_classes.
添加(
复杂双存储)
PyTorch._storage_classes.
添加(
ComplexFloat 存储)
类
_Triton 内核封装:
这是一个简单的包装器，用于存储一些测试目的的元数据。
定义 __init__(
我,
内核):
我.
内核 =
内核
我.
内核调用 =
假
定义
__调用__(
我, *
参数, **
关键字参数):
res = 我.
内核(*
参数, **
关键字参数)
我.
内核调用 =
真实
返回 res
定义
注册三叉戟内核():
如果
PyTorch.
部署运行中():
返回
@_WrappedTritonKernel
定义
内核实现(*
参数, **
关键字参数):
from torch.sparse._triton_ops 导入 bsr_dense_mm
返回 bsr_dense_mm(*
参数,
跳过检查=True, **
关键字参数)
@_WrappedTritonKernel
定义 addmm_kernel_impl(*
参数, **
关键字参数):
from torch.sparse._triton_ops 导入 bsr_dense_addmm
返回 bsr_dense_addmm(*
参数, skip_checks=True, **
关键字参数)
具有 triton = importlib.
工具.
查找规范(
triton) is not
无。
如果
具有 triton:
PyTorch.
三叉戟库.
注册操作(
"_triton_bsr_dense_mm_out",
"_triton_bsr_dense_mm_out(Tensor bsr, Tensor dense, *, Tensor(a!) out) -> Tensor(a!)",
kernel_impl,
"SparseCsrCUDA",
)
PyTorch.
三叉戟库.
注册操作(
"_triton_bsr_dense_addmm_out",
(
"_triton_bsr_dense_addmm_out(Tensor input, Tensor bsr, Tensor dense,"
" *, Scalar beta, Scalar alpha, Tensor(a!) out) -> Tensor(a!)"
),
addmm_kernel_impl,
SparseCsrCUDA,
)
_lazy_call(_register_triton_kernels)
from . 导入
转换, jiterator, nvtx,
分析器,
稀疏,
可调的
全部 = [
# 类型化存储和张量
"BFloat16 存储",
"BFloat16 张量",
"布尔存储",
"布尔张量",
"字节存储",
"字节张量",
"字符存储",
"字符张量",
"复杂数值存储",
"复杂数值浮点存储",
"双精度存储",
"双精度张量",
浮点存储,
浮点张量,
半精度存储,
半精度张量,
"内存存储",
"整数张量",
"长存储",
"长张量",
"短存储",
"短张量",
"CUDA 图",
"CUDA 错误",
"DeferredCudaCallError",
"事件",
"外部流",
"流",
"沉浸式上下文",
"增强型模块",
"缓存分配器分配",
"缓存分配器删除",
"缓存分配器启用",
"设备能否访问对等",
"检查错误",
"CUDA 状态",
"cudart",
"current_blas_handle",
"current_device",
current_stream,
"default_generators",
"默认流",
"设备",
"设备数量",
"设备内存使用",
"设备的",
清空缓存,
获取分配器后端,
CUDA 可插拔分配器,
更改当前分配器,
"获取架构列表",
"获取设备功能",
"获取设备名称",
"获取设备属性",
get_gencode_flags,
get_per_process_memory_fraction,
get_rng_state,
get_rng_state_all,
"从外部获取流",
"获取同步调试模式",
"图",
"图池句柄",
"图们",
"有半",
"has_magma",
"host_memory_stats",
"host_memory_stats_as_nested_dict",
"init",
"初始种子",
"IPC 收集",
"是否可用",
"是否支持 bf16",
"是否正在捕获当前流",
"已初始化",
"支持 TF32",
"jiterator",
"列出 GPU 进程",
"创建图形化可调用对象",
"手动播种",
"手动播种全部",
"最大分配内存",
"max_memory_cached",
"max_memory_reserved",
"mem_get_info",
"memory",
"内存分配",
"内存缓存",
"内存保留",
"内存快照",
内存统计,
内存统计嵌套字典,
内存摘要,
内存使用情况,
"内存池",
"内存池上下文",
"使用内存池",
"温度",
"功耗",
"时钟频率",
"nccl",
"nvtx",
"分析器",
"随机",
"重置累积的主机内存统计信息",
"重置累计内存统计",
"重置最大分配内存",
"重置最大缓存内存",
"重置峰值主机内存统计",
"重置峰值内存统计",
"种子",
"设置所有种子",
"设置设备",
"设置每个进程内存分数",
"设置随机数生成器状态",
"设置所有随机数生成器状态",
"设置流",
"设置同步调试模式",
"稀疏",
"流",
"流（复数）",
"同步",
"可调的",
"利用率",
]
