diff --git a/replayer/__main__.py b/replayer/__main__.py index 8dd279c..e9f344e 100644 --- a/replayer/__main__.py +++ b/replayer/__main__.py @@ -17,10 +17,8 @@ def main() -> None: p.add_argument("--endpoint", type=str, required=True, help="vLLM server URL (e.g. http://localhost:8000)") p.add_argument("--model", type=str, default="default", help="Model name for API") - p.add_argument("--time-scale", type=float, default=1.0, - help="Time compression (>1 = faster)") - p.add_argument("--max-inflight-sessions", type=int, default=32) - p.add_argument("--concurrency-limit", type=int, default=256) + p.add_argument("--concurrency-limit", type=int, default=2000, + help="Max concurrent HTTP requests (safety limit)") p.add_argument("--request-timeout", type=float, default=600.0) p.add_argument("--request-limit", type=int, default=None, help="Limit number of requests to replay") @@ -37,8 +35,6 @@ def main() -> None: output_path=args.output, endpoint_url=args.endpoint.rstrip("/"), model_name=args.model, - time_scale=args.time_scale, - max_inflight_sessions=args.max_inflight_sessions, concurrency_limit=args.concurrency_limit, request_timeout_s=args.request_timeout, request_limit=args.request_limit, diff --git a/replayer/replay.py b/replayer/replay.py index 787bc89..13f336a 100644 --- a/replayer/replay.py +++ b/replayer/replay.py @@ -1,16 +1,15 @@ """Trace replayer — send requests to vLLM following trace timing. -Supports both vLLM's /v1/completions (OpenAI-compatible) and /generate -(SGLang-style) endpoints. Uses hash_ids from the trace to construct -synthetic prompts that reproduce realistic prefix-cache hit patterns. +Uses hash_ids from the trace to construct synthetic prompts that +reproduce realistic prefix-cache hit patterns. Key behaviors: + - Trace-driven dispatch: each request is sent at its trace timestamp. + No artificial concurrency limits or time compression. - Per-session sequencing: turns within a session are sent in order, each waiting for the previous to complete before dispatching. - - Inter-session arrival: sessions start at their trace timestamps, - scaled by --time-scale. - - Concurrency control: --max-inflight-sessions caps concurrent sessions; - --concurrency-limit caps total in-flight requests. + If a turn completes after its successor's timestamp, the successor + fires immediately (no waiting for a past timestamp). """ from __future__ import annotations @@ -56,9 +55,7 @@ class ReplayConfig: trace_path: Path output_path: Path endpoint_url: str # comma-separated for round-robin: "http://host:8000,http://host:8001" - time_scale: float = 1.0 - max_inflight_sessions: int = 32 - concurrency_limit: int = 256 + concurrency_limit: int = 2000 request_timeout_s: float = 600.0 request_limit: int | None = None model_name: str = "default" @@ -214,34 +211,25 @@ async def _run_session( state: _SessionState, config: ReplayConfig, client: httpx.AsyncClient, - session_sem: asyncio.Semaphore, request_sem: asyncio.Semaphore, earliest_ts: float, sweep_start: float, sink: IncrementalMetricSink, ) -> list[RequestMetrics]: - async with session_sem: - # Wait until this session's start time - offset = (state.turns[0].timestamp_s - earliest_ts) / config.time_scale - wait = offset - (time.perf_counter() - sweep_start) - if wait > 0: - await asyncio.sleep(wait) + for req in state.turns: + # Wait until this request's trace timestamp + target_wall = (req.timestamp_s - earliest_ts) + elapsed = time.perf_counter() - sweep_start + if elapsed < target_wall: + await asyncio.sleep(target_wall - elapsed) - for req in state.turns: - # Intra-session: wait for turn's relative offset - if req != state.turns[0]: - target = (req.timestamp_s - state.turns[0].timestamp_s) / config.time_scale - elapsed = time.perf_counter() - sweep_start - offset - if elapsed < target: - await asyncio.sleep(target - elapsed) - - token_ids = _build_prompt_token_ids(req) - metric = await _dispatch_request( - client=client, config=config, req=req, - prompt_token_ids=token_ids, sem=request_sem, - ) - state.metrics.append(metric) - await sink.append(metric) + token_ids = _build_prompt_token_ids(req) + metric = await _dispatch_request( + client=client, config=config, req=req, + prompt_token_ids=token_ids, sem=request_sem, + ) + state.metrics.append(metric) + await sink.append(metric) return state.metrics @@ -283,16 +271,18 @@ async def replay_trace(config: ReplayConfig) -> list[RequestMetrics]: sessions = sorted(by_session.items(), key=lambda kv: kv[1][0].timestamp_s) earliest_ts = sessions[0][1][0].timestamp_s + latest_ts = max(r.timestamp_s for r in requests) + trace_span = latest_ts - earliest_ts - session_sem = asyncio.Semaphore(config.max_inflight_sessions) request_sem = asyncio.Semaphore(config.concurrency_limit) sink = IncrementalMetricSink(config.output_path) n_sessions = len(sessions) n_requests = len(requests) - logger.info("Replaying %d sessions (%d requests), time_scale=%.1f", - n_sessions, n_requests, config.time_scale) + qps = n_requests / trace_span if trace_span > 0 else 0 + logger.info("Replaying %d sessions (%d requests) over %.0fs (%.2f req/s)", + n_sessions, n_requests, trace_span, qps) pre_metrics = await _snapshot_prefix_cache_metrics(config.endpoint_url) sweep_start = time.perf_counter() @@ -312,7 +302,7 @@ async def replay_trace(config: ReplayConfig) -> list[RequestMetrics]: asyncio.create_task(_run_session( state=_SessionState(session_id=sid, turns=turns), config=config, client=client, - session_sem=session_sem, request_sem=request_sem, + request_sem=request_sem, earliest_ts=earliest_ts, sweep_start=sweep_start, sink=sink, )) diff --git a/scripts/bench.sh b/scripts/bench.sh index beb8e17..da29ae3 100755 --- a/scripts/bench.sh +++ b/scripts/bench.sh @@ -30,9 +30,7 @@ POLICY="linear" # linear | lmetric N_INSTANCES=8 BASE_PORT=8000 PROXY_PORT=9090 -REQUESTS=200 -TIME_SCALE=20 -MAX_SESSIONS=8 +REQUESTS="" # empty = all requests in trace HEAVY_THRESHOLD=20000 NO_OFFLOAD=false OVERLOAD_FACTOR_ARG="" @@ -46,8 +44,6 @@ while [[ $# -gt 0 ]]; do --policy) POLICY="$2"; shift 2 ;; --instances) N_INSTANCES="$2"; shift 2 ;; --requests) REQUESTS="$2"; shift 2 ;; - --time-scale) TIME_SCALE="$2"; shift 2 ;; - --sessions) MAX_SESSIONS="$2"; shift 2 ;; --heavy-threshold) HEAVY_THRESHOLD="$2"; shift 2 ;; --no-offload) NO_OFFLOAD=true; shift ;; --overload-factor) OVERLOAD_FACTOR_ARG="$2"; shift 2 ;; @@ -58,6 +54,7 @@ done if [ -z "$TAG" ]; then echo "Usage: bench.sh --tag NAME --mode {baseline|elastic} [--instances N] [--policy {linear|lmetric}] [--requests N]" + echo " Trace QPS is controlled by sample_trace.py --sample-ratio, not by bench.sh." exit 1 fi @@ -76,9 +73,7 @@ cat > "$OUTDIR/config.json" << CONF "policy": "$POLICY", "model": "$MODEL", "n_instances": $N_INSTANCES, - "requests": $REQUESTS, - "time_scale": $TIME_SCALE, - "max_sessions": $MAX_SESSIONS, + "requests": "${REQUESTS:-all}", "heavy_threshold": $HEAVY_THRESHOLD, "no_offload": "$NO_OFFLOAD", "overload_factor": "${OVERLOAD_FACTOR_ARG:-2.0}", @@ -245,7 +240,13 @@ launch_proxy() { # ─── Run benchmark ───────────────────────────────────────────────────────── run_benchmark() { - echo "[bench] Running $REQUESTS requests (time_scale=$TIME_SCALE, sessions=$MAX_SESSIONS)..." + local request_args="" + if [ -n "$REQUESTS" ]; then + request_args="--request-limit $REQUESTS" + echo "[bench] Running $REQUESTS requests (trace-driven timing)..." + else + echo "[bench] Running all requests in trace (trace-driven timing)..." + fi # Start GPU monitor in background bash "$PROJECT_DIR/scripts/gpu_monitor.sh" "$OUTDIR/gpu_util.csv" 5 & @@ -256,9 +257,7 @@ run_benchmark() { --output "$OUTDIR/metrics.jsonl" \ --endpoint "http://localhost:$PROXY_PORT" \ --model "$MODEL" \ - --time-scale "$TIME_SCALE" \ - --max-inflight-sessions "$MAX_SESSIONS" \ - --request-limit "$REQUESTS" \ + $request_args \ -v 2>&1 | tee "$OUTDIR/replayer.log" # Stop GPU monitor @@ -324,7 +323,7 @@ print('=' * 70) echo "================================================================" echo " bench.sh: $TAG" -echo " mode=$MODE policy=$POLICY requests=$REQUESTS overload_factor=${OVERLOAD_FACTOR_ARG:-2.0} max_batched_tokens=${MAX_BATCHED_TOKENS:-default}" +echo " mode=$MODE policy=$POLICY requests=${REQUESTS:-all} overload_factor=${OVERLOAD_FACTOR_ARG:-2.0}" echo " $(date)" echo "================================================================" diff --git a/scripts/sample_trace.py b/scripts/sample_trace.py index 5c18f2d..0c76348 100644 --- a/scripts/sample_trace.py +++ b/scripts/sample_trace.py @@ -2,22 +2,28 @@ Preserves: - Complete session structure (all turns within a session kept together) - - Original arrival timing (inter-session and intra-session gaps) + - Original arrival timing (re-zeroed to t=0 but NOT compressed) - hash_ids for KV cache reuse patterns - Request type distribution Sampling strategy: 1. Group requests by session (derived from parent_chat_id chains) - 2. Randomly sample N sessions (or until target request count reached) + 2. Randomly sample a fraction of sessions (--sample-ratio) + OR sample until target request count (--target-requests) 3. Re-zero timestamps so first event starts at t=0 - 4. Optionally compress time axis to increase load density + 4. The resulting QPS is proportional to the sample ratio, + preserving the production arrival pattern Usage: + # Sample 1.6% of sessions (e.g., 8 GPUs / 500 cluster GPUs) python scripts/sample_trace.py \\ --input ~/ali-trace/trace-glm5.1-formatted/051315-051317.jsonl \\ - --output traces/sampled.jsonl \\ - --target-requests 5000 \\ - --seed 42 + --output traces/sampled_ratio016.jsonl \\ + --sample-ratio 0.016 --seed 42 + + # Sample by request count (legacy) + python scripts/sample_trace.py \\ + --input ... --output ... --target-requests 1000 --seed 42 """ from __future__ import annotations @@ -58,39 +64,37 @@ def load_raw_rows(path: Path) -> dict[str, list[dict]]: def sample_sessions( rows_by_session: dict[str, list[dict]], *, - target_requests: int, + sample_ratio: float | None = None, + target_requests: int | None = None, seed: int, - strategy: str = "random", ) -> list[str]: - """Select sessions until target request count is reached.""" + """Select sessions by ratio or until target request count.""" all_sids = list(rows_by_session.keys()) rng = random.Random(seed) + rng.shuffle(all_sids) - if strategy == "random": - rng.shuffle(all_sids) - elif strategy == "sequential": - pass # keep file order - else: - raise ValueError(f"Unknown strategy: {strategy}") + if sample_ratio is not None: + n_select = max(1, int(len(all_sids) * sample_ratio)) + return all_sids[:n_select] - selected = [] - total = 0 - for sid in all_sids: - selected.append(sid) - total += len(rows_by_session[sid]) - if total >= target_requests: - break + if target_requests is not None: + selected = [] + total = 0 + for sid in all_sids: + selected.append(sid) + total += len(rows_by_session[sid]) + if total >= target_requests: + break + return selected - return selected + raise ValueError("Must specify --sample-ratio or --target-requests") def build_output( rows_by_session: dict[str, list[dict]], selected: list[str], - *, - time_scale: float = 1.0, ) -> list[dict]: - """Build output rows with re-zeroed timestamps.""" + """Build output rows with re-zeroed timestamps (no time compression).""" out_rows = [] for sid in selected: for row in rows_by_session[sid]: @@ -103,10 +107,9 @@ def build_output( if not out_rows: return out_rows - # Re-zero: subtract earliest timestamp t0 = float(out_rows[0]["timestamp"]) for row in out_rows: - row["timestamp"] = (float(row["timestamp"]) - t0) / time_scale + row["timestamp"] = float(row["timestamp"]) - t0 return out_rows @@ -125,17 +128,16 @@ def print_summary( output_lens = [r["output_length"] for r in out_rows] span_s = float(out_rows[-1]["timestamp"]) if out_rows else 0 + qps = n_requests / span_s if span_s > 0 else 0 + session_starts = {} for r in out_rows: sid = r["session_id"] ts = float(r["timestamp"]) if sid not in session_starts: session_starts[sid] = ts - starts_sorted = sorted(session_starts.values()) - deltas = [starts_sorted[i+1] - starts_sorted[i] - for i in range(len(starts_sorted) - 1)] - # hash_ids overlap: count unique hash_ids across all requests + # hash_ids overlap all_hashes = set() for r in out_rows: all_hashes.update(r.get("hash_ids", [])) @@ -149,12 +151,8 @@ def print_summary( print(f" Output length: min={min(output_lens)} max={max(output_lens)} " f"avg={sum(output_lens)/len(output_lens):.0f}") print(f" Trace span: {span_s:.1f}s ({span_s/60:.1f} min)") + print(f" QPS: {qps:.2f} req/s") print(f" Unique hash blocks: {len(all_hashes)}") - if deltas: - deltas.sort() - p = lambda q: deltas[min(int(q * len(deltas)), len(deltas) - 1)] - print(f" Session arrival deltas (s): p10={p(0.1):.2f} p50={p(0.5):.2f} " - f"p90={p(0.9):.2f} max={max(deltas):.2f}") def main() -> None: @@ -164,15 +162,16 @@ def main() -> None: help="Path to the full trace JSONL file") p.add_argument("--output", type=Path, required=True, help="Path to write sampled trace JSONL") - p.add_argument("--target-requests", type=int, default=5000, - help="Target number of requests (stops after session that crosses it)") - p.add_argument("--strategy", choices=["random", "sequential"], default="random", - help="Session selection strategy") - p.add_argument("--time-scale", type=float, default=1.0, - help="Compress time axis by this factor (>1 = faster arrival)") + p.add_argument("--sample-ratio", type=float, default=None, + help="Fraction of sessions to sample (e.g. 0.016 for 8/500 GPU ratio)") + p.add_argument("--target-requests", type=int, default=None, + help="Target number of requests (legacy, stops after session that crosses it)") p.add_argument("--seed", type=int, default=42) args = p.parse_args() + if args.sample_ratio is None and args.target_requests is None: + p.error("Must specify --sample-ratio or --target-requests") + print(f"Loading trace from {args.input} ...") rows_by_session = load_raw_rows(args.input) total_sessions = len(rows_by_session) @@ -181,15 +180,12 @@ def main() -> None: selected = sample_sessions( rows_by_session, + sample_ratio=args.sample_ratio, target_requests=args.target_requests, seed=args.seed, - strategy=args.strategy, ) - out_rows = build_output( - rows_by_session, selected, - time_scale=args.time_scale, - ) + out_rows = build_output(rows_by_session, selected) print_summary(rows_by_session, selected, out_rows)