#51481 [Bugfix][DP] Don't assume the engines started when forwarding a wake

原始 PR 作者 aoshen02 合并时间 2026-08-19 04:49 文件变更 3 提交数 5 评论 7 代码增减 +155 / -2

执行摘要

修复 DP 暂停后 engines_running 锁死，改为观察驱动

关联 issue #51476 明确指出：True 是预测而 False 是观察——coordinator 在转发唤醒时假定引擎会执行，而暂停共识后引擎会丢弃 START_DP_WAVE，于是 wave_complete 永远不会产生，标志锁死。这导致 wait_for_requests_to_drain 只能超时，前端也会因认为引擎在运行而停止发送 FIRST_REQ 通知，状态不可恢复。该问题影响所有 pause 模式（keep/abort/wait），且 sleep 路径同样会触发。

值得精读。该 PR 展示了如何把状态机中的“预测”改为“观察”来修复不可恢复的错误，并且用非空洞的测试设计锁住回归。建议重点关注 run_busy_loop 中的边沿检测位置和三个守卫条件，以及测试中 _poll_flag 的采样策略——它们共同构成了对这类分布式同步问题的可复用范式。

讨论亮点

该 PR 的 review 阶段没有实质技术争论，Claude bot 因 fork 自动评审被禁用，hao-aaron 与 njhill 直接批准。设计权衡体现在提交历史中：

测试断言策略：最初断言标志“永不置位”，但前端转发通知时会乐观置位，导致修复后也无法通过。提交 a7051a3 明确指出必须断言“能置位并自行回落”，否则测试失去区分度。
上报位置：第一版把 start_wave 通知放在 busy loop 顶部，但普通 ADD 请求只在 step 后的 _has_global_unfinished_reqs 才更新 engines_running，导致边沿常被错过、信号被破坏。提交 79fe8ec 将其移到状态真正转换的位置，用 was_running 捕获边沿。

实现拆解

coordinator.py：移除乐观置位
在 vllm/v1/engine/coordinator.py 的 process_input_socket 中，删除转发 START_DP_WAVE 时对 engines_running = True 和 wave_state_changed = True 的赋值。原因是暂停中的引擎会丢弃该广播，发送广播不能作为引擎已运行的证据；engines_running 从此只由引擎侧通知（wave_complete / start_wave）驱动。
core.py：新增 False→True 边沿上报
在 vllm/v1/engine/core.py 的 run_busy_loop 中，先保存本轮开始前的 was_running = self.engines_running，在 _has_global_unfinished_reqs 更新 engines_running 之后，新增 elif 分支：当 not was_running、has_coordinator、dp_rank == 0、not pending_pause 时，向协调器发送 EngineCoreOutputs(start_wave=...)。该上报与现有 wave_complete 对称，复用已有消息字段，不新增消息类型。
测试补充：非空洞的回归测试
在 tests/v1/distributed/test_async_llm_dp.py 新增 2 个测试。test_dp_pause_late_request_does_not_block_drain 先断言运行中的引擎被正确报告为 True，再在 pause(abort) 后验证晚到请求会让标志短暂置 True 但最终自行回落，并确认请求被保留、resume 后能完成；test_dp_sleep_late_request_does_not_block_drain 覆盖 sleep 路径，因为该路径会到达 _drain_requests_for_elastic_ep。测试刻意避免“永不置 True”的空洞断言，必须验证“能置位且能回落”。
配套说明：无配置、schema 或部署变更；控制面改动，不涉及模型输出。

文件	模块	状态	重要度
`vllm/v1/engine/core.py`	引擎核心	modified	6.34
`vllm/v1/engine/coordinator.py`	DP 协调	modified	5.66
`tests/v1/distributed/test_async_llm_dp.py`	DP 测试	modified	7.46

关键符号

run_busy_loop process_input_socket _poll_flag test_dp_pause_late_request_does_not_block_drain test_dp_sleep_late_request_does_not_block_drain

关键源码片段

vllm/v1/engine/core.py core-logic

核心修复点：在 busy loop 中新增 False→True 边沿上报，使引擎真正开始运行时通知协调器，替代原先的预测式置位。

