torchx.specs¶

这包含了TorchX AppDef及相关组件的定义。这些定义被组件用来定义应用程序，然后可以通过TorchX调度器或管道适配器启动。

AppDef¶

class torchx.specs.AppDef(name: str, roles: ~typing.List[~torchx.specs.api.Role] = <factory>, metadata: ~typing.Dict[str, str] = <factory>)[source]¶

表示由多个Roles和元数据组成的分布式应用程序。包含驱动程序将此应用程序提交给调度程序所需的信息。

Parameters:

name – 应用程序的名称
roles – 角色列表
metadata – 应用程序的元数据（元数据的处理取决于调度程序）

角色¶

class torchx.specs.Role(name: str, image: str, min_replicas: ~typing.Optional[int] = None, base_image: ~typing.Optional[str] = None, entrypoint: str = '<MISSING>', args: ~typing.List[str] = <factory>, env: ~typing.Dict[str, str] = <factory>, num_replicas: int = 1, max_retries: int = 0, retry_policy: ~torchx.specs.api.RetryPolicy = RetryPolicy.APPLICATION, resource: ~torchx.specs.api.Resource = <factory>, port_map: ~typing.Dict[str, int] = <factory>, metadata: ~typing.Dict[str, ~typing.Any] = <factory>, mounts: ~typing.List[~typing.Union[~torchx.specs.api.BindMount, ~torchx.specs.api.VolumeMount, ~torchx.specs.api.DeviceMount]] = <factory>)[source]¶

一组在AppDef中执行特定职责的节点。示例：

分布式数据并行应用程序 - 由单一角色（训练器）组成。
带有参数服务器的应用程序 - 由多个角色（训练器、参数服务器）组成。

注意

一个image是一个软件包，它被安装在由调度程序安排的容器上。调度程序上的容器决定了image实际上是什么。一个image可以像tar包一样简单，或者映射到一个docker image。调度程序通常知道如何根据image名称（str）“拉取”image，这可以是一个简单的名称（例如docker image）或一个url，例如s3://path/my_image.tar。

用法：

trainer = Role(name="trainer",
               image = "pytorch/torch:1",
               entrypoint = "my_trainer.py"
               args = ["--arg", "foo", ENV_VAR="FOOBAR"],
               num_replicas = 4,
               resource = Resource(cpu=1, gpu=1, memMB=500),
               port_map={"tcp_store":8080, "tensorboard": 8081},
               metadata={"local_cwd.property", value})

Parameters:

name – 角色的名称
image – 安装在容器上的软件包。
entrypoint – 在容器内调用角色的命令
args – 传递给入口点命令的命令行参数
env – 环境变量映射
num_replicas – 要运行的容器副本数量
min_replicas – 作业启动所需的最小副本数。当设置此参数时，作业大小可以根据集群资源和策略在min_replicas和num_replicas之间自动调整。如果调度器不支持自动扩展，则忽略此字段，作业大小将为num_replicas。实验性：对于HOT_SPARE重启策略，此字段用于指示作业运行所需的法定人数。
max_retries – 放弃前的最大重试次数
retry_policy – 副本失败时的重试行为
resource – 该角色的资源需求。调度器应在num_replicas个容器上调度该角色，每个容器至少应具有resource的保证。
port_map – 角色的端口映射。键是端口的唯一标识符例如 “tensorboard”: 9090
metadata – 与角色相关的自由形式信息，例如调度器特定数据。键应遵循以下模式：$scheduler.$key
mounts – 机器上的挂载点列表

pre_proc(scheduler: str, dryrun_info: AppDryRunInfo) → AppDryRunInfo[source]¶: 根据角色特定的配置修改调度器请求。该方法在调度器submit_dryrun期间为每个角色调用。如果有多个角色，则按照AppDef.roles列表定义的顺序为每个角色调用该方法。

class torchx.specs.RetryPolicy(value)[source]¶

定义了AppDef中Roles的重试策略。该策略定义了当角色副本遇到故障时的行为：

不成功（非零）退出代码
硬件/主机崩溃
抢占
驱逐

注意

并非所有调度程序都支持所有重试策略。然而，所有调度程序都必须支持RetryPolicy.APPLICATION。请参阅调度程序的文档以获取更多信息，了解它们支持的重试策略和行为注意事项（如果有的话）。

