Rendezvous

在 Torch 分布式弹性中，我们使用术语“rendezvous”来指代一种特定功能，它将 **分布式同步** 原语与 **对等发现** 相结合。

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

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

屏障:

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

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

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

还存在一个总体超时，如果从未达到 min 个节点，则会导致会合失败 - 这旨在作为一种简单的故障安全机制，以帮助释放部分分配的作业资源，以防资源管理器出现问题，并且被解释为不可重试。

排他性:

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

Torch Distributed Elastic 会合确保，如果一组节点已经完成了会合（因此可能已经开始训练），那么尝试会合的额外“迟到”节点只会宣布自己处于等待状态，并且必须等到（先前完成的）现有会合被销毁后才能继续。

一致性:

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

请注意，等级是不稳定的，这意味着同一个节点可以在下一个（重新）会合中被分配不同的等级。

容错:

Torch Distributed Elastic 会合旨在容忍会合过程中的节点故障。如果一个进程在加入会合并完成会合之间崩溃（或失去网络连接等），那么将自动与剩余的健康节点进行重新会合。

节点也可以在完成会合之后（或被其他节点观察到已经完成会合）失败 - 此场景将由 Torch Distributed Elastic train_loop 处理（它也会触发重新会合）。

共享键值存储:

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

此存储仅由已完成 rendezvous 的成员共享。它旨在由 Torch Distributed Elastic 用于交换初始化作业控制和数据平面所需的信息。

等待的 worker 和 rendezvous 关闭:

Torch Distributed Elastic rendezvous 处理程序对象提供额外的功能，这些功能在技术上不属于 rendezvous 过程

1. 查询有多少 worker 在屏障处到达迟了，哪些 worker 可以参与下一个 rendezvous。

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

DynamicRendezvousHandler:

Torch Distributed Elastic 带有 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 工作原理的状态图。

注册表

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 的端点，通常格式为 <主机名>[:<端口>]。

• 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。

返回类型

任何

get_as_bool(key, default=None)

将 key 的值作为 bool 返回。

返回类型

Optional[bool]

get_as_int(key, default=None)

将 key 的值作为 int 返回。

返回类型

Optional[int]

class torch.distributed.elastic.rendezvous.RendezvousHandlerRegistry

表示 RendezvousHandler 后端的注册表。

create_handler(params)

创建一个新的 RendezvousHandler。

返回类型

RendezvousHandler

register(backend, creator)

注册新的 rendezvous 后端。

参数

• backend (str) – 后端的名称。

• creator (Callable[[RendezvousParameters], RendezvousHandler]) – 用于构造 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，其他节点很快就会观察到这一点并停止运行。

返回类型

布尔值

abstract next_rendezvous()

rendezvous 障碍的主要入口点。

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

返回值

一个包含 torch.distributed.Store、rank 和 world size 的元组。

引发

• RendezvousClosedError – rendezvous 已关闭。

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

• RendezvousStateError – rendezvous 状态已损坏。

• RendezvousTimeoutError – rendezvous 未及时完成。

返回类型

Tuple[Store, int, int]

abstract num_nodes_waiting()

返回在集合点屏障到达过晚的节点数量，因此未包含在当前工作组中。

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

返回类型

int

abstract set_closed()

将集合点标记为已关闭。

abstract shutdown()

关闭为集合点打开的所有资源。

示例

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

返回类型

布尔值

异常

class torch.distributed.elastic.rendezvous.RendezvousError

表示集合点错误的基本类型。

class torch.distributed.elastic.rendezvous.RendezvousClosedError

当集合点关闭时引发。

class torch.distributed.elastic.rendezvous.RendezvousTimeoutError

当 rendezvous 未能在规定时间内完成时引发。

class torch.distributed.elastic.rendezvous.RendezvousConnectionError

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

class torch.distributed.elastic.rendezvous.RendezvousStateError

当 rendezvous 的状态损坏时引发。

class torch.distributed.elastic.rendezvous.RendezvousGracefulExitError

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

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

实现

动态 Rendezvous

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

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

参数

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

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

返回类型

DynamicRendezvousHandler

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

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

表示一个处理程序，它在节点集之间建立集合。

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

