Skip to content

Commit 99fa9f1

Browse files
committed
feat: runtime CP<->DP mode switching for xLLM workers.
Add support for flipping worker layout between CP_PREFILL (cp=N, dp=1) and DP_DECODE (cp=1, dp=N) on a live instance in ~500 ms, without a service restart. Traffic patterns rarely stay in one regime forever; this lets a single deployment adapt across the day rather than being statically sized for either lane. Design ------ * Startup builds BOTH ATB graphs (dual-mode) when `--enable_runtime_cp_dp_switch=true`, so a flip is a graph handle swap, not a rebuild. * Flip flow: scheduler drains in-flight requests -> pause workers -> swap the active `ParallelArgs` (cp/dp) and block-manager pool atomically -> resume. The block-manager pool is guarded by a shared-ptr for lockless reads on the hot path. * Worker-side call sites read the live `parallel_args_.cp_size()` (not the immutable `options_.cp_size()`), so already-scheduled requests observe the new layout after resume. New surfaces ------------ * `mode_switch.proto` + `ModeSwitchService`: RPC surface for flipping, exposed via the xLLM server. * `AutoFlipController` scaffolding for autoscaling callers. * `docs/en/features/runtime_cp_dp_switch.md`: user-facing design doc. Testing ------- * Unit: `dual_parallel_args_test`, `atomic_pool_handle_test`, `lopsided_defer_gate_test`. * E2E on 8*A3: `verify_switch` regression, 18/18 flips stable across 61 model layers; lm_eval CP == DP within tolerance; 10-min soak on 32-proc load did not regress HBM watermark. Related issue: #1967
1 parent 52eca61 commit 99fa9f1

60 files changed