REPLICA: Replaces the replica instance. Surviving replicas are untouched.
与dist.ddp组件一起使用，以便让torchelastic协调重启和成员变更。否则，应用程序需要自行处理失败的副本离开和替换副本的接纳。
应用程序：重新启动整个应用程序。
HOT_SPARE: Restarts the replicas for a role as long as quorum (min_replicas)
使用额外的主机作为备用不会违反规定。它并不真正支持弹性，只是将num_replicas和min_replicas之间的差异用作备用（实验性）。

资源¶

class torchx.specs.Resource(cpu: int, gpu: int, memMB: int, capabilities: ~typing.Dict[str, ~typing.Any] = <factory>, devices: ~typing.Dict[str, int] = <factory>)[source]¶

表示Role的资源需求。

Parameters:

cpu – 逻辑CPU核心的数量。CPU核心的定义取决于调度器。请参阅您的调度器文档，了解逻辑CPU核心如何映射到物理核心和线程。
gpu – GPU数量
memMB – 内存的MB数
capabilities – 额外的硬件规格（由调度程序解释）
devices – 一个包含设备名称及其数量的列表

注意：您应该优先使用named_resources，而不是直接指定原始资源需求。

static copy(original: 资源, **capabilities: Any) → 资源[source]¶: 复制资源并应用新功能。如果原始资源和参数中存在相同的功能，将使用参数中的功能。

torchx.specs.resource(cpu: Optional[int] = None, gpu: Optional[int] = None, memMB: Optional[int] = None, h: Optional[str] = None) → 资源[source]¶

便捷方法，用于从原始资源规格（cpu、gpu、memMB）或已注册的命名资源（h）创建Resource对象。请注意，（cpu、gpu、memMB）与h是互斥的，如果指定了h，则优先使用。

如果指定了h，则使用它从已注册的命名资源列表中查找资源规格。请参阅注册命名资源。

否则，将从原始资源规格创建一个Resource对象。

示例：

resource(cpu=1) # returns Resource(cpu=1)
resource(named_resource="foobar") # returns registered named resource "foo"
resource(cpu=1, named_resource="foobar") # returns registered named resource "foo" (cpu=1 ignored)
resource() # returns default resource values
resource(cpu=None, gpu=None, memMB=None) # throws

torchx.specs.get_named_resources(res: str) → 资源[source]¶

根据通过entrypoints.txt注册的字符串定义获取资源对象。

TorchX 实现了 named_resource 注册机制，该机制包括以下步骤：

创建一个模块并定义您的资源检索函数：

# my_module.resources
from typing import Dict
from torchx.specs import Resource

def gpu_x_1() -> Dict[str, Resource]:
    return Resource(cpu=2, memMB=64 * 1024, gpu = 2)

在入口点部分注册资源检索：

[torchx.named_resources]
gpu_x_1 = my_module.resources:gpu_x_1

gpu_x_1 可以作为字符串参数用于此函数：

from torchx.specs import named_resources
resource = named_resources["gpu_x_1"]

AWS 命名资源¶

torchx.specs.named_resources_aws 包含表示相应AWS实例类型的资源定义，这些定义取自 https://aws.amazon.com/ec2/instance-types/。这些资源在安装torchx库后通过入口点暴露。映射存储在 setup.py 文件中。

当前命名的资源并未指定AWS实例类型的能力，而仅表示内存、CPU和GPU数量上的等效资源。

注意

这些资源定义将来可能会发生变化。预计每个用户将管理自己的资源。请遵循https://pytorch.org/torchx/latest/specs.html#torchx.specs.get_named_resources来设置命名资源。

用法：

from torchx.specs import named_resources
print(named_resources["aws_t3.medium"])
print(named_resources["aws_m5.2xlarge"])
print(named_resources["aws_p3.2xlarge"])
print(named_resources["aws_p3.8xlarge"])

torchx.specs.named_resources_aws.aws_m5_2xlarge() → 资源[source]¶

torchx.specs.named_resources_aws.aws_p3_2xlarge() → 资源[source]¶

torchx.specs.named_resources_aws.aws_p3_8xlarge() → 资源[source]¶

