# mypy: 允许未类型化定义
导入 functools
导入 hashlib
导入 importlib.util
导入 itertools
导入 json
导入
记录日志
导入
操作系统
导入
os.path 模块
导入 pathlib
导入
正则表达式
导入
系统
导入 tempfile
导入
时间
from 集合
导入 defaultdict
from dataclasses 导入
数据类,
字段
from 打字
导入
任何,
可调用,
可选,
联合
from 弱引用
导入
弱集合
导入 torch._logging.structured
from torch._guards 导入
编译 ID
from torch._utils_internal 导入
记录结构化事件日志
from torch.utils._traceback 导入
捕获的 Traceback
日志 =
记录.
获取日志记录器(__name__)
# 这是一个合成型记录器，并不对应实际的记录器，
# 但处理我们所有的“跟踪”日志，这些日志是结构化的，并且不会
将标准错误输出到 stderr，但始终写入一个专门的日志文件。我们不将这些
经典模块层次结构中的记录器，因为我们不想抑制
日志数量，也会导致跟踪被抑制（通常跟踪不会被抑制）
收集，除非我们在生产环境中，在这种情况下它们总是被收集。）
#
# 可能我们应该允许一些子层次结构，以便您可以控制要收集哪些跟踪，出于性能考虑。
# # traces you want to collect, for performance reasons.
#
# 查看 https://docs.google.com/document/d/1CX_hJ0PNy9f3R1y8TJrfkSeLkvGjjjLU84BSXgS2AZ8/edit
跟踪日志 =
记录.
获取日志记录器(
torch.__trace)
默认日志级别 =
记录.
警告
日志环境变量 =
TORCH_LOGS
LOG_OUT_ENV_VAR = TORCH_LOGS_OUT
LOG_FORMAT_ENV_VAR = TORCH_LOGS_FORMAT
LOG_TRACE_ID_FILTER = "TORCH_LOGS_TRACE_ID_FILTER"
TRACE_ENV_VAR = "TORCH_TRACE"
DTRACE_ENV_VAR = "TORCH_DTRACE"
LOG_TRACE_HANDLER: 可选["LazyTraceHandler"] =
无
获取 DTrace 结构化信息 =
假
@dataclass
类
日志注册表:
# 简写名称到完全限定名称的缩写
# 注意：此列表仅包含已注册的日志记录器
# 从 register_log
# 例如 "dynamo" -> "torch._dynamo"
日志别名到日志全称:
字典[
字符串,
列表[
字符串]] =
字段(
默认工厂=
字典)
# 工具记录器合格名称，
# this is populated lazily, as calls to getArtifactLogger
# 当前格式为 <模块>.__<组件名称>
# 例如 "torch._dynamo.convert_frame.__guards"
artifact_log_qnames: 设置[
字符串] =
字段(
默认工厂=
设置)
# 如果通过打开指定，则记录注册日志的子日志
# 用户注册（例如，将 "torch._dynamo.output_graph" 放入环境变量中）
# 这些需要被跟踪，以便正确重置它们的级别
# 例如，"torch._dynamo.output_graph"
子日志 q 名称:
设置[
字符串] =
字段(
默认工厂=
设置)
# 艺术品名称，由 register_artifact 填充
# 例如："守卫"
艺术品名称:
设置[
字符串] =
字段(
默认工厂=
设置)
默认应显示在错误消息中的项目
可见项目:
设置[
字符串] =
字段(
默认工厂=
设置)
每个项目的简短描述
项目描述:
字典[
字符串,
字符串] =
字段(
默认工厂=
字典)
# 艺术品除非明确命名，否则不显示
设置。例如，即使有电感器，输出代码也不会显示。
日志级别设置为 DEBUG。必须在设置中明确指定。
默认不启用工件名称:
设置[
字符串] =
字段(
默认工厂=
设置)
日志格式字符串用于工件
工件日志格式化程序:
字典[
字符串,
记录.
格式化器] =
字段(
默认工厂=
字典)
定义
是否为工件(
我,
名称):
返回
名称
在
我.
艺术品名称
定义
是否为日志(
我,
别名):
返回
别名
在
我.
日志别名到日志名称映射
# 注册一个带有别名的日志
定义
注册日志(
我,
别名,
日志队列名称:
联盟[
字符串,
列表[
字符串
]]：
如果 isinstance(
日志队列名称,
字符串):
日志队列名称 = [
日志队列名称]
我.
将日志别名映射到日志全称[
别名] =
日志全称
# 注册一个工件名称
定义
注册工件名称(
我,
名称,
描述,
可见,
默认关闭,
日志格式
):
我.
艺术品名称.
添加(
名称)
如果
可见:
我.
可见文物.
添加(
名称)
我.
文物描述[
名称] =
描述
默认关闭
当 log_name 的 log_level 设置为 DEBUG 时
如果
默认关闭:
我.
默认关闭的工件名称.
添加(
名称)
如果
日志格式
是
不
无:
我.
艺术品日志格式化器[
名称] =
记录.
格式化器(
日志格式)
注册艺术品的日志名称
这是为了知道哪些日志需要重置
每当更改日志状态时
定义
注册艺术品日志(
我,
艺术品日志全名):
我.
艺术品日志全名列表.
添加(
艺术品日志全名)
定义
注册子日志(
我, log_qname):
我.
子日志 qname.
添加(log_qname)
# 将所有 qname 合并在一起（待定：考虑缓存？）
定义
获取日志 Q 名称(
我) ->
设置[
字符串
]：
返回
设置(itertools.chain.from_iterable(
我.
日志别名到日志 Q 名称.
值()))
定义
获取工件日志 Q 名称(
我):
返回
设置(
我.
工件日志 Q 名称)
定义
获取子日志 Q 名称(
我):
返回
设置(
我.
子日志 Q 名称)
定义
默认关闭(
我,
艺术品 Q 名称):
返回
艺术品全称
在
我.
默认关闭的艺术品名称
@dataclass
类
日志状态:
# 限定日志名称 -> 当前设置的日志级别
log_qname_to_level: 字典[
字符串,
字符串] =
字段(
默认工厂=
字典)
当前启用的工件集
工件名称:
设置[
字符串] =
字段(
默认工厂=
设置)
定义
启用工件(
我,
艺术品名称):
我.
艺术品名称列表.
添加(
艺术品名称)
定义
是否启用艺术品(
我,
名称):
返回
名称
在
我.
艺术品名称
定义
启用日志(
我,
日志队列名称,
日志级别):
如果 isinstance(
日志_qnames,
字符串):
日志_qnames = [
日志_qnames]
for 日志_qname
在
日志队列名称:
我.
日志队列名称到级别的映射[
日志队列名称] =
日志级别
定义
获取日志级别对(
我):
返回用户请求的所有合格模块名称
明确的日志设置。
..警告：
此函数曾用于返回所有记录器，无论是否
或不是用户指定了它们，或者没有；现在它只返回日志
用户明确提到的（以及 torch，）
总是在我们初始化日志时隐式请求
子系统。)
"文档"
返回
我.log_qname_to_level.
项目()
定义
清晰(
我):
我.log_qname_to_level.
清晰()
我.artifact_names.
清晰()
日志注册表 =
日志注册表()
日志状态 =
日志状态()
torch._logging.set_logs(**torch._logging.DEFAULT_LOGGING)
DEFAULT_LOGGING = {
"dynamo": 记录.INFO,
"aot": 记录.INFO,
电感器:
记录.INFO,
"fsdp": 记录.
信息,
ddp 图:
是,
图断点:
是,
守卫:
是,
重新编译:
是,
动态的:
记录.
信息,
}
[文档]
定义
设置日志(
*,
所有:
可选[
整数] =
无,
dynamo: 可选[
整数] =
无,
自动翻译:
可选[
整数] =
无,
自动微分:
可选[
整数] =
无,
动态:
可选[
整数] =
无,
电感器:
可选[
整数] =
无,
分布式:
可选[
整数] =
无,
c10d: 可选[
整数] =
无,
ddp: 可选[
整数] =
无,
fsdp: 可选[
整数] =
无,
dtensor: 可选[
整数] =
无,
onnx: 可选[
整数] =
无,
字节码:
布尔 =
错误,
aot_graphs: 布尔 =
错误,
沉浸式翻译图:
布尔 =
错误,
分布式数据并行图:
布尔 =
错误,
图:
布尔 =
错误,
图形代码:
布尔 =
错误,
图断点:
布尔 =
错误,
图大小:
布尔 =
错误,
守卫:
布尔 =
错误,
重新编译:
布尔 =
错误,
重新编译详细:
布尔 =
错误,
跟踪源:
布尔 =
错误,
跟踪调用:
布尔 =
错误,
trace_bytecode: 布尔 =
错误,
输出代码:
布尔 =
错误,
内核代码:
布尔 =
错误,
调度:
布尔 =
错误,
性能提示:
布尔 =
错误,
预毕业图:
布尔 =
错误,
毕业后图:
布尔 =
错误,
ir_pre_fusion: 布尔 =
错误,
穿透式翻译:
布尔 =
错误,
ONNX 诊断:
布尔 =
错误,
融合:
布尔 =
错误,
重叠:
布尔 =
错误,
导出:
可选[
整数] =
无,
模块:
可选[
字典[
字符串,
联盟[
整数,
布尔]]] =
无,
cuDAG 图:
布尔 =
错误,
sym_node: 布尔 =
错误,
编译自动微分:
布尔 =
错误,
编译自动微分详细模式:
布尔 =
错误,
cuDNN 图静态输入:
布尔 =
错误,
基准测试:
布尔 =
错误,
自动调整:
布尔 =
错误,
图区域扩展:
布尔 =
错误,
):
""
设置单个组件的日志级别并切换单个日志
艺术品类型。
..警告:: 此功能为原型，可能在未来存在兼容性问题。
可能会有破坏性变更。
..注意:: ``TORCH_LOGS`` 环境变量具有完全优先级。
在此函数上，所以如果设置了，此函数不做任何事情。
组件是 PyTorch 中相关特性的集合。所有日志
从给定组件发出的消息具有自己的日志级别。如果该
日志级别高于或等于特定消息的优先级
组件的日志级别设置，如果它被触发，则输出。否则，将被抑制。
这允许您，例如，静音大量日志消息
与您无关且会增加日志组件的冗余
与您相关。期望的日志级别值，从高到低排序为：
优先级，包括：
* ``logging.CRITICAL``
* ``日志错误``
* ``日志警告``
* ``日志信息``
* ``日志调试``
* ``logging.NOTSET``
请参阅 Python ``logging`` 模块的文档以获取更多关于日志级别的信息
日志级别请见 ``_
艺术品是一种特定的日志消息类型。每个艺术品都被分配
发送到父组件。组件可以发出许多不同类型的
文物。一般来说，如果其对应的
设置在以下参数列表中已开启或如果其父组件
设置为小于或等于该工件日志级别的日志级别。
关键字参数：
all (:class:`Optional[int]`):
所有组件的默认日志级别。默认：``logging.WARN``
dynamo (:class:`Optional[int]`):
TorchDynamo 组件的日志级别。默认：``logging.WARN``
aot (:class:`Optional[int]`):
AOTAutograd 组件的日志级别。默认：``logging.WARN``
autograd (:class:`Optional[int]`):
autograd 的日志级别。默认：``logging.WARN``
电感器 (:class:`Optional[int]`):
TorchInductor 组件的日志级别。默认：``logging.WARN``
动态 (:class:`Optional[int]`):
动态形状的日志级别。默认：``logging.WARN``
分布式 (:class:`Optional[int]`):
是否记录 PyTorch 分布式组件的 c10d 通信操作和其他调试信息。
默认：`logging.WARN`
c10d (:class:`Optional[int]`):
是否在 PyTorch 分布式组件中记录 c10d 通信操作的调试信息。
默认：`logging.WARN`
ddp (:class:`Optional[int]`):
是否在 PyTorch 分布式组件中记录与`DistributedDataParallel`(DDP)相关的调试信息。
默认：``logging.WARN``
fsdp (:class:`Optional[int]`)
是否在 PyTorch 分布式组件中记录与 ``FullyShardedDataParallel``(FSDP) 相关的调试信息。
默认：``logging.WARN``
dtensor (:class:`Optional[int]`):
是否在 PyTorch 分布式组件中记录与 ``DTensor``(DTensor) 相关的调试信息。
默认：``logging.WARN``
onnx (:class:`Optional[int]`):
ONNX 导出组件的日志级别。默认：`logging.WARN`
bytecode (:class:`bool`):
是否从 TorchDynamo 输出原始和生成的字节码。
默认：``False``
aot_graphs (:class:`bool`):
是否输出 AOTAutograd 生成的图。默认：``False``
aot_joint_graph (:class:`bool`):
是否输出 AOTAutograd 生成的联合前向-反向图。默认：``False``
ddp_graphs (:class:`bool`):
是否输出 DDPOptimizer 生成的图。默认：``False``
graph (:class:`bool`):
是否以表格形式输出 TorchDynamo 捕获的图。
默认：``False``
graph_code (:class:`bool`):
是否输出由 TorchDynamo 捕获的图的 Python 源代码。
默认：``False``
graph_breaks (:class:`bool`):
是否输出 TorchDynamo 遇到的图断点。
默认：``False``
graph_sizes (:class:`bool`):
是否输出由 TorchDynamo 捕获的图的张量大小。
默认：``False``
guards (:class:`bool`):
是否输出 TorchDynamo 为每个编译的
函数。默认：``False``
重新编译（:class:`bool`）:
是否每次在 TorchDynamo 重新编译函数时都发出守卫失败的原因和消息
TorchDynamo 重新编译函数时是否发出守卫失败的原因和消息。默认：``False``
recompiles_verbose (:class:`bool`):
是否在 TorchDynamo 重新编译函数时输出所有守卫失败的原因
一个函数，即使这些函数实际上并没有运行。默认：``False``
trace_source (:class:`bool`):
当 TorchDynamo 开始追踪新行时是否发出。默认：``False``
trace_call (:class:`bool`):
当 TorchDynamo 创建对应函数调用的 FX 节点时是否发出详细的行位置信息
Python 3.11+ 仅支持。默认：``False``
trace_bytecode (:class:`bool`):
是否输出字节码指令和追踪的堆栈状态作为 TorchDynamo
追踪字节码。默认：``False``
output_code (:class:`bool`):
是否按图生成 TorchInductor 输出代码。默认：``False``
kernel_code (:class:`bool`):
是否按内核生成 TorchInductor 输出代码。默认：``False``
schedule (:class:`bool`):
是否输出 TorchInductor 调度。默认：``False``
perf_hints (:class:`bool`):
是否输出 TorchInductor 性能提示。默认：``False``
pre_grad_graphs (:class:`bool`):
是否在电感梯度通过之前发出图形。默认：``False``
post_grad_graphs (:class:`bool`):
是否输出由后置梯度传递生成的图。默认：``False``
ir_pre_fusion (布尔类):
是否在电感器融合步骤之前输出图表。默认：``False``
ir_post_fusion (:class:`bool`):
是否在电感器融合步骤后输出图表。默认：``False``
onnx_diagnostics (:class:`bool`):
是否在日志中输出 ONNX 导出器的诊断信息。默认：``False``
fusion (:class:`bool`):
是否输出详细的电感器融合决策。默认：``False``
overlap (:class:`bool`):
是否输出详细的电感器计算/通信重叠决策。默认：``False``
sym_node (:class:`bool`):
是否输出各种 SymNode 操作调试信息。默认：``False``
导出 (:class:`Optional[int]`):
导出的日志级别。默认：``logging.WARN``
基准测试 (:class:`bool`):
是否输出详细的电感器基准测试信息。默认：``False``
modules (dict):
此参数提供了一种指定上述日志组件和工件设置的备选方法，格式为关键字参数
组件和工件设置，格式为关键字参数
字典作为单个参数给出。有两种情况
此处有用（1）如果有一个新的日志组件或工件
已注册但未添加关键字参数
此函数和（2）如果未注册模块的日志级别
需要设置。这可以通过提供完全限定的模块来完成
以名称作为键，以日志级别作为值。默认：``None``
cudagraph 静态输入(:class:`bool`):
是否为 cudagraph 静态输入检测输出调试信息。默认：``False``
自动调整（:class:`bool`）:
自动调整选择日志，例如内核源代码、性能和调整参数。默认：``False``
graph_region_expansion（:class:`bool`）:
是否输出重复图区域跟踪算法的详细步骤。默认：``False``
示例::
>>> # xdoctest: +SKIP
>>> 导入日志记录
# 以下代码将 "dynamo" 组件更改为输出 DEBUG 级别的日志
# 并输出 "graph_code" 艺术品
>>> torch._logging.set_logs(dynamo=logging.DEBUG, graph_code=True)
# 启用不同模块的日志
>>> torch._logging.set_logs(modules={"unregistered.module.name": logging.DEBUG})
"文档"
# 如果设置了环境变量则忽略
如果 LOG_ENV_VAR
在 os.
环境:
日志.
警告(
使用 TORCH_LOGS 环境变量进行日志设置，忽略对 set_logs 的调用
)
返回
日志状态.
清晰()
模块 =
模块
或者 {}
定义
_设置日志(**kwargs):
for 别名, val
在 itertools.chain(kwargs.
项目(),
模块.
项目()):
# 类型：忽略[联合属性]
如果 val
是
无:
continue
如果
日志注册表.
是否为工件(
别名):
如果
不 isinstance(val,
布尔):
raise ValueError(
f预期布尔值以启用工件{
别名}
，收到{val}"
)
如果 val:
日志状态.
启用工件(
别名)
elif 日志注册表.
是日志(
别名)
或者
别名
在
日志注册表.
子日志队列名称:
如果 val
不
在
记录.
_等级名称:
raise ValueError(
f"无法识别的日志等级，用于日志"{
别名}: {val}
,有效的级别值"
f"是：{
逗号.
连接
[
字符串(k) for k
在
记录.
_级别转名称.
键()])}"
)
日志状态.
启用日志(
日志注册表.
日志别名到日志名称映射.
获取(
别名,
别名), val
)
否则:
raise ValueError(
f"传递给 set_logs 的日志或工件名称未识别："{
别名}"
)
_初始化日志()
设置日志(
火把=
所有,
dynamo=dynamo,
预编译=
预编译,
自动微分=
自动微分,
电感器=
电感器,
动态=
动态,
字节码=
字节码,
aot_graphs=aot_graphs,
aot_联合图=
aot_联合图,
ddp_图=
ddp_图,
分布式=
分布式,
c10d=c10d,
ddp=ddp,
文件系统描述符=
文件系统描述符,
张量=
张量,
图=
图,
图形代码=
图形代码,
图断点=
图断点,
图大小=
图大小,
守卫=
守卫,
重新编译=
重新编译,
重新编译详细=
重新编译详细,
跟踪源=
跟踪源,
跟踪调用=
跟踪调用,
trace_bytecode=trace_bytecode,
输出代码=
输出代码,
内核代码=
内核代码,
调度=
调度,
性能提示=
性能提示,
预梯度图=
预梯度图,
后梯度图=
后梯度图,
红外预融合=
红外预融合,
红外后融合=
红外后融合,
onnx=onnx,
onnx 诊断=
onnx 诊断,
融合=
融合,
重叠=
重叠,
sym_node=sym_node,
导出=
导出,
cuDAG 图=
cuDAG 图,
compiled_autograd=compiled_autograd,
编译式自动微分详细模式=
编译式自动微分详细模式,
CuGraph 静态输入=
CuGraph 静态输入,
基准测试=
基准测试,
自动调优=
自动调优,
图形区域扩展=
图形区域扩展,
)
定义
获取日志记录器():
""
返回：所有已注册日志记录器的列表
"文档"
返回 [
记录.
获取日志记录器(qname) for qname
在
日志注册表.
获取日志 qname()]
定义
注册日志(
设置名称,
日志名称):
""
启用通过环境变量和用户 API 使用设置名称来控制日志
参数：
setting_name：在环境变量和用户 API 中使用的简称
log_name：与 setting_name 关联的日志名称
"文档"
日志注册表.
注册日志(
设置名称,
日志名称)
定义
注册工件(
设置名称,
描述,
可见=
错误,
默认关闭=
错误,
日志格式=
无
):
""
允许通过环境变量和用户 API 以名称控制工件
参数：
setting_name：在环境变量和用户 API 中使用的简称
description：对此输出的描述
visible：默认情况下是否建议给用户
默认关闭：是否在祖先记录器启用 DEBUG 级别时记录此项目
DEBUG 级别启用时是否记录
"文档"
日志注册表.
注册项目名称(
设置名称,
描述,
可见,
默认关闭,
日志格式
)
定义
获取 Artifact 记录器(
模块_qname,
艺术品名称):
如果
艺术品名称
不
在
日志注册表.
艺术品名称列表:
raise ValueError(
f"艺术品名称："{
表示(
艺术品名称)}
未注册，
f请调用 register_artifact({
表示(
艺术品名称)}
torch._logging.registrations. 中的 )
)
qname = module_qname + f".__"{
艺术品名称}"
日志 =
记录.
获取日志记录器(
命名空间)
日志.
艺术品名称 =
艺术品名称
# 类型: 忽略[attr-defined]
日志注册表.
注册艺术品日志(qname)
配置工件日志(
日志)
返回
日志
增加详细程度字符 =
加号
减少详细程度字符 =
"—"
详尽正则表达式 = (
“（”
+ |.
连接
[
正则表达式.
转义(
增加详尽字符),
正则表达式.
转义(
减少详尽字符
)]])
+ “?)”
)
定义
配置工件日志(
日志):
如果工件默认关闭，则只有在明确启用的情况下才应记录；将 propagate 设置为 False，以便不传播此工件
将 propagate 设置为 False，以便不传播此工件
传递给其祖先记录器
如果
日志注册表.
默认关闭(
日志.
艺术品名称):
日志.
传播 =
假
# 当显式启用时启用工件日志记录
如果
日志状态.
是否启用了工件(
日志.
艺术品名称):
日志.
设置等级(
记录.
调试)
日志.
传播 =
真实
匹配逗号分隔的日志名称列表（逗号后允许有空格）
定义 _gen_settings_regex():
返回
正则表达式.
编译(r"((\+|-)?[\w\.]+,\s*)*(\+|-)?[\w\.]+?")
定义 _validate_settings(
设置):
返回
正则表达式.
完全匹配(
_生成设置正则表达式(),
设置)
是
不
无
定义
帮助信息(
详细模式=
错误):
定义
填充到(s,
长度=30):
断言
长度(s)
≤
长度
返回 s +
输入文本为空，请提供需要翻译的文本 * (
长度 -
长度(s))
如果
详细模式:
打印的工件 =
日志注册表.
工件名称
否则:
打印艺术品 =
日志注册表.
可见艺术品
如果
详细模式:
标题 =
"所有已注册的名称"
否则:
标题 =
"可见的已注册名称（使用 TORCH_LOGS='+help'查看完整列表）"
行 = (
[所有]
+ 排序(
日志注册表.
日志别名到日志全称映射.
键())
+ 排序(
[
f"{填充到(
名称)}
制表符{
日志注册.
艺术品描述[
名称]}"
for 名称
在
打印艺术品
]
)
)
设置信息 = " " + "
输入文本翻译为简体中文为：\n ".
连接(
行)
示例 s =
""
示例：
设置 TorchDynamo 的日志级别为 logging.DEBUG 和 AOT 为 logging.INFO，请使用 TORCH_LOGS="+dynamo,aot
设置 TorchDynamo 的日志级别为 logging.DEBUG 和 AOT 为 logging.INFO，请使用 TORCH_LOGS="-dynamo,+inductor
设置 TorchDynamo 的日志级别为 logging.DEBUG 和 AOT 为 logging.INFO，请使用 TORCH_LOGS="-dynamo,+inductor
logging.ERROR 级别和 TorchInductor 调整为 logging.DEBUG 级别
设置 TORCH_LOGS="aot_graphs" 将启用 aot_graphs 艺术品
设置 TORCH_LOGS="+dynamo,schedule" 将启用设置 TorchDynamo 的日志级别
调整为 logging.DEBUG 级别并启用 schedule 艺术品
TORCH_LOGS="+some.random.module,schedule" 将设置 some.random.module 的日志级别为 logging.DEBUG 并启用 schedule 艺术品
some.random.module 的日志级别设置为 logging.DEBUG 并启用 schedule 艺术品
TORCH_LOGS_FORMAT=%(levelname)s: %(message)s
或任何提供的格式
字符串将设置输出格式
有效的键有 "levelname", "message", "pathname", "levelno", "lineno",
"文件名" 和 "名称"。
TORCH_LOGS_OUT=/tmp/output.txt 将日志输出到 /tmp/output.txt，作为
不错。当输出内容较长时很有用。
""
# flake8: 无 qa: B950
msg = f""
TORCH_LOGS 信息
{示例}
{标题}
{设置信息}
""
返回 msg
定义
_无效设置错误信息(
设置,
详细模式=
错误):
有效设置 = (
[所有]
+ 列表(
日志注册表.
日志别名到日志 Q 名称映射.
键())
+ 列表(
日志注册表.
艺术品名称)
)
有效的设置 =
“，”.
连接(
排序(
有效的设置))
msg = f""
无效的日志设置：{
设置}
，必须是一个逗号分隔的列表，包含完全
限定的模块名称、已注册的日志名称或已注册的工件名称。
关于各种设置的更多信息，请尝试设置 TORCH_LOGS="help"
有效的设置：
{valid_settings}
""
返回 msg
@functools.lru_cache
定义 _parse_log_settings(
设置):
如果
设置 ==
输入文本翻译为简体中文为："":
返回 {}
如果
设置 ==
帮助:
raise ValueError(帮助信息(
详细模式=
错误))
elif 设置 ==
+帮助:
raise ValueError(帮助信息(
详细模式=
是))
如果
不
_验证设置(
设置):
raise ValueError(无效设置错误信息(
设置))
设置 =
正则表达式.
子(r
\s+,
输入文本翻译为简体中文为："",
设置)
日志名称 =
设置.
分割(",")
定义
获取名称级别对(
名称):
清洁名称 =
名称.
替换(
增加详细字符,
输入文本翻译为简体中文为："")
清洁名称 =
清洁名称.
替换(
减少详细程度字符,
输入文本翻译为简体中文为："")
如果
名称[0] ==
增加详细程度字符:
等级 =
记录.
调试
elif 名称[0] == DECR_VERBOSITY_CHAR:
等级 =
记录.
错误
否则:
等级 =
记录.
信息
返回
清洁名称,
等级
日志状态 =
日志状态()
for 名称
在
日志名称:
名称,
等级 =
获取名称级别对(
名称)
如果
名称 ==
所有:
名称 =
torch
如果
日志注册表.
是日志(
名称):
断言
等级
是
不
无
日志_qnames =
日志注册表.
日志别名到日志_qnames[
名称]
日志状态.
启用日志(
日志名称,
级别)
elif 日志注册表.
是艺术品(
名称):
记录状态.
启用艺术品(
名称)
elif _是否有效模块(
名称):
如果
不
已注册父节点(
名称):
日志注册.
注册日志(
名称,
名称)
否则:
日志注册.
注册子日志(
名称)
日志状态.
启用日志(
名称,
级别)
否则:
raise ValueError(无效设置错误信息(
设置))
返回
日志状态
定义 _is_valid_module(qname):
规范 = importlib.
工具.
查找规范(qname)
返回
规范
是
不
无
定义
从环境更新日志状态():
全局
日志状态
日志设置 = os.
环境.
获取(
环境变量日志,
无)
如果
日志设置
是
不
无:
日志状态 =
解析日志设置(
日志设置)
定义
_已注册父项(
日志全名):
当前日志 =
记录.
获取日志记录器(
日志全名)
已注册的日志全名 =
日志注册表.
获取日志全名()
当
当前日志.
父节点:
如果
当前日志.
名称
在
已注册的日志 qname:
返回
真实
当前日志 =
当前日志.
父级
返回
假
定义
生成模块相对路径(
绝对路径):
""
给定一个对应 Python 模块的绝对文件路径
通过正常导入机制使用 sys.path 加载，将其转换为
相对于 Python 搜索路径之一的相对路径。
"文档"
绝对路径 = pathlib.
路径(
绝对路径).
解决()
for 路径
在
系统模块.
路径:
尝试:
相对路径 =
绝对路径.
相对于(
路径)
除了 ValueError:
continue
否则:
返回
字符串(
相对路径)
返回
字符串(
绝对路径)
当必要时对工件应用自定义格式
类 TorchLogsFormatter(
记录.
格式化器):
定义 __init__(
我, *,
跟踪:
布尔 =
错误,
trace_id 过滤器:
可选[
设置[
字符串]] =
无
):
超级().__init__()
我._is_trace =
跟踪
我.
_trace_id 过滤器 =
跟踪 ID 过滤器
定义
格式(
我,
记录):
艺术品名称 = getattr(
记录.
获取日志记录器(
记录.
名称),
"艺术品名称",
无)
如果
艺术品名称
是
不
无:
工件格式化器 =
日志注册.
艺术品日志格式化器.
获取(
艺术品名称,
无
)
如果
艺术品格式化器
是
不
无:
返回
艺术品格式化器.
格式(
记录)
记录.
消息 =
记录.
获取消息()
记录.asctime =
我.formatTime(
记录,
"%m"%d
%H:%M:%S)
异常处理 - 从 logging.Formatter.format 复制
s = 记录.
消息
如果
记录.exc_info:
避免多次转换，缓存 traceback 文本
（它本来就是常量）
如果
不
记录.exc_text:
记录.exc_text =
我.
格式异常(
记录.exc_info)
如果
记录.
异常文本:
如果 s[-1
] != "
输入文本翻译为简体中文为：\n":
s = s + "输入文本翻译为简体中文为：\n"
s = s + 记录.
激活文本
如果
记录.
栈信息:
如果 s[-1
] != "
输入文本翻译为简体中文为：\n":
s = s + "输入文本翻译为简体中文为：\n"
s = s + 我.
格式化栈(
记录.
栈信息)
记录.
排名前缀 =
请提供需要翻译的文本
如果
不
我._is_trace
并且
距离.
是否可用()
并且
距离.
已初始化():
记录.
排名前缀 = f
"[排名"{
距离.
获取排名()}
]
:
记录.
跟踪 ID =
请提供需要翻译的文本
如果 (
不
我.
_是否跟踪
并且 (
跟踪 ID :=
火把.
_守卫.
编译上下文.
当前跟踪 ID())
是
不
无
):
记录.
跟踪 ID = f" [{
跟踪 ID}]"
glog_level_to_abbr = {
"DEBUG": "V", # V 代表在 glog 中的 VERBOSE
"信息":
"我",
"警告": "W",
"错误": "E",
"严重": "C",
}
简短级别 =
日志级别缩写.
获取(
记录.
级别名称,
记录.
级别名称)
记录.
艺术品前缀 =
请提供需要翻译的文本
如果
艺术品名称
是
不
无:
记录.
艺术品前缀 = f
"[__"{
艺术品名称}]"
文件路径 =
生成模块相对路径(
记录.
路径名)
如果 (
我.
_跟踪 ID 过滤器
并且
记录.traceid.strip()
不
在
我._trace_id_filter
):
返回
请提供需要翻译的文本
前缀 = (
f"{记录.
排名前缀}{
简短级别}{
记录.
获取时间}.{
整数(
记录.
毫秒数 * 1000):06d} {
记录.
处理} "
f"{filepath}
输入文本:
:"
翻译:
：
f"{记录.
行号}]{
记录.traceid}{
记录.artifactprefix}"
)
如果
我._is_trace:
断言 s ==
请提供需要翻译的文本
尝试:
r = f"{前缀} {json.
压缩包(
记录.
元数据)}"
除了
类型错误:
日志.
警告(
失败的元数据：%r",
记录.
元数据)
raise
如果
记录.
负载数据
是
不
无:
r += 输入文本翻译为简体中文为："".
连接(f"\n\t{l}" for l
在
记录.
负载数据.
分割("
输入文本翻译为简体中文为：\n"))
返回 r
否则:
行 = s.
分割("
输入文本翻译为简体中文为：\n")
返回 "
输入文本翻译为简体中文为：\n".
连接(f"{
前缀} {l}" for l
在
行)
定义
默认格式化器():
fmt = os.环境.
获取(
LOG_FORMAT_ENV_VAR
日志格式环境变量, 无)
跟踪 ID 过滤器 = {
项目.strip()
for 项目
在 os.
环境.
获取(LOG_TRACE_ID_FILTER,
输入文本翻译为简体中文为："").
分割(",")
如果
项目.strip()
}
如果
格式化
是
无:
返回
Torch 日志格式化器(
跟踪 ID 过滤器=
跟踪 ID 过滤器)
否则:
如果 fmt
在 (
简短,
基本格式):
fmt = 记录.
基本格式
返回
记录.
格式化器(
格式化)
默认格式化器 =
_默认格式化器()
定义
_设置处理器(
创建处理函数,
日志):
调试处理程序 =
_跟踪处理程序(
创建处理函数())
调试处理器.
设置格式化器(
默认格式化器)
调试处理器.
设置等级(
记录.
调试)
日志.
添加处理器(
调试处理器)
处理器 = WeakSet() # type: ignore[var-annotated]
标记我们已创建的处理程序
以免修改用户处理程序
定义
_跟踪处理程序(
处理器):
处理程序.
添加(
处理器)
返回
处理器
定义 _is_torch_handler(
处理器):
返回
处理器
在
处理器
# 清除指定日志器上的所有 torch 处理器
定义 _clear_handlers(
日志):
删除 = [
处理器 for
处理器
在
日志.
处理程序
如果 _is_torch_handler(
处理器)]
for 处理器
在
删除:
日志.
移除处理程序(
处理器)
定义
_重置日志():
# 重置所有已注册的日志
for 日志_qname
在
日志注册表.
获取日志 Q 名称():
日志 =
记录.
获取日志记录器(
日志 Q 名称)
日志.
设置级别(
记录.
警告)
日志.
传播 =
假
_清除处理器(
日志)
# 重置所有工件和子日志
for 艺术品日志全名
在 itertools.chain(
日志注册表.
获取艺术品日志全名(),
日志注册表.
获取子日志 Q 名称()
):
日志 =
记录.
获取日志记录器(
艺术品日志 Q 名称)
日志.
设置级别(
记录.
未设置)
日志.
传播 =
真实
跟踪日志.
传播 =
假
清除处理器(
追踪日志)
定义
_获取日志状态():
返回
日志状态
定义
_设置日志状态(
状态):
全局
日志状态
日志状态 =
状态
定义
_初始化日志(
日志文件名=
无):
全局
获取 DTrace 结构化信息
重置日志()
从环境变量更新日志状态()
out = os.环境.
获取(
LOG_OUT 环境变量,
无)
如果 out
是
不
无:
日志文件名 = out
首先，将所有已知的（已注册的）记录器重置为 NOTSET，以便它们
尊重其父日志级别
for log_qname 在
日志注册表.
获取日志名称():
# 但不是顶级 torch 级别：这默认为 WARNING
# 以防止我们的日志消息泄露到较低级别
如果 log_qname ==
torch:
continue
日志 =
记录.
获取日志记录器(log_qname)
日志.
设置级别(
记录.
未设置)
# 现在，对于用户请求的非标准日志记录器，修改它们的日志级别
# logging behavior, modify their log levels
for log_qname, 等级
在
日志状态.get_log_level_pairs():
日志 =
记录.
获取日志记录器(log_qname)
日志.
设置级别(
级别)
最后，为所有已注册的记录器设置处理器
for log_qname 在
日志注册表.
获取日志 Q 名称():
日志 =
记录.
获取日志记录器(
日志 Q 名称)
设置处理程序(
记录.
流处理器,
日志,
)
如果
日志文件名
是
不
无:
_设置处理器(
lambda: 记录.
文件处理器(
日志文件名),
日志,
)
# 配置工件记录器，注意：这必须最后进行
# 由于考虑了祖先记录器的级别
for artifact_log_qname 在
日志注册表.
获取工件日志 Q 名称():
日志 =
记录.
获取日志记录器(
艺术品日志全名)
配置艺术品日志(
日志)
# 设置特殊跟踪日志的处理器，具有不同的默认配置
# 配置
跟踪目录名称 = os.
环境.
获取(
跟踪环境变量,
无)
如果
dtrace 目录名称 := os.
环境.
获取(
DTRACE 环境变量,
无):
获取 dtrace 结构化信息 =
真实
跟踪目录名称 =
dtrace 目录名称
# 此处理程序可能会在 trace_dir_name 为空且我们不在的情况下删除自己
实际上在一个 FB 环境中。这允许我们推迟实际
初始化它，直到我们实际上需要记录任何内容。这很重要
因为 JK 初始化了一个 C++单例，如果我们随后进行 fork 操作，这将会
损害我们的进程。
全局
日志跟踪处理器
如果
日志跟踪处理器
是
无:
日志跟踪处理器 =
懒人跟踪处理器(
跟踪目录名称)
此日志始终处于调试级别。我们还将进一步测试是否
在决定实际调用日志记录之前，是否有任何处理程序。
# 不手动调用
跟踪日志.
设置等级(
记录.
调试)
跟踪日志处理器 =
_跟踪处理器(
日志跟踪处理器)
跟踪日志处理器.
设置格式化器(
PyTorch 日志格式化器(
跟踪=
是))
跟踪日志.
添加处理器(
跟踪日志处理器)
类
懒加载跟踪处理器(
记录.
Stream 处理器):
“类似于 File 处理器，但文件仅在第一次日志消息时才被懒加载分配”
定义 __init__(
我,
根目录:
可选[
字符串
)]
# 这与延迟实现的实现方式相同
文件处理器
我.
根目录 =
根目录
记录.
处理器.__init__(
我)
我.
流 =
无
我.
内置打开 =
打开
# 从 cpython 中的 FileHandler 克隆而来
定义
关闭(
我):
我.
获取()
尝试:
尝试:
如果
我.
流:
尝试:
我.
清空()
最后:
流 =
我.
流
我.
流 =
无
如果
有属性(
流,
"关闭"):
流.
关闭()
最后:
# 调用无条件时
# 防止设置延迟时处理程序泄漏
# 参见问题 #42378：我们同样依赖于
# self._closed 被设置为 True
记录.
流处理器.
关闭(
我)
最后:
我.
释放()
定义
发射(
我,
记录):
如果
我.
流
是
无:
如果
我.
根目录
是
无:
跟踪日志目录 =
"/日志"
导入
torch 版本
是
torch 版本
如果 (
有属性(
torch 版本,
git 版本)
并且 os.
获取环境变量("MAST_HPC_JOB_NAME")
是
无
):
日志.
信息(
"LazyTraceHandler：已禁用，因为不在 fbcode 或 conda 上"
)
elif 不
火把.
_utils 内部.justknobs_check(
"pytorch/trace:启用"):
日志.
信息(
"LazyTraceHandler: 已禁用，因为 justknobs_check('pytorch/trace:启用')返回 False"
)
elif 不 os.
路径.
存在(TRACE_LOG_DIR):
日志.
信息(
"LazyTraceHandler: 已禁用，因为"%s
不存在,
跟踪日志目录,
)
elif 不 os.
访问(
追踪日志目录, os.
写操作权限):
日志.
信息(
LazyTraceHandler: 已禁用，因为%s
不可写,
追踪日志目录,
)
否则:
我.
根目录 =
追踪日志目录
如果
我.
根目录
是
不
无:
os.创建多级目录(
我.
根目录, exist_ok=
是)
等级后缀 =
请提供需要翻译的文本
如果
距离.
是否可用()
并且
距离.
已初始化():
等级后缀 = f
"等级_"{
距离.
获取排名()}
_
我.
流 =
临时文件.
命名临时文件(
模式="w+",
后缀=
".日志",
前缀=f
"专用_log_torch_trace_"{
rank 后缀}",
目录=
我.
根目录,
删除=
错误,
)
日志.
信息(
LazyTraceHandler：记录到%s",
我.
流.
名称)
否则:
我们消失，移除和空操作
跟踪日志.
移除处理程序(
我)
返回
如果
我.
流:
超级().
发射(
记录)
@functools.lru_cache(无)
定义
警告一次(
日志对象, *
参数, **kwargs):
""
该函数与 `logger.warning()` 类似，但只会发出相同消息的警告一次
注意：缓存是针对函数参数的，因此使用相同参数的 2 个不同的调用者将会命中缓存
这里假设所有警告消息在代码中都是唯一的。如果不是，则需要切换到另一种类型的缓存，这种缓存将调用帧信息包含在哈希函数中。
另一种类型的缓存，这种缓存将调用帧信息包含在哈希函数中。
"文档"
logger_obj.警告(*
参数, **kwargs)
类 LazyString:
定义 __init__(
我,
函数, *
参数, **kwargs):
我.
函数 =
函数
我.args = args
我.kwargs = kwargs
定义 __str__(
我):
返回
我.
函数(*
我.
参数, **
我.kwargs)
记录按帧/编译 ID 进行结构化日志记录所需的时间
键始终为 {frame_id}_{frame_compile_id}
结构化日志开销:
字典[
字符串, float] =
默认字典(float)
定义
添加结构化日志开销(
花费时间: float) ->
无:
全局
结构化日志开销
key = 无
如果 (
跟踪 ID :=
火把.
_守卫.
编译上下文.
当前跟踪 ID())
是
不
无:
帧 ID =
跟踪 ID.
编译 ID.
帧 ID
帧编译 ID =
跟踪 ID.
编译_id.
框架编译_id
# 为什么不追踪 trace_id.attempt，像结构化日志一样？
# 我们汇总所有尝试，因为
每次成功尝试都会记录一个编译指标
key = f"{帧 ID}_{
帧编译 ID}"
# TODO: 处理发生在特定编译 ID 之外的结构化日志
很难确定我们会在编译指标中记录哪个位置
它本身。
如果 key
是
不
无:
key = 字符串(
键)
结构化日志开销[
键] +=
花费的时间
定义
获取结构化日志开销() ->
可选[float
]：
key = 无
如果 (
跟踪 ID :=
火把.
_守卫.
编译上下文.
当前跟踪 ID())
是
不
无:
帧 ID =
跟踪 ID.
编译 ID.
帧 ID
帧编译 ID =
跟踪 ID.
编译 ID.
框架编译 ID
key = f"{框架 ID}_{
框架编译 ID}"
如果 key
是
不
无:
返回
结构化日志开销.
获取(
键)
否则:
返回
无
定义
跟踪结构化工件(
名称:
字符串,
# 将放入元数据中
编码:
字符串,
负载函数:
可调用
[]
可选[
联盟[
字符串,
对象]]] = lambda:
无,
) -> 无:
跟踪结构化(
"文物",
元数据函数=lambda: {
名称:
名称,
编码:
编码,
},
有效载荷函数=
有效载荷函数,
)
定义
结构化跟踪(
名称:
字符串,
# NB: metadata expected to be dict so adding more info is forward compatible
# Tuple[str, int] is a special case for string interning
元数据函数:
可调用
[]
联盟[
字典[
字符串,
任何
]，
元组[
字符串,
整数]]] =
字典,
*,
有效载荷函数:
可调用
[]
可选[
联盟[
字符串,
对象]]] = lambda:
无,
抑制上下文:
布尔 =
错误,
期望跟踪 ID:
布尔 =
是,
# 是否期望有一个当前的跟踪 ID
记录日志开销:
布尔 =
是,
是否记录结构化日志所花费的时间
编译 ID:
可选[
编译 ID] =
无,
# 可选，如果无法在跟踪中获取
) -> 无:
""
元数据是一个任意兼容 JSON 的结构，但预期不要过长（例如，小于 1MB）
负载是一个任意长度的字符串，但预期不会太长
负载是一个任意长度的字符串，但预期不会太长
换行，以确保没有太长的行
"文档"
断言
"名称"
不
在 [
"排名",
"编译自动微分 ID",
"帧 ID",
"帧编译 ID",
尝试,
]
断言
可调用(
元数据函数
), f元数据函数应该是可调用的，但得到了{
类型(
元数据函数)}"
断言
可调用(
payload_fn
), f"payload_fn 应该是可调用的，但得到了"{
类型(payload_fn)}"
# trace_log 永远不会传播，并且始终是 DEBUG 级别，因此也要检查这一点
# 替代检查日志级别
如果
跟踪日志.
处理器:
开始时间 =
时间.
纳秒时间()
记录:
字典[
字符串,
对象] = {}
记录[
名称] = metadata_fn()
如果
不
抑制上下文:
# TODO：实际上，排名可能应该只发出一次，在
# 顶部，而不是在所有日志中反复 spam，因为
永不改变，我们假设没有交错
如果
距离.
是否可用()
并且
距离.
已初始化():
记录[
"排名"] =
距离.
获取排名()
跟踪 ID =
火把.
_守卫.
编译上下文.
当前跟踪 ID()
如果
预期跟踪 ID
并且
跟踪 ID
是
无
并且
编译 ID
是
无:
记录日志调用的堆栈以更好地诊断原因
对于它没有帧 ID
记录[
"堆栈"] =
火把.
_日志记录.
结构化的.
来自 traceback(
捕获的 Traceback.
提取(
跳过=1).
摘要()
)
否则:
cid = 跟踪 ID.
编译 ID
如果
跟踪 ID
否则
编译 ID
如果 cid
是
不
无:
如果 cid.
编译自动微分 ID
是
不
无:
记录[
"编译自动微分 ID"] = cid.
编译自动微分 ID
如果 cid.
框架 ID
是
不
无:
记录[
frame_id] = cid.frame_id
如果 cid.frame_compile_id
是
不
无:
记录["frame_compile_id"] = cid.frame_compile_id
如果
跟踪 ID:
记录[
尝试] =
跟踪 ID.
尝试
有效载荷 =
有效载荷函数()
如果
负载数据
是
不
无:
如果
不 isinstance(
负载数据,
字符串):
如果 isinstance(
负载数据,
列表):
# 特殊情况以更佳的形式呈现
负载数据 = "[
输入文本翻译为简体中文为：\n" +
“
输入文本翻译为简体中文为：\n".
连接(json.
压缩包(i) for i
在
负载数据) + "
输入文本翻译为简体中文为：\n]"
否则:
定义
JSON 默认(
对象):
# 集合不是 JSON 可序列化的
如果 isinstance(
对象,
设置):
返回
列表(
对象)
raise 类型错误(
f"对象类型"{
类型(
对象)}
不是 JSON 可序列化的
)
# 强制换行，以避免超出行限制
有效载荷 = json.
压缩包(
负载数据,
默认=
JSON 默认,
缩进=0)
h = hashlib.md5(usedforsecurity=错误)
h.更新(
负载数据.
编码(
utf-8))
记录[
"携带负载"] = h.
摘要()
跟踪日志.
调试(
输入文本翻译为简体中文为："",
额外={
"元数据":
记录,
"负载":
负载},
栈级别=2
)
记录结构化跟踪事件(
名称,
记录)
如果
记录日志开销:
将纳秒转换为秒，并添加到帧编译总时间中
结构化日志开销_s = (
时间.
纳秒时间() -
开始时间) /
1000000000
添加结构化日志开销(
结构化日志开销_s)
定义 dtrace_structured(
名称:
字符串,
# NB: 元数据预期为字典，因此添加更多信息具有向前兼容性
# Tuple[str, int] 是字符串内嵌的特殊情况
元数据函数:
可调用
[]
联盟[
字典[
字符串,
任何
]，
元组[
字符串,
整数]]] =
字典,
*,
有效载荷函数:
可调用
[]
可选[
联盟[
字符串,
对象]]] = lambda:
无,
抑制上下文:
布尔 =
错误,
期望跟踪 ID:
布尔 =
是,
是否期望有当前跟踪 ID
记录日志开销:
布尔 =
是,
是否记录结构化日志所花费的时间
):
""
用于调试的更详细的信息记录。这可能会导致
程序变慢。
"文档"
如果 GET_DTRACE_STRUCTURED:
结构化跟踪(
名称,
元数据函数,
payload_fn=payload_fn,
suppress_context=suppress_context,
预期跟踪 ID=
预期跟踪 ID,
记录日志开销=
记录日志开销,
)
导入 torch._guards
导入 torch._utils_internal
导入 torch.distributed
是 dist
