Rendezvous

在 Torch 分布式弹性的上下文中，我们使用术语rendezvous来指代一种特定功能，该功能结合了分布式同步原语和对等发现。

Torch 分布式弹性使用它来收集训练作业的参与者（即节点），以便他们都同意相同的参与者列表和每个人的角色，并在何时可以开始/恢复训练方面做出一致的集体决策。

Torch 分布式弹性 rendezvous 提供以下关键功能

屏障 (Barrier):

执行 rendezvous 的节点都将阻塞，直到 rendezvous 被视为完成 - 当至少 min 个节点加入 rendezvous 屏障（对于同一作业）时，就会发生这种情况。 这也意味着屏障不一定是固定大小的。

在达到 min 个节点数后，还有额外的少量等待时间 - 这用于确保 rendezvous 不会“太快”完成（这可能会排除大约在同一时间尝试加入的其他节点）。

如果在屏障处收集到 max 个节点数，则 rendezvous 会立即完成。

还有一个总超时时间，如果永远无法达到 min 个节点数，则会导致 rendezvous 失败 - 这旨在成为一个简单的故障保护，以帮助释放部分分配的作业资源，以防资源管理器出现问题，并且应被解释为不可重试。

排他性 (Exclusivity):

一个简单的分布式屏障是不够的，因为我们还需要确保在任何给定时间（对于给定的作业）只存在一组节点。 换句话说，新节点（即稍后加入的节点）不应能够为同一作业形成独立的并行工作组。

Torch 分布式弹性 rendezvous 确保如果一组节点已经完成 rendezvous（因此可能已经在训练），那么尝试 rendezvous 的其他“延迟”节点将仅宣布自己正在等待，并且必须等到（先前完成的）现有 rendezvous 首先被销毁。

一致性 (Consistency):

当 rendezvous 完成时，其所有成员将就作业成员资格和每个人在其中的角色达成一致。 此角色使用一个整数表示，称为 rank，介于 0 和世界大小之间。

请注意，rank 是不稳定的，因为同一个节点在下一个（重新）rendezvous 中可能会被分配不同的 rank。

容错性 (Fault-tolerance):

Torch 分布式弹性 rendezvous 旨在容忍 rendezvous 过程中的节点故障。 如果进程在加入 rendezvous 和完成 rendezvous 之间崩溃（或失去网络连接等），则将自动使用剩余的健康节点重新 rendezvous。

节点也可能在完成 rendezvous 之后（或已被其他节点观察到已完成）失败 - 这种情况将由 Torch 分布式弹性 train_loop 处理（它也会触发重新 rendezvous）。

共享键值存储 (Shared key-value store):

当 rendezvous 完成时，将创建并返回共享键值存储。 此存储实现了 torch.distributed.Store API（请参阅分布式通信文档）。

此存储仅由已完成 rendezvous 的成员共享。 它旨在供 Torch 分布式弹性使用，以交换初始化作业控制和数据平面所需的信息。

等待工作进程和 rendezvous 关闭 (Waiting workers and rendezvous closing):

Torch 分布式弹性 rendezvous 处理程序对象提供额外的功能，这些功能在技术上不属于 rendezvous 过程

1. 查询有多少工作进程迟到屏障，谁可以参与下一个 rendezvous。

2. 设置 rendezvous已关闭以向所有节点发出信号，不要参与下一个 rendezvous。

DynamicRendezvousHandler:

Torch 分布式弹性附带 DynamicRendezvousHandler 类，该类实现了上述 rendezvous 机制。 它是一种后端无关类型，需要构造期间指定特定的 RendezvousBackend 实例。

Torch 分布式用户可以实现自己的后端类型，也可以使用 PyTorch 附带的以下实现之一

• C10dRendezvousBackend：使用 C10d 存储（默认 TCPStore）作为 rendezvous 后端。 使用 C10d 存储的主要优点是它不需要任何第三方依赖项（例如 etcd）来建立 rendezvous。

