diff --git a/.gitignore b/.gitignore index be8ff8a..6246c25 100644 --- a/.gitignore +++ b/.gitignore @@ -49,6 +49,9 @@ experiments/**/runtime/ # 无关项目 loomeval_yy/ +# 部署层运行时产物 +deploy/logs/ + .gitignore dsv4_dspark_h20_sglang_tp_dp_matrix diff --git a/README.md b/README.md index 35b5189..3184752 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,14 @@ tmux new-session -d -s h20-vllm-adaptive \ - 各平台的镜像 / venv / 模型路径见对应实验目录的 `config.env` 与 `platforms/.env`；**模型与数据集路径是机器相关的**，换机器时按实际路径调整。 - 环境搭建规范见 `envs/README.md` 与 `envs/UV_ENV_SETUP.md`。 +## 运维与部署分层 + +- 模型团队用 `deploy/profiles//-.env` 与 + `python -m sskj.deploy start|stop|status` 管理服务生命周期。 +- 运维只跑测试：`python -m sskj.bench run --url http://: --platform --experiment --profile smoke|matrix`。 +- 运维手册见 [`ops/README.md`](ops/README.md)，统一测试层复用现有 + `scripts/common/parse_backend.py` 等指标口径。 + ## 注意事项 - 提交内容：实验代码 + 最终产物（results.json / report.md / adaptive 的 jsonl 与 summary）；日志、raw_outputs、gpu_logs 不入库（`.gitignore` 已排除）。 diff --git a/deploy/profiles/p800/dsv4_w8a8_int8_sglang.env b/deploy/profiles/p800/dsv4_w8a8_int8_sglang.env new file mode 100644 index 0000000..665ae3a --- /dev/null +++ b/deploy/profiles/p800/dsv4_w8a8_int8_sglang.env @@ -0,0 +1,27 @@ +# P800 + SGLang-XPU + DeepSeek-V4-Flash-INT8 deployment profile. +# Model-team only. Ops should NOT run `python -m sskj.deploy`; they only +# run `python -m sskj.bench` against the URL this profile serves. + +PLATFORM=p800 +MODEL_NAME=DeepSeek-V4-Flash-INT8 +ENGINE=sglang +RUNTIME=docker +DOCKER_IMAGE=iregistry.baidu-int.com/xpu/sglang-p800-pd-disagg-0510:20260511_4202 +CONTAINER_NAME=sglang-dsv4-flash-tpdp +MODEL_PATH=/data1/models/DeepSeek-V4-Flash-INT8 +SERVED_MODEL_NAME=deepseek-v4-flash-int8 +PORT=30014 +HEALTH_PATH=/health +HEALTH_WAIT_S=600 +CONTAINER_PYTHON=/root/miniconda/envs/python310_torch25_cuda/bin/python +PATCH_ROOT=${ROOT_DIR}/platforms/patches/kunlun_p800 + +DEVICE_VARS="XPU_VISIBLE_DEVICES=0,1,2,3,4,5,6,7 CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7 CUDA_DEVICE_ORDER=OAM_ID" +ENGINE_ENV="SGLANG_USE_TRANSFORMERS_V5_TOKENIZER=1 XMLIR_FORCE_USE_XPU_GRAPH=1 SGLANG_DSV4_MODE=2604 PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True SGLANG_NSA_DUAL_STREAM=true SGLANG_NSA_QUANT_WQ_B_WK=false SGLANG_FP8_PAGED_MQA_LOGITS_TORCH=1 SGLANG_OPT_DEEPGEMM_HC_PRENORM=false SGLANG_OPT_USE_TILELANG_MHC_PRE=1 SGLANG_OPT_USE_TILELANG_MHC_POST=1 SGLANG_CLEAN_REQUEST_WHEN_RETRACT=1 SGLANG_SET_CPU_AFFINITY=1 SGLANG_OPT_USE_KLX_TOPK_KERNEL=1 XSGL_INTERTYPE_BFP16=1 ENABLE_FAST_BFP16_ATTN=1 XSGL_USE_DEEP_GEMM_BMM=1 XSGL_XDNN_QUANT=1 XSGL_FUSE_RMS_NORM_QUANT=1 XSGL_TRANSPOSE_MATMUL_WEIGHT=1 XINFER_QUANT_SDNN=1 XSGL_USE_MOE_SIGMOID_GROUP_TOPK_NORM=1 XSGL_EARLY_FIRST_TOKEN=1 XSGL_ENABLE_TGEMM_FP16=1 SGLANG_ENABLE_SPEC_V2=True SGLANG_ENABLE_OVERLAP_PLAN_STREAM=1 PYTHONDONTWRITEBYTECODE=1 XTORCH_OPS_LIB_DIR=/root/miniconda/envs/python310_torch25_cuda/lib/python3.10/site-packages/xtorch_ops XPU_RUNTIME_LIB_DIR=/root/miniconda/envs/python310_torch25_cuda/xcudart/lib BKCL_TREE_THRESHOLD=1048576 CUDA_ENABLE_P2P_NO_UVA=1 NCCL_IB_GID_INDEX=3 IS_DSV4=1 MC_CUSTOM_TOPO_JSON=/workspace/nic_priority_matrix_test.json SGLANG_DSV4_FP4_EXPERTS=false SGLANG_APPLY_CONFIG_BACKUP=auto BKCL_ENABLE_XDR=1 BKCL_RDMA_NICS=eth1,eth1,eth3,eth3,eth5,eth5,eth7,eth7 BKCL_RDMA_VERBS=1 XSGL_INT8_LM_HEAD=1 SGLANG_P800_ALL_GATHER_FALLBACK=0" + +DOCKER_FLAGS="--privileged --network host --ipc host --device /dev/xpu0:/dev/xpu0 --device /dev/xpu1:/dev/xpu1 --device /dev/xpu2:/dev/xpu2 --device /dev/xpu3:/dev/xpu3 --device /dev/xpu4:/dev/xpu4 --device /dev/xpu5:/dev/xpu5 --device /dev/xpu6:/dev/xpu6 --device /dev/xpu7:/dev/xpu7 --device /dev/xpuctrl:/dev/xpuctrl" +VOLUMES="${MODEL_PATH}:/models:ro ${MODEL_PATH}:${MODEL_PATH}:ro ${PATCH_ROOT}/nic_priority_matrix_test.json:/workspace/nic_priority_matrix_test.json:ro" + +LAUNCH_ARGS="--host 0.0.0.0 --port ${PORT} --model-path /models --attention-backend nsa --nsa-prefill klxdsa --nsa-decode klxdsa --trust-remote-code --disable-custom-all-reduce --page-size 64 --mem-fraction-static 0.8 --tensor-parallel-size ${TP} --disable-shared-experts-fusion --quantization w8a8_int8 --kv-cache-dtype float16 --disable-piecewise-cuda-graph --cuda-graph-max-bs 32 --watchdog-timeout 3000000 --tool-call-parser deepseekv4 --reasoning-parser deepseek-v4 --constrained-json-disable-any-whitespace --enable-metrics --enable-request-time-stats-logging --context-length 65536" +DP_FLAG="--dp-size ${DP}" +BOOTSTRAP="cd /workspace; find /root/miniconda/envs/python310_torch25_cuda/lib/python3.10/site-packages/sglang -type d -name __pycache__ -exec rm -rf {} + 2>/dev/null || true; ${CONTAINER_PYTHON} -m pip install --upgrade safetensors -q; ${CONTAINER_PYTHON} -m pip install https://files.pythonhosted.org/packages/14/8b/2a1333a6455c6fad401c2285dee6f58016c55b1cb44cae3a31f8a9cc7d83/apache_tvm_ffi-0.1.0b2-cp310-cp310-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl -q; ${CONTAINER_PYTHON} -c \"import torch; torch.float8_e8m0fnu = torch.uint8; import runpy, sys; sys.argv[0] = 'sglang.launch_server'; runpy.run_module('sglang.launch_server', run_name='__main__')\" ${LAUNCH_ARGS}" diff --git a/deploy/profiles/p800/glm52_w8a8_int8_dynamic.env b/deploy/profiles/p800/glm52_w8a8_int8_dynamic.env new file mode 100644 index 0000000..4ec4fb5 --- /dev/null +++ b/deploy/profiles/p800/glm52_w8a8_int8_dynamic.env @@ -0,0 +1,29 @@ +# P800 + SGLang-XPU + GLM-5.2-W8A8-INT8-Dynamic deployment profile. +# Supports the 2-node deploy previously living in deploy_glm52_2nodes.sh. +# Model-team only. Ops should not run `python -m sskj.deploy`. + +PLATFORM=p800 +MODEL_NAME=GLM-5.2-W8A8-INT8-Dynamic +ENGINE=sglang +RUNTIME=docker +DOCKER_IMAGE=iregistry.baidu-int.com/xpu/sglang-p800-pd-disagg-0510:20260511_4202 +CONTAINER_NAME=glm52_deploy +MODEL_PATH=/data1/models/GLM-5.2-W8A8-INT8-Dynamic +SERVED_MODEL_NAME=glm52 +PORT=30000 +HEALTH_PATH=/health +HEALTH_WAIT_S=600 +NNODES=2 +MASTER_IP=10.100.11.2 +DIST_PORT=6311 +DP_SIZE=2 + +DEVICE_VARS="XPU_VISIBLE_DEVICES=0,1,2,3,4,5,6,7 CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7 CUDA_DEVICE_ORDER=OAM_ID" +ENGINE_ENV="BKCL_ENABLE_XDR=1 BKCL_ENABLE_IBGDA=1 BKCL_SOCKET_IFNAME=eth0 BKCL_RDMA_NICS=eth1,eth2,eth3,eth4,eth5,eth6,eth7,eth8 BKCL_TREE_THRESHOLD=1048576 XSHMEM_MODE=1 XSHMEM_QP_NUM_PER_RANK=32 BKCL_RDMA_VERBS=1 NCCL_IB_GID_INDEX=3 CUDA_ENABLE_P2P_NO_UVA=1 SGLANG_USE_TRANSFORMERS_V5_TOKENIZER=1 XSGL_INTERTYPE_BFP16=1 ENABLE_FAST_BFP16_ATTN=1 XSGL_USE_DEEP_GEMM_BMM=1 XSGL_XDNN_QUANT=1 XSGL_FUSE_RMS_NORM_QUANT=1 XSGL_TRANSPOSE_MATMUL_WEIGHT=1 XINFER_QUANT_SDNN=1 XMLIR_FORCE_USE_XPU_GRAPH=1 XSGL_USE_MOE_SIGMOID_GROUP_TOPK_NORM=1 XSGL_INT8_LM_HEAD=1 SGLANG_CLEAN_REQUEST_WHEN_RETRACT=1 SGLANG_ENABLE_SPEC_V2=True SGLANG_ENABLE_OVERLAP_PLAN_STREAM=1 USE_FAST_BFP16_FC=1 XPUAPI_SDNN_BF16_ROUND_MODE=3 XMLIR_ENABLE_FAST_FC=true XSGL_FAST_SWIGLU=1 SGLANG_NSA_DUAL_STREAM=true SGLANG_NSA_QUANT_WQ_B_WK=false XPU_FLASH_ATTENTION_DECODER_USE_BALANCE=true" + +DOCKER_FLAGS="--privileged --network host --security-opt label=disable --shm-size 64m --device /dev/xpu0:/dev/xpu0 --device /dev/xpu1:/dev/xpu1 --device /dev/xpu2:/dev/xpu2 --device /dev/xpu3:/dev/xpu3 --device /dev/xpu4:/dev/xpu4 --device /dev/xpu5:/dev/xpu5 --device /dev/xpu6:/dev/xpu6 --device /dev/xpu7:/dev/xpu7 --device /dev/xpuctrl:/dev/xpuctrl" +VOLUMES="${MODEL_PATH}:/models:ro ${MODEL_PATH}:${MODEL_PATH}:ro" + +LAUNCH_ARGS="--host 0.0.0.0 --port ${PORT} --model-path /models --dist-init-addr ${MASTER_IP}:${DIST_PORT} --nnodes ${NNODES} --node-rank ${NODE_RANK} --attention-backend nsa --nsa-prefill klxdsa --nsa-decode klxdsa --trust-remote-code --disable-custom-all-reduce --page-size 64 --mem-fraction-static 0.82 --max-prefill-tokens 65536 --chunked-prefill-size 16384 --max-running-requests 128 --tp-size 16 --ep-size 16 --dp-size ${DP_SIZE} --enable-dp-attention --moe-dense-tp-size 1 --enable-dp-lm-head --moe-a2a-backend deepep --deepep-mode auto --quantization w8a8_int8 --disable-shared-experts-fusion --kv-cache-dtype float16 --disable-radix-cache --watchdog-timeout 3000000 --tool-call-parser glm47 --reasoning-parser glm45 --enable-nsa-prefill-context-parallel --enable-single-batch-overlap" + +BOOTSTRAP="source /root/miniconda/bin/activate python310_torch25_cuda; cd /workspace; pip install --no-index --find-links=/workspace/ /workspace/transformers-5.3.0-py3-none-any.whl 2>/dev/null || true; nohup python3 -m sglang.launch_server ${LAUNCH_ARGS} > /workspace/sglang_log_${NODE_RANK}.log 2>&1 & sleep infinity" diff --git a/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/run_adaptive_concurrency.sh b/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/run_adaptive_concurrency.sh index 16b6056..52d055e 100755 --- a/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/run_adaptive_concurrency.sh +++ b/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/run_adaptive_concurrency.sh @@ -16,6 +16,8 @@ source "${SCRIPT_DIR}/config.env" source "${SCRIPT_DIR}/adaptive_config.env" # shellcheck source=/dev/null source "${SCRIPT_DIR}/../../../scripts/common/adaptive_bench_lib.sh" +# shellcheck source=/dev/null +source "${SCRIPT_DIR}/../../../scripts/common/deploy_cli.sh" ENGINE="sglang" ENGINE_PORT="$SGLANG_PORT" @@ -65,13 +67,10 @@ engine_stop_server() { } engine_build_server_args() { - # Keep in sync with the launch args in start_sglang_docker.sh. local tp="$1" local dp="$2" - local args="--host 0.0.0.0 --port ${ENGINE_PORT} --model-path /models --attention-backend nsa --nsa-prefill klxdsa --nsa-decode klxdsa --trust-remote-code --disable-custom-all-reduce --page-size 64 --mem-fraction-static ${MEM_FRACTION_STATIC} --tensor-parallel-size ${tp} --disable-shared-experts-fusion --quantization w8a8_int8 --kv-cache-dtype float16 --disable-piecewise-cuda-graph --cuda-graph-max-bs 32 --watchdog-timeout 3000000 --tool-call-parser deepseekv4 --reasoning-parser deepseek-v4 --constrained-json-disable-any-whitespace --enable-metrics --enable-request-time-stats-logging --context-length ${CONTEXT_LENGTH}" - if (( dp > 1 )); then - args="${args} --dp-size ${dp}" - fi + local args + args="$(deploy_render_args "${DEPLOY_PROFILE:-p800/dsv4_w8a8_int8_sglang}" "$tp" "$dp" "$ENGINE_PORT" "$MODEL_PATH")" printf '%s' "$args" } diff --git a/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/run_adaptive_concurrency_add16.sh b/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/run_adaptive_concurrency_add16.sh index 91176a6..22efeb7 100755 --- a/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/run_adaptive_concurrency_add16.sh +++ b/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/run_adaptive_concurrency_add16.sh @@ -16,6 +16,8 @@ source "${SCRIPT_DIR}/config.env" source "${SCRIPT_DIR}/adaptive_config.env" # shellcheck source=/dev/null source "${SCRIPT_DIR}/../../../scripts/common/adaptive_bench_lib.sh" +# shellcheck source=/dev/null +source "${SCRIPT_DIR}/../../../scripts/common/deploy_cli.sh" ENGINE="sglang" ENGINE_PORT="$SGLANG_PORT" @@ -65,13 +67,10 @@ engine_stop_server() { } engine_build_server_args() { - # Keep in sync with the launch args in start_sglang_docker.sh. local tp="$1" local dp="$2" - local args="--host 0.0.0.0 --port ${ENGINE_PORT} --model-path /models --attention-backend nsa --nsa-prefill klxdsa --nsa-decode klxdsa --trust-remote-code --disable-custom-all-reduce --page-size 64 --mem-fraction-static ${MEM_FRACTION_STATIC} --tensor-parallel-size ${tp} --disable-shared-experts-fusion --quantization w8a8_int8 --kv-cache-dtype float16 --disable-piecewise-cuda-graph --cuda-graph-max-bs 32 --watchdog-timeout 3000000 --tool-call-parser deepseekv4 --reasoning-parser deepseek-v4 --constrained-json-disable-any-whitespace --enable-metrics --enable-request-time-stats-logging --context-length ${CONTEXT_LENGTH}" - if (( dp > 1 )); then - args="${args} --dp-size ${dp}" - fi + local args + args="$(deploy_render_args "${DEPLOY_PROFILE:-p800/dsv4_w8a8_int8_sglang}" "$tp" "$dp" "$ENGINE_PORT" "$MODEL_PATH")" printf '%s' "$args" } diff --git a/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/run_bench.sh b/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/run_bench.sh index bb643c8..d428cb0 100755 --- a/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/run_bench.sh +++ b/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/run_bench.sh @@ -11,6 +11,8 @@ source "${SCRIPT_DIR}/../../../scripts/common/lib.sh" source "${SCRIPT_DIR}/../../../scripts/common/platform.sh" # shellcheck source=/dev/null source "${SCRIPT_DIR}/config.env" +# shellcheck source=/dev/null +source "${SCRIPT_DIR}/../../../scripts/common/deploy_cli.sh" RUN_ID="${RUN_ID:-$(date '+%Y%m%d-%H%M%S')}" RESULT_BASE="${SCRIPT_DIR}/results" @@ -61,14 +63,10 @@ stop_server() { } build_server_args() { - # Keep in sync with the launch args in start_sglang_docker.sh. local tp="$1" local dp="$2" - - local args="--host 0.0.0.0 --port ${SGLANG_PORT} --model-path /models --attention-backend nsa --nsa-prefill klxdsa --nsa-decode klxdsa --trust-remote-code --disable-custom-all-reduce --page-size 64 --mem-fraction-static ${MEM_FRACTION_STATIC} --tensor-parallel-size ${tp} --disable-shared-experts-fusion --quantization w8a8_int8 --kv-cache-dtype float16 --disable-piecewise-cuda-graph --cuda-graph-max-bs 32 --watchdog-timeout 3000000 --tool-call-parser deepseekv4 --reasoning-parser deepseek-v4 --constrained-json-disable-any-whitespace --enable-metrics --enable-request-time-stats-logging --context-length ${CONTEXT_LENGTH}" - if [[ "$dp" -gt 1 ]]; then - args="${args} --dp-size ${dp}" - fi + local args + args="$(deploy_render_args "${DEPLOY_PROFILE:-p800/dsv4_w8a8_int8_sglang}" "$tp" "$dp" "$SGLANG_PORT" "$MODEL_PATH")" printf '%s' "$args" } diff --git a/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/start_sglang_docker.sh b/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/start_sglang_docker.sh index f19ebb4..f90b4f7 100755 --- a/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/start_sglang_docker.sh +++ b/experiments/p800/dsv4_p800_sglang_tp_dp_matrix/start_sglang_docker.sh @@ -1,9 +1,6 @@ #!/usr/bin/env bash -# Start the P800 SGLang INT8 server in Docker for a given TP×DP configuration. +# Start the P800 SGLang INT8 server through the shared deployment layer. # Usage: start_sglang_docker.sh -# -# Based on the proven P800 INT8 launch from dsv4_p800_long_context_matrix, -# with --tensor-parallel-size/--dp-size parameterized per config. set -Eeuo pipefail TP="${1:-}" @@ -20,157 +17,18 @@ source "${SCRIPT_DIR}/../../../scripts/common/lib.sh" source "${SCRIPT_DIR}/../../../scripts/common/platform.sh" # shellcheck source=/dev/null source "${SCRIPT_DIR}/config.env" +# shellcheck source=/dev/null +source "${SCRIPT_DIR}/../../../scripts/common/deploy_cli.sh" -PORT="${SGLANG_PORT:-30014}" RUNTIME_BASE="${RUNTIME_BASE:-${SCRIPT_DIR}/runtime}" mkdir -p "${RUNTIME_BASE}/logs" -SERVER_LOG="${RUNTIME_BASE}/logs/${EXPERIMENT}_sglang_docker_tp${TP}_dp${DP}_$(date +%Y%m%d_%H%M%S).log" -log "starting P800 SGLang INT8 server (tp=${TP}, dp=${DP})" -log "model: ${MODEL_PATH}" -log "port: ${PORT}" -log "container: ${CONTAINER_NAME}" - -# Stop any existing container with the same name. -docker rm -f "$CONTAINER_NAME" 2>/dev/null || true - -# Build device args. -device_args="" -for i in 0 1 2 3 4 5 6 7; do - device_args="${device_args} --device /dev/xpu${i}:/dev/xpu${i}" -done -device_args="${device_args} --device /dev/xpuctrl:/dev/xpuctrl" - -# Environment variables required by the P800 SGLang INT8 image. -env_args=( - -e XPU_VISIBLE_DEVICES="${XPU_VISIBLE_DEVICES}" - -e CUDA_VISIBLE_DEVICES="${CUDA_VISIBLE_DEVICES}" - -e CUDA_DEVICE_ORDER=OAM_ID - -e SGLANG_USE_TRANSFORMERS_V5_TOKENIZER=1 - -e XMLIR_FORCE_USE_XPU_GRAPH=1 - -e SGLANG_DSV4_MODE=2604 - -e PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True - -e SGLANG_NSA_DUAL_STREAM=true - -e SGLANG_NSA_QUANT_WQ_B_WK=false - -e SGLANG_FP8_PAGED_MQA_LOGITS_TORCH=1 - -e SGLANG_OPT_DEEPGEMM_HC_PRENORM=false - -e SGLANG_OPT_USE_TILELANG_MHC_PRE=1 - -e SGLANG_OPT_USE_TILELANG_MHC_POST=1 - -e SGLANG_CLEAN_REQUEST_WHEN_RETRACT=1 - -e SGLANG_SET_CPU_AFFINITY=1 - -e SGLANG_OPT_USE_KLX_TOPK_KERNEL=1 - -e XSGL_INTERTYPE_BFP16=1 - -e ENABLE_FAST_BFP16_ATTN=1 - -e XSGL_USE_DEEP_GEMM_BMM=1 - -e XSGL_XDNN_QUANT=1 - -e XSGL_FUSE_RMS_NORM_QUANT=1 - -e XSGL_TRANSPOSE_MATMUL_WEIGHT=1 - -e XINFER_QUANT_SDNN=1 - -e XSGL_USE_MOE_SIGMOID_GROUP_TOPK_NORM=1 - -e XSGL_EARLY_FIRST_TOKEN=1 - -e XSGL_ENABLE_TGEMM_FP16=1 - -e SGLANG_ENABLE_SPEC_V2=True - -e SGLANG_ENABLE_OVERLAP_PLAN_STREAM=1 - -e PYTHONDONTWRITEBYTECODE=1 - -e XTORCH_OPS_LIB_DIR=/root/miniconda/envs/python310_torch25_cuda/lib/python3.10/site-packages/xtorch_ops - -e XPU_RUNTIME_LIB_DIR=/root/miniconda/envs/python310_torch25_cuda/xcudart/lib - -e BKCL_TREE_THRESHOLD=1048576 - -e CUDA_ENABLE_P2P_NO_UVA=1 - -e NCCL_IB_GID_INDEX=3 - -e IS_DSV4=1 - -e MC_CUSTOM_TOPO_JSON=/workspace/nic_priority_matrix_test.json - # INT8 specific - -e SGLANG_DSV4_FP4_EXPERTS=false - -e SGLANG_APPLY_CONFIG_BACKUP=auto - -e BKCL_ENABLE_XDR=1 - -e BKCL_RDMA_NICS=eth1,eth1,eth3,eth3,eth5,eth5,eth7,eth7 - -e BKCL_RDMA_VERBS=1 - -e XSGL_INT8_LM_HEAD=1 - -e SGLANG_P800_ALL_GATHER_FALLBACK=0 -) - -# Launch args. Same proven P800 INT8 command as -# dsv4_p800_long_context_matrix/start_sglang.sh, with TP/DP parameterized. -# Note: whether this image supports --dp-size on XPU is part of what this -# experiment probes; TP=8/DP=1 matches the production configuration. -launch_args="--host 0.0.0.0 --port ${PORT} --model-path /models --attention-backend nsa --nsa-prefill klxdsa --nsa-decode klxdsa --trust-remote-code --disable-custom-all-reduce --page-size 64 --mem-fraction-static ${MEM_FRACTION_STATIC} --tensor-parallel-size ${TP} --disable-shared-experts-fusion --quantization w8a8_int8 --kv-cache-dtype float16 --disable-piecewise-cuda-graph --cuda-graph-max-bs 32 --watchdog-timeout 3000000 --tool-call-parser deepseekv4 --reasoning-parser deepseek-v4 --constrained-json-disable-any-whitespace --enable-metrics --enable-request-time-stats-logging --context-length ${CONTEXT_LENGTH}" - -if [[ "$DP" -gt 1 ]]; then - launch_args="${launch_args} --dp-size ${DP}" -fi - -# Base64-encode the bootstrap command to avoid host-shell quoting issues. -server_cmd=$(cat </dev/null || true -/root/miniconda/envs/python310_torch25_cuda/bin/pip install --upgrade safetensors -q -/root/miniconda/envs/python310_torch25_cuda/bin/pip install https://files.pythonhosted.org/packages/14/8b/2a1333a6455c6fad401c2285dee6f58016c55b1cb44cae3a31f8a9cc7d83/apache_tvm_ffi-0.1.0b2-cp310-cp310-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl -q -/root/miniconda/envs/python310_torch25_cuda/bin/python -c "import torch; torch.float8_e8m0fnu = torch.uint8; import runpy, sys; sys.argv[0] = 'sglang.launch_server'; runpy.run_module('sglang.launch_server', run_name='__main__')" ${launch_args} -EOF -) -server_cmd_b64=$(printf '%s' "$server_cmd" | base64 -w0) - -patch_mounts=( - -v "${PATCH_ROOT}/nic_priority_matrix_test.json:/workspace/nic_priority_matrix_test.json:ro" -) -# Patch the image's sglang.bench_serving to also report TTFT/TPOT/E2E P95. -# The P800 image only ships median/P99 for those metrics, and the adaptive -# search reads p95_ttft_ms for its TTFT SLO stop. The container is recreated -# per TP/DP config, so the patch must be mounted at every `docker run`. -BENCH_SERVING_PATCH="${SCRIPT_DIR}/patches/bench_serving.py" -if [[ -f "${BENCH_SERVING_PATCH}" ]]; then - patch_mounts+=( - -v "${BENCH_SERVING_PATCH}:/root/miniconda/envs/python310_torch25_cuda/lib/python3.10/site-packages/sglang/bench_serving.py:ro" - ) -fi -# Mount the ShareGPT seed dataset only when it exists on the host, so that -# BENCH_DATASET_NAME=random-ids (no seed needed) never makes docker create a -# stray directory at DATASET_PATH. -if [[ -f "${DATASET_PATH}" ]]; then - patch_mounts+=( -v "${DATASET_PATH}:${CONTAINER_DATASET_PATH}:ro" ) -fi - -echo "=== Starting P800 SGLang INT8 server in Docker (TP=${TP}, DP=${DP}) ===" -echo "Image: ${DOCKER_IMAGE}" -echo "Model: ${MODEL_PATH}" -echo "Container name: ${CONTAINER_NAME}" -echo "Host port: ${PORT}" -echo "Command: sglang.launch_server ${launch_args}" -echo "Log: docker logs ${CONTAINER_NAME}" - -docker run -d \ - --name "${CONTAINER_NAME}" \ - --privileged \ - --network host \ - --ipc host \ - ${device_args} \ - -v "${MODEL_PATH}:/models:ro" \ - -v "${MODEL_PATH}:${MODEL_PATH}:ro" \ - "${patch_mounts[@]}" \ - "${env_args[@]}" \ - "${DOCKER_IMAGE}" \ - bash -c "echo '${server_cmd_b64}' | base64 -d | bash" \ - >> "${SERVER_LOG}" 2>&1 - -log "container ${CONTAINER_NAME} started, waiting for health" -healthy=0 -for ((i = 1; i <= 600; i++)); do - if curl --fail --silent --show-error --max-time 5 "http://127.0.0.1:${PORT}/health" >/dev/null 2>&1; then - healthy=1 - break - fi - if [[ "$(docker inspect -f '{{.State.Running}}' "${CONTAINER_NAME}" 2>/dev/null)" != "true" ]]; then - log "ERROR: container ${CONTAINER_NAME} exited during startup; last server logs:" - docker logs --tail 100 "${CONTAINER_NAME}" 2>&1 || true - exit 1 - fi - sleep 1 -done -if [[ "$healthy" == "1" ]]; then - log "container ${CONTAINER_NAME} is healthy" - exit 0 -else - log "ERROR: container ${CONTAINER_NAME} failed health check; last server logs:" - docker logs --tail 100 "${CONTAINER_NAME}" 2>&1 || true - exit 1 -fi +log "starting P800 SGLang INT8 server (tp=${TP}, dp=${DP}) via deploy profile" +deploy_start \ + "${DEPLOY_PROFILE:-p800/dsv4_w8a8_int8_sglang}" \ + "$TP" \ + "$DP" \ + "${RUNTIME_BASE}/logs" \ + "${SGLANG_PORT:-30014}" \ + "$MODEL_PATH" \ + "$CONTAINER_NAME" diff --git a/ops/README.md b/ops/README.md new file mode 100644 index 0000000..65a32ab --- /dev/null +++ b/ops/README.md @@ -0,0 +1,90 @@ +# 运维一键测试手册 + +这份手册只说明怎么**跑测试**。模型部署由模型团队负责，运维不需要理解 +SGLang/vLLM 的引擎参数，也不需要执行 `python -m sskj.deploy`。 + +## 0. 准备 + +```bash +cd /data1/yy/sskj +python3 -m pip install -e . --no-deps +``` + +如果不想安装，也可以每次执行时加 `PYTHONPATH=src`： + +```bash +cd /data1/yy/sskj +PYTHONPATH=src python3 -m sskj.bench --help +``` + +## 1. 查看可测试的实验 + +```bash +PYTHONPATH=src python3 -m sskj.bench list --platform p800 +PYTHONPATH=src python3 -m sskj.bench list +``` + +## 2. 冒烟测试 + +```bash +PYTHONPATH=src python3 -m sskj.bench run \ + --url http://10.100.11.2:30014 \ + --platform p800 \ + --experiment dsv4_p800_sglang_tp_dp_matrix \ + --profile smoke +``` + +## 3. 完整矩阵测试 + +```bash +PYTHONPATH=src python3 -m sskj.bench run \ + --url http://10.100.11.2:30014 \ + --platform p800 \ + --experiment dsv4_p800_sglang_tp_dp_matrix \ + --profile matrix +``` + +`--profile matrix` 使用实验目录里的 `matrix.json`，只测试 `Y` 标记的场景， +每个 ISL 取低/高两个并发档位。 + +## 4. 先看计划再执行 + +任何 `run` 命令都可以加 `--dry-run`，只打印 URL、client 模式、场景清单和 +结果路径，不发送请求。 + +## 5. 结果在哪里 + +结果默认写回实验目录： + +```text +experiments///results// +├── results.json +├── report.md +├── run_manifest.json +├── raw_outputs/ +└── logs/ +``` + +把 `` 或 `report.md` 路径发给模型团队即可。 + +## 6. client 模式 + +默认自动选择： + +| 平台 | 默认 client | +|---|---| +| P800 | `server-container`（进入已部署容器执行，保留厂商 bench_serving） | +| H20/H200/6000D | `docker`（独立 sglang client 容器） | +| 910C | `server-container` 或 `native`（按实验配置自动选择） | + +跨机测试时可用 `--client docker --client-image lmsysorg/sglang:latest` +指定外部 client。 + +## 7. 常见问题 + +- 服务不健康：确认 `--url` 的端口可从测试机访问，服务已启动且 `/health` 可通。 +- 数据集缺失：仓库 `datasets/` 下没有 ShareGPT 文件时，统一 CLI 会自动退回 + `random-ids`，可用于连通性验证；正式矩阵前先放好数据集。 +- P95 指标：P800 使用容器内 `bench_serving`，解析层会从逐请求数据补算 P95， + 与旧实验口径一致。 +- 不要执行部署命令：`python -m sskj.deploy` 仅模型团队使用。 diff --git a/pyproject.toml b/pyproject.toml index aa3cf19..1481945 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,11 +9,19 @@ description = "Multi-platform LLM serving benchmark framework (sglang/vllm on P8 requires-python = ">=3.10" dependencies = [] +[project.scripts] +sskj-bench = "sskj.bench.cli:main" +sskj-deploy = "sskj.deploy.cli:main" + [project.optional-dependencies] dev = [ "ruff>=0.5.0", ] +[tool.setuptools] +package-dir = {"" = "src"} +packages = ["sskj", "sskj.bench", "sskj.common", "sskj.deploy"] + [tool.ruff] line-length = 120 target-version = "py310" @@ -25,4 +33,4 @@ ignore = ["E501"] [tool.ruff.format] quote-style = "double" -indent-style = "space" \ No newline at end of file +indent-style = "space" diff --git a/scripts/common/deploy_cli.sh b/scripts/common/deploy_cli.sh new file mode 100644 index 0000000..4766cec --- /dev/null +++ b/scripts/common/deploy_cli.sh @@ -0,0 +1,44 @@ +#!/usr/bin/env bash +# Helpers to delegate server lifecycle and argument rendering to +# `python -m sskj.deploy`. The deployment profile is the single source of +# engine launch args; experiment scripts should call these helpers instead of +# maintaining their own copies. + +DEPLOY_PYTHON="${PYTHON:-python3}" + +deploy_profile_abs() { + local profile="$1" + printf '%s' "${ROOT_DIR}/deploy/profiles/${profile}.env" +} + +deploy_start() { + local profile="$1" + local tp="$2" + local dp="$3" + local log_dir="$4" + local port="$5" + local model_path="$6" + local container="$7" + PYTHONPATH="${ROOT_DIR}/src" "${DEPLOY_PYTHON}" -m sskj.deploy start \ + --profile "$(deploy_profile_abs "$profile")" \ + --tp "$tp" \ + --dp "$dp" \ + --log-dir "$log_dir" \ + --port "$port" \ + --model-path "$model_path" \ + --container-name "$container" +} + +deploy_render_args() { + local profile="$1" + local tp="$2" + local dp="$3" + local port="$4" + local model_path="$5" + PYTHONPATH="${ROOT_DIR}/src" "${DEPLOY_PYTHON}" -m sskj.deploy render-args \ + --profile "$(deploy_profile_abs "$profile")" \ + --tp "$tp" \ + --dp "$dp" \ + --port "$port" \ + --model-path "$model_path" +} diff --git a/src/sskj/__init__.py b/src/sskj/__init__.py new file mode 100644 index 0000000..6dec3aa --- /dev/null +++ b/src/sskj/__init__.py @@ -0,0 +1,3 @@ +"""sskj: multi-platform LLM serving benchmark and deployment toolkit.""" + +__version__ = "0.1.0" diff --git a/src/sskj/bench/__init__.py b/src/sskj/bench/__init__.py new file mode 100644 index 0000000..88c0ea7 --- /dev/null +++ b/src/sskj/bench/__init__.py @@ -0,0 +1 @@ +"""Unified benchmark client for sskj.""" diff --git a/src/sskj/bench/__main__.py b/src/sskj/bench/__main__.py new file mode 100644 index 0000000..fad2de6 --- /dev/null +++ b/src/sskj/bench/__main__.py @@ -0,0 +1,4 @@ +from sskj.bench.cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/sskj/bench/cli.py b/src/sskj/bench/cli.py new file mode 100644 index 0000000..f5d7db5 --- /dev/null +++ b/src/sskj/bench/cli.py @@ -0,0 +1,238 @@ +"""CLI entrypoint: python -m sskj.bench.""" + +from __future__ import annotations + +import argparse +import sys +from datetime import datetime +from pathlib import Path +from urllib.parse import urlsplit + +from sskj.bench import config as bench_config +from sskj.bench.report import ( + append_status_scenario, + run_parse_backend, + write_manifest, + write_metadata, +) +from sskj.bench.runner import BenchClientOptions, run_scenario, wait_health +from sskj.bench.scenarios import generate_scenarios, smoke_scenario + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="python -m sskj.bench", description="sskj benchmark CLI") + parser.add_argument("--repo-root", type=Path, default=None, help="Repository root (default: auto)") + sub = parser.add_subparsers(dest="command", required=True) + + list_parser = sub.add_parser("list", help="List platforms and experiments") + list_parser.add_argument("--platform", choices=bench_config.PLATFORMS, default=None) + + run_parser = sub.add_parser("run", help="Run smoke or matrix benchmark against an existing endpoint") + run_parser.add_argument("--url", default=None, help="Serving endpoint, e.g. http://10.0.0.2:30000") + run_parser.add_argument("--platform", required=True, choices=bench_config.PLATFORMS) + run_parser.add_argument("--experiment", required=True) + run_parser.add_argument("--profile", choices=["smoke", "matrix"], default="smoke") + run_parser.add_argument("--run-id", default=None) + run_parser.add_argument("--backend", choices=["sglang", "vllm"], default=None) + run_parser.add_argument("--model", default=None) + run_parser.add_argument("--dataset", default=None) + run_parser.add_argument("--client", choices=["auto", "native", "docker", "server-container"], default="auto") + run_parser.add_argument("--client-image", default=None) + run_parser.add_argument("--client-python", default=None) + run_parser.add_argument("--server-container", default=None) + run_parser.add_argument("--container-python", default=None) + run_parser.add_argument("--isl", type=int, default=1024) + run_parser.add_argument("--osl", type=int, default=128) + run_parser.add_argument("--concurrency", type=int, default=8) + run_parser.add_argument("--num-prompts", type=int, default=40) + run_parser.add_argument("--output-dir", type=Path, default=None) + run_parser.add_argument("--timeout", type=int, default=1800) + run_parser.add_argument("--health-wait", type=int, default=600) + run_parser.add_argument("--dry-run", action="store_true") + return parser + + +def _normalize_url(url: str) -> tuple[str, str, int]: + if "://" not in url: + url = f"http://{url}" + parts = urlsplit(url) + host = parts.hostname or "127.0.0.1" + port = parts.port or 30000 + base = f"http://{host}:{port}" + return base, host, port + + +def _cmd_list(args: argparse.Namespace, root: Path) -> int: + platforms = [args.platform] if args.platform else bench_config.list_platforms(root) + for platform in platforms: + experiments = bench_config.list_experiments(platform, root) + print(f"[{platform}]") + for name in experiments: + print(f" {name}") + return 0 + + +def _cmd_run(args: argparse.Namespace, root: Path) -> int: + experiment_dir = bench_config.resolve_experiment(args.platform, args.experiment, root) + platform_env = bench_config.load_platform_env(args.platform, root) + experiment_env = bench_config.load_experiment_env(args.platform, args.experiment, root) + backend = args.backend or bench_config.detect_backend( + args.experiment, + experiment_env, + platform_env, + ) + engine = experiment_env.get("ENGINE") or platform_env.get("ENGINE") or backend + port_key = "SGLANG_PORT" if backend == "sglang" else "VLLM_PORT" + default_port = int(experiment_env.get(port_key) or platform_env.get("DEFAULT_PORT") or "30000") + url = args.url or f"http://127.0.0.1:{default_port}" + base_url, host, port = _normalize_url(url) + + model = ( + args.model + or experiment_env.get("MODEL_PATH") + or experiment_env.get("SERVED_MODEL_NAME") + or experiment_env.get("MODEL_NAME") + or "model" + ) + dataset = args.dataset or experiment_env.get("DATASET_PATH") + client_mode = bench_config.resolve_client_mode(args.client, args.platform, experiment_env) + + if args.profile == "matrix": + matrix_path = experiment_dir / "matrix.json" + if not matrix_path.exists(): + raise SystemExit(f"matrix.json not found for experiment: {matrix_path}") + scenarios = generate_scenarios(matrix_path, mode="Y", target_samples=2) + else: + scenarios = [ + smoke_scenario( + isl=args.isl, + osl=args.osl, + concurrency=args.concurrency, + num_prompts=args.num_prompts, + ) + ] + + run_id = args.run_id or f"ops_{datetime.now().strftime('%Y%m%d-%H%M%S')}" + result_root = args.output_dir / run_id if args.output_dir else experiment_dir / "results" / run_id + + server_container = args.server_container or experiment_env.get("CONTAINER_NAME") + container_python = args.container_python or experiment_env.get("CONTAINER_PYTHON") + client_image = args.client_image or experiment_env.get("DOCKER_CLIENT_IMAGE") or experiment_env.get("DOCKER_IMAGE") + client_python = args.client_python or experiment_env.get("VENV_CLIENT") + if client_python: + client_python = str(Path(client_python) / "bin" / "python") + + print(f"endpoint: {base_url}") + print(f"platform: {args.platform} experiment: {args.experiment}") + print(f"backend: {backend} client: {client_mode}") + print(f"profile: {args.profile} scenarios: {len(scenarios)}") + print(f"result: {result_root}") + for s in scenarios: + print( + f" c={s['concurrency']} isl={s['input_len']} osl={s['output_len']} " + f"num_prompts={s['num_prompts']}" + ) + if args.dry_run: + return 0 + + dataset_name = experiment_env.get("BENCH_DATASET_NAME", "random") + if dataset_name != "random" and not dataset_name.startswith("random"): + dataset_name = "random" + if (not dataset or not Path(dataset).exists()) and dataset_name == "random": + print(f"WARNING: dataset not found, will use random-ids: {dataset}") + dataset_name = "random-ids" + + print("checking service health ...") + if not wait_health(base_url, "/health", args.health_wait): + raise SystemExit(f"service health check failed: {base_url}/health") + + write_metadata( + result_root, + experiment=f"{args.platform}/{args.experiment}", + run_id=run_id, + model=model, + backend=backend, + engine=engine, + hardware=platform_env.get("HARDWARE", ""), + accelerator=platform_env.get("ACCELERATOR", ""), + chip=platform_env.get("CHIP", args.platform), + endpoint=base_url, + client_mode=client_mode, + root=root, + ) + write_manifest( + result_root, + { + "experiment": f"{args.platform}/{args.experiment}", + "run_id": run_id, + "endpoint": base_url, + "backend": backend, + "client_mode": client_mode, + "profile": args.profile, + "scenarios": scenarios, + "dataset": dataset, + "dataset_name": dataset_name, + }, + ) + (result_root / "raw_outputs").mkdir(parents=True, exist_ok=True) + (result_root / "logs").mkdir(parents=True, exist_ok=True) + + for s in scenarios: + output_file = ( + result_root + / "raw_outputs" + / f"{backend}_main_{s['concurrency']}_{s['input_len']}_{s['output_len']}.jsonl" + ) + detail_log = result_root / "logs" / f"{backend}_{s['concurrency']}_{s['input_len']}_{s['output_len']}.log" + options = BenchClientOptions( + backend=backend, + host=host, + port=port, + model=model, + dataset_name=dataset_name, + dataset_path=dataset, + container_dataset_path=experiment_env.get("CONTAINER_DATASET_PATH"), + random_range_ratio=float(experiment_env.get("RANDOM_RANGE_RATIO", "1.0")), + warmup_max_requests=int(experiment_env.get("BENCH_WARMUP_MAX_REQUESTS", "0") or 0), + client_mode=client_mode, + client_image=client_image, + client_python=client_python, + server_container=server_container, + container_python=container_python, + output_file=output_file, + detail_log=detail_log, + timeout_s=args.timeout, + root=root, + ) + name = f"c{s['concurrency']}_i{s['input_len']}_o{s['output_len']}" + print(f"running scenario {name} ...") + rc = run_scenario(s, options) + if rc == 0 and output_file.exists(): + print(f" ok: {output_file}") + else: + print(f" failed rc={rc}: {detail_log}") + append_status_scenario( + result_root, + name=name, + concurrency=s["concurrency"], + input_len=s["input_len"], + output_len=s["output_len"], + num_prompts=s["num_prompts"], + status="FAILED", + note=f"bench_serving rc={rc}", + ) + + print("parsing results ...") + run_parse_backend(sys.executable, root / "scripts" / "common" / "parse_backend.py", result_root, backend) + print(f"result saved: {result_root}") + return 0 + + +def main(argv: list[str] | None = None) -> int: + args = _build_parser().parse_args(argv) + root = args.repo_root or bench_config.repo_root() + if args.command == "list": + return _cmd_list(args, root) + if args.command == "run": + return _cmd_run(args, root) + return 2 diff --git a/src/sskj/bench/config.py b/src/sskj/bench/config.py new file mode 100644 index 0000000..24ece79 --- /dev/null +++ b/src/sskj/bench/config.py @@ -0,0 +1,109 @@ +"""Repository and experiment resolution helpers.""" + +from __future__ import annotations + +from pathlib import Path + +from sskj.common.env import parse_env_file + +PLATFORM_CHIP = { + "p800": "kunlun_p800", + "h20": "nvidia_h20", + "h200": "nvidia_h200", + "pro6000": "nvidia_rtx6000d", + "910c": "ascend_910c", +} + +PLATFORMS = ["p800", "h20", "h200", "pro6000", "910c"] + + +def repo_root() -> Path: + return Path(__file__).resolve().parents[3] + + +def experiments_dir(root: Path | None = None) -> Path: + return (root or repo_root()) / "experiments" + + +def list_platforms(root: Path | None = None) -> list[str]: + base = experiments_dir(root) + return [p.name for p in sorted(base.iterdir()) if p.is_dir() and p.name in PLATFORMS] + + +def list_experiments(platform: str, root: Path | None = None) -> list[str]: + base = experiments_dir(root) / platform + if not base.exists(): + return [] + return sorted(p.name for p in base.iterdir() if p.is_dir() and not p.name.startswith(".")) + + +def resolve_experiment(platform: str, experiment: str, root: Path | None = None) -> Path: + path = experiments_dir(root) / platform / experiment + if not path.exists(): + raise SystemExit(f"experiment not found: {path}") + return path + + +def load_platform_env(platform: str, root: Path | None = None) -> dict[str, str]: + root = root or repo_root() + chip = PLATFORM_CHIP[platform] + path = root / "platforms" / f"{chip}.env" + if not path.exists(): + return {} + return parse_env_file(path, base_env={"ROOT_DIR": str(root)}) + + +def load_experiment_env( + platform: str, + experiment: str, + root: Path | None = None, +) -> dict[str, str]: + root = root or repo_root() + experiment_dir = resolve_experiment(platform, experiment, root) + platform_env = load_platform_env(platform, root) + base = dict(platform_env) + base["ROOT_DIR"] = str(root) + base["SCRIPT_DIR"] = str(experiment_dir) + config_path = experiment_dir / "config.env" + if not config_path.exists(): + return base + return parse_env_file(config_path, base_env=base) + + +def detect_backend( + experiment_name: str, + experiment_env: dict[str, str], + platform_env: dict[str, str], +) -> str: + name = experiment_name.lower() + engine = (experiment_env.get("ENGINE") or platform_env.get("ENGINE") or "").lower() + if "sglang" in name: + return "sglang" + if "vllm" in name: + return "vllm" + if "vllm" in engine: + return "vllm" + return "sglang" + + +def resolve_client_mode( + mode: str, + platform: str, + experiment_env: dict[str, str], +) -> str: + if mode != "auto": + return mode + use_docker_client = experiment_env.get("USE_DOCKER_CLIENT", "1") + if use_docker_client == "0": + if experiment_env.get("CONTAINER_NAME") and experiment_env.get("CONTAINER_PYTHON"): + return "server-container" + return "native" + if experiment_env.get("DOCKER_CLIENT_IMAGE"): + return "docker" + if experiment_env.get("CONTAINER_NAME") and experiment_env.get("CONTAINER_PYTHON"): + return "server-container" + if experiment_env.get("DOCKER_IMAGE"): + return "docker" + if experiment_env.get("VENV_CLIENT") or experiment_env.get("VENV_SGLANG"): + return "native" + return "docker" diff --git a/src/sskj/bench/report.py b/src/sskj/bench/report.py new file mode 100644 index 0000000..c80e346 --- /dev/null +++ b/src/sskj/bench/report.py @@ -0,0 +1,126 @@ +"""Result metadata and report helpers for the unified benchmark CLI.""" + +from __future__ import annotations + +import json +import subprocess +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + + +def git_commit(root: Path) -> str: + try: + return ( + subprocess.run( + ["git", "rev-parse", "--short", "HEAD"], + cwd=root, + capture_output=True, + text=True, + timeout=10, + ) + .stdout.strip() + or "unknown" + ) + except Exception: + return "unknown" + + +def git_dirty(root: Path) -> str: + try: + out = subprocess.run( + ["git", "status", "--porcelain"], + cwd=root, + capture_output=True, + text=True, + timeout=10, + ).stdout.strip() + return "dirty" if out else "clean" + except Exception: + return "unknown" + + +def write_metadata( + result_root: Path, + *, + experiment: str, + run_id: str, + model: str, + backend: str, + engine: str, + hardware: str, + accelerator: str, + chip: str, + endpoint: str, + client_mode: str, + root: Path, +) -> None: + result_root.mkdir(parents=True, exist_ok=True) + metadata = { + "experiment": experiment, + "run_id": run_id, + "timestamp": datetime.now(timezone.utc).astimezone().isoformat(), + "model": model, + "backend": backend, + "engine": engine, + "hardware": hardware, + "accelerator": accelerator, + "chip": chip, + "script": "python -m sskj.bench", + "env": "", + "git_commit": git_commit(root), + "git_dirty": git_dirty(root), + "endpoint": endpoint, + "client_mode": client_mode, + } + data = {"metadata": metadata, "config": {}, "scenarios": []} + with open(result_root / "results.json", "w", encoding="utf-8") as f: + json.dump(data, f, indent=2, ensure_ascii=False) + + +def write_manifest(result_root: Path, payload: dict[str, Any]) -> None: + with open(result_root / "run_manifest.json", "w", encoding="utf-8") as f: + json.dump(payload, f, indent=2, ensure_ascii=False) + + +def append_status_scenario( + result_root: Path, + *, + name: str, + concurrency: int, + input_len: int, + output_len: int, + num_prompts: int, + status: str, + note: str, +) -> None: + path = result_root / "results.json" + data: dict[str, Any] = {"scenarios": []} + if path.exists(): + data = json.loads(path.read_text(encoding="utf-8")) + data.setdefault("scenarios", []).append( + { + "name": name, + "config": { + "phase": "main", + "concurrency": concurrency, + "input_len": input_len, + "output_len": output_len, + "dataset": "random", + "num_prompts": num_prompts, + }, + "status": status, + "note": note, + } + ) + path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") + + +def run_parse_backend(python_bin: str, parse_script: Path, result_root: Path, backend: str) -> int: + return subprocess.run( + [python_bin, str(parse_script), str(result_root), "--backend", backend], + cwd=parse_script.parents[2], + capture_output=True, + text=True, + timeout=120, + ).returncode diff --git a/src/sskj/bench/runner.py b/src/sskj/bench/runner.py new file mode 100644 index 0000000..15caa8d --- /dev/null +++ b/src/sskj/bench/runner.py @@ -0,0 +1,183 @@ +"""bench_serving invocation for native, docker, and server-container clients.""" + +from __future__ import annotations + +import shlex +import subprocess +import sys +import time +import urllib.request +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + + +@dataclass +class BenchClientOptions: + backend: str + host: str + port: int + model: str + dataset_name: str = "random" + dataset_path: str | None = None + container_dataset_path: str | None = None + random_range_ratio: float = 1.0 + warmup_max_requests: int = 0 + client_mode: str = "auto" + client_image: str | None = None + client_python: str | None = None + server_container: str | None = None + container_python: str | None = None + output_file: Path | None = None + detail_log: Path | None = None + timeout_s: int = 1800 + root: Path | None = None + extra_env: dict[str, str] = field(default_factory=dict) + + +def _bench_args(scenario: dict[str, Any], options: BenchClientOptions) -> list[str]: + args = [ + "--backend", + options.backend, + "--host", + options.host, + "--port", + str(options.port), + "--model", + options.model, + "--dataset-name", + options.dataset_name, + "--random-input-len", + str(scenario["input_len"]), + "--random-output-len", + str(scenario["output_len"]), + "--random-range-ratio", + str(options.random_range_ratio), + "--num-prompts", + str(scenario["num_prompts"]), + "--max-concurrency", + str(scenario["concurrency"]), + "--request-rate", + "10000", + "--output-details", + ] + if options.warmup_max_requests and options.warmup_max_requests > 0: + args += ["--warmup-requests", str(options.warmup_max_requests)] + + dataset_path = ( + options.container_dataset_path + if options.client_mode == "server-container" + else options.dataset_path + ) + if options.dataset_name == "random" and dataset_path: + args += ["--dataset-path", dataset_path] + elif options.dataset_name != "random": + args += ["--tokenize-prompt"] + return args + + +def _run(argv: list[str], log_path: Path | None, timeout_s: int) -> int: + log_path.parent.mkdir(parents=True, exist_ok=True) + try: + with open(log_path, "wb") as f: + proc = subprocess.run(argv, stdout=f, stderr=subprocess.STDOUT, timeout=timeout_s) + return proc.returncode + except subprocess.TimeoutExpired: + with open(log_path, "ab") as f: + f.write(b"\n[sskj] bench_serving timed out\n") + return 124 + + +def run_scenario(scenario: dict[str, Any], options: BenchClientOptions) -> int: + if options.output_file is None or options.detail_log is None: + raise ValueError("output_file and detail_log are required") + + base_args = _bench_args(scenario, options) + if options.client_mode == "native": + python_bin = options.client_python or sys.executable + return _run( + [python_bin, "-m", "sglang.bench_serving", *base_args, "--output-file", str(options.output_file)], + options.detail_log, + options.timeout_s, + ) + + if options.client_mode == "docker": + if not options.client_image: + raise SystemExit("docker client mode requires --client-image or DOCKER_CLIENT_IMAGE") + cmd = ["docker", "run", "--rm", "--network", "host"] + if options.root: + cmd += ["-v", f"{options.root}:{options.root}"] + if Path(options.model).exists(): + cmd += ["-v", f"{options.model}:{options.model}:ro"] + if options.dataset_path and Path(options.dataset_path).exists(): + cmd += ["-v", f"{options.dataset_path}:{options.dataset_path}:ro"] + if options.output_file: + cmd += ["-v", f"{options.output_file.parent}:{options.output_file.parent}"] + cmd += ["-e", "HF_HUB_OFFLINE=1", "-e", "TRANSFORMERS_OFFLINE=1", "-e", "HF_DATASETS_OFFLINE=1"] + cmd += [ + options.client_image, + "python", + "-m", + "sglang.bench_serving", + *base_args, + "--output-file", + str(options.output_file), + ] + return _run(cmd, options.detail_log, options.timeout_s) + + if options.client_mode == "server-container": + if not options.server_container or not options.container_python: + raise SystemExit("server-container client mode requires --server-container and --container-python") + container_output = f"/tmp/bench_outputs/{options.output_file.name}" + mkdir = subprocess.run( + ["docker", "exec", options.server_container, "mkdir", "-p", "/tmp/bench_outputs"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + if mkdir.returncode != 0: + raise SystemExit( + f"cannot prepare {options.server_container}:/tmp/bench_outputs (docker exec mkdir rc={mkdir.returncode})" + ) + cmd = [ + "docker", + "exec", + options.server_container, + "env", + "HF_HUB_OFFLINE=1", + "TRANSFORMERS_OFFLINE=1", + "HF_DATASETS_OFFLINE=1", + options.container_python, + "-m", + "sglang.bench_serving", + *base_args, + "--output-file", + container_output, + ] + rc = _run(cmd, options.detail_log, options.timeout_s) + if rc == 0: + options.output_file.parent.mkdir(parents=True, exist_ok=True) + cp = subprocess.run( + ["docker", "cp", f"{options.server_container}:{container_output}", str(options.output_file)], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + return cp.returncode + return rc + + raise SystemExit(f"unknown client mode: {options.client_mode}") + + +def wait_health(base_url: str, health_path: str = "/health", wait_s: int = 600) -> bool: + endpoint = base_url.rstrip("/") + (health_path or "/health") + for _ in range(max(1, wait_s)): + try: + with urllib.request.urlopen(endpoint, timeout=5): + return True + except Exception: + time.sleep(1) + return False + + +def split_env_assignments(value: str) -> list[str]: + """Split a space-separated `KEY=VALUE` list while preserving simple quotes.""" + return shlex.split(value) if value else [] diff --git a/src/sskj/bench/scenarios.py b/src/sskj/bench/scenarios.py new file mode 100644 index 0000000..c058718 --- /dev/null +++ b/src/sskj/bench/scenarios.py @@ -0,0 +1,71 @@ +"""Scenario generation for the unified benchmark CLI. + +The matrix format intentionally stays identical to the legacy +`matrix.json` + `generate_scenarios.py` convention. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + + +def generate_scenarios( + matrix_path: str | Path, + mode: str | None = None, + target_samples: int = 2, +) -> list[dict[str, Any]]: + """Expand a matrix.json into the low/high concurrency scenario list.""" + with open(matrix_path, encoding="utf-8") as f: + data = json.load(f) + + selected_mode = mode or data.get("mode", "Y") + matrix = data["matrix"] + concurrency_cfg = data.get("concurrency", {}) + + scenarios: list[dict[str, Any]] = [] + for isl_text in sorted(matrix.keys(), key=int): + isl = int(isl_text) + osl_map = matrix[isl_text] + cfg = concurrency_cfg.get(isl_text, {"low": 1, "high": 1}) + low = int(cfg.get("low", 1)) + high = int(cfg.get("high", low)) + concurrencies = [low, high] if low != high else [low] + if target_samples and target_samples > 0: + concurrencies = concurrencies[: max(1, target_samples)] + + for osl_text in sorted(osl_map.keys(), key=int): + mark = osl_map[osl_text] + if selected_mode == "Y" and mark != "Y": + continue + if selected_mode == "Y+P" and mark not in ("Y", "P"): + continue + if selected_mode == "all" and mark == "N": + continue + for conc in concurrencies: + scenarios.append( + { + "mark": mark, + "input_len": isl, + "output_len": int(osl_text), + "concurrency": conc, + "num_prompts": conc * 5, + } + ) + return scenarios + + +def smoke_scenario( + isl: int = 1024, + osl: int = 128, + concurrency: int = 8, + num_prompts: int = 40, +) -> dict[str, Any]: + return { + "mark": "Y", + "input_len": isl, + "output_len": osl, + "concurrency": concurrency, + "num_prompts": num_prompts, + } diff --git a/src/sskj/common/__init__.py b/src/sskj/common/__init__.py new file mode 100644 index 0000000..852bd2f --- /dev/null +++ b/src/sskj/common/__init__.py @@ -0,0 +1 @@ +"""Shared helpers for sskj packages.""" diff --git a/src/sskj/common/env.py b/src/sskj/common/env.py new file mode 100644 index 0000000..3bdc497 --- /dev/null +++ b/src/sskj/common/env.py @@ -0,0 +1,106 @@ +"""Minimal bash-style .env parser used by sskj CLI modules. + +The repository keeps deployment and platform configuration in shell `.env` +files so existing bash scripts can still source them. This parser understands +the scalar subset those files use: `KEY=value`, `export KEY=value`, quoted +values, comments, and `${VAR:-default}` substitution. +""" + +from __future__ import annotations + +import os +import re +from pathlib import Path + +_ASSIGN_RE = re.compile(r"^(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(.*)$") +_VAR_RE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)(?::-([^}]*))?\}") + + +def _strip_quotes(value: str) -> str: + value = value.strip() + if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'": + quote = value[0] + inner = value[1:-1] + if quote == '"': + inner = inner.replace('\\"', '"') + else: + inner = inner.replace("\\'", "'") + return inner + return value + + +def _strip_inline_comment(value: str) -> str: + in_single = False + in_double = False + for i, ch in enumerate(value): + if ch == "'" and not in_double: + in_single = not in_single + elif ch == '"' and not in_single: + in_double = not in_double + elif ch == "#" and not in_single and not in_double and (i == 0 or value[i - 1].isspace()): + return value[:i] + return value + + +def _expand(value: str, env: dict[str, str], depth: int = 0) -> str: + if depth > 6: + return value + + def repl(match: re.Match[str]) -> str: + name = match.group(1) + default = match.group(2) + if name in env and env[name] != "": + return env[name] + if default is not None: + return _expand(default, env, depth + 1) + return match.group(0) + + return _VAR_RE.sub(repl, value) + + +def parse_env_file( + path: str | Path, + base_env: dict[str, str] | None = None, + include_os_env: bool = True, +) -> dict[str, str]: + """Parse a shell `.env` file and return resolved scalar variables.""" + env: dict[str, str] = dict(os.environ if include_os_env else {}) + if base_env: + env.update(base_env) + path = Path(path) + if not path.exists(): + return env + + text = path.read_text(encoding="utf-8") + pending = "" + for raw_line in text.splitlines(): + line = raw_line.rstrip() + if line.endswith("\\"): + pending += line[:-1] + continue + if pending: + line = pending + line + pending = "" + line = line.strip() + if not line or line.startswith("#"): + continue + if line.startswith(("declare ", "if ", "fi", "for ", "done", "while ", "case ", "esac")): + continue + match = _ASSIGN_RE.match(line) + if not match: + continue + key, raw_value = match.group(1), match.group(2) + raw_value = _strip_inline_comment(raw_value) + raw_value = _strip_quotes(raw_value) + env[key] = _expand(raw_value, env) + + for key, value in list(env.items()): + env[key] = _expand(value, env) + return env + + +def expand_template(template: str, values: dict[str, str]) -> str: + """Expand `${VAR}` placeholders using the provided values.""" + from string import Template + + return Template(template).safe_substitute(values) diff --git a/src/sskj/deploy/__init__.py b/src/sskj/deploy/__init__.py new file mode 100644 index 0000000..1b0fc93 --- /dev/null +++ b/src/sskj/deploy/__init__.py @@ -0,0 +1 @@ +"""Model-team deployment layer for sskj.""" diff --git a/src/sskj/deploy/__main__.py b/src/sskj/deploy/__main__.py new file mode 100644 index 0000000..1199457 --- /dev/null +++ b/src/sskj/deploy/__main__.py @@ -0,0 +1,4 @@ +from sskj.deploy.cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/sskj/deploy/cli.py b/src/sskj/deploy/cli.py new file mode 100644 index 0000000..e713d67 --- /dev/null +++ b/src/sskj/deploy/cli.py @@ -0,0 +1,84 @@ +"""CLI entrypoint: python -m sskj.deploy (model-team only).""" + +from __future__ import annotations + +import argparse +from pathlib import Path + +from sskj.bench.config import repo_root +from sskj.deploy import runtime +from sskj.deploy.profile import load_profile, render_profile, validate_profile + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="python -m sskj.deploy", description="sskj deployment CLI") + parser.add_argument("--repo-root", type=Path, default=None) + sub = parser.add_subparsers(dest="command", required=True) + + for name in ("show", "render-args", "start", "stop", "status"): + p = sub.add_parser(name) + p.add_argument("--profile", required=True) + p.add_argument("--tp", type=int, default=None) + p.add_argument("--dp", type=int, default=None) + p.add_argument("--port", type=int, default=None) + p.add_argument("--model-path", default=None) + p.add_argument("--container-name", default=None) + p.add_argument("--node-rank", type=int, default=None) + p.add_argument("--dry-run", action="store_true") + p.add_argument("--log-dir", default=None) + return parser + + +def _overrides(args: argparse.Namespace) -> dict[str, str]: + values: dict[str, str] = {} + if args.tp is not None: + values["TP"] = str(args.tp) + if args.dp is not None: + values["DP"] = str(args.dp) + if args.port is not None: + values["PORT"] = str(args.port) + if args.model_path: + values["MODEL_PATH"] = args.model_path + if args.container_name: + values["CONTAINER_NAME"] = args.container_name + if args.node_rank is not None: + values["NODE_RANK"] = str(args.node_rank) + return values + + +def _cmd(args: argparse.Namespace, root: Path) -> int: + profile = render_profile(load_profile(root, args.profile), _overrides(args)) + validate_profile(profile) + if args.command == "show": + keys = [ + "PLATFORM", + "MODEL_NAME", + "ENGINE", + "RUNTIME", + "DOCKER_IMAGE", + "CONTAINER_NAME", + "MODEL_PATH", + "SERVED_MODEL_NAME", + "PORT", + "HEALTH_PATH", + "LAUNCH_ARGS", + ] + for key in keys: + print(f"{key}={profile.get(key, '')}") + return 0 + if args.command == "render-args": + print(profile.get("LAUNCH_ARGS", "")) + return 0 + if args.command == "start": + return runtime.start(profile, root, dry_run=args.dry_run, log_dir=args.log_dir) + if args.command == "stop": + return runtime.stop(profile, root) + if args.command == "status": + return runtime.status(profile, root) + return 2 + + +def main(argv: list[str] | None = None) -> int: + args = _build_parser().parse_args(argv) + root = args.repo_root or repo_root() + return _cmd(args, root) diff --git a/src/sskj/deploy/profile.py b/src/sskj/deploy/profile.py new file mode 100644 index 0000000..077a08a --- /dev/null +++ b/src/sskj/deploy/profile.py @@ -0,0 +1,75 @@ +"""Deployment profile loading and rendering.""" + +from __future__ import annotations + +from pathlib import Path + +from sskj.common.env import expand_template, parse_env_file + +REQUIRED_FIELDS = [ + "PLATFORM", + "MODEL_NAME", + "ENGINE", + "MODEL_PATH", + "SERVED_MODEL_NAME", + "PORT", + "HEALTH_PATH", +] + + +def resolve_profile(root: Path, profile: str | Path) -> Path: + path = Path(profile) + if path.exists(): + return path + candidates = [ + root / "deploy" / "profiles" / f"{profile}.env", + root / "deploy" / "profiles" / f"{profile}", + ] + for candidate in candidates: + if candidate.exists(): + return candidate + raise SystemExit(f"deploy profile not found: {profile}") + + +def load_profile(root: Path, profile: str | Path) -> dict[str, str]: + path = resolve_profile(root, profile) + return parse_env_file(path, base_env={"ROOT_DIR": str(root)}) + + +def render_profile(profile: dict[str, str], overrides: dict[str, str] | None = None) -> dict[str, str]: + values = dict(profile) + if overrides: + values.update(overrides) + values.setdefault("TP", "8") + values.setdefault("DP", "1") + + template_keys = [ + "CONTAINER_NAME", + "DEVICE_VARS", + "ENGINE_ENV", + "LAUNCH_ARGS", + "DP_FLAG", + "DOCKER_FLAGS", + "VOLUMES", + "PATCH_MOUNTS", + "BOOTSTRAP", + "MODEL_PATH", + "PORT", + ] + rendered = dict(values) + for key in template_keys: + if key in rendered: + rendered[key] = expand_template(str(rendered[key]), values) + + dp = int(values["DP"]) + if dp > 1 and rendered.get("DP_FLAG"): + rendered["LAUNCH_ARGS"] = f"{rendered.get('LAUNCH_ARGS', '')} {rendered['DP_FLAG']}".strip() + if not rendered.get("RUNTIME"): + rendered["RUNTIME"] = "docker" if rendered.get("DOCKER_IMAGE") else "native" + return rendered + + +def validate_profile(profile: dict[str, str]) -> None: + missing = [key for key in REQUIRED_FIELDS if not profile.get(key)] + if missing: + raise SystemExit(f"deploy profile missing required fields: {', '.join(missing)}") diff --git a/src/sskj/deploy/runtime.py b/src/sskj/deploy/runtime.py new file mode 100644 index 0000000..2e04488 --- /dev/null +++ b/src/sskj/deploy/runtime.py @@ -0,0 +1,147 @@ +"""Docker/native server lifecycle for deployment profiles.""" + +from __future__ import annotations + +import base64 +import shlex +import subprocess +from pathlib import Path + +from sskj.bench.runner import split_env_assignments, wait_health + + +def _log_dir(root: Path, log_dir: str | None) -> Path: + path = Path(log_dir) if log_dir else root / "deploy" / "logs" + path.mkdir(parents=True, exist_ok=True) + return path + + +def _health_url(profile: dict[str, str]) -> str: + host = profile.get("HEALTH_HOST", "127.0.0.1") + return f"http://{host}:{profile['PORT']}" + + +def start(profile: dict[str, str], root: Path, dry_run: bool = False, log_dir: str | None = None) -> int: + validate_runtime(profile) + logs = _log_dir(root, log_dir) + runtime = profile["RUNTIME"] + container = profile.get("CONTAINER_NAME", "") + health_wait = int(profile.get("HEALTH_WAIT_S", "600") or 600) + + if runtime == "docker": + if not profile.get("DOCKER_IMAGE"): + raise SystemExit("docker runtime requires DOCKER_IMAGE") + if not container: + raise SystemExit("docker runtime requires CONTAINER_NAME") + if not dry_run: + subprocess.run(["docker", "rm", "-f", container], capture_output=True, check=False) + + cmd = ["docker", "run", "-d", "--name", container] + if profile.get("DOCKER_FLAGS"): + cmd += shlex.split(profile["DOCKER_FLAGS"]) + if profile.get("NETWORK_MODE") == "bridge" and profile.get("PORT_MAP") == "1": + cmd += ["-p", f"{profile['PORT']}:{profile['PORT']}"] + for volume in split_env_assignments(profile.get("VOLUMES", "") + " " + profile.get("PATCH_MOUNTS", "")): + cmd += ["-v", volume] + for assignment in split_env_assignments( + profile.get("DEVICE_VARS", "") + " " + profile.get("ENGINE_ENV", "") + ): + cmd += ["-e", assignment] + cmd.append(profile["DOCKER_IMAGE"]) + + bootstrap = profile.get("BOOTSTRAP", "") + if bootstrap: + encoded = base64.b64encode(bootstrap.encode("utf-8")).decode("ascii") + cmd += ["bash", "-c", f"echo {encoded} | base64 -d | bash"] + else: + python_bin = profile.get("CONTAINER_PYTHON", "python") + cmd += ["bash", "-c", f"{python_bin} -m sglang.launch_server {profile['LAUNCH_ARGS']}"] + + (logs / f"{container}.cmd.txt").write_text(shlex.join(cmd) + "\n", encoding="utf-8") + if dry_run: + print(shlex.join(cmd)) + return 0 + print(f"starting container {container} ...") + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + print(result.stdout, result.stderr) + return result.returncode + if wait_health(_health_url(profile), profile.get("HEALTH_PATH", "/health"), health_wait): + print(f"container {container} is healthy") + return 0 + subprocess.run(["docker", "logs", "--tail", "100", container], check=False) + print(f"ERROR: container {container} failed health check") + return 1 + + # native runtime + python_bin = profile.get("PYTHON_BIN") or profile.get("CONTAINER_PYTHON") or "python3" + launch_args = shlex.split(profile.get("LAUNCH_ARGS", "")) + cmd = [python_bin, "-m", "sglang.launch_server", *launch_args] + log_path = logs / f"{profile.get('MODEL_NAME', 'server')}.log" + pid_file = logs / f"{profile.get('MODEL_NAME', 'server')}.pid" + if dry_run: + print(shlex.join(cmd)) + return 0 + with open(log_path, "wb") as f: + proc = subprocess.Popen(cmd, stdout=f, stderr=subprocess.STDOUT) + pid_file.write_text(str(proc.pid), encoding="utf-8") + print(f"started native server pid={proc.pid} log={log_path}") + if wait_health(_health_url(profile), profile.get("HEALTH_PATH", "/health"), health_wait): + print("native server is healthy") + return 0 + print(f"ERROR: native server failed health check; see {log_path}") + return 1 + + +def stop(profile: dict[str, str], root: Path) -> int: + validate_runtime(profile) + if profile["RUNTIME"] == "docker": + container = profile.get("CONTAINER_NAME") + if not container: + raise SystemExit("docker runtime requires CONTAINER_NAME") + subprocess.run(["docker", "rm", "-f", container], capture_output=True, check=False) + print(f"container {container} removed") + return 0 + logs = _log_dir(root, None) + pid_file = logs / f"{profile.get('MODEL_NAME', 'server')}.pid" + if pid_file.exists(): + pid = pid_file.read_text(encoding="utf-8").strip() + if pid: + subprocess.run(["kill", pid], capture_output=True, check=False) + print(f"killed native server pid={pid}") + pid_file.unlink(missing_ok=True) + return 0 + + +def status(profile: dict[str, str], root: Path) -> int: + validate_runtime(profile) + if profile["RUNTIME"] == "docker": + container = profile.get("CONTAINER_NAME", "") + result = subprocess.run( + ["docker", "inspect", "-f", "{{.State.Status}}", container], + capture_output=True, + text=True, + check=False, + ) + print(result.stdout.strip() or "missing") + return 0 if result.returncode == 0 else 1 + logs = _log_dir(root, None) + pid_file = logs / f"{profile.get('MODEL_NAME', 'server')}.pid" + if not pid_file.exists(): + print("stopped") + return 1 + pid = pid_file.read_text(encoding="utf-8").strip() + result = subprocess.run(["kill", "-0", pid], capture_output=True, check=False) + print("running" if result.returncode == 0 else "stopped") + return 0 if result.returncode == 0 else 1 + + +def validate_runtime(profile: dict[str, str]) -> None: + runtime = profile.get("RUNTIME", "docker" if profile.get("DOCKER_IMAGE") else "native") + if runtime not in ("docker", "native"): + raise SystemExit(f"unsupported runtime: {runtime}") + profile["RUNTIME"] = runtime + if runtime == "docker" and not profile.get("DOCKER_IMAGE"): + raise SystemExit("docker runtime requires DOCKER_IMAGE") + if runtime == "native" and not profile.get("PYTHON_BIN") and not profile.get("CONTAINER_PYTHON"): + raise SystemExit("native runtime requires PYTHON_BIN or CONTAINER_PYTHON")