torch.distributed.elastic.multiprocessing.subprocess_handler.subprocess_handler 的源代码

#!/usr/bin/env python3

版权所有（c）Facebook，Inc.及其关联公司
版权所有
#
此源代码遵循在源树根目录中的 LICENSE 文件中找到的 BSD 风格许可协议。
有关许可证文件，请参阅源树根目录。
导入 操作系统
导入 信号
导入 subprocess
导入 系统
来自 打字 导入 任意, 可选


__all__ = [SubprocessHandler]

是否为 Windows = 系统.平台 == win32


定义 获取默认信号() -> 信号.信号:
    获取默认终止信号。Unix 系统为 SIGTERM，Windows 系统为 CTRL_C_EVENT。
    如果 IS_WINDOWS:
        返回 信号.CTRL_C_EVENT  # type: ignore[attr-defined] # noqa: F821
    否则:
        返回 信号.SIGTERM


类 子进程处理器:
    ""
Python 的`subprocess.Popen`的便捷包装器。跟踪
与进程关联的元对象（例如，stdout 和 stderr 重定向文件描述符）。
```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)
```

    定义 初始化(
        自身,
        入口: 字符串,
        参数: 元组,
        环境: 字典[字符串, 字符串]
        标准输出: 可选[字符串]
        标准错误输出: 可选[字符串]
        本地排名 ID: int,
    ):
        自身._stdout = 打开(标准输出, w) 如果 stdout 否则 None
        自身._stderr = 打开(标准错误输出, w) 如果 标准错误 否则 None
        继承父环境变量
        环境变量 = 操作系统.环境.复制()
        环境变量.更新(环境)

        参数字符串 = (入口, *[字符串(e) 为 e 在 参数])
        自身.本地排名 ID = 本地排名 ID
        自身.进程: 子进程.Popen = 自身._popen(args_str, 环境变量)

    定义 _popen(自身, 参数: 元组, 环境: 字典[字符串, 字符串]) -> 子进程.Popen:
        kwargs: 字典[字符串, 任意] = {}
        如果 not IS_WINDOWS:
            kwargs["启动新会话"] = 真实
        返回 子进程.Popen(
            #  _PathLike[str], bytes, str]], bytes, str]` for 1st param but got
            #  `Tuple[str, *Tuple[Any, ...]]`.
            #  `Tuple[str, *Tuple[Any, ...]]`.
            参数=参数,
            环境=环境,
            标准输出=自身._stdout,
            标准错误输出=自身.标准错误输出,
            **kwargs,
        )

    定义 关闭(自身, 死亡信号: 可选[信号.信号] = 无) -> 无:
        如果 not 死亡信号:
            死亡信号 = 获取默认信号()
        如果 IS_WINDOWS:
            自身.进程.发送信号(死亡信号)
        否则:
            操作系统.杀死进程组(自身.进程.进程 ID, 死亡信号)
        如果 自身.标准输出:
            自身.标准输出.关闭()
        如果 自身.标准错误:
            自身.标准错误.关闭()