SGLang 源码完全拆解（十八）：弹性 EP 与故障恢复——扩容加入、rank 故障、专家布点重排与权重备份

承接上一篇：把”静态并行”变成”可变的并行”

SGLang 系列前十七篇已经完整走过一条从请求入口到设备执行的路径：请求在 TokenizerManager 里被分词、经 ZMQ 送进 Scheduler，调度器把 token 预算与 KV 页分配交给 PrefillAdder 与 RadixCache，再由 TpModelWorker 和 ModelRunner 组装成一次 ForwardBatch 交给注意力后端与 MoE 层，最后从采样、语法约束一路回到增量解码器。上一篇（SGLang 源码完全拆解（十七）：DP Attention 与专家并行）把这条路径上的”横向切分”讲透了：注意力用数据并行避开重复计算，MoE 用专家并行把专家摊到几十张卡上，用 DeepEP 做 all-to-all 搬 token，用 EPLB 动态重排专家位置。

但那篇讲的所有并行方案都有一个共同前提：进程组的形状在启动那一刻就固定了。tp_size、ep_size、nnodes 都是启动参数，torch.distributed 的进程组一旦建成，成员集合就是常量。这个前提在真实集群里非常昂贵：

• 训练好的模型权重、KV 池大小、CUDA Graph 的捕获形状，全部按启动时的并行度分配。想把 EP 从 32 扩到 72，唯一办法是停服、改参数、重启、重新加载几百 GB 权重，中间这段时间整个服务不可用。
• 集群里挂掉一张卡或一个节点，哪怕只是网卡抖动，整组 all-to-all 都会卡死或报错，因为集合通信默认要求”全体到场”。几十张卡的专家并行里，一张卡故障就让整台服务下线。
• 更糟的是，故障恢复后专家布点是按旧的世界大小算的，如果世界大小变了，每个 rank 上的专家存储布局也得跟着变——而权重是从磁盘按固定分片加载的，不可能随随便便重摆。

SGLang 的 Elastic EP（弹性专家并行）就是为了拆掉”启动即定型”这个前提而存在的一层机制。它要回答的问题是：在不重启进程组、不重新加载权重的前提下，如何让 EP 世界大小在运行时变大，以及如何在部分 rank 掉线后把服务拉回来。

上图来自 SGLang 官方博客 Large-Scale Expert Parallelism for DeepSeek（2025）。它是弹性 EP 要”改变形状”的那张原本固定的图：注意力部分与 MoE 部分用不同的并行方式切分，接缝处由 all-to-all 通信缝合。本篇讲的扩容与故障恢复，本质上都是在不重建这张图的前提下重排它的成员。

本篇的目标是把这层机制从外到内讲透，覆盖四条主线：

1. 预留：--max-ep-size 如何在启动时就把”未来可能加入的 rank”算进进程组与通信缓冲区的形状里；
2. 扩容：一次 /scale_elastic_ep 请求如何经过一个显式状态机，把新 rank 接进可扩展的 WORLD 组，并在新世界大小下重新同步专家布点与 CUDA Graph；
3. 故障恢复：某些 rank 掉线时，active_ranks 位图如何被用于检测、如何用 Mooncake 的 recover_ranks 把掉线的 rank 重新拉进每一个进程组；
4. 重排与补权重：世界大小变化后 EPLB 如何按”活跃 rank 数”重新算布点，以及 ExpertBackupManager 为什么必须在每台节点上留一份专家权重的 DRAM 副本。

本篇是 SGLang 子系列的第十八篇，也是核心机制的最后一块：运行时拓扑可变。下一篇会回到系列路线图的下一站——llama.cpp 源码完全拆解的总览篇，用同一套”先架构、再模块、后关键代码”的方法建立第三个框架的源码地图。

为什么必须”预留”：进程组形状不可变的根本原因

要理解弹性 EP 的全部设计，先要理解它对抗的是什么约束。

torch.distributed 的进程组（ProcessGroup）在创建时就确定了三样东西：成员 rank 列表、通信后端、通信缓冲区。以 NCCL 为例，ncclCommInitRank 拿到的是一份固定的 rank 数组，通信域内部的通道、树结构、环形拓扑全部按这份数组预计算。后续想加一个新 rank，NCCL 没有”增量加入”的接口——只能把原来的 communicator 销毁重建。

SGLang 的并行状态集中在 python/sglang/srt/distributed/parallel_state.py 的 GroupCoordinator 里。它对每个并行维度（TP、EP、DP、PP……）持有一个 device_group（GPU 侧通信）和一个 cpu_group（CPU 侧 gloo 通信）。弹性 EP 的做法不是去改 NCCL 的语义，而是在创建进程组时就把容量按最大值分配，把”未加入的槽位”标记为不活跃。看这段初始化逻辑：

1
# python/sglang/srt/distributed/parallel_state.py（节选）
2
if max_world_size is not None:
3
    assert max_world_size >= len(ranks), (
4
        f"max_world_size ({max_world_size}) must be >= "
5
        f"len of ranks ({len(ranks)})"
6
    )
7
    pg_active_size = max_world_size
8
pg_active_ranks = torch.zeros(pg_active_size, dtype=torch.int32, device="cuda")
9
pg_active_ranks[: len(ranks)] = 1
10
...
11
dev_opts = MooncakeBackendOptions(pg_active_ranks, recovered_rank)

三个关键点：

• 形状按 max_world_size 分配，而不是按当前的 len(ranks)。也就是说进程组从一开始就”知道”自己未来会有多少个座位。
• active_ranks 是一个 int32 位图，长度为 max_world_size，已加入的 rank 位置填 1，空座位填 0。这个位图不是事后统计出来的，而是进程组创建时构造、并交给通信后端的一份状态。
• MooncakeBackendOptions(pg_active_ranks, recovered_rank) 把这个位图直接传给通信后端。这意味着通信后端在发起集合通信时会知道哪些 rank 当前在场，从而可以对不在场的槽位做跳过或等待处理。

max_ep_size 这个启动参数决定了座位数：

1
max_ep_size: A[
2
    Optional[int],
3
    "Maximum EP size the server can scale to at runtime. Pre-allocates "
4
    "active-rank state and backend buffers to this size. Defaults to the "
5
    "launch-time world size.",
6
] = None

“Defaults to the launch-time world size” 意味着：不显式设置 --max-ep-size 时，弹性 EP 退化为普通 EP——座位数恰好等于启动时的世界大小，没有空位，也就没有扩容空间。这一条设计非常重要，它解释了为什么弹性能力是”选择性付费”的：预留的座位会占用通信缓冲区显存（比如 DeepEP 的 dispatch/combine 缓冲、MessageQueue 的共享内存广播环），不预留就不付这份代价。

max_ep_size 之外还有两个配套参数，专门给”以 joiner 身份启动”的新进程用：

1
ep_join_rank_offset: A[
2
    int,
3
    Arg(
4
        help=(
5
            "Global rank offset of an elastic EP joining group. Scale "
6
            "joiners must set this to the current effective EP size."
7
        ),
8
        cli_name="--elastic-ep-join-rank-offset",
9
    ),
10
] = 0
11
elastic_ep_initial_size: A[
12
    Optional[int],
13
    "EP size used to define the immutable per-rank expert storage layout. "
14
    "Scale joiners must use the primary deployment's launch-time EP size.",
15
] = None

ep_join_rank_offset 是 joiner 的全局 rank 起点。举例：主部署 EP=4，现在要扩到 6，那么新起来的 joiner 进程要把 --elastic-ep-join-rank-offset 设为 4，它的全局 rank 就从 4 开始编号，与主部署已有的 0–3 完全不冲突。注意 --elastic-ep-join-mode 的 help 里有一句硬约束：

1
'scale' joins as a new rank beyond the original group size and requires --node-rank 1.

扩容 joiner 必须是 --node-rank 1，这不是实现细节而是拓扑约定：弹性 EP 目前的实现假设每个 joiner 节点是”追加”的独立节点，不改动主节点内部的 rank 编号。

elastic_ep_initial_size 则锁定了每 rank 专家存储布局。它必须是主部署启动时的 EP 大小，而不是扩容后的目标大小。原因在后面讲权重备份时会看到：每个 rank 的 expert 权重张量形状是按初始 EP 大小切分的，扩容后一个 rank 负责的专家数量变了，但存储布局不能重排，否则加载权重与备份权重的指针映射全部失效。

弹性状态机：一次扩容到底经历了什么阶段

弹性 EP 的状态不是散落在各处的变量，而是一个显式的有限状态机，集中在 python/sglang/srt/elastic_ep/elastic_ep.py 的 ElasticEPState 与 ElasticEPStateManager 里。

状态数据本身是一组很朴素但信息量很大的字段：

