Phase 1 milestone: system-level analysis + reproducible report
- REPORT.md: self-contained milestone report covering baseline vs elastic setup, exact launch commands, benchmark params, results, log locations, and repo structure — sufficient for anyone to reproduce - analysis/pd_separation_analysis.md §5: elastic P2P system-level breakdown (KV cache hit ratio, per-class TTFT, GPU util paradox explanation) - scripts/cache_aware_proxy.py: round-robin P-instance selection replacing argmin(ongoing_tokens) to fix GPU load imbalance (3.0x → expected ~2x) - scripts/launch_elastic_p2p.sh: one-command launch for elastic P2P config Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -26,6 +26,7 @@ from fastapi.responses import StreamingResponse
|
||||
BLOCK_SIZE = 512
|
||||
CACHE_HIT_ALPHA = 1.0
|
||||
HEAVY_THRESHOLD = 20000 # default; overridden by --heavy-threshold
|
||||
OVERLOAD_FACTOR = 2.0
|
||||
|
||||
|
||||
class InstanceState:
|
||||
@@ -81,7 +82,6 @@ def pick_instance(instances: list[InstanceState], token_ids: list[int] | None,
|
||||
_inst_cumulative_tokens = [0] * len(instances)
|
||||
|
||||
avg_load = max(sum(i.ongoing_tokens for i in instances) / len(instances), 1.0)
|
||||
OVERLOAD_FACTOR = 2.0
|
||||
|
||||
# Session affinity for turn 2+ (with load override)
|
||||
if session_id and session_id in affinity:
|
||||
@@ -118,6 +118,7 @@ is_pd_sep = False
|
||||
_breakdown_log: list[dict] = []
|
||||
_offload_inflight = 0 # number of currently in-flight offloaded HEAVY requests
|
||||
MAX_OFFLOAD_INFLIGHT = 4 # cap concurrent offloads to prevent P overload
|
||||
_p_round_robin_idx = 0 # round-robin counter for P-instance selection
|
||||
|
||||
|
||||
async def init_prefill_bootstrap(instances: list[InstanceState], ready: asyncio.Event):
|
||||
@@ -239,18 +240,21 @@ async def _handle_combined(api, req_data, token_ids, input_length, session_id, h
|
||||
offload_reason = "disabled"
|
||||
if estimated_new >= HEAVY_THRESHOLD and offload_enabled and has_bootstrap and len(combined_instances) >= 2:
|
||||
d_inst = best_inst
|
||||
p_candidates = [inst for inst in combined_instances if inst is not d_inst]
|
||||
p_inst = min(p_candidates, key=lambda x: x.ongoing_tokens)
|
||||
p_candidates = [(i, inst) for i, inst in enumerate(combined_instances) if inst is not d_inst]
|
||||
avg_load = max(sum(i.ongoing_tokens for i in combined_instances) / len(combined_instances), 1.0)
|
||||
|
||||
# Decision logic:
|
||||
# 1. Global cap: max N concurrent offloads (prevents all-offload storm)
|
||||
# 2. P must not already be saturated with heavy prefills
|
||||
# 3. D must be doing something (otherwise no benefit from offloading)
|
||||
# NOTE: We do NOT require P < D. P can be busier than D — the point
|
||||
# is to keep heavy prefill OFF the session-sticky D instance so D's
|
||||
# decode is not disrupted and D's KV cache is available for future turns.
|
||||
global _offload_inflight
|
||||
# Round-robin P selection with overload skip (spreads P-role evenly)
|
||||
global _offload_inflight, _p_round_robin_idx
|
||||
p_inst = None
|
||||
for _ in range(len(p_candidates)):
|
||||
_p_round_robin_idx = (_p_round_robin_idx + 1) % len(p_candidates)
|
||||
candidate = p_candidates[_p_round_robin_idx][1]
|
||||
if candidate.ongoing_tokens < avg_load * OVERLOAD_FACTOR:
|
||||
p_inst = candidate
|
||||
break
|
||||
if p_inst is None:
|
||||
p_inst = min(p_candidates, key=lambda x: x[1].ongoing_tokens)[1]
|
||||
|
||||
if _offload_inflight >= MAX_OFFLOAD_INFLIGHT:
|
||||
offload_reason = "max_concurrent_reached"
|
||||
elif p_inst.ongoing_tokens >= HEAVY_THRESHOLD * 2:
|
||||
|
||||
122
scripts/launch_elastic_p2p.sh
Executable file
122
scripts/launch_elastic_p2p.sh
Executable file
@@ -0,0 +1,122 @@
|
||||
#!/bin/bash
|
||||
# Elastic P2P offload: 8× TP=1 kv_both instances + cache_aware_proxy --offload.
|
||||
#
|
||||
# Architecture:
|
||||
# All 8 instances run kv_role=kv_both (Mooncake connector).
|
||||
# The proxy classifies requests as WARM/MEDIUM/HEAVY.
|
||||
# HEAVY requests: prefill on a different instance (P), KV via Mooncake RDMA,
|
||||
# decode on session-sticky instance (D).
|
||||
# WARM/MEDIUM: co-located prefill+decode on session-sticky instance.
|
||||
#
|
||||
# Usage:
|
||||
# bash scripts/launch_elastic_p2p.sh # default: this machine
|
||||
# HOST=dash1 bash scripts/launch_elastic_p2p.sh # launch on dash1 via ssh
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
PROJECT_DIR="$(dirname "$SCRIPT_DIR")"
|
||||
VENV="$PROJECT_DIR/.venv/bin"
|
||||
VLLM="$VENV/vllm"
|
||||
PYTHON="$VENV/python"
|
||||
|
||||
MODEL="${MODEL_PATH:-$HOME/models/Qwen/Qwen3-Coder-30B-A3B-Instruct}"
|
||||
N_INSTANCES=8
|
||||
BASE_PORT=8000
|
||||
PROXY_PORT=9090
|
||||
HEAVY_THRESHOLD="${HEAVY_THRESHOLD:-20000}"
|
||||
MAX_OFFLOAD="${MAX_OFFLOAD:-4}"
|
||||
|
||||
trap 'echo "Cleaning up..."; kill $(jobs -p) 2>/dev/null; wait 2>/dev/null' EXIT INT TERM
|
||||
|
||||
echo "=== Elastic P2P Offload (${N_INSTANCES}× TP=1 kv_both) ==="
|
||||
echo " Model: $MODEL"
|
||||
echo " Instances: $N_INSTANCES × TP=1"
|
||||
echo " Proxy: port $PROXY_PORT"
|
||||
echo " Heavy threshold: $HEAVY_THRESHOLD tokens"
|
||||
echo " Max offload: $MAX_OFFLOAD concurrent"
|
||||
echo ""
|
||||
|
||||
# Step 1: Launch all instances with kv_role=kv_both
|
||||
combined_args=""
|
||||
bootstrap_ports=""
|
||||
for i in $(seq 0 $((N_INSTANCES - 1))); do
|
||||
port=$((BASE_PORT + i))
|
||||
bootstrap=$((8998 + i))
|
||||
master_port=$((29500 + i))
|
||||
logfile="/tmp/elastic_inst_${i}.log"
|
||||
|
||||
echo " Instance $i: GPU $i, port $port, bootstrap $bootstrap"
|
||||
|
||||
VLLM_MOONCAKE_BOOTSTRAP_PORT=$bootstrap \
|
||||
MASTER_PORT=$master_port \
|
||||
CUDA_VISIBLE_DEVICES=$i \
|
||||
$VLLM serve "$MODEL" \
|
||||
--host 0.0.0.0 --port $port \
|
||||
--tensor-parallel-size 1 \
|
||||
--trust-remote-code --enable-prefix-caching --enforce-eager \
|
||||
--dtype auto --gpu-memory-utilization 0.9 --max-model-len 200000 \
|
||||
--kv-transfer-config \
|
||||
'{"kv_connector":"MooncakeConnector","kv_role":"kv_both"}' \
|
||||
> "$logfile" 2>&1 &
|
||||
|
||||
combined_args="$combined_args http://127.0.0.1:$port"
|
||||
bootstrap_ports="${bootstrap_ports:+$bootstrap_ports,}$bootstrap"
|
||||
|
||||
sleep 2 # stagger startup to avoid port collision
|
||||
done
|
||||
|
||||
# Step 2: Wait for all instances
|
||||
echo ""
|
||||
echo "Waiting for instances..."
|
||||
for i in $(seq 0 $((N_INSTANCES - 1))); do
|
||||
port=$((BASE_PORT + i))
|
||||
timeout 600 bash -c "until curl -s localhost:$port/v1/models > /dev/null 2>&1; do sleep 5; done"
|
||||
echo " Instance $i (port $port) ready"
|
||||
done
|
||||
|
||||
# Step 3: Wait for bootstrap servers
|
||||
echo "Waiting for bootstrap servers..."
|
||||
for i in $(seq 0 $((N_INSTANCES - 1))); do
|
||||
bp=$((8998 + i))
|
||||
timeout 120 bash -c "until curl -s localhost:$bp/query > /dev/null 2>&1; do sleep 2; done"
|
||||
echo " Bootstrap $bp ready"
|
||||
done
|
||||
|
||||
# Step 4: Start proxy with --offload
|
||||
echo ""
|
||||
echo "Starting proxy (offload mode)..."
|
||||
$PYTHON "$PROJECT_DIR/scripts/cache_aware_proxy.py" \
|
||||
--combined $combined_args \
|
||||
--bootstrap-ports "$bootstrap_ports" \
|
||||
--offload \
|
||||
--heavy-threshold $HEAVY_THRESHOLD \
|
||||
--port $PROXY_PORT &
|
||||
sleep 5
|
||||
|
||||
# Step 5: Smoke test
|
||||
echo ""
|
||||
echo "Smoke test..."
|
||||
result=$(curl -s -m 120 http://localhost:$PROXY_PORT/v1/completions \
|
||||
-X POST -H "Content-Type: application/json" \
|
||||
-d "{\"model\":\"$MODEL\",\"prompt\":[100,200,300],\"max_tokens\":3,\"temperature\":0}" 2>&1)
|
||||
if echo "$result" | grep -q "choices"; then
|
||||
echo " Smoke test passed!"
|
||||
else
|
||||
echo " WARNING: Smoke test failed: $result"
|
||||
fi
|
||||
|
||||
echo ""
|
||||
echo "=== Elastic P2P ready ==="
|
||||
echo " Endpoint: http://localhost:$PROXY_PORT"
|
||||
echo " Breakdown: curl http://localhost:$PROXY_PORT/breakdown"
|
||||
echo " Instance logs: /tmp/elastic_inst_*.log"
|
||||
echo ""
|
||||
echo "Run benchmark:"
|
||||
echo " python -m replayer --trace traces/sampled_1000req_seed42.jsonl \\"
|
||||
echo " --output outputs/elastic_p2p/metrics.jsonl \\"
|
||||
echo " --endpoint http://localhost:$PROXY_PORT \\"
|
||||
echo " --time-scale 20 --max-inflight-sessions 8 -v"
|
||||
echo ""
|
||||
|
||||
wait
|
||||
Reference in New Issue
Block a user