# PR #50234 完整报告

- 仓库：`vllm-project/vllm`
- 标题：[PD][PushConnector] Record last activity of remotes to allow clean up of stale ones
- 合并时间：2026-08-07 17:43
- 原文链接：http://prhub.com.cn/vllm-project/vllm/pull/50234

---

# 执行摘要

- 一句话：PushConnector 记录远端活跃时间以清理过期引擎
- 推荐动作：值得精读。该 PR 虽小，但体现了分布式连接管理中一个典型陷阱：继承的清理逻辑需要正确接入活跃信号源才能生效。阅读重点：`_do_start_push_kv` 中刷新位置的选择及其注释，以及两个针对性测试如何用桩构造方式隔离真实 NIXL。对于想理解 vLLM KV connector 可靠性和 NIXL push 模式的工程师，这是很好的入门案例。

# 功能与动机

PR body 明确说明：PushConnector 继承了 base_worker 中清理 stale remote 数据结构、防止 CPU 泄漏的逻辑（源自 PR #44424），但自己没有记录每个远端引擎的最后活动时间，因此无法触发该清理流程。issue #48633 的 S1 条目指出 D 引擎在 scale up/down 后，P 侧 _remote_agents 和 NIXL agent 可能残留，导致连接泄漏。本 PR 通过记录 last activity 让 TTL eviction 真正生效。

# 实现拆解

实现分三步：

1. **在 push 热路径刷新活跃时间 **（vllm/distributed/kv_transfer/kv_connector/v1/nixl/push_worker.py）：在 `_do_start_push_kv` 中，当 `_ensure_handshake` 返回 `None`（即引擎已连接、WRITE 直接执行）时，新增 `self._engine_last_active[decode_engine_id] = time.perf_counter()`。关键点是：`_ensure_handshake` 只在握手建立时刷新活跃时间，对已连接引擎的持续 push 不会刷新，若不在此处记录，正在活跃传输的引擎也会被 TTL 误判为过期而回收。该行放在 WRITE 提交之前，与 pull 模式 `start_load_kv` 的语义对齐。
2. **修正 base_worker 注释 **（vllm/distributed/kv_transfer/kv_connector/v1/nixl/base_worker.py）：`_cleanup_remote_engine` 中原注释声称 push P 侧引擎不参与 eviction，现改为「刚完成的握手可能尚未记录活动时间，因此容忍缺失条目」，仅注释变更，逻辑不变，因为 `pop(engine_id, None)` 本就容忍缺失。
3. **补充单元测试 **（tests/v1/kv_connector/unit/test_nixl_push_connector.py）：新增 `_eviction_worker` 辅助构造器，创建具备 `_ensure_handshake`/`_cleanup_remote_engine` 所需全部字段的 worker 桩；新增 `test_active_push_refreshes_engine_last_active` 验证活跃 push 会刷新 last_active，避免被误回收；新增 `test_stale_engine_evicted_on_push` 验证超过 TTL 的 D 引擎在 push 触发握手时被删除，并调用 `remove_remote_agent`。测试均通过 stub `_xfer_blocks_for_req` 记录 WRITE 调用，不依赖真实 NIXL。

关键文件：
- `vllm/distributed/kv_transfer/kv_connector/v1/nixl/push_worker.py`（模块 KV 连接器；类别 source；类型 core-logic；符号 _do_start_push_kv）: 核心逻辑变更：在 push 直写路径刷新远端引擎最后活跃时间，使 TTL eviction 对 push 模式生效。
- `vllm/distributed/kv_transfer/kv_connector/v1/nixl/base_worker.py`（模块 KV 连接器；类别 source；类型 core-logic；符号 _cleanup_remote_engine）: 注释修正：说明 push 模式下 last_active 现在也会被记录，pop 容忍缺失的语义不变。
- `tests/v1/kv_connector/unit/test_nixl_push_connector.py`（模块 测试；类别 test；类型 test-coverage；符号 _eviction_worker, test_active_push_refreshes_engine_last_active, test_stale_engine_evicted_on_push）: 新增 eviction 场景的测试桩与两个关键测试，验证活跃刷新与过期回收。