1
@dataclass
2
class ElasticEPState:
3
    active_ranks: Optional[torch.Tensor]
4
    last_active_ranks: Optional[torch.Tensor]
5
    active_ranks_cpu: Optional[torch.Tensor]
6
    effective_ep_size: int = 0
7
    pending_ep_size: Optional[int] = None
8
    scale_phase: str = "idle"
9
    last_error: Optional[str] = None
10
    pending_since: Optional[float] = None
11
    original_ep_size: int = 0
12
    has_scaled: bool = False
13
    ep_join_rank_offset: int = 0

逐个解释：

• active_ranks：设备上的位图，长度 max_world_size，谁在场谁为 1。它是”当前世界的真值”。
• last_active_ranks：上一次快照。用于增量检测——比较它和 active_ranks 是否相等，就能判断世界是否刚刚发生了变化。
• active_ranks_cpu：CPU 副本。存在的理由是轮询路径（比如 HTTP 侧的 is_scaling 查询、rank 故障检测）不应该触发 GPU 同步，CPU 副本是”权威读取源”。源码注释直接点明了这一点。

1
@classmethod
2
def is_scaling(cls) -> bool:
3
    """Return whether a scale or recovery operation is pending.
4

5
    The CPU snapshot is authoritative because rank polling uses it too.
6
    """

• effective_ep_size：当前服务使用的 EP 大小。它和 active_ranks 的位图是两个不同视角：位图描述”座位上有谁”，effective_ep_size 描述”这次前向要按多大的世界算”。
• pending_ep_size：正在申请但尚未提交的目标大小。为空表示没有在途的缩放操作。
• scale_phase：状态机的当前阶段。
• original_ep_size：启动时的世界大小。用于判断”当前是否运行在扩展世界里”。
• has_scaled：是否发生过缩放。
• ep_join_rank_offset：joiner 视角的 rank 偏移。

状态机的推进由一组类方法完成，它们构成了一条清晰的线性流程：

1
request_scale(n)   → scale_phase = "waiting_for_cohort"
2
begin_scale()      → scale_phase = "pending"
3
mark_joining()     → scale_phase = "joining"
4
mark_configuring_data_plane() → scale_phase = "configuring_data_plane"
5
mark_syncing_new_world()      → scale_phase = "syncing_new_world"
6
commit_scale()     → scale_phase = "serving_expanded", effective_ep_size = pending
7
fail_scale(err)    → scale_phase = "failed"
8
fail_recovery(err) → scale_phase = "recovery_unsupported"

request_scale 是入口，它做了两件事：拒绝在已有在途请求或已进入 recovery_unsupported 时重复申请，然后记录目标大小与起始时间戳。

1
@classmethod
2
def request_scale(cls, n: int) -> bool:
3
    inst = cls._instance
4
    if inst is None:
5
        return False
6
    if (
7
        inst.pending_ep_size is not None
8
        or inst.scale_phase == "recovery_unsupported"
9
    ):
10
        return False
11
    inst.pending_ep_size = n
12
    inst.scale_phase = "waiting_for_cohort"
13
    inst.last_error = None
14
    inst.pending_since = time.monotonic()
15
    return True

注意 recovery_unsupported 是一个”终态”。一旦集群进入这个状态（通常意味着恢复所需的前提条件不满足，比如备份权重缺失、Mooncake 传输引擎注册失败），后续所有缩放申请都会被直接拒绝。这是一个刻意的”熔断”设计：与其在半坏的状态下继续缩放、把错误扩散到整个集群，不如明确标记能力不可用。

状态机里最有一手设计感的是 get_data_plane_ep_size：

1
@classmethod
2
def get_data_plane_ep_size(cls) -> int:
3
    inst = cls._instance
4
    assert inst is not None, "Elastic EP state is not initialized."
5
    if inst.pending_ep_size is not None and inst.scale_phase in (
6
        "configuring_data_plane",
7
        "syncing_new_world",
8
    ):
9
        return inst.pending_ep_size
10
    return inst.effective_ep_size

它区分了控制面（control plane）与数据面（data plane）两个 EP 大小。在 configuring_data_plane 与 syncing_new_world 这两个阶段，控制面已经把目标大小记为 pending_ep_size，但数据面（通信缓冲区、专家布点）还没有真正按新大小建好。这时候如果谁去问”现在的 EP 大小是多少”，答案必须是目标大小——因为所有正在构造的缓冲区都按目标大小在建；如果返回旧值，缓冲区就会按旧大小建、按新大小用，形状对不上。

这个 get_data_plane_ep_size 还被一个更宏观的判据使用：

1
def elastic_expanded_world_enabled() -> bool:
2
    """Return whether execution uses ranks admitted after server launch.
3

4
    Launch-time TP groups exclude ranks admitted during scale-up.
5
    """
6
    inst = ElasticEPStateManager.instance()
7
    if inst is None:
8
        return False
9
    if get_parallel().max_ep_size is None:
10
        return False
11
    return ElasticEPStateManager.get_data_plane_ep_size() > inst.original_ep_size

函数名直译就是”扩展世界已启用”。它的注释里有一句很容易被忽略但极其重要的约束：启动时建好的 TP 组不包含扩容时加入的 rank。这句话的原因在于扩容 joiner 是以 --node-rank 1 追加进来的独立节点，它参加的是 WORLD 组与 EP 组，而不参加主部署的 launch-time TP 组。因此”扩展世界”下代码必须区别对待两类 rank：一类在原始的 TP 组里，一类只在 WORLD 与 EP 组里。

is_scaling 里还有一个细节值得注意：判断”是否在缩放中”时不能只看 pending_ep_size：

1
if inst.pending_ep_size is not None:
2
    return True
3
active_count = int(inst.active_ranks_cpu[: inst.effective_ep_size].sum().item())
4
return active_count < inst.effective_ep_size

第二段逻辑说的是：即使没有在途的缩放请求，只要当前有效世界里还有 rank 是掉线状态（位图里 1 的个数小于有效大小），就认为系统处于”缩放/恢复中”。也就是说这个状态机同时覆盖了”主动扩容”和”被动恢复”两种触发源——这正是把它叫做 Elastic EP 而不是 Scale EP 的原因。

reset() 也要单独看一眼：

1
def reset(self):
2
    if self.active_ranks is not None:
3
        # Reserved slots stay inactive until their ranks join.
4
        self.active_ranks.zero_()
5
        self.active_ranks[: self.effective_ep_size] = 1
6
        self.snapshot_active_to_last()
7
        self.sync_active_to_cpu()

它先清零再按 effective_ep_size 重填前缀。注释 “Reserved slots stay inactive until their ranks join” 再次强调：预留座位默认是失活的，只有 rank 真正加入才置 1。这是一个安全默认值——如果反过来把预留座位默认置 1，集合通信会一直等一个永不到场的 rank。

扩容入口：一个 HTTP 请求如何走到调度器

弹性扩容对外的接口极简，只有两个 HTTP 端点，定义在 python/sglang/srt/entrypoints/elastic_ep.py 里：

1
@router.post("/scale_elastic_ep")
2
@auth_level(AuthLevel.ADMIN_OPTIONAL)
3
async def scale_elastic_ep(raw_request: Request):
4
    """Request an asynchronous EP scale-up."""
5
    ...
6
    new_ep_size = body.get("new_ep_size")
7
    ...
8
    if get_exec().moe.elastic_ep_backend is None:
9
        return ORJSONResponse(
10
            {"error": "elastic EP is not enabled (set --elastic-ep-backend)"},
11
            status_code=HTTPStatus.NOT_FOUND,
12
        )
13

14
    result = await _global_state.tokenizer_manager.scale_elastic_ep(
15
        ScaleElasticEPReqInput(new_ep_size=new_ep_size)
16
    )

1
@router.get("/is_scaling_elastic_ep")
2
@auth_level(AuthLevel.ADMIN_OPTIONAL)
3
async def is_scaling_elastic_ep(raw_request: Request):
4
    """Return the tokenizer's mirrored Elastic EP scale state."""
5
    ...
6
    return ORJSONResponse(_global_state.tokenizer_manager.get_elastic_ep_state())

两个端点都带 AuthLevel.ADMIN_OPTIONAL：缩放是改变集群拓扑的管理操作，不是普通推理请求，所以鉴权级别与 /generate 不同。端点本身不做任何决策，只做参数校验（new_ep_size 必须是正整数）与”功能是否开启”检查，然后把请求转给 TokenizerManager。

注意这里有一个容易被忽略的设计：HTTP 端点返回的是 result.success、old_ep_size、new_ep_size，但扩容并没有在 HTTP 请求返回时完成。端点注释写得很清楚——“Request an asynchronous EP scale-up”。这一点在整个机制里非常关键：新 rank 需要从别的节点启动进程、加入 WORLD 组、重新同步专家布点、重捕 CUDA Graph，这可能需要几十秒甚至几分钟。如果 HTTP 请求同步等待，客户端会超时；更糟的是，如果客户端在等待期间重试，就会触发重复的缩放请求。

