| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | """Optional runtime refinement for configs produced from vLLM Recipes. |
| 5 | |
| 6 | The recipe remains the baseline. Each policy may return an override only when |
| 7 | enough information is available. Keep policies independent so the tuning |
| 8 | algorithms can evolve without changing the converter or hardware detector. |
| 9 | |
| 10 | The policies in this draft are intentionally simple heuristics, not benchmark- |
| 11 | proven universal optima. |
| 12 | """ |
| 13 | |
| 14 | from __future__ import annotations |
| 15 | |
| 16 | from collections.abc import Callable |
| 17 | from dataclasses import dataclass, field |
| 18 | from math import ceil |
| 19 | from typing import Any |
| 20 | |
| 21 | from hardware_detection import HardwareInfo |
| 22 | |
| 23 | from vllm.config.scheduler import SchedulerConfig |
| 24 | |
| 25 | |
| 26 | @dataclass(frozen=True) |
| 27 | class WorkloadHints: |
| 28 | input_tokens: int | None = None |
| 29 | output_tokens: int | None = None |
| 30 | concurrency: int | None = None |
| 31 | ttft_sla_ms: float | None = None |
| 32 | tpot_sla_ms: float | None = None |
| 33 | target_qps: float | None = None |
| 34 | |
| 35 | def validate(self) -> None: |
| 36 | for name in ("input_tokens", "output_tokens", "concurrency"): |
| 37 | value = getattr(self, name) |
| 38 | if value is not None and value <= 0: |
| 39 | raise ValueError(f"{name} must be greater than zero") |
| 40 | |
| 41 | for name in ("ttft_sla_ms", "tpot_sla_ms", "target_qps"): |
| 42 | value = getattr(self, name) |
| 43 | if value is not None and value <= 0: |
| 44 | raise ValueError(f"{name} must be greater than zero") |
| 45 | |
| 46 | |
| 47 | @dataclass |
| 48 | class TuningResult: |
| 49 | overrides: dict[str, Any] = field(default_factory=dict) |
| 50 | notes: list[str] = field(default_factory=list) |
| 51 | |
| 52 | |
| 53 | Policy = Callable[ |
| 54 | [dict[str, Any], HardwareInfo | None, WorkloadHints, TuningResult], |
| 55 | None, |
| 56 | ] |
| 57 | |
| 58 | |
| 59 | def _record_override( |
| 60 | result: TuningResult, |
| 61 | key: str, |
| 62 | value: Any, |
| 63 | reason: str, |
| 64 | ) -> None: |
| 65 | result.overrides[key] = value |
| 66 | result.notes.append(f"{key}={value!r}: {reason}") |
| 67 | |
| 68 | |
| 69 | def _get_active_sequence_count( |
| 70 | config: dict[str, Any], |
| 71 | workload: WorkloadHints, |
| 72 | ) -> int: |
| 73 | """Return the sequence count used to size one scheduler iteration.""" |
| 74 | if workload.concurrency is not None: |
| 75 | return workload.concurrency |
| 76 | |
| 77 | configured = config.get("max-num-seqs") |
| 78 | if ( |
| 79 | isinstance(configured, int) |
| 80 | and not isinstance(configured, bool) |
| 81 | and configured > 0 |
| 82 | ): |
| 83 | return configured |
| 84 | |
| 85 | return SchedulerConfig.DEFAULT_MAX_NUM_SEQS |
| 86 | |
| 87 | |
| 88 | def _estimate_prefills_per_step( |
| 89 | workload: WorkloadHints, |
| 90 | active_sequences: int, |
| 91 | ) -> float: |
| 92 | """Estimate steady-state prompt arrivals per decode scheduler step.""" |
| 93 | # Reserve enough prefill budget for at least one representative prompt. |
| 94 | prefills_per_step = 1.0 |
| 95 | |
| 96 | # In steady state, active_sequences / output_tokens approximates how many |
| 97 | # requests finish and are replaced by new prefills per decode step. |
| 98 | if workload.output_tokens is not None: |
| 99 | prefills_per_step = max( |
| 100 | prefills_per_step, |
| 101 | active_sequences / workload.output_tokens, |
| 102 | ) |
| 103 | |
| 104 | # With target QPS + TPOT, approximate how many prompts arrive during one |
| 105 | # decode scheduler-step interval. |
| 106 | if workload.target_qps is not None and workload.tpot_sla_ms is not None: |
| 107 | prefills_per_step = max( |
| 108 | prefills_per_step, |
| 109 | workload.target_qps * workload.tpot_sla_ms / 1000.0, |
| 110 | ) |
| 111 | |
| 112 | # A step cannot replace more requests than are active. |
| 113 | return min(float(active_sequences), prefills_per_step) |
| 114 | |
| 115 | |
| 116 | def _resolve_tensor_parallel_size( |
| 117 | config: dict[str, Any], |
| 118 | hardware: HardwareInfo | None, |
| 119 | workload: WorkloadHints, |
| 120 | result: TuningResult, |
| 121 | ) -> None: |
| 122 | del config, workload |
| 123 | if hardware is None or hardware.numa_node_count < 1: |
| 124 | return |
| 125 | |
| 126 | # Draft policy: use the largest power-of-two TP that does not exceed the |
| 127 | # effective NUMA-node count. This deliberately avoids hard-coding unusual |
| 128 | # non-power-of-two topologies into the converter. |
| 129 | candidate = 1 << (hardware.numa_node_count.bit_length() - 1) |
| 130 | _record_override( |
| 131 | result, |
| 132 | "tensor-parallel-size", |
| 133 | candidate, |
| 134 | "derived from the effective NUMA-node count", |
| 135 | ) |
| 136 | |
| 137 | |
| 138 | def _resolve_gpu_memory_utilization( |
| 139 | config: dict[str, Any], |
| 140 | hardware: HardwareInfo | None, |
| 141 | workload: WorkloadHints, |
| 142 | result: TuningResult, |
| 143 | ) -> None: |
| 144 | del workload |
| 145 | if hardware is None or not hardware.numa_nodes: |
| 146 | return |
| 147 | |
| 148 | reserve_fraction = 0.10 |
| 149 | safe_fractions = [ |
| 150 | max( |
| 151 | 0.0, |
| 152 | node.available_memory_bytes / node.total_memory_bytes - reserve_fraction, |
| 153 | ) |
| 154 | for node in hardware.numa_nodes |
| 155 | if node.total_memory_bytes > 0 |
| 156 | ] |
| 157 | if not safe_fractions: |
| 158 | return |
| 159 | |
| 160 | safe_fraction = round(min(0.90, min(safe_fractions)), 2) |
| 161 | if safe_fraction <= 0: |
| 162 | return |
| 163 | |
| 164 | recipe_value = config.get("gpu-memory-utilization") |
| 165 | if isinstance(recipe_value, (int, float)): |
| 166 | candidate = min(float(recipe_value), safe_fraction) |
| 167 | else: |
| 168 | candidate = min(0.80, safe_fraction) |
| 169 | |
| 170 | _record_override( |
| 171 | result, |
| 172 | "gpu-memory-utilization", |
| 173 | round(candidate, 2), |
| 174 | "constrained by effective per-NUMA available memory", |
| 175 | ) |
| 176 | |
| 177 | |
| 178 | def _resolve_max_num_seqs( |
| 179 | config: dict[str, Any], |
| 180 | hardware: HardwareInfo | None, |
| 181 | workload: WorkloadHints, |
| 182 | result: TuningResult, |
| 183 | ) -> None: |
| 184 | del config, hardware |
| 185 | if workload.concurrency is None: |
| 186 | return |
| 187 | |
| 188 | _record_override( |
| 189 | result, |
| 190 | "max-num-seqs", |
| 191 | workload.concurrency, |
| 192 | "derived from the optional target concurrency", |
| 193 | ) |
| 194 | |
| 195 | |
| 196 | def _resolve_max_num_batched_tokens( |
| 197 | config: dict[str, Any], |
| 198 | hardware: HardwareInfo | None, |
| 199 | workload: WorkloadHints, |
| 200 | result: TuningResult, |
| 201 | ) -> None: |
| 202 | del hardware |
| 203 | if workload.input_tokens is None: |
| 204 | return |
| 205 | |
| 206 | active_sequences = _get_active_sequence_count(config, workload) |
| 207 | |
| 208 | # Decode requests consume roughly one token per active sequence in one |
| 209 | # scheduler iteration. Prefills share the remaining scheduler token budget. |
| 210 | decode_budget = active_sequences |
| 211 | prefills_per_step = _estimate_prefills_per_step(workload, active_sequences) |
| 212 | prefill_budget = ceil(workload.input_tokens * prefills_per_step) |
| 213 | |
| 214 | # Use vLLM's scheduler defaults/constraints as floors rather than local |
| 215 | # recipe-specific magic constants. |
| 216 | candidate = max( |
| 217 | SchedulerConfig.DEFAULT_MAX_NUM_BATCHED_TOKENS, |
| 218 | active_sequences, |
| 219 | decode_budget + prefill_budget, |
| 220 | ) |
| 221 | |
| 222 | # Chunked prefill normally lets a prompt span scheduler iterations. If the |
| 223 | # recipe explicitly disables it, vLLM requires the token budget to cover |
| 224 | # max_model_len when max_model_len is known. |
| 225 | if config.get("enable-chunked-prefill") is False: |
| 226 | max_model_len = config.get("max-model-len") |
| 227 | if ( |
| 228 | isinstance(max_model_len, int) |
| 229 | and not isinstance(max_model_len, bool) |
| 230 | and max_model_len > 0 |
| 231 | ): |
| 232 | candidate = max(candidate, max_model_len) |
| 233 | |
| 234 | _record_override( |
| 235 | result, |
| 236 | "max-num-batched-tokens", |
| 237 | candidate, |
| 238 | ( |
| 239 | "workload-derived scheduler budget " |
| 240 | f"(active_sequences={active_sequences}, " |
| 241 | f"prefills_per_step={prefills_per_step:.2f}, " |
| 242 | f"decode_budget={decode_budget}, " |
| 243 | f"prefill_budget={prefill_budget})" |
| 244 | ), |
| 245 | ) |
| 246 | |
| 247 | |
| 248 | def _resolve_data_parallel_size( |
| 249 | config: dict[str, Any], |
| 250 | hardware: HardwareInfo | None, |
| 251 | workload: WorkloadHints, |
| 252 | result: TuningResult, |
| 253 | ) -> None: |
| 254 | del config, hardware |
| 255 | if workload.target_qps is None: |
| 256 | return |
| 257 | |
| 258 | result.notes.append( |
| 259 | "data-parallel-size: target QPS supplied; kept recipe value because " |
| 260 | "per-replica capacity is not known yet" |
| 261 | ) |
| 262 | |
| 263 | |
| 264 | DEFAULT_POLICIES: tuple[Policy, ...] = ( |
| 265 | _resolve_tensor_parallel_size, |
| 266 | _resolve_gpu_memory_utilization, |
| 267 | _resolve_max_num_seqs, |
| 268 | _resolve_max_num_batched_tokens, |
| 269 | _resolve_data_parallel_size, |
| 270 | ) |
| 271 | |
| 272 | # Select tuning behavior from the hardware requested by the Recipes rendering, |
| 273 | # not from devices physically present in the host. A GPU server still exposes |
| 274 | # its host CPU topology, so host inspection alone cannot determine deployment |
| 275 | # intent. |
| 276 | HARDWARE_TUNING_POLICIES: dict[str, tuple[Policy, ...]] = { |
| 277 | "xeon6": DEFAULT_POLICIES, |
| 278 | } |
| 279 | |
| 280 | |
| 281 | def get_runtime_tuning_policies( |
| 282 | recipe_hardware: object, |
| 283 | ) -> tuple[Policy, ...]: |
| 284 | if not isinstance(recipe_hardware, str) or not recipe_hardware: |
| 285 | raise ValueError( |
| 286 | "Runtime tuning requires the resolved Recipes JSON to declare " |
| 287 | "a non-empty `hardware` field." |
| 288 | ) |
| 289 | |
| 290 | policies = HARDWARE_TUNING_POLICIES.get(recipe_hardware) |
| 291 | if policies is None: |
| 292 | supported = ", ".join(sorted(HARDWARE_TUNING_POLICIES)) |
| 293 | raise ValueError( |
| 294 | "Runtime tuning is not supported for recipe hardware " |
| 295 | f"{recipe_hardware!r}. Currently supported: {supported}." |
| 296 | ) |
| 297 | return policies |
| 298 | |
| 299 | |
| 300 | def finetune_runtime_config( |
| 301 | config: dict[str, Any], |
| 302 | *, |
| 303 | hardware: HardwareInfo | None = None, |
| 304 | workload: WorkloadHints | None = None, |
| 305 | policies: tuple[Policy, ...] = DEFAULT_POLICIES, |
| 306 | ) -> TuningResult: |
| 307 | """Return overrides; never mutate the recipe-derived config in place.""" |
| 308 | hints = workload or WorkloadHints() |
| 309 | hints.validate() |
| 310 | |
| 311 | result = TuningResult() |
| 312 | for policy in policies: |
| 313 | policy(config, hardware, hints, result) |
| 314 | return result |