sskj/experiments/pro6000/glm52_pro6000_sglang_multinode_tp16/run_bench.sh

#!/usr/bin/env bash
# TP×DP matrix benchmark for DeepSeek-V4-Flash on SGLang (Docker).
set -Eeuo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
EXPERIMENT_NAME="$(basename "$SCRIPT_DIR")"

# shellcheck source=/dev/null
source "${SCRIPT_DIR}/../../../scripts/common/lib.sh"
# shellcheck source=/dev/null
source "${SCRIPT_DIR}/../../../scripts/common/platform.sh"
# shellcheck source=/dev/null
source "${SCRIPT_DIR}/config.env"

RUN_ID="${RUN_ID:-$(date '+%Y%m%d-%H%M%S')}"
RESULT_BASE="${SCRIPT_DIR}/results"
MATRIX_FILE="${MATRIX_FILE:-${SCRIPT_DIR}/matrix.json}"
MATRIX_MODE="${MATRIX_MODE:-Y}"
SCENARIO_TIMEOUT_S="${SCENARIO_TIMEOUT_S:-1800}"
GPU_MEM_SAMPLE_INTERVAL_S="${GPU_MEM_SAMPLE_INTERVAL_S:-1}"
DRY_RUN="${DRY_RUN:-0}"
GRID_LIMIT="${GRID_LIMIT:-0}"

# Export variables used inside functions that are called via bash -c subshells.
export BENCH_DATASET_NAME DATASET_PATH MODEL_PATH RESULT_BASE DOCKER_IMAGE USE_DOCKER_CLIENT SGLANG_BENCH_MODULE

if [[ -x "${VENV_CLIENT}/bin/python" ]]; then
  PYTHON="${VENV_CLIENT}/bin/python"
else
  PYTHON="$(command -v python3)"
fi

DOCKER_IMAGE="${DOCKER_IMAGE:-lmsysorg/sglang:nightly-dev-cu13-20260720-b3570a45}"

log_dir_global="${RESULT_BASE}/${RUN_ID}/logs"
mkdir -p "$log_dir_global"
log_init "${log_dir_global}/orchestrator.log"

log "experiment=${EXPERIMENT_NAME} run_id=${RUN_ID} platform=${PLATFORM} hardware=${HARDWARE}"
log "matrix_mode=${MATRIX_MODE} matrix_file=${MATRIX_FILE} dry_run=${DRY_RUN} grid_limit=${GRID_LIMIT}"

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

is_server_healthy() {
  curl --fail --silent --show-error --max-time 10 "http://127.0.0.1:${SGLANG_PORT}/v1/models" >/dev/null 2>&1
}

stop_server() {
  local tp="$1"
  local dp="$2"

  log "stopping multi-node sglang (tp=${tp}, dp=${dp}) on both nodes"
  # Delegate to the multi-node stop orchestrator, which tears down node0
  # (local) and node1 (remote via SSH) together: pid file, docker rm, pkill.
  bash "${SCRIPT_DIR}/stop_sglang_multinode.sh" "$tp" "$dp" \
    >> "${log_dir_global}/sglang_tp${tp}_dp${dp}.stop.log" 2>&1 || true
}

build_server_args() {
  local tp="$1"
  local dp="$2"

  local args=(
    "python3 -m sglang.launch_server" --model-path "$MODEL_PATH"
    --trust-remote-code
    --tp-size "$tp"
    --dp-size "$dp"
    --quantization "$QUANTIZATION"
    --moe-runner-backend "$MOE_RUNNER_BACKEND"
    --mem-fraction-static "$MEM_FRACTION_STATIC"
    --context-length "$CONTEXT_LENGTH"
    --max-running-requests "$MAX_RUNNING_REQUESTS"
    --host 0.0.0.0
    --port "$SGLANG_PORT"
    --dist-init-addr "${NODE0_IP}:${DIST_INIT_PORT}"
    --nnodes "$NNODES"
    --node-rank 0
  )
  printf '%s ' "${args[@]}"
}

