会面

动态会合

torch.distributed.elastic.rendezvous.dynamic_rendezvous.create_handler(store, backend, params)[源代码]

使用指定的参数创建一个新的DynamicRendezvousHandler。

参数

• store (Store) - 作为rendezvous的一部分返回的C10d存储。

• backend (RendezvousBackend) – 用于持有会话状态的后端。

返回类型

DynamicRendezvousHandler

参数	描述
加入超时时间	会合预计完成的总时间（以秒为单位）。默认值为600秒。
上次呼叫超时	在达到最少节点数量后，完成会合之前的额外等待时间（以秒为单位）。默认为30秒。
关闭超时时间	从调用 RendezvousHandler.set_closed() 或 RendezvousHandler.shutdown() 开始，预期会话在多少秒后关闭。默认值为30秒。

类torch.distributed.elastic.rendezvous.dynamic_rendezvous.DynamicRendezvousHandler[源代码]

表示一个处理程序，用于在多个节点之间建立会合点。

classmethodfrom_backend(run_id, store, backend, min_nodes, max_nodes, local_addr=None, timeout=None)[源代码]

创建一个新的 DynamicRendezvousHandler。

参数

• run_id (str) – 会面的运行标识符。

• store (Store) - 作为rendezvous的一部分返回的C10d存储。

• backend (RendezvousBackend) – 用于持有会话状态的后端。

• min_nodes (int) – 会话接受的最少节点数。

• max_nodes (int) – 指定允许加入会合的最多节点数量。

• local_addr (Optional[str]) – 本地节点的地址。

• timeout (Optional[RendezvousTimeout]) – 会话的超时设置。

类torch.distributed.elastic.rendezvous.dynamic_rendezvous.RendezvousBackend[源代码]

表示一个保存会合状态的后端。

abstractget_state()[源代码]

获取会合的状态。

返回值

一个包含编码的会合状态及其栅栏令牌的元组，如果没有在后端找到任何状态，则返回None。

抛出异常

• RendezvousConnectionError — 与后端的连接失败。

• RendezvousStateError – 会话状态损坏。

返回类型

Optional[Tuple[bytes, Any]]

抽象属性 名称：str

获取后台的名称。

abstractset_state(state, token=None)[源代码]

设定会合状态。

新的会合状态根据条件进行设置：

• 如果指定的 token 与后端存储的围栏令牌匹配，则会更新状态，并将新的状态和其对应的围栏令牌返回给调用者。

• 如果指定的 token 与后端存储的 fencing token 不匹配，状态将不会被更新；相反，会将当前的状态和其 fencing token 返回给调用者。

• 如果指定的 token 为 None，则只有在后端没有现有状态的情况下才会设置新状态。然后将新的状态或者现有的状态及其 fencing token 返回给调用者。

参数

• state (bytes) – 代表编码后的会合状态。

• token (Optional[Any]) – 由先前对 get_state() 或 set_state() 的调用检索到的可选封条令牌。

返回值

一个元组，包含序列化的会合状态、对应的围栏令牌以及一个布尔值，表示设置尝试是否成功。

抛出异常

• RendezvousConnectionError — 与后端的连接失败。

• RendezvousStateError – 会话状态损坏。

返回类型

Optional[Tuple[bytes, Any, bool]]

类torch.distributed.elastic.rendezvous.dynamic_rendezvous.RendezvousTimeout(join=None, last_call=None, close=None, heartbeat=None)[源代码]

保留会合的超时设置。

参数

• join (Optional[timedelta]) – 会合预计完成的时间间隔。

• last_call (Optional[timedelta]) – 在会合达到最少所需参与者数量后，完成会合前的额外等待时间。

• close (Optional[timedelta]) – 在调用 RendezvousHandler.set_closed() 或 RendezvousHandler.shutdown() 之后，会面预计在什么时间内结束。

• keep_alive – 在此时间内预期完成保持连接活跃的心跳。

属性 close:timedelta

获取连接关闭之前的超时时长。

属性heartbeat:timedelta

获取保持连接活跃的心跳超时时间。

property join: timedelta

获取join超时时间。

属性< span>last_call:timedelta

获取上次调用的超时时间。

Etcd 会面（ Legacy）

类torch.distributed.elastic.rendezvous.etcd_rendezvous.EtcdRendezvousHandler(rdzv_impl, local_addr)[源代码]

