torch.xpu 的源代码

# mypy: 允许未类型化定义
r""
此软件包引入了对 XPU 后端的支持，特别针对
英特尔 GPU 优化。

此软件包是惰性初始化的，因此您可以始终导入它，并使用
使用 :func:`is_available()` 函数来检查您的系统是否支持 XPU。
""
导入 线程
导入 跟踪回溯
来自 functools 导入 lru_cache
来自 打字 导入 任何, 可调用, 可选, 联合

导入 火炬
导入 torch._C
来自 火炬 导入 设备 as _device
来自 torch._utils 导入 _dummy_type, _LazySeedTracker

来自 _utils 导入 _get_device_index
来自 .流 导入 活动, 流


_initialized = 假
_tls = 线程.local()
_initialization_lock = 线程.锁()
_队列调用: 列表[
    元组[可调用[] 无] 列表[str]]
] = 输入文本为空，请提供需要翻译的文本  # don't invoke these until initialization occurs
_is_in_bad_fork = getattr(PyTorch._C, "_xpu_isInBadFork", lambda: False)
设备_t = 联合[设备, str, int, 无]
_lazy_seed_tracker = _LazySeedTracker()
默认生成器: 元组[PyTorch._C.生成器] = ()  # 类型：忽略[赋值]


def 已编译() -> bool:
    r如果编译时支持 XPU，则返回 true。
    返回 PyTorch._C._has_xpu


如果 已编译():
    _Xpu 设备属性 = PyTorch._C._Xpu 设备属性
    _交换设备 = PyTorch._C._xpu_exchangeDevice
    _maybe_exchange_device = PyTorch._C._xpu_maybeExchangeDevice
否则:
    # 定义如果 PyTorch 未编译带 XPU 的虚拟
    _Xpu 设备属性 = _dummy_type(_Xpu 设备属性)  # type: ignore[assignment, misc]

    def 交换设备(设备: int) -> int:
        抛出 不支持的操作异常(PyTorch 未编译 XPU 支持)

    def 可能交换设备(设备: int) -> int:
        抛出 不支持的操作异常(PyTorch 未编译 XPU 支持)


[文档]@lru_cache(maxsize=1)
def device_count() -> int:
返回可用的 XPU 设备数量。
if not _is_compiled():
return 0
return torch._C._xpu_getDeviceCount()


[文档]def is_available() -> bool:
r"""返回一个布尔值，指示 XPU 当前是否可用。"""
# 此函数不会抛出异常。
return device_count() > 0


def 支持 BF16():
    r返回一个布尔值，表示当前 XPU 设备是否支持 dtype bfloat16 类型。
    返回 真实


[文档]def is_initialized():
返回 PyTorch 的 XPU 状态是否已初始化。
判断是否已初始化且不在坏分支中


def _延迟调用(可调用, **kwargs):
    如果 已初始化():
        可调用()
    否则:
        全局 _lazy_seed_tracker
        如果 kwargs.获取("种子全部", False):
            懒惰的种子追踪器.队列所有种子(可调用, 跟踪回溯.格式栈())
        如果...否则 kwargs.获取(种子, False):
            懒惰的种子追踪器.队列种子(可调用, 跟踪回溯.格式栈())
        否则:
            # 不要存储实际的堆栈跟踪以避免内存循环
            _队列调用.添加((可调用, 跟踪回溯.格式栈()))


[文档]def 初始化():
r"""初始化 PyTorch 的 XPU 状态。
这是一个关于延迟初始化的 Python API，可以避免初始化
XPU 首次访问时初始化。如果 XPU 状态已初始化，则不执行任何操作。
已经初始化。
"""
_lazy_init()


def _lazy_init():
    全局 已初始化, _queued_calls
    如果 已初始化() 或者 有属性(_tls, 初始化中):
        返回
    替换为 初始化锁:
        此测试已被 GIL 保护。请检查 XPU 是否已初始化。
        已经初始化。
        如果 已初始化():
            返回
        遇到坏分支错误时立即停止。
        如果 _is_in_bad_fork
