执行摘要

• 一句话：将路由器和会话服务器的启动从驱动子进程迁移至头节点 Ray Actor，统一生命周期管理。
• 推荐动作：此 PR 是 Rollout 模块一项重要的架构级重构，值得精读。核心设计决策是通过 Ray Actor 托管关键长运行进程，以解决子进程生命周期管理问题。_launch_command_on_head 和 _actor_is_alive 的实现模式，以及 wait_tcp_ready 对 Actor 存活状态的适配，都体现了在 Ray 环境下的工程权衡。理解这些变更有助于把握后续关联 PR（如 #2051, #2054）中“建立持久所有权”的上下文。

功能与动机

PR body 中指出，此变更是 Issue #1837 跟踪的一系列重构与增强的一部分。目标是提升系统组件的可观测性和生命周期管理能力。commit message 明确指出：“the sgl/miles router and the session servers stop being subprocesses the driver forks: the rendered command now runs inside a CommandActor pinned to the head node, so their lifetime is bound to an actor the cluster can see instead of a bare child process”。这解决了裸子进程一旦脱离驱动控制便无法被集群有效管理和监控的问题。

实现拆解

1. 引入通用头节点 Actor 创建器：新增 miles/utils/workers/cell_launch.py 文件，提供 create_head_worker_actor 函数。该函数封装了创建一个无 GPU、固定在集群头节点上的 Ray Actor 的逻辑，通过 compute_ray_pin_head_options() 确保调度正确。
2. 重构路由器启动逻辑：修改 miles/ray/rollout/router_manager.py 中的 start_router 函数，将调用 process_utils.launch_bound_subprocess 替换为新的 _launch_command_on_head 辅助函数。该函数调用 create_head_worker_actor 创建一个 CommandActor，并异步执行命令。同时，将 wait_for_server_ready 替换为新的 wait_tcp_ready，后者接受一个 is_alive 回调函数来检查 Actor 的存活状态。
3. 重构会话服务器启动逻辑：同样修改 router_manager.py 中的 start_session_server 函数，将启动多个服务器的逻辑从收集 processes 改为收集 launches（Actor handle），并使用相同的 _launch_command_on_head 和 wait_tcp_ready 模式。
4. 新增 Actor 存活检查函数：在 router_manager.py 中添加 _actor_is_alive 函数，通过尝试远程调用 actor_handle._get_node_ip 并设置超时来判断 Actor 是否存活。
5. 新增 TCP 就绪等待函数：在 miles/utils/http_utils.py 中添加 wait_tcp_ready 函数，它与 wait_for_server_ready 类似，但核心区别在于接受一个 is_alive: Callable[[], bool] 参数来检测服务提供方（此处为 Actor）的死亡。
6. 更新测试：在 tests/fast/ray/rollout/test_router_manager.py 中，新增 TestLaunchCommandOnHead 测试类，专门验证新引入的 _launch_command_on_head 函数行为。同时，更新现有的 TestStartRouterLaunchCommand 和 TestStartSessionServerLaunchCommand 测试，将 mock 目标从 process_utils.launch_bound_subprocess 改为 _launch_command_on_head，并调整断言以匹配新的 Actor 启动模式。

关键文件：

• miles/ray/rollout/router_manager.py（模块 路由器管理；类别 source；类型 entrypoint；符号 _launch_command_on_head, _actor_is_alive, start_router, start_session_server）: 这是本次重构的核心入口文件，包含了 start_router 和 start_session_server 两个主要函数的启动逻辑改造，以及新引入的 _launch_command_on_head 和 _actor_is_alive 辅助函数。
• miles/utils/workers/cell_launch.py（模块 工作器启动；类别 source；类型 core-logic；符号 create_head_worker_actor）: 新增的模块，提供了通用的 create_head_worker_actor 函数，是本次重构中“固定在头节点”这一核心概念的实现基础，被 router_manager.py 直接调用。
• miles/utils/http_utils.py（模块 HTTP工具；类别 source；类型 dependency-wiring；符号 wait_tcp_ready）: 新增了 wait_tcp_ready 函数，这是支持 Actor 存活检查的就绪等待机制的关键组件，修改了原有 TCP 端口等待函数的接口约定。
• tests/fast/ray/rollout/test_router_manager.py（模块 路由器测试；类别 test；类型 test-coverage；符号 TestLaunchCommandOnHead, test_runs_the_joined_command_on_a_head_command_actor, _create, fake_launch）: 测试文件，通过新增的 TestLaunchCommandOnHead 测试类覆盖了核心新函数，并更新了现有测试以匹配启动逻辑的变更。

