From daeb95eca08efd009360d5516032dbd1eadbbbb7 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Sat, 23 May 2026 22:33:10 +0800 Subject: [PATCH] =?UTF-8?q?RDMA=20overhead=202.0=E2=86=920.1s=20(direct=20?= =?UTF-8?q?read=20is=20raw=20memory,=20not=20scheduler=20flow)=20+=20retry?= =?UTF-8?q?=20on=20ConnectError=20to=20handle=20kv=5Fboth=20connection=20i?= =?UTF-8?q?nstability?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With RDMA_overhead=0.1s, offload triggers when C_s has just 700 tokens pending (0.1s queue), vs 38k tokens (5.4s) with the old 2.0s estimate. Co-Authored-By: Claude Opus 4.6 (1M context) --- scripts/cache_aware_proxy.py | 38 ++++++++++++++++++++++++------------ 1 file changed, 26 insertions(+), 12 deletions(-) diff --git a/scripts/cache_aware_proxy.py b/scripts/cache_aware_proxy.py index ae136e3..bebf4b6 100644 --- a/scripts/cache_aware_proxy.py +++ b/scripts/cache_aware_proxy.py @@ -23,6 +23,9 @@ from contextlib import asynccontextmanager from dataclasses import dataclass import httpx + +MAX_STREAM_RETRIES = 3 +RETRY_DELAY_S = 0.5 import uvicorn from fastapi import FastAPI, HTTPException, Request from fastapi.responses import StreamingResponse @@ -44,7 +47,7 @@ class Settings: max_offload_inflight: int = 4 # global cap on concurrent P-role offloads cache_gate_ratio: float = 0.3 # min cache_hit/input ratio to allow offload prefill_throughput: float = 7000.0 # tokens/s per GPU (H20 measurement) - rdma_overhead_s: float = 2.0 # RDMA transfer + decode-start overhead + rdma_overhead_s: float = 0.1 # direct RDMA read overhead (raw memory read ~10-50ms) cache_capacity_blocks: int = 200000 # per-instance LRU cap on shadow cached_blocks @@ -443,17 +446,28 @@ async def _handle_combined(api, req_data, token_ids, input_length, session_id, h async def generate(): prefill_done = False - try: - async with inst.client.stream("POST", api, json=req_data, headers=headers) as resp: - resp.raise_for_status() - async for chunk in resp.aiter_bytes(): - if not prefill_done: - inst.pending_prefill_tokens -= estimated_new - inst.ongoing_decode_tokens += input_length - breakdown["t_first_token"] = _time.monotonic() - prefill_done = True - yield chunk - inst.record_prefix(token_ids) + last_err = None + for attempt in range(MAX_STREAM_RETRIES): + try: + async with inst.client.stream("POST", api, json=req_data, headers=headers) as resp: + resp.raise_for_status() + async for chunk in resp.aiter_bytes(): + if not prefill_done: + inst.pending_prefill_tokens -= estimated_new + inst.ongoing_decode_tokens += input_length + breakdown["t_first_token"] = _time.monotonic() + prefill_done = True + yield chunk + inst.record_prefix(token_ids) + return + except (httpx.ConnectError, httpx.RemoteProtocolError) as e: + last_err = e + if prefill_done: + raise + if attempt < MAX_STREAM_RETRIES - 1: + await asyncio.sleep(RETRY_DELAY_S) + if last_err: + raise last_err finally: if not prefill_done: inst.pending_prefill_tokens -= estimated_new