• EtcdRendezvousBackend：取代了旧版 EtcdRendezvousHandler 类。 将 EtcdRendezvousBackend 实例传递给 DynamicRendezvousHandler 在功能上等同于实例化 EtcdRendezvousHandler。

  store = TCPStore("localhost")
  
  backend = C10dRendezvousBackend(store, "my_run_id")
  
  rdzv_handler = DynamicRendezvousHandler.from_backend(
      run_id="my_run_id",
      store=store,
      backend=backend,
      min_nodes=2,
      max_nodes=4
  )
  

下面是描述 rendezvous 工作原理的状态图。

注册表 (Registry)

class torch.distributed.elastic.rendezvous.RendezvousParameters(backend, endpoint, run_id, min_nodes, max_nodes, local_addr=None, **kwargs)

保存用于构造 RendezvousHandler 的参数。

参数

• backend (str) – 用于处理 rendezvous 的后端名称。

• endpoint (str) – rendezvous 的端点，通常采用 <hostname>[:<port>] 形式。

• run_id (str) – rendezvous 的 ID。

• min_nodes (int) – 允许加入 rendezvous 的最小节点数。

• max_nodes (int) – 允许加入 rendezvous 的最大节点数。

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

• **kwargs – 指定后端的其他参数。

get(key, default=None)

如果 key 存在，则返回 key 的值，否则返回 default。

返回类型

Any

get_as_bool(key, default=None)

以 bool 形式返回 key 的值。

返回类型

Optional[bool]

get_as_int(key, default=None)

以 int 形式返回 key 的值。

返回类型

Optional[int]

class torch.distributed.elastic.rendezvous.RendezvousHandlerRegistry

表示 RendezvousHandler 后端的注册表。

处理程序 (Handler)

class torch.distributed.elastic.rendezvous.RendezvousHandler

主 rendezvous 接口。

注意

分布式 Torch 用户通常不需要实现自己的 RendezvousHandler。 已经提供了基于 C10d Store 的实现，建议大多数用户使用。

abstract get_backend()

返回 rendezvous 后端的名称。

返回类型

str

abstract get_run_id()

返回 rendezvous 的运行 ID。

运行 ID 是用户定义的 ID，唯一标识分布式应用程序的实例。 它通常映射到作业 ID，用于允许节点加入正确的分布式应用程序。

返回类型

str

abstract is_closed()

检查 rendezvous 是否已关闭。

关闭的 rendezvous 表示将来在同一作业中重新 rendezvous 的所有尝试都将失败。

is_closed() 和 set_closed() 具有最终传播的语义，不应用于同步。 其目的是，如果至少一个节点确定作业已完成，它将关闭 rendezvous，其他节点将很快观察到这一点并停止运行。

返回类型

bool

abstract next_rendezvous()

进入 rendezvous 屏障的主要入口点。

阻塞直到 rendezvous 完成并且当前进程包含在形成的工作进程组中，或者发生超时，或者 rendezvous 被标记为已关闭。

返回

RendezvousInfo 的实例。

引发

• RendezvousClosedError – rendezvous 已关闭。

• RendezvousConnectionError – 与 rendezvous 后端的连接失败。

• RendezvousStateError – rendezvous 状态已损坏。

• RendezvousTimeoutError – rendezvous 未按时完成。

返回类型

RendezvousInfo

abstract num_nodes_waiting()

返回在 rendezvous 屏障处迟到的节点数，因此未包含在当前工作进程组中。

调用方应定期调用此方法以检查是否有新节点正在等待加入作业，如果是，则通过调用 next_rendezvous()（重新 rendezvous）来允许它们加入。

返回类型

int

abstract set_closed()

将 rendezvous 标记为已关闭。

abstract shutdown()

关闭为 rendezvous 打开的所有资源。

示例

rdzv_handler = ...
try:
    store, rank, world_size = rdzv_handler.next_rendezvous()
finally:
    rdzv_handler.shutdown()

返回类型

bool

property use_agent_store: bool

指示 next_rendezvous() 返回的存储引用可以与用户应用程序共享，并且在应用程序生命周期内可用。

Rendezous 处理程序实现将共享存储详细信息作为 RendezvousStoreInfo 的实例。 应用程序按照惯例使用 MASTER_ADDR/MASTER_PORT 环境变量来查找存储。

数据类 (Dataclasses)