torchx.specs.named_resources_aws.aws_t3_medium() → 资源[source]¶

宏¶

class torchx.specs.macros[source]¶

定义可以在Role.args元素和Role.env值中使用的宏。这些宏将在运行时替换为它们的实际值。

警告

除了上述提到的字段外，宏使用的角色的其他字段不会被替换。

可用的宏：

img_root - 拉取的容器镜像的根目录
app_id - 调度程序分配的应用程序ID
replica_id - unique id for each instance of a replica of a Role,
例如，一个具有3个副本的角色可以有0、1、2作为副本ID。请注意，当容器失败并被替换时，新容器将具有与它替换的容器相同的replica_id。例如，如果节点1失败并被调度程序替换，替换节点也将具有replica_id=1。

示例：

# runs: hello_world.py --app_id ${app_id}
trainer = Role(
           name="trainer",
           entrypoint="hello_world.py",
           args=["--app_id", macros.app_id],
           env={"IMAGE_ROOT_DIR": macros.img_root})
app = AppDef("train_app", roles=[trainer])
app_handle = session.run(app, scheduler="local_docker", cfg={})

class Values(img_root: str, app_id: str, replica_id: str, rank0_env: str, base_img_root: str = 'DEPRECATED')[source]¶

apply(role: 角色) → 角色[source]¶: apply 将值应用到指定角色的副本并返回它。

substitute(arg: str) → str[source]¶: substitute 将值应用于模板参数。

运行配置¶

class torchx.specs.runopts[source]¶

保存已接受的调度程序运行配置键、默认值（如果有）和帮助消息字符串。这些选项由Scheduler提供，并在Session.run中根据用户提供的运行配置进行验证。允许None默认值。必需的选项不得有非None的默认值。

重要

这个类没有访问器，因为它旨在由Scheduler.run_config_options构造并返回，并作为“帮助”工具或异常消息的一部分打印出来。

用法：

opts = runopts()

opts.add("run_as_user", type_=str, help="user to run the job as")
opts.add("cluster_id", type_=int, help="cluster to submit the job", required=True)
opts.add("priority", type_=float, default=0.5, help="job priority")
opts.add("preemptible", type_=bool, default=False, help="is the job preemptible")

# invalid
opts.add("illegal", default=10, required=True)
opts.add("bad_type", type=str, default=10)

opts.check(cfg)
print(opts)

add(cfg_key: str, type_: Type[Optional[Union[str, int, float, bool, List[str], Dict[str, str]]]], help: str, default: Optional[Union[str, int, float, bool, List[str], Dict[str, str]]] = None, required: bool = False) → None[source]¶: 添加带有给定帮助字符串和default值（如果有）的config选项。如果未指定default，则此选项为必填选项。

cfg_from_str(cfg_str: str) → Dict[str, Optional[Union[str, int, float, bool, List[str], Dict[str, str]]]][source]¶

从字符串字面量解析调度器 cfg 并返回一个 cfg 映射，其中 cfg 值已根据此 runopts 对象指定的适当类型进行转换。未知的键将被忽略，并且不会在结果映射中返回。

注意

与resolve方法不同，此方法不会解析默认选项或检查所需选项是否实际存在于给定的cfg_str中。此方法旨在在调用resolve()之前调用，当输入是字符串编码的运行配置时。也就是说，要完全解析配置，请调用opt.resolve(opt.cfg_from_str(cfg_literal))。

如果cfg_str是一个空字符串，那么将返回一个空的cfg。否则，至少需要一个由"="（等号）分隔的键值对。

可以使用","（逗号）或";"（分号）来分隔多个键值对。

CfgVal 允许 List 的基本类型，可以作为 "," 或 ";"（分号）分隔传递。由于相同的分隔符用于分隔配置键值对，此方法将最后一个（尾随的）"," 或 ";" 解释为键值对之间的分隔符。请参见下面的示例。

示例：

opts = runopts()
opts.add("FOO", type_=List[str], default=["a"], help="an optional list option")
opts.add("BAR", type_=str, required=True, help="a required str option")

# required and default options not checked
# method returns strictly parsed cfg from the cfg literal string
opts.cfg_from_str("") == {}