# vllm/v1/engine/core.py — run_busy_loop 中的关键改动
def run_busy_loop(self):
    """Core busy loop of the EngineCore for data parallel case."""
    # 循环直到收到 SIGINT/SIGTERM
    while self._handle_shutdown():
        # 1) 在每轮开头保存上一轮是否在运行，用于检测 False→True 边沿。
        # 必须放在这里：普通 ADD 请求只在 step 之后才会更新
        # self.engines_running，若在循环底部读取会读到旧值，
        # 完全错过边沿，导致协调器永远收不到启动通知。
        was_running = self.engines_running
        self._process_input_queue()
        self._maybe_publish_request_counts()

        # ... 弹性 EP 状态推进、engine step、dummy pass 等逻辑略 ...

        # 3) 全归约确认全局是否有未完成请求，并同步更新本地状态
        self.engines_running = self._has_global_unfinished_reqs(
            local_unfinished_reqs
        )

        if not self.engines_running:
            # False 边沿：wave 完成，向 client 发送 wave_complete
            if self.dp_rank == 0 or not self.has_coordinator:
                client_index = -1 if self.has_coordinator else 0
                self.output_queue.put_nowait(
                    (client_index,
                     EngineCoreOutputs(wave_complete=self.current_wave))
                )
            self.current_wave += 1
            self.step_counter = 0
        elif (not was_running
              and self.has_coordinator
              and self.dp_rank == 0
              and not self.pending_pause):
            # True 边沿：镜像 wave_complete。coordinator 不再在转发
            # START_DP_WAVE 时乐观置位，因此必须由真正完成 step 的
            # rank 0 上报 start_wave，作为“引擎已启动”的唯一证据。
            # 三个守卫保证：只在边沿触发、单一写者、排除暂停协商。
            self.output_queue.put_nowait(
                (-1, EngineCoreOutputs(start_wave=self.current_wave))
            )

vllm/v1/engine/coordinator.py core-logic

移除转发唤醒时的乐观置位，使 engines_running 只由引擎通知驱动，是整个修复的前提。

# vllm/v1/engine/coordinator.py — 转发唤醒时不再预测引擎状态
if self.enable_wave_coordination:
    # 收到前端新请求通知（XPUB socket），代表可能有引擎空闲等待唤醒
    engine_to_exclude, wave = decoded
    if not engines_running:
        if wave < current_wave:
            # 波次号过期时，让所有引擎都处理该消息，避免漏掉某个引擎
            engine_to_exclude = None

        # 此前这里会直接置位 engines_running = True，并假定引擎会执行。
        # 事实上暂停中的引擎会丢弃 START_DP_WAVE，导致引擎永远不会
        # 产生 wave_complete，标志再无机会回到 False。
        # 现在 engines_running 只由引擎侧通知（wave_complete / start_wave）
        # 驱动；这里仅负责把唤醒广播发出去，不再修改本地状态。
        self._send_start_wave(publish_back, current_wave, engine_to_exclude)

评论区精华

测试断言设计：settles back 而非 never sets 测试

最初的测试版本断言标志永不置位，但前端在转发通知时会乐观置位，导致修复后也失败。提交 a7051a3 指出必须断言标志能自行回落到 False，才具有区分度。

结论：改为先确认标志会置位（证明通知已到达）再等待其回落，使测试非空洞。 · 已解决

报告位置：循环顶部 vs step 之后 设计

最初将 start_wave 通知放在 busy loop 顶部，但普通 ADD 请求只在 step 后的 _has_global_unfinished_reqs 才更新 engines_running，导致边沿常被错过，协调器标志在正常运行时保持 False。

结论：改为在 _has_global_unfinished_reqs 之后用 was_running 检测 False→True 边沿，从实际状态转换点上报。 · 已解决

风险与影响

风险主要来自新增的时序窗口与广播开销：

额外的 START_DP_WAVE 广播：coordinator 处理 start_wave 消息时会产生一次回显广播，PR 实测 4 次 wave 从 4 次广播增至 7 次；引擎幂等，不会造成额外 step，但会轻微增加协调器负载。
早返回假阴性：wait_for_requests_to_drain 可能比真实状态早一个往返返回，PR 用“下游操作自带屏障（sleep 内含暂停共识）”来兜底。
单写者约束：dp_rank == 0 是唯一上报方，若该秩异常则协调器无法感知 True 边沿；但 wave_complete 已有同样约束，风险可控。
适用范围有限：enable_wave_coordination 仅对 MoE 模型启用，dense 模型不经过该路径；改动虽在核心循环，但受影响面明确。

影响范围集中在 vLLM v1 的 DP 协调器控制面：修复了暂停/休眠后 engines_running 锁死导致的 wait_for_requests_to_drain 超时、前端停止发送 FIRST_REQ、以及弹性 EP 伸缩 _drain_requests_for_elastic_ep 的错误 go/no-go 判断。对 offline SyncMPClient、SPMD 无 coordinator 场景、dense 模型均无影响。代码改动 3 文件、155 行，但属于核心状态机修复，显著提升 DP + pause/sleep + MoE 场景的稳定性。

核心路径变更时序窗口单写者约束新增广播开销

关联 Issue

#51476 [RFC]: DP pause latches the coordinator's engines_running, and per-mode request admission is unimplemented

完整报告

执行摘要