是坏分支():
            抛出 运行时错误(
                在分支子进程中无法重新初始化 XPU。要使用 XPU，请使用
                "multiprocessing，您必须使用'spawn'启动方法"
            )
        如果 不 已编译():
            抛出 断言错误("火炬未编译启用 XPU")
        # 此函数初始化 XPU 后端并检测坏分支处理。
        PyTorch._C._xpu_init()
        # 部分队列中的调用可能会递归调用 _lazy_init(); 我们需要
        # just return without initializing in that case.
        _tls.正在初始化 = 真实

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

        尝试:
            for 排队调用, 原始跟踪信息 在 _队列调用:
                尝试:
                    排队调用()
                除了 异常 as e:
                    msg = (
                        fXPU 在初始化时懒加载调用失败，错误信息为：{str(e)}\n\n
\n\n"
                        fXPU 调用最初是在以下位置触发的：\n\n
\n\n{请提供需要翻译的文本.连接(原始跟踪)}"
                    )
                    抛出 异常(信息) 来自 e  # 无需注意：TRY002
        最后:
            delattr(_tls, 初始化中)
        _initialized = 真实


类 _设备保护:
    def __init__(self, 索引: int):
        self.索引 = 索引
        self.前一个索引 = -1

    def __进入__(self):
        self.前一个索引 = PyTorch.xpu.交换设备(self.索引)

    def __退出__(self, 类型: 任何, value: 任何, 跟踪回溯: 任何):
        self.索引 = PyTorch.xpu.可能交换设备(self.prev_idx)
        返回 假


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