关键符号：_launch_command_on_head, _actor_is_alive, create_head_worker_actor, wait_tcp_ready

关键源码片段

miles/ray/rollout/router_manager.py

这是本次重构的核心入口文件，包含了 start_router 和 start_session_server 两个主要函数的启动逻辑改造，以及新引入的 _launch_command_on_head 和 _actor_is_alive 辅助函数。

# 重构后的 start_router 函数核心启动与等待逻辑
actor_handle = _launch_command_on_head(launch_argv) # 通过 Actor 启动命令
wait_tcp_ready(
    router_ip,
    router_port,
    # 关键：使用回调函数检查 Actor 的存活状态，替代原来的进程对象检查
    is_alive=functools.partial(_actor_is_alive, actor_handle),
    timeout=_SERVER_READY_TIMEOUT_SECS,
)

# 新增的辅助函数：在头节点创建一个无 GPU 的 CommandActor 来执行命令
def _launch_command_on_head(launch_argv: list[str]) -> ray.actor.ActorHandle:
    actor_handle = create_head_worker_actor(
        worker_cls=CommandActor, env_vars={}, num_cpus=0.2, ctor_kwargs={}
    )
    # 将命令列表合并为一个字符串，并异步发送给 Actor 执行
    actor_handle.run.remote(cmd=shlex.join(launch_argv), envs={})
    return actor_handle
​
# 新增的 Actor 存活检查函数：通过远程 RPC 调用的超时来判断 Actor 是否已死
def _actor_is_alive(actor_handle: ray.actor.ActorHandle) -> bool:
    try:
        # 尝试调用一个轻量级远程方法，如果 Actor 已死，此调用会超时
        ray.get(actor_handle._get_node_ip.remote(), timeout=30)
        return True
    except Exception:
        return False

miles/utils/workers/cell_launch.py

新增的模块，提供了通用的 create_head_worker_actor 函数，是本次重构中“固定在头节点”这一核心概念的实现基础，被 router_manager.py 直接调用。

import ray
from miles.utils.ray_utils import compute_ray_pin_head_options
​
def create_head_worker_actor(
    *,
    worker_cls: type, # Actor 的底层类，如 CommandActor
    env_vars: dict[str, str], # 传递给 Actor 运行时的环境变量
    num_cpus: float, # Actor 所需的 CPU 资源
    ctor_kwargs: dict, # 传递给 worker_cls 构造函数的关键字参数
) -> ray.actor.ActorHandle:
    """创建一个无 GPU 工作器 Actor 并将其固定在集群的头节点上。"""
    # 使用 ray.remote 将类包装为可远程调用的 Actor 类
    actor_cls = ray.remote(worker_cls)
    return actor_cls.options(
        num_cpus=num_cpus,
        num_gpus=0, # 关键：确保不调度到有 GPU 的节点上
        runtime_env={
            "env_vars": env_vars,
        },
        # 通过专用工具函数计算出将 Actor 固定到头节点的 Ray 调度选项
        **compute_ray_pin_head_options(),
    ).remote(**ctor_kwargs)

miles/utils/http_utils.py

新增了 wait_tcp_ready 函数，这是支持 Actor 存活检查的就绪等待机制的关键组件，修改了原有 TCP 端口等待函数的接口约定。