start_server() {
  local tp="$1"
  local dp="$2"

  log "starting sglang server tp=${tp} dp=${dp}"
  if [[ -n "${CONTAINER_NAME:-}" ]]; then
    bash "${SCRIPT_DIR}/run_sglang_in_container.sh" "$tp" "$dp" \
      >> "${log_dir_global}/sglang_tp${tp}_dp${dp}.server.outer.log" 2>&1
  else
    bash "${SCRIPT_DIR}/start_sglang_dp.sh" "$tp" "$dp" \
      >> "${log_dir_global}/sglang_tp${tp}_dp${dp}.server.outer.log" 2>&1
  fi

  if ! is_server_healthy; then
    log "error: sglang server tp=${tp} dp=${dp} failed health check on port ${SGLANG_PORT}"
    return 1
  fi
  log "sglang server tp=${tp} dp=${dp} is healthy on port ${SGLANG_PORT}"
}

restart_server() {
  local tp="$1"
  local dp="$2"
  log "restarting sglang server tp=${tp} dp=${dp} after non-OOM failure"
  stop_server "$tp" "$dp"
  sleep 10
  start_server "$tp" "$dp"
}

run_bench_serving() {
  # Inject the offline workload choice consistently for warmup and main runs.
  local -a dataset_args=(--dataset-name "$BENCH_DATASET_NAME")
  if [[ "$BENCH_DATASET_NAME" == "random" ]]; then
    dataset_args+=(--dataset-path "$DATASET_PATH")
  elif [[ "$BENCH_DATASET_NAME" == "random-ids" ]]; then
    : # random-ids does not need --tokenize-prompt
  else
    dataset_args+=(--tokenize-prompt)
  fi
  if [[ "${USE_DOCKER_CLIENT:-1}" == "1" ]]; then
    local vol_args=()
    vol_args+=("-v" "${MODEL_PATH}:${MODEL_PATH}:ro")
    if [[ "$BENCH_DATASET_NAME" == "random" ]]; then
      vol_args+=("-v" "${DATASET_PATH}:${DATASET_PATH}:ro")
    fi
    vol_args+=("-v" "${RESULT_BASE}:${RESULT_BASE}")
    docker run --rm \
      --network host \
      "${vol_args[@]}" \
      -e PYTHONUNBUFFERED=1 \
      --entrypoint python3 \
      "${DOCKER_IMAGE}" \
      -m "$SGLANG_BENCH_MODULE" "${dataset_args[@]}" "$@"
  else
    "$PYTHON" -m "$SGLANG_BENCH_MODULE" "${dataset_args[@]}" "$@"
  fi
}
export -f run_bench_serving

run_warmup() {
  local input_len="$1"
  local output_len="$2"

  log "warming up (input=${input_len}, output=${output_len}, num=1)"
  bash -c '
    run_bench_serving \
      --backend sglang \
      --host 127.0.0.1 \
      --port "'"$SGLANG_PORT"'" \
      --random-input-len "'"$input_len"'" \
      --random-output-len "'"$output_len"'" \
      --num-prompts 1 \
      --max-concurrency 1 \
      --request-rate 10000 \
      --output-file /dev/null \
      --output-details \
      >> "'"${log_dir_global}/warmup.log"'" 2>&1
  '
  log "warmup completed"
}