关键符号：_do_start_push_kv, _cleanup_remote_engine, _eviction_worker, test_active_push_refreshes_engine_last_active, test_stale_engine_evicted_on_push

## 关键源码片段

### `vllm/distributed/kv_transfer/kv_connector/v1/nixl/push_worker.py`

核心逻辑变更：在 push 直写路径刷新远端引擎最后活跃时间，使 TTL eviction 对 push 模式生效。

```python
# vllm/distributed/kv_transfer/kv_connector/v1/nixl/push_worker.py
# 摘取自 _do_start_push_kv 的关键路径：处理已连接引擎的直写分支

# 尝试建立 / 复用与 decode 引擎的握手；返回 Future 表明握手尚未完成
fut = self._ensure_handshake(
    decode_engine_id,
    registration_data["decode_host"],
    registration_data["decode_port"],
    registration_data["decode_tp_size"],
)
if fut is not None:
    # 后台 executor 正在握手，完成后的 WRITE 会通过 _deferred_push_inbox 重放；
    # 这里不能立即刷新 last_active，因为引擎此时尚未确认可用
    fut.add_done_callback(_on_handshake)
    return

# 引擎已连接：WRITE 直接执行。
# 关键点：_ensure_handshake 仅在首次握手时刷新 last_active，
# 持续 push 不会刷新；若不在此处更新时间戳，活跃引擎也会被
# _evict_stale_engines 按 TTL 误回收（pull 模式在 start_load_kv 中同样处理）
self._engine_last_active[decode_engine_id] = time.perf_counter()

# 以下为原有逻辑：构造 ReqMeta 并提交 NIXL WRITE
logical_local = self._as_grouped_block_ids(local_block_ids)
logical_remote = self._as_grouped_block_ids(remote_block_ids)
physical_local = self._logical_to_kernel_block_ids(
    logical_local, self._physical_blocks_per_logical_kv_block
)
push_meta = ReqMeta(
    local_block_ids=logical_local,
    local_physical_block_ids=physical_local,
    tp_size=self.world_size,
    remote=RemoteMeta(
        block_ids=logical_remote,
        host="",
        port=0,
        engine_id=decode_engine_id,
        request_id=decode_request_id,
    ),
)
self._xfer_blocks_for_req(req_id=request_id, meta=push_meta)

```

### `tests/v1/kv_connector/unit/test_nixl_push_connector.py`

新增 eviction 场景的测试桩与两个关键测试，验证活跃刷新与过期回收。

```python
# tests/v1/kv_connector/unit/test_nixl_push_connector.py
# 新增的 eviction 场景测试：验证 push 路径下的 TTL 行为


def _eviction_worker(engine_ttl: float) -> NixlPushConnectorWorker:
    """构造一个具备 base _ensure_handshake 清理路径所需字段的 push worker。"""
    w = _StubWriterWorker.fresh()
    w._engine_ttl = engine_ttl
    w._engine_last_active = {}
    w._engine_clock_offset = {}
    w._handshake_lock = threading.RLock()
    # _cleanup_remote_engine 回收引擎时会触碰这些结构
    w.nixl_wrapper = MagicMock()
    w.dst_xfer_side_handles = {}
    w.kv_caches_base_addr = {}
    w.dst_num_blocks = {}
    w.tp_mappings = {}
    w.transfer_topo = None
    w._logical_to_kernel_block_ids = lambda blocks, ratio: blocks
    w.writes = []
    w._xfer_blocks_for_req = lambda req_id, meta: w.writes.append(req_id)
    return w


def test_active_push_refreshes_engine_last_active():
    """活跃 push 必须刷新 last_active，否则正在传输的引擎会被误回收。"""
    w = _eviction_worker(engine_ttl=3600.0)
    # 已连接 D 引擎 -> _ensure_handshake 返回 None，WRITE 内联执行
    w._remote_agents["decode-engine"] = {(0, 0): "agent-decode"}
    stale = time.perf_counter() - 5.0
    w._engine_last_active["decode-engine"] = stale

    _real_do_start_push_kv(w, "req", ([1, 2, 3],), _registration_data("req"))

    assert w._engine_last_active["decode-engine"] > stale
    assert w.writes == ["req"]


def test_stale_engine_evicted_on_push():
    """push 触发握手清理：静默超过 TTL 的 D 引擎被回收。"""
    w = _eviction_worker(engine_ttl=30.0)
    w._remote_agents["D-old"] = {(0, 0): "agent-D-old"}
    w._engine_last_active["D-old"] = time.perf_counter() - 10_000.0
    # 被 push 的目标引擎本身已连接
    w._remote_agents["decode-engine"] = {(0, 0): "agent-decode"}

    _real_do_start_push_kv(w, "req", ([1, 2, 3],), _registration_data("req"))

    assert "D-old" not in w._remote_agents
    assert "D-old" not in w._engine_last_active
    w.nixl_wrapper.remove_remote_agent.assert_called_once_with("agent-D-old")

```