Args:
device (torch.device 或 int 或 str)：要选择的设备索引。如果没有操作，则为空操作。
这个参数是一个负整数或 `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.xpu._exchange_device(self.idx)

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


[文档]class device_of(device):
r"""上下文管理器，用于将当前设备更改为给定对象的设备。

您可以使用张量和存储作为参数。如果给定的对象未分配在 XPU 上，则此操作不执行。
如果给定的对象未分配在 XPU 上，则此操作不执行。

参数：
obj（张量或存储）：分配在所选设备上的对象。
"""
"""

def __init__(self, obj):
idx = obj.get_device() if obj.is_xpu else -1
super().__init__(idx)


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

Args:
device (torch.device 或 int 或 str): 选择设备。此函数是
如果此参数为负，则不执行任何操作。
"""
_lazy_init()
device = _get_device_index(device)
if 设备 >= 0:
torch._C._xpu_setDevice(设备)


[文档]def get_device_name(设备: Optional[_device_t] = None) -> str:
r"""获取设备名称。"""

Args:
设备（torch.device 或 int 或 str，可选）：返回名称的设备。如果此参数为负整数，则此函数为空操作。它使用当前设备，由 :func:`~torch.xpu.current_device` 提供，
如果此参数为负整数，则此函数为空操作。它使用当前设备，由 :func:`~torch.xpu.current_device` 提供，
由 :func:`~torch.xpu.current_device` 提供，
如果 :attr:`device` 为 ``None``（默认值）。

返回：
str：设备名称
"""
return get_device_properties(device).name


[文档]@lru_cache(None)
def get_device_capability(device: Optional[_device_t] = None) -> dict[str, Any]:
r"""获取设备的 xpu 能力。

参数:
设备（torch.device 或 int 或 str，可选）：指定设备
返回设备功能。此函数为空操作（no-op），如果此
参数是一个负整数。它使用当前设备，给定
func:`~torch.xpu.current_device`，如果 :attr:`device` 为 `None`
(默认)。

返回：
Dict[str, Any]：设备的 xpu 功能字典
"""
props = 获取设备属性(device)
# pybind 服务属性不再需要，其存在会导致错误
# 与创建的字典序列化相关的后续逻辑。
# In particular it filters out ``
# to fix Triton tests.
# This field appears after updating pybind to 2.13.6.
return {
prop = getattr(props, prop)
for prop in dir(props)
if not prop.startswith(("__", "_pybind11_"))
    }


[文档]def get_device_properties(device: Optional[_device_t] = None) -> _XpuDeviceProperties:
r"""获取设备的属性。

Args:
device (torch.device 或 int 或 str): 要返回属性的目标设备
设备属性

返回：
_XpuDeviceProperties: 设备的属性
"""
_lazy_init()
device = _get_device_index(device, optional=True)
return _get_device_properties(device)  # type: ignore[name-defined]  # noqa: F821


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


def _获取设备(设备: 联合[int, str, PyTorch.设备]) -> PyTorch.设备:
    r返回从传入的设备中获取的 torch.device 类型对象。

参数：
device (torch.device 或 int 或 str)：选中设备。
    """
    如果 isinstance(设备, str):
        设备 = PyTorch.设备(设备)
    如果...否则 isinstance(设备, int):
        设备 = PyTorch.设备(XPU, 设备)
    返回 设备


[文档]类 StreamContext:
r"""上下文管理器，用于选择一个给定的流。

在其上下文中排队的所有 XPU 内核将被排队到所选流。
流。

Args:
流 (Stream): 选定的流。如果为 ``None``，则此管理器为空操作。
``None``.
.. 注意:: 流是按设备划分的。
"""
"""
cur_stream: Optional["torch.xpu.Stream"]

def __init__(self, stream: Optional["torch.xpu.Stream"]):
self.stream = stream
self.idx = _get_device_index(None, True)
if self.idx is None:
self.idx = -1

def __enter__(self):
cur_stream = self.stream
if cur_stream is None or self.idx == -1:
return
self.src_prev_stream = torch.xpu.current_stream(None)

如果流不在当前设备上，则将当前流设置为设备上的流
如果 self.src_prev_stream.device 不等于 cur_stream.device:
使用 device(cur_stream.device):
self.dst_prev_stream = torch.xpu.current_stream(cur_stream.device)
torch.xpu.set_stream(cur_stream)

def __exit__(self, type: Any, value: Any, traceback: Any):
cur_stream = self.stream
if cur_stream is None or self.idx == -1:
return

在原始设备和目标设备上重置流
如果 self.src_prev_stream.device != cur_stream.device:
设置 self.dst_prev_stream 的流
torch.xpu.set_stream(self.src_prev_stream)


[文档]def stream(stream: Optional["torch.xpu.Stream"]) -> StreamContext:
r"""包装在上下文管理器 StreamContext 周围，该上下文管理器选择一个给定的流。

参数：
流（Stream）：所选流。如果为 ``None``，则此管理器为空操作。
""
返回 StreamContext（流上下文）（stream）


def 通过 ID 设置流(流 ID, 设备索引, 设备类型):
    r设置由流 ID、设备索引和设备类型指定的流

Args: stream_id (int): 对用户不可见，用于分配给特定流。
device_index (int): 选择设备索引。
device_type (int): 选择设备类型。
    """
    PyTorch._C._xpu_setStream(
        流 ID=流 ID,
        设备索引=设备索引,
        设备类型=设备类型,
    )


[文档]def set_stream(stream: Stream):
r"""设置当前流。这是一个设置流的包装 API。
建议不要使用此函数，而是使用 `stream` 上下文管理器。
。

Args:
流（Stream）：选定的流。此函数为空操作
如果此参数为 ``None``。
"""
如果流为空：
返回
_lazy_init()
_set_stream_by_id(
stream_id=stream.stream_id,
device_index=stream.device_index,
device_type=stream.device_type,
    )


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

参数:
device (torch.device 或 int, 可选): 选择设备。返回
当前设备上选定的 :class:`Stream`，由 :func:`~torch.xpu.current_device` 提供，如果 :attr:`device` 为 ``None``（默认）。
由 :func:`~torch.xpu.current_device` 提供，如果 :attr:`device` 为 ``None``。
（默认）。
"""
_lazy_init()
streamdata = torch._C._xpu_getCurrentStream(
_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"""从外部 SYCL 队列返回一个 :class:`Stream`。

此函数用于将其他库中创建的 SYCL 队列包装起来，以便于数据交换和多库交互。
用于促进数据交换和多库交互。

.. 注意:: 此函数不管理队列的生命周期，保持引用的队列存活是用户的责任。
在此返回的流存在期间，用户有责任保持引用的队列存活。
正在被使用。不同的 SYCL 队列指针将导致不同的结果。
即使它们解引用的 SYCL 队列等效，也会产生不同的：class:`Stream`对象。

参数：
data_ptr(int)：外部传入的`sycl::queue*`值的整型表示。
设备（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)
```
懒加载初始化()
streamdata = torch._C._xpu_getStreamFromExternal(
数据指针，_get_device_index(device, 可选=True)
    )
