torch.distributed.elastic.multiprocessing.errors.error_handler 源代码
#!/usr/bin/env python3
# mypy: 允许未类型化定义
版权所有（c）Facebook，Inc.及其关联公司
版权所有
#
此源代码遵循在源树根目录中的 LICENSE 文件中找到的 BSD 风格许可协议。
有关许可证文件，请参阅源树根目录。
导入 faulthandler
导入 json
导入
记录日志
导入
操作系统
导入
时间
导入
跟踪回溯
导入
警告
来自
打字
导入
任意,
可选
__all__ = [错误处理器]
日志记录器 =
记录日志.
获取日志记录器(__name__)
[文档]
类
错误处理器:
""
将提供的异常对象及其它元数据一起写入
将错误以结构化的方式写入 JSON 格式到指定的错误文件中
环境变量：`TORCHELASTIC_ERROR_FILE`。如果此环境
变量未设置，则简单地记录本应包含的内容
写入到错误文件中。
此处理程序可以被继承以自定义错误处理。
子类应重写 `initialize()` 和 `record_exception()`。
```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)
```
定义 _get_error_file_path(
自身) ->
可选[
字符串
]：
""
返回错误文件路径。
可能返回 `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)
```
返回
操作系统.
环境.
获取("TORCHELASTIC_ERROR_FILE",
无)
定义
初始化(
自身) ->
无:
""
在运行我们希望捕获错误/异常的代码之前调用。
通常注册信号/故障处理程序。用户可以覆盖此
函数用于添加自定义初始化/注册，以帮助
传播/信息错误/信号/异常/故障。
```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)
```
try:
错误处理程序.
启用(
所有线程=True)
除了
异常
作为 e:
warnings.警告(f
"无法启用故障处理器。"{
类型(e).__name__}: {e}")
定义
_写入错误文件(
自身,
文件路径:
字符串,
错误信息:
字符串) ->
无:
"""将错误信息写入文件。"""
try:
与
打开(
文件路径,
w)
作为 fp:
fp.写(
错误信息)
除了
异常
作为 e:
warnings.警告(f
"无法将错误写入文件。"{
类型(e).__name__}: {e}")
定义
记录异常(
自身, e:
基础异常) ->
无:
""
将异常的详细信息以结构化信息写入错误文件，格式为 JSON。
如果无法确定错误文件，则记录本应写入错误文件的内容。
那些内容将被记录在错误日志中。
```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)
```
文件 =
自身._get_error_file_path()
如果
文件:
数据 = {
"消息": {
"消息": f"{
类型(e).__name__}: {e}",
"额外信息": {
"Python 调用栈":
跟踪回溯.format_exc(),
"时间戳":
字符串(int(
时间.
时间())),
},
}
}
与
打开(
文件,
w)
作为 fp:
json.导出(
数据, fp)
定义
在 rootcause_data 中覆盖错误代码(
自身,
根因错误文件:
字符串,
根因错误:
字典[
字符串,
任意
]
错误代码:
整型 = 0,
):
"修改从文件中读取的 rootcause_error，以正确设置退出代码。"
如果
"消息" not
在 rootcause_error:
记录器.
警告(
"子错误文件（"%s
该字段没有 `message` 字段。
输入文本翻译为简体中文为：
\n"
无法覆盖错误代码：%s",
根本错误文件,
错误代码,
)
elif isinstance(根本原因错误[
"消息"
]
字符串):
记录器.
警告(
子错误文件（%s
）有新的消息格式。
输入文本翻译为简体中文为：
\n"
跳过错误代码覆盖,
根本原因错误文件,
)
否则:
根本原因错误[
"消息"
]
[错误代码] =
错误代码
定义
错误文件输出(
自身,
根本原因错误文件:
字符串,
错误代码:
整型 = 0):
“从子进程的根本原因错误中导出父错误文件，并显示错误代码。”
与
打开(
根本原因错误文件)
作为 fp:
根本原因错误 = json.
加载(fp)
覆盖错误代码，因为子进程无法捕获错误代码
# 被如 SIGSEGV 这样的信号终止。
如果
错误代码:
自身.
在 rootcause_data 中覆盖错误代码(
根因错误文件,
根因错误,
错误代码
)
记录器.
调试(
"子错误文件（"%s
）内容：
输入文本翻译为简体中文为：\n%s",
根原因错误文件,
json.压缩包(
根原因错误,
缩进=2),
)
我的错误文件 =
自身.
_获取错误文件路径()
如果
我的错误文件:
# 防止现有错误文件
这可能发生在使用多进程创建子进程时
并且在父进程和子进程中使用了相同的环境变量（TORCHELASTIC_ERROR_FILE）
来指定错误文件（分别）
因为子进程中的环境变量是在包装函数中设置的
# 默认情况下，子进程会继承父进程的环境变量，如果子进程
# 在包装函数启动之前，进程收到一个信号
# 然后，信号处理程序将写入错误文件，然后子进程
# 将写入父进程的错误文件。在这种情况下，只需记录
# 原始错误文件内容并覆盖错误文件。
自身.
删除(
我的错误文件)
自身.
写入错误文件(
我的错误文件, json.
压缩包(
根本原因错误))
记录器.
信息(
将错误文件导出到父目录%s",
我的错误文件)
否则:
记录器.
错误(
"未定义父错误文件，以复制子错误文件（"%s)",
根原因错误文件,
)
定义 _rm(
自身, my_error_file):
如果
操作系统.
路径.
判断是否为文件(my_error_file):
记录原始文件的内容。
与
打开(my_error_file)
作为 fp:
try:
原始 = json.
压缩包(json.
加载(fp),
缩进=2)
记录器.
警告(
"%s已存在
"将被覆盖。"
"原始内容："
输入文本翻译为简体中文为：\n%s",
我的错误文件,
原始,
)
除了 json.
解码器.
JSON 解码错误:
记录器.
警告(
"%s已存在
"将被覆盖。"
"无法加载原始内容："
输入文本翻译为简体中文为：\n",
my_error_file,
)
操作系统.
删除(
我的错误文件)