因此接口被设计成”提交 + 轮询”两段式：POST /scale_elastic_ep 只负责把目标大小写进状态机（scale_phase 变为 waiting_for_cohort），然后立即返回；客户端用 GET /is_scaling_elastic_ep 轮询状态，直到 scale_phase 变成 serving_expanded 或 failed。

TokenizerManager 侧的镜像状态说明了这套轮询的语义：

1
def get_elastic_ep_state(self):
2
    return {
3
        "is_scaling_elastic_ep": self.elastic_pending_ep_size is not None,
4
        "effective_ep_size": self.elastic_worker_count * self.attn_replica_size,
5
        "pending_ep_size": self.elastic_pending_ep_size,
6
        "scale_phase": self.elastic_scale_phase,
7
        "last_error": self.elastic_last_error,
8
    }

effective_ep_size 在 TokenizerManager 里是用 elastic_worker_count * attn_replica_size 算出来的，而不是直接读调度器的状态——因为 TokenizerManager 并不知道每个调度器内部的具体情况，它只维护一个”活跃的 DP worker 数”，乘以每个注意力副本的固定宽度就是有效 EP 大小。这个乘法背后的拓扑约束后面会展开。

而 scale_elastic_ep 的实现里有一个细节：它把请求发给每一个 DP 调度器（docstring 写的就是 “Send a scale request to every DP scheduler”），因为每个调度器都持有自己那份 ElasticEPStateManager 单例，状态必须同步推进：

1
async def scale_elastic_ep(
2
    self, obj: ScaleElasticEPReqInput
3
) -> ScaleElasticEPReqOutput:
4
    """Send a scale request to every DP scheduler."""
5
    if self.elastic_pending_ep_size is not None:
6
        return ScaleElasticEPReqOutput(
7
            success=False,
8
            message=(
9
                "A previous scale operation has not completed yet. Wait until "
10
                "all pending ranks have joined before issuing another scale."
11
            ),
12
            ...
13
        )
14
    self.auto_create_handle_loop()
15
    responses: List[ScaleElasticEPReqOutput] = await self.scale_elastic_ep_communicator(obj)
16
    for res in responses:
17
        if not res.success:
18
            ...
19
            return res

“任一调度器失败就整体失败”是一个全有或全无的语义。这很符合直觉：如果一部分 DP 组认为 EP 已经扩到 8，另一部分认为还是 4，那么后续的 all-to-all 通信双方对不上规模，整个服务会静默地跑出错误结果。所以这里宁可直接把整个请求判失败。

决策真正发生在调度器的 handle_scale_elastic_ep 里，它一次性做了全部合法性检查：

1
def handle_scale_elastic_ep(
2
    self, recv_req: ScaleElasticEPReqInput
3
) -> ScaleElasticEPReqOutput:
4
    """Begin a pending elastic EP scale-up request."""
5
    old_ep_size = ElasticEPStateManager.get_effective_ep_size()
6
    new_ep_size = recv_req.new_ep_size
7
    max_ep_size = get_parallel().max_world_size
8

9
    if new_ep_size <= old_ep_size:
10
        return ScaleElasticEPReqOutput(
11
            success=False,
12
            message=(
13
                f"new_ep_size ({new_ep_size}) must be greater than current "
14
                f"effective_ep_size ({old_ep_size})."
15
            ),
16
            ...
17
        )
18
    if new_ep_size > max_ep_size:
19
        return ScaleElasticEPReqOutput(
20
            success=False,
21
            message=(
22
                f"new_ep_size ({new_ep_size}) exceeds --max-ep-size "
23
                f"({max_ep_size}). Restart with a larger --max-ep-size."
24
            ),
25
            ...
26
        )

三条前置校验分别对应三种物理约束：

• 只能变大（new_ep_size > old_ep_size）。弹性 EP 当前的实现是单向的——只支持扩容，不支持缩容。原因不难理解：缩容意味着要撤掉一部分 rank 并把它们持有的专家权重转移出去，而权重的存储布局在 elastic_ep_initial_size 时就被锁定了，撤掉 rank 会让剩下的 rank 需要接收新专家、重新分配张量切片，工程量远大于扩容。错误信息里的措辞也很直白：“Restart with a larger —max-ep-size”——参数写小了没有别的办法，只能重启。
• 不能超过座位数（new_ep_size <= max_ep_size）。座位在启动时就分配好了，超出只能重启。
• 必须对齐注意力副本宽度：

1
from sglang.srt.elastic_ep.topology import attn_replica_size
2

3
replica_size = attn_replica_size()
4
if new_ep_size % replica_size != 0:
5
    return ScaleElasticEPReqOutput(
6
        success=False,
7
        message=(
8
            f"new_ep_size ({new_ep_size}) would admit an incomplete "
9
            f"attention replica with size {replica_size}."
10
        ),
11
        ...
12
    )

这条约束最能体现”SGLang 里注意力并行与专家并行必须错开”这一设计（上一篇讲过）的后果。topology.py 里的定义是：

1
def attn_replica_size() -> int:
2
    """Return the fixed physical width of one attention-DP replica."""
3
    parallel = get_parallel()
4
    return parallel.attn_tp_size * parallel.attn_cp_size

一个注意力副本（attention replica）是同时承载同一份注意力计算的一组 rank，宽度等于 attn_tp_size × attn_cp_size（cp 是 context parallel）。在 DP Attention 模式下，若干个这样的副本共存，每个副本处理自己那部分请求。于是：

1
DP size = EP size / attn_replica_size

也就是说，新增的 EP rank 必须是整数个注意力副本。如果新的大小只增加半个副本（比如副本宽度是 4，而 EP 从 8 扩到 10），那么新增的 2 张卡里没有一张能凑成完整的、能独立做注意力的副本——第 3 到第 4 号位置空着，构成一个”残缺副本”。这个副本内部的 TP 组不完整，注意力计算无法启动。

topology.py 里还有个专门的校验函数，用来保证 joiner 与主部署使用同样的注意力拓扑：

1
def validate_matching_attention_topology(
2
    *,
3
    primary_attn_tp_size: int,
4
    primary_attn_cp_size: int,
5
    joining_attn_tp_size: int,
6
    joining_attn_cp_size: int,
7
) -> None:
8
    primary_topology = (primary_attn_tp_size, primary_attn_cp_size)
9
    joining_topology = (joining_attn_tp_size, joining_attn_cp_size)
10
    if joining_topology != primary_topology:
11
        raise ValueError(
12
            "Primary and joining cohort must use the same attention topology "
13
            "for Elastic EP scale-up "
14
            ...
15
        )

“joining cohort”（加入队列）是一个重要概念：扩容不是一张卡一张卡地加，而是一整批同构的 rank 一起加入。这批 rank 必须与主部署在注意力侧使用完全一致的 attn_tp_size 与 attn_cp_size，否则它们算出的 hidden state 与主部署对不上，DP Attention 的 gather/scatter 就无法拼接。这解释了一个非常自然的实现选择：joiner 通常就是按与主部署完全相同的启动参数再起一个节点，只是额外带上 --node-rank 1、--elastic-ep-join-mode scale、--elastic-ep-join-rank-offset <当前有效EP大小> 和 --elastic-ep-initial-size <初始EP大小>。

topology.py 还提供了两个把”物理 rank 视角”折叠成”注意力副本视角”的工具：

1
def collapse_physical_rank_status(
2
    status: Sequence[bool], attn_replica_size: int
3
) -> list[bool]:
4
    """Collapse physical-rank health with an all-members replica rule."""
5
    physical_ep_size_to_dp_size(len(status), attn_replica_size)
6
    return [
7
        all(bool(value) for value in status[start : start + attn_replica_size])
8
        for start in range(0, len(status), attn_replica_size)
9
    ]

这个函数的规则是 “全成员才健康”：一个注意力副本只有在它全部的 attn_replica_size 个物理 rank 都健康时才算健康，任何一个掉线，整个副本判定为不健康。这条规则的价值在于把”rank 级健康信息”归约成”DP 级健康信息”，因为调度器与路由是按 DP 组来分配流量的——残缺的副本不能被派活。函数名里的 “collapse” 就是这个归约动作。

加入队列：cohort 注册与新世界同步

主部署收到缩放请求后，只是把状态推进到 waiting_for_cohort，然后等待。真正把事情推进的是 joiner 进程在自己的 ModelRunner 启动阶段执行的一段逻辑：

1
def _initialize_elastic_ep_joiner(self) -> None:
2
    if not (
3
        get_exec().moe.elastic_ep_backend is not None
4
        and get_exec().moe.is_ep_scale_joiner
5
    ):
