torch.mtia 源代码

# mypy: 允许未类型化定义
r""
此包为访问 MTIA 后端在 Python 中提供接口。
""

导入 线程
导入 警告
from 打字 导入 任何, 可调用, 可选, 联合

导入 火炬
from 火炬 导入 设备 as 设备, 张量
from torch._utils 导入 _dummy_type, _LazySeedTracker, 类属性
from torch 的类型 导入 设备

from _utils 导入 _get_device_index


设备_t = 联合[设备, str, int]

torch.mtia.Event/Stream 是 torch.Event/Stream 的别名
活动 = 火炬.活动
流 = 火炬.流

_initialized = 假
_队列调用: 列表[
    元组[可调用[] 无] 列表[str]]
] = 输入文本为空，请提供需要翻译的文本  # don't invoke these until initialization occurs
_tls = 线程.local()
_initialization_lock = 线程.锁()
_lazy_seed_tracker = _LazySeedTracker()


[文档]def 初始化():
_lazy_init()


[文档]def is_initialized():
r"""返回 PyTorch 的 MTIA 状态是否已初始化。"""
return _initialized and not _is_in_bad_fork()


定义 _is_in_bad_fork
是坏分支() -> bool:
    返回 火炬._C._mtia_isInBadFork()


定义 _lazy_init() -> 无:
    全局 已初始化, _queued_calls
    如果 已初始化() 或者 有属性(_tls, 初始化中):
        返回
    替换为 初始化锁:
        我们要再次检查锁定，伙计们！这是可以的，因为
        上述测试已经被 GIL 保护了。内部测试
        # 当一个线程因为其他线程正在初始化而被阻塞时
        # 当它们获得锁后，会发现没有剩下什么要做的。
        # 初始化完成。
        如果 已初始化():
            返回
        防止其他线程进入 _lazy_init 是很重要的
        立即，因为我们仍然保证有全局解释锁（GIL），因为某些
        以下 C 调用将释放全局解释器锁（GIL）
        如果 _is_in_bad_fork
是坏分支():
            抛出 RuntimeError(
                "不能在分叉的子进程中重新初始化 MTIA。要使用带有 MTIA 的 "
                "multiprocessing，您必须使用'spawn'启动方法"
            )
        如果 not 已编译():
            抛出 断言错误(
                "未使用 MTIA 启用编译的 Torch。"
                "确保你的 python 源文件中包含 `import mtia.host_runtime.torch_mtia.dynamic_library`"
                "并将 `//mtia/host_runtime/torch_mtia:torch_mtia` 添加为目标依赖！"
                "你的目标依赖！"
            )

        火炬._C._mtia_init()
        # 一些排队的调用可能会递归地调用 _lazy_init();
        # 在这种情况下，我们只需返回而不进行初始化即可。
        然而，我们绝不能让任何其他线程进入！
        _tls.正在初始化 = 真实

        _队列调用.扩展(电话 对于 电话 进入 懒惰的种子追踪器.获取电话() 如果 电话)

        尝试:
            对于 排队调用, 原始跟踪信息 进入 _队列调用:
                尝试:
                    排队调用()
                除了 异常 as e:
                    msg = (
                        f"MTIA 调用在初始化时懒加载失败，错误信息："{str(e)}\n\n
\n\n"
                        f"MTIA 调用最初是在以下位置被调用的："\n\n
\n\n{请提供需要翻译的文本.连接(原始跟踪)}"
                    )
                    抛出 DeferredMtiaCallError(信息) from e
        最后:
            delattr(_tls, 初始化中)
        _initialized = 真实


[文档]class DeferredMtiaCallError(Exception):
通过


定义 已编译() -> bool:
    r如果编译时支持 MTIA，则返回 true。
    返回 火炬._C._mtia_isBuilt()


[文档]def is_available() -> bool:
返回 MTIA 设备是否可用的结果
如果没有编译
返回 False
# MTIA 必须先初始化设备，才能知道是否有可用设备。
return 设备数量() > 0


