已合并
[feature] 支持引擎原生拉起: layerwise CDP调度模式适配 #773
[feature] 支持引擎原生拉起: layerwise CDP调度模式适配 #773
已合并
tobking创建于 8 天前
30 个文件变更+1535-71
M.agent/skills/motor-dev/references/coordinator.md+26-2
@@ -159,6 +159,30 @@ no routable topology at all → HTTP 503
159- `PDHybridRouter` (strategies/pd_hybrid.py): single instance runs prefill+decode together; also the degradation target when PD separation is unavailable (e.g., P/D instances circuit-broken or advertising no shared dispatch).159- `PDHybridRouter` (strategies/pd_hybrid.py): single instance runs prefill+decode together; also the degradation target when PD separation is unavailable (e.g., P/D instances circuit-broken or advertising no shared dispatch).
160- Both subclass `BaseRouter` (strategies/base.py); `_is_pd_hybrid_deploy` / `_is_pd_separation_fallback_to_hybrid_enabled` gates fallback (config `scheduler_config.enable_pd_separation_fallback_to_hybrid`, default true).160- Both subclass `BaseRouter` (strategies/base.py); `_is_pd_hybrid_deploy` / `_is_pd_separation_fallback_to_hybrid_enabled` gates fallback (config `scheduler_config.enable_pd_separation_fallback_to_hybrid`, default true).
161 161 
162+**vLLM P/D coordination modes** (`UnifiedPDRouter`):
163+ 
164+| Instance `dispatch_capabilities` | Mode | Order |
165+|---|---|---|
166+| homogeneous `prefill_handoff_decode` | HANDOFF | allocate P → prefill → allocate D → decode |
167+| homogeneous `concurrent_engine_sync` (vLLM layerwise / `dispatch_profile=trigger`) | TRIGGER | allocate D first → decode with `do_remote_prefill` + `metaserver` → D POSTs Worker `/v1/metaserver` → same Worker allocates P and forwards prefill |
168+| mixed handoff + trigger in one cluster | — | HTTP 503 |
169+ 
170+Mode selection uses allocated-instance `dispatch_capabilities` **and** cluster detection from the Worker-local instance cache (`get_local_instances`). That cache already holds `dispatch_capabilities`; SHM only has workload numbers. `GET_AVAILABLE_INSTANCES` remains a force-refresh RPC and must not run on every request. `ALLOCATE_ONLY` responses go through `_serialize_instance_minimal`, which must keep `dispatch_capabilities` (not only `id/role/job_name/model_name/engine_type`); otherwise Worker rebuilds empty caps and falls back to adapter HANDOFF while still allocating Decode first. If the selected mode is TRIGGER but the attempt has no decode resource (handoff-style P-first / D-deferred), fail closed with HTTP 503 — do not return TRIGGER and then `RuntimeError` into retry→500.
171+ 
172+SGLang stays on native bootstrap (`CoordinationMode.BOOTSTRAP`); that path is unchanged.
173+ 
174+**Trigger metaserver (per Worker, not on the infer port):**
175+ 
176+- `RequestInfo` is process-local. Infer workers share `coordinator_api_infer_port` via `SO_REUSEPORT`, so Decode's metaserver callback cannot land on the infer socket.
177+- `inference_workers_config.worker_metaserver_base_port` default **12000**. Worker `i` listens on `base+i`; set to `0` to disable.
178+- Dedicated uvicorn app (`InferenceServer.create_metaserver_app()`) exposes only `POST /v1/metaserver` — no API key, no infer TLS (`lifespan=off`). Default API-key / rate-limit skip sets include `/v1/metaserver`. Decode engine callbacks have no API key; do not require one on this socket. Infer is the primary uvicorn; metaserver is a sidecar. Bind/init/`serve()` failure logs ERROR, clears this process's `worker_metaserver_port`, and leaves the infer port running. Trigger requests then 503 via `_ensure_trigger_metaserver`. Infer exit sets `should_exit` and cancels the sidecar.
179+- The metaserver listen host prefers `POD_IP` when set, otherwise `api_config.coordinator_api_host` (same fallback as the advertised callback URL). Do not bind loopback: Decode may run on another node. Infer uvicorn still listens on `coordinator_api_host`.
180+- The callback URL advertises `POD_IP` when available, otherwise `api_config.coordinator_api_host`; IPv6 literals are RFC 3986 bracketed. `0.0.0.0`/`::` remain valid listen hosts at startup (including default `worker_metaserver_base_port=12000`). Trigger rejects them as advertised callback addresses when `POD_IP` is absent (HTTP 503 + error log), because wildcard listen addresses are not routable Decode callback destinations.
181+- Callback `request_id` is trimmed (`chatcmpl-` / `cmpl-…-0`) then looked up in that Worker's `RequestManager`. Query `?attempt=` must match the bound attempt (404 unknown request, 409 stale attempt).
182+- Each trigger attempt serializes callbacks with `AttemptContext.trigger_lock`. The active callback is registered as the attempt's Prefill task so disconnect/Decode failure during TTFT cancels it; a retry after Prefill completion returns idempotent success without allocating P again.
183+- If Scheduler allocation succeeds but Worker-local attempt workload registration fails, the allocation is rolled back directly with the returned workload delta.
184+- Runtime field `CoordinatorConfig.worker_metaserver_port` is per-process (`base+worker_index`) and is in the hot-reload skip-set.
185+ 
162**Request lifecycle:**186**Request lifecycle:**
163 187 
1641. `prepare_resource(plan)` — scheduling policy selects best instance → allocates workload slot1881. `prepare_resource(plan)` — scheduling policy selects best instance → allocates workload slot
@@ -191,14 +215,14 @@ Hot-reload is driven by a `ConfigWatcher` in the **Mgmt process** (not the daemo
191| `motor/coordinator/scheduler/runtime/workload_shm/` | | SHM layout (`layout.py`) + reader/writer |215| `motor/coordinator/scheduler/runtime/workload_shm/` | | SHM layout (`layout.py`) + reader/writer |
192| `motor/coordinator/domain/instance_manager.py` | | Central instance pool (available/unavailable, per-role sub-pools) |216| `motor/coordinator/domain/instance_manager.py` | | Central instance pool (available/unavailable, per-role sub-pools) |
193| `motor/coordinator/domain/request_manager.py` | | Request ID generation, workload tracking per request |217| `motor/coordinator/domain/request_manager.py` | | Request ID generation, workload tracking per request |
194-| `motor/coordinator/router/dispatch.py` | | `select_router_class` (dynamic router selection from live topology) + `handle_request` |218+| `motor/coordinator/router/dispatch.py` | | `select_router_class` (dynamic router selection from live topology) + `handle_request` + `handle_metaserver_request` |
195| `motor/coordinator/router/strategies/` | | `BaseRouter` + `PDHybridRouter` + `UnifiedPDRouter` implementations |219| `motor/coordinator/router/strategies/` | | `BaseRouter` + `PDHybridRouter` + `UnifiedPDRouter` implementations |
196| `motor/coordinator/router/dispatch_session.py` | | Dispatch attempt session/state tracking |220| `motor/coordinator/router/dispatch_session.py` | | Dispatch attempt session/state tracking |
197| `motor/coordinator/router/rescheduler/` | | `Rescheduler` (retry plans for failed requests) |221| `motor/coordinator/router/rescheduler/` | | `Rescheduler` (retry plans for failed requests) |
198| `motor/coordinator/api_client/` | | `ConductorApiClient` / `ControllerApiClient` / `NativeEngineApiClient` (HTTP clients to kv-conductor, controller, engine) |222| `motor/coordinator/api_client/` | | `ConductorApiClient` / `ControllerApiClient` / `NativeEngineApiClient` (HTTP clients to kv-conductor, controller, engine) |
199| `motor/coordinator/api_server/management_server.py` | | Mgmt: `/liveness`, `/readiness`, `/instances/refresh`, `/precision/alarm_cleared` |223| `motor/coordinator/api_server/management_server.py` | | Mgmt: `/liveness`, `/readiness`, `/instances/refresh`, `/precision/alarm_cleared` |
200| `motor/coordinator/api_server/observability_server.py` | | Obs: `/metrics`, `/health` (`/instance/metrics` deprecated → `GET /metrics?type=instance`) |224| `motor/coordinator/api_server/observability_server.py` | | Obs: `/metrics`, `/health` (`/instance/metrics` deprecated → `GET /metrics?type=instance`) |
201-| `motor/coordinator/api_server/inference_server.py` | | Infer: `/v1/completions`, `/v1/chat/completions`, `/v1/models`, `/v1/messages` + `/v1/messages/count_tokens` (Anthropic) |225+| `motor/coordinator/api_server/inference_server.py` | | Infer: `/v1/completions`, `/v1/chat/completions`, `/v1/models`, `/v1/messages` + `/v1/messages/count_tokens` (Anthropic); dedicated metaserver app `POST /v1/metaserver` |
202| `motor/coordinator/scheduler/runtime/scheduler_connection_manager.py` | | Shared Scheduler ZMQ connection (used by Mgmt/Obs/Infer) |226| `motor/coordinator/scheduler/runtime/scheduler_connection_manager.py` | | Shared Scheduler ZMQ connection (used by Mgmt/Obs/Infer) |
203| `motor/coordinator/domain/circuit_breaker.py` | | Per-instance circuit breaker state (closed/open) |227| `motor/coordinator/domain/circuit_breaker.py` | | Per-instance circuit breaker state (closed/open) |
204| `motor/coordinator/domain/scheduling_pin.py` | | Pinned-instance resolution, endpoint selection for an instance |228| `motor/coordinator/domain/scheduling_pin.py` | | Pinned-instance resolution, endpoint selection for an instance |
Mdocs/zh/design/pd_disaggregation.md+23-3
@@ -41,16 +41,18 @@ vLLM 引擎按 `kv_connector` 名称(大小写不敏感)推导 capability,
41- **`MultiConnector` 只看 `connectors[0]`(传输层)**。KV 池/存储类连接器(如 `AscendStoreConnector``MooncakeConnectorStoreV1``UCMConnector``LMCacheAscendConnector`)一般作为 `connectors[1]` 的后端使用,不参与 capability 判定,因此**无需**出现在白名单中。41- **`MultiConnector` 只看 `connectors[0]`(传输层)**。KV 池/存储类连接器(如 `AscendStoreConnector``MooncakeConnectorStoreV1``UCMConnector``LMCacheAscendConnector`)一般作为 `connectors[1]` 的后端使用,不参与 capability 判定,因此**无需**出现在白名单中。
42- 不在上表内、且 `connectors[0]` 也无法识别的连接器会被判为 `unknown`**不产生任何 capability**42- 不在上表内、且 `connectors[0]` 也无法识别的连接器会被判为 `unknown`**不产生任何 capability**
43 43 
44-> ⚠️ **fail-closed**:原生 vLLM P/D 启动只接受 handoff 语义;未知或不兼容 Connector 会在 NodeManager 构造启动命令时失败,避免把错误推迟到 KV 传输阶段。44+> ⚠️ **fail-closed**:原生 vLLM P/D 启动只接受 `handoff` 或 `trigger` 语义;未知或不兼容 Connector 会在 NodeManager 构造启动命令时失败,避免把错误推迟到 KV 传输阶段。
45 45 
46`dispatch_capabilities` 是 NodeManager 上报的兼容元数据,不支持在用户配置中直接填写。若需让**未被识别的连接器**作为 P/D 传输使用,请在 `motor_engine_prefill_config` / `motor_engine_decode_config` **顶层**(与 `engine_type` 同级,**不是** `engine_config` 内部)显式声明 `dispatch_profile`46`dispatch_capabilities` 是 NodeManager 上报的兼容元数据,不支持在用户配置中直接填写。若需让**未被识别的连接器**作为 P/D 传输使用,请在 `motor_engine_prefill_config` / `motor_engine_decode_config` **顶层**(与 `engine_type` 同级,**不是** `engine_config` 内部)显式声明 `dispatch_profile`
47 47 
48| `dispatch_profile` | 推导出的 capability | 协同行为 |48| `dispatch_profile` | 推导出的 capability | 协同行为 |
49|--------------------|---------------------|----------|49|--------------------|---------------------|----------|
50| `handoff` | `prefill_handoff_decode` | Prefill 完成后交给 Decode |50| `handoff` | `prefill_handoff_decode` | Prefill 完成后交给 Decode |
51-| `trigger` | `concurrent_engine_sync` | P/D 并发执行引擎同步 KV |51+| `trigger` | `concurrent_engine_sync` | Decode 先启动经 Worker metaserver 触发 Prefill;引擎按层同步 KV |
52 52 
53-当前原生 vLLM P/D 运行只接受 `handoff``trigger` 仅作兼容分类保留,不可用于原生 vLLM P/D 启动53+原生 vLLM P/D 支持 `handoff``trigger`(`MooncakeLayerwiseConnector` 推导 `trigger`)。同一集群内 **handoff 与 trigger 实例不可混部**,Coordinator 会返回 503。SGLang 仍使用自身 bootstrap 协议
54+ 
55+**Layerwise / trigger:** `motor_coordinator_config.inference_workers_config.worker_metaserver_base_port` 默认 `12000`。每个 Inference Worker 在独立端口 `base+worker_index` 上监听 `POST /v1/metaserver`(不与推理口 `SO_REUSEPORT` 共用)。设为 `0` 可关闭。Decode 引擎回调该地址时不携带 API Key,也不走推理面 TLS。metaserver 监听地址与 callback 广告地址一致:优先 `POD_IP`,否则用 `coordinator_api_host`(不绑 loopback,以便跨节点 Decode 回调)。`0.0.0.0`/`::` 可作为推理口监听地址启动,走 Trigger 时若没有可达广告地址则该请求 503。metaserver 端口冲突或启动失败只禁用该 Worker 的 Trigger(请求 503),不拖垮推理口。集群检测读 Worker 本地实例缓存(含 `dispatch_capabilities`),不在每个请求上走 `GET_AVAILABLE_INSTANCES`。若判定为 Trigger 但当前 attempt 没有 Decode 实例,返回 503。
54 56 
55**配置示例**(自定义 connector 不在白名单内时):57**配置示例**(自定义 connector 不在白名单内时):
56 58 
@@ -101,3 +103,21 @@ flowchart LR
101 Adapter --> EngineD[Decode instance]103 Adapter --> EngineD[Decode instance]
102 Hybrid --> EngineU[Union or fallback Prefill instance]104 Hybrid --> EngineU[Union or fallback Prefill instance]
103```105```
106+ 
107+vLLM **handoff**:Coordinator 先调度 Prefill,完成后再调度 Decode。
108+ 
109+vLLM **trigger / layerwise**`MooncakeLayerwiseConnector`):
110+ 
111+```mermaid
112+sequenceDiagram
113+ participant C as Client
114+ participant W as Inference Worker
115+ participant D as Decode
116+ participant P as Prefill
117+ C->>W: POST /v1/chat/completions
118+ W->>D: decode (do_remote_prefill, metaserver)
119+ D->>W: POST /v1/metaserver?attempt=N
120+ W->>P: prefill (remote_block_ids/host/port)
121+ P-->>D: layerwise KV
122+ D-->>C: tokens
123+```
Mdocs/zh/developer_guide/components/coordinator.md+3-2
@@ -28,7 +28,7 @@ Coordinator 进程入口为 `motor/coordinator/main.py`:异步 `main()` 中构
28 28 
29子进程由 `SubprocessSupervisor` 监控;Daemon 主循环中处理信号与退出(见 `CoordinatorDaemon.run` 后半部分)。29子进程由 `SubprocessSupervisor` 监控;Daemon 主循环中处理信号与退出(见 `CoordinatorDaemon.run` 后半部分)。
30 30 
31-推理面 OpenAI 兼容路径与 metaserver 行为见 [服务接口](../../user_guide/api/service_interfaces.md)。31+推理面 OpenAI 兼容路径见 [服务接口](../../user_guide/api/service_interfaces.md)。vLLM layerwise metaserver 行为见 [PD 分离](../../design/pd_disaggregation.md)。
32 32 
33### 路由前 Token 处理33### 路由前 Token 处理
34 34 
@@ -50,7 +50,8 @@ KV Cache 亲和调度复用。Chat 请求带有需要服务端管理的 `agent_h
50请以 [配置参考](../../user_guide/configuration/config_reference.md) 中 **`motor_coordinator_config`** 章节为权威字段说明。代码中与 Daemon 强相关的包括:50请以 [配置参考](../../user_guide/configuration/config_reference.md) 中 **`motor_coordinator_config`** 章节为权威字段说明。代码中与 Daemon 强相关的包括:
51 51 
52- `standby_config.enable_master_standby`:是否走主备与 Infer 启停分支。52- `standby_config.enable_master_standby`:是否走主备与 Infer 启停分支。
53-- `scheduler_config`:`deploy_mode`、`scheduler_type` 等,影响推理由(见 [PD 分离](../../design/pd_disaggregation.md))。53+- `scheduler_config`:`scheduler_type` 等推理 Router 当前实例角色与 `dispatch_capabilities` 动态选择,不再读取 `deploy_mode`(见 [PD 分离](../../design/pd_disaggregation.md))。
54+- `inference_workers_config.worker_metaserver_base_port`:vLLM layerwise/trigger 时每 Worker 独立 metaserver 端口;默认 `12000`,设为 `0` 关闭。监听地址优先 `POD_IP`,否则用 `coordinator_api_host`。端口冲突时推理口继续,Trigger 返回 503。
54- `api_config`:推理端口、管理端口等(与 `interface_description.md` 一致处为准)。55- `api_config`:推理端口、管理端口等(与 `interface_description.md` 一致处为准)。
55 56 
56## 使用样例57## 使用样例
Mdocs/zh/user_guide/configuration/config_reference.md+9-5
@@ -246,7 +246,8 @@ motor_coordinator_config字段配置样例如下所示:
246 }246 }
247 },247 },
248 "inference_workers_config": {248 "inference_workers_config": {
249- "num_workers": 4249+ "num_workers": 4,
250+ "worker_metaserver_base_port": 12000
250 },251 },
251 "timeout_config": {252 "timeout_config": {
252 "request_timeout": 30,253 "request_timeout": 30,
@@ -270,7 +271,8 @@ motor_coordinator_config字段配置样例如下所示:
270 "/openapi.json",271 "/openapi.json",
271 "/readiness",272 "/readiness",
272 "/redoc",273 "/redoc",
273- "/startup"274+ "/startup",
275+ "/v1/metaserver"
274 ],276 ],
275 "encryption_algorithm": "PBKDF2_SHA256"277 "encryption_algorithm": "PBKDF2_SHA256"
276 },278 },
@@ -288,7 +290,8 @@ motor_coordinator_config字段配置样例如下所示:
288 "/openapi.json",290 "/openapi.json",
289 "/readiness",291 "/readiness",
290 "/redoc",292 "/redoc",
291- "/startup"293+ "/startup",
294+ "/v1/metaserver"
292 ],295 ],
293 "error_message": "too many requests, please try again later",296 "error_message": "too many requests, please try again later",
294 "error_status_code": 429,297 "error_status_code": 429,
@@ -404,6 +407,7 @@ motor_coordinator_config字段配置样例如下所示:
404| w_disk | float | 互斥 Disk 命中块权重。默认值:`0.0` |407| w_disk | float | 互斥 Disk 命中块权重。默认值:`0.0` |
405| **inference_workers_config字段** |-|-|408| **inference_workers_config字段** |-|-|
406| num_workers | int | Coordinator中业务面worker个数,默认值:4。 |409| num_workers | int | Coordinator中业务面worker个数,默认值:4。 |
410+| worker_metaserver_base_port | int | vLLM layerwise/trigger PD 时每个 Inference Worker 的 metaserver 起始端口。默认值:`12000`。Worker `i` 监听 `base+i`,仅暴露 `POST /v1/metaserver`。设为 `0` 关闭。须保证 `base+num_workers-1 <= 65535`。同一集群不可混部 handoff 与 trigger。监听地址优先 `POD_IP`,否则用 `coordinator_api_host`(不绑 loopback)。`coordinator_api_host=0.0.0.0`/`::` 仍可启动;走 Trigger 时须有 `POD_IP` 或可达的 `coordinator_api_host`,否则该请求返回 503。端口占用或 metaserver 启动失败时推理口继续服务,该 Worker 的 Trigger 请求返回 503。 |
407| **timeout_config字段** |-|-|411| **timeout_config字段** |-|-|
408| request_timeout | int | 单次 HTTP 请求超时时间(秒)。默认值:`30` |412| request_timeout | int | 单次 HTTP 请求超时时间(秒)。默认值:`30` |
409| connection_timeout | int | 建立连接的超时时间(秒)。默认值:`10` |413| connection_timeout | int | 建立连接的超时时间(秒)。默认值:`10` |
@@ -415,7 +419,7 @@ motor_coordinator_config字段配置样例如下所示:
415| valid_keys | array | 合法的 API Key 字符串列表。默认值:`[]` |419| valid_keys | array | 合法的 API Key 字符串列表。默认值:`[]` |
416| header_name | string | 携带 API Key 的 HTTP 头名称。默认值:`Authorization` |420| header_name | string | 携带 API Key 的 HTTP 头名称。默认值:`Authorization` |
417| key_prefix | string | 头中 Key 的前缀,如`Bearer`。默认值:`Bearer`|421| key_prefix | string | 头中 Key 的前缀,如`Bearer`。默认值:`Bearer`|
418-| skip_paths | array | 不校验 API Key 的路径列表(如 `/metrics`、`/liveness`、`/docs` 等),可自定义 |422+| skip_paths | array | 不校验 API Key 的路径列表(如 `/metrics`、`/liveness`、`/docs`、`/v1/metaserver` 等),可自定义。代码默认包含 `/v1/metaserver`(Decode layerwise 回调不带 Key)。 |
419| encryption_algorithm | string | Key 校验使用的加密算法,如 `PBKDF2_SHA256`。默认值:`PBKDF2_SHA256` |423| encryption_algorithm | string | Key 校验使用的加密算法,如 `PBKDF2_SHA256`。默认值:`PBKDF2_SHA256` |
420| **rate_limit_config字段** |-|-|424| **rate_limit_config字段** |-|-|
421| enable_rate_limit | bool | 是否开启请求限流。可选:`true` / `false`。默认值:`false` |425| enable_rate_limit | bool | 是否开启请求限流。可选:`true` / `false`。默认值:`false` |
@@ -818,7 +822,7 @@ PD模式下P与D**各自独立配置**"health_check_config",未配置时使用
818 822 
819| 配置项 | 类型 | 说明 |823| 配置项 | 类型 | 说明 |
820|--------|------|------|824|--------|------|------|
821-| dispatch_profile | string | 可选值为 `handoff` 或 `trigger`。当前原生 vLLM P/D 启动仅接受 `handoff`;SGLang 使用自身 bootstrap 协议。Prefill 与 Decode 两端应保持一致。 |825+| dispatch_profile | string | 可选值为 `handoff` 或 `trigger`。原生 vLLM P/D 同时接受二者:`handoff` 为 Prefill 完成后交给 Decode`trigger` 为 Decode 先启动并经 Worker metaserver 触发 Prefill(`MooncakeLayerwiseConnector` 自动推导为 `trigger`)。SGLang 使用自身 bootstrap 协议。Prefill 与 Decode 两端应保持一致。 |
822 826 
823>[!NOTE]说明827>[!NOTE]说明
824> `dispatch_profile` 写在 `motor_engine_*_config` 顶层,不是在 `engine_config` 内部。`dispatch_capabilities` 为内部兼容字段,不支持用户直接填写。828> `dispatch_profile` 写在 `motor_engine_*_config` 顶层,不是在 `engine_config` 内部。`dispatch_capabilities` 为内部兼容字段,不支持用户直接填写。
Mdocs/zh/user_guide/features/kv_cache_store/README.md+2-2
@@ -108,7 +108,7 @@ KV池化主要通过 `user_config.json` 配置;使用 UCM 功能时还需要
108| [MemCache](backend/memcache.md) | `memcache` | 默认后端,天然支持,无需额外安装 |108| [MemCache](backend/memcache.md) | `memcache` | 默认后端,天然支持,无需额外安装 |
109| Yuanrong | `yuanrong` | TODO:后续版本支持 |109| Yuanrong | `yuanrong` | TODO:后续版本支持 |
110 110 
111-> 关于 Connector 的 handoff 白名单和 `MultiConnector` 传输层规则,请参见 [PD 分离特性说明](../../../design/pd_disaggregation.md#connector-驱动执行计划)。111+> 关于 Connector 的识别白名单和 `MultiConnector` 传输层规则,请参见 [PD 分离特性说明](../../../design/pd_disaggregation.md#connector-驱动执行计划)。
112 112 
113#### kv_cache_store_config(全局配置)113#### kv_cache_store_config(全局配置)
114 114 
@@ -201,7 +201,7 @@ KV池化主要通过 `user_config.json` 配置;使用 UCM 功能时还需要
201> - InferServiceSet 模板中若无 `kv-store` role(未使用 KV 池化的精简模板),deployer 会跳过 kv_store 域名解析,不影响部署。201> - InferServiceSet 模板中若无 `kv-store` role(未使用 KV 池化的精简模板),deployer 会跳过 kv_store 域名解析,不影响部署。
202> - 使用 `--update_instance_num` 扩缩容时,multi_deployment 模式同样会解析 `target_job_id`,确保新扩容的 engine Pod 能连上正确的 kv_store。202> - 使用 `--update_instance_num` 扩缩容时,multi_deployment 模式同样会解析 `target_job_id`,确保新扩容的 engine Pod 能连上正确的 kv_store。
203 203 
204-### 使用 `UCMConnector` 204+### 使用 `UCMConnector`
205 205 
206UCM 属于 KV池化功能,但不复用 `AscendStoreConnector` 的 backend 机制,而是通过 `UCMConnector` 接入:206UCM 属于 KV池化功能,但不复用 `AscendStoreConnector` 的 backend 机制,而是通过 `UCMConnector` 接入:
207 207 
Mexamples/features/config_sample.json+6-3
@@ -186,7 +186,8 @@
186 }186 }
187 },187 },
188 "inference_workers_config": {188 "inference_workers_config": {
189- "num_workers": 4189+ "num_workers": 4,
190+ "worker_metaserver_base_port": 12000
190 },191 },
191 "timeout_config": {192 "timeout_config": {
192 "request_timeout": 30,193 "request_timeout": 30,
@@ -210,7 +211,8 @@
210 "/liveness",211 "/liveness",
211 "/redoc",212 "/redoc",
212 "/instances/refresh",213 "/instances/refresh",
213- "/metrics"214+ "/metrics",
215+ "/v1/metaserver"
214 ],216 ],
215 "encryption_algorithm": "PBKDF2_SHA256"217 "encryption_algorithm": "PBKDF2_SHA256"
216 },218 },
@@ -228,7 +230,8 @@
228 "/redoc",230 "/redoc",
229 "/openapi.json",231 "/openapi.json",
230 "/favicon.ico",232 "/favicon.ico",
231- "/startup"233+ "/startup",
234+ "/v1/metaserver"
232 ],235 ],
233 "error_message": "too many requests, please try again later",236 "error_message": "too many requests, please try again later",
234 "error_status_code": 429,237 "error_status_code": 429,
Mmotor/common/utils/net.py+8-0
@@ -44,6 +44,14 @@ def detect_family(host: str) -> int:
44 return socket.AF_INET6 if _is_ipv6_literal(host) else socket.AF_INET44 return socket.AF_INET6 if _is_ipv6_literal(host) else socket.AF_INET
45 45 
46 46 
47+def is_unspecified_host(host: str) -> bool:
48+ """Return whether ``host`` is the IPv4 or IPv6 unspecified address."""
49+ try:
50+ return ipaddress.ip_address(_strip_brackets(host)).is_unspecified
51+ except ValueError:
52+ return False
53+ 
54+ 
47def format_host(host: str) -> str:55def format_host(host: str) -> str:
48 """Wrap an IPv6 literal in brackets so it is safe to embed in URLs."""56 """Wrap an IPv6 literal in brackets so it is safe to embed in URLs."""
49 if not host:57 if not host:
Mmotor/config/coordinator.py+22-2
@@ -144,6 +144,7 @@ def _default_skip_paths() -> set[str]:
144 "/redoc",144 "/redoc",
145 "/openapi.json",145 "/openapi.json",
146 "/favicon.ico",146 "/favicon.ico",
147+ "/v1/metaserver",
147 }148 }
148 149 
149 150 
@@ -157,6 +158,7 @@ def default_rate_limit_skip_paths() -> list[str]:
157 "/openapi.json",158 "/openapi.json",
158 "/favicon.ico",159 "/favicon.ico",
159 "/startup",160 "/startup",
161+ "/v1/metaserver",
160 ]162 ]
161 163 
162 164 
@@ -481,6 +483,8 @@ class APIKeyConfig:
481@dataclass483@dataclass
482class InferenceWorkersConfig:484class InferenceWorkersConfig:
483 num_workers: int = 4 # Number of inference API worker processes; >1 = multiprocess485 num_workers: int = 4 # Number of inference API worker processes; >1 = multiprocess
486+ # Base port for per-worker metaserver; 0=disabled. Worker i listens on base+i.
487+ worker_metaserver_base_port: int = 12000
wangyangG
wangyangwangyang7 天前

metaserver 端口默认开启,host=0.0.0.0 且无 POD_IP 的存量部署升级后校验直接失败,建议默认 0 或按需启用

likedislike
tobking
tobking
7 天前 评论:
Gganglv7 天前

严重程度: 严重

问题: worker_metaserver_base_port 默认值取 12000(本行),意味着未配置该字段的部署也会默认开启每个 Worker 的 metaserver 端口;同时 _validate_worker_metaserver_ports()(第 1102 行)在 base_port != 0 时无条件校验回调地址:Env.pod_ip or coordinator_api_host 为通配地址(0.0.0.0/::)即追加配置错误。任何配置了 coordinator_api_host: "0.0.0.0" 且未设置 POD_IP 的存量部署(Docker/裸机多节点常见配置,与 layerwise 完全无关),升级后 validate_config() 直接失败,Coordinator 无法启动。

原因: coordinator_api_host 的既有默认/常见取值就是绑定通配地址(_validate_ip_or_hostname 一直允许 0.0.0.0),校验没有把"是否启用 trigger 功能"与"是否触发校验"关联起来;且 PR 描述第 3 节明确写"默认 0 表示关闭",与最终代码(12000)矛盾,说明是中途改动遗留。

怎么改: 默认值改回 0(关闭),仅当显式配置 >0 时再做端口与地址校验;或把"回调地址不可达"的校验改为运行时判定(_trigger_metaserver_url 已能在 detect 到 trigger 集群时返回 503),避免无关部署在启动期被硬拦截。同时同步修正 PR 描述与文档口径。

likedislike
tobking
tobking
7 天前 评论:
tobking
tobking
7 天前 评论:
484 488 
485 489 
486@dataclass490@dataclass
@@ -621,6 +625,7 @@ class CoordinatorConfig:
621 last_modified: float | None = field(default=None, init=False)625 last_modified: float | None = field(default=None, init=False)
622 _errors: list[str] = field(default_factory=list, init=False)626 _errors: list[str] = field(default_factory=list, init=False)
623 worker_index: int | None = field(default=None, repr=False)627 worker_index: int | None = field(default=None, repr=False)
628+ worker_metaserver_port: int | None = field(default=None, repr=False)
624 629 
625 def __post_init__(self):630 def __post_init__(self):
626 """Validate configuration after initialization"""631 """Validate configuration after initialization"""
@@ -831,6 +836,7 @@ class CoordinatorConfig:
831 self.inference_workers_config.num_workers,836 self.inference_workers_config.num_workers,
832 "num_workers",837 "num_workers",
833 )838 )
839+ self._validate_worker_metaserver_ports()
834 840 
835 # Validate scheduler score configuration841 # Validate scheduler score configuration
836 self._validate_positive_number(842 self._validate_positive_number(
@@ -1001,7 +1007,7 @@ class CoordinatorConfig:
1001 return reload_dataclass_config_from_json(1007 return reload_dataclass_config_from_json(
1002 self,1008 self,
1003 self.from_json,1009 self.from_json,
1004- skip=frozenset({"worker_index"}),1010+ skip=frozenset({"worker_index", "worker_metaserver_port"}),
1005 skip_private=True,1011 skip_private=True,
1006 )1012 )
1007 1013 
@@ -1093,7 +1099,9 @@ class CoordinatorConfig:
1093 f" └─ Context Budget Mode: {self.context_budget_mode}\n"1099 f" └─ Context Budget Mode: {self.context_budget_mode}\n"
1094 "\n"1100 "\n"
1095 " Multiprocess (Inference Workers):\n"1101 " Multiprocess (Inference Workers):\n"
1096- f" ─ Num Workers: {self.inference_workers_config.num_workers}\n"1102+ f" ─ Num Workers: {self.inference_workers_config.num_workers}\n"
1103+ f" └─ Worker Metaserver Base: "
1104+ f"{self.inference_workers_config.worker_metaserver_base_port or 'disabled'}\n"
1097 "\n"1105 "\n"
1098 " Security:\n"1106 " Security:\n"
1099 f" ├─ Infer TLS: {'Enabled' if self.infer_tls_config.enable_tls else 'Disabled'}\n"1107 f" ├─ Infer TLS: {'Enabled' if self.infer_tls_config.enable_tls else 'Disabled'}\n"
@@ -1125,6 +1133,18 @@ class CoordinatorConfig:
1125 elif not allow_zero and value <= 0:1133 elif not allow_zero and value <= 0:
1126 self._errors.append(f"{field_name} must be greater than 0")1134 self._errors.append(f"{field_name} must be greater than 0")
1127 1135 
1136+ def _validate_worker_metaserver_ports(self) -> None:
1137+ """Allow 0 (disabled); otherwise each worker port must fit in 1-65535."""
1138+ base_port = self.inference_workers_config.worker_metaserver_base_port
1139+ if base_port == 0:
1140+ return
1141+ if not isinstance(base_port, int) or isinstance(base_port, bool) or base_port < 1:
1142+ self._errors.append("worker_metaserver_base_port must be 0 or a TCP port in 1-65535")
1143+ return
1144+ last_port = base_port + self.inference_workers_config.num_workers - 1
1145+ if last_port > 65535:
1146+ self._errors.append("worker_metaserver_base_port + num_workers - 1 must be in range 1-65535")
1147+ 
1128 def _validate_port_range(self, port: int, field_name: str) -> None:1148 def _validate_port_range(self, port: int, field_name: str) -> None:
1129 """Validate that a port number is in valid range (1-65535)"""1149 """Validate that a port number is in valid range (1-65535)"""
1130 if not (1 <= port <= 65535):1150 if not (1 <= port <= 65535):
Mmotor/config/node_manager.py+1-3
@@ -584,9 +584,7 @@ class NodeManagerConfig:
584 native_engine_config,584 native_engine_config,
585 explicit_profile=engine_config.get(DISPATCH_PROFILE_KEY),585 explicit_profile=engine_config.get(DISPATCH_PROFILE_KEY),
586 )586 )
587- # The native vLLM runtime implements only the explicit handoff contract.587+ capabilities = dispatch_capabilities_for_profile(profile)
588- # Do not advertise trigger/concurrent support that runtime validation rejects.
589- capabilities = dispatch_capabilities_for_profile(profile) if profile == DispatchProfile.HANDOFF else []
590 if not capabilities and profile == DispatchProfile.UNKNOWN:588 if not capabilities and profile == DispatchProfile.UNKNOWN:
591 logger.warning(589 logger.warning(
592 "Unable to infer vLLM dispatch capability from kv_transfer_config. "590 "Unable to infer vLLM dispatch capability from kv_transfer_config. "
Mmotor/coordinator/api_server/inference_server.py+18-1
@@ -11,6 +11,8 @@
11"""11"""
12Inference plane: Worker subprocess only; provides /v1/completions, /v1/chat/completions,12Inference plane: Worker subprocess only; provides /v1/completions, /v1/chat/completions,
13/v1/responses, /v1/messages, /v1/messages/count_tokens, /v1/models, etc.13/v1/responses, /v1/messages, /v1/messages/count_tokens, /v1/models, etc.
14+Dedicated per-worker metaserver (POST /v1/metaserver) is served on worker_metaserver_port
15+when inference_workers_config.worker_metaserver_base_port > 0.
14"""16"""
15 17 
16import asyncio18import asyncio
@@ -37,7 +39,7 @@ from motor.common.http.http_client import HTTPClientPool
37from motor.coordinator.models.constants import OpenAIField39from motor.coordinator.models.constants import OpenAIField
38from motor.coordinator.models.request import RequestType40from motor.coordinator.models.request import RequestType
39from motor.coordinator.domain.request_manager import RequestManager41from motor.coordinator.domain.request_manager import RequestManager
40-from motor.coordinator.router.dispatch import handle_request42+from motor.coordinator.router.dispatch import handle_metaserver_request, handle_request
41from motor.coordinator.tracer.tracing import TracerManager43from motor.coordinator.tracer.tracing import TracerManager
42from motor.coordinator.domain.agent_hint import agent_hint_implies_manage_request44from motor.coordinator.domain.agent_hint import agent_hint_implies_manage_request
43 45 
@@ -165,6 +167,21 @@ class InferenceServer(BaseCoordinatorServer):
165 """Inference FastAPI app, run by process_worker with uvicorn."""167 """Inference FastAPI app, run by process_worker with uvicorn."""
166 return self._inference_app168 return self._inference_app
167 169 
170+ def create_metaserver_app(self) -> FastAPI:
G
Gganglv7 天前

严重程度: 建议

问题: create_metaserver_app() 暴露的 POST /v1/metaserver 无 API Key、无 TLS,且绑定 coordinator_api_host(常为 0.0.0.0)。集群内任意可达该端口的进程都可以 POST;配合可猜测/可泄露的 req_id(时间戳+计数+8 位 hex 随机,约 32bit 熵)可触发 P 实例分配与 prefill 请求转发,造成资源滥用。

原因: 设计文档明确"Decode 引擎回调不携带 API Key、不走推理面 TLS"——可接受,但没有给出任何替代访问控制(绑定地址限定、来源网段、或可选的 key 校验)。

怎么改: 至少把监听地址限定为 Env.pod_ip(未设置时回退 loopback)而不是 coordinator_api_host;或在配置了 valid_keys 时允许回调注入 API Key 并开启校验,保持向后兼容。

likedislike
tobking
tobking
7 天前 评论:
171+ """Dedicated per-worker app for Decode layerwise callbacks. No API key / TLS."""
172+ app = FastAPI(title="Inference Worker Metaserver")
173+ 
174+ @app.post("/v1/metaserver")
175+ async def metaserver(request: Request):
176+ return await handle_metaserver_request(
177+ request,
178+ self.coordinator_config,
179+ scheduler=self._get_scheduler_client(),
180+ request_manager=self._request_manager,
181+ )
182+ 
183+ return app
184+ 
168 @asynccontextmanager185 @asynccontextmanager
169 async def _lifespan(self, app: FastAPI):186 async def _lifespan(self, app: FastAPI):
170 logger.info("Inference server is starting...")187 logger.info("Inference server is starting...")
Mmotor/coordinator/domain/scheduling.py+4-0
@@ -162,3 +162,7 @@ class SchedulingFacade(Protocol):
162 async def get_unblocked_instances(self, role: PDRole) -> list[int]:162 async def get_unblocked_instances(self, role: PDRole) -> list[int]:
163 """Return instance IDs of the given role that are NOT blocked by circuit breaker."""163 """Return instance IDs of the given role that are NOT blocked by circuit breaker."""
164 ...164 ...
165+ 
166+ async def get_local_instances(self, role: PDRole | None = None) -> dict[int, Instance]:
167+ """Return the local instance view; warm-up only when the cache is empty."""
168+ ...
Mmotor/coordinator/models/request.py+2-0
@@ -89,6 +89,8 @@ class RequestInfo(BaseModel):
89 _p_cancel_scope: anyio.CancelScope | None = PrivateAttr(default=None)89 _p_cancel_scope: anyio.CancelScope | None = PrivateAttr(default=None)
90 _d_cancel_scope: anyio.CancelScope | None = PrivateAttr(default=None)90 _d_cancel_scope: anyio.CancelScope | None = PrivateAttr(default=None)
91 _e_cancel_scope: anyio.CancelScope | None = PrivateAttr(default=None)91 _e_cancel_scope: anyio.CancelScope | None = PrivateAttr(default=None)
92+ # Bound UnifiedPD trigger attempt; looked up by the Worker metaserver callback.
93+ _trigger_attempt: object | None = PrivateAttr(default=None)
92 prompt_tokens_details: dict = Field(default={}, description="prefill prompt_tokens_details")94 prompt_tokens_details: dict = Field(default={}, description="prefill prompt_tokens_details")
93 prompt_token_ids: list = Field(default=[], description="prefill prompt_token_ids")95 prompt_token_ids: list = Field(default=[], description="prefill prompt_token_ids")
94 cached_token_ids: list = Field(default=[], description="Cached token_ids")96 cached_token_ids: list = Field(default=[], description="Cached token_ids")
Mmotor/coordinator/process/inference_manager.py+128-2
@@ -24,6 +24,7 @@ except ImportError:
24from motor.common.http.cert_util import CertUtil24from motor.common.http.cert_util import CertUtil
25from motor.common.utils.config_watcher import ConfigWatcher25from motor.common.utils.config_watcher import ConfigWatcher
26from motor.common.http.http_client import HTTPClientPool26from motor.common.http.http_client import HTTPClientPool
27+from motor.common.utils.env import Env
27from motor.common.utils.net import detect_family, format_address28from motor.common.utils.net import detect_family, format_address
28from motor.common.logger import get_logger, reconfigure_logging29from motor.common.logger import get_logger, reconfigure_logging
29from motor.config.coordinator import CoordinatorConfig30from motor.config.coordinator import CoordinatorConfig
@@ -42,6 +43,86 @@ def _socket_host(host: str) -> str:
42 return host43 return host
43 44 
44 45 
46+def metaserver_bind_host(config: CoordinatorConfig) -> str:
47+ """Listen address for the per-worker metaserver.
48+ 
49+ Prefer POD_IP so the socket is not bound on every interface when the infer
50+ API listens on 0.0.0.0. Fall back to coordinator_api_host (not loopback)
51+ so cross-node Decode callbacks still work without POD_IP.
52+ """
53+ return Env.pod_ip or config.api_config.coordinator_api_host
54+ 
55+ 
56+def _disable_worker_metaserver(config: CoordinatorConfig) -> None:
57+ """Drop this process's metaserver port so Trigger requests fail closed with 503."""
58+ config.worker_metaserver_port = None
59+ 
60+ 
61+async def serve_worker_metaserver(
62+ server: Any,
63+ config: CoordinatorConfig,
64+ worker_index: int,
65+ host: str,
66+ port: int,
67+) -> None:
68+ """Run the per-worker metaserver without taking down the infer socket.
69+ 
70+ Bind/serve failures are logged and disable Trigger on this Worker. Handoff
71+ traffic on the infer port must keep running (default base port 12000 can
72+ collide with unrelated services).
73+ """
74+ try:
75+ await server.serve()
76+ except Exception as e:
77+ logger.error(
78+ "Worker %s: metaserver failed on %s:%s: %s; infer port continues, trigger PD will return 503",
79+ worker_index,
80+ host,
81+ port,
82+ e,
83+ exc_info=True,
84+ )
85+ _disable_worker_metaserver(config)
86+ 
87+ 
88+async def run_inference_and_metaserver(
89+ infer_server: Any,
90+ infer_sockets: list[socket.socket] | None,
91+ metaserver_server: Any | None,
92+ config: CoordinatorConfig,
93+ worker_index: int,
94+ metaserver_host: str,
95+) -> None:
96+ """Serve infer as the primary task; metaserver is a best-effort sidecar."""
97+ infer_task = asyncio.create_task(
98+ infer_server.serve(sockets=infer_sockets),
99+ name=f"infer-{worker_index}",
100+ )
101+ meta_task: asyncio.Task | None = None
102+ if metaserver_server is not None:
103+ meta_port = config.worker_metaserver_port or 0
104+ meta_task = asyncio.create_task(
105+ serve_worker_metaserver(
106+ metaserver_server,
107+ config,
108+ worker_index,
109+ metaserver_host,
110+ meta_port,
111+ ),
112+ name=f"metaserver-{worker_index}",
113+ )
114+ try:
115+ await infer_task
116+ finally:
117+ if meta_task is not None and not meta_task.done() and metaserver_server is not None:
118+ metaserver_server.should_exit = True
119+ meta_task.cancel()
120+ try:
121+ await meta_task
122+ except asyncio.CancelledError:
123+ pass
124+ 
125+ 
45def run_inference_worker_proc(126def run_inference_worker_proc(
46 listen_address: tuple[str, int],127 listen_address: tuple[str, int],
47 sock: socket.socket,128 sock: socket.socket,
@@ -65,8 +146,12 @@ def run_inference_worker_proc(
65 146 
66 # Set process title147 # Set process title
67 set_process_title(name=str(worker_index))148 set_process_title(name=str(worker_index))
149+ config.worker_index = worker_index
150+ base_metaserver_port = config.inference_workers_config.worker_metaserver_base_port
151+ if base_metaserver_port > 0:
152+ config.worker_metaserver_port = base_metaserver_port + worker_index
68 153 
69- logger.info(f"Inference worker process {worker_index} starting (PID: {os.getpid()})")154+ logger.info("Inference worker process %s starting (PID: %s)", worker_index, os.getpid())
70 155 
71 # Create RequestManager first, then InferenceServer (business plane only)156 # Create RequestManager first, then InferenceServer (business plane only)
72 request_manager = RequestManager(config)157 request_manager = RequestManager(config)
@@ -125,9 +210,50 @@ def run_inference_worker_proc(
125 210 
126 # Create and run server(s)211 # Create and run server(s)
127 server = uvicorn.Server(uvicorn_config)212 server = uvicorn.Server(uvicorn_config)
213+ metaserver_server = None
214+ metaserver_host = ""
215+ if config.worker_metaserver_port:
216+ metaserver_host = metaserver_bind_host(config)
217+ metaserver_port = config.worker_metaserver_port
218+ try:
219+ metaserver_app = inference_server.create_metaserver_app()
220+ metaserver_kwargs = InferenceServer.create_base_uvicorn_config(
221+ metaserver_app,
222+ metaserver_host,
223+ metaserver_port,
224+ )
225+ metaserver_kwargs["lifespan"] = "off"
226+ metaserver_config = uvicorn.Config(**metaserver_kwargs)
227+ metaserver_config.load()
228+ metaserver_server = uvicorn.Server(metaserver_config)
229+ logger.info(
230+ "Worker %s: metaserver port enabled: %s (base=%s bind_host=%s)",
231+ worker_index,
232+ metaserver_port,
233+ base_metaserver_port,
234+ metaserver_host,
235+ )
236+ except Exception as e:
237+ logger.error(
238+ "Worker %s: metaserver failed to initialize on %s:%s: %s; infer port continues, trigger PD will return 503",
239+ worker_index,
240+ metaserver_host,
241+ metaserver_port,
242+ e,
243+ exc_info=True,
244+ )
245+ _disable_worker_metaserver(config)
246+ metaserver_server = None
128 247 
129 async def _run_servers():248 async def _run_servers():
130- await server.serve(sockets=[sock] if sock else None)249+ await run_inference_and_metaserver(
250+ server,
251+ [sock] if sock else None,
252+ metaserver_server,
253+ config,
254+ worker_index,
255+ metaserver_host,
256+ )
131 257 
132 try:258 try:
133 # Run server with shared socket.259 # Run server with shared socket.
Mmotor/coordinator/router/adapters/pd_protocol.py+49-0
@@ -18,10 +18,23 @@ from enum import Enum
18from types import MappingProxyType18from types import MappingProxyType
19from typing import Any, Protocol19from typing import Any, Protocol
20 20 
21+from motor.common.constants import CHAT_COMPLETION_PREFIX, COMPLETION_PREFIX, COMPLETION_SUFFIX
22+ 
21 23 
22class CoordinationMode(str, Enum):24class CoordinationMode(str, Enum):
23 HANDOFF = "handoff"25 HANDOFF = "handoff"
24 BOOTSTRAP = "bootstrap"26 BOOTSTRAP = "bootstrap"
27+ TRIGGER = "trigger"
28+ 
29+ 
30+def trim_vllm_engine_request_id(request_id: str) -> str:
31+ """Strip vLLM/OpenAI prefixes so Coordinator can look up the original req_id."""
32+ value = str(request_id or "").strip()
33+ if value.startswith(CHAT_COMPLETION_PREFIX):
34+ return value.removeprefix(CHAT_COMPLETION_PREFIX)
35+ if value.startswith(COMPLETION_PREFIX) and value.endswith(COMPLETION_SUFFIX):
36+ return value.removeprefix(COMPLETION_PREFIX).removesuffix(COMPLETION_SUFFIX)
37+ return value
25 38 
26 39 
27@dataclass(frozen=True)40@dataclass(frozen=True)
@@ -162,6 +175,42 @@ class VllmProtocolAdapter:
162 body.pop("rid", None)175 body.pop("rid", None)
163 body["request_id"] = request_id176 body["request_id"] = request_id
164 177 
178+ def build_trigger_decode_request(
179+ self,
180+ request: Mapping[str, Any],
181+ context: LegContext,
182+ metaserver_url: str,
183+ ) -> EngineRequest:
184+ body = deepcopy(dict(request))
185+ self.inject_request_id(body, context.engine_request_id)
186+ body["kv_transfer_params"] = {
187+ "do_remote_decode": False,
188+ "do_remote_prefill": True,
189+ "metaserver": metaserver_url,
190+ }
191+ return EngineRequest(api=context.api, body=body)
192+ 
193+ def build_trigger_prefill_request(
194+ self,
195+ request: Mapping[str, Any],
196+ context: LegContext,
197+ kv_transfer_params: Mapping[str, Any],
198+ ) -> EngineRequest:
199+ body = deepcopy(dict(request))
200+ self.inject_request_id(body, context.engine_request_id)
201+ body["stream"] = False
202+ body["max_tokens"] = 1
203+ body["min_tokens"] = 1
204+ body.pop("stream_options", None)
205+ if "max_completion_tokens" in body:
206+ body["max_completion_tokens"] = 1
207+ params = deepcopy(dict(kv_transfer_params))
208+ params["do_remote_decode"] = True
209+ params["do_remote_prefill"] = False
210+ params.pop("metaserver", None)
211+ body["kv_transfer_params"] = params
212+ return EngineRequest(api=context.api, body=body)
213+ 
165 def build_abort_request(self, context: LegContext) -> EngineRequest | None:214 def build_abort_request(self, context: LegContext) -> EngineRequest | None:
166 del context215 del context
167 216 
Mmotor/coordinator/router/dispatch.py+71-0
@@ -39,6 +39,8 @@ from motor.coordinator.domain.agent_hint import (
39 parse_agent_hint,39 parse_agent_hint,
40 ensure_minimum_messages_for_session_edits,40 ensure_minimum_messages_for_session_edits,
41)41)
42+from motor.coordinator.router.adapters.pd_protocol import trim_vllm_engine_request_id
43+from motor.coordinator.router.dispatch_session import AttemptContext
42from motor.coordinator.router.strategies.base import BaseRouter44from motor.coordinator.router.strategies.base import BaseRouter
43from motor.coordinator.router.strategies.pd_hybrid import PDHybridRouter45from motor.coordinator.router.strategies.pd_hybrid import PDHybridRouter
44from motor.coordinator.router.strategies.unified_pd import UnifiedPDRouter46from motor.coordinator.router.strategies.unified_pd import UnifiedPDRouter
@@ -337,6 +339,75 @@ async def handle_request(
337 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=safe_error_msg) from e339 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=safe_error_msg) from e
338 340 
339 341 
342+def _parse_trigger_attempt_seq(raw_request: Request) -> int:
343+ raw_value = raw_request.query_params.get("attempt")
344+ if raw_value is None:
345+ raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Missing attempt query parameter")
346+ try:
347+ return int(raw_value)
348+ except (TypeError, ValueError) as exc:
349+ raise HTTPException(
350+ status_code=status.HTTP_400_BAD_REQUEST,
351+ detail="Invalid attempt query parameter",
352+ ) from exc
353+ 
354+ 
355+@with_cancellation
356+async def handle_metaserver_request(
357+ raw_request: Request,
358+ config: CoordinatorConfig,
359+ scheduler=None,
360+ *,
361+ request_manager: RequestManager,
362+) -> dict:
363+ """Handle Decode-side layerwise callback and forward Prefill to a scheduled P instance."""
364+ try:
365+ body = await raw_request.json()
366+ except Exception as e:
367+ logger.warning("Metaserver JSON parse failed: %s", e)
368+ raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid JSON format") from e
369+ if not isinstance(body, dict) or not body:
370+ raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Empty request json")
371+ 
372+ request_id = trim_vllm_engine_request_id(str(body.get("request_id") or ""))
373+ if not request_id:
374+ raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Missing request_id")
375+ req_info = await request_manager.get_req_info(request_id)
376+ if req_info is None:
377+ raise HTTPException(
378+ status_code=status.HTTP_404_NOT_FOUND,
379+ detail=f"Request ID {request_id} not found",
380+ )
381+ 
382+ attempt_seq = _parse_trigger_attempt_seq(raw_request)
383+ attempt = req_info._trigger_attempt
384+ if not isinstance(attempt, AttemptContext):
385+ raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Trigger attempt not found")
386+ if attempt.attempt_seq != attempt_seq:
387+ raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Stale trigger attempt callback")
388+ 
389+ if scheduler is None:
390+ raise HTTPException(
391+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
392+ detail="Scheduler (SchedulingFacade) is required and must be injected by the server",
393+ )
394+ 
395+ router_impl = UnifiedPDRouter(
396+ req_info,
397+ config,
398+ scheduler=scheduler,
399+ request_manager=request_manager,
400+ )
401+ try:
402+ return await router_impl.handle_metaserver_request(body)
403+ except HTTPException:
404+ raise
405+ except Exception as e:
406+ logger.error("Error occurred in metaserver endpoint: %s", e, exc_info=True)
407+ safe_error_msg = sanitize_error_message(str(e))
408+ raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=safe_error_msg) from e
409+ 
410+ 
340async def __create_request_info(411async def __create_request_info(
341 raw_request: Request,412 raw_request: Request,
342 request_manager: RequestManager,413 request_manager: RequestManager,
Mmotor/coordinator/router/dispatch_session.py+18-6
@@ -59,6 +59,7 @@ class AttemptContext:
59 prefill_completed: bool = False59 prefill_completed: bool = False
60 decode_dispatched: bool = False60 decode_dispatched: bool = False
61 decode_completed: bool = False61 decode_completed: bool = False
62+ trigger_lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False)
62 stop_lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False)63 stop_lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False)
63 config: CoordinatorConfig | None = None64 config: CoordinatorConfig | None = None
64 65 
@@ -106,20 +107,31 @@ class AttemptContext:
106 107 
107 async def cancel(self, reason: str = ""):108 async def cancel(self, reason: str = ""):
108 tasks = []109 tasks = []
109- if self.prefill_task and not self.prefill_task.done() and not self.prefill_task.cancelled():110+ current_task = asyncio.current_task()
111+ if (
112+ self.prefill_task
113+ and self.prefill_task is not current_task
114+ and not self.prefill_task.done()
115+ and not self.prefill_task.cancelled()
116+ ):
110 logger.info(117 logger.info(
111 "Cancelling prefill task: %s %s because %s",118 "Cancelling prefill task: %s %s because %s",
112- self.prefill_resource.endpoint.ip,119+ self.prefill_resource.endpoint.ip if self.prefill_resource else "pending",
113- self.prefill_resource.instance.job_name,120+ self.prefill_resource.instance.job_name if self.prefill_resource else "pending",
114 reason,121 reason,
115 )122 )
116 self.prefill_task.cancel(msg=reason)123 self.prefill_task.cancel(msg=reason)
117 tasks.append(self.prefill_task)124 tasks.append(self.prefill_task)
118- if self.decode_task and not self.decode_task.done() and not self.decode_task.cancelled():125+ if (
126+ self.decode_task
127+ and self.decode_task is not current_task
128+ and not self.decode_task.done()
129+ and not self.decode_task.cancelled()
130+ ):
119 logger.info(131 logger.info(
120 "Cancelling decode task: %s %s because %s",132 "Cancelling decode task: %s %s because %s",
121- self.decode_resource.endpoint.ip,133+ self.decode_resource.endpoint.ip if self.decode_resource else "pending",
122- self.decode_resource.instance.job_name,134+ self.decode_resource.instance.job_name if self.decode_resource else "pending",
123 reason,135 reason,
124 )136 )
125 self.decode_task.cancel(msg=reason)137 self.decode_task.cancel(msg=reason)
Mmotor/coordinator/router/strategies/unified_pd.py+281-20
@@ -11,7 +11,7 @@
11import asyncio11import asyncio
12import time12import time
13from contextlib import aclosing13from contextlib import aclosing
14-from dataclasses import dataclass14+from dataclasses import dataclass, replace
15from typing import Any, AsyncGenerator15from typing import Any, AsyncGenerator
16 16 
17import httpx17import httpx
@@ -20,7 +20,10 @@ from fastapi.responses import JSONResponse, Response
20 20 
21import motor.common.utils.error as cancel_error21import motor.common.utils.error as cancel_error
22from motor.common.utils.error import RequestCancelledError22from motor.common.utils.error import RequestCancelledError
23+from motor.common.utils.env import Env
24+from motor.common.utils.net import format_address, is_unspecified_host
23from motor.common.resources.endpoint import WorkloadAction25from motor.common.resources.endpoint import WorkloadAction
26+from motor.common.resources.dispatch import DispatchPlan
24from motor.common.resources.instance import PDRole27from motor.common.resources.instance import PDRole
25from motor.config.coordinator import CoordinatorConfig28from motor.config.coordinator import CoordinatorConfig
26from motor.coordinator.domain import (29from motor.coordinator.domain import (
@@ -57,6 +60,7 @@ from motor.coordinator.router.adapters.pd_protocol import (
57 LegContext,60 LegContext,
58 PDProtocolAdapter,61 PDProtocolAdapter,
59 PrefillMetadata,62 PrefillMetadata,
63+ VllmProtocolAdapter,
60)64)
61from motor.coordinator.router.adapters.stream import (65from motor.coordinator.router.adapters.stream import (
62 parse_stream_chunk_json,66 parse_stream_chunk_json,
@@ -75,6 +79,8 @@ from motor.coordinator.router.upstream_error import (
75)79)
76 80 
77_SGLANG_PROTOCOL_ADAPTER = ADAPTERS["sglang"]81_SGLANG_PROTOCOL_ADAPTER = ADAPTERS["sglang"]
82+_TRIGGER_CAPABILITY = DispatchPlan.CONCURRENT_ENGINE_SYNC.value
83+_HANDOFF_CAPABILITY = DispatchPlan.PREFILL_HANDOFF_DECODE.value
78 84 
79 85 
80@dataclass(frozen=True)86@dataclass(frozen=True)
@@ -152,6 +158,7 @@ class UnifiedPDRouter(BaseRouter):
152 self._stream_body_sent = False158 self._stream_body_sent = False
153 self._active_retry_plan: RetryRequestPlan | None = None159 self._active_retry_plan: RetryRequestPlan | None = None
154 self._hybrid_stream_fallback_attempted = False # A request is only allowed to fallback once.160 self._hybrid_stream_fallback_attempted = False # A request is only allowed to fallback once.
161+ self._pd_uses_trigger: bool | None = None
155 # Task -> its bookkeeping record (dedup key, logging context, computed work item).162 # Task -> its bookkeeping record (dedup key, logging context, computed work item).
156 self._release_records: dict[asyncio.Task[bool], _ReleaseTaskRecord] = {}163 self._release_records: dict[asyncio.Task[bool], _ReleaseTaskRecord] = {}
157 # Dedup reverse index: release key -> the single in-flight task for that key.164 # Dedup reverse index: release key -> the single in-flight task for that key.
@@ -229,22 +236,23 @@ class UnifiedPDRouter(BaseRouter):
229 236 
230 @staticmethod237 @staticmethod
231 def _adapter_for_attempt(attempt: AttemptContext | None) -> PDProtocolAdapter | None:238 def _adapter_for_attempt(attempt: AttemptContext | None) -> PDProtocolAdapter | None:
232- if attempt is None or attempt.prefill_resource is None:239+ if attempt is None:
233 return None240 return None
234- engine_type = getattr(attempt.prefill_resource.instance, "engine_type", None)241+ resource = attempt.prefill_resource or attempt.decode_resource
242+ if resource is None:
243+ return None
244+ engine_type = getattr(resource.instance, "engine_type", None)
235 if not isinstance(engine_type, str):245 if not isinstance(engine_type, str):
236 return None246 return None
237- normalized = engine_type.strip().lower()247+ return ADAPTERS.get(engine_type.strip().lower())
238- return ADAPTERS.get(normalized)
239 248 
240 @staticmethod249 @staticmethod
241 def _require_adapter_for_attempt(attempt: AttemptContext) -> PDProtocolAdapter:250 def _require_adapter_for_attempt(attempt: AttemptContext) -> PDProtocolAdapter:
242 adapter = UnifiedPDRouter._adapter_for_attempt(attempt)251 adapter = UnifiedPDRouter._adapter_for_attempt(attempt)
243 if adapter is not None:252 if adapter is not None:
244 return adapter253 return adapter
245- engine_type = None254+ resource = attempt.prefill_resource or attempt.decode_resource
246- if attempt.prefill_resource is not None:255+ engine_type = getattr(resource.instance, "engine_type", None) if resource is not None else None
247- engine_type = getattr(attempt.prefill_resource.instance, "engine_type", None)
248 raise RuntimeError(f"Unsupported native engine type for P/D coordination: {engine_type!r}")256 raise RuntimeError(f"Unsupported native engine type for P/D coordination: {engine_type!r}")
249 257 
250 @staticmethod258 @staticmethod
@@ -280,8 +288,12 @@ class UnifiedPDRouter(BaseRouter):
280 async def handle_request(self) -> Response:288 async def handle_request(self) -> Response:
281 await self.do_encode()289 await self.do_encode()
282 self.is_meta = False290 self.is_meta = False
291+ uses_trigger = await self._pd_cluster_uses_trigger()
292+ if uses_trigger:
293+ self._ensure_trigger_metaserver()
283 if self.req_info.req_data.get("stream", False):294 if self.req_info.req_data.get("stream", False):
284- self._stream_commit_controller = StreamCommitController.requiring({"prefill", "decode"})295+ commit_parts = {"decode"} if uses_trigger else {"prefill", "decode"}
296+ self._stream_commit_controller = StreamCommitController.requiring(commit_parts)
285 return CommitAwareStreamingResponse(297 return CommitAwareStreamingResponse(
286 self._generate_stream_response(),298 self._generate_stream_response(),
287 self._stream_commit_controller,299 self._stream_commit_controller,
@@ -638,6 +650,11 @@ class UnifiedPDRouter(BaseRouter):
638 650 
639 async def _create_attempt(self, session: PDDispatchSession) -> AttemptContext:651 async def _create_attempt(self, session: PDDispatchSession) -> AttemptContext:
640 attempt_seq = session._attempt_seq + 1652 attempt_seq = session._attempt_seq + 1
653+ if await self._pd_cluster_uses_trigger():
654+ self._ensure_trigger_metaserver()
655+ d_resource = await self._prepare_attempt_resource(PDRole.ROLE_D, attempt_seq)
656+ return session.new_attempt(None, d_resource, self.config)
657+ 
641 p_resource = await self._prepare_attempt_resource(PDRole.ROLE_P, attempt_seq)658 p_resource = await self._prepare_attempt_resource(PDRole.ROLE_P, attempt_seq)
642 659 
643 # Handoff connectors (CPCD-style) do not need a concrete decode endpoint while prefill runs.660 # Handoff connectors (CPCD-style) do not need a concrete decode endpoint while prefill runs.
@@ -675,11 +692,93 @@ class UnifiedPDRouter(BaseRouter):
675 adapter = ADAPTERS.get(engine_type.strip().lower())692 adapter = ADAPTERS.get(engine_type.strip().lower())
676 return adapter is not None and adapter.coordination_mode == CoordinationMode.HANDOFF693 return adapter is not None and adapter.coordination_mode == CoordinationMode.HANDOFF
677 694 
695+ async def _pd_cluster_uses_trigger(self) -> bool:
696+ if self._pd_uses_trigger is not None:
697+ return self._pd_uses_trigger
698+ self._pd_uses_trigger = await self._detect_vllm_trigger_cluster()
699+ return self._pd_uses_trigger
700+ 
701+ async def _detect_vllm_trigger_cluster(self) -> bool:
702+ p_plans = await self._vllm_dispatch_plans(PDRole.ROLE_P)
703+ d_plans = await self._vllm_dispatch_plans(PDRole.ROLE_D)
704+ combined = p_plans | d_plans
705+ if not combined:
706+ return False
707+ if _TRIGGER_CAPABILITY in combined and _HANDOFF_CAPABILITY in combined:
708+ raise HTTPException(
709+ status_code=503,
710+ detail="Mixed vLLM handoff and layerwise/trigger P/D instances are not supported",
711+ )
712+ return combined == {_TRIGGER_CAPABILITY}
713+ 
714+ async def _vllm_dispatch_plans(self, role: PDRole) -> set[str]:
G
Gganglv7 天前

严重程度: 建议

问题: _detect_vllm_trigger_cluster(701-712 行)对 P、D 各调用一次 _vllm_dispatch_plans,而每次调用都会执行 get_available_instances——SchedulerClient.get_available_instances 每次无条件 send_request(无 TTL 缓存)。即每个推理请求(含 handoff/sglang 集群,非仅 trigger)都多 2 次 ZMQ 往返,随后还在调度器侧全量扫描实例。

原因: _pd_uses_trigger 只在单个 router 实例(单请求)内缓存,而集群拓扑在请求间变化不大,属于可缓存信息,放在热路径上性价比低。

怎么改: 按拓扑版本号或短 TTL(如 1-5s)做进程内缓存;或利用 get_available_instances 已维护的 instance cache 版本号,仅在版本变化时重新检测。

likedislike
tobking
tobking
7 天前 评论:
715+ get_instances = getattr(self._scheduler, "get_local_instances", None)
716+ if get_instances is None:
717+ get_instances = getattr(self._scheduler, "get_available_instances", None)
718+ if get_instances is None:
719+ return set()
720+ get_unblocked = getattr(self._scheduler, "get_unblocked_instances", None)
721+ unblocked_ids = set(await get_unblocked(role)) if get_unblocked is not None else None
722+ instances = await get_instances(role)
723+ plans: set[str] = set()
724+ for inst in instances.values():
725+ if unblocked_ids is not None and inst.id not in unblocked_ids:
726+ continue
727+ if str(getattr(inst, "engine_type", "") or "").strip().lower() != "vllm":
728+ continue
729+ caps = list(getattr(inst, "dispatch_capabilities", None) or [])
730+ if _TRIGGER_CAPABILITY in caps:
731+ plans.add(_TRIGGER_CAPABILITY)
732+ else:
733+ plans.add(_HANDOFF_CAPABILITY)
734+ return plans
735+ 
736+ def _ensure_trigger_metaserver(self) -> None:
737+ port = getattr(self.config, "worker_metaserver_port", None)
738+ if not isinstance(port, int) or isinstance(port, bool) or port <= 0:
739+ raise HTTPException(
740+ status_code=503,
741+ detail=("layerwise/trigger PD requires inference_workers_config.worker_metaserver_base_port > 0"),
742+ )
743+ 
744+ def _trigger_metaserver_url(self, attempt: AttemptContext) -> str:
745+ host = Env.pod_ip or self.config.api_config.coordinator_api_host
746+ if is_unspecified_host(host):
747+ self.logger.error(
748+ "Trigger metaserver callback host is unreachable: advertised_host=%s. "
749+ "Wildcard listen addresses (0.0.0.0/::) cannot be used as Decode callback targets; "
750+ "set POD_IP or a concrete coordinator_api_host",
751+ host,
752+ )
753+ raise HTTPException(
754+ status_code=503,
755+ detail="Trigger metaserver requires POD_IP or a concrete coordinator_api_host",
756+ )
757+ port = self.config.worker_metaserver_port
758+ return f"http://{format_address(host, port)}/v1/metaserver?attempt={attempt.attempt_seq}"
759+ 
760+ def _bind_trigger_attempt(self, attempt: AttemptContext) -> None:
761+ self.req_info._trigger_attempt = attempt
762+ 
763+ def _trigger_decode_request_for_attempt(self, attempt: AttemptContext) -> tuple[dict[str, Any], str]:
764+ adapter = self._require_adapter_for_attempt(attempt)
765+ if not isinstance(adapter, VllmProtocolAdapter):
766+ raise RuntimeError(f"Trigger decode requires vLLM adapter, got {adapter.engine_type}")
767+ req, api = self._base_request_for_attempt(PDRole.ROLE_D)
768+ context = replace(
769+ self._native_leg_context(attempt, PDRole.ROLE_D, api),
770+ engine_request_id=self.req_info.req_id,
771+ )
772+ engine_request = adapter.build_trigger_decode_request(req, context, self._trigger_metaserver_url(attempt))
773+ return engine_request.body, engine_request.api
774+ 
678 async def _run_stream_attempt(775 async def _run_stream_attempt(
679 self, attempt: AttemptContext, coordination_mode: CoordinationMode776 self, attempt: AttemptContext, coordination_mode: CoordinationMode
680 ) -> AsyncGenerator[str, None]:777 ) -> AsyncGenerator[str, None]:
681 if coordination_mode == CoordinationMode.HANDOFF:778 if coordination_mode == CoordinationMode.HANDOFF:
682 run_func = self._run_handoff_stream_attempt779 run_func = self._run_handoff_stream_attempt
780+ elif coordination_mode == CoordinationMode.TRIGGER:
781+ run_func = self._run_trigger_stream_attempt
683 else:782 else:
684 run_func = self._run_bootstrap_stream_attempt783 run_func = self._run_bootstrap_stream_attempt
685 async with aclosing(run_func(attempt)) as attempt_stream:784 async with aclosing(run_func(attempt)) as attempt_stream:
@@ -930,8 +1029,135 @@ class UnifiedPDRouter(BaseRouter):
930 ) -> dict[str, Any]:1029 ) -> dict[str, Any]:
931 if coordination_mode == CoordinationMode.HANDOFF:1030 if coordination_mode == CoordinationMode.HANDOFF:
932 return await self._run_handoff_nonstream_attempt(attempt)1031 return await self._run_handoff_nonstream_attempt(attempt)
933- else:1032+ if coordination_mode == CoordinationMode.TRIGGER:
934- return await self._run_bootstrap_nonstream_attempt(attempt)1033+ return await self._run_trigger_nonstream_attempt(attempt)
1034+ return await self._run_bootstrap_nonstream_attempt(attempt)
1035+ 
1036+ async def _run_trigger_stream_attempt(self, attempt: AttemptContext) -> AsyncGenerator[str, None]:
1037+ self._bind_trigger_attempt(attempt)
1038+ attempt.transition(AttemptState.ACTIVE)
1039+ d_req, d_api = self._trigger_decode_request_for_attempt(attempt)
1040+ stream_adapter_state = {}
1041+ sampling_state = self._init_sampling_state()
1042+ async with self._client_for(attempt.decode_resource) as d_client:
1043+ async with aclosing(
1044+ self._run_stream_decode_phase(
1045+ attempt,
1046+ d_client,
1047+ d_api,
1048+ d_req,
1049+ stream_adapter_state,
1050+ sampling_state=sampling_state,
1051+ )
1052+ ) as decode_stream:
1053+ async for chunk in decode_stream:
1054+ yield chunk
1055+ 
1056+ async def _run_trigger_nonstream_attempt(self, attempt: AttemptContext) -> dict[str, Any]:
1057+ self._bind_trigger_attempt(attempt)
1058+ attempt.transition(AttemptState.ACTIVE)
1059+ d_req, d_api = self._trigger_decode_request_for_attempt(attempt)
1060+ sampling_state = self._init_sampling_state()
1061+ async with self._client_for(attempt.decode_resource) as d_client:
1062+ return await self._await_nonstream_decode(
1063+ attempt,
1064+ d_api,
1065+ d_req,
1066+ d_client,
1067+ sampling_state=sampling_state,
1068+ )
1069+ 
1070+ async def handle_metaserver_request(self, kv_transfer_params: dict[str, Any]) -> dict[str, Any]:
1071+ """Forward Decode's layerwise metaserver callback to a scheduled Prefill instance."""
1072+ t0_metaserver = time.perf_counter()
1073+ attempt = self.req_info._trigger_attempt
1074+ if not isinstance(attempt, AttemptContext):
1075+ raise HTTPException(status_code=404, detail="Trigger attempt not found")
1076+ async with attempt.trigger_lock:
1077+ if self.req_info.is_cancelled:
1078+ raise HTTPException(status_code=409, detail="Request already cancelled")
1079+ if attempt.state in (AttemptState.STOPPING, AttemptState.STOPPED, AttemptState.DONE):
1080+ raise HTTPException(status_code=409, detail="Trigger attempt is no longer active")
1081+ if attempt.decode_resource is None:
1082+ raise HTTPException(status_code=409, detail="Trigger decode resource is missing")
1083+ if attempt.prefill_completed:
G
Gganglv7 天前

严重程度: 建议

问题: if attempt.prefill_completed: return {}——首个成功回调的响应体正是 Prefill 引擎的 kv_transfer_params(远端 block/host/port),decode 依赖它建立层间 KV 传输。若引擎重试的触发场景是"首个回调响应在途中丢失"(协调器侧已成功但引擎未收到),重试收到 {} 时拿不到任何 KV 参数,decode 可能挂起或失败。当前测试只覆盖"重试等首个完成"的乐观路径。

原因: 协调器未保存最近一次成功的 prefill 响应体,无法回放;_record_prefill_complete 只摘了 usage/状态。

怎么改: 在 attempt 上保存最近成功的 prefill 响应 body,幂等分支原样回放(响应体很小);若确认 vLLM 引擎对 {} 有明确语义,请在注释/文档中说明并补充"响应丢失后重试"的测试。

likedislike
tobking
tobking
7 天前 评论:
1084+ return {}
1085+ if attempt.prefill_dispatched:
1086+ raise HTTPException(status_code=409, detail="Prefill dispatch did not complete")
1087+ 
1088+ adapter = self._require_adapter_for_attempt(attempt)
1089+ if not isinstance(adapter, VllmProtocolAdapter):
1090+ raise HTTPException(status_code=500, detail="Trigger metaserver requires vLLM")
1091+ 
1092+ current_task = asyncio.current_task()
1093+ if current_task is None:
1094+ raise RuntimeError("Metaserver callback must run in an asyncio task")
1095+ attempt.register_prefill_task(current_task)
1096+ p_instance_id = None
1097+ try:
1098+ if attempt.prefill_resource is None:
1099+ attempt.prefill_resource = await self._prepare_attempt_resource(
1100+ PDRole.ROLE_P,
1101+ attempt.attempt_seq,
1102+ required_engine_type=str(attempt.decode_resource.instance.engine_type),
1103+ )
1104+ if self.req_info.is_cancelled or attempt.state in (
1105+ AttemptState.STOPPING,
1106+ AttemptState.STOPPED,
1107+ AttemptState.DONE,
1108+ ):
1109+ raise HTTPException(status_code=409, detail="Trigger attempt is no longer active")
1110+ 
1111+ req, api = self._base_request_for_attempt(PDRole.ROLE_P)
1112+ context = replace(
1113+ self._native_leg_context(attempt, PDRole.ROLE_P, api),
1114+ engine_request_id=self.req_info.req_id,
1115+ )
1116+ engine_request = adapter.build_trigger_prefill_request(req, context, kv_transfer_params)
1117+ p_instance_id = attempt.prefill_resource.instance.id
1118+ async with self._client_for(attempt.prefill_resource) as p_client:
1119+ attempt.register_canceller()
1120+ attempt.mark_dispatched(PDRole.ROLE_P.value)
1121+ response = await self.forward_request(
1122+ engine_request.api,
1123+ engine_request.body,
1124+ p_client,
1125+ self.config.exception_config.first_token_timeout,
1126+ )
1127+ attempt.mark_completed(PDRole.ROLE_P.value)
1128+ body = response.json()
1129+ self._record_prefill_complete(body)
1130+ await self._scheduler.report_cb_event(p_instance_id, "success")
1131+ self._submit_prefill_release_background(attempt, WorkloadAction.RELEASE_TOKENS)
1132+ elapsed_ms = (time.perf_counter() - t0_metaserver) * 1000
1133+ self.logger.info(
1134+ "Scheduling latency stage=metaserver_request_total elapsed_ms=%.2f role=ROLE_P req_id=%s",
1135+ elapsed_ms,
1136+ self.req_info.req_id,
1137+ )
1138+ return body
1139+ except asyncio.CancelledError:
1140+ self.logger.info("Metaserver request was cancelled req_id=%s", self.req_info.req_id)
1141+ if attempt.state not in (AttemptState.STOPPING, AttemptState.STOPPED, AttemptState.DONE):
1142+ await self._stop_attempt(attempt, AttemptStopReason.PEER_FAILED)
1143+ raise
1144+ except Exception as e:
1145+ elapsed_ms = (time.perf_counter() - t0_metaserver) * 1000
1146+ self.logger.warning(
1147+ "Scheduling latency stage=metaserver_request_total elapsed_ms=%.2f error=%s req_id=%s",
1148+ elapsed_ms,
1149+ e,
1150+ self.req_info.req_id,
1151+ )
1152+ if p_instance_id is not None and is_cb_reportable_failure(e):
1153+ await self._scheduler.report_cb_event(p_instance_id, "failure")
1154+ if isinstance(e, UpstreamHTTPError):
1155+ attempt.mark_completed(PDRole.ROLE_P.value)
1156+ await self._stop_attempt(attempt, AttemptStopReason.PEER_FAILED)
1157+ raise
1158+ finally:
1159+ if attempt.prefill_task is current_task:
1160+ attempt.prefill_task = None
935 1161 
936 async def _run_bootstrap_nonstream_attempt(self, attempt: AttemptContext) -> dict[str, Any]:1162 async def _run_bootstrap_nonstream_attempt(self, attempt: AttemptContext) -> dict[str, Any]:
937 attempt.transition(AttemptState.ACTIVE)1163 attempt.transition(AttemptState.ACTIVE)
@@ -1281,9 +1507,25 @@ class UnifiedPDRouter(BaseRouter):
1281 1507 
1282 def _select_coordination_mode(self, attempt: AttemptContext) -> CoordinationMode:1508 def _select_coordination_mode(self, attempt: AttemptContext) -> CoordinationMode:
1283 adapter = self._require_adapter_for_attempt(attempt)1509 adapter = self._require_adapter_for_attempt(attempt)
1284- if adapter.coordination_mode == CoordinationMode.BOOTSTRAP and attempt.decode_resource is None:1510+ if adapter.coordination_mode == CoordinationMode.BOOTSTRAP:
1285- raise RuntimeError(f"{adapter.engine_type} bootstrap requires a decode instance")1511+ if attempt.decode_resource is None:
1286- if attempt.decode_resource is not None:1512+ raise RuntimeError(f"{adapter.engine_type} bootstrap requires a decode instance")
1513+ if attempt.prefill_resource is not None:
1514+ decode_engine_type = getattr(attempt.decode_resource.instance, "engine_type", None)
1515+ if not isinstance(decode_engine_type, str) or decode_engine_type.strip().lower() != adapter.engine_type:
1516+ raise RuntimeError(
1517+ f"P/D engine types must match: prefill={adapter.engine_type}, decode={decode_engine_type!r}"
1518+ )
1519+ return CoordinationMode.BOOTSTRAP
1520+ # Prefer cluster detection already done in create_attempt: ALLOCATE_ONLY historically
1521+ # dropped dispatch_capabilities, which would otherwise fall back to adapter HANDOFF.
1522+ if self._pd_uses_trigger:
1523+ return self._require_trigger_decode(attempt)
1524+ resource = attempt.prefill_resource or attempt.decode_resource
1525+ caps = list(getattr(resource.instance, "dispatch_capabilities", None) or []) if resource is not None else []
1526+ if _TRIGGER_CAPABILITY in caps:
G
Gganglv7 天前

严重程度: 建议

问题: _select_coordination_mode 的 per-instance dispatch caps 兜底(本行)可独立返回 TRIGGER,而 _create_attempt(651 行)按 _pd_uses_trigger 决定"先分配 D(trigger)还是先分配 P 并延迟分配 D(非 trigger)"。检测缓存的是请求开始时 unblocked 实例快照,分配却走调度器独立的实例池。若检测时刻判定为非 trigger(trigger 实例当时 CB 阻断/未注册),随后调度器却分配到带 trigger caps 的 P 实例,_select_coordination_mode 返回 TRIGGER_run_trigger_stream_attempt_native_leg_context(attempt, ROLE_D)(1441 行)因 decode_resource is NoneRuntimeError,请求按 transport_retry_limit 重试全部失败最终 500——而不是 fail-closed 的 503。

原因: 检测(客户端快照)与分配(调度器实时池)两个判定源之间没有一致性约束,防御性 caps 兜底恰好覆盖了这条不一致路径。

怎么改: 在 caps 兜底分支返回 TRIGGER 时,先检查 attempt.decode_resource 是否已分配;若为 None 则抛出明确的 503(或回退 HANDOFF),避免以 RuntimeError 形态进入重试循环;同时补一个"检测后实例池变化"的测试用例。

likedislike
tobking
tobking
7 天前 评论:
1527+ return self._require_trigger_decode(attempt)
1528+ if attempt.decode_resource is not None and attempt.prefill_resource is not None:
1287 decode_engine_type = getattr(attempt.decode_resource.instance, "engine_type", None)1529 decode_engine_type = getattr(attempt.decode_resource.instance, "engine_type", None)
1288 if not isinstance(decode_engine_type, str) or decode_engine_type.strip().lower() != adapter.engine_type:1530 if not isinstance(decode_engine_type, str) or decode_engine_type.strip().lower() != adapter.engine_type:
1289 raise RuntimeError(1531 raise RuntimeError(
@@ -1291,6 +1533,22 @@ class UnifiedPDRouter(BaseRouter):
1291 )1533 )
1292 return adapter.coordination_mode1534 return adapter.coordination_mode
1293 1535 
1536+ def _require_trigger_decode(self, attempt: AttemptContext) -> CoordinationMode:
1537+ if attempt.decode_resource is None:
1538+ self.logger.error(
1539+ "Trigger mode selected without a decode resource req_id=%s attempt=%s",
1540+ self.req_info.req_id,
1541+ attempt.attempt_seq,
1542+ )
1543+ raise HTTPException(
1544+ status_code=503,
1545+ detail=(
1546+ "Trigger P/D requires a decode instance; cluster detection and "
1547+ "allocated instance capabilities are inconsistent"
1548+ ),
1549+ )
1550+ return CoordinationMode.TRIGGER
1551+ 
1294 async def _prepare_attempt_resource(1552 async def _prepare_attempt_resource(
1295 self,1553 self,
1296 role: PDRole,1554 role: PDRole,
@@ -1316,15 +1574,18 @@ class UnifiedPDRouter(BaseRouter):
1316 raise HTTPException(status_code=503, detail=error_message)1574 raise HTTPException(status_code=503, detail=error_message)
1317 raise RuntimeError(error_message)1575 raise RuntimeError(error_message)
1318 ins, endpoint, workload = result1576 ins, endpoint, workload = result
1319- await self._record_attempt_workload(attempt_seq, role, workload)1577+ if not await self._request_manager.add_req_attempt_workload(
1320- self.req_info.update_state(ReqState.P_ALLOCATED if role == PDRole.ROLE_P else ReqState.D_ALLOCATED)1578+ self.req_info.req_id,
1321- return ScheduledResource(instance=ins, endpoint=endpoint)1579+ attempt_seq,
1322- 1580+ role,
1323- async def _record_attempt_workload(self, attempt_seq: int, role: PDRole, workload) -> None:1581+ workload,
1324- if not await self._request_manager.add_req_attempt_workload(self.req_info.req_id, attempt_seq, role, workload):1582+ ):
1583+ await self._rollback_allocated_workload(ins, endpoint, role, workload)
1325 raise RuntimeError(1584 raise RuntimeError(
1326 f"Request {self.req_info.req_id} already allocated for attempt {attempt_seq} role {role}"1585 f"Request {self.req_info.req_id} already allocated for attempt {attempt_seq} role {role}"
1327 )1586 )
1587+ self.req_info.update_state(ReqState.P_ALLOCATED if role == PDRole.ROLE_P else ReqState.D_ALLOCATED)
1588+ return ScheduledResource(instance=ins, endpoint=endpoint)
1328 1589 
1329 async def _release_attempt(self, attempt: AttemptContext, *, wait: bool = True) -> None:1590 async def _release_attempt(self, attempt: AttemptContext, *, wait: bool = True) -> None:
1330 if attempt.prefill_resource:1591 if attempt.prefill_resource:
Mmotor/coordinator/scheduler/runtime/scheduler_client.py+20-0
@@ -1383,6 +1383,26 @@ class AsyncSchedulerClient:
1383 cached = self._cache.get_instances(role)1383 cached = self._cache.get_instances(role)
1384 return [inst.id for inst in cached if inst.id not in self._cb_blocked_instances]1384 return [inst.id for inst in cached if inst.id not in self._cb_blocked_instances]
1385 1385 
1386+ def _instances_from_cache(self, role: PDRole | None = None) -> dict[int, Instance]:
1387+ if role is not None:
1388+ return {inst.id: inst for inst in self._cache.get_instances(role)}
1389+ instances: dict[int, Instance] = {}
1390+ for cached_role in (PDRole.ROLE_E, PDRole.ROLE_P, PDRole.ROLE_D, PDRole.ROLE_U):
1391+ for inst in self._cache.get_instances(cached_role):
1392+ instances[inst.id] = inst
1393+ return instances
1394+ 
1395+ async def get_local_instances(self, role: PDRole | None = None) -> dict[int, Instance]:
1396+ """Return cached instances; RPC warm-up only when the local view is empty."""
1397+ instances = self._instances_from_cache(role)
1398+ if instances:
1399+ return instances
1400+ try:
1401+ await self.get_available_instances(None)
1402+ except Exception as e:
1403+ logger.debug("get_local_instances: warm-up fetch failed: %s", e)
1404+ return self._instances_from_cache(role)
1405+ 
1386 async def has_required_instances(self) -> InstanceReadiness:1406 async def has_required_instances(self) -> InstanceReadiness:
1387 """Return InstanceReadiness from cache; warm-up fetch if needed."""1407 """Return InstanceReadiness from cache; warm-up fetch if needed."""
1388 1408 
Mmotor/coordinator/scheduler/runtime/scheduler_server.py+6-1
@@ -161,7 +161,11 @@ def _instance_from_dict(data: dict) -> Instance | None:
161 161 
162 162 
163def _serialize_instance_minimal(instance: Instance | None) -> dict:163def _serialize_instance_minimal(instance: Instance | None) -> dict:
164- """Serialize minimal fields for select/allocate result (forward and release); reduce ZMQ payload."""164+ """Serialize minimal fields for select/allocate result (forward and release); reduce ZMQ payload.
165+ 
166+ Must keep ``dispatch_capabilities``: ALLOCATE_ONLY responses are rebuilt into Instance on the
167+ Worker and used by UnifiedPDRouter._select_coordination_mode (TRIGGER vs HANDOFF).
168+ """
165 if instance is None:169 if instance is None:
166 return {}170 return {}
167 return {171 return {
@@ -170,6 +174,7 @@ def _serialize_instance_minimal(instance: Instance | None) -> dict:
170 "job_name": instance.job_name,174 "job_name": instance.job_name,
171 "model_name": instance.model_name,175 "model_name": instance.model_name,
172 "engine_type": instance.engine_type,176 "engine_type": instance.engine_type,
177+ "dispatch_capabilities": list(instance.dispatch_capabilities or []),
173 }178 }
174 179 
175 180 
Mmotor/coordinator/scheduler/scheduler.py+4-0
@@ -232,6 +232,10 @@ class Scheduler:
232 """232 """
233 return dict(self._instance_provider.get_available_instances(role))233 return dict(self._instance_provider.get_available_instances(role))
234 234 
235+ async def get_local_instances(self, role: PDRole | None = None) -> dict[int, Instance]:
236+ """Return the in-process instance view without going through GET_AVAILABLE_INSTANCES."""
237+ return dict(self._instance_provider.get_available_instances(role))
238+ 
235 async def get_available_instance_roles(self) -> set[PDRole]:239 async def get_available_instance_roles(self) -> set[PDRole]:
236 """Return roles from the in-process instance provider without scheduler IPC."""240 """Return roles from the in-process instance provider without scheduler IPC."""
237 roles: set[PDRole] = set()241 roles: set[PDRole] = set()
Mmotor/node_manager/core/services/native_engine/backends/vllm/backend.py+3-2
@@ -28,7 +28,8 @@ class VllmBackend(BaseNativeEngineBackend):
28 deploy_config.engine_config,28 deploy_config.engine_config,
29 explicit_profile=deploy_config.dispatch_profile,29 explicit_profile=deploy_config.dispatch_profile,
30 )30 )
31- if profile != DispatchProfile.HANDOFF:31+ if profile not in (DispatchProfile.HANDOFF, DispatchProfile.TRIGGER):
32 raise ValueError(32 raise ValueError(
33- f"Native vLLM P/D launch only supports handoff connectors; resolved dispatch profile is {profile.value}"33+ "Native vLLM P/D launch only supports handoff or trigger connectors; "
34+ f"resolved dispatch profile is {profile.value}"
34 )35 )
Mtests/coordinator/core/test_config.py+37-0
@@ -554,6 +554,43 @@ def test_config_validation_errors(param, value, expected_error):
554 config.validate_config()554 config.validate_config()
555 555 
556 556 
557+def test_worker_metaserver_base_port_defaults_to_12000():
558+ config = CoordinatorConfig()
559+ assert config.inference_workers_config.worker_metaserver_base_port == 12000
560+ config.validate_config()
561+ 
562+ 
563+def test_worker_metaserver_base_port_zero_is_disabled():
564+ config = CoordinatorConfig()
565+ config.inference_workers_config.worker_metaserver_base_port = 0
566+ config.validate_config()
567+ 
568+ 
569+def test_worker_metaserver_base_port_overflow_is_rejected():
570+ config = CoordinatorConfig()
571+ config.inference_workers_config.num_workers = 4
572+ config.inference_workers_config.worker_metaserver_base_port = 65534
573+ with pytest.raises(ValueError, match="worker_metaserver_base_port \\+ num_workers - 1"):
574+ config.validate_config()
575+ 
576+ 
577+def test_worker_metaserver_startup_allows_unspecified_listen_host_without_pod_ip(monkeypatch):
578+ """Listen 0.0.0.0 is valid; unreachable callback host is checked only on Trigger."""
579+ monkeypatch.delenv("POD_IP", raising=False)
580+ config = CoordinatorConfig()
581+ config.api_config.coordinator_api_host = "0.0.0.0"
582+ 
583+ config.validate_config()
584+ 
585+ 
586+def test_worker_metaserver_accepts_pod_ip_with_unspecified_listen_host(monkeypatch):
587+ monkeypatch.setenv("POD_IP", "10.0.0.8")
588+ config = CoordinatorConfig()
589+ config.api_config.coordinator_api_host = "0.0.0.0"
590+ 
591+ config.validate_config()
592+ 
593+ 
557def test_config_validation_query_encoding_defaults_ok():594def test_config_validation_query_encoding_defaults_ok():
558 """Default query_encoding (msgpack) and json both validate."""595 """Default query_encoding (msgpack) and json both validate."""
559 config = CoordinatorConfig()596 config = CoordinatorConfig()
Mtests/coordinator/process/test_inference_manager_socket.py+22-1
@@ -11,10 +11,15 @@
11"""IPv6 single-stack: create_shared_socket must pick the right address family."""11"""IPv6 single-stack: create_shared_socket must pick the right address family."""
12 12 
13import socket13import socket
14+import sys
15+from unittest.mock import MagicMock
14 16 
15import pytest17import pytest
16 18 
17-from motor.coordinator.process.inference_manager import create_shared_socket19+sys.modules.setdefault("uvloop", MagicMock())
20+ 
21+from motor.config.coordinator import CoordinatorConfig # noqa: E402
22+from motor.coordinator.process.inference_manager import create_shared_socket, metaserver_bind_host # noqa: E402
18 23 
19 24 
20@pytest.mark.skipif(not hasattr(socket, "SO_REUSEPORT"), reason="SO_REUSEPORT not available")25@pytest.mark.skipif(not hasattr(socket, "SO_REUSEPORT"), reason="SO_REUSEPORT not available")
@@ -50,3 +55,19 @@ class TestCreateSharedSocket:
50 assert sock.family == socket.AF_INET655 assert sock.family == socket.AF_INET6
51 finally:56 finally:
52 sock.close()57 sock.close()
58+ 
59+ 
60+def test_metaserver_bind_host_prefers_pod_ip(monkeypatch):
61+ monkeypatch.setenv("POD_IP", "10.0.0.8")
62+ config = CoordinatorConfig()
63+ config.api_config.coordinator_api_host = "0.0.0.0"
64+ 
65+ assert metaserver_bind_host(config) == "10.0.0.8"
66+ 
67+ 
68+def test_metaserver_bind_host_falls_back_to_api_host(monkeypatch):
69+ monkeypatch.delenv("POD_IP", raising=False)
70+ config = CoordinatorConfig()
71+ config.api_config.coordinator_api_host = "192.168.1.10"
72+ 
73+ assert metaserver_bind_host(config) == "192.168.1.10"
Atests/coordinator/process/test_metaserver_sidecar.py+123-0
@@ -0,0 +1,123 @@
1+# Copyright (c) Huawei Technologies Co., Ltd. 2026. All rights reserved.
2+# MindIE is licensed under Mulan PSL v2.
3+# You can use this software according to the terms and conditions of the Mulan PSL v2.
4+# You may obtain a copy of Mulan PSL v2 at:
5+# http://license.coscl.org.cn/MulanPSL2
6+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
7+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
8+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
9+# See the Mulan PSL v2 for more details.
10+ 
11+"""Metaserver bind/serve failure must not take down the infer Worker."""
12+ 
13+from __future__ import annotations
14+ 
15+import asyncio
16+import sys
17+from unittest.mock import MagicMock
18+ 
19+import pytest
20+ 
21+sys.modules.setdefault("uvloop", MagicMock())
22+ 
23+from motor.config.coordinator import CoordinatorConfig # noqa: E402
24+from motor.coordinator.process.inference_manager import ( # noqa: E402
25+ run_inference_and_metaserver,
26+ serve_worker_metaserver,
27+)
28+ 
29+ 
30+class _HangServer:
31+ should_exit = False
32+ 
33+ def __init__(self):
34+ self.cancelled = asyncio.Event()
35+ 
36+ async def serve(self, sockets=None):
37+ del sockets
38+ try:
39+ await asyncio.Event().wait()
40+ except asyncio.CancelledError:
41+ self.cancelled.set()
42+ raise
43+ 
44+ 
45+class _FailingServer:
46+ should_exit = False
47+ 
48+ async def serve(self, sockets=None):
49+ del sockets
50+ raise OSError("Address already in use")
51+ 
52+ 
53+class _ReleaseServer:
54+ def __init__(self):
55+ self.started = asyncio.Event()
56+ self.release = asyncio.Event()
57+ 
58+ async def serve(self, sockets=None):
59+ del sockets
60+ self.started.set()
61+ await self.release.wait()
62+ 
63+ 
64+@pytest.mark.asyncio
65+async def test_serve_worker_metaserver_clears_port_on_bind_error(caplog):
66+ config = CoordinatorConfig()
67+ config.worker_metaserver_port = 12000
68+ 
69+ await serve_worker_metaserver(_FailingServer(), config, 0, "127.0.0.1", 12000)
70+ 
71+ assert config.worker_metaserver_port is None
72+ assert any("metaserver failed" in record.getMessage() for record in caplog.records)
73+ 
74+ 
75+@pytest.mark.asyncio
76+async def test_metaserver_serve_failure_does_not_stop_infer():
77+ config = CoordinatorConfig()
78+ config.worker_metaserver_port = 12000
79+ infer = _ReleaseServer()
80+ 
81+ task = asyncio.create_task(run_inference_and_metaserver(infer, None, _FailingServer(), config, 0, "127.0.0.1"))
82+ await infer.started.wait()
83+ for _ in range(50):
84+ if config.worker_metaserver_port is None:
85+ break
86+ await asyncio.sleep(0)
87+ assert config.worker_metaserver_port is None
88+ infer.release.set()
89+ await task
90+ 
91+ 
92+@pytest.mark.asyncio
93+async def test_infer_exit_stops_metaserver_without_clearing_port():
94+ config = CoordinatorConfig()
95+ config.worker_metaserver_port = 12001
96+ infer = _ReleaseServer()
97+ meta = _HangServer()
98+ 
99+ task = asyncio.create_task(run_inference_and_metaserver(infer, None, meta, config, 1, "10.0.0.1"))
100+ await infer.started.wait()
101+ infer.release.set()
102+ await task
103+ 
104+ assert meta.cancelled.is_set()
105+ assert config.worker_metaserver_port == 12001
106+ 
107+ 
108+@pytest.mark.asyncio
109+async def test_infer_failure_stops_metaserver_and_raises():
110+ config = CoordinatorConfig()
111+ config.worker_metaserver_port = 12002
112+ meta = _HangServer()
113+ 
114+ class _BoomInfer:
115+ async def serve(self, sockets=None):
116+ del sockets
117+ raise RuntimeError("infer boom")
118+ 
119+ with pytest.raises(RuntimeError, match="infer boom"):
120+ await run_inference_and_metaserver(_BoomInfer(), None, meta, config, 2, "10.0.0.1")
121+ 
122+ assert meta.cancelled.is_set()
123+ assert config.worker_metaserver_port == 12002
Mtests/coordinator/router/test_pd_protocol_adapter.py+55-0
@@ -269,3 +269,58 @@ def test_sglang_abort_uses_native_request_id():
269 269 
270def test_vllm_does_not_claim_an_unverified_abort_endpoint():270def test_vllm_does_not_claim_an_unverified_abort_endpoint():
271 assert VllmProtocolAdapter().build_abort_request(_context()) is None271 assert VllmProtocolAdapter().build_abort_request(_context()) is None
272+ 
273+ 
274+def test_trim_vllm_engine_request_id_strips_openai_prefixes():
275+ from motor.coordinator.router.adapters.pd_protocol import trim_vllm_engine_request_id
276+ 
277+ assert trim_vllm_engine_request_id("chatcmpl-abc") == "abc"
278+ assert trim_vllm_engine_request_id("cmpl-abc-0") == "abc"
279+ assert trim_vllm_engine_request_id("raw-id") == "raw-id"
280+ 
281+ 
282+def test_vllm_trigger_decode_request_keeps_generation_and_sets_metaserver():
283+ adapter = VllmProtocolAdapter()
284+ request = {
285+ "model": "glm",
286+ "messages": [{"role": "user", "content": "hello"}],
287+ "stream": True,
288+ "max_tokens": 128,
289+ "rid": "wrong",
290+ }
291+ original = deepcopy(request)
292+ 
293+ engine_request = adapter.build_trigger_decode_request(request, _context(), "http://127.0.0.1:12000/v1/metaserver")
294+ 
295+ assert request == original
296+ assert engine_request.body["request_id"] == "engine-1"
297+ assert engine_request.body["stream"] is True
298+ assert engine_request.body["max_tokens"] == 128
299+ assert engine_request.body["kv_transfer_params"] == {
300+ "do_remote_decode": False,
301+ "do_remote_prefill": True,
302+ "metaserver": "http://127.0.0.1:12000/v1/metaserver",
303+ }
304+ 
305+ 
306+def test_vllm_trigger_prefill_request_uses_decode_kv_params():
307+ adapter = VllmProtocolAdapter()
308+ request = {"model": "glm", "messages": [], "stream": True, "max_tokens": 32, "max_completion_tokens": 16}
309+ kv_params = {
310+ "request_id": "engine-1",
311+ "do_remote_decode": True,
312+ "remote_block_ids": [1, 2],
313+ "remote_host": "10.0.0.2",
314+ "metaserver": "http://ignored",
315+ }
316+ 
317+ engine_request = adapter.build_trigger_prefill_request(request, _context(), kv_params)
318+ 
319+ assert engine_request.body["stream"] is False
320+ assert engine_request.body["max_tokens"] == 1
321+ assert engine_request.body["min_tokens"] == 1
322+ assert engine_request.body["max_completion_tokens"] == 1
323+ assert engine_request.body["kv_transfer_params"]["do_remote_decode"] is True
324+ assert engine_request.body["kv_transfer_params"]["do_remote_prefill"] is False
325+ assert engine_request.body["kv_transfer_params"]["remote_block_ids"] == [1, 2]
326+ assert "metaserver" not in engine_request.body["kv_transfer_params"]
Atests/coordinator/router/test_router_native_trigger.py+520-0
@@ -0,0 +1,520 @@
1+# Copyright (c) Huawei Technologies Co., Ltd. 2026. All rights reserved.
2+# MindIE is licensed under Mulan PSL v2.
3+# You can use this software according to the terms and conditions of the Mulan PSL v2.
4+# You may obtain a copy of the Mulan PSL v2 at:
5+# http://license.coscl.org.cn/MulanPSL2
6+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
7+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
8+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
9+# See the Mulan PSL v2 for more details.
10+ 
11+import asyncio
12+from unittest.mock import AsyncMock, MagicMock
13+ 
14+import pytest
15+from fastapi import HTTPException, Request, status
16+ 
17+from motor.common.resources.dispatch import DispatchPlan
18+from motor.common.resources.endpoint import Endpoint, EndpointStatus, Workload
19+from motor.common.resources.instance import InsStatus, Instance, ParallelConfig, PDRole
20+from motor.config.coordinator import CoordinatorConfig
21+from motor.coordinator.domain import InstanceReadiness
22+from motor.coordinator.domain.instance_manager import InstanceManager
23+from motor.coordinator.domain.request_manager import RequestManager
24+from motor.coordinator.models.request import ReqState
25+from motor.coordinator.router.dispatch import handle_metaserver_request
26+from motor.coordinator.router.dispatch_session import AttemptState, AttemptStopReason, PDDispatchSession
27+from motor.coordinator.router.strategies.unified_pd import UnifiedPDRouter
28+from motor.coordinator.scheduler.scheduler import Scheduler
29+from tests.coordinator.router.mock_openai_request import create_mock_request_info
30+from tests.coordinator.router.test_router_native_handoff import (
31+ _UnifiedPDDecodeClient,
32+ _UnifiedPDPrefillClient,
33+ _patch_unified_pd_clients,
34+)
35+ 
36+ 
37+def _trigger_instance(instance_id: int, role: PDRole) -> Instance:
38+ return Instance(
39+ job_name=f"test-job-{instance_id}",
40+ model_name=f"test-model-{instance_id}",
41+ engine_type="vllm",
42+ dispatch_capabilities=[DispatchPlan.CONCURRENT_ENGINE_SYNC.value],
43+ id=instance_id,
44+ role=role,
45+ status=InsStatus.ACTIVE,
46+ parallel_config=ParallelConfig(dp_size=1, tp_size=1),
47+ endpoints={},
48+ )
49+ 
50+ 
51+def _handoff_instance(instance_id: int, role: PDRole) -> Instance:
52+ inst = _trigger_instance(instance_id, role)
53+ inst.dispatch_capabilities = [DispatchPlan.PREFILL_HANDOFF_DECODE.value]
54+ return inst
55+ 
56+ 
57+async def _hanging_receive():
58+ await asyncio.Event().wait()
59+ 
60+ 
61+def _metaserver_raw_request(body: dict, attempt: str) -> MagicMock:
62+ raw_request = MagicMock(spec=Request)
63+ raw_request.json = AsyncMock(return_value=body)
64+ raw_request.query_params = {"attempt": attempt}
65+ raw_request.receive = AsyncMock(side_effect=_hanging_receive)
66+ return raw_request
67+ 
68+ 
69+class _BlockingPrefillClient(_UnifiedPDPrefillClient):
70+ def __init__(self):
71+ super().__init__()
72+ self.started = asyncio.Event()
73+ self.release = asyncio.Event()
74+ self.cancelled = asyncio.Event()
75+ 
76+ async def post(self, path, json=None, headers=None, timeout=None):
77+ self.started.set()
78+ try:
79+ await self.release.wait()
80+ except asyncio.CancelledError:
81+ self.cancelled.set()
82+ raise
83+ return await super().post(path, json=json, headers=headers, timeout=timeout)
84+ 
85+ 
86+class TestRouterNativeTrigger:
87+ def _make_config(self) -> CoordinatorConfig:
88+ config = CoordinatorConfig()
89+ config.worker_metaserver_port = 12000
90+ return config
91+ 
92+ def _make_router(self, req_info, monkeypatch, p_client, d_client, *, config=None, scheduler=None):
93+ config = config or self._make_config()
94+ scheduler = scheduler or Scheduler(instance_provider=InstanceManager(config), config=config)
95+ router_obj = UnifiedPDRouter(
96+ req_info,
97+ config,
98+ scheduler=scheduler,
99+ request_manager=RequestManager(config),
100+ )
101+ _patch_unified_pd_clients(monkeypatch, router_obj, p_client, d_client)
102+ return router_obj
103+ 
104+ def _patch_instances(self, monkeypatch, instance_p, instance_d, endpoint_p, endpoint_d):
105+ def mock_get_available_instances(self, role=None):
106+ if role is None:
107+ return {instance_p.id: instance_p, instance_d.id: instance_d}
108+ if role == PDRole.ROLE_P:
109+ return {instance_p.id: instance_p}
110+ if role == PDRole.ROLE_D:
111+ return {instance_d.id: instance_d}
112+ return {}
113+ 
114+ async def mock_select_and_allocate(self, role, req_info, *, target_instance_id=None, required_engine_type=None):
115+ del req_info, target_instance_id, required_engine_type
116+ if role == PDRole.ROLE_P:
117+ return instance_p, endpoint_p, Workload(active_tokens=1)
118+ if role == PDRole.ROLE_D:
119+ return instance_d, endpoint_d, Workload(active_tokens=1)
120+ return None
121+ 
122+ async def mock_update_workload(self, params):
123+ del params
124+ return True
125+ 
126+ monkeypatch.setattr(InstanceManager, "get_available_instances", mock_get_available_instances)
127+ monkeypatch.setattr(
128+ InstanceManager,
129+ "get_required_instances_status",
130+ lambda self: InstanceReadiness.REQUIRED_MET,
131+ )
132+ monkeypatch.setattr(Scheduler, "select_and_allocate", mock_select_and_allocate)
133+ monkeypatch.setattr(Scheduler, "update_workload", mock_update_workload)
134+ 
135+ @pytest.fixture
136+ def trigger_pair(self, monkeypatch):
137+ host = "127.0.0.1"
138+ instance_p = _trigger_instance(0, PDRole.ROLE_P)
139+ endpoint_p = Endpoint(id=0, ip=host, business_port="8000", mgmt_port="8000", status=EndpointStatus.NORMAL)
140+ instance_p.endpoints = {host: {0: endpoint_p}}
141+ instance_d = _trigger_instance(1, PDRole.ROLE_D)
142+ endpoint_d = Endpoint(id=1, ip=host, business_port="8001", mgmt_port="8001", status=EndpointStatus.NORMAL)
143+ instance_d.endpoints = {host: {1: endpoint_d}}
144+ self._patch_instances(monkeypatch, instance_p, instance_d, endpoint_p, endpoint_d)
145+ return instance_p, instance_d, endpoint_p, endpoint_d
146+ 
147+ @pytest.mark.asyncio
148+ async def test_trigger_sends_decode_first_with_metaserver(self, monkeypatch, trigger_pair):
149+ del trigger_pair
150+ p_client = _UnifiedPDPrefillClient()
151+ d_client = _UnifiedPDDecodeClient()
152+ req_info = await create_mock_request_info()
153+ router = self._make_router(req_info, monkeypatch, p_client, d_client)
154+ 
155+ response = await router.handle_request()
156+ chunks = [chunk async for chunk in response.body_iterator]
157+ 
158+ assert chunks
159+ assert p_client.requests == []
160+ assert len(d_client.requests) == 1
161+ decode_body = d_client.requests[0]
162+ assert decode_body["request_id"] == req_info.req_id
163+ assert decode_body["kv_transfer_params"]["do_remote_prefill"] is True
164+ assert decode_body["kv_transfer_params"]["do_remote_decode"] is False
165+ assert decode_body["kv_transfer_params"]["metaserver"].startswith("http://")
166+ assert "attempt=1" in decode_body["kv_transfer_params"]["metaserver"]
167+ assert req_info.state == ReqState.DECODE_END
168+ assert ReqState.D_ALLOCATED in req_info.status
169+ assert ReqState.P_ALLOCATED not in req_info.status
170+ 
171+ @pytest.mark.asyncio
172+ async def test_select_coordination_mode_uses_cluster_trigger_when_allocate_caps_missing(
173+ self, monkeypatch, trigger_pair
174+ ):
175+ """Guards ALLOCATE_ONLY dropping caps: cluster detection must still drive TRIGGER."""
176+ del trigger_pair
177+ req_info = await create_mock_request_info()
178+ router = self._make_router(req_info, monkeypatch, _UnifiedPDPrefillClient(), _UnifiedPDDecodeClient())
179+ attempt = await router._create_attempt(PDDispatchSession(req_info.req_id))
180+ assert attempt.decode_resource is not None
181+ attempt.decode_resource.instance.dispatch_capabilities = []
182+ 
183+ from motor.coordinator.router.adapters.pd_protocol import CoordinationMode
184+ 
185+ assert router._select_coordination_mode(attempt) == CoordinationMode.TRIGGER
186+ 
187+ @pytest.mark.asyncio
188+ async def test_select_coordination_mode_trigger_caps_without_decode_returns_503(self, monkeypatch):
189+ """Handoff allocates P first; TRIGGER caps without D must fail-closed 503, not retry to 500."""
190+ host = "127.0.0.1"
191+ instance_p = _handoff_instance(0, PDRole.ROLE_P)
192+ endpoint_p = Endpoint(id=0, ip=host, business_port="8000", mgmt_port="8000", status=EndpointStatus.NORMAL)
193+ instance_p.endpoints = {host: {0: endpoint_p}}
194+ instance_d = _handoff_instance(1, PDRole.ROLE_D)
195+ endpoint_d = Endpoint(id=1, ip=host, business_port="8001", mgmt_port="8001", status=EndpointStatus.NORMAL)
196+ instance_d.endpoints = {host: {1: endpoint_d}}
197+ self._patch_instances(monkeypatch, instance_p, instance_d, endpoint_p, endpoint_d)
198+ 
199+ req_info = await create_mock_request_info()
200+ router = self._make_router(req_info, monkeypatch, _UnifiedPDPrefillClient(), _UnifiedPDDecodeClient())
201+ attempt = await router._create_attempt(PDDispatchSession(req_info.req_id))
202+ assert router._pd_uses_trigger is False
203+ assert attempt.decode_resource is None
204+ attempt.prefill_resource.instance.dispatch_capabilities = [DispatchPlan.CONCURRENT_ENGINE_SYNC.value]
205+ 
206+ with pytest.raises(HTTPException) as exc_info:
207+ router._select_coordination_mode(attempt)
208+ assert exc_info.value.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
209+ assert "decode instance" in str(exc_info.value.detail).lower()
210+ 
211+ @pytest.mark.asyncio
212+ async def test_pd_cluster_uses_trigger_reads_local_instances_not_rpc(self, monkeypatch, trigger_pair):
213+ """Trigger detection must use the Worker-local instance view, not GET_AVAILABLE_INSTANCES."""
214+ del trigger_pair
215+ req_info = await create_mock_request_info()
216+ router = self._make_router(req_info, monkeypatch, _UnifiedPDPrefillClient(), _UnifiedPDDecodeClient())
217+ scheduler = router._scheduler
218+ local_spy = AsyncMock(side_effect=scheduler.get_local_instances)
219+ avail_spy = AsyncMock(side_effect=scheduler.get_available_instances)
220+ monkeypatch.setattr(scheduler, "get_local_instances", local_spy)
221+ monkeypatch.setattr(scheduler, "get_available_instances", avail_spy)
222+ 
223+ assert await router._pd_cluster_uses_trigger() is True
224+ assert local_spy.await_count >= 1
225+ avail_spy.assert_not_awaited()
226+ 
227+ @pytest.mark.parametrize(
228+ "pod_ip,configured_host,expected_url",
229+ [
230+ ("10.0.0.8", "0.0.0.0", "http://10.0.0.8:12000/v1/metaserver?attempt=3"),
231+ ("2001:db8::8", "::", "http://[2001:db8::8]:12000/v1/metaserver?attempt=3"),
232+ (None, "coordinator.example.com", "http://coordinator.example.com:12000/v1/metaserver?attempt=3"),
233+ ],
234+ )
235+ def test_trigger_metaserver_url_uses_reachable_host(self, monkeypatch, pod_ip, configured_host, expected_url):
236+ if pod_ip is None:
237+ monkeypatch.delenv("POD_IP", raising=False)
238+ else:
239+ monkeypatch.setenv("POD_IP", pod_ip)
240+ config = self._make_config()
241+ config.api_config.coordinator_api_host = configured_host
242+ router = UnifiedPDRouter(MagicMock(), config, scheduler=MagicMock(), request_manager=MagicMock())
243+ attempt = MagicMock(attempt_seq=3)
244+ 
245+ assert router._trigger_metaserver_url(attempt) == expected_url
246+ 
247+ @pytest.mark.parametrize("configured_host", ["0.0.0.0", "::"])
248+ def test_trigger_metaserver_url_rejects_unspecified_host_without_pod_ip(self, monkeypatch, configured_host, caplog):
249+ monkeypatch.delenv("POD_IP", raising=False)
250+ config = self._make_config()
251+ config.api_config.coordinator_api_host = configured_host
252+ router = UnifiedPDRouter(MagicMock(), config, scheduler=MagicMock(), request_manager=MagicMock())
253+ 
254+ with caplog.at_level("ERROR"):
255+ with pytest.raises(HTTPException) as exc_info:
256+ router._trigger_metaserver_url(MagicMock(attempt_seq=1))
257+ 
258+ assert exc_info.value.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
259+ assert "POD_IP" in str(exc_info.value.detail)
260+ assert any(
261+ "callback host is unreachable" in record.getMessage() and configured_host in record.getMessage()
262+ for record in caplog.records
263+ )
264+ 
265+ @pytest.mark.asyncio
266+ async def test_metaserver_forwards_prefill_with_remote_blocks(self, monkeypatch, trigger_pair):
267+ del trigger_pair
268+ p_client = _UnifiedPDPrefillClient()
269+ d_client = _UnifiedPDDecodeClient()
270+ config = self._make_config()
271+ req_info = await create_mock_request_info()
272+ request_manager = RequestManager(config)
273+ await request_manager.add_req_info(req_info)
274+ router = UnifiedPDRouter(
275+ req_info,
276+ config,
277+ scheduler=Scheduler(instance_provider=InstanceManager(config), config=config),
278+ request_manager=request_manager,
279+ )
280+ _patch_unified_pd_clients(monkeypatch, router, p_client, d_client)
281+ attempt = await router._create_attempt(PDDispatchSession(req_info.req_id))
282+ router._bind_trigger_attempt(attempt)
283+ 
284+ kv_params = {
285+ "request_id": f"chatcmpl-{req_info.req_id}",
286+ "do_remote_decode": True,
287+ "remote_block_ids": [9, 8],
288+ "remote_host": "10.0.0.8",
289+ "remote_port": 15555,
290+ }
291+ body = await router.handle_metaserver_request(kv_params)
292+ assert body["kv_transfer_params"]["do_remote_prefill"] is True
293+ assert len(p_client.requests) == 1
294+ prefill_body = p_client.requests[0]
295+ assert prefill_body["stream"] is False
296+ assert prefill_body["max_tokens"] == 1
297+ assert prefill_body["kv_transfer_params"]["remote_block_ids"] == [9, 8]
298+ assert prefill_body["kv_transfer_params"]["do_remote_decode"] is True
299+ assert ReqState.P_ALLOCATED in req_info.status
300+ 
301+ @pytest.mark.asyncio
302+ async def test_metaserver_prefill_is_cancelled_with_attempt(self, monkeypatch, trigger_pair):
303+ del trigger_pair
304+ p_client = _BlockingPrefillClient()
305+ config = self._make_config()
306+ req_info = await create_mock_request_info()
307+ request_manager = RequestManager(config)
308+ await request_manager.add_req_info(req_info)
309+ router = UnifiedPDRouter(
310+ req_info,
311+ config,
312+ scheduler=Scheduler(instance_provider=InstanceManager(config), config=config),
313+ request_manager=request_manager,
314+ )
315+ _patch_unified_pd_clients(monkeypatch, router, p_client, _UnifiedPDDecodeClient())
316+ attempt = await router._create_attempt(PDDispatchSession(req_info.req_id))
317+ router._bind_trigger_attempt(attempt)
318+ 
319+ callback_task = asyncio.create_task(
320+ router.handle_metaserver_request(
321+ {
322+ "request_id": req_info.req_id,
323+ "do_remote_decode": True,
324+ "remote_block_ids": [9, 8],
325+ "remote_host": "10.0.0.8",
326+ "remote_port": 15555,
327+ }
328+ )
329+ )
330+ await p_client.started.wait()
331+ try:
332+ await router._stop_attempt(attempt, AttemptStopReason.CLIENT_DISCONNECT)
333+ assert callback_task.done()
334+ assert p_client.cancelled.is_set()
335+ assert attempt.state == AttemptState.STOPPED
336+ finally:
337+ callback_task.cancel()
338+ await asyncio.gather(callback_task, return_exceptions=True)
339+ 
340+ @pytest.mark.asyncio
341+ async def test_metaserver_disconnect_stops_active_attempt(self, monkeypatch, trigger_pair):
342+ del trigger_pair
343+ p_client = _BlockingPrefillClient()
344+ config = self._make_config()
345+ req_info = await create_mock_request_info()
346+ request_manager = RequestManager(config)
347+ await request_manager.add_req_info(req_info)
348+ router = UnifiedPDRouter(
349+ req_info,
350+ config,
351+ scheduler=Scheduler(instance_provider=InstanceManager(config), config=config),
352+ request_manager=request_manager,
353+ )
354+ _patch_unified_pd_clients(monkeypatch, router, p_client, _UnifiedPDDecodeClient())
355+ attempt = await router._create_attempt(PDDispatchSession(req_info.req_id))
356+ router._bind_trigger_attempt(attempt)
357+ 
358+ callback_task = asyncio.create_task(
359+ router.handle_metaserver_request(
360+ {
361+ "request_id": req_info.req_id,
362+ "do_remote_decode": True,
363+ "remote_block_ids": [9, 8],
364+ "remote_host": "10.0.0.8",
365+ "remote_port": 15555,
366+ }
367+ )
368+ )
369+ await p_client.started.wait()
370+ callback_task.cancel("decode metaserver connection closed")
371+ await asyncio.gather(callback_task, return_exceptions=True)
372+ 
373+ assert attempt.state == AttemptState.STOPPED
374+ assert await request_manager.get_req_attempt_workload(req_info.req_id, 1, PDRole.ROLE_P) is None
375+ assert await request_manager.get_req_attempt_workload(req_info.req_id, 1, PDRole.ROLE_D) is None
376+ 
377+ @pytest.mark.asyncio
378+ async def test_metaserver_retry_waits_for_first_prefill_and_is_idempotent(self, monkeypatch, trigger_pair):
379+ del trigger_pair
380+ p_client = _BlockingPrefillClient()
381+ config = self._make_config()
382+ req_info = await create_mock_request_info()
383+ request_manager = RequestManager(config)
384+ await request_manager.add_req_info(req_info)
385+ router = UnifiedPDRouter(
386+ req_info,
387+ config,
388+ scheduler=Scheduler(instance_provider=InstanceManager(config), config=config),
389+ request_manager=request_manager,
390+ )
391+ _patch_unified_pd_clients(monkeypatch, router, p_client, _UnifiedPDDecodeClient())
392+ attempt = await router._create_attempt(PDDispatchSession(req_info.req_id))
393+ router._bind_trigger_attempt(attempt)
394+ kv_params = {
395+ "request_id": req_info.req_id,
396+ "do_remote_decode": True,
397+ "remote_block_ids": [9, 8],
398+ "remote_host": "10.0.0.8",
399+ "remote_port": 15555,
400+ }
401+ 
402+ first_callback = asyncio.create_task(router.handle_metaserver_request(kv_params))
403+ await p_client.started.wait()
404+ retry_callback = asyncio.create_task(router.handle_metaserver_request(kv_params))
405+ await asyncio.sleep(0)
406+ assert not retry_callback.done()
407+ 
408+ p_client.release.set()
409+ first_result, retry_result = await asyncio.gather(first_callback, retry_callback)
410+ 
411+ assert first_result["kv_transfer_params"]["do_remote_prefill"] is True
412+ assert retry_result == {}
413+ assert len(p_client.requests) == 1
414+ 
415+ @pytest.mark.asyncio
416+ async def test_attempt_allocation_rolls_back_when_local_workload_already_exists(self, monkeypatch, trigger_pair):
417+ del trigger_pair
418+ config = self._make_config()
419+ req_info = await create_mock_request_info()
420+ request_manager = RequestManager(config)
421+ scheduler = Scheduler(instance_provider=InstanceManager(config), config=config)
422+ scheduler.update_workload = AsyncMock(return_value=True)
423+ router = UnifiedPDRouter(
424+ req_info,
425+ config,
426+ scheduler=scheduler,
427+ request_manager=request_manager,
428+ )
429+ await request_manager.add_req_attempt_workload(
430+ req_info.req_id,
431+ 1,
432+ PDRole.ROLE_P,
433+ Workload(active_tokens=1),
434+ )
435+ 
436+ with pytest.raises(RuntimeError, match="already allocated"):
437+ await router._prepare_attempt_resource(PDRole.ROLE_P, 1)
438+ 
439+ params = scheduler.update_workload.await_args.args[0]
440+ assert params.workload_change.active_tokens == -1
441+ 
442+ @pytest.mark.asyncio
443+ async def test_metaserver_unknown_request_id_returns_404(self, monkeypatch, trigger_pair):
444+ del trigger_pair, monkeypatch
445+ config = self._make_config()
446+ request_manager = RequestManager(config)
447+ raw_request = _metaserver_raw_request({"request_id": "missing"}, "1")
448+ 
449+ with pytest.raises(HTTPException) as exc_info:
450+ await handle_metaserver_request(
451+ raw_request,
452+ config,
453+ scheduler=Scheduler(instance_provider=InstanceManager(config), config=config),
454+ request_manager=request_manager,
455+ )
456+ assert exc_info.value.status_code == status.HTTP_404_NOT_FOUND
457+ 
458+ @pytest.mark.asyncio
459+ async def test_metaserver_stale_attempt_returns_409(self, monkeypatch, trigger_pair):
460+ del trigger_pair
461+ p_client = _UnifiedPDPrefillClient()
462+ d_client = _UnifiedPDDecodeClient()
463+ config = self._make_config()
464+ req_info = await create_mock_request_info()
465+ request_manager = RequestManager(config)
466+ await request_manager.add_req_info(req_info)
467+ router = UnifiedPDRouter(
468+ req_info,
469+ config,
470+ scheduler=Scheduler(instance_provider=InstanceManager(config), config=config),
471+ request_manager=request_manager,
472+ )
473+ _patch_unified_pd_clients(monkeypatch, router, p_client, d_client)
474+ attempt = await router._create_attempt(PDDispatchSession(req_info.req_id))
475+ router._bind_trigger_attempt(attempt)
476+ 
477+ raw_request = _metaserver_raw_request({"request_id": req_info.req_id}, "99")
478+ with pytest.raises(HTTPException) as exc_info:
479+ await handle_metaserver_request(
480+ raw_request,
481+ config,
482+ scheduler=router._scheduler,
483+ request_manager=request_manager,
484+ )
485+ assert exc_info.value.status_code == status.HTTP_409_CONFLICT
486+ 
487+ @pytest.mark.asyncio
488+ async def test_trigger_without_metaserver_port_fails(self, monkeypatch, trigger_pair):
489+ del trigger_pair
490+ config = CoordinatorConfig()
491+ req_info = await create_mock_request_info()
492+ router = self._make_router(
493+ req_info,
494+ monkeypatch,
495+ _UnifiedPDPrefillClient(),
496+ _UnifiedPDDecodeClient(),
497+ config=config,
498+ )
499+ with pytest.raises(HTTPException) as exc_info:
500+ await router.handle_request()
501+ assert exc_info.value.status_code == 503
502+ assert "worker_metaserver_base_port" in str(exc_info.value.detail)
503+ 
504+ @pytest.mark.asyncio
505+ async def test_mixed_handoff_and_trigger_returns_503(self, monkeypatch):
506+ host = "127.0.0.1"
507+ instance_p = _handoff_instance(0, PDRole.ROLE_P)
508+ endpoint_p = Endpoint(id=0, ip=host, business_port="8000", mgmt_port="8000", status=EndpointStatus.NORMAL)
509+ instance_p.endpoints = {host: {0: endpoint_p}}
510+ instance_d = _trigger_instance(1, PDRole.ROLE_D)
511+ endpoint_d = Endpoint(id=1, ip=host, business_port="8001", mgmt_port="8001", status=EndpointStatus.NORMAL)
512+ instance_d.endpoints = {host: {1: endpoint_d}}
513+ self._patch_instances(monkeypatch, instance_p, instance_d, endpoint_p, endpoint_d)
514+ 
515+ req_info = await create_mock_request_info()
516+ router = self._make_router(req_info, monkeypatch, _UnifiedPDPrefillClient(), _UnifiedPDDecodeClient())
517+ with pytest.raises(HTTPException) as exc_info:
518+ await router.handle_request()
519+ assert exc_info.value.status_code == 503
520+ assert "Mixed vLLM" in str(exc_info.value.detail)
Mtests/coordinator/scheduler/test_scheduler_client.py+42-0
@@ -688,6 +688,48 @@ class TestAsyncSchedulerClient:
688 assert result == {PDRole.ROLE_P, PDRole.ROLE_D}688 assert result == {PDRole.ROLE_P, PDRole.ROLE_D}
689 self.mock_transport.send_request.assert_not_awaited()689 self.mock_transport.send_request.assert_not_awaited()
690 690 
691+ @pytest.mark.asyncio
692+ async def test_get_local_instances_uses_cache_without_transport(self):
693+ """A warm cache is the local instance view; do not issue GET_AVAILABLE_INSTANCES."""
694+ mock_p = _make_instance(1, "prefill")
695+ 
696+ def _get_instances_side_effect(role):
697+ return [mock_p] if role == PDRole.ROLE_P else []
698+ 
699+ self.mock_cache.get_instances.side_effect = _get_instances_side_effect
700+ 
701+ result = await self.client.get_local_instances(PDRole.ROLE_P)
702+ 
703+ assert result[1] is mock_p
704+ self.mock_transport.send_request.assert_not_awaited()
705+ 
706+ @pytest.mark.asyncio
707+ async def test_get_local_instances_warms_up_when_cache_empty(self):
708+ """An empty local view may warm-up once via GET_AVAILABLE_INSTANCES."""
709+ inst_dict = _build_instance_dict(instance_id=7, role="prefill")
710+ self._mock_send_request(
711+ SchedulerResponseType.SUCCESS,
712+ {"instances": [inst_dict]},
713+ )
714+ cached: dict[PDRole, list] = {
715+ PDRole.ROLE_E: [],
716+ PDRole.ROLE_P: [],
717+ PDRole.ROLE_D: [],
718+ PDRole.ROLE_U: [],
719+ }
720+ 
721+ async def _replace_all(role, instances):
722+ cached[role] = list(instances)
723+ 
724+ self.mock_cache.replace_all = AsyncMock(side_effect=_replace_all)
725+ self.mock_cache.get_instances.side_effect = lambda role: cached.get(role, [])
726+ 
727+ result = await self.client.get_local_instances(PDRole.ROLE_P)
728+ 
729+ self.mock_transport.send_request.assert_awaited_once()
730+ assert 7 in result
731+ assert result[7].id == 7
732+ 
691 # -- test_has_required_instances ----------------------------------------733 # -- test_has_required_instances ----------------------------------------
692 734 
693 @pytest.mark.asyncio735 @pytest.mark.asyncio
Mtests/coordinator/scheduler/test_scheduler_server_main.py+13-3
@@ -193,14 +193,24 @@ class TestSerializeInstanceMinimal:
193 193 
194 def test_valid_instance_returns_minimal_fields(self):194 def test_valid_instance_returns_minimal_fields(self):
195 inst = _make_instance(7, (70,), role=PDRole.ROLE_D)195 inst = _make_instance(7, (70,), role=PDRole.ROLE_D)
196+ inst.dispatch_capabilities = ["concurrent_engine_sync"]
197+ inst.engine_type = "vllm"
196 result = _serialize_instance_minimal(inst)198 result = _serialize_instance_minimal(inst)
197 assert result["id"] == 7199 assert result["id"] == 7
198 assert result["role"] == PDRole.ROLE_D200 assert result["role"] == PDRole.ROLE_D
199 assert result["job_name"] == inst.job_name201 assert result["job_name"] == inst.job_name
200 assert result["model_name"] == "test_model"202 assert result["model_name"] == "test_model"
201- assert result["engine_type"] is None203+ assert result["engine_type"] == "vllm"
202- assert "dispatch_capabilities" not in result204+ assert result["dispatch_capabilities"] == ["concurrent_engine_sync"]
203- assert len(result) == 5205+ assert len(result) == 6
206+ 
207+ def test_allocate_only_roundtrip_keeps_trigger_capability(self):
208+ """ALLOCATE_ONLY minimal payload must keep caps so Worker can select TRIGGER."""
209+ inst = _make_instance(8, (80,), role=PDRole.ROLE_D, engine_type="vllm")
210+ inst.dispatch_capabilities = ["concurrent_engine_sync"]
211+ restored = _instance_from_dict(_serialize_instance_minimal(inst))
212+ assert restored is not None
213+ assert restored.dispatch_capabilities == ["concurrent_engine_sync"]
204 214 
205 215 
206class TestSerializeEndpointMinimal:216class TestSerializeEndpointMinimal:
Mtests/node_manager/core/services/native_engine/test_backends.py+13-9
@@ -169,7 +169,6 @@ def test_vllm_backend_accepts_multi_connector_with_handoff_transport():
169@pytest.mark.parametrize(169@pytest.mark.parametrize(
170 "connector,profile",170 "connector,profile",
171 [171 [
172- ("MooncakeLayerwiseConnector", "trigger"),
173 ("UnknownConnector", "unknown"),172 ("UnknownConnector", "unknown"),
174 ],173 ],
175)174)
@@ -186,7 +185,17 @@ def test_vllm_backend_rejects_non_handoff_pd_connector(connector, profile):
186 VllmBackend().prepare(context)185 VllmBackend().prepare(context)
187 186 
188 187 
189-def test_vllm_backend_rejects_explicit_trigger_profile():188+def test_vllm_backend_accepts_layerwise_connector():
189+ context = _context()
190+ spec = _build_with_config(
191+ VllmBackend(),
192+ context,
193+ _endpoint(engine_type="vllm", role="prefill", connector="MooncakeLayerwiseConnector"),
194+ )
195+ assert spec.argv[:2] == ("vllm", "serve")
196+ 
197+ 
198+def test_vllm_backend_accepts_explicit_trigger_profile():
190 context = _context()199 context = _context()
191 endpoint = _endpoint(200 endpoint = _endpoint(
192 engine_type="vllm",201 engine_type="vllm",
@@ -195,13 +204,8 @@ def test_vllm_backend_rejects_explicit_trigger_profile():
195 )204 )
196 endpoint.deploy_config.dispatch_profile = "trigger"205 endpoint.deploy_config.dispatch_profile = "trigger"
197 206 
198- with (207+ spec = _build_with_config(VllmBackend(), context, endpoint)
199- patch(208+ assert spec.argv[:2] == ("vllm", "serve")
200- "motor.node_manager.core.services.native_engine.backends.base.build_endpoint_config", return_value=endpoint
201- ),
202- pytest.raises(ValueError, match="resolved dispatch profile is trigger"),
203- ):
204- VllmBackend().prepare(context)
205 209 
206 210 
207def test_vllm_backend_allows_union_without_kv_connector():211def test_vllm_backend_allows_union_without_kv_connector():
Mtests/node_manager/test_config.py+6-4
@@ -700,7 +700,9 @@ def test_vllm_multi_connector_infers_transport_connector_capability():
700 assert NodeManagerConfig._infer_dispatch_capabilities(_multi_connector("NixlConnector")) == [700 assert NodeManagerConfig._infer_dispatch_capabilities(_multi_connector("NixlConnector")) == [
701 DispatchPlan.PREFILL_HANDOFF_DECODE.value701 DispatchPlan.PREFILL_HANDOFF_DECODE.value
702 ]702 ]
703- assert NodeManagerConfig._infer_dispatch_capabilities(_multi_connector("MooncakeLayerwiseConnector")) == []703+ assert NodeManagerConfig._infer_dispatch_capabilities(_multi_connector("MooncakeLayerwiseConnector")) == [
704+ DispatchPlan.CONCURRENT_ENGINE_SYNC.value
705+ ]
704 706 
705 707 
706def test_vllm_multi_connector_ignores_non_transport_connector_profiles():708def test_vllm_multi_connector_ignores_non_transport_connector_profiles():
@@ -720,7 +722,7 @@ def test_vllm_multi_connector_ignores_non_transport_connector_profiles():
720 },722 },
721 }723 }
722 724 
723- assert NodeManagerConfig._infer_dispatch_capabilities(engine_config) == []725+ assert NodeManagerConfig._infer_dispatch_capabilities(engine_config) == [DispatchPlan.CONCURRENT_ENGINE_SYNC.value]
724 726 
725 727 
726def test_vllm_multi_connector_requires_transport_and_store_connectors():728def test_vllm_multi_connector_requires_transport_and_store_connectors():
@@ -786,7 +788,7 @@ def test_user_dispatch_capabilities_cannot_enable_unknown_connector():
786 assert config_data["basic_config"]["dispatch_capabilities"] == []788 assert config_data["basic_config"]["dispatch_capabilities"] == []
787 789 
788 790 
789-def test_vllm_layerwise_connector_does_not_advertise_unsupported_native_capability():791+def test_vllm_layerwise_connector_advertises_concurrent_capability():
790 capabilities = NodeManagerConfig._infer_dispatch_capabilities(792 capabilities = NodeManagerConfig._infer_dispatch_capabilities(
791 {793 {
792 "engine_type": "vllm",794 "engine_type": "vllm",
@@ -798,7 +800,7 @@ def test_vllm_layerwise_connector_does_not_advertise_unsupported_native_capabili
798 }800 }
799 )801 )
800 802 
801- assert capabilities == []803+ assert capabilities == [DispatchPlan.CONCURRENT_ENGINE_SYNC.value]
802 804 
803 805 
804def test_sglang_infers_concurrent_capability():806def test_sglang_infers_concurrent_capability():