一句话：修复 DP 暂停后 engines_running 锁死，改为观察驱动
推荐动作：值得精读。该 PR 展示了如何把状态机中的“预测”改为“观察”来修复不可恢复的错误，并且用非空洞的测试设计锁住回归。建议重点关注 run_busy_loop 中的边沿检测位置和三个守卫条件，以及测试中 _poll_flag 的采样策略——它们共同构成了对这类分布式同步问题的可复用范式。

功能与动机

实现拆解

coordinator.py：移除乐观置位
在 vllm/v1/engine/coordinator.py 的 process_input_socket 中，删除转发 START_DP_WAVE 时对 engines_running = True 和 wave_state_changed = True 的赋值。原因是暂停中的引擎会丢弃该广播，发送广播不能作为引擎已运行的证据；engines_running 从此只由引擎侧通知（wave_complete / start_wave）驱动。
core.py：新增 False→True 边沿上报
在 vllm/v1/engine/core.py 的 run_busy_loop 中，先保存本轮开始前的 was_running = self.engines_running，在 _has_global_unfinished_reqs 更新 engines_running 之后，新增 elif 分支：当 not was_running、has_coordinator、dp_rank == 0、not pending_pause 时，向协调器发送 EngineCoreOutputs(start_wave=...)。该上报与现有 wave_complete 对称，复用已有消息字段，不新增消息类型。
测试补充：非空洞的回归测试
在 tests/v1/distributed/test_async_llm_dp.py 新增 2 个测试。test_dp_pause_late_request_does_not_block_drain 先断言运行中的引擎被正确报告为 True，再在 pause(abort) 后验证晚到请求会让标志短暂置 True 但最终自行回落，并确认请求被保留、resume 后能完成；test_dp_sleep_late_request_does_not_block_drain 覆盖 sleep 路径，因为该路径会到达 _drain_requests_for_elastic_ep。测试刻意避免“永不置 True”的空洞断言，必须验证“能置位且能回落”。
配套说明：无配置、schema 或部署变更；控制面改动，不涉及模型输出。

关键文件：

vllm/v1/engine/core.py（模块引擎核心；类别 source；类型 core-logic；符号 run_busy_loop, _has_global_unfinished_reqs）: 核心修复点：在 busy loop 中新增 False→True 边沿上报，使引擎真正开始运行时通知协调器，替代原先的预测式置位。
vllm/v1/engine/coordinator.py（模块 DP 协调；类别 source；类型 core-logic；符号 process_input_socket）: 移除转发唤醒时的乐观置位，使 engines_running 只由引擎通知驱动，是整个修复的前提。
tests/v1/distributed/test_async_llm_dp.py（模块 DP 测试；类别 test；类型 test-coverage；符号 _consume, _poll_flag, test_dp_pause_late_request_does_not_block_drain, test_dp_sleep_late_request_does_not_block_drain）: 新增 137 行回归测试，覆盖 pause 与 sleep 两条触发路径，验证标志能自行回落且请求被保留，防止修复被空洞断言破坏。

关键符号：run_busy_loop, process_input_socket, _poll_flag, test_dp_pause_late_request_does_not_block_drain, test_dp_sleep_late_request_does_not_block_drain

关键源码片段

`vllm/v1/engine/core.py`

核心修复点：在 busy loop 中新增 False→True 边沿上报，使引擎真正开始运行时通知协调器，替代原先的预测式置位。

# vllm/v1/engine/core.py — run_busy_loop 中的关键改动
def run_busy_loop(self):
    """Core busy loop of the EngineCore for data parallel case."""
    # 循环直到收到 SIGINT/SIGTERM
    while self._handle_shutdown():
        # 1) 在每轮开头保存上一轮是否在运行，用于检测 False→True 边沿。
        # 必须放在这里：普通 ADD 请求只在 step 之后才会更新
        # self.engines_running，若在循环底部读取会读到旧值，
        # 完全错过边沿，导致协调器永远收不到启动通知。
        was_running = self.engines_running
        self._process_input_queue()
        self._maybe_publish_request_counts()

        # ... 弹性 EP 状态推进、engine step、dummy pass 等逻辑略 ...

        # 3) 全归约确认全局是否有未完成请求，并同步更新本地状态
        self.engines_running = self._has_global_unfinished_reqs(
            local_unfinished_reqs
        )

        if not self.engines_running:
            # False 边沿：wave 完成，向 client 发送 wave_complete
            if self.dp_rank == 0 or not self.has_coordinator:
                client_index = -1 if self.has_coordinator else 0
                self.output_queue.put_nowait(
                    (client_index,
                     EngineCoreOutputs(wave_complete=self.current_wave))
                )
            self.current_wave += 1
            self.step_counter = 0
        elif (not was_running
              and self.has_coordinator
              and self.dp_rank == 0
              and not self.pending_pause):
            # True 边沿：镜像 wave_complete。coordinator 不再在转发
            # START_DP_WAVE 时乐观置位，因此必须由真正完成 step 的
            # rank 0 上报 start_wave，作为“引擎已启动”的唯一证据。
            # 三个守卫保证：只在边沿触发、单一写者、排除暂停协商。
            self.output_queue.put_nowait(
                (-1, EngineCoreOutputs(start_wave=self.current_wave))
            )