Lines changed: 2811 additions & 90 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.
Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,121 @@
1+
---
2+
title: "Runtime CP↔DP Mode Switching"
3+
sidebar:
4+
order: 91
5+
---
6+
## Background
7+
8+
xLLM supports two parallel layouts for a fixed world size:
9+
10+
- **CP_PREFILL** (`cp=N, dp=1`): context parallelism across all workers. Long prompts split their KV compute across workers, minimizing TTFT for large-input requests.
11+
- **DP_DECODE** (`cp=1, dp=N`): data parallelism, `N` independent decoder replicas. Higher concurrency and throughput for small requests, at the cost of no per-request work-splitting.
12+
13+
Traffic patterns rarely stay in one regime forever. Long-context ingest bursts favor CP; concurrent short-request traffic favors DP. Static provisioning per instance forces the operator to overprovision one lane or accept degraded metrics on the other.
14+
15+
**Runtime CP↔DP mode switching** lets an xLLM instance flip between the two layouts on demand, without a service restart, in ~500 ms.
16+
17+
## When to use it
18+
19+
- Traffic mix shifts across the day (long-context batch jobs at night, chat traffic during the day).
20+
- You want to test-drive both layouts on a single instance rather than maintain separate deployments.
21+
- You are building an autoscaling controller that decides layout per instance and needs a machine-actuated switch RPC.
22+
23+
Do **not** use it if:
24+
25+
- You are debugging a numerical accuracy problem — start with a fixed layout to eliminate flip as a variable.
26+
- Your workload is uniformly one shape — the resident cost of holding both ATB graphs (~600 MB / NPU + extra graph workspace) is not free.
27+
28+
## How it works
29+
30+
### Startup: dual-graph mode
31+
32+
When the worker starts with `--enable_runtime_cp_dp_switch=true`, it initializes both parallel layouts up front:
33+
34+
- Two `CollectiveCommunicator`s. One with `cp_size=N,dp_size=1`, one with `cp_size=1,dp_size=N`. Each brings up its own HCCL comm domain on a distinct master port (offset by `--dual_mode_port_stride`, default 256).
35+
- Two `ParallelArgs` snapshots wrapped in a `DualParallelArgs` with an `atomic<Mode>` discriminator. `active()` reads the current snapshot with acquire semantics.
36+
- Four ATB nodes per decoder layer: `cp_prefill`, `cp_decode`, `dp_prefill`, `dp_decode`. Weights are shared; only the graph binding differs.
37+
38+
The initial mode is chosen from `--cp_size` at startup (`cp>1 → CP_PREFILL`, else `DP_DECODE`).
39+
40+
### The switch: an RPC that atomically flips the layout
41+
42+
Callers invoke `xllm.proto.ModeSwitchService.SwitchMode` on the instance's `--disagg_pd_port` (both prefill and decode instances host the service):
43+
44+
```bash
45+
curl -sS -X POST \
46+
"http://<host>:<disagg_pd_port>/xllm.proto.ModeSwitchService/SwitchMode" \
47+
-d '{"target_mode":1,"timeout_ms":30000}'
48+
```
49+
50+
`target_mode` is `0` for CP_PREFILL, `1` for DP_DECODE. The response carries `{"ok":true,"current_mode":<0|1>}` on success.
51+
52+
**Serialize P then D.** The decode side rebuilds datadist links using the prefill side's advertised `dp_size` from etcd. If you flip both in parallel and D wins the race, D reads P's stale dp_size and links the wrong topology. Ad-hoc curl scripts should always flip the prefill instance first, wait for its response, then flip the decode instance.
53+
54+
### The orchestration steps
55+
56+
`ModeSwitchService::SwitchMode` on each instance runs, in order:
57+
58+
1. **`scheduler.pause(WAIT)`** — stop admitting new batches; existing in-flight steps finish.
59+
2. **`wait_until_paused(timeout_ms)`** — verify the drain completed; return an error to the caller if not (`{"error":"scheduler drain timeout"}`), leaving mode unchanged.
60+
3. **`begin_switch()`** — acquire a `std::shared_mutex` unique lock. Six read paths in `DisaggPDScheduler` (`dispatch_requests`, `try_allocate`, `decode_schedule`, `decode_recv_first_generation`, `link_instance`, `unlink_instance`) hold this shared; the unique lock waits them out.
61+
4. **`engine.switch_mode(target)`** — fan out `SwitchMode` RPC to every worker. Each worker flips its `DualParallelArgs::active_` atomic and refreshes its cached `parallel_args_` and `ModelContext`. The engine also updates its own `cp_size_/dp_size_/dp_local_size_` under the `paired = max(cp,dp)` invariant so subsequent scheduler polls see the new layout.
62+
5. **`rebuild_after_flip(new_dp_size)`** — reconstruct `BlockManagerPool` for the new dp_size. Reuses the old pool's `Options` so KV-cache-cap-derived config isn't recomputed. Publishes the new pool pointer to `XServiceClient` (release-store on `atomic<const Pool*>`), which the heartbeat / reconcile threads acquire-load once per iteration.
63+
6. **`re_register_dp_size(new)`** — write the new dp_size back to the etcd `InstanceInfo` so peers observe the post-flip topology when they reconcile.
64+
7. **`gate.unlock()` + `scheduler.resume()`** — release the write lock and start admitting new batches. Steady-state cost after this point is one atomic acquire-load per step.
65+
8. **`relink_after_flip`** — decode instances only. Unlink stale datadist connections and relink using the peer's fresh dp_size fetched from etcd. Runs *outside* the switch gate to avoid a deadlock between the datadist worker RPC handler and the caller thread that already holds the gate.
66+
67+
### Steady-state runtime concerns
68+
69+
**Lopsided-batch defer (DP mode).** DP forward requires every dp_rank to enter the collective with a shape-consistent input. If one dp_rank has a real batch and the other is empty, entering forward would either hang the peer (empty rank returns early) or race real KV blocks (empty rank fabricates a fake input). The scheduler defers such steps: if `active_dp_size > 1` and any dp_rank has an empty batch, `step()` returns without dispatching to the engine. A 100 ms backdoor forces the step if the imbalance persists — better to risk a fake-input path than to starve indefinitely. In practice the backdoor never fires (DP dispatch converges in <10 ms).
70+
71+
**Diagnostic logging.** The `enable_flip_verbose_log` gflag (default `false`) gates 12 per-request / per-step FLIPDIAG log sites. Flip lifecycle events (`switch_mode`, `rebuild_after_flip`, `relink_after_flip`, `lopsided_backdoor`, `EARLY_RETURN`) log unconditionally. Toggle at runtime via brpc's built-in endpoint (a `DEFINE_validator` marks the flag reloadable):
72+
73+
```bash
74+
curl "http://<host>:<disagg_pd_port>/flags/enable_flip_verbose_log?setvalue=true"
75+
```
76+
77+
## Failure modes and root causes
78+
79+
Six layers of concurrency issues surfaced during development. Each is documented here so future regressions have a starting point:
80+
81+
| # | Symptom | Root cause | Fix |
82+
|---|---------|------------|-----|
83+
| 1 | rank0 zombies during rebuild | dispatch handlers race `unique_ptr::reset` on `kv_cache_manager_` | `switch_gate_` unique_lock + pause/drain gate |
84+
| 2 | rank0 dies ~5 s after every flip | `XServiceClient` heartbeat holds a raw pointer to the old (destroyed) pool | `std::atomic<const BlockManagerPool*>` + release/acquire; heartbeat pins one snapshot per iteration |
85+
| 3 | flip-back CHECK failure `num_free_blocks_==free_blocks_.size()-1` | `BlockManagerImpl` destructor CHECK too strict; sequences release out of the original invariant window | Downgrade CHECK → WARNING |
86+
| 4 | `PushKvBlocks failed, ret=5010b007` (LLM_NOT_YET_LINK) | Datadist link topology stale after a peer's `dp_size` change | `re_register_dp_size` + `relink_after_flip` (P-side skip; D-side fetches fresh `InstanceInfo`; serialized P → D at the caller) |
87+
| 5 | DP forward hangs on `batch_sizes=[N,0]` | `dp_global_token_nums` says 0 for empty rank while worker's fake-input path made it 1; DP collective sizes disagree | Clamp `dp_global_token_nums[dp_rank] = max(1, ...)` in `LLMEngine::prepare_inputs` when `dp_size>1` |
88+
| 6 | DP forward still occasionally hangs on lopsided batches | ATB attention op hangs when one shard uses real block_tables and another the placeholder path | Scheduler-layer defer: skip lopsided batches, 100 ms backdoor as safety valve |
89+
90+
## Verification
91+
92+
The reference `verify_switch.sh` regression exercises the following pattern:
93+
94+
1. HTTP probe to confirm the service is up.
95+
2. Reset both P and D to CP mode as a preflight.
96+
3. Warmup request.
97+
4. **CP burst** (default 6 concurrent, `max_tokens=6`).
98+
5. Idle wait until both P and D report `Running requests: 0`.
99+
6. **Flip 0 → 1** (serialized P then D). Print each side's latency in ms.
100+
7. Idle wait, **DP burst**.
101+
8. Idle wait, **flip 1 → 0**.
102+
9. Idle wait, **CP-post burst**.
103+
104+
Success criteria: 18/18 responses across the three bursts, both flips complete without `drain timeout`, no `main process disappeared` in either instance's rank0 log.
105+
106+
## Configuration reference
107+
108+
| Flag | Default | Purpose |
109+
|------|---------|---------|
110+
| `--enable_runtime_cp_dp_switch` | `false` | Bring up two `CollectiveCommunicator`s + four ATB nodes at startup. Required for flip to work. |
111+
| `--dual_mode_port_stride` | `256` | Port offset between the CP and DP master ports. 256 avoids TIME_WAIT collisions on rapid restarts. |
112+
| `--enable_flip_verbose_log` | `false` | Emit per-request/per-step FLIPDIAG lines. Runtime-toggleable via brpc `/flags` endpoint. |
113+
| `--disagg_pd_port` | `7777` (non-disagg) / instance-specific (disagg) | Port `ModeSwitchService` listens on. |
114+
115+
## Not yet covered
116+
117+
- Real 61-layer DeepSeek-V3.2-w8a8 model validation on multi-host disagg (single-host validated).
118+
- Cross-machine disagg flip.
119+
- `lm_eval` numerical parity sweep between CP and DP with a shared seed.
120+
- Long-run stability (>1 h with periodic flips), including memory / connection leak checks.
121+
- Automatic mode-switch controller (framework scaffolding exists in this PR; wiring to real traffic signals is a follow-on PR).
Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,121 @@
1+
---
2+
title: "运行时 CP↔DP 模式切换"
3+
sidebar:
4+
order: 91
5+
---
6+
## 背景
7+
8+
在固定 world size 下,xLLM 支持两种并行布局:
9+
10+
- **CP_PREFILL** (`cp=N, dp=1`):全体 worker 上跑 context parallelism。长 prompt 的 KV 计算被拆到多个 worker 上,能显著降低大输入请求的 TTFT。
11+
- **DP_DECODE** (`cp=1, dp=N`):data parallelism,`N` 个独立 decoder 副本。小请求下并发和吞吐更高,代价是不再对单请求做工作切分。
12+
13+
真实流量的形态很少长期停留在同一种。长上下文的批量灌入场景更适合 CP;并发的短请求场景更适合 DP。按实例静态配置会迫使运维要么给一条通道超配,要么在另一条通道上接受指标退化。
14+
15+
**运行时 CP↔DP 模式切换** 允许一个 xLLM 实例在不重启服务的前提下,在两种布局之间按需切换,切换耗时约 500 ms。
16+
17+
## 什么时候用
18+
19+
- 一天内流量形态会切换(例如夜间跑长上下文批量任务,白天跑对话流量)。
20+
- 想在同一个实例上验证两种布局,而不是维护两套部署。
21+
- 你在开发一个自动扩缩控制器,需要按实例决策布局,并调用一个可以由机器发起的 switch RPC。
22+
23+
**不建议**在以下场景使用:
24+
25+
- 你在排查数值精度问题——先固定为一种布局,把 flip 当作变量排除掉。
26+
- 你的 workload 长期只有一种形态——同时驻留两张 ATB 图(约 600 MB / NPU + 额外 graph workspace)并非免费。
27+
28+
## 工作原理
29+
30+
### 启动:双图模式
31+
32+
worker 以 `--enable_runtime_cp_dp_switch=true` 启动时,会预先初始化两种并行布局:
33+
34+
- 两个 `CollectiveCommunicator`。一个 `cp_size=N,dp_size=1`,一个 `cp_size=1,dp_size=N`。每个都在独立的 master port 上拉起自己的 HCCL 通信域(端口偏移由 `--dual_mode_port_stride` 控制,默认 256)。
35+
- 两份 `ParallelArgs` 快照被 `DualParallelArgs` 包住,配一个 `atomic<Mode>` 判别器。`active()` 以 acquire 语义读取当前快照。
36+
- 每个 decoder layer 有 4 个 ATB 节点:`cp_prefill``cp_decode``dp_prefill``dp_decode`。权重共享,只是 graph 绑定不同。
37+
38+
初始模式由启动时的 `--cp_size` 决定(`cp>1 → CP_PREFILL`,否则 `DP_DECODE`)。
39+
40+
### 切换:通过 RPC 原子地翻转布局
41+
42+
调用方在实例的 `--disagg_pd_port` 上调用 `xllm.proto.ModeSwitchService.SwitchMode`(prefill 和 decode 实例都会承载这个服务):
43+
44+
```bash
45+
curl -sS -X POST \
46+
"http://<host>:<disagg_pd_port>/xllm.proto.ModeSwitchService/SwitchMode" \
47+
-d '{"target_mode":1,"timeout_ms":30000}'
48+
```
49+
50+
`target_mode``0` 表示 CP_PREFILL,`1` 表示 DP_DECODE。成功时响应体是 `{"ok":true,"current_mode":<0|1>}`
51+
52+
**必须先 P 再 D。** decode 侧会根据 etcd 里 prefill 侧广播的 `dp_size` 重建 datadist 链路。如果两侧并行 flip 且 D 抢先,D 读到的是 P 的旧 `dp_size`,会连到错误的拓扑。任何脚本都应确保先切完 prefill、等到响应,再切 decode。
53+
54+
### 编排流程
55+
56+
`ModeSwitchService::SwitchMode` 在每个实例上按以下顺序执行:
57+
58+
1. **`scheduler.pause(WAIT)`** —— 停止接收新 batch;已经在飞的 step 跑完。
59+
2. **`wait_until_paused(timeout_ms)`** —— 确认已经 drain 干净;若未 drain 干净则返回 `{"error":"scheduler drain timeout"}` 给调用方,模式保持不变。
60+
3. **`begin_switch()`** —— 拿 `std::shared_mutex` 的 unique lock。`DisaggPDScheduler` 中的 6 条读路径(`dispatch_requests``try_allocate``decode_schedule``decode_recv_first_generation``link_instance``unlink_instance`)持有 shared lock;unique lock 等它们全部释放。
61+
4. **`engine.switch_mode(target)`** —— 向所有 worker 广播 `SwitchMode` RPC。每个 worker 翻转自己的 `DualParallelArgs::active_` atomic,并刷新缓存的 `parallel_args_``ModelContext`。engine 也在 `paired = max(cp,dp)` 不变量下更新自己的 `cp_size_/dp_size_/dp_local_size_`,使后续 scheduler 轮询看到新布局。
62+
5. **`rebuild_after_flip(new_dp_size)`** —— 为新的 dp_size 重建 `BlockManagerPool`。复用旧 pool 的 `Options`,避免重新根据 KV cache cap 推导配置。把新 pool 指针以 release-store 发布给 `XServiceClient``atomic<const Pool*>`),心跳/reconcile 线程每轮以 acquire-load 拿一次快照。
63+
6. **`re_register_dp_size(new)`** —— 把新 dp_size 写回 etcd 的 `InstanceInfo`,让对端 reconcile 时观察到 flip 后的拓扑。
64+
7. **`gate.unlock()` + `scheduler.resume()`** —— 释放写锁并重新接收 batch。稳态代价:每个 step 一次原子 acquire-load。
65+
8. **`relink_after_flip`** —— 仅 decode 侧。unlink 旧的 datadist 链路,用刚从 etcd 拉到的 peer 侧 dp_size 重新 link。**这一步在 switch gate 之外执行**,否则会在 datadist worker RPC handler 和已经持有 gate 的调用线程之间产生死锁。
66+
67+
### 稳态运行时需要注意的问题
68+
69+
**Lopsided batch 延迟(DP 模式)。** DP forward 要求每个 dp_rank 以形状一致的输入进入 collective。如果一个 dp_rank 有真实 batch 而另一个是空的,直接进 forward 要么让空 rank 的对端 hang(空 rank 提前返回),要么让空 rank 伪造 fake input 从而竞争真的 KV block。scheduler 会延迟这类 step:若 `active_dp_size > 1` 且任一 dp_rank 是空 batch,`step()` 直接返回,不下发到 engine。100 ms 的 backdoor 会在长时间不平衡时强制放行一次—— 宁可走 fake-input 路径也不能无限饿死。实测 backdoor 几乎不触发(DP dispatch 一般 10 ms 内自动收敛)。
70+
71+
**诊断日志。** `enable_flip_verbose_log` gflag(默认 `false`)门控 12 处 per-request / per-step 的 FLIPDIAG 日志。flip 生命周期事件(`switch_mode``rebuild_after_flip``relink_after_flip``lopsided_backdoor``EARLY_RETURN`)无条件打印。可通过 brpc 内建端点在运行时切换(该 flag 用 `DEFINE_validator` 标记为可 reload):
72+
73+
```bash
74+
curl "http://<host>:<disagg_pd_port>/flags/enable_flip_verbose_log?setvalue=true"
75+
```
76+
77+
## 失效模式与根因
78+
79+
开发过程中暴露了 6 层并发问题,记录在此以便后续回归有起点:
80+
81+
| # | 现象 | 根因 | 修复 |
82+
|---|------|------|------|
83+
| 1 | rebuild 期间 rank0 变僵尸 | dispatch handler 与 `kv_cache_manager_` 上的 `unique_ptr::reset` 竞态 | `switch_gate_` 的 unique_lock + pause/drain gate |
84+
| 2 | 每次 flip 后约 5s rank0 死掉 | `XServiceClient` 心跳持有旧 pool 的裸指针(对象已销毁) | `std::atomic<const BlockManagerPool*>` + release/acquire;心跳每轮 pin 一份快照 |
85+
| 3 | flip 回来时 CHECK 失败 `num_free_blocks_==free_blocks_.size()-1` | `BlockManagerImpl` 析构 CHECK 太严;序列在原不变量窗口之外释放 | 把 CHECK 降级为 WARNING |
86+
| 4 | `PushKvBlocks failed, ret=5010b007`(LLM_NOT_YET_LINK) | peer 的 `dp_size` 改了以后 datadist 链路拓扑变陈旧 | `re_register_dp_size` + `relink_after_flip`(P 侧跳过;D 侧拉最新 `InstanceInfo`;调用方保证 P→D 串行) |
87+
| 5 | DP forward 在 `batch_sizes=[N,0]` 时 hang | `dp_global_token_nums` 上空 rank 显示 0,而 worker 的 fake-input 路径把它写成 1;DP collective 尺寸不一致 |`LLMEngine::prepare_inputs` 里,当 `dp_size>1` 时做 `dp_global_token_nums[dp_rank] = max(1, ...)` clamp |
88+
| 6 | DP forward 在 lopsided batch 上偶尔仍 hang | ATB attention op 在一个 shard 用真实 block_tables、另一个 shard 走 placeholder 路径时 hang | scheduler 层延迟:跳过 lopsided batch,100 ms backdoor 作为安全阀 |
89+
90+
## 验证
91+
92+
参考的 `verify_switch.sh` 回归脚本运行如下流程:
93+
94+
1. HTTP 探测确认服务已启动。
95+
2. 把 P 和 D 都重置到 CP 模式,作为 preflight。
96+
3. Warmup 请求。
97+
4. **CP burst**(默认 6 并发,`max_tokens=6`)。
98+
5. 空闲等待到 P 和 D 都报 `Running requests: 0`
99+
6. **Flip 0 → 1**(串行 P 后 D),打印两侧延迟(ms)。
100+
7. 空闲等待,跑 **DP burst**
101+
8. 空闲等待,**flip 1 → 0**
102+
9. 空闲等待,跑 **CP-post burst**
103+
104+
成功标准:3 轮 burst 累计 18/18 响应;两次 flip 都不出现 `drain timeout`;两个实例的 rank0 日志中都不出现 `main process disappeared`
105+
106+
## 配置参考
107+
108+
| Flag | 默认值 | 用途 |
109+
|------|--------|------|
110+
| `--enable_runtime_cp_dp_switch` | `false` | 启动时拉起两个 `CollectiveCommunicator` + 4 个 ATB 节点。flip 依赖此项。 |
111+
| `--dual_mode_port_stride` | `256` | CP 和 DP master port 之间的偏移。256 是为了避免快速重启时 TIME_WAIT 端口碰撞。 |
112+
| `--enable_flip_verbose_log` | `false` | 打印 per-request / per-step 的 FLIPDIAG 行。可通过 brpc `/flags` 端点在运行时切换。 |
113+
| `--disagg_pd_port` | 非 disagg `7777` / disagg 实例侧配置 | `ModeSwitchService` 的监听端口。 |
114+
115+
## 暂未覆盖
116+
117+
- 61 层真实 DeepSeek-V3.2-w8a8 模型在跨机 disagg 下的验证(单机 disagg 已验证)。
118+
- 跨机 disagg 的 flip。
119+
- CP 与 DP 在相同 seed 下的 `lm_eval` 数值对齐扫描。
120+
- 长时间稳定性(>1 h 周期性 flip),包含内存 / 连接泄漏检查。
121+
- 自动 mode-switch 控制器(本 PR 引入了 scaffolding;接入真实流量信号在后续 PR)。

tests/core/framework/parallel_state/CMakeLists.txt

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,14 @@
11
include(cc_test)
22

3+
# dual_parallel_args_test verifies the atomic Mode discriminator on
4+
# DualParallelArgs. Kept as a source file for reference / hand-run under a
5+
# CPU environment; not wired into CMake because parallel_args.h under
6+
# USE_NPU transitively includes xllm_atb_layers/atb which needs the full
7+
# ATB toolchain, and stubbing all of that just for a header-only test
8+
# yields more surface area than value. The lopsided_defer_gate_test
9+
# (tests/core/scheduler/) and atomic_pool_handle_test (tests/core/runtime/)
10+
# cover the runtime CP<->DP switch's higher-value invariants.
11+
312
if(USE_NPU)
413
cc_test(
514
NAME

0 commit comments

Comments
 (0)