# however, unknown options are ignored
# since the value type is unknown hence cannot cast to the correct type
opts.cfg_from_str("UNKNOWN=VALUE") == {}

opts.cfg_from_str("FOO=v1") == {"FOO": "v1"}

opts.cfg_from_str("FOO=v1,v2") == {"FOO": ["v1", "v2"]}
opts.cfg_from_str("FOO=v1;v2") == {"FOO": ["v1", "v2"]}

opts.cfg_from_str("FOO=v1,v2,BAR=v3") == {"FOO": ["v1", "v2"], "BAR": "v3"}
opts.cfg_from_str("FOO=v1;v2,BAR=v3") == {"FOO": ["v1", "v2"], "BAR": "v3"}
opts.cfg_from_str("FOO=v1;v2;BAR=v3") == {"FOO": ["v1", "v2"], "BAR": "v3"}

get(name: str) → Optional[runopt][source]¶: 如果注册了任何选项，则返回该选项，否则返回 None

static is_type(obj: Optional[Union[str, int, float, bool, List[str], Dict[str, str]]], tp: Type[Optional[Union[str, int, float, bool, List[str], Dict[str, str]]]]) → bool[source]¶: 如果obj是tp类型，则返回True。类似于isinstance()，但支持 tp = List[str]，因此可用于验证ConfigValue。

resolve(cfg: Mapping[str, Optional[Union[str, int, float, bool, List[str], Dict[str, str]]]]) → Dict[str, Optional[Union[str, int, float, bool, List[str], Dict[str, str]]]][source]¶: 检查给定的配置是否符合此runopts，如果未设置，则设置默认配置。

注意

此运行选项未知的额外配置将被忽略。

运行状态¶

class torchx.specs.AppStatus(state: ~torchx.specs.api.AppState, num_restarts: int = 0, msg: str = '', structured_error_msg: str = '<NONE>', ui_url: ~typing.Optional[str] = None, roles: ~typing.List[~torchx.specs.api.RoleStatus] = <factory>)[source]¶

AppDef 的运行状态。调度器可以返回任意的文本消息（msg 字段）。如果发生任何错误，调度器可以使用 json 响应填充 structured_error_msg。

replicas 表示作业中副本的状态。如果作业运行多次重试，该参数将包含最近一次重试的状态。注意：如果之前的重试失败，但最近一次重试成功或正在进行中，replicas 将不会包含发生的错误。

format(filter_roles: Optional[List[str]] = None) → str[source]¶

Format logs for app status. The app status include:

状态：应用程序的状态。
Num Restarts: 应用程序重启的次数。
角色：角色列表。
消息：调度程序返回的任意文本消息。
结构化错误信息：Json响应错误信息。
UI URL: 应用程序URL

raise_for_status() → None[source]¶: 如果状态不是SUCCEEDED，raise_for_status将引发一个AppStatusError。

class torchx.specs.AppState(value)[source]¶

应用程序的状态。应用程序从初始的UNSUBMITTED状态开始，经过SUBMITTED、PENDING、RUNNING状态，最终达到一个终止状态：SUCCEEDED、``FAILED``、CANCELLED。

如果调度器支持抢占，应用程序在抢占时将从RUNNING状态移动到PENDING状态。

如果用户停止应用程序，那么应用程序状态将移动到STOPPED，然后在调度程序实际取消作业时移动到CANCELLED。

未提交 - 应用程序尚未提交给调度程序
已提交 - 应用程序已成功提交到调度程序
待定 - 应用程序已提交给调度程序，等待分配
运行中 - 应用程序正在运行
成功 - 应用程序已成功完成
失败 - 应用程序未成功完成
已取消 - 应用程序在完成之前被取消
未知 - 应用程序状态未知

torchx.specs.ReplicaState¶: AppState的别名

挂载¶

torchx.specs.parse_mounts(opts: List[str]) → List[Union[BindMount, VolumeMount, DeviceMount]][source]¶

parse_mounts 将选项列表解析为类型化的挂载，遵循与 Docker 绑定挂载类似的格式。

可以在同一个列表中指定多个挂载点。每个挂载点必须首先指定type。

