torch.mps 的源代码

# mypy: 允许未类型化定义
r""
此包提供了一个用于访问 MPS（Metal Performance Shaders）后端的 Python 接口。
Metal 是苹果公司用于编程金属 GPU（图形处理器单元）的 API。使用 MPS 意味着可以通过在金属 GPU 上运行工作来提高性能。
通过在金属 GPU 上运行工作，可以提升性能。
查看更多详情，请访问 https://developer.apple.com/documentation/metalperformanceshaders
""
from 打字 导入 联合

导入 火炬
from 火炬 导入 张量


_is_in_bad_fork = getattr(火炬._C, `_mps_is_in_bad_fork`, lambda: False)
`_default_mps_generator`: 火炬._C.生成器 = 无  # 类型：忽略[赋值]


# 本地辅助函数（非公开或导出）
定义 _get_default_mps_generator() -> 火炬._C.生成器:
    全局 _default_mps_generator
    如果 _default_mps_generator 是 无:
        _default_mps_generator = 火炬._C._mps_get_default_generator()
    返回 _default_mps_generator


[文档]def device_count() -> int:
r"""返回可用的 MPS 设备数量。"""
return int(torch._C._has_mps and torch._C._mps_is_available())


[文档]def synchronize() -> None:
r"""等待所有流上的 MPS 设备中的所有内核完成。"""
return torch._C._mps_deviceSynchronize()


[文档]def get_rng_state(device: Union[int, str, torch.device] = "mps") -> Tensor:
返回随机数生成器状态作为 ByteTensor。

Args:
device (torch.device 或 int，可选)：返回 RNG 状态的设备。
默认：``'mps'``（即 ``torch.device('mps')``，当前 MPS 设备）。
"""
返回 `_get_default_mps_generator().get_state()`


[文档]def 设置随机数生成器状态(
new_state: Tensor, device: Union[int, str, torch.device] = "mps"
) -> None:
r"""设置随机数生成器状态。

Args:
new_state (torch.ByteTensor): 目标状态
device (torch.device 或 int, 可选): 设置 RNG 状态的设备。
默认: ``'mps'`` (即 ``torch.device('mps')``，当前 MPS 设备)。
"""
new_state_copy = new_state.clone(memory_format=torch.contiguous_format)
_get_default_mps_generator().set_state(new_state_copy)


[文档]def manual_seed(seed: int) -> None:
r"""设置生成随机数的种子。"""

Args:
种子 (int): 所需的种子。
"""
# torch.mps.manual_seed() 可以从全局调用
torch.manual_seed() 在 torch/random.py 中。因此，我们需要进行
确保 mps 可用（否则我们直接返回）
错误退出（erroring out）
如果 not torch._C._has_mps:
返回
seed = int(seed)
_get_default_mps_generator().manual_seed(seed)


[文档]def seed() -> None:
将生成随机数的种子设置为随机数。
获取默认的 MPS 生成器的种子()。


[文档]def empty_cache() -> None:
释放当前由缓存持有的所有未占用缓存内存。
分配器，以便它们可以在其他 GPU 应用程序中使用。
"""
torch._C._mps_emptyCache()


[文档]def set_per_process_memory_fraction(fraction) -> None:
设置 MPS 设备上限制进程内存分配的内存分数。
允许的值等于推荐最大设备内存的分数乘积
（从 Metal API 的 device.recommendedMaxWorkingSetSize 获取）。
如果在进程中尝试分配超过允许的值，将引发内存不足异常
内存分配器错误。

Args:
分数(float): 范围：0~2。允许的内存等于总内存 * 分数。

.. 注意::
传递 0 到分数表示无限分配
（内存不足时可能导致系统故障）。
通过大于 1.0 的分数允许超过数值的极限
从设备推荐的最大工作集大小返回。
"""

如果不是 float 类型：
抛出 TypeError 异常("fraction 参数类型无效，必须是 `float` 类型")
如果 fraction 小于 0 或者大于 2：
引发 ValueError 异常(f'无效的分数值：{fraction}。允许的范围：0~2')

torch._C._mps_setMemoryFraction(fraction)


[文档]def current_allocated_memory() -> int:
r"""返回当前由张量占用的 GPU 内存大小（以字节为单位）。""

.. 注意::
返回的大小不包括 MPSAllocator 内存池中的缓存分配。
内存池。
"""
return torch._C._mps_currentAllocatedMemory()


[文档]def driver_allocated_memory() -> int:
返回 Metal 驱动程序为该进程分配的总 GPU 内存（以字节为单位）。

.. 注意::
返回的大小包括 MPSAllocator 池中的缓存分配
以及 MPS/MPSGraph 框架的分配。
"""
return torch._C._mps_driverAllocatedMemory()


[文档]def recommended_max_memory() -> int:
返回推荐的最大工作集大小，单位为字节，用于 GPU 内存。

.. note::
推荐的最大工作集大小（针对 Metal）。
从设备.recommendedMaxWorkingSetSize 返回。
"""
返回 torch._C._mps_recommendedMaxMemory()。


[文档]定义编译着色器(source: str)函数。
编译从源代码生成的计算着色器，并允许调用内核
定义于 Python 运行时
示例：

>>> # xdoctest: +REQUIRES(环境:TORCH_DOCTEST_MPS)
>>> lib = torch.mps.compile_shader(
... "kernel void full(device float* out, constant float& val, uint idx [[thread_position_in_grid]]) { out[idx] = val; }"
        ...  )
>>> x = torch.zeros(16, device="mps")
>>> lib.full(x, 3.14)
"""
从 pathlib 模块导入 Path

从 torch.utils._cpp_embed_headers 模块导入 _embed_headers

如果没有 torch._C 的 _mps_compileShader 属性
则引发 RuntimeError("MPS 不可用")
source = _embed_headers(
[l + "\n" for l in source.split("\n")],
[当前文件父目录的父目录 / "include"]
空集合()
    )
返回 torch._C._mps_compileShader(source)


定义 是否可用() -> bool:
    返回 设备数量() > 0


from . 导入 分析器
from 事件 导入 活动


全部 = [
    编译着色器,
    设备数量,
    "获取随机数生成器状态",
    "手动播种",
    种子,
    "设置随机数生成器状态",
    "同步",
    清空缓存,
    "设置每个进程内存分数",
    当前已分配内存,
    驱动器已分配内存,
    "事件",
    "分析器",
    推荐最大内存,
    "是否可用",
]