from collections.abc import Callable
import socket
import time
​
def wait_tcp_ready(host: str, port: int, *, is_alive: Callable[[], bool], timeout: float = 30) -> None:
    """轮询直到 TCP 端口接受连接，期间将工作器死亡视为失败。"""
    deadline = time.time() + timeout
    while time.time() < deadline:
        # 关键变化：使用传入的 is_alive 回调来判断服务方（Actor）是否存活
        if not is_alive():
            raise RuntimeError(f"Server process died before port {port} became ready")
        try:
            # 处理 IPv6 地址，移除方括号
            with socket.create_connection((host.strip("[]"), port), timeout=1):
                return
        except OSError:
            time.sleep(0.5)
    raise RuntimeError(f"Server at {host}:{port} not ready after {timeout}s")

评论区精华

本 PR 只有 1 条由 reviewer guapisolo 留下的已合并状态评论，讨论内容为：“Reviewed [PR #1988 changes] in stack context: #1988 drops the router/session actor handles; #2051/#2054 establish durable ownership, with teardown completed by #2059/#2112 before safe-merge checkpoint #2176. The repair remains present through checked tip #2649.” 这条评论没有提出具体的代码问题或争议，而是明确了 PR #1988 在一个更大的重构 PR 链（涉及所有权建立、拆除和安全合并）中的位置和作用。它揭示了 #1988 是“丢弃旧句柄”的步骤，后续 PR 会建立更持久的归属关系（durable ownership）。

• PR 在大型重构链中的定位 (design): 本 PR 是建立持久 Actor 所有权（durable ownership）模型的准备步骤，主要负责“丢弃”旧的启动方式。更完整的生命周期管理（包括所有者建立和优雅终止）在后续 PR 链中完成。

风险与影响

• 风险：
  1. 启动机制变更：核心启动路径从驱动子进程切换为 Ray Actor，引入了对 Ray 调度和 Actor 生命周期管理的依赖。如果 Actor 创建失败或执行超时，错误处理路径与原来不同。
  2. 新增依赖：router_manager.py 现在直接依赖 ray 库和新建的 cell_launch、command_actor 模块，耦合度增加。
  3. 存活检查逻辑：新增的 _actor_is_alive 函数通过一个 30 秒超时的 RPC 调用来判断存活，这可能不是最高效或最可靠的方式，且在网络抖动时可能产生误判。
  4. 端口分配与就绪检查：端口分配逻辑（is_port_available）保持不变，但就绪等待从同步阻塞在子进程上，变为异步等待一个 Actor 可能启动的端口。wait_tcp_ready 中对 host 执行 strip("[]") 以处理 IPv6 地址，这是一个必要的适配点。
• 影响：
  1. 系统影响：提升了路由器和会话服务器在 Ray 集群中的可见性和可控性，为实现自动重启、资源跟踪和优雅关闭提供了基础。这是 Rollout 模块架构向更健壮、更云原生的方向演进的关键一步。
  2. 运维影响：变更后，这些组件的生命周期由 Ray 管理，理论上更容易监控。但调试时，问题排查路径会从查看子进程状态变为查看 Actor 日志和状态。
  3. 团队影响：作为大型重构链的一环，为后续更复杂的变更（如 #2051/#2054 建立持久所有权）铺平了道路。测试的更新确保了新逻辑的正确性，降低了回归风险。
  4. 用户影响：对最终用户透明，功能行为不变，但系统的稳定性和可维护性得到长期提升。
     - 风险标记：核心路径变更, 新增运行时依赖, 生命周期管理逻辑变更

关联脉络

• PR #2051 （推测）建立路由器/会话服务器的持久所有权: PR #1988 的审核评论明确指出 #1988 是“drops the router/session actor handles”，而 #2051 与 #2054 则“establish durable ownership”。两者是同一重构方向（提升 Actor 生命周期管理）的连续步骤。
• PR #2054 （推测）建立路由器/会话服务器的持久所有权: 同上，与 #1988 直接相关，共同完成 Actor 所有权模型的构建。
• PR #2049 Add a worker provider abstraction over RayWorkerManager: 同属 Issue #1837 的重构工作，涉及工作器管理抽象。PR #1988 将启动命令封装为 Actor，而 #2049 则进一步抽象了工作器地址查询等管理逻辑，两者在架构上互补。