6
        return
7

8
    parallel = get_parallel()
9
    join_effective_ep_size = parallel.ep_join_rank_offset + parallel.tp_size
10
    dist.barrier(group=parallel.tp_group.cpu_group)
11
    if parallel.tp_rank == 0:
12
        register_scale_cohort(
13
            parallel.ep_join_rank_offset,
14
            join_effective_ep_size,
15
            self._elastic_cuda_graph_enabled(),
16
        )
17
    join_scale_process_group()
18
    get_context().override("elastic_ep.scale_join", ep_size=join_effective_ep_size)

这段代码是整个扩容流程的”握手”核心，逐行拆开：

第一步：算出加入后的新世界大小。

1
join_effective_ep_size = parallel.ep_join_rank_offset + parallel.tp_size

ep_join_rank_offset 是 joiner 的全局 rank 起点（等于主部署当前的有效 EP 大小），加上 joiner 自己的 tp_size，就是加入后的新世界大小。举例：主部署 EP=4，joiner 是 2 张卡的一组（tp_size=2），并且把 --elastic-ep-join-rank-offset 设为 4，那么 join_effective_ep_size = 4 + 2 = 6。这与前面调度器里校验的 new_ep_size 是同一个数——双方必须独立算出相同的结果，否则握手就对不上。这也是为什么 --elastic-ep-join-rank-offset 必须”等于当前有效 EP 大小”，写错了会直接导致 new_ep_size 与实际加入人数不一致。

第二步：在 TCPStore 上登记 cohort。

1
def register_scale_cohort(
2
    rank_offset: int, target_ep_size: int, cuda_graph_enabled: bool
3
) -> None:
4
    store = get_global_tcp_store()
5
    if store is None:
6
        raise RuntimeError("Elastic EP scale-up requires the global TCPStore.")
7
    payload = msgspec.json.encode(
8
        ScaleCohort(
9
            target_ep_size=target_ep_size,
10
            cuda_graph_enabled=cuda_graph_enabled,
11
        )
12
    )
13
    store.set(f"{_SCALE_COHORT_KEY_PREFIX}/{rank_offset}", payload)

ScaleCohort 是一个极简的消息结构：

1
class ScaleCohort(msgspec.Struct, frozen=True, kw_only=True):
2
    target_ep_size: int
3
    cuda_graph_enabled: bool

它只携带两个字段，但每一个都必要：target_ep_size 告诉主部署”我们这批人加入后世界应该多大”，cuda_graph_enabled 告诉主部署”我们这批人启用了 CUDA Graph”——因为老 rank 必须据此决定要不要重新捕获图。这是异构配置下必须显式协商的东西：如果 joiner 开了图捕获而主部署没开，双方的执行路径就不一样，KV 池的内存占用也不同。

存储用的是 get_global_tcp_store()，一个 torch.distributed 的 TCPStore。选择 TCPStore 而不是集合通信来传递这条信息，是有原因的：此刻 joiner 还没有正常加入任何进程组，它只能通过一个外部的、所有进程都能访问的键值存储来”留言”。get_scale_cohort 是对称的读取端，先 store.check([key]) 再 store.get(key) 解码——用 check 而不是直接 get，是为了让主部署能区分”还没人登记”与”登记了但读失败”两种情况。

第三步：真正加入 WORLD 组。

1
def join_scale_process_group() -> None:
2
    """Join the expandable WORLD group for an append-only scale operation."""
3
    _join_world_group()
4
    _refresh_ep_members()

1
def _join_world_group() -> None:
2
    from mooncake.pg import join_group
3

4
    join_group(torch.distributed.group.WORLD)

注意这里调的不是 torch.distributed 的 API，而是 mooncake.pg.join_group。这是整套机制里最关键的一处”借力”：弹性 EP 依赖 Mooncake（Moonshot AI 开源的 KVCache 传输与通信库）提供的进程组扩展能力。标准 PyTorch 进程组没有”中途加入”的接口，Mooncake 的 mooncake.pg 模块补上了这一能力——它允许在一个已经建好的、容量被预留的进程组里，把新 rank 通过 join_group 接进来，并通过 recover_ranks 让掉线的 rank 重新入场。

_refresh_ep_members() 则是通知 MoE token dispatcher 更新成员表：

1
def _refresh_ep_members() -> None:
2
    from sglang.srt.layers.moe.token_dispatcher.mooncake import EPBuffer
3

4
    buffer = EPBuffer.get_existing_buffer()
5
    if buffer is not None:
6
        buffer.update_ep_member()

第四步：在新世界大小下重建依赖规模的组件。

1
get_context().override("elastic_ep.scale_join", ep_size=join_effective_ep_size)
2

3
global_ep_rank = parallel.tp_rank + parallel.ep_join_rank_offset
4
broadcast_global_expert_location_metadata(
5
    model_config=self.model_config,
6
    moe_ep_rank=global_ep_rank,
7
    src_rank=0,
8
)
9
set_global_expert_distribution_recorder(
10
    ExpertDistributionRecorder.init_new(
11
        get_global_expert_location_metadata(),
12
        rank=global_ep_rank,
13
    )
14
)
15

16
from sglang.srt.layers.dp_attention import enable_joiner_all_gather
17

18
enable_joiner_all_gather()
19
update_dp_attention_for_elastic_ep(
20
    physical_ep_size=join_effective_ep_size,
21
    physical_ep_rank=global_ep_rank,
22
)

这里四件事各自解决一个问题：

• get_context().override(...)：把”我是以 scale joiner 身份加入的、加入后的真实 EP 大小是多少”写进运行时上下文，供后续所有依赖 EP 大小的构造代码读取。
• broadcast_global_expert_location_metadata(..., src_rank=0)：从 rank 0 广播全局专家布点。joiner 手里没有主部署的专家布点信息（哪一层哪个专家在哪个物理位置），必须从主部署同步一份。src_rank=0 写死在这里，是因为扩容路径下主部署的 rank 0 一定活着——它是发起缩放的节点，不可能同时是掉线节点。
• ExpertDistributionRecorder.init_new(...)：初始化专家负载统计器。EPLB 的重排决策依赖每个 rank 上报的专家调用次数，joiner 必须从接入的一刻就开始记账。
• enable_joiner_all_gather() + update_dp_attention_for_elastic_ep(...)：把 DP Attention 的模块级状态改成”扩展世界”模式。因为 joiner 不参与主部署启动时的 TP 组，DP Attention 的 all-gather 必须走一条新的、覆盖全部活跃 rank 的路径。

最后，joiner 把新世界写回自己的状态机快照：

1
state = ElasticEPStateManager.instance()
2
if state is not None:
3
    state.active_ranks.zero_()
4
    state.active_ranks[:join_effective_ep_size] = 1
5
    state.snapshot_active_to_last()
6
    state.sync_active_to_cpu()
7
    state.scale_phase = "syncing_new_world"

active_ranks.zero_() 之后把前 join_effective_ep_size 位置为 1，等价于”承认 0 到 join_effective_ep_size-1 全部在场”。这与 reset() 的逻辑一致，都是”先清空、再按有效大小重填前缀”。

还有一个细节值得单独讲：joiner 在整段流程里做了一件很反直觉的事——主动关掉 EPLB：

1
if self.eplb_manager is not None:
2
    self.eplb_manager.disable_rebalance(
3
        "EPLB rebalance is disabled while elastic EP scale-up "
4
        "is being finalized"
5
    )

主部署在收到缩放请求时也做同样的事（eplb_manager.disable_rebalance("elastic EP scale-up is pending")）。双方一起禁掉 EPLB，原因是：扩容期间专家布点正在被重新计算与广播，此时如果 EPLB 并发地去做一次重排（它的重排也是搬专家权重、改 physical_to_logical_map），两者会互相覆盖对方的元数据，结果是某些 rank 按旧布点路由、某些按新布点路由，token 被送到错误的专家上。禁重排这个动作把风险窗口彻底关掉，等缩放走到 serving_expanded 再由主部署主动触发一次重排（见后文 maybe_rebalance_after_rank_fault）。

rank 故障：位图检测、peer 等待与逐组恢复

扩容讲完，接下来是另一半：故障恢复。这是弹性 EP 比”动态扩容”更难的部分，因为它面对的是不可预期的成员变化——rank 可能在任意时刻消失，而剩下的 rank 必须在不重启的前提下继续服务。

检测的入口在 ModelRunner 的前向路径上：

1
def maybe_recover_ep_ranks(
2
    *,
3
    tp_group: parallel_state.GroupCoordinator,
4
    eplb_manager: EPLBManager,
5
    model_config: ModelConfig,
6
    moe_ep_rank: int,
7
) -> bool:
8
    # TODO(perf): `active_ranks.all()` on a CUDA tensor triggers host-device
