torch.utils.module_tracker 的源代码

# mypy: 允许未类型化定义
导入 记录日志
导入 弱引用
from 打字 导入 类型检查

导入 火炬
from torch.autograd.graph 导入 注册多梯度钩子
from torch.nn.modules 下的 module 模块 导入 (
    注册模块前向钩子,
    注册模块前向预处理钩子,
)
from torch.utils._pytree 导入 树扁平化


如果 类型检查:
    from torch.utils.hooks 导入 可移除句柄


记录器 = 记录.获取日志记录器(__name__)


全部 = [模块跟踪器]


类 模块跟踪器:
    ""
``模块跟踪器`` 是一个上下文管理器，用于在执行过程中跟踪 nn.Module 层次结构
以便其他系统可以查询当前正在执行哪个模块（或其反向操作）
执行完毕。

您可以访问此上下文管理器的 `parents` 属性以获取所有父级的集合
当前正在通过其 fqn（完全限定名，也用作键）执行模块
状态字典。
您可以通过访问 `is_bw` 属性来了解您当前是在正向运行还是反向运行。

注意，`parents` 永远不为空，并且始终包含 "Global" 键。`is_bw` 标志
在执行另一个模块之前将保持 `True` 状态。
更准确，请提交一个请求此功能的 issue。从 fqn 到模块实例的映射尚未实现，如有需要，请提交一个请求此功能的 issue。
这是可能的，但尚未完成，如有需要，请提交一个请求此功能的 issue。

示例用法

.. 代码块 :: python

        mod = torch.nn.Linear(2, 2)

使用 ModuleTracker() as tracker:
# 在前向传播过程中访问任何内容
            def my_linear(m1, m2, bias):
打印(f"当前模块: {tracker.parents}")
                return torch.mm(m1, m2.t()) + bias
            torch.nn.functional.linear = my_linear

            mod(torch.rand(2, 2))

"文档"

    parents: 设置[字符串]
    ""
包含当前运行每个模块的前向 fqn 的集合
"文档"

    定义 __init__(我) -> 无:
        我.父母 = {全球}
        我.已知模块: 弱引用.弱密钥字典 = 弱引用.WeakKeyDictionary()
        我.已见模块: 弱引用.WeakSet = 弱引用.WeakSet()
        我._has_callback = 假
        我._hooks: 列表[可移除句柄] = 输入文本为空，请提供需要翻译的文本

    定义 _maybe_set_engine_callback(我):
        # 这假设没有并发调用回溯
        如果 我._has_callback:
            返回

        定义 回调():
            我.父母 = {全球}
            我._有回调 = 假

        火把.自动微分.变量._execution_engine.队列回调(回调)
        我._has_callback = 真实

    @property
    定义 is_bw(我):
        ""
当前是否正在反向传播过程中运行的布尔标记
"文档"
        返回 火把._C._当前图任务 ID() != -1

    定义 _get_mod_name(我, 模块):
        如果 修饰 不 在 我.已知模块:
            我.已知模块[模块] = 类型(模块).__name__
        模块名称 = 我.已知模块[模块]
        如果 修饰 不 在 我.已见模块:
            for 名称, 子模块 在 模块.命名子项():
                我.已知模块[子模块] = f"{模块名称}.{名称}"
                我.获取模块名称(子模块)
            我.已见模块.添加(模块)
        返回 模块名称

    定义 获取追加函数(我, 名称, 是黑白的):
        定义 函数(*参数):
            如果 是黑白的:
                我._可能设置引擎回调()
            如果 名称 在 我.parents:
                日志记录器.信息(
                    模块层次跟踪似乎出现了问题，因为这个模块已经被输入过了。%s在%s",
                    名称,
                    向后 如果 是黑白的 否则 前进,
                )
            我.parents.添加(名称)

        返回 fn

    定义 _获取流行函数(我, 名称, 是黑白的):
        定义 函数(*参数):
            如果 名称 在 我.parents:
                我.parents.删除(名称)
            否则:
                日志记录器.信息(
                    “当我们退出一个从未进入的模块时，模块层次跟踪变得混乱。”%s在%s",
                    名称,
                    “向后” 如果 is_bw 否则 前进,
                )

        返回 fn

    定义 _fw_pre_hook(我, 模块, 输入):
        名称 = 我._get_mod_name(模块)
        我._get_append_fn(名称, 错误)()

        参数, _ = 树形展平(输入)
        张量 = [a for a 在 args 如果 isinstance(a, 火把.张量) 并且 a.需要梯度]
        如果 张量:
            我._hooks.append(
                注册多梯度钩子(张量, 我._get_pop_fn(名称, 是))
            )

    定义 _fw_post_hook(我, 模块, 输入, 输出):
        名称 = 我._get_mod_name(模块)
        我._get_pop_fn(名称, 错误)()

        参数, _ = 树形展平(输出)
        张量 = [a for a 在 args 如果 isinstance(a, 火把.张量) 并且 a.需要梯度]
        如果 张量:
            我._hooks.append(
                注册多梯度钩子(张量, 我._get_append_fn(名称, 是))
            )

    定义 __进入__(我):
        我._fw_pre_handle = 注册模块前向预处理钩子(我._fw_pre_hook)
        我._fw 后处理 = 注册模块前向钩子(我._fw_post_hook)
        返回 self

    定义 __退出__(我, *参数):
        我._fw_pre_handle.删除()
        我._fw_post_handle.删除()
        for 钩子 在 我._hooks:
            hook.删除()
        我.钩子.清晰()