# 评论区精华

该 PR 的 review 过程非常简单：维护者 chaunceyjiang 直接批准（LGTM），没有 review comments。claude[bot] 自动评论说明 fork PR 默认关闭自动 review，可手动触发。因此没有实质性的技术争论或设计权衡讨论；核心设计决策（在 `_do_start_push_kv` 而非 `_ensure_handshake` 中刷新）由 PR 作者在代码注释和测试描述中自行说明。

- 暂无高价值评论线程

# 风险与影响

- 风险：风险点包括：
 1. **活跃时间刷新位置依赖**：刷新逻辑仅放在 `_ensure_handshake` 返回 `None` 的分支（已连接直写路径）。若未来新增其它 push 入口（如 deferred push 重放、不同调用方），可能遗漏刷新导致误回收。当前测试已覆盖直写路径，但未覆盖握手回调 `_on_handshake` 后的 deferred push 路径——该路径通过 `_deferred_push_inbox` 最终也会进入 `_do_start_push_kv`，因此间接安全。
 2. **并发安全**：`_engine_last_active` 由主线程（`_do_start_push_kv` 常在主线程执行）写入，而 `_evict_stale_engines` 在握手线程或主线程执行时遍历，均为 dict 单操作，CPython 下基本安全，但缺少显式锁，极端情况下多线程同时 push 到同一引擎可能产生非原子读改写。
 3. **TTL 误判**：刷新时间点是 WRITE 提交前而非完成后，若 WRITE 实际耗时超过 TTL 阈值（默认可能很大），理论上仍可能被回收；但 NIXL WRITE 是异步提交，风险很低。
 4. **行为回归**：base_worker 注释变更不影响逻辑；测试为新增，无既有测试改动。整体风险可控。
- 影响：影响范围集中在 NixlPushMode（PD 分离 push 场景）的 P 侧 worker：
 - **对用户**：修复长时间运行下 D 引擎 scale down 后 P 侧 agent 泄漏问题，提升大规模部署稳定性，减少 CPU 内存占用。
 - **对系统**：启用 base_worker 已有的 TTL eviction，使 `_evict_stale_engines` 在 push 模式下真正生效，与 pull 模式的行为对齐。
 - **对团队**：为 issue #48633 的可靠性路线图补齐 S1 关键项，后续可在同一框架上扩展更细粒度的 liveness 控制。改动小（源码 +3 行），风险低。
 - 风险标记：TTL 误判风险 , push 热路径变更 , 并发写无显式锁

# 关联脉络

- PR #51100 [Bugfix] Fix Mamba all-mode CPU offload boundary alignment: 同属 kv-connector v1 可靠性修复，与本次 push 模式 TTL 清理共同提升 KV 传输稳定性。
- PR #49644 [Feat][Core] Add disk offloading support to SimpleCPUOffloadConnector: 同为 KV connector 方向的演进，扩展连接器生命周期管理能力。