9
    # synchronization, and this function is on the forward-path.
10
    # This check only runs when `--elastic-ep-backend` is enabled, so the
11
    # synchronization overhead does not propagate to other configs.
12
    # Leave for future optimization of the elastic EP path.
13
    if tp_group.active_ranks.all() and tp_group.active_ranks_cpu.all():
14
        return False

这个函数的快路径就是这一行判断：如果两张位图（设备上的与 CPU 上的）全是 1，立即返回 False，什么也不做。绝大多数前向调用走的就是这条路径。

源码里的 TODO(perf) 注释很有价值，它诚实地记录了这个设计的代价：在 CUDA 张量上调用 .all() 会触发一次 host-device 同步。GPU 侧位图在显存里，要让 CPU 判断它是否为全 1，必须把数据拷回主机，这会打断 CPU 与 GPU 的异步流水线。作者的处理方式是”知道代价但接受”，理由是这条检查只在开启 --elastic-ep-backend 时才执行，不会污染普通配置的性能。这也解释了为什么状态里要维护一份 active_ranks_cpu：故障检测路径尽量只读 CPU 副本，避免为了一次”世界是否健康”的探测而付 GPU 同步的代价。

当快路径不成立（存在不在场的 rank），进入慢路径：

1
tp_active_ranks = tp_group.active_ranks.detach().cpu().numpy()
2
tp_active_ranks_cpu = tp_group.active_ranks_cpu.detach().numpy()
3
tp_active_ranks &= tp_active_ranks_cpu
4
# NOTE: `ranks_to_recover` uses indices in `tp_group`. For the current
5
# Mooncake elastic EP implementation we assume `--pp-size=1`, so the
6
# tp-group index is the same as the global rank index.
7
ranks_to_recover = [
8
    i for i in range(len(tp_active_ranks)) if not tp_active_ranks[i]
9
]

三件事：

1. 两份位图按位与（&=）。设备位图与 CPU 位图可能存在短暂不一致（一方已更新、另一方还没同步），取交集是保守做法：只有两边都认为不在场的 rank 才被当作需要恢复。取并集会导致把”其实已经回来了但设备位图还没刷新”的 rank 当成故障 rank 去恢复，凭空制造一次多余的集合通信。
2. ranks_to_recover 是”不在场”的 rank 列表，不是”在场”的。
3. 注释坦白了当前实现的拓扑假设：索引是 TP 组内的局部索引，之所以能直接当全局 rank 用，是因为当前 Mooncake 弹性 EP 实现假设 --pp-size=1。这是一个明确的适用范围声明——如果你用了流水线并行，这条恢复路径的前提就不成立了。

真正的恢复动作在 try_recover_ranks 里，它做的事情远比名字看上去复杂：

1
def try_recover_ranks(global_ranks: List[int]) -> bool:
2
    """Recover ranks in WORLD and every launch-time parallel group."""
3
    if not _try_recover_world(global_ranks):
4
        return False
5

6
    from mooncake.pg import recover_ranks
7

8
    for group in _iter_live_parallel_groups():
9
        local_ranks = _map_global_to_group_local_ranks(group.ranks, global_ranks)
10
        if not local_ranks:
11
            continue
12

13
        _wait_for_peer_state(group.device_group, local_ranks)
14
        recover_ranks(group.device_group, local_ranks)
15
        _wait_for_peer_state(group.cpu_group, local_ranks)
16
        recover_ranks(group.cpu_group, local_ranks)
17
        _maybe_create_message_queue(group)
18

19
    _refresh_ep_members()
20
    return True

这段代码回答了一个非常具体的问题：一个 rank 掉线时，它到底从”多少个进程组”里掉出去了？

答案是——每一个它参与过的进程组。一个 TP rank 可能同时属于 WORLD、TP 组、EP 组、DP 组、注意力组等等，_iter_live_parallel_groups() 把这些组全部遍历出来：

1
def _iter_live_parallel_groups() -> Iterator[parallel_state.GroupCoordinator]:
2
    groups = []
3
    for group_ref in parallel_state._groups.values():
4
        group = group_ref()
5
        if group is not None:
6
            groups.append(group)
7
    yield from sorted(groups, key=lambda group: group.unique_name)

两个细节：parallel_state._groups 里存的是弱引用（group_ref() 解引用后可能为 None，说明该组已被回收），所以要先过滤掉已死的组；排序用 unique_name 是为了让所有 rank 以完全一致的顺序遍历组——如果各 rank 以不同顺序调 recover_ranks，集合通信的匹配就会错位，这是分布式代码里最经典的一类 bug。

然后对每个组做四步操作：

1
local_ranks = _map_global_to_group_local_ranks(group.ranks, global_ranks)

第一步：全局 rank 转组内 rank。 因为每个进程组有自己的编号体系（组内 rank 0 未必是全局 rank 0），恢复操作必须用组内编号：

1
def _map_global_to_group_local_ranks(
2
    group_ranks: List[int], global_ranks: List[int]
3
) -> List[int]:
4
    rank_to_local = {rank: index for index, rank in enumerate(group_ranks)}
5
    return [rank_to_local[rank] for rank in global_ranks if rank in rank_to_local]

注意最后的 if rank in rank_to_local——如果一个故障 rank 不属于这个组，就跳过它（比如某个注意力副本组的成员范围不含这个 rank）。if not local_ranks: continue 处理的就是”这个组里一个故障 rank 都没有”的情况。

第二步：等待对端到达就绪态。

1
_PEER_STATE_POLL_INTERVAL_SEC = 0.01
2

3
def _wait_for_peer_state(backend, ranks: List[int]) -> None:
4
    from mooncake.pg import get_peer_state
5

6
    while not all(get_peer_state(backend, ranks)):
7
        time.sleep(_PEER_STATE_POLL_INTERVAL_SEC)

这里用的是轮询等待（间隔 10ms）而不是阻塞式等待。为什么必须等？因为恢复是”把掉线的 rank 拉回来”，而 rank 能不能被拉回来，取决于对方进程是否已经重启并处于可接入状态。轮询是唯一能表达”等对方准备好”的方式。

maybe_recover_ep_ranks 里有一段注释专门解释了为什么这个看起来”很危险”的轮询其实是安全的：

1
# try_recover_ranks polls peer state via Mooncake EP backend.
2
# Mooncake's internal semantics guarantee that all ranks observe
3
# consistent peer readiness state, so collective operations below
4
# are safe even though polling appears local.

也就是说：各个 rank 各自独立地判断”对端准备好了吗”，表面上它们可能在不同时刻得出不同结论，从而在不同时刻进入集合通信——这是分布式系统里的经典死锁模式（一部分 rank 已经在集合通信里等，另一部分还在轮询）。Mooncake 的语义保证消除了这个风险：所有 rank 观察到一致的 peer 就绪状态。这个保证是整个恢复机制能成立的前提，也是它必须绑定 Mooncake 这类特殊后端、无法直接用原生 PyTorch 进程组实现的原因。

第三步：设备组与 CPU 组各恢复一次。

1
recover_ranks(group.device_group, local_ranks)
2
_wait_for_peer_state(group.cpu_group, local_ranks)
3
recover_ranks(group.cpu_group, local_ranks)

每个 GroupCoordinator 都持有两个通信通道（device_group 走 NCCL/NVLink，cpu_group 走 gloo），两者必须分别恢复，因为它们的成员表、缓冲区、连接是各自独立的。先恢复设备组再恢复 CPU 组，并且各自都先 _wait_for_peer_state——顺序与等待都不能省。

第四步：重建共享内存广播环。

1
def _maybe_create_message_queue(group) -> None:
2
    if not group.use_message_queue_broadcaster or group.world_size <= 1:
3
        return
4

5
    from sglang.srt.distributed.device_communicators.shm_broadcast import MessageQueue
6

7
    group.mq_broadcaster = MessageQueue.create_from_process_group(
8
        group.cpu_group, 1 << 22, 6
9
    )

MessageQueue 是 SGLang 用来做”一份数据广播给多个 rank”的共享内存环（1 << 22 是 4 MiB 的环大小，6 是槽位数）。它在创建时就绑定了进程组的成员，rank 故障会让旧的消息队列失效，所以在恢复完 CPU 组之后必须重建。use_message_queue_broadcaster 判断保证只有真正用到广播的组才重建。

_try_recover_world 和 try_admit_scale_ranks 是 WORLD 组的对应版本：

1
def _try_recover_world(global_ranks: List[int]) -> bool:
2
    from mooncake.pg import get_peer_state, recover_ranks
3

4
    world_backend = torch.distributed.group.WORLD
5
    if not all(get_peer_state(world_backend, global_ranks)):
6
        return False
7

8
    recover_ranks(world_backend, global_ranks)
