torch.distributed.elastic.events.api 源代码

#!/usr/bin/env python3
# mypy: 允许未类型化定义

版权所有（c）Facebook，Inc.及其关联公司
版权所有
#
此源代码遵循在源树根目录中的 LICENSE 文件中找到的 BSD 风格许可协议。
有关许可证文件，请参阅源树根目录。

导入 json
来自 dataclasses 导入 asdict, 数据类, 字段
来自 枚举 导入 枚举
来自 打字 导入 可选, 联合


__all__ = [事件源, "事件", 节点状态, Rdzv 事件]

事件元数据值 = 联盟[字符串, int, float, 布尔, 无]


[文档]class 事件源(str, Enum):
事件生产者的已知标识符。

AGENT = "代理"
WORKER = "工作者"


[文档]@数据类
class Event:
"""
该类表示在 torchelastic 作业执行过程中发生的通用事件。

事件可以是任何有意义的行为。

Args:
名称：事件名称。
来源：事件生产者，例如代理或工作者
时间戳：事件发生时的毫秒时间戳。
metadata: 与事件相关联的附加数据。
"""

name: 字符串
source: 事件源
timestamp: int = 0
metadata: dict[str, EventMetadataValue] = field(default_factory=dict)

def __str__(self):
return self.serialize()

    @staticmethod
def deserialize(data: Union[str, "Event"]) -> "Event":
if isinstance(data, Event):
return data
if isinstance(data, str):
data_dict = json.loads(data)
data_dict["source"] = EventSource[data_dict["source"]]  # type: ignore[possibly-undefined]
return Event(**data_dict)

def serialize(self) -> str:
return json.dumps(asdict(self))


类 节点状态(字符串, 枚举):
    """节点可以处于 rendezvous 的状态。"""

    初始化 = 初始化
    运行中 = 运行中
    成功 = 成功
    失败 = 失败


@dataclass
类 Rdzv 事件:
    ""
表示任何会合事件的 Dataclass。

参数：
事件名称。（例如：当前执行的操作）
rendezvous_run_id: 约会运行的 ID
message: 描述事件的短信
hostname: 节点的主机名
节点的进程 ID
节点状态（INIT，RUNNING，SUCCEEDED，FAILED）
如果已知， rendezvous 存储的 master 端点
如果已知，节点的 rank
本地_id：在 dynamic_rendezvous.py 中定义的节点本地_id
错误堆栈跟踪，如果是错误事件
```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)
```

    名称: 字符串
    run_id: 字符串
    消息: 字符串
    hostname: 字符串
    进程 ID: 整型
    节点状态: 节点状态
    主端点: 字符串 = 请提供需要翻译的文本
    排名: 可选[int] = None
    本地 ID: 可选[int] = None
    错误跟踪: 字符串 = 请提供需要翻译的文本

    def __str__(自身):
        返回 自身.序列化()

    @staticmethod
    def 反序列化(数据: 联盟[字符串, RdzvEvent]) -> Rdzv 事件:
        如果 isinstance(数据, Rdzv 事件):
            返回 数据
        如果 isinstance(数据, 字符串):
            数据字典 = json.loads(数据)
        数据字典[节点状态] = 节点状态[数据字典[节点状态]]  # type: ignore[possibly-undefined]
        返回 Rdzv 事件(**数据字典)

    def 序列化(自身) -> 字符串:
        返回 json.压缩包(asdict(自身))