26O14m
#!/usr/bin/env bash
# ==========================================================================
# STAGE 3a-3 (신규): 혼합(Mixed) 워크로드 테스트
# 목적: 짧은 챗(1k in/512 out)과 긴 RAG형(32k in/512 out) 요청이 "동시에" 같은
# 서버로 들어올 때, 긴 프리필이 짧은 요청의 디코딩을 막는 Head-of-line
# blocking이 얼마나 발생하는지 확인. stage3a의 "격리 테스트" 결과와
# 비교해서 짧은 요청의 TTFT/ITL이 얼마나 나빠지는지가 핵심 지표.
# 사전조건: stage3a2 결과로 정한 최적 서버 설정값을 아래 SERVER_EXTRA_ARGS에 반영 권장
# ==========================================================================
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck disable=SC1091
source "${VALIDATION_ENV_FILE:-${SCRIPT_DIR}/validation_env.sh}"
export HF_HUB_OFFLINE=1
export TRANSFORMERS_OFFLINE=1
export HF_DATASETS_OFFLINE=1
LOCAL_MODEL_ROOT="${LOCAL_MODEL_ROOT:-/mnt/local-nvme-cache/models}"
MODEL_70B="${MODEL_70B:-${LOCAL_MODEL_ROOT}/Meta-Llama-3.1-70B-Instruct}"
[ -d "${MODEL_70B}" ] || { echo "모델 없음: ${MODEL_70B}"; exit 1; }
# stage3a2 결과 반영 지점 (기본값은 3a와 동일 보수적 설정)
SERVER_EXTRA_ARGS="${SERVER_EXTRA_ARGS:---gpu-memory-utilization 0.90 --enable-chunked-prefill}"
TP_SIZE=8
VLLM_PORT=8000
SERVER_HOST="127.0.0.1"
# 콜드스타트(CUDA 그래프 캡처/커널 컴파일 등, prefill=TTFT에 집중적으로 영향)가 "가장 먼저
# 실행되는 베이스라인(isolated_short)"에 전부 몰려서 기준선을 왜곡하는 문제를 막기 위한 워밍업.
NUM_WARMUPS=0 # legacy compatibility; explicit workload warm-up 사용
# v13.1 Stage3a3 optional stability controls.
# Enhanced crossover defaults: 5 total rounds = 1 discard + 4 balanced retained rounds.
# Set STAGE3A3_CROSSOVER_ENABLED=0 and/or STAGE3A3_CONVERGENCE_WARMUP_ENABLED=0 to disable.
STAGE3A3_CROSSOVER_ENABLED="${STAGE3A3_CROSSOVER_ENABLED:-1}"
STAGE3A3_CROSSOVER_ROUNDS="${STAGE3A3_CROSSOVER_ROUNDS:-5}"
STAGE3A3_CROSSOVER_DISCARD_ROUNDS="${STAGE3A3_CROSSOVER_DISCARD_ROUNDS:-1}"
STAGE3A3_ORDER_EFFECT_WARN_PCT="${STAGE3A3_ORDER_EFFECT_WARN_PCT:-10}"
STAGE3A3_MIN_RETAINED_ROUNDS="${STAGE3A3_MIN_RETAINED_ROUNDS:-4}"
STAGE3A3_MIN_DIRECTION_CONSISTENCY_PCT="${STAGE3A3_MIN_DIRECTION_CONSISTENCY_PCT:-75}"
STAGE3A3_CONVERGENCE_SCOPE="${STAGE3A3_CONVERGENCE_SCOPE:-initial}"
STAGE3A3_CONVERGENCE_WARMUP_ENABLED="${STAGE3A3_CONVERGENCE_WARMUP_ENABLED:-1}"
STAGE3A3_WARMUP_WINDOW_REQUESTS="${STAGE3A3_WARMUP_WINDOW_REQUESTS:-20}"
STAGE3A3_WARMUP_MIN_WINDOWS="${STAGE3A3_WARMUP_MIN_WINDOWS:-3}"
STAGE3A3_WARMUP_MAX_WINDOWS="${STAGE3A3_WARMUP_MAX_WINDOWS:-10}"
STAGE3A3_WARMUP_CONVERGENCE_PCT="${STAGE3A3_WARMUP_CONVERGENCE_PCT:-3}"
STAGE3A3_WARMUP_TTFT_CONVERGENCE_PCT="${STAGE3A3_WARMUP_TTFT_CONVERGENCE_PCT:-5}"
STAGE3A3_WARMUP_CONSECUTIVE_WINDOWS="${STAGE3A3_WARMUP_CONSECUTIVE_WINDOWS:-2}"
STAGE3A3_DIAG_METRICS_INTERVAL="${STAGE3A3_DIAG_METRICS_INTERVAL:-1}"
STAGE3A3_DIAG_KV_HIGH_PCT="${STAGE3A3_DIAG_KV_HIGH_PCT:-85}"
STAGE3A3_DIAG_WAITING_THRESHOLD="${STAGE3A3_DIAG_WAITING_THRESHOLD:-1}"
if [[ "${STAGE3A3_CROSSOVER_ENABLED}" != "1" ]]; then
echo "ERROR: crossover must be enabled for an order-balanced qualification (STAGE3A3_CROSSOVER_ENABLED=1)" >&2
exit 2
fi
if (( STAGE3A3_CROSSOVER_ROUNDS - STAGE3A3_CROSSOVER_DISCARD_ROUNDS < 4 )); then
echo "ERROR: at least four retained crossover rounds are required" >&2; exit 2
fi
LOG_DIR="${VALIDATION_ROOT}/stage3a3_mixed_$(date +%Y%m%d_%H%M%S)"
mkdir -p "${LOG_DIR}"
log() { echo -e "\n\033[1;32m[$(date '+%Y-%m-%d %H:%M:%S')] $1\033[0m"; }
validation_activate_venv
source "${SCRIPT_DIR}/stage3a_warmup_lib.sh"
log "vLLM 버전: $(python3 -c 'import vllm; print(vllm.__version__)' 2>/dev/null || echo '확인 불가')"
log "vLLM 서버 기동 (혼합 워크로드용, extra=[${SERVER_EXTRA_ARGS}])"
python3 -m vllm.entrypoints.openai.api_server \
--model "${MODEL_70B}" --served-model-name "llama-3.1-70b" \
--tensor-parallel-size "${TP_SIZE}" --max-model-len 65536 \
--port "${VLLM_PORT}" ${SERVER_EXTRA_ARGS} \
> "${LOG_DIR}/server.log" 2>&1 &
SERVER_PID=$!
until curl -s "http://${SERVER_HOST}:${VLLM_PORT}/v1/models" > /dev/null 2>&1; do
kill -0 "${SERVER_PID}" 2>/dev/null || { echo "서버 기동 실패: ${LOG_DIR}/server.log"; exit 1; }
sleep 5
done
log "서버 준비 완료"
stage3a_engine_warmup "${SERVER_HOST}" "${VLLM_PORT}" "llama-3.1-70b" "${MODEL_70B}" "mixed"
# --- 서버 /metrics의 TTFT/ITL 히스토그램에서 P50/P90/P99 직접 추출 (신규) ---
# 클라이언트가 계산한 median 하나만으로는 콜드스타트/이상치 때문인지 구분이 안 되므로,
# 서버가 직접 집계한 값을 별도 저장해서 교차검증. vLLM 0.30.0 metric family 기준.
capture_metrics_snapshot() {
local tag=$1
local raw="${LOG_DIR}/metrics_raw_${tag}.txt"
curl -s "http://${SERVER_HOST}:${VLLM_PORT}/metrics" > "${raw}" || { log " /metrics 조회 실패"; return; }
python3 - "${raw}" "${LOG_DIR}/metrics_percentiles_${tag}.json" << 'PYEOF'
import re, sys, json
raw_path, out_path = sys.argv[1], sys.argv[2]
text = open(raw_path).read()
def percentiles_from_histogram(metric_name):
pattern = re.compile(rf'{re.escape(metric_name)}_bucket\{{[^}}]*le="([^"]+)"[^}}]*\}}\s+([0-9.eE+-]+)')
buckets = []
for le, count in pattern.findall(text):
le_val = float("inf") if le == "+Inf" else float(le)
buckets.append((le_val, float(count)))
if not buckets:
return None
buckets.sort(key=lambda x: x[0])
total = buckets[-1][1]
if total <= 0:
return None
result = {}
for p in (0.5, 0.9, 0.99):
target = total * p
for le_val, cum in buckets:
if cum >= target:
result[f"p{int(p*100)}"] = None if le_val == float("inf") else round(le_val * 1000, 1)
break
result["sample_count"] = int(total)
return result
out = {"ttft_ms": percentiles_from_histogram("vllm:time_to_first_token_seconds"),
"itl_ms": percentiles_from_histogram("vllm:inter_token_latency_seconds")}
json.dump(out, open(out_path, "w"), indent=2)
if out["ttft_ms"]:
print(f" [서버 /metrics] TTFT P50={out['ttft_ms'].get('p50')}ms P90={out['ttft_ms'].get('p90')}ms "
f"P99={out['ttft_ms'].get('p99')}ms (n={out['ttft_ms']['sample_count']})")
else:
print(" [서버 /metrics] time_to_first_token_seconds 히스토그램 없음 — raw 파일 직접 확인 필요")
PYEOF
}
# Return 0 when the latest warm-up windows have converged. Writes per-window JSON files.
convergence_warmup() {
local tag=$1 in_len=$2 out_len=$3 concurrency=$4
local stable=0 w tmp prev="" cur="" decision
log "[convergence warm-up] ${tag}: min=${STAGE3A3_WARMUP_MIN_WINDOWS}, max=${STAGE3A3_WARMUP_MAX_WINDOWS}, n/window=${STAGE3A3_WARMUP_WINDOW_REQUESTS}"
for ((w=1; w<=STAGE3A3_WARMUP_MAX_WINDOWS; w++)); do
tmp="${LOG_DIR}/warmup_convergence_${tag}_w${w}.json"
"${VLLM_BIN}" bench serve \
--backend openai-chat --endpoint /v1/chat/completions \
--host "${SERVER_HOST}" --port "${VLLM_PORT}" \
--model "llama-3.1-70b" --tokenizer "${MODEL_70B}" \
--dataset-name random --random-input-len "${in_len}" --random-output-len "${out_len}" \
--num-prompts "${STAGE3A3_WARMUP_WINDOW_REQUESTS}" --max-concurrency "${concurrency}" --request-rate inf \
--save-result --result-dir "${LOG_DIR}" --result-filename "$(basename "${tmp}")" >/dev/null || true
cur=$(python3 - "${tmp}" <<'PYEOF'
import json,sys
try:d=json.load(open(sys.argv[1]))
except Exception:d={}
print(f"{float(d.get('request_throughput') or 0):.9f},{float(d.get('median_ttft_ms') or 0):.9f}")
PYEOF
)
decision=$(python3 - "${prev}" "${cur}" "${STAGE3A3_WARMUP_CONVERGENCE_PCT}" "${STAGE3A3_WARMUP_TTFT_CONVERGENCE_PCT}" <<'PYEOF'
import sys
prev,cur,tpct,lpct=sys.argv[1:]
if not prev:
print('0,NA,NA'); raise SystemExit
pr,pt=map(float,prev.split(',')); cr,ct=map(float,cur.split(','))
def delta(a,b): return abs(b-a)/abs(a)*100 if a else 999.0
dt,dl=delta(pr,cr),delta(pt,ct)
print(f"{int(dt<=float(tpct) and dl<=float(lpct))},{dt:.3f},{dl:.3f}")
PYEOF
)
IFS=, read -r ok throughput_delta ttft_delta <<< "${decision}"
log " window=${w} req/s,ttft=${cur} delta(req/s)=${throughput_delta}% delta(TTFT)=${ttft_delta}% stable=${ok}"
if (( w >= STAGE3A3_WARMUP_MIN_WINDOWS )) && [[ "${ok}" == "1" ]]; then stable=$((stable+1)); else stable=0; fi
if (( stable >= STAGE3A3_WARMUP_CONSECUTIVE_WINDOWS )); then
echo "CONVERGED,${w}" > "${LOG_DIR}/warmup_status_${tag}.txt"
return 0
fi
prev="${cur}"
done
echo "NOT_CONVERGED,${STAGE3A3_WARMUP_MAX_WINDOWS}" > "${LOG_DIR}/warmup_status_${tag}.txt"
return 1
}
declare -A STAGE3A3_CONVERGED_SHAPES=()
workload_warmup() {
local tag=$1 in_len=$2 out_len=$3 concurrency=$4
local shape="${in_len}_${out_len}_${concurrency}"
if [[ "${STAGE3A3_CONVERGENCE_WARMUP_ENABLED}" != "1" ]]; then
stage3a_load_warmup "${SERVER_HOST}" "${VLLM_PORT}" "llama-3.1-70b" "${MODEL_70B}" "${in_len}" "${out_len}" "${concurrency}" 0 0 "${tag}"
return
fi
if [[ "${STAGE3A3_CONVERGENCE_SCOPE}" == "per_mode" ]]; then
convergence_warmup "${tag}" "${in_len}" "${out_len}" "${concurrency}" || true
elif [[ "${STAGE3A3_CONVERGENCE_SCOPE}" == "initial" ]]; then
if [[ -z "${STAGE3A3_CONVERGED_SHAPES[${shape}]:-}" ]]; then
convergence_warmup "initial_${shape}" "${in_len}" "${out_len}" "${concurrency}" || true
STAGE3A3_CONVERGED_SHAPES["${shape}"]=1
else
stage3a_load_warmup "${SERVER_HOST}" "${VLLM_PORT}" "llama-3.1-70b" "${MODEL_70B}" "${in_len}" "${out_len}" "${concurrency}" 0 0 "${tag}"
fi
else
echo "ERROR: STAGE3A3_CONVERGENCE_SCOPE must be initial or per_mode" >&2
exit 2
fi
}
bench_measure() {
local tag=$1 in_len=$2 out_len=$3 concurrency=$4 num_prompts=$5
"${VLLM_BIN}" bench serve \
--backend openai-chat --endpoint /v1/chat/completions \
--host "${SERVER_HOST}" --port "${VLLM_PORT}" \
--model "llama-3.1-70b" --tokenizer "${MODEL_70B}" \
--dataset-name random --random-input-len "${in_len}" --random-output-len "${out_len}" \
--num-prompts "${num_prompts}" --max-concurrency "${concurrency}" --request-rate inf \
--save-result --result-dir "${LOG_DIR}" --result-filename "${tag}.json" || true
capture_metrics_snapshot "${tag}"
}
# Collect per-second vLLM scheduler/KV telemetry DURING each mode, not only cumulative histograms.
# Samples include running/waiting, KV usage and prefill/decode counters when exported.
diagnostic_sampler() {
local tag="$1"
local file="${LOG_DIR}/diagnostics_${tag}.jsonl"
while :; do
curl -fsS --max-time 3 "http://${SERVER_HOST}:${VLLM_PORT}/metrics" 2>/dev/null |
python3 -c '
import sys,re,json,time
names=("vllm:num_requests_running","vllm:num_requests_waiting",
"vllm:kv_cache_usage_perc","vllm:gpu_cache_usage_perc",
"vllm:num_requests_swapped","vllm:request_prefill_time_seconds_sum",
"vllm:request_decode_time_seconds_sum")
d={"timestamp":time.time()}
for line in sys.stdin:
if line.startswith("#"): continue
m=re.match(r"([^ {]+)(?:\\{[^}]*\\})?\\s+([-+0-9.eE]+)",line)
if m and m.group(1) in names:
try: d[m.group(1)]=max(float(m.group(2)),d.get(m.group(1),float("-inf")))
except ValueError: pass
print(json.dumps(d))
' >> "$file" || true
sleep "${STAGE3A3_DIAG_METRICS_INTERVAL}"
done
}
run_with_diagnostics() {
local tag="$1"; shift
diagnostic_sampler "$tag" &
local sampler_pid=$!
"$@"
local rc=$?
kill "$sampler_pid" 2>/dev/null || true
wait "$sampler_pid" 2>/dev/null || true
return "$rc"
}
run_isolated() {
local suffix=$1
workload_warmup "isolated_short_${suffix}" 1024 512 32
run_with_diagnostics "isolated_${suffix}" bench_measure "isolated_short_c32_${suffix}" 1024 512 32 160
workload_warmup "isolated_long_${suffix}" 32768 512 8
bench_measure "isolated_long_c8_${suffix}" 32768 512 8 16
}
run_mixed() {
local suffix=$1
# Warm both workload shapes before simultaneous measurement; warm-up itself is not concurrent.
workload_warmup "mixed_short_${suffix}" 1024 512 32
workload_warmup "mixed_long_${suffix}" 32768 512 8
diagnostic_sampler "mixed_${suffix}" &
local sampler_pid=$!
bench_measure "mixed_short_c32_${suffix}" 1024 512 32 160 & local spid=$!
bench_measure "mixed_long_c8_${suffix}" 32768 512 8 16 & local lpid=$!
wait "${spid}" "${lpid}"
kill "${sampler_pid}" 2>/dev/null || true
wait "${sampler_pid}" 2>/dev/null || true
}
if [[ "${STAGE3A3_CROSSOVER_ENABLED}" == "1" ]]; then
log "=== Stage3a3 crossover mode: rounds=${STAGE3A3_CROSSOVER_ROUNDS}, discard=${STAGE3A3_CROSSOVER_DISCARD_ROUNDS} ==="
for ((r=0; r<STAGE3A3_CROSSOVER_ROUNDS; r++)); do
if (( r % 2 == 0 )); then
log "=== Round ${r}: isolated -> mixed ==="
run_isolated "r${r}_first"
run_mixed "r${r}_second"
else
log "=== Round ${r}: mixed -> isolated ==="
run_mixed "r${r}_first"
run_isolated "r${r}_second"
fi
done
else
log "=== 1. 베이스라인(격리) — 짧은/긴 요청 ==="
run_isolated "legacy"
log "=== 2. 혼합 워크로드 ==="
run_mixed "legacy"
# Preserve legacy filenames for downstream consumers.
cp -f "${LOG_DIR}/isolated_short_c32_legacy.json" "${LOG_DIR}/isolated_short_c32.json"
cp -f "${LOG_DIR}/isolated_long_c8_legacy.json" "${LOG_DIR}/isolated_long_c8.json"
cp -f "${LOG_DIR}/mixed_short_c32_legacy.json" "${LOG_DIR}/mixed_short_c32.json"
cp -f "${LOG_DIR}/mixed_long_c8_legacy.json" "${LOG_DIR}/mixed_long_c8.json"
fi
kill -TERM "${SERVER_PID}" 2>/dev/null || true
wait "${SERVER_PID}" 2>/dev/null || true
export STAGE3A3_DIAG_KV_HIGH_PCT STAGE3A3_DIAG_WAITING_THRESHOLD
log "=== 결과 비교 ==="
python3 - "${LOG_DIR}" "${STAGE3A3_CROSSOVER_ENABLED}" "${STAGE3A3_CROSSOVER_ROUNDS}" "${STAGE3A3_CROSSOVER_DISCARD_ROUNDS}" "${STAGE3A3_ORDER_EFFECT_WARN_PCT}" "${STAGE3A3_CONVERGENCE_WARMUP_ENABLED}" "${STAGE3A3_MIN_RETAINED_ROUNDS}" "${STAGE3A3_MIN_DIRECTION_CONSISTENCY_PCT}" "${STAGE3A3_CONVERGENCE_SCOPE}" <<'PYEOF'
import glob,json,os,statistics,sys
(log_dir,crossover,rounds,discard,order_warn,conv,min_retained,min_direction,conv_scope)=sys.argv[1:]
rounds,discard=int(rounds),int(discard)
order_warn=float(order_warn); min_retained=int(min_retained); min_direction=float(min_direction)
def load(path):
try:return json.load(open(path))
except Exception:return {}
def metric(path,k):
v=load(path).get(k)
try:return float(v)
except Exception:return 0.0
def med(xs): return statistics.median(xs) if xs else 0.0
def pct(base,value): return (value-base)/base*100 if base else 0.0
def fmt(v): return f"{v:+.2f}%"
rows=[]
if crossover=='1':
for r in range(rounds):
if r < discard: continue
iso_pos='first' if r%2==0 else 'second'
mix_pos='second' if r%2==0 else 'first'
order='AB' if iso_pos=='first' else 'BA' # A=isolated, B=mixed
ip=os.path.join(log_dir,f'isolated_short_c32_r{r}_{iso_pos}.json')
mp=os.path.join(log_dir,f'mixed_short_c32_r{r}_{mix_pos}.json')
row={'round':r,'order':order,'iso_pos':iso_pos,'mix_pos':mix_pos,
'iso_ttft':metric(ip,'median_ttft_ms'),'mix_ttft':metric(mp,'median_ttft_ms'),
'iso_itl':metric(ip,'median_itl_ms'),'mix_itl':metric(mp,'median_itl_ms'),
'iso_rps':metric(ip,'request_throughput'),'mix_rps':metric(mp,'request_throughput')}
row['paired_ttft_degradation_pct']=pct(row['iso_ttft'],row['mix_ttft'])
row['paired_itl_degradation_pct']=pct(row['iso_itl'],row['mix_itl'])
rows.append(row)
else:
ip=os.path.join(log_dir,'isolated_short_c32.json'); mp=os.path.join(log_dir,'mixed_short_c32.json')
row={'round':0,'order':'AB','iso_pos':'first','mix_pos':'second',
'iso_ttft':metric(ip,'median_ttft_ms'),'mix_ttft':metric(mp,'median_ttft_ms'),
'iso_itl':metric(ip,'median_itl_ms'),'mix_itl':metric(mp,'median_itl_ms'),
'iso_rps':metric(ip,'request_throughput'),'mix_rps':metric(mp,'request_throughput')}
row['paired_ttft_degradation_pct']=pct(row['iso_ttft'],row['mix_ttft'])
row['paired_itl_degradation_pct']=pct(row['iso_itl'],row['mix_itl'])
rows=[row]
iso_ttft=med([x['iso_ttft'] for x in rows]); mix_ttft=med([x['mix_ttft'] for x in rows])
iso_itl=med([x['iso_itl'] for x in rows]); mix_itl=med([x['mix_itl'] for x in rows])
aggregate_ttft=pct(iso_ttft,mix_ttft)
aggregate_itl=pct(iso_itl,mix_itl)
paired_ttft=med([x['paired_ttft_degradation_pct'] for x in rows])
paired_itl=med([x['paired_itl_degradation_pct'] for x in rows])
ab=[x['paired_ttft_degradation_pct'] for x in rows if x['order']=='AB']
ba=[x['paired_ttft_degradation_pct'] for x in rows if x['order']=='BA']
ab_med=med(ab) if ab else None
ba_med=med(ba) if ba else None
order_delta=abs(ab_med-ba_med) if ab_med is not None and ba_med is not None else 0.0
# Direction consistency is relative to the paired median's sign. Zero median => require near-zero rounds only.
eps=1e-12
if abs(paired_ttft)<=eps:
agreeing=sum(1 for x in rows if abs(x['paired_ttft_degradation_pct'])<=eps)
else:
sign=1 if paired_ttft>0 else -1
agreeing=sum(1 for x in rows if (x['paired_ttft_degradation_pct']>0)==(sign>0))
direction_consistency=100.0*agreeing/len(rows) if rows else 0.0
warm_files=glob.glob(os.path.join(log_dir,'warmup_status_*.txt'))
warm_states=[open(x).read().strip() for x in warm_files]
not_conv=[x for x in warm_states if x.startswith('NOT_CONVERGED')]
reasons=[]
if crossover=='1' and len(rows)<min_retained:
reasons.append(f'retained_rounds<{min_retained}')
if crossover=='1' and direction_consistency<min_direction:
reasons.append(f'direction_consistency<{min_direction:.1f}%')
if crossover=='1' and order_delta>order_warn:
reasons.append(f'AB_BA_order_delta>{order_warn:.1f}%')
if conv=='1' and not_conv:
reasons.append('warmup_not_converged')
stability='UNSTABLE' if reasons else 'STABLE'
verdict='REVIEW_REQUIRED' if stability=='UNSTABLE' else 'READY_FOR_PASS_FAIL_EVALUATION'
# Scheduler pressure vs KV pressure is a diagnostic hypothesis, never an automatic causal proof.
def diag_summary(pattern):
samples=[]
for f in glob.glob(os.path.join(log_dir,pattern)):
for line in open(f):
try: samples.append(json.loads(line))
except Exception: pass
def maximum(*keys):
vals=[float(d[k]) for d in samples for k in keys if k in d]
return max(vals) if vals else None
return {'samples':len(samples),
'max_waiting':maximum('vllm:num_requests_waiting'),
'max_running':maximum('vllm:num_requests_running'),
'max_kv_usage':maximum('vllm:kv_cache_usage_perc','vllm:gpu_cache_usage_perc')}
diag_iso=diag_summary('diagnostics_isolated_r*.jsonl')
diag_mix=diag_summary('diagnostics_mixed_r*.jsonl')
kv=diag_mix['max_kv_usage']
waiting=diag_mix['max_waiting']
if kv is None or waiting is None:
cause='INSUFFICIENT_TELEMETRY'
elif kv>=float(os.environ.get('STAGE3A3_DIAG_KV_HIGH_PCT','85'))/100:
cause='KV_PRESSURE_POSSIBLE'
elif waiting>=float(os.environ.get('STAGE3A3_DIAG_WAITING_THRESHOLD','1')):
cause='SCHEDULER_QUEUE_OR_PREFILL_CONTENTION_POSSIBLE'
else:
cause='PREFILL_COMPUTE_CONTENTION_POSSIBLE'
summary={
'schema_version':4,'diagnostics':{'isolated':diag_iso,'mixed':diag_mix,'cause_hypothesis':cause},'crossover_enabled':crossover=='1','rounds':rounds,'discard_rounds':discard,
'retained_rounds':len(rows),'minimum_retained_rounds':min_retained,
'convergence_warmup_enabled':conv=='1','convergence_scope':conv_scope,
'warmup_not_converged_count':len(not_conv),
'rows':rows,
'isolated_median_ttft_ms':iso_ttft,'mixed_median_ttft_ms':mix_ttft,
'aggregate_median_ttft_degradation_pct':aggregate_ttft,
'paired_ttft_degradation_median_pct':paired_ttft,
'isolated_median_itl_ms':iso_itl,'mixed_median_itl_ms':mix_itl,
'aggregate_median_itl_degradation_pct':aggregate_itl,
'paired_itl_degradation_median_pct':paired_itl,
'ab_degradation_median_pct':ab_med,'ba_degradation_median_pct':ba_med,
'ab_ba_order_delta_pct':order_delta,'order_effect_warn_pct':order_warn,
'direction_consistency_pct':direction_consistency,
'minimum_direction_consistency_pct':min_direction,
'result_stability':stability,'stability_reasons':reasons,'verdict':verdict
}
json.dump(summary,open(os.path.join(log_dir,'stage3a3_stability_summary.json'),'w'),indent=2)
lines=[
'# Stage3a-3 Mixed Workload Result','',
f'- Crossover: {summary["crossover_enabled"]}',
f'- Rounds: {rounds}; discarded={discard}; retained={len(rows)} (minimum={min_retained})',
f'- Convergence warm-up: {summary["convergence_warmup_enabled"]} (scope={conv_scope})',
f'- Warm-up not converged: {len(not_conv)}',
f'- Paired TTFT degradation median: {fmt(paired_ttft)} **(primary)**',
f'- Aggregate median TTFT degradation: {fmt(aggregate_ttft)} (reference)',
f'- Direction consistency: {direction_consistency:.1f}% (minimum={min_direction:.1f}%)',
f'- AB degradation median: {"N/A" if ab_med is None else fmt(ab_med)}',
f'- BA degradation median: {"N/A" if ba_med is None else fmt(ba_med)}',
f'- AB/BA order delta: {order_delta:.1f}% (warn>{order_warn:.1f}%)',
f'- Stability: **{stability}**',
f'- Interference hypothesis: **{cause}** (not causal proof)',
f'- Mixed max waiting: {waiting}; max KV fraction: {kv}',
f'- Verdict: **{verdict}**'
]
if reasons:
lines += ['', '## Stability reasons'] + [f'- {x}' for x in reasons]
lines += [
'', '## Short request: isolated vs mixed',
'| Metric | Isolated median | Mixed median | Aggregate degradation | Paired degradation median |',
'|---|---:|---:|---:|---:|',
f'| TTFT | {iso_ttft:.1f} ms | {mix_ttft:.1f} ms | {aggregate_ttft:+.1f}% | {paired_ttft:+.1f}% |',
f'| ITL | {iso_itl:.1f} ms | {mix_itl:.1f} ms | {aggregate_itl:+.1f}% | {paired_itl:+.1f}% |',
'', '## Retained crossover rounds',
'| Round | Order | ISO TTFT | MIX TTFT | Paired degradation |',
'|---:|---|---:|---:|---:|'
]
for x in rows:
lines.append(f'| {x["round"]} | {x["order"]} | {x["iso_ttft"]:.1f} ms | {x["mix_ttft"]:.1f} ms | {x["paired_ttft_degradation_pct"]:+.1f}% |')
lines += [
'', '## Interpretation',
'- Primary interference estimate is the median of matched-pair per-round degradations.',
'- AB/BA order delta compares the degradation conclusion between isolated→mixed and mixed→isolated orders.',
'- Direction consistency measures how many retained rounds agree with the sign of the paired median.',
'- Scheduler-vs-KV attribution requires a controlled A/B run with varied max-num-batched-tokens or KV budget; telemetry alone is not proof.',
'- UNSTABLE results are REVIEW_REQUIRED; the observed degradation is informational and must not be declared PASS/FAIL.',
]
open(os.path.join(log_dir,'mixed_workload_report.md'),'w').write('\n'.join(lines)+'\n')
print('\n'.join(lines))
PYEOF
log "혼합 워크로드 테스트 완료! 리포트: ${LOG_DIR}/mixed_workload_report.md"
engineer