实现一个由 torch.distributed.elastic.rendezvous.etcd_rendezvous.EtcdRendezvous 支持的 torch.distributed.elastic.rendezvous.RendezvousHandler 接口。EtcdRendezvousHandler 使用 URL 来配置 rendezvous 类型，并传递特定于实现的配置给 rendezvous 模块。基本的 etcd rendezvous 配置 URL 如下所示：

etcd://<etcd_address>:<port>/<job_id>?min_workers=<min_workers>&max_workers=<max_workers>  # noqa: W605

-- example --

etcd://localhost:2379/1234?min_workers=1&max_workers=3

上述 URL 的含义如下：

1. 使用与 etcd 方案注册的会合处理程序

2. 要使用的 etcd 端点是 localhost:2379

3. job_id == 1234 用作 etcd 中的前缀。这样可以在多个作业之间共享同一个 etcd 服务器，前提是每个作业的 job_ids 必须是唯一的。需要注意的是，作业 ID 可以是任何字符串（不一定是数字），只要它是唯一的即可。

4. min_workers=1 和 max_workers=3 指定了成员数量的范围。Torch 分布式弹性会在集群大小大于或等于 min_workers 时开始运行作业，并允许最多 max_workers 的节点加入集群。

以下是可传递给etcd rendezvous的所有参数的完整列表：

参数	描述
最小工作线程数	确保会话有效的最少工人数
最大工作线程数	最大允许的工人数
超时	next_rendvezvous 预期成功的总超时时间（默认 600 秒）
上次呼叫超时	达到最少工人数后的额外等待时间（“最后呼叫”）（默认为30秒）
etcd 前缀	路径前缀（从etcd根目录开始），在此范围内创建所有etcd节点（默认为/torchelastic/p2p）

Etcd 存储

当使用 etcd 作为会面后端时，EtcdStore 是由 next_rendezvous() 返回的 C10d Store 实例类型。

类torch.distributed.elastic.rendezvous.etcd_store.EtcdStore(etcd_client, etcd_store_prefix, timeout=None)[源代码]

利用rendezvous etcd实例来实现c10 Store接口。

这是由 EtcdRendezvous 返回的 store 对象。

add(key, num)[源代码]

原子地将一个值通过指定的整数增量进行增加。

整数以十进制字符串的形式表示。如果键不存在，默认值将被假设为0。

返回值

新增的值

返回类型

int

check(keys)[源代码]

检查所有键是否立即可用（无需等待）。

返回类型

bool

get(key)[源代码]

通过键获取值，可能需要等待一段时间。

如果键不存在，将会进行阻塞等待，最长等待时间是 timeout，或者直到键被发布为止。

返回值

值 (字节)

抛出异常

LookupError - 如果在超时后键仍未发布 –

返回类型

字节(bytes)

set(key, value)[源代码]

将键值对写入 EtcdStore。

键和值都既可以是 Python str，也可以是 bytes。

wait(keys, override_timeout=None)[源代码]

等待所有密钥发布，或直到超时。

抛出异常

LookupError - 超时发生时

Etcd 服务器

EtcdServer 是一个方便的类，它使你在子进程中轻松启动和停止 etcd 服务器。这对于测试或单节点（多工作者）部署非常有用，在这些场景中手动设置 etcd 服务器会比较麻烦。

类torch.distributed.elastic.rendezvous.etcd_server.EtcdServer(data_dir=None)[源代码]

注意

已在版本 3.4.3 的 etcd 服务器上通过测试。

在随机可用的端口上启动和停止一个独立的本地 etcd 服务器。这对于单节点或多工作进程启动以及测试场景非常有用，在这些情况下，使用 sidecar 式的 etcd 服务器要比单独设置 etcd 服务器更加方便。

此类注册了一个终止处理器，在退出时关闭etcd子进程。但请注意，这并不是调用stop()方法的替代方案。

以下备用机制用于查找 etcd 二进制文件：

1. 使用 TORCHELASTIC_ETCD_BINARY_PATH 环境变量

2. 如果存在，使用 <文件根目录>/bin/etcd

3. 使用来自 PATH 中的 etcd

使用方法

server = EtcdServer("/usr/bin/etcd", 2379, "/tmp/default.etcd")
server.start()
client = server.get_client()
# use client
server.stop()

参数

etcd_binary_path - etcd服务器二进制文件的路径（参见上文中的备用路径）