scenario_already_completed() {
  local output_file="$1"
  local expected="$2"
  [[ -s "$output_file" ]] || return 1
  local completed
  completed="$("$PYTHON" -c "
import json, sys
path = sys.argv[1]
try:
    with open(path, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if line:
                data = json.loads(line)
                print(data.get('completed', 0))
                break
except Exception:
    print(0)
" "$output_file")"
  [[ "${completed:-0}" -ge "$expected" ]]
}

scenario_already_processed() {
  local result_root="$1"
  local scenario_name="$2"
  local json_path="${result_root}/results.json"
  [[ -f "$json_path" ]] || return 1
  "$PYTHON" -c "
import json, sys
path, name = sys.argv[1], sys.argv[2]
try:
    with open(path, 'r', encoding='utf-8') as f:
        data = json.load(f)
    for s in data.get('scenarios', []):
        if s.get('name') == name:
            if s.get('status') or s.get('metrics', {}).get('success', 0) > 0:
                sys.exit(0)
except Exception:
    pass
sys.exit(1)
" "$json_path" "$scenario_name"
}

detect_oom() {
  local detail_log="$1"
  local server_outer_log="$2"
  local pattern='CUDA out of memory|torch\.OutOfMemoryError|OutOfMemory|out of memory|OOM|RESOURCE_EXHAUSTED|Failed to allocate memory'
  if grep -Eiq "$pattern" "$detail_log" "$server_outer_log" 2>/dev/null; then
    return 0
  fi
  return 1
}

start_gpu_monitor() {
  local csv_path="$1"
  mkdir -p "$(dirname "$csv_path")"
  nvidia-smi \
    --query-gpu=timestamp,index,memory.used,memory.total,utilization.gpu \
    --format=csv \
    -l "$GPU_MEM_SAMPLE_INTERVAL_S" \
    > "$csv_path" 2>/dev/null &
  echo $!
}

stop_gpu_monitor() {
  local pid="$1"
  if kill -0 "$pid" 2>/dev/null; then
    kill "$pid" 2>/dev/null || true
    sleep 1
    kill -9 "$pid" 2>/dev/null || true
  fi
}

append_scenario_record() {
  local result_root="$1"
  local json_path="$result_root/results.json"
  shift
  local scenario_json
  scenario_json="$("$PYTHON" -c "
import json, sys
pairs = [a.split('=', 1) for a in sys.argv[1:]]
d = {}
for k, v in pairs:
    try:
        d[k] = json.loads(v)
    except json.JSONDecodeError:
        d[k] = v
print(json.dumps(d, ensure_ascii=False))
" "$@")"
  PYTHON="$PYTHON" append_scenario_to_json "$json_path" "$scenario_json"
}

record_skipped_csv() {
  local csv_path="$1"
  shift
  # Args: key=value
  local row
  row="$("$PYTHON" -c "
import csv, json, sys, io
pairs = [a.split('=', 1) for a in sys.argv[1:]]
d = {}
for k, v in pairs:
    try:
        d[k] = json.loads(v)
    except json.JSONDecodeError:
        d[k] = v
buf = io.StringIO()
writer = csv.DictWriter(buf, fieldnames=['engine','tp','dp','mark','isl','dsl','concurrency','status','reason','detail_log'], extrasaction='ignore')
writer.writerow(d)
print(buf.getvalue().strip())
" "$@")"
  echo "$row" >> "$csv_path"
}

skip_remaining_scenarios() {
  local result_root="$1"
  local scenario_tsv="$2"
  local start_index="$3"
  local status="$4"
  local reason="$5"
  local tp="$6"
  local dp="$7"
  local skipped_csv="${RESULT_BASE}/${RUN_ID}/skipped_after_oom.csv"

  local i=0
  tail -n +2 "$scenario_tsv" | while IFS=$'\t' read -r mark isl dsl conc num; do
    if (( i < start_index )); then
      i=$((i + 1))
      continue
    fi
    i=$((i + 1))
    local sname="c${conc}_i${isl}_o${dsl}"
    if scenario_already_processed "$result_root" "$sname"; then
      continue
    fi
    append_scenario_record "$result_root" \
      "name=${sname}" \
      "config=$(jq -n --arg phase main --argjson c "$conc" --argjson i "$isl" --argjson o "$dsl" --arg dataset "$BENCH_DATASET_NAME" --argjson n "$num" '{phase: $phase, concurrency: $c, input_len: $i, output_len: $o, dataset: $dataset, num_prompts: $n}')" \
      "status=\"${status}\"" \
      "note=\"${reason}\""
    record_skipped_csv "$skipped_csv" \
      "engine=sglang" "tp=${tp}" "dp=${dp}" "mark=${mark}" "isl=${isl}" "dsl=${dsl}" "concurrency=${conc}" "status=${status}" "reason=${reason}"
  done
}

# ---------------------------------------------------------------------------
# Per-configuration runner
# ---------------------------------------------------------------------------

run_parallel_config() {
  local tp="$1"
  local dp="$2"
  local config_label="tp${tp}_dp${dp}"
  local result_root="${RESULT_BASE}/${RUN_ID}/${config_label}"
  local raw_dir="${result_root}/raw_outputs"
  local gpu_log_dir="${result_root}/gpu_logs"
  local phase_log_dir="${result_root}/logs"
  mkdir -p "$raw_dir" "$gpu_log_dir" "$phase_log_dir"

  log "===== ${config_label} START ====="

  # Generate scenario list for this config.
  local scenario_tsv="${result_root}/scenarios.tsv"
  "$PYTHON" "${SCRIPT_DIR}/generate_scenarios.py" \
    --matrix "$MATRIX_FILE" \
    --mode "$MATRIX_MODE" \
    > "$scenario_tsv"
  local total_scenarios
  total_scenarios="$(tail -n +2 "$scenario_tsv" | wc -l)"
  log "generated ${total_scenarios} scenarios for ${config_label}"

  # Write metadata.
  ensure_result_root "$result_root"
  write_metadata_json \
    "${result_root}/results.json" \
    "${EXPERIMENT_NAME}_${config_label}" \
    "$RUN_ID" \
    "$MODEL_PATH" \
    "sglang" \
    "sglang" \
    "$HARDWARE" \
    "$ACCELERATOR" \
    "$CHIP" \
    "experiments/${EXPERIMENT_NAME}/run_bench.sh" \
    "$DOCKER_IMAGE" \
    "H200 SGLang TP×DP matrix for DeepSeek-V4-Flash on RTX 6000D"

  local server_args_str
  server_args_str="$(build_server_args "$tp" "$dp")"
  jq --arg tp "$tp" --arg dp "$dp" --arg cuda "$CUDA_VISIBLE_DEVICES" --arg args "$server_args_str" \
    '.config = {
       "tp": ($tp | tonumber),
       "dp": ($dp | tonumber),
       "cuda_visible_devices": $cuda,
       "backend": "sglang",
       "server_start_script": "experiments/'${EXPERIMENT_NAME}'/start_sglang_dp.sh",
       "server_args": $args
     }' "${result_root}/results.json" > "${result_root}/results.json.tmp" && \
    mv "${result_root}/results.json.tmp" "${result_root}/results.json"

  if [[ "$DRY_RUN" == "1" ]]; then
    log "DRY_RUN: would start server with args: ${server_args_str}"
    local line
    tail -n +2 "$scenario_tsv" | while IFS=$'\t' read -r mark isl dsl conc num; do
      log "DRY_RUN: ${config_label} scenario mark=${mark} c=${conc} i=${isl} o=${dsl} n=${num}"
    done
    log "===== ${config_label} DONE (dry run) ====="
    return 0
  fi

  # Initialize skipped_after_oom.csv for this run.
  local skipped_csv="${RESULT_BASE}/${RUN_ID}/skipped_after_oom.csv"
  if [[ ! -f "$skipped_csv" ]]; then
    echo "engine,tp,dp,mark,isl,dsl,concurrency,status,reason,detail_log" > "$skipped_csv"
  fi

  # Start server once for this TP×DP config.
  if ! start_server "$tp" "$dp"; then
    log "ERROR: ${config_label} failed to start; skipping all scenarios"
    skip_remaining_scenarios "$result_root" "$scenario_tsv" 0 "SKIPPED_SERVICE_START_FAILED" "service failed to start" "$tp" "$dp"
    log "===== ${config_label} DONE ====="
    return 0
  fi

  # Warmup with a small prompt before the first scenario.
  run_warmup 1024 128 || true

  # Read scenarios into an array so we can skip remaining entries on failure.
  local -a scenarios=()
  while IFS= read -r line; do
    scenarios+=("$line")
  done < <(tail -n +2 "$scenario_tsv")

  local i mark isl dsl conc num
  local output_file detail_log gpu_csv sname bench_rc
  for (( i = 0; i < ${#scenarios[@]}; i++ )); do
    IFS=$'\t' read -r mark isl dsl conc num <<< "${scenarios[$i]}"

    if [[ "$GRID_LIMIT" -gt 0 && "$i" -ge "$GRID_LIMIT" ]]; then
      log "GRID_LIMIT=${GRID_LIMIT} reached; skipping remaining scenarios"
      skip_remaining_scenarios "$result_root" "$scenario_tsv" "$i" "SKIPPED_GRID_LIMIT" "GRID_LIMIT reached" "$tp" "$dp"
      break
    fi

    sname="c${conc}_i${isl}_o${dsl}"
    output_file="${raw_dir}/sglang_main_${conc}_${isl}_${dsl}.jsonl"
    detail_log="${phase_log_dir}/sglang_${config_label}_${sname}.log"
    gpu_csv="${gpu_log_dir}/gpu_mem_${conc}_${isl}_${dsl}.csv"

    if scenario_already_completed "$output_file" "$num" || scenario_already_processed "$result_root" "$sname"; then
      log "skipping already-processed ${config_label} scenario: ${sname}"
      continue
    fi

    log "running ${config_label} scenario: mark=${mark} c=${conc} i=${isl} o=${dsl} n=${num}"

    local gpu_pid
    gpu_pid="$(start_gpu_monitor "$gpu_csv")"

    bench_rc=0
    timeout "$SCENARIO_TIMEOUT_S" bash -c '
      run_bench_serving \
        --backend sglang \
        --host 127.0.0.1 \
        --port "'"$SGLANG_PORT"'" \
        --random-input-len "'"$isl"'" \
        --random-output-len "'"$dsl"'" \
        --num-prompts "'"$num"'" \
        --max-concurrency "'"$conc"'" \
        --request-rate 10000 \
        --output-file "'"$output_file"'" \
        --output-details \
        > "'"$detail_log"'" 2>&1
    ' || bench_rc=$?

    stop_gpu_monitor "$gpu_pid"

    if [[ "$bench_rc" -eq 0 ]]; then
      log "finished ${config_label} scenario: output=${output_file}"
      append_scenario_record "$result_root" \
        "name=${sname}" \
        "config=$(jq -n --arg phase main --argjson c "$conc" --argjson i "$isl" --argjson o "$dsl" --arg dataset "$BENCH_DATASET_NAME" --argjson n "$num" '{phase: $phase, concurrency: $c, input_len: $i, output_len: $o, dataset: $dataset, num_prompts: $n}')" \
        "status=\"completed\"" \
        "note=\"benchmark finished successfully\""
      continue
    fi

    # Failure handling.
    if detect_oom "$detail_log" "${log_dir_global}/sglang_tp${tp}_dp${dp}.server.outer.log"; then
      log "ERROR: ${config_label} scenario ${sname} triggered OOM; stopping config"
      append_scenario_record "$result_root" \
        "name=${sname}" \
        "config=$(jq -n --arg phase main --argjson c "$conc" --argjson i "$isl" --argjson o "$dsl" --arg dataset "$BENCH_DATASET_NAME" --argjson n "$num" '{phase: $phase, concurrency: $c, input_len: $i, output_len: $o, dataset: $dataset, num_prompts: $n}')" \
        "status=\"OOM\"" \
        "note=\"detected CUDA out-of-memory\""
      record_skipped_csv "$skipped_csv" \
        "engine=sglang" "tp=${tp}" "dp=${dp}" "mark=${mark}" "isl=${isl}" "dsl=${dsl}" "concurrency=${conc}" "status=OOM" "reason=detected CUDA out-of-memory" "detail_log=${detail_log}"
      stop_server "$tp" "$dp"
      skip_remaining_scenarios "$result_root" "$scenario_tsv" "$((i + 1))" "SKIPPED_AFTER_OOM" "previous case OOM" "$tp" "$dp"
      break
    fi

    log "ERROR: ${config_label} scenario ${sname} failed (rc=${bench_rc}); see ${detail_log}"
    if [[ "$mark" == "P" ]]; then
      log "optional (P) scenario failed; recording as skipped and continuing"
      append_scenario_record "$result_root" \
        "name=${sname}" \
        "config=$(jq -n --arg phase main --argjson c "$conc" --argjson i "$isl" --argjson o "$dsl" --arg dataset "$BENCH_DATASET_NAME" --argjson n "$num" '{phase: $phase, concurrency: $c, input_len: $i, output_len: $o, dataset: $dataset, num_prompts: $n}')" \
        "status=\"skipped_optional\"" \
        "note=\"optional scenario failed (rc=${bench_rc})\""
      record_skipped_csv "$skipped_csv" \
        "engine=sglang" "tp=${tp}" "dp=${dp}" "mark=${mark}" "isl=${isl}" "dsl=${dsl}" "concurrency=${conc}" "status=skipped_optional" "reason=optional scenario failed (rc=${bench_rc})" "detail_log=${detail_log}"
      continue
    fi

    # Mandatory scenario failed but not OOM: try to restart the server.
    if restart_server "$tp" "$dp"; then
      run_warmup 1024 128 || true
      log "resuming ${config_label} after server restart"
      continue
    fi

    log "ERROR: ${config_label} server restart failed; skipping remaining scenarios"
    append_scenario_record "$result_root" \
      "name=${sname}" \
      "config=$(jq -n --arg phase main --argjson c "$conc" --argjson i "$isl" --argjson o "$dsl" --arg dataset "$BENCH_DATASET_NAME" --argjson n "$num" '{phase: $phase, concurrency: $c, input_len: $i, output_len: $o, dataset: $dataset, num_prompts: $n}')" \
      "status=\"FAILED\"" \
      "note=\"scenario failed and server restart failed (rc=${bench_rc})\""
    record_skipped_csv "$skipped_csv" \
      "engine=sglang" "tp=${tp}" "dp=${dp}" "mark=${mark}" "isl=${isl}" "dsl=${dsl}" "concurrency=${conc}" "status=FAILED" "reason=scenario failed and server restart failed" "detail_log=${detail_log}"
    skip_remaining_scenarios "$result_root" "$scenario_tsv" "$((i + 1))" "SKIPPED_RESTART_FAILED" "server restart failed" "$tp" "$dp"
    break
  done

  stop_server "$tp" "$dp"

  # Parse results.
  log "parsing ${config_label} results"
  "$PYTHON" "${SCRIPT_DIR}/../../../scripts/common/parse_backend.py" "$result_root" --backend sglang \
    >> "${phase_log_dir}/parse.log" 2>&1 || {
      log "WARNING: parser failed for ${config_label}; see ${phase_log_dir}/parse.log"
    }

  log "===== ${config_label} DONE ====="
}

# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

# Cleanup any leftovers.
for cfg in "${PARALLEL_CONFIGS[@]}"; do
  read -r tp dp <<< "$cfg"
  stop_server "$tp" "$dp"
done

# Run each parallel configuration.
for cfg in "${PARALLEL_CONFIGS[@]}"; do
  read -r tp dp <<< "$cfg"
  run_parallel_config "$tp" "$dp"
done

# Generate cross-configuration comparison.
log "generating comparison report"
"$PYTHON" "${SCRIPT_DIR}/compare.py" \
  --run-root "${RESULT_BASE}/${RUN_ID}" \
  --output "${RESULT_BASE}/${RUN_ID}/comparison.md" \
  >> "${log_dir_global}/compare.log" 2>&1 || {
    log "WARNING: comparison script failed; see ${log_dir_global}/compare.log"
  }

log "all results saved to ${RESULT_BASE}/${RUN_ID}"