class torch.distributed.elastic.rendezvous.RendezvousInfo(store, rank, world_size, bootstrap_store_info)

保存有关 rendezvous 的信息。

class torch.distributed.elastic.rendezvous.api.RendezvousStoreInfo(master_addr, master_port)

可用于引导训练器分布式通信的存储地址和端口

static build(rank, store)

工厂方法，在 rank0 主机上查找未使用的新端口，并使用所有 rank 查找 addr/port 信息。

如果 master_addr/master_port 是已知的（在共享现有 tcp store 服务器时很有用），请使用构造函数。

参数

• rank (int) – 当前节点的 rank

• store (Store) – 用于 rendezvous 的存储

• local_addr (Optional[str]) – 当前节点的地址，如果未提供，将从主机名解析

• server_port (Optional[int]) – TCPStore 服务器的端口，当 TCPStore 被共享时。

返回类型

RendezvousStoreInfo

异常 (Exceptions)

class torch.distributed.elastic.rendezvous.api.RendezvousError

表示 rendezvous 错误的基类型。

class torch.distributed.elastic.rendezvous.api.RendezvousClosedError

当 rendezvous 关闭时引发。

class torch.distributed.elastic.rendezvous.api.RendezvousTimeoutError

当 rendezvous 未按时完成时引发。

class torch.distributed.elastic.rendezvous.api.RendezvousConnectionError

当与 rendezvous 后端的连接失败时引发。

class torch.distributed.elastic.rendezvous.api.RendezvousStateError

当 rendezvous 的状态损坏时引发。

class torch.distributed.elastic.rendezvous.api.RendezvousGracefulExitError

当节点未包含在 rendezvous 中并优雅退出时引发。

异常是一种退出堆栈的机制，但不意味着失败。

实现

动态 Rendezvous

torch.distributed.elastic.rendezvous.dynamic_rendezvous.create_handler(store, backend, params)

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

参数

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

• backend (RendezvousBackend) – 用于保存 rendezvous 状态的后端。

返回类型

DynamicRendezvousHandler

参数	描述
join_timeout	rendezvous 预计完成的总时间，以秒为单位。默认为 600 秒。
last_call_timeout	在达到最少节点数后，完成 rendezvous 之前的额外等待时间，以秒为单位。默认为 30 秒。
close_timeout	在调用 RendezvousHandler.set_closed() 或 RendezvousHandler.shutdown() 后，rendezvous 预计关闭的时间，以秒为单位。默认为 30 秒。

class torch.distributed.elastic.rendezvous.dynamic_rendezvous.DynamicRendezvousHandler

表示一个处理程序，用于在一组节点之间建立 rendezvous。

classmethod from_backend(run_id, store, backend, min_nodes, max_nodes, local_addr=None, timeout=None)

创建一个新的 DynamicRendezvousHandler。

参数

• run_id (str) – rendezvous 的运行 id。

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

• backend (RendezvousBackend) – 用于保存 rendezvous 状态的后端。

• min_nodes (int) – 允许加入 rendezvous 的最小节点数。

• max_nodes (int) – 允许加入 rendezvous 的最大节点数。

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

• timeout (Optional[RendezvousTimeout]) – rendezvous 的超时配置。

class torch.distributed.elastic.rendezvous.dynamic_rendezvous.RendezvousBackend

表示保存 rendezvous 状态的后端。

abstract get_state()

获取 rendezvous 状态。

返回

编码后的 rendezvous 状态及其 fencing token 的元组，如果在后端中未找到状态，则为 None。

引发

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

• RendezvousStateError – rendezvous 状态已损坏。

返回类型

Optional[Tuple[bytes, Any]]

abstract property name: str

获取后端的名称。

abstract set_state(state, token=None)

设置 rendezvous 状态。

新的 rendezvous 状态是有条件设置的

• 如果指定的 token 与后端中存储的 fencing token 匹配，则状态将被更新。新的状态将连同其 fencing token 一起返回给调用者。

• 如果指定的 token 与后端中存储的 fencing token 不匹配，则状态将不会更新；而是将现有状态连同其 fencing token 一起返回给调用者。

• 如果指定的 token 为 None，则仅当后端中没有现有状态时，才会设置新状态。新的状态或现有状态将连同其 fencing token 一起返回给调用者。