Ex:: type=bind,src=/host,dst=/container,readonly,[type=bind,src=…,dst=…]
Supported types:: 绑定挂载：类型=绑定，源=<主机路径>，目标=<容器路径>[,只读] 卷挂载：类型=卷，源=<名称/ID>，目标=<容器路径>[,只读] 设备挂载：类型=设备，源=/dev/<设备>[,目标=<容器路径>][,权限=读写执行]

class torchx.specs.BindMount(src_path: str, dst_path: str, read_only: bool = False)[source]¶

定义一个绑定挂载，将主机路径mount –bind到工作环境中。请参阅调度程序文档，了解每个调度程序中绑定挂载的操作方式。

Parameters:

src_path – 主机上的路径
dst_path – 工作环境/容器中的路径
read_only – 挂载是否为只读

class torchx.specs.VolumeMount(src: str, dst_path: str, read_only: bool = False)[source]¶: 定义一个持久卷挂载到工作环境。 :param src: 要挂载的卷的名称或ID :param dst_path: 工作环境/容器中的路径 :param read_only: 挂载是否应为只读

class torchx.specs.DeviceMount(src_path: str, dst_path: str, permissions: str = 'rwm')[source]¶: 定义要挂载到容器中的主机设备。 :param src_path: 主机上的路径 :param dst_path: 工作环境/容器中的路径 :param permissions: 要设置的设备权限。默认值：读、写、创建节点

组件检查器¶

torchx.specs.file_linter.validate(path: str, component_function: str) → List[LinterMessage][source]¶

验证函数以确保其符合组件标准。

validate 找到 component_function 并根据以下规则对其进行验证：

该函数必须具有google-styple docs
所有函数参数都必须进行注释
函数必须返回 torchx.specs.api.AppDef

Parameters:

path – Python源文件的路径。
component_function – 要验证的函数名称。

Returns:

验证错误列表

Return type:

列表[LinterMessage]

torchx.specs.file_linter.get_fn_docstring(fn: Callable[[...], object]) → Tuple[str, Dict[str, str]][source]¶

从提供的函数中解析函数和参数描述。文档字符串应采用 google-style格式

如果函数没有文档字符串，函数描述将是函数的名称，TIP 关于如何改进帮助信息和参数描述将是参数的名称。

文档字符串中不存在的参数将包含默认/必需的信息

Parameters:

fn – 带有或不带有文档字符串的函数

Returns:

函数描述，参数描述，其中键是参数的名称，值是参数的描述: 如果描述

class torchx.specs.file_linter.LinterMessage(name: str, description: str, line: int, char: int, severity: str = 'error')[source]¶

class torchx.specs.file_linter.TorchFunctionVisitor(component_function_name: str)[source]¶

访问者找到component_function并在其上运行已注册的验证器。当前已注册的验证器：

TorchxFunctionArgsValidator - validates arguments of the function.
Criteria:
每个参数都应使用类型进行注释

The following types are supported:

原始类型: {int, str, float},

可选[原始类型],

Dict[原始类型, 原始类型],

List[primitive_types],

可选[字典[原始类型, 原始类型]],

可选[列表[原始类型]]

visit_FunctionDef(node: FunctionDef) → None[source]¶: 使用子验证器验证函数定义。

class torchx.specs.file_linter.TorchXArgumentHelpFormatter(prog, indent_increment=2, max_help_position=24, width=None)[source]¶

帮助消息格式化器，用于向参数帮助添加默认值和必需项。

如果参数是必需的，类会在帮助信息的末尾附加(required)。如果参数有默认值，类会在末尾附加(default: $DEFAULT)。格式化程序设计为仅用于torchx组件函数。这些函数没有同时具有必需和默认参数的情况。

class torchx.specs.file_linter.TorchxFunctionArgsValidator[source]¶

validate(app_specs_func_def: FunctionDef) → List[LinterMessage][source]¶: 调用方法来验证提供的函数定义。

class torchx.specs.file_linter.TorchxFunctionValidator[source]¶

abstract validate(app_specs_func_def: FunctionDef) → List[LinterMessage][source]¶: 调用方法来验证提供的函数定义。

class torchx.specs.file_linter.TorchxReturnValidator[source]¶

validate(app_specs_func_def: FunctionDef) → List[LinterMessage][source]¶

Validates return annotation of the torchx function. Current allowed annotations:

AppDef
specs.AppDef