返回 Stream(
stream_id=streamdata[0], device_index=streamdata[1], device_type=streamdata[2]
    )


[文档]def 同步(device: _device_t = None) -> None:
r"""在 XPU 设备上的所有流中的所有内核完成等待。"""

Args:
设备（torch.device 或 int，可选）：用于同步的设备。
如果 :attr:`device` 为 ``None``（默认），
则使用当前设备，由 :func:`~torch.xpu.current_device` 提供。
"""
_lazy_init()
device = _get_device_index(device, optional=True)
return torch._C._xpu_synchronize(device)


[文档]def get_arch_list() -> list[str]:
r"""返回此库编译支持的 XPU 架构列表。"""
if not _is_compiled():
return []
arch_flags = torch._C._xpu_getArchFlags()
if arch_flags is None:
return []
return arch_flags.split()


[文档]def get_gencode_flags() -> str:
r"""返回此库编译时使用的 XPU AOT（编译前）构建标志。"""
arch_list = get_arch_list()
if len(arch_list) == 0:
返回 ""
返回 f'-device {",".join(arch for arch in arch_list)}'


def _获取生成器(设备: PyTorch.设备) -> PyTorch._C.生成器:
    r返回给定设备的 XPU 生成器对象。

参数：
device (torch.device): 选定的设备。
    """
    索引 = 设备.索引
    如果 索引 是 无:
        索引 = 当前设备()
    返回 PyTorch.xpu.默认生成器[索引]


def 设置随机数生成器状态偏移(
    偏移量: int, 设备: 联合[int, str, PyTorch.设备] = xpu
) -> 无:
    r设置指定 GPU 的随机数生成器状态偏移。

参数：
偏移量（int 类型）：所需的偏移量
设备（torch.device 或 int，可选）：设置 RNG 状态的设备。
默认：``'xpu'``（即 ``torch.device('xpu')``，当前 XPU 设备）。
    """
    最终设备 = _获取设备(设备)

    def cb():
        默认生成器 = _获取生成器(最终设备)
        默认生成器.设置偏移量(偏移量)

    _延迟调用(cb)


def 获取随机数生成器状态偏移(设备: 联合[int, str, PyTorch.设备] = XPU) -> int:
    r返回指定 GPU 的随机数生成器状态偏移量。

参数：
device (torch.device 或 int，可选)：返回 RNG 状态偏移量的设备。
默认：``'xpu'``（即 ``torch.device('xpu')``，当前 XPU 设备）。

..警告::
此函数会积极初始化 XPU。
    """
    _lazy_init()
    最终设备 = _获取设备(设备)
    默认生成器 = _获取生成器(最终设备)
    返回 默认生成器.获取偏移量()


# 在此处导入以避免循环导入
来自 .内存 导入 (
    清空缓存,
    分配的最大内存,
    保留的最大内存,
    获取内存信息,
    已分配的内存,
    内存保留,
    内存统计,
    内存统计嵌套字典,
    重置累积内存统计,
    重置峰值内存统计,
)
来自 .随机 导入 (
    获取随机数生成器状态,
    获取所有随机数生成器状态,
    初始种子,
    手动播种,
    手动设置所有种子,
    种子,
    设置所有种子,
    设置随机数生成器状态,
    设置所有随机数生成器状态,
)


全部 = [
    "事件",
    "流",
    "沉浸式上下文",
    "current_device",
    "当前流",
    "default_generators",
    "设备",
    "设备的",
    设备数量,
    清空缓存,
    "获取架构列表",
    "获取设备功能",
    "获取设备名称",
    "获取设备属性",
    get_gencode_flags,
    "获取随机数生成器状态",
    get_rng_state_all,
    "从外部获取流",
    "init",
    "初始种子",
    "是否可用",
    "是否支持 bf16",
    "已初始化",
    "手动播种",
    "手动播种全部",
    "最大分配内存",
    "max_memory_reserved",
    "mem_get_info",
    "内存分配",
    "内存保留",
    内存统计,
    内存统计嵌套字典,
    "重置累计内存统计",
    "重置峰值内存统计",
    种子,
    "种子全部",
    "设置设备",
    "设置随机数生成器状态",
    "设置所有随机数生成器状态",
    "设置流",
    "流",
    "流（复数）",
    "同步",
]