`vllm/v1/engine/coordinator.py`

移除转发唤醒时的乐观置位，使 engines_running 只由引擎通知驱动，是整个修复的前提。

# vllm/v1/engine/coordinator.py — 转发唤醒时不再预测引擎状态
if self.enable_wave_coordination:
    # 收到前端新请求通知（XPUB socket），代表可能有引擎空闲等待唤醒
    engine_to_exclude, wave = decoded
    if not engines_running:
        if wave < current_wave:
            # 波次号过期时，让所有引擎都处理该消息，避免漏掉某个引擎
            engine_to_exclude = None

        # 此前这里会直接置位 engines_running = True，并假定引擎会执行。
        # 事实上暂停中的引擎会丢弃 START_DP_WAVE，导致引擎永远不会
        # 产生 wave_complete，标志再无机会回到 False。
        # 现在 engines_running 只由引擎侧通知（wave_complete / start_wave）
        # 驱动；这里仅负责把唤醒广播发出去，不再修改本地状态。
        self._send_start_wave(publish_back, current_wave, engine_to_exclude)

评论区精华

该 PR 的 review 阶段没有实质技术争论，Claude bot 因 fork 自动评审被禁用，hao-aaron 与 njhill 直接批准。设计权衡体现在提交历史中：

测试断言策略：最初断言标志“永不置位”，但前端转发通知时会乐观置位，导致修复后也无法通过。提交 a7051a3 明确指出必须断言“能置位并自行回落”，否则测试失去区分度。
上报位置：第一版把 start_wave 通知放在 busy loop 顶部，但普通 ADD 请求只在 step 后的 _has_global_unfinished_reqs 才更新 engines_running，导致边沿常被错过、信号被破坏。提交 79fe8ec 将其移到状态真正转换的位置，用 was_running 捕获边沿。
测试断言设计：settles back 而非 never sets (testing): 改为先确认标志会置位（证明通知已到达）再等待其回落，使测试非空洞。
报告位置：循环顶部 vs step 之后 (design): 改为在 _has_global_unfinished_reqs 之后用 was_running 检测 False→True 边沿，从实际状态转换点上报。

风险与影响

风险：风险主要来自新增的时序窗口与广播开销：
额外的 START_DP_WAVE 广播：coordinator 处理 start_wave 消息时会产生一次回显广播，PR 实测 4 次 wave 从 4 次广播增至 7 次；引擎幂等，不会造成额外 step，但会轻微增加协调器负载。
早返回假阴性：wait_for_requests_to_drain 可能比真实状态早一个往返返回，PR 用“下游操作自带屏障（sleep 内含暂停共识）”来兜底。
单写者约束：dp_rank == 0 是唯一上报方，若该秩异常则协调器无法感知 True 边沿；但 wave_complete 已有同样约束，风险可控。
适用范围有限：enable_wave_coordination 仅对 MoE 模型启用，dense 模型不经过该路径；改动虽在核心循环，但受影响面明确。
影响：影响范围集中在 vLLM v1 的 DP 协调器控制面：修复了暂停/休眠后 engines_running 锁死导致的 wait_for_requests_to_drain 超时、前端停止发送 FIRST_REQ、以及弹性 EP 伸缩 _drain_requests_for_elastic_ep 的错误 go/no-go 判断。对 offline SyncMPClient、SPMD 无 coordinator 场景、dense 模型均无影响。代码改动 3 文件、155 行，但属于核心状态机修复，显著提升 DP + pause/sleep + MoE 场景的稳定性。
风险标记：核心路径变更, 时序窗口, 单写者约束, 新增广播开销

关联脉络

PR #39366 Two-phase DP pause (introduced ignore_start_dp_wave): 引入两阶段暂停与 ignore_start_dp_wave，是导致本次锁死问题的直接前提。
PR #38009 Rejected per-engine running state: 被否决的替代方案：在引擎侧维护 per-engine 状态导致跨秩分歧和 all-reduce 死锁；本 PR 避免该问题，将决策保留在单一 coordinator。
PR #32103 Pause mode specification (frontend blocks new requests): 规范了 pause 模式的请求准入语义，本 PR 修复的问题与之相关，且 issue 中明确这是可分离的两件事。

#51481 [Bugfix][DP] Don't assume the engines started when forwarding a wake

执行摘要

修复 DP 暂停后 engines_running 锁死，改为观察驱动

实现拆解

实现拆解

评论区精华

风险与影响

关联 Issue

完整报告

参与讨论