9
    logger.debug("[Elastic EP][recover] WORLD recover_ranks(%s) done", global_ranks)
10
    return True
11

12
def try_admit_scale_ranks(global_ranks: List[int]) -> bool:
13
    """Admit append-only ranks into the expandable WORLD group."""
14
    if not _try_recover_world(global_ranks):
15
        return False
16
    _refresh_ep_members()
17
    return True

注意 _try_recover_world 与 _wait_for_peer_state 的行为差异：前者用 if not all(...): return False 立即返回失败，不做等待；后者是死等到就绪。这个差异有明确用意——扩容路径不能阻塞主部署的调度循环。新 rank 什么时候起来由外部运维决定，主部署不能为了等它而把整个前向路径堵住。所以扩容侧是”探测到就接入，探测不到就下次再来”，而恢复侧是”既然已经确认要恢复，就等到底”。

恢复完成后，还有两件收尾工作：

1
if ranks_to_recover and try_recover_ranks(ranks_to_recover):
2
    eplb_manager.reset_generator()
3
    broadcast_global_expert_location_metadata(
4
        model_config=model_config,
5
        moe_ep_rank=moe_ep_rank,
6
        src_rank=get_healthy_expert_location_src_rank(
7
            invoked_in_elastic_ep_rejoin_path=False
8
        ),
9
    )
10
    ElasticEPStateManager.instance().reset()
11
    logger.info(f"recover ranks {ranks_to_recover} done")
12
    return True

• eplb_manager.reset_generator()：把 EPLB 的重排生成器重置。因为世界大小变了，之前算到一半的布点已经作废。
• 重新广播专家布点，但这次 src_rank 不再写死 0，而是走一个专门的挑选函数：

1
def get_healthy_expert_location_src_rank(
2
    *, invoked_in_elastic_ep_rejoin_path: bool
3
) -> int:
4
    world_group = get_parallel().world_group
5
    # NOTE: do not key off the launch-time `ep_join_mode` here.
6
    # A rank that was started as a rejoin rank may later act as a healthy
7
    # rank in a subsequent recovery cycle.
8
    local_rejoin_flag = bool(invoked_in_elastic_ep_rejoin_path)
9
    gathered_rejoin_flags = world_group.all_gather_object(local_rejoin_flag)
10

11
    for rank_in_group, is_rejoin_rank in enumerate(gathered_rejoin_flags):
12
        if not is_rejoin_rank:
13
            return world_group.ranks[rank_in_group]
14

15
    raise RuntimeError(
16
        "No healthy rank found for broadcasting expert location metadata. "
17
        "All ranks are marked as elastic_ep_rejoin."
18
    )

这个函数解决的是一个身份漂移问题。注释说得很直白：“do not key off the launch-time ep_join_mode here”——不能用启动时写的 --elastic-ep-join-mode 来判断谁是健康 rank。原因是：一个以 recover 身份启动的 joiner，在故障恢复完成后就变成了正常的服务 rank；如果之后再发生一次故障，它完全可能充当”提供布点的健康源”。启动参数描述的是”这个进程出生时的角色”，而恢复需要的是”此刻谁手里有正确的布点”——这是两个不同的问题。

实现方式是用 all_gather_object 收集所有 rank 的”我是不是 rejoin rank”标志，取第一个不是 rejoin 的 rank 作为广播源。兜底逻辑也很有意思：如果所有人都自称 rejoin rank，直接抛错而不是随便挑一个——因为随便挑一个可能挑到手里没有正确布点的 rank，把错误的专家布点广播出去，那比直接失败糟糕得多。

世界变了，专家布点必须重算

扩容或恢复完成后，世界大小变了，但专家布点还是按旧世界算的。这时必须重排，而重排的第一件事是把”活跃 rank 数”喂给 EPLB 的算法。

在 python/sglang/srt/eplb/eplb_algorithms/__init__.py 里可以看到这个喂法：

1
ElasticEPStateManager.instance().active_ranks
2
if ElasticEPStateManager.instance() is not None
3
else ElasticEPStateManager.healthy_rank_state()

而 healthy_rank_state 是构造”全员健康”位图的工具：

1
@classmethod
2
def healthy_rank_state(
3
    cls, *, ep_size: Optional[int] = None, device: Optional[torch.device] = None
4
) -> torch.Tensor:
5
    size = ep_size if ep_size is not None else torch.distributed.get_world_size()
6
    dev = device if device is not None else cls._select_device()
7

8
    return torch.ones(size, dtype=torch.int32, device=dev)

注意这里 _select_device() 的存在：位图优先放在 CUDA 上，没有 CUDA 时回落到 CPU——

1
@staticmethod
2
def _select_device() -> torch.device:
3
    if is_cuda():
4
        return torch.device("cuda")
5
    elif is_cpu():
6
        return torch.device("cpu")
7
    else:
8
        raise NotImplementedError("Only CUDA and CPU support elastic ep now.")

只有 CUDA 与 CPU 两条路径，其它加速器（NPU、ROCm 等）在当前实现里会直接报错。这是一个明确的适用范围边界。

拿到位图之后，重排算法本身要做的事在 python/sglang/srt/eplb/eplb_algorithms/elasticity_aware.py 里：

1
def rebalance_experts(
2
    weight: torch.Tensor,
3
    num_replicas: int,
4
    num_groups: int,
5
    num_nodes: int,
6
    num_gpus: int,
7
    enable_hierarchical: bool,
8
    active_ranks: torch.Tensor,
9
) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
10
    """
11
    Entry point for expert-parallelism load balancer.
12

13
    Parameters:
14
        weight: [layers, num_logical_experts], the load statistics for all logical experts
15
        num_replicas: number of physical experts, must be a multiple of `num_gpus`
16
        ...
17
    """
18
    num_layers, num_logical_experts = weight.shape
19
    weight = weight.float().cpu()
20
    num_active_ranks = active_ranks.sum().item()
21
    num_local_experts = num_replicas // num_gpus
22
    if num_active_ranks < num_gpus:
23
        # Must fall back to global load-balance policy
24
        # and fix some params
25
        phy2log, phyrank, logcnt = rebalance_experts_hierarchical(
26
            weight,
27
            num_local_experts * num_active_ranks,
28
            1,
29
            1,
30
            num_active_ranks,
31
        )
32
    elif enable_hierarchical:
33
        ...

关键在于 num_active_ranks = active_ranks.sum().item()。布点算法不是按”物理 GPU 总数”算的，而是按”当前活跃的 GPU 数”算的。当有 rank 掉线时，num_active_ranks < num_gpus 成立，于是走一条特殊的退路：

1
rebalance_experts_hierarchical(
2
    weight,
3
    num_local_experts * num_active_ranks,
4
    1,
5
    1,
6
    num_active_ranks,
7
)

它把 num_replicas 从”完整规模”降到”每卡专家数 × 活跃卡数”，同时把 num_groups 与 num_nodes 都设成 1，也就是放弃层次化均衡、退回到全局扁平均衡。这个退路的动机是：层次化均衡的目标是”尽量减少跨节点通信”，它需要知道节点拓扑；而当有节点掉线时，拓扑已经不完整了，继续按旧拓扑做层次化均衡可能把专家搬到已经不存在的节点上。退回到扁平模式虽然通信代价更高，但至少保证布点落在真实存在的 rank 上。

后面还有一段专门处理”如何把布点结果重新对齐到完整的 rank 空间”：

1
if num_active_ranks < num_gpus:
2
    phy2log_slices = list(
3
        phy2log.view(num_layers, num_active_ranks, -1).unbind(dim=1)
4
    )
5
    active_ranks_list = active_ranks.tolist()
6
    for idx, active_rank in enumerate(active_ranks_list):
7
        if not active_rank:
8
            phy2log_slices.insert(idx, torch.zeros_like(phy2log_slices[0]))
9
            log2phy = torch.where(
10
                log2phy >= idx * num_local_experts,
11
                log2phy + num_local_experts,
12
                log2phy,
13
            )
14
    phy2log = torch.stack(phy2log_slices, dim=1).contiguous().view(num_layers, -1)

这段代码做的是”把压缩的布点结果张开回完整规模”：

• 循环遍历完整位图，遇到掉线的 rank 就插入一片全零（phy2log_slices.insert(idx, torch.zeros_like(...))），把掉线 rank 的位置占住，并把它之后所有 rank 的编号整体后移。
• log2phy 的修正要跟着做：log2phy 是”逻辑专家到物理位置”的反查表，当某个位置被插入的零片推开后，所有 >= idx * num_local_experts 的条目都要 + num_local_experts 才指向正确的物理专家。