[文档]def 同步(device: Optional[_device_t] = None) -> None:
r"""等待 MTIA 设备上所有流中的所有作业完成。"""
with torch.mtia.device(device):
return torch._C._mtia_deviceSynchronize()


[文档]def device_count() -> int:
返回可用的 MTIA 设备数量。
# TODO: 将 _accelerator_hooks_device_count 更改为抽象 MTIA 设备计数 API
return torch._C._mtia_getDeviceCount()


[文档]def current_device() -> int:
r"""返回当前选中设备的索引。"""
return torch._C._accelerator_hooks_get_current_device()


[文档]def current_stream(device: Optional[_device_t] = None) -> Stream:
返回给定设备的当前 :class:`Stream`。

参数:
device (torch.device 或 int, 可选): 选择设备。返回
当前选定的：class:`Stream`，针对当前设备
通过 :func:`~torch.mtia.current_device`，如果 :attr:`device` 是 ``None``
(默认)。
```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)
```
return torch._C._mtia_getCurrentStream(_get_device_index(device, optional=True))


[文档]def default_stream(device: Optional[_device_t] = None) -> Stream:
r"""返回给定设备的默认 :class:`Stream`。

Args:
设备（torch.device 或 int，可选）：选定的设备。如果 device 为 None（默认），
返回当前设备的默认：class:`Stream`，由：func:`~torch.mtia.current_device`给出，
如果：attr:`device`为 None（默认）。
。
"""
返回 torch._C._mtia_getDefaultStream(_get_device_index(device, optional=True))


[文档]def record_memory_history(
enabled: Optional[str] = "all", stacks: str = "python", max_entries: int = 0
)
    -> None:
启用/禁用 MTIA 分配器的内存分析器

Args:
enabled (all 或 state，可选) 选择设备。返回
当前设备的统计数据，由 current_device()提供
如果设备为 None（默认）

栈（"python"或"cpp"，可选）。选择要记录的堆栈跟踪

max_entries（int，可选）。记录的最大条目数
"""
如果未初始化()
返回
torch._C._mtia_recordMemoryHistory(enabled, stacks, max_entries)


[文档]def snapshot() -> dict[str, Any]:
返回 MTIA 内存分配器历史记录的字典

return torch._C._mtia_memorySnapshot()


[文档]def get_device_capability(device: Optional[_device_t] = None) -> tuple[int, int]:
返回给定设备的功能，以（主版本号，副版本号）元组的形式。

参数：
device (torch.device 或 int，可选) 选择设备。返回
当前设备的统计信息，由 current_device() 函数给出。
如果设备为空（默认）。
"""
返回 `torch._C._mtia_getDeviceCapability(_get_device_index(device, optional=True))`


[文档]def empty_cache() -> None:
清空 MTIA 设备缓存。
return torch._C._mtia_emptyCache()


[文档]def 设置流(stream: Stream):
设置当前流。这是一个设置流的包装 API。
使用此函数被建议，以 ``stream`` 优先
上下文管理器。

参数：
stream (Stream): 选定的流。此函数为空操作
如果此参数为 `None`。
"""
如果流为 `None`：
返回
torch._C._mtia_setCurrentStream(stream)


[文档]def set_device(device: _device_t) -> None:
r"""设置当前设备。

Args:
device (torch.device 或 int): 选择设备。此函数为空操作
如果此参数为负数。
"""
device = _get_device_index(device)
if 设备 >= 0:
torch._C._accelerator_hooks_set_current_device(设备)


[文档]class 设备:
r"""上下文管理器，用于更改选定的设备。

参数:
设备（torch.device 或 int）：选择设备索引。如果为空操作。
这个参数是一个负整数或“None”。
```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)
```

def __init__(self, device: Any):
self.idx = _get_device_index(device, optional=True)
self.prev_idx = -1

def __enter__(self):
self.prev_idx = torch._C._accelerator_hooks_maybe_exchange_device(self.idx)

def __exit__(self, type: Any, value: Any, traceback: Any):
self.idx = torch._C._accelerator_hooks_maybe_exchange_device(self.prev_idx)
return False