创建一个新的 DynamicRendezvousHandler.

参数

• run_id (str) – 集合的运行 ID。

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

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

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

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

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

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

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

表示保存 rendezvous 状态的后端。

abstract get_state()

获取 rendezvous 状态。

返回值

一个包含编码后的 rendezvous 状态和其围栏令牌的元组，如果后端中没有找到状态，则为 None。

引发

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

• RendezvousStateError – rendezvous 状态已损坏。

返回类型

Optional[Tuple[bytes, Any]]

abstract property name: str

获取后端名称。

abstract set_state(state, token=None)

设置 rendezvous 状态。

新的 rendezvous 状态将有条件地设置。

• 如果指定的 token 与后端存储的围栏令牌匹配，则状态将更新。新状态将与它的围栏令牌一起返回给调用者。

• 如果指定的 token 与后端存储的围栏令牌不匹配，则状态不会更新；相反，现有状态及其围栏令牌将返回给调用者。

• 如果指定的 token 为 None，则只有在后端不存在现有状态时才会设置新状态。新状态或现有状态及其围栏令牌将返回给调用者。

参数

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

• token (Optional[Any]) – 一个可选的围栏令牌，它是由先前对 get_state() 或 set_state() 的调用检索到的。

返回值

序列化 rendezvous 状态、其围栏令牌和一个布尔值元组，指示我们的设置尝试是否成功。

引发

• 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 (可选[timedelta]) – 预计集合完成的时间。

• last_call (可选[timedelta]) – 当集合达到最少所需参与者数量后，额外的等待时间，之后集合将完成。

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

• 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 存储的类型。当前支持的类型是“tcp”和“file”，分别对应于 torch.distributed.TCPStore 和 torch.distributed.FileStore。默认值为“tcp”。
read_timeout	存储操作的读取超时时间（以秒为单位）。默认值为 60 秒。 请注意，这仅适用于 torch.distributed.TCPStore。它与 torch.distributed.FileStore 无关，因为 torch.distributed.FileStore 不接受超时时间作为参数。
is_host	一个布尔值，指示此后端实例是否将托管 C10d 存储。如果未指定，它将通过将此机器的主机名或 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 存储进行通信的 torch.distributed.Store 实例。

• run_id (str) – 集合的运行 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。

返回类型

元组[EtcdRendezvousBackend, 存储]

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

表示基于 etcd 的 rendezvous 后端。

参数

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

• run_id (str) – 集合的运行 ID。

• key_prefix (可选[字符串]) – 在 etcd 中存储 rendezvous 状态的路径。

• ttl (可选[整数]) – rendezvous 状态的 TTL。如果未指定，则默认为两小时。

get_state()

参见基类。

返回类型

Optional[Tuple[bytes, Any]]

属性 name: 字符串

参见基类。

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)

实现一个由 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://:2379/1234?min_workers=1&max_workers=3

上面的 URL 解释如下

1. 使用注册到 etcd 方案的 rendezvous 处理程序

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

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

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

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

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

Etcd Store

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

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

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

这是 EtcdRendezvous 返回的存储对象。

add(key, num)

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

整数使用以 10 为基数的字符串表示。如果 key 不存在，将假设默认值为 0。

返回值

新的（增量）值

返回类型

int

check(keys)

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

返回类型

布尔值

get(key)

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

如果键不存在，将阻塞等待最多 timeout 持续时间或直到键被发布。

返回值

值 (bytes)

引发

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

返回类型

字节

set(key, value)

将键值对写入 EtcdStore。

键和值都可以是 Python str 或 bytes。

wait(keys, override_timeout=None)

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

引发

LookupError - 如果超时发生 –

Etcd 服务器

The EtcdServer 是一个方便的类，它使您能够轻松地在子进程上启动和停止 etcd 服务器。这对于测试或单节点（多工作器）部署很有用，在这些部署中，在侧面手动设置 etcd 服务器很麻烦。

警告

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

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

注意

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

在随机空闲端口上启动和停止本地独立 etcd 服务器。对于单节点、多工作器启动或测试很有用，在这些情况下，与单独设置 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 服务器二进制文件的路径（有关回退路径，请参见上文）