C/C++教程
[源码解析] PyTorch 分布式之弹性训练(5)---Rendezvous 引擎
本文主要是介绍[源码解析] PyTorch 分布式之弹性训练(5)---Rendezvous 引擎，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
[源码解析] PyTorch 分布式之弹性训练(5)---Rendezvous 引擎
目录- [源码解析] PyTorch 分布式之弹性训练(5)---Rendezvous 引擎
- 0x00 摘要
- 0x01 前言
- 1.1 总体系统
- 1.2 Rendezvous
- 1.3 解耦
- 0x02 引擎实现
- 2.1 基类
- 2.2 分布式操作引擎
- 2.2.1 定义
- 2.2.2 调用
- 2.2.2.1 _RendezvousKeepAliveOp
- 2.2.2.2 _RendezvousCloseOp
- 2.2.2.3 _RendezvousJoinOp
- 2.2.3 功能
- 2.2.3.1 主体循环
- 2.2.3.2 同步
- 后端
- _sanitize
- 0x03 算子
- 3.1 操作
- 3.2 算子
- 3.2.1 心跳
- 3.2.1.1 检查心跳
- 3.2.1.2 定期调用
- 3.2.1.2 设置心跳
- 3.2.2 关闭
- 3.2.3 结束
- 3.2.4 Join
- 3.2.1 心跳
- 0x04 业务操作
- 4.1 加入参与者
- 4.2 移除参与者
- 4.3 加入等待序列
- 4.4 移除等待序列
- 4.5 设置结束
- 4.6 心跳
- 0xFF 参考
0x00 摘要
在前面的文章之中，我们已经学习了PyTorch 分布式的基本模块，介绍了官方的几个例子，我们接下来会介绍PyTorch的弹性训练，本文是第五篇，看看Rendezvous 的内部引擎，比如如何处理节点加入，节点离开，等待，心跳等等。
弹性训练系列文章如下：
[源码解析] PyTorch 分布式之弹性训练(1) --- 总体思路
[源码解析] PyTorch 分布式之弹性训练(2)---启动&单节点流程
[源码解析] PyTorch 分布式之弹性训练(3)---代理
[源码解析] PyTorch 分布式之弹性训练(4)---Rendezvous 架构和逻辑
0x01 前言
1.1 总体系统
弹性训练可以理解为在 Rendezvous 基础之上的一个运行系统。
-
Agent 偏重具体节点上的逻辑
- Agent 负责具体业务逻辑相关操作，比如启动进程执行用户程序，监控用户程序运行情况，如果有异常就通知 Rendezvous。
- Agent 是一个 worker manager，负责启动/管理 workers 进程，组成一个 worker group，监控 workers 运行状态，捕获失效 workers，如果有故障/新加入worker，则重启 worker group。
- Agent负责维护 WORLD_SIZE 以及 RANK 信息。用户不需要再手动提供，Agent会自动处理这些。
- Agent 是具体节点上的后台进程，是独立个体。Agent自己无法实现整体上的弹性训练，所以需要一个机制来完成 worker 之间的相互发现，变更同步等等（WORLD_SIZE 和 RANK 这些信息其实也需要多个节点同步才能确定），这就是下面的 Rendezvous 概念。
-
Rendezvous 负责
集群逻辑
，保证节点之间对于""有哪些节点参与训练"达成强一致共识。
- 每一个 Agent 内部包括一个 Rendezvous handler，这些 handler 总体上构成了一个 Rendezvous 集群，从而构成了一个 Agent 集群。
- Rendezvous 完成之后，会创建一个共享键值存储（shared key-value store），这个store实现了一个
torch.distributed.StoreAPI。此存储仅由已完成Rendezvous的成员共享，它旨在让Torch Distributed Elastic在初始化作业过程之中交换控制和数据信息。 - Rendezvous 负责在每个agent之上维护当前 group 所有相关信息。每个 agent 之上有一个 rendezvous，它们会互相通信，总体维护一套信息，这些信息存储在上面提到的Store 之中。
- Rendezvous 负责集群逻辑相关，比如新加入节点，移除节点，分配rank等等。
1.2 Rendezvous
目前为止，Rendezvous 信息如下，DynamicRendezvousHandler 属于动态逻辑，其中，_RendezvousStateHolder 是状态等元信息存储（静态结构），大家会发现图中还有一个 _RendezvousOpExecutor 没有介绍，这就是运行时引擎，所以我们本文看看 _RendezvousOpExecutor 如何处理。
+-----------------------------+ +------------------------------------------------+
| LocalElasticAgent | | WorkerSpec |
| | | |
| +------------------------+ | | rdzv_handler = {DynamicRendezvousHandler} -------+
| |WorkerGroup | | | | |
| | spec +--------------> | entry = worker_fn | |
| | workers | | | | |
| | store | | | role = {str} 'trainer' | |
| | group_rank | | | | |
| | group_world_size | | +------------------------------------------------+ |
| | | | |
| +------------------------+ | |
| | |
| rdzv_run_id | |
| store | +-----------------------------------------+ |
| | |DynamicRendezvousHandler | |
+-----------------------------+ | | |
| | |
| _settings: RendezvousSettings | <--+
| |
| _store: Store |
| |
| _state_holder: _RendezvousStateHolder |
| |
| _op_executor: _RendezvousOpExecutor |
| |
+-----------------------------------------+
1.3 解耦
_RendezvousOpExecutor 把功能分割解耦：
- 业务逻辑被抽象成为一系列算子，比如
_RendevzousJoinOp。 - Rendezvous 内部维护了一套由业务函数组成的状态机，比如函数 _add_to_participants 用来添加参与者。
_RendezvousOpExecutor引擎来执行各种算子，依据算子结果，得到一个 Action，再利用 Action 调用业务函数进行操作。
本文主要介绍C10d 后端对应的 Rendezvous 引擎。
0x02 引擎实现
2.1 基类
_RendezvousOpExecutor 是引擎的基类，只是定义了run这个虚函数。
class _RendezvousOpExecutor(ABC):
"""Executes rendezvous operations."""
@abstractmethod
def run(
self, state_handler: Callable[[_RendezvousContext, float], _Action], deadline: float
) -> None:
"""Executes a rendezvous operation.
An operation is run inside a state machine and is expected to transition
the rendezvous from one state to another.
Args:
state_handler:
A callable that is expected to return the next state transition
action based on the current state of the rendezvous.
deadline:
The time, in seconds, at which the operation will be considered
timed-out.
"""
这里用到了 _RendezvousContext，其作用是把 Rendezvous 的各种信息封装了起来，提供给操作引擎。这里就有了 _RendezvousState 和 RendezvousSettings 的使用。
class _RendezvousContext:
"""Holds the context of the rendezvous.
Attributes:
node:
The node descriptor associated with the current rendezvous handler
instance.
state:
The current state of the rendezvous.
settings:
The rendezvous settings.
"""
node: _NodeDesc
state: _RendezvousState
settings: RendezvousSettings
def __init__(
self, node: _NodeDesc, state: _RendezvousState, settings: RendezvousSettings
) -> None:
self.node = node
self.state = state
self.settings = settings
2.2 分布式操作引擎
_DistributedRendezvousOpExecutor 拓展了 _RendezvousOpExecutor，是 ElasticTorch 的实际执行者。类似于 Looper，负责消息分发，调用业务，状态维护。
2.2.1 定义
与其基类相比，_DistributedRendezvousOpExecutor 加入了比如节点信息，状态，配置这样的成员变量。
class _DistributedRendezvousOpExecutor(_RendezvousOpExecutor):
"""Executes rendezvous operations using a shared state.
Args:
node:
The node descriptor associated with the current rendezvous handler
instance.
state_holder:
The ``RendezvousStateHolder`` to use to sync the rendezvous state
with other nodes.
settings:
The rendezvous settings.
"""
_node: _NodeDesc
_state: _RendezvousState
_state_holder: _RendezvousStateHolder
_settings: RendezvousSettings
def __init__(
self,
node: _NodeDesc,
state_holder: _RendezvousStateHolder,
settings: RendezvousSettings,
) -> None:
self._node = node
self._state_holder = state_holder
self._settings = settings
逻辑如下：
+---------------------------------------------------------------+ | _DistributedRendezvousOpExecutor | | | | +------------------------+ | | _state +---> | _RendezvousState | | | | | | | | participants | | | | wait_list | | | | last_heartbeats | | | | deadline | | | +------------------------+ | | | | +-------------------------+ | | _settings +--> | RendezvousSettings | | | | | | | +-------------------------+ | | | | +--------------------------------------+ | | _state_holder +---> | _BackendRendezvousStateHolder | | | | | | | | _backend: RendezvousBackend | | | | _state: _RendezvousState | | | | _settings: RendezvousSettings | | | | | | | +--------------------------------------+ | | +--------------------------------------+ | | | _NodeDesc | | | _node +-------> | fqdn: str | | | | pid: int | | | | local_id: int | | | | | | | +--------------------------------------+ | +---------------------------------------------------------------+
2.2.2 调用
我们举出几个例子来看看如何调用引擎，可以看到都是先设置算子，然后调用引擎的run函数。
2.2.2.1 _RendezvousKeepAliveOp
def _keep_alive(self) -> None:
self._heartbeat_lock.acquire()
op = _RendezvousKeepAliveOp() # 设置算子
deadline = self._get_deadline(self._settings.timeout.heartbeat)
self._op_executor.run(op, deadline) # 调用
2.2.2.2 _RendezvousCloseOp
def _close(self) -> None:
op = _RendezvousCloseOp() # 设置算子
deadline = self._get_deadline(self._settings.timeout.close)
self._op_executor.run(op, deadline) # 调用
2.2.2.3 _RendezvousJoinOp
def next_rendezvous(self) -> Tuple[Store, int, int]:
"""See base class."""
self._stop_heartbeats()
# Delay the execution for a small random amount of time if this is our
# first run. This will slightly skew the rendezvous attempts across the
# nodes and reduce the load on the backend.
if self._state_holder.state.round == 0:
_delay(seconds=(0, 0.3))
exit_op = _RendezvousExitOp() # 设置算子
join_op = _RendezvousJoinOp() # 设置算子
deadline = self._get_deadline(self._settings.timeout.join)
self._op_executor.run(exit_op, deadline) # 这里会进行调用
self._op_executor.run(join_op, deadline) # 调用
self._start_heartbeats()
rank, world_size = self._get_world()
store = self._get_store()
return store, rank, world_size
2.2.3 功能
_DistributedRendezvousOpExecutor 之中，run 函数实现了基础逻辑，就是依据 action 类型进行各种操作。
2.2.3.1 主体循环
run 具体代码如下：
def run(
self, state_handler: Callable[[_RendezvousContext, float], _Action], deadline: float
) -> None:
"""See base class."""
action = None
while action != _Action.FINISH: # 循环，一直到获得一个FINISH action 为止
# Reads or writes the latest rendezvous state shared by all nodes in
# the rendezvous. Note that our local changes might get overridden
# by another node if that node synced its changes before us.
# 这里很重要，在所有node之间做信息同步
has_set = self._state_holder.sync() # 因为最新状态在 rendezvous。
self._state = self._state_holder.state
ctx = _RendezvousContext(self._node, self._state, self._settings)
# Determine the next action to take based on the current state of
# the rendezvous.
action = state_handler(ctx, deadline) # 决定下一个操作，state_handler 就是算子
if action == _Action.FINISH:
continue
if action == _Action.ERROR_CLOSED:
raise RendezvousClosedError()
if action == _Action.ERROR_TIMEOUT:
raise RendezvousTimeoutError()
if action == _Action.SYNC:
# Delay the execution by one second to avoid overloading the
# backend if we are asked to poll for state changes.
_delay(seconds=1)
else:
if action == _Action.KEEP_ALIVE:
self._keep_alive()
elif action == _Action.ADD_TO_PARTICIPANTS:
self._add_to_participants()
elif action == _Action.ADD_TO_WAIT_LIST:
self._add_to_wait_list()
elif action == _Action.REMOVE_FROM_PARTICIPANTS:
self._remove_from_participants()
elif action == _Action.REMOVE_FROM_WAIT_LIST:
self._remove_from_wait_list()
elif action == _Action.MARK_RENDEZVOUS_COMPLETE:
self._mark_rendezvous_complete()
elif action == _Action.MARK_RENDEZVOUS_CLOSED:
self._mark_rendezvous_closed()
# Attempt to sync our changes back to other nodes.
self._state_holder.mark_dirty()
具体如下图。
+-----------------------------------------+ +---------------------------------------------------------------+
|DynamicRendezvousHandler | | _DistributedRendezvousOpExecutor |
| | | |
| | | +------------------------+ |
| _settings: RendezvousSettings | | _state +---> | _RendezvousState | |
| | | | | |
| | | | participants | |
| _store: Store | | | wait_list | |
| | | | last_heartbeats | |
| | | | deadline | |
| _state_holder: _RendezvousStateHolder | | +------------------------+ |
| | run(_RendezvousJoinOp()) | +-------------------------+ |
| | | _settings +--> | RendezvousSettings | |
| _op_executor +------------------------------------------------> | | | |
| | | +-------------------------+ |
| | | +--------------------------------------+ |
+-----------------------------------------+ | _state_holder +---> | _BackendRendezvousStateHolder | |
| | | |
| | _backend: RendezvousBackend | |
| | _state: _RendezvousState | |
| | _settings: RendezvousSettings | |
| | | |
| +--------------------------------------+ |
| +--------------------------------------+ |
| | _NodeDesc | |
| _node +-------> | fqdn: str | |
| | pid: int | |
| | local_id: int | |
| | | |
| +--------------------------------------+ |
+---------------------------------------------------------------+
手机如下：
2.2.3.2 同步
在 run 函数之中，需要注意的是：在执行各种算子操作之前，会调用 self._state_holder.sync() 在各个 worker 之间进行一个状态同步，达成共识 （consensus）。
def sync(self) -> Optional[bool]:
"""See base class."""
state_bits: Optional[bytes] = None
token = None
has_set: Optional[bool]
if self._dirty: # 如果本node状态变化了
has_set = False
state_bits = pickle.dumps(self._state)
# 把自己的状态设置到backend之中
set_response = self._backend.set_state(state_bits, self._token)
if set_response is not None:
state_bits, token, has_set = set_response
else: # 自己没变化，只能从后端获取
has_set = None
if self._cache_duration > 0:
# Avoid overloading the backend if we are asked to retrieve the
# state repeatedly. Try to serve the cached state.
if self._last_sync_time >= max(time.monotonic() - self._cache_duration, 0):
return None
get_response = self._backend.get_state() # 从backend获取其他节点最新状态
if get_response is not None:
state_bits, token = get_response
if state_bits is not None:
try:
self._state = pickle.loads(state_bits) # 用后端状态更新本身的状态
except pickle.PickleError as exc:
raise RendezvousStateError(
"The rendezvous state is corrupt. See inner exception for details."
) from exc
else:
self._state = _RendezvousState()
if has_set and self._dead_nodes and log.isEnabledFor(logging.DEBUG):
node_list = ", ".join(f"'{dead_node}'" for dead_node in self._dead_nodes)
msg = (
f"As part of the sync operation the node(s) {node_list} have been removed from the "
f"rendezvous '{self._settings.run_id}' since they had no heartbeat."
)
self._record(message=msg)
self._token = token
self._dirty = False
self._last_sync_time = time.monotonic()
self._sanitize()
return has_set
后端
torch/distributed/elastic/rendezvous/c10d_rendezvous_backend.py 之中是对应后端代码。
后端这里使用 store 作为一个集中式存储，是master。每个 node 是 client，会去master更新自己状态，并且获取其他node状态。这样所有node就会互通有无，达成共识。这里也会定期删除不更新元数据的clients。
get_state 就是简单的从 store 提取。
def get_state(self) -> Optional[Tuple[bytes, Token]]:
"""See base class."""
base64_state: bytes = self._call_store("get", self._key)
return self._decode_state(base64_state)
set_state 会做一个compare set，其返回new state和是否更新了state。
def set_state(
self, state: bytes, token: Optional[Token] = None
) -> Optional[Tuple[bytes, Token, bool]]:
"""See base class."""
base64_state_str: str = b64encode(state).decode()
if token:
# Shortcut if we know for sure that the token is not valid.
if not isinstance(token, bytes):
result = self.get_state()
if result is not None:
tmp = *result, False
# Python 3.6 does not support tuple unpacking in return
# statements.
return tmp
return None
token = token.decode()
else:
token = self._NULL_SENTINEL
base64_state: bytes = self._call_store("compare_set", self._key, token, base64_state_str)
state_token_pair = self._decode_state(base64_state)
if state_token_pair is None:
return None
new_state, new_token = state_token_pair
# C10d Store's compare_set method does not offer an easy way to find out
# whether our write attempt was successful. As a brute-force solution we
# perform a bitwise comparison of our local state and the remote state.
return new_state, new_token, new_state == state
_sanitize
_sanitize 方法用来依据其他节点消息做处理，比如清理故障节点。即，如果上一次的心跳时间超过了一定阈值范围，则会把这些节点标记为dead_node，并且从 participant或者wait list中清除这些节点。
def _sanitize(self) -> None:
state = self._state
expire_time = datetime.utcnow() - (
self._settings.keep_alive_interval * self._settings.keep_alive_max_attempt
)
# Filter out the dead nodes.
self._dead_nodes = [
node
for node, last_heartbeat in state.last_heartbeats.items()
if last_heartbeat < expire_time
]
participant_removed = False
for dead_node in self._dead_nodes:
del state.last_heartbeats[dead_node] # 移除故障节点
try:
del state.participants[dead_node] # 移除故障节点
participant_removed = True
except KeyError:
pass
try:
state.wait_list.remove(dead_node) # 移除故障节点
except KeyError:
pass
if participant_removed:
# Common epilogue shared with the _remove_from_participants()
# function of _DistributedRendezvousOpExecutor.
_remove_participant_epilogue(state, self._settings)
介绍完毕如何运行引擎，我们接下来看看具体算子。
0x03 算子
_RendezvousOpExecutor 引擎的业务逻辑被分成两层：用户操作 和 内部业务逻辑。用户操作和内部业务机制之间被解耦。
-
用户操作被分成各种算子，包括：心跳，Join，关闭，结束。比如Join 算子就是
_RendevzousJoinOp。 -
内部业务逻辑被分成各种业务函数，比如 _add_to_participants 方法从等待列表中移除节点，往 participants 加入这个节点。
-
算子和内部业务逻辑并不是一一对应，需要一个类似状态机的机制来控制。
- 比如，心跳操作算子的结果可能是：超时/keep alive/正常结束，所以应该根据这个结果调用不同的内部业务函数。这种对应关系逻辑就是通过 Action 来完成的。
- 各种算子联合起来，聚合成了一个状态机。
- 算子内部就是生成各种 Action，决定了状态机的下一步操作。
-
引擎内部就是根据 Action 来执行具体业务逻辑，或者可以说，是通过 Action 进行解耦。
具体如下，引擎从逻辑上可以分成三层：最上面是算子层，中间是 Action 层，下面是业务函数层。
+-----------------------------------------------------------------------------------------+
| |
| _RendezvousKeepAliveOp _RendezvousCloseOp _RendezvousExitOp _RendezvousJoinOp |
| |
+-------------+---------------------+--------------------+------------------+-------------+
| | | |
| | | |
| | | |
| | | |
v v v v
+-----------------------------------------------------------------------------------------+
| |
| KEEP_ALIVE ADD_TO_PARTICIPANTS ADD_TO_WAIT_LIST REMOVE_FROM_WAIT_LIST ...... |
| |
+-------------+----------+----------+----------+---------+---------+---------+------------+
| | | | | | |
| | | | | | |
| | | | | | |
| | | | | | |
v v v v v v v
+-----------------------------------------------------------------------------------------+
| |
| _add_to_participants _remove_from_participants _add_to_wait_list ...... |
| |
| |
+-----------------------------------------------------------------------------------------+
我们逐一解析。
3.1 操作
先来解析中间层 Action，看看有多少 Action。基于 rendezvous 的状态，引擎的actions具体如下。代码位于 torch/distributed/elastic/rendezvous/dynamic_rendezvous.py
class _Action(Enum):
"""Specifies the possible actions based on the state of the rendezvous."""
KEEP_ALIVE = 1
ADD_TO_PARTICIPANTS = 2
ADD_TO_WAIT_LIST = 3
REMOVE_FROM_PARTICIPANTS = 4
REMOVE_FROM_WAIT_LIST = 5
MARK_RENDEZVOUS_COMPLETE = 6
MARK_RENDEZVOUS_CLOSED = 7
SYNC = 8
ERROR_CLOSED = 9
ERROR_TIMEOUT = 10
FINISH = 11
3.2 算子
引擎之中实现了一些算子，基本上，一个操作对应一个算子，我们给出几个操作算子的例子，算子就是依据rendezvous的状态来设置操作类型。
3.2.1 心跳
3.2.1.1 检查心跳
_RendezvousKeepAliveOp 的作用是：依据当前状态和时间来确定下一步Action。主要是定期检查本Node是否故障。
class _RendezvousKeepAliveOp:
"""Represents a rendezvous keep-alive update operation."""
def __call__(self, ctx: _RendezvousContext, deadline: float) -> _Action:
if _should_keep_alive(ctx):
if time.monotonic() > deadline:
return _Action.ERROR_TIMEOUT
return _Action.KEEP_ALIVE
return _Action.FINISH
_should_keep_alive 方法为：
def _should_keep_alive(ctx: _RendezvousContext) -> bool:
"""Determines whether a keep-alive heartbeat should be sent."""
try:
last_heartbeat = ctx.state.last_heartbeats[ctx.node]
except KeyError:
return False
return last_heartbeat <= datetime.utcnow() - ctx.settings.keep_alive_interval
3.2.1.2 定期调用
这里要注意的是，因为做任何算子之前，都要调用 sync 操作，而 sync 会在 node 之间同步状态，因为心跳是定期的，所以同步状态也是定期的。
DynamicRendezvousHandler 之中会启动一个timer，定期调用_keep_alive_weak方法。
def _start_heartbeats(self) -> None:
self._keep_alive_timer = _PeriodicTimer(
self._settings.keep_alive_interval, self._keep_alive_weak, weakref.ref(self)
)
self._keep_alive_timer.set_name(f"RendezvousKeepAliveTimer_{self._this_node.local_id}")
self._keep_alive_timer.start()
其次，_keep_alive_weak 会调用 self._keep_alive()。
@staticmethod
def _keep_alive_weak(weak_self) -> None:
self = weak_self()
if self is not None:
self._keep_alive()
_keep_alive 会调用 _RendezvousKeepAliveOp。
def _keep_alive(self) -> None:
self._heartbeat_lock.acquire()
op = _RendezvousKeepAliveOp()
deadline = self._get_deadline(self._settings.timeout.heartbeat)
try:
self._op_executor.run(op, deadline)
msg = (
f"The node '{self._this_node}' has sent a keep-alive heartbeat to the rendezvous "
f"'{self._settings.run_id}'."
)
self._record(message=msg)
log.debug(msg)
except RendezvousError as ex:
msg = (
f"The node '{self._this_node}' has failed to send a keep-alive heartbeat to the "
f"rendezvous '{self._settings.run_id}' due to an error of type {type(ex).__name__}."
)
self._record(message=msg, node_state=NodeState.FAILED)
finally:
self._heartbeat_lock.release()
3.2.1.2 设置心跳
另外，_DistributedRendezvousOpExecutor 有一个 _keep_alive 同名函数，是用来实现内部逻辑，我们后续会讲到。
3.2.2 关闭
_RendezvousCloseOp 会依据当前状态和时间来确定下一步Action。
class _RendezvousCloseOp:
"""Represents a rendezvous close operation."""
def __call__(self, ctx: _RendezvousContext, deadline: float) -> _Action:
if ctx.state.closed:
return _Action.FINISH
if time.monotonic() > deadline:
return _Action.ERROR_TIMEOUT
return _Action.MARK_RENDEZVOUS_CLOSED
3.2.3 结束
_RendezvousExitOp 依据当前状态和时间来确定下一步Action。如果本Node不在participants之中，不处理。否则返回一个从 participants 列表删除的下一步Action。如果超时则返回对应Action。
class _RendezvousExitOp:
"""Represents a rendezvous exit operation."""
def __call__(self, ctx: _RendezvousContext, deadline: float) -> _Action:
if ctx.node in ctx.state.participants:
if time.monotonic() > deadline:
return _Action.ERROR_TIMEOUT
return _Action.REMOVE_FROM_PARTICIPANTS
return _Action.FINISH
3.2.4 Join
_RendezvousJoinOp 这里依据系统状态不同，做不同处理，比如试图把本Node加入到participant，或者 waiting list，或者继续等待，具体可以参见代码注释。
- 从上下文之中提取 _RendezvousState 状态，把结果存放在 state 之中。
- 如果状态是closed，则说明此时rendezvous已经结束，则返回_Action.ERROR_CLOSED。
- 看看是不是参与者，把结果存放在is_participant。
- 如果状态已经结束，且本节点已经是参与者，则说明 rendezvous 可以结束，返回 _Action.FINISH。
- 获取当前时间 now。
- 如果 now > deadline，说明已经超时。
- 如果还有时间做 rollback，说明本节点要返回之前的状态。
- 如果本节点已经是参与者，说明此时总节点数目没有达到 min，虽然已经是参与者，但是需要从参与者列表移除，所以返回 _Action.REMOVE_FROM_PARTICIPANTS。
- 如果本节点在等待列表之中，说明此时总节点数目没有达到 max，虽然在等待列表之中，但是需要从等待列表移除，所以返回_Action.REMOVE_FROM_WAIT_LIST。
- 否则返回_Action.ERROR_TIMEOUT。
- 如果还有时间做 rollback，说明本节点要返回之前的状态。
- 否则没有超时，继续处理。
- 如果state.complete 并且本节点不是参与者（如果节点是参与者，前面已经处理过了），说明rendezvous 已经结束，如果还没有达到最大节点数目，并且当前node不在等待列表之中，就需要添加到等待节点列表，等待下次监控周期到的时候，重新做rendezvous，就可以把等待列表中的节点加入到参与列表之中。所以返回_Action.ADD_TO_WAIT_LIST。
- 如果本节点是参与者并且state不是complete状态（如果是complete状态，前面已经处理过了），如果已经达到了最小节点数 & 已经超时了，则说明rendezvous 已经结束，则返回_Action.MARK_RENDEZVOUS_COMPLETE。
- 否则说明没结束，本节点也不是参与者，则直接加入到参与者列表，返回_Action.ADD_TO_PARTICIPANTS。
- 如果需要保持心跳，就返回 _Action.KEEP_ALIVE。
- 否则返回_Action.SYNC。
class _RendezvousJoinOp:
"""Represents a rendezvous join operation."""
def __call__(self, ctx: _RendezvousContext, deadline: float) -> _Action:
state = ctx.state # 从上下文之中提取 _RendezvousState 状态
# A closed rendezvous means that it no longer accepts new nodes.
if state.closed:
return _Action.ERROR_CLOSED # 如果已经结束，就返回 _Action.ERROR_CLOSED
is_participant = ctx.node in state.participants # 看看是不是参与者
# If we are part of the rendezvous and it is already complete there is
# no further action to take.
if state.complete and is_participant: # 如果是参与者且状态是结束，就返回 _Action.FINISH
return _Action.FINISH
now = time.monotonic()
if now > deadline: # 如果已经超时
rollback_period = 5 # 5 seconds
# If we still have time to rollback (a short period on top of the
# operation deadline), try to remove ourself from the rendezvous.
# It is okay if we can't though as our keep-alive will eventually
# expire.
if now <= deadline + rollback_period: # 如果还有时间来 rollback
# If we are part of the rendezvous, it means we couldn't find
# enough participants to complete it on time.
if is_participant: # 此时尚未达到min，虽然已经是参与者，但是需要移除
return _Action.REMOVE_FROM_PARTICIPANTS # 需要从参与者列表移除
# If we are in the wait list, it means we couldn't wait till the
# next round of the rendezvous.
if ctx.node in state.wait_list: # 此时已经达到 max，虽然已经在等待列表之中，需要移除
return _Action.REMOVE_FROM_WAIT_LIST # 需要从等待列表移除
return _Action.ERROR_TIMEOUT # 返回超时
if state.complete: # 如果 rendezvous 已经结束
# If we are here, it means we are not part of the rendezvous. In
# case the rendezvous has capacity for additional participants add
# ourself to the wait list for the next round.
if len(state.participants) < ctx.settings.max_nodes: # 如果还没有达到最大节点数
if ctx.node not in state.wait_list: # 如果当前node不在等待列表之中
return _Action.ADD_TO_WAIT_LIST # 就加入到等待列表，发送一个等待action
elif is_participant: # 如果已经在参与者列表
# If the rendezvous has enough number of participants including us,
# check whether we have passed the rendezvous deadline. If yes,
# complete it.
if len(state.participants) >= ctx.settings.min_nodes: # 如果达到了最小节点数
if cast(datetime, state.deadline) < datetime.utcnow(): # 如果达到了超时
return _Action.MARK_RENDEZVOUS_COMPLETE # 标示 rendezvous 已经结束
else: # 否则就直接加入到参与者
# The rendezvous is not complete yet and we are not part of it. Try
# to join.
return _Action.ADD_TO_PARTICIPANTS
if _should_keep_alive(ctx): # 如果需要保持心跳，就返回 _Action.KEEP_ALIVE
return _Action.KEEP_ALIVE
# At this point either the rendezvous is not complete, but we are part
# of it, which means we have to wait for other participants to join; or
# the rendezvous is complete, but we are not part of it, which means we
# have to wait for the next round.
return _Action.SYNC # 否则返回同步状态 _Action.SYNC
具体逻辑如下：
state.closed
+--------------------------> _Action.ERROR_CLOSED
|
|
| complete & participant
+--------------------------> _Action.FINISH
|
|
| timeout & participant
+--------------------------> _Action.REMOVE_FROM_PARTICIPANTS
|
|
| timeout & wait
+--------------------------> _Action.REMOVE_FROM_WAIT_LIST
|
+-------------------+ |
| | | timeout
| _RendezvousJoinOp +------------------------------> _Action.ERROR_TIMEOUT
| | |
+-------------------+ | complete & < max & not wait
|
+--------------------------> _Action.ADD_TO_WAIT_LIST
|
| complete & participant & > min & deadline
|
+--------------------------> _Action.MARK_RENDEZVOUS_COMPLETE
|
| not complete & not participant
|
+--------------------------> _Action.ADD_TO_PARTICIPANTS
|
| _should_keep_alive
|
+--------------------------> _Action.KEEP_ALIVE
|
| else
|
+--------------------------> _Action.SYNC
以下是源码之中 ETCD 后端 Rendezvous 状态描述图，我们可以大致参考比对 c10d的状态。
可见，etcd 后端的Join可以分为4个阶段：
- setup 阶段，会往固定目录写一个值，这是一个排他锁，如果写失败，说明目前正有一个
rendezvous过程在进行中。 - join（joinable） 阶段。如果写值成功，则进入join 阶段。如果在等待时间结束或者参与训练的节点达到了最大值，则进入 frozen 阶段。
- frozen（confirm）阶段。需要所有节点都确认，进入最后的 final 阶段。
- final 阶段。分配rank，
RANK 0的实例成为 master。
仿照上图，我们把 c10d 拓展如下。
+
|
|
v
+-----+------+
| |
| closed +---------------> ERROR_CLOSED
| |
+-----+------+
|
|
v
+-----+------+ is_participant
| |
| complete +---------------> FINISH
| |
+-----+------+
| is_participant
|
v +----> REMOVE_FROM_PARTICIPANTS
+-----+-------+ now > deadline +-----------+ now < rollback +-----------+ |
| | | | | | |
| join +----------------> | timeout +---------------------->+ rollback +-----+
| | | | | | |
+-----+-------+ +----+------+ +-----------+ |
| | | in state.wait_list
| | now > rollback |
| now < deadline | +----> REMOVE_FROM_WAIT_LIST
| +----------> ERROR_TIMEOUT
|
| complete && not is_participant && < max && not in state.wait_list
|
+------------------------------------------------------------------> ADD_TO_WAIT_LIST
|
| not complete && is_participant && > min && > deadline
|
+------------------------------------------------------------------> MARK_RENDEZVOUS_COMPLETE
|
| not complete && not is_participant
|
+-----------------------------------------> ADD_TO_PARTICIPANTS
|
| _should_keep_alive
|
+---------------------------> KEEP_ALIVE
|
|
v
SYNC
手机如下：
0x04 业务操作
_DistributedRendezvousOpExecutor.run 的内部就是依据 action 选择不同的业务函数来执行。
if action == _Action.KEEP_ALIVE:
self._keep_alive()
elif action == _Action.ADD_TO_PARTICIPANTS:
self._add_to_participants()
elif action == _Action.ADD_TO_WAIT_LIST:
self._add_to_wait_list()
elif action == _Action.REMOVE_FROM_PARTICIPANTS:
self._remove_from_participants()
elif action == _Action.REMOVE_FROM_WAIT_LIST:
self._remove_from_wait_list()
elif action == _Action.MARK_RENDEZVOUS_COMPLETE:
self._mark_rendezvous_complete()
elif action == _Action.MARK_RENDEZVOUS_CLOSED:
self._mark_rendezvous_closed()
我们接下来就看看具体这些内部函数逻辑。
4.1 加入参与者
接受到 ADD_TO_PARTICIPANTS 之后，调用 _add_to_participants 从等待列表中移除节点，往 participants 加入这个节点。
def _add_to_participants(self) -> None:
state = self._state
try:
state.wait_list.remove(self._node)
except KeyError:
pass
# The ranks of the participants will be set once the rendezvous is
# complete.
state.participants[self._node] = 0
self._keep_alive()
if len(state.participants) == self._settings.min_nodes:
state.deadline = datetime.utcnow() + self._settings.timeout.last_call
if len(state.participants) == self._settings.max_nodes:
self._mark_rendezvous_complete()
4.2 移除参与者
接受到 REMOVE_FROM_PARTICIPANTS 之后，调用 _remove_from_participants 从 participants 和 last_heartbeats 中删除参与者。
def _remove_from_participants(self) -> None:
state = self._state
del state.participants[self._node]
del state.last_heartbeats[self._node]
if state.complete:
# If we do not have any participants left, move to the next round.
if not state.participants:
state.complete = False
state.round += 1
else:
if len(state.participants) < self._settings.min_nodes:
state.deadline = None
4.3 加入等待序列
接受到 ADD_TO_WAIT_LIST 之后，调用 _add_to_wait_list 网 wait_list 中加入节点。
def _add_to_wait_list(self) -> None:
self._state.wait_list.add(self._node)
self._keep_alive()
4.4 移除等待序列
接受到 REMOVE_FROM_WAIT_LIST 之后，调用 _remove_from_wait_list 从 wait_list 移除节点。
def _remove_from_wait_list(self) -> None:
self._state.wait_list.remove(self._node)
del self._state.last_heartbeats[self._node]
4.5 设置结束
接受到 MARK_RENDEZVOUS_COMPLETE 之后，当 rendezvous 聚合操作结束之后，给每一个参与者设置 rank。
每个节点上都是按照同样算法排序，所以rank在每个节点上都是一样的。
def _mark_rendezvous_complete(self) -> None:
state = self._state
state.complete = True
state.deadline = None
# Assign the ranks.
for rank, node in enumerate(sorted(state.participants)):
state.participants[node] = rank
def _mark_rendezvous_closed(self) -> None:
self._state.closed = True
4.6 心跳
接收到 KEEP_ALIVE action之后，会调用到 _keep_alive 来维持心跳。另外，keep_alive 也会在 _add_to_participants等方法内被调用，会更新本地state之中的last heartbeats，下一次 sync 时候，会把 last_heartbeats 写入键值存储，这样其他Node就可以知道这个节点的状态了。而本地则会在 _sanitize 之中依据 last_heartbeats 做处理，我们之前提到过。
def _keep_alive(self) -> None:
msg = (
f"The node '{self._node}' updated its keep-alive heartbeat time for the rendezvous "
f"'{self._settings.run_id}'. Pending sync."
)
self._record(message=msg)
self._state.last_heartbeats[self._node] = datetime.utcnow()
_record 方法如下：
def _record(self, message: str, node_state: NodeState = NodeState.RUNNING) -> None:
construct_and_record_rdzv_event(
name=f"{self.__class__.__name__}.{get_method_name()}",
run_id=self._settings.run_id,
message=message,
node_state=node_state,
hostname=self._node.fqdn,
pid=self._node.pid,
local_id=self._node.local_id,
)
其就是调用如下代码记录log。
def record_rdzv_event(event: RdzvEvent) -> None:
_get_or_create_logger("dynamic_rendezvous").info(event.serialize())
def construct_and_record_rdzv_event(
run_id: str,
message: str,
node_state: NodeState,
name: str = "",
hostname: str = "",
pid: Optional[int] = None,
master_endpoint: str = "",
local_id: Optional[int] = None,
rank: Optional[int] = None,
) -> None:
# We don't want to perform an extra computation if not needed.
if isinstance(get_logging_handler("dynamic_rendezvous"), logging.NullHandler):
return
# Set up parameters.
if not hostname:
hostname = socket.getfqdn()
if not pid:
pid = os.getpid()
# Determines which file called this function.
callstack = inspect.stack()
filename = "no_file"
if len(callstack) > 1:
stack_depth_1 = callstack[1]
filename = os.path.basename(stack_depth_1.filename)
if not name:
name = stack_depth_1.function
# Delete the callstack variable. If kept, this can mess with python's
# garbage collector as we are holding on to stack frame information in
# the inspect module.
del callstack
# Set up error trace if this is an exception
if node_state == NodeState.FAILED:
error_trace = traceback.format_exc()
else:
error_trace = ""
# Initialize event object
event = RdzvEvent(
name=f"{filename}:{name}",
run_id=run_id,
message=message,
hostname=hostname,
pid=pid,
node_state=node_state,
master_endpoint=master_endpoint,
rank=rank,
local_id=local_id,
error_trace=error_trace,
)
# Finally, record the event.
record_rdzv_event(event)
至此，引擎部分也已经分析完毕，下一篇我们看看是否可以从整体角度再做一下全面梳理。
0xFF 参考
[源码解析] PyTorch 分布式之弹性训练(1) --- 总体思路
[源码解析] PyTorch 分布式之弹性训练(2)---启动&单节点流程
[源码解析] PyTorch 分布式之弹性训练(3)---代理
[源码解析] PyTorch 分布式之弹性训练(4)---Rendezvous 架构和逻辑
这篇关于[源码解析] PyTorch 分布式之弹性训练(5)---Rendezvous 引擎的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
您可能喜欢
-
增量更新怎么做？-icode9专业技术文章分享11-23
-
压缩包加密方案有哪些？-icode9专业技术文章分享11-23
-
用shell怎么写一个开机时自动同步远程仓库的代码？-icode9专业技术文章分享11-23
-
webman可以同步自己的仓库吗？-icode9专业技术文章分享11-23
-
在 Webman 中怎么判断是否有某命令进程正在运行？-icode9专业技术文章分享11-23
-
如何重置new Swiper?-icode9专业技术文章分享11-23
-
oss直传有什么好处？-icode9专业技术文章分享11-23
-
如何将oss直传封装成一个组件在其他页面调用时都可以使用？-icode9专业技术文章分享11-23
-
怎么使用laravel 11在代码里获取路由列表？-icode9专业技术文章分享11-23
-
怎么实现ansible playbook 备份代码中命名包含时间戳功能？-icode9专业技术文章分享11-22
-
ansible 的archive 参数是什么意思？-icode9专业技术文章分享11-22
-
ansible 中怎么只用archive 排除某个目录？-icode9专业技术文章分享11-22
-
exclude_path参数是什么作用？-icode9专业技术文章分享11-22
-
微信开放平台第三方平台什么时候调用数据预拉取和数据周期性更新接口？-icode9专业技术文章分享11-22
-
uniapp 实现聊天消息会话的列表功能怎么实现？-icode9专业技术文章分享11-22
栏目导航
