gahow/agentic-pd-hybrid
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: KVC v1-v4 debug journey + raise session soft_cap to 16

Browse Source 
Document the iterative debugging from v1 (broken KVC) through v4
(routing fixed + session cap raised), with code-level analysis of
the two main bugs encountered:

1. v2 root cause (mis-diagnosed previously as `allow_local_prefill`):
   `--policy default` for KVC mechanism caused replay's round-robin
   policy and the PD router's round-robin to diverge, sending requests
   with `session_params` to a D worker that did not have the session
   open. Resulted in 56-61% truncation with finish_reason
   "session id X does not exist".
   Fix: use `--policy kv-aware` (sweep_tp1_v3_kvaware.sh) so replay
   emits `x-smg-target-worker` and PD router uses consistent_hashing.

2. v3 new bottleneck: `pd-router-fallback-large-append-session-cap`
   dominated 52-65% of requests. Root cause was hardcoded
   `min(4, ...)` in `_decode_session_soft_cap`. With 7 D workers x 4
   sessions = 28 slots for 52 trace sessions, ~24 sessions starved
   permanently (bimodal direct-to-D rate of 0% or 99%).
   Fix: raise the cap to 16 (replay.py).

Also includes the v3 finding that direct-to-d-session path P50=0.495s
and TTFT P50=0.043s already beats the 8-way DP baseline (0.65s/0.093s)
- the KVC core mechanism works when fallback paths are avoided.

Files:
- docs/KVC_DEBUG_JOURNEY_V1_TO_V4.md: full journey + code location index
- docs/SWEBENCH_EXPERIMENT_{PROGRESS,RESULTS}.md: prior session notes
- scripts/sweep_tp1_v{2,3,4}*.sh: experiment driver scripts
- src/agentic_pd_hybrid/replay.py: cap 4 -> 16, audit fields
- src/agentic_pd_hybrid/pd_router.py: strip session_params from prefill
- src/agentic_pd_hybrid/metrics.py: truncated_request_count

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
kzlin
2026-04-28 21:10:41 +08:00
parent e9062b1d6e
commit c9d350b372
24 changed files with 1672 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/agentic_pd_hybrid/benchmark.py
View File

@@ -119,6 +119,8 @@ def run_live_benchmark(config: BenchmarkConfig) -> BenchmarkArtifacts:
try:
signal.signal(signal.SIGINT, _handle_termination)
signal.signal(signal.SIGTERM, _handle_termination)
_mechanisms_with_router = {"pd-disaggregation", "kvcache-centric", "pd-colo"}
_naive_dp = config.mechanism_name == "pd-colo"
if config.launch_stack:
stack = launch_pd_stack(
topology=topology,
@@ -132,18 +134,19 @@ def run_live_benchmark(config: BenchmarkConfig) -> BenchmarkArtifacts:
else config.timeout_s
),
include_router=(
config.mechanism_name in {"pd-disaggregation", "kvcache-centric"}
config.mechanism_name in _mechanisms_with_router
),
naive_dp=_naive_dp,
)
router_url = (
stack.router_url
if config.mechanism_name in {"pd-disaggregation", "kvcache-centric"}
if config.mechanism_name in _mechanisms_with_router
else None
)
else:
router_url = (
topology.router_url
if config.mechanism_name in {"pd-disaggregation", "kvcache-centric"}
if config.mechanism_name in _mechanisms_with_router
else None
)

9
src/agentic_pd_hybrid/cli.py
View File

@@ -455,11 +455,18 @@ def main() -> None:
if args.command == "print-launch":
topology = _topology_from_args(args)
has_pd = bool(topology.prefill_workers and topology.decode_workers)
has_direct_only = bool(
topology.direct_workers
and not topology.prefill_workers
and not topology.decode_workers
)
plan = build_launch_plan(
topology,
prefill_policy=args.prefill_policy,
decode_policy=args.decode_policy,
include_router=bool(topology.prefill_workers and topology.decode_workers),
include_router=has_pd or has_direct_only,
naive_dp=has_direct_only,
)
print(plan.render())
return

70
src/agentic_pd_hybrid/launcher.py
View File