参数

• state (bytes) – 编码后的 rendezvous 状态。

• token (Optional[Any]) – 通过先前调用 get_state() 或 set_state() 检索的可选 fencing token。

返回

序列化的 rendezvous 状态、其 fencing token 以及指示我们的设置尝试是否成功的布尔值的元组。

引发

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

• RendezvousStateError – rendezvous 状态已损坏。

返回类型

Optional[Tuple[bytes, Any, bool]]

class torch.distributed.elastic.rendezvous.dynamic_rendezvous.RendezvousTimeout(join=None, last_call=None, close=None, heartbeat=None)

保存 rendezvous 的超时配置。

参数

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

• last_call (Optional[timedelta]) – 一旦 rendezvous 达到所需的最少参与者人数，完成 rendezvous 之前的额外等待时间。

• close (Optional[timedelta]) – 在调用 RendezvousHandler.set_closed() 或 RendezvousHandler.shutdown() 后，rendezvous 预计关闭的时间。

• keep_alive – 保持活动心跳预计完成的时间。

property close: timedelta

获取关闭超时。

property heartbeat: timedelta

获取保持活动心跳超时。

property join: timedelta

获取加入超时。

property last_call: timedelta

获取最后一次调用超时。

C10d 后端

torch.distributed.elastic.rendezvous.c10d_rendezvous_backend.create_backend(params)

从指定的参数创建一个新的 C10dRendezvousBackend。

参数	描述
store_type	C10d store 的类型。当前支持的类型为 “tcp” 和 “file”，分别对应于 torch.distributed.TCPStore 和 torch.distributed.FileStore。默认为 “tcp”。
read_timeout	store 操作的读取超时时间，以秒为单位。默认为 60 秒。 请注意，这仅适用于 torch.distributed.TCPStore。它与 torch.distributed.FileStore 无关，后者不接受超时作为参数。
is_host	一个布尔值，指示此后端实例是否将托管 C10d store。如果未指定，将通过将此机器的主机名或 IP 地址与指定的 rendezvous 端点进行匹配来启发式地推断。默认为 None。 请注意，此配置选项仅适用于 torch.distributed.TCPStore。在正常情况下，您可以安全地跳过它；唯一需要它的时候是其值无法正确确定时（例如，rendezvous 端点具有 CNAME 作为主机名或与机器的 FQDN 不匹配）。

返回类型

Tuple[C10dRendezvousBackend, Store]

class torch.distributed.elastic.rendezvous.c10d_rendezvous_backend.C10dRendezvousBackend(store, run_id)

表示一个 C10d 支持的 rendezvous 后端。

参数

• store (Store) – 用于与 C10d store 通信的 torch.distributed.Store 实例。

• run_id (str) – rendezvous 的运行 id。

get_state()

参见基类。

返回类型

Optional[Tuple[bytes, Any]]

property name: str

参见基类。

set_state(state, token=None)

参见基类。

返回类型

Optional[Tuple[bytes, Any, bool]]

Etcd 后端

torch.distributed.elastic.rendezvous.etcd_rendezvous_backend.create_backend(params)

从指定的参数创建一个新的 EtcdRendezvousBackend。

参数	描述
read_timeout	etcd 操作的读取超时时间，以秒为单位。默认为 60 秒。
protocol	用于与 etcd 通信的协议。有效值为 “http” 和 “https”。默认为 “http”。
ssl_cert	与 HTTPS 一起使用的 SSL 客户端证书的路径。默认为 None。
ssl_cert_key	与 HTTPS 一起使用的 SSL 客户端证书的私钥的路径。默认为 None。
ca_cert	根 SSL 授权证书的路径。默认为 None。

返回类型

Tuple[EtcdRendezvousBackend, Store]

class torch.distributed.elastic.rendezvous.etcd_rendezvous_backend.EtcdRendezvousBackend(client, run_id, key_prefix=None, ttl=None)

表示一个基于 etcd 的 rendezvous 后端。

参数

• client (Client) – 用于与 etcd 通信的 etcd.Client 实例。

• run_id (str) – rendezvous 的运行 id。