为什么非要”张开”而不是干脆输出压缩后的表？ 因为下游的所有消费者——MoE 层的 physical_to_logical_map、token dispatcher 的 all-to-all 路由表、ExpertBackupClient 的物理专家编号运算——都假设表的下标就是物理 rank 编号，长度等于完整规模。如果这里输出一个长度只有活跃卡数的紧凑表，下游就得整体改接口。张开一次、让下游保持不变，是更省事也更不容易出错的选择。

ExpertLocationMetadata.logical_to_all_physical(...) 反查的用法在权重备份客户端里能看得更清楚：

1
physical_expert_ids = global_expert_location_metadata.logical_to_all_physical(
2
    layer_id, expert_id
3
)
4
for physical_expert_id in physical_expert_ids:
5
    if physical_expert_id not in range(
6
        num_local_experts * self.moe_ep_rank,
7
        num_local_experts * (self.moe_ep_rank + 1),
8
    ):
9
        continue
10
    param = weight_param[physical_expert_id % num_local_experts]

“逻辑专家 → 所有物理副本”是一对多的：一个逻辑专家可能因为冗余专家机制被复制到多个物理位置。% num_local_experts 把全局物理编号折算成当前 rank 内部的槽位编号。这套编号体系正是上一篇讲的 EPLB 机制的延续，弹性 EP 只是给它加上了”活跃集合会变”这个新维度。

1
def _compute_elastic_expert_layout(
2
    base_num_physical_experts: int,
3
    initial_ep_size: int,
4
    effective_ep_size: int,
5
) -> tuple[int, int]:
6
    assert base_num_physical_experts % initial_ep_size == 0
7
    num_local_physical_experts = base_num_physical_experts // initial_ep_size
8
    return (
9
        num_local_physical_experts * effective_ep_size,
10
        num_local_physical_experts,
11
    )

1
每 rank 槽位数（不变量） = base_num_physical_experts / initial_ep_size
2
全局槽位数（随世界变） = 每 rank 槽位数 × effective_ep_size

1
# elastic-EP scale-up rewrites ep_size on the published config
2
ep_size = get_parallel().ep_size

权重备份：为什么每台节点必须留一份 DRAM 副本

最后一个大问题：世界大小变了，专家在 rank 之间的分布变了，那么原本只加载了”自己那一份”专家的 rank，需要拿到新分给它的专家权重——这些权重从哪来？

最直接的办法是从磁盘重新加载。update_expert_location_with_recovery 里就保留着这条兜底路径：

1
if len(p2p_missing_logical_experts) > 0:
2
    # Load the missing expert weights from disk
3
    if callable(getattr(model, "generate_weight_name_filter", None)):
4
        # Filter and load only missing expert weights
5
        weight_name_filter = model.generate_weight_name_filter(
6
            p2p_missing_logical_experts
7
        )
8
    else:
9
        # Do a full reload from disk/DRAM
10
        logger.info(
11
            "[Elastic EP] Model does not implement generate_weight_name_filter. "
12
            "Performing full weight reload."
13
        )
14
        weight_name_filter = None
15

16
    if expert_backup_client is not None and expert_backup_client.use_backup:
17
        # Load the missing weights from the DRAM backup
18
        expert_backup_client.update_weights(weight_name_filter)
19
    else:
20
        # Load the missing weights from disk
21
        update_weights_from_disk_callable(
22
            get_model().model_path,
23
            get_model().load_format,
24
            weight_name_filter=weight_name_filter,
25
        )

代码里有一个清晰的三级降级链：

1. 模型实现了 generate_weight_name_filter：只加载缺失的那部分专家权重，代价最小。
2. 模型没实现：weight_name_filter = None，退化成全量重载——把所有专家权重重新读一遍，日志里专门打了一条 “Performing full weight reload” 提示。
3. 备份可用 vs 不可用：即使拿到了 filter，还有一个岔路——从 DRAM 备份读（expert_backup_client.update_weights(...)）还是从磁盘读（update_weights_from_disk_callable(...)）。

第三点就是 ExpertBackupManager 存在的意义。为什么需要 DRAM 备份，直接从磁盘读不行吗？

考虑一次恢复的真实时间线：一个 8 卡节点在运行中掉线，运维重启了它；主部署检测到 rank 缺失，发起恢复；恢复流程需要把这些 rank 原来持有的专家权重补齐（因为重排后布点变了）。如果从磁盘读，那么这次恢复的耗时 = 存储系统读取数百 GB 权重的耗时。对一个大 MoE 模型，专家权重可能占几十到上百 GB；如果是网络存储或慢盘，这段时间可能是几分钟到几十分钟。期间服务处于降级状态。

而专家权重在每台节点的 DRAM 里留一份副本，恢复时就变成一次内存到显存的搬运，走的是 Mooncake 的 RDMA 传输引擎（register_memory + batch_transfer_sync_read），带宽比磁盘高一个数量级以上。

ExpertBackupManager 的构造过程就是”把权重读进 DRAM 并注册给传输引擎”：

1
def backup_weights_from_disk(self):
2
    load_config = LoadConfig(load_format=self.load_format)
3
    loader = get_model_loader(load_config, self.model_config)
4

5
    with set_default_torch_dtype(self.model_config.dtype):
6
        iter = loader._get_weights_iterator(
7
            DefaultModelLoader.Source.init_new(self.model_config, None)
8
        )
9

10
        total_bytes = 0
11
        weight_info_dict = {}
12

13
        for name, weight in iter:
14
            expert_id = extract_expert_id(name)
15
            if expert_id < self.idmx and expert_id >= self.idmn:
16
                ...
17
                total_bytes += byte_size
18

19
        if total_bytes == 0:
20
            self.continuous_buffer = None
21
            self.weight_pointer_map = {}
22
            return
23

24
        self.continuous_buffer = torch.empty(
25
            total_bytes, dtype=torch.uint8, device="cpu"
26
        )

几个要点：

• extract_expert_id(name) 用正则 \.experts\.(\d+)\. 从参数名里挖出专家编号。这是”按参数名做数据分片”的典型做法——不需要额外的元数据文件，参数名本身就编码了专家编号。
• 按节点分片：self.idmn 与 self.idmx 是这台节点负责的专家编号区间：

