[Tools] Recipes sweep workflow (#57307)

Signed-off-by: louie-tsai <louie.tsai@intel.com>
This commit is contained in:
Louie Tsai
2026-10-09 19:44:09 +08:00
committed by GitHub
parent 1c47ebfc57
commit 196483e25b
17 changed files with 4737 additions and 581 deletions
+51 -38
View File
@@ -19,7 +19,7 @@ flowchart LR
```
The recipe is the baseline. Hardware and workload information can optionally
refine the initial configuration. Sweep tuning is an optional validation step.
refine the initial configuration. Sweep tuning is optional.
## Getting Started
@@ -54,7 +54,9 @@ For `xeon6`, the script enables hardware detection automatically. Generated
`config.yml` and `env.sh` files remain in the directory where the script was
invoked.
## 1. vLLM Recipes Only
## Detailed Deployment Workflow
### 1. vLLM Recipes Only
Use the converter directly when the recipe already contains the deployment
settings you need. This path requires only PyYAML; the vLLM Python package is
@@ -64,15 +66,7 @@ not required unless optional runtime tuning or sweep generation is requested.
pip install pyyaml
```
Choose whichever recipe-selection method fits the workflow:
**Interactive discovery** — search models, then choose hardware and strategy:
```bash
python3 tools/recipes/recipe_json_to_vllm_config.py
```
**Non-interactive discovery** — provide model and hardware and use the
For non-interactive discovery, provide the model and hardware and use the
Recipes-recommended strategy:
```bash
@@ -81,7 +75,17 @@ python3 tools/recipes/recipe_json_to_vllm_config.py \
--hardware xeon6
```
**Direct JSON input** — use a Recipes JSON URL or a local JSON file:
#### 1.1 Interactive Discovery
Search models, then choose hardware and strategy interactively:
```bash
python3 tools/recipes/recipe_json_to_vllm_config.py
```
#### 1.2 Direct JSON Input
Use a Recipes JSON URL or a local JSON file:
```bash
python3 tools/recipes/recipe_json_to_vllm_config.py \
@@ -90,15 +94,27 @@ python3 tools/recipes/recipe_json_to_vllm_config.py \
python3 tools/recipes/recipe_json_to_vllm_config.py recipe.json
```
All paths generate `config.yml` and `env.sh`. See
[REFERENCE.md](REFERENCE.md) for recipe discovery, strategy selection, direct
JSON input, custom output files, and deployment scope.
#### 1.3 Test a Preview Recipe API
## 2. Hardware Information (Optional)
A Recipes pull request can expose the same JSON API through its Vercel preview.
Use `--api-base` to validate it before the recipe is available in production:
Add `--detect-hardware` when the target host's effective CPU/NUMA/memory
resources should refine deployment-sensitive values such as
`tensor-parallel-size` and `gpu-memory-utilization`.
```bash
PREVIEW=https://vllm-recipes-git-fork-intel-ai-tce-dockerin-f4c148-inferact-inc.vercel.app
python3 tools/recipes/recipe_json_to_vllm_config.py \
--api-base "$PREVIEW" \
--model meta-llama/Llama-3.2-1B-Instruct \
--hardware xeon6
```
See [REFERENCE.md](REFERENCE.md) for recipe discovery, strategy selection,
custom output files, and deployment scope.
### 2. Hardware Information (Optional)
Add `--detect-hardware` when using the converter directly and the target host's
effective CPU/NUMA/memory resources should refine deployment-sensitive values:
```bash
python3 tools/recipes/recipe_json_to_vllm_config.py \
@@ -107,14 +123,13 @@ python3 tools/recipes/recipe_json_to_vllm_config.py \
--detect-hardware
```
Hardware detection is optional and uses vLLM CPU resource utilities only when
requested. See [RUNTIME_TUNING.md](RUNTIME_TUNING.md#hardware-information).
`serve_with_recipe.sh` enables this automatically for `xeon6`. See
[RUNTIME_TUNING.md](RUNTIME_TUNING.md#hardware-information) for details.
## 3. Workload Information (Optional)
### 3. Workload Information (Optional)
Workload hints can refine scheduler settings for one initial deployment
suggestion. Inputs include token lengths, concurrency, and optional latency or
capacity objectives.
Add workload hints when token lengths, concurrency, or latency objectives should
seed optional benchmark tuning:
```bash
python3 tools/recipes/recipe_json_to_vllm_config.py \
@@ -127,17 +142,13 @@ python3 tools/recipes/recipe_json_to_vllm_config.py \
--tpot-sla-ms 100
```
Hardware detection and workload information are independent optional inputs;
they can also be supplied together. See
[RUNTIME_TUNING.md](RUNTIME_TUNING.md#workload-information) for the supported
inputs and how runtime parameters are calculated.
See [RUNTIME_TUNING.md](RUNTIME_TUNING.md#workload-information) for supported
inputs and runtime calculations.
## 4. Sweep Tuning (Optional)
### 4. Sweep Tuning (Optional)
Use `--generate-sweep` when the initial scheduler suggestion should be validated
with `vllm bench sweep serve`. The sweep benchmarks nearby scheduler values and
`recommend.py` produces one measured `recommended-config.yml` plus the
selection evidence in `recommendation.json`.
Use `--generate-full-sweep` for benchmark-backed tuning of TP/DP, concurrency,
and scheduler settings:
```bash
python3 tools/recipes/recipe_json_to_vllm_config.py \
@@ -147,13 +158,15 @@ python3 tools/recipes/recipe_json_to_vllm_config.py \
--input-tokens 128 \
--output-tokens 128 \
--concurrency 32 \
--generate-sweep
--generate-full-sweep
```
See [SWEEP_TUNING.md](SWEEP_TUNING.md) for the benchmark, recommendation, and
vLLM CPU Docker-shell workflow.
See [SWEEP_TUNING.md](SWEEP_TUNING.md) for targeted sweep modes, NUMA binding,
failure handling, recommendations, reporting, and visualization.
## Start vLLM
### 5. Start vLLM
When using the converter directly:
```bash
source env.sh
+59 -23
View File
@@ -64,12 +64,10 @@ python3 tools/recipes/recipe_json_to_vllm_config.py \
--tpot-sla-ms 100
```
`--output-tokens` is used to estimate steady-state request turnover for
`max-num-batched-tokens`. When both `--target-qps` and `--tpot-sla-ms` are
available, they are also used to estimate prompt arrival pressure per scheduler
step. `--ttft-sla-ms` is collected but is not yet converted directly into a
batch-size formula because the relationship between TTFT and scheduler budget is
model- and hardware-dependent.
Scheduler calculation is opt-in with `--tune-scheduler`; workload hints do
not rewrite `max-num-seqs` or `max-num-batched-tokens` by default. The detailed
scheduler and staged-sweep behavior is documented in
[SWEEP_TUNING.md](SWEEP_TUNING.md).
## Deployment-Time Parameters
@@ -81,11 +79,30 @@ on a validation environment that differs from the user's target deployment.
| --- | --- | --- | --- |
| `tensor-parallel-size` | Available topology and model head counts constrain TP. | Hardware topology + model configuration | Use the largest head-compatible power-of-two TP within the effective NUMA-node count, for DP=1 and PP=1. |
| `gpu-memory-utilization` | Available memory can differ by machine size, container limits, and other memory use. The vLLM option name is also used by the CPU backend. | Hardware memory + recipe baseline | Calculate a conservative fraction from the most constrained NUMA node. |
| `max-num-seqs` | The useful scheduler concurrency depends on the number of requests expected to be active at the same time. | Workload concurrency | Set `max-num-seqs` to `--concurrency` when supplied. |
| `max-num-batched-tokens` | Each scheduler iteration must share its token budget between active decodes and incoming prefills. | Input/output token shape, concurrency, and optional QPS/TPOT | Calculate decode budget + expected prefill demand, with vLLM scheduler constraints as floors. |
| `max-num-seqs` | The useful scheduler concurrency depends on the number of requests expected to be active on each DP replica. | Workload concurrency + DP | Keep the recipe/vLLM value in `config.yml`; use `ceil(--concurrency / data-parallel-size)` as an explicit sweep seed. |
| `max-num-batched-tokens` | Each scheduler iteration must share its token budget between active decodes and incoming prefills. | Input/output token shape, concurrency, and optional QPS/TPOT | Keep the recipe/vLLM value in `config.yml`; use decode budget + expected prefill demand as an explicit sweep seed. |
| `data-parallel-size` | The required replica count depends on the requested throughput and the capacity of one replica. | Capacity target | Keep the recipe value today because per-replica SLA capacity is not known. |
## How Each Runtime Parameter Is Calculated
For `max-num-seqs` and `max-num-batched-tokens`, workload formulas now create
benchmark-only seeds instead of overriding the directly deployable
`config.yml`. The optional benchmark sweep measures those explicit seeds
against vLLM's normal serving defaults. Default-reference runs leave the
selected scheduler option unset at engine-configuration time so vLLM can
resolve its platform-, world-size-, model-, and usage-context-aware default
rather than assuming a fixed numeric value. See
[SWEEP_TUNING.md](SWEEP_TUNING.md).
This serving-default comparison is distinct from the `SchedulerConfig` fallback
constants used inside the initial tuning heuristics below.
## How Initial Runtime Tuning Seeds Are Calculated
The hardware formulas below produce directly deployable starting values. The
scheduler formulas produce explicit benchmark seeds and are not claimed to be
universal performance optima. This separation avoids baking workload guesses
into `config.yml` before measurements exist. The Tune All workflow benchmarks
TP/DP first, then concurrency, and finally scheduler settings before writing a
final recommendation. See [SWEEP_TUNING.md](SWEEP_TUNING.md).
### `tensor-parallel-size`
@@ -147,14 +164,20 @@ candidate = min(recipe value or 0.80, safe_fraction)
### `max-num-seqs`
When `--concurrency` is supplied, it directly represents the requested maximum
number of simultaneously active requests:
When `--concurrency` is supplied, it represents global client concurrency. Each
DP replica needs capacity for its expected share of those requests:
```text
max-num-seqs = concurrency
per_replica_concurrency = ceil(concurrency / data_parallel_size)
max-num-seqs = per_replica_concurrency
```
If concurrency is not supplied, the converter does not override this parameter.
For DP=1 this reduces to `max-num-seqs = concurrency`. This value seeds the
scheduler benchmark; it is not added to `config.yml`. If concurrency is not
supplied, no seed is calculated. Benchmark results show
that setting `max-num-seqs` below per-replica concurrency can create request
queueing and severe TTFT degradation; scheduler sweeps therefore do not use
below-concurrency candidates.
### `max-num-batched-tokens`
@@ -165,7 +188,7 @@ First, the policy determines the active sequence count:
```text
active_sequences =
--concurrency
ceil(--concurrency / data_parallel_size)
or recipe max-num-seqs
or vLLM default max_num_seqs (128)
```
@@ -181,17 +204,19 @@ The policy then estimates how many new prompts need prefill work per scheduler
step:
```text
per_replica_qps = target_qps / data_parallel_size
prefills_per_step = max(
1,
active_sequences / output_tokens,
target_qps * tpot_sla_ms / 1000
per_replica_qps * tpot_sla_ms / 1000
)
prefills_per_step = min(active_sequences, prefills_per_step)
prefill_budget = ceil(input_tokens * prefills_per_step)
```
The final scheduler budget is:
The final explicit scheduler sweep seed is:
```text
max-num-batched-tokens = max(
@@ -201,7 +226,7 @@ max-num-batched-tokens = max(
)
```
For example, with 128 input tokens, 128 output tokens, and concurrency 32:
For example, with 128 input tokens, 128 output tokens, concurrency 32, and DP=1:
```text
decode_budget = 32
@@ -210,6 +235,12 @@ prefill_budget = 128
candidate = max(2048, 32, 32 + 128) = 2048
```
The calculated token budget is a scheduler sweep seed and is not added to
`config.yml`. If P99 TTFT still
misses its objective while P99 TPOT has substantial headroom, Tune All should
reduce workload concurrency or select a different TP/DP layout rather than
reducing `max-num-seqs` below per-replica concurrency.
If chunked prefill is explicitly disabled and `max-model-len` is available, the
policy also ensures:
@@ -226,8 +257,10 @@ also needs measured per-replica throughput that still satisfies TTFT/TPOT:
DP = ceil(target_qps / qps_per_replica_at_SLO)
```
Because the converter does not have that measured capacity yet, it intentionally
keeps the recipe DP value rather than guessing.
Because the converter does not have that measured capacity yet, its initial
configuration intentionally keeps the recipe DP value rather than guessing.
Tune All subsequently benchmarks supported TP/DP layouts and replaces this seed
with a measured layout before concurrency and scheduler tuning.
## Precedence
@@ -235,8 +268,11 @@ keeps the recipe DP value rather than guessing.
vLLM defaults
-> vLLM Recipes baseline
-> hardware refinement (optional)
-> workload / SLO refinement (optional)
-> config.yml + env.sh
-> config.yml + env.sh
workload / SLO hints (optional)
-> explicit scheduler sweep seed
-> benchmark-backed recommended-config.yml
```
## Runtime-Tuning Hardware Scope
@@ -312,10 +348,10 @@ with [SWEEP_TUNING.md](SWEEP_TUNING.md).
## Implementation
- `hardware_detection.py` collects effective CPU/NUMA/memory information.
- `runtime_tuning.py` owns hardware-policy selection and independent parameter
- `sweep/runtime_tuning.py` owns hardware-policy selection and independent parameter
tuning functions.
- `recipe_json_to_vllm_config.py` resolves the recipe, collects optional inputs,
applies the selected policy, and generates `config.yml` and `env.sh`.
For benchmark-backed validation of the initial scheduler suggestion, see
For benchmark-backed validation of the explicit scheduler seed, see
[SWEEP_TUNING.md](SWEEP_TUNING.md).
+616 -76
View File
@@ -1,95 +1,130 @@
# Sweep Tuning
Sweep tuning is optional. The converter always creates one initial `config.yml`
first. Users can deploy it immediately or benchmark nearby scheduler settings
and generate one measured recommendation.
Sweep tuning is optional. The converter always creates an initial `config.yml`
that can be deployed directly. Scheduler calculation is disabled by default:
TP/DP and concurrency stages remove explicit `max-num-seqs` and
`max-num-batched-tokens` so vLLM resolves its normal scheduler defaults.
## Generate
```bash
python3 tools/recipes/recipe_json_to_vllm_config.py \
--model meta-llama/Llama-3.1-8B-Instruct \
--hardware xeon6 \
--detect-hardware \
--input-tokens 128 \
--output-tokens 128 \
--concurrency 32 \
--ttft-sla-ms 3000 \
--tpot-sla-ms 100 \
--generate-sweep
```
The optional package is:
Use `--tune-scheduler` only when you explicitly want workload-derived scheduler
values before the dedicated scheduler stage. For benchmark-backed tuning, the
recommended workflow is **Tune All**, which measures the serving stack in
dependency order:
```text
sweep/
├── serve_params.json
├── bench_params.json
├── run_sweep.sh
├── recommend.py
└── SWEEP.md
TP/DP -> concurrency -> scheduler
```
The sweep varies only `max-num-seqs` and `max-num-batched-tokens`. It uses an
eight-point directed design: a batch-budget curve at the initial sequence count
plus lower/higher batch interactions at three-quarters and one-half of that
count. This gives broader coverage than a one-parameter-at-a-time sweep without
the cost of a full Cartesian grid.
The examples below assume tuning is run inside the target vLLM CPU container so
hardware detection sees the same CPU, NUMA, memory, and cgroup limits as the
deployment.
## Run and Recommend
## 1. Start the vLLM CPU Docker Shell
From the vLLM source tree:
```bash
sweep/run_sweep.sh --dry-run
sweep/run_sweep.sh
sweep/recommend.py
mkdir -p recipe-output
docker run --rm -it \
--entrypoint bash \
--security-opt seccomp=unconfined \
--cap-add SYS_NICE \
--shm-size=4g \
-p 8000:8000 \
-v "$PWD/tools/recipes:/recipes:ro" \
-v "$PWD/recipe-output:/output" \
-v "$HOME/.cache/huggingface:/root/.cache/huggingface" \
-w /output \
vllm/vllm-openai-cpu:latest-x86_64
```
Resume an interrupted sweep with:
`/recipes` is mounted read-only from the source tree and `/output` is writable.
Run the converter, generated sweep scripts, and hardware detection inside this
container.
```bash
sweep/run_sweep.sh --resume
For additional container details, see
[RUNTIME_TUNING.md](RUNTIME_TUNING.md#vllm-cpu-docker-shell).
## 2. Recommended: Tune All
Use `--generate-full-sweep` by default when you want benchmark-backed runtime
tuning. It measures TP/DP first, then the highest useful concurrency, then
scheduler parameters using the selected layout and concurrency.
```mermaid
flowchart TD
I[["Input<br/>Recipe + detected hardware<br/>fixed input/output workload shape"]]
A1("1. Tune TP/DP<br/>tensor-parallel-size + data-parallel-size")
O1[/"parallel-layout-config.yml<br/>selected TP/DP"/]
A2("2. Tune concurrency<br/>max_concurrency")
O2[/"concurrency-recommendation.json<br/>recommended max_concurrency"/]
A3("3. Tune scheduler<br/>max-num-seqs + max-num-batched-tokens")
O3[/"recommended-config.yml<br/>final vLLM server config"/]
I --> A1
A1 --> O1
O1 --> A2
A2 --> O2
O2 --> A3
A3 --> O3
```
`recommend.py` writes:
### Workflow options
The table is ordered by the tuning sequence. **Tune All** is the recommended
default; the other modes are useful when only one part of the serving stack
needs to be re-measured.
| Goal | Option | Tuning sequence |
| --- | --- | --- |
| **Tune all (recommended)** | `--generate-full-sweep` | **1. TP/DP -> 2. concurrency -> 3. scheduler** |
| Tune TP/DP only | `--generate-parallel-layout-sweep` | 1. TP/DP |
| Tune concurrency only | `--generate-concurrency-sweep` | 2. concurrency |
| Tune scheduler only | `--generate-scheduler-sweep` (`--generate-sweep` alias) | 3. scheduler |
All workflows keep one fixed input/output workload shape and require
`--input-tokens`, `--output-tokens`, and `--concurrency`.
For **Tune All**, the supplied `--concurrency` is the representative load used
for the initial TP/DP comparison and to size the benchmark request set. The
concurrency stage then measures the final SLA-feasible `max_concurrency`.
### Scheduler defaults and `--tune-scheduler`
By default, `--concurrency`, token lengths, and SLA values are workload hints;
they do not cause the converter to calculate `max-num-seqs` or
`max-num-batched-tokens`. This keeps scheduler capacity independent from the
seed concurrency while TP/DP and maximum concurrency are being measured.
The default Tune All policy is:
```text
sweep/recommended-config.yml
sweep/recommendation.json
Stage 1: TP/DP -> vLLM scheduler defaults
Stage 2: concurrency -> vLLM scheduler defaults
Stage 3: scheduler -> calculate and benchmark scheduler candidates
```
With TTFT/TPOT objectives, the benchmark uses vLLM `--goodput`. The recommender
calculates duration-weighted combined compliance across repeated runs and
requires both median P99 latency objectives and the minimum compliance ratio
(default `0.99`). Among eligible configurations it selects the highest mean
output-token throughput. Use `recommend.py --minimum-compliance VALUE` to
change the compliance requirement.
If no configuration is eligible, `recommend.py` records the highest-goodput
configuration as `best_effort` in `recommendation.json`, does not write a
deployable `recommended-config.yml`, and exits with status 2. Without latency
objectives it selects highest mean output-token throughput. Failed benchmark
configurations are excluded in either mode.
When an eligible configuration exists, `recommended-config.yml` changes only
the two swept scheduler parameters. `recommendation.json` records mean, median,
and worst-run P99 values, combined compliance, and the measured evidence for
every candidate.
Deploy:
For a directly deployable heuristic configuration, or to intentionally carry
workload-derived scheduler values into the earlier sweep stages, opt in with:
```bash
source env.sh
vllm serve --config sweep/recommended-config.yml
--tune-scheduler
```
## vLLM CPU Docker Shell
With that flag, the existing workload formulas are enabled. In particular,
`max-num-seqs` uses the per-replica workload concurrency and
`max-num-batched-tokens` uses the estimated decode and prefill pressure.
Use the common CPU container setup in
[RUNTIME_TUNING.md](RUNTIME_TUNING.md#vllm-cpu-docker-shell). It ensures
hardware detection runs against the same effective CPU, NUMA, memory, and
cgroup limits used by the deployment.
A scheduler-only sweep is also an explicit opt-in to scheduler calculation: it
uses the workload formulas as benchmark seeds and compares them with vLLM
defaults.
Inside that container, generate the initial suggestion plus sweep package:
### Generate the Tune All package
Inside the container:
```bash
python3 /recipes/recipe_json_to_vllm_config.py \
@@ -103,35 +138,540 @@ python3 /recipes/recipe_json_to_vllm_config.py \
--tpot-sla-ms 100 \
--config-out /output/config.yml \
--env-out /output/env.sh \
--generate-sweep \
--generate-full-sweep \
--sweep-out-dir /output/sweep
```
The initial configuration can be deployed directly:
The initial configuration remains directly deployable:
```bash
source /output/env.sh
vllm serve --config /output/config.yml
```
Stop that manually started server before running the sweep, because
`run_sweep.sh` starts and stops its own vLLM servers.
Before measurements, `config.yml` retains the recipe/vLLM scheduler defaults.
The sweep package separately carries explicit workload-derived values for
`max-num-seqs` and `max-num-batched-tokens`, so the benchmark can compare both
policies without making the workload heuristic the deployment default.
Stop a manually started server before running the sweep because the generated
sweep scripts start and stop their own vLLM servers.
Run all tuning stages:
```bash
/output/sweep/run_sweep.sh --dry-run
/output/sweep/run_sweep.sh
/output/sweep/recommend.py
/output/sweep/run_full_sweep.sh
```
Inspect:
The end-to-end flow produces intermediate recommendations for TP/DP and
concurrency and finishes with:
```text
/output/sweep/recommended-config.yml
/output/sweep/recommendation.json
/output/sweep/sweep-report.html
```
Inspect the final recommendation:
```bash
cat /output/sweep/recommendation.json
```
Deploy the measured result:
## 3. Tune All Stage Details
### Stage 1: Tune TP/DP
The parallel-layout stage measures NUMA-aware combinations of:
- `tensor-parallel-size`
- `data-parallel-size`
Primary candidates use all effective NUMA nodes:
```text
tensor-parallel-size * data-parallel-size = effective NUMA nodes
```
The effective NUMA-node count comes from hardware detection; it is not fixed by
the sweep generator. The table below shows representative detected topologies.
TP is restricted to the supported values `1`, `2`, `4`, and `8`. The sweep also
includes the largest supported TP size that does not exceed the effective
NUMA-node count, even if that layout leaves some NUMA nodes idle.
| Effective NUMA nodes | Generated layouts |
| ---: | --- |
| 2 | `TP=2, DP=1`; `TP=1, DP=2` |
| 4 | `TP=4, DP=1`; `TP=2, DP=2`; `TP=1, DP=4` |
| 6 | `TP=4, DP=1` (4 of 6 nodes); `TP=2, DP=3`; `TP=1, DP=6` |
| 8 | `TP=8, DP=1`; `TP=4, DP=2`; `TP=2, DP=4`; `TP=1, DP=8` |
The generator then applies model- and runtime-aware safety policy before any
server is started:
1. For an MoE model, if the CPU communicator does not override the variable-size
`all_gatherv` and `reduce_scatterv` collectives used by MoE data parallelism,
all generated candidates use `DP=1`. Every supported TP value is retained,
so a 6-NUMA-node host tests `TP=4`, `TP=2`, and `TP=1` rather than only `TP=4`.
2. When MoE dimensions are known and expert parallelism is disabled, the fused
MoE path flattens TP, DP, and PCP. A candidate is valid only when
`moe_intermediate_size % (TP * DP * PCP) == 0`.
3. Failures not covered by a preflight rule remain isolated by
`--continue-on-error` when the installed CLI supports it, with bounded
`--resume` retries as the compatibility fallback.
Skipped candidates and reasons are saved in
`sweep/parallel_layout_skips.json`. If model metadata cannot be loaded, the
generator keeps the generic layouts and relies on the third policy.
By default, parallel-layout candidates change only TP and DP. The generated
benchmark config removes explicit `max-num-seqs` and
`max-num-batched-tokens`, so every layout uses vLLM-resolved scheduler
defaults. This prevents a scheduler value derived from the seed concurrency
from becoming part of the TP/DP selection.
With `--tune-scheduler`, the workload-derived per-replica scheduler baseline is
restored for this stage.
The stage writes:
```text
sweep/parallel-layout-config.yml
sweep/parallel-layout-recommendation.json
```
The selected TP/DP layout becomes the fixed server layout for Stage 2.
### Temporary TP/DP NUMA-binding workaround
Xeon TP/DP and Tune All sweeps currently enable a temporary explicit CPU-binding
workaround by default:
```text
--tp-dp-numa-bind-workaround
```
Hardware detection builds one `VLLM_CPU_OMP_THREADS_BIND` CPU list per effective
NUMA node. On x86 it selects one logical CPU per physical core, matching vLLM's
auto-binding SMT policy, and excludes one physical core from each NUMA node for
non-OMP work. The resulting value is written to `env.sh`, for example:
```bash
export VLLM_CPU_OMP_THREADS_BIND='48-62|64-78|80-94|96-110'
```
The exact CPU IDs depend on the effective container/cgroup cpuset and NUMA
topology. Discontinuous CPU IDs are preserved as comma-separated ranges.
`VLLM_CPU_NUM_OF_RESERVED_CPU` is not used by this workaround because vLLM only
applies its reserved-CPU logic in automatic binding mode. With an explicit
`VLLM_CPU_OMP_THREADS_BIND`, reserved cores must already be omitted from the
generated lists.
After the vLLM CPU DP NUMA-binding issue is fixed, disable the workaround
without removing the implementation:
```bash
--no-tp-dp-numa-bind-workaround
```
An explicit non-`auto` `VLLM_CPU_OMP_THREADS_BIND` supplied by the recipe is
preserved.
### Stage 2: Tune Concurrency
The concurrency stage keeps the selected TP/DP layout fixed while finding the
largest useful client concurrency. By default the generated benchmark config
removes explicit `max-num-seqs` and `max-num-batched-tokens`, so the seed
concurrency cannot become an artificial scheduler admission limit.
With `--tune-scheduler`, explicit scheduler values are preserved as an opt-in
behavior.
When TTFT and/or TPOT objectives are supplied, Recipes uses an **adaptive SLA
boundary search** instead of uniformly sampling the entire workload range.
The supplied `--concurrency` is the seed. The controller grows upward until it
finds the first SLA failure (or shrinks downward when the seed fails), then
binary-searches the integer PASS/FAIL bracket. One vLLM server remains running
throughout this refinement, so midpoint probes do not reload the model.
For example:
```text
32 PASS, 64 FAIL
-> 48
-> 40 or 56
-> ...
-> adjacent PASS / FAIL boundary
```
A point is SLA-feasible only when all of the following hold:
1. Every measured run completes without failed requests.
2. Median P99 TTFT is at or below the supplied TTFT objective.
3. Median P99 TPOT is at or below the supplied TPOT objective.
4. Duration-weighted combined request compliance is at least the configured
minimum (`0.99`, or 99%, by default).
The combined-compliance check is intentionally stricter than looking at the
TTFT and TPOT P99 columns independently. `vllm bench serve --goodput` counts
requests that satisfy all supplied latency objectives, and the recommender
estimates the combined compliant-request fraction across repeated runs as:
```text
sum(request_goodput * duration) / sum(completed_requests)
```
Therefore a candidate can have P99 TTFT below 3000 ms and P99 TPOT below
100 ms but still be SLA-ineligible if fewer than 99% of requests satisfy both
objectives together. For example, a reported combined compliance of `98.96%`
fails the default `99.00%` requirement even when both displayed P99 metrics
pass. The HTML report shows the combined compliance, required threshold, and
the specific SLA failure reason for each candidate.
Adaptive concurrency uses a `1800` second server-readiness timeout by default,
which is intentionally longer than the generic benchmark helper default for
CPU model loading and initialization. `run_full_sweep.sh
--server-ready-timeout VALUE` also exports the selected timeout explicitly to
the generated concurrency runner, in addition to forwarding the CLI option.
This avoids falling back to a shorter adaptive-helper timeout when the
concurrency stage is invoked through the full staged workflow.
The default growth factor is `2` and the default hard cap is `1000`.
Generated runners accept `--growth-factor VALUE` and
`--max-concurrency-cap VALUE`.
Without TTFT/TPOT objectives there is no SLA boundary. In that mode the
generated runner keeps vLLM Workload Explorer:
```text
vllm bench sweep serve_workload --workload-var max_concurrency
```
The stage writes:
```text
sweep/concurrency-recommendation.json
```
`max_concurrency` is benchmark/deployment metadata. It is **not** written as a
`vllm serve` configuration parameter.
### Stage 3: Tune Scheduler
Before the scheduler sweep, `prepare_scheduler.py` recalculates the scheduler
baseline from:
```text
selected TP/DP
+
selected max_concurrency
+
fixed input/output token shape
```
The scheduler stage keeps an explicit `max-num-seqs` at the per-replica
concurrency lower bound while tuning `max-num-batched-tokens`. It also compares
the explicit sequence limit with the vLLM default.
The per-replica sequence baseline is recalculated as:
```text
max-num-seqs = ceil(selected max_concurrency / selected data-parallel-size)
```
The directed scheduler sweep does not test `max-num-seqs` below this value;
doing so limits immediately active requests and can turn scheduler queueing into
severe TTFT degradation. It keeps the existing batch-budget curve around the
baseline and also measures three vLLM-default reference candidates:
| Reference candidate | `max-num-seqs` | `max-num-batched-tokens` |
| --- | --- | --- |
| `vllm_default_max_num_seqs` | vLLM default | generated value |
| `vllm_default_max_num_batched_tokens` | generated value | vLLM default |
| `vllm_defaults` | vLLM default | vLLM default |
The default references do not hard-code values such as 128 or 2048. The sweep
omits the selected scheduler argument and lets the installed vLLM resolve its
normal platform-, world-size-, model-, and usage-context-aware default.
If a default-reference candidate wins, the corresponding scheduler key is
omitted from `recommended-config.yml`.
## 4. Benchmark Size, Warmup, and Failure Handling
Measured request count scales with concurrency using:
```text
num_prompts = min(1000, max(100, concurrency * 10))
```
This targets about ten concurrency turnovers when neither bound applies. The
100-prompt floor avoids very small samples for P99/compliance measurements, and
the 1000-prompt cap bounds sweep runtime.
Before a measured parameter combination, generated scripts use one unmeasured
warmup containing one full concurrency window. Warmup output is retained as
`warmup.json` for auditing but is excluded from recommendation statistics.
Generated scripts also handle sweep-run failures without discarding completed
work:
- If the installed vLLM CLI supports `--continue-on-error`, the generated script
enables it.
- If the installed vLLM does not expose that option, the script omits it and
automatically retries an interrupted sweep with `--resume`.
- The default is two automatic resume retries. Override it with
`VLLM_RECIPE_SWEEP_RETRIES`.
For example:
```bash
export VLLM_RECIPE_SWEEP_RETRIES=3
/output/sweep/run_full_sweep.sh
```
## 5. Targeted Tuning Workflows
Use these modes when a full re-tune is unnecessary.
### Tune TP/DP Only
Use this when only the parallel layout needs to be measured. This mode stops
after selecting `tensor-parallel-size` and `data-parallel-size`:
```bash
python3 /recipes/recipe_json_to_vllm_config.py \
--model meta-llama/Llama-3.1-8B-Instruct \
--hardware xeon6 \
--detect-hardware \
--input-tokens 128 \
--output-tokens 128 \
--concurrency 32 \
--ttft-sla-ms 3000 \
--tpot-sla-ms 100 \
--config-out /output/config.yml \
--env-out /output/env.sh \
--generate-parallel-layout-sweep \
--sweep-out-dir /output/sweep
```
Run:
```bash
/output/sweep/run_parallel_layout_sweep.sh
/output/sweep/recommend_parallel_layout.py
```
The result is `/output/sweep/parallel-layout-config.yml`. Use Tune All when
concurrency and scheduler tuning should follow the selected layout.
### Tune Concurrency Only
Use this when TP/DP and scheduler configuration are already fixed and only
concurrent serving capacity needs to be measured:
```bash
python3 /recipes/recipe_json_to_vllm_config.py \
--model meta-llama/Llama-3.1-8B-Instruct \
--hardware xeon6 \
--detect-hardware \
--input-tokens 128 \
--output-tokens 128 \
--concurrency 32 \
--ttft-sla-ms 3000 \
--tpot-sla-ms 100 \
--config-out /output/config.yml \
--env-out /output/env.sh \
--generate-concurrency-sweep \
--sweep-out-dir /output/sweep
```
Run:
```bash
/output/sweep/run_concurrency_sweep.sh
/output/sweep/recommend_concurrency.py
```
### Tune Scheduler Only
Use this when TP/DP and target concurrency are already known:
```bash
python3 /recipes/recipe_json_to_vllm_config.py \
--model meta-llama/Llama-3.1-8B-Instruct \
--hardware xeon6 \
--detect-hardware \
--input-tokens 128 \
--output-tokens 128 \
--concurrency 32 \
--ttft-sla-ms 3000 \
--tpot-sla-ms 100 \
--config-out /output/config.yml \
--env-out /output/env.sh \
--generate-scheduler-sweep \
--sweep-out-dir /output/sweep
```
`--generate-sweep` remains a backward-compatible alias for
`--generate-scheduler-sweep`.
Run:
```bash
/output/sweep/run_sweep.sh
/output/sweep/recommend.py
```
## 6. Recommendation Policy
With TTFT/TPOT objectives, benchmark commands use vLLM `--goodput`.
For TP/DP and scheduler selection, the recommender:
1. Excludes invalid configurations and failed benchmark measurements.
2. Calculates duration-weighted combined request compliance across successful
runs from request goodput and run duration.
3. Requires the median P99 TTFT and P99 TPOT objectives independently.
4. Also requires the combined request-compliance ratio to meet
`--minimum-compliance`, which defaults to `0.99`.
5. Establishes the highest mean output-token throughput among eligible
configurations, then applies the selection policy below.
The P99 checks and combined-compliance check answer different questions. P99
verifies the tail of each latency metric independently; combined compliance
verifies that the required fraction of requests satisfy all configured
objectives together. A candidate must pass both forms of validation to be
marked SLA eligible.
For a scheduler-only comparison with one fixed TP/DP layout, candidates within
1% of the highest eligible output-token throughput are treated as practically
equivalent. Within that set, the recommender prefers the candidate with fewer
explicit scheduler overrides, followed by compliance, goodput, and throughput.
This avoids hard-coding a scheduler value for a difference that is likely within
normal run-to-run variation. Change or disable the equivalence range with:
```bash
/output/sweep/recommend.py --throughput-equivalence-percent VALUE
```
Use `0` to restore exact highest-throughput selection. TP/DP comparisons still
use exact highest-throughput selection because changing a parallel layout is a
material deployment decision.
For concurrency selection, the policy instead selects the **highest
SLA-feasible `max_concurrency`**, with throughput/compliance metrics used to
break ties.
Change the compliance threshold with:
```bash
/output/sweep/recommend.py --minimum-compliance VALUE
```
If no configuration is SLA-feasible, the recommendation JSON records a
best-effort candidate. A deployable final configuration is produced only when
the required recommendation stages succeed.
## 7. Post-Benchmark Reporting
After Tune All finishes, `run_full_sweep.sh` automatically runs `report.py` and
writes a self-contained summary:
```text
/output/sweep/sweep-report.html
```
The HTML report includes the selected configuration, a stage summary, SLA
metrics, candidate tables, and automatic coverage observations. It uses only
Python's standard library. A report-generation error is printed as a warning
and does not change the success of the completed sweep.
The TTFT and TPOT objectives supplied to `recipe_json_to_vllm_config.py` are
embedded in every generated recommender and in `report.py`.
`run_full_sweep.sh` also passes those values explicitly to every recommendation
stage and the report. Report generation stops with an error if a stale or
manually generated recommendation JSON contains different SLA values.
Regenerate the report manually when recommendation files change:
```bash
cd /output/sweep
./report.py
```
### Optional figures
Every generated sweep package includes a standalone analysis helper. It is not
called by `run_full_sweep.sh` or any stage runner, so figure generation cannot
start, resume, or alter a sweep.
After one or more benchmark stages finish, install the plotting dependencies:
```bash
cd /output/sweep
python3 -m pip install -r requirements.txt
```
Preview the planned figures, then generate them from the completed
`summary.json` files:
```bash
./visualize.py --dry-run
./visualize.py
```
Use `--stage parallel-layout`, `--stage concurrency-tuning`, or
`--stage runtime-tuning` to analyze selected stages. The helper writes PNG files
under each stage's `figures/` directory. See the generated `VISUALIZATION.md`
for the figure set and additional examples.
## 8. Deploy the Tuned Configuration
After Tune All completes:
```bash
source /output/env.sh
vllm serve --config /output/sweep/recommended-config.yml
```
## Select sweep stages
Sweep implementation and reporting helpers live in `tools/recipes/sweep/`.
The converter remains `tools/recipes/recipe_json_to_vllm_config.py`.
Use `--sweep-stage concurrency` to generate only a concurrency sweep against
the supplied recipe's fixed TP/DP layout, for example:
```bash
python3 tools/recipes/recipe_json_to_vllm_config.py \
--model meta-llama/Llama-3.1-8B-Instruct --hardware xeon6 \
--input-tokens 128 --output-tokens 128 --concurrency 32 \
--ttft-sla-ms 3000 --tpot-sla-ms 100 \
--sweep-stage concurrency --sweep-out-dir sweep
./sweep/run_concurrency_sweep.sh
./sweep/recommend_concurrency.py
```
`--sweep-stage` accepts `parallel-layout`, `concurrency`, `scheduler`, or `all`.
The existing `--generate-*-sweep` flags remain supported; do not combine them
with `--sweep-stage`. `parallel-layout` and `all` require `--detect-hardware`.
In a package generated with `--sweep-stage all` or `--generate-full-sweep`,
select the stages to execute:
```bash
./sweep/run_full_sweep.sh --stages concurrency
./sweep/run_full_sweep.sh --stages concurrency,scheduler
./sweep/run_full_sweep.sh --stages scheduler --resume
```
The default is all three stages. Selected stages run in dependency order.
Skipping parallel-layout reuses `parallel-layout-config.yml` if present;
otherwise the runner copies the generated `initial-config.yml` as the fixed
layout. Scheduler-only execution requires `concurrency-recommendation.json`
from an earlier concurrency run. Remaining arguments go to the selected sweep
runners. Use separate output directories for unrelated experiments.
+73
View File
@@ -0,0 +1,73 @@
# Post-Benchmark Reporting
## HTML summary
`run_full_sweep.sh` automatically generates a self-contained HTML report after
the final recommendation:
```text
sweep-report.html
```
The report uses Python's standard library and does not require plotting
dependencies. It summarizes selected settings, stage results, SLA metrics, and
all measured candidates. Report-generation failures are warnings and do not
change sweep success.
The generated `report.py` uses the same TTFT and TPOT objectives supplied when
the sweep package was created. It displays both objectives and validates them
against every available recommendation JSON before writing the report.
Regenerate it manually or select a different output path:
```bash
./report.py
./report.py --output results/my-sweep-report.html
```
## Optional figures
Visualization is an independent analysis step. It reads completed sweep
results and does not start a server, run a benchmark, or change a tuning
recommendation.
Install the plotting dependencies in the same environment as vLLM:
```bash
python3 -m pip install -r requirements.txt
```
Preview the figures that would be generated:
```bash
./visualize.py --dry-run
```
Generate figures for every completed stage:
```bash
./visualize.py
```
Select one or more stages when needed:
```bash
./visualize.py --stage concurrency-tuning
./visualize.py --stage parallel-layout --stage runtime-tuning
```
Figures are written beside the source results:
```text
results/parallel-layout/figures/
results/concurrency-tuning/figures/
results/runtime-tuning/figures/
```
The analysis produces output-throughput, P99 TTFT, and P99 TPOT figures. The
concurrency stage also produces a throughput/TTFT tradeoff figure and a goodput
figure when the benchmark results contain that metric.
The native vLLM Pareto command is intentionally not used because its efficiency
axis is expressed as tokens per second per GPU, which is not an accurate label
for Xeon CPU layouts.
+84 -1
View File
@@ -4,7 +4,7 @@
"""CPU hardware discovery for optional vLLM Recipes runtime tuning.
Keep this module focused on facts exposed to the current process/container.
Tuning decisions belong in runtime_tuning.py.
Tuning decisions belong in sweep/runtime_tuning.py.
"""
from __future__ import annotations
@@ -46,6 +46,89 @@ class HardwareInfo:
return asdict(self)
def _compress_cpu_ids(cpu_ids: list[int] | tuple[int, ...]) -> str:
"""Return a compact Linux CPU-list string while preserving CPU IDs."""
values = sorted(set(cpu_ids))
if not values:
return ""
ranges: list[str] = []
start = previous = values[0]
for cpu_id in values[1:]:
if cpu_id == previous + 1:
previous = cpu_id
continue
ranges.append(str(start) if start == previous else f"{start}-{previous}")
start = previous = cpu_id
ranges.append(str(start) if start == previous else f"{start}-{previous}")
return ",".join(ranges)
def build_numa_omp_threads_bind(
hardware: HardwareInfo,
*,
reserved_cores_per_numa: int = 1,
) -> str:
"""Build explicit x86 OMP CPU lists for the effective NUMA nodes.
Temporary TP/DP sweep workaround:
- select one logical CPU per physical core, matching vLLM x86 auto-binding;
- reserve cores by omitting them from each explicit NUMA CPU list;
- emit one pipe-separated list per effective NUMA node.
Explicit VLLM_CPU_OMP_THREADS_BIND lists bypass vLLM's auto-binding reserve
logic, so the reserved cores must be excluded here.
"""
if reserved_cores_per_numa < 0:
raise ValueError("reserved_cores_per_numa must be at least 0")
effective_nodes = {node.node_id for node in hardware.numa_nodes}
allowed_cpus = get_allowed_cpu_list()
groups: list[str] = []
for node in hardware.numa_nodes:
node_cpus = [
cpu
for cpu in allowed_cpus
if cpu.numa_node == node.node_id and cpu.numa_node in effective_nodes
]
core_to_logical_ids: dict[int, list[int]] = {}
for cpu in node_cpus:
if cpu.physical_core < 0:
continue
core_to_logical_ids.setdefault(cpu.physical_core, []).append(cpu.id)
if not core_to_logical_ids:
raise RuntimeError(
f"NUMA node {node.node_id} has no physical-core topology "
"available for explicit OMP binding."
)
# Match vLLM's x86 auto-binding selector: for SMT siblings on a physical
# core, use the highest logical CPU ID.
selected_cpu_ids = sorted(
max(logical_ids) for _, logical_ids in sorted(core_to_logical_ids.items())
)
if len(selected_cpu_ids) <= reserved_cores_per_numa:
raise RuntimeError(
f"NUMA node {node.node_id} has only {len(selected_cpu_ids)} "
"usable physical cores, which is not enough after reserving "
f"{reserved_cores_per_numa} core(s)."
)
if reserved_cores_per_numa:
selected_cpu_ids = selected_cpu_ids[:-reserved_cores_per_numa]
groups.append(_compress_cpu_ids(selected_cpu_ids))
if not groups:
raise RuntimeError("No effective NUMA CPU lists are available for OMP binding.")
return "|".join(groups)
def _read_socket_id(cpu_id: int) -> int | None:
path = Path(f"/sys/devices/system/cpu/cpu{cpu_id}/topology/physical_package_id")
try:
+236 -26
View File
@@ -189,17 +189,67 @@ def parse_args() -> argparse.Namespace:
type=float,
help="Optional capacity target for future DP/capacity tuning.",
)
tuning.add_argument(
"--tune-scheduler",
action="store_true",
help=(
"Opt in to workload-derived max-num-seqs and "
"max-num-batched-tokens. By default scheduler parameters remain "
"at recipe/vLLM defaults."
),
)
sweep = p.add_argument_group(
"optional performance sweep",
("Generate benchmark files after creating one initial runtime suggestion."),
("Generate benchmark files alongside the directly deployable runtime config."),
)
sweep.add_argument(
"--sweep-stage",
choices=("parallel-layout", "concurrency", "scheduler", "all"),
help="Generate only the selected stage, or all stages in dependency order.",
)
sweep.add_argument(
"--generate-sweep",
action="store_true",
help="Backward-compatible alias for --generate-scheduler-sweep.",
)
sweep.add_argument(
"--generate-scheduler-sweep",
action="store_true",
help=("Generate the max-num-seqs/max-num-batched-tokens scheduler sweep."),
)
sweep.add_argument(
"--generate-parallel-layout-sweep",
action="store_true",
help=(
"Generate an optional vllm bench sweep package around the single "
"initial runtime suggestion."
"Generate a standalone NUMA-aware TP/DP sweep. Requires --detect-hardware."
),
)
sweep.add_argument(
"--generate-concurrency-sweep",
action="store_true",
help=(
"Generate a max_concurrency workload sweep using "
"vllm bench sweep serve_workload."
),
)
sweep.add_argument(
"--generate-full-sweep",
action="store_true",
help=(
"Generate the end-to-end TP/DP -> max_concurrency -> scheduler "
"tuning pipeline. Requires --detect-hardware."
),
)
sweep.add_argument(
"--tp-dp-numa-bind-workaround",
action=argparse.BooleanOptionalAction,
default=True,
help=(
"Temporary Xeon TP/DP NUMA-binding workaround. For parallel-layout "
"and full sweeps, generate an explicit VLLM_CPU_OMP_THREADS_BIND "
"from detected NUMA topology. Enabled by default; disable with "
"--no-tp-dp-numa-bind-workaround after the vLLM DP binding fix."
),
)
sweep.add_argument(
@@ -766,10 +816,22 @@ def write_config(
Path(path).write_text("\n".join(metadata) + "\n" + body, encoding="utf-8")
def write_env(path: str, source: str, recipe: dict[str, Any]) -> None:
env = recipe.get("env") or {}
if not isinstance(env, dict):
raise ValueError(f"Recipe `env` must be an object, got {type(env).__name__}")
def write_env(
path: str,
source: str,
recipe: dict[str, Any],
*,
env_overrides: dict[str, object] | None = None,
) -> None:
recipe_env = recipe.get("env") or {}
if not isinstance(recipe_env, dict):
raise ValueError(
f"Recipe `env` must be an object, got {type(recipe_env).__name__}"
)
env = dict(recipe_env)
overrides = env_overrides or {}
env.update(overrides)
lines = [
"#!/usr/bin/env bash",
@@ -778,6 +840,9 @@ def write_env(path: str, source: str, recipe: dict[str, Any]) -> None:
"",
]
if overrides:
lines.append("# Temporary runtime overrides derived from detected hardware.")
if env:
for key, value in env.items():
lines.append(f"export {key}={shlex.quote(str(value))}")
@@ -793,6 +858,34 @@ def main() -> int:
args = parse_args()
try:
sweep_modes = {
"--generate-sweep": args.generate_sweep,
"--generate-scheduler-sweep": args.generate_scheduler_sweep,
"--generate-parallel-layout-sweep": args.generate_parallel_layout_sweep,
"--generate-concurrency-sweep": args.generate_concurrency_sweep,
"--generate-full-sweep": args.generate_full_sweep,
}
selected_sweep_modes = [
name for name, enabled in sweep_modes.items() if enabled
]
if len(selected_sweep_modes) > 1:
raise ValueError(
"Choose exactly one sweep-generation mode: "
+ ", ".join(selected_sweep_modes)
)
if args.sweep_stage is not None:
if selected_sweep_modes:
raise ValueError(
"Do not combine --sweep-stage with --generate-*-sweep options."
)
stage_flags = {
"parallel-layout": "generate_parallel_layout_sweep",
"concurrency": "generate_concurrency_sweep",
"scheduler": "generate_scheduler_sweep",
"all": "generate_full_sweep",
}
setattr(args, stage_flags[args.sweep_stage], True)
selected_sweep_modes = ["--sweep-stage=" + args.sweep_stage]
source = args.source
if source is None:
source = discover_recipe_source(
@@ -813,7 +906,8 @@ def main() -> int:
tuning_requested = (
args.detect_hardware
or args.generate_sweep
or args.tune_scheduler
or bool(selected_sweep_modes)
or any(
value is not None
for value in (
@@ -828,12 +922,15 @@ def main() -> int:
)
tuning = None
sweep_config = None
workload = None
sweep_writer = None
hardware = None
env_overrides: dict[str, object] = {}
if tuning_requested:
# Keep plain Recipes conversion lightweight. vLLM-specific modules
# are imported only for optional runtime tuning or sweep generation.
from runtime_tuning import (
from sweep.runtime_tuning import (
WorkloadHints,
finetune_runtime_config,
get_runtime_tuning_policies,
@@ -848,24 +945,60 @@ def main() -> int:
target_qps=args.target_qps,
)
if args.generate_sweep:
from sweep_generation import (
validate_sweep_workload,
write_sweep_files,
)
if selected_sweep_modes:
from sweep.sweep_generation import validate_sweep_workload
validate_sweep_workload(workload)
if args.generate_sweep or args.generate_scheduler_sweep:
from sweep.sweep_generation import write_sweep_files
sweep_writer = write_sweep_files
recipe_hardware = recipe.get("hardware")
policies = get_runtime_tuning_policies(recipe_hardware)
scheduler_tuning_requested = bool(
args.tune_scheduler
or args.generate_sweep
or args.generate_scheduler_sweep
)
policies = get_runtime_tuning_policies(
recipe_hardware,
tune_scheduler=scheduler_tuning_requested,
)
hardware = None
if args.detect_hardware:
from hardware_detection import detect_hardware
from hardware_detection import (
build_numa_omp_threads_bind,
detect_hardware,
)
hardware = detect_hardware()
if (
args.tp_dp_numa_bind_workaround
and str(recipe_hardware).lower() == "xeon6"
and (
args.generate_parallel_layout_sweep or args.generate_full_sweep
)
):
recipe_env = recipe.get("env") or {}
current_binding = (
recipe_env.get("VLLM_CPU_OMP_THREADS_BIND")
if isinstance(recipe_env, dict)
else None
)
# Preserve an explicit recipe-provided manual binding.
# Replace an unset or "auto" value with the temporary
# topology-derived binding.
if current_binding in (None, "auto"):
env_overrides["VLLM_CPU_OMP_THREADS_BIND"] = (
build_numa_omp_threads_bind(
hardware,
reserved_cores_per_numa=1,
)
)
tuning = finetune_runtime_config(
config,
hardware=hardware,
@@ -873,20 +1006,74 @@ def main() -> int:
policies=policies,
)
config.update(tuning.overrides)
if args.tune_scheduler:
config.update(tuning.sweep_overrides)
sweep_config = dict(config)
sweep_config.update(tuning.sweep_overrides)
write_config(args.config_out, source, recipe, config)
write_env(args.env_out, source, recipe)
write_env(
args.env_out,
source,
recipe,
env_overrides=env_overrides,
)
sweep_files: list[Path] = []
if args.generate_sweep:
if args.generate_full_sweep:
if not args.detect_hardware or hardware is None:
raise ValueError("--generate-full-sweep requires --detect-hardware.")
assert workload is not None
assert sweep_writer is not None
sweep_files = sweep_writer(
from sweep.sweep_generation import write_full_sweep_files
sweep_files = write_full_sweep_files(
args.sweep_out_dir,
config_path=args.config_out,
env_path=args.env_out,
config=config,
workload=workload,
numa_node_count=hardware.numa_node_count,
tune_scheduler=args.tune_scheduler,
)
elif args.generate_parallel_layout_sweep:
if not args.detect_hardware or hardware is None:
raise ValueError(
"--generate-parallel-layout-sweep requires --detect-hardware."
)
assert workload is not None
from sweep.sweep_generation import write_parallel_layout_sweep_files
sweep_files = write_parallel_layout_sweep_files(
args.sweep_out_dir,
config_path=args.config_out,
env_path=args.env_out,
config=config,
workload=workload,
numa_node_count=hardware.numa_node_count,
tune_scheduler=args.tune_scheduler,
)
elif args.generate_concurrency_sweep:
assert workload is not None
from sweep.sweep_generation import write_concurrency_sweep_files
sweep_files = write_concurrency_sweep_files(
args.sweep_out_dir,
config_path=args.config_out,
env_path=args.env_out,
config=config,
workload=workload,
tune_scheduler=args.tune_scheduler,
)
elif args.generate_sweep or args.generate_scheduler_sweep:
assert workload is not None
assert sweep_writer is not None
assert sweep_config is not None
sweep_files = sweep_writer(
args.sweep_out_dir,
config_path=args.config_out,
env_path=args.env_out,
config=sweep_config,
workload=workload,
)
if tuning is not None:
@@ -894,6 +1081,13 @@ def main() -> int:
print("Initial runtime suggestion:")
for key, value in tuning.overrides.items():
print(f" {key}: {value}")
if tuning.sweep_overrides:
if args.tune_scheduler:
print("Workload-derived scheduler tuning:")
elif selected_sweep_modes:
print("Explicit scheduler sweep seed:")
for key, value in tuning.sweep_overrides.items():
print(f" {key}: {value}")
for note in tuning.notes:
print(f" tuning: {note}")
except Exception as exc:
@@ -910,15 +1104,31 @@ def main() -> int:
print(f" vllm serve --config {shlex.quote(args.config_out)}")
if sweep_files:
sweep_dir = Path(args.sweep_out_dir)
run_parallel = sweep_dir / "run_parallel_layout_sweep.sh"
recommend_parallel = sweep_dir / "recommend_parallel_layout.py"
run_concurrency = sweep_dir / "run_concurrency_sweep.sh"
recommend_concurrency = sweep_dir / "recommend_concurrency.py"
run_sweep = sweep_dir / "run_sweep.sh"
recommend = sweep_dir / "recommend.py"
run_full = sweep_dir / "run_full_sweep.sh"
print()
print("Optional performance sweep:")
print(f" {shlex.quote(str(run_sweep))} --dry-run")
print(f" {shlex.quote(str(run_sweep))}")
print()
print("After the sweep:")
print(f" {shlex.quote(str(recommend))}")
if args.generate_full_sweep:
print(f" {shlex.quote(str(run_full))}")
elif args.generate_concurrency_sweep:
print(f" {shlex.quote(str(run_concurrency))} --dry-run")
print(f" {shlex.quote(str(run_concurrency))}")
print(f" {shlex.quote(str(recommend_concurrency))}")
elif args.generate_parallel_layout_sweep:
print(f" {shlex.quote(str(run_parallel))} --dry-run")
print(f" {shlex.quote(str(run_parallel))}")
print(f" {shlex.quote(str(recommend_parallel))}")
else:
print(f" {shlex.quote(str(run_sweep))} --dry-run")
print(f" {shlex.quote(str(run_sweep))}")
print()
print("After the sweep:")
print(f" {shlex.quote(str(recommend))}")
return 0
+4
View File
@@ -0,0 +1,4 @@
# Dependencies used by `vllm bench sweep plot`.
matplotlib
pandas
seaborn
+4
View File
@@ -0,0 +1,4 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Runtime tuning and benchmark sweep generation helpers."""
+261
View File
@@ -0,0 +1,261 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Find the maximum SLA-feasible serving concurrency adaptively."""
from __future__ import annotations
import argparse
import shlex
from pathlib import Path
from statistics import median
from typing import Any
from vllm.benchmarks.sweep.param_sweep import ParameterSweepItem
from vllm.benchmarks.sweep.serve import run_server
from vllm.benchmarks.sweep.serve_workload import run_comb_workload
DEFAULT_MINIMUM_COMPLIANCE = 0.99
DEFAULT_MAX_CONCURRENCY_CAP = 1000
DEFAULT_GROWTH_FACTOR = 2.0
def _number(value: object) -> float | None:
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
return float(value)
def _num_prompts(concurrency: int) -> int:
return min(1000, max(100, concurrency * 10))
def _sla_eligible(
rows: list[dict[str, Any]],
*,
expected_runs: int,
ttft_sla_ms: float | None,
tpot_sla_ms: float | None,
minimum_compliance: float,
) -> bool:
if len(rows) < expected_runs:
return False
if any(int(_number(row.get("failed")) or 0) for row in rows):
return False
completed = sum(int(_number(row.get("completed")) or 0) for row in rows)
if completed <= 0:
return False
compliant = sum(
(_number(row.get("request_goodput")) or 0.0)
* (_number(row.get("duration")) or 0.0)
for row in rows
)
if compliant / completed < minimum_compliance:
return False
if ttft_sla_ms is not None:
values = [
value
for row in rows
if (value := _number(row.get("p99_ttft_ms"))) is not None
]
if not values or median(values) > ttft_sla_ms:
return False
if tpot_sla_ms is not None:
values = [
value
for row in rows
if (value := _number(row.get("p99_tpot_ms"))) is not None
]
if not values or median(values) > tpot_sla_ms:
return False
return True
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description=(
"Find the highest SLA-feasible max_concurrency using exponential "
"bracketing followed by integer binary search."
)
)
parser.add_argument("--serve-cmd", required=True)
parser.add_argument("--bench-cmd", required=True)
parser.add_argument("--results-dir", required=True)
parser.add_argument("--seed-concurrency", type=int, required=True)
parser.add_argument("--input-tokens", type=int, required=True)
parser.add_argument("--output-tokens", type=int, required=True)
parser.add_argument("--ttft-sla-ms", type=float)
parser.add_argument("--tpot-sla-ms", type=float)
parser.add_argument(
"--minimum-compliance",
type=float,
default=DEFAULT_MINIMUM_COMPLIANCE,
)
parser.add_argument(
"--max-concurrency-cap",
type=int,
default=DEFAULT_MAX_CONCURRENCY_CAP,
)
parser.add_argument(
"--growth-factor",
type=float,
default=DEFAULT_GROWTH_FACTOR,
)
parser.add_argument("--num-runs", type=int, default=3)
parser.add_argument("--server-ready-timeout", type=int, default=1800)
parser.add_argument("--show-stdout", action="store_true")
parser.add_argument("--continue-on-error", action="store_true")
parser.add_argument("--resume", action="store_true")
parser.add_argument("--dry-run", action="store_true")
return parser.parse_args()
def main() -> int:
args = parse_args()
if args.seed_concurrency <= 0:
raise ValueError("--seed-concurrency must be positive.")
if not 0.0 < args.minimum_compliance <= 1.0:
raise ValueError("--minimum-compliance must be greater than 0 and at most 1.")
if args.growth_factor <= 1.0:
raise ValueError("--growth-factor must be greater than 1.")
if args.max_concurrency_cap <= 0:
raise ValueError("--max-concurrency-cap must be positive.")
if args.num_runs <= 0:
raise ValueError("--num-runs must be positive.")
if args.ttft_sla_ms is None and args.tpot_sla_ms is None:
raise ValueError("Adaptive search requires TTFT and/or TPOT SLA.")
seed = args.seed_concurrency
cap = max(seed, args.max_concurrency_cap)
results_dir = Path(args.results_dir)
results_dir.mkdir(parents=True, exist_ok=True)
if args.dry_run:
first_upper = min(
cap,
max(seed + 1, int(round(seed * args.growth_factor))),
)
print("Adaptive concurrency search plan:")
print(f" seed: {seed}")
print(f" first upper probe: {first_upper}")
print(f" hard cap: {cap}")
print(" strategy: bracket SLA boundary, then integer binary search")
return 0
serve_cmd = shlex.split(args.serve_cmd)
bench_cmd = shlex.split(args.bench_cmd)
serve_comb = ParameterSweepItem()
def run_point(server: object, concurrency: int) -> bool:
bench_comb = ParameterSweepItem(
{
"_benchmark_name": "user_workload",
"random_input_len": args.input_tokens,
"random_output_len": args.output_tokens,
"num_prompts": _num_prompts(concurrency),
}
)
print(f"[ADAPTIVE CONCURRENCY] testing max_concurrency={concurrency}")
rows = (
run_comb_workload(
server,
bench_cmd,
serve_comb=serve_comb,
bench_comb=bench_comb,
link_vars=[],
experiment_dir=results_dir,
num_runs=args.num_runs,
dry_run=False,
warmup_num_prompts=min(concurrency, 1000),
continue_on_error=args.continue_on_error,
workload_var="max_concurrency",
workload_value=concurrency,
)
or []
)
eligible = _sla_eligible(
rows,
expected_runs=args.num_runs,
ttft_sla_ms=args.ttft_sla_ms,
tpot_sla_ms=args.tpot_sla_ms,
minimum_compliance=args.minimum_compliance,
)
print(
"[ADAPTIVE CONCURRENCY] "
f"max_concurrency={concurrency}: {'PASS' if eligible else 'FAIL'}"
)
return eligible
with run_server(
serve_cmd,
[],
show_stdout=args.show_stdout,
serve_overrides=serve_comb,
dry_run=False,
server_ready_timeout=args.server_ready_timeout,
) as server:
seed_passes = run_point(server, seed)
if seed_passes:
low = seed
high: int | None = None
while low < cap:
probe = min(
cap,
max(low + 1, int(round(low * args.growth_factor))),
)
if run_point(server, probe):
low = probe
if low == cap:
print(
"[ADAPTIVE CONCURRENCY] reached cap without an SLA failure"
)
return 0
else:
high = probe
break
if high is None:
return 0
else:
high = seed
low: int | None = None
probe = seed
while probe > 1:
probe = max(1, probe // 2)
if run_point(server, probe):
low = probe
break
high = probe
if low is None:
print(
"[ADAPTIVE CONCURRENCY] no SLA-feasible concurrency found, "
"including max_concurrency=1"
)
return 0
assert low is not None
assert high is not None
print(f"[ADAPTIVE CONCURRENCY] refining bracket PASS={low}, FAIL={high}")
while high - low > 1:
midpoint = (low + high) // 2
if run_point(server, midpoint):
low = midpoint
else:
high = midpoint
print(f"[ADAPTIVE CONCURRENCY] boundary resolved: PASS={low}, FAIL={high}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,330 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Select one max-concurrency recommendation from a workload sweep."""
from __future__ import annotations
import argparse
import json
from collections import defaultdict
from pathlib import Path
from statistics import mean, median
from typing import Any
DEFAULT_TTFT_SLA_MS: float | None = None
DEFAULT_TPOT_SLA_MS: float | None = None
DEFAULT_SEED_CONCURRENCY: int | None = None
DEFAULT_MINIMUM_COMPLIANCE: float = 0.99
DEFAULT_RESULTS_DIR = "results/concurrency-tuning"
DEFAULT_OUTPUT_JSON = "concurrency-recommendation.json"
def _number(value: object) -> float | None:
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
return float(value)
def _mean(rows: list[dict[str, Any]], key: str) -> float | None:
values = [_number(row.get(key)) for row in rows]
numbers = [value for value in values if value is not None]
return mean(numbers) if numbers else None
def _percentile_summary(
rows: list[dict[str, Any]], key: str
) -> tuple[float | None, float | None, float | None]:
values = [_number(row.get(key)) for row in rows]
numbers = [value for value in values if value is not None]
if not numbers:
return None, None, None
return mean(numbers), median(numbers), max(numbers)
def _load_runs(results_dir: Path) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
for path in sorted(results_dir.rglob("summary.json")):
data = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(data, list):
continue
for row in data:
if isinstance(row, dict):
copied = dict(row)
copied["_summary_path"] = str(path)
rows.append(copied)
return rows
def _aggregate_candidates(
rows: list[dict[str, Any]],
*,
use_goodput: bool,
ttft_sla_ms: float | None,
tpot_sla_ms: float | None,
minimum_compliance: float,
) -> list[dict[str, Any]]:
grouped: dict[int, list[dict[str, Any]]] = defaultdict(list)
for row in rows:
concurrency = row.get("max_concurrency")
if (
isinstance(concurrency, bool)
or not isinstance(concurrency, int)
or concurrency <= 0
):
continue
grouped[concurrency].append(row)
candidates: list[dict[str, Any]] = []
for concurrency, runs in sorted(grouped.items()):
failed_requests = sum(int(_number(run.get("failed")) or 0) for run in runs)
output_throughput = _mean(runs, "output_throughput")
request_throughput = _mean(runs, "request_throughput")
request_goodput = _mean(runs, "request_goodput")
mean_p99_ttft_ms, median_p99_ttft_ms, worst_p99_ttft_ms = _percentile_summary(
runs, "p99_ttft_ms"
)
mean_p99_tpot_ms, median_p99_tpot_ms, worst_p99_tpot_ms = _percentile_summary(
runs, "p99_tpot_ms"
)
completed_requests = sum(
int(_number(run.get("completed")) or 0) for run in runs
)
estimated_compliant_requests = sum(
(_number(run.get("request_goodput")) or 0.0)
* (_number(run.get("duration")) or 0.0)
for run in runs
)
combined_compliance_ratio = (
estimated_compliant_requests / completed_requests
if completed_requests > 0
else None
)
valid = failed_requests == 0 and output_throughput is not None
reason = None
if failed_requests:
valid = False
reason = f"{failed_requests} failed request(s) across successful runs"
elif use_goodput and request_goodput is None:
valid = False
reason = "request_goodput is missing; rerun with SLO goodput enabled"
p99_sla_eligible = valid
if ttft_sla_ms is not None:
p99_sla_eligible = bool(
p99_sla_eligible
and median_p99_ttft_ms is not None
and median_p99_ttft_ms <= ttft_sla_ms
)
if tpot_sla_ms is not None:
p99_sla_eligible = bool(
p99_sla_eligible
and median_p99_tpot_ms is not None
and median_p99_tpot_ms <= tpot_sla_ms
)
compliance_eligible = bool(
valid
and combined_compliance_ratio is not None
and combined_compliance_ratio >= minimum_compliance
)
candidates.append(
{
"max_concurrency": concurrency,
"run_count": len(runs),
"failed_requests": failed_requests,
"mean_request_goodput": request_goodput,
"mean_request_throughput": request_throughput,
"mean_output_throughput": output_throughput,
"mean_p99_ttft_ms": mean_p99_ttft_ms,
"median_p99_ttft_ms": median_p99_ttft_ms,
"worst_p99_ttft_ms": worst_p99_ttft_ms,
"mean_p99_tpot_ms": mean_p99_tpot_ms,
"median_p99_tpot_ms": median_p99_tpot_ms,
"worst_p99_tpot_ms": worst_p99_tpot_ms,
"estimated_compliant_requests": estimated_compliant_requests,
"completed_requests": completed_requests,
"combined_compliance_ratio": combined_compliance_ratio,
"combined_compliance_percent": (
combined_compliance_ratio * 100
if combined_compliance_ratio is not None
else None
),
"benchmark_valid": valid,
"p99_sla_eligible": p99_sla_eligible,
"compliance_eligible": compliance_eligible,
"sla_eligible": p99_sla_eligible and compliance_eligible,
"valid": valid,
"invalid_reason": reason,
"summary_files": sorted({str(run["_summary_path"]) for run in runs}),
}
)
return candidates
def _select_candidate(
candidates: list[dict[str, Any]], *, use_goodput: bool
) -> tuple[dict[str, Any] | None, dict[str, Any], str]:
valid = [candidate for candidate in candidates if candidate["valid"]]
if not valid:
raise ValueError(
"No valid workload-sweep point was found. Inspect failed runs and metrics."
)
if use_goodput:
objective = "highest_sla_feasible_max_concurrency"
eligible = [candidate for candidate in valid if candidate["sla_eligible"]]
winner = (
max(
eligible,
key=lambda candidate: (
candidate["max_concurrency"],
candidate["mean_output_throughput"],
candidate["combined_compliance_ratio"] or 0.0,
candidate["mean_request_goodput"] or 0.0,
),
)
if eligible
else None
)
best_effort = max(
valid,
key=lambda candidate: (
candidate["mean_request_goodput"] or 0.0,
candidate["combined_compliance_ratio"] or 0.0,
candidate["mean_output_throughput"],
candidate["max_concurrency"],
),
)
else:
objective = "highest_mean_output_throughput"
winner = max(
valid,
key=lambda candidate: (
candidate["mean_output_throughput"],
candidate["max_concurrency"],
),
)
best_effort = winner
return winner, best_effort, objective
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Select max_concurrency from vLLM Workload Explorer results."
)
parser.add_argument("--results-dir", default=DEFAULT_RESULTS_DIR)
parser.add_argument("--ttft-sla-ms", type=float, default=DEFAULT_TTFT_SLA_MS)
parser.add_argument("--tpot-sla-ms", type=float, default=DEFAULT_TPOT_SLA_MS)
parser.add_argument(
"--seed-concurrency", type=int, default=DEFAULT_SEED_CONCURRENCY
)
parser.add_argument(
"--minimum-compliance",
type=float,
default=DEFAULT_MINIMUM_COMPLIANCE,
)
parser.add_argument("--output-json", default=DEFAULT_OUTPUT_JSON)
return parser.parse_args()
def main() -> int:
args = parse_args()
if not 0.0 < args.minimum_compliance <= 1.0:
raise ValueError("--minimum-compliance must be greater than 0 and at most 1.")
script_dir = Path(__file__).resolve().parent
results_dir = Path(args.results_dir)
if not results_dir.is_absolute():
results_dir = script_dir / results_dir
output_json = Path(args.output_json)
if not output_json.is_absolute():
output_json = script_dir / output_json
if not results_dir.exists():
raise ValueError(f"Workload-sweep results were not found at {results_dir}.")
rows = _load_runs(results_dir)
if not rows:
raise ValueError(f"No summary.json results found under {results_dir}.")
use_goodput = args.ttft_sla_ms is not None or args.tpot_sla_ms is not None
candidates = _aggregate_candidates(
rows,
use_goodput=use_goodput,
ttft_sla_ms=args.ttft_sla_ms,
tpot_sla_ms=args.tpot_sla_ms,
minimum_compliance=args.minimum_compliance,
)
winner, best_effort, objective = _select_candidate(
candidates, use_goodput=use_goodput
)
recommendation = {
"status": (
"sla_feasible" if winner is not None else "no_sla_feasible_concurrency"
),
"selection_objective": objective,
"seed_concurrency": args.seed_concurrency,
"minimum_compliance_ratio": args.minimum_compliance,
"slo": {
"ttft_ms": args.ttft_sla_ms,
"tpot_ms": args.tpot_sla_ms,
},
"recommended": (
{"max_concurrency": winner["max_concurrency"]}
if winner is not None
else None
),
"measured": (
{
key: winner[key]
for key in (
"run_count",
"mean_request_throughput",
"mean_request_goodput",
"mean_output_throughput",
"combined_compliance_ratio",
"combined_compliance_percent",
"mean_p99_ttft_ms",
"median_p99_ttft_ms",
"worst_p99_ttft_ms",
"mean_p99_tpot_ms",
"median_p99_tpot_ms",
"worst_p99_tpot_ms",
)
}
if winner is not None
else None
),
"best_effort": {
"max_concurrency": best_effort["max_concurrency"],
"mean_request_goodput": best_effort["mean_request_goodput"],
"mean_output_throughput": best_effort["mean_output_throughput"],
"combined_compliance_ratio": best_effort["combined_compliance_ratio"],
"p99_sla_eligible": best_effort["p99_sla_eligible"],
},
"candidates": candidates,
}
output_json.write_text(
json.dumps(recommendation, indent=2) + "\n", encoding="utf-8"
)
if winner is None:
print("No SLA-feasible max_concurrency was found.")
print(f"Wrote diagnostic evidence to {output_json}")
return 2
print(f"Recommended max_concurrency: {winner['max_concurrency']}")
print(f"Selection objective: {objective}")
print(f"Wrote {output_json}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -47,7 +47,10 @@ class WorkloadHints:
@dataclass
class TuningResult:
"""Runtime overrides and benchmark-only seeds produced by tuning policies."""
overrides: dict[str, Any] = field(default_factory=dict)
sweep_overrides: dict[str, Any] = field(default_factory=dict)
notes: list[str] = field(default_factory=list)
@@ -67,13 +70,24 @@ def _record_override(
result.notes.append(f"{key}={value!r}: {reason}")
def _record_sweep_override(
result: TuningResult,
key: str,
value: Any,
reason: str,
) -> None:
"""Record an explicit benchmark seed without changing deployment config."""
result.sweep_overrides[key] = value
result.notes.append(f"sweep {key}={value!r}: {reason}")
def _get_active_sequence_count(
config: dict[str, Any],
workload: WorkloadHints,
) -> int:
"""Return the sequence count used to size one scheduler iteration."""
"""Return the per-replica sequence count for one scheduler iteration."""
if workload.concurrency is not None:
return workload.concurrency
return ceil(workload.concurrency / _get_data_parallel_size(config))
configured = config.get("max-num-seqs")
if (
@@ -86,9 +100,22 @@ def _get_active_sequence_count(
return SchedulerConfig.DEFAULT_MAX_NUM_SEQS
def _get_data_parallel_size(config: dict[str, Any]) -> int:
"""Return a valid configured DP size, falling back to one replica."""
configured = config.get("data-parallel-size", 1)
if (
isinstance(configured, int)
and not isinstance(configured, bool)
and configured > 0
):
return configured
return 1
def _estimate_prefills_per_step(
workload: WorkloadHints,
active_sequences: int,
data_parallel_size: int,
) -> float:
"""Estimate steady-state prompt arrivals per decode scheduler step."""
# Reserve enough prefill budget for at least one representative prompt.
@@ -105,9 +132,10 @@ def _estimate_prefills_per_step(
# With target QPS + TPOT, approximate how many prompts arrive during one
# decode scheduler-step interval.
if workload.target_qps is not None and workload.tpot_sla_ms is not None:
per_replica_qps = workload.target_qps / data_parallel_size
prefills_per_step = max(
prefills_per_step,
workload.target_qps * workload.tpot_sla_ms / 1000.0,
per_replica_qps * workload.tpot_sla_ms / 1000.0,
)
# A step cannot replace more requests than are active.
@@ -213,15 +241,22 @@ def _resolve_max_num_seqs(
workload: WorkloadHints,
result: TuningResult,
) -> None:
del config, hardware
del hardware
if workload.concurrency is None:
return
_record_override(
data_parallel_size = _get_data_parallel_size(config)
per_replica_concurrency = ceil(workload.concurrency / data_parallel_size)
_record_sweep_override(
result,
"max-num-seqs",
workload.concurrency,
"derived from the optional target concurrency",
per_replica_concurrency,
(
"benchmark seed derived from the optional global target concurrency "
f"and data_parallel_size={data_parallel_size}; config.yml keeps the "
"recipe/vLLM scheduler default"
),
)
@@ -235,12 +270,17 @@ def _resolve_max_num_batched_tokens(
if workload.input_tokens is None:
return
data_parallel_size = _get_data_parallel_size(config)
active_sequences = _get_active_sequence_count(config, workload)
# Decode requests consume roughly one token per active sequence in one
# scheduler iteration. Prefills share the remaining scheduler token budget.
decode_budget = active_sequences
prefills_per_step = _estimate_prefills_per_step(workload, active_sequences)
prefills_per_step = _estimate_prefills_per_step(
workload,
active_sequences,
data_parallel_size,
)
prefill_budget = ceil(workload.input_tokens * prefills_per_step)
# Use vLLM's scheduler defaults/constraints as floors rather than local
@@ -263,16 +303,18 @@ def _resolve_max_num_batched_tokens(
):
candidate = max(candidate, max_model_len)
_record_override(
_record_sweep_override(
result,
"max-num-batched-tokens",
candidate,
(
"workload-derived scheduler budget "
f"(active_sequences={active_sequences}, "
"workload-derived benchmark seed "
f"(per_replica_active_sequences={active_sequences}, "
f"data_parallel_size={data_parallel_size}, "
f"prefills_per_step={prefills_per_step:.2f}, "
f"decode_budget={decode_budget}, "
f"prefill_budget={prefill_budget})"
f"prefill_budget={prefill_budget}); config.yml keeps the "
"recipe/vLLM scheduler default"
),
)
@@ -296,9 +338,12 @@ def _resolve_data_parallel_size(
DEFAULT_POLICIES: tuple[Policy, ...] = (
_resolve_tensor_parallel_size,
_resolve_gpu_memory_utilization,
_resolve_data_parallel_size,
)
SCHEDULER_TUNING_POLICIES: tuple[Policy, ...] = (
_resolve_max_num_seqs,
_resolve_max_num_batched_tokens,
_resolve_data_parallel_size,
)
# Select tuning behavior from the hardware requested by the Recipes rendering,
@@ -312,6 +357,8 @@ HARDWARE_TUNING_POLICIES: dict[str, tuple[Policy, ...]] = {
def get_runtime_tuning_policies(
recipe_hardware: object,
*,
tune_scheduler: bool = False,
) -> tuple[Policy, ...]:
if not isinstance(recipe_hardware, str) or not recipe_hardware:
raise ValueError(
@@ -326,6 +373,8 @@ def get_runtime_tuning_policies(
"Runtime tuning is not supported for recipe hardware "
f"{recipe_hardware!r}. Currently supported: {supported}."
)
if tune_scheduler:
policies = (*policies, *SCHEDULER_TUNING_POLICIES)
return policies
@@ -336,7 +385,7 @@ def finetune_runtime_config(
workload: WorkloadHints | None = None,
policies: tuple[Policy, ...] = DEFAULT_POLICIES,
) -> TuningResult:
"""Return overrides; never mutate the recipe-derived config in place."""
"""Return deployment overrides and sweep seeds without mutating config."""
hints = workload or WorkloadHints()
hints.validate()
@@ -0,0 +1,225 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Prepare scheduler sweep inputs after TP/DP and concurrency selection."""
from __future__ import annotations
import argparse
import json
import math
from pathlib import Path
from typing import Any
try:
import yaml
except ImportError as exc:
raise SystemExit("PyYAML is required. Install it with: pip install pyyaml") from exc
PROMPTS_PER_CONCURRENCY = 10
MIN_NUM_PROMPTS = 100
MAX_NUM_PROMPTS = 1000
def _positive_int(config: dict[str, Any], key: str) -> int:
value = config.get(key)
if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
raise ValueError(f"Expected positive integer {key!r}, got {value!r}.")
return value
def _num_prompts_for_concurrency(concurrency: int) -> int:
return min(
MAX_NUM_PROMPTS,
max(MIN_NUM_PROMPTS, concurrency * PROMPTS_PER_CONCURRENCY),
)
def _strict_lower_power_of_two(value: int) -> int:
if value <= 1:
return 1
return 1 << ((value - 1).bit_length() - 1)
def _strict_upper_power_of_two(value: int) -> int:
return 1 << value.bit_length()
def _scheduler_baseline(
config: dict[str, Any],
*,
concurrency: int,
input_tokens: int,
output_tokens: int,
tpot_sla_ms: float | None,
target_qps: float | None,
) -> tuple[int, int]:
data_parallel_size = int(config.get("data-parallel-size", 1))
if data_parallel_size <= 0:
raise ValueError("data-parallel-size must be positive.")
per_replica_concurrency = math.ceil(concurrency / data_parallel_size)
prefills_per_step = max(1.0, per_replica_concurrency / max(output_tokens, 1))
if target_qps is not None and tpot_sla_ms is not None:
per_replica_qps = target_qps / data_parallel_size
prefills_per_step = max(
prefills_per_step,
per_replica_qps * tpot_sla_ms / 1000.0,
)
prefills_per_step = min(float(per_replica_concurrency), prefills_per_step)
batch = max(
2048,
per_replica_concurrency,
per_replica_concurrency + math.ceil(input_tokens * prefills_per_step),
)
if config.get("enable-chunked-prefill") is False:
max_model_len = config.get("max-model-len")
if isinstance(max_model_len, int) and not isinstance(max_model_len, bool):
batch = max(batch, max_model_len)
return per_replica_concurrency, batch
def _build_serve_params(config: dict[str, Any]) -> list[dict[str, Any]]:
initial_seqs = _positive_int(config, "max-num-seqs")
initial_batch = _positive_int(config, "max-num-batched-tokens")
minimum_batch = initial_seqs
if config.get("enable-chunked-prefill") is False:
max_model_len = config.get("max-model-len")
if (
isinstance(max_model_len, int)
and not isinstance(max_model_len, bool)
and max_model_len > 0
):
minimum_batch = max(minimum_batch, max_model_len)
lower_batch = max(minimum_batch, _strict_lower_power_of_two(initial_batch))
smaller_batch = max(minimum_batch, _strict_lower_power_of_two(lower_batch))
higher_batch = max(minimum_batch, _strict_upper_power_of_two(initial_batch))
candidates: list[dict[str, Any]] = []
seen: set[tuple[int | None, int | None]] = set()
def add(
name: str,
max_num_seqs: int | None,
max_num_batched_tokens: int | None,
) -> None:
if max_num_seqs is not None and max_num_batched_tokens is not None:
max_num_batched_tokens = max(max_num_batched_tokens, max_num_seqs)
signature = (max_num_seqs, max_num_batched_tokens)
if signature in seen:
return
seen.add(signature)
candidate: dict[str, Any] = {"_benchmark_name": name}
if max_num_seqs is not None:
candidate["max_num_seqs"] = max_num_seqs
if max_num_batched_tokens is not None:
candidate["max_num_batched_tokens"] = max_num_batched_tokens
candidates.append(candidate)
add("initial", initial_seqs, initial_batch)
add("smaller_batch_budget", initial_seqs, smaller_batch)
add("lower_batch_budget", initial_seqs, lower_batch)
add("higher_batch_budget", initial_seqs, higher_batch)
add("vllm_default_max_num_seqs", None, initial_batch)
add("vllm_default_max_num_batched_tokens", initial_seqs, None)
add("vllm_defaults", None, None)
return candidates
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("--config", required=True)
parser.add_argument("--concurrency-recommendation", required=True)
parser.add_argument("--input-tokens", type=int, required=True)
parser.add_argument("--output-tokens", type=int, required=True)
parser.add_argument("--tpot-sla-ms", type=float)
parser.add_argument("--target-qps", type=float)
parser.add_argument("--output-config", required=True)
parser.add_argument("--serve-params", required=True)
parser.add_argument("--bench-params", required=True)
return parser.parse_args()
def main() -> int:
args = parse_args()
config_path = Path(args.config)
recommendation_path = Path(args.concurrency_recommendation)
config = yaml.safe_load(config_path.read_text(encoding="utf-8"))
if not isinstance(config, dict):
raise ValueError(f"{config_path} does not contain a YAML object.")
recommendation = json.loads(recommendation_path.read_text(encoding="utf-8"))
selected = recommendation.get("recommended")
if not isinstance(selected, dict):
raise ValueError(
f"{recommendation_path} has no SLA-feasible concurrency recommendation."
)
concurrency = selected.get("max_concurrency")
if (
isinstance(concurrency, bool)
or not isinstance(concurrency, int)
or concurrency <= 0
):
raise ValueError("Recommended max_concurrency is missing or invalid.")
seqs, batch = _scheduler_baseline(
config,
concurrency=concurrency,
input_tokens=args.input_tokens,
output_tokens=args.output_tokens,
tpot_sla_ms=args.tpot_sla_ms,
target_qps=args.target_qps,
)
scheduler_seed = dict(config)
scheduler_seed["max-num-seqs"] = seqs
scheduler_seed["max-num-batched-tokens"] = batch
scheduler_config = dict(config)
scheduler_config.pop("max-num-seqs", None)
scheduler_config.pop("max-num-batched-tokens", None)
Path(args.output_config).write_text(
yaml.safe_dump(
scheduler_config,
sort_keys=False,
default_flow_style=False,
allow_unicode=True,
),
encoding="utf-8",
)
Path(args.serve_params).write_text(
json.dumps(_build_serve_params(scheduler_seed), indent=2) + "\n",
encoding="utf-8",
)
Path(args.bench_params).write_text(
json.dumps(
[
{
"random_input_len": args.input_tokens,
"random_output_len": args.output_tokens,
"max_concurrency": concurrency,
"num_prompts": _num_prompts_for_concurrency(concurrency),
}
],
indent=2,
)
+ "\n",
encoding="utf-8",
)
print(f"Prepared scheduler sweep for max_concurrency={concurrency}")
print(f"Scheduler seed max-num-seqs={seqs}")
print(f"Scheduler seed max-num-batched-tokens={batch}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
File diff suppressed because it is too large Load Diff
+756
View File
@@ -0,0 +1,756 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Generate a self-contained HTML summary from completed recipe sweeps."""
from __future__ import annotations
import argparse
import json
from datetime import datetime, timezone
from html import escape
from pathlib import Path
from typing import Any
STAGES = (
(
"parallel-layout",
"Parallel layout",
"parallel-layout-recommendation.json",
),
(
"concurrency-tuning",
"Concurrency",
"concurrency-recommendation.json",
),
("runtime-tuning", "Scheduler", "recommendation.json"),
)
DEFAULT_TTFT_SLA_MS: float | None = None
DEFAULT_TPOT_SLA_MS: float | None = None
def _read_object(path: Path) -> dict[str, Any] | None:
if not path.is_file():
return None
value = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(value, dict):
raise ValueError(f"{path} does not contain a JSON object.")
return value
def _summary_runs(stage_dir: Path) -> list[dict[str, Any]]:
runs: list[dict[str, Any]] = []
for path in sorted(stage_dir.glob("**/summary.json")):
value = json.loads(path.read_text(encoding="utf-8"))
if isinstance(value, list):
runs.extend(item for item in value if isinstance(item, dict))
return runs
def _number(value: Any) -> float | None:
if isinstance(value, bool) or not isinstance(value, int | float):
return None
return float(value)
def _fmt(value: Any, digits: int = 1) -> str:
number = _number(value)
if number is None:
return "-"
return f"{number:.{digits}f}"
def _fmt_int(value: Any) -> str:
number = _number(value)
return "-" if number is None else f"{number:.0f}"
def _fmt_percent(value: Any) -> str:
number = _number(value)
return "-" if number is None else f"{number:.1f}%"
def _text(value: Any) -> str:
if value is None:
return "-"
return escape(str(value))
def _resolve_sla(
reports: dict[str, dict[str, Any] | None],
key: str,
expected: float | None,
) -> float | None:
values: dict[str, float | None] = {}
for stage, label, _ in STAGES:
report = reports[stage]
if report is not None:
values[label] = _number((report.get("slo") or {}).get(key))
non_null = {value for value in values.values() if value is not None}
if expected is not None:
mismatched = {
label: value for label, value in values.items() if value != expected
}
if mismatched:
details = ", ".join(
f"{label}={value}" for label, value in mismatched.items()
)
raise ValueError(
f"Recommendation {key} SLA does not match the command-line "
f"value {expected}: {details}"
)
return expected
if len(non_null) > 1:
details = ", ".join(f"{label}={value}" for label, value in values.items())
raise ValueError(f"Recommendation {key} SLA values disagree: {details}")
return next(iter(non_null), None)
def _resolve_minimum_compliance(
reports: dict[str, dict[str, Any] | None],
) -> float | None:
values = {
_number(report.get("minimum_compliance_ratio"))
for report in reports.values()
if report is not None
}
non_null = {value for value in values if value is not None}
if len(non_null) > 1:
raise ValueError(
"Recommendation minimum-compliance values disagree: "
+ ", ".join(str(value) for value in sorted(non_null))
)
return next(iter(non_null), None)
def _table(headers: list[str], rows: list[list[Any]]) -> str:
if not rows:
return '<p class="muted">No completed candidates were found.</p>'
head = "".join(f"<th>{escape(header)}</th>" for header in headers)
body = []
for row in rows:
cells = "".join(f"<td>{_text(cell)}</td>" for cell in row)
body.append("<tr>" + cells + "</tr>")
return (
'<div class="table-wrap"><table><thead><tr>'
+ head
+ "</tr></thead><tbody>"
+ "".join(body)
+ "</tbody></table></div>"
)
def _status_badge(value: Any) -> str:
status = str(value or "unknown").replace("_", " ")
css_class = "pass" if value == "sla_feasible" else "warn"
return f'<span class="badge {css_class}">{escape(status)}</span>'
def _sla_label(candidate: dict[str, Any]) -> str:
if candidate.get("sla_eligible") is True:
return "PASS"
if candidate.get("sla_eligible") is False:
return "FAIL"
return "-"
def _sla_reason(
candidate: dict[str, Any],
*,
ttft_sla_ms: float | None,
tpot_sla_ms: float | None,
minimum_compliance: float | None,
) -> str:
if candidate.get("valid") is False or candidate.get("benchmark_valid") is False:
return str(candidate.get("invalid_reason") or "benchmark invalid")
reasons: list[str] = []
ttft = _number(candidate.get("median_p99_ttft_ms"))
tpot = _number(candidate.get("median_p99_tpot_ms"))
compliance = _number(candidate.get("combined_compliance_ratio"))
if ttft_sla_ms is not None and (ttft is None or ttft > ttft_sla_ms):
reasons.append(
"TTFT "
+ ("missing" if ttft is None else f"{ttft:.1f} ms > {ttft_sla_ms:.1f} ms")
)
if tpot_sla_ms is not None and (tpot is None or tpot > tpot_sla_ms):
reasons.append(
"TPOT "
+ ("missing" if tpot is None else f"{tpot:.1f} ms > {tpot_sla_ms:.1f} ms")
)
if minimum_compliance is not None and (
compliance is None or compliance < minimum_compliance
):
reasons.append(
"combined compliance "
+ (
"missing"
if compliance is None
else f"{compliance * 100:.2f}% < {minimum_compliance * 100:.2f}%"
)
)
return "PASS" if not reasons else "; ".join(reasons)
def _scheduler_value(candidate: dict[str, Any], key: str) -> str:
value = candidate.get(key)
return "vLLM default" if value is None else str(value)
def _selected_settings(
parallel: dict[str, Any] | None,
concurrency: dict[str, Any] | None,
scheduler: dict[str, Any] | None,
) -> list[list[Any]]:
parallel_rec = (parallel or {}).get("recommended") or {}
concurrency_rec = (concurrency or {}).get("recommended") or {}
scheduler_rec = (scheduler or {}).get("recommended") or {}
max_num_seqs = (
_scheduler_value(scheduler_rec, "max_num_seqs")
if scheduler is not None
else "-"
)
max_num_batched_tokens = (
_scheduler_value(scheduler_rec, "max_num_batched_tokens")
if scheduler is not None
else "-"
)
return [
["tensor-parallel-size", parallel_rec.get("tensor_parallel_size")],
["data-parallel-size", parallel_rec.get("data_parallel_size")],
["max concurrency", concurrency_rec.get("max_concurrency")],
["max-num-seqs", max_num_seqs],
["max-num-batched-tokens", max_num_batched_tokens],
]
def _stage_summary(
reports: dict[str, dict[str, Any] | None],
) -> list[list[Any]]:
rows = []
for stage, label, _ in STAGES:
report = reports[stage]
if report is None:
rows.append([label, "not available", "-", "-", "-", "-"])
continue
measured = report.get("measured") or {}
rows.append(
[
label,
str(report.get("status", "unknown")).replace("_", " "),
_fmt(measured.get("mean_output_throughput")),
_fmt(measured.get("mean_request_goodput"), 3),
_fmt_int(measured.get("mean_p99_ttft_ms")),
_fmt(measured.get("mean_p99_tpot_ms")),
]
)
return rows
def _parallel_rows(
report: dict[str, Any] | None,
*,
ttft_sla_ms: float | None,
tpot_sla_ms: float | None,
minimum_compliance: float | None,
) -> list[list[Any]]:
rows = []
for candidate in (report or {}).get("candidates", []):
rows.append(
[
candidate.get("tensor_parallel_size"),
candidate.get("data_parallel_size"),
candidate.get("max_num_seqs"),
candidate.get("max_num_batched_tokens"),
_fmt(candidate.get("mean_output_throughput")),
_fmt_int(candidate.get("mean_p99_ttft_ms")),
_fmt(candidate.get("mean_p99_tpot_ms")),
_fmt_percent(candidate.get("combined_compliance_percent")),
_sla_label(candidate),
_sla_reason(
candidate,
ttft_sla_ms=ttft_sla_ms,
tpot_sla_ms=tpot_sla_ms,
minimum_compliance=minimum_compliance,
),
]
)
return rows
def _concurrency_rows(
report: dict[str, Any] | None,
*,
ttft_sla_ms: float | None,
tpot_sla_ms: float | None,
minimum_compliance: float | None,
) -> list[list[Any]]:
candidates = (report or {}).get("candidates", [])
candidates = sorted(candidates, key=lambda item: item.get("max_concurrency", 0))
rows = []
for candidate in candidates:
rows.append(
[
candidate.get("max_concurrency"),
_fmt(candidate.get("mean_output_throughput")),
_fmt(candidate.get("mean_request_throughput"), 3),
_fmt(candidate.get("mean_request_goodput"), 3),
_fmt_percent(candidate.get("combined_compliance_percent")),
_fmt_int(candidate.get("mean_p99_ttft_ms")),
_fmt(candidate.get("mean_p99_tpot_ms")),
_sla_label(candidate),
_sla_reason(
candidate,
ttft_sla_ms=ttft_sla_ms,
tpot_sla_ms=tpot_sla_ms,
minimum_compliance=minimum_compliance,
),
]
)
return rows
def _scheduler_rows(
report: dict[str, Any] | None,
*,
ttft_sla_ms: float | None,
tpot_sla_ms: float | None,
minimum_compliance: float | None,
) -> list[list[Any]]:
rows = []
for candidate in (report or {}).get("candidates", []):
rows.append(
[
_scheduler_value(candidate, "max_num_seqs"),
_scheduler_value(candidate, "max_num_batched_tokens"),
_fmt(candidate.get("mean_output_throughput")),
_fmt(candidate.get("mean_request_goodput"), 3),
_fmt_int(candidate.get("mean_p99_ttft_ms")),
_fmt(candidate.get("mean_p99_tpot_ms")),
_fmt_percent(candidate.get("combined_compliance_percent")),
_sla_label(candidate),
_sla_reason(
candidate,
ttft_sla_ms=ttft_sla_ms,
tpot_sla_ms=tpot_sla_ms,
minimum_compliance=minimum_compliance,
),
]
)
return rows
def _metadata(runs: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
for stage, _, _ in STAGES:
if runs[stage]:
first = runs[stage][0]
return {
"model": first.get("model_id", "unknown"),
"input_tokens": first.get("random_input_len"),
"output_tokens": first.get("random_output_len"),
"request_rate": first.get("request_rate"),
}
return {
"model": "unknown",
"input_tokens": None,
"output_tokens": None,
"request_rate": None,
}
def _observations(
reports: dict[str, dict[str, Any] | None],
runs: dict[str, list[dict[str, Any]]],
) -> list[tuple[str, str]]:
observations: list[tuple[str, str]] = []
parallel = reports["parallel-layout"]
concurrency = reports["concurrency-tuning"]
scheduler = reports["runtime-tuning"]
parallel_candidates = (parallel or {}).get("candidates", [])
if len(parallel_candidates) == 1:
observations.append(
(
"warning",
(
"Only one parallel-layout candidate completed. The result "
"validates that layout but does not establish an optimum."
),
)
)
if parallel and concurrency and runs["parallel-layout"]:
representative = runs["parallel-layout"][0].get("max_concurrency")
concurrency_values = {
item.get("max_concurrency") for item in concurrency.get("candidates", [])
}
parallel_passed = parallel_candidates and parallel_candidates[0].get(
"p99_sla_eligible"
)
if representative not in concurrency_values and parallel_passed:
observations.append(
(
"warning",
(
f"Concurrency {representative} passed in the "
"parallel-layout stage but was not included in the "
"concurrency grid. Treat the automated concurrency "
"choice as conservative."
),
)
)
concurrency_candidates = (concurrency or {}).get("candidates", [])
first_failure = next(
(
item
for item in sorted(
concurrency_candidates,
key=lambda candidate: candidate.get("max_concurrency", 0),
)
if item.get("p99_sla_eligible") is False
),
None,
)
if first_failure and concurrency:
slo = concurrency.get("slo") or {}
ttft = _number(first_failure.get("mean_p99_ttft_ms"))
tpot = _number(first_failure.get("mean_p99_tpot_ms"))
ttft_slo = _number(slo.get("ttft_ms"))
tpot_slo = _number(slo.get("tpot_ms"))
if (
ttft
and ttft_slo
and ttft > ttft_slo
and tpot is not None
and (tpot_slo is None or tpot <= tpot_slo)
):
observations.append(
(
"info",
(
"TTFT is the first limiting SLA; TPOT remains within "
"its objective at the first failing concurrency."
),
)
)
scheduler_candidates = (scheduler or {}).get("candidates", [])
scheduler_throughputs = [
number
for number in (
_number(item.get("mean_output_throughput")) for item in scheduler_candidates
)
if number is not None
]
if len(scheduler_throughputs) > 1 and min(scheduler_throughputs) > 0:
spread = (max(scheduler_throughputs) / min(scheduler_throughputs) - 1) * 100
if spread <= 3:
observations.append(
(
"info",
(
"Scheduler output throughput is tightly grouped: "
f"{spread:.2f}% total spread across measured candidates."
),
)
)
if not observations:
observations.append(
("info", "No automatic coverage or SLA warnings were detected.")
)
return observations
def _observation_html(observations: list[tuple[str, str]]) -> str:
return "".join(
f'<div class="notice {escape(level)}">{escape(message)}</div>'
for level, message in observations
)
def generate_report(
root: Path,
output: Path,
title: str,
ttft_sla_ms: float | None = None,
tpot_sla_ms: float | None = None,
) -> None:
reports = {stage: _read_object(root / filename) for stage, _, filename in STAGES}
if not any(reports.values()):
names = ", ".join(filename for _, _, filename in STAGES)
raise ValueError(f"No recommendation files found under {root}: {names}")
ttft_sla_ms = _resolve_sla(reports, "ttft_ms", ttft_sla_ms)
tpot_sla_ms = _resolve_sla(reports, "tpot_ms", tpot_sla_ms)
minimum_compliance = _resolve_minimum_compliance(reports)
runs = {stage: _summary_runs(root / "results" / stage) for stage, _, _ in STAGES}
meta = _metadata(runs)
observations = _observations(reports, runs)
concurrency = reports["concurrency-tuning"] or {}
scheduler = reports["runtime-tuning"] or {}
measured = scheduler.get("measured") or concurrency.get("measured") or {}
recommended_concurrency = (concurrency.get("recommended") or {}).get(
"max_concurrency"
)
generated = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
styles = """
:root { color-scheme: light; --ink: #17324d; --blue: #146c94;
--teal: #2a9d8f; --green: #2e7d32; --amber: #d97706;
--red: #c0392b; --line: #d8dee4; --soft: #f3f5f7; }
* { box-sizing: border-box; }
body { margin: 0; background: #eef2f5; color: var(--ink);
font: 15px/1.45 Arial, sans-serif; }
main { max-width: 1180px; margin: 28px auto; padding: 0 20px 40px; }
header, section { background: white; border: 1px solid var(--line);
border-radius: 12px; margin-bottom: 18px; padding: 24px; }
h1 { margin: 0 0 6px; font-size: 30px; }
h2 { margin: 0 0 14px; font-size: 21px; }
h3 { margin: 20px 0 10px; font-size: 16px; color: var(--blue); }
.sub, .muted { color: #607486; }
.cards { display: grid; grid-template-columns: repeat(4, minmax(0, 1fr));
gap: 12px; margin-top: 20px; }
.card { background: var(--soft); border-radius: 9px; padding: 14px; }
.card strong { display: block; font-size: 22px; }
.card span { color: #607486; font-size: 12px; }
.badge { border-radius: 99px; display: inline-block; font-size: 12px;
font-weight: 700; padding: 4px 9px; text-transform: uppercase; }
.badge.pass { background: #eaf5ec; color: var(--green); }
.badge.warn { background: #fff4e5; color: var(--amber); }
.notice { border-left: 5px solid var(--blue); margin: 9px 0;
padding: 11px 14px; background: #eaf4f8; }
.notice.warning { border-color: var(--amber); background: #fff4e5; }
.table-wrap { overflow-x: auto; }
table { width: 100%; border-collapse: collapse; font-size: 13px; }
th { background: var(--ink); color: white; text-align: left; }
th, td { border: 1px solid var(--line); padding: 8px 9px; white-space: nowrap; }
tbody tr:nth-child(even) { background: var(--soft); }
footer { color: #607486; font-size: 12px; text-align: center; }
@media (max-width: 760px) { .cards { grid-template-columns: 1fr 1fr; } }
@media print { body { background: white; } main { margin: 0; max-width: none; }
header, section { border: 0; break-inside: avoid; padding: 12px 0; }
.table-wrap { overflow: visible; } }
"""
body = f"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{escape(title)}</title>
<style>{styles}</style>
</head>
<body><main>
<header>
<div class="sub">vLLM recipe sweep report</div>
<h1>{escape(title)}</h1>
<div class="sub">{_text(meta["model"])} | {_text(meta["input_tokens"])}
input / {_text(meta["output_tokens"])} output tokens |
request rate {_text(meta["request_rate"])}</div>
<div class="cards">
<div class="card"><strong>{_text(recommended_concurrency)}</strong>
<span>recommended concurrency</span></div>
<div class="card"><strong>{_fmt(measured.get("mean_output_throughput"))}</strong>
<span>output tokens/s</span></div>
<div class="card"><strong>{_fmt_int(measured.get("mean_p99_ttft_ms"))} ms</strong>
<span>mean p99 TTFT</span></div>
<div class="card"><strong>{_fmt(measured.get("mean_p99_tpot_ms"))} ms</strong>
<span>mean p99 TPOT</span></div>
<div class="card"><strong>{_fmt_int(ttft_sla_ms)} ms</strong>
<span>TTFT SLA</span></div>
<div class="card"><strong>{_fmt(tpot_sla_ms)} ms</strong>
<span>TPOT SLA</span></div>
<div class="card"><strong>{
_fmt_percent(
minimum_compliance * 100 if minimum_compliance is not None else None
)
}</strong>
<span>required combined compliance</span></div>
</div>
</header>
<section>
<h2>Recommended configuration</h2>
{
_table(
["Setting", "Value"],
_selected_settings(
reports["parallel-layout"],
reports["concurrency-tuning"],
reports["runtime-tuning"],
),
)
}
</section>
<section>
<h2>Stage summary</h2>
{
_table(
[
"Stage",
"Status",
"Output tok/s",
"Goodput req/s",
"p99 TTFT ms",
"p99 TPOT ms",
],
_stage_summary(reports),
)
}
<h3>Automatic observations</h3>
{_observation_html(observations)}
</section>
<section>
<h2>Parallel-layout candidates {
_status_badge((reports["parallel-layout"] or {}).get("status"))
}</h2>
{
_table(
[
"TP",
"DP",
"Max seqs",
"Batch tokens",
"Output tok/s",
"p99 TTFT ms",
"p99 TPOT ms",
"Compliance",
"SLA",
"SLA reason",
],
_parallel_rows(
reports["parallel-layout"],
ttft_sla_ms=ttft_sla_ms,
tpot_sla_ms=tpot_sla_ms,
minimum_compliance=minimum_compliance,
),
)
}
</section>
<section>
<h2>Concurrency candidates {
_status_badge((reports["concurrency-tuning"] or {}).get("status"))
}</h2>
{
_table(
[
"Concurrency",
"Output tok/s",
"Req/s",
"Goodput req/s",
"Compliance",
"p99 TTFT ms",
"p99 TPOT ms",
"SLA",
"SLA reason",
],
_concurrency_rows(
reports["concurrency-tuning"],
ttft_sla_ms=ttft_sla_ms,
tpot_sla_ms=tpot_sla_ms,
minimum_compliance=minimum_compliance,
),
)
}
</section>
<section>
<h2>Scheduler candidates {
_status_badge((reports["runtime-tuning"] or {}).get("status"))
}</h2>
{
_table(
[
"Max seqs",
"Batch tokens",
"Output tok/s",
"Goodput req/s",
"p99 TTFT ms",
"p99 TPOT ms",
"Compliance",
"SLA",
"SLA reason",
],
_scheduler_rows(
reports["runtime-tuning"],
ttft_sla_ms=ttft_sla_ms,
tpot_sla_ms=tpot_sla_ms,
minimum_compliance=minimum_compliance,
),
)
}
</section>
<footer>Generated {escape(generated)} from completed recommendation and
summary JSON files. Warmup results are excluded by the recommenders.</footer>
</main></body>
</html>
"""
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(body, encoding="utf-8")
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Generate a self-contained HTML report from recipe sweeps."
)
parser.add_argument(
"--sweep-dir",
default=None,
help="Sweep package directory (default: beside this script).",
)
parser.add_argument(
"--output",
default=None,
help="Output HTML path (default: <sweep-dir>/sweep-report.html).",
)
parser.add_argument(
"--title",
default="vLLM Recipe Sweep",
help="Report title.",
)
parser.add_argument(
"--ttft-sla-ms",
type=float,
default=DEFAULT_TTFT_SLA_MS,
help="Expected TTFT objective; must match every recommendation file.",
)
parser.add_argument(
"--tpot-sla-ms",
type=float,
default=DEFAULT_TPOT_SLA_MS,
help="Expected TPOT objective; must match every recommendation file.",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
for name in ("ttft_sla_ms", "tpot_sla_ms"):
value = getattr(args, name)
if value is not None and value <= 0:
option = "--" + name.replace("_", "-")
raise ValueError(f"{option} must be greater than zero.")
script_dir = Path(__file__).absolute().parent
root = Path(args.sweep_dir).resolve() if args.sweep_dir else script_dir
output = Path(args.output).resolve() if args.output else root / "sweep-report.html"
generate_report(
root,
output,
args.title,
ttft_sla_ms=args.ttft_sla_ms,
tpot_sla_ms=args.tpot_sla_ms,
)
print(f"Wrote HTML sweep report: {output}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -2,7 +2,7 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Select one measured scheduler recommendation from vLLM sweep results."""
"""Select one measured parallel-layout or scheduler sweep recommendation."""
from __future__ import annotations
@@ -24,6 +24,10 @@ DEFAULT_ENV_PATH: str | None = None
DEFAULT_TTFT_SLA_MS: float | None = None
DEFAULT_TPOT_SLA_MS: float | None = None
DEFAULT_MINIMUM_COMPLIANCE: float = 0.99
DEFAULT_THROUGHPUT_EQUIVALENCE_PERCENT: float = 1.0
DEFAULT_RESULTS_DIR = "results/runtime-tuning"
DEFAULT_OUTPUT_CONFIG = "recommended-config.yml"
DEFAULT_OUTPUT_JSON = "recommendation.json"
def _number(value: object) -> float | None:
@@ -48,11 +52,75 @@ def _percentile_summary(
return mean(numbers), median(numbers), max(numbers)
def _positive_int(row: dict[str, Any], key: str) -> int | None:
value = row.get(key)
def _scheduler_value(row: dict[str, Any], key: str) -> tuple[bool, int | None]:
if key not in row:
# Default-reference sweep candidates omit scheduler keys entirely.
return True, None
value = row[key]
if value is None:
return True, None
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
return None
return value
return False, None
return True, value
def _parallel_value(row: dict[str, Any], key: str, default: int) -> tuple[bool, int]:
if key not in row:
return True, default
value = row[key]
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
return False, default
return True, value
def _scheduler_sort_value(value: int | None) -> int:
return -1 if value is None else value
def _scheduler_preference_key(value: object) -> tuple[int, int]:
"""Prefer vLLM default, then the smaller explicit value on exact ties."""
if value is None:
return (1, 0)
if isinstance(value, int) and not isinstance(value, bool):
return (0, -value)
return (0, 0)
def _has_fixed_parallel_layout(candidates: list[dict[str, Any]]) -> bool:
"""Return whether candidates belong to one scheduler-only comparison."""
layouts = {
(candidate["tensor_parallel_size"], candidate["data_parallel_size"])
for candidate in candidates
}
return len(layouts) == 1
def _select_near_equivalent_scheduler_candidate(
candidates: list[dict[str, Any]],
*,
throughput_equivalence_percent: float,
) -> dict[str, Any]:
"""Prefer fewer scheduler overrides among near-equal configurations."""
best_throughput = max(
candidate["mean_output_throughput"] for candidate in candidates
)
throughput_floor = best_throughput * (1.0 - throughput_equivalence_percent / 100.0)
near_equivalent = [
candidate
for candidate in candidates
if candidate["mean_output_throughput"] >= throughput_floor
]
return max(
near_equivalent,
key=lambda candidate: (
len(candidate["vllm_default_parameters"]),
candidate["combined_compliance_ratio"] or 0.0,
candidate["mean_request_goodput"] or 0.0,
candidate["mean_output_throughput"],
_scheduler_preference_key(candidate["max_num_batched_tokens"]),
_scheduler_preference_key(candidate["max_num_seqs"]),
),
)
def _resolve(script_dir: Path, value: str | None) -> Path | None:
@@ -80,21 +148,40 @@ def _aggregate_candidates(
rows: list[dict[str, Any]],
*,
use_goodput: bool,
default_tensor_parallel_size: int = 1,
default_data_parallel_size: int = 1,
ttft_sla_ms: float | None = None,
tpot_sla_ms: float | None = None,
minimum_compliance: float = DEFAULT_MINIMUM_COMPLIANCE,
) -> list[dict[str, Any]]:
grouped: dict[tuple[int, int], list[dict[str, Any]]] = defaultdict(list)
grouped: dict[tuple[int, int, int | None, int | None], list[dict[str, Any]]] = (
defaultdict(list)
)
for row in rows:
seqs = _positive_int(row, "max_num_seqs")
batch = _positive_int(row, "max_num_batched_tokens")
if seqs is None or batch is None:
seqs_valid, seqs = _scheduler_value(row, "max_num_seqs")
batch_valid, batch = _scheduler_value(row, "max_num_batched_tokens")
tp_valid, tp = _parallel_value(
row, "tensor_parallel_size", default_tensor_parallel_size
)
dp_valid, dp = _parallel_value(
row, "data_parallel_size", default_data_parallel_size
)
if not seqs_valid or not batch_valid or not tp_valid or not dp_valid:
continue
grouped[(seqs, batch)].append(row)
grouped[(tp, dp, seqs, batch)].append(row)
candidates: list[dict[str, Any]] = []
for (seqs, batch), runs in sorted(grouped.items()):
sorted_groups = sorted(
grouped.items(),
key=lambda item: (
item[0][0],
item[0][1],
_scheduler_sort_value(item[0][2]),
_scheduler_sort_value(item[0][3]),
),
)
for (tp, dp, seqs, batch), runs in sorted_groups:
failed_requests = sum(int(_number(run.get("failed")) or 0) for run in runs)
output_throughput = _mean(runs, "output_throughput")
request_throughput = _mean(runs, "request_throughput")
@@ -150,9 +237,19 @@ def _aggregate_candidates(
candidates.append(
{
"tensor_parallel_size": tp,
"data_parallel_size": dp,
"max_num_seqs": seqs,
"max_num_batched_tokens": batch,
"run_count": len(runs),
"vllm_default_parameters": [
name
for name, value in (
("max_num_seqs", seqs),
("max_num_batched_tokens", batch),
)
if value is None
],
"failed_requests": failed_requests,
"mean_request_goodput": request_goodput,
"mean_request_throughput": request_throughput,
@@ -188,6 +285,7 @@ def _select_candidate(
candidates: list[dict[str, Any]],
*,
use_goodput: bool,
throughput_equivalence_percent: float = (DEFAULT_THROUGHPUT_EQUIVALENCE_PERCENT),
) -> tuple[dict[str, Any] | None, dict[str, Any], str]:
valid = [candidate for candidate in candidates if candidate["valid"]]
if not valid:
@@ -204,8 +302,10 @@ def _select_candidate(
candidate["mean_request_goodput"],
candidate["combined_compliance_ratio"] or 0.0,
candidate["mean_output_throughput"],
-candidate["max_num_batched_tokens"],
-candidate["max_num_seqs"],
-candidate["data_parallel_size"],
-candidate["tensor_parallel_size"],
_scheduler_preference_key(candidate["max_num_batched_tokens"]),
_scheduler_preference_key(candidate["max_num_seqs"]),
),
)
if best_effort["mean_request_goodput"] <= 0:
@@ -214,30 +314,46 @@ def _select_candidate(
"the supplied TTFT/TPOT objectives."
)
eligible = [candidate for candidate in valid if candidate["sla_eligible"]]
winner = (
max(
if eligible and _has_fixed_parallel_layout(valid):
winner = _select_near_equivalent_scheduler_candidate(
eligible,
key=lambda candidate: (
candidate["mean_output_throughput"],
candidate["combined_compliance_ratio"],
candidate["mean_request_goodput"],
-candidate["max_num_batched_tokens"],
-candidate["max_num_seqs"],
),
throughput_equivalence_percent=throughput_equivalence_percent,
)
else:
winner = (
max(
eligible,
key=lambda candidate: (
candidate["mean_output_throughput"],
candidate["combined_compliance_ratio"],
candidate["mean_request_goodput"],
-candidate["data_parallel_size"],
-candidate["tensor_parallel_size"],
_scheduler_preference_key(candidate["max_num_batched_tokens"]),
_scheduler_preference_key(candidate["max_num_seqs"]),
),
)
if eligible
else None
)
if eligible
else None
)
else:
objective = "highest_mean_output_throughput"
winner = max(
valid,
key=lambda candidate: (
candidate["mean_output_throughput"],
-candidate["max_num_batched_tokens"],
-candidate["max_num_seqs"],
),
)
if _has_fixed_parallel_layout(valid):
winner = _select_near_equivalent_scheduler_candidate(
valid,
throughput_equivalence_percent=throughput_equivalence_percent,
)
else:
winner = max(
valid,
key=lambda candidate: (
candidate["mean_output_throughput"],
-candidate["data_parallel_size"],
-candidate["tensor_parallel_size"],
_scheduler_preference_key(candidate["max_num_batched_tokens"]),
_scheduler_preference_key(candidate["max_num_seqs"]),
),
)
best_effort = winner
return winner, best_effort, objective
@@ -250,6 +366,31 @@ def _load_config(path: Path) -> dict[str, Any]:
return data
def _build_recommended_config(
initial_config: dict[str, Any],
winner: dict[str, Any],
) -> dict[str, Any]:
recommended_config = dict(initial_config)
recommended_config["tensor-parallel-size"] = winner["tensor_parallel_size"]
recommended_config["data-parallel-size"] = winner["data_parallel_size"]
for config_key, result_key in (
("max-num-seqs", "max_num_seqs"),
("max-num-batched-tokens", "max_num_batched_tokens"),
):
value = winner[result_key]
if value is None:
recommended_config.pop(config_key, None)
else:
recommended_config[config_key] = value
return recommended_config
def _format_scheduler_value(value: object) -> str:
if value is None:
return "vLLM default"
return str(value)
def _write_config(
path: Path,
*,
@@ -274,13 +415,13 @@ def _write_config(
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description=(
"Select one recommended max-num-seqs/max-num-batched-tokens pair "
"from generated vLLM sweep results."
"Select one recommended TP/DP layout and scheduler setting from "
"generated vLLM sweep results."
)
)
parser.add_argument(
"--results-dir",
default="results/runtime-tuning",
default=DEFAULT_RESULTS_DIR,
help="Sweep experiment directory (default: results/runtime-tuning).",
)
parser.add_argument(
@@ -314,14 +455,24 @@ def parse_args() -> argparse.Namespace:
"meet all supplied objectives (default: 0.99)."
),
)
parser.add_argument(
"--throughput-equivalence-percent",
type=float,
default=DEFAULT_THROUGHPUT_EQUIVALENCE_PERCENT,
help=(
"For a scheduler-only comparison, prefer fewer explicit scheduler "
"overrides when throughput is within this percentage of the best "
"eligible result (default: 1.0)."
),
)
parser.add_argument(
"--output-config",
default="recommended-config.yml",
default=DEFAULT_OUTPUT_CONFIG,
help="Recommended config output path.",
)
parser.add_argument(
"--output-json",
default="recommendation.json",
default=DEFAULT_OUTPUT_JSON,
help="Recommendation evidence output path.",
)
return parser.parse_args()
@@ -331,6 +482,10 @@ def main() -> int:
args = parse_args()
if not 0.0 < args.minimum_compliance <= 1.0:
raise ValueError("--minimum-compliance must be greater than 0 and at most 1.")
if not 0.0 <= args.throughput_equivalence_percent < 100.0:
raise ValueError(
"--throughput-equivalence-percent must be at least 0 and less than 100."
)
script_dir = Path(__file__).resolve().parent
config_path = _resolve(script_dir, args.config)
@@ -350,6 +505,7 @@ def main() -> int:
f"Sweep results were not found at {results_dir}. Run the sweep first."
)
initial_config = _load_config(config_path)
rows = _load_runs(results_dir)
if not rows:
raise ValueError(f"No summary.json sweep results found under {results_dir}.")
@@ -358,19 +514,20 @@ def main() -> int:
candidates = _aggregate_candidates(
rows,
use_goodput=use_goodput,
default_tensor_parallel_size=int(initial_config.get("tensor-parallel-size", 1)),
default_data_parallel_size=int(initial_config.get("data-parallel-size", 1)),
ttft_sla_ms=args.ttft_sla_ms,
tpot_sla_ms=args.tpot_sla_ms,
minimum_compliance=args.minimum_compliance,
)
winner, best_effort, objective = _select_candidate(
candidates, use_goodput=use_goodput
candidates,
use_goodput=use_goodput,
throughput_equivalence_percent=args.throughput_equivalence_percent,
)
initial_config = _load_config(config_path)
if winner is not None:
recommended_config = dict(initial_config)
recommended_config["max-num-seqs"] = winner["max_num_seqs"]
recommended_config["max-num-batched-tokens"] = winner["max_num_batched_tokens"]
recommended_config = _build_recommended_config(initial_config, winner)
_write_config(
output_config,
source_path=config_path,
@@ -384,8 +541,11 @@ def main() -> int:
recommended = None
if winner is not None:
recommended = {
"tensor_parallel_size": winner["tensor_parallel_size"],
"data_parallel_size": winner["data_parallel_size"],
"max_num_seqs": winner["max_num_seqs"],
"max_num_batched_tokens": winner["max_num_batched_tokens"],
"vllm_default_parameters": winner["vllm_default_parameters"],
}
measured = {
key: winner[key]
@@ -413,19 +573,25 @@ def main() -> int:
),
"selection_objective": objective,
"minimum_compliance_ratio": args.minimum_compliance,
"throughput_equivalence_percent": args.throughput_equivalence_percent,
"slo": {
"ttft_ms": args.ttft_sla_ms,
"tpot_ms": args.tpot_sla_ms,
},
"initial": {
"tensor_parallel_size": initial_config.get("tensor-parallel-size"),
"data_parallel_size": initial_config.get("data-parallel-size", 1),
"max_num_seqs": initial_config.get("max-num-seqs"),
"max_num_batched_tokens": initial_config.get("max-num-batched-tokens"),
},
"recommended": recommended,
"measured": measured,
"best_effort": {
"tensor_parallel_size": best_effort["tensor_parallel_size"],
"data_parallel_size": best_effort["data_parallel_size"],
"max_num_seqs": best_effort["max_num_seqs"],
"max_num_batched_tokens": best_effort["max_num_batched_tokens"],
"vllm_default_parameters": best_effort["vllm_default_parameters"],
"mean_request_goodput": best_effort["mean_request_goodput"],
"combined_compliance_ratio": best_effort["combined_compliance_ratio"],
"p99_sla_eligible": best_effort["p99_sla_eligible"],
@@ -445,8 +611,15 @@ def main() -> int:
print("Recommended runtime configuration")
print()
print(f" max-num-seqs: {winner['max_num_seqs']}")
print(f" max-num-batched-tokens: {winner['max_num_batched_tokens']}")
print(f" tensor-parallel-size: {winner['tensor_parallel_size']}")
print(f" data-parallel-size: {winner['data_parallel_size']}")
print(
" max-num-seqs: " + _format_scheduler_value(winner["max_num_seqs"])
)
print(
" max-num-batched-tokens: "
+ _format_scheduler_value(winner["max_num_batched_tokens"])
)
print()
print(f"Selection objective: {objective}")
if use_goodput:
+226
View File
@@ -0,0 +1,226 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Generate post-benchmark figures from completed recipe sweep results."""
from __future__ import annotations
import argparse
import json
import os
import shutil
import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any
@dataclass(frozen=True)
class PlotSpec:
var_x: str
var_y: str
fig_name: str
STAGE_PLOTS: dict[str, tuple[PlotSpec, ...]] = {
"parallel-layout": (
PlotSpec("recipe_candidate", "output_throughput", "parallel_throughput"),
PlotSpec("recipe_candidate", "p99_ttft_ms", "parallel_p99_ttft"),
PlotSpec("recipe_candidate", "p99_tpot_ms", "parallel_p99_tpot"),
),
"concurrency-tuning": (
PlotSpec("max_concurrency", "output_throughput", "concurrency_throughput"),
PlotSpec("max_concurrency", "p99_ttft_ms", "concurrency_p99_ttft"),
PlotSpec("max_concurrency", "p99_tpot_ms", "concurrency_p99_tpot"),
PlotSpec("output_throughput", "p99_ttft_ms", "throughput_p99_ttft"),
),
"runtime-tuning": (
PlotSpec("recipe_candidate", "output_throughput", "scheduler_throughput"),
PlotSpec("recipe_candidate", "p99_ttft_ms", "scheduler_p99_ttft"),
PlotSpec("recipe_candidate", "p99_tpot_ms", "scheduler_p99_tpot"),
),
}
def _serve_candidate(summary_path: Path, record: dict[str, Any]) -> str:
directory_name = summary_path.parent.name
prefix = "SERVE--"
separator = "-BENCH--"
if directory_name.startswith(prefix) and separator in directory_name:
return directory_name[len(prefix) :].split(separator, 1)[0]
benchmark_name = record.get("_benchmark_name")
if isinstance(benchmark_name, str) and benchmark_name:
return benchmark_name
return "fixed_configuration"
def _load_stage_records(stage_dir: Path) -> list[dict[str, Any]]:
records: list[dict[str, Any]] = []
for summary_path in sorted(stage_dir.rglob("summary.json")):
data = json.loads(summary_path.read_text(encoding="utf-8"))
if not isinstance(data, list):
continue
for record in data:
if not isinstance(record, dict):
continue
normalized = dict(record)
normalized["recipe_candidate"] = _serve_candidate(
summary_path,
normalized,
)
records.append(normalized)
return records
def _has_numeric_metric(records: list[dict[str, Any]], key: str) -> bool:
return any(
isinstance(record.get(key), (int, float))
and not isinstance(record.get(key), bool)
for record in records
)
def _run_plot(
normalized_dir: Path,
figure_dir: Path,
spec: PlotSpec,
*,
dry_run: bool,
) -> None:
command = [
"vllm",
"bench",
"sweep",
"plot",
str(normalized_dir),
"--fig-dir",
str(figure_dir.resolve()),
"--var-x",
spec.var_x,
"--var-y",
spec.var_y,
"--fig-name",
spec.fig_name,
]
if dry_run:
command.append("--dry-run")
environment = dict(os.environ)
environment.setdefault("MPLBACKEND", "Agg")
subprocess.run(command, check=True, env=environment)
def generate_figures(
results_dir: Path,
*,
selected_stages: list[str] | None = None,
dry_run: bool = False,
) -> list[Path]:
if shutil.which("vllm") is None:
raise RuntimeError(
"The vllm executable was not found. Run this analysis inside the "
"vLLM benchmark environment."
)
stages = selected_stages or list(STAGE_PLOTS)
discovered: list[Path] = []
with tempfile.TemporaryDirectory(prefix="vllm-recipe-plot-") as temp_dir:
temporary_root = Path(temp_dir)
for stage in stages:
stage_dir = results_dir / stage
if not stage_dir.is_dir():
continue
records = _load_stage_records(stage_dir)
if not records:
continue
discovered.append(stage_dir)
normalized_dir = temporary_root / stage
normalized_dir.mkdir(parents=True, exist_ok=True)
(normalized_dir / "summary.json").write_text(
json.dumps(records, indent=2) + "\n",
encoding="utf-8",
)
figure_dir = stage_dir / "figures"
plots = list(STAGE_PLOTS[stage])
if stage == "concurrency-tuning" and _has_numeric_metric(
records, "request_goodput"
):
plots.append(
PlotSpec(
"max_concurrency",
"request_goodput",
"concurrency_goodput",
)
)
for spec in plots:
_run_plot(
normalized_dir,
figure_dir,
spec,
dry_run=dry_run,
)
if not discovered:
requested = ", ".join(stages)
raise ValueError(
f"No completed sweep summaries found under {results_dir} "
f"for stages: {requested}."
)
return discovered
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Plot completed recipe sweep results without running a sweep."
)
parser.add_argument(
"--results-dir",
default=None,
help="Results root (default: results beside this script).",
)
parser.add_argument(
"--stage",
action="append",
choices=tuple(STAGE_PLOTS),
help="Plot only this stage; repeat to select multiple stages.",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Preview vLLM plot operations without writing figures.",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
script_dir = Path(__file__).resolve().parent
results_dir = (
Path(args.results_dir).resolve()
if args.results_dir is not None
else script_dir / "results"
)
stages = generate_figures(
results_dir,
selected_stages=args.stage,
dry_run=args.dry_run,
)
print()
if args.dry_run:
print("Visualization dry run completed for:")
else:
print("Wrote post-benchmark figures under:")
for stage_dir in stages:
print(f" {stage_dir / 'figures'}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
-358
View File
@@ -1,358 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Generate optional vLLM benchmark sweep files from one initial suggestion."""
from __future__ import annotations
import json
import os
import shlex
from pathlib import Path
from typing import Any
from runtime_tuning import WorkloadHints
def _positive_int(config: dict[str, Any], key: str) -> int:
value = config.get(key)
if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
raise ValueError(f"Sweep generation requires a positive initial {key!r} value.")
return value
def _strict_lower_power_of_two(value: int) -> int:
if value <= 1:
return 1
return 1 << ((value - 1).bit_length() - 1)
def _strict_upper_power_of_two(value: int) -> int:
return 1 << value.bit_length()
def validate_sweep_workload(workload: WorkloadHints) -> None:
required = {
"--input-tokens": workload.input_tokens,
"--output-tokens": workload.output_tokens,
"--concurrency": workload.concurrency,
}
missing = [name for name, value in required.items() if value is None]
if missing:
raise ValueError("--generate-sweep requires " + ", ".join(missing) + ".")
def build_serve_params(config: dict[str, Any]) -> list[dict[str, Any]]:
"""Build a diverse, bounded sweep around the initial suggestion.
The initial five-point sweep changed only one parameter at a time and
skipped the useful middle sequence count. Eight directed points cover the
batch-budget curve at full concurrency plus interactions at 3/4 and 1/2 of
the initial scheduler concurrency without paying for a full Cartesian grid.
"""
initial_seqs = _positive_int(config, "max-num-seqs")
initial_batch = _positive_int(config, "max-num-batched-tokens")
minimum_batch = initial_seqs
if config.get("enable-chunked-prefill") is False:
max_model_len = config.get("max-model-len")
if (
isinstance(max_model_len, int)
and not isinstance(max_model_len, bool)
and max_model_len > 0
):
minimum_batch = max(minimum_batch, max_model_len)
lower_seqs = max(1, (initial_seqs + 1) // 2)
middle_seqs = max(lower_seqs, (3 * initial_seqs + 3) // 4)
lower_batch = max(
minimum_batch,
_strict_lower_power_of_two(initial_batch),
)
smaller_batch = max(
minimum_batch,
_strict_lower_power_of_two(lower_batch),
)
higher_batch = max(
minimum_batch,
_strict_upper_power_of_two(initial_batch),
)
candidates: list[dict[str, Any]] = []
seen: set[tuple[int, int]] = set()
def add(name: str, max_num_seqs: int, max_num_batched_tokens: int) -> None:
max_num_batched_tokens = max(
max_num_batched_tokens,
max_num_seqs,
)
signature = (max_num_seqs, max_num_batched_tokens)
if signature in seen:
return
seen.add(signature)
candidates.append(
{
"_benchmark_name": name,
"max_num_seqs": max_num_seqs,
"max_num_batched_tokens": max_num_batched_tokens,
}
)
# Keep the exact initial suggestion as the measured baseline. Additional
# values exist only in the optional sweep package.
add("initial", initial_seqs, initial_batch)
add("smaller_batch_budget", initial_seqs, smaller_batch)
add("lower_batch_budget", initial_seqs, lower_batch)
add("higher_batch_budget", initial_seqs, higher_batch)
add("middle_seqs_lower_batch", middle_seqs, lower_batch)
add("middle_seqs_higher_batch", middle_seqs, higher_batch)
add("lower_seqs_lower_batch", lower_seqs, lower_batch)
add("lower_seqs_higher_batch", lower_seqs, higher_batch)
return candidates
def build_bench_params(workload: WorkloadHints) -> list[dict[str, Any]]:
validate_sweep_workload(workload)
assert workload.input_tokens is not None
assert workload.output_tokens is not None
assert workload.concurrency is not None
return [
{
"_benchmark_name": "user_workload",
"random_input_len": workload.input_tokens,
"random_output_len": workload.output_tokens,
"max_concurrency": workload.concurrency,
}
]
def _benchmark_models(config: dict[str, Any]) -> tuple[str, str]:
model = config.get("model")
if not isinstance(model, str) or not model:
raise ValueError("Sweep generation requires a model in config.yml.")
served_model_name = config.get("served-model-name")
if isinstance(served_model_name, str) and served_model_name:
return served_model_name, model
if (
isinstance(served_model_name, list)
and served_model_name
and isinstance(served_model_name[0], str)
):
return served_model_name[0], model
return model, model
def _relative_to(directory: Path, target: str) -> str:
return os.path.relpath(Path(target).resolve(), directory.resolve())
def _write_json(path: Path, value: object) -> None:
path.write_text(
json.dumps(value, indent=2) + "\n",
encoding="utf-8",
)
def _write_run_script(
path: Path,
*,
config_rel: str,
env_rel: str,
request_model: str,
tokenizer: str,
workload: WorkloadHints,
) -> None:
bench_parts = [
"vllm bench serve",
"--backend vllm",
f"--model {shlex.quote(request_model)}",
f"--tokenizer {shlex.quote(tokenizer)}",
"--dataset-name random",
"--request-rate inf",
"--ignore-eos",
"--metric-percentiles 99",
]
goodput_pairs: list[str] = []
if workload.ttft_sla_ms is not None:
goodput_pairs.append(f"ttft:{workload.ttft_sla_ms:g}")
if workload.tpot_sla_ms is not None:
goodput_pairs.append(f"tpot:{workload.tpot_sla_ms:g}")
if goodput_pairs:
bench_parts.append(
"--goodput " + " ".join(shlex.quote(pair) for pair in goodput_pairs)
)
bench_cmd = " ".join(bench_parts)
script = f"""#!/usr/bin/env bash
set -euo pipefail
SCRIPT_DIR="$(cd -- "$(dirname -- "${{BASH_SOURCE[0]}}")" && pwd)"
CONFIG_PATH="${{SCRIPT_DIR}}/{config_rel}"
ENV_PATH="${{SCRIPT_DIR}}/{env_rel}"
source "${{ENV_PATH}}"
vllm bench sweep serve \
--serve-cmd "vllm serve --config '${{CONFIG_PATH}}'" \
--bench-cmd "{bench_cmd}" \
--serve-params "${{SCRIPT_DIR}}/serve_params.json" \
--bench-params "${{SCRIPT_DIR}}/bench_params.json" \
--output-dir "${{SCRIPT_DIR}}/results" \
--experiment-name runtime-tuning \
"$@"
"""
path.write_text(script, encoding="utf-8")
path.chmod(path.stat().st_mode | 0o111)
def _write_recommend_script(
path: Path,
*,
config_rel: str,
env_rel: str,
workload: WorkloadHints,
) -> None:
template_path = Path(__file__).with_name("sweep_recommendation.py")
source = template_path.read_text(encoding="utf-8")
replacements = {
"DEFAULT_CONFIG_PATH: str | None = None": (
f"DEFAULT_CONFIG_PATH: str | None = {config_rel!r}"
),
"DEFAULT_ENV_PATH: str | None = None": (
f"DEFAULT_ENV_PATH: str | None = {env_rel!r}"
),
"DEFAULT_TTFT_SLA_MS: float | None = None": (
f"DEFAULT_TTFT_SLA_MS: float | None = {workload.ttft_sla_ms!r}"
),
"DEFAULT_TPOT_SLA_MS: float | None = None": (
f"DEFAULT_TPOT_SLA_MS: float | None = {workload.tpot_sla_ms!r}"
),
}
for marker, replacement in replacements.items():
if marker not in source:
raise ValueError(f"Recommender template marker not found: {marker}")
source = source.replace(marker, replacement, 1)
path.write_text(source, encoding="utf-8")
path.chmod(path.stat().st_mode | 0o111)
def _write_guide(path: Path, workload: WorkloadHints) -> None:
sla_lines: list[str] = []
if workload.ttft_sla_ms is not None:
sla_lines.append(f"- TTFT objective: `{workload.ttft_sla_ms:g} ms`")
if workload.tpot_sla_ms is not None:
sla_lines.append(f"- TPOT objective: `{workload.tpot_sla_ms:g} ms`")
sla_text = "\n".join(sla_lines)
if sla_text:
sla_text = (
"\n## Supplied latency objectives\n\n"
+ sla_text
+ "\n\nThe generated benchmark uses these values with vLLM "
"`--goodput`.\n"
)
content = f"""# Optional Runtime Tuning Sweep
The generated `config.yml` is the **single initial suggestion** and can be
deployed directly. The sweep benchmarks nearby values for:
- `max-num-seqs`
- `max-num-batched-tokens`
## Run
```bash
./run_sweep.sh --dry-run
./run_sweep.sh
./recommend.py
```
For a quick one-run experiment:
```bash
rm -rf results/runtime-tuning
./run_sweep.sh --num-runs 1
./recommend.py
```
Resume an interrupted sweep:
```bash
./run_sweep.sh --resume
```
By default, vLLM benchmarks each parameter combination three times.
{sla_text}
## Outputs
`recommend.py` writes:
```text
recommended-config.yml
recommendation.json
```
With TTFT/TPOT objectives it requires duration-weighted combined compliance of
at least 99% plus median P99 TTFT/TPOT compliance, then selects highest mean
output-token throughput. If no candidate qualifies, it records a best-effort
candidate but does not write `recommended-config.yml`. Without latency
objectives it selects highest mean output-token throughput. Configurations with
failed requests are excluded.
`recommended-config.yml` copies the initial configuration and changes only
`max-num-seqs` and `max-num-batched-tokens`.
"""
path.write_text(content, encoding="utf-8")
def write_sweep_files(
output_dir: str,
*,
config_path: str,
env_path: str,
config: dict[str, Any],
workload: WorkloadHints,
) -> list[Path]:
"""Write an optional sweep package around the initial config suggestion."""
validate_sweep_workload(workload)
directory = Path(output_dir)
directory.mkdir(parents=True, exist_ok=True)
serve_params = directory / "serve_params.json"
bench_params = directory / "bench_params.json"
run_script = directory / "run_sweep.sh"
recommend_script = directory / "recommend.py"
guide = directory / "SWEEP.md"
config_rel = _relative_to(directory, config_path)
env_rel = _relative_to(directory, env_path)
_write_json(serve_params, build_serve_params(config))
_write_json(bench_params, build_bench_params(workload))
request_model, tokenizer = _benchmark_models(config)
_write_run_script(
run_script,
config_rel=config_rel,
env_rel=env_rel,
request_model=request_model,
tokenizer=tokenizer,
workload=workload,
)
_write_recommend_script(
recommend_script,
config_rel=config_rel,
env_rel=env_rel,
workload=workload,
)
_write_guide(guide, workload)
return [serve_params, bench_params, run_script, recommend_script, guide]