• key_prefix (Optional[str]) – 在 etcd 中存储 rendezvous 状态的路径前缀。

• ttl (Optional[int]) – rendezvous 状态的 TTL（Time To Live，生存时间）。如果未指定，则默认为两小时。

get_state()

参见基类。

返回类型

Optional[Tuple[bytes, Any]]

property name: str

参见基类。

set_state(state, token=None)

参见基类。

返回类型

Optional[Tuple[bytes, Any, bool]]

Etcd Rendezvous (旧版)

警告

DynamicRendezvousHandler 类取代了 EtcdRendezvousHandler 类，并且推荐大多数用户使用。EtcdRendezvousHandler 处于维护模式，将来将被弃用。

class 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://127.0.0.1:2379/1234?min_workers=1&max_workers=3

上面的 URL 解释如下

1. 使用使用 etcd scheme 注册的 rendezvous 处理程序

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

3. job_id == 1234 用作 etcd 中的前缀（这允许为多个作业共享一个公共的 etcd 服务器，只要 job_ids 保证是唯一的）。请注意，作业 id 可以是任何字符串（例如，不需要是数字），只要它是唯一的。

4. min_workers=1 和 max_workers=3 指定成员大小的范围 - 只要集群大小大于或等于 min_workers，Torch Distributed Elastic 就会开始运行作业，并允许最多 max_workers 加入集群。

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

参数	描述
min_workers	rendezvous 有效的最小 worker 数
max_workers	允许加入的最大 worker 数
超时	在 expected_rendezvous 预期成功完成的总超时时间（默认为 600 秒）
last_call_timeout	达到最少 worker 数量后额外的等待时间（“最后呼叫”）（默认为 30 秒）
etcd_prefix	路径前缀（从 etcd 根目录），所有 etcd 节点将在其中创建（默认为 /torchelastic/p2p）

Etcd 存储

EtcdStore 是 C10d Store 实例类型，当 etcd 用作 rendezvous 后端时，由 next_rendezvous() 返回。

class torch.distributed.elastic.rendezvous.etcd_store.EtcdStore(etcd_client, etcd_store_prefix, timeout=None)

通过 piggybacking 在 rendezvous etcd 实例上来实现 c10 Store 接口。

这是由 EtcdRendezvous 返回的 store 对象。

add(key, num)

原子地将一个值增加一个整数值。

整数以 base 10 的字符串形式表示。如果 key 不存在，将假定默认值为 0。

返回

新的（递增的）值

返回类型

int

check(keys)

检查所有 key 是否立即存在（不等待）。

返回类型

bool

get(key)

通过 key 获取值，可能会进行阻塞等待。

如果 key 没有立即出现，将阻塞等待最多 timeout 时长，或者直到 key 发布。

返回

值 (bytes)

引发

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

返回类型

bytes

set(key, value)

将 key/value 对写入 EtcdStore。

key 和 value 都可以是 Python str 或 bytes。

wait(keys, override_timeout=None)

等待直到所有 key 都发布，或直到超时。

引发

LookupError - 如果发生超时 –

Etcd 服务器

EtcdServer 是一个便捷类，可让您轻松地在子进程中启动和停止 etcd 服务器。这对于测试或单节点（多 worker）部署非常有用，在这些部署中，手动设置 etcd 服务器很麻烦。

警告

对于生产和多节点部署，请考虑正确部署高可用的 etcd 服务器，因为这是分布式作业的单点故障。

class torch.distributed.elastic.rendezvous.etcd_server.EtcdServer(data_dir=None)

注意

在 etcd 服务器 v3.4.3 上测试。

在随机空闲端口上启动和停止本地独立 etcd 服务器。适用于单节点、多 worker 启动或测试，在这些情况下，sidecar etcd 服务器比单独设置 etcd 服务器更方便。

此类注册了一个终止处理程序，以便在退出时关闭 etcd 子进程。此终止处理程序不能替代调用 stop() 方法。

以下回退机制用于查找 etcd 二进制文件

1. 使用环境变量 TORCHELASTIC_ETCD_BINARY_PATH

2. 如果存在，则使用 <this file root>/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 服务器二进制文件的路径（有关回退路径，请参见上文）