@@ -34,7 +34,24 @@ def build_launch_plan(
decode_policy: str = "manual",
include_router: bool = True,
router_request_timeout_s: float | None = None,
naive_dp: bool = False,
) -> LaunchPlan:
router_command: tuple[str, ...] | None = None
if include_router:
if topology.prefill_workers and topology.decode_workers:
router_command = _build_router_command(
topology,
prefill_policy=prefill_policy,
decode_policy=decode_policy,
request_timeout_s=router_request_timeout_s,
)
elif naive_dp and topology.direct_workers:
router_command = _build_dp_router_command(
topology,
backend_policy=decode_policy,
request_timeout_s=router_request_timeout_s,
)
return LaunchPlan(
prefill_commands=tuple(
_build_server_command(topology, worker) for worker in topology.prefill_workers
@@ -43,24 +60,17 @@ def build_launch_plan(
_build_server_command(topology, worker) for worker in topology.decode_workers
),
direct_commands=tuple(
_build_server_command(topology, worker) for worker in topology.direct_workers
),
router_command=(
_build_router_command(
topology,
prefill_policy=prefill_policy,
decode_policy=decode_policy,
request_timeout_s=router_request_timeout_s,
)
if include_router and topology.prefill_workers and topology.decode_workers
else None
_build_server_command(topology, worker, naive_dp=naive_dp)
for worker in topology.direct_workers
),
router_command=router_command,
)
def _build_server_command(
topology: SingleNodeTopology,
worker: WorkerSpec,
naive_dp: bool = False,
) -> tuple[str, ...]:
command = [
sys.executable,
@@ -76,11 +86,15 @@ def _build_server_command(
str(worker.port),
"--base-gpu-id",
str(worker.gpu_id),
"--disaggregation-mode",
_disaggregation_mode_for(worker),
"--disaggregation-transfer-backend",
topology.transfer_backend,
]
# Naive DP direct workers: no disaggregation flags at all
if not (naive_dp and worker.role == "direct"):
command.extend([
"--disaggregation-mode",
_disaggregation_mode_for(worker),
"--disaggregation-transfer-backend",
topology.transfer_backend,
])
if worker.tp_size > 1:
command.extend(["--tp-size", str(worker.tp_size)])
if topology.trust_remote_code:
@@ -135,6 +149,32 @@ def _build_router_command(
return tuple(command)
def _build_dp_router_command(
topology: SingleNodeTopology,
*,
backend_policy: str,
request_timeout_s: float | None,
) -> tuple[str, ...]:
command: list[str] = [
sys.executable,
"-B",
"-u",
"-m",
"agentic_pd_hybrid.pd_router",
"--host",
topology.router_host,
"--port",
str(topology.router_port),
"--backend-policy",
backend_policy,
]
if request_timeout_s is not None:
command.extend(["--request-timeout-s", str(request_timeout_s)])
for worker in topology.direct_workers:
command.extend(["--backend", worker.url])
return tuple(command)
def _render_named_command(name: str, command: tuple[str, ...]) -> str:
return f"# {name}\n" + " ".join(shlex.quote(part) for part in command)

20
src/agentic_pd_hybrid/metrics.py
View File

@@ -43,6 +43,9 @@ class RequestMetrics:
ttft_s: float | None
tpot_s: float | None
error: str | None = None
actual_output_tokens: int | None = None
requested_output_tokens: int | None = None
finish_reason: str | None = None
@classmethod
def from_decision(
@@ -63,6 +66,9 @@ class RequestMetrics:
prefill_request_priority: int | None = None,
decode_request_priority: int | None = None,
error: str | None = None,
actual_output_tokens: int | None = None,
requested_output_tokens: int | None = None,
finish_reason: str | None = None,
) -> "RequestMetrics":
return cls(
request_id=request.request_id,
@@ -95,6 +101,9 @@ class RequestMetrics:
ttft_s=ttft_s,
tpot_s=tpot_s,
error=error,
actual_output_tokens=actual_output_tokens,
requested_output_tokens=requested_output_tokens,
finish_reason=finish_reason,
)
@@ -158,6 +167,17 @@ def write_summary_json(
str(key): value for key, value in sorted(decode_priorities.items())
},
"error_count": sum(1 for row in rows if row.error is not None),
"truncated_request_count": sum(
1
for row in rows
if row.actual_output_tokens is not None
and row.requested_output_tokens is not None
and row.requested_output_tokens > 1
and row.actual_output_tokens < row.requested_output_tokens * 0.5
),
"actual_output_tokens_stats": _stats(
[float(row.actual_output_tokens) for row in rows if row.actual_output_tokens is not None]
),
}
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as handle:

193
src/agentic_pd_hybrid/pd_router.py
View File

@@ -74,8 +74,58 @@ class RouterState:
return idx
@dataclass
class DpRouterConfig:
host: str
port: int
backend_urls: list[str]
backend_policy: str = "round_robin"
request_timeout_s: float = 1800.0
class DpRouterState:
"""DP (data-parallel) router: forward each request to exactly one backend."""
def __init__(self, config: DpRouterConfig):
if not config.backend_urls:
raise ValueError("At least one backend worker is required")
self.config = config
self.cursor = 0
self.sticky_map: dict[str, int] = {}
def select_backend(self, headers: dict[str, str]) -> str:
idx = self._select_index(headers)
return self.config.backend_urls[idx]
def _select_index(self, headers: dict[str, str]) -> int:
target_worker = headers.get("x-smg-target-worker")
routing_key = headers.get("x-smg-routing-key")
if (
self.config.backend_policy == "consistent_hashing"
and target_worker is not None
):
idx = int(target_worker)
if 0 <= idx < len(self.config.backend_urls):
return idx
if self.config.backend_policy == "manual" and routing_key:
cached = self.sticky_map.get(routing_key)
if cached is not None:
return cached
idx = self.cursor % len(self.config.backend_urls)
self.cursor += 1
self.sticky_map[routing_key] = idx
return idx
idx = self.cursor % len(self.config.backend_urls)
self.cursor += 1
return idx
app = FastAPI()
router_state: RouterState | None = None
dp_state: DpRouterState | None = None
@app.get("/health")
@@ -85,6 +135,16 @@ async def health() -> Response:
@app.get("/health_generate")
async def health_generate() -> Response:
if dp_state is not None:
async with aiohttp.ClientSession() as session:
tasks = [
session.get(f"{url}/health_generate")
for url in dp_state.config.backend_urls
]
for response in asyncio.as_completed(tasks):
async with await response:
pass
return Response(status_code=200)
state = _require_state()
async with aiohttp.ClientSession() as session:
tasks = []
@@ -101,6 +161,11 @@ async def health_generate() -> Response:
@app.get("/v1/models")
async def models() -> ORJSONResponse:
if dp_state is not None:
async with aiohttp.ClientSession() as session:
async with session.get(f"{dp_state.config.backend_urls[0]}/v1/models") as resp:
payload = await resp.json()
return ORJSONResponse(payload, status_code=resp.status)
state = _require_state()
async with aiohttp.ClientSession() as session:
async with session.get(f"{state.config.prefill_urls[0][0]}/v1/models") as response:
@@ -147,6 +212,15 @@ async def _forward_to_backend(
headers: dict[str, str],
endpoint_name: str,
) -> Response:
# DP mode: forward to a single backend
if dp_state is not None:
return await _forward_to_dp_backend(
request_data=request_data,
headers=headers,
endpoint_name=endpoint_name,
)
# PD mode: coordinate prefill + decode
state = _require_state()
prefill_server, bootstrap_port, decode_server = state.select_pair(headers)
prefill_request, decode_request = _build_backend_requests(
@@ -186,6 +260,63 @@ async def _forward_to_backend(
)
async def _forward_to_dp_backend(
*,
request_data: dict,
headers: dict[str, str],
endpoint_name: str,
) -> Response:
assert dp_state is not None
backend_server = dp_state.select_backend(headers)
cleaned = _strip_internal_fields(request_data)
timeout_s = dp_state.config.request_timeout_s
if request_data.get("stream", False):
return StreamingResponse(
_stream_dp_generate(
request_data=cleaned,
backend_server=backend_server,
endpoint_name=endpoint_name,
timeout_s=timeout_s,
),
media_type="text/event-stream",
)
async with aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=timeout_s)
) as session:
async with session.post(
f"{backend_server}/{endpoint_name}", json=cleaned
) as response:
body = await response.read()
return Response(
content=body,
status_code=response.status,
media_type=response.content_type,
)
async def _stream_dp_generate(
*,
request_data: dict,
backend_server: str,
endpoint_name: str,
timeout_s: float,
) -> AsyncIterator[bytes]:
async with aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=timeout_s)
) as session:
async with session.post(
f"{backend_server}/{endpoint_name}", json=request_data
) as response:
if response.status != HTTPStatus.OK:
payload = await response.read()
yield payload
return
async for chunk in response.content.iter_chunked(_STREAM_CHUNK_SIZE):
yield chunk
async def _stream_generate(
*,
prefill_request: dict,
@@ -241,6 +372,12 @@ def _build_backend_requests(
prefill_request.update(bootstrap_payload)
decode_request.update(bootstrap_payload)
# session_params is only meaningful for the decode worker (streaming session
# KV reuse). Sending it to the prefill worker causes the D side to
# short-circuit with local-prefill on already-open sessions, returning
# truncated responses while P's KV transfer gets aborted.
prefill_request.pop("session_params", None)
if prefill_priority is not None:
prefill_request["priority"] = int(prefill_priority)
if decode_priority is not None:
@@ -262,7 +399,7 @@ def _require_state() -> RouterState:
def main() -> None:
parser = argparse.ArgumentParser(description="Minimal local PD router")
parser = argparse.ArgumentParser(description="Minimal local PD / DP router")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=8000)
parser.add_argument(
@@ -270,30 +407,58 @@ def main() -> None:
nargs=2,
metavar=("URL", "BOOTSTRAP_PORT"),
action="append",
required=True,
default=None,
)
parser.add_argument(
"--decode",
action="append",
required=True,
default=None,
)
parser.add_argument("--prefill-policy", default="round_robin")
parser.add_argument("--decode-policy", default="manual")
parser.add_argument(
"--backend",
action="append",
default=None,
help="Backend URL for DP (data-parallel) mode. Repeat for each worker.",
)
parser.add_argument(
"--backend-policy",
default="round_robin",
help="Routing policy for DP mode: round_robin, manual, consistent_hashing.",
)
parser.add_argument("--request-timeout-s", type=float, default=1800.0)
args = parser.parse_args()
global router_state
router_state = RouterState(
RouterConfig(
host=args.host,
port=args.port,
prefill_urls=[(url, int(port)) for url, port in args.prefill],
decode_urls=list(args.decode),
prefill_policy=args.prefill_policy,
decode_policy=args.decode_policy,
request_timeout_s=args.request_timeout_s,
global router_state, dp_state
if args.backend:
# DP mode: simple forward to one of N backends
dp_state = DpRouterState(
DpRouterConfig(
host=args.host,
port=args.port,
backend_urls=list(args.backend),
backend_policy=args.backend_policy,
request_timeout_s=args.request_timeout_s,
)
)
)
elif args.prefill and args.decode:
# PD mode: prefill/decode coordination
router_state = RouterState(
RouterConfig(
host=args.host,
port=args.port,
prefill_urls=[(url, int(port)) for url, port in args.prefill],
decode_urls=list(args.decode),
prefill_policy=args.prefill_policy,
decode_policy=args.decode_policy,
request_timeout_s=args.request_timeout_s,
)
)
else:
parser.error("Either --backend (DP mode) or both --prefill and --decode (PD mode) are required")
uvicorn.run(app, host=args.host, port=args.port, log_level="info")

118
src/agentic_pd_hybrid/replay.py
View File

@@ -124,6 +124,9 @@ class ExecutionResult:
prefill_request_priority: int | None = None
decode_request_priority: int | None = None
error: str | None = None
actual_output_tokens: int | None = None
requested_output_tokens: int | None = None
finish_reason: str | None = None
async def replay_trace(config: ReplayConfig) -> list[RequestMetrics]:
@@ -274,6 +277,9 @@ async def _run_request(
prefill_request_priority=execution.prefill_request_priority,
decode_request_priority=execution.decode_request_priority,
error=execution.error,
actual_output_tokens=execution.actual_output_tokens,
requested_output_tokens=execution.requested_output_tokens,
finish_reason=execution.finish_reason,
)
@@ -286,7 +292,7 @@ async def _invoke_router(
session_id: str | None = None,
prefill_request_priority: int | None = None,
decode_request_priority: int | None = None,
) -> tuple[float, float | None, float | None, int]:
) -> GenerateResult:
headers = _build_headers(
request=request,
header_mode=config.header_mode,
@@ -414,6 +420,18 @@ async def _invoke_chat_completion(
return latency_s, ttft_s, tpot_s, cached_tokens
@dataclass(frozen=True)
class GenerateResult:
latency_s: float
ttft_s: float | None
tpot_s: float | None
cached_tokens: int
actual_output_tokens: int
requested_output_tokens: int
finish_reason: str | None
server_meta_info: dict | None
async def _invoke_generate(
*,
client: httpx.AsyncClient,
@@ -423,12 +441,16 @@ async def _invoke_generate(
timeout_s: float,
stream_idle_timeout_s: float | None,
stream: bool,
) -> tuple[float, float | None, float | None, int]:
) -> GenerateResult:
start = time.perf_counter()
ttft_s: float | None = None
cached_tokens = 0
sampling_params = payload.get("sampling_params", {})
generated_tokens = int(sampling_params.get("max_new_tokens", 1))
requested_output_tokens = int(sampling_params.get("max_new_tokens", 1))
actual_token_count = 0
finish_reason: str | None = None
last_meta_info: dict | None = None
if stream:
async with client.stream(
"POST",
@@ -452,8 +474,19 @@ async def _invoke_generate(
if isinstance(error, dict):
raise ValueError(error.get("message", json.dumps(error)))
cached_tokens = max(cached_tokens, _extract_generate_cached_tokens(parsed))
if _contains_generate_token(parsed) and ttft_s is None:
ttft_s = time.perf_counter() - start
if _contains_generate_token(parsed):
actual_token_count += 1
if ttft_s is None:
ttft_s = time.perf_counter() - start
meta_info = parsed.get("meta_info")
if isinstance(meta_info, dict):
last_meta_info = meta_info
completion_tokens = int(meta_info.get("completion_tokens", 0))
if completion_tokens > actual_token_count:
actual_token_count = completion_tokens
fr = meta_info.get("finish_reason")
if fr is not None:
finish_reason = str(fr)
if _is_generate_terminal_chunk(parsed):
break
else:
@@ -469,15 +502,33 @@ async def _invoke_generate(
if isinstance(error, dict):
raise ValueError(error.get("message", json.dumps(error)))
cached_tokens = _extract_generate_cached_tokens(parsed)
meta_info = parsed.get("meta_info")
if isinstance(meta_info, dict):
last_meta_info = meta_info
actual_token_count = int(meta_info.get("completion_tokens", 0))
finish_reason = meta_info.get("finish_reason")
latency_s = time.perf_counter() - start
if stream and ttft_s is None and generated_tokens > 0:
if stream and ttft_s is None and requested_output_tokens > 0:
raise RuntimeError("generate stream ended before producing any token")
# Use actual token count for TPOT (not requested count)
effective_tokens = max(1, actual_token_count) if actual_token_count > 0 else max(1, requested_output_tokens)
if ttft_s is None:
tpot_s = None
else:
tpot_s = max(0.0, latency_s - ttft_s) / max(1, generated_tokens)
return latency_s, ttft_s, tpot_s, cached_tokens
tpot_s = max(0.0, latency_s - ttft_s) / effective_tokens
return GenerateResult(
latency_s=latency_s,
ttft_s=ttft_s,
tpot_s=tpot_s,
cached_tokens=cached_tokens,
actual_output_tokens=actual_token_count,
requested_output_tokens=requested_output_tokens,
finish_reason=finish_reason,
server_meta_info=last_meta_info,
)
async def _open_streaming_session(
@@ -850,8 +901,8 @@ def _decode_session_soft_cap(
- residency.headroom_tokens.get(server_url, 0),
)
if usable_capacity_tokens <= 0:
return 4
return max(1, min(4, usable_capacity_tokens // target_tokens))
return 16
return max(1, min(16, usable_capacity_tokens // target_tokens))
def _should_admit_new_decode_session(
@@ -1587,7 +1638,7 @@ async def _invoke_plain_router(
config=config,
direct_to_d_predicted=False,
)
latency_s, ttft_s, tpot_s, cached_tokens = await _invoke_router(
gen = await _invoke_router(
client=client,
request=request,
config=config,
@@ -1598,13 +1649,16 @@ async def _invoke_plain_router(
execution_mode=execution_mode,
actual_kv_transfer_blocks=decision.kv_transfer_blocks,
effective_input_length=request.input_length,
cached_tokens=cached_tokens,
cached_tokens=gen.cached_tokens,
prefill_request_priority=prefill_priority,
session_reused=False,
session_reset=False,
latency_s=latency_s,
ttft_s=ttft_s,
tpot_s=tpot_s,
latency_s=gen.latency_s,
ttft_s=gen.ttft_s,
tpot_s=gen.tpot_s,
actual_output_tokens=gen.actual_output_tokens,
requested_output_tokens=gen.requested_output_tokens,
finish_reason=gen.finish_reason,
)
@@ -1676,7 +1730,7 @@ async def _invoke_kvcache_seeded_router(
decode_session.opened = True
decode_session_newly_opened = True
decode_session.active_requests += 1
latency_s, ttft_s, tpot_s, cached_tokens = await _invoke_router(
gen = await _invoke_router(
client=client,
request=request,
config=config,
@@ -1742,13 +1796,16 @@ async def _invoke_kvcache_seeded_router(
execution_mode=execution_mode,
actual_kv_transfer_blocks=decision.kv_transfer_blocks,
effective_input_length=request.input_length,
cached_tokens=cached_tokens,
cached_tokens=gen.cached_tokens,
prefill_request_priority=prefill_priority,
session_reused=False,
session_reset=False,
latency_s=latency_s,
ttft_s=ttft_s,
tpot_s=tpot_s,
latency_s=gen.latency_s,
ttft_s=gen.ttft_s,
tpot_s=gen.tpot_s,
actual_output_tokens=gen.actual_output_tokens,
requested_output_tokens=gen.requested_output_tokens,
finish_reason=gen.finish_reason,
)
@@ -1774,14 +1831,16 @@ async def _execute_request(
)
if config.mechanism_name == "pd-colo":
return await _invoke_direct(
if not config.router_url:
raise ValueError("router_url is required for pd-colo replay")
result = await _invoke_plain_router(
client=client,
request=request,
config=config,
decision=decision,
direct_sessions=direct_sessions,
direct_session_lock=direct_session_lock,
execution_mode="dp-colo-router",
)
return replace(result, actual_kv_transfer_blocks=0)
if config.mechanism_name == "kvcache-centric":
if not config.router_url:
@@ -2238,7 +2297,7 @@ async def _invoke_session_direct(
session.active_requests += 1
try:
latency_s, ttft_s, tpot_s, cached_tokens = await _invoke_generate(
gen = await _invoke_generate(
client=client,
base_url=session.server_url,
headers={"x-request-id": request.request_id},
@@ -2277,12 +2336,15 @@ async def _invoke_session_direct(
execution_mode=execution_mode,
actual_kv_transfer_blocks=0,
effective_input_length=len(input_ids),
cached_tokens=cached_tokens,
cached_tokens=gen.cached_tokens,
session_reused=session_reused,
session_reset=session_reset,
latency_s=latency_s,
ttft_s=ttft_s,
tpot_s=tpot_s,
latency_s=gen.latency_s,
ttft_s=gen.ttft_s,
tpot_s=gen.tpot_s,
actual_output_tokens=gen.actual_output_tokens,
requested_output_tokens=gen.requested_output_tokens,
finish_reason=gen.finish_reason,
)

5
src/agentic_pd_hybrid/stack.py
View File

@@ -66,6 +66,7 @@ def launch_pd_stack(
timeout_s: float = 1200.0,
router_request_timeout_s: float | None = None,
include_router: bool = True,
naive_dp: bool = False,
) -> ManagedPdStack:
run_dir.mkdir(parents=True, exist_ok=True)
logs_dir = run_dir / "logs"
@@ -77,6 +78,7 @@ def launch_pd_stack(
decode_policy=decode_policy,
include_router=include_router,
router_request_timeout_s=router_request_timeout_s,
naive_dp=naive_dp,
)
prefill_processes = [
@@ -195,6 +197,9 @@ def _build_process_env(topology: SingleNodeTopology) -> dict[str, str]:
env["MC_MS_AUTO_DISC"] = "0"
if topology.ib_device:
env["MOONCAKE_DEVICE"] = topology.ib_device
elif topology.transfer_backend == "mooncake":
# Default to TCP when RDMA is not forced (e.g. loopback on same node)
env.setdefault("MOONCAKE_PROTOCOL", "tcp")
repo_root = Path(__file__).resolve().parents[2]
python_paths = [