类 Stream 上下文:
    r选择给定流的上下文管理器。

所有在该上下文中排队的 MTIA 内核都将被加入到一个选定的队列中
流。

参数：
流（Stream）：选择的流。如果它是空操作，则此管理器无作用。
``None``。
.. 注意:: 流是按设备划分的。
    """

    当前流: 可选[torch.mtia.Stream]

    定义 __init__(self, 流: 可选["torch.mtia.Stream")]
        self.cur_stream = 无
        self.流 = 流
        self.索引 = 获取设备索引(无, True)
        如果 not 火炬.算子.是否正在脚本化():
            如果 self.索引 是 无:
                self.索引 = -1

        self.src_prev_stream = (
            无 如果 not 火炬.算子.是否正在脚本化() 否则 火炬.管理培训师.默认流(无)
        )
        self.目标流 = (
            无 如果 not 火炬.算子.是否正在脚本化() 否则 火炬.管理培训师.默认流(无)
        )

    定义 __进入__(self):
        # 本地 cur_stream 变量用于类型细化
        cur_stream = self.流
        # 如果流为空或 MTIA 设备不可用，则返回
        如果 cur_stream 是 无 或者 self.索引 == -1:
            返回
        self.src_prev_stream = 火炬.管理培训师.current_stream(无)

        # 如果流不在当前设备上，则
        # 在设备上设置当前流
        如果 self.原始流.设备 != 当前流.设备:
            替换为 设备(当前流.设备):
                self.目标流 = 火炬.管理培训师.current_stream(当前流.设备)
        火炬.管理培训师.设置流(当前流)

    定义 __退出__(self, 类型: 任何, value: 任何, 跟踪回溯: 任何):
        # 本地 cur_stream 变量用于类型细化
        cur_stream = self.流
        # 如果流为空或没有 MTIA 设备可用，则返回
        如果 cur_stream 是 无 或者 self.索引 == -1:
            返回

        在原始设备上重置流
        # 和目标设备
        如果 self.原始流.设备 != 当前流.设备:  # 类型：忽略[联合属性]
            火炬.管理培训师.设置流(self.目标前流)  # type: ignore[arg-type]
        火炬.管理培训师.设置流(self.原始流)  # 类型：忽略[arg-type]



[文档]def stream(stream: Optional["torch.mtia.Stream"]) -> StreamContext:
r"""围绕选择给定流的上下文管理器 StreamContext 进行包装。

参数：
stream (Stream): 选定的流。如果为 ``None``，则此管理器为空操作。
``None``。
.. 注意:: 在 eager 模式下流是 Stream 类型的，而在 JIT 下不支持 torch.mtia.stream
"""
返回 StreamContext(stream)


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

参数：
device (torch.device 或 int，可选)：返回 RNG 状态的设备。
默认：``'mtia'``（即``torch.device('mtia')``，当前 mtia 设备）。
"""
warnings.warn(
"get_rng_state 函数在 torch.mtia 中未实现",
UserWarning
stacklevel=2,
    )
返回 torch.zeros([1], dtype=torch.uint8, device=device)


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

Args:
new_state (torch.ByteTensor): 目标状态
device (torch.device 或 int, 可选): 设置 RNG 状态的设备。
默认: ``'mtia'`` (即 ``torch.device('mtia')``，当前 mtia 设备)。
"""
warnings.warn(
"set_rng_state 函数在 torch.mtia 中未实现",
UserWarning,
stacklevel=2,
    )


from .内存 导入 *  # noqa: F403


全部 = [
    "init",
    "是否可用",
    "已初始化",
    "同步",
    设备数量,
    "current_device",
    "当前流",
    "默认流",
    内存统计,
    "最大分配内存",
    "重置峰值内存统计",
    "获取设备功能",
    "记录内存历史",
    "快照",
    清空缓存,
    "设置设备",
    "设置流",
    "流",
    "设备",
    "设置随机数生成器状态",
    "获取随机数生成器状态",
]