1
self.expert_num = self.model_config.hf_config.n_routed_experts
2
self.idmn = (self.expert_num // self.engine_num) * self.engine_rank
3
self.idmx = (self.expert_num // self.engine_num) * (self.engine_rank + 1)

engine_num 是节点数（nnodes），engine_rank 是节点编号（node_rank）。每个节点备份的是”全局专家编号里属于自己那一段”的权重。这个划分是静态的、按节点的，与世界大小（EP 大小）无关——这正是前面 elastic_ep_initial_size 要”锁定每 rank 专家存储布局”的原因：备份的划分是按初始布局算的，扩容后布局不能变，否则备份的区间就对不上。

• continuous_buffer 是一整块连续 CPU 内存：所有本节点的专家权重被按名字排序后顺序拷贝进一块 torch.uint8 缓冲：

1
for name in sorted(weight_info_dict.keys()):
2
    weight_info = weight_info_dict[name]
3
    weight = weight_info["weight"]
4
    byte_size = weight_info["byte_size"]
5
    weight_flat = weight.flatten().contiguous()
6
    weight_bytes = weight_flat.view(torch.uint8)
7
    start_byte = current_byte_offset
8
    end_byte = current_byte_offset + byte_size
9
    weight_ptr = buffer_base_ptr + current_byte_offset
10
    self.continuous_buffer[start_byte:end_byte].copy_(weight_bytes)
11
    self.weight_pointer_map[name] = ExpertWeightPointer(
12
        weight_ptr=weight_ptr,
13
        byte_size=byte_size,
14
    )
15
    current_byte_offset = end_byte

为什么必须是一整块连续缓冲，而不是每个权重单独一个张量？ 因为 RDMA 传输引擎需要注册一段连续的、物理地址稳定的内存区域（register_memory(server_ptr, server_len)）。零散的张量意味着要为每一块单独注册，注册开销与元数据管理都会爆炸。合成一块连续缓冲后，只需注册一次，并记录每个权重在这块缓冲里的字节偏移（weight_ptr + byte_size），传输时按偏移取出即可。

上图来自 SGLang 官方博客 Large-Scale Expert Parallelism for DeepSeek（LMSYS，2025-05-05）。它展示并行设计下专家权重的落位方式——正是”哪张卡持有哪些专家”这件事，在弹性扩容与故障恢复之后必须被重新计算，也是 DRAM 备份要按节点分片保存专家权重的原因。

• weight_pointer_map 是”参数名 → (指针, 字节数)“的映射表。这张表会被序列化后通过 ZMQ 发给所有客户端：

1
back_req = BackupDramReq(
2
    rank=self.engine_rank,
3
    weight_pointer_map=self.weight_pointer_map,
4
    session_id=self.session_id,
5
    buffer_size=self.continuous_buffer.numel()
6
    * self.continuous_buffer.element_size(),
7
)
8
sock_send(self.send_to_expert_backup_client, back_req)

session_id 是 Mooncake 传输引擎的会话标识，客户端拿到它才能发起跨机的内存读取。还有一段”等所有客户端报到”的逻辑，注释解释了它存在的原因：

1
# Block until all expert backup clients have reported readiness, to avoid
2
# losing the initial PUB message due to slow joiners.
3
num_ready_clients = 0
4

5
while num_ready_clients < get_parallel().tp_size:
6
    sock_recv(self.recv_from_expert_backup_client)
7
    num_ready_clients += 1

这是 ZMQ PUB/SUB 模式的经典陷阱：PUB 端发出的消息只送给”此刻已连接”的 SUB 端，晚连接的订阅者会静默丢掉在它连接之前发出的消息（“slow joiner” 问题）。如果 manager 不等客户端就绪就发 BackupDramReq，部分客户端永远收不到，use_backup 会一直是 False。解决方案是加一条 PUSH/PULL 同步通道：客户端连上后先往 PULL 端口发一条 UpdateExpertBackupReq，manager 数够 tp_size 条才发正式消息。代码里那句注释 “Synchronization socket to avoid PUB/SUB slow joiner issues” 说的就是这件事。

客户端侧接收完后进入线程循环：

1
def _receive_loop(self):
2
    cnt = 0
3
    while cnt < self.engine_num:
4
        response = sock_recv(self.recv_list[cnt])
5
        self.dram_map_list[response.rank] = response.weight_pointer_map
6
        self.session_id_list[response.rank] = response.session_id
7
        self.buffer_size = max(self.buffer_size, response.buffer_size)
8
        cnt += 1
9

10
    self.use_backup = True
11
    self.start_transfer_client()

每个客户端保存全部节点的映射表（dram_map_list 是长度为 engine_num 的列表），而不是只保存自己节点的。这是必需的：恢复时一个 rank 需要的专家可能原本属于任意节点，客户端必须能从任何节点拉权重。buffer_size 取所有节点的最大值，用于分配足够大的传输缓冲。

客户端的注册是把自己的全部参数注册给传输引擎：

1
def start_transfer_client(self):
2
    from sglang.srt.distributed.parallel_state import get_mooncake_transfer_engine
3

4
    self.transfer_engine = get_mooncake_transfer_engine()
5

6
    self.params_dict = dict(self._get_model().named_parameters())
7
    for name, param in self.params_dict.items():
8
        param_data = param.data
9
        ret_value = self.transfer_engine.engine.register_memory(
10
            param_data.data_ptr(), param_data.numel() * param_data.element_size()
11
        )
12
        if ret_value != 0:
13
            self.use_backup = False
14
            logger.warning("Register fails. Stop using expert weight backup!")
15
            break

注册失败就关掉备份，而不是抛异常。这是一处很务实的容错：备份只是”加速恢复”的手段，不是正确性的必要条件（磁盘路径始终兜底），所以注册失败时降级到磁盘读取即可，不必让整个服务起不来。

真正搬运权重的是 update_weights，它是全篇最复杂的一段逻辑，因为它同时要处理三件事：名字映射、专家编号映射、张量切片。

1
def update_weights(self, weight_name_filter=None):
2
    global_expert_location_metadata = get_global_expert_location_metadata()
3
    num_experts = (
4
        self.model_config.hf_config.n_routed_experts
5
        + get_exec().moe.ep_num_redundant_experts
6
    )
7
    num_local_experts = num_experts // self.moe_ep_size
8
    for i in range(self.engine_num):
9
        server_ptr_list = []
10
        local_ptr_list = []
11
        weight_size_list = []
12

13
        for name, weight_info in self.dram_map_list[i].items():
14
            ...
15
            layer_id, expert_id, weight_name = extract_layer_and_expert_id(name)

名字映射：磁盘上的参数名（HuggingFace 命名）与运行时模型里的参数名不同。客户端做了一次规范化替换：

1
if weight_name == "gate_proj":
2
    shard_id = "w1"
3
    param_name = "experts.w13_"
4
elif weight_name == "down_proj":
5
    shard_id = "w2"
6
    param_name = "experts.w2_"
7
elif weight_name == "up_proj":
8
    shard_id = "w3"
9
    param_name = "experts.w13_"
10
else:
11
    raise RuntimeError(f"Unknown weight name {weight_name}")
12

13
name = name.replace(f"experts.{expert_id}.{weight_name}.", param_name)
14
weight_param = self.params_dict[name]

这里有三处值得说明的细节：

• gate_proj 与 up_proj 都被映射到 experts.w13_。运行时模型把 MLP 的两个投影合并存储在一张 w13 张量里（这是为了配合 fused MoE kernel 的 group GEMM，一次读出两个投影的权重），而磁盘上是分开的两份。
• shard_id 记录了”这一份对应合并张量的哪一半”，下面切片时就靠它：

1
param = weight_param[physical_expert_id % num_local_experts]
2
if shard_id == "w1":
3
    param = param.narrow(0, 0, param.shape[0] // 2)
4
elif shard_id == "w3":
5
    param = param.narrow(
6
        0, param.shape[0] // 2, param.shape[0] // 2
7
    )

w1（gate）取合并张量的前半段，w3（up）取后半段，w2（down）不切。这段切片逻辑是”存储布局与磁盘布局不同构”的必然代价——它之所以出现在恢复路径上，是因为恢复是从磁盘格式的备份里取值，要往运行时格式的张量里写。

• else: raise RuntimeError：遇到不认识的投影名直接报错，不静默跳过。静默跳过会造成”权重看起来加载成功了但某些投影没更新”的最坏情况——模型能跑、输出却是错的。

专家编号映射：这是恢复路径里最绕的一段。

1
physical_expert_ids = (
2
    global_expert_location_metadata.logical_to_all_physical(
3
        layer_id, expert_id
4
    )
5
)
6
for physical_expert_id in physical_expert_ids:
7
    if physical_expert_id not in range(
8
        num_local_experts * self.moe_ep_rank,
9
        num_local_experts * (self.moe_ep_rank + 1),
10
    ):
11
        continue
12
    param = weight_param[physical_expert_id % num_local_experts]

三个编号空间在这里交汇：

• expert_id：备份文件里的编号，是磁盘布局的专家下标（对应 elastic_ep_initial_size 时的划分）。
• physical_expert_id：全局物理专家编号，由 EPLB 的 physical_to_logical_map 决定，可能包含冗余专家。
• num_local_experts * self.moe_ep_rank 到 ... * (moe_ep_rank + 1) 的区间：当前 rank 负责的物理专家区间。只有落在这个区间里的物理专家副本才需要被写入本 rank。同一个逻辑专家被复制到多个物理位置时，logical_to_all_physical 返回多个编号，客户端只为”属于自己”的那些写入。

最后是传输本身：

1
server_ptr_list.append(weight_info.weight_ptr)
2
local_ptr_list.append(param.data_ptr())
3
assert param.numel() * param.element_size() == weight_info.byte_size
4
weight_size_list.append(weight_info.byte_size)
5
...
6
ret = self.transfer_engine.engine.batch_transfer_sync_read(
7
    self.session_id_list[i],
8
    local_ptr_list,
9
    server_ptr_list,
10
    weight_size_list,
11
)

assert param.numel() * param.element_size() == weight_info.byte_size 是一个”安全阀”：它保证目标张量的大小与源数据的字节数完全一致。这个断言如果失败，说明名字映射或切片逻辑错了；如果不做这个断言，RDMA 会按错误的长度读写，结果是显存越界或静默写入错误数据——后者远比崩溃难排查。用一句 assert 把可能的内存损坏转化成一次明确的失败，是这段代码里最有工程价值的一处设计。

batch_transfer_sync_read 是同步批量读：一次性提交多个 (源指针, 目的指针, 长度) 三元组，由传输引擎批量发起 RDMA 读。批量提交而不是逐条提交，是因为 RDMA 的每次操作都有固定开销（QP 提交、完成队列轮询），把几百个小传输合并成一次批量提交，开销可以摊薄一个数量级以上。

工程实践：从启动命令到一次完整的扩缩演练

前面把机制讲完了，这一节把它落到可执行的形态上。

1
sglang serve \
2
  --model-path deepseek-ai/DeepSeek-V3 \
3
  --tp 8 \
4
  --attn-dp-size 8 \
5
  --moe-a2a-backend mooncake \
6
  --elastic-ep-backend mooncake \
7
  --max-ep-size 16 \
8
  --enable-eplb \
9
  --ep-num-redundant-experts 32 \
10
  --deepep-mode low_latency \
11
  --enable-dp-lm-head \
12
  --disable-custom-all-reduce \
13
  --mem-fraction-static 0.5